johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package infracollection

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"slices"
	"strings"
	"sync"
	"time"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/validation"
	"k8s.io/utils/ptr"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	exposurev1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/exposure/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bus"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/atprotorepo"
	infractl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/infra"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/infra"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/binding"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/conds"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/netretry"
)

const (
	Name = "infracollection"

	EphemeralName = "ephemeralinfracollection"
)

type Options struct {
	Bus *bus.Bus

	Dynamic kube.Factory

	Scope func(workspace string) bool

	Log *slog.Logger
}

type Controller struct {
	bus *bus.Bus

	dynamic kube.Factory

	log *slog.Logger

	repos map[string]*binding.Table[struct{}]

	retries *netretry.Tracker[retryKey]

	mu sync.Mutex

	over map[retryKey]map[string]bool

	absent map[retryKey]absentAt
}

type absentAt struct {
	index string

	uris map[string]bool
}

type retryKey struct {
	kind string

	cluster string

	name string
}

func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) {
	if opts.Bus == nil {
		return nil, errors.New("infracollection: the bus is required")
	}
	if opts.Dynamic == nil {
		return nil, errors.New("infracollection: the dynamic client factory is required")
	}
	if err := m.AddToScheme(v1alpha1.AddToScheme, exposurev1alpha1.AddToScheme); err != nil {
		return nil, err
	}
	log := opts.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	c := &Controller{bus: opts.Bus, dynamic: opts.Dynamic, log: log.With("controller", Name), retries: netretry.NewTracker[retryKey](), over: map[retryKey]map[string]bool{}, absent: map[retryKey]absentAt{}, repos: map[string]*binding.Table[struct{}]{
		infractl.CollectionKind: binding.New[struct{}](), infractl.EphemeralCollectionKind: binding.New[struct{}]()}}
	if err := register[*v1alpha1.InfraCollection, *v1alpha1.Infra](m, c, Name, infraOps, opts.Scope); err != nil {
		return nil, err
	}
	if err := register[*v1alpha1.EphemeralInfraCollection, *v1alpha1.EphemeralInfra](m, c, EphemeralName, ephemeralOps, opts.Scope); err != nil {
		return nil, err
	}
	return c, nil
}

func register[T, C ctlruntime.Object](m *ctlruntime.Manager, c *Controller, name string, ops collectionOps[T], scope func(string) bool) error {
	r := &reconciler[T]{c: c, ops: ops, manager: ctlruntime.FieldManagerPrefix + name}
	options := []ctlruntime.Option{ctlruntime.Export(infractl.Export), ctlruntime.Owns[C](),
		ctlruntime.WatchesIn(infractl.RecordExport, func(_ context.Context, _ ctlruntime.Request, obj *v1alpha1.AtprotoRecordWatch) []ctlruntime.Request {
			labels := obj.GetLabels()
			if labels[v1alpha1.InfraKindLabel] != ops.kind || labels[v1alpha1.InfraNameLabel] == "" {
				return nil
			}
			return []ctlruntime.Request{{Name: labels[v1alpha1.InfraNameLabel]}}
		}),
		ctlruntime.WatchesFiltered(infractl.RepoExport, ctlruntime.WatchFilter[*v1alpha1.AtprotoRepo]{Changed: func(old, new *v1alpha1.AtprotoRepo) bool {
			return old.Status.DID != new.Status.DID
		}}, func(_ context.Context, src ctlruntime.Request, obj *v1alpha1.AtprotoRepo) []ctlruntime.Request {
			return c.repoUsers(ops.kind, src.Workspace, obj.Name)
		}),
		infractl.RepoLinkWatch(func(workspace, name string) []ctlruntime.Request { return c.repoUsers(ops.kind, workspace, name) }),
		ctlruntime.Events(infractl.RepoBindEvents(c.bus, func(workspace, name string) []ctlruntime.Request { return c.repoUsers(ops.kind, workspace, name) })),
		ctlruntime.Gone(func(_ context.Context, req ctlruntime.Request) {
			c.repos[ops.kind].Set(binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}, nil)
			c.retries.Reset(retryKey{kind: ops.kind, cluster: req.Cluster, name: req.Name})
			c.forget(retryKey{kind: ops.kind, cluster: req.Cluster, name: req.Name})
		})}
	if scope != nil {
		options = append(options, ctlruntime.InWorkspaces(scope))
	}
	if err := ctlruntime.For[T](m, name, r, options...); err != nil {
		return fmt.Errorf("infracollection: registering the %s controller: %w", ops.kind, err)
	}
	return nil
}

