this repo has no description
some improvements because it worked and hopefully I haven't borked it now
+36
-35
+26
-17
consumer.go
+26
-17
consumer.go
···
6
6
"fmt"
7
7
"log"
8
8
"log/slog"
9
-
"slices"
9
+
"sync"
10
10
"time"
11
11
12
12
apibsky "github.com/bluesky-social/indigo/api/bsky"
···
28
28
"app.bsky.feed.post",
29
29
}
30
30
cfg.WantedDids = []string{
31
-
"did:plc:dadhhalkfcq3gucaq25hjqon",
31
+
// "did:plc:dadhhalkfcq3gucaq25hjqon",
32
32
}
33
33
return &consumer{
34
34
cfg: cfg,
···
37
37
38
38
func (con *consumer) Consume(ctx context.Context, feedGen *FeedGenerator, logger *slog.Logger) error {
39
39
h := &handler{
40
-
seenSeqs: make(map[int64]struct{}),
41
-
feedGenerator: feedGen,
40
+
seenSeqs: make(map[int64]struct{}),
41
+
feedGenerator: feedGen,
42
+
parentsToLookFor: make(map[string]struct{}),
42
43
}
43
44
44
45
scheduler := sequential.NewScheduler("jetstream_localdev", logger, h.HandleEvent)
···
63
64
seenSeqs map[int64]struct{}
64
65
highwater int64
65
66
feedGenerator *FeedGenerator
66
-
parentsToLookFor []string
67
+
mu sync.Mutex
68
+
parentsToLookFor map[string]struct{}
67
69
}
68
70
69
71
func (h *handler) HandleEvent(ctx context.Context, event *models.Event) error {
70
72
// Unmarshal the record if there is one
71
-
if event.Commit != nil && (event.Commit.Operation == models.CommitOperationCreate || event.Commit.Operation == models.CommitOperationUpdate) {
73
+
if event.Commit == nil {
74
+
return nil
75
+
}
76
+
if event.Commit.Operation == models.CommitOperationCreate || event.Commit.Operation == models.CommitOperationUpdate {
72
77
switch event.Commit.Collection {
73
78
case "app.bsky.feed.post":
74
79
var post apibsky.FeedPost
···
76
81
return fmt.Errorf("failed to unmarshal post: %w", err)
77
82
}
78
83
84
+
// we only care about posts that have parents which are replies
85
+
if post.Reply == nil || post.Reply.Parent == nil || post.Reply.Parent.Uri == "" {
86
+
return nil
87
+
}
88
+
89
+
h.mu.Lock()
90
+
defer h.mu.Unlock()
91
+
79
92
// look for posts where I've "subsribed" so that we can add the parent URI to a list of replies to that parent to look for
80
-
if post.Text == "/subscribe" {
81
-
if post.Reply != nil && post.Reply.Parent != nil && post.Reply.Parent.Uri != "" {
82
-
slog.Info("it's a reply with a parent! Adding to parents to look for", "parent URI", post.Reply.Parent.Uri)
83
-
h.parentsToLookFor = append(h.parentsToLookFor, post.Reply.Parent.Uri)
84
-
}
93
+
if post.Text == "/subscribe" && event.Did == "did:plc:dadhhalkfcq3gucaq25hjqon" {
94
+
slog.Info("it's a reply with a parent! Adding to parents to look for", "parent URI", post.Reply.Parent.Uri)
95
+
h.parentsToLookFor[post.Reply.Parent.Uri] = struct{}{}
85
96
return nil
86
97
}
87
98
88
-
if post.Reply != nil && post.Reply.Parent != nil && post.Reply.Parent.Uri != "" {
89
-
if slices.Contains(h.parentsToLookFor, post.Reply.Parent.Uri) {
90
-
slog.Info("Event", "data", fmt.Sprintf("%+v", event.Commit))
91
-
h.feedGenerator.AddToFeed(fmt.Sprintf("at://%s/app.bsky.feed.post/%s", event.Did, event.Commit.RKey))
92
-
}
99
+
// see if the post is a reply to a post we are subscribed to
100
+
if _, ok := h.parentsToLookFor[post.Reply.Parent.Uri]; ok {
101
+
slog.Info("post is a reply to a parent we are subscribed to", "parent URI", post.Reply.Parent.Uri, "did", event.Did, "RKey", event.Commit.RKey)
102
+
h.feedGenerator.AddToFeedPosts(fmt.Sprintf("at://%s/app.bsky.feed.post/%s", event.Did, event.Commit.RKey))
93
103
}
94
104
}
95
105
}
96
-
97
106
return nil
98
107
}
+10
-18
feed.go
+10
-18
feed.go
···
7
7
8
8
type FeedGenerator struct {
9
9
mu sync.Mutex
10
-
feeds []string
10
+
posts map[string]struct{}
11
11
}
12
12
13
-
// TODO: pass in feeds or something
14
13
func NewFeedGenerator() *FeedGenerator {
15
-
return &FeedGenerator{}
14
+
return &FeedGenerator{
15
+
posts: make(map[string]struct{}),
16
+
}
16
17
}
17
18
18
19
func (f *FeedGenerator) GetFeed(ctx context.Context, feed, cursor string, limit int) (*FeedReponse, error) {
19
-
// TODO: get something from a database instead
20
-
// resp := &FeedReponse{
21
-
// Feed: []FeedItem{
22
-
// {
23
-
// Post: "at://did:plc:dadhhalkfcq3gucaq25hjqon/app.bsky.feed.post/3l7j5ma2si42r",
24
-
// FeedContext: "this is some additional context",
25
-
// },
26
-
// },
27
-
// Cursor: "",
28
-
// }
29
20
f.mu.Lock()
30
21
defer f.mu.Unlock()
31
-
feedItems := make([]FeedItem, 0, len(f.feeds))
32
-
for _, feed := range f.feeds {
22
+
23
+
feedItems := make([]FeedItem, 0, len(f.posts))
24
+
for post := range f.posts {
33
25
feedItems = append(feedItems, FeedItem{
34
-
Post: feed,
26
+
Post: post,
35
27
})
36
28
}
37
29
···
43
35
return resp, nil
44
36
}
45
37
46
-
func (f *FeedGenerator) AddToFeed(postURI string) {
38
+
func (f *FeedGenerator) AddToFeedPosts(postURI string) {
47
39
f.mu.Lock()
48
40
defer f.mu.Unlock()
49
41
// TODO: store this in DB instead
50
-
f.feeds = append(f.feeds, postURI)
42
+
f.posts[postURI] = struct{}{}
51
43
}