johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package marketrequester

import (
	"context"
	"errors"
	"fmt"
	"strings"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"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"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/manifests"
	role "github.com/johnandersen777/socialweb-computer-kcp/internal/market/requester"
	pdssvc "github.com/johnandersen777/socialweb-computer-kcp/internal/services/pds"
)

const (
	Name = "marketrequester"

	Kind = string(v1alpha1.ApplicationKindMarketRequester)

	RequesterSet            = "market-requester"
	RequesterComponent      = "requester"
	RequesterApp            = "market-requester"
	RequesterVaultServer    = "requester"
	RequesterVaultClient    = "requester"
	RequesterAccount        = "requester"
	RequesterIdentity       = "requester"
	RequesterContract       = "contract"
	RequesterProfileNetwork = "market-network"
	RequesterProfileDefault = "default"
)

var Workflows = []string{"requester-tangled-vouch", "requester-only-me", "requester-mutuals", "open"}

type Controller struct {
	c *compose.Composer
}

func Register(m *ctlruntime.Manager, c *compose.Composer) error {
	ctl := &Controller{c: c}
	c.ProvideTrust(schema.GroupKind{Group: v1alpha1.GroupName, Kind: Kind})
	opts := append([]ctlruntime.Option{ctlruntime.Export(compose.ExportApplications), ctlruntime.Finalizer(compose.Finalizer)}, c.Watches(Kind)...)
	if err := ctlruntime.For[*v1alpha1.MarketRequester](m, Name, ctl, opts...); err != nil {
		return fmt.Errorf("marketrequester: registering the MarketRequester controller: %w", err)
	}
	return nil
}

type pass struct {
	ctx context.Context

	ctl *Controller

	c *compose.Composer

	r compose.Root

	o *v1alpha1.MarketRequester

	st v1alpha1.ApplicationStatus

	links []compose.Link

	identityPhase v1alpha1.MarketIdentityPhase
}

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.MarketRequesterGVR, 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) ingressDown(snap compose.Snapshot, plan compose.AccountPlan) string {
	name := compose.ServiceNameFor(compose.ComponentKindRequester, RequesterComponent) + compose.ExposureIngressSuffix
	mode := compose.RequesterCommsMode(p.o.Spec.Comms, compose.OAuthReceiveOnly(plan.Mode))
	if down := compose.SetIngressRegisteredCondition(&p.st.Conditions, p.o.Generation, snap, name, mode, p.o.Spec.SecretsFallback); down != "" {
		return "; " + down
	}
	return ""
}

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.MarketRequester) (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.st.Phase, p.identityPhase); reason != "" && !r.Suspend {
			c.Park(ctx, r, compose.RequesterStreamRole, reason)
			return res, nil
		}
		c.Adopt(ctx, r)
		c.ApplyQuotas(ctx, r, o.Spec.Quota)
	}
	return res, err
}

func (p *pass) delete(ctx context.Context) (ctlruntime.Result, error) {
	deployments, err := p.c.DeleteDeployments(ctx, p.r.Workspace)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if deployments > 0 {
		p.st.Phase, p.st.Message = v1alpha1.ApplicationPhaseDeleting, fmt.Sprintf("%d deployment(s) releasing", deployments)
		return ctlruntime.Result{Waiting: true}, p.c.WriteStatus(ctx, p.r, compose.MarketRequesterGVR, p.st)
	}
	contracts, err := p.c.DeleteAll(ctx, p.r.Workspace, compose.ComputeContractGVR)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if contracts > 0 {
		p.st.Phase, p.st.Message = v1alpha1.ApplicationPhaseDeleting, fmt.Sprintf("%d ComputeContract(s) releasing", contracts)
		return ctlruntime.Result{Waiting: true}, p.c.WriteStatus(ctx, p.r, compose.MarketRequesterGVR, p.st)
	}
	left, err := p.c.StopComponents(ctx, p.r, "")
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if left == 0 {
		if left, err = p.c.DeleteVaults(ctx, p.r); err != nil {
			return ctlruntime.Result{}, err
		}
	}
	if left > 0 {
		p.st.Phase, p.st.Message = v1alpha1.ApplicationPhaseDeleting, fmt.Sprintf("%d component(s) stopping", left)
		return ctlruntime.Result{Waiting: true}, p.c.WriteStatus(ctx, p.r, compose.MarketRequesterGVR, p.st)
	}
	p.c.Drop(p.r)
	return ctlruntime.Result{}, nil
}

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

