johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package bidder

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

	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"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/market/participant"
)

type outboxAdapter struct {
	kind string

	observe func(ctx context.Context, request participant.OutboxRequest) (participant.Observation, error)

	apply func(ctx context.Context, request participant.OutboxRequest) (participant.Observation, error)

	remove func(ctx context.Context, request participant.OutboxRequest) error
}

var _ participant.Adapter = (*outboxAdapter)(nil)

func (a *outboxAdapter) Name() string { return a.kind }

func (a *outboxAdapter) Observe(ctx context.Context, request participant.OutboxRequest) (participant.Observation, error) {
	return a.observe(ctx, request)
}

func (a *outboxAdapter) Apply(ctx context.Context, request participant.OutboxRequest) (participant.Observation, error) {
	return a.apply(ctx, request)
}

func (a *outboxAdapter) Delete(ctx context.Context, request participant.OutboxRequest) error {
	if a.remove == nil {
		return participant.ErrRemoteNotFound
	}
	return a.remove(ctx, request)
}

func (b *Bidder) eventAdapter() participant.Adapter {
	return &outboxAdapter{kind: b.guests.EventAdapter(), observe: b.observeEvent, apply: b.applyEvent, remove: b.deleteEvent}
}

func (b *Bidder) provisionAdapter() participant.Adapter {
	return &outboxAdapter{kind: ProvisionSyncAdapter, observe: b.observeProvision, apply: b.applyProvision}
}

func (b *Bidder) releaseAdapter() participant.Adapter {
	return &outboxAdapter{kind: ReleaseSyncAdapter, observe: b.observeRelease, apply: b.applyRelease}
}

func (b *Bidder) inboxAdapter() participant.Adapter {
	return &outboxAdapter{kind: MarketInboxAdapter, observe: b.observeInbox, apply: b.applyInbox}
}

func payloadHash(raw json.RawMessage) string {
	sum := sha256.Sum256(raw)
	return "sha256:" + hex.EncodeToString(sum[:])
}

func (b *Bidder) stamp() string {
	return b.now().UTC().Format(time.RFC3339)
}

type marketEventPayload struct {
	Collection string `json:"collection"`

	Receipt struct {
		URI string `json:"uri"`
		CID string `json:"cid"`
	} `json:"receipt"`

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

	SubmitEventURL string `json:"submitEventUrl"`

	PayloadRKey string `json:"payloadRKey"`

	EventRKey string `json:"eventRKey"`

	RequestEvent *marketEventRequest `json:"requestEvent,omitempty"`

	RequesterDID string `json:"requesterDid,omitempty"`
}

func decodeMarketEventPayload(raw json.RawMessage) (marketEventPayload, map[string]any, error) {
	var payload marketEventPayload
	if len(raw) == 0 {
		return payload, nil, errors.New("bidder: the market event payload is empty")
	}
	if err := json.Unmarshal(raw, &payload); err != nil {
		return payload, nil, fmt.Errorf("bidder: the market event payload is not JSON: %w", err)
	}
	var envelope map[string]any
	if err := json.Unmarshal(raw, &envelope); err != nil {
		return payload, nil, fmt.Errorf("bidder: the market event payload is not JSON: %w", err)
	}
	switch {
	case strings.TrimSpace(payload.Collection) == "":
		return payload, nil, errors.New("bidder: the market event payload names no collection")
	case payload.Receipt.URI == "" || payload.Receipt.CID == "":
		return payload, nil, errors.New("bidder: the market event payload names no receipt")
	case payload.Record == nil:
		return payload, nil, errors.New("bidder: the market event payload names no record")
	}
	return payload, envelope, nil
}

func (p marketEventPayload) recordKeys(key string) (string, string) {
	payloadRKey, eventRKey := market.EventRKeys(key)
	return cmp.Or(strings.TrimSpace(p.PayloadRKey), payloadRKey), cmp.Or(strings.TrimSpace(p.EventRKey), eventRKey)
}

func unsignedRecord(record map[string]any) map[string]any {
	out := make(map[string]any, len(record))
	for key, value := range record {
		if key != "signatures" {
			out[key] = value
		}
	}
	return out
}

func marketEventHash(envelope map[string]any, record map[string]any) (string, error) {
	unsigned := make(map[string]any, len(envelope))
	maps.Copy(unsigned, envelope)
	unsigned["record"] = record
	raw, err := json.Marshal(unsigned)
	if err != nil {
		return "", err
	}
	return payloadHash(raw), nil
}

func recordObservedAt(record map[string]any) string {
	for _, key := range []string{"observedAt", "createdAt"} {
		if value, ok := record[key].(string); ok && value != "" {
			return value
		}
	}
	return ""
}

