zzstoatzz.io
search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
17 kB
468 lines
1#!/usr/bin/env -S uv run --script --quiet
2# /// script
3# requires-python = ">=3.12"
4# dependencies = [
5# "httpx",
6# "pydantic-settings",
7# "atproto",
8# ]
9# ///
10"""
11Backfill publication records via CAR walk.
12
13Sibling of backfill-subscriptions. Our `publications` table is populated ONLY
14when we directly witness a publication record (its create/update on the
15firehose, or a backfill). Indexing a publication's *documents* does NOT
16materialize the publication row. So any publication record that predates our
17consumption window and hasn't been edited since has its documents indexed but
18no publication row — and a subscription pointing at it finds no JOIN match.
19
20That's a ~25% gap (measured: 393 / 1593 distinct subscribed-to publications had
21no row), which silently understates the publisher and most-subscribed
22leaderboards. This walks the two publication collections directly and upserts
23the records ground-truth.
24
25 com.atproto.sync.listReposByCollection -> DIDs holding the collection
26 slingshot resolveMiniDoc -> DID -> PDS endpoint
27 com.atproto.sync.getRepo -> full repo as CAR
28 walk MST locally -> each publication record
29 UPSERT into publications -> idempotent on (uri)
30
31indexed_at is set to now() on every upsert so the frozen local read-replica's
32incremental sync (indexed_at >= last_sync) picks the rows up and rebuilds its
33publications_fts — same contract every Turso-mutating script honors.
34
35Usage:
36 TURSO_URL=libsql://... TURSO_TOKEN=... ./scripts/backfill-publications
37 ./scripts/backfill-publications --dry-run
38 ./scripts/backfill-publications --workers 16
39 ./scripts/backfill-publications --limit 50 # debug
40"""
41
42from __future__ import annotations
43
44import argparse
45import os
46import sys
47import time
48from concurrent.futures import ThreadPoolExecutor, as_completed
49from dataclasses import dataclass
50from typing import NamedTuple
51
52import httpx
53from atproto_core.car import CAR
54from atproto_core.cid import CID
55from pydantic_settings import BaseSettings, SettingsConfigDict
56
57SLINGSHOT = "https://slingshot.microcosm.blue"
58
59# corpus admission policy — mirror of backend/src/policy.zig + ingest/ingester.zig
60# isBridgyPds. Every write path (ingester, indexer, snapshot builder) drops
61# these; a backfill must enforce the same gates or it re-admits exactly what we
62# deliberately keep out. Registry of who/why/evidence: docs/exclusions.md.
63BANNED_DIDS = frozenset(
64 {
65 "did:plc:oql6ds5vnff4ugar6rruliwd", # drivepatents.com patent bot
66 "did:plc:2s32mlusc66sjb256aenynfc", # destinationcharged.com NHTSA recall mirror
67 }
68)
69
70
71def is_bridgy_pds(pds: str | None) -> bool:
72 """Host-match a PDS endpoint against brid.gy (mirrors backend isBridgyPds)."""
73 if not pds:
74 return False
75 host = pds.split("://", 1)[-1].split("/", 1)[0].split(":", 1)[0].lower()
76 return host == "brid.gy" or host.endswith(".brid.gy")
77
78
79# both publication lexicons. leaflet stores the host in `base_path`; standard
80# stores a full `url` we strip to its host (matches backend ingester.zig).
81COLLECTIONS = ["pub.leaflet.publication", "site.standard.publication"]
82
83
84class PubRow(NamedTuple):
85 uri: str
86 did: str
87 rkey: str
88 name: str
89 description: str
90 base_path: str
91
92# Union DIDs across multiple relays — lightrail's index lags the Bluesky
93# reference relay for niche collections.
94RELAY_ENDPOINTS = [
95 "https://relay1.us-east.bsky.network",
96 "https://lightrail.microcosm.blue",
97 "https://relay.waow.tech",
98]
99
100
101def strip_url_scheme(url: str | None) -> str | None:
102 """site.standard.publication.url is a full URL (e.g. https://devlog.pckt.blog);
103 we keep only the host, mirroring backend stripUrlScheme."""
104 if not url:
105 return None
106 s = url
107 for scheme in ("https://", "http://"):
108 if s.startswith(scheme):
109 s = s[len(scheme):]
110 break
111 host = s.split("/", 1)[0]
112 return host or None
113
114
115class Settings(BaseSettings):
116 model_config = SettingsConfigDict(
117 env_file=os.environ.get("ENV_FILE", ".env"), extra="ignore"
118 )
119 turso_url: str
120 turso_token: str
121
122 @property
123 def turso_host(self) -> str:
124 url = self.turso_url
125 if url.startswith("libsql://"):
126 url = url[len("libsql://"):]
127 return url
128
129
130# ---- Turso ----------------------------------------------------------------
131
132
133def turso_pipeline(settings: Settings, requests: list[dict]) -> list[dict]:
134 last_err: Exception | None = None
135 for attempt in range(3):
136 try:
137 resp = httpx.post(
138 f"https://{settings.turso_host}/v2/pipeline",
139 headers={
140 "Authorization": f"Bearer {settings.turso_token}",
141 "Content-Type": "application/json",
142 },
143 json={"requests": [*requests, {"type": "close"}]},
144 timeout=180,
145 )
146 resp.raise_for_status()
147 return resp.json()["results"]
148 except (httpx.ReadTimeout, httpx.ConnectError, httpx.RemoteProtocolError) as e:
149 last_err = e
150 print(
151 f" turso pipeline attempt {attempt + 1} failed: {type(e).__name__}; retrying",
152 file=sys.stderr,
153 )
154 time.sleep(2 * (attempt + 1))
155 assert last_err is not None
156 raise last_err
157
158
159def turso_upsert_publications(settings: Settings, rows: list[PubRow]) -> int:
160 BATCH = 50
161 written = 0
162 for i in range(0, len(rows), BATCH):
163 chunk = rows[i: i + BATCH]
164 reqs = [
165 {
166 "type": "execute",
167 "stmt": {
168 "sql": (
169 "INSERT INTO publications "
170 "(uri, did, rkey, name, description, base_path, indexed_at) "
171 "VALUES (?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%S','now')) "
172 "ON CONFLICT(uri) DO UPDATE SET "
173 " did = excluded.did, "
174 " rkey = excluded.rkey, "
175 " name = excluded.name, "
176 " description = excluded.description, "
177 " base_path = excluded.base_path, "
178 " indexed_at = strftime('%Y-%m-%dT%H:%M:%S','now')"
179 ),
180 "args": [
181 {"type": "text", "value": row.uri},
182 {"type": "text", "value": row.did},
183 {"type": "text", "value": row.rkey},
184 {"type": "text", "value": row.name},
185 {"type": "text", "value": row.description},
186 {"type": "text", "value": row.base_path},
187 ],
188 },
189 }
190 for row in chunk
191 ]
192 results = turso_pipeline(settings, reqs)
193 for r in results:
194 if r["type"] == "error":
195 print(f" upsert error: {r['error']}", file=sys.stderr)
196 continue
197 response = r.get("response") or {}
198 if response.get("type") != "execute":
199 continue
200 written += int(response.get("result", {}).get("affected_row_count", 0))
201 return written
202
203
204# ---- Enumeration + resolution -------------------------------------------
205
206
207def _enumerate_from_relay(client: httpx.Client, base: str, collection: str) -> list[str]:
208 dids: list[str] = []
209 cursor: str | None = None
210 while True:
211 params: dict = {"collection": collection, "limit": 1000}
212 if cursor:
213 params["cursor"] = cursor
214 try:
215 r = client.get(
216 f"{base}/xrpc/com.atproto.sync.listReposByCollection",
217 params=params,
218 timeout=30,
219 )
220 r.raise_for_status()
221 except Exception as e:
222 print(f" {base} ({collection}): relay error {type(e).__name__}", file=sys.stderr)
223 return dids
224 body = r.json()
225 for repo in body.get("repos", []):
226 dids.append(repo["did"])
227 cursor = body.get("cursor")
228 if not cursor:
229 return dids
230
231
232def enumerate_owner_dids() -> list[str]:
233 all_dids: set[str] = set()
234 with httpx.Client() as client:
235 for collection in COLLECTIONS:
236 for base in RELAY_ENDPOINTS:
237 from_relay = _enumerate_from_relay(client, base, collection)
238 print(f" {base} ({collection}): {len(from_relay)} DIDs")
239 all_dids.update(from_relay)
240 return sorted(all_dids)
241
242
243def _did_from_pub_uri(uri: str) -> str | None:
244 # at://<did>/<collection>/<rkey>
245 if not uri.startswith("at://"):
246 return None
247 rest = uri[len("at://"):]
248 did = rest.split("/", 1)[0]
249 return did or None
250
251
252def owner_dids_with_unindexed_subscribed_pubs(settings: Settings) -> list[str]:
253 """The surgical target set: owners of publications that someone subscribes to
254 but which have no row in our publications table. Directly closes the measured
255 JOIN gap with a few hundred CAR fetches instead of walking the whole network."""
256 results = turso_pipeline(
257 settings,
258 [
259 {
260 "type": "execute",
261 "stmt": {
262 "sql": (
263 "SELECT DISTINCT s.publication_uri "
264 "FROM subscriptions s "
265 "LEFT JOIN publications p ON p.uri = s.publication_uri "
266 "WHERE p.uri IS NULL"
267 )
268 },
269 }
270 ],
271 )
272 dids: set[str] = set()
273 for r in results:
274 if r.get("type") != "ok":
275 continue
276 response = r.get("response") or {}
277 if response.get("type") != "execute":
278 continue
279 for row in response.get("result", {}).get("rows", []):
280 if not row:
281 continue
282 uri = row[0].get("value")
283 did = _did_from_pub_uri(uri) if isinstance(uri, str) else None
284 if did:
285 dids.add(did)
286 return sorted(dids)
287
288
289def resolve_pds(client: httpx.Client, did: str) -> str | None:
290 try:
291 r = client.get(
292 f"{SLINGSHOT}/xrpc/blue.microcosm.identity.resolveMiniDoc",
293 params={"identifier": did},
294 timeout=15,
295 )
296 r.raise_for_status()
297 return r.json().get("pds")
298 except Exception:
299 return None
300
301
302# ---- CAR walk ------------------------------------------------------------
303
304
305def _walk_mst(car: CAR, prefixes: list[bytes]) -> list[tuple[str, str, dict]]:
306 """Return [(collection, rkey, decoded_value_dict), ...] for every entry whose
307 key starts with one of the f"{collection}/" prefixes."""
308 blocks = car.blocks
309 root = blocks[car.root]
310 data_cid = CID.decode(root["data"])
311
312 out: list[tuple[str, str, dict]] = []
313 stack = [data_cid]
314 while stack:
315 node = blocks[stack.pop()]
316 if (left := node.get("l")) is not None:
317 stack.append(CID.decode(left))
318 prev_key = b""
319 for entry in node["e"]:
320 key = prev_key[: entry["p"]] + entry["k"]
321 prev_key = key
322 for prefix in prefixes:
323 if key.startswith(prefix):
324 value_cid = CID.decode(entry["v"])
325 value = blocks[value_cid]
326 collection = prefix[:-1].decode() # strip trailing "/"
327 rkey = key[len(prefix):].decode("utf-8", errors="replace")
328 out.append((collection, rkey, value))
329 break
330 if (t := entry.get("t")) is not None:
331 stack.append(CID.decode(t))
332 return out
333
334
335# ---- Fetch one DID's publications ----------------------------------------
336
337
338@dataclass
339class FetchResult:
340 did: str
341 rows: list[PubRow]
342 status: str # "ok", "no-pds", "fetch-err:<type>", "parse-err"
343
344
345def fetch_publications_for_did(client: httpx.Client, did: str) -> FetchResult:
346 # policy gate 1: explicitly banned bulk-archive repos (docs/exclusions.md).
347 if did in BANNED_DIDS:
348 return FetchResult(did, [], "banned")
349 pds = resolve_pds(client, did)
350 if not pds:
351 return FetchResult(did, [], "no-pds")
352 # policy gate 2: bridgy-hosted PDS — same drop as every other write path.
353 if is_bridgy_pds(pds):
354 return FetchResult(did, [], "bridgy")
355 try:
356 r = client.get(
357 f"{pds}/xrpc/com.atproto.sync.getRepo",
358 params={"did": did},
359 timeout=60,
360 )
361 r.raise_for_status()
362 car = CAR.from_bytes(r.content)
363 except httpx.HTTPStatusError as e:
364 return FetchResult(did, [], f"fetch-err:{e.response.status_code}")
365 except Exception as e:
366 return FetchResult(did, [], f"fetch-err:{type(e).__name__}")
367 try:
368 prefixes = [f"{c}/".encode() for c in COLLECTIONS]
369 entries = _walk_mst(car, prefixes)
370 except Exception:
371 return FetchResult(did, [], "parse-err")
372
373 rows: list[PubRow] = []
374 for collection, rkey, value in entries:
375 if not isinstance(value, dict):
376 continue
377 name = value.get("name")
378 if not isinstance(name, str) or not name:
379 continue # name is required (matches backend processPublication)
380 description = value.get("description")
381 description = description if isinstance(description, str) else ""
382 # leaflet: base_path; standard: host of url
383 base_path = value.get("base_path")
384 if not isinstance(base_path, str) or not base_path:
385 base_path = strip_url_scheme(value.get("url")) or ""
386 # skip dev/staging .test domains (matches backend)
387 if base_path.endswith(".test"):
388 continue
389 rows.append(
390 PubRow(
391 uri=f"at://{did}/{collection}/{rkey}",
392 did=did,
393 rkey=rkey,
394 name=name,
395 description=description,
396 base_path=base_path,
397 )
398 )
399 return FetchResult(did, rows, "ok")
400
401
402# ---- Main ---------------------------------------------------------------
403
404
405def main() -> int:
406 parser = argparse.ArgumentParser(description="Backfill publication records via CAR walk")
407 parser.add_argument("--dry-run", action="store_true", help="Don't write to Turso")
408 parser.add_argument("--workers", type=int, default=8, help="Concurrent PDS fetches")
409 parser.add_argument("--limit", type=int, default=None, help="Cap DIDs (debug)")
410 parser.add_argument(
411 "--from-subscriptions",
412 action="store_true",
413 help="Surgical mode: only walk owners of subscribed-but-unindexed pubs "
414 "(closes the measured JOIN gap with a few hundred fetches instead of the "
415 "whole network).",
416 )
417 args = parser.parse_args()
418
419 try:
420 settings = Settings() # type: ignore
421 except Exception as e:
422 print(f"error loading settings: {e}", file=sys.stderr)
423 print("required env vars: TURSO_URL, TURSO_TOKEN", file=sys.stderr)
424 return 1
425
426 if args.from_subscriptions:
427 print("=== surgical: owners of subscribed-but-unindexed publications ===")
428 dids = owner_dids_with_unindexed_subscribed_pubs(settings)
429 else:
430 print(f"=== {' + '.join(COLLECTIONS)} ===")
431 print("enumerating owner DIDs...")
432 dids = enumerate_owner_dids()
433 if args.limit:
434 dids = dids[: args.limit]
435 print(f" {len(dids)} DIDs")
436 if not dids:
437 return 0
438
439 started = time.time()
440 done = 0
441 statuses: dict[str, int] = {}
442 rows_acc: list[PubRow] = []
443
444 # follow_redirects: some PDSes 301 getRepo to a canonical host.
445 with httpx.Client(follow_redirects=True) as client, ThreadPoolExecutor(max_workers=args.workers) as pool:
446 futures = {pool.submit(fetch_publications_for_did, client, did): did for did in dids}
447 for fut in as_completed(futures):
448 res = fut.result()
449 statuses[res.status] = statuses.get(res.status, 0) + 1
450 rows_acc.extend(res.rows)
451 done += 1
452 if done % 25 == 0 or done == len(dids):
453 elapsed = time.time() - started
454 rate = done / elapsed if elapsed > 0 else 0
455 print(f" [{done}/{len(dids)}] {rate:.1f} dids/s, {len(rows_acc)} records, statuses={statuses}")
456
457 print(f"collected {len(rows_acc)} records from {statuses.get('ok', 0)} DIDs")
458 if args.dry_run:
459 print("dry-run: skipping writes")
460 return 0
461 print("upserting...")
462 written = turso_upsert_publications(settings, rows_acc)
463 print(f" wrote {written} rows")
464 return 0
465
466
467if __name__ == "__main__":
468 sys.exit(main())