johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package main

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"maps"
	"os"
	"os/exec"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"

	"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/marketrequester"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/sshgateway"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/clicfg"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/devinstance"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostsvc"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/fsx"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/stack/request"
)

const (
	requesterVerb      = "run requester"
	requesterReadyRead = 10 * time.Second
	requesterRootApply = 3 * time.Minute
)

var requesterOptions = []string{backendOpt, "network", "name", "identity", "state-dir", "ready-file", "fg", "ready-timeout", "network-timeout",
	"kcp-path", "kine-path", "bao-path", "dumbpipe-path", "asset-cache", "log-level", "unit-prefix", "with-gateway", "node-download-base"}

var requesterPasswordOptions = []string{"name", "state-dir", "network-timeout"}

var requesterSigninOptions = []string{"name", "state-dir", "json"}

var requesterHelperOptions = []string{"name", "state-dir", "ready-file", "json"}

var requesterSSHOptions = []string{"name", "state-dir", "ready-file", "contract", "transport", "tag", "exec", "contract-timeout", "network-timeout",
	"dumbpipe-path", "asset-cache", "log-level"}

func parseRequesterSpec(args []string, verb string, verbOptions []string, network bool, other func(compose.AppArg) (bool, error)) (v1alpha1.MarketRequesterSpec, error) {
	root := v1alpha1.MarketRequesterSpec{Profile: marketrequester.RequesterProfileDefault,
		Identity: v1alpha1.ApplicationIdentity{Mode: v1alpha1.MarketIdentityModeLocalRepo}}
	parsed, err := compose.ParseAppArgs(marketrequester.RequesterApp, args)
	if err != nil {
		return root, err
	}
	var nv v1alpha1.MarketNetworkValue
	var options []v1alpha1.ApplicationOption
	nvSet, expectHandle := false, false
	for _, p := range parsed {
		switch p.Name {
		case "identity", "pds-state-dir", "expected-pds-url":
			return root, fmt.Errorf("--%s is reserved; pass global flags before --", p.Name)
		}
		networkIdentity := network && slices.Contains(compose.NetworkRequesterIdentityOptions, p.Name)
		if networkIdentity {
			if p.Name == compose.OptionIdentityMode && v1alpha1.MarketIdentityMode(p.Value) != v1alpha1.MarketIdentityModePdsAccount {
				return root, fmt.Errorf("--%s %s: a market network requester takes the identity mode %s only", p.Name, p.Value, v1alpha1.MarketIdentityModePdsAccount)
			}
			options = append(options, v1alpha1.ApplicationOption{Name: p.Name, Value: p.Value})
		}
		perApp := networkIdentity || p.To == "spec.secretsFallback" || p.To == "spec.comms.mode" || p.To == "spec.comms.eventSourceRefs" || strings.HasPrefix(p.To, "spec.requesterIngress")
		if network && !perApp && (strings.HasPrefix(p.To, compose.MapToSpec) || strings.HasPrefix(p.To, compose.MapToNetwork)) {
			return root, fmt.Errorf("--%s maps to %s, which the market network sets for its child applications", p.Name, p.To)
		}
		switch {
		case p.To == compose.MapToOptions && !p.Mapped && slices.Contains(verbOptions, p.Name):
			return root, fmt.Errorf("--%s is a %s option; pass it before --", p.Name, verb)
		case p.To == compose.MapToOptions && !p.Mapped:
			return root, fmt.Errorf("--%s is no %s option (%s); %s does not pass unknown options on", p.Name, marketrequester.RequesterApp, compose.OptionMapPath, verb)
		case p.To == compose.MapToOptions:
			options = append(options, v1alpha1.ApplicationOption{Name: p.Name, Value: p.Value})
		case p.To == "spec.identity.mode":
			switch p.Name {
			case "atproto-oauth":
				root.Identity.Mode = v1alpha1.MarketIdentityModeOAuthBrowser
			case "atproto-oauth-qr":
				root.Identity.Mode = v1alpha1.MarketIdentityModeOAuthQr
			default:
				root.Identity.Mode = v1alpha1.MarketIdentityMode(p.Value)
			}
		case p.To == "spec.identity.handle":
			root.Identity.Handle = p.Value
			expectHandle = expectHandle || p.Name == "expect-handle"
		case p.To == "spec.identity.pdsUrl":
			root.Identity.PdsURL = p.Value
		case p.To == "spec.identity.association":
			if p.Value != "false" {
				root.Identity.Association = v1alpha1.AssociationSkip
			}
		case p.To == "spec.bidderDids":
			root.BidderDids = append(root.BidderDids, splitList(p.Value)...)
		case strings.HasPrefix(p.To, "spec.requesterIngress"):
			var applied bool
			if root.RequesterIngress, applied, err = applyIngressOption(root.RequesterIngress, p.Name, p.Value); err != nil {
				return root, err
			}
			if !applied {
				return root, fmt.Errorf("--%s maps to %s, which %s does not take", p.Name, p.To, verb)
			}
		case strings.HasPrefix(p.To, "spec.ingress"):
			var applied bool
			if root.Ingress, applied, err = applyIngressOption(root.Ingress, p.Name, p.Value); err != nil {
				return root, err
			}
			if !applied {
				return root, fmt.Errorf("--%s maps to %s, which %s does not take", p.Name, p.To, verb)
			}
		case p.To == "spec.secretsFallback":
			switch fallback := v1alpha1.SecretsFallback(strings.TrimSpace(p.Value)); fallback {
			case v1alpha1.SecretsFallbackNone, v1alpha1.SecretsFallbackSealed:
				root.SecretsFallback = fallback
			default:
				return root, fmt.Errorf("--secrets-fallback %q: want None or Sealed", p.Value)
			}
		case strings.HasPrefix(p.To, "spec.comms"):
			var applied bool
			if root.Comms, applied, err = applyCommsOption(root.Comms, p.Name, p.Value); err != nil {
				return root, err
			}
			if !applied {
				return root, fmt.Errorf("--%s maps to %s, which %s does not take", p.Name, p.To, verb)
			}
		case strings.HasPrefix(p.To, compose.MapToNetwork):
			nvSet = true
			switch p.To {
			case "network.plcDirectoryUrl":
				nv.PLCDirectoryURL = p.Value
			case "network.ingressProxyHost":
				nv.IngressProxyHost = p.Value
			case "network.qrOrigin":
				nv.QrOrigin = p.Value
			case "network.hosts":
				for _, h := range splitList(p.Value) {
					name, addr, ok := strings.Cut(h, "=")
					if !ok {
						return root, fmt.Errorf("--network-hosts %q: want name=address,...", p.Value)
					}
					nv.Hosts = append(nv.Hosts, v1alpha1.NetworkHost{Name: name, Address: addr})
				}
			}
		default:
			handled := false
			if other != nil {
				if handled, err = other(p); err != nil {
					return root, err
				}
			}
			if !handled {
				return root, fmt.Errorf("--%s maps to %s, which %s does not take", p.Name, p.To, verb)
			}
		}
	}
	if expectHandle {
		switch root.Identity.Mode {
		case v1alpha1.MarketIdentityModeOAuthQr, v1alpha1.MarketIdentityModeOAuthBrowser, v1alpha1.MarketIdentityModeOAuthDelegated:
		default:
			return root, fmt.Errorf("--expect-handle pins the account of an oauth sign-in; the identity mode %s signs in no existing account, add --atproto-oauth-qr", root.Identity.Mode)
		}
		if _, err := syntax.ParseHandle(strings.TrimSpace(root.Identity.Handle)); err != nil {
			return root, fmt.Errorf("--expect-handle %q: %w", root.Identity.Handle, err)
		}
	}
	if network && root.Identity.Mode == v1alpha1.MarketIdentityModePdsAccount && strings.TrimSpace(root.Identity.Handle) == "" {
		return root, fmt.Errorf("--%s %s needs --atproto-handle naming the account", compose.OptionIdentityMode, v1alpha1.MarketIdentityModePdsAccount)
	}
	if nvSet || !network {
		nv.PLCDirectoryURL = cmp.Or(nv.PLCDirectoryURL, "https://plc.directory")
		nv.IngressProxyHost = cmp.Or(nv.IngressProxyHost, "xrpc.fedproxy.com")
		root.Network = &nv
	}
	if !network && root.Ingress == nil {
		root.Ingress = compose.DefaultIngressConfig()
		if nv.IngressProxyHost != "" {
			root.Ingress.Xrpc.IngressProxyHost = nv.IngressProxyHost
		}
	}
	root.Options = options
	return root, nil
}

