johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package marketpresence

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

	"github.com/bluesky-social/indigo/atproto/syntax"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	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/market"
	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"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt"
	presence "github.com/johnandersen777/socialweb-computer-kcp/internal/marketpresence"
)

const (
	Name = "marketpresence"

	Export = "applications"

	BidderKind = "Bidder"

	ConditionReady = "Ready"

	ConditionBidderReady = "BidderReady"

	ConditionHeartbeat = "Heartbeat"

	DefaultRefresh = time.Minute
)

var errOtherWorkspace = errors.New("the bidderRef names another workspace and no cross-workspace Bidder reader is configured")

type Heartbeats interface {
	Observe(ctx context.Context, q presence.Query) presence.Observation
}

type Bidders interface {
	Get(ctx context.Context, workspace, name string) (*v1alpha1.Bidder, error)
}

var _ Heartbeats = (*presence.Heartbeats)(nil)

type Options struct {
	Heartbeats Heartbeats

	Bidders Bidders

	Scope func(workspace string) bool

	Refresh time.Duration

	Now func() time.Time

	Log *slog.Logger
}

type Controller struct {
	heartbeats Heartbeats

	bidders Bidders

	refresh time.Duration

	now func() time.Time

	log *slog.Logger

	refs *binding.Table[struct{}]
}

func New(opts Options) (*Controller, error) {
	if opts.Heartbeats == nil {
		return nil, errors.New("marketpresence: a heartbeat reader is required")
	}
	log := opts.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	now := opts.Now
	if now == nil {
		now = time.Now
	}
	return &Controller{heartbeats: opts.Heartbeats, bidders: opts.Bidders, refresh: ctlruntime.Period(opts.Refresh, DefaultRefresh), now: now,
		log: log.With("component", "marketpresence-controller"), refs: binding.New[struct{}]()}, nil
}

func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) {
	if err := m.AddToScheme(v1alpha1.AddToScheme); err != nil {
		return nil, err
	}
	ctl, err := New(opts)
	if err != nil {
		return nil, err
	}
	options := []ctlruntime.Option{ctlruntime.Export(Export), ctlruntime.Watches(ctl.presencesOf), ctlruntime.Gone(ctl.gone)}
	if opts.Scope != nil {
		options = append(options, ctlruntime.InWorkspaces(opts.Scope))
	}
	if err := ctlruntime.For[*v1alpha1.MarketPresence](m, Name, ctl, options...); err != nil {
		return nil, fmt.Errorf("marketpresence: registering the MarketPresence controller: %w", err)
	}
	return ctl, nil
}

func (c *Controller) presencesOf(_ context.Context, src ctlruntime.Request, obj *v1alpha1.Bidder) []ctlruntime.Request {
	derived := presence.PresenceName(obj.Name)
	out := []ctlruntime.Request{{Cluster: src.Cluster, Workspace: src.Workspace, Name: derived}}
	for _, t := range c.refs.Users(binding.Key(src.Workspace, BidderKind, obj.Name)) {
		if t.Cluster == src.Cluster && t.Name == derived {
			continue
		}
		out = append(out, ctlruntime.Request{Cluster: t.Cluster, Workspace: t.Workspace, Name: t.Name})
	}
	return out
}

