johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package marketbidder

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"sync"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"

	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/app/compose"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/guestcp"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
)

const (
	Name = "marketbidder"

	Kind = string(v1alpha1.ApplicationKindMarketBidder)

	BidderSet             = "market-bidder"
	BidderComponent       = "bidder"
	BidderApp             = "market-bidder"
	BidderIdentity        = "bidder"
	BidderAccount         = "bidder"
	BidderVaultServer     = "bidder"
	BidderVaultClient     = "bidder"
	DefaultPolicy         = v1alpha1.BidderDefaultPolicy
	BidderProfileProdQr   = "prod-qr"
	BidderProfileNetwork  = "market-network"
	BidderProfileDevLocal = "dev-local"
	GuestComponent        = "guest-kcp"
	GuestVaultClient      = "guest-kcp"
	GuestAccount          = "guest-kcp"
	MarketRootNameArg     = "--market-root-name"

	GuestDownloadsExecutable = "socialweb-guest-downloads"

	identityKeysPath = "identity/bidder/keys"
)

var BidderProfiles = []string{BidderProfileProdQr, BidderProfileNetwork, BidderProfileDevLocal}

type SignInGuard func(spec v1alpha1.MarketBidderSpec, id *v1alpha1.MarketIdentity) error

type Controller struct {
	c *compose.Composer

	guard SignInGuard

	mu sync.Mutex

	rotated map[string]int64
}

func Register(m *ctlruntime.Manager, c *compose.Composer, guard SignInGuard) error {
	ctl := &Controller{c: c, guard: guard, rotated: map[string]int64{}}
	c.ProvideTrust(schema.GroupKind{Group: v1alpha1.GroupName, Kind: Kind})
	opts := append([]ctlruntime.Option{ctlruntime.Export(compose.ExportApplications), ctlruntime.Finalizer(compose.Finalizer)}, c.Watches(Kind)...)
	opts = append(opts, compose.ComputeVMWatch(c, compose.ExportComputeVMs, Kind),
		compose.DrainWatch[*v1alpha1.AtprotoRecord](c, compose.ExportMarketIdentities, Kind),
		compose.OperatorRecordWatch(c, compose.ExportMarketIdentities, Kind))
	if err := ctlruntime.For[*v1alpha1.MarketBidder](m, Name, ctl, opts...); err != nil {
		return fmt.Errorf("marketbidder: registering the MarketBidder controller: %w", err)
	}
	return nil
}

func (ctl *Controller) rotatedAt(r compose.Root) int64 {
	ctl.mu.Lock()
	defer ctl.mu.Unlock()
	return ctl.rotated[string(r.UID)]
}

func (ctl *Controller) setRotated(r compose.Root, n int64) {
	ctl.mu.Lock()
	defer ctl.mu.Unlock()
	ctl.rotated[string(r.UID)] = n
}

func (ctl *Controller) forget(r compose.Root) {
	ctl.mu.Lock()
	defer ctl.mu.Unlock()
	delete(ctl.rotated, string(r.UID))
}

type pass struct {
	ctx context.Context

	ctl *Controller

	c *compose.Composer

	r compose.Root

	o *v1alpha1.MarketBidder

	st v1alpha1.ApplicationStatus

	identityPhase v1alpha1.MarketIdentityPhase
}

func (p *pass) log() *slog.Logger { return p.c.Log().With("marketbidder", p.o.Name) }

func (p *pass) write(phase v1alpha1.ApplicationPhase, msg string) (ctlruntime.Result, error) {
	p.st.Phase, p.st.Message = phase, msg
	return ctlruntime.Result{}, p.c.WriteStatus(p.ctx, p.r, compose.MarketBidderGVR, p.st)
}

func (p *pass) fail(err error) (ctlruntime.Result, error) {
	compose.SetOverrideErrorIfAny(&p.st.Conditions, &p.st.ComponentSummary, p.o.Generation, err)
	return p.write(v1alpha1.ApplicationPhaseFailed, err.Error())
}

func (p *pass) retry(what string, err error) (ctlruntime.Result, error) {
	msg := err.Error()
	if what != "" {
		msg = what + ": " + msg
	}
	if _, werr := p.write(v1alpha1.ApplicationPhaseProvisioning, msg); werr != nil {
		return ctlruntime.Result{}, errors.Join(err, werr)
	}
	return ctlruntime.Result{}, err
}

func (p *pass) wait(msg string) (ctlruntime.Result, error) {
	return p.write(v1alpha1.ApplicationPhaseProvisioning, msg)
}

