johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package compose

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"reflect"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
)

const (
	ReasonChildEditReverted = "ChildEditReverted"

	eventNamespace = "default"
)

type Desired struct {
	Kind string

	Name string

	Spec any
}

func DesiredKeys(desired []Desired) []string {
	out := make([]string, 0, len(desired))
	for _, d := range desired {
		out = append(out, ComponentKey(d.Kind, d.Name))
	}
	return out
}

func ToMap(obj any) (map[string]any, error) {
	if v := reflect.ValueOf(obj); v.IsValid() && v.Kind() != reflect.Pointer {
		ptr := reflect.New(v.Type())
		ptr.Elem().Set(v)
		obj = ptr.Interface()
	}
	return runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
}

func SpecMap(spec any) (map[string]any, error) {
	if spec == nil {
		return nil, nil
	}
	if m, ok := spec.(map[string]any); ok {
		return m, nil
	}
	m, err := ToMap(spec)
	if err != nil {
		return nil, failf("the component spec: %w", err)
	}
	if len(m) == 0 {
		return nil, nil
	}
	return m, nil
}

type Generic struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec kube.Doc `json:"spec,omitempty"`

	Status kube.Doc `json:"status,omitempty"`
}

func DocMeta(d kube.Doc) map[string]any {
	m, _ := d["metadata"].(map[string]any)
	if m == nil {
		m = map[string]any{}
		d["metadata"] = m
	}
	return m
}

func docField(d kube.Doc, key string) string {
	s, _ := d[key].(string)
	return s
}

func docMetaField(d kube.Doc, key string) string {
	m, _ := d["metadata"].(map[string]any)
	s, _ := m[key].(string)
	return s
}

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

func (c *Composer) Object(r Root, ws, apiVersion, kind, name string, labels map[string]string) kube.Doc {
	d := NewDoc(apiVersion, kind, name)
	merged := r.Labels("")
	maps.Copy(merged, labels)
	meta := DocMeta(d)
	meta["labels"] = merged
	if ws == "" || ws == r.Workspace {
		meta["ownerReferences"] = []metav1.OwnerReference{r.OwnerRef()}
	}
	return d
}

func (c *Composer) ApplyComponent(ctx context.Context, r Root, ws string, d Desired) error {
	ct, err := componentTypeFor(d.Kind)
	if err != nil {
		return err
	}
	spec, err := SpecMap(d.Spec)
	if err != nil {
		return err
	}
	ws = workspaceOr(ws, r.Workspace)
	u := c.Object(r, ws, ct.APIVersion(), ct.Kind, d.Name, map[string]string{ComponentLabel: d.Kind})
	if spec != nil {
		u["spec"] = spec
	}
	current, err := c.applyChild(ctx, r, r.Manager(), ws, ct.GVR, u)
	if err != nil || current {
		return err
	}
	return c.revertForeignSpec(ctx, r, ws, ct.GVR, u)
}

func (c *Composer) EnsureComponents(ctx context.Context, r Root, desired []Desired) error {
	for _, d := range desired {
		if err := c.ApplyComponent(ctx, r, r.Workspace, d); err != nil {
			return fmt.Errorf("%s %s: %w", d.Kind, d.Name, err)
		}
	}
	return nil
}

func workspaceOr(ws, fallback string) string {
	if ws = strings.TrimSpace(ws); ws != "" {
		return ws
	}
	return fallback
}

func (c *Composer) Apply(ctx context.Context, r Root, ws string, gvr schema.GroupVersionResource, u kube.Doc) error {
	return c.applyAs(ctx, r, r.Manager(), ws, gvr, u)
}

func (c *Composer) applyAs(ctx context.Context, r Root, manager, ws string, gvr schema.GroupVersionResource, u kube.Doc) error {
	_, err := c.applyChild(ctx, r, manager, ws, gvr, u)
	return err
}

func (c *Composer) applyChild(ctx context.Context, r Root, manager, ws string, gvr schema.GroupVersionResource, u kube.Doc) (bool, error) {
	ws = workspaceOr(ws, r.Workspace)
	ri, err := c.Client(ws, gvr)
	if err != nil {
		return false, err
	}
	delete(u, "status")
	if m, ok := u["metadata"].(map[string]any); ok {
		delete(m, "creationTimestamp")
	}
	namespace := docMetaField(u, "namespace")
	rc := ri.In(namespace)
	kind, name := docField(u, "kind"), docMetaField(u, "name")
	key, digest := applyKey(ws, gvr, namespace, name, manager), applyDigest(u)
	if c.applies.current(key, digest) {
		return true, nil
	}
	got, err := kube.Apply[kube.Doc](ctx, rc, u, kube.ApplyOptions{FieldManager: manager})
	if err == nil {
		c.applies.record(key, kind, digest, docObject(*got))
		return false, nil
	}
	c.applies.drop(key)
	if !apierrors.IsConflict(err) {
		return false, fmt.Errorf("compose: applying %s %s in %s: %w", kind, name, ws, err)
	}
	c.recordConflict(ctx, r, ws, u, err)
	got, err = kube.Apply[kube.Doc](ctx, rc, u, kube.ApplyOptions{FieldManager: manager, Force: true})
	if err != nil {
		return false, fmt.Errorf("compose: applying %s %s in %s: %w", kind, name, ws, err)
	}
	c.applies.record(key, kind, digest, docObject(*got))
	return false, nil
}