func (c *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, object *v1alpha1.MarketPresence) (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
	}
	ref := object.Spec.BidderRef
	workspace := cmp.Or(strings.TrimSpace(ref.Workspace), req.Workspace)
	kind := cmp.Or(strings.TrimSpace(ref.Kind), BidderKind)
	bidderName := strings.TrimSpace(ref.Name)
	c.refs.Set(target, &struct{}{}, binding.Key(workspace, kind, bidderName))

	gen := object.Generation
	status := object.Status
	status.Conditions = append([]metav1.Condition(nil), object.Status.Conditions...)
	status.ObservedGeneration = gen

	if !strings.EqualFold(kind, BidderKind) || bidderName == "" {
		message := "the bidderRef must name a Bidder"
		conds.Set(&status.Conditions, gen, ConditionBidderReady, false, "BidderRefInvalid", message)
		expire(&status, gen, "BidderRefInvalid", message)
		return ctlruntime.Result{}, c.apply(ctx, req, object, status)
	}
	bidder, err := c.bidder(ctx, req, workspace, bidderName)
	switch {
	case apierrors.IsNotFound(err):
		if workspace == req.Workspace && ownedBy(object, bidderName) {
			c.log.Info("a MarketPresence goes with its gone Bidder", "workspace", req.Workspace, "marketpresence", req.Name)
			return ctlruntime.Result{}, req.Delete(ctx, object)
		}
		message := "the Bidder named by the MarketPresence does not exist"
		conds.Set(&status.Conditions, gen, ConditionBidderReady, false, "BidderMissing", message)
		expire(&status, gen, "BidderMissing", message)
		return ctlruntime.Result{}, c.apply(ctx, req, object, status)
	case errors.Is(err, errOtherWorkspace):
		conds.Set(&status.Conditions, gen, ConditionBidderReady, false, "BidderUnreadable", err.Error())
		expire(&status, gen, "BidderUnreadable", err.Error())
		return ctlruntime.Result{}, c.apply(ctx, req, object, status)
	case err != nil:
		message := "the Bidder named by the MarketPresence could not be read"
		conds.Set(&status.Conditions, gen, ConditionBidderReady, false, "BidderUnreadable", message)
		expire(&status, gen, "BidderUnreadable", message)
		return ctlruntime.Result{}, errors.Join(fmt.Errorf("marketpresence: reading the Bidder %s/%s: %w", workspace, bidderName, err),
			c.apply(ctx, req, object, status))
	}

	bidderReady, phase := readiness(bidder)
	if bidderReady {
		conds.Set(&status.Conditions, gen, ConditionBidderReady, true, "BidderReady", "the Bidder reports Ready")
	} else {
		message := "the Bidder is not Ready"
		if phase != "" && phase != string(v1alpha1.ComponentPhaseReady) {
			message = "the Bidder is " + phase
		}
		conds.Set(&status.Conditions, gen, ConditionBidderReady, false, "BidderNotReady", message)
	}

	offering := strings.TrimSpace(object.Spec.OfferingURI)
	did := bidderDID(bidder)
	if offering != "" {
		aturi, err := syntax.ParseATURI(offering)
		if err != nil || aturi.Collection() == "" || aturi.RecordKey() == "" {
			expire(&status, gen, "OfferingInvalid", "spec.offeringUri is not an atproto record URI")
			return ctlruntime.Result{}, c.apply(ctx, req, object, status)
		}
		if did == "" {
			did = string(aturi.Authority())
		}
	}

	now := c.now().UTC()
	ttl := opt.Positive(time.Duration(object.Spec.TtlSeconds)*time.Second, market.HeartbeatTTL)
	plc := ""
	if bidder.Spec.Network != nil {
		plc = bidder.Spec.Network.PLCDirectoryURL
	}
	observed := c.heartbeats.Observe(ctx, presence.Query{DID: did, OfferingURI: offering, PLCDirectoryURL: plc, Now: now})
	last := status.LastHeartbeatAt
	if !observed.At.IsZero() {
		if prior, err := time.Parse(time.RFC3339, last); err != nil || observed.At.After(prior) {
			last = observed.At.UTC().Format(time.RFC3339)
		}
	}
	status.LastHeartbeatAt = last

	fresh := false
	if at, err := time.Parse(time.RFC3339, last); err == nil {
		age := now.Sub(at)
		fresh = age >= -market.HeartbeatMaxSkew && age <= ttl
	}

	switch {
	case !observed.At.IsZero():
		conds.Set(&status.Conditions, gen, ConditionHeartbeat, true, "HeartbeatFresh", "the market heartbeat is readable")
	case fresh:
		conds.Set(&status.Conditions, gen, ConditionHeartbeat, true, "HeartbeatCached",
			"the market heartbeat is unreadable; the last observed heartbeat is within the TTL")
	default:
		conds.Set(&status.Conditions, gen, ConditionHeartbeat, false, observed.Reason, "no fresh market heartbeat is readable")
	}

	status.Phase = v1alpha1.MarketPresencePhaseExpired
	status.Ready = false
	switch {
	case bidderReady && fresh:
		status.Phase = v1alpha1.MarketPresencePhaseActive
		status.Ready = true
		conds.Set(&status.Conditions, gen, ConditionReady, true, "Active", "the Bidder is Ready and its heartbeat is within the TTL")
	case !fresh && last != "":
		conds.Set(&status.Conditions, gen, ConditionReady, false, "HeartbeatExpired", "the last market heartbeat is older than ttlSeconds")
	case !fresh:
		conds.Set(&status.Conditions, gen, ConditionReady, false, observed.Reason, "no fresh market heartbeat is readable")
	default:
		conds.Set(&status.Conditions, gen, ConditionReady, false, "BidderNotReady", "the Bidder is not Ready")
	}

	if err := c.apply(ctx, req, object, status); err != nil {
		return ctlruntime.Result{}, err
	}
	if did == "" {
		return ctlruntime.Result{}, nil
	}
	return ctlruntime.RefreshAfter(c.refresh), nil
}

