johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package compose

import (
	"cmp"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"slices"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

	tenancyv1alpha1 "github.com/kcp-dev/sdk/apis/tenancy/v1alpha1"

	"github.com/johnandersen777/socialweb-computer-kcp/api/kcpgvr"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/clusterhome"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kcpapply"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/manifests"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/diskusage"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostprocess"
)

const (
	GCSampleEvery       = 30 * time.Second
	GCRescanEvery       = 5 * time.Minute
	AdoptRescanEvery    = 5 * time.Minute
	AdoptSetRescanAfter = 30 * time.Second
	StateDirSampleEvery = 3 * time.Minute
	GCStuckAfter        = 10 * time.Minute
	GCExamples          = 5
	GCListCap           = 40
	GCWorkspaceCap      = 200
	ClusterLabel        = "socialweb.computer/cluster"
	AppWorkspaceLabel   = "apps.socialweb.computer/application"
	providerWS          = "socialweb-provider"
	accountWorkspace    = "account"

	QuotaName      = "socialweb-bounded"
	QuotaNamespace = "default"
)

var ResourceQuotaGVR = kcpgvr.ResourceQuotas

var QuotaDefaults = struct {
	Bids, ComputeVMs, ComputeContracts, NodeClaims, PolicyWorkflowRuns, Components int64
}{Bids: 10000, ComputeVMs: 64, ComputeContracts: 1024, NodeClaims: 64, PolicyWorkflowRuns: 500, Components: 16}

var FinalizerControllers = map[string]string{
	Finalizer:                                       "app: the composition controller of the root",
	"computevm.socialweb.computer/release":          "computevm: the ComputeVM controller of the core",
	"socialweb.computer/postgres":                   "postgresctl: the Postgres controller of the core",
	"socialweb.computer/cluster-member":             "clusterprovider: the cluster provider of the gateway",
	"cluster.socialweb.computer/teardown":           "clusterprovider: the cluster provider of the gateway",
	"node.socialweb.computer/teardown":              "clusterprovider: the cluster provider of the gateway",
	"nodeclaim.socialweb.computer/release":          "socialweb-computer-k8s web app: the node-market claim mapper",
	"atprotoidentity.socialweb.computer/revocation": "socialweb-computer-k8s web app: the revocation sweep",
}

func infraSet(setName, profile string, include []string) (manifests.Set, error) {
	ix, err := manifests.Load()
	if err != nil {
		return manifests.Set{}, err
	}
	set, ok := ix.Set(setName)
	if !ok {
		return manifests.Set{}, fmt.Errorf("compose: deploy/sets.yaml names no %s set", setName)
	}
	out := set
	out.Phases = nil
	for _, p := range set.Phases {
		base, only, scoped := strings.Cut(p.Name, "@")
		if !scoped {
			out.Phases = append(out.Phases, p)
			continue
		}
		if only == profile && slices.Contains(include, base) {
			out.Phases = append(out.Phases, p)
		}
	}
	return out, nil
}

func phaseSet(setName string, phases []string) (manifests.Set, error) {
	ix, err := manifests.Load()
	if err != nil {
		return manifests.Set{}, err
	}
	set, ok := ix.Set(setName)
	if !ok {
		return manifests.Set{}, fmt.Errorf("compose: deploy/sets.yaml names no %s set", setName)
	}
	var keep []manifests.Phase
	for _, p := range set.Phases {
		if slices.Contains(phases, p.Name) {
			keep = append(keep, p)
		}
	}
	if len(keep) == 0 {
		return manifests.Set{}, fmt.Errorf("compose: the %s set names none of the phases %v", setName, phases)
	}
	set.Phases = keep
	return set, nil
}

func setDigest(set manifests.Set, values, extra any) (string, error) {
	rendered, err := set.Render(values)
	if err != nil {
		return "", err
	}
	raw, err := json.Marshal(struct {
		Rendered []manifests.Rendered
		Extra    any
	}{rendered, extra})
	if err != nil {
		return "", err
	}
	sum := sha256.Sum256(raw)
	return hex.EncodeToString(sum[:]), nil
}

