···
subscribers = hashmap.New[string, *SubscriberData]()
38
+
listeningTo = hashmap.New[string, *hashmap.Map[string, *SubscriberData]]()
likeStream *client.Client
subscriberStream *client.Client
···
logger.Info("fetched reposts")
125
-
subscriber := &SubscriberData{
126
+
sd := &SubscriberData{
// use user follows as default listen to
···
133
-
subscribers.Set(did, subscriber)
134
+
subscribers.Set(sd.DID, sd)
135
+
for listenDid := range sd.ListenTo {
136
+
listenTo(sd, listenDid)
updateSubscriberStreamOpts()
// delete subscriber after we are done
138
-
subscribers.Del(did)
143
+
for listenDid := range sd.ListenTo {
144
+
stopListeningTo(sd.DID, listenDid)
146
+
subscribers.Del(sd.DID)
updateSubscriberStreamOpts()
···
163
+
func listenTo(sd *SubscriberData, did string) {
164
+
targetDids, _ := listeningTo.GetOrInsert(did, hashmap.New[string, *SubscriberData]())
165
+
targetDids.Insert(sd.DID, sd)
168
+
func stopListeningTo(subscriberDid, did string) {
169
+
if targetDids, exists := listeningTo.Get(did); exists {
170
+
targetDids.Del(subscriberDid)
func getLikeStreamOpts() models.SubscriberOptionsUpdatePayload {
return models.SubscriberOptionsUpdatePayload{
WantedCollections: []string{"app.bsky.feed.like"},
157
-
WantedDIDs: getFollowsDids(),
177
+
// WantedDIDs: getFollowsDids(),
···
221
+
// skip handling event if its not from a source we are listening to
222
+
targets, exists := listeningTo.Get(event.Did)
if err := json.Unmarshal(event.Commit.Record, &like); err != nil {
203
-
logger.Error("Failed to unmarshal like", "error", err)
229
+
logger.Error("failed to unmarshal like", "error", err)
207
-
subscribers.Range(func(s string, sd *SubscriberData) bool {
233
+
targets.Range(func(s string, sd *SubscriberData) bool {
for repostURI, _ := range sd.Reposts {
// (un)liked a post the subscriber reposted
if like.Subject.Uri == repostURI {
···
if err := sd.Conn.WriteJSON(notification); err != nil {
218
-
logger.Error("Failed to send notification", "subscriber", sd.DID, "error", err)
244
+
logger.Error("failed to send notification", "subscriber", sd.DID, "error", err)
···
case "app.bsky.graph.follow":
modifySubscribersWithEvent(
245
-
func(s *SubscriberData, r bsky.GraphFollow) { delete(s.ListenTo, r.Subject) },
271
+
func(s *SubscriberData, r bsky.GraphFollow) {
272
+
delete(s.ListenTo, r.Subject)
273
+
stopListeningTo(s.DID, r.Subject)
func(s *SubscriberData, r bsky.GraphFollow) {
s.ListenTo[r.Subject] = struct{}{}
277
+
listenTo(s, r.Subject)
ptr.pet