func (b *Bidder) observeEvent(ctx context.Context, request participant.OutboxRequest) (participant.Observation, error) {
	payload, envelope, err := decodeMarketEventPayload(request.Payload)
	if err != nil {
		return participant.Observation{}, err
	}
	payloadRKey, eventRKey := payload.recordKeys(request.Key)
	payloadRec, err := b.records.GetRecord(ctx, b.cfg.DID, payload.Collection, payloadRKey)
	if errors.Is(err, market.ErrNotFound) {
		return participant.Observation{}, nil
	}
	if err != nil {
		return participant.Observation{}, err
	}
	eventRec, err := b.records.GetRecord(ctx, b.cfg.DID, market.EventNSID, eventRKey)
	if errors.Is(err, market.ErrNotFound) {
		return participant.Observation{}, nil
	}
	if err != nil {
		return participant.Observation{}, err
	}
	hash, err := marketEventHash(envelope, unsignedRecord(payloadRec.Value))
	if err != nil {
		return participant.Observation{}, err
	}
	return participant.Observation{
		Exists: payload.submitTarget(request) == "" || b.eventDelivered(request),
		Ref:    payloadRec.URI,
		CID:    payloadRec.CID,
		Hash:   hash,
		At:     recordObservedAt(payloadRec.Value),
		Meta: map[string]string{
			"payloadUri": payloadRec.URI,
			"payloadCid": payloadRec.CID,
			"eventUri":   eventRec.URI,
			"eventCid":   eventRec.CID,
		},
	}, nil
}

func (p marketEventPayload) submitTarget(request participant.OutboxRequest) string {
	return cmp.Or(strings.TrimSpace(p.SubmitEventURL), strings.TrimSpace(request.Sink.SubmitEventURL))
}

func deliveredKey(uid string, payload json.RawMessage) string {
	return uid + "#" + payloadHash(payload)
}

func (b *Bidder) eventDelivered(request participant.OutboxRequest) bool {
	_, ok := b.delivered.Load(deliveredKey(request.UID, request.Payload))
	return ok
}

func (b *Bidder) applyEvent(ctx context.Context, request participant.OutboxRequest) (participant.Observation, error) {
	payload, _, err := decodeMarketEventPayload(request.Payload)
	if err != nil {
		return participant.Observation{}, err
	}
	payloadRKey, eventRKey := payload.recordKeys(request.Key)
	event, err := b.writeMarketEvent(ctx, request, payload, payloadRKey, eventRKey)
	if err != nil {
		return participant.Observation{}, err
	}
	if target := payload.submitTarget(request); target != "" {
		body := map[string]any{"uri": event.URI, "cid": event.CID, "record": event.Record}
		postCtx, cancel := context.WithTimeout(ctx, EventSubmitTimeout)
		err := b.outbound.Post(postCtx, target, market.SubmitEventNSID, body)
		cancel()
		if err != nil {
			if ctx.Err() != nil {
				return participant.Observation{}, ctx.Err()
			}
			return participant.Observation{}, b.eventSubmitFailed(request, payload, target, err)
		}
		b.delivered.Store(deliveredKey(request.UID, request.Payload), true)
	}
	return b.observeEvent(ctx, request)
}

func (b *Bidder) writeMarketEvent(ctx context.Context, request participant.OutboxRequest, payload marketEventPayload, payloadRKey, eventRKey string) (market.SignedRecord, error) {
	stored, err := b.records.GetRecord(ctx, b.cfg.DID, market.EventNSID, eventRKey)
	if err == nil {
		return market.SignedRecord{URI: stored.URI, CID: stored.CID, Record: stored.Value}, nil
	}
	if !errors.Is(err, market.ErrNotFound) {
		return market.SignedRecord{}, fmt.Errorf("bidder: reading the market event: %w", err)
	}
	if payload.RequestEvent != nil {
		if err := b.applyInboundEvent(ctx, request, payload); err != nil {
			return market.SignedRecord{}, err
		}
	}
	record := unsignedRecord(payload.Record)
	signed, err := market.SignRecord(record, b.cfg.DID, b.signer)
	if err != nil {
		return market.SignedRecord{}, fmt.Errorf("bidder: signing the market event payload: %w", err)
	}
	writer := participant.CreateOnly{Records: b.records}
	written, err := writer.PutRecord(ctx, b.cfg.DID, payload.Collection, payloadRKey, signed)
	if err != nil {
		return market.SignedRecord{}, fmt.Errorf("bidder: writing the market event payload: %w", err)
	}
	event, err := market.PutEvent(ctx, writer, b.cfg.DID, eventRKey, market.EventOptions{
		Receipt:   market.RecordRef{URI: payload.Receipt.URI, CID: payload.Receipt.CID},
		Payload:   market.RecordRef{URI: written.URI, CID: written.CID},
		CreatedAt: recordObservedAt(record),
	}, b.signer)
	if err != nil {
		return market.SignedRecord{}, fmt.Errorf("bidder: writing the market event: %w", err)
	}
	return event, nil
}

