its for when you want to get like notifications for your reposts
feat: the rest of the owl
+224
-178
+4
go.mod
+4
go.mod
···
22
22
github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
23
23
github.com/hashicorp/go-retryablehttp v0.7.5 // indirect
24
24
github.com/hashicorp/golang-lru v1.0.2 // indirect
25
+
github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
25
26
github.com/ipfs/bbloom v0.0.4 // indirect
26
27
github.com/ipfs/go-block-format v0.2.0 // indirect
27
28
github.com/ipfs/go-cid v0.4.1 // indirect
···
53
54
github.com/prometheus/procfs v0.15.1 // indirect
54
55
github.com/spaolacci/murmur3 v1.1.0 // indirect
55
56
github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e // indirect
57
+
gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b // indirect
58
+
gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02 // indirect
56
59
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect
57
60
go.opentelemetry.io/otel v1.21.0 // indirect
58
61
go.opentelemetry.io/otel/metric v1.21.0 // indirect
···
62
65
go.uber.org/zap v1.26.0 // indirect
63
66
golang.org/x/crypto v0.22.0 // indirect
64
67
golang.org/x/sys v0.22.0 // indirect
68
+
golang.org/x/time v0.5.0 // indirect
65
69
golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
66
70
google.golang.org/protobuf v1.34.2 // indirect
67
71
lukechampine.com/blake3 v1.2.1 // indirect
+8
go.sum
+8
go.sum
···
42
42
github.com/hashicorp/go-retryablehttp v0.7.5/go.mod h1:Jy/gPYAdjqffZ/yFGCFV2doI5wjtH1ewM9u8iYVjtX8=
43
43
github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c=
44
44
github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
45
+
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
46
+
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
45
47
github.com/ipfs/bbloom v0.0.4 h1:Gi+8EGJ2y5qiD5FbsbpX/TMNcJw8gSqr7eyjHa4Fhvs=
46
48
github.com/ipfs/bbloom v0.0.4/go.mod h1:cS9YprKXpoZ9lT0n/Mw/a6/aFV6DTjTLYHeA+gyqMG0=
47
49
github.com/ipfs/go-block-format v0.2.0 h1:ZqrkxBA2ICbDRbK8KJs/u0O3dlp6gmAuuXUJNiW1Ycs=
···
130
132
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
131
133
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
132
134
github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
135
+
gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b h1:CzigHMRySiX3drau9C6Q5CAbNIApmLdat5jPMqChvDA=
136
+
gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b/go.mod h1:/y/V339mxv2sZmYYR64O07VuCpdNZqCTwO8ZcouTMI8=
137
+
gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02 h1:qwDnMxjkyLmAFgcfgTnfJrmYKWhHnci3GjDqcZp1M3Q=
138
+
gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02/go.mod h1:JTnUj0mpYiAsuZLmKjTx/ex3AtMowcCgnE7YNyCEP0I=
133
139
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 h1:aFJWCqJMNjENlcleuuOkGAPH82y0yULBScfXcIEdS24=
134
140
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1/go.mod h1:sEGXWArGqc3tVa+ekntsN65DmVbVeW+7lTKTjZF3/Fo=
135
141
go.opentelemetry.io/otel v1.21.0 h1:hzLeKBZEL7Okw2mGzZ0cc4k/A7Fta0uoPgaJCr8fsFc=
···
187
193
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
188
194
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
189
195
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
196
+
golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
197
+
golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
190
198
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
191
199
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
192
200
golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+59
jetstream.go
+59
jetstream.go
···
1
+
package main
2
+
3
+
import (
4
+
"context"
5
+
"log/slog"
6
+
"time"
7
+
8
+
"github.com/bluesky-social/jetstream/pkg/client"
9
+
"github.com/bluesky-social/jetstream/pkg/client/schedulers/sequential"
10
+
"github.com/bluesky-social/jetstream/pkg/models"
11
+
)
12
+
13
+
type HandleEvent func(context.Context, *models.Event) error
14
+
15
+
func startJetstreamLoop(logger *slog.Logger, outStream **client.Client, name string, handleEvent HandleEvent, optsFn func() models.SubscriberOptionsUpdatePayload) {
16
+
for {
17
+
stream, startFn, err := startJetstreamClient(name, optsFn(), handleEvent)
18
+
*outStream = stream
19
+
if startFn != nil {
20
+
err = startFn()
21
+
}
22
+
if err != nil {
23
+
logger.Error("stream failed", "name", name, "error", err)
24
+
}
25
+
}
26
+
}
27
+
28
+
func startJetstreamClient(name string, opts models.SubscriberOptionsUpdatePayload, handleEvent HandleEvent) (*client.Client, func() error, error) {
29
+
ctx := context.Background()
30
+
31
+
config := client.DefaultClientConfig()
32
+
config.WebsocketURL = "wss://jetstream1.us-west.bsky.network/subscribe"
33
+
config.Compress = true
34
+
config.WantedCollections = opts.WantedCollections
35
+
config.WantedDids = opts.WantedDIDs
36
+
config.RequireHello = len(config.WantedDids) == 0
37
+
38
+
scheduler := sequential.NewScheduler(name, logger, handleEvent)
39
+
40
+
c, err := client.NewClient(config, logger, scheduler)
41
+
if err != nil {
42
+
logger.Error("failed to create jetstream client", "name", name, "error", err)
43
+
return nil, nil, err
44
+
}
45
+
46
+
startFn := func() error {
47
+
cursor := time.Now().UnixMicro()
48
+
49
+
logger.Info("starting jetstream client", "name", name, "collections", opts.WantedCollections, "wanted_dids", len(opts.WantedDIDs))
50
+
if err := c.ConnectAndRead(ctx, &cursor); err != nil {
51
+
logger.Error("jetstream client failed", "name", name, "error", err)
52
+
return err
53
+
}
54
+
55
+
return nil
56
+
}
57
+
58
+
return c, startFn, nil
59
+
}
+87
-178
main.go
+87
-178
main.go
···
3
3
import (
4
4
"context"
5
5
"encoding/json"
6
-
"fmt"
7
6
"log"
8
7
"log/slog"
9
8
"net/http"
10
9
"sync"
11
-
"time"
12
10
13
-
"github.com/bluesky-social/indigo/api/atproto"
14
11
"github.com/bluesky-social/indigo/api/bsky"
15
12
"github.com/bluesky-social/indigo/xrpc"
16
13
"github.com/bluesky-social/jetstream/pkg/client"
17
-
"github.com/bluesky-social/jetstream/pkg/client/schedulers/sequential"
18
14
"github.com/bluesky-social/jetstream/pkg/models"
19
15
"github.com/gorilla/mux"
20
16
"github.com/gorilla/websocket"
···
24
20
25
21
// Data structures
26
22
type SubscriberData struct {
27
-
DID string
28
-
Conn *websocket.Conn
29
-
Follows Set[string]
30
-
Reposts Set[string]
23
+
DID string
24
+
Conn *websocket.Conn
25
+
ListenTo Set[string]
26
+
Reposts Set[string]
31
27
}
32
28
33
29
type NotificationMessage struct {
···
44
40
likeStream *client.Client
45
41
subscriberStream *client.Client
46
42
47
-
xrpcClient *xrpc.Client
48
-
49
43
upgrader = websocket.Upgrader{
50
44
CheckOrigin: func(r *http.Request) bool {
51
45
return true
···
55
49
logger *slog.Logger
56
50
)
57
51
58
-
func main() {
59
-
logger = slog.Default()
52
+
func getFollowsDids() []string {
53
+
subscribersMux.RLock()
54
+
defer subscribersMux.RUnlock()
60
55
61
-
xrpcClient = &xrpc.Client{
62
-
Client: &http.Client{
63
-
Timeout: 30 * time.Second,
64
-
},
65
-
Host: "https://bsky.social",
56
+
var dids []string
57
+
for _, subscriber := range subscribers {
58
+
for follow, _ := range subscriber.ListenTo {
59
+
dids = append(dids, follow)
60
+
}
66
61
}
67
62
68
-
if err := initializeJetstreams(); err != nil {
69
-
log.Fatalf("cannot start jetstream: %s", err)
63
+
return dids
64
+
}
65
+
66
+
func getSubscriberDids() []string {
67
+
subscribersMux.RLock()
68
+
defer subscribersMux.RUnlock()
69
+
70
+
var dids []string
71
+
for did := range subscribers {
72
+
dids = append(dids, did)
70
73
}
71
74
75
+
return dids
76
+
}
77
+
78
+
func main() {
79
+
logger = slog.Default()
80
+
81
+
go likeStreamLoop(logger)
82
+
go subscriberStreamLoop(logger)
83
+
72
84
r := mux.NewRouter()
73
85
r.HandleFunc("/subscribe/{did}", handleSubscribe).Methods("GET")
74
86
···
82
94
vars := mux.Vars(r)
83
95
did := vars["did"]
84
96
97
+
logger = logger.With("did", did)
98
+
85
99
conn, err := upgrader.Upgrade(w, r, nil)
86
100
if err != nil {
87
101
logger.Error("WebSocket upgrade failed", "error", err)
···
89
103
}
90
104
defer conn.Close()
91
105
92
-
logger.Info("New subscriber", "did", did)
106
+
logger.Info("new subscriber")
93
107
94
-
follows, err := fetchFollows(r.Context(), did)
108
+
pdsURI, err := findUserPDS(r.Context(), did)
95
109
if err != nil {
96
-
logger.Error("Error fetching follows", "did", did, "error", err)
110
+
logger.Error("cant resolve user pds", "error", err)
97
111
return
98
112
}
113
+
logger = logger.With("pds", pdsURI)
99
114
100
-
reposts, err := fetchReposts(r.Context(), did)
115
+
xrpcClient := &xrpc.Client{
116
+
Host: pdsURI,
117
+
}
118
+
// todo: implement skipping fetching follows and allow specifying users to listen to via websocket
119
+
follows, err := fetchFollows(r.Context(), xrpcClient, did)
120
+
if err != nil {
121
+
logger.Error("error fetching follows", "error", err)
122
+
return
123
+
}
124
+
logger.Info("fetched follows")
125
+
reposts, err := fetchReposts(r.Context(), xrpcClient, did)
101
126
if err != nil {
102
-
logger.Error("Error fetching reposts", "did", did, "error", err)
127
+
logger.Error("error fetching reposts", "error", err)
103
128
return
104
129
}
130
+
logger.Info("fetched reposts")
105
131
106
-
// Store subscriber data
107
132
subscriber := &SubscriberData{
108
-
DID: did,
109
-
Conn: conn,
110
-
Follows: follows,
111
-
Reposts: reposts,
133
+
DID: did,
134
+
Conn: conn,
135
+
// use user follows as default listen to
136
+
ListenTo: follows,
137
+
Reposts: reposts,
112
138
}
113
139
114
140
subscribersMux.Lock()
115
141
subscribers[did] = subscriber
116
142
subscribersMux.Unlock()
117
143
updateSubscriberStreamOpts()
144
+
updateLikeStreamOpts()
118
145
// delete subscriber after we are done
119
146
defer func() {
120
147
subscribersMux.Lock()
121
148
delete(subscribers, did)
122
149
subscribersMux.Unlock()
123
150
updateSubscriberStreamOpts()
151
+
updateLikeStreamOpts()
124
152
}()
125
153
126
-
for {
127
-
_, _, err := conn.ReadMessage()
128
-
if err != nil {
129
-
logger.Info("WebSocket connection closed", "did", did, "error", err)
130
-
break
131
-
}
132
-
}
133
-
}
134
-
135
-
func fetchReposts(ctx context.Context, did string) (Set[string], error) {
136
-
all := make(Set[string])
137
-
cursor := ""
154
+
logger.Info("serving subscriber")
138
155
139
156
for {
140
-
out, err := atproto.RepoListRecords(ctx, &xrpc.Client{}, "app.bsky.feed.repost", cursor, 100, did, false)
157
+
_, _, err := conn.ReadMessage()
141
158
if err != nil {
142
-
return nil, err
143
-
}
144
-
145
-
for _, record := range out.Records {
146
-
all[record.Uri] = struct{}{}
147
-
}
148
-
149
-
if out.Cursor == nil || *out.Cursor == "" {
159
+
logger.Info("WebSocket connection closed", "error", err)
150
160
break
151
161
}
152
-
cursor = *out.Cursor
153
162
}
154
-
155
-
return all, nil
156
-
}
157
-
158
-
func fetchFollows(ctx context.Context, did string) (Set[string], error) {
159
-
all := make(Set[string])
160
-
cursor := ""
161
-
162
-
for {
163
-
out, err := bsky.GraphGetFollows(ctx, &xrpc.Client{}, did, cursor, 100)
164
-
if err != nil {
165
-
return nil, err
166
-
}
167
-
168
-
for _, record := range out.Follows {
169
-
all[record.Did] = struct{}{}
170
-
}
171
-
172
-
if out.Cursor == nil || *out.Cursor == "" {
173
-
break
174
-
}
175
-
cursor = *out.Cursor
176
-
}
177
-
178
-
return all, nil
179
-
}
180
-
181
-
func initializeJetstreams() error {
182
-
if err := startLikeClient(); err != nil {
183
-
return fmt.Errorf("like stream: %w", err)
184
-
}
185
-
if err := startSubscriberClient(); err != nil {
186
-
return fmt.Errorf("subscriber stream: %w", err)
187
-
}
188
-
return nil
189
163
}
190
164
191
165
func getLikeStreamOpts() models.SubscriberOptionsUpdatePayload {
···
203
177
}
204
178
205
179
func updateLikeStreamOpts() {
206
-
err := likeStream.SendOptionsUpdate(getLikeStreamOpts())
207
-
if err != nil {
208
-
// reinit like stream
209
-
}
210
-
}
211
-
212
-
func updateSubscriberStreamOpts() {
213
-
err := subscriberStream.SendOptionsUpdate(getSubscriberStreamOpts())
214
-
if err != nil {
215
-
// reinit subscriber stream
216
-
}
217
-
}
218
-
219
-
func startLikeClient() error {
220
180
opts := getLikeStreamOpts()
221
-
if len(opts.WantedDIDs) == 0 {
222
-
return nil // No follows to track
223
-
}
224
-
225
-
handler := &likeHandler{}
226
-
var err error
227
-
likeStream, err = startJetstreamClient("like_tracker", opts, handler.HandleEvent)
181
+
err := likeStream.SendOptionsUpdate(opts)
228
182
if err != nil {
229
-
return err
183
+
logger.Error("couldnt update like stream opts", "error", err)
184
+
return
230
185
}
231
-
232
-
return nil
186
+
logger.Info("updated like stream opts", "requestedDids", len(opts.WantedDIDs))
233
187
}
234
188
235
-
func startSubscriberClient() error {
189
+
func updateSubscriberStreamOpts() {
236
190
opts := getSubscriberStreamOpts()
237
-
if len(opts.WantedDIDs) == 0 {
238
-
return nil // No subscribers to track
239
-
}
240
-
241
-
handler := &subscriberHandler{}
242
-
var err error
243
-
subscriberStream, err = startJetstreamClient("subscriber", opts, handler.HandleEvent)
191
+
err := subscriberStream.SendOptionsUpdate(opts)
244
192
if err != nil {
245
-
return err
193
+
logger.Error("couldnt update subscriber stream opts", "error", err)
194
+
return
246
195
}
247
-
248
-
return nil
196
+
logger.Info("updated subscriber stream opts", "userCount", len(opts.WantedDIDs))
249
197
}
250
198
251
-
func startJetstreamClient(name string, opts models.SubscriberOptionsUpdatePayload, handleEvent func(context.Context, *models.Event) error) (*client.Client, error) {
252
-
ctx := context.Background()
253
-
254
-
config := client.DefaultClientConfig()
255
-
config.WebsocketURL = "wss://jetstream.atproto.tools/subscribe"
256
-
config.Compress = true
257
-
config.WantedCollections = opts.WantedCollections
258
-
config.WantedDids = opts.WantedDIDs
259
-
260
-
scheduler := sequential.NewScheduler(name, logger, handleEvent)
261
-
262
-
c, err := client.NewClient(config, logger, scheduler)
263
-
if err != nil {
264
-
logger.Error("Failed to create client", "name", name, "error", err)
265
-
return nil, err
266
-
}
267
-
268
-
cursor := time.Now().UnixMicro()
269
-
270
-
logger.Info("Starting client", "name", name, "collections", opts.WantedCollections, "wanted_dids", len(opts.WantedDIDs))
271
-
if err := c.ConnectAndRead(ctx, &cursor); err != nil {
272
-
logger.Error("Client failed", "name", name, "error", err)
273
-
return nil, err
274
-
}
275
-
276
-
return c, nil
199
+
func likeStreamLoop(logger *slog.Logger) {
200
+
startJetstreamLoop(logger, &likeStream, "like_tracker", HandleLikeEvent, getLikeStreamOpts)
277
201
}
278
202
279
-
func getFollowsDids() []string {
280
-
subscribersMux.RLock()
281
-
defer subscribersMux.RUnlock()
282
-
283
-
var dids []string
284
-
for _, subscriber := range subscribers {
285
-
for follow, _ := range subscriber.Follows {
286
-
dids = append(dids, follow)
287
-
}
288
-
}
289
-
290
-
return dids
203
+
func subscriberStreamLoop(logger *slog.Logger) {
204
+
startJetstreamLoop(logger, &subscriberStream, "subscriber", HandleSubscriberEvent, getSubscriberStreamOpts)
291
205
}
292
206
293
-
func getSubscriberDids() []string {
294
-
subscribersMux.RLock()
295
-
defer subscribersMux.RUnlock()
296
-
297
-
var dids []string
298
-
for did := range subscribers {
299
-
dids = append(dids, did)
207
+
func HandleLikeEvent(ctx context.Context, event *models.Event) error {
208
+
if event == nil || event.Commit == nil || len(event.Commit.Record) == 0 {
209
+
return nil
300
210
}
301
211
302
-
return dids
303
-
}
304
-
305
-
type likeHandler struct{}
306
-
307
-
func (h *likeHandler) HandleEvent(ctx context.Context, event *models.Event) error {
308
212
var like bsky.FeedLike
309
213
if err := json.Unmarshal(event.Commit.Record, &like); err != nil {
310
214
logger.Error("Failed to unmarshal like", "error", err)
···
334
238
return nil
335
239
}
336
240
337
-
type subscriberHandler struct{}
241
+
func HandleSubscriberEvent(ctx context.Context, event *models.Event) error {
242
+
if event == nil || event.Commit == nil {
243
+
return nil
244
+
}
338
245
339
-
func (h *subscriberHandler) HandleEvent(ctx context.Context, event *models.Event) error {
340
246
switch event.Commit.Collection {
341
247
case "app.bsky.feed.repost":
342
248
modifySubscribersWithEvent(
···
349
255
case "app.bsky.graph.follow":
350
256
modifySubscribersWithEvent(
351
257
event,
352
-
func(s *SubscriberData, r bsky.GraphFollow) { delete(s.Follows, r.Subject) },
258
+
func(s *SubscriberData, r bsky.GraphFollow) { delete(s.ListenTo, r.Subject) },
353
259
func(s *SubscriberData, r bsky.GraphFollow) {
354
-
s.Follows[r.Subject] = struct{}{}
260
+
s.ListenTo[r.Subject] = struct{}{}
355
261
},
356
262
)
357
-
updateLikeStreamOpts()
358
263
}
359
264
360
265
return nil
···
363
268
type ModifyFunc[v any] func(*SubscriberData, v)
364
269
365
270
func modifySubscribersWithEvent[v any](event *models.Event, onDelete ModifyFunc[v], onUpdate ModifyFunc[v]) error {
271
+
if len(event.Commit.Record) == 0 {
272
+
return nil
273
+
}
274
+
366
275
var data v
367
276
if err := json.Unmarshal(event.Commit.Record, &data); err != nil {
368
-
logger.Error("Failed to unmarshal repost", "error", err)
277
+
logger.Error("Failed to unmarshal repost", "error", err, "raw", event.Commit.Record)
369
278
return nil
370
279
}
371
280
+66
xrpc.go
+66
xrpc.go
···
1
+
package main
2
+
3
+
import (
4
+
"context"
5
+
"encoding/json"
6
+
"fmt"
7
+
8
+
"github.com/bluesky-social/indigo/api/atproto"
9
+
"github.com/bluesky-social/indigo/api/bsky"
10
+
"github.com/bluesky-social/indigo/atproto/identity"
11
+
"github.com/bluesky-social/indigo/atproto/syntax"
12
+
"github.com/bluesky-social/indigo/xrpc"
13
+
)
14
+
15
+
func findUserPDS(ctx context.Context, did string) (string, error) {
16
+
id, err := identity.DefaultDirectory().LookupDID(ctx, syntax.DID(did))
17
+
if err != nil {
18
+
return "", err
19
+
}
20
+
pdsURI := id.PDSEndpoint()
21
+
if len(pdsURI) == 0 {
22
+
return "", fmt.Errorf("no PDS URL was found in identity document")
23
+
}
24
+
25
+
return pdsURI, nil
26
+
}
27
+
28
+
func fetchRecords[v any](ctx context.Context, xrpcClient *xrpc.Client, collection, did string, extractFn func(v) string) (Set[string], error) {
29
+
all := make(Set[string])
30
+
cursor := ""
31
+
32
+
for {
33
+
// todo: ratelimits?? idk what this does for those
34
+
out, err := atproto.RepoListRecords(ctx, xrpcClient, collection, cursor, 100, did, false)
35
+
if err != nil {
36
+
return nil, err
37
+
}
38
+
39
+
for _, record := range out.Records {
40
+
raw, _ := record.Value.MarshalJSON()
41
+
var val v
42
+
if err := json.Unmarshal(raw, &val); err != nil {
43
+
return nil, err
44
+
}
45
+
s := extractFn(val)
46
+
if len(s) > 0 {
47
+
all[s] = struct{}{}
48
+
}
49
+
}
50
+
51
+
if out.Cursor == nil || *out.Cursor == "" {
52
+
break
53
+
}
54
+
cursor = *out.Cursor
55
+
}
56
+
57
+
return all, nil
58
+
}
59
+
60
+
func fetchReposts(ctx context.Context, xrpcClient *xrpc.Client, did string) (Set[string], error) {
61
+
return fetchRecords(ctx, xrpcClient, "app.bsky.feed.repost", did, func(v bsky.FeedRepost) string { return v.Subject.Uri })
62
+
}
63
+
64
+
func fetchFollows(ctx context.Context, xrpcClient *xrpc.Client, did string) (Set[string], error) {
65
+
return fetchRecords(ctx, xrpcClient, "app.bsky.graph.follow", did, func(v bsky.GraphFollow) string { return v.Subject })
66
+
}