type requesterSummary struct {
	request.Ready
	Mode      string  `json:"mode"`
	Network   string  `json:"network,omitempty"`
	Instance  string  `json:"instance"`
	ReadyFile string  `json:"readyFile"`
	Restarted bool    `json:"restarted"`
	Seconds   float64 `json:"seconds"`
}

type requesterRun struct {
	*requestRun
	dir        string
	readyFiles []string
	startedAt  time.Time
	sshKey     string
	mu         sync.Mutex
	last       request.Ready
	gateway    *request.ReadyGateway
	gatewayOn  bool
	gatewayExe map[string]devinstance.Executable
	instance   *devinstance.Record
}

func requesterHome(a *app, vals clicfg.Values, name string) (string, string, error) {
	root, err := devinstance.StateRoot(vals.String("state-dir"), a.env)
	if err != nil {
		return "", "", err
	}
	if root, err = filepath.Abs(root); err != nil {
		return "", "", err
	}
	if err := request.CheckName(name); err != nil {
		return "", "", err
	}
	return root, request.RequesterDir(root, name), nil
}

func requesterReadyPath(vals clicfg.Values, dir string) (string, error) {
	path := strings.TrimSpace(vals.String("ready-file"))
	if path == "" {
		return filepath.Join(dir, request.ReadyFile), nil
	}
	return filepath.Abs(path)
}

