johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package request

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"

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

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/marketnetwork"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/marketrequester"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpserver"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/fsx"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/supervise"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/market/requester"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/vault/openbao"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
)

const (
	RequestsDir                  = "requests"
	InvocationFile               = "invocation.json"
	SummaryFile                  = "summary.json"
	SSHDir                       = "ssh"
	ModeNetwork                  = "network"
	ModeStandalone               = "standalone"
	RootName                     = "app"
	FieldManager                 = "socialweb-computer-run-request"
	ReleaseFieldManager          = "socialweb.computer/run-request"
	AssociateFieldManager        = "socialweb.computer/run-request-associate"
	NetworkAppFieldManager       = "socialweb.computer/network-app"
	ContractTTLSeconds     int32 = 8 * 24 * 60 * 60
	SecretsPath                  = "contracts/" + marketrequester.RequesterContract + "/secrets"
	PdsAccountKeysPath           = "identity/" + marketrequester.RequesterIdentity + "/keys"
	AtircParentPath              = requester.AtircParentPath
	Poll                         = time.Second
	RetryMax                     = 5 * time.Second
	irohScheme                   = "iroh://"
	maxNameLength                = 30
)

var nameRE = regexp.MustCompile(`^[a-z0-9]([-a-z0-9]*[a-z0-9])?$`)

func CheckName(name string) error {
	if !nameRE.MatchString(name) || len(name) > maxNameLength {
		return fmt.Errorf("request: name %q is not a DNS label of at most %d characters", name, maxNameLength)
	}
	return nil
}

func Root(stateRoot string) string { return filepath.Join(stateRoot, RequestsDir) }

func Dir(stateRoot, name string) string { return filepath.Join(Root(stateRoot), name) }

type Invocation struct {
	Name      string    `json:"name"`
	Mode      string    `json:"mode"`
	Network   string    `json:"network,omitempty"`
	Instance  string    `json:"instance"`
	Identity  string    `json:"identity,omitempty"`
	StateRoot string    `json:"stateRoot"`
	CreatedAt time.Time `json:"createdAt"`
}

type ContractSummary struct {
	Phase         string          `json:"phase,omitempty"`
	WinnerDid     string          `json:"winnerDid,omitempty"`
	Bids          int             `json:"bids,omitempty"`
	ReceiptUri    string          `json:"receiptUri,omitempty"`
	VMAddress     string          `json:"vmAddress,omitempty"`
	Networks      []string        `json:"networks,omitempty"`
	HostKeys      []string        `json:"hostKeys,omitempty"`
	ReleaseOK     *bool           `json:"releaseOk,omitempty"`
	ReleaseStatus int             `json:"releaseStatus,omitempty"`
	Termination   string          `json:"termination,omitempty"`
	Sandbox       *SandboxSummary `json:"sandbox,omitempty"`
	R2Release     string          `json:"r2Release,omitempty"`
	AtircRevokes  string          `json:"atircRevokes,omitempty"`
}

type SandboxSummary struct {
	ExitCode int    `json:"exitCode"`
	TimedOut bool   `json:"timedOut,omitempty"`
	Stdout   string `json:"stdout,omitempty"`
	Stderr   string `json:"stderr,omitempty"`
}

type Summary struct {
	Name             string           `json:"name"`
	Mode             string           `json:"mode"`
	Network          string           `json:"network,omitempty"`
	Instance         string           `json:"instance"`
	Identity         string           `json:"identity,omitempty"`
	KcpURL           string           `json:"kcpURL,omitempty"`
	KcpColdSeconds   float64          `json:"kcpColdSeconds,omitempty"`
	Did              string           `json:"did,omitempty"`
	Phase            string           `json:"phase,omitempty"`
	Message          string           `json:"message,omitempty"`
	SecretsUrl       string           `json:"secretsUrl,omitempty"`
	SecretsFetchedAt string           `json:"secretsFetchedAt,omitempty"`
	Contract         *ContractSummary `json:"contract,omitempty"`
	KcpRunning       bool             `json:"kcpRunning"`
	CreatedAt        time.Time        `json:"createdAt"`
	UpdatedAt        time.Time        `json:"updatedAt"`
}

