johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package embedded

import (
	"bytes"
	"cmp"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"maps"
	"slices"
	"time"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/rest"

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

const (
	ProviderIdentitiesVersion = 1

	providerWorkspacePhase = "workspace"
)

type ProviderIdentity struct {
	Export string `json:"export"`

	Namespace string `json:"namespace"`

	Name string `json:"name"`

	Type corev1.SecretType `json:"type,omitempty"`

	Data map[string][]byte `json:"data"`
}

type ProviderIdentities struct {
	Version int `json:"version"`

	Identities []ProviderIdentity `json:"identities"`
}

type apiExportIdentity struct {
	Metadata metav1.ObjectMeta `json:"metadata"`
	Spec     struct {
		Identity *struct {
			SecretRef *corev1.SecretReference `json:"secretRef,omitempty"`
		} `json:"identity,omitempty"`
	} `json:"spec"`
}

func ParseProviderIdentities(raw []byte) (ProviderIdentities, error) {
	if len(bytes.TrimSpace(raw)) == 0 {
		return ProviderIdentities{}, nil
	}
	var out ProviderIdentities
	if err := json.Unmarshal(raw, &out); err != nil {
		return ProviderIdentities{}, errors.New("embedded: the provider identities are not valid JSON")
	}
	if out.Version != ProviderIdentitiesVersion {
		return ProviderIdentities{}, fmt.Errorf("embedded: unsupported provider identities version %d (want %d)", out.Version, ProviderIdentitiesVersion)
	}
	for _, id := range out.Identities {
		if id.Namespace == "" || id.Name == "" || len(id.Data) == 0 {
			return ProviderIdentities{}, fmt.Errorf("embedded: the provider identity of %q names no secret or holds no key", id.Export)
		}
	}
	return out, nil
}

func (r *Requester) ProviderIdentities(ctx context.Context) ([]byte, error) {
	if r == nil || r.rest == nil {
		return nil, errors.New("embedded: the requester is not running")
	}
	kc, err := kube.ForWorkspace(r.rest, r.providerWorkspace)
	if err != nil {
		return nil, err
	}
	exports, err := kube.List[apiExportIdentity](ctx, kc.Resource(kcpgvr.APIExports), metav1.ListOptions{})
	if err != nil {
		return nil, fmt.Errorf("embedded: listing the provider APIExports: %w", err)
	}
	out := ProviderIdentities{Version: ProviderIdentitiesVersion}
	for _, export := range exports {
		if export.Spec.Identity == nil || export.Spec.Identity.SecretRef == nil {
			return nil, fmt.Errorf("embedded: the APIExport %s has no identity yet", export.Metadata.Name)
		}
		ref := export.Spec.Identity.SecretRef
		secret, err := kube.Get[corev1.Secret](ctx, kc.Resource(kcpgvr.Secrets).In(ref.Namespace), ref.Name)
		if err != nil {
			return nil, fmt.Errorf("embedded: reading the identity of APIExport %s: %w", export.Metadata.Name, err)
		}
		out.Identities = append(out.Identities, ProviderIdentity{Export: export.Metadata.Name, Namespace: ref.Namespace, Name: ref.Name, Type: secret.Type, Data: secret.Data})
	}
	slices.SortFunc(out.Identities, func(a, b ProviderIdentity) int {
		return cmp.Compare(a.Export, b.Export)
	})
	return json.Marshal(out)
}

func restoreProviderIdentities(ctx context.Context, restCfg *rest.Config, workspace string, identities ProviderIdentities, log *slog.Logger) error {
	if len(identities.Identities) == 0 {
		return nil
	}
	kc, err := kube.ForWorkspace(restCfg, workspace)
	if err != nil {
		return err
	}
	namespaces := map[string]bool{}
	created, kept := 0, 0
	for _, id := range identities.Identities {
		secrets := kc.Resource(kcpgvr.Secrets).In(id.Namespace)
		live, err := kube.Get[corev1.Secret](ctx, secrets, id.Name)
		switch {
		case err == nil:
			if !maps.EqualFunc(live.Data, id.Data, bytes.Equal) {
				return fmt.Errorf("embedded: the provider workspace holds a different identity for APIExport %s than the one restored; refusing to start", id.Export)
			}
			kept++
			continue
		case !apierrors.IsNotFound(err):
			return fmt.Errorf("embedded: reading the identity of APIExport %s: %w", id.Export, err)
		}
		if !namespaces[id.Namespace] {
			ns := &corev1.Namespace{TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "Namespace"}, ObjectMeta: metav1.ObjectMeta{Name: id.Namespace}}
			if _, err := kube.Create[corev1.Namespace](ctx, kc.Resource(kcpgvr.Namespaces), ns, metav1.CreateOptions{FieldManager: FieldManager}); err != nil && !apierrors.IsAlreadyExists(err) {
				return fmt.Errorf("embedded: creating the %s namespace for the provider identities: %w", id.Namespace, err)
			}
			namespaces[id.Namespace] = true
		}
		secret := &corev1.Secret{
			TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Secret"},
			ObjectMeta: metav1.ObjectMeta{Name: id.Name, Namespace: id.Namespace},
			Type:       id.Type,
			Data:       id.Data,
		}
		if _, err := kube.Create[corev1.Secret](ctx, secrets, secret, metav1.CreateOptions{FieldManager: FieldManager}); err != nil {
			return fmt.Errorf("embedded: restoring the identity of APIExport %s: %w", id.Export, err)
		}
		created++
	}
	log.Info("embedded: provider export identities restored", "created", created, "kept", kept)
	return nil
}

