From 708414eccaeda15e25c51ec1d9d71779234199e4 Mon Sep 17 00:00:00 2001
From: oppiliappan <me@oppi.li>
Date: Tue, 17 Jun 2025 10:41:22 +0100
Subject: [PATCH] spindle: filter jetstream by wanted dids

Signed-off-by: oppiliappan <me@oppi.li>
---
 knotclient/events.go |  6 ++++++
 spindle/server.go    | 13 +++++++------
 2 files changed, 13 insertions(+), 6 deletions(-)

diff --git a/knotclient/events.go b/knotclient/events.go
index 7d14e555..f0ceebf2 100644
--- a/knotclient/events.go
+++ b/knotclient/events.go
@@ -150,6 +150,12 @@ func (c *EventConsumer) Stop() {
 }
 
 func (c *EventConsumer) AddSource(ctx context.Context, s EventSource) {
+	// we are already listening to this source
+	if _, ok := c.cfg.Sources[s]; ok {
+		c.logger.Info("source already present", "source", s)
+		return
+	}
+
 	c.cfgMu.Lock()
 	c.cfg.Sources[s] = struct{}{}
 	c.wg.Add(1)
diff --git a/spindle/server.go b/spindle/server.go
index eacbbf49..e2e6618a 100644
--- a/spindle/server.go
+++ b/spindle/server.go
@@ -70,10 +70,11 @@ func Run(ctx context.Context) error {
 		tangled.SpindleMemberNSID,
 		tangled.RepoNSID,
 	}
-	jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, "spindle", collections, nil, logger, d, false, false)
+	jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, "spindle", collections, nil, logger, d, true, true)
 	if err != nil {
 		return fmt.Errorf("failed to setup jetstream client: %w", err)
 	}
+	jc.AddDid(cfg.Server.Owner)
 
 	spindle := Spindle{
 		jc:  jc,
@@ -118,19 +119,19 @@ func Run(ctx context.Context) error {
 	ccfg.Dev = cfg.Server.Dev
 	ccfg.ProcessFunc = spindle.processPipeline
 	ccfg.CursorStore = cursorStore
-	knotstream := knotclient.NewEventConsumer(*ccfg)
 	knownKnots, err := d.Knots()
 	if err != nil {
 		return err
 	}
 	for _, knot := range knownKnots {
-		knotstream.AddSource(ctx, knotclient.NewEventSource(knot))
+		logger.Info("adding source start", "knot", knot)
+		ccfg.Sources[knotclient.EventSource{knot}] = struct{}{}
 	}
-	spindle.ks = knotstream
+	spindle.ks = knotclient.NewEventConsumer(*ccfg)
 
 	go func() {
-		logger.Info("starting knot event consumer", "knots")
-		knotstream.Start(ctx)
+		logger.Info("starting knot event consumer")
+		spindle.ks.Start(ctx)
 	}()
 
 	logger.Info("starting spindle server", "address", cfg.Server.ListenAddr)
-- 
2.54.0