johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package clusterkcp

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strings"

	kcpv1alpha1 "github.com/kcp-dev/sdk/apis/tenancy/v1alpha1"
	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"
	"k8s.io/apimachinery/pkg/runtime/serializer"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/rest"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/clusterprovider"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/clusterhome"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/materialiser"
)

var (
	socialwebResource = v1alpha1.GroupVersion
	tenancyResource   = kcpv1alpha1.SchemeGroupVersion
)

var _ clusterprovider.Clusters = (*Registry)(nil)

type RegistryOptions struct {
	Host string

	StateDir string

	RestConfig *rest.Config

	Transport http.RoundTripper

	Account string
}

type Registry struct {
	cfg      *rest.Config
	http     *http.Client
	codecs   runtime.NegotiatedSerializer
	stateDir string
	account  string
}

func NewRegistry(opts RegistryOptions) (*Registry, error) {
	if opts.Host == "" {
		return nil, errors.New("clusterprovider: Host is required")
	}
	cfg := &rest.Config{}
	if opts.RestConfig != nil {
		cfg = rest.CopyConfig(opts.RestConfig)
	}
	cfg.Host = strings.TrimSuffix(opts.Host, "/")
	cfg.ContentType = "application/json"
	cfg.AcceptContentTypes = "application/json"
	if opts.Transport != nil {
		cfg.Transport = opts.Transport
	}
	httpClient, err := rest.HTTPClientFor(cfg)
	if err != nil {
		return nil, fmt.Errorf("clusterprovider: %w", err)
	}
	scheme, err := newScheme()
	if err != nil {
		return nil, err
	}
	return &Registry{
		cfg:      cfg,
		http:     httpClient,
		codecs:   serializer.NewCodecFactory(scheme).WithoutConversion(),
		stateDir: opts.StateDir,
		account:  strings.TrimSpace(opts.Account),
	}, nil
}

func (r *Registry) Refs(ctx context.Context) ([]clusterprovider.Ref, error) {
	accounts, _, err := r.workspaces(ctx, clusterhome.RootWorkspace)
	if err != nil {
		return nil, err
	}
	var out []clusterprovider.Ref
	for _, account := range accounts {
		path := clusterhome.RootWorkspace + ":" + account
		if r.account != "" && path != clusterhome.AccountPath(r.account) {
			continue
		}
		homes, deleting, err := r.workspaces(ctx, path)
		if apierrors.IsNotFound(err) {
			continue
		}
		if err != nil {
			return nil, err
		}
		for _, home := range homes {
			ref := clusterprovider.Ref{Account: path, Home: path + ":" + home}
			list, err := r.listClusters(ctx, ref.Home)
			if apierrors.IsNotFound(err) || (err != nil && (deleting[home] || r.goneOrDeleting(ctx, path, home))) {
				continue
			}
			if err != nil {
				return nil, err
			}
			for i := range list.Items {
				cr := list.Items[i]
				out = append(out, clusterprovider.Ref{
					Account: path,
					DID:     cr.Spec.Account,
					Home:    ref.Home,
					Name:    cr.Name,
				})
			}
		}
	}
	return out, nil
}

func (r *Registry) AccountClusters(ctx context.Context, account string) ([]v1alpha1.Cluster, error) {
	if r.account != "" && account != clusterhome.AccountPath(r.account) {
		return nil, nil
	}
	homes, deleting, err := r.workspaces(ctx, account)
	if apierrors.IsNotFound(err) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	var out []v1alpha1.Cluster
	for _, home := range homes {
		list, err := r.listClusters(ctx, account+":"+home)
		if apierrors.IsNotFound(err) || (err != nil && (deleting[home] || r.goneOrDeleting(ctx, account, home))) {
			continue
		}
		if err != nil {
			return nil, err
		}
		out = append(out, list.Items...)
	}
	return out, nil
}

func (r *Registry) listClusters(ctx context.Context, home string) (*v1alpha1.ClusterList, error) {
	c, err := r.clusters(ctx, home)
	if err != nil {
		return nil, err
	}
	var list v1alpha1.ClusterList
	if err := c.Get().Resource("clusters").Do(ctx).Into(&list); err != nil {
		return nil, fmt.Errorf("clusterprovider: list clusters in %s: %w", home, err)
	}
	for i := range list.Items {
		if list.Items[i].Spec.Account == "" {
			return nil, fmt.Errorf(
				"clusterprovider: Cluster %s in %s has no spec.account, so no mount path can be derived for it",
				list.Items[i].Name, home)
		}
	}
	return &list, nil
}

