johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package exposurelocal

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"

	apierrors "k8s.io/apimachinery/pkg/api/errors"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	exposurev1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/exposure/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bus"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/binding"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/strs"
)

const (
	Name = "exposurelocal"

	Export = "exposures"

	ServiceKind = "Service"
)

type Options struct {
	Origin string

	Handlers *bus.Bus

	Scope func(workspace string) bool

	Log *slog.Logger
}

type Controller struct {
	origin Origin

	handlers *bus.Bus

	registry *Registry

	refs *binding.Table[struct{}]

	events *ctlruntime.Queue

	cancel func()

	log *slog.Logger
}

func New(opts Options) (*Controller, error) {
	origin, err := ParseOrigin(opts.Origin)
	if err != nil {
		return nil, err
	}
	log := opts.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	ctl := &Controller{origin: origin, handlers: opts.Handlers, registry: NewRegistry(), refs: binding.New[struct{}](),
		events: ctlruntime.NewQueue(), log: log.With("component", "exposurelocal-controller")}
	return ctl, nil
}

func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) {
	if err := m.AddToScheme(exposurev1alpha1.AddToScheme); err != nil {
		return nil, err
	}
	ctl, err := New(opts)
	if err != nil {
		return nil, err
	}
	events, err := ctl.events.Run(m, Name+": handlers")
	if err != nil {
		return nil, fmt.Errorf("exposurelocal: forwarding handler changes: %w", err)
	}
	options := []ctlruntime.Option{
		ctlruntime.Export(Export),
		ctlruntime.Events(events),
		ctlruntime.Gone(ctl.gone),
		ctlruntime.Watches(func(_ context.Context, src ctlruntime.Request, obj *exposurev1alpha1.Service) []ctlruntime.Request {
			return ctl.users(binding.Key(src.Workspace, ServiceKind, obj.Name))
		}),
	}
	if opts.Scope != nil {
		options = append(options, ctlruntime.InWorkspaces(opts.Scope))
	}
	if err := ctlruntime.For[*exposurev1alpha1.Exposure](m, Name, ctl, options...); err != nil {
		return nil, fmt.Errorf("exposurelocal: registering the Exposure controller: %w", err)
	}
	if opts.Handlers != nil {
		ctl.cancel = opts.Handlers.Subscribe(ctl.HandlerChanged)
	}
	return ctl, nil
}

func (c *Controller) Close() {
	if c.cancel != nil {
		c.cancel()
	}
}

func (c *Controller) HandlerChanged(key bus.Key) {
	c.events.Add(c.users(binding.Key(key.Workspace, key.Kind, key.Name))...)
}

func (c *Controller) Registry() *Registry { return c.registry }

func (c *Controller) Lookup(host, path string) (http.Handler, bool) {
	return c.registry.Lookup(host, path)
}

func (c *Controller) HandlerFor(host, path string) (http.Handler, bool) {
	return c.registry.Lookup(host, path)
}

func (c *Controller) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	c.registry.ServeHTTP(w, req)
}

func (c *Controller) users(ref string) []ctlruntime.Request {
	var out []ctlruntime.Request
	for _, t := range c.refs.Users(ref) {
		out = append(out, ctlruntime.Request{Cluster: t.Cluster, Workspace: t.Workspace, Name: t.Name})
	}
	return out
}

func (c *Controller) gone(_ context.Context, req ctlruntime.Request) {
	c.remove(ownerOf(req))
	c.refs.Set(binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}, nil)
}