func (b *Bidder) eventSubmitFailed(request participant.OutboxRequest, payload marketEventPayload, target string, err error) error {
	if participant.Refused(err) {
		b.log.Warn("the requester refused the market event", "collection", payload.Collection, "url", target, "err", err)
		return fmt.Errorf("bidder: the requester refused the market event: %w: %w", err, participant.ErrRejected)
	}
	sync, known := b.syncs.Find(func(sync v1alpha1.StateSync) bool { return string(sync.UID) == request.UID })
	if !b.eventWanted(request, payload) {
		b.log.Info("the market event is no longer sent: its ComputeVM is released", "collection", payload.Collection, "url", target,
			"stateSync", sync.Name, "err", err)
		return fmt.Errorf("bidder: the ComputeVM is released; the market event is no longer sent: %w: %w", err, participant.ErrObsolete)
	}
	wrapped := fmt.Errorf("bidder: submitting the market event to the requester: %w: %w", err, participant.ErrUnreachable)
	if !known || sync.Status.LastError != wrapped.Error() {
		b.log.Warn("submitting the market event to the requester failed; it is retried while the ComputeVM lives", "collection", payload.Collection,
			"url", target, "stateSync", sync.Name, "attempts", sync.Status.Attempts+1, "err", err)
	}
	return wrapped
}

func (b *Bidder) eventWanted(request participant.OutboxRequest, payload marketEventPayload) bool {
	if request.Source == nil {
		return false
	}
	vm, ok := b.vms.Get(request.Source.Name)
	if !ok || (request.Source.UID != "" && string(vm.UID) != request.Source.UID) {
		return false
	}
	if payload.Collection == market.ComputeEventsVMTerminatedNSID {
		return true
	}
	return vm.DeletionTimestamp == nil && !b.guests.Ended(&vm) && !b.guests.Releasing(&vm)
}

func (b *Bidder) deleteEvent(ctx context.Context, request participant.OutboxRequest) error {
	payload, _, err := decodeMarketEventPayload(request.Payload)
	if err != nil {
		return err
	}
	payloadRKey, eventRKey := payload.recordKeys(request.Key)
	for _, item := range [][2]string{{payload.Collection, payloadRKey}, {market.EventNSID, eventRKey}} {
		if err := b.records.DeleteRecord(ctx, b.cfg.DID, item[0], item[1]); err != nil && !errors.Is(err, market.ErrNotFound) {
			return err
		}
	}
	return nil
}

func (b *Bidder) verifyInboundDelete(ctx context.Context, event marketEventRequest, receiptURI, receiptCID, requesterDID string) (*market.Event, error) {
	uri := strings.TrimSpace(event.URI)
	if uri == "" || event.Record == nil {
		return nil, errors.New("bidder: the inbound event is empty")
	}
	parsed, err := market.ParseEvent(event.Record)
	if err != nil {
		return nil, fmt.Errorf("bidder: the inbound event is invalid: %w", err)
	}
	if market.NSIDFromURI(parsed.Payload.URI) != market.ComputeEventsVMDeleteNSID {
		return nil, errors.New("bidder: the inbound event is not a vm delete")
	}
	if err := b.verifier.VerifyRecord(ctx, market.ResolvedRecord{URI: uri, CID: event.CID, Value: event.Record}); err != nil {
		return nil, fmt.Errorf("bidder: the inbound event is not signed: %w", err)
	}
	if market.ATURIAuthority(uri) != requesterDID {
		return nil, errors.New("bidder: the inbound event is not from the requester")
	}
	if parsed.Receipt.URI != receiptURI || parsed.Receipt.CID != receiptCID {
		return nil, errors.New("bidder: the inbound event names another receipt")
	}
	resolved, err := b.records.Resolve(ctx, market.RecordRef{URI: parsed.Payload.URI, CID: parsed.Payload.CID})
	if err != nil {
		return nil, fmt.Errorf("bidder: resolving the inbound event payload: %w", err)
	}
	if kind, _ := resolved.Value["$type"].(string); kind != market.ComputeEventsVMDeleteNSID {
		return nil, errors.New("bidder: the inbound event payload is not a vm delete")
	}
	if err := b.verifier.VerifyRecord(ctx, resolved); err != nil {
		return nil, fmt.Errorf("bidder: the inbound event payload is not signed: %w", err)
	}
	return parsed, nil
}

func (b *Bidder) applyInboundEvent(ctx context.Context, request participant.OutboxRequest, payload marketEventPayload) error {
	event, err := b.verifyInboundDelete(ctx, *payload.RequestEvent, payload.Receipt.URI, payload.Receipt.CID, payload.RequesterDID)
	if err != nil {
		return err
	}
	if request.Source == nil || strings.TrimSpace(request.Source.Name) == "" {
		return errors.New("bidder: the inbound event names no ComputeVM")
	}
	vm, ok := b.vms.Get(request.Source.Name)
	if !ok {
		b.log.Warn("the ComputeVM of an inbound delete event is gone", "computevm", request.Source.Name)
		return nil
	}
	if vm.Spec.Requester.Did != "" && vm.Spec.Requester.Did != payload.RequesterDID {
		return errors.New("bidder: the inbound event is not from the contract requester")
	}
	observedAt := b.stamp()
	release := v1alpha1.ComputeVMRelease{RequestedAt: observedAt}
	if event.Payload.URI != "" {
		release.Request = &commonv1.StrongRef{URI: event.Payload.URI, CID: event.Payload.CID}
	}
	if err := b.intents.ApplyComputeVMRelease(ctx, vm.Name, release); err != nil {
		return fmt.Errorf("bidder: recording the release request on the ComputeVM: %w", err)
	}
	info := contractInfo{computeVM: vm.Name, receiptURI: payload.Receipt.URI, receiptCID: payload.Receipt.CID, acceptAuthor: payload.RequesterDID}
	if terms := vm.Spec.Contract; terms != nil {
		info.acceptURI, info.acceptCID, info.submitEventURL = terms.Accept.URI, terms.Accept.CID, terms.SubmitEventURL
	}
	if bid := b.bidForVM(&vm); bid != nil {
		info.bidName = bid.Name
	}
	b.markReleaseRequested(ctx, info.bidName, info,
		&commonv1.StrongRef{URI: event.Payload.URI, CID: event.Payload.CID},
		&commonv1.StrongRef{URI: payload.RequestEvent.URI, CID: payload.RequestEvent.CID},
		observedAt)
	return nil
}