func (p *pass) ready(msg string) (ctlruntime.Result, error) {
	p.st = compose.Ready(p.st, msg)
	return ctlruntime.Result{}, p.c.WriteStatus(p.ctx, p.r, compose.MarketBidderGVR, p.st)
}

func (p *pass) snapshot(ctx context.Context) (compose.Snapshot, error) {
	snap, err := p.c.Snapshot(ctx, p.r.Workspace, p.r.Owner())
	if err != nil {
		return snap, err
	}
	p.st.Components, p.st.Listeners = snap.Components, snap.Listeners
	return snap, nil
}

func (ctl *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, o *v1alpha1.MarketBidder) (ctlruntime.Result, error) {
	c := ctl.c
	if !c.Mine(req) {
		return ctlruntime.Result{}, nil
	}
	r := compose.RootOf(req, Kind, o, o.Spec.Suspend)
	c.Observe(r)
	p := &pass{ctx: ctx, ctl: ctl, c: c, r: r, o: o, st: compose.NewStatus(o.Generation, o.Spec.Profile, o.Status)}
	if r.Deleting {
		return p.delete(ctx)
	}
	if c.Parked(r) {
		return ctlruntime.Result{}, nil
	}
	res, err := p.reconcile(ctx)
	if err == nil {
		if reason := compose.ParkReason("", p.identityPhase); reason != "" && !r.Suspend {
			c.Park(ctx, r, "", reason)
			return res, nil
		}
		c.Adopt(ctx, r)
		c.ApplyQuotas(ctx, r, o.Spec.Quota)
	}
	return res, err
}

func (p *pass) deleting(msg string) (ctlruntime.Result, error) {
	p.st.Phase, p.st.Message = v1alpha1.ApplicationPhaseDeleting, msg
	return ctlruntime.Result{Waiting: true}, p.c.WriteStatus(p.ctx, p.r, compose.MarketBidderGVR, p.st)
}

func (p *pass) delete(ctx context.Context) (ctlruntime.Result, error) {
	c, r := p.c, p.r
	vms, err := c.DeleteAll(ctx, r.Workspace, v1alpha1.ComputeVMGVR)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if vms > 0 {
		return p.deleting(fmt.Sprintf("%d ComputeVM(s) releasing", vms))
	}
	pdsUp := false
	if v, found, err := c.GetComponent(ctx, r.Workspace, compose.ComponentKindPds, compose.PdsComponent); err == nil && found && v.Status.Ready {
		pdsUp = true
	}
	records, err := c.ReleaseRecords(ctx, r.Workspace, pdsUp)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if records > 0 {
		return p.deleting(fmt.Sprintf("%d AtprotoRecord(s) leaving the repo before the Pds stops", records))
	}
	left, err := c.StopComponents(ctx, r, "")
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if left == 0 {
		if left, err = c.DeleteVaults(ctx, r); err != nil {
			return ctlruntime.Result{}, err
		}
	}
	if left > 0 {
		return p.deleting(fmt.Sprintf("%d component(s) stopping", left))
	}
	p.ctl.forget(r)
	c.Drop(r)
	return ctlruntime.Result{}, nil
}

func (p *pass) self() commonv1.ObjectRef {
	return compose.AppRef(compose.ComponentKindBidder, BidderComponent, p.r.Workspace)
}

func (p *pass) guest() commonv1.ObjectRef {
	return compose.AppRef(compose.ComponentKindGuestControlPlane, GuestComponent, p.r.Workspace)
}

func (p *pass) baseLinks(mount string) []compose.Link {
	self, ws := p.self(), p.r.Workspace
	return []compose.Link{
		compose.PolicyRunLink(self, compose.PolicyEngineRefIn(compose.PolicyEngineComponent, ws), cmp.Or(strings.TrimSpace(p.o.Spec.Policy), DefaultPolicy)),
		compose.VaultLink(self, BidderVaultServer, true, mount+"/guests/*"),
		compose.ComputeVMVaultLink(ws, BidderVaultServer, mount+"/guests/*"),
		compose.RootVaultLink(p.r, BidderVaultServer, mount+"/identity/*", mount+"/oidc-rotation/*", guestOIDCKeyPath),
		compose.RefLink(self, p.guest(), compose.Cap(guestcp.CapabilityEvents)),
	}
}

func (p *pass) ensure(ctx context.Context, desired []compose.Desired) error {
	return p.c.EnsureLinked(ctx, p.r, desired, desired)
}

func (p *pass) prune(ctx context.Context, desired []compose.Desired) error {
	_, err := p.c.DeleteComponentsExcept(ctx, p.r.Workspace, p.r.Owner(), compose.DesiredKeys(desired))
	return err
}