func (r *Registry) Read(ctx context.Context, ref clusterprovider.Ref) (*v1alpha1.Cluster, error) {
	c, err := r.clusters(ctx, ref.Home)
	if err != nil {
		return nil, err
	}
	var cl v1alpha1.Cluster
	if err := c.Get().Resource("clusters").Name(ref.Name).Do(ctx).Into(&cl); err != nil {
		return nil, fmt.Errorf("clusterprovider: read cluster %s in %s: %w", ref.Name, ref.Home, err)
	}
	return &cl, nil
}

func (r *Registry) Nodes(ctx context.Context, ref clusterprovider.Ref) ([]v1alpha1.Node, error) {
	c, err := r.clusters(ctx, ref.Home)
	if err != nil {
		return nil, err
	}
	var list v1alpha1.NodeList
	if err := c.Get().Resource("nodes").Do(ctx).Into(&list); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, nil
		}
		return nil, fmt.Errorf("clusterprovider: list nodes in %s: %w", ref.Home, err)
	}
	out := make([]v1alpha1.Node, 0, len(list.Items))
	for i := range list.Items {
		if list.Items[i].Spec.ClusterRef.Name == ref.Name {
			out = append(out, list.Items[i])
		}
	}
	return out, nil
}

func (r *Registry) WriteNode(ctx context.Context, ref clusterprovider.Ref, nodeName string, status v1alpha1.NodeStatus) error {
	c, err := r.clusters(ctx, ref.Home)
	if err != nil {
		return err
	}
	if err := kcpclient.WriteNodeTo(ctx, c, ref.Name, nodeName, status); err != nil {
		return fmt.Errorf("clusterprovider: node %s in %s: %w", nodeName, ref.Home, err)
	}
	return nil
}

func (r *Registry) WriteStatus(ctx context.Context, ref clusterprovider.Ref, st v1alpha1.ClusterStatus) error {
	c, err := r.clusters(ctx, ref.Home)
	if err != nil {
		return err
	}
	body, err := statusPatch(st)
	if err != nil {
		return err
	}
	if err := c.Patch(types.MergePatchType).SubResource("status").Resource("clusters").
		Name(ref.Name).Body(body).Do(ctx).Into(&v1alpha1.Cluster{}); err != nil {
		return fmt.Errorf("clusterprovider: write status for %s in %s: %w", ref.Name, ref.Home, err)
	}
	return nil
}

func (r *Registry) RemoveFinalizer(ctx context.Context, ref clusterprovider.Ref) error {
	c, err := r.clusters(ctx, ref.Home)
	if err != nil {
		return err
	}
	var cl v1alpha1.Cluster
	if err := c.Get().Resource("clusters").Name(ref.Name).Do(ctx).Into(&cl); err != nil {
		return fmt.Errorf("clusterprovider: reading %s in %s to release its finalizer: %w", ref.Name, ref.Home, err)
	}
	remaining := make([]string, 0, len(cl.Finalizers))
	for _, f := range cl.Finalizers {
		if f != v1alpha1.FinalizerCluster {
			remaining = append(remaining, f)
		}
	}
	if len(remaining) == len(cl.Finalizers) {
		return nil
	}
	body, err := json.Marshal([]map[string]any{
		{"op": "test", "path": "/metadata/finalizers", "value": cl.Finalizers},
		{"op": "add", "path": "/metadata/finalizers", "value": remaining},
	})
	if err != nil {
		return fmt.Errorf("clusterprovider: releasing the finalizer of %s: %w", ref.Name, err)
	}
	if err := c.Patch(types.JSONPatchType).Resource("clusters").Name(ref.Name).
		Body(body).Do(ctx).Into(&v1alpha1.Cluster{}); err != nil {
		return fmt.Errorf("clusterprovider: releasing the finalizer of %s in %s: %w", ref.Name, ref.Home, err)
	}
	return nil
}

func (r *Registry) TenantPath(ctx context.Context, ref clusterprovider.Ref) materialiser.PathProbe {
	if r.stateDir == "" {
		return materialiser.PathProbe{Err: "no cluster state dir to find the tenant front in"}
	}
	key, err := ref.Key()
	if err != nil {
		return materialiser.PathProbe{Err: err.Error()}
	}
	dir, err := clusterprovider.ClusterDataDir(r.stateDir, key)
	if err != nil {
		return materialiser.PathProbe{Err: err.Error()}
	}
	return clusterprovider.ProbeTenantFront(ctx, dir)
}