func WriteInvocation(inv Invocation) error {
	return WriteInvocationIn(Dir(inv.StateRoot, inv.Name), inv)
}

func WriteInvocationIn(dir string, inv Invocation) error {
	return fsx.WriteJSON(filepath.Join(dir, InvocationFile), inv, 0o600)
}

func ReadInvocation(stateRoot, name string) (Invocation, error) {
	var inv Invocation
	if err := CheckName(name); err != nil {
		return inv, err
	}
	if err := fsx.ReadJSON(filepath.Join(Dir(stateRoot, name), InvocationFile), &inv); err != nil {
		return inv, fmt.Errorf("request: no invocation %s under %s: %w", name, Root(stateRoot), err)
	}
	return inv, nil
}

func WriteSummary(stateRoot string, s Summary) error {
	s.UpdatedAt = time.Now().UTC()
	return fsx.WriteJSON(filepath.Join(Dir(stateRoot, s.Name), SummaryFile), s, 0o600)
}

func ReadSummary(stateRoot, name string) (Summary, error) {
	var s Summary
	err := fsx.ReadJSON(filepath.Join(Dir(stateRoot, name), SummaryFile), &s)
	return s, err
}

func List(stateRoot string) ([]Summary, error) {
	entries, err := os.ReadDir(Root(stateRoot))
	if errors.Is(err, os.ErrNotExist) {
		return []Summary{}, nil
	}
	if err != nil {
		return nil, err
	}
	out := []Summary{}
	for _, e := range entries {
		if !e.IsDir() {
			continue
		}
		s, err := ReadSummary(stateRoot, e.Name())
		if err != nil {
			inv, ierr := ReadInvocation(stateRoot, e.Name())
			if ierr != nil {
				continue
			}
			s = Summary{Name: inv.Name, Mode: inv.Mode, Network: inv.Network, Instance: inv.Instance, CreatedAt: inv.CreatedAt,
				Message: "no summary yet: the invocation did not finish writing one"}
		}
		out = append(out, s)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].CreatedAt.Before(out[j].CreatedAt) })
	return out, nil
}

func SummaryOf(c *v1alpha1.ComputeContract) *ContractSummary {
	if c == nil || c.Status.Contract == nil {
		return nil
	}
	k := c.Status.Contract
	s := &ContractSummary{Phase: string(k.Phase), VMAddress: k.VMAddress}
	for _, n := range k.Networks {
		s.Networks = append(s.Networks, n.Tag+"="+n.Address)
	}
	if k.Winner != nil {
		s.WinnerDid = k.Winner.DID
	}
	if k.Bids != nil {
		s.Bids = *k.Bids
	}
	if k.Receipt != nil {
		s.ReceiptUri = k.Receipt.URI
	}
	if k.R2Release != nil {
		s.R2Release = string(k.R2Release.Phase)
		if k.R2Release.Message != "" {
			s.R2Release += ": " + k.R2Release.Message
		}
	}
	s.AtircRevokes = string(c.Status.AtircRevokes)
	if k.Sandbox != nil {
		s.Sandbox = &SandboxSummary{ExitCode: k.Sandbox.ExitCode, TimedOut: k.Sandbox.TimedOut, Stdout: k.Sandbox.Stdout, Stderr: k.Sandbox.Stderr}
	}
	if k.HostKeys != nil {
		for _, h := range k.HostKeys.Keys {
			s.HostKeys = append(s.HostKeys, h.Type+" "+h.Fingerprint)
		}
	}
	if k.Release != nil {
		ok := k.Release.OK
		s.ReleaseOK = &ok
		if k.Release.Status != nil {
			s.ReleaseStatus = *k.Release.Status
		}
	}
	if k.Termination != nil {
		s.Termination = k.Termination.Reason
	}
	return s
}

type Kcp struct {
	rest      *rest.Config
	dyn       *kube.Client
	workspace string
	contract  string
}

func (k *Kcp) ForContract(name string) *Kcp {
	out := *k
	out.contract = name
	return &out
}

func (k *Kcp) ContractName() string {
	return cmp.Or(k.contract, marketrequester.RequesterContract)
}

