package atprotorepo import ( "context" "fmt" "log/slog" "maps" "slices" "strings" 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/runtime/schema" commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/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" ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/binding" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/conds" ) const ( Name = "atprotorepo" Export = "applications" Kind = "AtprotoRepo" LocalRepoKind = "LocalRepo" RemoteRepoKind = "RemoteRepo" ReasonBound = "Bound" ReasonBackendMissing = "BackendMissing" ReasonInvalidBackend = "InvalidBackend" ReasonBackendReady = "BackendReady" ReasonBackendNotReady = "BackendNotReady" ReasonBackendStale = "BackendStale" ) var GroupKind = schema.GroupKind{Group: v1alpha1.GroupName, Kind: Kind} type Options struct { Bus *bus.Bus Log *slog.Logger } type repoBinding struct { self commonv1.ObjectRef backend commonv1.ObjectRef labels map[string]string } type Controller struct { bus *bus.Bus log *slog.Logger bindings *binding.Table[repoBinding] } func New(opts Options) *Controller { log := opts.Log if log == nil { log = slog.New(slog.DiscardHandler) } return &Controller{bus: opts.Bus, log: log.With("component", "atprotorepo-controller"), bindings: binding.New[repoBinding]()} } func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) { if err := m.AddToScheme(v1alpha1.AddToScheme); err != nil { return nil, err } ctl := New(opts) ctl.Provide() ctl.relayBinds() if err := ctlruntime.For[*v1alpha1.AtprotoRepo](m, Name, ctl, ctlruntime.Export(Export), ctlruntime.Watches(func(_ context.Context, src ctlruntime.Request, obj *v1alpha1.LocalRepo) []ctlruntime.Request { return ctl.users(src.Workspace, LocalRepoKind, obj.Name) }), ctlruntime.Watches(func(_ context.Context, src ctlruntime.Request, obj *v1alpha1.RemoteRepo) []ctlruntime.Request { return ctl.users(src.Workspace, RemoteRepoKind, obj.Name) })); err != nil { return nil, fmt.Errorf("atprotorepo: registering the AtprotoRepo controller: %w", err) } return ctl, nil } func (c *Controller) Provide() { provide[atprotorepo.Repo](c, atprotorepo.CapabilityWrite) provide[atprotorepo.RecordWriter](c, atprotorepo.CapabilityWrite) provide[atprotorepo.RecordReader](c, atprotorepo.CapabilityRead) provide[atprotorepo.Subscriber](c, atprotorepo.CapabilitySubscribe) } func provide[T any](c *Controller, capability string) { bus.Provide(c.bus, GroupKind, capability, func(ctx context.Context, to commonv1.ObjectRef, scope []string) (T, error) { var zero T b, ok := c.bindings.Get(to.Workspace, to.Name) if !ok { return zero, fmt.Errorf("%w: %s", atprotorepo.ErrNotBound, to.String()) } return bus.Resolve[T](bus.WithCallerLabels(ctx, b.self, b.labels), c.bus, b.self, b.backend, bus.Capability{Name: capability, Scope: scope}) }) } func (c *Controller) relayBinds() { c.bus.Subscribe(func(key bus.Key) { if !strings.EqualFold(key.Kind, LocalRepoKind) && !strings.EqualFold(key.Kind, RemoteRepoKind) { return } for _, t := range c.bindings.Users(binding.Key(key.Workspace, key.Kind, key.Name)) { c.bus.NotifyBound(bus.Key{Workspace: t.Workspace, Kind: Kind, Name: t.Name}) } }) } func (c *Controller) users(workspace, kind, name string) []ctlruntime.Request { var out []ctlruntime.Request for _, t := range c.bindings.Users(binding.Key(workspace, kind, name)) { out = append(out, ctlruntime.Request{Cluster: t.Cluster, Workspace: t.Workspace, Name: t.Name}) } return out } func (c *Controller) bind(req ctlruntime.Request, b *repoBinding, backend string) { prior, held := c.bindings.Get(req.Workspace, req.Name) c.bindings.Set(binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}, b, backend) if b != nil && (!held || prior.self != b.self || prior.backend != b.backend || !maps.Equal(prior.labels, b.labels)) { c.bus.NotifyBound(bus.Key{Workspace: req.Workspace, Kind: Kind, Name: req.Name}) } } func contractKind(kind string) bool { for _, res := range v1alpha1.APIResources { if strings.EqualFold(res.Kind, strings.TrimSpace(kind)) && slices.Contains(res.Contracts, commonv1.ContractAtprotoRepos) { return true } } return false } func (c *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, object *v1alpha1.AtprotoRepo) (ctlruntime.Result, error) { if !object.DeletionTimestamp.IsZero() { c.bind(req, nil, "") return ctlruntime.Result{}, nil } generation := object.Generation status := v1alpha1.AtprotoRepoStatus{ObservedGeneration: generation, Conditions: slices.Clone(object.Status.Conditions)} ref := object.Spec.BackendRef workspace := ref.InWorkspace(req.Workspace) kind := strings.TrimSpace(ref.Kind) name := strings.TrimSpace(ref.Name) switch { case kind == "" || name == "": c.bind(req, nil, "") return ctlruntime.Result{}, c.unbound(ctx, req, object, status, ReasonInvalidBackend, "the backendRef needs a kind and a name") case !contractKind(kind): c.bind(req, nil, "") return ctlruntime.Result{}, c.unbound(ctx, req, object, status, ReasonInvalidBackend, fmt.Sprintf("the backendRef kind %s does not carry the %s contract", kind, commonv1.ContractLabelAtprotoRepos)) case workspace != req.Workspace: c.bind(req, nil, "") return ctlruntime.Result{}, c.unbound(ctx, req, object, status, ReasonInvalidBackend, fmt.Sprintf("the backend lives in %s; an AtprotoRepo binds a backend in its own workspace %s", workspace, req.Workspace)) } bk := binding.Key(workspace, kind, name) backend, backendGeneration, err := c.readBackend(ctx, req, kind, name) if apierrors.IsNotFound(err) { c.bind(req, nil, bk) return ctlruntime.Result{}, c.unbound(ctx, req, object, status, ReasonBackendMissing, fmt.Sprintf("the %s %s does not exist", kind, name)) } if err != nil { return ctlruntime.Result{}, err } self := commonv1.ObjectRef{Group: v1alpha1.GroupName, Kind: Kind, Name: object.Name, Workspace: req.Workspace} c.bind(req, &repoBinding{self: self, backend: commonv1.ObjectRef{Group: v1alpha1.GroupName, Kind: kind, Name: name, Workspace: workspace}, labels: maps.Clone(object.Labels)}, bk) status.DID = backend.DID status.Head = backend.Head status.Rev = backend.Rev status.Writable = backend.Writable status.Address = backend.Address setCondition(&status, v1alpha1.AtprotoRepoConditionBackendBound, true, ReasonBound, fmt.Sprintf("bound to %s %s", kind, name), generation) ready := meta.FindStatusCondition(backend.Conditions, commonv1.ConditionReady) switch { case backend.ObservedGeneration != backendGeneration: setCondition(&status, v1alpha1.AtprotoRepoConditionReady, false, ReasonBackendStale, fmt.Sprintf("the %s status is at generation %d, the spec at %d", kind, backend.ObservedGeneration, backendGeneration), generation) case ready == nil: setCondition(&status, v1alpha1.AtprotoRepoConditionReady, false, ReasonBackendNotReady, fmt.Sprintf("the %s reports no Ready condition", kind), generation) case ready.Status != metav1.ConditionTrue: setCondition(&status, v1alpha1.AtprotoRepoConditionReady, false, ReasonBackendNotReady, ready.Message, generation) default: setCondition(&status, v1alpha1.AtprotoRepoConditionReady, true, ReasonBackendReady, ready.Message, generation) } return ctlruntime.Result{}, c.apply(ctx, req, object, status) } func (c *Controller) readBackend(ctx context.Context, req ctlruntime.Request, kind, name string) (v1alpha1.AtprotoRepoBackendStatus, int64, error) { switch strings.ToLower(kind) { case strings.ToLower(LocalRepoKind): var obj v1alpha1.LocalRepo if err := req.Get(ctx, "", name, &obj); err != nil { return v1alpha1.AtprotoRepoBackendStatus{}, 0, err } return obj.Status, obj.Generation, nil case strings.ToLower(RemoteRepoKind): var obj v1alpha1.RemoteRepo if err := req.Get(ctx, "", name, &obj); err != nil { return v1alpha1.AtprotoRepoBackendStatus{}, 0, err } return obj.Status, obj.Generation, nil } return v1alpha1.AtprotoRepoBackendStatus{}, 0, fmt.Errorf("atprotorepo: no reader for the backend kind %s", kind) } func (c *Controller) unbound(ctx context.Context, req ctlruntime.Request, object *v1alpha1.AtprotoRepo, status v1alpha1.AtprotoRepoStatus, reason, message string) error { generation := object.Generation setCondition(&status, v1alpha1.AtprotoRepoConditionBackendBound, false, reason, message, generation) setCondition(&status, v1alpha1.AtprotoRepoConditionReady, false, reason, message, generation) return c.apply(ctx, req, object, status) } func (c *Controller) apply(ctx context.Context, req ctlruntime.Request, object *v1alpha1.AtprotoRepo, status v1alpha1.AtprotoRepoStatus) error { next := object.DeepCopy() next.Status = status return req.ApplyStatus(ctx, next) } func setCondition(status *v1alpha1.AtprotoRepoStatus, conditionType string, value bool, reason, message string, generation int64) { if message == "" { message = reason } conds.Set(&status.Conditions, generation, conditionType, value, reason, message) }