johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package kcpstore

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

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"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/internal/kcplibs/common/clientlimit"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcplibs/common/ref"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/kcp"
)

func IsNotFound(err error) bool {
	return apierrors.IsNotFound(err)
}

var ErrNoPath = errors.New("kcpstore: the logical cluster carries no kcp.io/path annotation")

type Options struct {
	Host string

	RestConfig *rest.Config

	Transport http.RoundTripper

	QPS float32

	Burst int
}

func (o Options) tuned(cfg *rest.Config) *rest.Config {
	return clientlimit.Apply(cfg, o.QPS, o.Burst)
}

type Store struct {
	cfg *rest.Config

	http *http.Client

	codecs runtime.NegotiatedSerializer

	mu sync.Mutex

	clients map[string]rest.Interface
}

func New(opts Options) (*Store, error) {
	if opts.Host == "" {
		return nil, errors.New("kcpstore: Host is required")
	}
	cfg := opts.tuned(&rest.Config{})
	if opts.RestConfig != nil {
		cfg = opts.tuned(opts.RestConfig)
	}
	cfg.Host = ref.BaseHost(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("kcpstore: %w", err)
	}
	return &Store{
		cfg:     cfg,
		http:    httpClient,
		codecs:  serializer.NewCodecFactory(runtime.NewScheme()).WithoutConversion(),
		clients: map[string]rest.Interface{},
	}, nil
}

func (s *Store) Config() *rest.Config {
	return s.cfg
}

func (s *Store) ClientFor(logicalCluster string, gv schema.GroupVersion) (rest.Interface, error) {
	key := logicalCluster + "|" + gv.String()
	s.mu.Lock()
	if client, ok := s.clients[key]; ok {
		s.mu.Unlock()
		return client, nil
	}
	s.mu.Unlock()

	cfg := rest.CopyConfig(s.cfg)
	cfg.GroupVersion = &gv
	apiPath := ref.APIPathPrefix + logicalCluster + "/apis"
	if gv.Group == "" {
		apiPath = ref.APIPathPrefix + logicalCluster + "/api"
	}
	cfg.APIPath = apiPath
	cfg.NegotiatedSerializer = s.codecs
	client, err := rest.RESTClientForConfigAndClient(cfg, s.http)
	if err != nil {
		return nil, fmt.Errorf("kcpstore: %s: %w", logicalCluster, err)
	}
	s.mu.Lock()
	s.clients[key] = client
	s.mu.Unlock()
	return client, nil
}

type Resource[T any] struct {
	store *Store

	gvr schema.GroupVersionResource

	clusterScoped bool
}

type ResourceOptions struct {
	ClusterScoped bool
}

func Of[T any](s *Store, gvr schema.GroupVersionResource) *Resource[T] {
	return OfWithOptions[T](s, gvr, ResourceOptions{})
}

func OfClusterScoped[T any](s *Store, gvr schema.GroupVersionResource) *Resource[T] {
	return OfWithOptions[T](s, gvr, ResourceOptions{ClusterScoped: true})
}

func OfWithOptions[T any](s *Store, gvr schema.GroupVersionResource, opts ResourceOptions) *Resource[T] {
	return &Resource[T]{store: s, gvr: gvr, clusterScoped: opts.ClusterScoped}
}

func (r *Resource[T]) ClusterScoped() *Resource[T] {
	if r == nil {
		return nil
	}
	out := *r
	out.clusterScoped = true
	return &out
}

func (r *Resource[T]) withNamespace(req *rest.Request, namespace string) *rest.Request {
	if r.clusterScoped {
		return req
	}
	return req.Namespace(namespace)
}

func (r *Resource[T]) client(logicalCluster string) (rest.Interface, error) {
	return r.store.ClientFor(logicalCluster, r.gvr.GroupVersion())
}

func (r *Resource[T]) GetRaw(ctx context.Context, target ref.Ref) ([]byte, error) {
	c, err := r.client(target.LogicalCluster)
	if err != nil {
		return nil, err
	}
	raw, err := r.withNamespace(c.Get(), target.Namespace).Resource(r.gvr.Resource).Name(target.Name).Do(ctx).Raw()
	if err != nil {
		return nil, fmt.Errorf("kcpstore: read %s %s in %s: %w", r.gvr.Resource, target.Name, target.LogicalCluster, err)
	}
	return raw, nil
}

