johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package eventstream

import (
	"cmp"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/identity"
	"github.com/bluesky-social/indigo/atproto/syntax"
	cid "github.com/ipfs/go-cid"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"

	v1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/pdsclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/events"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/events/relay"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/kcp"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/statesync"
)

const (
	marketBidderMarketInbox = "bidder.market-inbox"

	marketRequesterMarketInbox = "requester.market-inbox"

	marketRequesterBidInbox = "requester.bid-inbox"

	marketInboxLabel = "socialweb.computer/market-inbox"

	marketCollectionLabel = "socialweb.computer/market-collection"

	marketNamePrefix = "market-"

	MarketDedupWindow = 15 * time.Second

	MarketDedupCap = 10000
)

type DynamicFactory = kube.Factory

func newDirectory(httpClient *http.Client, plcURL, workspace string) identity.Directory {
	return directoryAt(httpClient, cmp.Or(strings.TrimSpace(plcURL), "http://"+kcp.ServiceFQDN("plc", "default", workspace, kcp.DefaultServiceDomain)))
}

type MarketOptions struct {
	Workspace string

	Role string

	Dynamic DynamicFactory

	Resolver *pdsclient.Resolver

	PLCDirectoryURL string

	Context context.Context

	Log *slog.Logger
}

func MarketHandlerName(role, workspace string) string {
	return marketNamePrefix + strings.ToLower(strings.TrimSpace(role)) + "@" + strings.TrimSpace(workspace)
}

type marketHandler struct {
	opts MarketOptions

	log *slog.Logger

	directory identity.Directory

	client *http.Client

	mu    sync.Mutex
	pdses map[syntax.DID]pdsclient.PDS

	seen *seenSet

	addressees *addressees
}

type marketPayload struct {
	DID string `json:"did"`

	URI string `json:"uri"`

	CID string `json:"cid"`

	Collection string `json:"collection"`

	Record map[string]any `json:"record"`
}

var _ events.Sink = (*MarketSink)(nil)

func MarketRole(name string) (string, bool) {
	rest, ok := strings.CutPrefix(strings.TrimSpace(name), marketNamePrefix)
	if !ok {
		return "", false
	}
	role, _, _ := strings.Cut(rest, "@")
	switch role {
	case "bidder", "requester":
		return role, true
	}
	return "", false
}

type MarketSink = marketHandler

func NewMarketSink(opts MarketOptions) (*MarketSink, error) {
	opts.Workspace = strings.TrimSpace(opts.Workspace)
	if opts.Workspace == "" {
		return nil, errors.New("eventstream: a market handler needs a workspace")
	}
	opts.Role = strings.ToLower(strings.TrimSpace(opts.Role))
	switch opts.Role {
	case "bidder", "requester":
	default:
		return nil, fmt.Errorf("eventstream: unknown market role %q", opts.Role)
	}
	if opts.Dynamic == nil {
		return nil, errors.New("eventstream: a market handler needs a dynamic client factory")
	}
	if opts.Resolver == nil {
		return nil, errors.New("eventstream: a market handler needs a pdsclient resolver")
	}
	if opts.Log == nil {
		opts.Log = slog.New(slog.DiscardHandler)
	}
	if opts.Context == nil {
		return nil, errors.New("eventstream: a market handler needs the run context")
	}
	addressees, err := newAddressees(opts.Context, opts.Role, opts.Workspace, opts.Dynamic, opts.Log)
	if err != nil {
		return nil, fmt.Errorf("eventstream: the market addressees of %s: %w", opts.Workspace, err)
	}
	httpClient := httpClientWithFallback(opts.Resolver.HTTPClient, nil)
	return &marketHandler{
		opts:       opts,
		log:        opts.Log,
		directory:  newDirectory(httpClient, opts.PLCDirectoryURL, opts.Workspace),
		client:     httpClient,
		pdses:      map[syntax.DID]pdsclient.PDS{},
		seen:       newSeenSet(MarketDedupWindow, MarketDedupCap),
		addressees: addressees,
	}, nil
}

func (h *marketHandler) Deliver(ctx context.Context, event events.Event) error {
	repo, err := syntax.ParseDID(event.Repo)
	if err != nil {
		return fmt.Errorf("eventstream: the event repo %q is not a DID: %w", event.Repo, err)
	}
	for _, op := range event.Ops {
		switch strings.ToLower(strings.TrimSpace(op.Action)) {
		case "create", "update":
		default:
			continue
		}
		collection, rkey, ok := splitPath(op.Path)
		if !ok {
			continue
		}
		adapter := marketAdapter(h.opts.Role, collection)
		if adapter == "" {
			h.addressees.skipped(collection)
			continue
		}
		if !h.addressees.wants(ctx, collection, event.Repo) {
			if err := ctx.Err(); err != nil {
				return err
			}
			continue
		}
		uri := "at://" + event.Repo + "/" + collection + "/" + rkey
		if h.addressees.closed(event.Source, collection, uri, nil) {
			continue
		}
		if op.CID == "" {
			if err := h.deliverOp(ctx, event, adapter, repo, uri, collection, rkey, op); err != nil {
				return err
			}
			continue
		}
		if err := h.deliverOnce(ctx, event, adapter, repo, uri, collection, rkey, op); err != nil {
			return err
		}
	}
	return nil
}

