johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package componentcore

import (
	"cmp"
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/identity"
	"k8s.io/client-go/rest"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/apiserverfront"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose"
	atprotooauth "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/oauth"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/oauthclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/pdsclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/component"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/component/ports"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/atprotosession"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/childapplication"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/cluster"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/clusterroll"
	componentctl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/component"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/componentlink"
	computevmctl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/computevm"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/contractevent"
	eventsctl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/events"
	exposurectl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/exposure"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/guestidentitygrant"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/hostinstall"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/hostrelease"
	infractl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/infra"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/infracollection"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/kcpinstallation"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/marketidentity"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/marketintents"
	marketpresencectl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/marketpresence"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/openbaoserver"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/pdsaccount"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/plcidentity"
	policyctl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/policy"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/portlease"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/postgres"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/relaystatus"
	remoteopenbaoctl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/remoteopenbao"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/repos"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	servicectl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/service"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/subnetlease"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/ttl"
	vaultctl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/vault"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/vaultcertificate"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/vaultcredential"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/eventstream"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/exposure"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostlifecycle"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/install"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcplibs/common/ref"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcplibs/impl/kcpstore"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpserver"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/hostport"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/kcp"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/supervise"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/lease/local"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/market/participant"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/marketidentity/atproto"
	presence "github.com/johnandersen777/socialweb-computer-kcp/internal/marketpresence"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/openbaoctl"
	policycontroller "github.com/johnandersen777/socialweb-computer-kcp/internal/policyengine/controller"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/statusstate"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/subnetallocator"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/vault/openbao"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/vaultmain"
)

const (
	pdsClientTimeout = 30 * time.Second

	identityResolverTimeout = 10 * time.Second

	identityHTTPTimeout = 20 * time.Second

	defaultQROrigin = "https://qr.fedfork.com"
)

func (c *Core) runControllers(ctx context.Context) error {
	cfg := c.cfg
	log := cfg.Log
	restCfg, err := c.managerRestConfig()
	if err != nil {
		return fmt.Errorf("componentcore: the controller manager: %w", err)
	}
	mgr, err := ctlruntime.NewManager(ctlruntime.Config{Rest: restCfg, AddToScheme: schemes, Workspaces: c.scope.Contains, Log: log})
	if err != nil {
		return fmt.Errorf("componentcore: the controller manager: %w", err)
	}
	clients, err := newWorkspaceClients(restCfg)
	if err != nil {
		return fmt.Errorf("componentcore: the workspace clients: %w", err)
	}
	g := supervise.New(ctx)
	defer g.Stop()
	ctx = g.Context()
	c.scope.start(ctx)
	c.registerLocal(ctx, mgr, clients)
	if err := c.registerNetwork(mgr, clients, g); err != nil {
		log.Warn("the network controllers are not wired", "err", err.Error())
	}
	if cfg.Compute != nil {
		compute := *cfg.Compute
		compute.Logins = c.logins
		g.Go("computevm", func(ctx context.Context) error { return computevmctl.Run(ctx, compute) })
	}
	g.Go("controller-manager", mgr.Start)
	return g.Wait()
}