func (p *pass) baseLinks(mount string) []compose.Link {
	self := p.self()
	return []compose.Link{
		compose.PolicyRunLink(self, compose.PolicyEngineRefIn(compose.PolicyEngineComponent, p.r.Workspace), Workflows...),
		compose.VaultLink(self, RequesterVaultServer, false, mount+"/*"),
		compose.VaultLink(self, RequesterVaultServer, true, mount+"/contracts/*"),
		compose.RootVaultLink(p.r, RequesterVaultServer, mount+"/identity/*"),
	}
}

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, "requester",
		compose.AccountVault{Server: RequesterVaultServer, Client: RequesterVaultClient, Mount: RequesterVaultClient},
		v1alpha1.AssociationRequireApproval)
	if err != nil {
		return p.fail(err)
	}
	mount := plan.Vault.MountOr(RequesterVaultClient)
	pdsAccountHandle := compose.PdsHandle(ws, "requester", o.Spec.Identity.Handle)
	ingressSpec := compose.AppIngressOrDefault(o.Spec.Profile == RequesterProfileDefault, o.Spec.Ingress, o.Spec.Network)
	requireIngress, ingressRef, ingressConfig, ingressErr := compose.PdsAppIngress(ingressSpec, o.Spec.Network, o.Spec.Profile == RequesterProfileDefault)
	pdsPublicURL, _ := c.PdsPublicURL(ctx, ws, compose.PdsComponent, requireIngress)
	if r.Suspend {
		return p.suspended(ctx, plan, pdsAccountHandle, pdsPublicURL, ingressConfig, requireIngress, ingressRef)
	}
	if err := c.ApplyVault(ctx, r, RequesterVaultServer); err != nil {
		return ctlruntime.Result{}, err
	}
	p.links = append(p.links, p.baseLinks(mount)...)
	intact, err := c.CheckWorkloads(ctx, r)(c.SetWorkloads(RequesterSet, o.Spec.Profile, &manifests.MarketRequester{}))
	if err != nil {
		return p.retry("its set workloads", err)
	}
	if !c.Applied(r) {
		digest, digestErr := compose.InfraDigest(RequesterSet, o.Spec.Profile, &manifests.MarketRequester{}, plan.Identity)
		if intact && digestErr == nil && c.AppliedBefore(r, o.Status.AppliedSet, digest) {
			c.MarkApplied(r)
		} else {
			if _, err := p.write(v1alpha1.ApplicationPhaseProvisioning, "applying its schemas, its own OpenBaoServer and its workspace"); err != nil {
				return ctlruntime.Result{}, err
			}
			if err := c.ApplyInfra(ctx, r, RequesterSet, o.Spec.Profile, &manifests.MarketRequester{}); err != nil {
				return p.retry("", err)
			}
			if err := c.ApplyIdentitySpec(ctx, r, ws, RequesterIdentity, plan.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}
			}
		}
	}
	if err := c.ApplyLinks(ctx, r, "root", p.links); err != nil {
		return ctlruntime.Result{}, err
	}
	if _, _, err := c.VaultRefs(ctx, ws, RequesterVaultServer, RequesterVaultClient); err != nil {
		return p.retry("its vault", err)
	}
	if ok, msg := c.LinksAccepted(ctx, r, p.links); !ok {
		return p.wait(msg)
	}
	if err := c.SeedPdsAccountPassword(ctx, r, plan); err != nil {
		return p.retry("its account password", err)
	}
	id, err := c.Identity(ctx, ws, RequesterIdentity)
	if apierrors.IsNotFound(err) {
		if err := c.ApplyIdentitySpec(ctx, r, ws, RequesterIdentity, plan.Identity); 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)
	}
	return p.local(ctx, plan, id, pdsAccountHandle, pdsPublicURL, ingressConfig, requireIngress, ingressRef, ingressErr)
}