func (h *marketHandler) deliverOnce(ctx context.Context, event events.Event, adapter string, repo syntax.DID, uri, collection, rkey string, op events.Op) error {
	key := uri + "|" + op.CID
	for {
		entry, owner := h.seen.claim(key, event.Source, time.Now())
		if owner {
			err := h.deliverOp(ctx, event, adapter, repo, uri, collection, rkey, op)
			h.seen.finish(key, entry, err == nil)
			return err
		}
		select {
		case <-entry.done:
		case <-ctx.Done():
			return ctx.Err()
		}
		if !entry.ok {
			continue
		}
		if dup := h.seen.drop(entry); dup.Drops == 1 {
			h.log.Info("market_event_duplicate_dropped", "role", h.opts.Role, "workspace", h.opts.Workspace, "uri", uri, "cid", op.CID,
				"source", event.Source, "first", dup.First, "dropped", dup.Dropped)
		}
		return nil
	}
}

func (h *marketHandler) deliverOp(ctx context.Context, event events.Event, adapter string, repo syntax.DID, uri, collection, rkey string, op events.Op) error {
	var expected *cid.Cid
	if op.CID != "" {
		parsed, err := cid.Decode(op.CID)
		if err != nil {
			return fmt.Errorf("eventstream: the market op %s has an invalid cid: %w", uri, err)
		}
		expected = &parsed
	}
	value, recordCID, err := h.fetchMarketRecord(ctx, event.Source, repo, collection, rkey, expected)
	if err != nil {
		return fmt.Errorf("eventstream: fetching the market record %s: %w", uri, err)
	}
	if h.addressees.closed(event.Source, collection, uri, value) {
		return nil
	}
	owner, ok := h.addressees.owner(ctx, collection, event.Repo, value, func(ctx context.Context, receiptURI string) (map[string]any, error) {
		return h.fetchReceipt(ctx, event.Source, receiptURI)
	})
	if !ok {
		return ctx.Err()
	}
	return h.createSync(ctx, adapter, event.Repo, uri, recordCID, collection, value, owner)
}

func (h *marketHandler) fetchReceipt(ctx context.Context, source, receiptURI string) (map[string]any, error) {
	aturi, err := syntax.ParseATURI(receiptURI)
	if err != nil {
		return nil, err
	}
	repo, err := syntax.ParseDID(aturi.Authority().String())
	if err != nil {
		return nil, err
	}
	value, _, err := h.fetchMarketRecord(ctx, source, repo, aturi.Collection().String(), aturi.RecordKey().String(), nil)
	return value, err
}

func (h *marketHandler) createSync(ctx context.Context, adapter, did, uri, cid, collection string, record map[string]any, owner *market.ContractRef) error {
	payload, err := json.Marshal(marketPayload{DID: did, URI: uri, CID: cid, Collection: collection, Record: record})
	if err != nil {
		return fmt.Errorf("eventstream: encoding the market payload for %s: %w", uri, err)
	}
	object := &v1alpha1.StateSync{
		TypeMeta: metav1.TypeMeta{APIVersion: v1alpha1.GroupVersion.String(), Kind: "StateSync"},
		ObjectMeta: metav1.ObjectMeta{
			Name: marketSyncName(uri, cid),
			Labels: map[string]string{
				marketInboxLabel:      "true",
				marketCollectionLabel: collection,
			},
			OwnerReferences: contractOwners(owner),
		},
		Spec: v1alpha1.StateSyncSpec{
			Sink:           v1alpha1.StateSyncSink{Adapter: adapter},
			Direction:      v1alpha1.StateSyncDirectionOutbound,
			Payload:        payload,
			IdempotencyKey: "market:" + uri + "#" + cid,
			ConflictPolicy: v1alpha1.StateSyncConflictSourceWins,
			DeletionPolicy: v1alpha1.StateSyncDeletionRetain,
		},
	}
	stored, err := toUnstructured(object)
	if err != nil {
		return fmt.Errorf("eventstream: encoding the market StateSync for %s: %w", uri, err)
	}
	dyn, err := h.opts.Dynamic(h.opts.Workspace)
	if err != nil {
		return fmt.Errorf("eventstream: opening workspace %s: %w", h.opts.Workspace, err)
	}
	if _, err := kube.Create[kube.Doc](ctx, dyn.Resource(statesync.StateSyncGVR), stored, metav1.CreateOptions{}); err != nil {
		if apierrors.IsAlreadyExists(err) {
			return nil
		}
		return fmt.Errorf("eventstream: creating the market StateSync %s: %w", object.Name, err)
	}
	return nil
}

