rickub
nandi/gleanpublic Fork
⑂ 81e384f
Branches
claude/modal-container-min-76527a main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/glean.git
git clone ssh://git@rickub.com/nandi/glean.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

Download ZIP

Port the article and feed queries

The core of internal/db/feed.go and article.go, with the SQLite wrapper
grown to carry them: typed nullable row access, transactions, and prepared
statements stepped many times.

Read state sets the shape. It is per-user in its own table, so a listing is
always a LEFT JOIN and unread means no row -- hence COALESCE(r.is_read, 0)
rather than a column on the article.

Two behaviours carried over deliberately, both about not resurfacing what a
reader has dealt with: articles already past the retention window are not
ingested at all, and read state is restored from history after ingest, since
a purged article that reappears gets a new surrogate id.

Four bugs the probe caught, none of which would fail at compile time. likes
keys on author_did, not user_did -- it is an ATProto record with an author.
FTS5 needs the bare table name on the left of MATCH, so neither an alias nor
a schema-qualified name works; the match moved to a subquery that also
carries rank out for ordering. The shared feed column list used bare names,
which go ambiguous the moment a query joins subscriptions. And the probe
itself assumed a handle column on users, which does not exist.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
nandithebull committed 2026-09-21T00:30:08-07:00 Browse files
81e384f parent: 51371bc
modified spike/nim/.gitignore +1 -0
@@ -11,3 +11,4 @@ store_probe
1111 *.db-wal
1212 *.db-shm
1313 gleandb_probe
14+stores_probe
@@ -11,3 +11,4 @@ store_probe
11 *.db-wal11 *.db-wal
12 *.db-shm12 *.db-shm
13 gleandb_probe13 gleandb_probe
14+stores_probe
added spike/nim/atproto/articlestore.nim +270 -0
new file mode 100644
@@ -0,0 +1,270 @@
1+## Articles and read state, ported from internal/db/article.go.
2+##
3+## Read state is per-user and lives in its own table, so "the article list"
4+## is always a LEFT JOIN: an article with no read_state row is unread. That
5+## shape is why COALESCE(r.is_read, 0) appears everywhere rather than a
6+## boolean column on the article itself.
7+
8+import std/[options, strutils, times]
9+import ./feedstore
10+import ./gleandb
11+import ./sqlite
12+
13+type
14+ Article* = object
15+ id*: int64
16+ feedUrl*: string
17+ feedTitle*: string
18+ feedFaviconUrl*: string
19+ guid*: string
20+ title*: string
21+ url*: string
22+ author*: string
23+ summary*: string
24+ content*: string
25+ fullContent*: string
26+ published*: Option[DateTime]
27+ updated*: Option[DateTime]
28+ isRead*: bool
29+ likeCount*: int
30+ hasLiked*: bool
31+
32+ ArticleFilter* = enum
33+ afAll = "all"
34+ afUnread = "unread"
35+ afRead = "read"
36+
37+ NewArticle* = object ## What a feed parser produces, before it has an id.
38+ feedUrl*: string
39+ guid*: string
40+ title*: string
41+ url*: string
42+ author*: string
43+ summary*: string
44+ content*: string
45+ published*: Option[DateTime]
46+ updated*: Option[DateTime]
47+
48+proc readArticle(r: Row): Article =
49+ Article(
50+ id: r.i64(0),
51+ feedUrl: r.str(1),
52+ feedTitle: r.str(2),
53+ feedFaviconUrl: r.str(3),
54+ guid: r.str(4),
55+ title: r.str(5),
56+ url: r.str(6),
57+ author: r.str(7),
58+ summary: r.str(8),
59+ content: r.str(9),
60+ fullContent: r.str(10),
61+ published: parseSqliteTime(r.str(11)),
62+ updated: parseSqliteTime(r.str(12)),
63+ isRead: r.b(13),
64+ likeCount: r.i(14),
65+ hasLiked: r.b(15),
66+ )
67+
68+## The column list is shared so every listing returns the same shape and
69+## readArticle can stay positional. A mismatch here reads as silently shifted
70+## fields rather than an error.
71+const articleColumns = """
72+ a.id, a.feed_url, COALESCE(f.title, a.feed_url), f.favicon_url,
73+ a.guid, a.title, a.url, a.author, a.summary, a.content, a.full_content,
74+ a.published, a.updated,
75+ COALESCE(r.is_read, 0),
76+ (SELECT count(*) FROM articles.likes l WHERE l.article_url = a.url),
77+ EXISTS(SELECT 1 FROM articles.likes l
78+ WHERE l.article_url = a.url AND l.author_did = ?)"""
79+
80+const articleFrom = """
81+ FROM articles.articles a
82+ LEFT JOIN articles.feeds f ON f.feed_url = a.feed_url
83+ LEFT JOIN articles.read_state r ON r.article_id = a.id AND r.user_did = ?"""
84+
85+proc getArticle*(g: GleanDb, userDid: string, id: int64): Option[Article] =
86+ g.db.queryFirst("SELECT " & articleColumns & articleFrom & " WHERE a.id = ?",
87+ [p(userDid), p(userDid), p(id)], readArticle)
88+
89+proc getArticleByUrl*(g: GleanDb, userDid, url: string): Option[Article] =
90+ g.db.queryFirst("SELECT " & articleColumns & articleFrom & " WHERE a.url = ?",
91+ [p(userDid), p(userDid), p(url)], readArticle)
92+
93+proc listArticles*(g: GleanDb, userDid: string, filter = afAll,
94+ feedUrl = "", category = "", sortOldest = false,
95+ limit = 25, offset = 0): seq[Article] =
96+ ## The main listing, scoped to what the user subscribes to.
97+ var sql = "SELECT " & articleColumns & articleFrom & """
98+ JOIN articles.subscriptions s
99+ ON s.feed_url = a.feed_url AND s.user_did = ?
100+ WHERE 1=1"""
101+ var params = @[p(userDid), p(userDid), p(userDid)]
102+
103+ case filter
104+ of afUnread: sql &= " AND COALESCE(r.is_read, 0) = 0"
105+ of afRead: sql &= " AND COALESCE(r.is_read, 0) = 1"
106+ of afAll: discard
107+
108+ if feedUrl.len > 0:
109+ sql &= " AND a.feed_url = ?"
110+ params.add p(feedUrl)
111+ if category.len > 0:
112+ sql &= " AND s.category = ?"
113+ params.add p(category)
114+
115+ # NULL published sorts last either way: an article with no date is almost
116+ # always a parse failure, and burying it beats putting it at the top.
117+ sql &= (if sortOldest:
118+ " ORDER BY a.published IS NULL, a.published ASC, a.id ASC"
119+ else:
120+ " ORDER BY a.published IS NULL, a.published DESC, a.id DESC")
121+ sql &= " LIMIT ? OFFSET ?"
122+ params.add p(limit.int64)
123+ params.add p(offset.int64)
124+
125+ g.db.queryAll(sql, params, readArticle)
126+
127+proc searchArticles*(g: GleanDb, userDid, query: string,
128+ limit = 25, offset = 0): seq[Article] =
129+ ## Full-text search over the user's subscriptions.
130+ ##
131+ ## The query goes to FTS5 as-is, so its operators (AND, OR, NEAR, "phrase")
132+ ## work. That also means a stray quote is a syntax error rather than a
133+ ## literal, which is why callers should treat a raised error as "no results"
134+ ## rather than a failure.
135+ ##
136+ ## MATCH is in a subquery rather than joined directly: FTS5 requires the
137+ ## bare table name on the left of MATCH, so an aliased or schema-qualified
138+ ## reference is rejected. The subquery also carries `rank` out for ordering.
139+ if query.strip.len == 0: return @[]
140+ g.db.queryAll("SELECT " & articleColumns & articleFrom & """
141+ JOIN articles.subscriptions s
142+ ON s.feed_url = a.feed_url AND s.user_did = ?
143+ JOIN (SELECT rowid, rank FROM articles.articles_fts
144+ WHERE articles_fts MATCH ?) m ON m.rowid = a.id
145+ ORDER BY m.rank
146+ LIMIT ? OFFSET ?""",
147+ [p(userDid), p(userDid), p(userDid), p(query),
148+ p(limit.int64), p(offset.int64)], readArticle)
149+
150+proc batchUpsertArticles*(g: GleanDb, articles: seq[NewArticle],
151+ retentionDays = 0) =
152+ ## Ingest a fetch. One transaction, one prepared statement.
153+ ##
154+ ## Two behaviours carried over from the Go version, both about not
155+ ## resurfacing things the reader has dealt with:
156+ ## * Articles already older than the retention window are skipped --
157+ ## ingesting them only to purge them churns the database.
158+ ## * Read state is restored from history afterwards, because a purged
159+ ## article that comes back gets a new surrogate id and would otherwise
160+ ## read as unread again.
161+ if articles.len == 0: return
162+
163+ var cutoff = none(DateTime)
164+ if retentionDays > 0:
165+ cutoff = some(now().utc - initDuration(days = retentionDays))
166+
167+ var feedUrls: seq[string]
168+ g.db.transaction:
169+ var ps = g.db.prepared("""
170+ INSERT INTO articles.articles
171+ (feed_url, guid, title, url, author, summary, content, published, updated, language)
172+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, '')
173+ ON CONFLICT(feed_url, guid) DO NOTHING""")
174+ defer: ps.finalize()
175+
176+ for a in articles:
177+ if cutoff.isSome and a.published.isSome and a.published.get < cutoff.get:
178+ continue
179+ if a.feedUrl notin feedUrls:
180+ feedUrls.add a.feedUrl
181+ ps.exec(
182+ p(a.feedUrl), p(a.guid), p(a.title), pOrNull(a.url), pOrNull(a.author),
183+ pOrNull(a.summary), pOrNull(a.content),
184+ (if a.published.isSome: p(formatSqliteTime(a.published.get)) else: nullParam()),
185+ (if a.updated.isSome: p(formatSqliteTime(a.updated.get)) else: nullParam()))
186+
187+ var restore = g.db.prepared("""
188+ INSERT INTO articles.read_state (user_did, article_id, is_read, read_at)
189+ SELECT h.user_did, a.id, h.is_read, h.read_at
190+ FROM articles.read_state_history h
191+ JOIN articles.articles a ON a.feed_url = h.feed_url AND a.guid = h.guid
192+ WHERE h.feed_url = ?
193+ ON CONFLICT(user_did, article_id) DO NOTHING""")
194+ defer: restore.finalize()
195+ for url in feedUrls:
196+ restore.exec(p(url))
197+
198+proc updateArticleFullContent*(g: GleanDb, id: int64, fullContent: string) =
199+ g.db.run("UPDATE articles.articles SET full_content = ? WHERE id = ?",
200+ pOrNull(fullContent), p(id))
201+
202+# --- read state ------------------------------------------------------------
203+
204+proc setRead(g: GleanDb, userDid: string, articleId: int64, isRead: bool) =
205+ g.db.run("""INSERT INTO articles.read_state (user_did, article_id, is_read, read_at)
206+ VALUES (?, ?, ?, CURRENT_TIMESTAMP)
207+ ON CONFLICT(user_did, article_id) DO UPDATE SET
208+ is_read = excluded.is_read,
209+ read_at = excluded.read_at""",
210+ p(userDid), p(articleId), p(if isRead: 1'i64 else: 0'i64))
211+
212+proc markArticleRead*(g: GleanDb, userDid: string, articleId: int64) =
213+ setRead(g, userDid, articleId, true)
214+
215+proc markArticleUnread*(g: GleanDb, userDid: string, articleId: int64) =
216+ setRead(g, userDid, articleId, false)
217+
218+proc markAllRead*(g: GleanDb, userDid, feedUrl: string) =
219+ ## Scoped to one feed when given, otherwise everything the user subscribes
220+ ## to. Getting this scope wrong is the difference between clearing a feed
221+ ## and clearing the whole account, so the two cases are separate statements
222+ ## rather than one with an optional predicate.
223+ if feedUrl.len > 0:
224+ g.db.run("""INSERT INTO articles.read_state (user_did, article_id, is_read, read_at)
225+ SELECT ?, a.id, 1, CURRENT_TIMESTAMP
226+ FROM articles.articles a
227+ WHERE a.feed_url = ?
228+ ON CONFLICT(user_did, article_id) DO UPDATE SET
229+ is_read = 1, read_at = CURRENT_TIMESTAMP""",
230+ p(userDid), p(feedUrl))
231+ else:
232+ g.db.run("""INSERT INTO articles.read_state (user_did, article_id, is_read, read_at)
233+ SELECT ?, a.id, 1, CURRENT_TIMESTAMP
234+ FROM articles.articles a
235+ JOIN articles.subscriptions s
236+ ON s.feed_url = a.feed_url AND s.user_did = ?
237+ ON CONFLICT(user_did, article_id) DO UPDATE SET
238+ is_read = 1, read_at = CURRENT_TIMESTAMP""",
239+ p(userDid), p(userDid))
240+
241+proc getUnreadCount*(g: GleanDb, userDid, feedUrl, category: string): int =
242+ var sql = """
243+ SELECT count(*)
244+ FROM articles.articles a
245+ JOIN articles.subscriptions s
246+ ON s.feed_url = a.feed_url AND s.user_did = ?
247+ LEFT JOIN articles.read_state r
248+ ON r.article_id = a.id AND r.user_did = ?
249+ WHERE COALESCE(r.is_read, 0) = 0"""
250+ var params = @[p(userDid), p(userDid)]
251+ if feedUrl.len > 0:
252+ sql &= " AND a.feed_url = ?"
253+ params.add p(feedUrl)
254+ if category.len > 0:
255+ sql &= " AND s.category = ?"
256+ params.add p(category)
257+ g.db.queryInt(sql, params).get(0).int
258+
259+proc isRead*(g: GleanDb, userDid: string, articleId: int64): bool =
260+ g.db.queryInt("""SELECT COALESCE(is_read, 0) FROM articles.read_state
261+ WHERE user_did = ? AND article_id = ?""",
262+ p(userDid), p(articleId)).get(0) != 0
263+
264+proc countNewArticles*(g: GleanDb, userDid: string, since: DateTime): int =
265+ g.db.queryInt("""
266+ SELECT count(*) FROM articles.articles a
267+ JOIN articles.subscriptions s
268+ ON s.feed_url = a.feed_url AND s.user_did = ?
269+ WHERE a.fetched_at > ?""",
270+ p(userDid), p(formatSqliteTime(since))).get(0).int
new file mode 100644
@@ -0,0 +1,270 @@
1+## Articles and read state, ported from internal/db/article.go.
2+##
3+## Read state is per-user and lives in its own table, so "the article list"
4+## is always a LEFT JOIN: an article with no read_state row is unread. That
5+## shape is why COALESCE(r.is_read, 0) appears everywhere rather than a
6+## boolean column on the article itself.
7+
8+import std/[options, strutils, times]
9+import ./feedstore
10+import ./gleandb
11+import ./sqlite
12+
13+type
14+ Article* = object
15+ id*: int64
16+ feedUrl*: string
17+ feedTitle*: string
18+ feedFaviconUrl*: string
19+ guid*: string
20+ title*: string
21+ url*: string
22+ author*: string
23+ summary*: string
24+ content*: string
25+ fullContent*: string
26+ published*: Option[DateTime]
27+ updated*: Option[DateTime]
28+ isRead*: bool
29+ likeCount*: int
30+ hasLiked*: bool
31+
32+ ArticleFilter* = enum
33+ afAll = "all"
34+ afUnread = "unread"
35+ afRead = "read"
36+
37+ NewArticle* = object ## What a feed parser produces, before it has an id.
38+ feedUrl*: string
39+ guid*: string
40+ title*: string
41+ url*: string
42+ author*: string
43+ summary*: string
44+ content*: string
45+ published*: Option[DateTime]
46+ updated*: Option[DateTime]
47+
48+proc readArticle(r: Row): Article =
49+ Article(
50+ id: r.i64(0),
51+ feedUrl: r.str(1),
52+ feedTitle: r.str(2),
53+ feedFaviconUrl: r.str(3),
54+ guid: r.str(4),
55+ title: r.str(5),
56+ url: r.str(6),
57+ author: r.str(7),
58+ summary: r.str(8),
59+ content: r.str(9),
60+ fullContent: r.str(10),
61+ published: parseSqliteTime(r.str(11)),
62+ updated: parseSqliteTime(r.str(12)),
63+ isRead: r.b(13),
64+ likeCount: r.i(14),
65+ hasLiked: r.b(15),
66+ )
67+
68+## The column list is shared so every listing returns the same shape and
69+## readArticle can stay positional. A mismatch here reads as silently shifted
70+## fields rather than an error.
71+const articleColumns = """
72+ a.id, a.feed_url, COALESCE(f.title, a.feed_url), f.favicon_url,
73+ a.guid, a.title, a.url, a.author, a.summary, a.content, a.full_content,
74+ a.published, a.updated,
75+ COALESCE(r.is_read, 0),
76+ (SELECT count(*) FROM articles.likes l WHERE l.article_url = a.url),
77+ EXISTS(SELECT 1 FROM articles.likes l
78+ WHERE l.article_url = a.url AND l.author_did = ?)"""
79+
80+const articleFrom = """
81+ FROM articles.articles a
82+ LEFT JOIN articles.feeds f ON f.feed_url = a.feed_url
83+ LEFT JOIN articles.read_state r ON r.article_id = a.id AND r.user_did = ?"""
84+
85+proc getArticle*(g: GleanDb, userDid: string, id: int64): Option[Article] =
86+ g.db.queryFirst("SELECT " & articleColumns & articleFrom & " WHERE a.id = ?",
87+ [p(userDid), p(userDid), p(id)], readArticle)
88+
89+proc getArticleByUrl*(g: GleanDb, userDid, url: string): Option[Article] =
90+ g.db.queryFirst("SELECT " & articleColumns & articleFrom & " WHERE a.url = ?",
91+ [p(userDid), p(userDid), p(url)], readArticle)
92+
93+proc listArticles*(g: GleanDb, userDid: string, filter = afAll,
94+ feedUrl = "", category = "", sortOldest = false,
95+ limit = 25, offset = 0): seq[Article] =
96+ ## The main listing, scoped to what the user subscribes to.
97+ var sql = "SELECT " & articleColumns & articleFrom & """
98+ JOIN articles.subscriptions s
99+ ON s.feed_url = a.feed_url AND s.user_did = ?
100+ WHERE 1=1"""
101+ var params = @[p(userDid), p(userDid), p(userDid)]
102+
103+ case filter
104+ of afUnread: sql &= " AND COALESCE(r.is_read, 0) = 0"
105+ of afRead: sql &= " AND COALESCE(r.is_read, 0) = 1"
106+ of afAll: discard
107+
108+ if feedUrl.len > 0:
109+ sql &= " AND a.feed_url = ?"
110+ params.add p(feedUrl)
111+ if category.len > 0:
112+ sql &= " AND s.category = ?"
113+ params.add p(category)
114+
115+ # NULL published sorts last either way: an article with no date is almost
116+ # always a parse failure, and burying it beats putting it at the top.
117+ sql &= (if sortOldest:
118+ " ORDER BY a.published IS NULL, a.published ASC, a.id ASC"
119+ else:
120+ " ORDER BY a.published IS NULL, a.published DESC, a.id DESC")
121+ sql &= " LIMIT ? OFFSET ?"
122+ params.add p(limit.int64)
123+ params.add p(offset.int64)
124+
125+ g.db.queryAll(sql, params, readArticle)
126+
127+proc searchArticles*(g: GleanDb, userDid, query: string,
128+ limit = 25, offset = 0): seq[Article] =
129+ ## Full-text search over the user's subscriptions.
130+ ##
131+ ## The query goes to FTS5 as-is, so its operators (AND, OR, NEAR, "phrase")
132+ ## work. That also means a stray quote is a syntax error rather than a
133+ ## literal, which is why callers should treat a raised error as "no results"
134+ ## rather than a failure.
135+ ##
136+ ## MATCH is in a subquery rather than joined directly: FTS5 requires the
137+ ## bare table name on the left of MATCH, so an aliased or schema-qualified
138+ ## reference is rejected. The subquery also carries `rank` out for ordering.
139+ if query.strip.len == 0: return @[]
140+ g.db.queryAll("SELECT " & articleColumns & articleFrom & """
141+ JOIN articles.subscriptions s
142+ ON s.feed_url = a.feed_url AND s.user_did = ?
143+ JOIN (SELECT rowid, rank FROM articles.articles_fts
144+ WHERE articles_fts MATCH ?) m ON m.rowid = a.id
145+ ORDER BY m.rank
146+ LIMIT ? OFFSET ?""",
147+ [p(userDid), p(userDid), p(userDid), p(query),
148+ p(limit.int64), p(offset.int64)], readArticle)
149+
150+proc batchUpsertArticles*(g: GleanDb, articles: seq[NewArticle],
151+ retentionDays = 0) =
152+ ## Ingest a fetch. One transaction, one prepared statement.
153+ ##
154+ ## Two behaviours carried over from the Go version, both about not
155+ ## resurfacing things the reader has dealt with:
156+ ## * Articles already older than the retention window are skipped --
157+ ## ingesting them only to purge them churns the database.
158+ ## * Read state is restored from history afterwards, because a purged
159+ ## article that comes back gets a new surrogate id and would otherwise
160+ ## read as unread again.
161+ if articles.len == 0: return
162+
163+ var cutoff = none(DateTime)
164+ if retentionDays > 0:
165+ cutoff = some(now().utc - initDuration(days = retentionDays))
166+
167+ var feedUrls: seq[string]
168+ g.db.transaction:
169+ var ps = g.db.prepared("""
170+ INSERT INTO articles.articles
171+ (feed_url, guid, title, url, author, summary, content, published, updated, language)
172+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, '')
173+ ON CONFLICT(feed_url, guid) DO NOTHING""")
174+ defer: ps.finalize()
175+
176+ for a in articles:
177+ if cutoff.isSome and a.published.isSome and a.published.get < cutoff.get:
178+ continue
179+ if a.feedUrl notin feedUrls:
180+ feedUrls.add a.feedUrl
181+ ps.exec(
182+ p(a.feedUrl), p(a.guid), p(a.title), pOrNull(a.url), pOrNull(a.author),
183+ pOrNull(a.summary), pOrNull(a.content),
184+ (if a.published.isSome: p(formatSqliteTime(a.published.get)) else: nullParam()),
185+ (if a.updated.isSome: p(formatSqliteTime(a.updated.get)) else: nullParam()))
186+
187+ var restore = g.db.prepared("""
188+ INSERT INTO articles.read_state (user_did, article_id, is_read, read_at)
189+ SELECT h.user_did, a.id, h.is_read, h.read_at
190+ FROM articles.read_state_history h
191+ JOIN articles.articles a ON a.feed_url = h.feed_url AND a.guid = h.guid
192+ WHERE h.feed_url = ?
193+ ON CONFLICT(user_did, article_id) DO NOTHING""")
194+ defer: restore.finalize()
195+ for url in feedUrls:
196+ restore.exec(p(url))
197+
198+proc updateArticleFullContent*(g: GleanDb, id: int64, fullContent: string) =
199+ g.db.run("UPDATE articles.articles SET full_content = ? WHERE id = ?",
200+ pOrNull(fullContent), p(id))
201+
202+# --- read state ------------------------------------------------------------
203+
204+proc setRead(g: GleanDb, userDid: string, articleId: int64, isRead: bool) =
205+ g.db.run("""INSERT INTO articles.read_state (user_did, article_id, is_read, read_at)
206+ VALUES (?, ?, ?, CURRENT_TIMESTAMP)
207+ ON CONFLICT(user_did, article_id) DO UPDATE SET
208+ is_read = excluded.is_read,
209+ read_at = excluded.read_at""",
210+ p(userDid), p(articleId), p(if isRead: 1'i64 else: 0'i64))
211+
212+proc markArticleRead*(g: GleanDb, userDid: string, articleId: int64) =
213+ setRead(g, userDid, articleId, true)
214+
215+proc markArticleUnread*(g: GleanDb, userDid: string, articleId: int64) =
216+ setRead(g, userDid, articleId, false)
217+
218+proc markAllRead*(g: GleanDb, userDid, feedUrl: string) =
219+ ## Scoped to one feed when given, otherwise everything the user subscribes
220+ ## to. Getting this scope wrong is the difference between clearing a feed
221+ ## and clearing the whole account, so the two cases are separate statements
222+ ## rather than one with an optional predicate.
223+ if feedUrl.len > 0:
224+ g.db.run("""INSERT INTO articles.read_state (user_did, article_id, is_read, read_at)
225+ SELECT ?, a.id, 1, CURRENT_TIMESTAMP
226+ FROM articles.articles a
227+ WHERE a.feed_url = ?
228+ ON CONFLICT(user_did, article_id) DO UPDATE SET
229+ is_read = 1, read_at = CURRENT_TIMESTAMP""",
230+ p(userDid), p(feedUrl))
231+ else:
232+ g.db.run("""INSERT INTO articles.read_state (user_did, article_id, is_read, read_at)
233+ SELECT ?, a.id, 1, CURRENT_TIMESTAMP
234+ FROM articles.articles a
235+ JOIN articles.subscriptions s
236+ ON s.feed_url = a.feed_url AND s.user_did = ?
237+ ON CONFLICT(user_did, article_id) DO UPDATE SET
238+ is_read = 1, read_at = CURRENT_TIMESTAMP""",
239+ p(userDid), p(userDid))
240+
241+proc getUnreadCount*(g: GleanDb, userDid, feedUrl, category: string): int =
242+ var sql = """
243+ SELECT count(*)
244+ FROM articles.articles a
245+ JOIN articles.subscriptions s
246+ ON s.feed_url = a.feed_url AND s.user_did = ?
247+ LEFT JOIN articles.read_state r
248+ ON r.article_id = a.id AND r.user_did = ?
249+ WHERE COALESCE(r.is_read, 0) = 0"""
250+ var params = @[p(userDid), p(userDid)]
251+ if feedUrl.len > 0:
252+ sql &= " AND a.feed_url = ?"
253+ params.add p(feedUrl)
254+ if category.len > 0:
255+ sql &= " AND s.category = ?"
256+ params.add p(category)
257+ g.db.queryInt(sql, params).get(0).int
258+
259+proc isRead*(g: GleanDb, userDid: string, articleId: int64): bool =
260+ g.db.queryInt("""SELECT COALESCE(is_read, 0) FROM articles.read_state
261+ WHERE user_did = ? AND article_id = ?""",
262+ p(userDid), p(articleId)).get(0) != 0
263+
264+proc countNewArticles*(g: GleanDb, userDid: string, since: DateTime): int =
265+ g.db.queryInt("""
266+ SELECT count(*) FROM articles.articles a
267+ JOIN articles.subscriptions s
268+ ON s.feed_url = a.feed_url AND s.user_did = ?
269+ WHERE a.fetched_at > ?""",
270+ p(userDid), p(formatSqliteTime(since))).get(0).int
added spike/nim/atproto/feedstore.nim +269 -0
new file mode 100644
@@ -0,0 +1,269 @@
1+## Feeds and subscriptions, ported from internal/db/feed.go.
2+##
3+## `feeds` is global -- one row per URL however many people subscribe -- and
4+## `subscriptions` is the per-user join onto it. That split is why
5+## subscriber_count exists as a stored column rather than a COUNT(*): it is
6+## read on every listing and written only when someone subscribes.
7+
8+import std/[options, times]
9+import ./gleandb
10+import ./sqlite
11+
12+type
13+ Feed* = object
14+ feedUrl*: string
15+ title*: string
16+ siteUrl*: string
17+ description*: string
18+ feedType*: string
19+ faviconUrl*: string
20+ subscriberCount*: int
21+ errorCount*: int
22+ lastError*: string
23+ lastFetchedAt*: Option[DateTime]
24+
25+ Subscription* = object
26+ id*: int64
27+ userDid*: string
28+ feedUrl*: string
29+ feedTitle*: string
30+ category*: string
31+ uri*: string
32+ cid*: string
33+ addedAt*: Option[DateTime]
34+ unreadCount*: int
35+ faviconUrl*: string
36+
37+const SqliteTimeFormat = "yyyy-MM-dd HH:mm:ss"
38+
39+proc parseSqliteTime*(s: string): Option[DateTime] =
40+ ## SQLite has no date type; CURRENT_TIMESTAMP writes "YYYY-MM-DD HH:MM:SS"
41+ ## in UTC. Values written by Go's driver can carry a timezone and
42+ ## fractional seconds, so both shapes are accepted.
43+ if s.len == 0: return none(DateTime)
44+ for fmt in [SqliteTimeFormat, "yyyy-MM-dd'T'HH:mm:ss'Z'",
45+ "yyyy-MM-dd HH:mm:ss'.'fff", "yyyy-MM-dd'T'HH:mm:sszzz"]:
46+ try:
47+ return some(parse(s, fmt, utc()))
48+ except TimeParseError:
49+ continue
50+ none(DateTime)
51+
52+proc formatSqliteTime*(t: DateTime): string =
53+ t.utc.format(SqliteTimeFormat)
54+
55+proc readFeed(r: Row): Feed =
56+ Feed(
57+ feedUrl: r.str(0),
58+ title: r.str(1),
59+ siteUrl: r.str(2),
60+ description: r.str(3),
61+ feedType: r.str(4),
62+ faviconUrl: r.str(5),
63+ subscriberCount: r.i(6),
64+ errorCount: r.i(7),
65+ lastError: r.str(8),
66+ lastFetchedAt: parseSqliteTime(r.str(9)),
67+ )
68+
69+## Qualified with the `f` alias, which every query below uses. Bare names
70+## work until a query joins subscriptions -- which also has feed_url -- and
71+## then fail as ambiguous rather than picking the wrong one.
72+const feedColumns = """f.feed_url, f.title, f.site_url, f.description,
73+ f.feed_type, f.favicon_url, f.subscriber_count, f.error_count, f.last_error,
74+ f.last_fetched_at"""
75+
76+proc upsertFeed*(g: GleanDb, f: Feed) =
77+ ## Insert, or refresh the metadata of an existing feed.
78+ ##
79+ ## subscriber_count and the error columns are deliberately not touched: they
80+ ## are owned by subscribe/unsubscribe and by the fetcher, and a metadata
81+ ## refresh that reset them would silently revive dead feeds.
82+ g.db.run("""INSERT INTO articles.feeds
83+ (feed_url, title, site_url, description, feed_type, favicon_url)
84+ VALUES (?, ?, ?, ?, ?, ?)
85+ ON CONFLICT(feed_url) DO UPDATE SET
86+ title = excluded.title,
87+ site_url = excluded.site_url,
88+ description = excluded.description,
89+ feed_type = COALESCE(excluded.feed_type, articles.feeds.feed_type),
90+ favicon_url = COALESCE(excluded.favicon_url, articles.feeds.favicon_url)""",
91+ p(f.feedUrl), pOrNull(f.title), pOrNull(f.siteUrl),
92+ pOrNull(f.description), pOrNull(f.feedType), pOrNull(f.faviconUrl))
93+
94+proc getFeed*(g: GleanDb, feedUrl: string): Option[Feed] =
95+ g.db.queryFirst("SELECT " & feedColumns &
96+ " FROM articles.feeds f WHERE f.feed_url = ?",
97+ [p(feedUrl)], readFeed)
98+
99+proc getFeedsToFetch*(g: GleanDb, olderThan: Duration, limit: int): seq[Feed] =
100+ ## Feeds due for a refresh: never fetched, or last fetched before the
101+ ## cutoff. Dead feeds are excluded by error_count so a permanently broken
102+ ## URL does not consume a fetch slot every cycle.
103+ let cutoff = formatSqliteTime(now().utc - olderThan)
104+ g.db.queryAll("""SELECT """ & feedColumns & """
105+ FROM articles.feeds f
106+ WHERE f.subscriber_count > 0
107+ AND f.error_count < 10
108+ AND (f.last_fetched_at IS NULL OR f.last_fetched_at < ?)
109+ ORDER BY f.last_fetched_at IS NOT NULL, f.last_fetched_at
110+ LIMIT ?""",
111+ [p(cutoff), p(limit.int64)], readFeed)
112+
113+proc markFeedFetched*(g: GleanDb, feedUrl: string) =
114+ ## A successful fetch clears the error state, so a feed that recovers
115+ ## re-enters the rotation instead of staying dead.
116+ g.db.run("""UPDATE articles.feeds
117+ SET last_fetched_at = CURRENT_TIMESTAMP,
118+ last_error = NULL,
119+ error_count = 0
120+ WHERE feed_url = ?""", p(feedUrl))
121+
122+proc markFeedFetchError*(g: GleanDb, feedUrl, lastError: string) =
123+ g.db.run("""UPDATE articles.feeds
124+ SET last_fetched_at = CURRENT_TIMESTAMP,
125+ last_error = ?,
126+ error_count = error_count + 1
127+ WHERE feed_url = ?""", p(lastError), p(feedUrl))
128+
129+proc updateFeedFavicon*(g: GleanDb, feedUrl, faviconUrl: string) =
130+ g.db.run("UPDATE articles.feeds SET favicon_url = ? WHERE feed_url = ?",
131+ pOrNull(faviconUrl), p(feedUrl))
132+
133+# --- subscriptions ---------------------------------------------------------
134+
135+proc readSubscription(r: Row): Subscription =
136+ Subscription(
137+ id: r.i64(0),
138+ userDid: r.str(1),
139+ feedUrl: r.str(2),
140+ feedTitle: r.str(3),
141+ category: r.str(4),
142+ uri: r.str(5),
143+ cid: r.str(6),
144+ addedAt: parseSqliteTime(r.str(7)),
145+ unreadCount: r.i(8),
146+ faviconUrl: r.str(9),
147+ )
148+
149+proc createSubscription*(g: GleanDb, userDid, feedUrl, title, category,
150+ uri, cid: string) =
151+ ## Subscribe, and keep subscriber_count in step.
152+ ##
153+ ## Both statements run in one transaction because the count is derived: a
154+ ## crash between them leaves a number that no longer matches reality, and
155+ ## nothing recomputes it outside the nightly recount.
156+ g.db.transaction:
157+ g.db.run("""INSERT INTO articles.subscriptions
158+ (user_did, feed_url, title, category, uri, cid)
159+ VALUES (?, ?, ?, ?, ?, ?)
160+ ON CONFLICT(user_did, feed_url) DO UPDATE SET
161+ title = excluded.title,
162+ category = excluded.category,
163+ uri = COALESCE(excluded.uri, articles.subscriptions.uri),
164+ cid = COALESCE(excluded.cid, articles.subscriptions.cid)""",
165+ p(userDid), p(feedUrl), pOrNull(title), pOrNull(category),
166+ pOrNull(uri), pOrNull(cid))
167+ # Only on a genuine insert: re-subscribing must not inflate the count.
168+ if g.db.changes() > 0:
169+ g.db.run("""UPDATE articles.feeds
170+ SET subscriber_count = (
171+ SELECT count(*) FROM articles.subscriptions
172+ WHERE feed_url = ?)
173+ WHERE feed_url = ?""", p(feedUrl), p(feedUrl))
174+
175+proc deleteSubscription*(g: GleanDb, userDid, feedUrl: string) =
176+ g.db.transaction:
177+ g.db.run("""DELETE FROM articles.subscriptions
178+ WHERE user_did = ? AND feed_url = ?""", p(userDid), p(feedUrl))
179+ g.db.run("""UPDATE articles.feeds
180+ SET subscriber_count = (
181+ SELECT count(*) FROM articles.subscriptions
182+ WHERE feed_url = ?)
183+ WHERE feed_url = ?""", p(feedUrl), p(feedUrl))
184+
185+proc deleteAllSubscriptions*(g: GleanDb, userDid: string) =
186+ g.db.transaction:
187+ g.db.run("DELETE FROM articles.subscriptions WHERE user_did = ?", p(userDid))
188+ g.db.run("""UPDATE articles.feeds SET subscriber_count = (
189+ SELECT count(*) FROM articles.subscriptions s
190+ WHERE s.feed_url = articles.feeds.feed_url)""")
191+
192+proc getSubscription*(g: GleanDb, userDid, feedUrl: string): Option[Subscription] =
193+ g.db.queryFirst("""
194+ SELECT s.id, s.user_did, s.feed_url,
195+ COALESCE(s.title, f.title, s.feed_url), s.category, s.uri, s.cid,
196+ s.added_at, 0, f.favicon_url
197+ FROM articles.subscriptions s
198+ LEFT JOIN articles.feeds f ON f.feed_url = s.feed_url
199+ WHERE s.user_did = ? AND s.feed_url = ?""",
200+ [p(userDid), p(feedUrl)], readSubscription)
201+
202+proc listSubscriptions*(g: GleanDb, userDid, category: string,
203+ limit, offset: int): seq[Subscription] =
204+ ## The unread count is computed per row here rather than stored: it is
205+ ## per-user and changes constantly, so a cached column would be wrong more
206+ ## often than right.
207+ var sql = """
208+ SELECT s.id, s.user_did, s.feed_url,
209+ COALESCE(s.title, f.title, s.feed_url), s.category, s.uri, s.cid,
210+ s.added_at,
211+ (SELECT count(*) FROM articles.articles a
212+ LEFT JOIN articles.read_state r
213+ ON r.article_id = a.id AND r.user_did = s.user_did
214+ WHERE a.feed_url = s.feed_url AND COALESCE(r.is_read, 0) = 0),
215+ f.favicon_url
216+ FROM articles.subscriptions s
217+ LEFT JOIN articles.feeds f ON f.feed_url = s.feed_url
218+ WHERE s.user_did = ?"""
219+ var params = @[p(userDid)]
220+ if category.len > 0:
221+ sql &= " AND s.category = ?"
222+ params.add p(category)
223+ sql &= " ORDER BY COALESCE(s.title, f.title, s.feed_url) COLLATE NOCASE LIMIT ? OFFSET ?"
224+ params.add p(limit.int64)
225+ params.add p(offset.int64)
226+ g.db.queryAll(sql, params, readSubscription)
227+
228+proc getSubscriptionCount*(g: GleanDb, userDid: string): int =
229+ g.db.queryInt("SELECT count(*) FROM articles.subscriptions WHERE user_did = ?",
230+ p(userDid)).get(0).int
231+
232+proc getCategories*(g: GleanDb, userDid: string): seq[string] =
233+ for r in g.db.query("""SELECT DISTINCT category FROM articles.subscriptions
234+ WHERE user_did = ? AND category IS NOT NULL
235+ AND category != ''
236+ ORDER BY category COLLATE NOCASE""", p(userDid)):
237+ result.add r.str(0)
238+
239+proc listDeadFeeds*(g: GleanDb, userDid: string, threshold: int): seq[Feed] =
240+ ## Subscribed feeds that keep failing, so the UI can offer a retry rather
241+ ## than silently showing nothing.
242+ g.db.queryAll("""SELECT """ & feedColumns & """
243+ FROM articles.feeds f
244+ JOIN articles.subscriptions s ON s.feed_url = f.feed_url
245+ WHERE s.user_did = ? AND f.error_count >= ?
246+ ORDER BY f.error_count DESC""",
247+ [p(userDid), p(threshold.int64)], readFeed)
248+
249+proc recountSubscriberCounts*(g: GleanDb) =
250+ ## Repair pass. The counts are maintained incrementally, so anything that
251+ ## interrupts a subscribe leaves them drifted.
252+ g.db.run("""UPDATE articles.feeds SET subscriber_count = (
253+ SELECT count(*) FROM articles.subscriptions s
254+ WHERE s.feed_url = articles.feeds.feed_url)""")
255+
256+proc batchUpsertFeeds*(g: GleanDb, feeds: seq[Feed]) =
257+ if feeds.len == 0: return
258+ g.db.transaction:
259+ var ps = g.db.prepared("""
260+ INSERT INTO articles.feeds
261+ (feed_url, title, site_url, description, feed_type, favicon_url)
262+ VALUES (?, ?, ?, ?, ?, ?)
263+ ON CONFLICT(feed_url) DO UPDATE SET
264+ title = COALESCE(excluded.title, articles.feeds.title),
265+ site_url = COALESCE(excluded.site_url, articles.feeds.site_url)""")
266+ defer: ps.finalize()
267+ for f in feeds:
268+ ps.exec(p(f.feedUrl), pOrNull(f.title), pOrNull(f.siteUrl),
269+ pOrNull(f.description), pOrNull(f.feedType), pOrNull(f.faviconUrl))
new file mode 100644
@@ -0,0 +1,269 @@
1+## Feeds and subscriptions, ported from internal/db/feed.go.
2+##
3+## `feeds` is global -- one row per URL however many people subscribe -- and
4+## `subscriptions` is the per-user join onto it. That split is why
5+## subscriber_count exists as a stored column rather than a COUNT(*): it is
6+## read on every listing and written only when someone subscribes.
7+
8+import std/[options, times]
9+import ./gleandb
10+import ./sqlite
11+
12+type
13+ Feed* = object
14+ feedUrl*: string
15+ title*: string
16+ siteUrl*: string
17+ description*: string
18+ feedType*: string
19+ faviconUrl*: string
20+ subscriberCount*: int
21+ errorCount*: int
22+ lastError*: string
23+ lastFetchedAt*: Option[DateTime]
24+
25+ Subscription* = object
26+ id*: int64
27+ userDid*: string
28+ feedUrl*: string
29+ feedTitle*: string
30+ category*: string
31+ uri*: string
32+ cid*: string
33+ addedAt*: Option[DateTime]
34+ unreadCount*: int
35+ faviconUrl*: string
36+
37+const SqliteTimeFormat = "yyyy-MM-dd HH:mm:ss"
38+
39+proc parseSqliteTime*(s: string): Option[DateTime] =
40+ ## SQLite has no date type; CURRENT_TIMESTAMP writes "YYYY-MM-DD HH:MM:SS"
41+ ## in UTC. Values written by Go's driver can carry a timezone and
42+ ## fractional seconds, so both shapes are accepted.
43+ if s.len == 0: return none(DateTime)
44+ for fmt in [SqliteTimeFormat, "yyyy-MM-dd'T'HH:mm:ss'Z'",
45+ "yyyy-MM-dd HH:mm:ss'.'fff", "yyyy-MM-dd'T'HH:mm:sszzz"]:
46+ try:
47+ return some(parse(s, fmt, utc()))
48+ except TimeParseError:
49+ continue
50+ none(DateTime)
51+
52+proc formatSqliteTime*(t: DateTime): string =
53+ t.utc.format(SqliteTimeFormat)
54+
55+proc readFeed(r: Row): Feed =
56+ Feed(
57+ feedUrl: r.str(0),
58+ title: r.str(1),
59+ siteUrl: r.str(2),
60+ description: r.str(3),
61+ feedType: r.str(4),
62+ faviconUrl: r.str(5),
63+ subscriberCount: r.i(6),
64+ errorCount: r.i(7),
65+ lastError: r.str(8),
66+ lastFetchedAt: parseSqliteTime(r.str(9)),
67+ )
68+
69+## Qualified with the `f` alias, which every query below uses. Bare names
70+## work until a query joins subscriptions -- which also has feed_url -- and
71+## then fail as ambiguous rather than picking the wrong one.
72+const feedColumns = """f.feed_url, f.title, f.site_url, f.description,
73+ f.feed_type, f.favicon_url, f.subscriber_count, f.error_count, f.last_error,
74+ f.last_fetched_at"""
75+
76+proc upsertFeed*(g: GleanDb, f: Feed) =
77+ ## Insert, or refresh the metadata of an existing feed.
78+ ##
79+ ## subscriber_count and the error columns are deliberately not touched: they
80+ ## are owned by subscribe/unsubscribe and by the fetcher, and a metadata
81+ ## refresh that reset them would silently revive dead feeds.
82+ g.db.run("""INSERT INTO articles.feeds
83+ (feed_url, title, site_url, description, feed_type, favicon_url)
84+ VALUES (?, ?, ?, ?, ?, ?)
85+ ON CONFLICT(feed_url) DO UPDATE SET
86+ title = excluded.title,
87+ site_url = excluded.site_url,
88+ description = excluded.description,
89+ feed_type = COALESCE(excluded.feed_type, articles.feeds.feed_type),
90+ favicon_url = COALESCE(excluded.favicon_url, articles.feeds.favicon_url)""",
91+ p(f.feedUrl), pOrNull(f.title), pOrNull(f.siteUrl),
92+ pOrNull(f.description), pOrNull(f.feedType), pOrNull(f.faviconUrl))
93+
94+proc getFeed*(g: GleanDb, feedUrl: string): Option[Feed] =
95+ g.db.queryFirst("SELECT " & feedColumns &
96+ " FROM articles.feeds f WHERE f.feed_url = ?",
97+ [p(feedUrl)], readFeed)
98+
99+proc getFeedsToFetch*(g: GleanDb, olderThan: Duration, limit: int): seq[Feed] =
100+ ## Feeds due for a refresh: never fetched, or last fetched before the
101+ ## cutoff. Dead feeds are excluded by error_count so a permanently broken
102+ ## URL does not consume a fetch slot every cycle.
103+ let cutoff = formatSqliteTime(now().utc - olderThan)
104+ g.db.queryAll("""SELECT """ & feedColumns & """
105+ FROM articles.feeds f
106+ WHERE f.subscriber_count > 0
107+ AND f.error_count < 10
108+ AND (f.last_fetched_at IS NULL OR f.last_fetched_at < ?)
109+ ORDER BY f.last_fetched_at IS NOT NULL, f.last_fetched_at
110+ LIMIT ?""",
111+ [p(cutoff), p(limit.int64)], readFeed)
112+
113+proc markFeedFetched*(g: GleanDb, feedUrl: string) =
114+ ## A successful fetch clears the error state, so a feed that recovers
115+ ## re-enters the rotation instead of staying dead.
116+ g.db.run("""UPDATE articles.feeds
117+ SET last_fetched_at = CURRENT_TIMESTAMP,
118+ last_error = NULL,
119+ error_count = 0
120+ WHERE feed_url = ?""", p(feedUrl))
121+
122+proc markFeedFetchError*(g: GleanDb, feedUrl, lastError: string) =
123+ g.db.run("""UPDATE articles.feeds
124+ SET last_fetched_at = CURRENT_TIMESTAMP,
125+ last_error = ?,
126+ error_count = error_count + 1
127+ WHERE feed_url = ?""", p(lastError), p(feedUrl))
128+
129+proc updateFeedFavicon*(g: GleanDb, feedUrl, faviconUrl: string) =
130+ g.db.run("UPDATE articles.feeds SET favicon_url = ? WHERE feed_url = ?",
131+ pOrNull(faviconUrl), p(feedUrl))
132+
133+# --- subscriptions ---------------------------------------------------------
134+
135+proc readSubscription(r: Row): Subscription =
136+ Subscription(
137+ id: r.i64(0),
138+ userDid: r.str(1),
139+ feedUrl: r.str(2),
140+ feedTitle: r.str(3),
141+ category: r.str(4),
142+ uri: r.str(5),
143+ cid: r.str(6),
144+ addedAt: parseSqliteTime(r.str(7)),
145+ unreadCount: r.i(8),
146+ faviconUrl: r.str(9),
147+ )
148+
149+proc createSubscription*(g: GleanDb, userDid, feedUrl, title, category,
150+ uri, cid: string) =
151+ ## Subscribe, and keep subscriber_count in step.
152+ ##
153+ ## Both statements run in one transaction because the count is derived: a
154+ ## crash between them leaves a number that no longer matches reality, and
155+ ## nothing recomputes it outside the nightly recount.
156+ g.db.transaction:
157+ g.db.run("""INSERT INTO articles.subscriptions
158+ (user_did, feed_url, title, category, uri, cid)
159+ VALUES (?, ?, ?, ?, ?, ?)
160+ ON CONFLICT(user_did, feed_url) DO UPDATE SET
161+ title = excluded.title,
162+ category = excluded.category,
163+ uri = COALESCE(excluded.uri, articles.subscriptions.uri),
164+ cid = COALESCE(excluded.cid, articles.subscriptions.cid)""",
165+ p(userDid), p(feedUrl), pOrNull(title), pOrNull(category),
166+ pOrNull(uri), pOrNull(cid))
167+ # Only on a genuine insert: re-subscribing must not inflate the count.
168+ if g.db.changes() > 0:
169+ g.db.run("""UPDATE articles.feeds
170+ SET subscriber_count = (
171+ SELECT count(*) FROM articles.subscriptions
172+ WHERE feed_url = ?)
173+ WHERE feed_url = ?""", p(feedUrl), p(feedUrl))
174+
175+proc deleteSubscription*(g: GleanDb, userDid, feedUrl: string) =
176+ g.db.transaction:
177+ g.db.run("""DELETE FROM articles.subscriptions
178+ WHERE user_did = ? AND feed_url = ?""", p(userDid), p(feedUrl))
179+ g.db.run("""UPDATE articles.feeds
180+ SET subscriber_count = (
181+ SELECT count(*) FROM articles.subscriptions
182+ WHERE feed_url = ?)
183+ WHERE feed_url = ?""", p(feedUrl), p(feedUrl))
184+
185+proc deleteAllSubscriptions*(g: GleanDb, userDid: string) =
186+ g.db.transaction:
187+ g.db.run("DELETE FROM articles.subscriptions WHERE user_did = ?", p(userDid))
188+ g.db.run("""UPDATE articles.feeds SET subscriber_count = (
189+ SELECT count(*) FROM articles.subscriptions s
190+ WHERE s.feed_url = articles.feeds.feed_url)""")
191+
192+proc getSubscription*(g: GleanDb, userDid, feedUrl: string): Option[Subscription] =
193+ g.db.queryFirst("""
194+ SELECT s.id, s.user_did, s.feed_url,
195+ COALESCE(s.title, f.title, s.feed_url), s.category, s.uri, s.cid,
196+ s.added_at, 0, f.favicon_url
197+ FROM articles.subscriptions s
198+ LEFT JOIN articles.feeds f ON f.feed_url = s.feed_url
199+ WHERE s.user_did = ? AND s.feed_url = ?""",
200+ [p(userDid), p(feedUrl)], readSubscription)
201+
202+proc listSubscriptions*(g: GleanDb, userDid, category: string,
203+ limit, offset: int): seq[Subscription] =
204+ ## The unread count is computed per row here rather than stored: it is
205+ ## per-user and changes constantly, so a cached column would be wrong more
206+ ## often than right.
207+ var sql = """
208+ SELECT s.id, s.user_did, s.feed_url,
209+ COALESCE(s.title, f.title, s.feed_url), s.category, s.uri, s.cid,
210+ s.added_at,
211+ (SELECT count(*) FROM articles.articles a
212+ LEFT JOIN articles.read_state r
213+ ON r.article_id = a.id AND r.user_did = s.user_did
214+ WHERE a.feed_url = s.feed_url AND COALESCE(r.is_read, 0) = 0),
215+ f.favicon_url
216+ FROM articles.subscriptions s
217+ LEFT JOIN articles.feeds f ON f.feed_url = s.feed_url
218+ WHERE s.user_did = ?"""
219+ var params = @[p(userDid)]
220+ if category.len > 0:
221+ sql &= " AND s.category = ?"
222+ params.add p(category)
223+ sql &= " ORDER BY COALESCE(s.title, f.title, s.feed_url) COLLATE NOCASE LIMIT ? OFFSET ?"
224+ params.add p(limit.int64)
225+ params.add p(offset.int64)
226+ g.db.queryAll(sql, params, readSubscription)
227+
228+proc getSubscriptionCount*(g: GleanDb, userDid: string): int =
229+ g.db.queryInt("SELECT count(*) FROM articles.subscriptions WHERE user_did = ?",
230+ p(userDid)).get(0).int
231+
232+proc getCategories*(g: GleanDb, userDid: string): seq[string] =
233+ for r in g.db.query("""SELECT DISTINCT category FROM articles.subscriptions
234+ WHERE user_did = ? AND category IS NOT NULL
235+ AND category != ''
236+ ORDER BY category COLLATE NOCASE""", p(userDid)):
237+ result.add r.str(0)
238+
239+proc listDeadFeeds*(g: GleanDb, userDid: string, threshold: int): seq[Feed] =
240+ ## Subscribed feeds that keep failing, so the UI can offer a retry rather
241+ ## than silently showing nothing.
242+ g.db.queryAll("""SELECT """ & feedColumns & """
243+ FROM articles.feeds f
244+ JOIN articles.subscriptions s ON s.feed_url = f.feed_url
245+ WHERE s.user_did = ? AND f.error_count >= ?
246+ ORDER BY f.error_count DESC""",
247+ [p(userDid), p(threshold.int64)], readFeed)
248+
249+proc recountSubscriberCounts*(g: GleanDb) =
250+ ## Repair pass. The counts are maintained incrementally, so anything that
251+ ## interrupts a subscribe leaves them drifted.
252+ g.db.run("""UPDATE articles.feeds SET subscriber_count = (
253+ SELECT count(*) FROM articles.subscriptions s
254+ WHERE s.feed_url = articles.feeds.feed_url)""")
255+
256+proc batchUpsertFeeds*(g: GleanDb, feeds: seq[Feed]) =
257+ if feeds.len == 0: return
258+ g.db.transaction:
259+ var ps = g.db.prepared("""
260+ INSERT INTO articles.feeds
261+ (feed_url, title, site_url, description, feed_type, favicon_url)
262+ VALUES (?, ?, ?, ?, ?, ?)
263+ ON CONFLICT(feed_url) DO UPDATE SET
264+ title = COALESCE(excluded.title, articles.feeds.title),
265+ site_url = COALESCE(excluded.site_url, articles.feeds.site_url)""")
266+ defer: ps.finalize()
267+ for f in feeds:
268+ ps.exec(p(f.feedUrl), pOrNull(f.title), pOrNull(f.siteUrl),
269+ pOrNull(f.description), pOrNull(f.feedType), pOrNull(f.faviconUrl))
modified spike/nim/atproto/sqlite.nim +128 -0
@@ -45,6 +45,10 @@ proc sqlite3_bind_int64(s: pointer, col: cint, v: int64): cint
4545 proc sqlite3_bind_blob(s: pointer, col: cint, data: pointer, n: cint,
4646 destructor: pointer): cint
4747 proc sqlite3_libversion(): cstring
48+proc sqlite3_column_type(s: pointer, col: cint): cint
49+proc sqlite3_last_insert_rowid(db: pointer): int64
50+proc sqlite3_changes(db: pointer): cint
51+proc sqlite3_clear_bindings(s: pointer): cint
4852 {.pop.}
4953
5054 proc sqlite3_vec_init(db: pointer, errMsg: ptr cstring,
@@ -163,3 +167,127 @@ iterator rows*(db: Db, sql: string, params: varargs[Param]): seq[string] =
163167 let raw = sqlite3_column_text(stmt, i.cint)
164168 row[i] = if raw == nil: "" else: $raw
165169 yield row
170+
171+
172+# --- typed row access ------------------------------------------------------
173+
174+const SQLITE_NULL* = 5
175+
176+type Row* = object
177+ ## A cursor positioned on one result row. Valid only until the next step.
178+ stmt: pointer
179+
180+proc isNull*(r: Row, col: int): bool =
181+ sqlite3_column_type(r.stmt, col.cint) == SQLITE_NULL
182+
183+proc str*(r: Row, col: int): string =
184+ ## NULL reads as "". Most of Glean's text columns are nullable but mean the
185+ ## empty string, and the Go side collapses them the same way at the wire
186+ ## boundary (see nullStr in internal/server/api.go).
187+ if r.isNull(col): return ""
188+ let raw = sqlite3_column_text(r.stmt, col.cint)
189+ if raw == nil: "" else: $raw
190+
191+proc strOpt*(r: Row, col: int): Option[string] =
192+ ## For the few places where NULL and "" genuinely differ.
193+ if r.isNull(col): none(string) else: some(r.str(col))
194+
195+proc i64*(r: Row, col: int): int64 =
196+ if r.isNull(col): 0'i64 else: sqlite3_column_int64(r.stmt, col.cint)
197+
198+proc i*(r: Row, col: int): int = r.i64(col).int
199+
200+proc f*(r: Row, col: int): float =
201+ if r.isNull(col): 0.0 else: sqlite3_column_double(r.stmt, col.cint).float
202+
203+proc b*(r: Row, col: int): bool = r.i64(col) != 0
204+
205+iterator query*(db: Db, sql: string, params: varargs[Param]): Row =
206+ ## Typed row iteration. The Row borrows the statement, so copy anything you
207+ ## need out of it before the next iteration.
208+ let stmt = prepare(db, sql, @params)
209+ defer: discard sqlite3_finalize(stmt)
210+ while sqlite3_step(stmt) == SQLITE_ROW:
211+ yield Row(stmt: stmt)
212+
213+proc queryFirst*[T](db: Db, sql: string, params: openArray[Param],
214+ read: proc(r: Row): T): Option[T] =
215+ let stmt = prepare(db, sql, params)
216+ defer: discard sqlite3_finalize(stmt)
217+ if sqlite3_step(stmt) != SQLITE_ROW:
218+ return none(T)
219+ some(read(Row(stmt: stmt)))
220+
221+proc queryAll*[T](db: Db, sql: string, params: openArray[Param],
222+ read: proc(r: Row): T): seq[T] =
223+ let stmt = prepare(db, sql, params)
224+ defer: discard sqlite3_finalize(stmt)
225+ while sqlite3_step(stmt) == SQLITE_ROW:
226+ result.add read(Row(stmt: stmt))
227+
228+proc lastInsertId*(db: Db): int64 = sqlite3_last_insert_rowid(db.handle)
229+proc changes*(db: Db): int = sqlite3_changes(db.handle).int
230+
231+# --- transactions ----------------------------------------------------------
232+
233+template transaction*(db: Db, body: untyped) =
234+ ## Run `body` in a transaction, rolling back if it raises.
235+ ##
236+ ## Batch ingest is the reason this exists: a few thousand inserts outside a
237+ ## transaction is a few thousand fsyncs, which turns a feed refresh into a
238+ ## minutes-long operation.
239+ db.exec "BEGIN IMMEDIATE"
240+ try:
241+ body
242+ db.exec "COMMIT"
243+ except CatchableError:
244+ try: db.exec "ROLLBACK"
245+ except CatchableError: discard
246+ raise
247+
248+# --- reusable prepared statements ------------------------------------------
249+
250+type Prepared* = object
251+ ## A statement prepared once and stepped many times. Worth it inside a
252+ ## batch; pointless outside one.
253+ stmt: pointer
254+ db: pointer
255+
256+proc prepared*(db: Db, sql: string): Prepared =
257+ var stmt: pointer
258+ check(db.handle, sqlite3_prepare_v2(db.handle, sql.cstring, -1, stmt, nil),
259+ "prepare")
260+ Prepared(stmt: stmt, db: db.handle)
261+
262+proc exec*(ps: Prepared, params: varargs[Param]) =
263+ discard sqlite3_reset(ps.stmt)
264+ discard sqlite3_clear_bindings(ps.stmt)
265+ for i, prm in params:
266+ let col = (i + 1).cint
267+ let rc =
268+ if prm.isNull: SQLITE_OK.cint
269+ else:
270+ case prm.kind
271+ of 0: sqlite3_bind_text(ps.stmt, col, prm.s.cstring, prm.s.len.cint,
272+ SQLITE_TRANSIENT)
273+ of 1: sqlite3_bind_int64(ps.stmt, col, prm.i)
274+ of 2:
275+ if prm.b.len == 0:
276+ sqlite3_bind_blob(ps.stmt, col, nil, 0, SQLITE_TRANSIENT)
277+ else:
278+ sqlite3_bind_blob(ps.stmt, col, prm.b[0].unsafeAddr, prm.b.len.cint,
279+ SQLITE_TRANSIENT)
280+ if rc != SQLITE_OK:
281+ raise newException(SqliteError, "binding parameter " & $col)
282+ check(ps.db, sqlite3_step(ps.stmt), "step")
283+
284+proc finalize*(ps: var Prepared) =
285+ if ps.stmt != nil:
286+ discard sqlite3_finalize(ps.stmt)
287+ ps.stmt = nil
288+
289+proc nullParam*(): Param = Param(isNull: true)
290+
291+proc pOrNull*(s: string): Param =
292+ ## Empty text becomes NULL, matching the Go side's nilIfEmpty.
293+ if s.len == 0: nullParam() else: p(s)
@@ -45,6 +45,10 @@ proc sqlite3_bind_int64(s: pointer, col: cint, v: int64): cint
45 proc sqlite3_bind_blob(s: pointer, col: cint, data: pointer, n: cint,45 proc sqlite3_bind_blob(s: pointer, col: cint, data: pointer, n: cint,
46 destructor: pointer): cint46 destructor: pointer): cint
47 proc sqlite3_libversion(): cstring47 proc sqlite3_libversion(): cstring
48+proc sqlite3_column_type(s: pointer, col: cint): cint
49+proc sqlite3_last_insert_rowid(db: pointer): int64
50+proc sqlite3_changes(db: pointer): cint
51+proc sqlite3_clear_bindings(s: pointer): cint
48 {.pop.}52 {.pop.}
49 53
50 proc sqlite3_vec_init(db: pointer, errMsg: ptr cstring,54 proc sqlite3_vec_init(db: pointer, errMsg: ptr cstring,
@@ -163,3 +167,127 @@ iterator rows*(db: Db, sql: string, params: varargs[Param]): seq[string] =
163 let raw = sqlite3_column_text(stmt, i.cint)167 let raw = sqlite3_column_text(stmt, i.cint)
164 row[i] = if raw == nil: "" else: $raw168 row[i] = if raw == nil: "" else: $raw
165 yield row169 yield row
170+
171+
172+# --- typed row access ------------------------------------------------------
173+
174+const SQLITE_NULL* = 5
175+
176+type Row* = object
177+ ## A cursor positioned on one result row. Valid only until the next step.
178+ stmt: pointer
179+
180+proc isNull*(r: Row, col: int): bool =
181+ sqlite3_column_type(r.stmt, col.cint) == SQLITE_NULL
182+
183+proc str*(r: Row, col: int): string =
184+ ## NULL reads as "". Most of Glean's text columns are nullable but mean the
185+ ## empty string, and the Go side collapses them the same way at the wire
186+ ## boundary (see nullStr in internal/server/api.go).
187+ if r.isNull(col): return ""
188+ let raw = sqlite3_column_text(r.stmt, col.cint)
189+ if raw == nil: "" else: $raw
190+
191+proc strOpt*(r: Row, col: int): Option[string] =
192+ ## For the few places where NULL and "" genuinely differ.
193+ if r.isNull(col): none(string) else: some(r.str(col))
194+
195+proc i64*(r: Row, col: int): int64 =
196+ if r.isNull(col): 0'i64 else: sqlite3_column_int64(r.stmt, col.cint)
197+
198+proc i*(r: Row, col: int): int = r.i64(col).int
199+
200+proc f*(r: Row, col: int): float =
201+ if r.isNull(col): 0.0 else: sqlite3_column_double(r.stmt, col.cint).float
202+
203+proc b*(r: Row, col: int): bool = r.i64(col) != 0
204+
205+iterator query*(db: Db, sql: string, params: varargs[Param]): Row =
206+ ## Typed row iteration. The Row borrows the statement, so copy anything you
207+ ## need out of it before the next iteration.
208+ let stmt = prepare(db, sql, @params)
209+ defer: discard sqlite3_finalize(stmt)
210+ while sqlite3_step(stmt) == SQLITE_ROW:
211+ yield Row(stmt: stmt)
212+
213+proc queryFirst*[T](db: Db, sql: string, params: openArray[Param],
214+ read: proc(r: Row): T): Option[T] =
215+ let stmt = prepare(db, sql, params)
216+ defer: discard sqlite3_finalize(stmt)
217+ if sqlite3_step(stmt) != SQLITE_ROW:
218+ return none(T)
219+ some(read(Row(stmt: stmt)))
220+
221+proc queryAll*[T](db: Db, sql: string, params: openArray[Param],
222+ read: proc(r: Row): T): seq[T] =
223+ let stmt = prepare(db, sql, params)
224+ defer: discard sqlite3_finalize(stmt)
225+ while sqlite3_step(stmt) == SQLITE_ROW:
226+ result.add read(Row(stmt: stmt))
227+
228+proc lastInsertId*(db: Db): int64 = sqlite3_last_insert_rowid(db.handle)
229+proc changes*(db: Db): int = sqlite3_changes(db.handle).int
230+
231+# --- transactions ----------------------------------------------------------
232+
233+template transaction*(db: Db, body: untyped) =
234+ ## Run `body` in a transaction, rolling back if it raises.
235+ ##
236+ ## Batch ingest is the reason this exists: a few thousand inserts outside a
237+ ## transaction is a few thousand fsyncs, which turns a feed refresh into a
238+ ## minutes-long operation.
239+ db.exec "BEGIN IMMEDIATE"
240+ try:
241+ body
242+ db.exec "COMMIT"
243+ except CatchableError:
244+ try: db.exec "ROLLBACK"
245+ except CatchableError: discard
246+ raise
247+
248+# --- reusable prepared statements ------------------------------------------
249+
250+type Prepared* = object
251+ ## A statement prepared once and stepped many times. Worth it inside a
252+ ## batch; pointless outside one.
253+ stmt: pointer
254+ db: pointer
255+
256+proc prepared*(db: Db, sql: string): Prepared =
257+ var stmt: pointer
258+ check(db.handle, sqlite3_prepare_v2(db.handle, sql.cstring, -1, stmt, nil),
259+ "prepare")
260+ Prepared(stmt: stmt, db: db.handle)
261+
262+proc exec*(ps: Prepared, params: varargs[Param]) =
263+ discard sqlite3_reset(ps.stmt)
264+ discard sqlite3_clear_bindings(ps.stmt)
265+ for i, prm in params:
266+ let col = (i + 1).cint
267+ let rc =
268+ if prm.isNull: SQLITE_OK.cint
269+ else:
270+ case prm.kind
271+ of 0: sqlite3_bind_text(ps.stmt, col, prm.s.cstring, prm.s.len.cint,
272+ SQLITE_TRANSIENT)
273+ of 1: sqlite3_bind_int64(ps.stmt, col, prm.i)
274+ of 2:
275+ if prm.b.len == 0:
276+ sqlite3_bind_blob(ps.stmt, col, nil, 0, SQLITE_TRANSIENT)
277+ else:
278+ sqlite3_bind_blob(ps.stmt, col, prm.b[0].unsafeAddr, prm.b.len.cint,
279+ SQLITE_TRANSIENT)
280+ if rc != SQLITE_OK:
281+ raise newException(SqliteError, "binding parameter " & $col)
282+ check(ps.db, sqlite3_step(ps.stmt), "step")
283+
284+proc finalize*(ps: var Prepared) =
285+ if ps.stmt != nil:
286+ discard sqlite3_finalize(ps.stmt)
287+ ps.stmt = nil
288+
289+proc nullParam*(): Param = Param(isNull: true)
290+
291+proc pOrNull*(s: string): Param =
292+ ## Empty text becomes NULL, matching the Go side's nilIfEmpty.
293+ if s.len == 0: nullParam() else: p(s)
modified spike/nim/readme.md +50 -0
@@ -17,6 +17,7 @@ nim c -r -d:ssl jetstream_probe.nim # firehose
1717 nim c -r -d:ssl reconnect_probe.nim # supervisor
1818 nim c -r -d:ssl store_probe.nim # persistence
1919 nim c -r -d:ssl gleandb_probe.nim # database
20+nim c -r -d:ssl stores_probe.nim # queries
2021 ```
2122
2223 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.
@@ -285,6 +286,55 @@ The remaining query surface — `article.go`, `feed.go`, `social.go`,
285286 `retention.go` and friends — is mechanical against this base, and large. It
286287 is the first part of the port where the work is volume rather than risk.
287288
289+## 9. Article and feed queries
290+
291+`atproto/feedstore.nim` and `atproto/articlestore.nim` port the core of
292+`internal/db/feed.go` and `internal/db/article.go`, with `sqlite.nim` grown to
293+carry them: typed nullable row access, transactions, and prepared statements
294+that can be stepped many times.
295+
296+Read state is the shape everything else follows. It is per-user and lives in
297+its own table, so a listing is always a LEFT JOIN and "unread" means *no row*
298+— hence `COALESCE(r.is_read, 0)` throughout rather than a column on the
299+article.
300+
301+Two behaviours carried over deliberately, both about not resurfacing things a
302+reader has dealt with:
303+
304+- Articles already older than the retention window are **not ingested**.
305+ Writing them only to purge them churns the database.
306+- After ingest, read state is **restored from history**, because a purged
307+ article that reappears gets a new surrogate id and would otherwise come
308+ back unread.
309+
310+The probe targets the ways these can be quietly wrong: a mark-all-read whose
311+scope clears the account instead of one feed, a subscriber count that drifts,
312+a listing that leaks another user's feeds, a NULL date sorting to the top.
313+
314+### Four bugs the probe caught
315+
316+All four were wrong assumptions about the schema or about SQLite, and none
317+would have raised at compile time:
318+
319+- `likes` keys on **`author_did`**, not `user_did` — a like is an ATProto
320+ record with an author, not a user-scoped row.
321+- FTS5 requires the **bare table name** on the left of `MATCH`. Neither an
322+ alias nor a schema-qualified name is accepted, so the match moved into a
323+ subquery, which also carries `rank` out for ordering.
324+- The shared feed column list used bare names, which work until a query joins
325+ `subscriptions` — which also has `feed_url` — and then fail as ambiguous.
326+ Now qualified with the table alias.
327+- The probe's own first draft assumed `users` had a `handle` column. It does
328+ not; handles are resolved from atproto rather than stored.
329+
330+### Scope
331+
332+Roughly the core of the two files by usage: feeds, subscriptions, listing,
333+filtering, read state, counts, search, retention, dead feeds. Not ported:
334+`BatchReconcileSubscriptions`, OPML-adjacent helpers,
335+`GetNextArticleID`'s navigation logic, and the several list variants the
336+recommendation engine uses. They are mechanical against these patterns.
337+
288338 ## What this does and does not prove
289339
290340 Proven: the cryptography, the storage layer, and the OAuth flow up to user
@@ -17,6 +17,7 @@ nim c -r -d:ssl jetstream_probe.nim # firehose
17 nim c -r -d:ssl reconnect_probe.nim # supervisor17 nim c -r -d:ssl reconnect_probe.nim # supervisor
18 nim c -r -d:ssl store_probe.nim # persistence18 nim c -r -d:ssl store_probe.nim # persistence
19 nim c -r -d:ssl gleandb_probe.nim # database19 nim c -r -d:ssl gleandb_probe.nim # database
20+nim c -r -d:ssl stores_probe.nim # queries
20 ```21 ```
21 22
22 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.23 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.
@@ -285,6 +286,55 @@ The remaining query surface — `article.go`, `feed.go`, `social.go`,
285 `retention.go` and friends — is mechanical against this base, and large. It286 `retention.go` and friends — is mechanical against this base, and large. It
286 is the first part of the port where the work is volume rather than risk.287 is the first part of the port where the work is volume rather than risk.
287 288
289+## 9. Article and feed queries
290+
291+`atproto/feedstore.nim` and `atproto/articlestore.nim` port the core of
292+`internal/db/feed.go` and `internal/db/article.go`, with `sqlite.nim` grown to
293+carry them: typed nullable row access, transactions, and prepared statements
294+that can be stepped many times.
295+
296+Read state is the shape everything else follows. It is per-user and lives in
297+its own table, so a listing is always a LEFT JOIN and "unread" means *no row*
298+— hence `COALESCE(r.is_read, 0)` throughout rather than a column on the
299+article.
300+
301+Two behaviours carried over deliberately, both about not resurfacing things a
302+reader has dealt with:
303+
304+- Articles already older than the retention window are **not ingested**.
305+ Writing them only to purge them churns the database.
306+- After ingest, read state is **restored from history**, because a purged
307+ article that reappears gets a new surrogate id and would otherwise come
308+ back unread.
309+
310+The probe targets the ways these can be quietly wrong: a mark-all-read whose
311+scope clears the account instead of one feed, a subscriber count that drifts,
312+a listing that leaks another user's feeds, a NULL date sorting to the top.
313+
314+### Four bugs the probe caught
315+
316+All four were wrong assumptions about the schema or about SQLite, and none
317+would have raised at compile time:
318+
319+- `likes` keys on **`author_did`**, not `user_did` — a like is an ATProto
320+ record with an author, not a user-scoped row.
321+- FTS5 requires the **bare table name** on the left of `MATCH`. Neither an
322+ alias nor a schema-qualified name is accepted, so the match moved into a
323+ subquery, which also carries `rank` out for ordering.
324+- The shared feed column list used bare names, which work until a query joins
325+ `subscriptions` — which also has `feed_url` — and then fail as ambiguous.
326+ Now qualified with the table alias.
327+- The probe's own first draft assumed `users` had a `handle` column. It does
328+ not; handles are resolved from atproto rather than stored.
329+
330+### Scope
331+
332+Roughly the core of the two files by usage: feeds, subscriptions, listing,
333+filtering, read state, counts, search, retention, dead feeds. Not ported:
334+`BatchReconcileSubscriptions`, OPML-adjacent helpers,
335+`GetNextArticleID`'s navigation logic, and the several list variants the
336+recommendation engine uses. They are mechanical against these patterns.
337+
288 ## What this does and does not prove338 ## What this does and does not prove
289 339
290 Proven: the cryptography, the storage layer, and the OAuth flow up to user340 Proven: the cryptography, the storage layer, and the OAuth flow up to user
added spike/nim/stores_probe.nim +224 -0
new file mode 100644
@@ -0,0 +1,224 @@
1+## Exercises the article and feed queries against a real database.
2+##
3+## nim c -r -d:ssl stores_probe.nim
4+##
5+## The checks lean on the places these queries can be quietly wrong: a scope
6+## that clears more than it should, a count that drifts, a listing that
7+## returns other people's feeds, a NULL that sorts to the top.
8+
9+import std/[options, os, strformat, strutils, times]
10+import atproto/[articlestore, feedstore, gleandb, sqlite]
11+
12+var failures = 0
13+
14+proc report(name: string, ok: bool, detail = "") =
15+ if not ok: inc failures
16+ let label = if ok: "PASS" else: "FAIL"
17+ if detail.len > 0: echo &" {label} {name} -- {detail}"
18+ else: echo &" {label} {name}"
19+
20+proc cleanup(base: string) =
21+ for suffix in ["_users", "_articles", "_recs"]:
22+ for ext in ["", "-wal", "-shm"]:
23+ removeFile(base & suffix & ext)
24+
25+const
26+ alice = "did:plc:alice"
27+ bob = "did:plc:bob"
28+ feedA = "https://a.test/feed"
29+ feedB = "https://b.test/feed"
30+
31+proc main() =
32+ echo "Article / feed store probe"
33+ echo ""
34+
35+ let base = getTempDir() / "glean_stores_probe"
36+ cleanup(base)
37+ defer: cleanup(base)
38+
39+ var g = gleandb.open(base)
40+ defer: g.close()
41+ g.migrate()
42+
43+ for did in [alice, bob]:
44+ g.db.run("INSERT INTO users (did) VALUES (?)", p(did))
45+
46+ # --- feeds --------------------------------------------------------------
47+ block:
48+ g.upsertFeed(Feed(feedUrl: feedA, title: "Feed A", feedType: "rss",
49+ siteUrl: "https://a.test"))
50+ g.upsertFeed(Feed(feedUrl: feedB, title: "Feed B", feedType: "atom"))
51+ let f = g.getFeed(feedA)
52+ report("a feed round-trips", f.isSome and f.get.title == "Feed A",
53+ f.get(Feed()).title)
54+
55+ # A metadata refresh must not reset state the fetcher owns.
56+ g.markFeedFetchError(feedA, "boom")
57+ g.upsertFeed(Feed(feedUrl: feedA, title: "Feed A renamed", feedType: "rss"))
58+ let after = g.getFeed(feedA).get
59+ report("upsert refreshes metadata", after.title == "Feed A renamed")
60+ report("upsert preserves error state", after.errorCount == 1,
61+ &"error_count={after.errorCount}")
62+
63+ g.markFeedFetched(feedA)
64+ let healed = g.getFeed(feedA).get
65+ report("a successful fetch clears the error state",
66+ healed.errorCount == 0 and healed.lastError.len == 0)
67+ report("last_fetched_at is recorded", healed.lastFetchedAt.isSome)
68+
69+ # --- subscriptions ------------------------------------------------------
70+ block:
71+ g.createSubscription(alice, feedA, "", "news", "", "")
72+ g.createSubscription(alice, feedB, "", "", "", "")
73+ g.createSubscription(bob, feedA, "", "", "", "")
74+
75+ report("subscriber_count tracks subscribers",
76+ g.getFeed(feedA).get.subscriberCount == 2,
77+ $g.getFeed(feedA).get.subscriberCount)
78+
79+ # Re-subscribing is an update, not a second subscriber.
80+ g.createSubscription(alice, feedA, "My title", "tech", "", "")
81+ report("re-subscribing does not inflate the count",
82+ g.getFeed(feedA).get.subscriberCount == 2,
83+ $g.getFeed(feedA).get.subscriberCount)
84+ report("re-subscribing updates the row",
85+ g.getSubscription(alice, feedA).get.category == "tech")
86+
87+ report("subscription count is per user",
88+ g.getSubscriptionCount(alice) == 2 and g.getSubscriptionCount(bob) == 1)
89+ report("categories exclude empties",
90+ g.getCategories(alice) == @["tech"], g.getCategories(alice).join(","))
91+
92+ let listed = g.listSubscriptions(alice, "", 10, 0)
93+ report("listing returns only this user's subscriptions", listed.len == 2,
94+ $listed.len)
95+ report("listing falls back to the feed title when none is set",
96+ g.listSubscriptions(bob, "", 10, 0)[0].feedTitle == "Feed A renamed")
97+
98+ # --- articles -----------------------------------------------------------
99+ block:
100+ let t0 = now().utc - initDuration(days = 2)
101+ g.batchUpsertArticles(@[
102+ NewArticle(feedUrl: feedA, guid: "a1", title: "Nim systems programming",
103+ url: "https://a.test/1", summary: "about nim",
104+ published: some(t0)),
105+ NewArticle(feedUrl: feedA, guid: "a2", title: "Second post",
106+ url: "https://a.test/2", published: some(t0 + initDuration(hours = 1))),
107+ NewArticle(feedUrl: feedB, guid: "b1", title: "Elsewhere",
108+ url: "https://b.test/1", published: some(t0)),
109+ ])
110+ let all = g.listArticles(alice, limit = 50)
111+ report("articles are ingested and listed", all.len == 3, $all.len)
112+
113+ # Re-ingesting the same guids must not duplicate.
114+ g.batchUpsertArticles(@[
115+ NewArticle(feedUrl: feedA, guid: "a1", title: "Nim systems programming",
116+ published: some(t0))])
117+ report("re-ingesting is idempotent",
118+ g.listArticles(alice, limit = 50).len == 3)
119+
120+ report("bob only sees his own subscriptions",
121+ g.listArticles(bob, limit = 50).len == 2,
122+ $g.listArticles(bob, limit = 50).len)
123+
124+ # Retention: an article older than the window is not worth ingesting.
125+ g.batchUpsertArticles(@[
126+ NewArticle(feedUrl: feedA, guid: "ancient", title: "Old news",
127+ published: some(now().utc - initDuration(days = 90)))],
128+ retentionDays = 30)
129+ report("articles past the retention window are skipped",
130+ g.listArticles(alice, limit = 50).len == 3)
131+
132+ let newest = g.listArticles(alice, feedUrl = feedA, limit = 10)
133+ report("newest first by default", newest[0].guid == "a2", newest[0].guid)
134+ let oldest = g.listArticles(alice, feedUrl = feedA, sortOldest = true, limit = 10)
135+ report("oldest first when asked", oldest[0].guid == "a1", oldest[0].guid)
136+
137+ report("filtering by feed scopes the listing",
138+ g.listArticles(alice, feedUrl = feedB, limit = 10).len == 1)
139+ report("filtering by category scopes the listing",
140+ g.listArticles(alice, category = "tech", limit = 10).len == 2,
141+ $g.listArticles(alice, category = "tech", limit = 10).len)
142+
143+ # --- read state ---------------------------------------------------------
144+ block:
145+ let arts = g.listArticles(alice, feedUrl = feedA, limit = 10)
146+ let first = arts[0]
147+
148+ report("everything starts unread", g.getUnreadCount(alice, "", "") == 3,
149+ $g.getUnreadCount(alice, "", ""))
150+
151+ g.markArticleRead(alice, first.id)
152+ report("marking read is reflected in the count",
153+ g.getUnreadCount(alice, "", "") == 2)
154+ report("read state is per user",
155+ g.getUnreadCount(bob, "", "") == 2, $g.getUnreadCount(bob, "", ""))
156+ report("the article reads back as read",
157+ g.getArticle(alice, first.id).get.isRead)
158+
159+ report("unread filter excludes it",
160+ g.listArticles(alice, filter = afUnread, limit = 50).len == 2)
161+ report("read filter includes only it",
162+ g.listArticles(alice, filter = afRead, limit = 50).len == 1)
163+
164+ g.markArticleUnread(alice, first.id)
165+ report("marking unread reverses it", g.getUnreadCount(alice, "", "") == 3)
166+
167+ # The scope that matters: one feed, not the account.
168+ g.markAllRead(alice, feedA)
169+ report("mark-all-read on a feed clears only that feed",
170+ g.getUnreadCount(alice, feedA, "") == 0 and
171+ g.getUnreadCount(alice, feedB, "") == 1,
172+ &"A={g.getUnreadCount(alice, feedA, \"\")} B={g.getUnreadCount(alice, feedB, \"\")}")
173+ report("and leaves other users alone",
174+ g.getUnreadCount(bob, "", "") == 2)
175+
176+ g.markAllRead(alice, "")
177+ report("unscoped mark-all-read clears everything subscribed",
178+ g.getUnreadCount(alice, "", "") == 0)
179+
180+ # --- search -------------------------------------------------------------
181+ block:
182+ let hits = g.searchArticles(alice, "systems")
183+ report("full-text search finds an article", hits.len == 1,
184+ if hits.len > 0: hits[0].title else: "no hits")
185+ report("search is scoped to subscriptions",
186+ g.searchArticles(bob, "Elsewhere").len == 0,
187+ "bob is not subscribed to feed B")
188+ report("an empty query returns nothing", g.searchArticles(alice, " ").len == 0)
189+
190+ # --- unsubscribe --------------------------------------------------------
191+ block:
192+ g.deleteSubscription(alice, feedA)
193+ report("unsubscribing decrements the count",
194+ g.getFeed(feedA).get.subscriberCount == 1,
195+ $g.getFeed(feedA).get.subscriberCount)
196+ report("unsubscribing hides the feed's articles",
197+ g.listArticles(alice, limit = 50).len == 1)
198+
199+ # The counts are maintained incrementally, so the repair pass must agree.
200+ g.db.run("UPDATE articles.feeds SET subscriber_count = 99")
201+ g.recountSubscriberCounts()
202+ report("the recount repairs drifted counts",
203+ g.getFeed(feedA).get.subscriberCount == 1,
204+ $g.getFeed(feedA).get.subscriberCount)
205+
206+ # --- dead feeds ---------------------------------------------------------
207+ block:
208+ for _ in 0 ..< 3:
209+ g.markFeedFetchError(feedB, "still broken")
210+ let dead = g.listDeadFeeds(alice, threshold = 3)
211+ report("dead feeds are listed for the subscriber",
212+ dead.len == 1 and dead[0].feedUrl == feedB, $dead.len)
213+ report("a healthy feed is not listed as dead",
214+ g.listDeadFeeds(bob, threshold = 3).len == 0)
215+
216+ echo ""
217+ if failures == 0:
218+ echo "RESULT: the article and feed queries behave."
219+ else:
220+ echo &"RESULT: {failures} check(s) failed."
221+ quit 1
222+
223+when isMainModule:
224+ main()
new file mode 100644
@@ -0,0 +1,224 @@
1+## Exercises the article and feed queries against a real database.
2+##
3+## nim c -r -d:ssl stores_probe.nim
4+##
5+## The checks lean on the places these queries can be quietly wrong: a scope
6+## that clears more than it should, a count that drifts, a listing that
7+## returns other people's feeds, a NULL that sorts to the top.
8+
9+import std/[options, os, strformat, strutils, times]
10+import atproto/[articlestore, feedstore, gleandb, sqlite]
11+
12+var failures = 0
13+
14+proc report(name: string, ok: bool, detail = "") =
15+ if not ok: inc failures
16+ let label = if ok: "PASS" else: "FAIL"
17+ if detail.len > 0: echo &" {label} {name} -- {detail}"
18+ else: echo &" {label} {name}"
19+
20+proc cleanup(base: string) =
21+ for suffix in ["_users", "_articles", "_recs"]:
22+ for ext in ["", "-wal", "-shm"]:
23+ removeFile(base & suffix & ext)
24+
25+const
26+ alice = "did:plc:alice"
27+ bob = "did:plc:bob"
28+ feedA = "https://a.test/feed"
29+ feedB = "https://b.test/feed"
30+
31+proc main() =
32+ echo "Article / feed store probe"
33+ echo ""
34+
35+ let base = getTempDir() / "glean_stores_probe"
36+ cleanup(base)
37+ defer: cleanup(base)
38+
39+ var g = gleandb.open(base)
40+ defer: g.close()
41+ g.migrate()
42+
43+ for did in [alice, bob]:
44+ g.db.run("INSERT INTO users (did) VALUES (?)", p(did))
45+
46+ # --- feeds --------------------------------------------------------------
47+ block:
48+ g.upsertFeed(Feed(feedUrl: feedA, title: "Feed A", feedType: "rss",
49+ siteUrl: "https://a.test"))
50+ g.upsertFeed(Feed(feedUrl: feedB, title: "Feed B", feedType: "atom"))
51+ let f = g.getFeed(feedA)
52+ report("a feed round-trips", f.isSome and f.get.title == "Feed A",
53+ f.get(Feed()).title)
54+
55+ # A metadata refresh must not reset state the fetcher owns.
56+ g.markFeedFetchError(feedA, "boom")
57+ g.upsertFeed(Feed(feedUrl: feedA, title: "Feed A renamed", feedType: "rss"))
58+ let after = g.getFeed(feedA).get
59+ report("upsert refreshes metadata", after.title == "Feed A renamed")
60+ report("upsert preserves error state", after.errorCount == 1,
61+ &"error_count={after.errorCount}")
62+
63+ g.markFeedFetched(feedA)
64+ let healed = g.getFeed(feedA).get
65+ report("a successful fetch clears the error state",
66+ healed.errorCount == 0 and healed.lastError.len == 0)
67+ report("last_fetched_at is recorded", healed.lastFetchedAt.isSome)
68+
69+ # --- subscriptions ------------------------------------------------------
70+ block:
71+ g.createSubscription(alice, feedA, "", "news", "", "")
72+ g.createSubscription(alice, feedB, "", "", "", "")
73+ g.createSubscription(bob, feedA, "", "", "", "")
74+
75+ report("subscriber_count tracks subscribers",
76+ g.getFeed(feedA).get.subscriberCount == 2,
77+ $g.getFeed(feedA).get.subscriberCount)
78+
79+ # Re-subscribing is an update, not a second subscriber.
80+ g.createSubscription(alice, feedA, "My title", "tech", "", "")
81+ report("re-subscribing does not inflate the count",
82+ g.getFeed(feedA).get.subscriberCount == 2,
83+ $g.getFeed(feedA).get.subscriberCount)
84+ report("re-subscribing updates the row",
85+ g.getSubscription(alice, feedA).get.category == "tech")
86+
87+ report("subscription count is per user",
88+ g.getSubscriptionCount(alice) == 2 and g.getSubscriptionCount(bob) == 1)
89+ report("categories exclude empties",
90+ g.getCategories(alice) == @["tech"], g.getCategories(alice).join(","))
91+
92+ let listed = g.listSubscriptions(alice, "", 10, 0)
93+ report("listing returns only this user's subscriptions", listed.len == 2,
94+ $listed.len)
95+ report("listing falls back to the feed title when none is set",
96+ g.listSubscriptions(bob, "", 10, 0)[0].feedTitle == "Feed A renamed")
97+
98+ # --- articles -----------------------------------------------------------
99+ block:
100+ let t0 = now().utc - initDuration(days = 2)
101+ g.batchUpsertArticles(@[
102+ NewArticle(feedUrl: feedA, guid: "a1", title: "Nim systems programming",
103+ url: "https://a.test/1", summary: "about nim",
104+ published: some(t0)),
105+ NewArticle(feedUrl: feedA, guid: "a2", title: "Second post",
106+ url: "https://a.test/2", published: some(t0 + initDuration(hours = 1))),
107+ NewArticle(feedUrl: feedB, guid: "b1", title: "Elsewhere",
108+ url: "https://b.test/1", published: some(t0)),
109+ ])
110+ let all = g.listArticles(alice, limit = 50)
111+ report("articles are ingested and listed", all.len == 3, $all.len)
112+
113+ # Re-ingesting the same guids must not duplicate.
114+ g.batchUpsertArticles(@[
115+ NewArticle(feedUrl: feedA, guid: "a1", title: "Nim systems programming",
116+ published: some(t0))])
117+ report("re-ingesting is idempotent",
118+ g.listArticles(alice, limit = 50).len == 3)
119+
120+ report("bob only sees his own subscriptions",
121+ g.listArticles(bob, limit = 50).len == 2,
122+ $g.listArticles(bob, limit = 50).len)
123+
124+ # Retention: an article older than the window is not worth ingesting.
125+ g.batchUpsertArticles(@[
126+ NewArticle(feedUrl: feedA, guid: "ancient", title: "Old news",
127+ published: some(now().utc - initDuration(days = 90)))],
128+ retentionDays = 30)
129+ report("articles past the retention window are skipped",
130+ g.listArticles(alice, limit = 50).len == 3)
131+
132+ let newest = g.listArticles(alice, feedUrl = feedA, limit = 10)
133+ report("newest first by default", newest[0].guid == "a2", newest[0].guid)
134+ let oldest = g.listArticles(alice, feedUrl = feedA, sortOldest = true, limit = 10)
135+ report("oldest first when asked", oldest[0].guid == "a1", oldest[0].guid)
136+
137+ report("filtering by feed scopes the listing",
138+ g.listArticles(alice, feedUrl = feedB, limit = 10).len == 1)
139+ report("filtering by category scopes the listing",
140+ g.listArticles(alice, category = "tech", limit = 10).len == 2,
141+ $g.listArticles(alice, category = "tech", limit = 10).len)
142+
143+ # --- read state ---------------------------------------------------------
144+ block:
145+ let arts = g.listArticles(alice, feedUrl = feedA, limit = 10)
146+ let first = arts[0]
147+
148+ report("everything starts unread", g.getUnreadCount(alice, "", "") == 3,
149+ $g.getUnreadCount(alice, "", ""))
150+
151+ g.markArticleRead(alice, first.id)
152+ report("marking read is reflected in the count",
153+ g.getUnreadCount(alice, "", "") == 2)
154+ report("read state is per user",
155+ g.getUnreadCount(bob, "", "") == 2, $g.getUnreadCount(bob, "", ""))
156+ report("the article reads back as read",
157+ g.getArticle(alice, first.id).get.isRead)
158+
159+ report("unread filter excludes it",
160+ g.listArticles(alice, filter = afUnread, limit = 50).len == 2)
161+ report("read filter includes only it",
162+ g.listArticles(alice, filter = afRead, limit = 50).len == 1)
163+
164+ g.markArticleUnread(alice, first.id)
165+ report("marking unread reverses it", g.getUnreadCount(alice, "", "") == 3)
166+
167+ # The scope that matters: one feed, not the account.
168+ g.markAllRead(alice, feedA)
169+ report("mark-all-read on a feed clears only that feed",
170+ g.getUnreadCount(alice, feedA, "") == 0 and
171+ g.getUnreadCount(alice, feedB, "") == 1,
172+ &"A={g.getUnreadCount(alice, feedA, \"\")} B={g.getUnreadCount(alice, feedB, \"\")}")
173+ report("and leaves other users alone",
174+ g.getUnreadCount(bob, "", "") == 2)
175+
176+ g.markAllRead(alice, "")
177+ report("unscoped mark-all-read clears everything subscribed",
178+ g.getUnreadCount(alice, "", "") == 0)
179+
180+ # --- search -------------------------------------------------------------
181+ block:
182+ let hits = g.searchArticles(alice, "systems")
183+ report("full-text search finds an article", hits.len == 1,
184+ if hits.len > 0: hits[0].title else: "no hits")
185+ report("search is scoped to subscriptions",
186+ g.searchArticles(bob, "Elsewhere").len == 0,
187+ "bob is not subscribed to feed B")
188+ report("an empty query returns nothing", g.searchArticles(alice, " ").len == 0)
189+
190+ # --- unsubscribe --------------------------------------------------------
191+ block:
192+ g.deleteSubscription(alice, feedA)
193+ report("unsubscribing decrements the count",
194+ g.getFeed(feedA).get.subscriberCount == 1,
195+ $g.getFeed(feedA).get.subscriberCount)
196+ report("unsubscribing hides the feed's articles",
197+ g.listArticles(alice, limit = 50).len == 1)
198+
199+ # The counts are maintained incrementally, so the repair pass must agree.
200+ g.db.run("UPDATE articles.feeds SET subscriber_count = 99")
201+ g.recountSubscriberCounts()
202+ report("the recount repairs drifted counts",
203+ g.getFeed(feedA).get.subscriberCount == 1,
204+ $g.getFeed(feedA).get.subscriberCount)
205+
206+ # --- dead feeds ---------------------------------------------------------
207+ block:
208+ for _ in 0 ..< 3:
209+ g.markFeedFetchError(feedB, "still broken")
210+ let dead = g.listDeadFeeds(alice, threshold = 3)
211+ report("dead feeds are listed for the subscriber",
212+ dead.len == 1 and dead[0].feedUrl == feedB, $dead.len)
213+ report("a healthy feed is not listed as dead",
214+ g.listDeadFeeds(bob, threshold = 3).len == 0)
215+
216+ echo ""
217+ if failures == 0:
218+ echo "RESULT: the article and feed queries behave."
219+ else:
220+ echo &"RESULT: {failures} check(s) failed."
221+ quit 1
222+
223+when isMainModule:
224+ main()