type ingress struct {
	config *v1alpha1.IngressConfig

	required bool

	ref *commonv1.ObjectRef

	err error
}

func defaultIngressProfile(profile string) bool {
	return profile == BidderProfileProdQr || profile == BidderProfileNetwork
}

func (p *pass) reconcile(ctx context.Context) (ctlruntime.Result, error) {
	c, r, o := p.c, p.r, p.o
	ws := r.Workspace
	plan, err := c.PlanAccount(ctx, ws, o.Spec.Identity, "bidder",
		compose.AccountVault{Server: BidderVaultServer, Client: BidderVaultClient, Mount: BidderVaultClient},
		v1alpha1.AssociationRequireApproval)
	if err != nil {
		return p.fail(err)
	}
	mount := plan.Vault.MountOr(BidderVaultClient)
	o.Spec.Identity.Mode = cmp.Or(o.Spec.Identity.Mode, plan.Mode)
	pdsHandle := compose.PdsHandle(ws, "bidder", o.Spec.Identity.Handle)
	in := ingress{}
	spec := compose.AppIngressOrDefault(defaultIngressProfile(o.Spec.Profile), o.Spec.Ingress, o.Spec.Network)
	in.required, in.ref, in.config, in.err = compose.PdsAppIngress(spec, o.Spec.Network, o.Spec.Profile == BidderProfileProdQr)
	pdsPublicURL, _ := c.PdsPublicURL(ctx, ws, compose.PdsComponent, in.required)
	if r.Suspend {
		return p.suspended(ctx, plan, pdsHandle, pdsPublicURL, in)
	}
	if err := c.ApplyVault(ctx, r, BidderVaultServer); err != nil {
		return ctlruntime.Result{}, err
	}
	links := p.baseLinks(mount)
	if err := c.ApplyLinks(ctx, r, "root", links); err != nil {
		return ctlruntime.Result{}, err
	}
	var phases []string
	if plan.UsesAssociation() {
		phases = []string{"association-watch"}
	}
	values := p.infraValues()
	intact, err := c.CheckWorkloads(ctx, r)(c.SetWorkloads(BidderSet, o.Spec.Profile, values, phases...))
	if err != nil {
		return p.retry("its set workloads", err)
	}
	if !c.Applied(r) {
		identity := compose.OperatorWatched(plan.Identity, o.Spec.Operator, compose.BidderAssociate)
		digest, digestErr := compose.InfraDigest(BidderSet, o.Spec.Profile, values, identity, phases...)
		if intact && digestErr == nil && c.AppliedBefore(r, o.Status.AppliedSet, digest) {
			c.MarkApplied(r)
		} else {
			if _, err := p.write(v1alpha1.ApplicationPhaseProvisioning, "applying its own OpenBaoServer and workspace"); err != nil {
				return ctlruntime.Result{}, err
			}
			if err := c.ApplyInfra(ctx, r, BidderSet, o.Spec.Profile, values, phases...); err != nil {
				return p.retry("", err)
			}
			if err := c.ApplyIdentitySpec(ctx, r, ws, BidderIdentity, identity); err != nil {
				return p.retry("its MarketIdentity", err)
			}
			c.MarkApplied(r)
			if digestErr == nil {
				p.st.AppliedSet = &v1alpha1.AppliedSet{Generation: o.Generation, Digest: digest}
			}
		}
		p.log().Warn("no ComputeVM PolicyTrigger action handler is registered; ComputeVM triggers stay fail-closed")
	}
	if _, _, err := c.VaultRefs(ctx, ws, BidderVaultServer, BidderVaultClient); err != nil {
		return p.wait("its vault: " + err.Error())
	}
	if ok, msg := c.LinksAccepted(ctx, r, links); !ok {
		return p.wait(msg)
	}
	if err := c.SeedPdsAccountPassword(ctx, r, plan); err != nil {
		return p.retry("its account password", err)
	}
	if plan.Local || o.Spec.Identity.Mode == v1alpha1.MarketIdentityModeLocalRepo {
		if err := c.EnsureRepoKey(ctx, r, BidderVaultServer, mount, identityKeysPath); err != nil {
			return p.retry("its repository key", err)
		}
	}
	id, err := c.Identity(ctx, ws, BidderIdentity)
	if apierrors.IsNotFound(err) {
		if err := c.ApplyIdentitySpec(ctx, r, ws, BidderIdentity, compose.OperatorWatched(plan.Identity, o.Spec.Operator, compose.BidderAssociate)); err != nil {
			return p.retry("its MarketIdentity", err)
		}
		return p.wait("its MarketIdentity was missing and is applied again")
	}
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if !plan.Local {
		return p.account(ctx, plan, id, mount)
	}
	return p.local(ctx, plan, id, mount, pdsHandle, pdsPublicURL, in)
}