func (c *Controller) bindRepo(kind string, req ctlruntime.Request, repo commonv1.ObjectRef) {
	c.repos[kind].Set(binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}, &struct{}{},
		binding.Key(repo.Workspace, infractl.RepoKind, repo.Name))
}

func (c *Controller) repoUsers(kind, workspace, name string) []ctlruntime.Request {
	var out []ctlruntime.Request
	for _, t := range c.repos[kind].Users(binding.Key(workspace, infractl.RepoKind, name)) {
		out = append(out, ctlruntime.Request{Cluster: t.Cluster, Workspace: t.Workspace, Name: t.Name})
	}
	return out
}

type child struct {
	name string

	owners []metav1.OwnerReference

	record *v1alpha1.InfraRecordBinding

	uri string
}

type collectionOps[T ctlruntime.Object] struct {
	kind string

	child infra.Kind

	view func(T) (v1alpha1.InfraCollectionSpec, v1alpha1.InfraCollectionStatus)

	statusObject func(name string, status v1alpha1.InfraCollectionStatus) T

	children func(ctx context.Context, dyn *kube.Client) ([]child, error)

	newChild func(meta metav1.ObjectMeta, binding v1alpha1.InfraRecordBinding) ctlruntime.Object
}

func childOf(meta metav1.ObjectMeta, spec v1alpha1.InfraSpec, status v1alpha1.InfraStatus) child {
	out := child{name: meta.Name, owners: meta.OwnerReferences, record: spec.Record}
	if status.Record != nil {
		out.uri = status.Record.URI
	}
	return out
}

var infraOps = collectionOps[*v1alpha1.InfraCollection]{
	kind:  infractl.CollectionKind,
	child: infra.KindInfra,
	view: func(o *v1alpha1.InfraCollection) (v1alpha1.InfraCollectionSpec, v1alpha1.InfraCollectionStatus) {
		return o.Spec, o.Status
	},
	statusObject: func(name string, status v1alpha1.InfraCollectionStatus) *v1alpha1.InfraCollection {
		return &v1alpha1.InfraCollection{ObjectMeta: metav1.ObjectMeta{Name: name}, Status: status}
	},
	children: func(ctx context.Context, dyn *kube.Client) ([]child, error) {
		items, err := kube.List[v1alpha1.Infra](ctx, dyn.Resource(v1alpha1.InfraGVR), metav1.ListOptions{})
		var out []child
		for _, o := range items {
			out = append(out, childOf(o.ObjectMeta, o.Spec, o.Status))
		}
		return out, err
	},
	newChild: func(meta metav1.ObjectMeta, binding v1alpha1.InfraRecordBinding) ctlruntime.Object {
		return &v1alpha1.Infra{ObjectMeta: meta, Spec: v1alpha1.InfraSpec{Record: &binding}}
	},
}

