A social RSS reader built on the AT Protocol.
glean.at
glean
atproto
atmosphere
rss
feed
social
app
Refactor database schema and sessions implementation
+129
-373
+9
-9
internal/atproto/xrpc.go
+9
-9
internal/atproto/xrpc.go
···
29
29
30
30
query := `
31
31
SELECT s.id, f.feed_url, COALESCE(s.title, f.title), s.category, s.added_at
32
-
FROM subscriptions s
33
-
JOIN feeds f ON s.feed_url = f.feed_url
32
+
FROM articles.subscriptions s
33
+
JOIN articles.feeds f ON s.feed_url = f.feed_url
34
34
WHERE s.user_did = ?`
35
35
args := []any{repo}
36
36
···
95
95
query := `
96
96
SELECT a.uri, a.cid, u.did, a.feed_url, a.article_url,
97
97
a.quote, a.note, a.tags, a.rating, a.created_at
98
-
FROM annotations a
98
+
FROM articles.annotations a
99
99
JOIN users u ON a.author_did = u.did
100
100
WHERE 1=1`
101
101
args := []any{}
···
180
180
181
181
query := `
182
182
SELECT l.uri, l.cid, u.did, l.feed_url, l.article_url, l.created_at
183
-
FROM likes l
183
+
FROM articles.likes l
184
184
JOIN users u ON l.author_did = u.did
185
185
WHERE 1=1`
186
186
args := []any{}
···
250
250
251
251
query := `
252
252
SELECT l.feed_url, l.article_url, a.title, COUNT(*) as like_count
253
-
FROM likes l
254
-
LEFT JOIN articles a ON l.article_url = a.url
253
+
FROM articles.likes l
254
+
LEFT JOIN articles.articles a ON l.article_url = a.url
255
255
WHERE 1=1`
256
256
args := []any{}
257
257
···
372
372
query := `
373
373
SELECT u.did, COUNT(s.id) as subscription_count
374
374
FROM users u
375
-
LEFT JOIN subscriptions s ON u.did = s.user_did
375
+
LEFT JOIN articles.subscriptions s ON u.did = s.user_did
376
376
WHERE u.did IN (` + strings.Join(placeholders, ",") + `)`
377
377
378
378
if cursor != "" {
···
417
417
}
418
418
subRows, err := h.db.QueryContext(r.Context(), `
419
419
SELECT s.user_did, s.feed_url, COALESCE(s.title, f.title), s.category
420
-
FROM subscriptions s
421
-
JOIN feeds f ON s.feed_url = f.feed_url
420
+
FROM articles.subscriptions s
421
+
JOIN articles.feeds f ON s.feed_url = f.feed_url
422
422
WHERE s.user_did IN (`+strings.Join(ph, ",")+`)
423
423
ORDER BY s.user_did, s.added_at DESC
424
424
`, args...)
-226
internal/db/db.go
-226
internal/db/db.go
···
57
57
type DB struct {
58
58
*sql.DB
59
59
}
60
-
61
-
func (d *DB) Close() error {
62
-
return d.DB.Close()
63
-
}
64
-
65
-
func Open(path string) (*DB, error) {
66
-
db, err := sql.Open("sqlite3_glean", path+"?cache=shared&"+DSN)
67
-
if err != nil {
68
-
return nil, err
69
-
}
70
-
71
-
db.SetMaxOpenConns(10)
72
-
db.SetMaxIdleConns(5)
73
-
db.SetConnMaxLifetime(30 * time.Minute)
74
-
75
-
wrapped := &DB{db}
76
-
if err := initSchema(wrapped); err != nil {
77
-
db.Close()
78
-
return nil, err
79
-
}
80
-
81
-
return wrapped, nil
82
-
}
83
-
84
-
func initSchema(db *DB) error {
85
-
for _, s := range schema {
86
-
if _, err := db.Exec(s); err != nil {
87
-
return err
88
-
}
89
-
}
90
-
return nil
91
-
}
92
-
93
-
var schema = []string{
94
-
`CREATE TABLE IF NOT EXISTS users (
95
-
did TEXT PRIMARY KEY,
96
-
indexed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
97
-
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
98
-
)`,
99
-
`CREATE TABLE IF NOT EXISTS feeds (
100
-
feed_url TEXT PRIMARY KEY,
101
-
title TEXT,
102
-
site_url TEXT,
103
-
description TEXT,
104
-
feed_type TEXT CHECK(feed_type IN ('rss', 'atom', 'json')),
105
-
last_fetched_at DATETIME,
106
-
last_error TEXT,
107
-
subscriber_count INTEGER NOT NULL DEFAULT 0,
108
-
etag TEXT,
109
-
last_modified TEXT,
110
-
consecutive_empty_fetches INTEGER NOT NULL DEFAULT 0,
111
-
error_count INTEGER NOT NULL DEFAULT 0,
112
-
favicon_url TEXT
113
-
)`,
114
-
`CREATE TABLE IF NOT EXISTS subscriptions (
115
-
id INTEGER PRIMARY KEY AUTOINCREMENT,
116
-
user_did TEXT NOT NULL,
117
-
feed_url TEXT NOT NULL,
118
-
title TEXT,
119
-
category TEXT,
120
-
added_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
121
-
uri TEXT,
122
-
cid TEXT,
123
-
UNIQUE(user_did, feed_url)
124
-
)`,
125
-
`CREATE TABLE IF NOT EXISTS articles (
126
-
id INTEGER PRIMARY KEY AUTOINCREMENT,
127
-
feed_url TEXT NOT NULL,
128
-
guid TEXT NOT NULL,
129
-
title TEXT NOT NULL DEFAULT '',
130
-
url TEXT,
131
-
author TEXT,
132
-
summary TEXT,
133
-
content TEXT,
134
-
full_content TEXT,
135
-
published DATETIME,
136
-
updated DATETIME,
137
-
fetched_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
138
-
UNIQUE(feed_url, guid)
139
-
)`,
140
-
`CREATE TABLE IF NOT EXISTS read_state (
141
-
user_did TEXT NOT NULL,
142
-
article_id INTEGER NOT NULL,
143
-
is_read BOOLEAN NOT NULL DEFAULT 0,
144
-
read_at DATETIME,
145
-
PRIMARY KEY (user_did, article_id)
146
-
)`,
147
-
`CREATE TABLE IF NOT EXISTS annotations (
148
-
id INTEGER PRIMARY KEY AUTOINCREMENT,
149
-
uri TEXT NOT NULL UNIQUE,
150
-
author_did TEXT NOT NULL,
151
-
feed_url TEXT NOT NULL,
152
-
article_url TEXT NOT NULL,
153
-
quote TEXT,
154
-
note TEXT,
155
-
tags TEXT,
156
-
rating INTEGER,
157
-
created_at DATETIME NOT NULL,
158
-
cid TEXT
159
-
)`,
160
-
`CREATE TABLE IF NOT EXISTS likes (
161
-
id INTEGER PRIMARY KEY AUTOINCREMENT,
162
-
uri TEXT NOT NULL UNIQUE,
163
-
author_did TEXT NOT NULL,
164
-
feed_url TEXT NOT NULL,
165
-
article_url TEXT NOT NULL,
166
-
created_at DATETIME NOT NULL,
167
-
cid TEXT,
168
-
UNIQUE(author_did, feed_url, article_url)
169
-
)`,
170
-
`CREATE TABLE IF NOT EXISTS feed_similarity (
171
-
feed_a TEXT NOT NULL,
172
-
feed_b TEXT NOT NULL,
173
-
jaccard REAL NOT NULL,
174
-
computed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
175
-
PRIMARY KEY (feed_a, feed_b),
176
-
CHECK(feed_a < feed_b)
177
-
)`,
178
-
`CREATE TABLE IF NOT EXISTS user_similarity (
179
-
user_a TEXT NOT NULL,
180
-
user_b TEXT NOT NULL,
181
-
jaccard REAL NOT NULL,
182
-
common_feeds INTEGER NOT NULL,
183
-
common_likes INTEGER NOT NULL DEFAULT 0,
184
-
common_tags INTEGER NOT NULL DEFAULT 0,
185
-
computed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
186
-
PRIMARY KEY (user_a, user_b),
187
-
CHECK(user_a < user_b)
188
-
)`,
189
-
`CREATE TABLE IF NOT EXISTS follows (
190
-
user_did TEXT NOT NULL,
191
-
target_did TEXT NOT NULL,
192
-
uri TEXT,
193
-
cid TEXT,
194
-
followed_at DATETIME,
195
-
PRIMARY KEY (user_did, target_did)
196
-
)`,
197
-
`CREATE TABLE IF NOT EXISTS oauth_auth_requests (
198
-
state TEXT PRIMARY KEY,
199
-
data TEXT NOT NULL
200
-
)`,
201
-
`CREATE TABLE IF NOT EXISTS oauth_sessions (
202
-
account_did TEXT NOT NULL,
203
-
session_id TEXT NOT NULL,
204
-
data TEXT NOT NULL,
205
-
PRIMARY KEY (account_did, session_id)
206
-
)`,
207
-
`CREATE TABLE IF NOT EXISTS dismissed_recommendations (
208
-
user_did TEXT NOT NULL,
209
-
target_type TEXT NOT NULL CHECK(target_type IN ('feed', 'article')),
210
-
target_id TEXT NOT NULL,
211
-
reason TEXT,
212
-
dismissed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
213
-
PRIMARY KEY (user_did, target_type, target_id)
214
-
)`,
215
-
`CREATE TABLE IF NOT EXISTS recommendation_impressions (
216
-
user_did TEXT NOT NULL,
217
-
target_type TEXT NOT NULL CHECK(target_type IN ('feed', 'article')),
218
-
target_id TEXT NOT NULL,
219
-
first_shown_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
220
-
last_shown_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
221
-
shown_count INTEGER NOT NULL DEFAULT 1,
222
-
acted BOOLEAN NOT NULL DEFAULT 0,
223
-
PRIMARY KEY (user_did, target_type, target_id)
224
-
)`,
225
-
`CREATE TABLE IF NOT EXISTS follow_distances (
226
-
user_a TEXT NOT NULL,
227
-
user_b TEXT NOT NULL,
228
-
distance INTEGER NOT NULL CHECK(distance IN (1, 2)),
229
-
PRIMARY KEY (user_a, user_b)
230
-
)`,
231
-
`CREATE TABLE IF NOT EXISTS user_signal_weights (
232
-
user_did TEXT PRIMARY KEY,
233
-
w_sub REAL NOT NULL DEFAULT 1.0,
234
-
w_like REAL NOT NULL DEFAULT 0.5,
235
-
w_tag REAL NOT NULL DEFAULT 0.3,
236
-
w_social REAL NOT NULL DEFAULT 0.7,
237
-
w_pop REAL NOT NULL DEFAULT 0.2,
238
-
w_category REAL NOT NULL DEFAULT 0.4,
239
-
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
240
-
)`,
241
-
`CREATE TABLE IF NOT EXISTS user_signal_profiles (
242
-
user_did TEXT PRIMARY KEY,
243
-
total_likes INTEGER NOT NULL DEFAULT 0,
244
-
total_tags INTEGER NOT NULL DEFAULT 0,
245
-
top_categories TEXT,
246
-
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
247
-
)`,
248
-
`CREATE INDEX IF NOT EXISTS idx_subscriptions_feed ON subscriptions(feed_url)`,
249
-
`CREATE INDEX IF NOT EXISTS idx_subscriptions_feed_user ON subscriptions(feed_url, user_did)`,
250
-
`CREATE INDEX IF NOT EXISTS idx_subscriptions_user ON subscriptions(user_did)`,
251
-
`CREATE INDEX IF NOT EXISTS idx_subscriptions_uri ON subscriptions(uri)`,
252
-
`CREATE INDEX IF NOT EXISTS idx_likes_author_feed ON likes(author_did, feed_url, created_at)`,
253
-
`CREATE INDEX IF NOT EXISTS idx_articles_feed ON articles(feed_url)`,
254
-
`CREATE INDEX IF NOT EXISTS idx_articles_published ON articles(published DESC)`,
255
-
`CREATE INDEX IF NOT EXISTS idx_articles_url ON articles(url)`,
256
-
`CREATE INDEX IF NOT EXISTS idx_read_state_unread ON read_state(user_did, is_read) WHERE is_read = 0`,
257
-
`CREATE INDEX IF NOT EXISTS idx_annotations_article ON annotations(article_url)`,
258
-
`CREATE INDEX IF NOT EXISTS idx_annotations_author ON annotations(author_did)`,
259
-
`CREATE INDEX IF NOT EXISTS idx_annotations_created_at ON annotations(created_at DESC)`,
260
-
`CREATE INDEX IF NOT EXISTS idx_likes_article ON likes(feed_url, article_url)`,
261
-
`CREATE INDEX IF NOT EXISTS idx_likes_author ON likes(author_did)`,
262
-
`CREATE INDEX IF NOT EXISTS idx_likes_created_at ON likes(created_at DESC)`,
263
-
`CREATE INDEX IF NOT EXISTS idx_follows_user ON follows(user_did)`,
264
-
`CREATE INDEX IF NOT EXISTS idx_follows_target ON follows(target_did)`,
265
-
`CREATE INDEX IF NOT EXISTS idx_follows_uri ON follows(uri)`,
266
-
`CREATE INDEX IF NOT EXISTS idx_user_similarity_b ON user_similarity(user_b)`,
267
-
`CREATE INDEX IF NOT EXISTS idx_user_similarity_a ON user_similarity(user_a)`,
268
-
`CREATE INDEX IF NOT EXISTS idx_dismissed_user_type ON dismissed_recommendations(user_did, target_type)`,
269
-
`CREATE INDEX IF NOT EXISTS idx_impressions_user_unacted ON recommendation_impressions(user_did, acted, shown_count)`,
270
-
`CREATE INDEX IF NOT EXISTS idx_impressions_last_shown ON recommendation_impressions(last_shown_at)`,
271
-
`CREATE INDEX IF NOT EXISTS idx_follow_distances_b ON follow_distances(user_b)`,
272
-
`CREATE INDEX IF NOT EXISTS idx_follow_distances_a_dist ON follow_distances(user_a, distance)`,
273
-
`CREATE INDEX IF NOT EXISTS idx_follows_followed_at ON follows(followed_at)`,
274
-
`CREATE VIRTUAL TABLE IF NOT EXISTS articles_fts USING fts5(title, summary, content, author, content=articles, content_rowid=id)`,
275
-
`CREATE TRIGGER IF NOT EXISTS articles_ai AFTER INSERT ON articles BEGIN
276
-
INSERT INTO articles_fts(rowid, title, summary, content, author) VALUES (new.id, new.title, new.summary, new.content, new.author);
277
-
END`,
278
-
`CREATE TRIGGER IF NOT EXISTS articles_ad AFTER DELETE ON articles BEGIN
279
-
INSERT INTO articles_fts(articles_fts, rowid, title, summary, content, author) VALUES('delete', old.id, old.title, old.summary, old.content, old.author);
280
-
END`,
281
-
`CREATE TRIGGER IF NOT EXISTS articles_au AFTER UPDATE ON articles BEGIN
282
-
INSERT INTO articles_fts(articles_fts, rowid, title, summary, content, author) VALUES('delete', old.id, old.title, old.summary, old.content, old.author);
283
-
INSERT INTO articles_fts(rowid, title, summary, content, author) VALUES (new.id, new.title, new.summary, new.content, new.author);
284
-
END`,
285
-
}
+47
-68
internal/db/feed.go
+47
-68
internal/db/feed.go
···
6
6
"errors"
7
7
"strings"
8
8
"time"
9
+
10
+
"pkg.rbrt.fr/glean/internal/feed"
9
11
)
10
12
11
13
var ErrDuplicateSubscription = errors.New("already subscribed to this feed")
···
24
26
ConsecutiveEmptyFetches int
25
27
ErrorCount int
26
28
FaviconURL sql.NullString
29
+
}
30
+
31
+
func (f *Feed) ToFeed() *feed.Feed {
32
+
return &feed.Feed{
33
+
URL: f.FeedURL,
34
+
Title: f.Title.String,
35
+
SiteURL: f.SiteURL.String,
36
+
Description: f.Description.String,
37
+
Type: f.FeedType.String,
38
+
FaviconURL: f.FaviconURL.String,
39
+
ETag: f.Etag.String,
40
+
LastModified: f.LastModified.String,
41
+
}
27
42
}
28
43
29
44
type Subscription struct {
···
39
54
FaviconURL sql.NullString
40
55
}
41
56
42
-
func (s *ArticleStore) UpsertFeed(ctx context.Context, feed *Feed) error {
43
-
return s.BatchUpsertFeeds(ctx, []*Feed{feed})
44
-
}
57
+
const feedSelectCols = `feed_url, title, site_url, description, feed_type,
58
+
last_fetched_at, last_error, subscriber_count, etag, last_modified,
59
+
consecutive_empty_fetches, error_count, favicon_url`
45
60
46
-
func (s *ArticleStore) GetFeed(ctx context.Context, feedURL string) (*Feed, error) {
61
+
func scanFeed(scanner interface{ Scan(...any) error }) (*Feed, error) {
47
62
f := &Feed{}
48
-
err := s.db.QueryRowContext(ctx, `
49
-
SELECT feed_url, title, site_url, description, feed_type,
50
-
last_fetched_at, last_error, subscriber_count, etag, last_modified,
51
-
consecutive_empty_fetches, error_count, favicon_url
52
-
FROM articles.feeds WHERE feed_url = ?
53
-
`, feedURL).Scan(&f.FeedURL, &f.Title, &f.SiteURL, &f.Description, &f.FeedType,
63
+
if err := scanner.Scan(&f.FeedURL, &f.Title, &f.SiteURL, &f.Description, &f.FeedType,
54
64
&f.LastFetchedAt, &f.LastError, &f.SubscriberCount, &f.Etag, &f.LastModified,
55
-
&f.ConsecutiveEmptyFetches, &f.ErrorCount, &f.FaviconURL)
56
-
if err != nil {
65
+
&f.ConsecutiveEmptyFetches, &f.ErrorCount, &f.FaviconURL); err != nil {
57
66
return nil, err
58
67
}
59
68
return f, nil
60
69
}
61
70
71
+
func (s *ArticleStore) UpsertFeed(ctx context.Context, feed *Feed) error {
72
+
return s.BatchUpsertFeeds(ctx, []*Feed{feed})
73
+
}
74
+
75
+
func (s *ArticleStore) GetFeed(ctx context.Context, feedURL string) (*Feed, error) {
76
+
row := s.db.QueryRowContext(ctx, `SELECT `+feedSelectCols+` FROM articles.feeds WHERE feed_url = ?`, feedURL)
77
+
return scanFeed(row)
78
+
}
79
+
62
80
func (s *ArticleStore) GetFeedsToFetch(ctx context.Context, olderThan time.Duration, limit int) ([]*Feed, error) {
63
81
cutoff := time.Now().Add(-olderThan)
64
-
rows, err := s.db.QueryContext(ctx, `
65
-
SELECT feed_url, title, site_url, description, feed_type,
66
-
last_fetched_at, last_error, subscriber_count, etag, last_modified,
67
-
consecutive_empty_fetches, error_count, favicon_url
68
-
FROM articles.feeds
82
+
rows, err := s.db.QueryContext(ctx, `SELECT `+feedSelectCols+` FROM articles.feeds
69
83
WHERE subscriber_count > 0 AND error_count < 25 AND (last_fetched_at IS NULL OR last_fetched_at <= ?)
70
-
ORDER BY last_fetched_at ASC NULLS FIRST
71
-
LIMIT ?
72
-
`, cutoff, limit)
84
+
ORDER BY last_fetched_at ASC NULLS FIRST LIMIT ?`, cutoff, limit)
73
85
if err != nil {
74
86
return nil, err
75
87
}
···
77
89
78
90
var feeds []*Feed
79
91
for rows.Next() {
80
-
f := &Feed{}
81
-
if err := rows.Scan(&f.FeedURL, &f.Title, &f.SiteURL, &f.Description, &f.FeedType,
82
-
&f.LastFetchedAt, &f.LastError, &f.SubscriberCount, &f.Etag, &f.LastModified,
83
-
&f.ConsecutiveEmptyFetches, &f.ErrorCount, &f.FaviconURL); err != nil {
92
+
f, err := scanFeed(rows)
93
+
if err != nil {
84
94
return nil, err
85
95
}
86
96
feeds = append(feeds, f)
···
137
147
return err
138
148
}
139
149
140
-
func uriOrNil(v string) any {
141
-
if v == "" {
142
-
return nil
143
-
}
144
-
return v
145
-
}
146
-
147
150
func nilIfEmpty(v string) any {
148
151
if v == "" {
149
152
return nil
···
306
309
}
307
310
308
311
func (s *ArticleStore) ListDeadFeeds(ctx context.Context, userDID string, threshold int) ([]*Feed, error) {
309
-
rows, err := s.db.QueryContext(ctx, `
310
-
SELECT f.feed_url, f.title, f.site_url, f.description, f.feed_type,
311
-
f.last_fetched_at, f.last_error, f.subscriber_count, f.etag, f.last_modified,
312
-
f.consecutive_empty_fetches, f.error_count, f.favicon_url
313
-
FROM articles.feeds f
312
+
rows, err := s.db.QueryContext(ctx, `SELECT `+feedSelectCols+` FROM articles.feeds f
314
313
JOIN articles.subscriptions s ON s.feed_url = f.feed_url AND s.user_did = ?
315
-
WHERE f.error_count >= ?
316
-
ORDER BY f.error_count DESC
317
-
`, userDID, threshold)
314
+
WHERE f.error_count >= ? ORDER BY f.error_count DESC`, userDID, threshold)
318
315
if err != nil {
319
316
return nil, err
320
317
}
···
322
319
323
320
var feeds []*Feed
324
321
for rows.Next() {
325
-
f := &Feed{}
326
-
if err := rows.Scan(&f.FeedURL, &f.Title, &f.SiteURL, &f.Description, &f.FeedType,
327
-
&f.LastFetchedAt, &f.LastError, &f.SubscriberCount, &f.Etag, &f.LastModified,
328
-
&f.ConsecutiveEmptyFetches, &f.ErrorCount, &f.FaviconURL); err != nil {
322
+
f, err := scanFeed(rows)
323
+
if err != nil {
329
324
return nil, err
330
325
}
331
326
feeds = append(feeds, f)
···
334
329
}
335
330
336
331
func (s *ArticleStore) ListAllFeeds(ctx context.Context, limit, offset int) ([]*Feed, error) {
337
-
rows, err := s.db.QueryContext(ctx, `
338
-
SELECT feed_url, title, site_url, description, feed_type,
339
-
last_fetched_at, last_error, subscriber_count, etag, last_modified,
340
-
consecutive_empty_fetches, error_count, favicon_url
341
-
FROM articles.feeds
342
-
ORDER BY subscriber_count DESC
343
-
LIMIT ? OFFSET ?
344
-
`, limit, offset)
332
+
rows, err := s.db.QueryContext(ctx, `SELECT `+feedSelectCols+` FROM articles.feeds
333
+
ORDER BY subscriber_count DESC LIMIT ? OFFSET ?`, limit, offset)
345
334
if err != nil {
346
335
return nil, err
347
336
}
···
349
338
350
339
var feeds []*Feed
351
340
for rows.Next() {
352
-
f := &Feed{}
353
-
if err := rows.Scan(&f.FeedURL, &f.Title, &f.SiteURL, &f.Description, &f.FeedType,
354
-
&f.LastFetchedAt, &f.LastError, &f.SubscriberCount, &f.Etag, &f.LastModified,
355
-
&f.ConsecutiveEmptyFetches, &f.ErrorCount, &f.FaviconURL); err != nil {
341
+
f, err := scanFeed(rows)
342
+
if err != nil {
356
343
return nil, err
357
344
}
358
345
feeds = append(feeds, f)
···
457
444
}
458
445
continue
459
446
}
460
-
result, err := insertStmt.ExecContext(ctx, userDID, sub.FeedURL, nilIfEmpty(sub.Title), sub.Category, uriOrNil(sub.URI), uriOrNil(sub.CID))
447
+
result, err := insertStmt.ExecContext(ctx, userDID, sub.FeedURL, nilIfEmpty(sub.Title), sub.Category, nilIfEmpty(sub.URI), nilIfEmpty(sub.CID))
461
448
if err != nil {
462
449
return err
463
450
}
···
472
459
}
473
460
474
461
func (s *ArticleStore) ListUnsubscribedFeeds(ctx context.Context, userDID string, limit, offset int) ([]*Feed, error) {
475
-
rows, err := s.db.QueryContext(ctx, `
476
-
SELECT feed_url, title, site_url, description, feed_type,
477
-
last_fetched_at, last_error, subscriber_count, etag, last_modified,
478
-
consecutive_empty_fetches, error_count, favicon_url
479
-
FROM articles.feeds
462
+
rows, err := s.db.QueryContext(ctx, `SELECT `+feedSelectCols+` FROM articles.feeds
480
463
WHERE feed_url NOT IN (SELECT feed_url FROM articles.subscriptions WHERE user_did = ?)
481
-
ORDER BY subscriber_count DESC
482
-
LIMIT ? OFFSET ?
483
-
`, userDID, limit, offset)
464
+
ORDER BY subscriber_count DESC LIMIT ? OFFSET ?`, userDID, limit, offset)
484
465
if err != nil {
485
466
return nil, err
486
467
}
···
488
469
489
470
var feeds []*Feed
490
471
for rows.Next() {
491
-
f := &Feed{}
492
-
if err := rows.Scan(&f.FeedURL, &f.Title, &f.SiteURL, &f.Description, &f.FeedType,
493
-
&f.LastFetchedAt, &f.LastError, &f.SubscriberCount, &f.Etag, &f.LastModified,
494
-
&f.ConsecutiveEmptyFetches, &f.ErrorCount, &f.FaviconURL); err != nil {
472
+
f, err := scanFeed(rows)
473
+
if err != nil {
495
474
return nil, err
496
475
}
497
476
feeds = append(feeds, f)
+1
-1
internal/db/follow.go
+1
-1
internal/db/follow.go
+1
-10
internal/db/store.go
+1
-10
internal/db/store.go
···
22
22
}
23
23
var feeds []*feed.Feed
24
24
for _, df := range dbFeeds {
25
-
feeds = append(feeds, &feed.Feed{
26
-
URL: df.FeedURL,
27
-
Title: df.Title.String,
28
-
SiteURL: df.SiteURL.String,
29
-
Description: df.Description.String,
30
-
Type: df.FeedType.String,
31
-
FaviconURL: df.FaviconURL.String,
32
-
ETag: df.Etag.String,
33
-
LastModified: df.LastModified.String,
34
-
})
25
+
feeds = append(feeds, df.ToFeed())
35
26
}
36
27
return feeds, nil
37
28
}
+4
-13
internal/feed/discover.go
+4
-13
internal/feed/discover.go
···
12
12
"pkg.rbrt.fr/glean/internal/httpclient"
13
13
)
14
14
15
-
type imageContentTypePrefixes []string
16
-
17
-
func (p imageContentTypePrefixes) matches(contentType string) bool {
18
-
for _, prefix := range p {
19
-
if strings.HasPrefix(contentType, prefix) {
20
-
return true
21
-
}
22
-
}
23
-
return false
15
+
func isImageContentType(ct string) bool {
16
+
return strings.HasPrefix(ct, "image/")
24
17
}
25
-
26
-
var imageContentTypes = imageContentTypePrefixes{"image/"}
27
18
28
19
type DiscoveryResult struct {
29
20
FeedURLs []string
···
140
131
return
141
132
}
142
133
resp.Body.Close()
143
-
if resp.StatusCode == http.StatusOK && imageContentTypes.matches(resp.Header.Get("Content-Type")) {
134
+
if resp.StatusCode == http.StatusOK && isImageContentType(resp.Header.Get("Content-Type")) {
144
135
select {
145
136
case found <- result{url: cleanFavicon(resolved.String()), found: true}:
146
137
default:
···
170
161
return false
171
162
}
172
163
resp.Body.Close()
173
-
return resp.StatusCode == http.StatusOK && imageContentTypes.matches(resp.Header.Get("Content-Type"))
164
+
return resp.StatusCode == http.StatusOK && isImageContentType(resp.Header.Get("Content-Type"))
174
165
}
175
166
176
167
func extractHref(link string) string {
+4
-1
internal/feed/fetcher.go
+4
-1
internal/feed/fetcher.go
···
203
203
func (s *Scheduler) FetchFeed(ctx context.Context, feed *Feed) {
204
204
call := &fetchCall{done: make(chan struct{})}
205
205
if actual, loaded := s.inFlight.LoadOrStore(feed.URL, call); loaded {
206
-
<-actual.(*fetchCall).done
206
+
select {
207
+
case <-actual.(*fetchCall).done:
208
+
case <-ctx.Done():
209
+
}
207
210
return
208
211
}
209
212
defer func() {
+1
-1
internal/httpclient/httpclient.go
+1
-1
internal/httpclient/httpclient.go
-1
internal/server/annotations_handler.go
-1
internal/server/annotations_handler.go
+1
-1
internal/server/auth_handler.go
+1
-1
internal/server/auth_handler.go
···
102
102
PDSURL: sessData.HostURL,
103
103
SessionID: sessData.SessionID,
104
104
}
105
-
encoded, err := encodeSession(sessionData)
105
+
encoded, err := encodeSession(s.sessionKey, sessionData)
106
106
if err != nil {
107
107
s.logger.Error("failed to encode session", "error", err)
108
108
s.renderError(w, r, http.StatusInternalServerError, "Session error", "Could not create your session. Please try again.")
-1
internal/server/dashboard_handler.go
-1
internal/server/dashboard_handler.go
+21
-25
internal/server/feeds_handler.go
+21
-25
internal/server/feeds_handler.go
···
8
8
"net/http"
9
9
"time"
10
10
11
+
"golang.org/x/sync/errgroup"
12
+
11
13
"pkg.rbrt.fr/glean/internal/atproto"
12
14
"pkg.rbrt.fr/glean/internal/cluster"
13
15
"pkg.rbrt.fr/glean/internal/db"
···
259
261
feedURLs := feed.ExtractFeedURLs(opml)
260
262
var added int
261
263
client := s.pdsClientForUser(r)
264
+
265
+
var favGoroutines []struct{ feedURL, siteURL string }
262
266
for _, fu := range feedURLs {
263
267
f := &db.Feed{
264
268
FeedURL: fu.URL,
···
271
275
continue
272
276
}
273
277
274
-
go func(feedURL, siteURL string) {
275
-
if fav := feed.ResolveFavicon(context.Background(), feedURL, siteURL); fav != "" {
276
-
if err := s.dbs.Articles.UpdateFeedFavicon(context.Background(), feedURL, fav); err != nil {
277
-
s.logger.Warn("failed to update favicon", "error", err, "feed", feedURL)
278
-
}
279
-
}
280
-
}(fu.URL, fu.SiteURL)
278
+
favGoroutines = append(favGoroutines, struct{ feedURL, siteURL string }{fu.URL, fu.SiteURL})
281
279
282
280
var subURI, subCID string
283
281
if client != nil {
···
304
302
}
305
303
added++
306
304
}
305
+
306
+
go func() {
307
+
g, ctx := errgroup.WithContext(context.Background())
308
+
g.SetLimit(5)
309
+
for _, fav := range favGoroutines {
310
+
g.Go(func() error {
311
+
if f := feed.ResolveFavicon(ctx, fav.feedURL, fav.siteURL); f != "" {
312
+
_ = s.dbs.Articles.UpdateFeedFavicon(ctx, fav.feedURL, f)
313
+
}
314
+
return nil
315
+
})
316
+
}
317
+
_ = g.Wait()
318
+
}()
307
319
308
320
w.Header().Set("HX-Redirect", "/feeds")
309
321
w.WriteHeader(http.StatusOK)
···
385
397
s.logger.Warn("failed to get feed", "error", err, "feed", sub.FeedURL)
386
398
continue
387
399
}
388
-
ff := &feed.Feed{
389
-
URL: f.FeedURL,
390
-
Title: f.Title.String,
391
-
SiteURL: f.SiteURL.String,
392
-
Description: f.Description.String,
393
-
Type: f.FeedType.String,
394
-
ETag: f.Etag.String,
395
-
LastModified: f.LastModified.String,
396
-
}
400
+
ff := f.ToFeed()
397
401
s.scheduler.FetchFeed(ctx, ff)
398
402
}
399
403
}
···
411
415
return
412
416
}
413
417
414
-
ff := &feed.Feed{
415
-
URL: f.FeedURL,
416
-
Title: f.Title.String,
417
-
SiteURL: f.SiteURL.String,
418
-
Description: f.Description.String,
419
-
Type: f.FeedType.String,
420
-
ETag: f.Etag.String,
421
-
LastModified: f.LastModified.String,
422
-
}
418
+
ff := f.ToFeed()
423
419
s.scheduler.FetchFeed(r.Context(), ff)
424
420
425
421
user := currentUser(r)
+30
-8
internal/server/middleware.go
+30
-8
internal/server/middleware.go
···
4
4
"crypto/rand"
5
5
"encoding/hex"
6
6
"net/http"
7
+
"net/url"
7
8
"strings"
8
9
"time"
9
-
10
-
"github.com/go-chi/chi/v5/middleware"
11
10
)
12
11
13
12
func (s *Server) sessionMiddleware(next http.Handler) http.Handler {
···
33
32
34
33
func csrfToken() string {
35
34
b := make([]byte, 32)
36
-
rand.Read(b)
35
+
if _, err := rand.Read(b); err != nil {
36
+
return ""
37
+
}
37
38
return hex.EncodeToString(b)
38
39
}
39
40
···
82
83
}
83
84
84
85
func sameOrigin(origin, host string) bool {
85
-
return strings.HasPrefix(origin, "http://"+host) || strings.HasPrefix(origin, "https://"+host)
86
+
u, err := url.Parse(origin)
87
+
if err != nil {
88
+
return false
89
+
}
90
+
return u.Host == host
86
91
}
87
92
88
93
func (s *Server) realIPLogger(next http.Handler) http.Handler {
89
94
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
90
95
start := time.Now()
91
-
ww := middleware.NewWrapResponseWriter(w, r.ProtoMajor)
96
+
sw := &statusWriter{ResponseWriter: w}
92
97
93
-
next.ServeHTTP(ww, r)
98
+
next.ServeHTTP(sw, r)
94
99
95
100
ip := r.RemoteAddr
96
101
if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
···
106
111
"method", r.Method,
107
112
"url", scheme+"://"+r.Host+r.RequestURI,
108
113
"from", ip,
109
-
"status", ww.Status(),
110
-
"bytes", ww.BytesWritten(),
114
+
"status", sw.status,
115
+
"bytes", sw.bytes,
111
116
"duration", time.Since(start).Round(time.Microsecond),
112
117
)
113
118
})
114
119
}
120
+
121
+
type statusWriter struct {
122
+
http.ResponseWriter
123
+
status int
124
+
bytes int
125
+
}
126
+
127
+
func (w *statusWriter) WriteHeader(code int) {
128
+
w.status = code
129
+
w.ResponseWriter.WriteHeader(code)
130
+
}
131
+
132
+
func (w *statusWriter) Write(b []byte) (int, error) {
133
+
n, err := w.ResponseWriter.Write(b)
134
+
w.bytes += n
135
+
return n, err
136
+
}
+2
internal/server/server.go
+2
internal/server/server.go
···
68
68
scraper *scraper.Scraper
69
69
clientID string
70
70
callbackURL string
71
+
sessionKey []byte
71
72
}
72
73
73
74
func New(dbs *db.Databases, clientID, callbackURL, addr string, scheduler *feed.Scheduler, engine *cluster.Engine, logger *slog.Logger) *Server {
···
98
99
scraper: scraper.New(logger),
99
100
clientID: clientID,
100
101
callbackURL: callbackURL,
102
+
sessionKey: loadSessionKey(),
101
103
}
102
104
103
105
s.setupMiddleware()
+8
-8
internal/server/session.go
+8
-8
internal/server/session.go
···
30
30
return nil
31
31
}
32
32
33
-
data, err := decodeSession(cookie.Value)
33
+
data, err := decodeSession(s.sessionKey, cookie.Value)
34
34
if err != nil {
35
35
return nil
36
36
}
···
49
49
50
50
func (s *Server) setUserSession(w http.ResponseWriter, user *db.User) {
51
51
data := sessionData{DID: user.DID}
52
-
encoded, err := encodeSession(data)
52
+
encoded, err := encodeSession(s.sessionKey, data)
53
53
if err != nil {
54
54
s.logger.Error("failed to encode session", "error", err)
55
55
return
···
89
89
if err != nil {
90
90
return nil
91
91
}
92
-
data, err := decodeSession(cookie.Value)
92
+
data, err := decodeSession(s.sessionKey, cookie.Value)
93
93
if err != nil {
94
94
return nil
95
95
}
96
96
return data
97
97
}
98
98
99
-
func sessionKey() []byte {
99
+
func loadSessionKey() []byte {
100
100
key := os.Getenv("GLEAN_SESSION_KEY")
101
101
if key == "" {
102
102
key = "default-dev-key-change-in-production"
···
104
104
return []byte(key)
105
105
}
106
106
107
-
func encodeSession(data sessionData) (string, error) {
107
+
func encodeSession(key []byte, data sessionData) (string, error) {
108
108
payload, err := json.Marshal(data)
109
109
if err != nil {
110
110
return "", err
111
111
}
112
112
113
-
mac := hmac.New(sha256.New, sessionKey())
113
+
mac := hmac.New(sha256.New, key)
114
114
mac.Write(payload)
115
115
sig := mac.Sum(nil)
116
116
···
118
118
return base64.URLEncoding.EncodeToString(raw), nil
119
119
}
120
120
121
-
func decodeSession(encoded string) (*sessionData, error) {
121
+
func decodeSession(key []byte, encoded string) (*sessionData, error) {
122
122
raw, err := base64.URLEncoding.DecodeString(encoded)
123
123
if err != nil {
124
124
return nil, errInvalidSession
···
131
131
payload := raw[:len(raw)-sha256.Size]
132
132
sig := raw[len(raw)-sha256.Size:]
133
133
134
-
mac := hmac.New(sha256.New, sessionKey())
134
+
mac := hmac.New(sha256.New, key)
135
135
mac.Write(payload)
136
136
expectedSig := mac.Sum(nil)
137
137