johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package compose

import (
	"context"
	"maps"
	"reflect"
	"sort"
	"strings"
	"sync"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	eventsv1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/events/v1alpha1"
	exposurev1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/exposure/v1alpha1"
	pdsv1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/pds/v1alpha1"
	policyv1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/policy/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
)

const (
	ExportApplications     = "applications"
	ExportPolicy           = "policy"
	ExportExposures        = "exposures"
	ExportHostProcesses    = "hostprocesses"
	ExportMarketIdentities = "marketidentities"
	ExportComputeVMs       = "computevms"
	ExportComputeContracts = "computecontracts"
	ExportEvents           = "events"
	ExportPDS              = "pds"
	ExportClusters         = "clusters"
)

type Root struct {
	APIVersion string

	Kind string

	Name string

	UID types.UID

	Generation int64

	Workspace string

	Cluster string

	Deleting bool

	Suspend bool
}

func RootOf(req ctlruntime.Request, kind string, obj ctlruntime.Object, suspend bool) Root {
	return Root{APIVersion: v1alpha1.GroupVersion.String(), Kind: kind, Name: obj.GetName(), UID: obj.GetUID(),
		Generation: obj.GetGeneration(), Workspace: req.Workspace, Cluster: req.Cluster,
		Deleting: !obj.GetDeletionTimestamp().IsZero(), Suspend: suspend}
}

func OwnerName(kind, name string) string {
	return strings.ToLower(kind) + "." + name
}

func (r Root) Owner() string { return OwnerName(r.Kind, r.Name) }

func (r Root) Manager() string { return ctlruntime.FieldManagerPrefix + strings.ToLower(r.Kind) }

func (r Root) AdoptManager() string { return r.Manager() + "/adopt" }

func (r Root) IdentityManager() string { return r.Manager() + "/identity" }

func (r Root) ManagedBy() string { return "socialweb.computer." + strings.ToLower(r.Kind) }

func (r Root) Ref() commonv1.ObjectRef {
	return commonv1.ObjectRef{Group: v1alpha1.GroupName, Kind: r.Kind, Name: r.Name, Workspace: r.Workspace}
}

func (r Root) OwnerRef() metav1.OwnerReference {
	return metav1.OwnerReference{APIVersion: r.APIVersion, Kind: r.Kind, Name: r.Name, UID: r.UID, Controller: new(true)}
}

func (r Root) Labels(componentKind string) map[string]string {
	out := map[string]string{
		OwnerLabel:     r.Owner(),
		RootKindLabel:  r.Kind,
		ManagedByLabel: r.ManagedBy(),
	}
	if componentKind != "" {
		out[ComponentLabel] = componentKind
	}
	if r.Cluster != "" {
		out[RootClusterLabel] = r.Cluster
	}
	return out
}

type rootIndex struct {
	mu sync.Mutex

	byCluster map[string]map[string]map[string]bool

	deleting map[types.UID]Root
}

func (c *Composer) Observe(r Root) {
	c.roots.mu.Lock()
	defer c.roots.mu.Unlock()
	kinds := c.roots.byCluster[r.Cluster]
	if kinds == nil {
		kinds = map[string]map[string]bool{}
		c.roots.byCluster[r.Cluster] = kinds
	}
	if kinds[r.Kind] == nil {
		kinds[r.Kind] = map[string]bool{}
	}
	kinds[r.Kind][r.Name] = true
	if r.Deleting && r.UID != "" {
		if c.roots.deleting == nil {
			c.roots.deleting = map[types.UID]Root{}
		}
		c.roots.deleting[r.UID] = r
	} else {
		delete(c.roots.deleting, r.UID)
	}
}

func (c *Composer) Drop(r Root) {
	c.roots.mu.Lock()
	defer c.roots.mu.Unlock()
	if kinds := c.roots.byCluster[r.Cluster]; kinds != nil && kinds[r.Kind] != nil {
		delete(kinds[r.Kind], r.Name)
	}
	delete(c.roots.deleting, r.UID)
	c.mu.Lock()
	delete(c.applied, string(r.UID))
	c.mu.Unlock()
	c.adoptions.forget(r)
	c.linkSets.forget(linkRoot(r))
}

func (c *Composer) rootsIn(cluster, kind string) []string {
	c.roots.mu.Lock()
	defer c.roots.mu.Unlock()
	var out []string
	for name := range c.roots.byCluster[cluster][kind] {
		out = append(out, name)
	}
	sort.Strings(out)
	return out
}

func (c *Composer) deletingIn(rootKind, workspace string) []ctlruntime.Request {
	c.roots.mu.Lock()
	defer c.roots.mu.Unlock()
	var out []ctlruntime.Request
	for _, r := range c.roots.deleting {
		if r.Kind == rootKind && (r.Workspace == workspace || r.Workspace+":"+ComponentWorkspace == workspace) {
			out = append(out, ctlruntime.Request{Cluster: r.Cluster, Name: r.Name})
		}
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Cluster != out[j].Cluster {
			return out[i].Cluster < out[j].Cluster
		}
		return out[i].Name < out[j].Name
	})
	return out
}

