johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package marketidentity

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"

	apierrors "k8s.io/apimachinery/pkg/api/errors"

	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/atproto/pdsclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bus"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	vaultctl "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/vault"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/binding"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/marketidentity/atproto"
)

const (
	Name = "marketidentity"

	Export = "marketidentities"

	PdsExport = "applications"

	RecordWatchKind = "AtprotoRecordWatch"

	VaultExport = "hostprocesses"

	VaultKind = "OpenBaoServer"

	VaultObjectKind = "Vault"

	eventBuffer = 256
)

type Options struct {
	Scope func(workspace string) bool

	Endpoints atproto.EndpointFunc

	Vault atproto.VaultFunc

	Resolver *pdsclient.Resolver

	HTTPClient *http.Client

	Bus *bus.Bus

	OAuth atproto.OAuthOptions

	PLCURL string

	RecordHTTP *http.Client

	Log *slog.Logger
}

type known struct {
	cluster string

	req ctlruntime.Request

	identity *v1alpha1.MarketIdentity
}

type Controller struct {
	accounts *atproto.Accounts

	roots *vaultctl.Roots

	log *slog.Logger

	identities *binding.Table[known]

	events chan ctlruntime.Request

	bus *bus.Bus
}

func New(opts Options) *Controller {
	log := opts.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	log = log.With("component", "marketidentity-controller")
	roots := vaultctl.NewRoots(opts.Bus, log)
	trust := func(ctx context.Context, workspace string, id *v1alpha1.MarketIdentity) (*tls.Config, error) {
		pool, err := roots.Pool(ctx, Self(workspace, id.Name), id.Spec.CARef, nil)
		if err != nil || pool == nil {
			return nil, err
		}
		return pool.PublicClientConfig(nil), nil
	}
	accounts := atproto.New(atproto.Config{Endpoints: opts.Endpoints, Vault: opts.Vault, Resolver: opts.Resolver, HTTPClient: opts.HTTPClient,
		TLS: trust, Sessions: sessions(opts.Bus), OAuth: opts.OAuth, PLCURL: opts.PLCURL, RecordHTTP: opts.RecordHTTP, Log: log})
	return &Controller{accounts: accounts, roots: roots, log: log, identities: binding.New[known](), events: make(chan ctlruntime.Request, eventBuffer), bus: opts.Bus}
}

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(opts.Bus)
	ctl.wakeOnVault(opts.Bus)
	options := []ctlruntime.Option{ctlruntime.Export(Export), ctlruntime.Events(ctl.events), ctlruntime.Gone(ctl.gone),
		ctlruntime.Watches(func(_ context.Context, src ctlruntime.Request, obj *v1alpha1.AtprotoRecordWatch) []ctlruntime.Request {
			return ctl.users(binding.Key(src.Workspace, RecordWatchKind, obj.Name))
		}),
		ctlruntime.WatchesIn(PdsExport, func(_ context.Context, src ctlruntime.Request, obj *v1alpha1.Pds) []ctlruntime.Request {
			return ctl.users(binding.Key(src.Workspace, atproto.ComponentKind, obj.Name))
		}),
		ctlruntime.WatchesIn(VaultExport, func(_ context.Context, src ctlruntime.Request, _ *v1alpha1.OpenBaoServer) []ctlruntime.Request {
			return ctl.users(binding.Key(src.Workspace, VaultKind))
		})}
	if opts.Scope != nil {
		options = append(options, ctlruntime.InWorkspaces(opts.Scope))
	}
	if err := ctlruntime.For[*v1alpha1.MarketIdentity](m, Name, ctl, options...); err != nil {
		return nil, fmt.Errorf("marketidentity: registering the MarketIdentity controller: %w", err)
	}
	return ctl, nil
}

func (c *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, object *v1alpha1.MarketIdentity) (ctlruntime.Result, error) {
	if !object.DeletionTimestamp.IsZero() {
		c.gone(ctx, req)
		return ctlruntime.Result{}, nil
	}
	prior, held := c.identities.Get(req.Workspace, req.Name)
	wasReady := held && prior.identity != nil && prior.identity.Status.Phase == v1alpha1.MarketIdentityPhaseReady
	c.remember(req, object)
	status := object.Status
	watch, err := c.associationWatch(ctx, req, object)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	atproto.Associate(object, watch, &status)
	result, reconcileErr := c.accounts.Reconcile(ctx, req.Workspace, object, status)
	refAfter, refErr := c.accounts.ApplyAssociationRef(ctx, object, &result.Status)
	if errors.Is(refErr, atproto.ErrAssociationRefStale) && refAfter > 0 {
		refErr, refAfter = nil, min(refAfter, ctlruntime.DefaultBackoffMax)
	}
	reconcileErr = errors.Join(reconcileErr, refErr)
	if refAfter > 0 && (result.RequeueAfter == 0 || refAfter < result.RequeueAfter) {
		result.RequeueAfter = refAfter
	}
	next := object.DeepCopy()
	next.Status = result.Status
	if err := req.ApplyStatus(ctx, next); err != nil {
		return ctlruntime.Result{}, errors.Join(reconcileErr, err)
	}
	c.remember(req, next)
	if !wasReady && next.Status.Phase == v1alpha1.MarketIdentityPhaseReady {
		c.log.Info("marketidentity_ready_notified", "workspace", req.Workspace, "identity", req.Name)
		c.bus.NotifyBound(bus.Key{Workspace: req.Workspace, Kind: GroupKind.Kind, Name: req.Name})
	}
	if reconcileErr != nil {
		return ctlruntime.Result{}, reconcileErr
	}
	return ctlruntime.Result{RequeueAfter: result.RequeueAfter}, nil
}