var ephemeralOps = collectionOps[*v1alpha1.EphemeralInfraCollection]{
	kind:  infractl.EphemeralCollectionKind,
	child: infra.KindEphemeralInfra,
	view: func(o *v1alpha1.EphemeralInfraCollection) (v1alpha1.InfraCollectionSpec, v1alpha1.InfraCollectionStatus) {
		return o.Spec, o.Status
	},
	statusObject: func(name string, status v1alpha1.InfraCollectionStatus) *v1alpha1.EphemeralInfraCollection {
		return &v1alpha1.EphemeralInfraCollection{ObjectMeta: metav1.ObjectMeta{Name: name}, Status: status}
	},
	children: func(ctx context.Context, dyn *kube.Client) ([]child, error) {
		items, err := kube.List[v1alpha1.EphemeralInfra](ctx, dyn.Resource(v1alpha1.EphemeralInfraGVR), metav1.ListOptions{})
		var out []child
		for _, o := range items {
			out = append(out, childOf(o.ObjectMeta, o.Spec.InfraSpec, o.Status.InfraStatus))
		}
		return out, err
	},
	newChild: func(meta metav1.ObjectMeta, binding v1alpha1.InfraRecordBinding) ctlruntime.Object {
		return &v1alpha1.EphemeralInfra{ObjectMeta: meta, Spec: v1alpha1.EphemeralInfraSpec{InfraSpec: v1alpha1.InfraSpec{Record: &binding}}}
	},
}

type reconciler[T ctlruntime.Object] struct {
	c *Controller

	ops collectionOps[T]

	manager string
}

func ChildName(rkey string) string {
	if len(validation.IsDNS1123Subdomain(rkey)) == 0 {
		return rkey
	}
	sum := sha256.Sum256([]byte(rkey))
	return "r-" + hex.EncodeToString(sum[:6])
}