func (p *pass) desired(ctx context.Context, plan compose.AccountPlan, id *v1alpha1.MarketIdentity, pdsHandle, pdsPublicURL string,
	ingress *v1alpha1.IngressConfig, requireIngress bool, ingressRef *commonv1.ObjectRef) ([]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, RequesterVaultServer)
	var hostnames []string
	if ident.Handle != "" {
		hostnames = []string{ident.Handle}
	}
	mode, guestSSHKeysFile := compose.RequesterComponentFields(o.Spec.Options, plan.Mode)
	spec := v1alpha1.RequesterSpec{
		ComponentSpec:        p.c.NetworkComponentSpec(p.r, compose.ComponentCRStateDir(stateRoot, compose.ComponentKindRequester, RequesterComponent), network),
		Identity:             ident,
		PolicyEngineRef:      compose.ComponentRefFor(compose.ComponentKindPolicyEngine, compose.PolicyEngineComponent),
		VaultRef:             vault,
		Hostnames:            hostnames,
		Mode:                 mode,
		MarketExposureRef:    &commonv1.LocalObjectRef{Name: compose.ServiceNameFor(compose.ComponentKindRequester, RequesterComponent) + compose.ExposureIngressSuffix},
		OperatorDID:          compose.OperatorDID(o.Spec.Identity.ApprovedAssociations, id),
		GuestSSHKeysFile:     guestSSHKeysFile,
		BidderDIDs:           append([]string(nil), o.Spec.BidderDids...),
		Association:          o.Spec.Association.DeepCopy(),
		AcceptCopies:         int32(compose.OptionInt(o.Spec.Options, "accept-copies")),
		Comms:                o.Spec.Comms,
		Ingress:              o.Spec.RequesterIngress,
		SecretsFallback:      o.Spec.SecretsFallback,
		AtircParentRef:       o.Spec.AtircParentRef.DeepCopy(),
		AtircMintedRef:       &commonv1.SecretRef{Mount: plan.Vault.MountOr(RequesterVaultClient), Path: role.AtircMintedPath},
		SuspendParticipation: o.Spec.Suspend,
	}
	out := []compose.Desired{compose.PolicyEngineDesired()}
	if plan.Local {
		pdsStateDir := compose.OptionValue(o.Spec.Options, "pds-state-dir")
		identityStore := strings.TrimSpace(pdsStateDir) != ""
		existing := identityStore && (contractDID(plan, id) != "" || strings.TrimSpace(compose.OptionValue(o.Spec.Options, "expected-pds-url")) != "")
		pdsSpec, err := compose.AppPdsSpec(stateRoot, pdsStateDir, existing, network, pdsHandle, vault, hostnames, pdsPublicURL, requireIngress, ingressRef)
		if err != nil {
			return nil, err
		}
		if existing {
			dir := compose.PdsStateDir(stateRoot, pdsStateDir)
			if pdsSpec.StateDir != dir || pdsSpec.Storage != nil {
				return nil, fmt.Errorf("compose: the reused Pds state dir is %s, not %s, or carries storage overrides", pdsSpec.StateDir, dir)
			}
		}
		out = append(out, compose.Desired{Kind: compose.ComponentKindPds, Name: compose.PdsComponent, Spec: pdsSpec})
		spec.PdsRef = compose.ComponentRefFor(compose.ComponentKindPds, compose.PdsComponent)
	} else {
		spec.PdsRef = plan.PdsRef
		spec.PdsDID = plan.PdsDID
	}
	out = append(out, compose.Desired{Kind: compose.ComponentKindRequester, Name: RequesterComponent, Spec: spec})
	inbound := o.Spec.Comms == nil || o.Spec.Comms.Mode != v1alpha1.MarketCommsModeFirehose
	out = append(out, compose.HTTPServiceExposures(ws, compose.ComponentKindRequester, RequesterComponent, "http", network, hostnames, false, inbound)...)
	if plan.Local {
		out = append(out, compose.PdsExposures(ws, compose.PdsComponent, network, hostnames, ingress)...)
	}
	return compose.ApplyOverrides(o.Spec.Components, out)
}

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
}

func (p *pass) suspended(ctx context.Context, plan compose.AccountPlan, pdsHandle, pdsPublicURL string,
	ingress *v1alpha1.IngressConfig, requireIngress bool, ingressRef *commonv1.ObjectRef) (ctlruntime.Result, error) {
	c, r, o := p.c, p.r, p.o
	id, _ := c.Identity(ctx, r.Workspace, RequesterIdentity)
	desired, err := p.desired(ctx, plan, id, pdsHandle, pdsPublicURL, ingress, requireIngress, ingressRef)
	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
	}
	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.MarketRequesterGVR, p.st)
}

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

