johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package remoterepo

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"maps"
	"slices"
	"strings"
	"sync"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"k8s.io/apimachinery/pkg/runtime/schema"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	pdsv1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/pds/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	reporemote "github.com/johnandersen777/socialweb-computer-kcp/internal/atprotorepo/remote"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atprotorepo/scoped"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bus"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/atprotorepo"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/pds"
	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 = "remoterepo"

	Export = "applications"

	PDSExport = "pds"

	IdentityExport = "marketidentities"

	Kind = "RemoteRepo"

	RemotePDSKind = "RemotePDS"

	ReasonReady = "Ready"

	ReasonInvalidSpec = "InvalidSpec"

	ReasonLinkDenied = "LinkDenied"

	ReasonPDSUnavailable = "PDSUnavailable"

	ReasonCredentialsUnavailable = "CredentialsUnavailable"

	ReasonObserveFailed = "ObserveFailed"
)

var GroupKind = schema.GroupKind{Group: v1alpha1.GroupName, Kind: Kind}

type Options struct {
	Bus *bus.Bus

	Network atprotorepo.RecordReader

	Log *slog.Logger
}

type backend struct {
	self commonv1.ObjectRef

	labels map[string]string

	pds commonv1.ObjectRef

	credentials commonv1.ObjectRef

	did string
}

type Controller struct {
	bus *bus.Bus

	network atprotorepo.RecordReader

	log *slog.Logger

	backends *binding.Table[backend]

	mu sync.Mutex

	ready map[binding.Target]bool
}

func New(opts Options) *Controller {
	log := opts.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	return &Controller{bus: opts.Bus, network: opts.Network, log: log.With("component", "remoterepo-controller"),
		backends: binding.New[backend](), ready: map[binding.Target]bool{}}
}

func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) {
	if err := m.AddToScheme(v1alpha1.AddToScheme, pdsv1alpha1.AddToScheme); err != nil {
		return nil, err
	}
	ctl := New(opts)
	ctl.Provide()
	if err := ctlruntime.For[*v1alpha1.RemoteRepo](m, Name, ctl, ctlruntime.Export(Export),
		ctlruntime.WatchesIn(PDSExport, func(_ context.Context, src ctlruntime.Request, obj *pdsv1alpha1.RemotePDS) []ctlruntime.Request {
			return ctl.users(binding.Key(src.Workspace, RemotePDSKind, obj.Name))
		}),
		ctlruntime.WatchesIn(IdentityExport, func(_ context.Context, src ctlruntime.Request, obj *v1alpha1.MarketIdentity) []ctlruntime.Request {
			return ctl.users(binding.Key(src.Workspace, "MarketIdentity", obj.Name))
		})); err != nil {
		return nil, fmt.Errorf("remoterepo: registering the RemoteRepo controller: %w", err)
	}
	return ctl, nil
}

func (c *Controller) users(ref string) []ctlruntime.Request {
	var out []ctlruntime.Request
	for _, t := range c.backends.Users(ref) {
		out = append(out, ctlruntime.Request{Cluster: t.Cluster, Workspace: t.Workspace, Name: t.Name})
	}
	return out
}

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.backends.Get(to.Workspace, to.Name)
		if !ok {
			return zero, fmt.Errorf("%w: the RemoteRepo %s is not reconciled", atprotorepo.ErrNotBound, to.String())
		}
		repo, err := c.open(ctx, b, scope)
		if err != nil {
			return zero, err
		}
		out, ok := repo.(T)
		if !ok {
			return zero, fmt.Errorf("remoterepo: the repo handle is not a %T", zero)
		}
		return out, nil
	})
}

func (b backend) caller(ctx context.Context) context.Context {
	return bus.WithCallerLabels(ctx, b.self, b.labels)
}

func (c *Controller) transport(ctx context.Context, b backend) (pds.Transport, error) {
	return bus.Resolve[pds.Transport](b.caller(ctx), c.bus, b.self, b.pds, bus.Capability{Name: pds.CapabilityAccount, Scope: []string{b.did}})
}

func (c *Controller) authenticator(ctx context.Context, b backend) (pds.Authenticator, error) {
	return bus.Resolve[pds.Authenticator](b.caller(ctx), c.bus, b.self, b.credentials,
		bus.Capability{Name: pds.CapabilitySession, Scope: []string{b.did}})
}

func (c *Controller) open(ctx context.Context, b backend, scope []string) (atprotorepo.Repo, error) {
	transport, err := c.transport(ctx, b)
	if err != nil {
		return nil, err
	}
	auth, err := c.authenticator(ctx, b)
	if err != nil {
		return nil, err
	}
	repo, err := reporemote.New(reporemote.Config{DID: b.did, Transport: transport, Auth: reporemote.NewAuth(auth)})
	if err != nil {
		return nil, err
	}
	return scoped.New(scoped.Route(repo, c.network), scope), nil
}