func OpenKcp(kubeconfig, workspace string) (*Kcp, error) {
	cfg, err := kcpserver.FrontRestConfig(kubeconfig)
	if err != nil {
		return nil, err
	}
	workspace = cmp.Or(workspace, compose.DefaultWorkspace)
	root := rest.CopyConfig(cfg)
	root.Host = strings.TrimSuffix(cfg.Host, "/") + "/clusters/" + workspace
	dc, err := kube.New(root)
	if err != nil {
		return nil, err
	}
	return &Kcp{rest: cfg, dyn: dc, workspace: workspace}, nil
}

func typeMeta(kind string) metav1.TypeMeta {
	return metav1.TypeMeta{APIVersion: v1alpha1.GroupVersion.String(), Kind: kind}
}

func (k *Kcp) ApplyRoot(ctx context.Context, spec v1alpha1.MarketRequesterSpec) error {
	if err := compose.AwaitBound(ctx, k.dyn.Resource(compose.APIBindingGVR), compose.RootBindTimeout, compose.RootExports...); err != nil {
		return err
	}
	obj := &v1alpha1.MarketRequester{TypeMeta: typeMeta("MarketRequester"), ObjectMeta: metav1.ObjectMeta{Name: RootName}, Spec: spec}
	_, err := kube.Apply[kube.Doc](ctx, k.dyn.Resource(compose.MarketRequesterGVR), obj, kube.ApplyOptions{FieldManager: FieldManager, Force: true})
	return err
}

func (k *Kcp) Root(ctx context.Context) (v1alpha1.ApplicationStatus, error) {
	o, err := kube.Get[v1alpha1.MarketRequester](ctx, k.dyn.Resource(compose.MarketRequesterGVR), RootName)
	if err != nil {
		return v1alpha1.ApplicationStatus{}, err
	}
	return o.Status, nil
}

func (k *Kcp) WaitReady(ctx context.Context, timeout time.Duration, progress func(v1alpha1.ApplicationStatus)) error {
	last := ""
	var failed error
	err := await(ctx, k.dyn.Resource(compose.MarketRequesterGVR), compose.MarketRequesterGVR.GroupResource(), RootName, timeout, "waiting for the MarketRequester to be Ready",
		func(o *v1alpha1.MarketRequester) bool {
			st := o.Status
			if key := string(st.Phase) + st.Message; key != last && progress != nil {
				progress(st)
				last = key
			}
			if st.Phase == v1alpha1.ApplicationPhaseFailed {
				failed = fmt.Errorf("request: the MarketRequester failed: %s", st.Message)
				return true
			}
			if st.Phase != v1alpha1.ApplicationPhaseReady || len(st.Components) == 0 {
				return false
			}
			for _, c := range st.Components {
				if !c.Ready {
					return false
				}
			}
			return true
		})
	if err != nil {
		return err
	}
	return failed
}

func (k *Kcp) FollowRoot(ctx context.Context, each func(*v1alpha1.MarketRequester)) error {
	_, err := kube.Await(ctx, k.dyn.Resource(compose.MarketRequesterGVR), RootName, 0, func(o *v1alpha1.MarketRequester) (bool, error) {
		if o != nil {
			each(o)
		}
		return false, nil
	})
	return err
}

func (k *Kcp) Contract(ctx context.Context) (*v1alpha1.ComputeContract, error) {
	return kube.Get[v1alpha1.ComputeContract](ctx, k.dyn.Resource(compose.ComputeContractGVR), k.ContractName())
}

func await[T any](ctx context.Context, r kube.Resource, gr schema.GroupResource, name string, timeout time.Duration, what string, done func(*T) bool) error {
	var missing error
	_, err := kube.Await(ctx, r, name, timeout, func(o *T) (bool, error) {
		if o == nil {
			missing = apierrors.NewNotFound(gr, name)
			return false, nil
		}
		return done(o), nil
	})
	return awaitErr(err, what, timeout, missing)
}

