johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package atprotosession

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	exposurev1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/exposure/v1alpha1"
	pdsv1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/pds/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/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 = "atprotosession"

	Export = "applications"

	PDSExport = "pds"

	LinkExport = "exposures"

	Kind = "AtprotoSession"

	AccountKind = "PdsAccount"

	PdsKind = "Pds"

	RemotePDSKind = "RemotePDS"

	ConditionReady = "Ready"

	ReasonObserved = "Observed"

	ReasonWaiting = "Waiting"

	ReasonInvalidSpec = "InvalidSpec"

	ReasonSourceError = "SourceError"

	ReasonPdsNotReady = "PdsNotReady"

	DefaultRefresh = 30 * time.Second
)

type Options struct {
	Bus *bus.Bus

	Scope func(workspace string) bool

	Refresh time.Duration

	Log *slog.Logger
}

type Controller struct {
	bus *bus.Bus

	log *slog.Logger

	refresh time.Duration

	refs *binding.Table[struct{}]
}

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", "atprotosession-controller"), refresh: ctlruntime.Period(opts.Refresh, DefaultRefresh),
		refs: binding.New[struct{}]()}
}

func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) {
	if opts.Bus == nil {
		return nil, errors.New("atprotosession: a bus is required")
	}
	if err := m.AddToScheme(v1alpha1.AddToScheme, pdsv1alpha1.AddToScheme, exposurev1alpha1.AddToScheme); err != nil {
		return nil, err
	}
	ctl := New(opts)
	options := []ctlruntime.Option{ctlruntime.Export(Export), ctlruntime.Gone(ctl.gone),
		ctlruntime.Watches(func(_ context.Context, src ctlruntime.Request, obj *v1alpha1.PdsAccount) []ctlruntime.Request {
			return ctl.users(binding.Key(src.Workspace, AccountKind, obj.Name))
		}),
		ctlruntime.Watches(func(_ context.Context, src ctlruntime.Request, obj *v1alpha1.Pds) []ctlruntime.Request {
			return ctl.users(binding.Key(src.Workspace, PdsKind, obj.Name))
		}),
		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(LinkExport, func(_ context.Context, src ctlruntime.Request, obj *exposurev1alpha1.ComponentLink) []ctlruntime.Request {
			to := obj.Spec.To
			return ctl.users(binding.Key(to.InWorkspace(src.Workspace), to.Kind, to.Name))
		})}
	if opts.Scope != nil {
		options = append(options, ctlruntime.InWorkspaces(opts.Scope))
	}
	if err := ctlruntime.For[*v1alpha1.AtprotoSession](m, Name, ctl, options...); err != nil {
		return nil, fmt.Errorf("atprotosession: registering the AtprotoSession controller: %w", err)
	}
	return ctl, nil
}

func (c *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, object *v1alpha1.AtprotoSession) (ctlruntime.Result, error) {
	target := binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}
	if !object.DeletionTimestamp.IsZero() {
		c.refs.Set(target, nil)
		return ctlruntime.Result{}, nil
	}
	generation := object.Generation
	status := v1alpha1.AtprotoSessionStatus{ObservedGeneration: generation, Conditions: object.Status.Conditions}
	to := pdsRef(req.Workspace, object.Spec.PdsRef)
	refs := []string{binding.Key(req.Workspace, AccountKind, object.Spec.AccountRef.Name)}
	if strings.TrimSpace(to.Name) != "" {
		refs = append(refs, binding.Key(to.Workspace, to.Kind, to.Name))
	}
	c.refs.Set(target, &struct{}{}, refs...)
	didText := strings.TrimSpace(object.Spec.DID)
	if didText == "" || strings.TrimSpace(to.Name) == "" {
		conds.Set(&status.Conditions, generation, ConditionReady, false, ReasonWaiting, "the AtprotoSession needs a did and a pdsRef")
		return ctlruntime.Result{}, c.apply(ctx, req, object, status)
	}
	did, err := syntax.ParseDID(didText)
	if err != nil {
		conds.Set(&status.Conditions, generation, ConditionReady, false, ReasonInvalidSpec, "the AtprotoSession did is invalid: "+err.Error())
		return ctlruntime.Result{}, c.apply(ctx, req, object, status)
	}
	self := commonv1.ObjectRef{Group: v1alpha1.GroupName, Kind: Kind, Name: object.Name, Workspace: req.Workspace}
	sessions, err := bus.Resolve[pds.Sessions](ctx, c.bus, self, to, bus.Capability{Name: pds.CapabilitySession, Scope: []string{did.String()}})
	if errors.Is(err, pds.ErrNotFound) {
		c.log.Info("an AtprotoSession waits for its PDS", "workspace", req.Workspace, "name", object.Name, "err", err.Error())
		conds.Set(&status.Conditions, generation, ConditionReady, false, ReasonPdsNotReady, "the PDS is not ready: "+err.Error())
		return ctlruntime.Result{}, c.apply(ctx, req, object, status)
	}
	var session pds.Session
	if err == nil {
		session, err = c.session(ctx, sessions, did.String())
	}
	if err != nil {
		conds.Set(&status.Conditions, generation, ConditionReady, false, ReasonSourceError, "reading the session: "+err.Error())
		return ctlruntime.Result{}, errors.Join(fmt.Errorf("atprotosession: reading %s from %s: %w", did, to.String(), err), c.apply(ctx, req, object, status))
	}
	status.Active = session.Active
	conds.Set(&status.Conditions, generation, ConditionReady, true, ReasonObserved, "the session status is observed without its tokens")
	if err := c.apply(ctx, req, object, status); err != nil {
		return ctlruntime.Result{}, err
	}
	if !strings.EqualFold(strings.TrimSpace(to.Kind), RemotePDSKind) {
		return ctlruntime.Result{}, nil
	}
	return ctlruntime.RefreshAfter(c.refresh), nil
}

func (c *Controller) session(ctx context.Context, sessions pds.Sessions, did string) (pds.Session, error) {
	session, err := sessions.Session(ctx, did)
	if err != nil {
		return pds.Session{}, err
	}
	if session.DID != "" && session.DID != did {
		return pds.Session{}, fmt.Errorf("the session is for %s, not %s", session.DID, did)
	}
	return session, nil
}

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

func (c *Controller) gone(_ context.Context, req ctlruntime.Request) {
	c.refs.Set(binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}, nil)
}

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

func pdsRef(workspace string, ref commonv1.ObjectRef) commonv1.ObjectRef {
	out := ref
	out.Workspace = ref.InWorkspace(workspace)
	if out.Group == "" {
		out.Group = v1alpha1.GroupName
		if strings.EqualFold(strings.TrimSpace(ref.Kind), RemotePDSKind) {
			out.Group = pdsv1alpha1.GroupName
		}
	}
	return out
}