johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package gatewaykcp

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

	tenancyv1alpha1 "github.com/kcp-dev/sdk/apis/tenancy/v1alpha1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"

	"github.com/johnandersen777/socialweb-computer-kcp/api/kcpgvr"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose"
	appgateway "github.com/johnandersen777/socialweb-computer-kcp/internal/app/gateway"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/clusterhome"
	"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"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/sweepkcp"
)

const (
	DefaultBootstrapDID   = "did:web:example.com"
	bootstrapWorkspaceTTL = 2 * time.Minute
	bootstrapRetryMax     = 5 * time.Second
)

var (
	workspaceGVR = kcpgvr.Workspaces
	identityGVR  = v1alpha1.AtprotoIdentityGVR
	clusterGVR   = v1alpha1.ClusterGVR
)

func ssa(ctx context.Context, r kube.Resource, obj kube.Doc) error {
	_, err := kube.Apply[kube.Doc](ctx, r, obj, kube.ApplyOptions{FieldManager: FieldManager, Force: true})
	return err
}

func workspace(name, typ string) kube.Doc {
	return kube.Doc{"apiVersion": "tenancy.kcp.io/v1alpha1", "kind": "Workspace", "metadata": map[string]any{"name": name},
		"spec": map[string]any{"type": map[string]any{"name": typ, "path": clusterhome.RootWorkspace}}}
}

func applyWorkspace(ctx context.Context, ap *kcpapply.Applier, parent, name, typ string) error {
	body, err := json.Marshal(workspace(name, typ))
	if err != nil {
		return err
	}
	objs, err := manifests.Decode("gatewaykcp bootstrap workspace "+parent+":"+name, body)
	if err != nil {
		return err
	}
	_, err = ap.ApplyObjects(ctx, parent, objs)
	return err
}

func retry(ctx context.Context, what string, f func() error) error {
	deadline := time.Now().Add(bootstrapWorkspaceTTL)
	delay := Poll
	for {
		err := f()
		if err == nil {
			return nil
		}
		left := time.Until(deadline)
		if left <= 0 {
			return fmt.Errorf("gatewaykcp: bootstrap cluster: %s: %w", what, err)
		}
		again := time.NewTimer(min(delay, left))
		select {
		case <-ctx.Done():
			again.Stop()
			return ctx.Err()
		case <-again.C:
		}
		delay = min(2*delay, bootstrapRetryMax)
	}
}

func waitReady(ctx context.Context, r kube.Resource, name string) error {
	var last error
	_, err := kube.Await(ctx, r, name, bootstrapWorkspaceTTL, func(ws *tenancyv1alpha1.Workspace) (bool, error) {
		if ws == nil {
			last = apierrors.NewNotFound(workspaceGVR.GroupResource(), name)
			return false, nil
		}
		if p := string(ws.Status.Phase); p != "Ready" {
			last = fmt.Errorf("phase %q", p)
			return false, nil
		}
		return true, nil
	})
	if errors.Is(err, kube.ErrAwaitTimeout) && last != nil {
		return fmt.Errorf("gatewaykcp: bootstrap cluster: %s: %w", "workspace "+name+" Ready", last)
	}
	return err
}

func bootstrapCluster(ctx context.Context, kubeconfig, base, did, cluster string, log *slog.Logger) error {
	did = cmp.Or(did, DefaultBootstrapDID)
	restCfg, err := memberRestConfig(kubeconfig)
	if err != nil {
		return err
	}
	client := func(ws string) (*kube.Client, error) { return kube.New(clusterConfig(restCfg, ws)) }
	ap, err := kcpapply.New(restCfg, kcpapply.Options{FieldManager: FieldManager, WaitTimeout: bootstrapWorkspaceTTL, Log: log})
	if err != nil {
		return err
	}
	account := clusterhome.AccountHash(did)
	root, err := client(clusterhome.RootWorkspace)
	if err != nil {
		return err
	}
	if err := retry(ctx, "account workspace", func() error {
		return applyWorkspace(ctx, ap, clusterhome.RootWorkspace, account, string(kcpclient.AccountWorkspaceType))
	}); err != nil {
		return err
	}
	if err := waitReady(ctx, root.Resource(workspaceGVR), account); err != nil {
		return err
	}
	acct, err := client(clusterhome.AccountPath(did))
	if err != nil {
		return err
	}
	identity := kube.Doc{"apiVersion": v1alpha1.GroupVersion.String(), "kind": "AtprotoIdentity",
		"metadata": map[string]any{"name": kcpclient.IdentityRecordName(did),
			"labels": map[string]any{sweepkcp.SweepExemptLabel: sweepkcp.SweepExemptValue}},
		"spec": map[string]any{"did": did, "scope": string(v1alpha1.IdentityScopeAccount), "revoked": false}}
	if err := retry(ctx, "account identity", func() error { return ssa(ctx, acct.Resource(identityGVR), identity) }); err != nil {
		return err
	}
	if err := retry(ctx, "home workspace", func() error {
		return applyWorkspace(ctx, ap, clusterhome.AccountPath(did), cluster, string(kcpclient.HomeWorkspaceType))
	}); err != nil {
		return err
	}
	if err := waitReady(ctx, acct.Resource(workspaceGVR), cluster); err != nil {
		return err
	}
	home, err := client(clusterhome.HomePath(did, cluster))
	if err != nil {
		return err
	}
	obj := kube.Doc{"apiVersion": v1alpha1.GroupVersion.String(), "kind": "Cluster",
		"metadata": map[string]any{"name": cluster, "finalizers": []any{v1alpha1.FinalizerCluster}},
		"spec":     map[string]any{"account": did, "running": true}}
	if err := retry(ctx, "cluster", func() error { return ssa(ctx, home.Resource(clusterGVR), obj) }); err != nil {
		return err
	}
	agent, err := kcpclient.ParseServiceAccountRef(appgateway.WebAppAccountRef(base))
	if err != nil {
		return err
	}
	kc, err := kcpclient.New(kcpclient.Options{Host: restCfg.Host, RestConfig: restCfg, NodeClaimAgent: agent})
	if err != nil {
		return err
	}
	if err := retry(ctx, "web app grant on the home", func() error { return kc.GrantAgentHome(ctx, clusterhome.HomePath(did, cluster)) }); err != nil {
		return err
	}
	log.Info("the bootstrap cluster exists", "cluster", cluster, "account", did, "home", clusterhome.HomePath(did, cluster))
	return nil
}

func BootstrapCluster(ctx context.Context, kubeconfig, base, did, cluster string, log *slog.Logger) error {
	base = cmp.Or(base, compose.DefaultWorkspace)
	return bootstrapCluster(ctx, kubeconfig, base, did, cluster, log)
}