From 2c52d7850c42d2081772ddb6f2e74776d0729ede Mon Sep 17 00:00:00 2001 From: oppiliappan Date: Thu, 21 May 2026 11:46:22 +0000 Subject: [PATCH] appview/pipelines: introduce status notifier allows web/ssh handlers to subscribe to status updates for a particular pipeline by ATURI. we can then live update UI based on data arriving on this channel. spindlestream can now notify interested parties about new pipeline statuses. Signed-off-by: oppiliappan --- appview/pipelines/notifier.go | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++++ appview/state/spindlestream.go | 13 ++++++++----- appview/state/state.go | 8 +++++++- 3 file(s) changed, 67 insertion(s)(+), 6 deletion(s)(-) diff --git a/appview/pipelines/notifier.go b/appview/pipelines/notifier.go new file mode 100644 --- /dev/null +++ b/appview/pipelines/notifier.go @@ -0,0 +1,52 @@ +package pipelines + +import ( + "sync" + + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/notifier" +) + +// StatusNotifier is a keyed broadcast notifier for pipeline status changes, keyed by the pipeline's AT URI. +// +// subscribers are notified whenever a status update arrives for that pipeline +type StatusNotifier struct { + mu sync.Mutex + keys map[syntax.ATURI]*notifier.Notifier +} + +func NewStatusNotifier() *StatusNotifier { + return &StatusNotifier{ + keys: make(map[syntax.ATURI]*notifier.Notifier), + } +} + +func (n *StatusNotifier) Publish(uri syntax.ATURI) { + n.mu.Lock() + p, ok := n.keys[uri] + n.mu.Unlock() + if ok { + p.NotifyAll() + } +} + +func (n *StatusNotifier) Subscribe(uri syntax.ATURI) chan struct{} { + n.mu.Lock() + p, ok := n.keys[uri] + if !ok { + nb := notifier.New() + p = &nb + n.keys[uri] = p + } + n.mu.Unlock() + return p.Subscribe() +} + +func (n *StatusNotifier) Unsubscribe(uri syntax.ATURI, ch chan struct{}) { + n.mu.Lock() + p, ok := n.keys[uri] + n.mu.Unlock() + if ok { + p.Unsubscribe(ch) + } +} diff --git a/appview/state/spindlestream.go b/appview/state/spindlestream.go --- a/appview/state/spindlestream.go +++ b/appview/state/spindlestream.go @@ -14,6 +14,7 @@ "tangled.org/core/appview/config" "tangled.org/core/appview/db" "tangled.org/core/appview/models" + "tangled.org/core/appview/pipelines" ec "tangled.org/core/eventconsumer" "tangled.org/core/eventconsumer/cursor" "tangled.org/core/log" @@ -22,7 +23,7 @@ spindle "tangled.org/core/spindle/models" ) -func Spindlestream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac.Enforcer) (*ec.Consumer, error) { +func Spindlestream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac.Enforcer, pn *pipelines.StatusNotifier) (*ec.Consumer, error) { logger := log.FromContext(ctx) logger = log.SubLogger(logger, "spindlestream") @@ -46,7 +47,7 @@ cfg := ec.ConsumerConfig{ Sources: srcs, - ProcessFunc: spindleIngester(ctx, logger, d), + ProcessFunc: spindleIngester(ctx, logger, d, pn), RetryInterval: c.Spindlestream.RetryInterval, MaxRetryInterval: c.Spindlestream.MaxRetryInterval, ConnectionTimeout: c.Spindlestream.ConnectionTimeout, @@ -60,18 +61,18 @@ return ec.NewConsumer(cfg), nil } -func spindleIngester(ctx context.Context, logger *slog.Logger, d *db.DB) ec.ProcessFunc { +func spindleIngester(ctx context.Context, logger *slog.Logger, d *db.DB, pn *pipelines.StatusNotifier) ec.ProcessFunc { return func(ctx context.Context, source ec.Source, msg ec.Message) error { switch msg.Nsid { case tangled.PipelineStatusNSID: - return ingestPipelineStatus(ctx, logger, d, source, msg) + return ingestPipelineStatus(ctx, logger, d, pn, source, msg) } return nil } } -func ingestPipelineStatus(ctx context.Context, logger *slog.Logger, d *db.DB, source ec.Source, msg ec.Message) error { +func ingestPipelineStatus(ctx context.Context, logger *slog.Logger, d *db.DB, pn *pipelines.StatusNotifier, source ec.Source, msg ec.Message) error { var record tangled.PipelineStatus err := json.Unmarshal(msg.EventJson, &record) if err != nil { @@ -110,6 +111,8 @@ if err != nil { return fmt.Errorf("failed to add pipeline status: %w", err) } + + pn.Publish(pipelineUri) return nil } diff --git a/appview/state/state.go b/appview/state/state.go --- a/appview/state/state.go +++ b/appview/state/state.go @@ -28,6 +28,8 @@ whnotify "tangled.org/core/appview/notify/webhook" "tangled.org/core/appview/oauth" "tangled.org/core/appview/pages" + "tangled.org/core/appview/pipelines" + pipelinessh "tangled.org/core/appview/pipelines/ssh" "tangled.org/core/appview/reporesolver" "tangled.org/core/appview/repoverify" "tangled.org/core/appview/validator" @@ -67,6 +69,7 @@ repoResolver *reporesolver.RepoResolver knotstream *eventconsumer.Consumer spindlestream *eventconsumer.Consumer + pipelineNotifier *pipelines.StatusNotifier logger *slog.Logger validator *validator.Validator cfClient *cloudflare.Client @@ -210,7 +213,9 @@ } knotstream.Start(ctx) - spindlestream, err := Spindlestream(ctx, config, d, enforcer) + pipelineNotifier := pipelines.NewStatusNotifier() + + spindlestream, err := Spindlestream(ctx, config, d, enforcer, pipelineNotifier) if err != nil { return nil, fmt.Errorf("failed to start spindlestream consumer: %w", err) } @@ -232,6 +237,7 @@ repoResolver: repoResolver, knotstream: knotstream, spindlestream: spindlestream, + pipelineNotifier: pipelineNotifier, logger: logger, validator: validator, cfClient: cfClient, -- tangled.sh