johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package marketintents

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	kruntime "k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	atmarket "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market"
	market "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/market"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
)

const (
	FieldManagerBidSpec = ctlruntime.FieldManagerPrefix + "bidder-spec"

	FieldManagerComputeVMContract = ctlruntime.FieldManagerPrefix + "bidder-contract"

	FieldManagerComputeVMRelease = ctlruntime.FieldManagerPrefix + "bidder-release"

	FieldManagerStateSyncStatus = ctlruntime.FieldManagerPrefix + "statesync"

	FieldManagerStateSyncFinalizer = ctlruntime.FieldManagerPrefix + "statesync/finalizer"

	ownerCacheTTL = time.Minute
)

type handle struct {
	c *Controller

	owner commonv1.ObjectRef

	manager string
}

var _ market.Intents = (*handle)(nil)

func (c *Controller) open(_ context.Context, to commonv1.ObjectRef, _ []string) (market.Intents, error) {
	if strings.TrimSpace(to.Workspace) == "" || strings.TrimSpace(to.Name) == "" || strings.TrimSpace(to.Kind) == "" {
		return nil, fmt.Errorf("marketintents: %s names no component", to.String())
	}
	return &handle{c: c, owner: to, manager: ctlruntime.FieldManagerPrefix + strings.ToLower(strings.TrimSpace(to.Kind))}, nil
}

func (h *handle) Subscribe(ctx context.Context, sub market.Subscription) (func(), error) {
	return h.c.subscribe(ctx, h.owner.Workspace, sub)
}

func (h *handle) client() (client.Client, error) {
	return h.c.clients(h.owner.Workspace)
}

func (h *handle) ownerRef(ctx context.Context) (metav1.OwnerReference, error) {
	cacheKey := h.owner.Workspace + "/" + h.owner.Kind + "/" + h.owner.Name
	h.c.mu.Lock()
	cached, ok := h.c.owners[cacheKey]
	h.c.mu.Unlock()
	ref := metav1.OwnerReference{APIVersion: v1alpha1.GroupVersion.String(), Kind: h.owner.Kind, Name: h.owner.Name}
	if ok && time.Since(cached.at) < ownerCacheTTL {
		ref.UID = types.UID(cached.uid)
		return ref, nil
	}
	cl, err := h.client()
	if err != nil {
		return ref, err
	}
	live, err := h.meta(ctx, cl, h.owner.Kind, h.owner.Name)
	if err != nil {
		return ref, fmt.Errorf("marketintents: reading the owner %s: %w", h.owner.String(), err)
	}
	ref.UID = live.UID
	h.c.mu.Lock()
	h.c.owners[cacheKey] = ownerUID{uid: string(live.UID), at: time.Now()}
	h.c.mu.Unlock()
	return ref, nil
}

func encode(kind string, value any) (map[string]any, error) {
	raw, err := json.Marshal(value)
	if err != nil {
		return nil, fmt.Errorf("marketintents: encoding the %s: %w", kind, err)
	}
	dec := json.NewDecoder(bytes.NewReader(raw))
	dec.UseNumber()
	out := map[string]any{}
	if err := dec.Decode(&out); err != nil {
		return nil, fmt.Errorf("marketintents: encoding the %s: %w", kind, err)
	}
	return out, nil
}

func kindOf(kind string) schema.GroupVersionKind {
	return v1alpha1.GroupVersion.WithKind(kind)
}

func skeleton(kind, name string) kube.Doc {
	return kube.Doc{"apiVersion": v1alpha1.GroupVersion.String(), "kind": kind, "metadata": map[string]any{"name": name}}
}

func metadataOf(doc kube.Doc) map[string]any {
	meta, _ := doc["metadata"].(map[string]any)
	return meta
}

