johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package bidder

import (
	"cmp"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market"
	contract "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/market"
)

const (
	provisionSyncPrefix = "accept-"

	releaseSyncPrefix = "release-"

	reportSyncPrefix = "report-"

	guestEventSyncPrefix = "guest-"

	guestEventSyncKeyPrefix = "guest:"
)

type marketEventRequest struct {
	URI string `json:"uri"`

	CID string `json:"cid"`

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

type provisionRef struct {
	URI string `json:"uri"`

	CID string `json:"cid"`
}

type provisionPayload struct {
	Accept provisionRef `json:"accept"`

	RequesterDID string `json:"requesterDid"`
}

type releasePayload struct {
	ReceiptURI string `json:"receiptUri"`

	RequesterDID string `json:"requesterDid"`

	Event marketEventRequest `json:"event"`
}

func normalizedPayload(value any) (json.RawMessage, error) {
	raw, err := json.Marshal(value)
	if err != nil {
		return nil, err
	}
	var stored map[string]any
	if err := json.Unmarshal(raw, &stored); err != nil {
		return nil, err
	}
	return json.Marshal(stored)
}

func outboundSync(name string, labels map[string]string, owners []metav1.OwnerReference, source *v1alpha1.StateSyncSource,
	adapter, key string, payload json.RawMessage) v1alpha1.StateSync {
	return v1alpha1.StateSync{
		ObjectMeta: metav1.ObjectMeta{Name: name, Labels: labels, OwnerReferences: owners},
		Spec: v1alpha1.StateSyncSpec{
			Source:         source,
			Sink:           v1alpha1.StateSyncSink{Adapter: adapter},
			Direction:      v1alpha1.StateSyncDirectionOutbound,
			Payload:        payload,
			IdempotencyKey: key,
			ConflictPolicy: v1alpha1.StateSyncConflictSourceWins,
			DeletionPolicy: v1alpha1.StateSyncDeletionRetain,
		},
	}
}

func (b *Bidder) createSync(ctx context.Context, sync v1alpha1.StateSync, kind string, fields ...any) error {
	err := b.intents.CreateStateSync(ctx, sync)
	if errors.Is(err, contract.ErrExists) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("bidder: creating the %s StateSync %s: %w", kind, sync.Name, err)
	}
	b.log.Info("a "+kind+" StateSync is created", append(fields, "stateSync", sync.Name)...)
	return nil
}

func vmSource(workspace string, vm *v1alpha1.ComputeVM) *v1alpha1.StateSyncSource {
	return &v1alpha1.StateSyncSource{Workspace: workspace, ApiVersion: v1alpha1.GroupVersion.String(), Kind: "ComputeVM", Name: vm.Name, UID: string(vm.UID)}
}

func reportSyncName(uid, collection, tag string) string {
	key := collection
	if tag != "" {
		key += "#" + tag
	}
	sum := sha256.Sum256([]byte(key + "\x00" + uid))
	return reportSyncPrefix + hex.EncodeToString(sum[:8])
}

func reportSyncKey(collection, tag, receiptURI string) string {
	switch {
	case collection == market.ComputeEventsVMOnNetworkNSID && tag != "":
		return "onNetwork:" + tag + ":" + receiptURI
	case collection == market.ComputeEventsVMOnNetworkNSID:
		return "onNetwork:" + receiptURI
	case collection == market.ComputeEventsVMHostKeysNSID:
		return "hostKeys:" + receiptURI
	default:
		return collection + ":" + receiptURI
	}
}

func guestEventSyncName(uid, collection, tag, kind, recordSum string) string {
	sum := sha256.Sum256([]byte(collection + "\x00" + tag + "\x00" + kind + "\x00" + recordSum + "\x00" + uid))
	return guestEventSyncPrefix + hex.EncodeToString(sum[:8])
}

func guestEventSyncKey(collection, tag, kind, recordSum, receiptURI string) string {
	return guestEventSyncKeyPrefix + collection + ":" + tag + ":" + kind + ":" + recordSum + ":" + receiptURI
}

func (b *Bidder) ensureGuestEventSync(ctx context.Context, vm *v1alpha1.ComputeVM, collection, tag, kind string, record map[string]any) (string, error) {
	raw, err := json.Marshal(record)
	if err != nil {
		return "", err
	}
	sum := sha256.Sum256(raw)
	recordSum := hex.EncodeToString(sum[:])
	name := guestEventSyncName(string(vm.UID), collection, tag, kind, recordSum)
	key := func(receiptURI string) string { return guestEventSyncKey(collection, tag, kind, recordSum, receiptURI) }
	return name, b.ensureEventSync(ctx, vm, name, key, collection, record, "guest event")
}

func (b *Bidder) ensureReportSync(ctx context.Context, vm *v1alpha1.ComputeVM, collection, tag string, record map[string]any) (string, error) {
	name := reportSyncName(string(vm.UID), collection, tag)
	key := func(receiptURI string) string { return reportSyncKey(collection, tag, receiptURI) }
	return name, b.ensureEventSync(ctx, vm, name, key, collection, record, "market report")
}

