Monorepo for Tangled
spindle/{ingester,server},jetstream: ingest repo.pull
Signed-off-by: dawn <dawn@tangled.org>
+22
-4
+16
-3
jetstream/jetstream.go
+16
-3
jetstream/jetstream.go
···
30
30
ident string
31
31
l *slog.Logger
32
32
33
-
logDids bool
34
-
wantedDids Set[string]
33
+
logDids bool
34
+
wantedDids Set[string]
35
+
unfilteredNsids Set[string]
35
36
db DB
36
37
waitForDid bool
37
38
mu sync.RWMutex
···
52
53
}
53
54
j.mu.Lock()
54
55
j.wantedDids[did] = struct{}{}
56
+
j.mu.Unlock()
57
+
}
58
+
59
+
func (j *JetstreamClient) ExemptCollection(nsid string) {
60
+
j.mu.Lock()
61
+
j.unfilteredNsids[nsid] = struct{}{}
55
62
j.mu.Unlock()
56
63
}
57
64
···
82
89
matches = true
83
90
}
84
91
}
92
+
if !matches && evt.Commit != nil {
93
+
if _, ok := j.unfilteredNsids[evt.Commit.Collection]; ok {
94
+
matches = true
95
+
}
96
+
}
85
97
j.mu.RUnlock()
86
98
87
99
var err error
···
106
118
ident: ident,
107
119
db: db,
108
120
l: logger,
109
-
wantedDids: make(map[string]struct{}),
121
+
wantedDids: make(map[string]struct{}),
122
+
unfilteredNsids: make(map[string]struct{}),
110
123
111
124
logDids: logDids,
112
125
+3
-1
spindle/ingester.go
+3
-1
spindle/ingester.go
···
27
27
switch e.Commit.Collection {
28
28
case tangled.SpindleMemberNSID:
29
29
err = s.ingestMember(ctx, e)
30
-
case tangled.RepoNSID, tangled.RepoCollaboratorNSID:
30
+
case tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID:
31
31
if evt, ok := jetstreamToTapEvent(e); ok {
32
32
err = s.tap.processEvent(ctx, evt)
33
33
}
···
68
68
Collection: syntax.NSID(e.Commit.Collection),
69
69
Action: action,
70
70
Record: e.Commit.Record,
71
+
// jetstream is only used for live
72
+
Live: true,
71
73
},
72
74
}, true
73
75
}
+3
spindle/server.go
+3
spindle/server.go
···
123
123
tangled.SpindleMemberNSID,
124
124
tangled.RepoNSID,
125
125
tangled.RepoCollaboratorNSID,
126
+
tangled.RepoPullNSID,
126
127
}
127
128
jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, "spindle", collections, nil, log.SubLogger(logger, "jetstream"), d, true, true)
128
129
if err != nil {
129
130
return nil, fmt.Errorf("failed to setup jetstream client: %w", err)
130
131
}
131
132
jc.AddDid(cfg.Server.Owner)
133
+
// pull records are created by arbitrary users too, same hack as in tap
134
+
jc.ExemptCollection(tangled.RepoPullNSID)
132
135
133
136
// Check if the spindle knows about any Dids;
134
137
dids, err := d.GetAllDids()