func (h *handle) meta(ctx context.Context, cl client.Client, kind, name string) (*metav1.PartialObjectMetadata, error) {
	doc, err := kube.GetWith(ctx, cl, kindOf(kind), types.NamespacedName{Name: name})
	if err != nil {
		return nil, err
	}
	return kube.Decode[metav1.PartialObjectMetadata](doc)
}

func (h *handle) exists(ctx context.Context, kind, name string) (bool, error) {
	cl, err := h.client()
	if err != nil {
		return false, err
	}
	_, err = kube.GetWith(ctx, cl, kindOf(kind), types.NamespacedName{Name: name})
	switch {
	case err == nil:
		return true, nil
	case apierrors.IsNotFound(err):
		return false, nil
	default:
		return false, err
	}
}

func (h *handle) apply(ctx context.Context, kind, name string, doc kube.Doc, manager string) error {
	cl, err := h.client()
	if err != nil {
		return err
	}
	if err := kube.ApplyWith(ctx, cl, doc, kube.ApplyOptions{FieldManager: manager, Force: true}); err != nil {
		return fmt.Errorf("marketintents: applying %s %s: %w", kind, name, err)
	}
	return nil
}

func (h *handle) applyStatus(ctx context.Context, kind, name string, status map[string]any, manager string) error {
	cl, err := h.client()
	if err != nil {
		return err
	}
	doc := skeleton(kind, name)
	doc["status"] = status
	if err := kube.ApplyWith(ctx, cl, doc, kube.ApplyOptions{FieldManager: manager, Force: true, Status: true}); err != nil {
		return fmt.Errorf("marketintents: applying the status of %s %s: %w", kind, name, err)
	}
	return nil
}

func (h *handle) create(ctx context.Context, kind string, meta metav1.ObjectMeta, spec any, manager string) error {
	name := strings.TrimSpace(meta.Name)
	if name == "" {
		return fmt.Errorf("marketintents: the %s names no object", kind)
	}
	found, err := h.exists(ctx, kind, name)
	if err != nil {
		return err
	}
	if found {
		return market.ErrExists
	}
	owners := meta.OwnerReferences
	if len(owners) == 0 {
		owner, err := h.ownerRef(ctx)
		if err != nil {
			return err
		}
		owners = []metav1.OwnerReference{owner}
	}
	doc := skeleton(kind, name)
	held := metadataOf(doc)
	if len(meta.Labels) > 0 {
		held["labels"] = stringMap(meta.Labels)
	}
	if len(meta.Annotations) > 0 {
		held["annotations"] = stringMap(meta.Annotations)
	}
	refs := make([]any, 0, len(owners))
	for i := range owners {
		ref, err := kruntime.DefaultUnstructuredConverter.ToUnstructured(&owners[i])
		if err != nil {
			continue
		}
		refs = append(refs, ref)
	}
	held["ownerReferences"] = refs
	value, err := encode(kind+" spec", spec)
	if err != nil {
		return err
	}
	doc["spec"] = value
	return h.apply(ctx, kind, name, doc, manager)
}

func (h *handle) applySpecField(ctx context.Context, kind, name, field string, value any, manager string) error {
	found, err := h.exists(ctx, kind, name)
	if err != nil {
		return err
	}
	if !found {
		return fmt.Errorf("%w: %s %s", market.ErrNotFound, kind, name)
	}
	encoded, err := encode(kind+" "+field, map[string]any{field: value})
	if err != nil {
		return err
	}
	doc := skeleton(kind, name)
	doc["spec"] = encoded
	return h.apply(ctx, kind, name, doc, manager)
}

func (h *handle) remove(ctx context.Context, kind, name string) error {
	cl, err := h.client()
	if err != nil {
		return err
	}
	if err := kube.DeleteWith(ctx, cl, kindOf(kind), types.NamespacedName{Name: name}, client.PropagationPolicy(metav1.DeletePropagationBackground)); err != nil && !apierrors.IsNotFound(err) {
		return fmt.Errorf("marketintents: deleting %s %s: %w", kind, name, err)
	}
	return nil
}