func awaitErr(err error, what string, timeout time.Duration, last error) error {
	if !errors.Is(err, kube.ErrAwaitTimeout) {
		return err
	}
	if last != nil {
		return fmt.Errorf("request: %s within %s: %w", what, timeout, last)
	}
	return fmt.Errorf("request: %s within %s", what, timeout)
}

func retry(ctx context.Context, deadline time.Time, timeout time.Duration, what string, try func() (bool, error)) error {
	delay := Poll
	var last error
	for {
		done, err := try()
		if done {
			return err
		}
		if err != nil {
			last = err
		}
		left := time.Until(deadline)
		if left <= 0 {
			return awaitErr(kube.ErrAwaitTimeout, what, timeout, last)
		}
		t := time.NewTimer(min(delay, left))
		select {
		case <-ctx.Done():
			t.Stop()
			return ctx.Err()
		case <-t.C:
		}
		delay = min(2*delay, RetryMax)
	}
}

func (k *Kcp) CreateContract(ctx context.Context, spec v1alpha1.ComputeContractSpec, timeout time.Duration) error {
	if spec.TTLSecondsAfterFinished == nil {
		ttl := ContractTTLSeconds
		spec.TTLSecondsAfterFinished = &ttl
	}
	body, err := kube.ApplyBody(&v1alpha1.ComputeContract{TypeMeta: typeMeta("ComputeContract"),
		ObjectMeta: metav1.ObjectMeta{Name: marketrequester.RequesterContract}, Spec: spec}, false)
	if err != nil {
		return err
	}
	return retry(ctx, time.Now().Add(timeout), timeout, "creating the ComputeContract (its schema is bound by the requester's core)", func() (bool, error) {
		_, err := kube.Create[kube.Doc](ctx, k.dyn.Resource(compose.ComputeContractGVR), body, metav1.CreateOptions{FieldManager: FieldManager})
		if err == nil || apierrors.IsAlreadyExists(err) {
			return true, nil
		}
		return false, err
	})
}

func (k *Kcp) Release(ctx context.Context, requestedAt string) error {
	contracts := k.dyn.Resource(compose.ComputeContractGVR)
	if _, err := kube.Get[kube.Doc](ctx, contracts, marketrequester.RequesterContract); err != nil {
		return err
	}
	doc := kube.Doc{"apiVersion": v1alpha1.GroupVersion.String(), "kind": "ComputeContract",
		"metadata": map[string]any{"name": marketrequester.RequesterContract}, "spec": map[string]any{"release": true, "releaseRequestedAt": requestedAt}}
	_, err := kube.Apply[kube.Doc](ctx, contracts, doc, kube.ApplyOptions{FieldManager: ReleaseFieldManager, Force: true})
	return err
}

func (k *Kcp) Associate(ctx context.Context, subject string) error {
	contracts := k.dyn.Resource(compose.ComputeContractGVR)
	if _, err := kube.Get[kube.Doc](ctx, contracts, marketrequester.RequesterContract); err != nil {
		return err
	}
	doc := kube.Doc{"apiVersion": v1alpha1.GroupVersion.String(), "kind": "ComputeContract",
		"metadata": map[string]any{"name": marketrequester.RequesterContract}, "spec": map[string]any{"association": map[string]any{"subject": subject}}}
	_, err := kube.Apply[kube.Doc](ctx, contracts, doc, kube.ApplyOptions{FieldManager: AssociateFieldManager, Force: true})
	return err
}