func (r *reconciler[T]) Reconcile(ctx context.Context, req ctlruntime.Request, obj T) (ctlruntime.Result, error) {
	if !obj.GetDeletionTimestamp().IsZero() {
		return ctlruntime.Result{}, nil
	}
	spec, previous := r.ops.view(obj)
	nsid := r.ops.child.NSID()
	gen := obj.GetGeneration()
	status := v1alpha1.InfraCollectionStatus{ObservedGeneration: gen, Collection: nsid, Did: previous.Did, Records: previous.Records,
		Children: previous.Children, Conditions: slices.Clone(previous.Conditions)}
	meta.RemoveStatusCondition(&status.Conditions, v1alpha1.InfraCollectionConditionListed)
	attrs := []any{"kind", r.ops.kind, "workspace", req.Workspace, "name", req.Name}
	key := retryKey{kind: r.ops.kind, cluster: req.Cluster, name: req.Name}
	fail := func(err error) (ctlruntime.Result, error) {
		status.Message = err.Error()
		reason, retry := "IndexReadFailed", err
		unresolvedReason, unresolved := infractl.Unresolved(err)
		delay := infractl.ReadRetry(r.c.log, r.c.retries, key, err, attrs...)
		switch {
		case errors.Is(err, infractl.ErrRepoNoDID):
			reason, retry = "RepoPending", nil
		case unresolved:
			reason, retry = unresolvedReason, nil
			infractl.NoteUnresolved(r.c.log, previous.Conditions, &status.Conditions, gen, unresolvedReason, err, attrs...)
		case delay > 0:
			reason, retry = v1alpha1.InfraReasonRecordReadRetrying, nil
			status.Message = v1alpha1.InfraMessageRecordReadRetrying
		}
		conds.Set(&status.Conditions, gen, v1alpha1.InfraCollectionConditionIndexed, false, reason, status.Message)
		return ctlruntime.Result{Backoff: unresolved, RequeueAfter: delay}, errors.Join(retry, req.ApplyStatus(ctx, r.ops.statusObject(req.Name, status)))
	}
	repo, err := infractl.RepoRef(req.Workspace, spec.RepoRef)
	if err != nil {
		return fail(err)
	}
	r.c.bindRepo(r.ops.kind, req, repo)
	self := commonv1.ObjectRef{Group: v1alpha1.GroupName, Kind: r.ops.kind, Name: req.Name, Workspace: req.Workspace}
	did, reader, err := infractl.Repo(ctx, r.c.bus, r.c.dynamic, self, repo, infra.KindInfra.NSID(), nsid)
	if err != nil {
		return fail(err)
	}
	infractl.NoteUnresolved(r.c.log, previous.Conditions, &status.Conditions, gen, "", nil, attrs...)
	owner := metav1.OwnerReference{APIVersion: v1alpha1.GroupVersion.String(), Kind: r.ops.kind, Name: req.Name, UID: obj.GetUID(), Controller: ptr.To(true)}
	labels := map[string]string{v1alpha1.InfraKindLabel: r.ops.kind, v1alpha1.InfraNameLabel: req.Name}
	watchName := infra.ChildName(r.ops.child, "coll", req.Name)
	if err := infractl.EnsureWatch(ctx, r.c.dynamic, r.manager, req.Workspace, watchName, labels, owner,
		repo, infra.KindInfra.NSID(), []string{infra.ReservedRkey}, spec.EventSourceRefs); err != nil {
		return fail(err)
	}
	dyn, err := r.c.dynamic(req.Workspace)
	if err != nil {
		return fail(err)
	}
	raw, indexCID, present, err := readIndex(ctx, dyn, reader, watchName, did)
	if err != nil {
		return fail(infractl.ReadFailed(err))
	}
	now := time.Now().UTC()
	var live []infra.IndexEntry
	invalid := ""
	if present {
		entries, err := infra.DecodeIndex(raw)
		if err != nil {
			invalid = err.Error()
		}
		for _, e := range entries {
			if e.Collection == nsid && !e.Expired(now) {
				live = append(live, e)
			}
		}
	}
	r.c.retries.Reset(key)
	kids, err := r.ops.children(ctx, dyn)
	if err != nil {
		return fail(err)
	}
	byURI := map[string]child{}
	byName := map[string]child{}
	for _, k := range kids {
		byName[k.name] = k
		if uri := boundURI(k, req.Workspace, repo, did, nsid); uri != "" {
			byURI[uri] = k
		}
	}
	var children []v1alpha1.InfraCollectionChild
	created := 0
	for _, e := range live {
		uri := "at://" + did + "/" + nsid + "/" + e.Rkey
		entry := v1alpha1.InfraCollectionChild{Rkey: e.Rkey, URI: uri}
		if k, ok := byURI[uri]; ok {
			if ownedBy(k, obj.GetUID()) {
				entry.Name = k.name
			} else {
				entry.BoundBy = k.name
			}
			children = append(children, entry)
			continue
		}
		if r.c.ended(key, uri) {
			entry.Message = "Ended: the record ended; it is kept and not adopted"
			children = append(children, entry)
			continue
		}
		name := ChildName(e.Rkey)
		if k, ok := byName[name]; ok {
			entry.Message = fmt.Sprintf("NameConflict: %s %s exists and binds no record of this collection", r.ops.child, k.name)
			children = append(children, entry)
			continue
		}
		if r.c.isAbsent(key, indexCID, uri) {
			entry.Message = "NotFound: the index names a record the repo does not hold"
			children = append(children, entry)
			continue
		}
		got, err := reader.GetRecord(ctx, did, nsid, e.Rkey)
		if errors.Is(err, atprotorepo.ErrNotFound) {
			r.c.noteAbsent(key, indexCID, uri)
			entry.Message = "NotFound: the index names a record the repo does not hold"
			children = append(children, entry)
			continue
		}
		if err != nil {
			return fail(infractl.ReadFailed(fmt.Errorf("infracollection: reading %s: %w", uri, err)))
		}
		entry.CID = got.CID
		decoded, derr := infra.Decode(r.ops.child, got.Value)
		if errors.Is(derr, infra.ErrReserved) || decoded.Over(r.ops.child, now) {
			r.c.noteEnded(key, uri)
			entry.Message = "Ended: the record ended; it is kept and not adopted"
			children = append(children, entry)
			continue
		}
		meta := metav1.ObjectMeta{Name: name, Labels: map[string]string{v1alpha1.InfraCollectionLabel: req.Name},
			OwnerReferences: []metav1.OwnerReference{owner}}
		binding := v1alpha1.InfraRecordBinding{RepoRef: spec.RepoRef, URI: uri, EventSourceRefs: spec.EventSourceRefs}
		if err := req.Apply(ctx, r.ops.newChild(meta, binding)); err != nil {
			return fail(fmt.Errorf("infracollection: creating %s %s for %s: %w", r.ops.child, name, uri, err))
		}
		r.c.log.Info("infra_collection_child_created", "kind", r.ops.kind, "workspace", req.Workspace, "name", req.Name, "child", name,
			"uri", uri)
		entry.Name = name
		created++
		children = append(children, entry)
	}
	status.Did, status.Records, status.Children = did, len(live), children
	reason := "Indexed"
	status.Message = fmt.Sprintf("%d live %s entr(ies) in %s, %d child object(s) created this pass", len(live), nsid, infra.IndexURI(did), created)
	switch {
	case invalid != "":
		reason, status.Message = "IndexInvalid", "the index record does not decode: "+invalid
	case !present:
		reason, status.Message = "IndexEmpty", "the repo holds no index at "+infra.IndexURI(did)
	case len(live) == 0:
		reason = "IndexEmpty"
	}
	conds.Set(&status.Conditions, gen, v1alpha1.InfraCollectionConditionIndexed, invalid == "", reason, status.Message)
	if err := req.ApplyStatus(ctx, r.ops.statusObject(req.Name, status)); err != nil {
		return ctlruntime.Result{}, err
	}
	return ctlruntime.Result{}, nil
}