func (h *handle) CreateBid(ctx context.Context, bid v1alpha1.Bid) error {
	return h.create(ctx, market.KindBid, bid.ObjectMeta, bid.Spec, v1alpha1.FieldManagerBid)
}

func (h *handle) ApplyBidSpec(ctx context.Context, name string, spec v1alpha1.BidSpec) error {
	found, err := h.exists(ctx, market.KindBid, name)
	if err != nil {
		return err
	}
	if !found {
		return fmt.Errorf("%w: Bid %s", market.ErrNotFound, name)
	}
	value, err := encode("Bid spec", spec)
	if err != nil {
		return err
	}
	doc := skeleton(market.KindBid, name)
	doc["spec"] = value
	return h.apply(ctx, market.KindBid, name, doc, FieldManagerBidSpec)
}

func (h *handle) ApplyBidStatus(ctx context.Context, name string, status v1alpha1.BidStatus) error {
	value, err := encode("Bid status", status)
	if err != nil {
		return err
	}
	return h.applyStatus(ctx, market.KindBid, name, value, v1alpha1.FieldManagerBid)
}

func (h *handle) CreateComputeVM(ctx context.Context, vm v1alpha1.ComputeVM) error {
	return h.create(ctx, market.KindComputeVM, vm.ObjectMeta, vm.Spec, v1alpha1.FieldManagerComputeVMCreator)
}

func (h *handle) DeleteComputeVM(ctx context.Context, name string) error {
	return h.remove(ctx, market.KindComputeVM, name)
}

func (h *handle) ApplyComputeVMContract(ctx context.Context, name string, contract v1alpha1.ComputeVMContract) error {
	return h.applySpecField(ctx, market.KindComputeVM, name, "contract", contract, FieldManagerComputeVMContract)
}

func (h *handle) ApplyComputeVMRelease(ctx context.Context, name string, release v1alpha1.ComputeVMRelease) error {
	return h.applySpecField(ctx, market.KindComputeVM, name, "release", release, FieldManagerComputeVMRelease)
}

func (h *handle) ApplyComputeVMPublished(ctx context.Context, name string, published v1alpha1.ComputeVMPublished) error {
	value, err := encode("ComputeVM published", map[string]any{"market": map[string]any{"published": published}})
	if err != nil {
		return err
	}
	return h.applyStatus(ctx, market.KindComputeVM, name, value, v1alpha1.FieldManagerComputeVMMarketPublished)
}

func (h *handle) CreateGuestIdentityGrant(ctx context.Context, grant v1alpha1.GuestIdentityGrant) error {
	return h.create(ctx, market.KindGuestIdentityGrant, grant.ObjectMeta, grant.Spec, h.manager)
}

func (h *handle) CreateStateSync(ctx context.Context, sync v1alpha1.StateSync) error {
	return h.create(ctx, market.KindStateSync, sync.ObjectMeta, sync.Spec, h.manager)
}

func (h *handle) DeleteStateSync(ctx context.Context, name string) error {
	return h.remove(ctx, market.KindStateSync, name)
}

func (h *handle) ApplyStateSyncStatus(ctx context.Context, name string, status v1alpha1.StateSyncStatus) error {
	value, err := encode("StateSync status", status)
	if err != nil {
		return err
	}
	err = h.applyStatus(ctx, market.KindStateSync, name, value, FieldManagerStateSyncStatus)
	if apierrors.IsNotFound(errors.Unwrap(err)) {
		return nil
	}
	return err
}

func (h *handle) SetStateSyncFinalizer(ctx context.Context, name, finalizer string, present bool) error {
	cl, err := h.client()
	if err != nil {
		return err
	}
	live, err := h.meta(ctx, cl, market.KindStateSync, name)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil
		}
		return err
	}
	has := false
	for _, f := range live.Finalizers {
		if f == finalizer {
			has = true
		}
	}
	if has == present {
		return nil
	}
	doc := skeleton(market.KindStateSync, name)
	if present {
		metadataOf(doc)["finalizers"] = []any{finalizer}
	}
	return h.apply(ctx, market.KindStateSync, name, doc, FieldManagerStateSyncFinalizer)
}

