its for when you want to get like notifications for your reposts
feat: let users specify if they want to handle specifying what dids to listen to
+95
-44
+95
-44
main.go
+95
-44
main.go
···
18
18
19
19
type Set[T comparable] map[T]struct{}
20
20
21
+
const ListenTypeNone = "none"
22
+
const ListenTypeFollows = "follows"
23
+
21
24
type SubscriberData struct {
22
-
DID string
23
-
Conn *websocket.Conn
24
-
ListenTo Set[string]
25
-
Reposts Set[string]
25
+
DID string
26
+
Conn *websocket.Conn
27
+
ListenType string
28
+
ListenTo Set[string]
29
+
Reposts Set[string]
26
30
}
27
31
28
32
type NotificationMessage struct {
···
31
35
RepostURI string `json:"repost_uri"`
32
36
}
33
37
38
+
type SubscriberMessage struct {
39
+
Type string `json:"type"`
40
+
Content json.RawMessage `json:"content"`
41
+
}
42
+
43
+
type SubscriberUpdateListenTo struct {
44
+
ListenTo []string `json:"listen_to"`
45
+
}
46
+
34
47
var (
35
48
// storing the subscriber data in both Should Be Fine
36
49
// we dont modify subscriber data at the same time in two places
···
72
85
func main() {
73
86
logger = slog.Default()
74
87
75
-
go likeStreamLoop(logger)
76
-
go subscriberStreamLoop(logger)
88
+
go startJetstreamLoop(logger, &likeStream, "like_tracker", HandleLikeEvent, getLikeStreamOpts)
89
+
go startJetstreamLoop(logger, &subscriberStream, "subscriber", HandleSubscriberEvent, getSubscriberStreamOpts)
77
90
78
91
r := mux.NewRouter()
79
92
r.HandleFunc("/subscribe/{did}", handleSubscribe).Methods("GET")
80
93
81
-
log.Println("Server starting on :8080")
94
+
log.Println("server starting on :8080")
82
95
if err := http.ListenAndServe(":8080", r); err != nil {
83
96
log.Fatalf("error while serving: %s", err)
84
97
}
···
88
101
vars := mux.Vars(r)
89
102
did := vars["did"]
90
103
104
+
query := r.URL.Query()
105
+
// "follows", everything else is considered as "none"
106
+
listenType := query.Get("listenTo")
107
+
if len(listenType) == 0 {
108
+
listenType = ListenTypeFollows
109
+
}
110
+
91
111
logger = logger.With("did", did)
92
112
93
113
conn, err := upgrader.Upgrade(w, r, nil)
···
109
129
xrpcClient := &xrpc.Client{
110
130
Host: pdsURI,
111
131
}
112
-
// todo: implement skipping fetching follows and allow specifying users to listen to via websocket
113
-
follows, err := fetchFollows(r.Context(), xrpcClient, did)
114
-
if err != nil {
115
-
logger.Error("error fetching follows", "error", err)
132
+
133
+
var subbedTo Set[string]
134
+
135
+
switch listenType {
136
+
case ListenTypeFollows:
137
+
follows, err := fetchFollows(r.Context(), xrpcClient, did)
138
+
if err != nil {
139
+
logger.Error("error fetching follows", "error", err)
140
+
return
141
+
}
142
+
logger.Info("fetched follows")
143
+
subbedTo = follows
144
+
case ListenTypeNone:
145
+
subbedTo = make(Set[string])
146
+
default:
147
+
logger.Error("invalid listen type", "requestedType", listenType)
116
148
return
117
149
}
118
-
logger.Info("fetched follows")
150
+
119
151
reposts, err := fetchReposts(r.Context(), xrpcClient, did)
120
152
if err != nil {
121
153
logger.Error("error fetching reposts", "error", err)
···
124
156
logger.Info("fetched reposts")
125
157
126
158
sd := &SubscriberData{
127
-
DID: did,
128
-
Conn: conn,
129
-
// use user follows as default listen to
130
-
ListenTo: follows,
131
-
Reposts: reposts,
159
+
DID: did,
160
+
Conn: conn,
161
+
ListenType: listenType,
162
+
ListenTo: subbedTo,
163
+
Reposts: reposts,
132
164
}
133
165
134
166
subscribers.Set(sd.DID, sd)
···
152
184
logger.Info("serving subscriber")
153
185
154
186
for {
155
-
_, _, err := conn.ReadMessage()
187
+
var msg SubscriberMessage
188
+
err := conn.ReadJSON(&msg)
156
189
if err != nil {
157
190
logger.Info("WebSocket connection closed", "error", err)
158
191
break
159
192
}
193
+
switch msg.Type {
194
+
case "update_listen_to":
195
+
// only allow this if we arent managing listen to
196
+
if sd.ListenType != ListenTypeNone {
197
+
continue
198
+
}
199
+
200
+
var innerMsg SubscriberUpdateListenTo
201
+
if err := json.Unmarshal(msg.Content, &innerMsg); err != nil {
202
+
logger.Info("invalid message", "error", err)
203
+
break
204
+
}
205
+
// remove all current listens and add the ones the user requested
206
+
for listenDid := range sd.ListenTo {
207
+
stopListeningTo(sd.DID, listenDid)
208
+
delete(sd.ListenTo, listenDid)
209
+
}
210
+
for _, listenDid := range innerMsg.ListenTo {
211
+
sd.ListenTo[listenDid] = struct{}{}
212
+
listenTo(sd, listenDid)
213
+
}
214
+
}
160
215
}
161
216
}
162
217
···
192
247
return
193
248
}
194
249
logger.Info("updated subscriber stream opts", "userCount", len(opts.WantedDIDs))
195
-
}
196
-
197
-
func likeStreamLoop(logger *slog.Logger) {
198
-
startJetstreamLoop(logger, &likeStream, "like_tracker", HandleLikeEvent, getLikeStreamOpts)
199
-
}
200
-
201
-
func subscriberStreamLoop(logger *slog.Logger) {
202
-
startJetstreamLoop(logger, &subscriberStream, "subscriber", HandleSubscriberEvent, getSubscriberStreamOpts)
203
250
}
204
251
205
252
func HandleLikeEvent(ctx context.Context, event *models.Event) error {
···
249
296
case "app.bsky.feed.repost":
250
297
modifySubscribersWithEvent(
251
298
event,
252
-
func(s *SubscriberData, r bsky.FeedRepost) { delete(s.Reposts, r.Subject.Uri) },
253
-
func(s *SubscriberData, r bsky.FeedRepost) {
254
-
s.Reposts[r.Subject.Uri] = struct{}{}
299
+
func(s *SubscriberData, r bsky.FeedRepost, deleted bool) {
300
+
if deleted {
301
+
delete(s.Reposts, r.Subject.Uri)
302
+
} else {
303
+
s.Reposts[r.Subject.Uri] = struct{}{}
304
+
}
255
305
},
256
306
)
257
307
case "app.bsky.graph.follow":
258
308
modifySubscribersWithEvent(
259
309
event,
260
-
func(s *SubscriberData, r bsky.GraphFollow) {
261
-
delete(s.ListenTo, r.Subject)
262
-
stopListeningTo(s.DID, r.Subject)
263
-
},
264
-
func(s *SubscriberData, r bsky.GraphFollow) {
265
-
s.ListenTo[r.Subject] = struct{}{}
266
-
listenTo(s, r.Subject)
310
+
func(s *SubscriberData, r bsky.GraphFollow, deleted bool) {
311
+
// if we arent managing then we dont need to update anything
312
+
if s.ListenType != ListenTypeFollows {
313
+
return
314
+
}
315
+
if deleted {
316
+
stopListeningTo(s.DID, r.Subject)
317
+
delete(s.ListenTo, r.Subject)
318
+
} else {
319
+
s.ListenTo[r.Subject] = struct{}{}
320
+
listenTo(s, r.Subject)
321
+
}
267
322
},
268
323
)
269
324
}
···
271
326
return nil
272
327
}
273
328
274
-
type ModifyFunc[v any] func(*SubscriberData, v)
329
+
type ModifyFunc[v any] func(*SubscriberData, v, bool)
275
330
276
-
func modifySubscribersWithEvent[v any](event *models.Event, onDelete ModifyFunc[v], onUpdate ModifyFunc[v]) error {
331
+
func modifySubscribersWithEvent[v any](event *models.Event, handle ModifyFunc[v]) error {
277
332
if len(event.Commit.Record) == 0 {
278
333
return nil
279
334
}
280
335
281
336
var data v
282
337
if err := json.Unmarshal(event.Commit.Record, &data); err != nil {
283
-
logger.Error("Failed to unmarshal repost", "error", err, "raw", event.Commit.Record)
338
+
logger.Error("failed to unmarshal repost", "error", err, "raw", event.Commit.Record)
284
339
return nil
285
340
}
286
341
287
342
if subscriber, exists := subscribers.Get(event.Did); exists {
288
-
if event.Commit.Operation == models.CommitOperationDelete {
289
-
onDelete(subscriber, data)
290
-
} else {
291
-
onUpdate(subscriber, data)
292
-
}
343
+
handle(subscriber, data, event.Commit.Operation == models.CommitOperationDelete)
293
344
}
294
345
295
346
return nil