func (c *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, object *v1alpha1.RemoteRepo) (ctlruntime.Result, error) {
	target := binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}
	if !object.DeletionTimestamp.IsZero() {
		c.backends.Set(target, nil)
		c.settle(target, false)
		return ctlruntime.Result{}, nil
	}
	generation := object.Generation
	status := v1alpha1.AtprotoRepoBackendStatus{ObservedGeneration: generation, DID: strings.TrimSpace(object.Spec.DID),
		Conditions: slices.Clone(object.Status.Conditions)}
	pdsRef := object.Spec.PdsRef
	pdsRef.Group = pdsv1alpha1.GroupName
	pdsRef.Workspace = pdsRef.InWorkspace(req.Workspace)
	credentials := object.Spec.CredentialsRef.ObjectRef
	if credentials.Group == "" {
		credentials.Group = v1alpha1.GroupName
	}
	credentials.Workspace = credentials.InWorkspace(req.Workspace)
	_, didErr := syntax.ParseDID(status.DID)
	switch {
	case !strings.EqualFold(strings.TrimSpace(pdsRef.Kind), RemotePDSKind) || strings.TrimSpace(pdsRef.Name) == "":
		c.backends.Set(target, nil)
		return c.notReady(ctx, req, object, status, ReasonInvalidSpec, "the pdsRef must name a RemotePDS", nil)
	case strings.TrimSpace(credentials.Kind) == "" || strings.TrimSpace(credentials.Name) == "":
		c.backends.Set(target, nil)
		return c.notReady(ctx, req, object, status, ReasonInvalidSpec, "the credentialsRef must name a credentials provider", nil)
	case didErr != nil:
		c.backends.Set(target, nil)
		return c.notReady(ctx, req, object, status, ReasonInvalidSpec, fmt.Sprintf("the did %q is not a DID", status.DID), nil)
	}
	b := backend{self: commonv1.ObjectRef{Group: v1alpha1.GroupName, Kind: Kind, Name: object.Name, Workspace: req.Workspace},
		labels: maps.Clone(object.Labels), pds: pdsRef, credentials: credentials, did: status.DID}
	prior, held := c.backends.Get(req.Workspace, req.Name)
	c.backends.Set(target, &b, binding.Key(pdsRef.Workspace, RemotePDSKind, pdsRef.Name),
		binding.Key(credentials.Workspace, credentials.Kind, credentials.Name))
	if !held || prior.self != b.self || prior.pds != b.pds || prior.credentials != b.credentials || prior.did != b.did || !maps.Equal(prior.labels, b.labels) {
		c.bus.NotifyBound(bus.Key{Workspace: req.Workspace, Kind: Kind, Name: req.Name})
	}
	transport, err := c.transport(ctx, b)
	if err != nil {
		return c.resolveFailed(ctx, req, object, status, ReasonPDSUnavailable, err)
	}
	status.Address = transport.Address()
	syncer, err := bus.Resolve[pds.Sync](b.caller(ctx), c.bus, b.self, b.pds, bus.Capability{Name: pds.CapabilitySync})
	if err != nil {
		return c.resolveFailed(ctx, req, object, status, ReasonPDSUnavailable, err)
	}
	head, err := syncer.LatestCommit(ctx, status.DID)
	if err != nil {
		return c.notReady(ctx, req, object, status, ReasonObserveFailed, err.Error(), err)
	}
	status.Head, status.Rev = head.CID, head.Rev
	if _, err := c.authenticator(ctx, b); err != nil {
		return c.resolveFailed(ctx, req, object, status, ReasonCredentialsUnavailable, err)
	}
	status.Writable = true
	conds.Set(&status.Conditions, generation, commonv1.ConditionReady, true, ReasonReady,
		fmt.Sprintf("the RemotePDS %s serves %s with the credentials of %s %s", pdsRef.Name, status.DID, credentials.Kind, credentials.Name))
	if err := c.apply(ctx, req, object, status); err != nil {
		return ctlruntime.Result{}, err
	}
	c.settle(target, true)
	return ctlruntime.Result{}, nil
}

func (c *Controller) settle(target binding.Target, ready bool) {
	c.mu.Lock()
	was := c.ready[target]
	if ready {
		c.ready[target] = true
	} else {
		delete(c.ready, target)
	}
	c.mu.Unlock()
	if ready && !was {
		c.bus.NotifyBound(bus.Key{Workspace: target.Workspace, Kind: Kind, Name: target.Name})
	}
}

func (c *Controller) resolveFailed(ctx context.Context, req ctlruntime.Request, object *v1alpha1.RemoteRepo,
	status v1alpha1.AtprotoRepoBackendStatus, reason string, err error) (ctlruntime.Result, error) {
	switch {
	case errors.Is(err, bus.ErrLinkDenied):
		return c.notReady(ctx, req, object, status, ReasonLinkDenied, err.Error(), err)
	case errors.Is(err, pds.ErrNotFound), errors.Is(err, bus.ErrNoProvider):
		return c.notReady(ctx, req, object, status, reason, err.Error(), nil)
	}
	return c.notReady(ctx, req, object, status, reason, err.Error(), err)
}

func (c *Controller) notReady(ctx context.Context, req ctlruntime.Request, object *v1alpha1.RemoteRepo, status v1alpha1.AtprotoRepoBackendStatus,
	reason, message string, cause error) (ctlruntime.Result, error) {
	c.settle(binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}, false)
	status.Writable = false
	conds.Set(&status.Conditions, object.Generation, commonv1.ConditionReady, false, reason, message)
	if err := c.apply(ctx, req, object, status); err != nil {
		return ctlruntime.Result{}, errors.Join(cause, err)
	}
	return ctlruntime.Result{}, cause
}

func (c *Controller) apply(ctx context.Context, req ctlruntime.Request, object *v1alpha1.RemoteRepo, status v1alpha1.AtprotoRepoBackendStatus) error {
	next := object.DeepCopy()
	next.Status = status
	return req.ApplyStatus(ctx, next)
}