func InfraDigest(setName, profile string, values, extra any, include ...string) (string, error) {
	set, err := infraSet(setName, profile, include)
	if err != nil {
		return "", err
	}
	return setDigest(set, values, extra)
}

func PhasesDigest(setName string, values, extra any, phases ...string) (string, error) {
	set, err := phaseSet(setName, phases)
	if err != nil {
		return "", err
	}
	return setDigest(set, values, extra)
}

func (c *Composer) ApplyInfra(ctx context.Context, r Root, setName, profile string, values any, include ...string) error {
	set, err := infraSet(setName, profile, include)
	if err != nil {
		return err
	}
	return c.applySet(ctx, r, set, values)
}

func (c *Composer) ApplySetPhases(ctx context.Context, r Root, setName string, values any, phases ...string) error {
	set, err := phaseSet(setName, phases)
	if err != nil {
		return err
	}
	return c.applySet(ctx, r, set, values)
}

func (c *Composer) applySet(ctx context.Context, r Root, set manifests.Set, values any) error {
	defer c.adoptions.setApplied(r)
	ap, err := kcpapply.New(c.restCfg, kcpapply.Options{WaitTimeout: ApplyTimeout, FieldManager: r.Manager(), Base: c.Workspace(),
		Labels: map[string]string{OwnerLabel: r.Owner(), RootKindLabel: r.Kind, RootClusterLabel: r.Cluster}, Log: c.log})
	if err != nil {
		return err
	}
	_, err = ap.Apply(ctx, set, values)
	return err
}

func ApplySet(ctx context.Context, c *Composer) error {
	ap, err := kcpapply.New(c.restCfg, kcpapply.Options{WaitTimeout: ApplyTimeout, Base: c.Workspace(), Log: c.log})
	if err != nil {
		return err
	}
	ix, err := manifests.Load()
	if err != nil {
		return err
	}
	set, ok := ix.Set(ManifestSet)
	if !ok {
		return fmt.Errorf("compose: deploy/sets.yaml names no %s set", ManifestSet)
	}
	values, err := manifests.DecodeValues(set.Values, nil)
	if err != nil {
		return err
	}
	_, err = ap.Apply(ctx, set, values)
	return err
}

type adoptKind struct {
	Kind string
	GVR  schema.GroupVersionResource
}

var adoptKinds = []adoptKind{
	{KindBid, v1alpha1.BidGVR},
	{KindComputeContract, ComputeContractGVR},
	{KindStateSync, v1alpha1.StateSyncGVR},
	{KindMarketPresence, v1alpha1.MarketPresenceGVR},
	{KindMarketIdentity, MarketIdentityGVR},
	{KindOpenBaoServer, OpenBaoServerGVR},
	{KindHostProcess, HostProcessGVR},
	{"AtprotoRecord", AtprotoRecordGVR},
	{"AtprotoRecordWatch", RecordWatchGVR},
	{"Workspace", WorkspaceGVR},
}

func adoptable(rootKind, kind string, u metav1.Object, owner string) bool {
	if len(u.GetOwnerReferences()) > 0 || u.GetDeletionTimestamp() != nil {
		return false
	}
	if kind == KindHostProcess && strings.HasPrefix(u.GetName(), hostprocess.MirrorPrefix) {
		return false
	}
	if u.GetLabels()[OwnerLabel] == owner {
		return true
	}
	switch kind {
	case KindBid, KindStateSync, KindMarketPresence:
		return rootKind == string(v1alpha1.ApplicationKindMarketBidder) || (kind != KindBid && rootKind == string(v1alpha1.ApplicationKindMarketRequester))
	case KindComputeContract:
		return rootKind == string(v1alpha1.ApplicationKindMarketRequester)
	}
	return false
}

type adoptState struct {
	uid types.UID

	generation int64

	scannedAt time.Time

	setApplied bool

	pending map[string]map[string]adoptObject
}

type adoptObject struct {
	name string

	namespace string

	uid types.UID
}