func DrainWatch[C ctlruntime.Object](c *Composer, export, rootKind string) ctlruntime.Option {
	return ctlruntime.WatchesIn(export, func(_ context.Context, src ctlruntime.Request, _ C) []ctlruntime.Request {
		return c.deletingIn(rootKind, src.Workspace)
	})
}

func OperatorRecordWatch(c *Composer, export, rootKind string) ctlruntime.Option {
	filter := ctlruntime.WatchFilter[*v1alpha1.AtprotoRecord]{Changed: specChanged[*v1alpha1.AtprotoRecord]}
	return ctlruntime.WatchesFiltered(export, filter, func(_ context.Context, src ctlruntime.Request, rec *v1alpha1.AtprotoRecord) []ctlruntime.Request {
		if rec.GetLabels()[OperatorRecordLabel] != OperatorRepoName {
			return nil
		}
		var out []ctlruntime.Request
		for _, name := range c.rootsIn(src.Cluster, rootKind) {
			out = append(out, ctlruntime.Request{Cluster: src.Cluster, Name: name})
		}
		return c.unparked(rootKind, out, src)
	})
}

func TeardownWatch[C ctlruntime.Object](c *Composer, export, rootKind string) ctlruntime.Option {
	return ctlruntime.WatchesIn(export, func(_ context.Context, _ ctlruntime.Request, _ C) []ctlruntime.Request {
		return c.deletingOf(rootKind)
	})
}

func SiblingWatch[C ctlruntime.Object](c *Composer, export, rootKind string) ctlruntime.Option {
	return ctlruntime.WatchesIn(export, func(_ context.Context, src ctlruntime.Request, _ C) []ctlruntime.Request {
		var out []ctlruntime.Request
		for _, name := range c.rootsIn(src.Cluster, rootKind) {
			out = append(out, ctlruntime.Request{Cluster: src.Cluster, Name: name})
		}
		return out
	})
}

func (c *Composer) deletingOf(rootKind string) []ctlruntime.Request {
	c.roots.mu.Lock()
	defer c.roots.mu.Unlock()
	var out []ctlruntime.Request
	for _, r := range c.roots.deleting {
		if r.Kind == rootKind {
			out = append(out, ctlruntime.Request{Cluster: r.Cluster, Name: r.Name})
		}
	}
	return out
}

func (c *Composer) Mine(req ctlruntime.Request) bool {
	return strings.TrimSpace(req.Workspace) == c.Workspace()
}

func mapByLabel(rootKind string, src ctlruntime.Request, labels map[string]string) []ctlruntime.Request {
	if labels[RootKindLabel] != rootKind {
		return nil
	}
	owner := labels[OwnerLabel]
	prefix := strings.ToLower(rootKind) + "."
	name, ok := strings.CutPrefix(owner, prefix)
	if !ok || name == "" {
		return nil
	}
	cluster := src.Cluster
	if lc := strings.TrimSpace(labels[RootClusterLabel]); lc != "" {
		cluster = lc
	}
	return []ctlruntime.Request{{Cluster: cluster, Name: name}}
}

func observed[C ctlruntime.Object](c *Composer, changed func(old, new C) bool) ctlruntime.WatchFilter[C] {
	if t := reflect.TypeFor[C](); t.Kind() == reflect.Pointer {
		c.applies.watch(t.Elem().Name())
	}
	return ctlruntime.WatchFilter[C]{Changed: changed, Observe: func(obj C, gone bool) { c.applies.observe(obj, gone) }}
}

func specChanged[C ctlruntime.Object](old, new C) bool {
	return old.GetGeneration() != new.GetGeneration() || !maps.Equal(old.GetLabels(), new.GetLabels()) ||
		old.GetDeletionTimestamp().IsZero() != new.GetDeletionTimestamp().IsZero() ||
		!reflect.DeepEqual(old.GetOwnerReferences(), new.GetOwnerReferences())
}

func linkChanged(old, new *exposurev1alpha1.ComponentLink) bool {
	return specChanged(old, new) || old.Status.ObservedGeneration != new.Status.ObservedGeneration || linkAccepted(old) != linkAccepted(new)
}

func childWatch[C ctlruntime.Object](c *Composer, export, rootKind string) ctlruntime.Option {
	return childWatchIf[C](c, export, rootKind, nil)
}

func childWatchIf[C ctlruntime.Object](c *Composer, export, rootKind string, changed func(old, new C) bool) ctlruntime.Option {
	return ctlruntime.WatchesFiltered(export, observed(c, changed), func(_ context.Context, src ctlruntime.Request, obj C) []ctlruntime.Request {
		return c.unparked(rootKind, mapByLabel(rootKind, src, obj.GetLabels()), src)
	})
}

func appliedWatch[C ctlruntime.Object](c *Composer, export string) ctlruntime.Option {
	return ctlruntime.WatchesFiltered(export, observed(c, func(_, _ C) bool { return false }), func(context.Context, ctlruntime.Request, C) []ctlruntime.Request {
		return nil
	})
}