func decodeProvisionPayload(raw json.RawMessage) (provisionPayload, error) {
	var payload provisionPayload
	if len(raw) == 0 {
		return payload, errors.New("bidder: the provision payload is empty")
	}
	if err := json.Unmarshal(raw, &payload); err != nil {
		return payload, fmt.Errorf("bidder: the provision payload is not JSON: %w", err)
	}
	switch {
	case payload.Accept.URI == "" || payload.Accept.CID == "":
		return payload, errors.New("bidder: the provision payload names no accept")
	case strings.TrimSpace(payload.RequesterDID) == "":
		return payload, errors.New("bidder: the provision payload names no requester")
	}
	return payload, nil
}

func (b *Bidder) settledObservation(ref string, request participant.OutboxRequest, meta map[string]string) participant.Observation {
	return participant.Observation{Exists: true, Ref: ref, Hash: payloadHash(request.Payload), At: b.stamp(), Meta: meta}
}

func (b *Bidder) observeProvision(_ context.Context, request participant.OutboxRequest) (participant.Observation, error) {
	payload, err := decodeProvisionPayload(request.Payload)
	if err != nil {
		return participant.Observation{}, err
	}
	acceptMeta := map[string]string{"acceptUri": payload.Accept.URI, "acceptCid": payload.Accept.CID}
	bid := b.bidForAcceptStatus(payload.Accept.URI, payload.Accept.CID)
	if bid != nil && bid.Status.Terminated != nil {
		return b.settledObservation(market.TerminationKey(bid.Status.Terminated.ReceiptURI), request, acceptMeta), nil
	}
	var vm *v1alpha1.ComputeVM
	if bid != nil && bid.Status.ComputeVM != "" {
		if held, ok := b.vms.Get(bid.Status.ComputeVM); ok {
			vm = &held
		}
	}
	if vm == nil {
		key := b.guests.AcceptKey(payload.Accept.URI, payload.Accept.CID)
		if held, ok := b.vms.Find(func(vm v1alpha1.ComputeVM) bool { return b.guests.AcceptKeyOf(&vm) == key }); ok {
			vm = &held
		}
	}
	if vm == nil {
		if bid != nil && bid.Status.ComputeVM != "" {
			return b.settledObservation("released:"+bid.Status.ComputeVM, request, acceptMeta), nil
		}
		return participant.Observation{}, nil
	}
	terms := vm.Spec.Contract
	if terms == nil || terms.Accept.URI != payload.Accept.URI || terms.Accept.CID != payload.Accept.CID || terms.Receipt.URI == "" {
		return participant.Observation{}, nil
	}
	meta := map[string]string{"computevm": vm.Name, "acceptUri": payload.Accept.URI, "acceptCid": payload.Accept.CID,
		"receiptUri": terms.Receipt.URI, "receiptCid": terms.Receipt.CID}
	observation := b.settledObservation(vm.Name, request, meta)
	observation.CID = string(vm.UID)
	return observation, nil
}