func runRequester(ctx context.Context, a *app, vals clicfg.Values, rest []string) int {
	started := time.Now()
	pos, pass := splitWords(rest)
	if len(pos) > 0 {
		return a.fail(exitRefused, "run requester: unexpected %q; MarketRequester options go after --", strings.Join(pos, " "))
	}
	backend := hostsvc.BackendProc
	if b := vals.String(backendOpt); b != "" {
		parsed, err := hostsvc.ParseBackend(b)
		if err != nil {
			return a.fail(exitRefused, "run requester: %v", err)
		}
		backend = parsed
	}
	if _, err := exec.LookPath("ssh-keygen"); err != nil {
		return a.fail(exitRefused, "run requester: ssh-keygen is not on PATH; install the OpenSSH client (apt install openssh-client, dnf install openssh-clients)")
	}
	name := cmp.Or(strings.TrimSpace(vals.String("name")), "requester")
	stateRoot, dir, err := requesterHome(a, vals, name)
	if err != nil {
		return a.fail(exitRefused, "run requester: %v", err)
	}
	network := strings.TrimSpace(vals.String("network"))
	if network != "" && backend == hostsvc.BackendSystemdUser {
		return a.fail(exitRefused, "run requester: --network places the requester under the market network's devsup; pass --host-backend=proc")
	}
	idName := strings.TrimSpace(vals.String("identity"))
	parsed, err := parseRequesterSpec(pass, requesterVerb, requesterOptions, network != "", nil)
	if err != nil {
		return a.fail(exitRefused, "run requester: %v", err)
	}
	readyFile, err := requesterReadyPath(vals, dir)
	if err != nil {
		return a.fail(exitRefused, "run requester: --ready-file: %v", err)
	}
	networkTimeout, err := time.ParseDuration(vals.String("network-timeout"))
	if err != nil {
		return a.fail(exitRefused, "run requester: network-timeout: %v", err)
	}
	if err := request.EnsureRequesterDir(dir); err != nil {
		return a.fail(exitFailed, "run requester: %v", err)
	}
	owner, owned, err := request.TakeOwnerDir(dir)
	if err != nil {
		return a.fail(exitFailed, "run requester: %v", err)
	}
	if !owned {
		return a.fail(exitRefused, "run requester: another process runs the requester %s (%s); run requester env|ssh %s reach it", name, dir, name)
	}
	defer owner.Close()
	rec := request.RequesterRecord{Schema: request.RequesterSchema, Name: name, Network: network, Identity: idName, Spec: parsed,
		WithGateway: vals.Bool("with-gateway"), CreatedAt: time.Now().UTC()}
	saved, restarted, err := request.ReadRequesterRecord(dir)
	if err != nil {
		return a.fail(exitFailed, "run requester: %v", err)
	}
	if restarted {
		switch {
		case len(pass) == 0 && saved.Network == network && saved.Identity == idName:
			rec = saved
		case !saved.SameShape(rec):
			return a.fail(exitRefused, "run requester: the requester %s was created with other options (network %q, identity %q, its own -- options in %s); start it with the same options, or with none after --",
				name, saved.Network, saved.Identity, filepath.Join(dir, request.RequesterFile))
		default:
			rec.CreatedAt = saved.CreatedAt
		}
		if want := saved.WithGateway || vals.Bool("with-gateway"); want != saved.WithGateway {
			rec.WithGateway = want
			if err := request.WriteRequesterRecord(dir, rec); err != nil {
				return a.fail(exitFailed, "run requester: %v", err)
			}
		} else {
			rec.WithGateway = want
		}
	} else if err := request.WriteRequesterRecord(dir, rec); err != nil {
		return a.fail(exitFailed, "run requester: %v", err)
	}
	mode := request.ModeStandalone
	if network != "" {
		mode = request.ModeNetwork
	}
	rr := &requestRun{a: a, vals: vals, stateRoot: stateRoot, homeDir: dir, verb: requesterVerb, root: rec.Spec, appOptions: rec.Spec.Options,
		networkTimeout: networkTimeout, resuming: true, transport: transportIroh}
	rr.inv = request.Invocation{Name: name, Mode: mode, Network: network, StateRoot: stateRoot, CreatedAt: rec.CreatedAt}
	rr.summary = request.Summary{Name: name, Mode: mode, Network: network, CreatedAt: rec.CreatedAt}
	r := &requesterRun{requestRun: rr, dir: dir, readyFiles: []string{filepath.Join(dir, request.ReadyFile)}, startedAt: time.Now().UTC()}
	if readyFile != r.readyFiles[0] {
		r.readyFiles = append(r.readyFiles, readyFile)
	}
	if rec.WithGateway {
		if r.gatewayExe, err = r.gatewayPreflight(vals); err != nil {
			return a.fail(exitRefused, "run requester: %v", err)
		}
		if err := r.markGateway(ctx); err != nil {
			return a.fail(exitRefused, "run requester: %v", err)
		}
		r.gatewayOn = true
	}
	if err := r.openIdentity(name); err != nil {
		return a.fail(exitRefused, "run requester: %v", err)
	}
	defer func() {
		if r.identity != nil {
			r.identity.Close()
		}
	}()
	pub, key, err := request.EnsureKey(ctx, filepath.Join(dir, request.SSHDir))
	if err != nil {
		return a.fail(exitFailed, "run requester: %v", err)
	}
	r.sshKey = key + ".pub"
	if err := r.linkIdentityExposure(); err != nil {
		return a.fail(exitFailed, "run requester: %v", err)
	}
	for _, path := range r.readyFiles {
		if err := request.RemoveReady(path); err != nil {
			return a.fail(exitFailed, "run requester: %v", err)
		}
	}
	fresh := true
	if mode == request.ModeNetwork {
		_, netRec, err := r.networkLayout()
		if err != nil {
			return a.fail(exitRefused, "run requester: %v", err)
		}
		_, found, err := request.NetworkApp(ctx, netRec.AdminKubeconfig, name)
		if err != nil {
			return a.fail(exitFailed, "run requester: %v", err)
		}
		fresh = !found
	}
	if backend == hostsvc.BackendSystemdUser {
		r.log("run requester: the installed requester; its unit stops the core and keeps its kcp units and state", "dir", dir, "restart", restarted)
		return r.installedCore(ctx, stateRoot, func(ctx context.Context, rec devinstance.Record, notifyReady func() error) error {
			r.inv.Instance, r.summary.Instance, r.summary.KcpURL = rec.Layout.ID, rec.Layout.ID, rec.KcpURL
			r.instance = &rec
			if err := r.attach(ctx, rec.AdminKubeconfig); err != nil {
				return err
			}
			return r.serve(ctx, started, readyFile, pub, restarted, notifyReady)
		})
	}
	r.log("run requester: a long-lived requester; Ctrl-C stops its kcp and keeps its state", "dir", dir, "mode", mode, "network", network, "restart", restarted)
	if err := r.start(ctx, fresh); err != nil {
		r.stop(context.WithoutCancel(ctx), false)
		if ctx.Err() != nil {
			return a.fail(exitFailed, "run requester: stopped before the requester was Ready; its kcp is stopped and its state is kept in %s", dir)
		}
		return a.fail(exitFailed, "run requester: %v", err)
	}
	err = r.serve(ctx, started, readyFile, pub, restarted, func() error { return nil })
	dctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), fgDownTimeout)
	defer cancel()
	r.stop(dctx, false)
	if err != nil {
		return a.fail(exitFailed, "run requester: %v", err)
	}
	return exitOK
}