func (h *handle) CreateComputeContract(ctx context.Context, contract v1alpha1.ComputeContract) error {
	return h.create(ctx, market.KindComputeContract, contract.ObjectMeta, contract.Spec, h.manager)
}

func (h *handle) ReleaseComputeContract(ctx context.Context, name string) error {
	cl, err := h.client()
	if err != nil {
		return err
	}
	live, err := kube.GetWith(ctx, cl, kindOf(market.KindComputeContract), types.NamespacedName{Name: name})
	if apierrors.IsNotFound(err) {
		return fmt.Errorf("%w: %s %s", market.ErrNotFound, market.KindComputeContract, name)
	}
	if err != nil {
		return fmt.Errorf("marketintents: reading %s %s: %w", market.KindComputeContract, name, err)
	}
	spec, _ := kruntime.DeepCopyJSONValue(live["spec"]).(map[string]any)
	if spec == nil {
		spec = map[string]any{}
	}
	spec["release"] = true
	spec["releaseRequestedAt"] = atmarket.FormatTime(time.Now())
	doc := skeleton(market.KindComputeContract, name)
	composed := metadataOf(doc)
	liveMeta, _ := live["metadata"].(map[string]any)
	for _, field := range []string{"labels", "ownerReferences"} {
		if value, ok := liveMeta[field]; ok {
			composed[field] = kruntime.DeepCopyJSONValue(value)
		}
	}
	doc["spec"] = spec
	return h.apply(ctx, market.KindComputeContract, name, doc, h.manager)
}

func (h *handle) DeleteComputeContract(ctx context.Context, name string) error {
	return h.remove(ctx, market.KindComputeContract, name)
}

func (h *handle) MergeComputeContractStatus(ctx context.Context, name string, status v1alpha1.ComputeContractStatus) error {
	value, err := encode("ComputeContract status", status)
	if err != nil {
		return err
	}
	return h.mergeStatus(ctx, market.KindComputeContract, name, value)
}

func (h *handle) SetComputeContractPendingEvents(ctx context.Context, name string, events []v1alpha1.ComputeContractEvent) error {
	if events == nil {
		events = []v1alpha1.ComputeContractEvent{}
	}
	value, err := encode("ComputeContract pending events", map[string]any{"pendingEvents": events})
	if err != nil {
		return err
	}
	return h.mergeStatus(ctx, market.KindComputeContract, name, value)
}

func (h *handle) mergeStatus(ctx context.Context, kind, name string, patch map[string]any) error {
	unlock := h.c.lockStatus(h.owner.Workspace + "/" + kind + "/" + name)
	defer unlock()
	cl, err := h.client()
	if err != nil {
		return err
	}
	live, err := kube.GetWith(ctx, cl, kindOf(kind), types.NamespacedName{Name: name})
	if err != nil {
		return fmt.Errorf("marketintents: reading %s %s: %w", kind, name, err)
	}
	current, _ := live["status"].(map[string]any)
	merged := mergePatch(kruntime.DeepCopyJSON(current), patch)
	return h.applyStatus(ctx, kind, name, merged, h.manager)
}

func stringMap(in map[string]string) map[string]any {
	out := make(map[string]any, len(in))
	for k, v := range in {
		out[k] = v
	}
	return out
}

func mergePatch(target, patch map[string]any) map[string]any {
	if target == nil {
		target = map[string]any{}
	}
	for k, v := range patch {
		if v == nil {
			delete(target, k)
			continue
		}
		next, isMap := v.(map[string]any)
		if !isMap {
			target[k] = v
			continue
		}
		existing, _ := target[k].(map[string]any)
		target[k] = mergePatch(existing, next)
	}
	return target
}