func splitAfterPhase(set manifests.Set, phase string) []manifests.Set {
	for i, p := range set.Phases {
		if p.Name != phase {
			continue
		}
		head, tail := set, set
		head.Phases = slices.Clone(set.Phases[:i+1])
		tail.Phases = slices.Clone(set.Phases[i+1:])
		return []manifests.Set{head, tail}
	}
	return []manifests.Set{set}
}

func endsWithPhase(set manifests.Set, phase string) bool {
	return len(set.Phases) > 0 && set.Phases[len(set.Phases)-1].Name == phase
}

const requesterSet = "market-requester"

var requesterBindingPhases = []string{"schemas", "bindings"}

func applyRequesterBindings(parent context.Context, cfg Config, log *slog.Logger) error {
	ctx, cancel := context.WithTimeout(parent, ApplyStallTimeout)
	defer cancel()
	applier, err := kcpapply.New(cfg.Rest, kcpapply.Options{Base: RootWorkspace, FieldManager: FieldManager, WaitTimeout: 3 * time.Minute, Log: log})
	if err != nil {
		return fmt.Errorf("embedded: manifest applier: %w", err)
	}
	ix, err := manifests.Load()
	if err != nil {
		return err
	}
	set, ok := ix.Set(requesterSet)
	if !ok {
		return fmt.Errorf("embedded: deploy/sets.yaml names no %s set", requesterSet)
	}
	var phases []manifests.Phase
	for _, p := range set.Phases {
		if slices.Contains(requesterBindingPhases, p.Name) {
			phases = append(phases, p)
		}
	}
	if len(phases) != len(requesterBindingPhases) {
		return fmt.Errorf("embedded: the %s set has no %v phases", requesterSet, requesterBindingPhases)
	}
	set.Phases = phases
	values, err := manifests.DecodeValues(set.Values, nil)
	if err != nil {
		return fmt.Errorf("embedded: decoding values for %s: %w", requesterSet, err)
	}
	if _, err := applier.Apply(ctx, set, values); err != nil {
		return fmt.Errorf("embedded: applying the %s schemas and bindings: %w", requesterSet, err)
	}
	return nil
}