func (p *pass) desired(ctx context.Context, plan compose.AccountPlan, id *v1alpha1.MarketIdentity, pdsHandle, pdsPublicURL string, in ingress) ([]compose.Desired, error) {
	c, o, ws := p.c, p.o, p.r.Workspace
	stateRoot := c.StateRoot()
	network := o.Spec.Network
	ident := compose.ComponentIdentity(id, o.Spec.Identity.Handle)
	vault := compose.ComponentRefFor(compose.KindOpenBaoServer, BidderVaultServer)
	var hostnames []string
	if ident.Handle != "" {
		hostnames = []string{ident.Handle}
	}
	guestPdsRef := plan.ComponentPdsRef()
	guestSpec := v1alpha1.GuestControlPlaneSpec{
		ComponentSpec: p.c.NetworkComponentSpec(p.r, compose.ComponentCRStateDir(stateRoot, compose.ComponentKindGuestControlPlane, GuestComponent), network),
		Identity:      ident,
		PdsRef:        guestPdsRef,
		VaultRef:      vault,
		Hostnames:     hostnames,
	}
	downloads, err := p.guestDownloadsDir()
	if err != nil {
		return nil, err
	}
	if downloads != "" {
		guestSpec.Storage = &v1alpha1.ComponentStorage{BlobsDir: downloads}
	}
	guest := compose.Desired{Kind: compose.ComponentKindGuestControlPlane, Name: GuestComponent, Spec: guestSpec}
	var image string
	var firecracker *v1alpha1.FirecrackerProvider
	var kcp *v1alpha1.KcpProvider
	if o.Spec.Kcp != nil {
		k := *o.Spec.Kcp
		kcp = &k
		image = k.GuestImageSource
	}
	if o.Spec.Firecracker != nil {
		f := *o.Spec.Firecracker
		f.VmruntimeCommand = append([]string(nil), f.VmruntimeCommand...)
		firecracker = &f
	}
	bidder := compose.Desired{Kind: compose.ComponentKindBidder, Name: BidderComponent, Spec: v1alpha1.BidderSpec{
		ComponentSpec:        p.c.NetworkComponentSpec(p.r, compose.ComponentCRStateDir(stateRoot, compose.ComponentKindBidder, BidderComponent), network),
		MarketExposureRef:    &commonv1.LocalObjectRef{Name: compose.ServiceNameFor(compose.ComponentKindBidder, BidderComponent) + compose.ExposureIngressSuffix},
		Mode:                 string(plan.Mode),
		Identity:             ident,
		PdsRef:               plan.ComponentPdsRef(),
		PolicyEngineRef:      compose.ComponentRefFor(compose.ComponentKindPolicyEngine, compose.PolicyEngineComponent),
		VaultRef:             vault,
		Hostnames:            hostnames,
		GuestImageSource:     image,
		Firecracker:          firecracker,
		Kcp:                  kcp,
		ComputeKinds:         append([]string(nil), o.Spec.ComputeKinds...),
		Operator:             o.Spec.Operator,
		Policy:               cmp.Or(strings.TrimSpace(o.Spec.Policy), DefaultPolicy),
		PolicyArgs:           compose.OptionValue(o.Spec.Options, "policy-args"),
		Association:          o.Spec.Association.DeepCopy(),
		GuestNetwork:         o.Spec.GuestNetwork,
		BidderDIDs:           append([]string(nil), o.Spec.BidderDids...),
		Comms:                o.Spec.Comms,
		SuspendParticipation: o.Spec.Suspend,
		Storage:              guestSpec.Storage,
	}}
	out := []compose.Desired{compose.PolicyEngineDesired(), guest, bidder}
	if plan.Local {
		pdsSpec, err := compose.AppPdsSpec(stateRoot, "", false, network, pdsHandle, vault, hostnames, pdsPublicURL, in.required, in.ref)
		if err != nil {
			return nil, err
		}
		out = append([]compose.Desired{{Kind: compose.ComponentKindPds, Name: compose.PdsComponent, Spec: pdsSpec}}, out...)
	}
	out = append(out, compose.HTTPServiceExposures(ws, compose.ComponentKindBidder, BidderComponent, "market", network, hostnames, false, true)...)
	out = append(out, compose.HTTPServiceExposures(ws, compose.ComponentKindGuestControlPlane, GuestComponent, "http", network, hostnames, true, true)...)
	if plan.Local {
		out = append(out, compose.PdsExposures(ws, compose.PdsComponent, network, hostnames, in.config)...)
	}
	return compose.ApplyOverrides(o.Spec.Components, out)
}