type adoptions struct {
	mu sync.Mutex

	roots map[string]*adoptState
}

func adoptKey(cluster, rootKind, name string) string {
	return cluster + "/" + rootKind + "/" + name
}

func (a *adoptions) due(r Root, now time.Time) (map[string]bool, map[string]map[string]adoptObject) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.roots == nil {
		a.roots = map[string]*adoptState{}
	}
	key := adoptKey(r.Cluster, r.Kind, r.Name)
	st := a.roots[key]
	if st == nil || st.uid != r.UID || st.generation != r.Generation || now.Sub(st.scannedAt) >= AdoptRescanEvery ||
		(st.setApplied && now.Sub(st.scannedAt) >= AdoptSetRescanAfter) {
		a.roots[key] = &adoptState{uid: r.UID, generation: r.Generation, scannedAt: now}
		all := make(map[string]bool, len(adoptKinds))
		for _, k := range adoptKinds {
			all[k.Kind] = true
		}
		return all, nil
	}
	objects := st.pending
	st.pending = nil
	return nil, objects
}

func (a *adoptions) mark(cluster, rootKind, name, kind string, obj metav1.Object) {
	a.mu.Lock()
	defer a.mu.Unlock()
	st := a.roots[adoptKey(cluster, rootKind, name)]
	if st == nil {
		return
	}
	if st.pending == nil {
		st.pending = map[string]map[string]adoptObject{}
	}
	if st.pending[kind] == nil {
		st.pending[kind] = map[string]adoptObject{}
	}
	st.pending[kind][obj.GetNamespace()+"/"+obj.GetName()] = adoptObject{name: obj.GetName(), namespace: obj.GetNamespace(), uid: obj.GetUID()}
}

func (a *adoptions) setApplied(r Root) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if st := a.roots[adoptKey(r.Cluster, r.Kind, r.Name)]; st != nil {
		st.setApplied = true
	}
}

func (a *adoptions) forget(r Root) {
	a.mu.Lock()
	defer a.mu.Unlock()
	delete(a.roots, adoptKey(r.Cluster, r.Kind, r.Name))
}

func (c *Composer) noteAdoptable(rootKind, kind string, reqs []ctlruntime.Request, src ctlruntime.Request, obj metav1.Object) {
	for _, req := range reqs {
		cluster := cmp.Or(req.Cluster, src.Cluster)
		if adoptable(rootKind, kind, obj, OwnerName(rootKind, req.Name)) {
			c.adoptions.mark(cluster, rootKind, req.Name, kind, obj)
		}
	}
}

func (c *Composer) Adopt(ctx context.Context, r Root) {
	if r.Deleting || r.UID == "" {
		return
	}
	kinds, objects := c.adoptions.due(r, time.Now())
	for _, k := range adoptKinds {
		switch {
		case kinds[k.Kind]:
			c.adoptAll(ctx, r, k)
		case len(objects[k.Kind]) > 0:
			apiVersion := k.GVR.GroupVersion().String()
			for _, o := range objects[k.Kind] {
				c.adoptOne(ctx, r, k, apiVersion, k.Kind, o)
			}
		}
	}
}

func (c *Composer) adoptAll(ctx context.Context, r Root, k adoptKind) {
	ri, err := c.Client(r.Workspace, k.GVR)
	if err != nil {
		return
	}
	items, err := kube.List[Generic](ctx, ri, metav1.ListOptions{})
	if err != nil {
		return
	}
	for i := range items {
		u := &items[i]
		if adoptable(r.Kind, k.Kind, u, r.Owner()) {
			c.adoptOne(ctx, r, k, u.APIVersion, u.Kind, adoptObject{name: u.Name, namespace: u.Namespace, uid: u.UID})
		}
	}
}

