zzstoatzz.io
search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
12 kB
349 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 site.standard.graph.recommend records via CAR walk.
12
13Pattern (cribbed from prefect-pack/collection_creators):
14 lightrail listReposByCollection -> enumerate DIDs holding the collection
15 slingshot resolveMiniDoc -> map DID -> PDS endpoint
16 com.atproto.sync.getRepo -> fetch full repo as CAR
17 walk MST locally -> read each key+value with prefix
18 UPSERT into recommends -> idempotent on (uri)
19
20Why not tap? Tap is for streaming new records going forward. For a one-shot
21backfill, a CAR walk is one HTTP call per DID and gives ground truth.
22
23Usage:
24 TURSO_URL=libsql://... TURSO_TOKEN=... ./scripts/backfill-recommends
25 ./scripts/backfill-recommends --dry-run
26"""
27
28from __future__ import annotations
29
30import argparse
31import os
32import sys
33import time
34from concurrent.futures import ThreadPoolExecutor, as_completed
35from dataclasses import dataclass
36
37import httpx
38from atproto_core.car import CAR
39from atproto_core.cid import CID
40from pydantic_settings import BaseSettings, SettingsConfigDict
41
42SLINGSHOT = "https://slingshot.microcosm.blue"
43
44# Recommend-shaped lexicons in the wild, surveyed via constellation 2026-05-21.
45# Each entry: (collection NSID, field in record holding the document at-uri).
46DEFAULT_LEXICONS = [
47 ("site.standard.graph.recommend", "document"),
48 ("pub.leaflet.interactions.recommend", "subject"),
49]
50
51# Union DIDs across multiple relays — lightrail's index lags the Bluesky
52# reference relay by ~15% for niche collections (216 vs 254 at last check).
53RELAY_ENDPOINTS = [
54 "https://relay1.us-east.bsky.network",
55 "https://lightrail.microcosm.blue",
56 "https://relay.waow.tech",
57]
58
59
60class Settings(BaseSettings):
61 model_config = SettingsConfigDict(
62 env_file=os.environ.get("ENV_FILE", ".env"), extra="ignore"
63 )
64 turso_url: str
65 turso_token: str
66
67 @property
68 def turso_host(self) -> str:
69 url = self.turso_url
70 if url.startswith("libsql://"):
71 url = url[len("libsql://") :]
72 return url
73
74
75# ---- Turso ----------------------------------------------------------------
76
77
78def turso_pipeline(settings: Settings, requests: list[dict]) -> list[dict]:
79 last_err: Exception | None = None
80 for attempt in range(3):
81 try:
82 resp = httpx.post(
83 f"https://{settings.turso_host}/v2/pipeline",
84 headers={
85 "Authorization": f"Bearer {settings.turso_token}",
86 "Content-Type": "application/json",
87 },
88 json={"requests": [*requests, {"type": "close"}]},
89 timeout=180,
90 )
91 resp.raise_for_status()
92 return resp.json()["results"]
93 except (httpx.ReadTimeout, httpx.ConnectError, httpx.RemoteProtocolError) as e:
94 last_err = e
95 print(
96 f" turso pipeline attempt {attempt + 1} failed: {type(e).__name__}; retrying",
97 file=sys.stderr,
98 )
99 time.sleep(2 * (attempt + 1))
100 assert last_err is not None
101 raise last_err
102
103
104def turso_upsert_recommends(settings: Settings, rows: list[tuple[str, str, str, str, str]]) -> int:
105 """rows: (uri, did, rkey, document_uri, created_at)."""
106 BATCH = 50
107 written = 0
108 for i in range(0, len(rows), BATCH):
109 chunk = rows[i : i + BATCH]
110 reqs = [
111 {
112 "type": "execute",
113 "stmt": {
114 "sql": (
115 "INSERT INTO recommends (uri, did, rkey, document_uri, created_at, indexed_at) "
116 "VALUES (?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%S','now')) "
117 "ON CONFLICT(uri) DO UPDATE SET "
118 " did = excluded.did, "
119 " rkey = excluded.rkey, "
120 " document_uri = excluded.document_uri, "
121 " created_at = excluded.created_at, "
122 " indexed_at = strftime('%Y-%m-%dT%H:%M:%S','now')"
123 ),
124 "args": [
125 {"type": "text", "value": uri},
126 {"type": "text", "value": did},
127 {"type": "text", "value": rkey},
128 {"type": "text", "value": doc_uri},
129 {"type": "text", "value": created_at},
130 ],
131 },
132 }
133 for (uri, did, rkey, doc_uri, created_at) in chunk
134 ]
135 results = turso_pipeline(settings, reqs)
136 for r in results:
137 if r["type"] == "error":
138 print(f" upsert error: {r['error']}", file=sys.stderr)
139 continue
140 response = r.get("response") or {}
141 if response.get("type") != "execute":
142 continue
143 written += int(response.get("result", {}).get("affected_row_count", 0))
144 return written
145
146
147# ---- Enumeration + resolution -------------------------------------------
148
149
150def _enumerate_from_relay(client: httpx.Client, base: str, collection: str) -> list[str]:
151 dids: list[str] = []
152 cursor: str | None = None
153 while True:
154 params: dict = {"collection": collection, "limit": 1000}
155 if cursor:
156 params["cursor"] = cursor
157 try:
158 r = client.get(
159 f"{base}/xrpc/com.atproto.sync.listReposByCollection",
160 params=params,
161 timeout=30,
162 )
163 r.raise_for_status()
164 except Exception as e:
165 print(f" {base}: relay error {type(e).__name__}", file=sys.stderr)
166 return dids
167 body = r.json()
168 for repo in body.get("repos", []):
169 dids.append(repo["did"])
170 cursor = body.get("cursor")
171 if not cursor:
172 return dids
173
174
175def enumerate_creator_dids(collection: str) -> list[str]:
176 all_dids: set[str] = set()
177 with httpx.Client() as client:
178 for base in RELAY_ENDPOINTS:
179 from_relay = _enumerate_from_relay(client, base, collection)
180 print(f" {base}: {len(from_relay)} DIDs")
181 all_dids.update(from_relay)
182 return sorted(all_dids)
183
184
185def resolve_pds(client: httpx.Client, did: str) -> str | None:
186 try:
187 r = client.get(
188 f"{SLINGSHOT}/xrpc/blue.microcosm.identity.resolveMiniDoc",
189 params={"identifier": did},
190 timeout=15,
191 )
192 r.raise_for_status()
193 return r.json().get("pds")
194 except Exception:
195 return None
196
197
198# ---- CAR walk ------------------------------------------------------------
199
200
201def _walk_mst(car: CAR, collection: str) -> list[tuple[str, dict]]:
202 """Return [(rkey, decoded_value_dict), ...] for every entry whose key
203 starts with f"{collection}/".
204
205 MST nodes are nested via CIDs in `l` (left subtree) and `e[i].t` (right
206 subtree after entry i). Keys are byte-encoded with prefix compression
207 (`p` = chars shared with previous key, `k` = unique suffix).
208 """
209 blocks = car.blocks
210 root = blocks[car.root]
211 data_cid = CID.decode(root["data"])
212
213 prefix_bytes = f"{collection}/".encode()
214 out: list[tuple[str, dict]] = []
215 stack = [data_cid]
216 while stack:
217 node = blocks[stack.pop()]
218 if (left := node.get("l")) is not None:
219 stack.append(CID.decode(left))
220 prev_key = b""
221 for entry in node["e"]:
222 key = prev_key[: entry["p"]] + entry["k"]
223 prev_key = key
224 if key.startswith(prefix_bytes):
225 value_cid = CID.decode(entry["v"])
226 value = blocks[value_cid]
227 rkey = key[len(prefix_bytes) :].decode("utf-8", errors="replace")
228 out.append((rkey, value))
229 if (t := entry.get("t")) is not None:
230 stack.append(CID.decode(t))
231 return out
232
233
234# ---- Fetch one DID's recommends ------------------------------------------
235
236
237@dataclass
238class FetchResult:
239 did: str
240 rows: list[tuple[str, str, str, str, str]]
241 status: str # "ok", "unresolved", "no-pds", "fetch-err:<type>", "parse-err"
242
243
244def fetch_recommends_for_did(client: httpx.Client, did: str, collection: str, doc_field: str) -> FetchResult:
245 pds = resolve_pds(client, did)
246 if not pds:
247 return FetchResult(did, [], "no-pds")
248 try:
249 r = client.get(
250 f"{pds}/xrpc/com.atproto.sync.getRepo",
251 params={"did": did},
252 timeout=60,
253 )
254 r.raise_for_status()
255 car = CAR.from_bytes(r.content)
256 except httpx.HTTPStatusError as e:
257 return FetchResult(did, [], f"fetch-err:{e.response.status_code}")
258 except Exception as e:
259 return FetchResult(did, [], f"fetch-err:{type(e).__name__}")
260 try:
261 entries = _walk_mst(car, collection)
262 except Exception:
263 return FetchResult(did, [], "parse-err")
264
265 rows: list[tuple[str, str, str, str, str]] = []
266 for rkey, value in entries:
267 uri = f"at://{did}/{collection}/{rkey}"
268 document_uri = value.get(doc_field) if isinstance(value, dict) else None
269 created_at = value.get("createdAt") if isinstance(value, dict) else None
270 if not isinstance(document_uri, str):
271 continue # malformed; skip
272 rows.append((uri, did, rkey, document_uri, str(created_at) if created_at else ""))
273 return FetchResult(did, rows, "ok")
274
275
276# ---- Main ---------------------------------------------------------------
277
278
279def run_one_lexicon(settings: Settings, collection: str, doc_field: str, args) -> int:
280 print(f"\n=== {collection} (.{doc_field}) ===")
281 print("enumerating creator DIDs...")
282 dids = enumerate_creator_dids(collection)
283 if args.limit:
284 dids = dids[: args.limit]
285 print(f" {len(dids)} DIDs")
286 if not dids:
287 return 0
288
289 started = time.time()
290 done = 0
291 statuses: dict[str, int] = {}
292 rows_acc: list[tuple[str, str, str, str, str]] = []
293
294 with httpx.Client() as client, ThreadPoolExecutor(max_workers=args.workers) as pool:
295 futures = {
296 pool.submit(fetch_recommends_for_did, client, did, collection, doc_field): did
297 for did in dids
298 }
299 for fut in as_completed(futures):
300 res = fut.result()
301 statuses[res.status] = statuses.get(res.status, 0) + 1
302 rows_acc.extend(res.rows)
303 done += 1
304 if done % 25 == 0 or done == len(dids):
305 elapsed = time.time() - started
306 rate = done / elapsed if elapsed > 0 else 0
307 print(f" [{done}/{len(dids)}] {rate:.1f} dids/s, {len(rows_acc)} records, statuses={statuses}")
308
309 print(f"collected {len(rows_acc)} records from {statuses.get('ok', 0)} DIDs")
310 if args.dry_run:
311 print("dry-run: skipping writes")
312 return 0
313 print("upserting...")
314 written = turso_upsert_recommends(settings, rows_acc)
315 print(f" wrote {written} rows")
316 return written
317
318
319def main() -> int:
320 parser = argparse.ArgumentParser(description="Backfill recommend-shaped records via CAR walk")
321 parser.add_argument("--dry-run", action="store_true", help="Don't write to Turso")
322 parser.add_argument("--workers", type=int, default=8, help="Concurrent PDS fetches")
323 parser.add_argument("--limit", type=int, default=None, help="Cap DIDs per lexicon (debug)")
324 parser.add_argument("--collection", default=None, help="Override: only process this NSID")
325 parser.add_argument("--doc-field", default=None, help="Field holding the document at-uri (paired with --collection)")
326 args = parser.parse_args()
327
328 try:
329 settings = Settings() # type: ignore
330 except Exception as e:
331 print(f"error loading settings: {e}", file=sys.stderr)
332 print("required env vars: TURSO_URL, TURSO_TOKEN", file=sys.stderr)
333 return 1
334
335 if args.collection:
336 field = args.doc_field or "document"
337 lexicons = [(args.collection, field)]
338 else:
339 lexicons = DEFAULT_LEXICONS
340
341 total = 0
342 for collection, doc_field in lexicons:
343 total += run_one_lexicon(settings, collection, doc_field, args)
344 print(f"\ngrand total: {total} rows written across {len(lexicons)} lexicon(s)")
345 return 0
346
347
348if __name__ == "__main__":
349 sys.exit(main())