func (p *pass) guestDownloadsDir() (string, error) {
	record := p.c.Config().Record
	if record == nil {
		return "", nil
	}
	rec, err := record()
	if err != nil {
		return "", fmt.Errorf("marketbidder: reading the instance record for the guest downloads: %w", err)
	}
	return rec.Executables[GuestDownloadsExecutable].Path, nil
}

func (p *pass) suspended(ctx context.Context, plan compose.AccountPlan, pdsHandle, pdsPublicURL string, in ingress) (ctlruntime.Result, error) {
	c, r, o := p.c, p.r, p.o
	id, _ := c.Identity(ctx, r.Workspace, BidderIdentity)
	desired, err := p.desired(ctx, plan, id, pdsHandle, pdsPublicURL, in)
	if err != nil {
		return p.fail(err)
	}
	snap, err := p.snapshot(ctx)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	held := compose.Existing(snap, desired)
	if err := p.ensure(ctx, held); err != nil {
		return p.retry("", err)
	}
	if snap, err = p.snapshot(ctx); err != nil {
		return ctlruntime.Result{}, err
	}
	if id != nil {
		p.st.Did = id.Status.Did
	}
	if o.Spec.Kcp != nil {
		p.st.Components = append(p.st.Components, p.guestComponents(ctx)...)
	}
	compose.FillApplicationStatus(&p.st, o.Generation, true, o.Spec.Components, held, snap, nil)
	p.st = compose.SuspendedOnceApplied(p.st, snap, held)
	return ctlruntime.Result{}, c.WriteStatus(ctx, r, compose.MarketBidderGVR, p.st)
}

func (p *pass) streams(ctx context.Context, plan compose.AccountPlan, localDID string) error {
	streams, err := p.c.ApplyMarketStreams(ctx, p.r, compose.Streams{Role: compose.BidderStreamRole, Comms: compose.OverrideComms(p.o.Spec.Components, compose.ComponentKindBidder, BidderComponent, p.o.Spec.Comms),
		Sources: compose.SourceInput{Network: p.o.Spec.Network, LocalDID: localDID}, ForceFirehose: compose.OAuthReceiveOnly(plan.Mode)})
	if err != nil {
		return err
	}
	return p.c.ApplyLinks(ctx, p.r, "events", streams.Links)
}

func (p *pass) oauthEndpoints(ctx context.Context, snap compose.Snapshot) error {
	return p.c.ApplyOAuthIdentityEndpoints(ctx, p.r, snap, p.r.Workspace, BidderIdentity,
		compose.ServiceNameFor(compose.ComponentKindBidder, BidderComponent)+compose.ExposureIngressSuffix, "bidder", p.o.Spec.Identity)
}