func (c *Composer) adoptOne(ctx context.Context, r Root, k adoptKind, apiVersion, kind string, o adoptObject) {
	patch := NewDoc(apiVersion, kind, o.name)
	meta := DocMeta(patch)
	if o.namespace != "" {
		meta["namespace"] = o.namespace
	}
	if o.uid != "" {
		meta["uid"] = string(o.uid)
	}
	meta["ownerReferences"] = []metav1.OwnerReference{r.OwnerRef()}
	if err := c.applyAs(ctx, r, r.AdoptManager(), r.Workspace, k.GVR, patch); err != nil {
		if !apierrors.IsNotFound(err) && !apierrors.IsConflict(err) {
			c.log.Warn("adopting an intent object under its root", "root", r.Owner(), "kind", k.Kind, "name", o.name, "err", err.Error())
		}
		return
	}
	c.log.Info("an intent object hangs from its root", "root", r.Owner(), "kind", k.Kind, "name", o.name)
	c.samples.gcDirty.Store(true)
}

type quotaEntry struct {
	component string
	gvr       schema.GroupVersionResource
	field     string
	def       int64
	gate      bool
}

func (e quotaEntry) key() string { return "count/" + e.gvr.Resource + "." + e.gvr.Group }

func (e quotaEntry) name() string { return e.gvr.Resource + "." + e.gvr.Group }

func quotaPlan(rootKind string) []quotaEntry {
	runs := []quotaEntry{{"", PolicyWorkflowRunGVR, "policyWorkflowRuns", QuotaDefaults.PolicyWorkflowRuns, false}}
	var entries []quotaEntry
	switch rootKind {
	case string(v1alpha1.ApplicationKindMarketBidder):
		entries = append([]quotaEntry{
			{"", v1alpha1.BidGVR, "bids", QuotaDefaults.Bids, true},
			{"", v1alpha1.ComputeVMGVR, "computeVMs", QuotaDefaults.ComputeVMs, true},
		}, runs...)
	case string(v1alpha1.ApplicationKindMarketRequester):
		entries = append([]quotaEntry{{"", ComputeContractGVR, "computeContracts", QuotaDefaults.ComputeContracts, false}}, runs...)
	case string(v1alpha1.ApplicationKindGateway):
		entries = append([]quotaEntry{{"", ComputeContractGVR, "computeContracts", QuotaDefaults.ComputeContracts, false}}, runs...)
	case string(v1alpha1.ApplicationKindSshGateway), "MarketNetwork":
		entries = runs
	default:
		return nil
	}
	for _, kind := range ComponentKindsFor(rootKind) {
		if ct, ok := ComponentTypeFor(kind); ok {
			entries = append(entries, quotaEntry{"", ct.GVR, "components", QuotaDefaults.Components, false})
		}
	}
	return entries
}

func limitOf(configured int32, def int64) int64 {
	if configured != 0 {
		return int64(configured)
	}
	return def
}

func BidderQuotaArgs(q *v1alpha1.ApplicationQuota) []string {
	var bids, vms int32
	if q != nil {
		bids, vms = q.Bids, q.ComputeVMs
	}
	var out []string
	if n := limitOf(bids, QuotaDefaults.Bids); n > 0 {
		out = append(out, "--bid-quota", strconv.FormatInt(n, 10))
	}
	if n := limitOf(vms, QuotaDefaults.ComputeVMs); n > 0 {
		out = append(out, "--computevm-quota", strconv.FormatInt(n, 10))
	}
	return out
}

func quotaLimit(q *v1alpha1.ApplicationQuota, e quotaEntry) int64 {
	if q == nil {
		return e.def
	}
	var n int32
	switch e.field {
	case "bids":
		n = q.Bids
	case "computeVMs":
		n = q.ComputeVMs
	case "computeContracts":
		n = q.ComputeContracts
	case "policyWorkflowRuns":
		n = q.PolicyWorkflowRuns
	}
	return limitOf(n, e.def)
}

func (c *Composer) quotaWorkspace(r Root, component string) string {
	if component == "" {
		return r.Workspace
	}
	return r.Workspace + ":" + component
}