func (k *Kcp) WaitContract(ctx context.Context, timeout time.Duration, done func(*v1alpha1.ComputeContract) bool,
	progress func(*v1alpha1.ComputeContract)) (*v1alpha1.ComputeContract, error) {
	wctx, cancel := context.WithCancel(ctx)
	defer cancel()
	watch := supervise.New(wctx)
	defer func() { _ = watch.Stop() }()
	var mu sync.Mutex
	why := ""
	watch.Go("request: requester failure", func(ctx context.Context) error {
		_, _ = kube.Await(ctx, k.dyn.Resource(compose.MarketRequesterGVR), RootName, 0, func(o *v1alpha1.MarketRequester) (bool, error) {
			if o == nil {
				return false, nil
			}
			w := requesterFailure(o.Status)
			mu.Lock()
			why = w
			mu.Unlock()
			if w != "" {
				cancel()
			}
			return w != "", nil
		})
		return nil
	})
	failure := func() string {
		mu.Lock()
		defer mu.Unlock()
		return why
	}
	var cur *v1alpha1.ComputeContract
	var missing error
	last := ""
	_, err := kube.Await(wctx, k.dyn.Resource(compose.ComputeContractGVR), k.ContractName(), timeout,
		func(c *v1alpha1.ComputeContract) (bool, error) {
			if c == nil {
				missing = apierrors.NewNotFound(compose.ComputeContractGVR.GroupResource(), k.ContractName())
				return false, nil
			}
			cur = c
			key := string(c.Status.Phase) + "|" + c.Status.Message
			if c.Status.Contract != nil {
				key += "|" + string(c.Status.Contract.Phase)
			}
			if key != last && progress != nil {
				progress(c)
				last = key
			}
			return done(c), nil
		})
	if cur != nil && done(cur) {
		return cur, nil
	}
	if w := failure(); w != "" && ctx.Err() == nil {
		return cur, fmt.Errorf("request: the requester application cannot carry the contract: %s", w)
	}
	return cur, awaitErr(err, "waiting for the ComputeContract", timeout, missing)
}

func requesterFailure(st v1alpha1.ApplicationStatus) string {
	if st.Phase == v1alpha1.ApplicationPhaseFailed {
		return "phase Failed: " + st.Message
	}
	for _, c := range st.Components {
		if c.Phase == string(v1alpha1.ComponentPhaseFailed) {
			return fmt.Sprintf("%s %s/%s is Failed", c.Kind, c.Workspace, c.Name)
		}
	}
	return ""
}

func (k *Kcp) PutSecrets(ctx context.Context, bundle []byte, timeout time.Duration) (commonv1.SecretRef, error) {
	return k.PutSecret(ctx, SecretsPath, map[string]any{"bundle": string(bundle)}, "the secrets bundle", timeout)
}

func (k *Kcp) PutPdsAccountPassword(ctx context.Context, password string, timeout time.Duration) (commonv1.SecretRef, error) {
	return k.writeVault(ctx, PdsAccountKeysPath, "the pdsAccount password", timeout, func(c *openbao.Client, full string) error {
		data := map[string]any{}
		cur, err := c.GetKV(ctx, full)
		var re *openbao.ResponseError
		if err != nil && !(errors.As(err, &re) && re.Status == http.StatusNotFound) {
			return err
		}
		if inner, ok := cur["data"].(map[string]any); ok {
			data = inner
		}
		data["password"] = password
		return c.PutKV(ctx, full, data)
	})
}

func (k *Kcp) PutSecret(ctx context.Context, path string, data map[string]any, what string, timeout time.Duration) (commonv1.SecretRef, error) {
	return k.writeVault(ctx, path, what, timeout, func(c *openbao.Client, full string) error { return c.PutKV(ctx, full, data) })
}

func (k *Kcp) writeVault(ctx context.Context, path, what string, timeout time.Duration, write func(c *openbao.Client, full string) error) (commonv1.SecretRef, error) {
	ref := commonv1.SecretRef{Mount: marketrequester.RequesterVaultClient, Path: path}
	what = "writing " + what + " into the requester's own OpenBao"
	deadline := time.Now().Add(timeout)
	ri := k.dyn.Resource(compose.OpenBaoServerGVR)
	server, client := marketrequester.RequesterVaultServer, marketrequester.RequesterVaultClient
	var pending error
	_, err := kube.Await(ctx, ri, server, timeout, func(o *v1alpha1.OpenBaoServer) (bool, error) {
		pending = vaultPending(o, server, client)
		return pending == nil, nil
	})
	if err != nil {
		return ref, awaitErr(err, what, timeout, pending)
	}
	err = retry(ctx, deadline, timeout, what, func() (bool, error) {
		c, _, refs, err := compose.ServerClient(ctx, ri, server, client)
		if err != nil {
			return false, err
		}
		defer c.RevokeSelf(context.WithoutCancel(ctx))
		ref.Mount = refs.KVMount
		if err := write(c, refs.KVMount+"/data/"+path); err != nil {
			return false, err
		}
		return true, nil
	})
	return ref, err
}