func (p *pass) account(ctx context.Context, plan compose.AccountPlan, id *v1alpha1.MarketIdentity) (ctlruntime.Result, error) {
	c, r, o := p.c, p.r, p.o
	ws := r.Workspace
	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:
		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, "", "", nil, false, nil)
	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, RequesterIdentity, plan, id.Status.Did, id.Status.PdsURL)
	if err != nil {
		c.Log().Warn("the existing-account repo", "root", r.Owner(), "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, RequesterIdentity, ws), id.Status.Did))
	repoLinks = append(repoLinks, compose.InfraLinks(ws, compose.RepoRefIn(id.Status.Did, ws))...)
	if err := c.ApplyLinks(ctx, r, "repo", repoLinks); err != nil {
		return ctlruntime.Result{}, err
	}
	snap, err := p.snapshot(ctx)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if err := c.ApplyOAuthIdentityEndpoints(ctx, r, snap, ws, RequesterIdentity, compose.ServiceNameFor(compose.ComponentKindRequester, RequesterComponent)+compose.ExposureIngressSuffix, "requester", o.Spec.Identity); 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)
	down := p.ingressDown(snap, plan)
	if snap.Satisfied(desired) {
		p.st = compose.Ready(p.st, "the existing account is signed in and every component is ready"+down)
		return ctlruntime.Result{}, c.WriteStatus(ctx, r, compose.MarketRequesterGVR, p.st)
	}
	return p.wait("waiting for every component to be ready")
}