func (p *pass) account(ctx context.Context, plan compose.AccountPlan, id *v1alpha1.MarketIdentity, mount string) (ctlruntime.Result, error) {
	c, r, o := p.c, p.r, p.o
	ws := r.Workspace
	if id.Status.Phase != v1alpha1.MarketIdentityPhaseAwaitingScan {
		p.identityPhase = id.Status.Phase
	}
	switch id.Status.Phase {
	case v1alpha1.MarketIdentityPhaseFailed, v1alpha1.MarketIdentityPhaseSessionExpired:
		return p.write(v1alpha1.ApplicationPhaseFailed, "its MarketIdentity: "+id.Status.Message)
	case v1alpha1.MarketIdentityPhaseReady:
	default:
		switch plan.Mode {
		case v1alpha1.MarketIdentityModeOAuthBrowser, v1alpha1.MarketIdentityModeOAuthQr, v1alpha1.MarketIdentityModeOAuthDelegated:
			desired, err := p.desired(ctx, plan, id, "", "", ingress{})
			if err != nil {
				return p.fail(err)
			}
			if err := p.ensure(ctx, desired); err != nil {
				return p.retry("", err)
			}
			if err := p.prune(ctx, desired); err != nil {
				return ctlruntime.Result{}, err
			}
			snap, err := p.snapshot(ctx)
			if err != nil {
				return ctlruntime.Result{}, err
			}
			if err := p.oauthEndpoints(ctx, snap); err != nil {
				return ctlruntime.Result{}, err
			}
		}
		return p.wait("waiting for its existing-account MarketIdentity to sign in")
	}
	if strings.TrimSpace(id.Status.Did) == "" {
		return p.wait("waiting for its existing-account MarketIdentity to report a did")
	}
	if plan.Mode == v1alpha1.MarketIdentityModePdsAccount && strings.TrimSpace(id.Status.PdsURL) == "" {
		return p.write(v1alpha1.ApplicationPhaseFailed, "the existing-account MarketIdentity reports no pdsUrl; it fails closed")
	}
	desired, err := p.desired(ctx, plan, id, "", "", ingress{})
	if err != nil {
		return p.fail(err)
	}
	if err := p.ensure(ctx, desired); err != nil {
		return p.retry("", err)
	}
	if err := p.streams(ctx, plan, ""); err != nil {
		return p.fail(err)
	}
	if err := p.prune(ctx, desired); err != nil {
		return ctlruntime.Result{}, err
	}
	repoLinks, err := c.AccountRepo(ctx, r, ws, BidderIdentity, plan, id.Status.Did, id.Status.PdsURL)
	if err != nil {
		p.log().Warn("the existing-account repo", "err", err.Error())
	}
	repoLinks = append(repoLinks, compose.RepoWriteLink(p.self(), compose.RepoRefIn(id.Status.Did, ws), compose.ScopeAll),
		compose.ServiceAuthLink(p.self(), compose.AppRef(compose.KindMarketIdentity, BidderIdentity, ws), id.Status.Did))
	if ref := plan.ComponentPdsRef(); ref != nil && strings.EqualFold(ref.Kind, compose.KindRemotePDS) && strings.TrimSpace(ref.Name) != "" {
		repoLinks = append(repoLinks, compose.ServiceAuthLink(p.guest(), compose.RemotePDSRefIn(strings.TrimSpace(ref.Name), ws), id.Status.Did))
	}
	if err := c.ApplyLinks(ctx, r, "repo", repoLinks); err != nil {
		return ctlruntime.Result{}, err
	}
	p.operatorSignIn(ctx, mount)
	if err := p.rotateSigningKey(ctx, mount); err != nil {
		p.log().Warn("rotating the bidder's OIDC signing key", "err", err.Error())
	}
	snap, err := p.snapshot(ctx)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	p.st.Did = id.Status.Did
	compose.FillApplicationStatus(&p.st, o.Generation, false, o.Spec.Components, desired, snap, nil)
	if snap.Satisfied(desired) {
		return p.ready("the existing account is signed in and every component is ready")
	}
	return p.wait("waiting for every component to be ready")
}