func (c *Core) registerLocal(ctx context.Context, mgr *ctlruntime.Manager, clients *workspaceClients) {
	cfg := c.cfg
	local := c.scope.Contains
	base := c.scope.Base
	warn := func(what string, err error) {
		if err != nil {
			cfg.Log.Warn("a controller is not wired", "controllers", what, "err", err.Error())
		}
	}
	exposures, err := c.registerExposure(mgr, local)
	warn("Service, Exposure, ComponentLink", err)
	for _, changed := range exposures {
		c.closers = append(c.closers, c.registry.Subscribe(changed), c.handlers.Subscribe(changed))
	}
	_, err = componentctl.Register(mgr, componentctl.Options{Runtime: c.runtime, Scope: local, Changed: exposures, Log: cfg.Log})
	warn("components", err)
	warn("GuestControlPlane inputs", c.registerGuestControlPlane(mgr, local))
	warn("MarketIdentity", c.registerIdentity(mgr, local))
	warn("NodeClaim", c.registerNodeClaims(mgr, c.nodeClaims))
	_, err = marketintents.Register(mgr, marketintents.Options{Bus: c.handlers, Clients: clients.client, Scope: local, Log: cfg.Log})
	warn("market intents", err)
	_, err = contractevent.Register(mgr, contractevent.Options{Bus: c.handlers, Clients: clients.client, Scope: local, Log: cfg.Log})
	warn("ContractEvent", err)
	if cfg.DynamicFactory != nil {
		_, err = infractl.Register(mgr, infractl.Options{Bus: c.handlers, Dynamic: cfg.DynamicFactory, Scope: local, Writer: "core", Log: cfg.Log})
		warn("Infra, EphemeralInfra", err)
		_, err = infracollection.Register(mgr, infracollection.Options{Bus: c.handlers, Dynamic: cfg.DynamicFactory, Scope: local, Log: cfg.Log})
		warn("InfraCollection, EphemeralInfraCollection", err)
	}
	_, err = marketpresencectl.Register(mgr, marketpresencectl.Options{Heartbeats: presence.New(presence.Config{HTTPClient: c.handlers.Client()}),
		Bidders: bidders{clients: clients}, Scope: local, Log: cfg.Log})
	warn("MarketPresence", err)
	warn("leases", c.registerLeases(ctx, mgr, local, base))
	warn("status objects", c.registerStatus(mgr, base))
	warn("host lifecycle", c.registerHost(mgr, clients, base))
	if cfg.Apps != nil {
		warn("application roots", app.Register(ctx, mgr, *cfg.Apps))
	}
	if cfg.TTLScope != nil {
		tc, err := ttl.Register(mgr, ttl.Options{Log: cfg.Log, Scope: cfg.TTLScope})
		warn("TTL", err)
		if err == nil && cfg.OnTTL != nil {
			cfg.OnTTL(tc.Report)
		}
	}
}

func (c *Core) registerExposure(mgr *ctlruntime.Manager, scope func(string) bool) ([]func(component.Key), error) {
	log := c.cfg.Log
	services, err := servicectl.Register(mgr, servicectl.Options{Handlers: c.handlers, Registry: c.registry, Scope: scope, Log: log})
	if err != nil {
		return nil, err
	}
	publisher := exposure.NewPublisher(exposure.PublisherConfig{Handlers: c.handlers, Ports: c.ports, TLS: c.tls,
		Registrar: c.ingressRegistrar, Log: log})
	exposures, err := exposurectl.Register(mgr, exposurectl.Options{Publisher: publisher, Scope: scope, Log: log})
	if err != nil {
		return nil, err
	}
	c.closers = append(c.closers, exposures.Close)
	links, err := componentlink.Register(mgr, componentlink.Options{Policy: c.linkPolicy, Registry: c.registry, Providers: c.handlers, Scope: scope, Log: log})
	if err != nil {
		return nil, err
	}
	return []func(component.Key){services.ComponentChanged, exposures.ComponentChanged, links.ComponentChanged}, nil
}

func (c *Core) registerIdentity(mgr *ctlruntime.Manager, scope func(string) bool) error {
	cfg := c.cfg
	oauth := atproto.OAuthOptions{Directory: identity.DefaultDirectory(), HTTPClient: c.networkClient(identityHTTPTimeout),
		States: &oauthclient.MemoryStateStore{}, Scopes: append([]string(nil), cfg.OAuthScopes...), QRScopes: atprotooauth.DIDKeyAssociatorOAuthScope}
	if key, err := loadOAuthCLIKey(cfg.StateDir); err == nil {
		oauth.CLIKey = key
		oauth.QROrigin = cmp.Or(strings.TrimSpace(os.Getenv("SOCIALWEB_QR_ORIGIN")), defaultQROrigin)
	}
	var mu sync.Mutex
	vaults := map[string]atproto.SecretStore{}
	ctl, err := marketidentity.Register(mgr, marketidentity.Options{Scope: scope, Endpoints: c.pdsEndpoint,
		Vault: func(workspace string) atproto.SecretStore {
			mu.Lock()
			defer mu.Unlock()
			if store, ok := vaults[workspace]; ok {
				return store
			}
			store, err := newIdentityVaultStore(c.store, workspace, cfg.CredentialClient, cfg.CredentialServer)
			if err != nil {
				return nil
			}
			vaults[workspace] = store
			return store
		},
		Resolver: &pdsclient.Resolver{Handlers: c.handlers, Timeout: identityResolverTimeout}, Bus: c.handlers, OAuth: oauth,
		PLCURL: c.plcBaseURL(), RecordHTTP: c.networkClient(networkReadTimeout), Log: cfg.Log})
	if err != nil {
		return err
	}
	base := c.scope.base()
	for _, name := range []string{"bidder", "requester"} {
		c.oauthCallbacks.Register(name, func(ctx context.Context, _ http.ResponseWriter, r *http.Request) error {
			return ctl.CompleteBrowserAuthorization(ctx, base, name, r)
		})
	}
	return nil
}