func (p *pass) local(ctx context.Context, plan compose.AccountPlan, id *v1alpha1.MarketIdentity, pdsHandle, pdsPublicURL string,
	ingress *v1alpha1.IngressConfig, requireIngress bool, ingressRef *commonv1.ObjectRef, ingressErr error) (ctlruntime.Result, error) {
	c, r, o := p.c, p.r, p.o
	ws := r.Workspace
	pdsStateDir := compose.PdsStateDir(c.StateRoot(), compose.OptionValue(o.Spec.Options, "pds-state-dir"))
	identityStore := strings.TrimSpace(compose.OptionValue(o.Spec.Options, "pds-state-dir")) != ""
	expectedURL := strings.TrimSpace(compose.OptionValue(o.Spec.Options, "expected-pds-url"))
	didHint := ""
	if identityStore {
		didHint = contractDID(plan, id)
	}
	existing := identityStore && (didHint != "" || expectedURL != "")
	if expectedURL != "" && didHint == "" {
		return p.fail(errors.New("compose: the identity store names a PDS URL but no DID; the store is partial"))
	}
	if existing && didHint != "" {
		wantOAuth := expectedURL != "" || pdsPublicURL != "" || !requireIngress
		if err := validatePDSStateDir(pdsStateDir, didHint, wantOAuth); err != nil {
			return p.fail(err)
		}
	}
	if err := checkExpectedPDSURL(expectedURL, pdsPublicURL, existing); err != nil {
		return p.fail(err)
	}
	desired, err := p.desired(ctx, plan, id, pdsHandle, pdsPublicURL, ingress, requireIngress, ingressRef)
	if err != nil {
		return p.fail(err)
	}
	first := compose.PdsApplySet(desired, compose.PdsComponent, compose.ComponentRefName(ingressRef))
	if requireIngress && 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 requester's app Pds to be ready")
	}
	if requireIngress {
		if ingressErr != nil {
			compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, "", true, ingressErr)
			return p.wait("the app Pds ingress: " + ingressErr.Error())
		}
		if err := snap.PdsIngressReady(compose.PdsComponent, ingressRef); err != nil {
			compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, "", true, err)
			return p.wait("waiting for the 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 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 app Pds public URL")
	}
	if err := checkExpectedPDSURL(expectedURL, pdsPublicURL, existing); err != nil {
		return p.fail(err)
	}
	if desired, err = p.desired(ctx, plan, id, pdsHandle, pdsPublicURL, ingress, requireIngress, ingressRef); err != nil {
		return p.fail(err)
	}
	pds, _ := compose.DesiredPds(desired)
	if spec, ok := pds.Spec.(v1alpha1.PdsSpec); ok {
		if err := checkExpectedPDSURL(expectedURL, spec.PublicURL, existing); err != nil {
			return p.fail(err)
		}
	}
	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 app Pds to apply its public URL")
	}
	account := compose.PdsAccount{Vault: RequesterVaultServer, Mount: plan.Vault.MountOr(RequesterVaultClient), Path: compose.PdsSessionPath("requester"),
		Handle: pdsHandle, AdoptDID: didHint, Sync: true}
	if err := c.ApplyLinks(ctx, r, "account", compose.WithoutLinks(compose.PdsAccountLinks(r, account), p.links)); err != nil {
		return ctlruntime.Result{}, err
	}
	if ok, msg := c.LinksAccepted(ctx, r, compose.PdsAccountLinks(r, account)); !ok {
		return p.wait(msg)
	}
	did, err := c.EnsurePdsAccount(ctx, r, account)
	if err != nil {
		if errors.Is(err, compose.ErrIdentityAccountMissing) || errors.Is(err, compose.ErrIdentityInvalid) {
			return p.fail(err)
		}
		return p.retry("its app Pds account", err)
	}
	acct := compose.LocalAccount{DID: did, Handle: pdsHandle, SessionID: RequesterVaultServer + ":" + account.Path, PLC: o.Spec.Network != nil}
	repoLinks := append(compose.LocalAccountLinks(ws, acct), compose.RepoWriteLink(p.self(), compose.RepoRefIn(did, ws), compose.ScopeAll),
		compose.RecordWriteLink(ws, r.Owner(), compose.RepoRefIn(did, ws), compose.ScopeAll))
	repoLinks = append(repoLinks, compose.InfraLinks(ws, compose.RepoRefIn(did, ws))...)
	if requireIngress && localPdsSignsAtirc(o.Spec.Network, pdsPublicURL) {
		repoLinks = append(repoLinks, compose.ServiceAuthLink(p.self(), 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 plan.Identity.AssociationRef != nil {
		if err := c.ApplyAssociationRecord(ctx, r, ws, compose.RepoRefIn(did, ws), RequesterIdentity, compose.RequesterAssociate); err != nil {
			return ctlruntime.Result{}, err
		}
	}
	if req := o.Spec.Identity.AssociationRequest; req != nil {
		if err := c.ApplyAssociationRequestRecord(ctx, r, ws, compose.RepoRefIn(did, ws), RequesterIdentity, compose.RequesterAssociate, *req); err != nil {
			return p.fail(err)
		}
	}
	requesterDir := compose.ComponentCRStateDir(c.StateRoot(), compose.ComponentKindRequester, RequesterComponent)
	changed, err := compose.SyncRequesterRepoKey(c.StateRoot(), pdsStateDir, did, requesterDir)
	if err != nil {
		return p.retry("its app Pds signing key", err)
	}
	if changed {
		if err := c.RestartComponent(ctx, r, ws, compose.ComponentKindRequester, RequesterComponent); err != nil && !apierrors.IsNotFound(err) {
			return p.retry("restarting the requester", err)
		}
	}
	identity := plan.Identity
	identity.Did, identity.PdsURL, identity.Handle = did, pdsPublicURL, pdsHandle
	if err := c.ApplyIdentitySpec(ctx, r, ws, RequesterIdentity, identity); err != nil {
		return p.retry("its app Pds identity", err)
	}
	if id, err = c.Identity(ctx, ws, RequesterIdentity); err != nil {
		return ctlruntime.Result{}, err
	}
	p.identityPhase = id.Status.Phase
	if desired, err = p.desired(ctx, plan, id, pdsHandle, pdsPublicURL, ingress, requireIngress, ingressRef); 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
	}
	if snap, err = p.snapshot(ctx); 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)
	down := p.ingressDown(snap, plan)
	switch msg := compose.AssociationWait(plan, id, compose.RequesterAssociate); {
	case msg != "":
		return p.wait(msg)
	case snap.Satisfied(desired) && p.st.Did != "":
		p.st = compose.Ready(p.st, "the requester is signed in and its components are ready"+down)
		return ctlruntime.Result{}, c.WriteStatus(ctx, r, compose.MarketRequesterGVR, p.st)
	case p.st.Did == "":
		return p.wait("waiting for the requester's MarketIdentity to sign in")
	}
	return p.wait("waiting for every component to be ready")
}

func localPdsSignsAtirc(network *v1alpha1.MarketNetworkValue, pdsPublicURL string) bool {
	if network == nil {
		return false
	}
	ok, _ := role.LocalPdsSignsAtirc(network.PLCDirectoryURL, pdsPublicURL)
	return ok
}

func validatePDSStateDir(dir, did string, wantOAuth bool) error {
	_, err := pdssvc.ValidateStateDir(dir, did, wantOAuth)
	return err
}

func checkExpectedPDSURL(expected, got string, existing bool) error {
	if !existing || expected == "" || got == "" {
		return nil
	}
	if compose.SamePDSURL(expected, got) {
		return nil
	}
	return fmt.Errorf("compose: the identity PDS public URL is %s, not the recorded %s; refusing to rewrite the DID document", got, expected)
}

func contractDID(plan compose.AccountPlan, id *v1alpha1.MarketIdentity) string {
	if did := strings.TrimSpace(plan.Identity.Did); did != "" {
		return did
	}
	if id == nil {
		return ""
	}
	if did := strings.TrimSpace(id.Status.Did); did != "" {
		return did
	}
	return strings.TrimSpace(id.Spec.Did)
}