func readIndex(ctx context.Context, dyn *kube.Client, reader atprotorepo.RecordReader, watchName, did string) (json.RawMessage, string, bool, error) {
	uri := infra.IndexURI(did)
	watch, err := kube.Get[v1alpha1.AtprotoRecordWatch](ctx, dyn.Resource(v1alpha1.AtprotoRecordWatchGVR), watchName)
	if err != nil && !apierrors.IsNotFound(err) {
		return nil, "", false, err
	}
	if err == nil {
		if i := slices.IndexFunc(watch.Status.Records, func(rec v1alpha1.WatchedRecord) bool { return rec.URI == uri }); i >= 0 {
			return watch.Status.Records[i].Value, watch.Status.Records[i].Cid, true, nil
		}
		if slices.Contains(watch.Status.Deleted, uri) {
			return nil, "", false, nil
		}
	}
	got, err := reader.GetRecord(ctx, did, infra.KindInfra.NSID(), infra.ReservedRkey)
	if errors.Is(err, atprotorepo.ErrNotFound) {
		return nil, "", false, nil
	}
	if err != nil {
		return nil, "", false, fmt.Errorf("infracollection: reading %s: %w", uri, err)
	}
	return got.Value, got.CID, true, nil
}

func (c *Controller) isAbsent(key retryKey, index, uri string) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	memo, ok := c.absent[key]
	return ok && index != "" && memo.index == index && memo.uris[uri]
}

func (c *Controller) noteAbsent(key retryKey, index, uri string) {
	if index == "" {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	memo, ok := c.absent[key]
	if !ok || memo.index != index {
		memo = absentAt{index: index, uris: map[string]bool{}}
	}
	memo.uris[uri] = true
	c.absent[key] = memo
}

func (c *Controller) ended(key retryKey, uri string) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.over[key][uri]
}

func (c *Controller) noteEnded(key retryKey, uri string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.over[key] == nil {
		c.over[key] = map[string]bool{}
	}
	c.over[key][uri] = true
}

func (c *Controller) forget(key retryKey) {
	c.mu.Lock()
	defer c.mu.Unlock()
	delete(c.over, key)
	delete(c.absent, key)
}

func ownedBy(k child, uid types.UID) bool {
	for _, o := range k.owners {
		if o.UID == uid {
			return true
		}
	}
	return false
}

func boundURI(k child, workspace string, repo commonv1.ObjectRef, did, nsid string) string {
	if k.uri != "" {
		return k.uri
	}
	if k.record == nil {
		return ""
	}
	if uri := strings.TrimSpace(k.record.URI); uri != "" {
		return uri
	}
	ref, err := infractl.RepoRef(workspace, k.record.RepoRef)
	if err != nil || ref.Name != repo.Name || ref.Workspace != repo.Workspace {
		return ""
	}
	return "at://" + did + "/" + nsid + "/" + k.name
}