func (r *requesterRun) attach(ctx context.Context, kubeconfig string) error {
	k, err := request.OpenKcp(kubeconfig, "")
	if err != nil {
		return err
	}
	r.kubeconfig, r.workspace = kubeconfig, ""
	r.kcp = k.ForContract(r.target)
	if err := applyRequesterRoot(ctx, k, r.root, r.a.log); err != nil {
		return err
	}
	r.summary.KcpRunning = true
	r.log("the requester's kcp runs", "kcp", r.summary.KcpURL, "instance", r.inv.Instance)
	if err := r.signIn(ctx, true); err != nil {
		return fmt.Errorf("signing in: %w", err)
	}
	if err := r.kcp.WaitReady(ctx, r.networkTimeout, func(st v1alpha1.ApplicationStatus) {
		r.log("requester", "phase", string(st.Phase), "message", st.Message, "components", len(st.Components))
	}); err != nil {
		return err
	}
	return request.WriteInvocationIn(r.home(), r.inv)
}

func applyRequesterRoot(ctx context.Context, k *request.Kcp, spec v1alpha1.MarketRequesterSpec, log *slog.Logger) error {
	started := time.Now()
	delay := time.Second
	for attempt := 1; ; attempt++ {
		actx, cancel := context.WithTimeout(ctx, requesterRootApply)
		err := k.ApplyRoot(actx, spec)
		cancel()
		if err == nil {
			return nil
		}
		if time.Since(started) >= requesterRootApply {
			return fmt.Errorf("applying the MarketRequester root object after %d attempts: %w", attempt, err)
		}
		log.Warn("applying the MarketRequester root object; retrying", "attempt", attempt, "err", err.Error())
		t := time.NewTimer(delay)
		select {
		case <-ctx.Done():
			t.Stop()
			return ctx.Err()
		case <-t.C:
		}
		delay = min(2*delay, 15*time.Second)
	}
}