func (b *Bidder) applyProvision(ctx context.Context, request participant.OutboxRequest) (participant.Observation, error) {
	payload, err := decodeProvisionPayload(request.Payload)
	if err != nil {
		return participant.Observation{}, err
	}
	acceptResolved, err := b.records.Resolve(ctx, market.RecordRef{URI: payload.Accept.URI, CID: payload.Accept.CID})
	if err != nil {
		return participant.Observation{}, fmt.Errorf("bidder: resolving the accept: %w", err)
	}
	accept, err := market.ParseAccept(acceptResolved.Value)
	if err != nil {
		return participant.Observation{}, fmt.Errorf("bidder: the accept is invalid: %w", err)
	}
	if err := b.verifier.VerifyRecord(ctx, acceptResolved); err != nil {
		return participant.Observation{}, fmt.Errorf("bidder: the accept is not signed by its author: %w", err)
	}
	if scope := b.policy.Scope(ctx, participant.PolicyInput{SelfDid: b.cfg.DID, CounterpartyDid: payload.RequesterDID,
		RFP: &participant.PolicyRef{URI: accept.RFP.URI, CID: accept.RFP.CID}}); !scope.Allow {
		b.log.Info("the scope policy declined the accept", "accept", payload.Accept.URI, "reason", scope.Reason)
		return participant.Observation{}, errors.New("bidder: the scope policy declined the accept: " + scope.Reason)
	}
	found, err := b.bidForAccept(accept)
	if err != nil {
		return participant.Observation{}, fmt.Errorf("bidder: finding the bid of the accept: %w", err)
	}
	unlock := b.lockBid(found.Name)
	defer unlock()
	bid, ok := b.bids.Get(found.Name)
	if !ok {
		return participant.Observation{}, fmt.Errorf("bidder: the bid %s is gone", found.Name)
	}
	contract, done, err := b.acceptContract(ctx, request, payload, accept, acceptResolved, &bid)
	if err != nil || done != nil {
		return ptrValue(done), err
	}
	name, err := b.provision(ctx, provisionRequest{
		rfpURI:         accept.RFP.URI,
		rfpCID:         accept.RFP.CID,
		vmURI:          contract.rfp.Payload.URI,
		vmCID:          contract.rfp.Payload.CID,
		acceptURI:      payload.Accept.URI,
		acceptCID:      payload.Accept.CID,
		bidURI:         accept.Bid.URI,
		bidCID:         accept.Bid.CID,
		requesterDID:   payload.RequesterDID,
		acceptedAt:     accept.CreatedAt,
		vmPayload:      contract.vmPayload,
		bidConfig:      contract.bidConfig,
		bidConfigValue: contract.bidConfigValue,
		owner:          bidOwner(&bid),
	})
	if err != nil {
		return participant.Observation{}, err
	}
	now := contract.now
	terms := v1alpha1.ComputeVMContract{
		Receipt:        commonv1.StrongRef{URI: contract.receiptURI, CID: contract.receiptCID},
		Accept:         commonv1.StrongRef{URI: payload.Accept.URI, CID: payload.Accept.CID},
		SubmitEventURL: contract.submitEventURL,
		AcceptedAt:     now.Format(time.RFC3339),
	}
	if err := b.intents.ApplyComputeVMContract(ctx, name, terms); err != nil {
		return participant.Observation{}, err
	}
	if held, ok := b.vms.Get(name); ok {
		held.Spec.Contract = &terms
		b.vms.Put(name, held)
	}
	if current, ok := b.bids.Get(bid.Name); ok {
		if err := b.bindAccepted(ctx, &current, name, now); err != nil {
			return participant.Observation{}, err
		}
		if terminated := current.Status.Terminated; terminated != nil {
			info := contractInfo{bidName: bid.Name, computeVM: name, receiptURI: contract.receiptURI, receiptCID: contract.receiptCID,
				acceptURI: payload.Accept.URI, acceptCID: payload.Accept.CID, acceptAuthor: payload.RequesterDID, submitEventURL: contract.submitEventURL}
			if err := b.releaseAccepted(ctx, info, terminated, now); err != nil {
				return participant.Observation{}, err
			}
		}
	}
	return b.observeProvision(ctx, request)
}

func ptrValue(o *participant.Observation) participant.Observation {
	if o == nil {
		return participant.Observation{}
	}
	return *o
}

type acceptedContract struct {
	rfp *market.RFP

	vmPayload market.ResolvedRecord

	bidConfig market.StrongRef

	bidConfigValue map[string]any

	submitEventURL string

	receiptURI string

	receiptCID string

	now time.Time
}

