Port likes, annotations, follows and trending
Completes the database layer with internal/db/social.go. Likes and annotations are ATProto records first and rows second, keyed by their at:// URI because that is the identity the PDS owns. Each can arrive twice -- once from the firehose, once from a PDS sync -- so every write is idempotent by construction rather than by checking first, and an annotation edit keeps the URI while changing the content. Two details carried over deliberately. The language filter always keeps language = '', which means not-yet-classified rather than no-language; excluding it would make new articles vanish for anyone with a filter set and reappear once classified. And future-dated articles sort last in trending, or a feed publishing with a scheduled timestamp pins itself to the top. The probe weights toward the destructive path: deleteOrphanedLikes and its annotation twin remove rows the PDS no longer has, so the checks assert both that the dropped rows go and that other users' rows stay. The empty-active- set case is documented rather than defended, since "you unliked everything" and "the fetch failed" are indistinguishable from inside the store. Also adds a nimble file, which the growing pile of probes had outgrown: `nimble test` for the offline suite, `testnet` for the ones that talk to real servers, `schema` to regenerate the extracted DDL. The split matters because a network failure can mean bsky.social is having a bad day rather than that the port broke. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
4bdee40 parent: 81e384f modified
spike/nim/.gitignore +2 -0 | @@ -12,3 +12,5 @@ store_probe | ||
| 12 | 12 | *.db-shm |
| 13 | 13 | gleandb_probe |
| 14 | 14 | stores_probe |
| 15 | +social_probe | |
| 16 | +nimcache/ | |
| @@ -12,3 +12,5 @@ store_probe | |||
| 12 | *.db-shm | 12 | *.db-shm |
| 13 | gleandb_probe | 13 | gleandb_probe |
| 14 | stores_probe | 14 | stores_probe |
| 15 | +social_probe | ||
| 16 | +nimcache/ | ||
modified
spike/nim/atproto/articlestore.nim +8 -5 | @@ -45,7 +45,10 @@ type | ||
| 45 | 45 | published*: Option[DateTime] |
| 46 | 46 | updated*: Option[DateTime] |
| 47 | 47 | |
| 48 | -proc readArticle(r: Row): Article = | |
| 48 | +proc readArticleRow*(r: Row): Article = | |
| 49 | + ## Exported so other stores returning the same column list can share it -- | |
| 50 | + ## socialstore's liked-articles listing, in particular. The positional | |
| 51 | + ## reader is only safe while every caller selects `articleColumns`. | |
| 49 | 52 | Article( |
| 50 | 53 | id: r.i64(0), |
| 51 | 54 | feedUrl: r.str(1), |
| @@ -84,11 +87,11 @@ const articleFrom = """ | ||
| 84 | 87 | |
| 85 | 88 | proc getArticle*(g: GleanDb, userDid: string, id: int64): Option[Article] = |
| 86 | 89 | g.db.queryFirst("SELECT " & articleColumns & articleFrom & " WHERE a.id = ?", |
| 87 | - [p(userDid), p(userDid), p(id)], readArticle) | |
| 90 | + [p(userDid), p(userDid), p(id)], readArticleRow) | |
| 88 | 91 | |
| 89 | 92 | proc getArticleByUrl*(g: GleanDb, userDid, url: string): Option[Article] = |
| 90 | 93 | g.db.queryFirst("SELECT " & articleColumns & articleFrom & " WHERE a.url = ?", |
| 91 | - [p(userDid), p(userDid), p(url)], readArticle) | |
| 94 | + [p(userDid), p(userDid), p(url)], readArticleRow) | |
| 92 | 95 | |
| 93 | 96 | proc listArticles*(g: GleanDb, userDid: string, filter = afAll, |
| 94 | 97 | feedUrl = "", category = "", sortOldest = false, |
| @@ -122,7 +125,7 @@ proc listArticles*(g: GleanDb, userDid: string, filter = afAll, | ||
| 122 | 125 | params.add p(limit.int64) |
| 123 | 126 | params.add p(offset.int64) |
| 124 | 127 | |
| 125 | - g.db.queryAll(sql, params, readArticle) | |
| 128 | + g.db.queryAll(sql, params, readArticleRow) | |
| 126 | 129 | |
| 127 | 130 | proc searchArticles*(g: GleanDb, userDid, query: string, |
| 128 | 131 | limit = 25, offset = 0): seq[Article] = |
| @@ -145,7 +148,7 @@ proc searchArticles*(g: GleanDb, userDid, query: string, | ||
| 145 | 148 | ORDER BY m.rank |
| 146 | 149 | LIMIT ? OFFSET ?""", |
| 147 | 150 | [p(userDid), p(userDid), p(userDid), p(query), |
| 148 | - p(limit.int64), p(offset.int64)], readArticle) | |
| 151 | + p(limit.int64), p(offset.int64)], readArticleRow) | |
| 149 | 152 | |
| 150 | 153 | proc batchUpsertArticles*(g: GleanDb, articles: seq[NewArticle], |
| 151 | 154 | retentionDays = 0) = |
| @@ -45,7 +45,10 @@ type | |||
| 45 | published*: Option[DateTime] | 45 | published*: Option[DateTime] |
| 46 | updated*: Option[DateTime] | 46 | updated*: Option[DateTime] |
| 47 | 47 | ||
| 48 | -proc readArticle(r: Row): Article = | 48 | +proc readArticleRow*(r: Row): Article = |
| 49 | + ## Exported so other stores returning the same column list can share it -- | ||
| 50 | + ## socialstore's liked-articles listing, in particular. The positional | ||
| 51 | + ## reader is only safe while every caller selects `articleColumns`. | ||
| 49 | Article( | 52 | Article( |
| 50 | id: r.i64(0), | 53 | id: r.i64(0), |
| 51 | feedUrl: r.str(1), | 54 | feedUrl: r.str(1), |
| @@ -84,11 +87,11 @@ const articleFrom = """ | |||
| 84 | 87 | ||
| 85 | proc getArticle*(g: GleanDb, userDid: string, id: int64): Option[Article] = | 88 | proc getArticle*(g: GleanDb, userDid: string, id: int64): Option[Article] = |
| 86 | g.db.queryFirst("SELECT " & articleColumns & articleFrom & " WHERE a.id = ?", | 89 | g.db.queryFirst("SELECT " & articleColumns & articleFrom & " WHERE a.id = ?", |
| 87 | - [p(userDid), p(userDid), p(id)], readArticle) | 90 | + [p(userDid), p(userDid), p(id)], readArticleRow) |
| 88 | 91 | ||
| 89 | proc getArticleByUrl*(g: GleanDb, userDid, url: string): Option[Article] = | 92 | proc getArticleByUrl*(g: GleanDb, userDid, url: string): Option[Article] = |
| 90 | g.db.queryFirst("SELECT " & articleColumns & articleFrom & " WHERE a.url = ?", | 93 | g.db.queryFirst("SELECT " & articleColumns & articleFrom & " WHERE a.url = ?", |
| 91 | - [p(userDid), p(userDid), p(url)], readArticle) | 94 | + [p(userDid), p(userDid), p(url)], readArticleRow) |
| 92 | 95 | ||
| 93 | proc listArticles*(g: GleanDb, userDid: string, filter = afAll, | 96 | proc listArticles*(g: GleanDb, userDid: string, filter = afAll, |
| 94 | feedUrl = "", category = "", sortOldest = false, | 97 | feedUrl = "", category = "", sortOldest = false, |
| @@ -122,7 +125,7 @@ proc listArticles*(g: GleanDb, userDid: string, filter = afAll, | |||
| 122 | params.add p(limit.int64) | 125 | params.add p(limit.int64) |
| 123 | params.add p(offset.int64) | 126 | params.add p(offset.int64) |
| 124 | 127 | ||
| 125 | - g.db.queryAll(sql, params, readArticle) | 128 | + g.db.queryAll(sql, params, readArticleRow) |
| 126 | 129 | ||
| 127 | proc searchArticles*(g: GleanDb, userDid, query: string, | 130 | proc searchArticles*(g: GleanDb, userDid, query: string, |
| 128 | limit = 25, offset = 0): seq[Article] = | 131 | limit = 25, offset = 0): seq[Article] = |
| @@ -145,7 +148,7 @@ proc searchArticles*(g: GleanDb, userDid, query: string, | |||
| 145 | ORDER BY m.rank | 148 | ORDER BY m.rank |
| 146 | LIMIT ? OFFSET ?""", | 149 | LIMIT ? OFFSET ?""", |
| 147 | [p(userDid), p(userDid), p(userDid), p(query), | 150 | [p(userDid), p(userDid), p(userDid), p(query), |
| 148 | - p(limit.int64), p(offset.int64)], readArticle) | 151 | + p(limit.int64), p(offset.int64)], readArticleRow) |
| 149 | 152 | ||
| 150 | proc batchUpsertArticles*(g: GleanDb, articles: seq[NewArticle], | 153 | proc batchUpsertArticles*(g: GleanDb, articles: seq[NewArticle], |
| 151 | retentionDays = 0) = | 154 | retentionDays = 0) = |
added
spike/nim/atproto/socialstore.nim +388 -0 | new file mode 100644 | ||
| @@ -0,0 +1,388 @@ | ||
| 1 | +## Likes, annotations and the social reads built on them, ported from | |
| 2 | +## internal/db/social.go. | |
| 3 | +## | |
| 4 | +## Likes and annotations are ATProto records first and rows second. They are | |
| 5 | +## keyed by `uri` (the at:// record URI) because that is the identity the PDS | |
| 6 | +## owns, and every one of them can arrive twice -- once from the Jetstream | |
| 7 | +## firehose and once from a PDS sync -- so every write here is idempotent by | |
| 8 | +## construction rather than by checking first. | |
| 9 | + | |
| 10 | +import std/[options, strutils, times] | |
| 11 | +import ./articlestore | |
| 12 | +import ./feedstore | |
| 13 | +import ./gleandb | |
| 14 | +import ./sqlite | |
| 15 | + | |
| 16 | +type | |
| 17 | + Annotation* = object | |
| 18 | + id*: int64 | |
| 19 | + uri*: string | |
| 20 | + authorDid*: string | |
| 21 | + feedUrl*: string | |
| 22 | + articleUrl*: string | |
| 23 | + quote*: string | |
| 24 | + note*: string | |
| 25 | + tags*: seq[string] | |
| 26 | + rating*: Option[int] | |
| 27 | + createdAt*: Option[DateTime] | |
| 28 | + cid*: string | |
| 29 | + | |
| 30 | + Like* = object | |
| 31 | + id*: int64 | |
| 32 | + uri*: string | |
| 33 | + authorDid*: string | |
| 34 | + feedUrl*: string | |
| 35 | + articleUrl*: string | |
| 36 | + createdAt*: Option[DateTime] | |
| 37 | + cid*: string | |
| 38 | + | |
| 39 | + TrendingItem* = object | |
| 40 | + articleId*: int64 | |
| 41 | + title*: string | |
| 42 | + url*: string | |
| 43 | + author*: string | |
| 44 | + summary*: string | |
| 45 | + feedUrl*: string | |
| 46 | + feedTitle*: string | |
| 47 | + faviconUrl*: string | |
| 48 | + likeCount*: int | |
| 49 | + annotationCount*: int | |
| 50 | + hasLiked*: bool | |
| 51 | + | |
| 52 | +# --- helpers --------------------------------------------------------------- | |
| 53 | + | |
| 54 | +proc splitTags(s: string): seq[string] = | |
| 55 | + for t in s.split(','): | |
| 56 | + let v = t.strip() | |
| 57 | + if v.len > 0: result.add v | |
| 58 | + | |
| 59 | +proc joinTags(tags: seq[string]): string = tags.join(",") | |
| 60 | + | |
| 61 | +proc langFilter*(languages: seq[string], prefix: string): (string, seq[Param]) = | |
| 62 | + ## Restrict to the reader's languages, always keeping unknown ones. | |
| 63 | + ## | |
| 64 | + ## `language = ''` means detection has not run yet, not "no language". It is | |
| 65 | + ## included deliberately: excluding it would make new articles vanish for | |
| 66 | + ## anyone with a filter set, and reappear minutes later once classified. | |
| 67 | + if languages.len == 0: | |
| 68 | + return ("", @[]) | |
| 69 | + var holes: seq[string] | |
| 70 | + var params: seq[Param] | |
| 71 | + for l in languages: | |
| 72 | + holes.add "?" | |
| 73 | + params.add p(l) | |
| 74 | + (" AND (" & prefix & "language IN (" & holes.join(",") & ") OR " & | |
| 75 | + prefix & "language = '')", params) | |
| 76 | + | |
| 77 | +# --- annotations ----------------------------------------------------------- | |
| 78 | + | |
| 79 | +proc readAnnotation(r: Row): Annotation = | |
| 80 | + Annotation( | |
| 81 | + id: r.i64(0), | |
| 82 | + uri: r.str(1), | |
| 83 | + authorDid: r.str(2), | |
| 84 | + feedUrl: r.str(3), | |
| 85 | + articleUrl: r.str(4), | |
| 86 | + quote: r.str(5), | |
| 87 | + note: r.str(6), | |
| 88 | + tags: splitTags(r.str(7)), | |
| 89 | + rating: (if r.isNull(8): none(int) else: some(r.i(8))), | |
| 90 | + createdAt: parseSqliteTime(r.str(9)), | |
| 91 | + cid: r.str(10), | |
| 92 | + ) | |
| 93 | + | |
| 94 | +const annotationColumns = """an.id, an.uri, an.author_did, an.feed_url, | |
| 95 | + an.article_url, an.quote, an.note, an.tags, an.rating, an.created_at, an.cid""" | |
| 96 | + | |
| 97 | +proc createAnnotation*(g: GleanDb, a: Annotation) = | |
| 98 | + ## Upsert on `uri`: the same record can arrive from Jetstream and from a | |
| 99 | + ## sync, and an edit keeps the URI while changing the content. | |
| 100 | + g.db.run("""INSERT INTO articles.annotations | |
| 101 | + (uri, author_did, feed_url, article_url, quote, note, tags, | |
| 102 | + rating, created_at, cid) | |
| 103 | + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) | |
| 104 | + ON CONFLICT(uri) DO UPDATE SET | |
| 105 | + quote = excluded.quote, | |
| 106 | + note = excluded.note, | |
| 107 | + tags = excluded.tags, | |
| 108 | + rating = excluded.rating, | |
| 109 | + cid = excluded.cid""", | |
| 110 | + p(a.uri), p(a.authorDid), p(a.feedUrl), p(a.articleUrl), | |
| 111 | + pOrNull(a.quote), pOrNull(a.note), pOrNull(joinTags(a.tags)), | |
| 112 | + (if a.rating.isSome: p(a.rating.get.int64) else: nullParam()), | |
| 113 | + p(formatSqliteTime(a.createdAt.get(now().utc))), | |
| 114 | + pOrNull(a.cid)) | |
| 115 | + | |
| 116 | +proc getAnnotation*(g: GleanDb, id: int64): Option[Annotation] = | |
| 117 | + g.db.queryFirst("SELECT " & annotationColumns & | |
| 118 | + " FROM articles.annotations an WHERE an.id = ?", | |
| 119 | + [p(id)], readAnnotation) | |
| 120 | + | |
| 121 | +proc annotationExists*(g: GleanDb, uri: string): bool = | |
| 122 | + g.db.queryInt("SELECT 1 FROM articles.annotations WHERE uri = ?", | |
| 123 | + p(uri)).isSome | |
| 124 | + | |
| 125 | +proc deleteAnnotation*(g: GleanDb, uri: string) = | |
| 126 | + g.db.run("DELETE FROM articles.annotations WHERE uri = ?", p(uri)) | |
| 127 | + | |
| 128 | +proc listAnnotations*(g: GleanDb, feedUrl = "", articleUrl = "", | |
| 129 | + authorDid = "", limit = 50, offset = 0): seq[Annotation] = | |
| 130 | + var sql = "SELECT " & annotationColumns & | |
| 131 | + " FROM articles.annotations an WHERE 1=1" | |
| 132 | + var params: seq[Param] | |
| 133 | + if feedUrl.len > 0: | |
| 134 | + sql &= " AND an.feed_url = ?" | |
| 135 | + params.add p(feedUrl) | |
| 136 | + if articleUrl.len > 0: | |
| 137 | + sql &= " AND an.article_url = ?" | |
| 138 | + params.add p(articleUrl) | |
| 139 | + if authorDid.len > 0: | |
| 140 | + sql &= " AND an.author_did = ?" | |
| 141 | + params.add p(authorDid) | |
| 142 | + sql &= " ORDER BY an.created_at DESC LIMIT ? OFFSET ?" | |
| 143 | + params.add p(limit.int64) | |
| 144 | + params.add p(offset.int64) | |
| 145 | + g.db.queryAll(sql, params, readAnnotation) | |
| 146 | + | |
| 147 | +proc batchCreateAnnotations*(g: GleanDb, annotations: seq[Annotation]) = | |
| 148 | + if annotations.len == 0: return | |
| 149 | + g.db.transaction: | |
| 150 | + var ps = g.db.prepared(""" | |
| 151 | + INSERT INTO articles.annotations | |
| 152 | + (uri, author_did, feed_url, article_url, quote, note, tags, rating, | |
| 153 | + created_at, cid) | |
| 154 | + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) | |
| 155 | + ON CONFLICT(uri) DO NOTHING""") | |
| 156 | + defer: ps.finalize() | |
| 157 | + for a in annotations: | |
| 158 | + ps.exec(p(a.uri), p(a.authorDid), p(a.feedUrl), p(a.articleUrl), | |
| 159 | + pOrNull(a.quote), pOrNull(a.note), pOrNull(joinTags(a.tags)), | |
| 160 | + (if a.rating.isSome: p(a.rating.get.int64) else: nullParam()), | |
| 161 | + p(formatSqliteTime(a.createdAt.get(now().utc))), pOrNull(a.cid)) | |
| 162 | + | |
| 163 | +# --- likes ----------------------------------------------------------------- | |
| 164 | + | |
| 165 | +proc readLike(r: Row): Like = | |
| 166 | + Like( | |
| 167 | + id: r.i64(0), | |
| 168 | + uri: r.str(1), | |
| 169 | + authorDid: r.str(2), | |
| 170 | + feedUrl: r.str(3), | |
| 171 | + articleUrl: r.str(4), | |
| 172 | + createdAt: parseSqliteTime(r.str(5)), | |
| 173 | + cid: r.str(6), | |
| 174 | + ) | |
| 175 | + | |
| 176 | +const likeColumns = """l.id, l.uri, l.author_did, l.feed_url, l.article_url, | |
| 177 | + l.created_at, l.cid""" | |
| 178 | + | |
| 179 | +proc createLike*(g: GleanDb, l: Like) = | |
| 180 | + g.db.run("""INSERT INTO articles.likes | |
| 181 | + (uri, author_did, feed_url, article_url, created_at, cid) | |
| 182 | + VALUES (?, ?, ?, ?, ?, ?) | |
| 183 | + ON CONFLICT(uri) DO NOTHING""", | |
| 184 | + p(l.uri), p(l.authorDid), p(l.feedUrl), p(l.articleUrl), | |
| 185 | + p(formatSqliteTime(l.createdAt.get(now().utc))), pOrNull(l.cid)) | |
| 186 | + | |
| 187 | +proc batchCreateLikes*(g: GleanDb, likes: seq[Like]) = | |
| 188 | + if likes.len == 0: return | |
| 189 | + g.db.transaction: | |
| 190 | + var ps = g.db.prepared(""" | |
| 191 | + INSERT INTO articles.likes | |
| 192 | + (uri, author_did, feed_url, article_url, created_at, cid) | |
| 193 | + VALUES (?, ?, ?, ?, ?, ?) | |
| 194 | + ON CONFLICT(uri) DO NOTHING""") | |
| 195 | + defer: ps.finalize() | |
| 196 | + for l in likes: | |
| 197 | + ps.exec(p(l.uri), p(l.authorDid), p(l.feedUrl), p(l.articleUrl), | |
| 198 | + p(formatSqliteTime(l.createdAt.get(now().utc))), pOrNull(l.cid)) | |
| 199 | + | |
| 200 | +proc deleteLike*(g: GleanDb, uri: string) = | |
| 201 | + g.db.run("DELETE FROM articles.likes WHERE uri = ?", p(uri)) | |
| 202 | + | |
| 203 | +proc deleteLikeByUserArticle*(g: GleanDb, authorDid, feedUrl, articleUrl: string) = | |
| 204 | + g.db.run("""DELETE FROM articles.likes | |
| 205 | + WHERE author_did = ? AND feed_url = ? AND article_url = ?""", | |
| 206 | + p(authorDid), p(feedUrl), p(articleUrl)) | |
| 207 | + | |
| 208 | +proc getLike*(g: GleanDb, authorDid, feedUrl, articleUrl: string): Option[Like] = | |
| 209 | + g.db.queryFirst("SELECT " & likeColumns & """ | |
| 210 | + FROM articles.likes l | |
| 211 | + WHERE l.author_did = ? AND l.feed_url = ? AND l.article_url = ?""", | |
| 212 | + [p(authorDid), p(feedUrl), p(articleUrl)], readLike) | |
| 213 | + | |
| 214 | +proc hasLiked*(g: GleanDb, authorDid, feedUrl, articleUrl: string): bool = | |
| 215 | + getLike(g, authorDid, feedUrl, articleUrl).isSome | |
| 216 | + | |
| 217 | +proc getLikeCount*(g: GleanDb, feedUrl, articleUrl: string): int = | |
| 218 | + g.db.queryInt("""SELECT count(*) FROM articles.likes | |
| 219 | + WHERE feed_url = ? AND article_url = ?""", | |
| 220 | + p(feedUrl), p(articleUrl)).get(0).int | |
| 221 | + | |
| 222 | +proc listLikes*(g: GleanDb, authorDid: string; feedUrl = "", | |
| 223 | + limit = 50, offset = 0): seq[Like] = | |
| 224 | + var sql = "SELECT " & likeColumns & | |
| 225 | + " FROM articles.likes l WHERE l.author_did = ?" | |
| 226 | + var params = @[p(authorDid)] | |
| 227 | + if feedUrl.len > 0: | |
| 228 | + sql &= " AND l.feed_url = ?" | |
| 229 | + params.add p(feedUrl) | |
| 230 | + sql &= " ORDER BY l.created_at DESC LIMIT ? OFFSET ?" | |
| 231 | + params.add p(limit.int64) | |
| 232 | + params.add p(offset.int64) | |
| 233 | + g.db.queryAll(sql, params, readLike) | |
| 234 | + | |
| 235 | +# --- reconciliation -------------------------------------------------------- | |
| 236 | + | |
| 237 | +proc deleteOrphanedLikes*(g: GleanDb, userDid: string, activeUris: seq[string]) = | |
| 238 | + ## Remove local likes the user's PDS no longer has. | |
| 239 | + ## | |
| 240 | + ## This is the destructive half of a sync, so an empty active set is treated | |
| 241 | + ## as "delete everything this user liked" only because that is what an empty | |
| 242 | + ## PDS collection means. A caller that failed to *fetch* the collection must | |
| 243 | + ## not reach here -- the difference between "you unliked everything" and | |
| 244 | + ## "the request failed" is not visible from this side. | |
| 245 | + if activeUris.len == 0: | |
| 246 | + g.db.run("DELETE FROM articles.likes WHERE author_did = ?", p(userDid)) | |
| 247 | + return | |
| 248 | + g.db.transaction: | |
| 249 | + g.db.run("CREATE TEMP TABLE IF NOT EXISTS active_uris (uri TEXT PRIMARY KEY)") | |
| 250 | + g.db.run("DELETE FROM active_uris") | |
| 251 | + var ps = g.db.prepared( | |
| 252 | + "INSERT OR IGNORE INTO active_uris (uri) VALUES (?)") | |
| 253 | + defer: ps.finalize() | |
| 254 | + for uri in activeUris: | |
| 255 | + ps.exec(p(uri)) | |
| 256 | + g.db.run("""DELETE FROM articles.likes | |
| 257 | + WHERE author_did = ? | |
| 258 | + AND uri NOT IN (SELECT uri FROM active_uris)""", p(userDid)) | |
| 259 | + | |
| 260 | +proc deleteOrphanedAnnotations*(g: GleanDb, userDid: string, | |
| 261 | + activeUris: seq[string]) = | |
| 262 | + if activeUris.len == 0: | |
| 263 | + g.db.run("DELETE FROM articles.annotations WHERE author_did = ?", p(userDid)) | |
| 264 | + return | |
| 265 | + g.db.transaction: | |
| 266 | + g.db.run("CREATE TEMP TABLE IF NOT EXISTS active_uris (uri TEXT PRIMARY KEY)") | |
| 267 | + g.db.run("DELETE FROM active_uris") | |
| 268 | + var ps = g.db.prepared( | |
| 269 | + "INSERT OR IGNORE INTO active_uris (uri) VALUES (?)") | |
| 270 | + defer: ps.finalize() | |
| 271 | + for uri in activeUris: | |
| 272 | + ps.exec(p(uri)) | |
| 273 | + g.db.run("""DELETE FROM articles.annotations | |
| 274 | + WHERE author_did = ? | |
| 275 | + AND uri NOT IN (SELECT uri FROM active_uris)""", p(userDid)) | |
| 276 | + | |
| 277 | +# --- follows --------------------------------------------------------------- | |
| 278 | + | |
| 279 | +proc upsertFollow*(g: GleanDb, userDid, targetDid, uri, cid: string) = | |
| 280 | + g.db.run("""INSERT INTO follows (user_did, target_did, uri, cid, followed_at) | |
| 281 | + VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) | |
| 282 | + ON CONFLICT(user_did, target_did) DO UPDATE SET | |
| 283 | + uri = COALESCE(excluded.uri, follows.uri), | |
| 284 | + cid = COALESCE(excluded.cid, follows.cid)""", | |
| 285 | + p(userDid), p(targetDid), pOrNull(uri), pOrNull(cid)) | |
| 286 | + | |
| 287 | +proc deleteFollow*(g: GleanDb, userDid, targetDid: string) = | |
| 288 | + g.db.run("DELETE FROM follows WHERE user_did = ? AND target_did = ?", | |
| 289 | + p(userDid), p(targetDid)) | |
| 290 | + | |
| 291 | +proc listFollows*(g: GleanDb, userDid: string): seq[string] = | |
| 292 | + for r in g.db.query( | |
| 293 | + "SELECT target_did FROM follows WHERE user_did = ? ORDER BY target_did", | |
| 294 | + p(userDid)): | |
| 295 | + result.add r.str(0) | |
| 296 | + | |
| 297 | +# --- trending -------------------------------------------------------------- | |
| 298 | + | |
| 299 | +proc readTrending(r: Row): TrendingItem = | |
| 300 | + TrendingItem( | |
| 301 | + articleId: r.i64(0), | |
| 302 | + title: r.str(1), | |
| 303 | + url: r.str(2), | |
| 304 | + author: r.str(3), | |
| 305 | + summary: r.str(4), | |
| 306 | + feedUrl: r.str(5), | |
| 307 | + feedTitle: r.str(6), | |
| 308 | + faviconUrl: r.str(7), | |
| 309 | + likeCount: r.i(8), | |
| 310 | + annotationCount: r.i(9), | |
| 311 | + hasLiked: r.b(10), | |
| 312 | + ) | |
| 313 | + | |
| 314 | +const trendingSelect = """ | |
| 315 | + SELECT ar.id, ar.title, COALESCE(ar.url, ''), COALESCE(ar.author, ''), | |
| 316 | + COALESCE(ar.summary, ''), l.feed_url, COALESCE(f.title, ''), | |
| 317 | + COALESCE(f.favicon_url, ''), | |
| 318 | + COUNT(DISTINCT l.id) AS like_count, | |
| 319 | + COUNT(DISTINCT an.id) AS annotation_count, | |
| 320 | + COALESCE(MAX(CASE WHEN ul.id IS NOT NULL THEN 1 ELSE 0 END), 0) | |
| 321 | + FROM articles.likes l | |
| 322 | + JOIN articles.articles ar | |
| 323 | + ON ar.url = l.article_url AND ar.feed_url = l.feed_url | |
| 324 | + LEFT JOIN articles.feeds f ON f.feed_url = l.feed_url | |
| 325 | + LEFT JOIN articles.annotations an | |
| 326 | + ON an.feed_url = l.feed_url AND an.article_url = l.article_url | |
| 327 | + AND an.created_at >= ? | |
| 328 | + LEFT JOIN articles.likes ul | |
| 329 | + ON ul.feed_url = l.feed_url AND ul.article_url = l.article_url | |
| 330 | + AND ul.author_did = ? | |
| 331 | + WHERE l.created_at >= ?""" | |
| 332 | + | |
| 333 | +## Future-dated articles sort last: a feed that publishes with a scheduled | |
| 334 | +## timestamp would otherwise pin itself to the top of trending indefinitely. | |
| 335 | +const trendingOrder = """ | |
| 336 | + GROUP BY ar.id | |
| 337 | + ORDER BY like_count DESC, annotation_count DESC, | |
| 338 | + (CASE WHEN ar.published > datetime('now') THEN 1 ELSE 0 END), | |
| 339 | + ar.published DESC | |
| 340 | + LIMIT ? OFFSET ?""" | |
| 341 | + | |
| 342 | +proc listTrendingArticles*(g: GleanDb, userDid, since: string, | |
| 343 | + limit = 25, offset = 0): seq[TrendingItem] = | |
| 344 | + ## Global trending: everyone's likes. | |
| 345 | + g.db.queryAll(trendingSelect & trendingOrder, | |
| 346 | + [p(since), p(userDid), p(since), p(limit.int64), p(offset.int64)], | |
| 347 | + readTrending) | |
| 348 | + | |
| 349 | +proc listTrendingArticlesForUser*(g: GleanDb, userDid, since: string, | |
| 350 | + languages: seq[string] = @[], | |
| 351 | + limit = 25, offset = 0): seq[TrendingItem] = | |
| 352 | + ## Trending within the reader's own graph: people they follow, people the | |
| 353 | + ## similarity table pairs them with, and themselves. | |
| 354 | + let (filter, langParams) = langFilter(languages, "ar.") | |
| 355 | + var params = @[p(since), p(userDid), p(since), | |
| 356 | + p(userDid), p(userDid), p(userDid), p(userDid), p(userDid)] | |
| 357 | + params.add langParams | |
| 358 | + params.add p(limit.int64) | |
| 359 | + params.add p(offset.int64) | |
| 360 | + g.db.queryAll(trendingSelect & """ | |
| 361 | + AND l.author_did IN ( | |
| 362 | + SELECT CASE WHEN us.user_a = ? THEN us.user_b ELSE us.user_a END | |
| 363 | + FROM recs.user_similarity us | |
| 364 | + WHERE us.user_a = ? OR us.user_b = ? | |
| 365 | + UNION SELECT ? | |
| 366 | + UNION SELECT fo.target_did FROM follows fo WHERE fo.user_did = ? | |
| 367 | + )""" & filter & trendingOrder, params, readTrending) | |
| 368 | + | |
| 369 | +proc listLikedArticles*(g: GleanDb, userDid: string, | |
| 370 | + limit = 25, offset = 0): seq[Article] = | |
| 371 | + ## The reader's own likes, as articles. | |
| 372 | + g.db.queryAll(""" | |
| 373 | + SELECT a.id, a.feed_url, COALESCE(f.title, a.feed_url), f.favicon_url, | |
| 374 | + a.guid, a.title, a.url, a.author, a.summary, a.content, | |
| 375 | + a.full_content, a.published, a.updated, | |
| 376 | + COALESCE(r.is_read, 0), | |
| 377 | + (SELECT count(*) FROM articles.likes l2 WHERE l2.article_url = a.url), | |
| 378 | + 1 | |
| 379 | + FROM articles.likes l | |
| 380 | + JOIN articles.articles a | |
| 381 | + ON a.url = l.article_url AND a.feed_url = l.feed_url | |
| 382 | + LEFT JOIN articles.feeds f ON f.feed_url = a.feed_url | |
| 383 | + LEFT JOIN articles.read_state r ON r.article_id = a.id AND r.user_did = ? | |
| 384 | + WHERE l.author_did = ? | |
| 385 | + ORDER BY l.created_at DESC | |
| 386 | + LIMIT ? OFFSET ?""", | |
| 387 | + [p(userDid), p(userDid), p(limit.int64), p(offset.int64)], | |
| 388 | + proc(r: Row): Article = readArticleRow(r)) | |
| new file mode 100644 | |||
| @@ -0,0 +1,388 @@ | |||
| 1 | +## Likes, annotations and the social reads built on them, ported from | ||
| 2 | +## internal/db/social.go. | ||
| 3 | +## | ||
| 4 | +## Likes and annotations are ATProto records first and rows second. They are | ||
| 5 | +## keyed by `uri` (the at:// record URI) because that is the identity the PDS | ||
| 6 | +## owns, and every one of them can arrive twice -- once from the Jetstream | ||
| 7 | +## firehose and once from a PDS sync -- so every write here is idempotent by | ||
| 8 | +## construction rather than by checking first. | ||
| 9 | + | ||
| 10 | +import std/[options, strutils, times] | ||
| 11 | +import ./articlestore | ||
| 12 | +import ./feedstore | ||
| 13 | +import ./gleandb | ||
| 14 | +import ./sqlite | ||
| 15 | + | ||
| 16 | +type | ||
| 17 | + Annotation* = object | ||
| 18 | + id*: int64 | ||
| 19 | + uri*: string | ||
| 20 | + authorDid*: string | ||
| 21 | + feedUrl*: string | ||
| 22 | + articleUrl*: string | ||
| 23 | + quote*: string | ||
| 24 | + note*: string | ||
| 25 | + tags*: seq[string] | ||
| 26 | + rating*: Option[int] | ||
| 27 | + createdAt*: Option[DateTime] | ||
| 28 | + cid*: string | ||
| 29 | + | ||
| 30 | + Like* = object | ||
| 31 | + id*: int64 | ||
| 32 | + uri*: string | ||
| 33 | + authorDid*: string | ||
| 34 | + feedUrl*: string | ||
| 35 | + articleUrl*: string | ||
| 36 | + createdAt*: Option[DateTime] | ||
| 37 | + cid*: string | ||
| 38 | + | ||
| 39 | + TrendingItem* = object | ||
| 40 | + articleId*: int64 | ||
| 41 | + title*: string | ||
| 42 | + url*: string | ||
| 43 | + author*: string | ||
| 44 | + summary*: string | ||
| 45 | + feedUrl*: string | ||
| 46 | + feedTitle*: string | ||
| 47 | + faviconUrl*: string | ||
| 48 | + likeCount*: int | ||
| 49 | + annotationCount*: int | ||
| 50 | + hasLiked*: bool | ||
| 51 | + | ||
| 52 | +# --- helpers --------------------------------------------------------------- | ||
| 53 | + | ||
| 54 | +proc splitTags(s: string): seq[string] = | ||
| 55 | + for t in s.split(','): | ||
| 56 | + let v = t.strip() | ||
| 57 | + if v.len > 0: result.add v | ||
| 58 | + | ||
| 59 | +proc joinTags(tags: seq[string]): string = tags.join(",") | ||
| 60 | + | ||
| 61 | +proc langFilter*(languages: seq[string], prefix: string): (string, seq[Param]) = | ||
| 62 | + ## Restrict to the reader's languages, always keeping unknown ones. | ||
| 63 | + ## | ||
| 64 | + ## `language = ''` means detection has not run yet, not "no language". It is | ||
| 65 | + ## included deliberately: excluding it would make new articles vanish for | ||
| 66 | + ## anyone with a filter set, and reappear minutes later once classified. | ||
| 67 | + if languages.len == 0: | ||
| 68 | + return ("", @[]) | ||
| 69 | + var holes: seq[string] | ||
| 70 | + var params: seq[Param] | ||
| 71 | + for l in languages: | ||
| 72 | + holes.add "?" | ||
| 73 | + params.add p(l) | ||
| 74 | + (" AND (" & prefix & "language IN (" & holes.join(",") & ") OR " & | ||
| 75 | + prefix & "language = '')", params) | ||
| 76 | + | ||
| 77 | +# --- annotations ----------------------------------------------------------- | ||
| 78 | + | ||
| 79 | +proc readAnnotation(r: Row): Annotation = | ||
| 80 | + Annotation( | ||
| 81 | + id: r.i64(0), | ||
| 82 | + uri: r.str(1), | ||
| 83 | + authorDid: r.str(2), | ||
| 84 | + feedUrl: r.str(3), | ||
| 85 | + articleUrl: r.str(4), | ||
| 86 | + quote: r.str(5), | ||
| 87 | + note: r.str(6), | ||
| 88 | + tags: splitTags(r.str(7)), | ||
| 89 | + rating: (if r.isNull(8): none(int) else: some(r.i(8))), | ||
| 90 | + createdAt: parseSqliteTime(r.str(9)), | ||
| 91 | + cid: r.str(10), | ||
| 92 | + ) | ||
| 93 | + | ||
| 94 | +const annotationColumns = """an.id, an.uri, an.author_did, an.feed_url, | ||
| 95 | + an.article_url, an.quote, an.note, an.tags, an.rating, an.created_at, an.cid""" | ||
| 96 | + | ||
| 97 | +proc createAnnotation*(g: GleanDb, a: Annotation) = | ||
| 98 | + ## Upsert on `uri`: the same record can arrive from Jetstream and from a | ||
| 99 | + ## sync, and an edit keeps the URI while changing the content. | ||
| 100 | + g.db.run("""INSERT INTO articles.annotations | ||
| 101 | + (uri, author_did, feed_url, article_url, quote, note, tags, | ||
| 102 | + rating, created_at, cid) | ||
| 103 | + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) | ||
| 104 | + ON CONFLICT(uri) DO UPDATE SET | ||
| 105 | + quote = excluded.quote, | ||
| 106 | + note = excluded.note, | ||
| 107 | + tags = excluded.tags, | ||
| 108 | + rating = excluded.rating, | ||
| 109 | + cid = excluded.cid""", | ||
| 110 | + p(a.uri), p(a.authorDid), p(a.feedUrl), p(a.articleUrl), | ||
| 111 | + pOrNull(a.quote), pOrNull(a.note), pOrNull(joinTags(a.tags)), | ||
| 112 | + (if a.rating.isSome: p(a.rating.get.int64) else: nullParam()), | ||
| 113 | + p(formatSqliteTime(a.createdAt.get(now().utc))), | ||
| 114 | + pOrNull(a.cid)) | ||
| 115 | + | ||
| 116 | +proc getAnnotation*(g: GleanDb, id: int64): Option[Annotation] = | ||
| 117 | + g.db.queryFirst("SELECT " & annotationColumns & | ||
| 118 | + " FROM articles.annotations an WHERE an.id = ?", | ||
| 119 | + [p(id)], readAnnotation) | ||
| 120 | + | ||
| 121 | +proc annotationExists*(g: GleanDb, uri: string): bool = | ||
| 122 | + g.db.queryInt("SELECT 1 FROM articles.annotations WHERE uri = ?", | ||
| 123 | + p(uri)).isSome | ||
| 124 | + | ||
| 125 | +proc deleteAnnotation*(g: GleanDb, uri: string) = | ||
| 126 | + g.db.run("DELETE FROM articles.annotations WHERE uri = ?", p(uri)) | ||
| 127 | + | ||
| 128 | +proc listAnnotations*(g: GleanDb, feedUrl = "", articleUrl = "", | ||
| 129 | + authorDid = "", limit = 50, offset = 0): seq[Annotation] = | ||
| 130 | + var sql = "SELECT " & annotationColumns & | ||
| 131 | + " FROM articles.annotations an WHERE 1=1" | ||
| 132 | + var params: seq[Param] | ||
| 133 | + if feedUrl.len > 0: | ||
| 134 | + sql &= " AND an.feed_url = ?" | ||
| 135 | + params.add p(feedUrl) | ||
| 136 | + if articleUrl.len > 0: | ||
| 137 | + sql &= " AND an.article_url = ?" | ||
| 138 | + params.add p(articleUrl) | ||
| 139 | + if authorDid.len > 0: | ||
| 140 | + sql &= " AND an.author_did = ?" | ||
| 141 | + params.add p(authorDid) | ||
| 142 | + sql &= " ORDER BY an.created_at DESC LIMIT ? OFFSET ?" | ||
| 143 | + params.add p(limit.int64) | ||
| 144 | + params.add p(offset.int64) | ||
| 145 | + g.db.queryAll(sql, params, readAnnotation) | ||
| 146 | + | ||
| 147 | +proc batchCreateAnnotations*(g: GleanDb, annotations: seq[Annotation]) = | ||
| 148 | + if annotations.len == 0: return | ||
| 149 | + g.db.transaction: | ||
| 150 | + var ps = g.db.prepared(""" | ||
| 151 | + INSERT INTO articles.annotations | ||
| 152 | + (uri, author_did, feed_url, article_url, quote, note, tags, rating, | ||
| 153 | + created_at, cid) | ||
| 154 | + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) | ||
| 155 | + ON CONFLICT(uri) DO NOTHING""") | ||
| 156 | + defer: ps.finalize() | ||
| 157 | + for a in annotations: | ||
| 158 | + ps.exec(p(a.uri), p(a.authorDid), p(a.feedUrl), p(a.articleUrl), | ||
| 159 | + pOrNull(a.quote), pOrNull(a.note), pOrNull(joinTags(a.tags)), | ||
| 160 | + (if a.rating.isSome: p(a.rating.get.int64) else: nullParam()), | ||
| 161 | + p(formatSqliteTime(a.createdAt.get(now().utc))), pOrNull(a.cid)) | ||
| 162 | + | ||
| 163 | +# --- likes ----------------------------------------------------------------- | ||
| 164 | + | ||
| 165 | +proc readLike(r: Row): Like = | ||
| 166 | + Like( | ||
| 167 | + id: r.i64(0), | ||
| 168 | + uri: r.str(1), | ||
| 169 | + authorDid: r.str(2), | ||
| 170 | + feedUrl: r.str(3), | ||
| 171 | + articleUrl: r.str(4), | ||
| 172 | + createdAt: parseSqliteTime(r.str(5)), | ||
| 173 | + cid: r.str(6), | ||
| 174 | + ) | ||
| 175 | + | ||
| 176 | +const likeColumns = """l.id, l.uri, l.author_did, l.feed_url, l.article_url, | ||
| 177 | + l.created_at, l.cid""" | ||
| 178 | + | ||
| 179 | +proc createLike*(g: GleanDb, l: Like) = | ||
| 180 | + g.db.run("""INSERT INTO articles.likes | ||
| 181 | + (uri, author_did, feed_url, article_url, created_at, cid) | ||
| 182 | + VALUES (?, ?, ?, ?, ?, ?) | ||
| 183 | + ON CONFLICT(uri) DO NOTHING""", | ||
| 184 | + p(l.uri), p(l.authorDid), p(l.feedUrl), p(l.articleUrl), | ||
| 185 | + p(formatSqliteTime(l.createdAt.get(now().utc))), pOrNull(l.cid)) | ||
| 186 | + | ||
| 187 | +proc batchCreateLikes*(g: GleanDb, likes: seq[Like]) = | ||
| 188 | + if likes.len == 0: return | ||
| 189 | + g.db.transaction: | ||
| 190 | + var ps = g.db.prepared(""" | ||
| 191 | + INSERT INTO articles.likes | ||
| 192 | + (uri, author_did, feed_url, article_url, created_at, cid) | ||
| 193 | + VALUES (?, ?, ?, ?, ?, ?) | ||
| 194 | + ON CONFLICT(uri) DO NOTHING""") | ||
| 195 | + defer: ps.finalize() | ||
| 196 | + for l in likes: | ||
| 197 | + ps.exec(p(l.uri), p(l.authorDid), p(l.feedUrl), p(l.articleUrl), | ||
| 198 | + p(formatSqliteTime(l.createdAt.get(now().utc))), pOrNull(l.cid)) | ||
| 199 | + | ||
| 200 | +proc deleteLike*(g: GleanDb, uri: string) = | ||
| 201 | + g.db.run("DELETE FROM articles.likes WHERE uri = ?", p(uri)) | ||
| 202 | + | ||
| 203 | +proc deleteLikeByUserArticle*(g: GleanDb, authorDid, feedUrl, articleUrl: string) = | ||
| 204 | + g.db.run("""DELETE FROM articles.likes | ||
| 205 | + WHERE author_did = ? AND feed_url = ? AND article_url = ?""", | ||
| 206 | + p(authorDid), p(feedUrl), p(articleUrl)) | ||
| 207 | + | ||
| 208 | +proc getLike*(g: GleanDb, authorDid, feedUrl, articleUrl: string): Option[Like] = | ||
| 209 | + g.db.queryFirst("SELECT " & likeColumns & """ | ||
| 210 | + FROM articles.likes l | ||
| 211 | + WHERE l.author_did = ? AND l.feed_url = ? AND l.article_url = ?""", | ||
| 212 | + [p(authorDid), p(feedUrl), p(articleUrl)], readLike) | ||
| 213 | + | ||
| 214 | +proc hasLiked*(g: GleanDb, authorDid, feedUrl, articleUrl: string): bool = | ||
| 215 | + getLike(g, authorDid, feedUrl, articleUrl).isSome | ||
| 216 | + | ||
| 217 | +proc getLikeCount*(g: GleanDb, feedUrl, articleUrl: string): int = | ||
| 218 | + g.db.queryInt("""SELECT count(*) FROM articles.likes | ||
| 219 | + WHERE feed_url = ? AND article_url = ?""", | ||
| 220 | + p(feedUrl), p(articleUrl)).get(0).int | ||
| 221 | + | ||
| 222 | +proc listLikes*(g: GleanDb, authorDid: string; feedUrl = "", | ||
| 223 | + limit = 50, offset = 0): seq[Like] = | ||
| 224 | + var sql = "SELECT " & likeColumns & | ||
| 225 | + " FROM articles.likes l WHERE l.author_did = ?" | ||
| 226 | + var params = @[p(authorDid)] | ||
| 227 | + if feedUrl.len > 0: | ||
| 228 | + sql &= " AND l.feed_url = ?" | ||
| 229 | + params.add p(feedUrl) | ||
| 230 | + sql &= " ORDER BY l.created_at DESC LIMIT ? OFFSET ?" | ||
| 231 | + params.add p(limit.int64) | ||
| 232 | + params.add p(offset.int64) | ||
| 233 | + g.db.queryAll(sql, params, readLike) | ||
| 234 | + | ||
| 235 | +# --- reconciliation -------------------------------------------------------- | ||
| 236 | + | ||
| 237 | +proc deleteOrphanedLikes*(g: GleanDb, userDid: string, activeUris: seq[string]) = | ||
| 238 | + ## Remove local likes the user's PDS no longer has. | ||
| 239 | + ## | ||
| 240 | + ## This is the destructive half of a sync, so an empty active set is treated | ||
| 241 | + ## as "delete everything this user liked" only because that is what an empty | ||
| 242 | + ## PDS collection means. A caller that failed to *fetch* the collection must | ||
| 243 | + ## not reach here -- the difference between "you unliked everything" and | ||
| 244 | + ## "the request failed" is not visible from this side. | ||
| 245 | + if activeUris.len == 0: | ||
| 246 | + g.db.run("DELETE FROM articles.likes WHERE author_did = ?", p(userDid)) | ||
| 247 | + return | ||
| 248 | + g.db.transaction: | ||
| 249 | + g.db.run("CREATE TEMP TABLE IF NOT EXISTS active_uris (uri TEXT PRIMARY KEY)") | ||
| 250 | + g.db.run("DELETE FROM active_uris") | ||
| 251 | + var ps = g.db.prepared( | ||
| 252 | + "INSERT OR IGNORE INTO active_uris (uri) VALUES (?)") | ||
| 253 | + defer: ps.finalize() | ||
| 254 | + for uri in activeUris: | ||
| 255 | + ps.exec(p(uri)) | ||
| 256 | + g.db.run("""DELETE FROM articles.likes | ||
| 257 | + WHERE author_did = ? | ||
| 258 | + AND uri NOT IN (SELECT uri FROM active_uris)""", p(userDid)) | ||
| 259 | + | ||
| 260 | +proc deleteOrphanedAnnotations*(g: GleanDb, userDid: string, | ||
| 261 | + activeUris: seq[string]) = | ||
| 262 | + if activeUris.len == 0: | ||
| 263 | + g.db.run("DELETE FROM articles.annotations WHERE author_did = ?", p(userDid)) | ||
| 264 | + return | ||
| 265 | + g.db.transaction: | ||
| 266 | + g.db.run("CREATE TEMP TABLE IF NOT EXISTS active_uris (uri TEXT PRIMARY KEY)") | ||
| 267 | + g.db.run("DELETE FROM active_uris") | ||
| 268 | + var ps = g.db.prepared( | ||
| 269 | + "INSERT OR IGNORE INTO active_uris (uri) VALUES (?)") | ||
| 270 | + defer: ps.finalize() | ||
| 271 | + for uri in activeUris: | ||
| 272 | + ps.exec(p(uri)) | ||
| 273 | + g.db.run("""DELETE FROM articles.annotations | ||
| 274 | + WHERE author_did = ? | ||
| 275 | + AND uri NOT IN (SELECT uri FROM active_uris)""", p(userDid)) | ||
| 276 | + | ||
| 277 | +# --- follows --------------------------------------------------------------- | ||
| 278 | + | ||
| 279 | +proc upsertFollow*(g: GleanDb, userDid, targetDid, uri, cid: string) = | ||
| 280 | + g.db.run("""INSERT INTO follows (user_did, target_did, uri, cid, followed_at) | ||
| 281 | + VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) | ||
| 282 | + ON CONFLICT(user_did, target_did) DO UPDATE SET | ||
| 283 | + uri = COALESCE(excluded.uri, follows.uri), | ||
| 284 | + cid = COALESCE(excluded.cid, follows.cid)""", | ||
| 285 | + p(userDid), p(targetDid), pOrNull(uri), pOrNull(cid)) | ||
| 286 | + | ||
| 287 | +proc deleteFollow*(g: GleanDb, userDid, targetDid: string) = | ||
| 288 | + g.db.run("DELETE FROM follows WHERE user_did = ? AND target_did = ?", | ||
| 289 | + p(userDid), p(targetDid)) | ||
| 290 | + | ||
| 291 | +proc listFollows*(g: GleanDb, userDid: string): seq[string] = | ||
| 292 | + for r in g.db.query( | ||
| 293 | + "SELECT target_did FROM follows WHERE user_did = ? ORDER BY target_did", | ||
| 294 | + p(userDid)): | ||
| 295 | + result.add r.str(0) | ||
| 296 | + | ||
| 297 | +# --- trending -------------------------------------------------------------- | ||
| 298 | + | ||
| 299 | +proc readTrending(r: Row): TrendingItem = | ||
| 300 | + TrendingItem( | ||
| 301 | + articleId: r.i64(0), | ||
| 302 | + title: r.str(1), | ||
| 303 | + url: r.str(2), | ||
| 304 | + author: r.str(3), | ||
| 305 | + summary: r.str(4), | ||
| 306 | + feedUrl: r.str(5), | ||
| 307 | + feedTitle: r.str(6), | ||
| 308 | + faviconUrl: r.str(7), | ||
| 309 | + likeCount: r.i(8), | ||
| 310 | + annotationCount: r.i(9), | ||
| 311 | + hasLiked: r.b(10), | ||
| 312 | + ) | ||
| 313 | + | ||
| 314 | +const trendingSelect = """ | ||
| 315 | + SELECT ar.id, ar.title, COALESCE(ar.url, ''), COALESCE(ar.author, ''), | ||
| 316 | + COALESCE(ar.summary, ''), l.feed_url, COALESCE(f.title, ''), | ||
| 317 | + COALESCE(f.favicon_url, ''), | ||
| 318 | + COUNT(DISTINCT l.id) AS like_count, | ||
| 319 | + COUNT(DISTINCT an.id) AS annotation_count, | ||
| 320 | + COALESCE(MAX(CASE WHEN ul.id IS NOT NULL THEN 1 ELSE 0 END), 0) | ||
| 321 | + FROM articles.likes l | ||
| 322 | + JOIN articles.articles ar | ||
| 323 | + ON ar.url = l.article_url AND ar.feed_url = l.feed_url | ||
| 324 | + LEFT JOIN articles.feeds f ON f.feed_url = l.feed_url | ||
| 325 | + LEFT JOIN articles.annotations an | ||
| 326 | + ON an.feed_url = l.feed_url AND an.article_url = l.article_url | ||
| 327 | + AND an.created_at >= ? | ||
| 328 | + LEFT JOIN articles.likes ul | ||
| 329 | + ON ul.feed_url = l.feed_url AND ul.article_url = l.article_url | ||
| 330 | + AND ul.author_did = ? | ||
| 331 | + WHERE l.created_at >= ?""" | ||
| 332 | + | ||
| 333 | +## Future-dated articles sort last: a feed that publishes with a scheduled | ||
| 334 | +## timestamp would otherwise pin itself to the top of trending indefinitely. | ||
| 335 | +const trendingOrder = """ | ||
| 336 | + GROUP BY ar.id | ||
| 337 | + ORDER BY like_count DESC, annotation_count DESC, | ||
| 338 | + (CASE WHEN ar.published > datetime('now') THEN 1 ELSE 0 END), | ||
| 339 | + ar.published DESC | ||
| 340 | + LIMIT ? OFFSET ?""" | ||
| 341 | + | ||
| 342 | +proc listTrendingArticles*(g: GleanDb, userDid, since: string, | ||
| 343 | + limit = 25, offset = 0): seq[TrendingItem] = | ||
| 344 | + ## Global trending: everyone's likes. | ||
| 345 | + g.db.queryAll(trendingSelect & trendingOrder, | ||
| 346 | + [p(since), p(userDid), p(since), p(limit.int64), p(offset.int64)], | ||
| 347 | + readTrending) | ||
| 348 | + | ||
| 349 | +proc listTrendingArticlesForUser*(g: GleanDb, userDid, since: string, | ||
| 350 | + languages: seq[string] = @[], | ||
| 351 | + limit = 25, offset = 0): seq[TrendingItem] = | ||
| 352 | + ## Trending within the reader's own graph: people they follow, people the | ||
| 353 | + ## similarity table pairs them with, and themselves. | ||
| 354 | + let (filter, langParams) = langFilter(languages, "ar.") | ||
| 355 | + var params = @[p(since), p(userDid), p(since), | ||
| 356 | + p(userDid), p(userDid), p(userDid), p(userDid), p(userDid)] | ||
| 357 | + params.add langParams | ||
| 358 | + params.add p(limit.int64) | ||
| 359 | + params.add p(offset.int64) | ||
| 360 | + g.db.queryAll(trendingSelect & """ | ||
| 361 | + AND l.author_did IN ( | ||
| 362 | + SELECT CASE WHEN us.user_a = ? THEN us.user_b ELSE us.user_a END | ||
| 363 | + FROM recs.user_similarity us | ||
| 364 | + WHERE us.user_a = ? OR us.user_b = ? | ||
| 365 | + UNION SELECT ? | ||
| 366 | + UNION SELECT fo.target_did FROM follows fo WHERE fo.user_did = ? | ||
| 367 | + )""" & filter & trendingOrder, params, readTrending) | ||
| 368 | + | ||
| 369 | +proc listLikedArticles*(g: GleanDb, userDid: string, | ||
| 370 | + limit = 25, offset = 0): seq[Article] = | ||
| 371 | + ## The reader's own likes, as articles. | ||
| 372 | + g.db.queryAll(""" | ||
| 373 | + SELECT a.id, a.feed_url, COALESCE(f.title, a.feed_url), f.favicon_url, | ||
| 374 | + a.guid, a.title, a.url, a.author, a.summary, a.content, | ||
| 375 | + a.full_content, a.published, a.updated, | ||
| 376 | + COALESCE(r.is_read, 0), | ||
| 377 | + (SELECT count(*) FROM articles.likes l2 WHERE l2.article_url = a.url), | ||
| 378 | + 1 | ||
| 379 | + FROM articles.likes l | ||
| 380 | + JOIN articles.articles a | ||
| 381 | + ON a.url = l.article_url AND a.feed_url = l.feed_url | ||
| 382 | + LEFT JOIN articles.feeds f ON f.feed_url = a.feed_url | ||
| 383 | + LEFT JOIN articles.read_state r ON r.article_id = a.id AND r.user_did = ? | ||
| 384 | + WHERE l.author_did = ? | ||
| 385 | + ORDER BY l.created_at DESC | ||
| 386 | + LIMIT ? OFFSET ?""", | ||
| 387 | + [p(userDid), p(userDid), p(limit.int64), p(offset.int64)], | ||
| 388 | + proc(r: Row): Article = readArticleRow(r)) | ||
added
spike/nim/glean_spike.nimble +58 -0 | new file mode 100644 | ||
| @@ -0,0 +1,58 @@ | ||
| 1 | +version = "0.1.0" | |
| 2 | +author = "glean" | |
| 3 | +description = "Nim port spike for glean: atproto, storage and the database layer" | |
| 4 | +license = "MIT" | |
| 5 | +srcDir = "." | |
| 6 | + | |
| 7 | +requires "nim >= 2.2.0" | |
| 8 | +requires "bearssl" | |
| 9 | +requires "nimcrypto" | |
| 10 | +requires "chronos" | |
| 11 | +requires "websock" | |
| 12 | + | |
| 13 | +# Probes that touch only this machine. Safe to run anywhere, including CI. | |
| 14 | +const offlineProbes = @[ | |
| 15 | + "sqlite_probe", | |
| 16 | + "dpop_probe", | |
| 17 | + "store_probe", | |
| 18 | + "gleandb_probe", | |
| 19 | + "stores_probe", | |
| 20 | + "social_probe", | |
| 21 | + "reconnect_probe", # uses a local websocket server, not the network | |
| 22 | +] | |
| 23 | + | |
| 24 | +# Probes that talk to real servers. Kept separate because a failure here can | |
| 25 | +# mean "bsky.social is having a bad day" rather than "the port is broken", and | |
| 26 | +# a test suite that cannot tell those apart stops being believed. | |
| 27 | +const networkProbes = @[ | |
| 28 | + "oauth_probe", | |
| 29 | + "xrpc_probe", | |
| 30 | + "jetstream_probe", | |
| 31 | +] | |
| 32 | + | |
| 33 | +proc runProbe(name: string) = | |
| 34 | + echo "\n=== " & name | |
| 35 | + exec "nim c -r --hints:off -d:ssl " & name & ".nim" | |
| 36 | + | |
| 37 | +task test, "Run the probes that need no network": | |
| 38 | + for probe in offlineProbes: | |
| 39 | + runProbe(probe) | |
| 40 | + echo "\nAll offline probes passed." | |
| 41 | + | |
| 42 | +task testnet, "Run the probes that talk to real servers": | |
| 43 | + for probe in networkProbes: | |
| 44 | + runProbe(probe) | |
| 45 | + echo "\nAll network probes passed." | |
| 46 | + | |
| 47 | +task testall, "Run every probe": | |
| 48 | + for probe in offlineProbes & networkProbes: | |
| 49 | + runProbe(probe) | |
| 50 | + echo "\nAll probes passed." | |
| 51 | + | |
| 52 | +task build, "Compile every module without running anything": | |
| 53 | + for probe in offlineProbes & networkProbes: | |
| 54 | + exec "nim c --hints:off -d:ssl -c " & probe & ".nim" | |
| 55 | + echo "\nEverything compiles." | |
| 56 | + | |
| 57 | +task schema, "Regenerate atproto/schema.nim from the Go source": | |
| 58 | + exec "python3 tools/extract_schema.py" | |
| new file mode 100644 | |||
| @@ -0,0 +1,58 @@ | |||
| 1 | +version = "0.1.0" | ||
| 2 | +author = "glean" | ||
| 3 | +description = "Nim port spike for glean: atproto, storage and the database layer" | ||
| 4 | +license = "MIT" | ||
| 5 | +srcDir = "." | ||
| 6 | + | ||
| 7 | +requires "nim >= 2.2.0" | ||
| 8 | +requires "bearssl" | ||
| 9 | +requires "nimcrypto" | ||
| 10 | +requires "chronos" | ||
| 11 | +requires "websock" | ||
| 12 | + | ||
| 13 | +# Probes that touch only this machine. Safe to run anywhere, including CI. | ||
| 14 | +const offlineProbes = @[ | ||
| 15 | + "sqlite_probe", | ||
| 16 | + "dpop_probe", | ||
| 17 | + "store_probe", | ||
| 18 | + "gleandb_probe", | ||
| 19 | + "stores_probe", | ||
| 20 | + "social_probe", | ||
| 21 | + "reconnect_probe", # uses a local websocket server, not the network | ||
| 22 | +] | ||
| 23 | + | ||
| 24 | +# Probes that talk to real servers. Kept separate because a failure here can | ||
| 25 | +# mean "bsky.social is having a bad day" rather than "the port is broken", and | ||
| 26 | +# a test suite that cannot tell those apart stops being believed. | ||
| 27 | +const networkProbes = @[ | ||
| 28 | + "oauth_probe", | ||
| 29 | + "xrpc_probe", | ||
| 30 | + "jetstream_probe", | ||
| 31 | +] | ||
| 32 | + | ||
| 33 | +proc runProbe(name: string) = | ||
| 34 | + echo "\n=== " & name | ||
| 35 | + exec "nim c -r --hints:off -d:ssl " & name & ".nim" | ||
| 36 | + | ||
| 37 | +task test, "Run the probes that need no network": | ||
| 38 | + for probe in offlineProbes: | ||
| 39 | + runProbe(probe) | ||
| 40 | + echo "\nAll offline probes passed." | ||
| 41 | + | ||
| 42 | +task testnet, "Run the probes that talk to real servers": | ||
| 43 | + for probe in networkProbes: | ||
| 44 | + runProbe(probe) | ||
| 45 | + echo "\nAll network probes passed." | ||
| 46 | + | ||
| 47 | +task testall, "Run every probe": | ||
| 48 | + for probe in offlineProbes & networkProbes: | ||
| 49 | + runProbe(probe) | ||
| 50 | + echo "\nAll probes passed." | ||
| 51 | + | ||
| 52 | +task build, "Compile every module without running anything": | ||
| 53 | + for probe in offlineProbes & networkProbes: | ||
| 54 | + exec "nim c --hints:off -d:ssl -c " & probe & ".nim" | ||
| 55 | + echo "\nEverything compiles." | ||
| 56 | + | ||
| 57 | +task schema, "Regenerate atproto/schema.nim from the Go source": | ||
| 58 | + exec "python3 tools/extract_schema.py" | ||
modified
spike/nim/readme.md +39 -9 | @@ -9,18 +9,21 @@ Run both: | ||
| 9 | 9 | |
| 10 | 10 | ``` |
| 11 | 11 | cd spike/nim |
| 12 | -nim c -r sqlite_probe.nim # storage | |
| 13 | -nim c -r dpop_probe.nim && ./dpop_probe | (cd verify && go run .) # crypto | |
| 14 | -nim c -r -d:ssl oauth_probe.nim [handle] # the flow | |
| 15 | -nim c -r -d:ssl xrpc_probe.nim # signing | |
| 16 | -nim c -r -d:ssl jetstream_probe.nim # firehose | |
| 17 | -nim c -r -d:ssl reconnect_probe.nim # supervisor | |
| 18 | -nim c -r -d:ssl store_probe.nim # persistence | |
| 19 | -nim c -r -d:ssl gleandb_probe.nim # database | |
| 20 | -nim c -r -d:ssl stores_probe.nim # queries | |
| 12 | +nimble test # everything that needs no network | |
| 13 | +nimble testnet # the probes that talk to real servers | |
| 14 | +nimble testall # both | |
| 15 | +nimble schema # regenerate atproto/schema.nim from the Go source | |
| 21 | 16 | ``` |
| 22 | 17 | |
| 23 | 18 | Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks. |
| 19 | +They are split by whether they touch the network: a failure against | |
| 20 | +`bsky.social` can mean that server is having a bad day rather than that the | |
| 21 | +port is broken, and a suite that cannot tell those apart stops being | |
| 22 | +believed. Individual probes still run on their own: | |
| 23 | + | |
| 24 | +``` | |
| 25 | +nim c -r -d:ssl <name>_probe.nim | |
| 26 | +``` | |
| 24 | 27 | |
| 25 | 28 | ## 1. sqlite-vec + FTS5 — works |
| 26 | 29 | |
| @@ -335,6 +338,33 @@ filtering, read state, counts, search, retention, dead feeds. Not ported: | ||
| 335 | 338 | `GetNextArticleID`'s navigation logic, and the several list variants the |
| 336 | 339 | recommendation engine uses. They are mechanical against these patterns. |
| 337 | 340 | |
| 341 | +## 10. Likes, annotations, follows and trending | |
| 342 | + | |
| 343 | +`atproto/socialstore.nim` ports `internal/db/social.go`, completing the | |
| 344 | +database layer. | |
| 345 | + | |
| 346 | +Likes and annotations are ATProto records first and rows second. They are | |
| 347 | +keyed by their `at://` URI because that is the identity the PDS owns, and | |
| 348 | +each can arrive twice -- once from the firehose and once from a PDS sync -- | |
| 349 | +so every write is idempotent by construction rather than by checking first. | |
| 350 | +An edit keeps the URI and changes the content, which is why annotations | |
| 351 | +upsert rather than insert. | |
| 352 | + | |
| 353 | +Two details worth keeping: | |
| 354 | + | |
| 355 | +- The language filter always keeps `language = ''`. That means *not yet | |
| 356 | + classified*, not *no language*; excluding it would make new articles vanish | |
| 357 | + for anyone with a filter set and reappear minutes later once classified. | |
| 358 | +- Future-dated articles sort last in trending. A feed publishing with a | |
| 359 | + scheduled timestamp would otherwise pin itself to the top indefinitely. | |
| 360 | + | |
| 361 | +The probe weights toward the destructive path. `deleteOrphanedLikes` and its | |
| 362 | +annotation twin delete local rows the PDS no longer has, so the checks assert | |
| 363 | +both halves: the dropped rows go, and *other* users' rows do not. The empty | |
| 364 | +active set is called out in the code, because "you unliked everything" and | |
| 365 | +"the fetch failed" are indistinguishable from inside the store — the caller | |
| 366 | +has to not get there. | |
| 367 | + | |
| 338 | 368 | ## What this does and does not prove |
| 339 | 369 | |
| 340 | 370 | Proven: the cryptography, the storage layer, and the OAuth flow up to user |
| @@ -9,18 +9,21 @@ Run both: | |||
| 9 | 9 | ||
| 10 | ``` | 10 | ``` |
| 11 | cd spike/nim | 11 | cd spike/nim |
| 12 | -nim c -r sqlite_probe.nim # storage | 12 | +nimble test # everything that needs no network |
| 13 | -nim c -r dpop_probe.nim && ./dpop_probe | (cd verify && go run .) # crypto | 13 | +nimble testnet # the probes that talk to real servers |
| 14 | -nim c -r -d:ssl oauth_probe.nim [handle] # the flow | 14 | +nimble testall # both |
| 15 | -nim c -r -d:ssl xrpc_probe.nim # signing | 15 | +nimble schema # regenerate atproto/schema.nim from the Go source |
| 16 | -nim c -r -d:ssl jetstream_probe.nim # firehose | ||
| 17 | -nim c -r -d:ssl reconnect_probe.nim # supervisor | ||
| 18 | -nim c -r -d:ssl store_probe.nim # persistence | ||
| 19 | -nim c -r -d:ssl gleandb_probe.nim # database | ||
| 20 | -nim c -r -d:ssl stores_probe.nim # queries | ||
| 21 | ``` | 16 | ``` |
| 22 | 17 | ||
| 23 | Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks. | 18 | Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks. |
| 19 | +They are split by whether they touch the network: a failure against | ||
| 20 | +`bsky.social` can mean that server is having a bad day rather than that the | ||
| 21 | +port is broken, and a suite that cannot tell those apart stops being | ||
| 22 | +believed. Individual probes still run on their own: | ||
| 23 | + | ||
| 24 | +``` | ||
| 25 | +nim c -r -d:ssl <name>_probe.nim | ||
| 26 | +``` | ||
| 24 | 27 | ||
| 25 | ## 1. sqlite-vec + FTS5 — works | 28 | ## 1. sqlite-vec + FTS5 — works |
| 26 | 29 | ||
| @@ -335,6 +338,33 @@ filtering, read state, counts, search, retention, dead feeds. Not ported: | |||
| 335 | `GetNextArticleID`'s navigation logic, and the several list variants the | 338 | `GetNextArticleID`'s navigation logic, and the several list variants the |
| 336 | recommendation engine uses. They are mechanical against these patterns. | 339 | recommendation engine uses. They are mechanical against these patterns. |
| 337 | 340 | ||
| 341 | +## 10. Likes, annotations, follows and trending | ||
| 342 | + | ||
| 343 | +`atproto/socialstore.nim` ports `internal/db/social.go`, completing the | ||
| 344 | +database layer. | ||
| 345 | + | ||
| 346 | +Likes and annotations are ATProto records first and rows second. They are | ||
| 347 | +keyed by their `at://` URI because that is the identity the PDS owns, and | ||
| 348 | +each can arrive twice -- once from the firehose and once from a PDS sync -- | ||
| 349 | +so every write is idempotent by construction rather than by checking first. | ||
| 350 | +An edit keeps the URI and changes the content, which is why annotations | ||
| 351 | +upsert rather than insert. | ||
| 352 | + | ||
| 353 | +Two details worth keeping: | ||
| 354 | + | ||
| 355 | +- The language filter always keeps `language = ''`. That means *not yet | ||
| 356 | + classified*, not *no language*; excluding it would make new articles vanish | ||
| 357 | + for anyone with a filter set and reappear minutes later once classified. | ||
| 358 | +- Future-dated articles sort last in trending. A feed publishing with a | ||
| 359 | + scheduled timestamp would otherwise pin itself to the top indefinitely. | ||
| 360 | + | ||
| 361 | +The probe weights toward the destructive path. `deleteOrphanedLikes` and its | ||
| 362 | +annotation twin delete local rows the PDS no longer has, so the checks assert | ||
| 363 | +both halves: the dropped rows go, and *other* users' rows do not. The empty | ||
| 364 | +active set is called out in the code, because "you unliked everything" and | ||
| 365 | +"the fetch failed" are indistinguishable from inside the store — the caller | ||
| 366 | +has to not get there. | ||
| 367 | + | ||
| 338 | ## What this does and does not prove | 368 | ## What this does and does not prove |
| 339 | 369 | ||
| 340 | Proven: the cryptography, the storage layer, and the OAuth flow up to user | 370 | Proven: the cryptography, the storage layer, and the OAuth flow up to user |
added
spike/nim/social_probe.nim +216 -0 | new file mode 100644 | ||
| @@ -0,0 +1,216 @@ | ||
| 1 | +## Exercises likes, annotations, follows and trending. | |
| 2 | +## | |
| 3 | +## nim c -r -d:ssl social_probe.nim | |
| 4 | +## | |
| 5 | +## Weighted towards the destructive and the social-graph paths: orphan | |
| 6 | +## reconciliation deletes rows, and trending decides what a reader sees, so | |
| 7 | +## both are worth more than "the insert worked". | |
| 8 | + | |
| 9 | +import std/[options, os, strformat, strutils, times] | |
| 10 | +import atproto/[articlestore, feedstore, gleandb, socialstore, 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 | + carol = "did:plc:carol" | |
| 29 | + feedA = "https://a.test/feed" | |
| 30 | + urlA1 = "https://a.test/1" | |
| 31 | + urlA2 = "https://a.test/2" | |
| 32 | + | |
| 33 | +proc main() = | |
| 34 | + echo "Social store probe" | |
| 35 | + echo "" | |
| 36 | + | |
| 37 | + let base = getTempDir() / "glean_social_probe" | |
| 38 | + cleanup(base) | |
| 39 | + defer: cleanup(base) | |
| 40 | + | |
| 41 | + var g = gleandb.open(base) | |
| 42 | + defer: g.close() | |
| 43 | + g.migrate() | |
| 44 | + | |
| 45 | + for did in [alice, bob, carol]: | |
| 46 | + g.db.run("INSERT INTO users (did) VALUES (?)", p(did)) | |
| 47 | + g.upsertFeed(Feed(feedUrl: feedA, title: "Feed A", feedType: "rss")) | |
| 48 | + for did in [alice, bob, carol]: | |
| 49 | + g.createSubscription(did, feedA, "", "", "", "") | |
| 50 | + | |
| 51 | + let t = now().utc | |
| 52 | + g.batchUpsertArticles(@[ | |
| 53 | + NewArticle(feedUrl: feedA, guid: "a1", title: "First", url: urlA1, | |
| 54 | + published: some(t - initDuration(hours = 2))), | |
| 55 | + NewArticle(feedUrl: feedA, guid: "a2", title: "Second", url: urlA2, | |
| 56 | + published: some(t - initDuration(hours = 1))), | |
| 57 | + ]) | |
| 58 | + | |
| 59 | + # --- likes -------------------------------------------------------------- | |
| 60 | + block: | |
| 61 | + g.createLike(Like(uri: "at://alice/like/1", authorDid: alice, | |
| 62 | + feedUrl: feedA, articleUrl: urlA1, createdAt: some(t))) | |
| 63 | + report("a like round-trips", g.hasLiked(alice, feedA, urlA1)) | |
| 64 | + report("a like is per author", not g.hasLiked(bob, feedA, urlA1)) | |
| 65 | + | |
| 66 | + # The same record arrives from Jetstream and from a PDS sync. | |
| 67 | + g.createLike(Like(uri: "at://alice/like/1", authorDid: alice, | |
| 68 | + feedUrl: feedA, articleUrl: urlA1, createdAt: some(t))) | |
| 69 | + report("re-delivering a like does not duplicate it", | |
| 70 | + g.getLikeCount(feedA, urlA1) == 1, $g.getLikeCount(feedA, urlA1)) | |
| 71 | + | |
| 72 | + g.batchCreateLikes(@[ | |
| 73 | + Like(uri: "at://bob/like/1", authorDid: bob, feedUrl: feedA, | |
| 74 | + articleUrl: urlA1, createdAt: some(t)), | |
| 75 | + Like(uri: "at://carol/like/1", authorDid: carol, feedUrl: feedA, | |
| 76 | + articleUrl: urlA1, createdAt: some(t)), | |
| 77 | + Like(uri: "at://bob/like/2", authorDid: bob, feedUrl: feedA, | |
| 78 | + articleUrl: urlA2, createdAt: some(t)), | |
| 79 | + ]) | |
| 80 | + report("batch insert works", g.getLikeCount(feedA, urlA1) == 3, | |
| 81 | + $g.getLikeCount(feedA, urlA1)) | |
| 82 | + report("likes are listed per author", | |
| 83 | + g.listLikes(bob).len == 2, $g.listLikes(bob).len) | |
| 84 | + report("likes can be scoped to a feed", | |
| 85 | + g.listLikes(bob, feedUrl = feedA).len == 2) | |
| 86 | + | |
| 87 | + g.deleteLikeByUserArticle(alice, feedA, urlA1) | |
| 88 | + report("unliking removes only that author's like", | |
| 89 | + g.getLikeCount(feedA, urlA1) == 2 and not g.hasLiked(alice, feedA, urlA1), | |
| 90 | + $g.getLikeCount(feedA, urlA1)) | |
| 91 | + | |
| 92 | + # --- annotations -------------------------------------------------------- | |
| 93 | + block: | |
| 94 | + g.createAnnotation(Annotation( | |
| 95 | + uri: "at://alice/ann/1", authorDid: alice, feedUrl: feedA, | |
| 96 | + articleUrl: urlA1, quote: "a quote", note: "a note", | |
| 97 | + tags: @["nim", "systems"], rating: some(4), createdAt: some(t))) | |
| 98 | + let listed = g.listAnnotations(articleUrl = urlA1) | |
| 99 | + report("an annotation round-trips", listed.len == 1, $listed.len) | |
| 100 | + if listed.len > 0: | |
| 101 | + report("tags round-trip as a list", listed[0].tags == @["nim", "systems"], | |
| 102 | + listed[0].tags.join("|")) | |
| 103 | + report("rating round-trips", listed[0].rating == some(4)) | |
| 104 | + | |
| 105 | + # An edit keeps the URI and changes the content. | |
| 106 | + g.createAnnotation(Annotation( | |
| 107 | + uri: "at://alice/ann/1", authorDid: alice, feedUrl: feedA, | |
| 108 | + articleUrl: urlA1, note: "edited", tags: @["nim"], createdAt: some(t))) | |
| 109 | + let edited = g.listAnnotations(articleUrl = urlA1) | |
| 110 | + report("re-delivering an annotation updates rather than duplicates", | |
| 111 | + edited.len == 1 and edited[0].note == "edited", | |
| 112 | + &"{edited.len} rows, note={edited[0].note}") | |
| 113 | + report("a cleared rating round-trips as none", edited[0].rating.isNone) | |
| 114 | + | |
| 115 | + report("annotationExists finds it", g.annotationExists("at://alice/ann/1")) | |
| 116 | + report("annotationExists rejects an unknown uri", | |
| 117 | + not g.annotationExists("at://nobody/ann/9")) | |
| 118 | + | |
| 119 | + g.batchCreateAnnotations(@[ | |
| 120 | + Annotation(uri: "at://bob/ann/1", authorDid: bob, feedUrl: feedA, | |
| 121 | + articleUrl: urlA1, note: "bob's", createdAt: some(t))]) | |
| 122 | + report("annotations are listed per author", | |
| 123 | + g.listAnnotations(authorDid = bob).len == 1) | |
| 124 | + | |
| 125 | + # --- follows ------------------------------------------------------------ | |
| 126 | + block: | |
| 127 | + g.upsertFollow(alice, bob, "at://alice/follow/1", "cid1") | |
| 128 | + g.upsertFollow(alice, bob, "at://alice/follow/1", "cid1") | |
| 129 | + report("following is idempotent", g.listFollows(alice) == @[bob], | |
| 130 | + g.listFollows(alice).join(",")) | |
| 131 | + g.upsertFollow(alice, carol, "", "") | |
| 132 | + report("follows accumulate", g.listFollows(alice).len == 2) | |
| 133 | + g.deleteFollow(alice, carol) | |
| 134 | + report("unfollowing removes one edge", g.listFollows(alice) == @[bob]) | |
| 135 | + | |
| 136 | + # --- trending ----------------------------------------------------------- | |
| 137 | + block: | |
| 138 | + let since = formatSqliteTime(t - initDuration(days = 1)) | |
| 139 | + let global = g.listTrendingArticles(alice, since) | |
| 140 | + report("global trending ranks by likes", | |
| 141 | + global.len == 2 and global[0].url == urlA1, | |
| 142 | + if global.len > 0: &"top={global[0].url} likes={global[0].likeCount}" | |
| 143 | + else: "empty") | |
| 144 | + report("trending counts annotations too", | |
| 145 | + global.len > 0 and global[0].annotationCount == 2, | |
| 146 | + if global.len > 0: $global[0].annotationCount else: "-") | |
| 147 | + report("hasLiked is per viewer", | |
| 148 | + not global[0].hasLiked, "alice unliked hers earlier") | |
| 149 | + | |
| 150 | + let viewedByBob = g.listTrendingArticles(bob, since) | |
| 151 | + report("hasLiked is true for a viewer who liked it", | |
| 152 | + viewedByBob[0].hasLiked) | |
| 153 | + | |
| 154 | + # Alice follows only bob, so carol's like should not pull an article in | |
| 155 | + # that bob did not also like. | |
| 156 | + let scoped = g.listTrendingArticlesForUser(alice, since) | |
| 157 | + report("graph-scoped trending only counts followed authors", | |
| 158 | + scoped.len > 0, $scoped.len) | |
| 159 | + let scopedTop = scoped[0] | |
| 160 | + report("a followed author's likes are counted", | |
| 161 | + scopedTop.likeCount >= 1, $scopedTop.likeCount) | |
| 162 | + | |
| 163 | + # An author nobody follows contributes nothing to a graph-scoped view. | |
| 164 | + let carolOnly = g.listTrendingArticlesForUser(carol, since) | |
| 165 | + report("a reader with no follows still sees their own likes", | |
| 166 | + carolOnly.len == 1, $carolOnly.len) | |
| 167 | + | |
| 168 | + # --- language filter ---------------------------------------------------- | |
| 169 | + block: | |
| 170 | + let (sql, params) = langFilter(@["en", "fr"], "ar.") | |
| 171 | + report("language filter builds placeholders", | |
| 172 | + sql.count("?") == 2 and params.len == 2, sql.strip()) | |
| 173 | + # Unclassified articles must survive the filter, or new items vanish for | |
| 174 | + # a few minutes and then reappear. | |
| 175 | + report("language filter keeps unclassified articles", | |
| 176 | + "language = ''" in sql) | |
| 177 | + report("an empty language list adds no filter", | |
| 178 | + langFilter(@[], "ar.")[0].len == 0) | |
| 179 | + | |
| 180 | + # --- liked articles ----------------------------------------------------- | |
| 181 | + block: | |
| 182 | + let liked = g.listLikedArticles(bob) | |
| 183 | + report("liked articles are listed newest first", liked.len == 2, $liked.len) | |
| 184 | + report("liked articles carry article fields", | |
| 185 | + liked[0].title.len > 0 and liked[0].feedTitle == "Feed A", | |
| 186 | + liked[0].title) | |
| 187 | + | |
| 188 | + # --- reconciliation ----------------------------------------------------- | |
| 189 | + # The destructive path: local rows the PDS no longer has must go, and | |
| 190 | + # everyone else's must not. | |
| 191 | + block: | |
| 192 | + report("bob starts with 2 likes", g.listLikes(bob).len == 2) | |
| 193 | + g.deleteOrphanedLikes(bob, @["at://bob/like/1"]) | |
| 194 | + report("reconciling deletes likes the PDS dropped", | |
| 195 | + g.listLikes(bob).len == 1, $g.listLikes(bob).len) | |
| 196 | + report("reconciling leaves other users alone", | |
| 197 | + g.listLikes(carol).len == 1, $g.listLikes(carol).len) | |
| 198 | + | |
| 199 | + g.deleteOrphanedAnnotations(alice, @["at://alice/ann/1"]) | |
| 200 | + report("reconciling keeps annotations the PDS still has", | |
| 201 | + g.listAnnotations(authorDid = alice).len == 1) | |
| 202 | + g.deleteOrphanedAnnotations(alice, @[]) | |
| 203 | + report("an empty active set clears that author's annotations", | |
| 204 | + g.listAnnotations(authorDid = alice).len == 0) | |
| 205 | + report("and leaves another author's annotations", | |
| 206 | + g.listAnnotations(authorDid = bob).len == 1) | |
| 207 | + | |
| 208 | + echo "" | |
| 209 | + if failures == 0: | |
| 210 | + echo "RESULT: likes, annotations, follows and trending behave." | |
| 211 | + else: | |
| 212 | + echo &"RESULT: {failures} check(s) failed." | |
| 213 | + quit 1 | |
| 214 | + | |
| 215 | +when isMainModule: | |
| 216 | + main() | |
| new file mode 100644 | |||
| @@ -0,0 +1,216 @@ | |||
| 1 | +## Exercises likes, annotations, follows and trending. | ||
| 2 | +## | ||
| 3 | +## nim c -r -d:ssl social_probe.nim | ||
| 4 | +## | ||
| 5 | +## Weighted towards the destructive and the social-graph paths: orphan | ||
| 6 | +## reconciliation deletes rows, and trending decides what a reader sees, so | ||
| 7 | +## both are worth more than "the insert worked". | ||
| 8 | + | ||
| 9 | +import std/[options, os, strformat, strutils, times] | ||
| 10 | +import atproto/[articlestore, feedstore, gleandb, socialstore, 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 | + carol = "did:plc:carol" | ||
| 29 | + feedA = "https://a.test/feed" | ||
| 30 | + urlA1 = "https://a.test/1" | ||
| 31 | + urlA2 = "https://a.test/2" | ||
| 32 | + | ||
| 33 | +proc main() = | ||
| 34 | + echo "Social store probe" | ||
| 35 | + echo "" | ||
| 36 | + | ||
| 37 | + let base = getTempDir() / "glean_social_probe" | ||
| 38 | + cleanup(base) | ||
| 39 | + defer: cleanup(base) | ||
| 40 | + | ||
| 41 | + var g = gleandb.open(base) | ||
| 42 | + defer: g.close() | ||
| 43 | + g.migrate() | ||
| 44 | + | ||
| 45 | + for did in [alice, bob, carol]: | ||
| 46 | + g.db.run("INSERT INTO users (did) VALUES (?)", p(did)) | ||
| 47 | + g.upsertFeed(Feed(feedUrl: feedA, title: "Feed A", feedType: "rss")) | ||
| 48 | + for did in [alice, bob, carol]: | ||
| 49 | + g.createSubscription(did, feedA, "", "", "", "") | ||
| 50 | + | ||
| 51 | + let t = now().utc | ||
| 52 | + g.batchUpsertArticles(@[ | ||
| 53 | + NewArticle(feedUrl: feedA, guid: "a1", title: "First", url: urlA1, | ||
| 54 | + published: some(t - initDuration(hours = 2))), | ||
| 55 | + NewArticle(feedUrl: feedA, guid: "a2", title: "Second", url: urlA2, | ||
| 56 | + published: some(t - initDuration(hours = 1))), | ||
| 57 | + ]) | ||
| 58 | + | ||
| 59 | + # --- likes -------------------------------------------------------------- | ||
| 60 | + block: | ||
| 61 | + g.createLike(Like(uri: "at://alice/like/1", authorDid: alice, | ||
| 62 | + feedUrl: feedA, articleUrl: urlA1, createdAt: some(t))) | ||
| 63 | + report("a like round-trips", g.hasLiked(alice, feedA, urlA1)) | ||
| 64 | + report("a like is per author", not g.hasLiked(bob, feedA, urlA1)) | ||
| 65 | + | ||
| 66 | + # The same record arrives from Jetstream and from a PDS sync. | ||
| 67 | + g.createLike(Like(uri: "at://alice/like/1", authorDid: alice, | ||
| 68 | + feedUrl: feedA, articleUrl: urlA1, createdAt: some(t))) | ||
| 69 | + report("re-delivering a like does not duplicate it", | ||
| 70 | + g.getLikeCount(feedA, urlA1) == 1, $g.getLikeCount(feedA, urlA1)) | ||
| 71 | + | ||
| 72 | + g.batchCreateLikes(@[ | ||
| 73 | + Like(uri: "at://bob/like/1", authorDid: bob, feedUrl: feedA, | ||
| 74 | + articleUrl: urlA1, createdAt: some(t)), | ||
| 75 | + Like(uri: "at://carol/like/1", authorDid: carol, feedUrl: feedA, | ||
| 76 | + articleUrl: urlA1, createdAt: some(t)), | ||
| 77 | + Like(uri: "at://bob/like/2", authorDid: bob, feedUrl: feedA, | ||
| 78 | + articleUrl: urlA2, createdAt: some(t)), | ||
| 79 | + ]) | ||
| 80 | + report("batch insert works", g.getLikeCount(feedA, urlA1) == 3, | ||
| 81 | + $g.getLikeCount(feedA, urlA1)) | ||
| 82 | + report("likes are listed per author", | ||
| 83 | + g.listLikes(bob).len == 2, $g.listLikes(bob).len) | ||
| 84 | + report("likes can be scoped to a feed", | ||
| 85 | + g.listLikes(bob, feedUrl = feedA).len == 2) | ||
| 86 | + | ||
| 87 | + g.deleteLikeByUserArticle(alice, feedA, urlA1) | ||
| 88 | + report("unliking removes only that author's like", | ||
| 89 | + g.getLikeCount(feedA, urlA1) == 2 and not g.hasLiked(alice, feedA, urlA1), | ||
| 90 | + $g.getLikeCount(feedA, urlA1)) | ||
| 91 | + | ||
| 92 | + # --- annotations -------------------------------------------------------- | ||
| 93 | + block: | ||
| 94 | + g.createAnnotation(Annotation( | ||
| 95 | + uri: "at://alice/ann/1", authorDid: alice, feedUrl: feedA, | ||
| 96 | + articleUrl: urlA1, quote: "a quote", note: "a note", | ||
| 97 | + tags: @["nim", "systems"], rating: some(4), createdAt: some(t))) | ||
| 98 | + let listed = g.listAnnotations(articleUrl = urlA1) | ||
| 99 | + report("an annotation round-trips", listed.len == 1, $listed.len) | ||
| 100 | + if listed.len > 0: | ||
| 101 | + report("tags round-trip as a list", listed[0].tags == @["nim", "systems"], | ||
| 102 | + listed[0].tags.join("|")) | ||
| 103 | + report("rating round-trips", listed[0].rating == some(4)) | ||
| 104 | + | ||
| 105 | + # An edit keeps the URI and changes the content. | ||
| 106 | + g.createAnnotation(Annotation( | ||
| 107 | + uri: "at://alice/ann/1", authorDid: alice, feedUrl: feedA, | ||
| 108 | + articleUrl: urlA1, note: "edited", tags: @["nim"], createdAt: some(t))) | ||
| 109 | + let edited = g.listAnnotations(articleUrl = urlA1) | ||
| 110 | + report("re-delivering an annotation updates rather than duplicates", | ||
| 111 | + edited.len == 1 and edited[0].note == "edited", | ||
| 112 | + &"{edited.len} rows, note={edited[0].note}") | ||
| 113 | + report("a cleared rating round-trips as none", edited[0].rating.isNone) | ||
| 114 | + | ||
| 115 | + report("annotationExists finds it", g.annotationExists("at://alice/ann/1")) | ||
| 116 | + report("annotationExists rejects an unknown uri", | ||
| 117 | + not g.annotationExists("at://nobody/ann/9")) | ||
| 118 | + | ||
| 119 | + g.batchCreateAnnotations(@[ | ||
| 120 | + Annotation(uri: "at://bob/ann/1", authorDid: bob, feedUrl: feedA, | ||
| 121 | + articleUrl: urlA1, note: "bob's", createdAt: some(t))]) | ||
| 122 | + report("annotations are listed per author", | ||
| 123 | + g.listAnnotations(authorDid = bob).len == 1) | ||
| 124 | + | ||
| 125 | + # --- follows ------------------------------------------------------------ | ||
| 126 | + block: | ||
| 127 | + g.upsertFollow(alice, bob, "at://alice/follow/1", "cid1") | ||
| 128 | + g.upsertFollow(alice, bob, "at://alice/follow/1", "cid1") | ||
| 129 | + report("following is idempotent", g.listFollows(alice) == @[bob], | ||
| 130 | + g.listFollows(alice).join(",")) | ||
| 131 | + g.upsertFollow(alice, carol, "", "") | ||
| 132 | + report("follows accumulate", g.listFollows(alice).len == 2) | ||
| 133 | + g.deleteFollow(alice, carol) | ||
| 134 | + report("unfollowing removes one edge", g.listFollows(alice) == @[bob]) | ||
| 135 | + | ||
| 136 | + # --- trending ----------------------------------------------------------- | ||
| 137 | + block: | ||
| 138 | + let since = formatSqliteTime(t - initDuration(days = 1)) | ||
| 139 | + let global = g.listTrendingArticles(alice, since) | ||
| 140 | + report("global trending ranks by likes", | ||
| 141 | + global.len == 2 and global[0].url == urlA1, | ||
| 142 | + if global.len > 0: &"top={global[0].url} likes={global[0].likeCount}" | ||
| 143 | + else: "empty") | ||
| 144 | + report("trending counts annotations too", | ||
| 145 | + global.len > 0 and global[0].annotationCount == 2, | ||
| 146 | + if global.len > 0: $global[0].annotationCount else: "-") | ||
| 147 | + report("hasLiked is per viewer", | ||
| 148 | + not global[0].hasLiked, "alice unliked hers earlier") | ||
| 149 | + | ||
| 150 | + let viewedByBob = g.listTrendingArticles(bob, since) | ||
| 151 | + report("hasLiked is true for a viewer who liked it", | ||
| 152 | + viewedByBob[0].hasLiked) | ||
| 153 | + | ||
| 154 | + # Alice follows only bob, so carol's like should not pull an article in | ||
| 155 | + # that bob did not also like. | ||
| 156 | + let scoped = g.listTrendingArticlesForUser(alice, since) | ||
| 157 | + report("graph-scoped trending only counts followed authors", | ||
| 158 | + scoped.len > 0, $scoped.len) | ||
| 159 | + let scopedTop = scoped[0] | ||
| 160 | + report("a followed author's likes are counted", | ||
| 161 | + scopedTop.likeCount >= 1, $scopedTop.likeCount) | ||
| 162 | + | ||
| 163 | + # An author nobody follows contributes nothing to a graph-scoped view. | ||
| 164 | + let carolOnly = g.listTrendingArticlesForUser(carol, since) | ||
| 165 | + report("a reader with no follows still sees their own likes", | ||
| 166 | + carolOnly.len == 1, $carolOnly.len) | ||
| 167 | + | ||
| 168 | + # --- language filter ---------------------------------------------------- | ||
| 169 | + block: | ||
| 170 | + let (sql, params) = langFilter(@["en", "fr"], "ar.") | ||
| 171 | + report("language filter builds placeholders", | ||
| 172 | + sql.count("?") == 2 and params.len == 2, sql.strip()) | ||
| 173 | + # Unclassified articles must survive the filter, or new items vanish for | ||
| 174 | + # a few minutes and then reappear. | ||
| 175 | + report("language filter keeps unclassified articles", | ||
| 176 | + "language = ''" in sql) | ||
| 177 | + report("an empty language list adds no filter", | ||
| 178 | + langFilter(@[], "ar.")[0].len == 0) | ||
| 179 | + | ||
| 180 | + # --- liked articles ----------------------------------------------------- | ||
| 181 | + block: | ||
| 182 | + let liked = g.listLikedArticles(bob) | ||
| 183 | + report("liked articles are listed newest first", liked.len == 2, $liked.len) | ||
| 184 | + report("liked articles carry article fields", | ||
| 185 | + liked[0].title.len > 0 and liked[0].feedTitle == "Feed A", | ||
| 186 | + liked[0].title) | ||
| 187 | + | ||
| 188 | + # --- reconciliation ----------------------------------------------------- | ||
| 189 | + # The destructive path: local rows the PDS no longer has must go, and | ||
| 190 | + # everyone else's must not. | ||
| 191 | + block: | ||
| 192 | + report("bob starts with 2 likes", g.listLikes(bob).len == 2) | ||
| 193 | + g.deleteOrphanedLikes(bob, @["at://bob/like/1"]) | ||
| 194 | + report("reconciling deletes likes the PDS dropped", | ||
| 195 | + g.listLikes(bob).len == 1, $g.listLikes(bob).len) | ||
| 196 | + report("reconciling leaves other users alone", | ||
| 197 | + g.listLikes(carol).len == 1, $g.listLikes(carol).len) | ||
| 198 | + | ||
| 199 | + g.deleteOrphanedAnnotations(alice, @["at://alice/ann/1"]) | ||
| 200 | + report("reconciling keeps annotations the PDS still has", | ||
| 201 | + g.listAnnotations(authorDid = alice).len == 1) | ||
| 202 | + g.deleteOrphanedAnnotations(alice, @[]) | ||
| 203 | + report("an empty active set clears that author's annotations", | ||
| 204 | + g.listAnnotations(authorDid = alice).len == 0) | ||
| 205 | + report("and leaves another author's annotations", | ||
| 206 | + g.listAnnotations(authorDid = bob).len == 1) | ||
| 207 | + | ||
| 208 | + echo "" | ||
| 209 | + if failures == 0: | ||
| 210 | + echo "RESULT: likes, annotations, follows and trending behave." | ||
| 211 | + else: | ||
| 212 | + echo &"RESULT: {failures} check(s) failed." | ||
| 213 | + quit 1 | ||
| 214 | + | ||
| 215 | +when isMainModule: | ||
| 216 | + main() | ||