func vaultPending(o *v1alpha1.OpenBaoServer, server, client string) error {
	if o == nil {
		return apierrors.NewNotFound(compose.OpenBaoServerGVR.GroupResource(), server)
	}
	st := o.Status
	if st.Address == "" || st.Refs == nil {
		return fmt.Errorf("the OpenBaoServer %s reports no address yet (phase %s)", server, st.Phase)
	}
	for _, x := range st.Clients {
		if x.Name == client && x.ClientCert != "" {
			return nil
		}
	}
	return fmt.Errorf("the OpenBaoServer %s reports no client %s yet", server, client)
}

func (k *Kcp) Identities() (*kcpclient.MarketIdentities, error) {
	c, err := kcpclient.New(kcpclient.Options{Host: k.rest.Host, RestConfig: k.rest})
	if err != nil {
		return nil, err
	}
	return c.MarketIdentities(k.workspace)
}

func (k *Kcp) WaitIdentityObject(ctx context.Context, timeout time.Duration) error {
	return await(ctx, k.dyn.Resource(compose.MarketIdentityGVR), compose.MarketIdentityGVR.GroupResource(), marketrequester.RequesterIdentity,
		timeout, "waiting for the requester's MarketIdentity (its core creates it)", func(*v1alpha1.MarketIdentity) bool { return true })
}

func (k *Kcp) AwaitIdentity(ctx context.Context, name string, each func(*v1alpha1.MarketIdentity) (bool, error)) (*v1alpha1.MarketIdentity, error) {
	return kube.Await(ctx, k.dyn.Resource(compose.MarketIdentityGVR), name, 0, each)
}

func (k *Kcp) AwaitContract(ctx context.Context, timeout time.Duration, done func(*v1alpha1.ComputeContract) bool) (*v1alpha1.ComputeContract, error) {
	return kube.Await(ctx, k.dyn.Resource(compose.ComputeContractGVR), k.ContractName(), timeout,
		func(c *v1alpha1.ComputeContract) (bool, error) { return c != nil && done(c), nil })
}

func (k *Kcp) AwaitRecord(ctx context.Context, name string, timeout time.Duration, done func(*v1alpha1.AtprotoRecord) (bool, error)) (*v1alpha1.AtprotoRecord, error) {
	return kube.Await(ctx, k.dyn.Resource(compose.AtprotoRecordGVR), name, timeout,
		func(r *v1alpha1.AtprotoRecord) (bool, error) {
			if r == nil {
				return false, nil
			}
			return done(r)
		})
}

func networkClient(kubeconfig string) (kube.Resource, error) {
	cfg, err := kcpserver.FrontRestConfig(kubeconfig)
	if err != nil {
		return kube.Resource{}, err
	}
	cfg.Host = strings.TrimSuffix(cfg.Host, "/") + "/clusters/" + compose.DefaultWorkspace
	dc, err := kube.New(cfg)
	if err != nil {
		return kube.Resource{}, err
	}
	return dc.Resource(compose.MarketNetworkGVR), nil
}

func networkApps(ctx context.Context, ri kube.Resource) ([]v1alpha1.NetworkApp, error) {
	o, err := kube.Get[v1alpha1.MarketNetwork](ctx, ri, marketnetwork.NetworkName)
	if err != nil {
		return nil, fmt.Errorf("request: the market network's MarketNetwork: %w", err)
	}
	return o.Spec.Apps, nil
}

func applyNetworkApp(ctx context.Context, ri kube.Resource, name string, app *v1alpha1.NetworkApp) error {
	spec := map[string]any{}
	if app != nil {
		entry, err := runtime.DefaultUnstructuredConverter.ToUnstructured(app)
		if err != nil {
			return err
		}
		spec["apps"] = []any{entry}
	}
	doc := kube.Doc{"apiVersion": v1alpha1.GroupVersion.String(), "kind": "MarketNetwork",
		"metadata": map[string]any{"name": marketnetwork.NetworkName}, "spec": spec}
	if _, err := kube.Apply[kube.Doc](ctx, ri, doc, kube.ApplyOptions{FieldManager: NetworkAppFieldManager + "/" + name}); err != nil {
		return fmt.Errorf("request: applying the market network's application %s: %w", name, err)
	}
	return nil
}