func (p *pass) local(ctx context.Context, plan compose.AccountPlan, id *v1alpha1.MarketIdentity, mount, pdsHandle, pdsPublicURL string, in ingress) (ctlruntime.Result, error) {
	c, r, o := p.c, p.r, p.o
	ws := r.Workspace
	desired, err := p.desired(ctx, plan, id, pdsHandle, pdsPublicURL, in)
	if err != nil {
		return p.fail(err)
	}
	first := compose.PdsApplySet(desired, compose.PdsComponent, compose.ComponentRefName(in.ref))
	if in.required && pdsPublicURL == "" {
		if applied, err := c.PdsPublicURLApplied(ctx, ws, compose.PdsComponent); err != nil {
			return ctlruntime.Result{}, err
		} else if applied {
			first = compose.WithoutPds(first)
		}
	}
	if err := p.c.EnsureAdding(ctx, p.r, first); err != nil {
		return p.retry("", err)
	}
	snap, err := p.snapshot(ctx)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if !snap.Ready(compose.ComponentKindPds, compose.PdsComponent) {
		return p.wait("waiting for the bidder's app Pds to be ready")
	}
	if in.required {
		if in.err != nil {
			compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, "", true, in.err)
			return p.wait("the bidder's app Pds ingress: " + in.err.Error())
		}
		if err := snap.PdsIngressReady(compose.PdsComponent, in.ref); err != nil {
			compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, "", true, err)
			return p.wait("waiting for the bidder's app Pds ingress: " + err.Error())
		}
		if pdsPublicURL, err = c.PdsIngressURL(ctx, ws, compose.PdsComponent); err != nil {
			compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, "", true, err)
			return p.write(compose.PdsIngressPhase(err), "the bidder's app Pds ingress: "+err.Error())
		}
		compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, "", true, nil)
	} else {
		pdsPublicURL, _ = c.PdsPublicURL(ctx, ws, compose.PdsComponent, false)
		compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, "", false, nil)
	}
	if pdsPublicURL == "" {
		return p.wait("waiting for the bidder's app Pds public URL")
	}
	if desired, err = p.desired(ctx, plan, id, pdsHandle, pdsPublicURL, in); err != nil {
		return p.fail(err)
	}
	pds, _ := compose.DesiredPds(desired)
	if err := p.c.EnsureAdding(ctx, p.r, []compose.Desired{pds}); err != nil {
		return p.retry("", err)
	}
	if snap, err = p.snapshot(ctx); err != nil {
		return ctlruntime.Result{}, err
	}
	if !snap.PdsApplied(compose.PdsComponent) {
		return p.wait("waiting for the bidder's app Pds to apply its public URL")
	}
	account := compose.PdsAccount{Vault: BidderVaultServer, Mount: mount, Path: compose.PdsSessionPath("bidder"), Handle: pdsHandle}
	accountLinks := compose.PdsAccountLinks(r, account)
	if err := c.ApplyLinks(ctx, r, "account", compose.WithoutLinks(accountLinks, p.baseLinks(mount))); err != nil {
		return ctlruntime.Result{}, err
	}
	if ok, msg := c.LinksAccepted(ctx, r, accountLinks); !ok {
		return p.wait(msg)
	}
	did, err := c.EnsurePdsAccount(ctx, r, account)
	if err != nil {
		return p.retry("its app Pds account", err)
	}
	acct := compose.LocalAccount{DID: did, Handle: pdsHandle, SessionID: BidderVaultServer + ":" + account.Path, PLC: o.Spec.Network != nil}
	repo := compose.RepoRefIn(did, ws)
	repoLinks := append(compose.LocalAccountLinks(ws, acct), compose.RepoWriteLink(p.self(), repo, compose.ScopeAll),
		compose.RecordWriteLink(ws, r.Owner(), repo, compose.ScopeAll),
		compose.ServiceAuthLink(p.self(), compose.AppRef(compose.ComponentKindPds, compose.PdsComponent, ws), did),
		compose.ServiceAuthLink(p.guest(), compose.AppRef(compose.ComponentKindPds, compose.PdsComponent, ws), did))
	if err := c.ApplyLinks(ctx, r, "repo", repoLinks); err != nil {
		return ctlruntime.Result{}, err
	}
	if err := c.ApplyLocalAccount(ctx, r, ws, acct); err != nil {
		return ctlruntime.Result{}, err
	}
	if err := c.ApplyAssociationRecord(ctx, r, ws, repo, BidderIdentity, compose.BidderAssociate); err != nil {
		return ctlruntime.Result{}, err
	}
	identity := compose.OperatorWatched(plan.Identity, o.Spec.Operator, compose.BidderAssociate)
	identity.Did, identity.PdsURL, identity.Handle = did, pdsPublicURL, pdsHandle
	if err := c.ApplyIdentitySpec(ctx, r, ws, BidderIdentity, identity); err != nil {
		return p.retry("its app Pds identity", err)
	}
	if id, err = c.Identity(ctx, ws, BidderIdentity); err != nil {
		return ctlruntime.Result{}, err
	}
	if desired, err = p.desired(ctx, plan, id, pdsHandle, pdsPublicURL, in); err != nil {
		return p.fail(err)
	}
	if err := p.c.EnsureLinked(ctx, p.r, desired, compose.WithoutPds(desired)); err != nil {
		return p.retry("", err)
	}
	if err := p.streams(ctx, plan, did); err != nil {
		return p.fail(err)
	}
	if err := p.prune(ctx, desired); err != nil {
		return ctlruntime.Result{}, err
	}
	p.st.Did = id.Status.Did
	if err := p.rotateSigningKey(ctx, mount); err != nil {
		p.log().Warn("rotating the bidder's OIDC signing key", "err", err.Error())
	}
	if refused, err := p.guardSignIn(ctx, id, mount); err != nil || refused != "" {
		if err != nil {
			return ctlruntime.Result{}, err
		}
		return p.wait(refused)
	}
	if snap, err = p.snapshot(ctx); err != nil {
		return ctlruntime.Result{}, err
	}
	if err := p.oauthEndpoints(ctx, snap); err != nil {
		return ctlruntime.Result{}, err
	}
	if o.Spec.Kcp != nil {
		p.st.Components = append(p.st.Components, p.guestComponents(ctx)...)
	}
	compose.FillApplicationStatus(&p.st, o.Generation, false, o.Spec.Components, desired, snap, nil)
	if o.Spec.Profile == BidderProfileNetwork {
		p.operator(ctx, id, mount)
	}
	p.operatorSignIn(ctx, mount)
	if snap.Satisfied(desired) && p.st.Did != "" {
		return p.ready("every component is ready")
	}
	return p.wait("waiting for every component to be ready and the bidder's MarketIdentity for a did")
}