func (c *Composer) ApplyQuotas(ctx context.Context, r Root, q *v1alpha1.ApplicationQuota) {
	if r.Deleting || r.UID == "" {
		return
	}
	byWS := map[string]map[string]string{}
	for _, e := range quotaPlan(r.Kind) {
		if e.gate {
			continue
		}
		ws := c.quotaWorkspace(r, e.component)
		if byWS[ws] == nil {
			byWS[ws] = map[string]string{}
		}
		if n := quotaLimit(q, e); n > 0 {
			byWS[ws][e.key()] = fmt.Sprint(n)
		}
	}
	for ws, hard := range byWS {
		if err := c.applyQuota(ctx, r, ws, hard); err != nil && ctx.Err() == nil {
			c.log.Warn("applying the object-count quota of a workspace", "workspace", ws, "err", err.Error())
		}
	}
	c.samples.recordQuotas(r, q)
}

func (c *Composer) applyQuota(ctx context.Context, r Root, ws string, hard map[string]string) error {
	ri, err := c.Client(ws, ResourceQuotaGVR)
	if err != nil {
		return err
	}
	if len(hard) == 0 {
		if err := ri.In(QuotaNamespace).Delete(ctx, QuotaName, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
			return err
		}
		return nil
	}
	hardAny := map[string]any{}
	for k, v := range hard {
		hardAny[k] = v
	}
	u := NewDoc("v1", "ResourceQuota", QuotaName)
	u["spec"] = map[string]any{"hard": hardAny}
	meta := DocMeta(u)
	meta["namespace"] = QuotaNamespace
	if ws == r.Workspace {
		meta["ownerReferences"] = []metav1.OwnerReference{r.OwnerRef()}
	}
	return c.applyAs(ctx, r, r.Manager(), ws, ResourceQuotaGVR, u)
}

type samples struct {
	mu sync.Mutex

	stateDir *v1alpha1.StateDirUsage

	stateDirAt time.Time

	gc *v1alpha1.GCStatus

	gcAt time.Time

	gcDirty atomic.Bool

	quotaAt time.Time

	quotas map[string]rootQuota

	quotaUsage []v1alpha1.QuotaUsage
}

type rootQuota struct {
	root Root

	quota *v1alpha1.ApplicationQuota
}

func (s *samples) recordQuotas(r Root, q *v1alpha1.ApplicationQuota) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.quotas == nil {
		s.quotas = map[string]rootQuota{}
	}
	s.quotas[r.Owner()] = rootQuota{root: r, quota: q}
}

func (s *samples) into(ctx context.Context, c *Composer, m map[string]any, declared map[string]bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	now := time.Now()
	if dir := c.cfg.Layout.StateDir; dir != "" && now.Sub(s.stateDirAt) > StateDirSampleEvery {
		s.stateDirAt = now
		if u, err := diskusage.Of(dir); err == nil {
			s.stateDir = &v1alpha1.StateDirUsage{Path: dir, Bytes: u.Bytes, Files: u.Files, SampledAt: now.UTC().Format(time.RFC3339)}
		}
	}
	if s.gcAt.IsZero() {
		s.gcAt = now
		s.gcDirty.Store(true)
	}
	if now.Sub(s.gcAt) > GCSampleEvery && (s.gcDirty.Load() || now.Sub(s.gcAt) > GCRescanEvery) {
		s.gcAt = now
		s.gcDirty.Store(false)
		if st := c.scanGC(ctx); st != nil {
			s.gc = st
		}
	}
	if now.Sub(s.quotaAt) > GCSampleEvery {
		s.quotaAt = now
		s.quotaUsage = c.readQuotas(ctx, s.quotas)
	}
	put := func(key string, v any) {
		if !declared[key] {
			return
		}
		var out any
		if b, err := json.Marshal(v); err == nil && json.Unmarshal(b, &out) == nil {
			m[key] = out
		}
	}
	if s.stateDir != nil {
		put("stateDir", s.stateDir)
	}
	if s.gc != nil {
		put("gc", s.gc)
	}
	if len(s.quotaUsage) > 0 {
		put("quotas", s.quotaUsage)
	}
	if c.cfg.TTLReport != nil {
		if st := c.cfg.TTLReport(); st != nil {
			put("ttl", st)
		}
	}
	if c.cfg.StoreUsage != nil {
		if u := c.cfg.StoreUsage(); u != nil {
			put("store", u)
		}
	}
}