func (r *Registry) IdentityState(ctx context.Context, ref clusterprovider.Ref) (v1alpha1.IdentityState, error) {
	c, err := r.resource(ref.Account, socialwebResource)
	if err != nil {
		return "", err
	}
	var list v1alpha1.AtprotoIdentityList
	if err := c.Get().Resource("atprotoidentities").Do(ctx).Into(&list); err != nil {
		if apierrors.IsNotFound(err) {
			return "", nil
		}
		return "", fmt.Errorf("clusterprovider: list identities in %s: %w", ref.Account, err)
	}
	for i := range list.Items {
		if list.Items[i].Spec.Scope != v1alpha1.IdentityScopeAccount {
			continue
		}
		if list.Items[i].Spec.Revoked && list.Items[i].Status.State == "" {
			return v1alpha1.IdentityRevocationPending, nil
		}
		return list.Items[i].Status.State, nil
	}
	return "", nil
}

func (r *Registry) workspaces(ctx context.Context, logicalCluster string) ([]string, map[string]bool, error) {
	c, err := r.resource(logicalCluster, tenancyResource)
	if err != nil {
		return nil, nil, err
	}
	var list kcpv1alpha1.WorkspaceList
	if err := c.Get().Resource("workspaces").Do(ctx).Into(&list); err != nil {
		return nil, nil, fmt.Errorf("clusterprovider: list workspaces in %s: %w", logicalCluster, err)
	}
	var out []string
	deleting := map[string]bool{}
	for i := range list.Items {
		if list.Items[i].Spec.Mount == nil {
			out = append(out, list.Items[i].Name)
			deleting[list.Items[i].Name] = list.Items[i].DeletionTimestamp != nil
		}
	}
	return out, deleting, nil
}

func (r *Registry) goneOrDeleting(ctx context.Context, parent, name string) bool {
	c, err := r.resource(parent, tenancyResource)
	if err != nil {
		return false
	}
	var ws kcpv1alpha1.Workspace
	err = c.Get().Resource("workspaces").Name(name).Do(ctx).Into(&ws)
	return apierrors.IsNotFound(err) || (err == nil && ws.DeletionTimestamp != nil)
}

func (r *Registry) clusters(ctx context.Context, logicalCluster string) (rest.Interface, error) {
	return r.resource(logicalCluster, socialwebResource)
}

func (r *Registry) resource(logicalCluster string, gv schema.GroupVersion) (rest.Interface, error) {
	cfg := rest.CopyConfig(r.cfg)
	cfg.GroupVersion = &gv
	cfg.APIPath = "/clusters/" + logicalCluster + "/apis"
	cfg.NegotiatedSerializer = r.codecs
	c, err := rest.RESTClientForConfigAndClient(cfg, r.http)
	if err != nil {
		return nil, fmt.Errorf("clusterprovider: %s: %w", logicalCluster, err)
	}
	return c, nil
}

func newScheme() (*runtime.Scheme, error) {
	s := runtime.NewScheme()
	if err := v1alpha1.AddToScheme(s); err != nil {
		return nil, fmt.Errorf("clusterprovider: %w", err)
	}
	if err := kcpv1alpha1.AddToScheme(s); err != nil {
		return nil, fmt.Errorf("clusterprovider: %w", err)
	}
	metav1.AddToGroupVersion(s, schema.GroupVersion{Version: "v1"})
	return s, nil
}

type clusterStatusPatch struct {
	Phase      v1alpha1.ClusterPhase `json:"phase,omitempty"`
	URL        string                `json:"URL,omitempty"`
	Instance   string                `json:"instance,omitempty"`
	NodeExpiry *int64                `json:"nodeExpiry"`
	Conditions []metav1.Condition    `json:"conditions,omitempty"`
}

func statusPatch(st v1alpha1.ClusterStatus) ([]byte, error) {
	body, err := json.Marshal(map[string]any{"status": clusterStatusPatch{
		Phase:      st.Phase,
		URL:        st.URL,
		Instance:   st.Instance,
		NodeExpiry: st.NodeExpiry,
		Conditions: st.Conditions,
	}})
	if err != nil {
		return nil, fmt.Errorf("clusterprovider: status patch: %w", err)
	}
	return body, nil
}