johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package jetstreameventsource

import (
	"context"
	"fmt"

	eventsv1 "github.com/johnandersen777/socialweb-computer-kcp/api/events/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/eventsource"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/events/jetstream"
)

const Name = "jetstreameventsource"

func Register(ctx context.Context, m *ctlruntime.Manager, opts eventsource.BackendOptions) (*eventsource.Backend[*eventsv1.JetstreamEventSource], error) {
	dial := eventsource.BusDial(opts.Bus)
	b := eventsource.NewBackend(ctx, eventsv1.JetstreamEventSourceKind, opts,
		func(_ context.Context, _ ctlruntime.Request, obj *eventsv1.JetstreamEventSource) (eventsource.Upstream, error) {
			source, err := jetstream.New(jetstream.Config{URL: obj.Spec.URL}, jetstream.Deps{Dial: dial, Log: opts.Log})
			if err != nil {
				return eventsource.Upstream{}, err
			}
			return eventsource.Upstream{Address: source.Address(), Source: source}, nil
		},
		func(o *eventsv1.JetstreamEventSource) *eventsv1.EventSourceStatus { return &o.Status })
	if err := ctlruntime.For[*eventsv1.JetstreamEventSource](m, Name, b, ctlruntime.Export(eventsource.Export)); err != nil {
		return nil, fmt.Errorf("jetstreameventsource: registering the JetstreamEventSource controller: %w", err)
	}
	return b, nil
}