func (b *Bidder) acceptContract(ctx context.Context, request participant.OutboxRequest, payload provisionPayload, accept *market.Accept,
	acceptResolved market.ResolvedRecord, bid *v1alpha1.Bid) (acceptedContract, *participant.Observation, error) {
	var out acceptedContract
	if bid.Spec.Requester != "" && bid.Spec.Requester != payload.RequesterDID {
		return out, nil, errors.New("bidder: the accept is not from the requester that opened the contract")
	}
	if terminated := bid.Status.Terminated; terminated != nil {
		if bid.Status.Accept == nil || bid.Status.Accept.URI != payload.Accept.URI || bid.Status.Accept.CID != payload.Accept.CID {
			return out, nil, errors.New("bidder: the bid's contract was released")
		}
		settled := b.settledObservation(market.TerminationKey(terminated.ReceiptURI), request,
			map[string]string{"acceptUri": payload.Accept.URI, "acceptCid": payload.Accept.CID})
		return out, &settled, nil
	}
	if bid.Status.Phase == v1alpha1.BidPhaseFailed {
		return out, nil, errors.New("bidder: the bid failed: " + bid.Status.Message)
	}
	out.now = b.now().UTC()
	if bidExpired(bid, out.now) || bid.Status.Phase == v1alpha1.BidPhaseExpired {
		b.expireBid(ctx, bid, out.now)
		return out, nil, errors.New("bidder: the bid has expired")
	}
	rfpResolved, err := b.records.Resolve(ctx, market.RecordRef{URI: accept.RFP.URI, CID: accept.RFP.CID})
	if err != nil {
		return out, nil, fmt.Errorf("bidder: resolving the accepted RFP: %w", err)
	}
	if out.rfp, err = market.ParseRFP(rfpResolved.Value); err != nil {
		return out, nil, fmt.Errorf("bidder: the accepted RFP is invalid: %w", err)
	}
	if err := b.verifier.VerifyRecord(ctx, rfpResolved); err != nil {
		return out, nil, fmt.Errorf("bidder: the accepted RFP is not signed by its author: %w", err)
	}
	decision := b.policy.EvaluatePolicies(ctx, out.rfp.Policies, b.fullInput(payload.RequesterDID, accept.RFP.URI, accept.RFP.CID))
	if !decision.Allow {
		b.log.Warn("policy rejected the accept", "accept", payload.Accept.URI, "reason", decision.Reason, "run", decision.Run)
		return out, nil, errors.New("bidder: the policy rejected the accept: " + decision.Reason)
	}
	if out.vmPayload, err = b.records.Resolve(ctx, market.RecordRef{URI: out.rfp.Payload.URI, CID: out.rfp.Payload.CID}); err != nil {
		return out, nil, fmt.Errorf("bidder: resolving the RFP payload: %w", err)
	}
	bidResolved, err := b.records.Resolve(ctx, market.RecordRef{URI: accept.Bid.URI, CID: accept.Bid.CID})
	if err != nil {
		return out, nil, fmt.Errorf("bidder: resolving the accepted bid: %w", err)
	}
	parsedBid, err := market.ParseBid(bidResolved.Value)
	if err != nil {
		return out, nil, fmt.Errorf("bidder: the accepted bid is invalid: %w", err)
	}
	if parsedBid.Config == nil {
		return out, nil, errors.New("bidder: the accepted bid names no bid config")
	}
	configResolved, err := b.records.Resolve(ctx, market.RecordRef{URI: parsedBid.Config.URI, CID: parsedBid.Config.CID})
	if err != nil {
		return out, nil, fmt.Errorf("bidder: resolving the bid config: %w", err)
	}
	out.bidConfig = market.StrongRef{URI: parsedBid.Config.URI, CID: parsedBid.Config.CID}
	out.bidConfigValue = configResolved.Value
	out.submitEventURL = b.eventEndpoint(ctx, payload.RequesterDID, accept.SubmitEvent)
	if strings.TrimSpace(accept.SubmitEvent) != "" && out.submitEventURL == "" {
		b.log.Warn("the accept's submitEvent does not resolve; the firehose carries the event", "accept", payload.Accept.URI, "submitEvent", accept.SubmitEvent)
	}
	receiptSubmitEvent := ""
	if b.cfg.InboundXRPC {
		receiptSubmitEvent = accept.SubmitEvent
	}
	out.receiptURI, out.receiptCID, err = b.createReceipt(ctx, market.ReceiptRefs{
		RFP:         market.RecordRef{URI: accept.RFP.URI, CID: accept.RFP.CID},
		Bid:         market.RecordRef{URI: accept.Bid.URI, CID: accept.Bid.CID},
		Accept:      market.RecordRef{URI: payload.Accept.URI, CID: payload.Accept.CID},
		Payload:     &market.RecordRef{URI: out.rfp.Payload.URI, CID: out.rfp.Payload.CID},
		SubmitEvent: receiptSubmitEvent,
	}, acceptResolved, payload.RequesterDID)
	if err != nil {
		return out, nil, fmt.Errorf("bidder: writing the receipt: %w", err)
	}
	status := cloneBidStatus(bid.Status)
	if status.Accept == nil || status.Accept.URI != payload.Accept.URI || status.Accept.CID != payload.Accept.CID {
		status.Accept = &commonv1.StrongRef{URI: payload.Accept.URI, CID: payload.Accept.CID}
		if status.Timings.AcceptedAt == "" {
			status.Timings.AcceptedAt = out.now.Format(time.RFC3339)
		}
		status.Phase = v1alpha1.BidPhaseWon
		if err := b.writeBidStatus(ctx, bid, status); err != nil {
			return out, nil, fmt.Errorf("bidder: recording the accepted contract: %w", err)
		}
	}
	return out, nil, nil
}

func (b *Bidder) writeBidStatus(ctx context.Context, bid *v1alpha1.Bid, status v1alpha1.BidStatus) error {
	if err := b.intents.ApplyBidStatus(ctx, bid.Name, status); err != nil {
		return err
	}
	bid.Status = status
	b.bids.Put(bid.Name, *bid)
	return nil
}

func (b *Bidder) bindAccepted(ctx context.Context, bid *v1alpha1.Bid, vmName string, now time.Time) error {
	if bid.Status.ComputeVM == vmName {
		return nil
	}
	status := cloneBidStatus(bid.Status)
	status.ComputeVM = vmName
	status.Phase = v1alpha1.BidPhaseWon
	if status.Timings.AcceptedAt == "" {
		status.Timings.AcceptedAt = now.Format(time.RFC3339)
	}
	return b.writeBidStatus(ctx, bid, status)
}

func (b *Bidder) releaseAccepted(ctx context.Context, info contractInfo, terminated *v1alpha1.BidTerminated, now time.Time) error {
	release := v1alpha1.ComputeVMRelease{RequestedAt: terminated.At}
	var requestRef *commonv1.StrongRef
	if terminated.PayloadURI != "" {
		requestRef = &commonv1.StrongRef{URI: terminated.PayloadURI, CID: terminated.PayloadCid}
		release.Request = &commonv1.StrongRef{URI: terminated.PayloadURI, CID: terminated.PayloadCid}
	}
	if err := b.intents.ApplyComputeVMRelease(ctx, info.computeVM, release); err != nil {
		return err
	}
	if _, err := b.ensureTerminationSync(ctx, info, cmp.Or(terminated.At, now.Format(time.RFC3339)), requestRef); err != nil {
		b.log.Warn("resuming the termination of a released accept failed", "bid", info.bidName, "err", err.Error())
	}
	return nil
}