func AddToNetwork(ctx context.Context, kubeconfig string, app v1alpha1.NetworkApp) error {
	ri, err := networkClient(kubeconfig)
	if err != nil {
		return err
	}
	apps, err := networkApps(ctx, ri)
	if err != nil {
		return err
	}
	for _, a := range apps {
		if a.Name == app.Name {
			return fmt.Errorf("request: the market network already has an application %s", app.Name)
		}
	}
	return applyNetworkApp(ctx, ri, app.Name, &app)
}

func SetStopped(ctx context.Context, kubeconfig, name string, stopped bool) error {
	ri, err := networkClient(kubeconfig)
	if err != nil {
		return err
	}
	apps, err := networkApps(ctx, ri)
	if err != nil {
		return err
	}
	for _, a := range apps {
		if a.Name == name {
			a.Stopped = stopped
			return applyNetworkApp(ctx, ri, name, &a)
		}
	}
	return fmt.Errorf("request: the market network has no application %s", name)
}

func RemoveFromNetwork(ctx context.Context, kubeconfig, name string) error {
	ri, err := networkClient(kubeconfig)
	if err != nil {
		return err
	}
	if err := applyNetworkApp(ctx, ri, name, nil); err != nil {
		return err
	}
	apps, err := networkApps(ctx, ri)
	if err != nil {
		return err
	}
	for _, a := range apps {
		if a.Name == name {
			return fmt.Errorf("request: another writer of the MarketNetwork still applies its application %s", name)
		}
	}
	return nil
}

func NetworkApp(ctx context.Context, kubeconfig, name string) (v1alpha1.NetworkAppStatus, bool, error) {
	ri, err := networkClient(kubeconfig)
	if err != nil {
		return v1alpha1.NetworkAppStatus{}, false, err
	}
	o, err := kube.Get[v1alpha1.MarketNetwork](ctx, ri, marketnetwork.NetworkName)
	if err != nil {
		return v1alpha1.NetworkAppStatus{}, false, err
	}
	for _, a := range o.Status.Apps {
		if a.Name == name {
			return a, true, nil
		}
	}
	return v1alpha1.NetworkAppStatus{}, false, nil
}

func WaitNetworkApp(ctx context.Context, kubeconfig, name string, timeout time.Duration, done func(v1alpha1.NetworkAppStatus) bool,
	progress func(v1alpha1.NetworkAppStatus)) (v1alpha1.NetworkAppStatus, error) {
	ri, err := networkClient(kubeconfig)
	if err != nil {
		return v1alpha1.NetworkAppStatus{}, err
	}
	var cur v1alpha1.NetworkAppStatus
	var failed error
	last := ""
	err = await(ctx, ri, compose.MarketNetworkGVR.GroupResource(), marketnetwork.NetworkName, timeout, "waiting for the market network to place "+name,
		func(o *v1alpha1.MarketNetwork) bool {
			i := slices.IndexFunc(o.Status.Apps, func(a v1alpha1.NetworkAppStatus) bool { return a.Name == name })
			if i < 0 {
				return false
			}
			a := o.Status.Apps[i]
			cur = a
			if key := string(a.Phase) + a.Message; key != last && progress != nil {
				progress(a)
				last = key
			}
			if a.Phase == v1alpha1.ApplicationPhaseFailed {
				failed = fmt.Errorf("request: the market network failed %s: %s", name, a.Message)
				return true
			}
			return done(a)
		})
	if err != nil {
		return cur, err
	}
	return cur, failed
}

func workspaceClient(kubeconfig, ws string, gvr schema.GroupVersionResource) (kube.Resource, error) {
	cfg, err := kcpserver.FrontRestConfig(kubeconfig)
	if err != nil {
		return kube.Resource{}, err
	}
	cfg.Host = strings.TrimSuffix(cfg.Host, "/") + "/clusters/" + ws
	dc, err := kube.New(cfg)
	if err != nil {
		return kube.Resource{}, err
	}
	return dc.Resource(gvr), nil
}