func (r *requesterRun) installedCore(ctx context.Context, stateRoot string, serve func(context.Context, devinstance.Record, func() error) error) int {
	a, vals := r.a, r.vals
	exes := map[string]devinstance.Executable{}
	for _, x := range []struct{ name, option string }{{request.ExeKcp, "kcp-path"}, {request.ExeKine, "kine-path"}, {"bao", "bao-path"},
		{sshgateway.DumbpipeExecutable, "dumbpipe-path"}} {
		e, err := a.optionalExe(ctx, vals, x.name, x.option)
		if err != nil {
			return a.fail(exitRefused, "run requester: %v", err)
		}
		if e.Path != "" {
			exes[x.name] = e
		}
	}
	plugins, err := a.openBaoPluginExes(vals)
	if err != nil {
		return a.fail(exitRefused, "run requester: %v", err)
	}
	maps.Copy(exes, plugins)
	exe, err := selfExe()
	if err != nil {
		return a.fail(exitFailed, "run requester: %v", err)
	}
	exes["socialweb-computer"] = devinstance.Executable{Path: exe}
	readyTimeout, err := time.ParseDuration(vals.String("ready-timeout"))
	if err != nil || readyTimeout < 0 {
		return a.fail(exitRefused, "run requester: ready-timeout %q: want a positive duration", vals.String("ready-timeout"))
	}
	prefix := vals.String("unit-prefix")
	m, err := hostsvc.Open(ctx, hostsvc.Config{Backend: hostsvc.BackendSystemdUser, UnitPrefix: prefix})
	if err != nil {
		return a.fail(exitFailed, "run requester: the systemd-user backend: %v", err)
	}
	defer m.Close()
	err = request.RunCore(ctx, request.CoreConfig{StateDir: stateRoot, Prefix: prefix, Name: r.inv.Name, Executables: exes,
		ReadyTimeout: readyTimeout, Log: a.log, Serve: serve}, m)
	if err != nil && !errors.Is(err, context.Canceled) {
		return a.fail(exitFailed, "run requester: %v", err)
	}
	return exitOK
}

func (r *requesterRun) serve(ctx context.Context, started time.Time, readyFile, pub string, restarted bool, notifyReady func() error) error {
	dir := r.dir
	if err := r.checkIdentity(); err != nil {
		return err
	}
	if err := request.WriteWorkspaceKubeconfig(filepath.Join(dir, request.KubeconfigFile), r.kubeconfig, r.workspace); err != nil {
		return fmt.Errorf("the workspace kubeconfig: %w", err)
	}
	st, err := r.kcp.Root(ctx)
	if err != nil {
		return err
	}
	ready := r.ready(ctx, st)
	gctx, stopGateway := context.WithCancel(ctx)
	defer stopGateway()
	var gatewayDone <-chan error
	if r.gatewayOn {
		done, gw, err := r.startGateway(gctx, r.gatewayExe, ready.Did)
		if err != nil {
			return fmt.Errorf("--with-gateway: %w", err)
		}
		gatewayDone = done
		r.mu.Lock()
		r.gateway = gw
		r.mu.Unlock()
		ready = r.ready(ctx, st)
		r.log("the requester's gateway is Ready; the ready file names it", "workspace", gw.Workspace, "clusters", gw.Clusters)
	}
	if err := r.writeReady(ready); err != nil {
		return fmt.Errorf("the ready file: %w", err)
	}
	r.log("the requester is Ready; the ready file is written", "readyFile", readyFile, "did", ready.Did, "kcp", ready.KcpURL, "workspace", ready.Workspace,
		"sshPublicKey", pub, "gateway", ready.Gateway != nil)
	r.a.printJSON(requesterSummary{Ready: ready, Mode: r.inv.Mode, Network: r.inv.Network, Instance: r.inv.Instance, ReadyFile: readyFile, Restarted: restarted,
		Seconds: time.Since(started).Seconds()})
	if err := notifyReady(); err != nil {
		return err
	}
	follow, stopFollow := context.WithCancel(ctx)
	var wg sync.WaitGroup
	wg.Go(func() {
		if err := r.kcp.FollowRoot(follow, func(o *v1alpha1.MarketRequester) {
			if err := r.writeReady(r.ready(follow, o.Status)); err != nil {
				r.a.log.Warn("rewriting the ready file", "err", err.Error())
			}
		}); err != nil && follow.Err() == nil {
			r.a.log.Warn("following the MarketRequester for the ready file", "err", err.Error())
		}
	})
	var ended error
	select {
	case <-ctx.Done():
	case err := <-gatewayDone:
		if ctx.Err() == nil {
			ended = fmt.Errorf("the requester's gateway ended: %v", err)
		}
	}
	stopGateway()
	stopFollow()
	wg.Wait()
	for _, path := range r.readyFiles {
		if err := request.RemoveReady(path); err != nil {
			r.a.log.Warn("removing the ready file", "path", path, "err", err.Error())
		}
	}
	return ended
}