func decodeReleasePayload(raw json.RawMessage) (releasePayload, error) {
	var payload releasePayload
	if len(raw) == 0 {
		return payload, errors.New("bidder: the release payload is empty")
	}
	if err := json.Unmarshal(raw, &payload); err != nil {
		return payload, fmt.Errorf("bidder: the release payload is not JSON: %w", err)
	}
	switch {
	case strings.TrimSpace(payload.ReceiptURI) == "":
		return payload, errors.New("bidder: the release payload names no receipt")
	case strings.TrimSpace(payload.RequesterDID) == "":
		return payload, errors.New("bidder: the release payload names no requester")
	case strings.TrimSpace(payload.Event.URI) == "" || payload.Event.Record == nil:
		return payload, errors.New("bidder: the release payload names no event")
	}
	return payload, nil
}

func releaseAppliedKey(payload releasePayload) string {
	return "release:" + payload.ReceiptURI + "#" + payload.Event.URI
}

func (b *Bidder) releaseSettled(payload releasePayload, request participant.OutboxRequest, ref string) participant.Observation {
	observation := b.settledObservation(cmp.Or(ref, "released:"+payload.ReceiptURI), request, nil)
	return observation
}

func (b *Bidder) observeRelease(ctx context.Context, request participant.OutboxRequest) (participant.Observation, error) {
	payload, err := decodeReleasePayload(request.Payload)
	if err != nil {
		return participant.Observation{}, err
	}
	if _, ok := b.released.Load(releaseAppliedKey(payload)); ok {
		return b.releaseSettled(payload, request, ""), nil
	}
	if market.ATURIAuthority(payload.ReceiptURI) != b.cfg.DID {
		return participant.Observation{}, nil
	}
	resolved, err := b.records.Resolve(ctx, market.RecordRef{URI: payload.ReceiptURI})
	if errors.Is(err, market.ErrNotFound) {
		return participant.Observation{}, nil
	}
	if err != nil {
		return participant.Observation{}, err
	}
	receipt, err := market.ParseReceipt(resolved.Value)
	if err != nil {
		return participant.Observation{}, err
	}
	if vm := b.vmForAccept(receipt.Accept.URI, receipt.Accept.CID); vm != nil {
		if b.guests.ReleaseRequested(vm) {
			return b.releaseSettled(payload, request, vm.Name), nil
		}
		return participant.Observation{}, nil
	}
	bid := b.bidForAcceptStatus(receipt.Accept.URI, receipt.Accept.CID)
	if bid != nil && bid.Status.Terminated != nil && bid.Status.Terminated.ReceiptURI == payload.ReceiptURI {
		return b.releaseSettled(payload, request, ""), nil
	}
	return participant.Observation{}, nil
}

func (b *Bidder) applyRelease(ctx context.Context, request participant.OutboxRequest) (participant.Observation, error) {
	payload, err := decodeReleasePayload(request.Payload)
	if err != nil {
		return participant.Observation{}, err
	}
	if market.ATURIAuthority(payload.ReceiptURI) != b.cfg.DID {
		return participant.Observation{}, errors.New("bidder: the release intent names another bidder's receipt")
	}
	resolved, err := b.records.Resolve(ctx, market.RecordRef{URI: payload.ReceiptURI})
	if err != nil {
		return participant.Observation{}, fmt.Errorf("bidder: resolving the receipt of the release intent: %w", err)
	}
	receipt, err := market.ParseReceipt(resolved.Value)
	if err != nil {
		return participant.Observation{}, fmt.Errorf("bidder: the receipt of the release intent is invalid: %w", err)
	}
	if market.ATURIAuthority(receipt.Accept.URI) != payload.RequesterDID {
		return participant.Observation{}, errors.New("bidder: the release intent is not from the accept author")
	}
	event, err := b.verifyInboundDelete(ctx, payload.Event, payload.ReceiptURI, resolved.CID, payload.RequesterDID)
	if err != nil {
		return participant.Observation{}, err
	}
	requestRef := &commonv1.StrongRef{URI: event.Payload.URI, CID: event.Payload.CID}
	eventRef := &commonv1.StrongRef{URI: payload.Event.URI, CID: payload.Event.CID}
	observedAt := b.stamp()
	info := contractInfo{receiptURI: payload.ReceiptURI, receiptCID: resolved.CID,
		acceptURI: receipt.Accept.URI, acceptCID: receipt.Accept.CID, acceptAuthor: payload.RequesterDID,
		submitEventURL: b.eventEndpoint(ctx, payload.RequesterDID, receipt.SubmitEvent)}
	vm := b.vmForAccept(receipt.Accept.URI, receipt.Accept.CID)
	if vm == nil {
		bid := b.bidForAcceptStatus(receipt.Accept.URI, receipt.Accept.CID)
		if bid == nil {
			return participant.Observation{}, errors.New("bidder: the release intent names no known contract")
		}
		info.bidName = bid.Name
		b.markReleaseRequested(ctx, bid.Name, info, requestRef, eventRef, observedAt)
		b.released.Store(releaseAppliedKey(payload), true)
		return b.releaseSettled(payload, request, ""), nil
	}
	info.computeVM = vm.Name
	release := v1alpha1.ComputeVMRelease{RequestedAt: observedAt}
	if requestRef.URI != "" {
		release.Request = &commonv1.StrongRef{URI: requestRef.URI, CID: requestRef.CID}
	}
	if err := b.intents.ApplyComputeVMRelease(ctx, vm.Name, release); err != nil {
		return participant.Observation{}, fmt.Errorf("bidder: recording the release request on the ComputeVM: %w", err)
	}
	if bid := b.bidForVM(vm); bid != nil {
		info.bidName = bid.Name
	}
	b.markReleaseRequested(ctx, info.bidName, info, requestRef, eventRef, observedAt)
	if _, err := b.ensureTerminationSync(ctx, info, observedAt, requestRef); err != nil {
		return participant.Observation{}, err
	}
	b.released.Store(releaseAppliedKey(payload), true)
	return b.releaseSettled(payload, request, vm.Name), nil
}

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

	URI string `json:"uri"`

	CID string `json:"cid"`

	Collection string `json:"collection"`

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