func adoptWatch[C ctlruntime.Object](c *Composer, export, rootKind, kind string) ctlruntime.Option {
	return ctlruntime.WatchesFiltered(export, observed[C](c, nil), func(_ context.Context, src ctlruntime.Request, obj C) []ctlruntime.Request {
		out := mapByLabel(rootKind, src, obj.GetLabels())
		c.noteAdoptable(rootKind, kind, out, src, obj)
		if kind == KindMarketIdentity {
			c.unpark(rootKind, out, src)
			return out
		}
		return c.unparked(rootKind, out, src)
	})
}

func intentWatch[C ctlruntime.Object](c *Composer, export, rootKind, kind string) ctlruntime.Option {
	return ctlruntime.WatchesFiltered(export, observed[C](c, nil), func(_ context.Context, src ctlruntime.Request, obj C) []ctlruntime.Request {
		if out := mapByLabel(rootKind, src, obj.GetLabels()); len(out) > 0 {
			c.noteAdoptable(rootKind, kind, out, src, obj)
			return c.unparked(rootKind, out, src)
		}
		if len(obj.GetOwnerReferences()) > 0 && obj.GetDeletionTimestamp().IsZero() {
			return nil
		}
		c.samples.gcDirty.Store(true)
		var out []ctlruntime.Request
		for _, name := range c.rootsIn(src.Cluster, rootKind) {
			out = append(out, ctlruntime.Request{Cluster: src.Cluster, Name: name})
		}
		c.noteAdoptable(rootKind, kind, out, src, obj)
		return c.unparked(rootKind, out, src)
	})
}

func (c *Composer) Watches(rootKind string) []ctlruntime.Option {
	return []ctlruntime.Option{
		childWatch[*v1alpha1.Bidder](c, "", rootKind),
		childWatch[*v1alpha1.GuestControlPlane](c, "", rootKind),
		childWatch[*v1alpha1.Pds](c, "", rootKind),
		childWatch[*v1alpha1.Requester](c, "", rootKind),
		childWatch[*v1alpha1.GatewayAPI](c, "", rootKind),
		childWatch[*v1alpha1.SshServer](c, "", rootKind),
		childWatch[*v1alpha1.Plc](c, "", rootKind),
		childWatch[*v1alpha1.Relay](c, "", rootKind),
		childWatch[*v1alpha1.IngressProxy](c, "", rootKind),
		childWatch[*v1alpha1.DispatcherTLS](c, "", rootKind),
		childWatchIf(c, "", rootKind, specChanged[*v1alpha1.LocalRepo]),
		childWatchIf(c, "", rootKind, specChanged[*v1alpha1.RemoteRepo]),
		childWatchIf(c, "", rootKind, specChanged[*v1alpha1.AtprotoRepo]),
		intentWatch[*v1alpha1.MarketPresence](c, "", rootKind, KindMarketPresence),
		intentWatch[*v1alpha1.StateSync](c, "", rootKind, KindStateSync),
		childWatch[*policyv1alpha1.PolicyEngine](c, ExportPolicy, rootKind),
		childWatch[*exposurev1alpha1.Service](c, ExportExposures, rootKind),
		childWatch[*exposurev1alpha1.Exposure](c, ExportExposures, rootKind),
		childWatchIf(c, ExportExposures, rootKind, linkChanged),
		adoptWatch[*v1alpha1.OpenBaoServer](c, ExportHostProcesses, rootKind, KindOpenBaoServer),
		childWatch[*v1alpha1.Vault](c, ExportHostProcesses, rootKind),
		adoptWatch[*v1alpha1.MarketIdentity](c, ExportMarketIdentities, rootKind, KindMarketIdentity),
		adoptWatch[*v1alpha1.AtprotoRecordWatch](c, ExportMarketIdentities, rootKind, "AtprotoRecordWatch"),
		intentWatch[*v1alpha1.Bid](c, ExportComputeVMs, rootKind, KindBid),
		intentWatch[*v1alpha1.ComputeContract](c, ExportComputeContracts, rootKind, KindComputeContract),
		intentWatch[*v1alpha1.Infra](c, ExportComputeContracts, rootKind, "Infra"),
		intentWatch[*v1alpha1.EphemeralInfra](c, ExportComputeContracts, rootKind, "EphemeralInfra"),
		childWatchIf(c, ExportEvents, rootKind, specChanged[*eventsv1alpha1.EventStream]),
		childWatchIf(c, ExportEvents, rootKind, specChanged[*eventsv1alpha1.EventSource]),
		childWatchIf(c, ExportEvents, rootKind, specChanged[*eventsv1alpha1.PdsEventSource]),
		childWatchIf(c, ExportEvents, rootKind, specChanged[*eventsv1alpha1.RelayEventSource]),
		childWatchIf(c, ExportEvents, rootKind, specChanged[*eventsv1alpha1.JetstreamEventSource]),
		childWatchIf(c, ExportPDS, rootKind, specChanged[*pdsv1alpha1.RemotePDS]),
		appliedWatch[*v1alpha1.PdsAccount](c, ""),
		appliedWatch[*v1alpha1.AtprotoSession](c, ""),
		appliedWatch[*v1alpha1.PlcIdentity](c, ""),
		appliedWatch[*v1alpha1.VaultCredential](c, ""),
	}
}