func (c *Controller) gone(ctx context.Context, req ctlruntime.Request) {
	c.refs.Set(binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}, nil)
	if !strings.HasPrefix(req.Name, "bidder-") {
		return
	}
	bidder, err := c.derivedBidder(ctx, req)
	if err != nil {
		c.log.Warn("finding the Bidder of a missing MarketPresence", "workspace", req.Workspace, "marketpresence", req.Name, "err", err.Error())
		return
	}
	if bidder == nil {
		return
	}
	referenced, err := c.referenced(ctx, req, bidder.Name)
	if err != nil {
		c.log.Warn("listing the MarketPresences of a Bidder", "workspace", req.Workspace, "bidder", bidder.Name, "err", err.Error())
		return
	}
	if referenced {
		return
	}
	owner := metav1.OwnerReference{APIVersion: v1alpha1.GroupVersion.String(), Kind: BidderKind, Name: bidder.Name, UID: bidder.UID}
	next := &v1alpha1.MarketPresence{
		ObjectMeta: metav1.ObjectMeta{Name: req.Name, OwnerReferences: []metav1.OwnerReference{owner}},
		Spec: v1alpha1.MarketPresenceSpec{
			BidderRef:  commonv1.ObjectRef{Workspace: req.Workspace, Kind: BidderKind, Name: bidder.Name},
			TtlSeconds: int(market.HeartbeatTTL.Seconds()),
		},
	}
	if err := req.Apply(ctx, next); err != nil {
		c.log.Warn("applying the MarketPresence of a Bidder", "workspace", req.Workspace, "bidder", bidder.Name, "err", err.Error())
		return
	}
	c.log.Info("a MarketPresence follows its Bidder", "workspace", req.Workspace, "bidder", bidder.Name, "marketpresence", req.Name)
}

func (c *Controller) derivedBidder(ctx context.Context, req ctlruntime.Request) (*v1alpha1.Bidder, error) {
	var list v1alpha1.BidderList
	if err := req.List(ctx, &list); err != nil {
		return nil, err
	}
	for i := range list.Items {
		item := &list.Items[i]
		if presence.PresenceName(item.Name) == req.Name && item.DeletionTimestamp.IsZero() {
			return item, nil
		}
	}
	return nil, nil
}

func (c *Controller) referenced(ctx context.Context, req ctlruntime.Request, bidderName string) (bool, error) {
	if len(c.refs.Users(binding.Key(req.Workspace, BidderKind, bidderName))) > 0 {
		return true, nil
	}
	var list v1alpha1.MarketPresenceList
	if err := req.List(ctx, &list); err != nil {
		return false, err
	}
	for i := range list.Items {
		ref := list.Items[i].Spec.BidderRef
		if cmp.Or(strings.TrimSpace(ref.Workspace), req.Workspace) == req.Workspace &&
			strings.EqualFold(cmp.Or(strings.TrimSpace(ref.Kind), BidderKind), BidderKind) && strings.TrimSpace(ref.Name) == bidderName {
			return true, nil
		}
	}
	return false, nil
}

func (c *Controller) bidder(ctx context.Context, req ctlruntime.Request, workspace, name string) (*v1alpha1.Bidder, error) {
	if workspace != req.Workspace {
		if c.bidders == nil {
			return nil, errOtherWorkspace
		}
		return c.bidders.Get(ctx, workspace, name)
	}
	var bidder v1alpha1.Bidder
	if err := req.Get(ctx, "", name, &bidder); err != nil {
		return nil, err
	}
	return &bidder, nil
}

func (c *Controller) apply(ctx context.Context, req ctlruntime.Request, object *v1alpha1.MarketPresence, status v1alpha1.MarketPresenceStatus) error {
	if reflect.DeepEqual(object.Status, status) {
		return nil
	}
	next := object.DeepCopy()
	next.Status = status
	if err := req.ApplyStatus(ctx, next); err != nil {
		return err
	}
	if object.Status.Phase != status.Phase || object.Status.Ready != status.Ready {
		c.log.Info("market presence", "workspace", req.Workspace, "marketpresence", req.Name, "phase", string(status.Phase),
			"ready", status.Ready, "lastHeartbeatAt", status.LastHeartbeatAt)
	}
	return nil
}

func expire(status *v1alpha1.MarketPresenceStatus, gen int64, reason, message string) {
	status.Phase = v1alpha1.MarketPresencePhaseExpired
	status.Ready = false
	conds.Set(&status.Conditions, gen, ConditionReady, false, reason, message)
	conds.Set(&status.Conditions, gen, ConditionHeartbeat, false, reason, message)
}

func ownedBy(object *v1alpha1.MarketPresence, bidder string) bool {
	for _, owner := range object.OwnerReferences {
		if owner.Kind == BidderKind && owner.Name == bidder && strings.HasPrefix(owner.APIVersion, v1alpha1.GroupName+"/") {
			return true
		}
	}
	return false
}

func readiness(bidder *v1alpha1.Bidder) (bool, string) {
	phase := string(bidder.Status.Phase)
	if phase != "" && phase != string(v1alpha1.ComponentPhaseReady) {
		return false, phase
	}
	return bidder.Status.Ready, phase
}

func bidderDID(bidder *v1alpha1.Bidder) string {
	if bidder.Spec.Identity == nil {
		return ""
	}
	return strings.TrimSpace(bidder.Spec.Identity.Did)
}