func (p *marketInboxPayload) decode(raw json.RawMessage) error {
	if len(raw) == 0 {
		return errors.New("bidder: the market inbox payload is empty")
	}
	if err := json.Unmarshal(raw, p); err != nil {
		return fmt.Errorf("bidder: the market inbox payload is not JSON: %w", err)
	}
	if strings.TrimSpace(p.URI) == "" || strings.TrimSpace(p.CID) == "" {
		return errors.New("bidder: the market inbox payload names no record")
	}
	p.Collection = cmp.Or(strings.TrimSpace(p.Collection), market.NSIDFromURI(p.URI))
	return nil
}

func (b *Bidder) observeInbox(_ context.Context, request participant.OutboxRequest) (participant.Observation, error) {
	var payload marketInboxPayload
	if err := payload.decode(request.Payload); err != nil {
		return participant.Observation{}, err
	}
	if _, ok := b.inboxed.Load(inboxKey(request)); ok {
		return participant.InboxObservation(request, payload.URI), nil
	}
	if payload.Collection == market.RFPNSID {
		if _, ok := b.bids.Get(bidName(b.cfg.DID, payload.URI, payload.CID)); ok {
			return participant.InboxObservation(request, payload.URI), nil
		}
	}
	return participant.Observation{At: b.stamp()}, nil
}

func (b *Bidder) applyInbox(ctx context.Context, request participant.OutboxRequest) (participant.Observation, error) {
	var payload marketInboxPayload
	if err := payload.decode(request.Payload); err != nil {
		return participant.Observation{}, err
	}
	b.log.Info("market inbox applied", "collection", payload.Collection, "uri", payload.URI, "did", b.cfg.DID)
	if b.cfg.Suspend && payload.Collection == market.RFPNSID {
		b.log.Info("market inbox declined: the bidder's participation is suspended", "collection", payload.Collection, "uri", payload.URI)
		b.inboxed.Store(inboxKey(request), true)
		return participant.InboxObservation(request, payload.URI), nil
	}
	switch payload.Collection {
	case market.RFPNSID:
		if b.rfpClosed(payload.URI, payload.Record, "inbox") {
			break
		}
		if _, served := b.inboxServesKind(ctx, payload.URI, payload.CID); !served {
			b.log.Info("market inbox dropped an RFP of a compute kind this bidder does not serve", "rfp", payload.URI)
			b.inboxed.Store(inboxKey(request), true)
			return participant.InboxObservation(request, payload.URI), nil
		}
		bid, err := b.enqueueBid(ctx, payload.URI, payload.CID, payload.DID)
		if err != nil {
			return participant.Observation{}, fmt.Errorf("bidder: recording the bid intent: %w", err)
		}
		b.log.Info("market inbox enqueued a bid", "rfp", payload.URI, "bid", bid.Name, "phase", string(bid.Status.Phase))
	case market.AcceptNSID:
		if accept, err := market.ParseAccept(payload.Record); err == nil && market.ATURIAuthority(accept.Bid.URI) != b.cfg.DID {
			break
		}
		provision := provisionPayload{RequesterDID: payload.DID, Accept: provisionRef{URI: payload.URI, CID: payload.CID}}
		if _, err := b.ensureProvisionSync(ctx, provision); err != nil {
			return participant.Observation{}, fmt.Errorf("bidder: recording the provision intent: %w", err)
		}
	case market.EventNSID:
		if err := b.applyMarketEvent(ctx, payload.URI, payload.CID, payload.Record, payload.DID); err != nil {
			return participant.Observation{}, err
		}
	}
	b.inboxed.Store(inboxKey(request), true)
	return participant.InboxObservation(request, payload.URI), nil
}

func inboxKey(request participant.OutboxRequest) string {
	return request.UID + "#" + request.Key
}