func (c *Composer) recordConflict(ctx context.Context, r Root, ws string, u kube.Doc, cause error) {
	kind, name := docField(u, "kind"), docMetaField(u, "name")
	message := fmt.Sprintf("%s %s: a field another manager set is reverted to the spec %s %s composes: %s", kind, name, r.Kind, r.Name, cause.Error())
	c.log.Warn("a direct edit of a composed child is reverted", "root", r.Owner(), "kind", kind, "name", name, "workspace", ws, "err", cause.Error())
	c.Event(ctx, r, ws, docField(u, "apiVersion"), kind, name, "Warning", ReasonChildEditReverted, message)
}

func (c *Composer) Event(ctx context.Context, r Root, ws, apiVersion, kind, name, eventType, reason, message string) {
	ri, err := c.Client(ws, EventGVR)
	if err != nil {
		return
	}
	now := time.Now().UTC().Format(time.RFC3339)
	if len(message) > 1024 {
		message = message[:1024]
	}
	ev := kube.Doc{
		"apiVersion":         "v1",
		"kind":               "Event",
		"metadata":           map[string]any{"generateName": strings.ToLower(kind) + "-" + name + "-", "namespace": eventNamespace},
		"involvedObject":     map[string]any{"apiVersion": apiVersion, "kind": kind, "name": name},
		"reason":             reason,
		"message":            message,
		"type":               eventType,
		"source":             map[string]any{"component": r.ManagedBy()},
		"reportingComponent": r.ManagedBy(),
		"firstTimestamp":     now,
		"lastTimestamp":      now,
		"count":              int64(1),
	}
	if _, err := kube.Create[kube.Doc](ctx, ri.In(eventNamespace), ev, metav1.CreateOptions{FieldManager: r.Manager()}); err != nil && ctx.Err() == nil {
		c.log.Warn("recording an event", "root", r.Owner(), "reason", reason, "err", err.Error())
	}
}

func (c *Composer) ApplyTyped(ctx context.Context, r Root, ws string, gvr schema.GroupVersionResource, apiVersion, kind string, obj any) error {
	m, err := ToMap(obj)
	if err != nil {
		return err
	}
	u := kube.Doc(m)
	u["apiVersion"], u["kind"] = apiVersion, kind
	ws = workspaceOr(ws, r.Workspace)
	meta := DocMeta(u)
	labels := r.Labels(kind)
	if own, ok := meta["labels"].(map[string]any); ok {
		for k, v := range own {
			if s, ok := v.(string); ok {
				labels[k] = s
			}
		}
	}
	meta["labels"] = labels
	if ws == r.Workspace {
		meta["ownerReferences"] = []metav1.OwnerReference{r.OwnerRef()}
	}
	return c.Apply(ctx, r, ws, gvr, u)
}

func Get[T any](ctx context.Context, c *Composer, ws string, gvr schema.GroupVersionResource, name string) (*T, error) {
	ri, err := c.Client(ws, gvr)
	if err != nil {
		return nil, err
	}
	return kube.Get[T](ctx, ri, name)
}

func (c *Composer) GetComponent(ctx context.Context, ws, kind, name string) (ComponentView, bool, error) {
	ct, err := componentTypeFor(kind)
	if err != nil {
		return ComponentView{}, false, err
	}
	o, err := Get[Generic](ctx, c, ws, ct.GVR, name)
	if apierrors.IsNotFound(err) {
		return ComponentView{Kind: kind, Name: name, Workspace: ws}, false, nil
	}
	if err != nil {
		return ComponentView{}, false, err
	}
	v, err := ComponentViewOf(ws, kind, o)
	return v, err == nil, err
}

func (c *Composer) ListComponents(ctx context.Context, ws, owner string) ([]ComponentView, error) {
	views := make([][]ComponentView, len(ComponentTypes))
	errs := make([]error, len(ComponentTypes))
	var wg sync.WaitGroup
	for i, ct := range ComponentTypes {
		wg.Go(func() { views[i], errs[i] = c.listComponents(ctx, ws, owner, ct.Kind, ct.GVR) })
	}
	wg.Wait()
	if err := errors.Join(errs...); err != nil {
		return nil, err
	}
	out := slices.Concat(views...)
	sort.Slice(out, func(i, j int) bool {
		if out[i].Kind != out[j].Kind {
			return out[i].Kind < out[j].Kind
		}
		return out[i].Name < out[j].Name
	})
	return out, nil
}

func (c *Composer) DeleteComponent(ctx context.Context, ws, kind, name string) error {
	ct, err := componentTypeFor(kind)
	if err != nil {
		return err
	}
	ri, err := c.Client(ws, ct.GVR)
	if err != nil {
		return err
	}
	if err := ri.Delete(ctx, name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
		return err
	}
	return nil
}