func (r *Resource[T]) Get(ctx context.Context, target ref.Ref) (*T, error) {
	raw, err := r.GetRaw(ctx, target)
	if err != nil {
		return nil, err
	}
	return decode[T](raw)
}

func (r *Resource[T]) List(ctx context.Context, logicalCluster string) ([]T, error) {
	c, err := r.client(logicalCluster)
	if err != nil {
		return nil, err
	}
	raw, err := c.Get().Resource(r.gvr.Resource).Do(ctx).Raw()
	if err != nil {
		return nil, fmt.Errorf("kcpstore: list %s in %s: %w", r.gvr.Resource, logicalCluster, err)
	}
	return decodeList[T](raw)
}

func (r *Resource[T]) ListIn(ctx context.Context, logicalCluster, namespace string) ([]T, error) {
	c, err := r.client(logicalCluster)
	if err != nil {
		return nil, err
	}
	raw, err := c.Get().Namespace(namespace).Resource(r.gvr.Resource).Do(ctx).Raw()
	if err != nil {
		return nil, fmt.Errorf("kcpstore: list %s in %s/%s: %w", r.gvr.Resource, logicalCluster, namespace, err)
	}
	return decodeList[T](raw)
}

func (r *Resource[T]) Create(ctx context.Context, logicalCluster string, obj *T) error {
	c, err := r.client(logicalCluster)
	if err != nil {
		return err
	}
	raw, err := json.Marshal(obj)
	if err != nil {
		return fmt.Errorf("kcpstore: encode %s: %w", r.gvr.Resource, err)
	}
	namespace := ""
	if !r.clusterScoped {
		namespace, err = namespaceOf(raw)
		if err != nil {
			return err
		}
	}
	response, err := r.withNamespace(c.Post(), namespace).Resource(r.gvr.Resource).Body(raw).Do(ctx).Raw()
	if err != nil {
		return fmt.Errorf("kcpstore: create %s %s in %s: %w%s", r.gvr.Resource, NameOf(raw), logicalCluster, err, ResponseDetail(response))
	}
	return nil
}

func ResponseDetail(response []byte) string {
	if len(response) == 0 {
		return ""
	}
	return ": " + strings.TrimSpace(string(response))
}

func (r *Resource[T]) Delete(ctx context.Context, target ref.Ref) error {
	c, err := r.client(target.LogicalCluster)
	if err != nil {
		return err
	}
	err = r.withNamespace(c.Delete(), target.Namespace).Resource(r.gvr.Resource).Name(target.Name).
		Body([]byte(`{"propagationPolicy":"Background"}`)).Do(ctx).Error()
	if err != nil && !IsNotFound(err) {
		return fmt.Errorf("kcpstore: delete %s %s in %s: %w", r.gvr.Resource, target.Name, target.LogicalCluster, err)
	}
	return nil
}

func (r *Resource[T]) ApplyStatus(ctx context.Context, target ref.Ref, body []byte, fieldManager string) error {
	c, err := r.client(target.LogicalCluster)
	if err != nil {
		return err
	}
	response, err := r.withNamespace(c.Patch(types.ApplyPatchType), target.Namespace).SubResource("status").Resource(r.gvr.Resource).
		Name(target.Name).Param("fieldManager", fieldManager).Param("force", "true").Body(body).Do(ctx).Raw()
	if err != nil {
		return fmt.Errorf("kcpstore: apply %s status for %s in %s: %w%s", r.gvr.Resource, target.Name, target.LogicalCluster, err, ResponseDetail(response))
	}
	return nil
}

