this repo has no description
feat: use transactions
+30
-3
+30
-3
pkg/mostliked/handler.go
+30
-3
pkg/mostliked/handler.go
···
88
88
89
89
go trimPostsTable(ctx, queries)
90
90
91
+
var (
92
+
dbTx *sql.Tx
93
+
queriesTx *db.Queries
94
+
txOpen bool
95
+
err error
96
+
eventCount int
97
+
)
98
+
91
99
for evt := range events {
100
+
if !txOpen {
101
+
dbTx, err = dbCnx.BeginTx(ctx, nil)
102
+
if err != nil {
103
+
log.Fatal(err)
104
+
}
105
+
txOpen = true
106
+
queriesTx = queries.WithTx(dbTx)
107
+
}
108
+
92
109
var like appbsky.FeedLike
93
110
var event jetstream.Event
94
111
if err := json.Unmarshal(evt, &event); err != nil {
···
142
159
}
143
160
drafts.Delete(like.Subject.Uri)
144
161
log.Println("storing", like.Subject.Uri, "in database")
145
-
err := queries.InsertPost(ctx, db.InsertPostParams{
162
+
err := queriesTx.InsertPost(ctx, db.InsertPostParams{
146
163
Uri: like.Subject.Uri,
147
164
CreateTs: draftPost.Created,
148
165
Likes: draftPost.Likes,
···
151
168
log.Println("error inserting post")
152
169
}
153
170
for _, lang := range draftPost.Languages {
154
-
err = queries.InsertLang(ctx, db.InsertLangParams{
171
+
err = queriesTx.InsertLang(ctx, db.InsertLangParams{
155
172
Uri: like.Subject.Uri,
156
173
Lang: strings.ToLower(lang.IsoCode639_1().String()),
157
174
})
···
160
177
}
161
178
}
162
179
} else {
163
-
err := queries.UpdateLikes(ctx, like.Subject.Uri)
180
+
err := queriesTx.UpdateLikes(ctx, like.Subject.Uri)
164
181
if err != nil {
165
182
log.Println("error updating likes")
166
183
}
184
+
}
185
+
186
+
eventCount += 1
187
+
if eventCount % 500 == 0 {
188
+
if err := dbTx.Commit(); err != nil {
189
+
log.Fatalf("commit failed: %v\n", err)
190
+
} else {
191
+
txOpen = false
192
+
}
193
+
log.Println("db committed")
167
194
}
168
195
}
169
196
}