func (c *Core) registerLeases(ctx context.Context, mgr *ctlruntime.Manager, scope, base func(string) bool) error {
	cfg := c.cfg
	portsFile := filepath.Join(cfg.StateDir, PortsDir, ports.PortsFile)
	networksFile := filepath.Join(cfg.StateDir, "networks.json")
	watch := func(paths ...string) <-chan struct{} {
		ch, err := local.Watch(ctx, paths...)
		if err != nil {
			cfg.Log.Warn("the lease controllers do not follow host file changes", "paths", strings.Join(paths, ", "), "err", err.Error())
			return nil
		}
		return ch
	}
	if _, err := portlease.Register(mgr, portlease.Options{Ports: c.ports, Changes: watch(portsFile), Scope: scope, Log: cfg.Log}); err != nil {
		return err
	}
	subnets, err := subnetallocator.Open(subnetallocator.Config{LeasePath: networksFile, AppInstance: cfg.InstallAccount})
	if err != nil {
		cfg.Log.Warn("the SubnetLease controller is not wired: no subnet allocator", "err", err.Error())
	} else if _, err := subnetlease.Register(mgr, subnetlease.Options{Subnets: subnets, Changes: watch(networksFile), Scope: scope,
		Log: cfg.Log}); err != nil {
		return err
	}
	if _, err := cluster.Register(mgr, cluster.Options{Ports: c.ports, Front: apiserverfront.Client{Path: cfg.FrontControl},
		Changes: watch(cfg.SlotsFile, portsFile), Scope: base}); err != nil {
		return err
	}
	if _, err := clusterroll.Register(mgr, clusterroll.Options{Rolls: local.SlotFile{Path: cfg.SlotsFile}, Changes: watch(cfg.SlotsFile),
		Scope: base}); err != nil {
		return err
	}
	return nil
}

func (c *Core) registerStatus(mgr *ctlruntime.Manager, scope func(string) bool) error {
	base := c.scope.base()
	plc := plcidentity.Options{Scope: scope}
	if url := c.plcBaseURL(); url != "" {
		directory, err := statusstate.NewHTTPPLCClient(url, c.busOrNetworkClient())
		if err != nil {
			return err
		}
		plc.Directory = directory
	}
	vault := openBaoMetadata(c.store, base)
	_, accountErr := pdsaccount.Register(mgr, pdsaccount.Options{Bus: c.handlers, Scope: scope, Log: c.cfg.Log})
	_, sessionErr := atprotosession.Register(mgr, atprotosession.Options{Bus: c.handlers, Scope: scope, Log: c.cfg.Log})
	_, plcErr := plcidentity.Register(mgr, plc)
	_, relayErr := relaystatus.Register(mgr, relaystatus.Options{Source: statusstate.NewHTTPRelayHealthSource(c.relayEndpoint(base), c.busOrNetworkClient()),
		Scope: scope})
	_, credentialErr := vaultcredential.Register(mgr, vaultcredential.Options{Source: vault, Scope: scope})
	_, certificateErr := vaultcertificate.Register(mgr, vaultcertificate.Options{Source: vault, Scope: scope})
	return errors.Join(accountErr, sessionErr, plcErr, relayErr, credentialErr, certificateErr)
}

func (c *Core) registerHost(mgr *ctlruntime.Manager, clients *workspaceClients, scope func(string) bool) error {
	cfg := c.cfg
	authority := cfg.HostAuthority
	if authority == nil {
		authority = c.fileAuthority()
	}
	_, kcpErr := kcpinstallation.Register(mgr, kcpinstallation.Options{Log: cfg.Log, Authority: authority, Scope: scope})
	_, installErr := hostinstall.Register(mgr, hostinstall.Options{Log: cfg.Log, Authority: authority, Scope: scope})
	_, releaseErr := hostrelease.Register(mgr, hostrelease.Options{Log: cfg.Log, Authority: authority, Scope: scope})
	_, childErr := childapplication.Register(mgr, childapplication.Options{Log: cfg.Log, Workspaces: workspaceTree{clients: clients},
		Objects: clients, Scope: scope})
	if err := errors.Join(kcpErr, installErr, releaseErr, childErr); err != nil {
		return err
	}
	if cfg.Postgres != nil {
		if _, err := postgres.Register(mgr, postgres.Options{Log: cfg.Log, Host: cfg.Postgres, Scope: scope}); err != nil {
			return err
		}
	}
	if cfg.OpenBao != nil {
		servers, err := openbaoserver.Register(mgr, openbaoserver.Options{Servers: cfg.OpenBao, Grants: guestGrants{clients: clients}, Scope: scope,
			Log: cfg.Log})
		if err != nil {
			return err
		}
		if _, err := guestidentitygrant.Register(mgr, guestidentitygrant.Options{Outcomes: servers, Events: servers.GrantEvents(), Scope: scope,
			Log: cfg.Log}); err != nil {
			return err
		}
	}
	return nil
}

