johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package guestcontrolplane

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"sync"

	"k8s.io/apimachinery/pkg/runtime"

	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/bus"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/guestcp"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/guestcp/queue"
)

func (c *Controller) Provide() {
	bus.Provide(c.bus, GroupKind, guestcp.CapabilityInputs, func(_ context.Context, to commonv1.ObjectRef, _ []string) (guestcp.Source, error) {
		workspace, name, err := target(to)
		if err != nil {
			return nil, err
		}
		return c.feedFor(workspace, name), nil
	})
	bus.Provide(c.bus, GroupKind, guestcp.CapabilityGuests, func(_ context.Context, to commonv1.ObjectRef, _ []string) (guestcp.Guests, error) {
		workspace, _, err := target(to)
		if err != nil {
			return nil, err
		}
		return guests{store: c.store, workspace: workspace}, nil
	})
	bus.Provide(c.bus, GroupKind, guestcp.CapabilityGrants, func(_ context.Context, to commonv1.ObjectRef, _ []string) (guestcp.Grants, error) {
		workspace, _, err := target(to)
		if err != nil {
			return nil, err
		}
		return grants{store: c.store, workspace: workspace}, nil
	})
	bus.Provide(c.bus, GroupKind, guestcp.CapabilityEvents, func(_ context.Context, to commonv1.ObjectRef, _ []string) (guestcp.Events, error) {
		workspace, name, err := target(to)
		if err != nil {
			return nil, err
		}
		return c.hubFor(workspace, name), nil
	})
	bus.Provide(c.bus, GroupKind, guestcp.CapabilityQueue, func(_ context.Context, to commonv1.ObjectRef, _ []string) (guestcp.Queue, error) {
		workspace, name, err := target(to)
		if err != nil {
			return nil, err
		}
		return c.hubFor(workspace, name), nil
	})
}

func (c *Controller) hubFor(workspace, name string) *queue.Hub {
	key := feedKey(workspace, name)
	c.mu.Lock()
	defer c.mu.Unlock()
	h := c.hubs[key]
	if h == nil {
		h = queue.New()
		c.hubs[key] = h
	}
	return h
}

func target(to commonv1.ObjectRef) (string, string, error) {
	workspace, name := strings.TrimSpace(to.Workspace), strings.TrimSpace(to.Name)
	if workspace == "" || name == "" {
		return "", "", fmt.Errorf("guestcontrolplane: the handle target %s names no workspace or name", to.String())
	}
	return workspace, name, nil
}

type feed struct {
	mu sync.Mutex

	inputs guestcp.Inputs

	held bool

	next int

	watchers map[int]func(guestcp.Inputs)
}

var _ guestcp.Source = (*feed)(nil)

func newFeed() *feed {
	return &feed{watchers: map[int]func(guestcp.Inputs){}}
}

func (f *feed) Inputs(context.Context) (guestcp.Inputs, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.inputs, nil
}

func (f *feed) Watch(fn func(guestcp.Inputs)) func() {
	if fn == nil {
		return func() {}
	}
	f.mu.Lock()
	id := f.next
	f.next++
	f.watchers[id] = fn
	in, held := f.inputs, f.held
	f.mu.Unlock()
	if held {
		fn(in)
	}
	return func() {
		f.mu.Lock()
		delete(f.watchers, id)
		f.mu.Unlock()
	}
}

func (f *feed) publish(in guestcp.Inputs) bool {
	f.mu.Lock()
	if f.held && reflect.DeepEqual(f.inputs, in) {
		f.mu.Unlock()
		return false
	}
	f.inputs, f.held = in, true
	watchers := make([]func(guestcp.Inputs), 0, len(f.watchers))
	for _, fn := range f.watchers {
		watchers = append(watchers, fn)
	}
	f.mu.Unlock()
	for _, fn := range watchers {
		fn(in)
	}
	return true
}

func (f *feed) close() {
	f.mu.Lock()
	f.watchers = map[int]func(guestcp.Inputs){}
	f.mu.Unlock()
}

type guests struct {
	store Store

	workspace string
}

var _ guestcp.Guests = guests{}

func (g guests) RecordGuest(ctx context.Context, name string, report v1alpha1.ComputeVMGuestReport) error {
	name = strings.TrimSpace(name)
	if name == "" {
		return errors.New("guestcontrolplane: the guest report names no ComputeVM")
	}
	guest, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&report)
	if err != nil {
		return err
	}
	doc := kube.Doc{
		"apiVersion": v1alpha1.GroupVersion.String(),
		"kind":       "ComputeVM",
		"metadata":   map[string]any{"name": name},
		"status":     map[string]any{"market": map[string]any{"guest": guest}},
	}
	return g.store.Apply(ctx, g.workspace, FieldManager, doc, true)
}

type grants struct {
	store Store

	workspace string
}

var _ guestcp.Grants = grants{}

func (g grants) Apply(ctx context.Context, grant v1alpha1.GuestIdentityGrant) error {
	if strings.TrimSpace(grant.Name) == "" {
		return errors.New("guestcontrolplane: applying a GuestIdentityGrant with no name")
	}
	object, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&grant)
	if err != nil {
		return err
	}
	metadata := map[string]any{"name": grant.Name}
	held, _ := object["metadata"].(map[string]any)
	if owners, _ := held["ownerReferences"].([]any); len(owners) > 0 {
		metadata["ownerReferences"] = owners
	}
	if labels, _ := held["labels"].(map[string]any); len(labels) > 0 {
		metadata["labels"] = labels
	}
	doc := kube.Doc{
		"apiVersion": v1alpha1.GroupVersion.String(),
		"kind":       "GuestIdentityGrant",
		"metadata":   metadata,
		"spec":       object["spec"],
	}
	return g.store.Apply(ctx, g.workspace, FieldManager, doc, false)
}

func (g grants) Delete(ctx context.Context, name string) error {
	name = strings.TrimSpace(name)
	if name == "" {
		return nil
	}
	grant := &v1alpha1.GuestIdentityGrant{}
	grant.Name = name
	return g.store.Delete(ctx, g.workspace, grant)
}