func runRequesterPassword(ctx context.Context, a *app, vals clicfg.Values, rest []string) int {
	pos, pass := splitWords(rest)
	if len(pos) != 1 || len(pass) > 0 {
		return a.fail(exitRefused, "run requester password <name>: one requester name; the app password comes on stdin")
	}
	_, dir, err := requesterHome(a, vals, pos[0])
	if err != nil {
		return a.fail(exitRefused, "run requester password: %v", err)
	}
	timeout, err := time.ParseDuration(vals.String("network-timeout"))
	if err != nil {
		return a.fail(exitRefused, "run requester password: network-timeout: %v", err)
	}
	var rec devinstance.Record
	if err := fsx.ReadJSON(filepath.Join(request.InstanceStateDirIn(dir, pos[0]), "instance.json"), &rec); err != nil || rec.AdminKubeconfig == "" {
		return a.fail(exitRefused, "run requester password: the requester %s has no running kcp to hold the password (start it first): %v", pos[0], err)
	}
	b, err := io.ReadAll(io.LimitReader(os.Stdin, 4096))
	if err != nil {
		return a.fail(exitFailed, "run requester password: reading stdin: %v", err)
	}
	password := strings.TrimSpace(string(b))
	if password == "" {
		return a.fail(exitRefused, "run requester password: stdin held no password")
	}
	k, err := request.OpenKcp(rec.AdminKubeconfig, rec.BaseWorkspace)
	if err != nil {
		return a.fail(exitFailed, "run requester password: %v", err)
	}
	ref, err := k.PutPdsAccountPassword(ctx, password, timeout)
	if err != nil {
		return a.fail(exitFailed, "run requester password: %v", err)
	}
	a.printJSON(map[string]string{"name": pos[0], "mount": ref.Mount, "path": ref.Path})
	return exitOK
}

type requesterSignin struct {
	Name          string                                `json:"name"`
	Phase         v1alpha1.MarketIdentityPhase          `json:"phase"`
	Did           string                                `json:"did,omitempty"`
	Handle        string                                `json:"handle,omitempty"`
	Message       string                                `json:"message,omitempty"`
	PendingAction *v1alpha1.MarketIdentityPendingAction `json:"pendingAction,omitempty"`
}