func (h *marketHandler) fetchMarketRecord(ctx context.Context, source string, repo syntax.DID, collection, rkey string, cidPtr *cid.Cid) (map[string]any, string, error) {
	cidText := ""
	if cidPtr != nil {
		cidText = cidPtr.String()
	}
	if pds, err := h.authorPDS(ctx, repo); err == nil {
		record, err := pds.GetRecord(ctx, pdsclient.RepoPath{Repo: repo, Collection: syntax.NSID(collection), Rkey: syntax.RecordKey(rkey)}, cidPtr)
		if err == nil {
			value := map[string]any{}
			if err := json.Unmarshal(record.Value, &value); err != nil {
				return nil, "", err
			}
			return value, record.CID.String(), nil
		}
	}
	hostname, err := h.relayHost(ctx, source, repo.String())
	if err != nil {
		if hostname = sourceHost(source); hostname == "" {
			return nil, "", err
		}
	}
	for _, scheme := range []string{"http", "https"} {
		target := scheme + "://" + hostname + "/xrpc/com.atproto.repo.getRecord?repo=" + url.QueryEscape(repo.String()) +
			"&collection=" + url.QueryEscape(collection) + "&rkey=" + url.QueryEscape(rkey) + "&cid=" + url.QueryEscape(cidText)
		req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
		if err != nil {
			return nil, "", err
		}
		req.Header.Set("Accept", "application/json")
		resp, err := h.httpClient().Do(req)
		if err != nil {
			continue
		}
		data, readErr := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
		resp.Body.Close()
		if readErr != nil || resp.StatusCode != http.StatusOK {
			continue
		}
		var body struct {
			CID   string         `json:"cid"`
			Value map[string]any `json:"value"`
		}
		if err := json.Unmarshal(data, &body); err != nil {
			continue
		}
		if body.Value == nil {
			continue
		}
		body.CID = cmp.Or(body.CID, cidText)
		return body.Value, body.CID, nil
	}
	return nil, "", errors.New("the author PDS is not reachable through its identity or the relay host list")
}

func (h *marketHandler) relayHost(ctx context.Context, source, did string) (string, error) {
	base, ok := strings.CutPrefix(strings.TrimSpace(source), relay.SourcePrefix)
	if !ok || base == "" {
		return "", errors.New("no relay source names the author PDS")
	}
	target := strings.TrimRight(base, "/") + "/xrpc/com.atproto.sync.getRepoStatus?did=" + url.QueryEscape(did)
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
	if err != nil {
		return "", err
	}
	req.Header.Set("Accept", "application/json")
	resp, err := h.httpClient().Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("relay getRepoStatus returned HTTP %d", resp.StatusCode)
	}
	var body struct {
		Hostname string `json:"hostname"`
	}
	if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&body); err != nil {
		return "", err
	}
	if strings.TrimSpace(body.Hostname) == "" {
		return "", fmt.Errorf("relay getRepoStatus names no host for %s", did)
	}
	return strings.TrimSpace(body.Hostname), nil
}

func sourceHost(source string) string {
	base, ok := strings.CutPrefix(strings.TrimSpace(source), relay.SourcePrefix)
	if !ok {
		return ""
	}
	u, err := url.Parse(strings.TrimSpace(base))
	if err != nil {
		return ""
	}
	return u.Host
}

func (h *marketHandler) authorPDS(ctx context.Context, did syntax.DID) (pdsclient.PDS, error) {
	h.mu.Lock()
	cached := h.pdses[did]
	h.mu.Unlock()
	if cached != nil {
		return cached, nil
	}
	ident, err := h.directory.LookupDID(ctx, did)
	if err != nil {
		return nil, err
	}
	endpoint := strings.TrimSpace(ident.PDSEndpoint())
	if endpoint == "" {
		return nil, fmt.Errorf("%s declares no atproto_pds service", did)
	}
	pds, err := pdsclient.NewRemote(pdsclient.RemoteConfig{
		BaseURL:    endpoint,
		DID:        did,
		HTTPClient: h.httpClient(),
		Timeout:    20 * time.Second,
	})
	if err != nil {
		return nil, err
	}
	h.mu.Lock()
	h.pdses[did] = pds
	h.mu.Unlock()
	return pds, nil
}

func (h *marketHandler) httpClient() *http.Client {
	if h.client != nil {
		return h.client
	}
	return httpClientWithFallback(nil, nil)
}

func contractOwners(owner *market.ContractRef) []metav1.OwnerReference {
	if owner == nil || owner.Name == "" || owner.UID == "" {
		return nil
	}
	return []metav1.OwnerReference{{APIVersion: v1alpha1.GroupVersion.String(), Kind: "ComputeContract", Name: owner.Name, UID: types.UID(owner.UID)}}
}

func marketSyncName(uri, cid string) string {
	sum := sha256.Sum256([]byte(uri + "#" + cid))
	return marketNamePrefix + hex.EncodeToString(sum[:16])
}

func marketAdapter(role, collection string) string {
	switch role {
	case "bidder":
		if market.BidderConsumes(collection) {
			return marketBidderMarketInbox
		}
	case "requester":
		switch {
		case collection == market.BidNSID:
			return marketRequesterBidInbox
		case market.RequesterConsumes(collection):
			return marketRequesterMarketInbox
		}
	}
	return ""
}