func (b *Bidder) ensureEventSync(ctx context.Context, vm *v1alpha1.ComputeVM, name string, key func(receiptURI string) string, collection string,
	record map[string]any, kind string) error {
	terms := vm.Spec.Contract
	if terms == nil {
		return fmt.Errorf("bidder: a %s StateSync needs a contract", kind)
	}
	if terms.Receipt.URI == "" || terms.Receipt.CID == "" {
		return fmt.Errorf("bidder: a %s StateSync needs a receipt", kind)
	}
	if vm.UID == "" {
		return fmt.Errorf("bidder: a %s StateSync needs a ComputeVM UID", kind)
	}
	envelope := map[string]any{
		"collection": collection,
		"receipt":    map[string]any{"uri": terms.Receipt.URI, "cid": terms.Receipt.CID},
		"record":     record,
	}
	if terms.SubmitEventURL != "" {
		envelope["submitEventUrl"] = terms.SubmitEventURL
	}
	payload, err := normalizedPayload(envelope)
	if err != nil {
		return err
	}
	owners := []metav1.OwnerReference{{APIVersion: v1alpha1.GroupVersion.String(), Kind: "ComputeVM", Name: vm.Name, UID: vm.UID}}
	sync := outboundSync(name, b.guests.SyncLabels(vm), owners, vmSource(b.cfg.Workspace, vm), b.guests.EventAdapter(),
		key(terms.Receipt.URI), payload)
	return b.createSync(ctx, sync, kind, "computevm", vm.Name, "collection", collection)
}

func (b *Bidder) ensureTerminationSync(ctx context.Context, info contractInfo, observedAt string, request *commonv1.StrongRef) (string, error) {
	held, ok := b.vms.Get(info.computeVM)
	if !ok {
		return "", fmt.Errorf("bidder: the ComputeVM %s does not exist", info.computeVM)
	}
	vm := held.DeepCopy()
	if terms := vm.Spec.Contract; terms == nil || terms.Receipt.URI == "" || terms.SubmitEventURL == "" {
		if vm.Spec.Contract == nil {
			vm.Spec.Contract = &v1alpha1.ComputeVMContract{}
		}
		if vm.Spec.Contract.Receipt.URI == "" {
			vm.Spec.Contract.Receipt.URI = info.receiptURI
			vm.Spec.Contract.Receipt.CID = info.receiptCID
		}
		vm.Spec.Contract.SubmitEventURL = cmp.Or(vm.Spec.Contract.SubmitEventURL, info.submitEventURL)
	}
	sync, err := b.guests.Termination(vm, b.cfg.Workspace, releaseReason, observedAt, request)
	if err != nil {
		return "", err
	}
	if err := b.createSync(ctx, sync, "termination", "computevm", vm.Name); err != nil {
		return "", err
	}
	return sync.Name, nil
}

func (b *Bidder) ensureInboundTerminationSync(ctx context.Context, vm *v1alpha1.ComputeVM, request *commonv1.StrongRef, requesterDID, eventURI, eventCID string, record map[string]any) error {
	sync, err := b.guests.Termination(vm, b.cfg.Workspace, releaseReason, b.now().UTC().Format(time.RFC3339), request)
	if err != nil {
		return err
	}
	envelope := map[string]any{}
	if err := json.Unmarshal(sync.Spec.Payload, &envelope); err != nil {
		return fmt.Errorf("bidder: the termination StateSync has no payload: %w", err)
	}
	envelope["requestEvent"] = map[string]any{"uri": eventURI, "cid": eventCID, "record": record}
	envelope["requesterDid"] = requesterDID
	if sync.Spec.Payload, err = normalizedPayload(envelope); err != nil {
		return err
	}
	return b.createSync(ctx, sync, "termination", "computevm", vm.Name)
}

func provisionSyncName(acceptURI string) string {
	sum := sha256.Sum256([]byte(strings.TrimSpace(acceptURI)))
	return provisionSyncPrefix + hex.EncodeToString(sum[:8])
}

func (b *Bidder) ensureProvisionSync(ctx context.Context, request provisionPayload) (string, error) {
	if b.guests.AcceptKey(request.Accept.URI, request.Accept.CID) == "" {
		return "", errors.New("bidder: a provision StateSync needs an accept")
	}
	payload, err := normalizedPayload(request)
	if err != nil {
		return "", err
	}
	name := provisionSyncName(request.Accept.URI)
	sync := outboundSync(name, map[string]string{adapterLabel: ProvisionSyncAdapter}, nil, &v1alpha1.StateSyncSource{Workspace: b.cfg.Workspace},
		ProvisionSyncAdapter, "accept:"+request.Accept.URI, payload)
	if err := b.createSync(ctx, sync, "provision", "accept", request.Accept.URI); err != nil {
		return "", err
	}
	return name, nil
}

func releaseSyncName(receiptURI, eventURI string) string {
	sum := sha256.Sum256([]byte(strings.TrimSpace(receiptURI) + "\x00" + strings.TrimSpace(eventURI)))
	return releaseSyncPrefix + hex.EncodeToString(sum[:8])
}

func (b *Bidder) ensureReleaseSync(ctx context.Context, receiptURI, requesterDID, eventURI, eventCID string, record map[string]any) error {
	request := releasePayload{ReceiptURI: receiptURI, RequesterDID: requesterDID,
		Event: marketEventRequest{URI: eventURI, CID: eventCID, Record: record}}
	payload, err := normalizedPayload(request)
	if err != nil {
		return err
	}
	sync := outboundSync(releaseSyncName(receiptURI, eventURI), map[string]string{adapterLabel: ReleaseSyncAdapter}, nil,
		&v1alpha1.StateSyncSource{Workspace: b.cfg.Workspace}, ReleaseSyncAdapter, "release:"+receiptURI+"#"+eventURI, payload)
	return b.createSync(ctx, sync, "release", "receipt", receiptURI)
}