func (c *Controller) CompleteBrowserAuthorization(ctx context.Context, workspace, name string, r *http.Request) error {
	entry, ok := c.identities.Get(workspace, name)
	if !ok || entry.identity == nil {
		return fmt.Errorf("marketidentity: the MarketIdentity %s in %s is not reconciled", name, workspace)
	}
	err := c.accounts.CompleteBrowserAuthorization(ctx, workspace, entry.identity, r)
	c.enqueue(ctlruntime.Request{Cluster: entry.cluster, Workspace: workspace, Name: name})
	return err
}

func (c *Controller) SignOut(ctx context.Context, workspace, name string) error {
	entry, ok := c.identities.Get(workspace, name)
	if !ok || entry.identity == nil {
		return fmt.Errorf("marketidentity: the MarketIdentity %s in %s is not reconciled", name, workspace)
	}
	var live v1alpha1.MarketIdentity
	if err := entry.req.Get(ctx, "", name, &live); err != nil {
		return fmt.Errorf("marketidentity: reading the MarketIdentity %s in %s: %w", name, workspace, err)
	}
	if err := atproto.SignOutRefusal(live.Spec.Mode); err != nil {
		return err
	}
	next := max(live.Spec.SignOutGeneration, live.Status.SignedOutGeneration) + 1
	if err := entry.req.ApplyFieldsAs(ctx, v1alpha1.FieldManagerMarketIdentitySignOut, &live,
		map[string]any{"spec": map[string]any{"signOutGeneration": next}}); err != nil {
		return fmt.Errorf("marketidentity: requesting the sign out of %s in %s: %w", name, workspace, err)
	}
	c.log.Info("sign out requested", "workspace", workspace, "identity", name, "signOutGeneration", next)
	return nil
}

func (c *Controller) ComponentChanged(workspace, kind, name string) {
	for _, r := range c.users(binding.Key(workspace, kind, name)) {
		c.enqueue(r)
	}
}

func (c *Controller) associationWatch(ctx context.Context, req ctlruntime.Request, object *v1alpha1.MarketIdentity) (*v1alpha1.AtprotoRecordWatch, error) {
	name := strings.TrimSpace(object.Spec.AssociationWatch)
	if name == "" || object.Spec.Mode != v1alpha1.MarketIdentityModeLocalRepo || object.Spec.Association == v1alpha1.AssociationSkip {
		return nil, nil
	}
	var watch v1alpha1.AtprotoRecordWatch
	err := req.Get(ctx, "", name, &watch)
	if apierrors.IsNotFound(err) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("marketidentity: reading the association watch %s: %w", name, err)
	}
	return &watch, nil
}

func (c *Controller) remember(req ctlruntime.Request, object *v1alpha1.MarketIdentity) {
	identity := object.DeepCopy()
	refs := []string{binding.Key(req.Workspace, VaultKind)}
	if name := strings.TrimSpace(object.Spec.AssociationWatch); name != "" {
		refs = append(refs, binding.Key(req.Workspace, RecordWatchKind, name))
	}
	if object.Spec.Mode == v1alpha1.MarketIdentityModeLocalRepo {
		refs = append(refs, binding.Key(req.Workspace, atproto.ComponentKind, atproto.ComponentName))
	}
	c.identities.Set(binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}, &known{cluster: req.Cluster, req: req, identity: identity}, refs...)
}

func (c *Controller) gone(_ context.Context, req ctlruntime.Request) {
	c.identities.Set(binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}, nil)
	c.accounts.Forget(req.Workspace, req.Name)
	c.roots.Drop(Self(req.Workspace, req.Name))
}

func Self(workspace, name string) commonv1.ObjectRef {
	return commonv1.ObjectRef{Group: v1alpha1.GroupName, Kind: "MarketIdentity", Name: name, Workspace: workspace}
}

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

func (c *Controller) wakeOnVault(b *bus.Bus) {
	b.Subscribe(func(key bus.Key) {
		if key.Kind != VaultObjectKind {
			return
		}
		reqs := c.users(binding.Key(key.Workspace, VaultKind))
		if len(reqs) == 0 {
			return
		}
		go func() {
			for _, r := range reqs {
				c.events <- r
			}
		}()
	})
}

func (c *Controller) enqueue(r ctlruntime.Request) {
	select {
	case c.events <- r:
	default:
		c.log.Warn("the MarketIdentity event queue is full; the event is dropped", "workspace", r.Workspace, "name", r.Name)
	}
}