func (c *Composer) countObjects(ctx context.Context, ws string, res schema.GroupVersionResource) (int64, bool) {
	ri, err := c.Client(ws, res)
	if err != nil {
		return 0, false
	}
	items, meta, err := kube.ListPage[metav1.PartialObjectMetadata](ctx, ri, metav1.ListOptions{Limit: 1})
	if err != nil {
		return 0, false
	}
	n := int64(len(items))
	if rem := meta.RemainingItemCount; rem != nil {
		return n + *rem, true
	}
	if meta.Continue != "" {
		all, err := kube.List[metav1.PartialObjectMetadata](ctx, ri, metav1.ListOptions{})
		if err != nil {
			return 0, false
		}
		return int64(len(all)), true
	}
	return n, true
}

func (c *Composer) readQuotas(ctx context.Context, roots map[string]rootQuota) []v1alpha1.QuotaUsage {
	var out []v1alpha1.QuotaUsage
	for _, rq := range roots {
		for _, e := range quotaPlan(rq.root.Kind) {
			hard := quotaLimit(rq.quota, e)
			if hard <= 0 {
				continue
			}
			ws := c.quotaWorkspace(rq.root, e.component)
			used, ok := c.countObjects(ctx, ws, e.gvr)
			if !ok {
				continue
			}
			out = append(out, v1alpha1.QuotaUsage{Workspace: ws, Resource: e.name(), Hard: hard, Used: used})
		}
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Workspace != out[j].Workspace {
			return out[i].Workspace < out[j].Workspace
		}
		return out[i].Resource < out[j].Resource
	})
	if len(out) > GCListCap {
		out = out[:GCListCap]
	}
	return out
}

var gcKinds = []adoptKind{
	{KindBid, v1alpha1.BidGVR},
	{KindComputeVM, v1alpha1.ComputeVMGVR},
	{"GuestIdentityGrant", v1alpha1.GuestIdentityGrantGVR},
	{KindHostProcess, HostProcessGVR},
	{KindOpenBaoServer, OpenBaoServerGVR},
	{"Postgres", v1alpha1.PostgresGVR},
	{KindMarketIdentity, MarketIdentityGVR},
	{KindComputeContract, ComputeContractGVR},
	{"NodeClaim", v1alpha1.NodeClaimGVR},
	{KindStateSync, v1alpha1.StateSyncGVR},
	{KindMarketPresence, v1alpha1.MarketPresenceGVR},
}

func gcAllKinds() []adoptKind {
	out := slices.Clone(gcKinds)
	for _, t := range ComponentTypes {
		out = append(out, adoptKind{t.Kind, t.GVR})
	}
	for _, t := range PolicyTypes {
		if t.Kind != ComponentKindPolicyEngine {
			out = append(out, adoptKind{t.Kind, t.GVR})
		}
	}
	return out
}

func (c *Composer) gcWorkspaces(ctx context.Context) []string {
	root := c.Workspace()
	out := []string{root}
	ri, err := c.Client(root, WorkspaceGVR)
	if err != nil {
		return out
	}
	list, err := kube.List[tenancyv1alpha1.Workspace](ctx, ri, metav1.ListOptions{})
	if err != nil {
		return out
	}
	for _, w := range list {
		name := w.Name
		if _, app := w.Labels[AppWorkspaceLabel]; app || name == providerWS || w.DeletionTimestamp != nil {
			continue
		}
		if w.Status.Phase != "Ready" {
			continue
		}
		out = append(out, root+":"+name)
		if w.Spec.Type != nil && w.Spec.Type.Name == accountWorkspace && len(out) < GCWorkspaceCap {
			sub, err := c.Client(root+":"+name, WorkspaceGVR)
			if err != nil {
				continue
			}
			homes, err := kube.List[metav1.PartialObjectMetadata](ctx, sub, metav1.ListOptions{})
			if err != nil {
				continue
			}
			for _, h := range homes {
				if h.DeletionTimestamp == nil && len(out) < GCWorkspaceCap {
					out = append(out, root+":"+name+":"+h.Name)
				}
			}
		}
	}
	return out
}