func (p *pass) operator(ctx context.Context, id *v1alpha1.MarketIdentity, mount string) {
	c, r, o := p.c, p.r, p.o
	if o.Spec.Network == nil || o.Spec.Identity.Handle == "" || o.Spec.Identity.PdsURL == "" {
		return
	}
	op := compose.Operator{PdsURL: o.Spec.Identity.PdsURL, CARef: c.PublishNetworkTrust(r, o.Spec.Network), Identifier: o.Spec.Identity.Handle,
		Vault: BidderVaultServer, Mount: mount, Create: true}
	if err := c.ApplyLinks(ctx, r, "operator", compose.OperatorLinks(r, op)); err != nil {
		p.log().Warn("the operator account links", "err", err.Error())
		return
	}
	if compose.ClaimOpen(id) {
		operator, err := c.WriteClaim(ctx, r, op, id, "bidder-", compose.BidderAssociate)
		switch {
		case err != nil:
			p.log().Warn("associating the bidder's DID with the market network operator account", "err", err.Error())
		case operator != "":
			p.log().Info("the market network operator account wrote its claim record naming the bidder's own DID", "operator", operator,
				"handle", o.Spec.Identity.Handle, "bidderDid", id.Status.Did)
		}
	}
	if compose.OperatorSignInSet(o.Spec.Options) {
		return
	}
	if err := c.ApplyOperatorRepo(ctx, r, r.Workspace, op); err != nil {
		p.log().Warn("composing the repo the market network operator account's declared records write through", "err", err.Error())
	}
}

func (p *pass) operatorSignIn(ctx context.Context, mount string) {
	c, r, o := p.c, p.r, p.o
	if !compose.OperatorSignInSet(o.Spec.Options) {
		return
	}
	var caRef *commonv1.ObjectRef
	if compose.OptionValue(o.Spec.Options, compose.OperatorSignInModeOption) == string(v1alpha1.MarketIdentityModePdsAccount) {
		caRef = c.PublishNetworkTrust(r, o.Spec.Network)
	}
	s, err := compose.OperatorSignInOf(o.Spec.Options, BidderVaultServer, mount, caRef)
	if err != nil {
		p.log().Warn("bidder_operator_signin_refused", "err", err.Error())
		return
	}
	did, err := c.ApplyOperatorSignIn(ctx, r, s)
	switch {
	case err != nil:
		p.log().Warn("bidder_operator_signin", "account", s.Account, "mode", string(s.Mode), "err", err.Error())
	case did == "":
		p.log().Info("bidder_operator_signin_waiting", "account", s.Account, "mode", string(s.Mode), "identity", compose.OperatorRepoName)
	}
}

func (p *pass) guardSignIn(ctx context.Context, id *v1alpha1.MarketIdentity, mount string) (string, error) {
	c, r, o := p.c, p.r, p.o
	if p.ctl.guard == nil || id == nil {
		return "", nil
	}
	refusal := p.ctl.guard(o.Spec, id)
	if refusal == nil {
		return "", nil
	}
	p.log().Error("bidder_account_refused", "handle", o.Spec.Identity.Handle, "signedIn", id.Status.Handle,
		"did", id.Status.Did, "reason", refusal.Error())
	if ref := id.Status.SessionRef; ref != nil {
		refMount := cmp.Or(strings.Trim(ref.Mount, "/"), mount)
		if kv, err := c.KV(ctx, r, BidderVaultServer, refMount+"/identity/*"); err == nil {
			if err := kv.Delete(ctx, refMount, ref.Path); err != nil {
				p.log().Warn("removing the refused sign-in session", "path", ref.Path, "err", err.Error())
			}
		} else {
			p.log().Warn("opening the bidder's vault to remove the refused session", "err", err.Error())
		}
	}
	if _, err := c.DeleteComponentsExcept(ctx, r.Workspace, r.Owner(), nil); err != nil {
		return "", err
	}
	ri, err := c.Client(r.Workspace, compose.MarketIdentityGVR)
	if err != nil {
		return "", err
	}
	if err := ri.Delete(ctx, BidderIdentity, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
		return "", err
	}
	c.Forget(r)
	return "the sign-in was refused and the bidder restarts with a fresh QR: " + refusal.Error(), nil
}