func WaitNetworkVouches(ctx context.Context, kubeconfig, name string, timeout time.Duration, progress func(ws, record string)) error {
	ri, err := networkClient(kubeconfig)
	if err != nil {
		return err
	}
	var own, did string
	o, err := kube.Await(ctx, ri, marketnetwork.NetworkName, timeout, func(o *v1alpha1.MarketNetwork) (bool, error) {
		if o == nil {
			return false, nil
		}
		i := slices.IndexFunc(o.Status.Apps, func(a v1alpha1.NetworkAppStatus) bool { return a.Name == name })
		if i < 0 {
			return false, nil
		}
		own, did = o.Status.Apps[i].Workspace, strings.TrimSpace(o.Status.Apps[i].Did)
		return own != "" && did != "", nil
	})
	if errors.Is(err, kube.ErrAwaitTimeout) {
		return fmt.Errorf("request: the market network reports no workspace and DID for %s after %s", name, timeout)
	}
	if err != nil {
		return err
	}
	type want struct{ ws, record string }
	var wants []want
	for _, a := range o.Status.Apps {
		if a.Kind == v1alpha1.ApplicationKindMarketBidder && a.Workspace != "" && a.Did != "" && a.Phase == v1alpha1.ApplicationPhaseReady {
			wants = append(wants, want{ws: a.Workspace, record: marketnetwork.RequesterVouchName(did)}, want{record: marketnetwork.BidderVouchName(a.Name)})
		}
	}
	for _, w := range wants {
		ws := cmp.Or(w.ws, own)
		rr, err := workspaceClient(kubeconfig, ws, compose.AtprotoRecordGVR)
		if err != nil {
			return err
		}
		if progress != nil {
			progress(ws, w.record)
		}
		last, err := kube.Await(ctx, rr, w.record, timeout, func(rec *v1alpha1.AtprotoRecord) (bool, error) {
			return rec != nil && rec.Status.Phase == v1alpha1.AtprotoRecordPhaseApplied, nil
		})
		if errors.Is(err, kube.ErrAwaitTimeout) {
			msg := "not composed"
			if last != nil {
				msg = cmp.Or(last.Status.Message, string(last.Status.Phase))
			}
			return fmt.Errorf("request: the network's vouch %s in %s is not Applied after %s: %s", w.record, ws, timeout, msg)
		}
		if err != nil {
			return err
		}
	}
	return nil
}

func WaitNetworkGone(ctx context.Context, kubeconfig, name string, timeout time.Duration) error {
	ri, err := networkClient(kubeconfig)
	if err != nil {
		return err
	}
	_, err = kube.Await(ctx, ri, marketnetwork.NetworkName, timeout, func(o *v1alpha1.MarketNetwork) (bool, error) {
		return o != nil && !slices.ContainsFunc(o.Status.Apps, func(a v1alpha1.NetworkAppStatus) bool { return a.Name == name }), nil
	})
	if errors.Is(err, kube.ErrAwaitTimeout) {
		return fmt.Errorf("the market network still lists %s", name)
	}
	return err
}

func HostKeyAlias(receiptURI string) string {
	clean := func(v string) string {
		v = strings.ToLower(v)
		var b strings.Builder
		dash := false
		for _, r := range v {
			if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
				b.WriteRune(r)
				dash = false
				continue
			}
			if !dash {
				b.WriteByte('-')
				dash = true
			}
		}
		return strings.Trim(b.String(), "-")
	}
	rest := strings.TrimPrefix(receiptURI, "at://")
	authority, _, _ := strings.Cut(rest, "/")
	rkey := receiptURI[strings.LastIndex(receiptURI, "/")+1:]
	return "swc-guest-" + clean(authority) + "-" + clean(rkey)
}

func Ticket(vmAddress string) (string, error) {
	if !strings.HasPrefix(vmAddress, irohScheme) {
		return "", fmt.Errorf("request: the guest address %q is not an iroh ticket; iroh is the only guest transport", vmAddress)
	}
	return strings.TrimPrefix(vmAddress, irohScheme), nil
}