func specField(u *Generic, field string) bool {
	_, found := u.Spec[field]
	return found
}

func ownedByLink(u *Generic) bool {
	if len(u.OwnerReferences) > 0 {
		return true
	}
	if specField(u, "ttlSecondsAfterFinished") {
		return true
	}
	labels := u.Labels
	if _, ok := labels[ClusterLabel]; ok {
		return true
	}
	_, ok := labels[OwnerLabel]
	return ok
}

func addExample(ex []string, v string) []string {
	if len(ex) < GCExamples {
		return append(ex, v)
	}
	return ex
}

func finalizerController(f string) string {
	if c, ok := FinalizerControllers[f]; ok {
		return c
	}
	return "unknown: no controller is registered for this finalizer"
}

func (c *Composer) scanGC(ctx context.Context) *v1alpha1.GCStatus {
	var total, orphanTotal, stuckTotal int64
	orphans := map[string]*v1alpha1.GCKindCount{}
	stuck := map[string]*v1alpha1.GCStuck{}
	kinds := gcAllKinds()
	for _, ws := range c.gcWorkspaces(ctx) {
		for _, k := range kinds {
			ri, err := c.Client(ws, k.GVR)
			if err != nil {
				continue
			}
			items, err := kube.List[Generic](ctx, ri, metav1.ListOptions{})
			if err != nil {
				if ctx.Err() != nil {
					return nil
				}
				continue
			}
			for i := range items {
				u := &items[i]
				total++
				ref := ws + "/" + u.Name
				if dt := u.DeletionTimestamp; dt != nil {
					if time.Since(dt.Time) > GCStuckAfter {
						for _, f := range u.Finalizers {
							key := k.Kind + "|" + f
							s := stuck[key]
							if s == nil {
								s = &v1alpha1.GCStuck{Kind: k.Kind, Finalizer: f, Controller: finalizerController(f)}
								stuck[key] = s
							}
							s.Count++
							s.Examples = addExample(s.Examples, ref)
							stuckTotal++
						}
					}
					continue
				}
				if (k.Kind == KindHostProcess && strings.HasPrefix(u.Name, hostprocess.MirrorPrefix)) || ownedByLink(u) {
					continue
				}
				cnt := orphans[k.Kind]
				if cnt == nil {
					cnt = &v1alpha1.GCKindCount{Kind: k.Kind}
					orphans[k.Kind] = cnt
				}
				cnt.Count++
				cnt.Examples = addExample(cnt.Examples, ref)
				orphanTotal++
			}
		}
	}
	st := &v1alpha1.GCStatus{SampledAt: time.Now().UTC().Format(time.RFC3339), Objects: total, OrphanTotal: orphanTotal, StuckTotal: stuckTotal}
	for _, cnt := range orphans {
		st.Orphans = append(st.Orphans, *cnt)
	}
	sort.Slice(st.Orphans, func(i, j int) bool { return st.Orphans[i].Kind < st.Orphans[j].Kind })
	if len(st.Orphans) > GCListCap {
		st.Orphans = st.Orphans[:GCListCap]
	}
	for _, s := range stuck {
		st.Terminating = append(st.Terminating, *s)
	}
	sort.Slice(st.Terminating, func(i, j int) bool {
		if st.Terminating[i].Kind != st.Terminating[j].Kind {
			return st.Terminating[i].Kind < st.Terminating[j].Kind
		}
		return st.Terminating[i].Finalizer < st.Terminating[j].Finalizer
	})
	if len(st.Terminating) > GCListCap {
		st.Terminating = st.Terminating[:GCListCap]
	}
	return st
}