func (c *Core) fileAuthority() *hostlifecycle.FileAuthority {
	cfg := c.cfg
	return hostlifecycle.NewFileAuthority(hostlifecycle.FileAuthorityOptions{
		StateDir:     cfg.StateDir,
		Account:      cfg.InstallAccount,
		FrontControl: cfg.FrontControl,
		Record: func(account string) (hostlifecycle.InstallRecord, error) {
			rec, err := install.ReadRecord(cfg.InstallStateRoot, account)
			if err != nil {
				return hostlifecycle.InstallRecord{}, err
			}
			return hostlifecycle.InstallRecord{StateDir: rec.StateDir, Steps: rec.Steps}, nil
		},
		ActiveSlot: func(context.Context) (string, error) {
			slots, err := kcpserver.ReadSlots(cfg.SlotsFile)
			if err != nil {
				return "", err
			}
			return slots.Active, nil
		},
		CoreVersion: func(ctx context.Context, stateDir, link string) (string, error) {
			asset, err := kcpserver.ReadAsset(ctx, filepath.Join(stateDir, "bin", link, install.CoreName))
			if err != nil {
				return "", err
			}
			return cmp.Or(asset.Version, asset.SHA256), nil
		},
	})
}

func (c *Core) registerNetwork(mgr *ctlruntime.Manager, clients *workspaceClients, g *supervise.Group) error {
	cfg := c.cfg
	log := cfg.Log
	remoteBao := vaultctl.Remote(c.handlers, log)
	if err := repos.Register(mgr, c.repoOptions()); err != nil {
		return fmt.Errorf("componentcore: the repo controllers: %w", err)
	}
	if err := policyctl.Register(mgr, policyctl.Options{Bus: c.handlers, PLCURL: cfg.PLCBaseURL, HTTP: c.networkClient(networkReadTimeout), Log: log, Disabled: cfg.DisablePolicy,
		Sandbox: newSandbox(),
		Actions: func(workspace string) (map[string]policycontroller.ActionHandler, error) {
			return policyActionHandlers(c.store, workspace)
		}}); err != nil {
		return fmt.Errorf("componentcore: the policy controllers: %w", err)
	}
	if _, err := vaultctl.Register(mgr, c.handlers, vaultctl.Options{Remote: remoteBao, Log: log}); err != nil {
		return fmt.Errorf("componentcore: the vault controller: %w", err)
	}
	if _, err := remoteopenbaoctl.Register(mgr, remoteopenbaoctl.Options{Remote: remoteBao, Log: log}); err != nil {
		return fmt.Errorf("componentcore: the RemoteOpenBao controller: %w", err)
	}
	watchChildren, watchRepos, watchValues := eventsctl.RecordWatch(clients.client)
	run, err := eventsctl.Register(g.Context(), mgr, eventsctl.Options{Bus: c.handlers, Dynamic: eventstream.DynamicFactory(cfg.DynamicFactory),
		Resolver: &pdsclient.Resolver{Handlers: c.handlers, Timeout: pdsClientTimeout}, PLCURL: c.plcBaseURL(), HTTPClient: c.networkClient(networkReadTimeout), StateDir: cfg.StateDir,
		RecordWatchChildren: watchChildren, RecordWatchRepos: watchRepos, RecordWatchValues: watchValues, Log: log})
	if err != nil {
		return fmt.Errorf("componentcore: the event controllers: %w", err)
	}
	g.Go("events", run)
	return nil
}