func runRequesterSignin(ctx context.Context, a *app, vals clicfg.Values, rest []string) int {
	pos, pass := splitWords(rest)
	if len(pos) != 1 || len(pass) > 0 {
		return a.fail(exitRefused, "run requester signin <name>: one requester name")
	}
	_, dir, err := requesterHome(a, vals, pos[0])
	if err != nil {
		return a.fail(exitRefused, "run requester signin: %v", err)
	}
	var rec devinstance.Record
	if err := fsx.ReadJSON(filepath.Join(request.InstanceStateDirIn(dir, pos[0]), "instance.json"), &rec); err != nil || rec.AdminKubeconfig == "" {
		return a.fail(exitRefused, "run requester signin: the requester %s has no kcp of its own under %s (start it first; a --network requester renders its sign-in in its own terminal): %v", pos[0], dir, err)
	}
	k, err := request.OpenKcp(rec.AdminKubeconfig, rec.BaseWorkspace)
	if err != nil {
		return a.fail(exitFailed, "run requester signin: %v", err)
	}
	ids, err := k.Identities()
	if err != nil {
		return a.fail(exitFailed, "run requester signin: %v", err)
	}
	gctx, cancel := context.WithTimeout(ctx, requesterReadyRead)
	defer cancel()
	obj, err := ids.GetMarketIdentity(gctx, marketrequester.RequesterIdentity)
	if err != nil {
		return a.fail(exitFailed, "run requester signin: the requester %s's kcp does not answer or holds no identity yet (is its unit running?): %v", pos[0], err)
	}
	out := requesterSignin{Name: pos[0], Phase: obj.Status.Phase, Did: obj.Status.Did, Handle: obj.Status.Handle, Message: obj.Status.Message,
		PendingAction: obj.Status.PendingAction}
	if vals.Bool("json") {
		a.printJSON(out)
		return exitOK
	}
	switch {
	case out.PendingAction != nil && out.PendingAction.URL != "":
		fmt.Fprintf(a.stderr, "identity %s is %s: %s; open the URL below (scripts/oauth-signin.ts --url) before %s\n", marketrequester.RequesterIdentity,
			out.Phase, cmp.Or(out.Message, string(out.PendingAction.Kind)), cmp.Or(out.PendingAction.ExpiresAt, "it is replaced"))
		fmt.Fprintln(a.stdout, out.PendingAction.URL)
		return exitOK
	case out.Phase == v1alpha1.MarketIdentityPhaseReady:
		fmt.Fprintf(a.stderr, "identity %s is Ready as %s (%s); no sign-in is pending\n", marketrequester.RequesterIdentity, cmp.Or(out.Handle, "no handle"), out.Did)
		return exitOK
	case out.PendingAction != nil:
		return a.fail(exitFailed, "run requester signin: identity %s is %s and waits on %s, not on a sign-in URL: %s", marketrequester.RequesterIdentity, out.Phase,
			out.PendingAction.Kind, cmp.Or(out.PendingAction.Message, out.Message))
	}
	return a.fail(exitFailed, "run requester signin: identity %s is %s with no sign-in URL yet: %s", marketrequester.RequesterIdentity, cmp.Or(string(out.Phase), "unset"), out.Message)
}

func (r *requesterRun) ready(ctx context.Context, st v1alpha1.ApplicationStatus) request.Ready {
	var id *v1alpha1.MarketIdentity
	if ids, err := r.kcp.Identities(); err == nil {
		ictx, cancel := context.WithTimeout(ctx, requesterReadyRead)
		if obj, err := ids.GetMarketIdentity(ictx, marketrequester.RequesterIdentity); err == nil {
			id = obj
		}
		cancel()
	}
	did, handle, mode, pdsURL, ingress := request.ReadyOf(st, id)
	r.mu.Lock()
	gw := r.gateway
	r.mu.Unlock()
	return request.Ready{Gateway: gw, Schema: request.ReadySchema, Name: r.inv.Name, Pid: os.Getpid(), StartedAt: r.startedAt, KcpURL: r.summary.KcpURL,
		Kubeconfig: filepath.Join(r.dir, request.KubeconfigFile), Workspace: cmp.Or(r.workspace, compose.DefaultWorkspace), Did: did, Handle: handle,
		IdentityMode: mode, PdsURL: pdsURL, Ingress: ingress, SSHPublicKey: r.sshKey}
}

func (r *requesterRun) writeReady(ready request.Ready) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	if ready.Did == "" {
		ready.Did = r.last.Did
	}
	if r.last.Schema != 0 && r.last.Same(ready) {
		return nil
	}
	for _, path := range r.readyFiles {
		if err := request.WriteReady(path, ready); err != nil {
			return err
		}
	}
	if r.last.Schema != 0 {
		r.log("the ready file is rewritten", "did", ready.Did, "kcp", ready.KcpURL, "pdsIngress", ready.Ingress.Pds, "requesterIngress", ready.Ingress.Requester)
	}
	r.last = ready
	return nil
}

func readRequesterReady(a *app, vals clicfg.Values, rest []string, verb string) (request.Ready, string, string, int) {
	pos, pass := splitWords(rest)
	if len(pos) != 1 || len(pass) > 0 {
		return request.Ready{}, "", "", a.fail(exitRefused, "run requester %s <name>: one requester name", verb)
	}
	stateRoot, dir, err := requesterHome(a, vals, pos[0])
	if err != nil {
		return request.Ready{}, "", "", a.fail(exitRefused, "run requester %s: %v", verb, err)
	}
	path, err := requesterReadyPath(vals, dir)
	if err != nil {
		return request.Ready{}, "", "", a.fail(exitRefused, "run requester %s: --ready-file: %v", verb, err)
	}
	ready, err := request.ReadReady(path)
	if err != nil {
		return ready, "", "", a.fail(exitRefused, "run requester %s: %v", verb, err)
	}
	if ready.Name != pos[0] {
		return ready, "", "", a.fail(exitRefused, "run requester %s: the ready file %s is the requester %s's, not %s", verb, path, ready.Name, pos[0])
	}
	return ready, stateRoot, dir, exitOK
}