func componentRank(kind string) int {
	switch kind {
	case ComponentKindService, ComponentKindExposure:
		return 0
	case ComponentKindGatewayAPI, ComponentKindSshServer, ComponentKindBidder:
		return 1
	case ComponentKindRequester, ComponentKindGuestControlPlane, ComponentKindPolicyEngine:
		return 2
	case ComponentKindIngressProxy, ComponentKindDispatcherTLS:
		return 3
	case ComponentKindPlc, ComponentKindRelay:
		return 4
	case ComponentKindPds:
		return 5
	}
	return 6
}

const AccountWorkspaceType = "account"

func (c *Composer) DeleteAccountClusters(ctx context.Context) (int, error) {
	ri, err := c.Client(clusterhome.RootWorkspace, WorkspaceGVR)
	if err != nil {
		return 0, err
	}
	accounts, err := kube.List[tenancyv1alpha1.Workspace](ctx, ri, metav1.ListOptions{})
	if err != nil {
		return 0, err
	}
	left := 0
	for _, account := range accounts {
		if account.Spec.Type == nil || account.Spec.Type.Name != AccountWorkspaceType {
			continue
		}
		n, err := c.deleteClustersIn(ctx, clusterhome.RootWorkspace+":"+account.Name)
		left += n
		if err != nil {
			return left, err
		}
	}
	return left, nil
}

func (c *Composer) DeleteClustersOf(ctx context.Context, did string) (int, error) {
	return c.deleteClustersIn(ctx, clusterhome.AccountPath(did))
}

func (c *Composer) deleteClustersIn(ctx context.Context, path string) (int, error) {
	hri, err := c.Client(path, WorkspaceGVR)
	if err != nil {
		return 0, err
	}
	homes, err := kube.List[tenancyv1alpha1.Workspace](ctx, hri, metav1.ListOptions{})
	if apierrors.IsNotFound(err) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	left := 0
	for _, home := range homes {
		if home.Spec.Mount != nil {
			continue
		}
		n, err := c.DeleteAll(ctx, path+":"+home.Name, v1alpha1.ClusterGVR)
		if err != nil && home.DeletionTimestamp == nil {
			return left, err
		}
		left += n
	}
	return left, nil
}

func (c *Composer) StopComponents(ctx context.Context, r Root, ws string) (int, error) {
	ws = workspaceOr(ws, r.Workspace)
	views, err := c.ListComponents(ctx, ws, r.Owner())
	if err != nil {
		return 0, err
	}
	if len(views) == 0 {
		return 0, c.DeleteLinks(ctx, r, ws)
	}
	lowest := componentRank(views[0].Kind)
	for _, v := range views {
		lowest = min(lowest, componentRank(v.Kind))
	}
	for _, v := range views {
		if componentRank(v.Kind) == lowest {
			if err := c.DeleteComponent(ctx, ws, v.Kind, v.Name); err != nil {
				return len(views), err
			}
		}
	}
	return len(views), nil
}

func (c *Composer) ReleaseRecords(ctx context.Context, recordsWs string, pdsUp bool) (int, error) {
	ri, err := c.Client(recordsWs, AtprotoRecordGVR)
	if err != nil {
		return 0, err
	}
	items, err := kube.List[metav1.PartialObjectMetadata](ctx, ri, metav1.ListOptions{})
	if apierrors.IsNotFound(err) || (err == nil && len(items) == 0) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	for i := range items {
		o := &items[i]
		switch {
		case o.DeletionTimestamp == nil:
			err = ri.Delete(ctx, o.Name, metav1.DeleteOptions{})
		case !pdsUp:
			_, err = kube.Patch[metav1.PartialObjectMetadata](ctx, ri, o.Name, types.MergePatchType, []byte(`{"metadata":{"finalizers":null}}`), metav1.PatchOptions{})
		}
		if err != nil && !apierrors.IsNotFound(err) {
			return len(items), err
		}
	}
	return len(items), nil
}

func Stamp(parts ...string) string {
	sum := sha256.Sum256([]byte(strings.Join(parts, "|")))
	return hex.EncodeToString(sum[:8])
}