func (c *Core) networkClient(timeout time.Duration) *http.Client {
	egress := component.Egress{Hosts: c.hosts, TLS: c.tls, Log: c.cfg.Log}
	if n := c.cfg.Network; n != nil {
		if hosts := participant.NetworkHosts(*n, c.hosts); hosts != nil {
			egress.Hosts = hosts
		}
		if pool, err := x509.SystemCertPool(); err == nil && pool.AppendCertsFromPEM([]byte(n.CABundle)) {
			egress.TLS = &tls.Config{MinVersion: tls.VersionTLS12, RootCAs: pool}
		}
	}
	return egress.Client(timeout)
}

func (c *Core) busOrNetworkClient() *http.Client {
	external := c.networkClient(networkReadTimeout).Transport
	if external == nil {
		external = http.DefaultTransport
	}
	inProcess := c.handlers.RoundTripper()
	return &http.Client{Timeout: networkReadTimeout, Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
		if _, ok := c.handlers.HandlerForHost(hostport.Normalize(req.URL.Host)); ok {
			return inProcess.RoundTrip(req)
		}
		return external.RoundTrip(req)
	})}
}

type roundTripFunc func(*http.Request) (*http.Response, error)

func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { return f(req) }

const PublicPLCDirectoryURL = "https://plc.directory"

func (c *Core) plcBaseURL() string {
	if url := strings.TrimSpace(c.cfg.PLCBaseURL); url != "" {
		return url
	}
	if c.cfg.Network == nil {
		return PublicPLCDirectoryURL
	}
	return "http://" + kcp.ServiceFQDN(compose.NetworkPLCName, "default", c.cfg.Workspaces[0], kcp.DefaultServiceDomain)
}

func (c *Core) managerRestConfig() (*rest.Config, error) {
	if c.cfg.Kubeconfig != "" {
		return kcpserver.FrontRestConfig(c.cfg.Kubeconfig)
	}
	if c.cfg.KCP == nil {
		return nil, ErrNoKubeconfig
	}
	client, err := c.cfg.KCP.Client(c.cfg.Workspaces[0])
	if err != nil {
		return nil, err
	}
	restCfg := client.RESTConfig()
	if restCfg == nil {
		return nil, errors.New("componentcore: the kcp client has no rest config")
	}
	return restCfg, nil
}

func (c *Core) relayEndpoint(base string) statusstate.RelayEndpointFunc {
	return func(ref string) (string, bool) {
		ref = strings.TrimSpace(ref)
		ref = cmp.Or(ref, compose.ComponentKindRelay+"/"+compose.NetworkRelayName)
		if strings.Contains(ref, "://") {
			return ref, true
		}
		kind, name := compose.ComponentKindRelay, ref
		if before, after, ok := strings.Cut(ref, "/"); ok {
			kind, name = before, after
		}
		kind, name = strings.TrimSpace(kind), strings.TrimSpace(name)
		kind = cmp.Or(kind, compose.ComponentKindRelay)
		name = cmp.Or(name, compose.NetworkRelayName)
		if endpoint, ok := c.registry.Endpoint(base, kind, name, "http"); ok {
			if endpoint.URL != "" {
				return endpoint.URL, true
			}
			if endpoint.Address != "" {
				return "http://" + endpoint.Address, true
			}
		}
		for _, url := range c.cfg.Relays {
			if url = strings.TrimSpace(url); url != "" {
				return url, true
			}
		}
		return "http://" + kcp.ServiceFQDN(strings.ToLower(kind), "default", base, kcp.DefaultServiceDomain), true
	}
}

func openBaoMetadata(store *kcpstore.Store, workspace string) statusstate.VaultSource {
	if store == nil {
		return nil
	}
	return &statusstate.OpenBaoMetadataSource{Open: func(ctx context.Context, server, mount string) (statusstate.MetadataReader, error) {
		server = strings.TrimSpace(server)
		if server == "" {
			return nil, errors.New("componentcore: the vault status object names no OpenBaoServer")
		}
		obj, err := kcpstore.OfClusterScoped[v1alpha1.OpenBaoServer](store, openbaoctl.GVR).Get(ctx, ref.New(workspace, "", server))
		if err != nil {
			return nil, fmt.Errorf("componentcore: reading OpenBaoServer %s: %w", server, err)
		}
		if client := openbao.ClientForMount(obj.Status.Clients, mount); client != "" {
			srv, err := openbao.ServerOf(workspace, server, obj.Status, client)
			if err != nil {
				return nil, err
			}
			return openbao.Open(ctx, srv)
		}
		access, err := vaultmain.RootAccessOf(*obj)
		if err != nil {
			return nil, err
		}
		return access.Client(ctx)
	}}
}