johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package eventstream

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"maps"
	"strings"

	"github.com/bluesky-social/indigo/atproto/identity"
	"github.com/bluesky-social/indigo/atproto/syntax"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	eventsapi "github.com/johnandersen777/socialweb-computer-kcp/api/events/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
)

const AtprotoRepoKind = "AtprotoRepo"

type fallbackDirectory []identity.Directory

func (d fallbackDirectory) lookup(fn func(identity.Directory) (*identity.Identity, error)) (*identity.Identity, error) {
	var last error
	for _, dir := range d {
		ident, err := fn(dir)
		if err == nil {
			return ident, nil
		}
		last = err
	}
	if last == nil {
		last = identity.ErrDIDNotFound
	}
	return nil, last
}

func (d fallbackDirectory) LookupHandle(ctx context.Context, handle syntax.Handle) (*identity.Identity, error) {
	return d.lookup(func(dir identity.Directory) (*identity.Identity, error) {
		return dir.LookupHandle(ctx, handle)
	})
}

func (d fallbackDirectory) LookupDID(ctx context.Context, did syntax.DID) (*identity.Identity, error) {
	return d.lookup(func(dir identity.Directory) (*identity.Identity, error) {
		return dir.LookupDID(ctx, did)
	})
}

func (d fallbackDirectory) Lookup(ctx context.Context, atid syntax.AtIdentifier) (*identity.Identity, error) {
	return d.lookup(func(dir identity.Directory) (*identity.Identity, error) {
		return dir.Lookup(ctx, atid)
	})
}

func (d fallbackDirectory) Purge(ctx context.Context, atid syntax.AtIdentifier) error {
	var last error
	for _, dir := range d {
		if err := dir.Purge(ctx, atid); err != nil {
			last = err
		}
	}
	return last
}

type SourcePlan struct {
	Name string

	Key string

	SourceRef commonv1.ObjectRef

	RepoRef *commonv1.ObjectRef

	Filter eventsapi.EventStreamFilter
}

type WatchRepo struct {
	DID string

	Ref *commonv1.ObjectRef
}

func EventStreamNameForWatchSource(watchName, sourceKey string) string {
	if strings.HasPrefix(sourceKey, "pds/") {
		return EventStreamNameForWatch(watchName)
	}
	sum := sha256.Sum256([]byte(sourceKey))
	suffix := hex.EncodeToString(sum[:6])
	candidate := EventStreamNamePrefix + watchName + "-" + suffix
	if len(candidate) <= maxEventStreamName {
		return candidate
	}
	keep := maxEventStreamName - len(suffix) - 1
	return candidate[:keep] + "-" + suffix
}

func SpecRepo(spec v1alpha1.AtprotoRecordWatchSpec) (string, error) {
	repo := strings.TrimSpace(spec.Repo)
	switch {
	case strings.HasPrefix(repo, "did:"):
		return repo, nil
	case repo == "" && len(spec.Source.EventSourceRefs) == 0:
		return "", errors.New("eventstream: spec.repo or spec.repoRef is required")
	}
	return "", nil
}

func PlanSources(workspace, watchName string, spec v1alpha1.AtprotoRecordWatchSpec, repo WatchRepo) ([]SourcePlan, error) {
	filter := eventsapi.EventStreamFilter{}
	if collection := strings.TrimSpace(spec.Collection); collection != "" {
		filter.Collections = []string{collection}
	}
	if repo.DID != "" {
		filter.DIDs = []string{repo.DID}
	}
	var plans []SourcePlan
	seen := map[string]bool{}
	for _, ref := range spec.Source.EventSourceRefs {
		name := strings.TrimSpace(ref.Name)
		if name == "" {
			continue
		}
		if kind := strings.TrimSpace(ref.Kind); kind != "" && !strings.EqualFold(kind, eventsapi.EventSourceKind) {
			return nil, fmt.Errorf("eventstream: source.eventSourceRefs must name EventSources, not %s %s", kind, name)
		}
		source := commonv1.ObjectRef{Group: eventsapi.GroupName, Kind: eventsapi.EventSourceKind, Name: name, Workspace: ref.InWorkspace(workspace)}
		key := "source/" + source.Workspace + "/" + name
		if seen[key] {
			continue
		}
		seen[key] = true
		plans = append(plans, SourcePlan{Name: EventStreamNameForWatchSource(watchName, key), Key: key, SourceRef: source, Filter: filter})
	}
	if len(plans) > 0 {
		return plans, nil
	}
	if repo.Ref == nil {
		return nil, errors.New("eventstream: the record watch names no source.eventSourceRefs and no AtprotoRepo repoRef")
	}
	key := "pds/" + repo.Ref.Name
	name := EventStreamNameForWatchSource(watchName, key)
	plans = append(plans, SourcePlan{Name: name, Key: key, RepoRef: repo.Ref, Filter: filter,
		SourceRef: commonv1.ObjectRef{Group: eventsapi.GroupName, Kind: eventsapi.EventSourceKind, Name: name, Workspace: workspace}})
	return plans, nil
}

func EventStreamFor(owner metav1.OwnerReference, workspace string, plan SourcePlan) *eventsapi.EventStream {
	source := plan.SourceRef
	if source.Workspace == workspace {
		source.Workspace = ""
	}
	return &eventsapi.EventStream{
		TypeMeta: metav1.TypeMeta{APIVersion: eventsapi.GroupVersion.String(), Kind: "EventStream"},
		ObjectMeta: metav1.ObjectMeta{Name: plan.Name, Labels: map[string]string{RecordWatchLabel: owner.Name},
			OwnerReferences: []metav1.OwnerReference{owner}},
		Spec: eventsapi.EventStreamSpec{
			SourceRef: source,
			Filter:    plan.Filter,
			Sink:      eventsapi.EventStreamSink{Kind: eventsapi.EventStreamSinkRecordWatch, Name: owner.Name, Namespace: workspace},
		},
	}
}

func PdsSourcesFor(owner metav1.OwnerReference, workspace string, plan SourcePlan) (*eventsapi.PdsEventSource, *eventsapi.EventSource, bool) {
	if plan.RepoRef == nil {
		return nil, nil, false
	}
	repo := *plan.RepoRef
	if repo.Workspace == workspace {
		repo.Workspace = ""
	}
	name := plan.SourceRef.Name
	labels := map[string]string{RecordWatchLabel: owner.Name}
	owners := []metav1.OwnerReference{owner}
	backend := &eventsapi.PdsEventSource{TypeMeta: metav1.TypeMeta{APIVersion: eventsapi.GroupVersion.String(), Kind: eventsapi.PdsEventSourceKind},
		ObjectMeta: metav1.ObjectMeta{Name: name, Labels: labels, OwnerReferences: owners}, Spec: eventsapi.PdsEventSourceSpec{RepoRef: repo}}
	abstract := &eventsapi.EventSource{TypeMeta: metav1.TypeMeta{APIVersion: eventsapi.GroupVersion.String(), Kind: eventsapi.EventSourceKind},
		ObjectMeta: metav1.ObjectMeta{Name: name, Labels: maps.Clone(labels), OwnerReferences: owners},
		Spec:       eventsapi.EventSourceSpec{BackendRef: commonv1.ObjectRef{Group: eventsapi.GroupName, Kind: eventsapi.PdsEventSourceKind, Name: name}}}
	return backend, abstract, true
}