From bd32ea8f781385f16da67148476fcf965d2063e4 Mon Sep 17 00:00:00 2001
From: dawn <dawn@tangled.org>
Date: Thu, 16 Jul 2026 17:24:31 +0000
Subject: [PATCH] spindle/{ingester,server},jetstream: ingest repo.pull

Signed-off-by: dawn <dawn@tangled.org>
---
 jetstream/jetstream.go | 19 ++++++++++++++++---
 spindle/ingester.go | 4 +++-
 spindle/server.go | 3 +++
 3 file(s) changed, 22 insertion(s)(+), 4 deletion(s)(-)

diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -30,8 +30,9 @@ client *client.Client
 	ident  string
 	l      *slog.Logger
 
-	logDids    bool
-	wantedDids Set[string]
+	logDids           bool
+	wantedDids        Set[string]
+	unfilteredNsids   Set[string]
 	db         DB
 	waitForDid bool
 	mu         sync.RWMutex
@@ -55,6 +56,12 @@ j.wantedDids[did] = struct{}{}
 	j.mu.Unlock()
 }
 
+func (j *JetstreamClient) ExemptCollection(nsid string) {
+	j.mu.Lock()
+	j.unfilteredNsids[nsid] = struct{}{}
+	j.mu.Unlock()
+}
+
 func (j *JetstreamClient) RemoveDid(did string) {
 	if did == "" {
 		return
@@ -82,6 +89,11 @@ if _, ok := j.wantedDids[evt.Did]; ok {
 				matches = true
 			}
 		}
+		if !matches && evt.Commit != nil {
+			if _, ok := j.unfilteredNsids[evt.Commit.Collection]; ok {
+				matches = true
+			}
+		}
 		j.mu.RUnlock()
 
 		var err error
@@ -106,7 +118,8 @@ cfg:        cfg,
 		ident:      ident,
 		db:         db,
 		l:          logger,
-		wantedDids: make(map[string]struct{}),
+		wantedDids:      make(map[string]struct{}),
+		unfilteredNsids: make(map[string]struct{}),
 
 		logDids: logDids,
 
diff --git a/spindle/ingester.go b/spindle/ingester.go
--- a/spindle/ingester.go
+++ b/spindle/ingester.go
@@ -27,7 +27,7 @@ var err error
 		switch e.Commit.Collection {
 		case tangled.SpindleMemberNSID:
 			err = s.ingestMember(ctx, e)
-		case tangled.RepoNSID, tangled.RepoCollaboratorNSID:
+		case tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID:
 			if evt, ok := jetstreamToTapEvent(e); ok {
 				err = s.tap.processEvent(ctx, evt)
 			}
@@ -68,6 +68,8 @@ Rkey:       syntax.RecordKey(e.Commit.RKey),
 			Collection: syntax.NSID(e.Commit.Collection),
 			Action:     action,
 			Record:     e.Commit.Record,
+			// jetstream is only used for live
+			Live: true,
 		},
 	}, true
 }
diff --git a/spindle/server.go b/spindle/server.go
--- a/spindle/server.go
+++ b/spindle/server.go
@@ -123,12 +123,15 @@ collections := []string{
 		tangled.SpindleMemberNSID,
 		tangled.RepoNSID,
 		tangled.RepoCollaboratorNSID,
+		tangled.RepoPullNSID,
 	}
 	jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, "spindle", collections, nil, log.SubLogger(logger, "jetstream"), d, true, true)
 	if err != nil {
 		return nil, fmt.Errorf("failed to setup jetstream client: %w", err)
 	}
 	jc.AddDid(cfg.Server.Owner)
+	// pull records are created by arbitrary users too, same hack as in tap
+	jc.ExemptCollection(tangled.RepoPullNSID)
 
 	// Check if the spindle knows about any Dids;
 	dids, err := d.GetAllDids()

--
tangled.sh