func runRequesterEnv(ctx context.Context, a *app, vals clicfg.Values, rest []string) int {
	ready, _, dir, code := readRequesterReady(a, vals, rest, "env")
	if code != exitOK {
		return code
	}
	ex := []devinstance.Export{
		{Name: "KUBECONFIG", Value: ready.Kubeconfig},
		{Name: "SOCIALWEB_KCP_HOST", Value: ready.KcpURL},
		{Name: "SOCIALWEB_KCP_WORKSPACE", Value: ready.Workspace},
		{Name: "SOCIALWEB_REQUESTER_NAME", Value: ready.Name},
		{Name: "SOCIALWEB_REQUESTER_DID", Value: ready.Did},
		{Name: "SOCIALWEB_REQUESTER_DIR", Value: dir},
		{Name: "SOCIALWEB_REQUESTER_SSH_PUBLIC_KEY", Value: ready.SSHPublicKey},
	}
	if vals.Bool("json") {
		a.printJSON(exportMap(ex))
		return exitOK
	}
	fmt.Fprint(a.stdout, devinstance.ShellExports(ex))
	return exitOK
}

func (a *app) openRequester(ctx context.Context, vals clicfg.Values, rest []string, verb string) (*requestRun, int) {
	ready, stateRoot, dir, code := readRequesterReady(a, vals, rest, verb)
	if code != exitOK {
		return nil, code
	}
	r, err := newRequestRun(a, vals)
	if err != nil {
		return nil, a.fail(exitRefused, "run requester %s: %v", verb, err)
	}
	r.stateRoot, r.homeDir, r.verb = stateRoot, dir, requesterVerb
	r.inv = request.Invocation{Name: ready.Name, StateRoot: stateRoot}
	k, err := request.OpenKcp(ready.Kubeconfig, ready.Workspace)
	if err != nil {
		return nil, a.fail(exitFailed, "run requester %s: %v", verb, err)
	}
	actx, cancel := context.WithTimeout(ctx, requestAttachProbe)
	defer cancel()
	st, err := k.Root(actx)
	if err != nil {
		return nil, a.fail(exitFailed, "run requester %s: the requester's kcp does not answer (is run requester %s running?): %v", verb, ready.Name, err)
	}
	if st.Phase != v1alpha1.ApplicationPhaseReady {
		return nil, a.fail(exitFailed, "run requester %s: the MarketRequester is %s: %s", verb, st.Phase, st.Message)
	}
	if r.target == "" {
		return nil, a.fail(exitRefused, "run requester %s: --contract names the ComputeContract (a run requester has no contract of its own)", verb)
	}
	r.kcp = k.ForContract(r.target)
	return r, exitOK
}

func runRequesterSSH(ctx context.Context, a *app, vals clicfg.Values, rest []string) int {
	r, code := a.openRequester(ctx, vals, rest, "ssh")
	if r == nil {
		return code
	}
	if _, err := exec.LookPath("ssh"); err != nil {
		return a.fail(exitRefused, "run requester ssh: ssh is not on PATH; install the OpenSSH client (apt install openssh-client, dnf install openssh-clients)")
	}
	c, err := r.kcp.WaitContract(ctx, r.contractTimeout, contractSettled, r.progress)
	if err != nil {
		return a.fail(exitFailed, "run requester ssh: %v", err)
	}
	if c.Status.Phase != v1alpha1.ComputeContractHeld {
		return a.fail(exitFailed, "run requester ssh: the contract is %s, not Held: %s", c.Status.Phase, c.Status.Message)
	}
	ec, err := r.ssh(ctx, c)
	if err != nil {
		return a.fail(exitFailed, "run requester ssh: %v", err)
	}
	if ec != 0 {
		return exitFailed
	}
	return exitOK
}

func runRequesterProxy(ctx context.Context, a *app, vals clicfg.Values, rest []string) int {
	return proxyRequest(ctx, a, vals, requesterVerb+" proxy", func(q *app) (*requestRun, bool, int) {
		r, code := q.openRequester(ctx, vals, rest, "proxy")
		return r, false, code
	})
}