func (c *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, object *exposurev1alpha1.Exposure) (ctlruntime.Result, error) {
	self := binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}
	if !object.DeletionTimestamp.IsZero() {
		c.remove(ownerOf(req))
		c.refs.Set(self, nil)
		return ctlruntime.Result{}, nil
	}
	spec := object.Spec
	serviceName := strs.LastSegment(spec.ServiceRef)
	var refs []string
	if serviceName != "" {
		refs = append(refs, binding.Key(req.Workspace, ServiceKind, serviceName))
	}
	c.refs.Set(self, &struct{}{}, refs...)
	var svc *exposurev1alpha1.Service
	if serviceName != "" {
		var found exposurev1alpha1.Service
		switch err := req.Get(ctx, "", serviceName, &found); {
		case err == nil:
			svc = &found
		case apierrors.IsNotFound(err):
		default:
			return ctlruntime.Result{}, err
		}
	}
	switch {
	case spec.Suspend:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "Suspended", "the Exposure is suspended")
	case spec.Mode == exposurev1alpha1.ExposureModeIngress:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "NoIngress",
			"an in-process host has no ingress; only a local Exposure is served")
	case spec.Mode != exposurev1alpha1.ExposureModeLocal:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposureFailed, "InvalidMode", fmt.Sprintf("unknown mode %q", spec.Mode))
	case serviceName == "":
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "ServiceMissing", "the Exposure has no serviceRef")
	case svc == nil:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "ServiceMissing",
			"the Service "+serviceName+" is not present in workspace "+req.Workspace)
	case svc.Spec.Suspend:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "ServiceSuspended",
			"the Service "+serviceName+" is suspended")
	}
	target, err := targetOf(svc.Spec.TargetRef, req.Workspace)
	if err != nil {
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposureFailed, "InvalidTarget", err.Error())
	}
	c.refs.Set(self, &struct{}{}, append(refs, binding.Key(target.Workspace, target.Kind, target.Name))...)
	owner := ownerOf(req)
	host, base := c.origin.Host, c.origin.URL
	if c.handlers != nil && isHTTP(svc.Spec.Protocol) {
		_, ok := c.handlers.Handler(target, svc.Spec.Handler)
		if !ok {
			message := fmt.Sprintf("no handler %q is registered for %s/%s in workspace %s", handlerName(svc.Spec.Handler),
				target.Kind, target.Name, target.Workspace)
			return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "HandlerMissing", message)
		}
		if err := c.registry.Register(owner, host, spec.Path, withCaller(c.current(target, svc.Spec.Handler), target)); err != nil {
			return ctlruntime.Result{}, err
		}
		c.log.Debug("the in-process Exposure route is registered", "host", host, "path", spec.Path,
			"target", target.Kind+"/"+target.Name)
	} else {
		c.registry.Remove(owner)
	}
	next := object.DeepCopy()
	next.Status = Status(object.Generation, base, spec.Path)
	return ctlruntime.Result{}, req.ApplyStatus(ctx, next)
}

func (c *Controller) withdraw(ctx context.Context, req ctlruntime.Request, object *exposurev1alpha1.Exposure,
	phase exposurev1alpha1.ExposurePhase, reason, message string) (ctlruntime.Result, error) {
	c.remove(ownerOf(req))
	next := object.DeepCopy()
	next.Status = NotReady(object.Generation, phase, reason, message)
	return ctlruntime.Result{}, req.ApplyStatus(ctx, next)
}

func (c *Controller) remove(owner string) {
	c.registry.Remove(owner)
}

func ownerOf(req ctlruntime.Request) string { return req.Workspace + "/" + req.Name }

func targetOf(ref commonv1.ObjectRef, workspace string) (bus.Key, error) {
	kind := strings.TrimSpace(ref.Kind)
	name := strings.TrimSpace(ref.Name)
	if kind == "" {
		return bus.Key{}, errors.New("the targetRef has no kind")
	}
	if name == "" {
		return bus.Key{}, errors.New("the targetRef has no name")
	}
	targetWorkspace := strings.TrimSpace(ref.Workspace)
	if targetWorkspace == "" {
		targetWorkspace = workspace
	}
	return bus.Key{Workspace: targetWorkspace, Kind: kind, Name: name}, nil
}

func handlerName(value string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		return bus.DefaultHandlerName
	}
	return value
}

func isHTTP(protocol exposurev1alpha1.ServiceProtocol) bool {
	switch exposurev1alpha1.ServiceProtocol(strings.ToLower(strings.TrimSpace(string(protocol)))) {
	case exposurev1alpha1.ServiceProtocolHTTP, exposurev1alpha1.ServiceProtocolWS:
		return true
	}
	return false
}

func (c *Controller) current(target bus.Key, name string) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		handler, ok := c.handlers.Handler(target, name)
		if !ok {
			http.Error(w, "the exposed component has no handler registered", http.StatusServiceUnavailable)
			return
		}
		handler.ServeHTTP(w, r)
	})
}

func withCaller(handler http.Handler, target bus.Key) http.Handler {
	ref := target.Ref()
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		handler.ServeHTTP(w, r.WithContext(bus.WithCaller(r.Context(), ref)))
	})
}