func (c *Composer) DeleteComponentsExcept(ctx context.Context, ws, owner string, keep []string) (int, error) {
	views, err := c.ListComponents(ctx, ws, owner)
	if err != nil {
		return 0, err
	}
	left := 0
	for _, v := range views {
		if slices.Contains(keep, ComponentKey(v.Kind, v.Name)) {
			continue
		}
		left++
		if err := c.DeleteComponent(ctx, ws, v.Kind, v.Name); err != nil {
			return left, err
		}
		c.log.Info("deleted a component the root object omits", "owner", owner, "kind", v.Kind, "name", v.Name, "workspace", ws)
	}
	return left, nil
}

func (c *Composer) RestartComponent(ctx context.Context, r Root, ws, kind, name string) error {
	ct, err := componentTypeFor(kind)
	if err != nil {
		return err
	}
	u := NewDoc(ct.APIVersion(), ct.Kind, name)
	u["spec"] = map[string]any{"restartedAt": time.Now().UTC().Format(time.RFC3339Nano)}
	return c.applyAs(ctx, r, r.Manager()+"/restart", ws, ct.GVR, u)
}

func (c *Composer) Prune(ctx context.Context, ws string, gvr schema.GroupVersionResource, selector string, keep map[string]bool) error {
	ri, err := c.Client(ws, gvr)
	if err != nil {
		return err
	}
	items, err := kube.List[metav1.PartialObjectMetadata](ctx, ri, metav1.ListOptions{LabelSelector: selector})
	if apierrors.IsNotFound(err) || apierrors.IsForbidden(err) {
		return nil
	}
	if err != nil {
		return err
	}
	for i := range items {
		name := items[i].Name
		if keep[name] {
			continue
		}
		if err := ri.Delete(ctx, name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
			return err
		}
	}
	return nil
}

func (c *Composer) listComponents(ctx context.Context, ws, owner, kind string, gvr schema.GroupVersionResource) ([]ComponentView, error) {
	ri, err := c.Client(ws, gvr)
	if err != nil {
		return nil, err
	}
	items, err := kube.List[Generic](ctx, ri, metav1.ListOptions{LabelSelector: OwnerLabel + "=" + owner})
	if apierrors.IsNotFound(err) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	out := make([]ComponentView, 0, len(items))
	for i := range items {
		v, err := ComponentViewOf(ws, kind, &items[i])
		if err != nil {
			return nil, err
		}
		out = append(out, v)
	}
	return out, nil
}

func (c *Composer) DeleteOwned(ctx context.Context, ws string, gvr schema.GroupVersionResource, owner string) (int, error) {
	ri, err := c.Client(ws, gvr)
	if err != nil {
		return 0, err
	}
	items, err := kube.List[metav1.PartialObjectMetadata](ctx, ri, metav1.ListOptions{LabelSelector: OwnerLabel + "=" + owner})
	if apierrors.IsNotFound(err) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	for _, o := range items {
		if o.DeletionTimestamp == nil {
			if err := ri.Delete(ctx, o.Name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
				return len(items), err
			}
		}
	}
	return len(items), nil
}

func (c *Composer) DeleteAll(ctx context.Context, ws string, gvr schema.GroupVersionResource) (int, error) {
	ri, err := c.Client(ws, gvr)
	if err != nil {
		return 0, err
	}
	items, err := kube.List[metav1.PartialObjectMetadata](ctx, ri, metav1.ListOptions{})
	if apierrors.IsNotFound(err) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	for _, o := range items {
		if o.DeletionTimestamp == nil {
			if err := ri.Delete(ctx, o.Name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
				return len(items), err
			}
		}
	}
	return len(items), nil
}

func (c *Composer) DeleteDeployments(ctx context.Context, ws string) (int, error) {
	left := 0
	for _, gvr := range []schema.GroupVersionResource{v1alpha1.InfraGVR, v1alpha1.EphemeralInfraGVR} {
		n, err := c.DeleteAll(ctx, ws, gvr)
		if err != nil {
			return left + n, err
		}
		left += n
	}
	return left, nil
}

func (c *Composer) HeldContracts(ctx context.Context, ws string) (int, error) {
	ri, err := c.Client(ws, ComputeContractGVR)
	if err != nil {
		return 0, err
	}
	items, err := kube.List[v1alpha1.ComputeContract](ctx, ri, metav1.ListOptions{})
	if apierrors.IsNotFound(err) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	held := 0
	for i := range items {
		if items[i].HoldsGuest() {
			held++
		}
	}
	return held, nil
}

func (c *Composer) SetSpecField(ctx context.Context, r Root, ws string, gvr schema.GroupVersionResource, apiVersion, kind, name string, spec map[string]any) error {
	u := NewDoc(apiVersion, kind, name)
	u["spec"] = spec
	return c.Apply(ctx, r, ws, gvr, u)
}

func jsonCopy(in any, out any) error {
	b, err := json.Marshal(in)
	if err != nil {
		return err
	}
	return json.Unmarshal(b, out)
}