A social RSS reader built on the AT Protocol.
glean.at
glean
atproto
atmosphere
rss
feed
social
app
Implement orphan cleanup for sync operations
+355
-77
+4
internal/atproto/stream_handler.go
+4
internal/atproto/stream_handler.go
···
202
202
return h.articles.UpdateAnnotation(ctx, a)
203
203
204
204
case actionDelete:
205
+
// Margin notes are converted to annotations; their URI is tracked
206
+
// by sync so orphan cleanup handles deletion during backfill.
205
207
}
206
208
return nil
207
209
}
···
252
254
return err
253
255
254
256
case actionDelete:
257
+
// Skyreader subscriptions are imported into glean subscriptions;
258
+
// sync's orphan cleanup handles deletion during backfill.
255
259
}
256
260
return nil
257
261
}
+73
-70
internal/atproto/sync.go
+73
-70
internal/atproto/sync.go
···
33
33
func (s *Sync) Run(ctx context.Context, userDID string) error {
34
34
s.logger.Info("syncing from PDS", "did", userDID)
35
35
36
-
if err := s.syncCollection(ctx, userDID, CollectionSubscription, s.batchReconcileSubscriptions); err != nil {
36
+
if err := s.syncSubscriptions(ctx, userDID); err != nil {
37
37
s.logger.Error("sync subscriptions failed", "error", err, "did", userDID)
38
38
}
39
-
if err := s.syncCollection(ctx, userDID, CollectionSkyreaderSubscription, s.batchReconcileSkyreaderSubscriptions); err != nil {
40
-
s.logger.Error("sync skyreader subscriptions failed", "error", err, "did", userDID)
41
-
}
42
-
if err := s.syncCollection(ctx, userDID, CollectionLike, s.batchReconcileLikes); err != nil {
39
+
if err := s.syncLikes(ctx, userDID); err != nil {
43
40
s.logger.Error("sync likes failed", "error", err, "did", userDID)
44
41
}
45
-
if err := s.syncCollection(ctx, userDID, CollectionAnnotation, s.batchReconcileAnnotations); err != nil {
42
+
if err := s.syncAnnotations(ctx, userDID); err != nil {
46
43
s.logger.Error("sync annotations failed", "error", err, "did", userDID)
47
44
}
48
-
if err := s.syncCollection(ctx, userDID, CollectionMarginNote, s.batchReconcileMarginNotes); err != nil {
49
-
s.logger.Error("sync margin notes failed", "error", err, "did", userDID)
50
-
}
51
45
if err := s.syncFollows(ctx, userDID); err != nil {
52
46
s.logger.Error("sync follows failed", "error", err, "did", userDID)
53
47
}
···
55
49
return nil
56
50
}
57
51
58
-
func (s *Sync) syncCollection(ctx context.Context, userDID, collection string, fn func(ctx context.Context, userDID string, records []Record) error) error {
52
+
func (s *Sync) listRecords(ctx context.Context, userDID, collection string) ([]Record, error) {
59
53
var allRecords []Record
60
54
cursor := ""
61
55
for {
62
56
records, next, err := s.client.ListRecords(ctx, userDID, collection, 100, cursor)
63
57
if err != nil {
64
-
return err
58
+
return nil, err
65
59
}
66
60
allRecords = append(allRecords, records...)
67
61
if next == "" || len(records) == 0 {
···
69
63
}
70
64
cursor = next
71
65
}
72
-
if len(allRecords) == 0 {
73
-
return nil
74
-
}
75
-
return fn(ctx, userDID, allRecords)
66
+
return allRecords, nil
76
67
}
77
68
78
-
func (s *Sync) batchReconcileSubscriptions(ctx context.Context, userDID string, records []Record) error {
69
+
func (s *Sync) syncSubscriptions(ctx context.Context, userDID string) error {
70
+
gleanRecs, err := s.listRecords(ctx, userDID, CollectionSubscription)
71
+
if err != nil {
72
+
return err
73
+
}
74
+
skyRecs, err := s.listRecords(ctx, userDID, CollectionSkyreaderSubscription)
75
+
if err != nil {
76
+
return err
77
+
}
78
+
79
79
var feeds []*db.Feed
80
80
var subs []db.SubData
81
+
activeFeedURLs := make(map[string]bool)
81
82
82
-
for _, r := range records {
83
+
for _, r := range gleanRecs {
83
84
var rec SubscriptionRecord
84
85
if err := json.Unmarshal(r.Value, &rec); err != nil {
85
86
continue
···
87
88
if rec.FeedURL == "" {
88
89
continue
89
90
}
91
+
activeFeedURLs[rec.FeedURL] = true
90
92
feeds = append(feeds, &db.Feed{FeedURL: rec.FeedURL, Title: db.NullStr(rec.Title)})
91
93
subs = append(subs, db.SubData{
92
94
FeedURL: rec.FeedURL,
···
97
99
})
98
100
}
99
101
100
-
if len(feeds) > 0 {
101
-
if err := s.articles.BatchUpsertFeeds(ctx, feeds); err != nil {
102
-
return fmt.Errorf("upsert feeds: %w", err)
103
-
}
104
-
}
105
-
return s.articles.BatchReconcileSubscriptions(ctx, userDID, subs)
106
-
}
107
-
108
-
func (s *Sync) batchReconcileSkyreaderSubscriptions(ctx context.Context, userDID string, records []Record) error {
109
-
var feeds []*db.Feed
110
-
var subs []db.SubData
111
-
112
-
for _, r := range records {
102
+
for _, r := range skyRecs {
113
103
var rec SkyreaderSubscriptionRecord
114
104
if err := json.Unmarshal(r.Value, &rec); err != nil {
115
105
continue
···
117
107
if rec.FeedURL == "" {
118
108
continue
119
109
}
110
+
activeFeedURLs[rec.FeedURL] = true
120
111
feeds = append(feeds, &db.Feed{FeedURL: rec.FeedURL, Title: db.NullStr(rec.Title), SiteURL: db.NullStr(rec.SiteURL)})
121
112
subs = append(subs, db.SubData{
122
113
FeedURL: rec.FeedURL,
···
128
119
129
120
if len(feeds) > 0 {
130
121
if err := s.articles.BatchUpsertFeeds(ctx, feeds); err != nil {
131
-
return fmt.Errorf("failed to upsert feeds: %w", err)
122
+
return fmt.Errorf("upsert feeds: %w", err)
132
123
}
133
124
}
134
-
135
-
return s.articles.BatchReconcileSubscriptions(ctx, userDID, subs)
125
+
if err := s.articles.BatchReconcileSubscriptions(ctx, userDID, subs); err != nil {
126
+
return err
127
+
}
128
+
return s.articles.DeleteOrphanedSubscriptions(ctx, userDID, activeFeedURLs)
136
129
}
137
130
138
-
func (s *Sync) batchReconcileLikes(ctx context.Context, userDID string, records []Record) error {
139
-
var likes []*db.Like
131
+
func (s *Sync) syncLikes(ctx context.Context, userDID string) error {
132
+
records, err := s.listRecords(ctx, userDID, CollectionLike)
133
+
if err != nil {
134
+
return err
135
+
}
140
136
137
+
var likes []*db.Like
138
+
activeURIs := make(map[string]bool)
141
139
for _, r := range records {
142
140
var rec LikeRecord
143
141
if err := json.Unmarshal(r.Value, &rec); err != nil {
···
146
144
if rec.FeedURL == "" || rec.ArticleURL == "" {
147
145
continue
148
146
}
147
+
activeURIs[r.URI] = true
149
148
t, _ := time.Parse(time.RFC3339, rec.CreatedAt)
150
149
likes = append(likes, &db.Like{
151
150
URI: r.URI,
···
157
156
})
158
157
}
159
158
160
-
return s.articles.BatchCreateLikes(ctx, likes)
159
+
if err := s.articles.BatchCreateLikes(ctx, likes); err != nil {
160
+
return err
161
+
}
162
+
return s.articles.DeleteOrphanedLikes(ctx, userDID, activeURIs)
161
163
}
162
164
163
-
func (s *Sync) batchReconcileAnnotations(ctx context.Context, userDID string, records []Record) error {
165
+
func (s *Sync) syncAnnotations(ctx context.Context, userDID string) error {
166
+
annRecs, err := s.listRecords(ctx, userDID, CollectionAnnotation)
167
+
if err != nil {
168
+
return err
169
+
}
170
+
marginRecs, err := s.listRecords(ctx, userDID, CollectionMarginNote)
171
+
if err != nil {
172
+
return err
173
+
}
174
+
164
175
var annotations []*db.Annotation
176
+
activeURIs := make(map[string]bool)
165
177
166
-
for _, r := range records {
178
+
for _, r := range annRecs {
167
179
var rec AnnotationRecord
168
180
if err := json.Unmarshal(r.Value, &rec); err != nil {
169
181
continue
···
171
183
if rec.FeedURL == "" || rec.ArticleURL == "" {
172
184
continue
173
185
}
186
+
activeURIs[r.URI] = true
174
187
t, _ := time.Parse(time.RFC3339, rec.CreatedAt)
175
188
a := &db.Annotation{
176
189
URI: r.URI,
···
189
202
annotations = append(annotations, a)
190
203
}
191
204
192
-
return s.articles.BatchCreateAnnotations(ctx, annotations)
193
-
}
194
-
195
-
func (s *Sync) batchReconcileMarginNotes(ctx context.Context, userDID string, records []Record) error {
196
-
var annotations []*db.Annotation
197
-
198
-
for _, r := range records {
205
+
for _, r := range marginRecs {
199
206
var rec MarginNoteRecord
200
207
if err := json.Unmarshal(r.Value, &rec); err != nil {
201
208
continue
···
204
211
if articleURL == "" {
205
212
continue
206
213
}
214
+
activeURIs[r.URI] = true
207
215
208
216
feedURL := ""
209
217
if article, err := s.articles.GetArticleByURL(ctx, articleURL); err == nil {
···
224
232
})
225
233
}
226
234
227
-
return s.articles.BatchCreateAnnotations(ctx, annotations)
235
+
if err := s.articles.BatchCreateAnnotations(ctx, annotations); err != nil {
236
+
return err
237
+
}
238
+
return s.articles.DeleteOrphanedAnnotations(ctx, userDID, activeURIs)
228
239
}
229
240
230
241
func (s *Sync) syncFollows(ctx context.Context, userDID string) error {
231
242
activeFollows := make(map[string]db.Follow)
232
243
233
244
for _, collection := range []string{CollectionBskyFollow, CollectionTangledFollow} {
234
-
cursor := ""
235
-
for {
236
-
records, next, err := s.client.ListRecords(ctx, userDID, collection, 100, cursor)
237
-
if err != nil {
238
-
return err
245
+
records, err := s.listRecords(ctx, userDID, collection)
246
+
if err != nil {
247
+
return err
248
+
}
249
+
250
+
for _, r := range records {
251
+
var rec FollowRecord
252
+
if err := json.Unmarshal(r.Value, &rec); err != nil {
253
+
continue
239
254
}
240
-
241
-
for _, r := range records {
242
-
var rec FollowRecord
243
-
if err := json.Unmarshal(r.Value, &rec); err != nil {
244
-
continue
245
-
}
246
-
if rec.Subject == "" {
247
-
continue
248
-
}
249
-
250
-
t, _ := time.Parse(time.RFC3339, rec.CreatedAt)
251
-
activeFollows[rec.Subject] = db.Follow{
252
-
URI: db.NullStr(r.URI),
253
-
CID: db.NullStr(r.CID),
254
-
FollowedAt: db.NullTime(t),
255
-
}
255
+
if rec.Subject == "" {
256
+
continue
256
257
}
257
258
258
-
if next == "" || len(records) == 0 {
259
-
break
259
+
t, _ := time.Parse(time.RFC3339, rec.CreatedAt)
260
+
activeFollows[rec.Subject] = db.Follow{
261
+
URI: db.NullStr(r.URI),
262
+
CID: db.NullStr(r.CID),
263
+
FollowedAt: db.NullTime(t),
260
264
}
261
-
cursor = next
262
265
}
263
266
}
264
267
+133
internal/db/batch_test.go
+133
internal/db/batch_test.go
···
292
292
err = dbs.Articles.BatchCreateAnnotations(ctx, annotations)
293
293
assert.NilError(t, err)
294
294
}
295
+
296
+
func TestBatchReconcileSubscriptions_DeletesOrphaned(t *testing.T) {
297
+
ctx := context.Background()
298
+
dbs := setupTestDB(t)
299
+
userDID := seedSubscriptionData(t, ctx, dbs)
300
+
301
+
_ = dbs.Articles.UpsertFeed(ctx, &Feed{FeedURL: "https://a.com/feed.xml", Title: NullStr("A")})
302
+
_ = dbs.Articles.UpsertFeed(ctx, &Feed{FeedURL: "https://b.com/feed.xml", Title: NullStr("B")})
303
+
_ = dbs.Articles.UpsertFeed(ctx, &Feed{FeedURL: "https://c.com/feed.xml", Title: NullStr("C")})
304
+
305
+
subs := []SubData{
306
+
{FeedURL: "https://a.com/feed.xml", Title: "A", URI: "at://uri-a", CID: "cid-a"},
307
+
{FeedURL: "https://b.com/feed.xml", Title: "B", URI: "at://uri-b", CID: "cid-b"},
308
+
{FeedURL: "https://c.com/feed.xml", Title: "C", URI: "at://uri-c", CID: "cid-c"},
309
+
}
310
+
err := dbs.Articles.BatchReconcileSubscriptions(ctx, userDID, subs)
311
+
assert.NilError(t, err)
312
+
313
+
err = dbs.Articles.DeleteOrphanedSubscriptions(ctx, userDID, map[string]bool{
314
+
"https://a.com/feed.xml": true,
315
+
"https://c.com/feed.xml": true,
316
+
})
317
+
assert.NilError(t, err)
318
+
319
+
list, err := dbs.Articles.ListSubscriptions(ctx, userDID, "", 10, 0)
320
+
assert.NilError(t, err)
321
+
assert.Equal(t, len(list), 2)
322
+
323
+
feedURLs := map[string]bool{}
324
+
for _, s := range list {
325
+
feedURLs[s.FeedURL] = true
326
+
}
327
+
assert.Assert(t, feedURLs["https://a.com/feed.xml"])
328
+
assert.Assert(t, feedURLs["https://c.com/feed.xml"])
329
+
330
+
f, err := dbs.Articles.GetFeed(ctx, "https://b.com/feed.xml")
331
+
assert.NilError(t, err)
332
+
assert.Equal(t, f.SubscriberCount, 0)
333
+
}
334
+
335
+
func TestBatchReconcileSubscriptions_PreservesLocalOnly(t *testing.T) {
336
+
ctx := context.Background()
337
+
dbs := setupTestDB(t)
338
+
userDID := seedSubscriptionData(t, ctx, dbs)
339
+
340
+
_ = dbs.Articles.UpsertFeed(ctx, &Feed{FeedURL: "https://a.com/feed.xml"})
341
+
_ = dbs.Articles.UpsertFeed(ctx, &Feed{FeedURL: "https://b.com/feed.xml"})
342
+
343
+
err := dbs.Articles.CreateSubscription(ctx, userDID, "https://a.com/feed.xml", "A", "", "at://uri-a", "cid")
344
+
assert.NilError(t, err)
345
+
err = dbs.Articles.CreateSubscription(ctx, userDID, "https://b.com/feed.xml", "B", "", "", "")
346
+
assert.NilError(t, err)
347
+
348
+
err = dbs.Articles.DeleteOrphanedSubscriptions(ctx, userDID, map[string]bool{
349
+
"https://a.com/feed.xml": true,
350
+
})
351
+
assert.NilError(t, err)
352
+
353
+
list, err := dbs.Articles.ListSubscriptions(ctx, userDID, "", 10, 0)
354
+
assert.NilError(t, err)
355
+
assert.Equal(t, len(list), 2)
356
+
}
357
+
358
+
func TestBatchReconcileSubscriptions_DeletesAllWhenEmpty(t *testing.T) {
359
+
ctx := context.Background()
360
+
dbs := setupTestDB(t)
361
+
userDID := seedSubscriptionData(t, ctx, dbs)
362
+
363
+
_ = dbs.Articles.UpsertFeed(ctx, &Feed{FeedURL: "https://a.com/feed.xml"})
364
+
_ = dbs.Articles.UpsertFeed(ctx, &Feed{FeedURL: "https://b.com/feed.xml"})
365
+
366
+
subs := []SubData{
367
+
{FeedURL: "https://a.com/feed.xml", URI: "at://uri-a", CID: "cid-a"},
368
+
{FeedURL: "https://b.com/feed.xml", URI: "at://uri-b", CID: "cid-b"},
369
+
}
370
+
err := dbs.Articles.BatchReconcileSubscriptions(ctx, userDID, subs)
371
+
assert.NilError(t, err)
372
+
373
+
err = dbs.Articles.DeleteOrphanedSubscriptions(ctx, userDID, map[string]bool{})
374
+
assert.NilError(t, err)
375
+
376
+
list, err := dbs.Articles.ListSubscriptions(ctx, userDID, "", 10, 0)
377
+
assert.NilError(t, err)
378
+
assert.Equal(t, len(list), 0)
379
+
}
380
+
381
+
func TestDeleteOrphanedLikes_RemovesOrphaned(t *testing.T) {
382
+
ctx := context.Background()
383
+
dbs := setupTestDB(t)
384
+
385
+
now := NullTime(time.Now())
386
+
likes := []*Like{
387
+
{URI: "at://like1", AuthorDID: "did:test:u1", FeedURL: "https://a.com/feed", ArticleURL: "https://a.com/1", CreatedAt: now, CID: NullStr("cid1")},
388
+
{URI: "at://like2", AuthorDID: "did:test:u1", FeedURL: "https://a.com/feed", ArticleURL: "https://a.com/2", CreatedAt: now, CID: NullStr("cid2")},
389
+
}
390
+
err := dbs.Articles.BatchCreateLikes(ctx, likes)
391
+
assert.NilError(t, err)
392
+
393
+
err = dbs.Articles.DeleteOrphanedLikes(ctx, "did:test:u1", map[string]bool{"at://like1": true})
394
+
assert.NilError(t, err)
395
+
396
+
exists, err := dbs.Articles.HasLiked(ctx, "did:test:u1", "https://a.com/feed", "https://a.com/1")
397
+
assert.NilError(t, err)
398
+
assert.Equal(t, exists, true)
399
+
400
+
exists, err = dbs.Articles.HasLiked(ctx, "did:test:u1", "https://a.com/feed", "https://a.com/2")
401
+
assert.NilError(t, err)
402
+
assert.Equal(t, exists, false)
403
+
}
404
+
405
+
func TestDeleteOrphanedAnnotations_RemovesOrphaned(t *testing.T) {
406
+
ctx := context.Background()
407
+
dbs := setupTestDB(t)
408
+
409
+
now := NullTime(time.Now())
410
+
annotations := []*Annotation{
411
+
{URI: "at://ann1", AuthorDID: "did:test:u1", FeedURL: "https://a.com/feed", ArticleURL: "https://a.com/1", Note: NullStr("Keep"), CreatedAt: now},
412
+
{URI: "at://ann2", AuthorDID: "did:test:u1", FeedURL: "https://a.com/feed", ArticleURL: "https://a.com/2", Note: NullStr("Remove"), CreatedAt: now},
413
+
}
414
+
err := dbs.Articles.BatchCreateAnnotations(ctx, annotations)
415
+
assert.NilError(t, err)
416
+
417
+
err = dbs.Articles.DeleteOrphanedAnnotations(ctx, "did:test:u1", map[string]bool{"at://ann1": true})
418
+
assert.NilError(t, err)
419
+
420
+
exists, err := dbs.Articles.AnnotationExists(ctx, "at://ann1")
421
+
assert.NilError(t, err)
422
+
assert.Equal(t, exists, true)
423
+
424
+
exists, err = dbs.Articles.AnnotationExists(ctx, "at://ann2")
425
+
assert.NilError(t, err)
426
+
assert.Equal(t, exists, false)
427
+
}
+60
-1
internal/db/feed.go
+60
-1
internal/db/feed.go
···
119
119
return err
120
120
}
121
121
122
+
func (s *ArticleStore) incrementSubscriberCount(ctx context.Context, feedURL string) error {
123
+
_, err := s.db.ExecContext(ctx, `
124
+
UPDATE articles.feeds SET subscriber_count = subscriber_count + 1 WHERE feed_url = ?
125
+
`, feedURL)
126
+
return err
127
+
}
128
+
122
129
func (s *ArticleStore) CreateSubscription(ctx context.Context, userDID, feedURL, title, category, uri, cid string) error {
123
130
existing, err := s.GetSubscription(ctx, userDID, feedURL)
124
131
if err != nil || existing == nil {
125
-
return s.BatchReconcileSubscriptions(ctx, userDID, []SubData{{FeedURL: feedURL, Title: title, Category: category, URI: uri, CID: cid}})
132
+
result, err := s.db.ExecContext(ctx, `
133
+
INSERT OR IGNORE INTO articles.subscriptions (user_did, feed_url, title, category, uri, cid)
134
+
VALUES (?, ?, ?, ?, ?, ?)
135
+
`, userDID, feedURL, nilIfEmpty(title), nilIfEmpty(category), nilIfEmpty(uri), nilIfEmpty(cid))
136
+
if err != nil {
137
+
return err
138
+
}
139
+
if n, _ := result.RowsAffected(); n > 0 {
140
+
return s.incrementSubscriberCount(ctx, feedURL)
141
+
}
142
+
return nil
126
143
}
127
144
128
145
unchanged := existing.FeedTitle == title && existing.Category.String == category && existing.CID.String == cid
···
462
479
}
463
480
}
464
481
}
482
+
return tx.Commit()
483
+
}
484
+
485
+
func (s *ArticleStore) DeleteOrphanedSubscriptions(ctx context.Context, userDID string, activeFeedURLs map[string]bool) error {
486
+
rows, err := s.db.QueryContext(ctx,
487
+
`SELECT feed_url FROM articles.subscriptions WHERE user_did = ? AND uri IS NOT NULL AND uri != ''`,
488
+
userDID)
489
+
if err != nil {
490
+
return err
491
+
}
492
+
var toDelete []string
493
+
for rows.Next() {
494
+
var feedURL string
495
+
if err := rows.Scan(&feedURL); err != nil {
496
+
rows.Close()
497
+
return err
498
+
}
499
+
if !activeFeedURLs[feedURL] {
500
+
toDelete = append(toDelete, feedURL)
501
+
}
502
+
}
503
+
rows.Close()
504
+
505
+
if len(toDelete) == 0 {
506
+
return nil
507
+
}
508
+
509
+
tx, err := s.db.BeginTx(ctx, nil)
510
+
if err != nil {
511
+
return err
512
+
}
513
+
defer tx.Rollback()
514
+
515
+
for _, feedURL := range toDelete {
516
+
if _, err := tx.ExecContext(ctx, `DELETE FROM articles.subscriptions WHERE user_did = ? AND feed_url = ?`, userDID, feedURL); err != nil {
517
+
return err
518
+
}
519
+
if _, err := tx.ExecContext(ctx, `UPDATE articles.feeds SET subscriber_count = MAX(subscriber_count - 1, 0) WHERE feed_url = ?`, feedURL); err != nil {
520
+
return err
521
+
}
522
+
}
523
+
465
524
return tx.Commit()
466
525
}
467
526