this repo has no description
huge refactor. split DB out into a store package. renamed things to make more sense
+290
-293
-21
OLDDockerfile
-21
OLDDockerfile
···
1
-
# # Use the Go 1.23 alpine official image
2
-
# # https://hub.docker.com/_/golang
3
-
# FROM golang:1.23-alpine
4
-
5
-
# # Create and change to the app directory.
6
-
# WORKDIR /app
7
-
8
-
# # Copy go mod and sum files
9
-
# COPY go.mod go.sum ./
10
-
11
-
# # Copy local code to the container image.
12
-
# COPY . ./
13
-
14
-
# # Install project dependencies
15
-
# RUN CGO_ENABLED=1 go mod download
16
-
17
-
# # Build the app
18
-
# RUN go build -o app
19
-
20
-
# # Run the service on container startup.
21
-
# ENTRYPOINT ["./app"]
+28
-29
consumer.go
+28
-29
consumer.go
···
2
2
3
3
import (
4
4
"context"
5
-
"database/sql"
5
+
6
6
"encoding/json"
7
7
"fmt"
8
8
"log/slog"
···
14
14
"github.com/bluesky-social/jetstream/pkg/client/schedulers/sequential"
15
15
"github.com/bluesky-social/jetstream/pkg/models"
16
16
"github.com/bugsnag/bugsnag-go/v2"
17
+
"github.com/willdot/bskyfeedgen/store"
17
18
)
18
19
19
20
type consumer struct {
···
38
39
h := &handler{
39
40
seenSeqs: make(map[int64]struct{}),
40
41
feedGenerator: feedGen,
41
-
db: feedGen.db,
42
+
store: *feedGen.store,
42
43
}
43
44
44
45
scheduler := sequential.NewScheduler("jetstream_localdev", logger, h.HandleEvent)
···
64
65
seenSeqs map[int64]struct{}
65
66
highwater int64
66
67
feedGenerator *FeedGenerator
67
-
db *sql.DB
68
+
store store.Store
68
69
}
69
70
70
71
func (h *handler) HandleEvent(ctx context.Context, event *models.Event) error {
···
98
99
return nil
99
100
}
100
101
101
-
parentURI := post.Reply.Parent.Uri
102
+
subscribedPostURI := post.Reply.Parent.Uri
102
103
103
-
// 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
104
+
// look for posts that are "subscribe" so that we can add the post URI to a list of posts we want to find replies for
104
105
if strings.Contains(post.Text, "/subscribe") && event.Did == "did:plc:dadhhalkfcq3gucaq25hjqon" {
105
-
slog.Info("a post that's subscribing to a parent. Adding to parents to look for", "parent URI", parentURI)
106
-
return h.addDidToSubscribedParent(parentURI, event.Did, event.Commit.RKey)
106
+
slog.Info("a post that's subscribing to another post. Adding to posts to look for", "subscribed post URI", subscribedPostURI)
107
+
return h.addDidToSubscribedPost(subscribedPostURI, event.Did, event.Commit.RKey)
107
108
}
108
109
109
110
// see if the post is a reply to a post we are subscribed to
110
-
subscribedDids := h.getSubscribedDidsForParent(parentURI)
111
+
subscribedDids := h.getSubscribedDidsForPost(subscribedPostURI)
111
112
if len(subscribedDids) == 0 {
112
113
return nil
113
114
}
114
115
115
-
slog.Info("post is a reply to a parent that users are subscribed to", "parent URI", parentURI, "dids", subscribedDids, "RKey", event.Commit.RKey)
116
+
slog.Info("post is a reply to a post that users are subscribed to", "subscribed post URI", subscribedPostURI, "dids", subscribedDids, "RKey", event.Commit.RKey)
116
117
117
-
h.feedGenerator.AddToFeedPosts(subscribedDids, parentURI, fmt.Sprintf("at://%s/app.bsky.feed.post/%s", event.Did, event.Commit.RKey))
118
+
replyPostURI := fmt.Sprintf("at://%s/app.bsky.feed.post/%s", event.Did, event.Commit.RKey)
119
+
h.feedGenerator.AddToFeedPosts(subscribedDids, subscribedPostURI, replyPostURI)
118
120
return nil
119
121
}
120
122
···
128
130
return nil
129
131
}
130
132
slog.Info("delete event received", "did", event.Did, "rkey", event.Commit.RKey)
131
-
132
-
parentURI, err := getSubscribingPostParentURI(h.db, event.Did, event.Commit.RKey)
133
+
subscribedPostURI, err := h.store.GetSubscribedPostURI(event.Did, event.Commit.RKey)
133
134
if err != nil {
134
-
slog.Error("get subscribing post parent URI", "error", err, "rkey", event.Commit.RKey, "user DID", event.Did)
135
-
return fmt.Errorf("get subscribing post parent URI: %w", err)
135
+
slog.Error("get subscribed post URI", "error", err, "rkey", event.Commit.RKey, "user DID", event.Did)
136
+
return fmt.Errorf("get subscribed post URI: %w", err)
136
137
}
137
138
138
-
slog.Info("delete parent URI", "parent URI", parentURI, "rkey", event.Commit.RKey)
139
-
140
-
// delete from feeds for the parentURI and the users DID first. This is so that if this fails, it can be tried again and the
139
+
// delete from feeds for the subscribedPostURI and the users DID first. This is so that if this fails, it can be tried again and the
141
140
// subscription will be still there
142
-
err = deleteFeedItemsForParentURIandUserDID(h.db, parentURI, event.Did)
141
+
err = h.store.DeleteFeedItemsForSubscribedPostURIandUserDID(subscribedPostURI, event.Did)
143
142
if err != nil {
144
-
slog.Error("delete feed items for parentURI and user", "error", err, "parentURI", parentURI, "user DID", event.Did)
145
-
return fmt.Errorf("delete feed items for parentURI and user: %w", err)
143
+
slog.Error("delete feed items for subscribedPostURI and user", "error", err, "subscribedPostURI", subscribedPostURI, "user DID", event.Did)
144
+
return fmt.Errorf("delete feed items for subscribedPostURI and user: %w", err)
146
145
}
147
146
148
-
// delete from subscriptions for the parentURI and the users DID now that we have cleaned up the feeds
149
-
err = deleteSubscriptionForUser(h.db, event.Did, parentURI)
147
+
// delete from subscriptions for the postURI and the users DID now that we have cleaned up the feeds
148
+
err = h.store.DeleteSubscriptionForUser(event.Did, subscribedPostURI)
150
149
if err != nil {
151
-
slog.Error("delete subscription for user", "error", err, "parentURI", parentURI, "user DID", event.Did)
150
+
slog.Error("delete subscription for user", "error", err, "subscribedPostURI", subscribedPostURI, "user DID", event.Did)
152
151
return fmt.Errorf("delete subscription and user: %w", err)
153
152
}
154
153
155
154
return nil
156
155
}
157
156
158
-
func (h *handler) addDidToSubscribedParent(parentURI, userDid, rkey string) error {
159
-
err := addSubscriptionForParent(h.db, parentURI, userDid, rkey)
157
+
func (h *handler) addDidToSubscribedPost(subscribedPostURI, userDid, rkey string) error {
158
+
err := h.store.AddSubscriptionForPost(subscribedPostURI, userDid, rkey)
160
159
if err != nil {
161
-
return fmt.Errorf("add subscription for parent: %w", err)
160
+
return fmt.Errorf("add subscription for post: %w", err)
162
161
}
163
162
return nil
164
163
}
165
164
166
-
func (h *handler) getSubscribedDidsForParent(parentURI string) []string {
167
-
dids, err := getSubscriptionsForParent(h.db, parentURI)
165
+
func (h *handler) getSubscribedDidsForPost(postURI string) []string {
166
+
dids, err := h.store.GetSubscriptionsForPost(postURI)
168
167
if err != nil {
169
-
slog.Error("getting subscriptions for parent", "error", err)
168
+
slog.Error("getting subscriptions for post", "error", err)
170
169
bugsnag.Notify(err)
171
170
}
172
171
-226
database.go
-226
database.go
···
1
-
package main
2
-
3
-
import (
4
-
"context"
5
-
"database/sql"
6
-
"errors"
7
-
"fmt"
8
-
"log/slog"
9
-
"os"
10
-
11
-
_ "github.com/glebarez/go-sqlite"
12
-
)
13
-
14
-
func NewDatabase(dbPath string) (*sql.DB, error) {
15
-
err := createDbFile(dbPath)
16
-
if err != nil {
17
-
return nil, fmt.Errorf("create db file: %w", err)
18
-
}
19
-
20
-
db, err := sql.Open("sqlite", dbPath)
21
-
if err != nil {
22
-
return nil, fmt.Errorf("open database: %w", err)
23
-
}
24
-
25
-
err = db.Ping()
26
-
if err != nil {
27
-
return nil, fmt.Errorf("ping db: %w", err)
28
-
}
29
-
30
-
err = createFeedTable(db)
31
-
if err != nil {
32
-
return nil, fmt.Errorf("creating feed table: %w", err)
33
-
}
34
-
35
-
err = createSubscriptionsTable(db)
36
-
if err != nil {
37
-
return nil, fmt.Errorf("creating subscription table: %w", err)
38
-
}
39
-
40
-
return db, nil
41
-
}
42
-
43
-
func createDbFile(dbFilename string) error {
44
-
if _, err := os.Stat(dbFilename); !errors.Is(err, os.ErrNotExist) {
45
-
return nil
46
-
}
47
-
48
-
f, err := os.Create(dbFilename)
49
-
if err != nil {
50
-
return fmt.Errorf("create db file : %w", err)
51
-
}
52
-
f.Close()
53
-
return nil
54
-
}
55
-
56
-
func createFeedTable(db *sql.DB) error {
57
-
createFeedTableSQL := `CREATE TABLE IF NOT EXISTS feed (
58
-
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
59
-
"uri" TEXT,
60
-
"userDID" TEXT,
61
-
"parentURI" TEXT,
62
-
UNIQUE(uri, userDID)
63
-
);`
64
-
65
-
slog.Info("Create feed table...")
66
-
statement, err := db.Prepare(createFeedTableSQL)
67
-
if err != nil {
68
-
return fmt.Errorf("prepare DB statement to create feeds table: %w", err)
69
-
}
70
-
_, err = statement.Exec()
71
-
if err != nil {
72
-
return fmt.Errorf("exec sql statement to create feeds table: %w", err)
73
-
}
74
-
slog.Info("feed table created")
75
-
76
-
return nil
77
-
}
78
-
79
-
func createSubscriptionsTable(db *sql.DB) error {
80
-
createSubscriptionsTableSQL := `CREATE TABLE IF NOT EXISTS subscriptions (
81
-
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
82
-
"parentURI" TEXT,
83
-
"userDID" TEXT,
84
-
"subscriptionRkey" TEXT,
85
-
UNIQUE(parentURI, userDID)
86
-
);`
87
-
88
-
slog.Info("Create subscriptions table...")
89
-
statement, err := db.Prepare(createSubscriptionsTableSQL)
90
-
if err != nil {
91
-
return fmt.Errorf("prepare DB statement to create subscriptions table: %w", err)
92
-
}
93
-
_, err = statement.Exec()
94
-
if err != nil {
95
-
return fmt.Errorf("exec sql statement to create subscriptions table: %w", err)
96
-
}
97
-
slog.Info("subscriptions table created")
98
-
99
-
return nil
100
-
}
101
-
102
-
type feedItem struct {
103
-
ID int
104
-
URI string
105
-
UserDID string
106
-
parentURI string
107
-
}
108
-
109
-
func addFeedItem(_ context.Context, db *sql.DB, feedItem feedItem) error {
110
-
slog.Info("add feed item", "parenturi", feedItem.parentURI, "user did", feedItem.UserDID)
111
-
sql := `INSERT INTO feed (uri, userDID, parentURI) VALUES (?, ?, ?) ON CONFLICT(uri, userDID) DO NOTHING;`
112
-
_, err := db.Exec(sql, feedItem.URI, feedItem.UserDID, feedItem.parentURI)
113
-
if err != nil {
114
-
return fmt.Errorf("exec insert feed item: %w", err)
115
-
}
116
-
return nil
117
-
}
118
-
119
-
func getUsersFeedItems(db *sql.DB, usersDID string) ([]feedItem, error) {
120
-
sql := "SELECT id, uri, userDID FROM feed WHERE userDID = ?;"
121
-
rows, err := db.Query(sql, usersDID)
122
-
if err != nil {
123
-
return nil, fmt.Errorf("run query to get users feed item: %w", err)
124
-
}
125
-
defer rows.Close()
126
-
127
-
feedItems := make([]feedItem, 0)
128
-
for rows.Next() {
129
-
var feedItem feedItem
130
-
if err := rows.Scan(&feedItem.ID, &feedItem.URI, &feedItem.UserDID); err != nil {
131
-
return nil, fmt.Errorf("scan row: %w", err)
132
-
}
133
-
feedItems = append(feedItems, feedItem)
134
-
}
135
-
136
-
return feedItems, nil
137
-
}
138
-
139
-
func deleteFeedItemsForParentURIandUserDID(db *sql.DB, parentURI, userDID string) error {
140
-
slog.Info("delete feed", "parent uri", parentURI, "userdid", userDID)
141
-
142
-
sql := "DELETE FROM feed WHERE parentURI = ? AND userDID = ?;"
143
-
statement, err := db.Prepare(sql)
144
-
if err != nil {
145
-
return fmt.Errorf("prepare delete feed items: %w", err)
146
-
}
147
-
res, err := statement.Exec(parentURI, userDID)
148
-
if err != nil {
149
-
return fmt.Errorf("exec delete feed items: %w", err)
150
-
}
151
-
152
-
n, _ := res.RowsAffected()
153
-
154
-
slog.Info("delete feed res", "affected rows", n)
155
-
return nil
156
-
}
157
-
158
-
type subscription struct {
159
-
ID int
160
-
ParentURI string
161
-
UserDID string
162
-
SubecriptionRkey string
163
-
}
164
-
165
-
func getSubscriptionsForParent(db *sql.DB, parentURI string) ([]string, error) {
166
-
sql := "SELECT id, parentURI, userDID FROM subscriptions WHERE parentURI = ?"
167
-
rows, err := db.Query(sql, parentURI)
168
-
if err != nil {
169
-
return nil, fmt.Errorf("run query to get subscriptions: %w", err)
170
-
}
171
-
defer rows.Close()
172
-
173
-
dids := make([]string, 0)
174
-
for rows.Next() {
175
-
var subscription subscription
176
-
if err := rows.Scan(&subscription.ID, &subscription.ParentURI, &subscription.UserDID); err != nil {
177
-
return nil, fmt.Errorf("scan row: %w", err)
178
-
}
179
-
dids = append(dids, subscription.UserDID)
180
-
}
181
-
182
-
return dids, nil
183
-
}
184
-
185
-
// urh
186
-
func addSubscriptionForParent(db *sql.DB, parentURI, userDid, subscriptionRkey string) error {
187
-
sql := `INSERT INTO subscriptions (parentURI, userDID, subscriptionRkey) VALUES (?, ?, ?) ON CONFLICT(parentURI, userDID) DO NOTHING;`
188
-
_, err := db.Exec(sql, parentURI, userDid, subscriptionRkey)
189
-
if err != nil {
190
-
return fmt.Errorf("exec insert subscrptions: %w", err)
191
-
}
192
-
return nil
193
-
}
194
-
195
-
func getSubscribingPostParentURI(db *sql.DB, userDID, rkey string) (string, error) {
196
-
slog.Info("params", "rkey", rkey, "did", userDID)
197
-
sql := "SELECT id, parentURI FROM subscriptions WHERE subscriptionRkey = ? AND userDID = ?;"
198
-
rows, err := db.Query(sql, rkey, userDID)
199
-
if err != nil {
200
-
return "", fmt.Errorf("run query to get subscribing post parent URI: %w", err)
201
-
}
202
-
defer rows.Close()
203
-
204
-
parentURI := ""
205
-
for rows.Next() {
206
-
var subscription subscription
207
-
if err := rows.Scan(&subscription.ID, &subscription.ParentURI); err != nil {
208
-
return "", fmt.Errorf("scan row: %w", err)
209
-
}
210
-
211
-
slog.Info("record", "val", subscription)
212
-
213
-
parentURI = subscription.ParentURI
214
-
break
215
-
}
216
-
return parentURI, nil
217
-
}
218
-
219
-
func deleteSubscriptionForUser(db *sql.DB, userDID, parentURI string) error {
220
-
sql := "DELETE FROM subscriptions WHERE parentURI = ? AND userDID = ?;"
221
-
_, err := db.Exec(sql, parentURI, userDID)
222
-
if err != nil {
223
-
return fmt.Errorf("exec delete subscription for user: %w", err)
224
-
}
225
-
return nil
226
-
}
+13
-13
feed.go
+13
-13
feed.go
···
2
2
3
3
import (
4
4
"context"
5
-
"database/sql"
6
5
"fmt"
7
6
"log/slog"
8
7
9
8
"github.com/bugsnag/bugsnag-go/v2"
9
+
"github.com/willdot/bskyfeedgen/store"
10
10
)
11
11
12
12
type FeedGenerator struct {
13
-
db *sql.DB
13
+
store *store.Store
14
14
}
15
15
16
-
func NewFeedGenerator(db *sql.DB) *FeedGenerator {
16
+
func NewFeedGenerator(store *store.Store) *FeedGenerator {
17
17
return &FeedGenerator{
18
-
db: db,
18
+
store: store,
19
19
}
20
20
}
21
21
···
24
24
Feed: make([]FeedItem, 0, 0),
25
25
}
26
26
27
-
usersFeed, err := getUsersFeedItems(f.db, userDID)
27
+
usersFeed, err := f.store.GetUsersFeedItems(userDID)
28
28
if err != nil {
29
29
return resp, fmt.Errorf("get users feed items from DB: %w", err)
30
30
}
···
32
32
feedItems := make([]FeedItem, 0, len(usersFeed))
33
33
for _, post := range usersFeed {
34
34
feedItems = append(feedItems, FeedItem{
35
-
Post: post.URI,
35
+
Post: post.ReplyURI,
36
36
})
37
37
}
38
38
···
42
42
return resp, nil
43
43
}
44
44
45
-
func (f *FeedGenerator) AddToFeedPosts(usersDids []string, parentURI, postURI string) {
45
+
func (f *FeedGenerator) AddToFeedPosts(usersDids []string, subscribedPostURI, replyPostURI string) {
46
46
for _, did := range usersDids {
47
-
feedItem := feedItem{
48
-
URI: postURI,
49
-
UserDID: did,
50
-
parentURI: parentURI,
47
+
feedItem := store.FeedItem{
48
+
ReplyURI: replyPostURI,
49
+
UserDID: did,
50
+
SubscribedPostURI: subscribedPostURI,
51
51
}
52
-
err := addFeedItem(context.Background(), f.db, feedItem)
52
+
err := f.store.AddFeedItem(feedItem)
53
53
if err != nil {
54
-
slog.Error("add users feed item", "error", err, "did", did, "uri", postURI)
54
+
slog.Error("add users feed item", "error", err, "did", did, "reply post URI", replyPostURI)
55
55
bugsnag.Notify(err)
56
56
continue
57
57
}
+6
-4
main.go
+6
-4
main.go
···
13
13
14
14
"github.com/avast/retry-go/v4"
15
15
"github.com/bugsnag/bugsnag-go/v2"
16
+
"github.com/willdot/bskyfeedgen/store"
16
17
)
17
18
18
19
const (
···
44
45
return
45
46
}
46
47
dbFilename := path.Join(dbMountPath, "database.db")
47
-
db, err := NewDatabase(dbFilename)
48
+
49
+
store, err := store.New(dbFilename)
48
50
if err != nil {
49
-
slog.Error("create new database", "error", err)
51
+
slog.Error("create new store", "error", err)
50
52
bugsnag.Notify(err)
51
53
return
52
54
}
53
-
defer db.Close()
55
+
defer store.Close()
54
56
55
-
feeder := NewFeedGenerator(db)
57
+
feeder := NewFeedGenerator(store)
56
58
57
59
feedDidBase := os.Getenv("FEED_DID_BASE")
58
60
if feedDidBase == "" {
+64
store/database.go
+64
store/database.go
···
1
+
package store
2
+
3
+
import (
4
+
"database/sql"
5
+
"errors"
6
+
"fmt"
7
+
"log/slog"
8
+
"os"
9
+
10
+
_ "github.com/glebarez/go-sqlite"
11
+
)
12
+
13
+
type Store struct {
14
+
db *sql.DB
15
+
}
16
+
17
+
func New(dbPath string) (*Store, error) {
18
+
err := createDbFile(dbPath)
19
+
if err != nil {
20
+
return nil, fmt.Errorf("create db file: %w", err)
21
+
}
22
+
23
+
db, err := sql.Open("sqlite", dbPath)
24
+
if err != nil {
25
+
return nil, fmt.Errorf("open database: %w", err)
26
+
}
27
+
28
+
err = db.Ping()
29
+
if err != nil {
30
+
return nil, fmt.Errorf("ping db: %w", err)
31
+
}
32
+
33
+
err = createFeedTable(db)
34
+
if err != nil {
35
+
return nil, fmt.Errorf("creating feed table: %w", err)
36
+
}
37
+
38
+
err = createSubscriptionsTable(db)
39
+
if err != nil {
40
+
return nil, fmt.Errorf("creating subscription table: %w", err)
41
+
}
42
+
43
+
return &Store{db: db}, nil
44
+
}
45
+
46
+
func (s *Store) Close() {
47
+
err := s.db.Close()
48
+
if err != nil {
49
+
slog.Error("failed to close db", "error", err)
50
+
}
51
+
}
52
+
53
+
func createDbFile(dbFilename string) error {
54
+
if _, err := os.Stat(dbFilename); !errors.Is(err, os.ErrNotExist) {
55
+
return nil
56
+
}
57
+
58
+
f, err := os.Create(dbFilename)
59
+
if err != nil {
60
+
return fmt.Errorf("create db file : %w", err)
61
+
}
62
+
f.Close()
63
+
return nil
64
+
}
+83
store/feed.go
+83
store/feed.go
···
1
+
package store
2
+
3
+
import (
4
+
"database/sql"
5
+
"fmt"
6
+
"log/slog"
7
+
)
8
+
9
+
func createFeedTable(db *sql.DB) error {
10
+
createFeedTableSQL := `CREATE TABLE IF NOT EXISTS feed (
11
+
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
12
+
"replyURI" TEXT,
13
+
"userDID" TEXT,
14
+
"subscribedPostURI" TEXT,
15
+
UNIQUE(replyURI, userDID)
16
+
);`
17
+
18
+
slog.Info("Create feed table...")
19
+
statement, err := db.Prepare(createFeedTableSQL)
20
+
if err != nil {
21
+
return fmt.Errorf("prepare DB statement to create feeds table: %w", err)
22
+
}
23
+
_, err = statement.Exec()
24
+
if err != nil {
25
+
return fmt.Errorf("exec sql statement to create feeds table: %w", err)
26
+
}
27
+
slog.Info("feed table created")
28
+
29
+
return nil
30
+
}
31
+
32
+
type FeedItem struct {
33
+
ID int
34
+
ReplyURI string
35
+
UserDID string
36
+
SubscribedPostURI string
37
+
}
38
+
39
+
func (s *Store) AddFeedItem(feedItem FeedItem) error {
40
+
sql := `INSERT INTO feed (replyURI, userDID, subscribedPostURI) VALUES (?, ?, ?) ON CONFLICT(replyURI, userDID) DO NOTHING;`
41
+
_, err := s.db.Exec(sql, feedItem.ReplyURI, feedItem.UserDID, feedItem.SubscribedPostURI)
42
+
if err != nil {
43
+
return fmt.Errorf("exec insert feed item: %w", err)
44
+
}
45
+
return nil
46
+
}
47
+
48
+
func (s *Store) GetUsersFeedItems(usersDID string) ([]FeedItem, error) {
49
+
sql := "SELECT id, replyURI, userDID FROM feed WHERE userDID = ?;"
50
+
rows, err := s.db.Query(sql, usersDID)
51
+
if err != nil {
52
+
return nil, fmt.Errorf("run query to get users feed item: %w", err)
53
+
}
54
+
defer rows.Close()
55
+
56
+
feedItems := make([]FeedItem, 0)
57
+
for rows.Next() {
58
+
var feedItem FeedItem
59
+
if err := rows.Scan(&feedItem.ID, &feedItem.ReplyURI, &feedItem.UserDID); err != nil {
60
+
return nil, fmt.Errorf("scan row: %w", err)
61
+
}
62
+
feedItems = append(feedItems, feedItem)
63
+
}
64
+
65
+
return feedItems, nil
66
+
}
67
+
68
+
func (s *Store) DeleteFeedItemsForSubscribedPostURIandUserDID(subscribedPostURI, userDID string) error {
69
+
sql := "DELETE FROM feed WHERE subscribedPostURI = ? AND userDID = ?;"
70
+
statement, err := s.db.Prepare(sql)
71
+
if err != nil {
72
+
return fmt.Errorf("prepare delete feed items: %w", err)
73
+
}
74
+
res, err := statement.Exec(subscribedPostURI, userDID)
75
+
if err != nil {
76
+
return fmt.Errorf("exec delete feed items: %w", err)
77
+
}
78
+
79
+
n, _ := res.RowsAffected()
80
+
81
+
slog.Info("delete feed res", "affected rows", n)
82
+
return nil
83
+
}
+96
store/subscription.go
+96
store/subscription.go
···
1
+
package store
2
+
3
+
import (
4
+
"database/sql"
5
+
"fmt"
6
+
"log/slog"
7
+
)
8
+
9
+
func createSubscriptionsTable(db *sql.DB) error {
10
+
createSubscriptionsTableSQL := `CREATE TABLE IF NOT EXISTS subscriptions (
11
+
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
12
+
"subscribedPostURI" TEXT,
13
+
"userDID" TEXT,
14
+
"subscriptionPostRkey" TEXT,
15
+
UNIQUE(subscribedPostURI, userDID)
16
+
);`
17
+
18
+
slog.Info("Create subscriptions table...")
19
+
statement, err := db.Prepare(createSubscriptionsTableSQL)
20
+
if err != nil {
21
+
return fmt.Errorf("prepare DB statement to create subscriptions table: %w", err)
22
+
}
23
+
_, err = statement.Exec()
24
+
if err != nil {
25
+
return fmt.Errorf("exec sql statement to create subscriptions table: %w", err)
26
+
}
27
+
slog.Info("subscriptions table created")
28
+
29
+
return nil
30
+
}
31
+
32
+
type Subscription struct {
33
+
ID int
34
+
SubscribedPostURI string
35
+
UserDID string
36
+
SubscriptionPostRkey string
37
+
}
38
+
39
+
func (s *Store) GetSubscriptionsForPost(postURI string) ([]string, error) {
40
+
sql := "SELECT userDID FROM subscriptions WHERE subscribedPostURI = ?"
41
+
rows, err := s.db.Query(sql, postURI)
42
+
if err != nil {
43
+
return nil, fmt.Errorf("run query to get subscriptions: %w", err)
44
+
}
45
+
defer rows.Close()
46
+
47
+
dids := make([]string, 0)
48
+
for rows.Next() {
49
+
var subscription Subscription
50
+
if err := rows.Scan(&subscription.UserDID); err != nil {
51
+
return nil, fmt.Errorf("scan row: %w", err)
52
+
}
53
+
dids = append(dids, subscription.UserDID)
54
+
}
55
+
56
+
return dids, nil
57
+
}
58
+
59
+
func (s *Store) AddSubscriptionForPost(subscribedPostURI, userDid, subscriptionRkey string) error {
60
+
sql := `INSERT INTO subscriptions (subscribedPostURI, userDID, subscriptionRkey) VALUES (?, ?, ?) ON CONFLICT(subscribedPostURI, userDID) DO NOTHING;`
61
+
_, err := s.db.Exec(sql, subscribedPostURI, userDid, subscriptionRkey)
62
+
if err != nil {
63
+
return fmt.Errorf("exec insert subscrptions: %w", err)
64
+
}
65
+
return nil
66
+
}
67
+
68
+
func (s *Store) GetSubscribedPostURI(userDID, rkey string) (string, error) {
69
+
sql := "SELECT id, subscribedPostURI FROM subscriptions WHERE subscriptionRkey = ? AND userDID = ?;"
70
+
rows, err := s.db.Query(sql, rkey, userDID)
71
+
if err != nil {
72
+
return "", fmt.Errorf("run query to get subscribed post URI: %w", err)
73
+
}
74
+
defer rows.Close()
75
+
76
+
subscribedPostURI := ""
77
+
for rows.Next() {
78
+
var subscription Subscription
79
+
if err := rows.Scan(&subscription.ID, &subscription.SubscribedPostURI); err != nil {
80
+
return "", fmt.Errorf("scan row: %w", err)
81
+
}
82
+
83
+
subscribedPostURI = subscription.SubscribedPostURI
84
+
break
85
+
}
86
+
return subscribedPostURI, nil
87
+
}
88
+
89
+
func (s *Store) DeleteSubscriptionForUser(userDID, postURI string) error {
90
+
sql := "DELETE FROM subscriptions WHERE subscribedPostURI = ? AND userDID = ?;"
91
+
_, err := s.db.Exec(sql, postURI, userDID)
92
+
if err != nil {
93
+
return fmt.Errorf("exec delete subscription for user: %w", err)
94
+
}
95
+
return nil
96
+
}