func (r *Resource[T]) Apply(ctx context.Context, target ref.Ref, body []byte, fieldManager string) error {
	c, err := r.client(target.LogicalCluster)
	if err != nil {
		return err
	}
	response, err := r.withNamespace(c.Patch(types.ApplyPatchType), target.Namespace).Resource(r.gvr.Resource).
		Name(target.Name).Param("fieldManager", fieldManager).Param("force", "true").Body(body).Do(ctx).Raw()
	if err != nil {
		return fmt.Errorf("kcpstore: apply %s %s in %s: %w%s", r.gvr.Resource, target.Name, target.LogicalCluster, err, ResponseDetail(response))
	}
	return nil
}

func decode[T any](raw []byte) (*T, error) {
	var out T
	if err := json.Unmarshal(raw, &out); err != nil {
		return nil, fmt.Errorf("kcpstore: decode object: %w", err)
	}
	return &out, nil
}

func decodeList[T any](raw []byte) ([]T, error) {
	var list struct {
		Items []T `json:"items"`
	}
	if err := json.Unmarshal(raw, &list); err != nil {
		return nil, fmt.Errorf("kcpstore: decode list: %w", err)
	}
	return list.Items, nil
}

func namespaceOf(raw []byte) (string, error) {
	var obj struct {
		Metadata struct {
			Namespace string `json:"namespace"`
		} `json:"metadata"`
	}
	if err := json.Unmarshal(raw, &obj); err != nil {
		return "", fmt.Errorf("kcpstore: encode does not round trip: %w", err)
	}
	return obj.Metadata.Namespace, nil
}

func NameOf(raw []byte) string {
	var obj struct {
		Metadata struct {
			Name string `json:"name"`
		} `json:"metadata"`
	}
	_ = json.Unmarshal(raw, &obj)
	return obj.Metadata.Name
}

func (s *Store) ClusterPath(ctx context.Context, logicalCluster string) (string, error) {
	c, err := s.ClientFor(logicalCluster, schema.GroupVersion{Group: "core.kcp.io", Version: "v1alpha1"})
	if err != nil {
		return "", err
	}
	raw, err := c.Get().Resource("logicalclusters").Name("cluster").Do(ctx).Raw()
	if err != nil {
		return "", fmt.Errorf("kcpstore: read logical cluster %s: %w", logicalCluster, err)
	}
	var obj struct {
		Metadata struct {
			Annotations map[string]string `json:"annotations"`
		} `json:"metadata"`
	}
	if err := json.Unmarshal(raw, &obj); err != nil {
		return "", fmt.Errorf("kcpstore: parse logical cluster %s: %w", logicalCluster, err)
	}
	path := obj.Metadata.Annotations[kcp.PathAnnotation]
	if path == "" {
		return "", fmt.Errorf("%w: %s", ErrNoPath, logicalCluster)
	}
	return path, nil
}

func (s *Store) MintServiceAccountToken(ctx context.Context, logicalCluster, namespace, name string, ttl time.Duration) (string, error) {
	namespace = cmp.Or(namespace, "default")
	c, err := s.ClientFor(logicalCluster, schema.GroupVersion{Group: "", Version: "v1"})
	if err != nil {
		return "", err
	}
	seconds := int64(ttl / time.Second)
	body, err := json.Marshal(map[string]any{
		"apiVersion": "authentication.k8s.io/v1",
		"kind":       "TokenRequest",
		"spec": map[string]any{
			"audiences":         []string{},
			"expirationSeconds": seconds,
		},
	})
	if err != nil {
		return "", fmt.Errorf("kcpstore: encode the token request: %w", err)
	}
	raw, err := c.Post().Namespace(namespace).Resource("serviceaccounts").Name(name).
		SubResource("token").Body(body).Do(ctx).Raw()
	if err != nil {
		return "", fmt.Errorf("kcpstore: mint token for %s/%s in %s: %w", namespace, name, logicalCluster, err)
	}
	var out struct {
		Status struct {
			Token string `json:"token"`
		} `json:"status"`
	}
	if err := json.Unmarshal(raw, &out); err != nil {
		return "", fmt.Errorf("kcpstore: parse the token request response: %w", err)
	}
	if out.Status.Token == "" {
		return "", fmt.Errorf("kcpstore: the TokenRequest for %s/%s in %s returned no token", namespace, name, logicalCluster)
	}
	return out.Status.Token, nil
}