zzstoatzz.io
search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
35 kB
864 lines
1#!/usr/bin/env -S uv run --script --quiet
2# /// script
3# requires-python = ">=3.12"
4# dependencies = ["httpx"]
5# ///
6"""Read-only, resumable source-to-index corpus audit.
7
8Inventories every repo advertising ``site.standard.document`` at the relay,
9walks authoritative PDS records for policy-eligible repos, and compares them
10with Turso, the adopted serving snapshot, and turbopuffer. The only writes are
11to a local SQLite checkpoint (default: /tmp/pub-search-corpus-audit.sqlite).
12"""
13
14from __future__ import annotations
15
16import argparse
17import asyncio
18import base64
19import datetime as dt
20import hashlib
21import json
22import os
23from pathlib import Path
24import sqlite3
25import subprocess
26import sys
27import time
28from typing import Any
29from urllib.parse import urlparse
30
31import httpx
32
33
34COLLECTION = "site.standard.document"
35RELAY = "https://relay1.us-east.bsky.network"
36TPUF_NAMESPACE = "leaflet-search"
37_BACKEND_MACHINE: str | None = None
38PLAIN_BLOCKS = {
39 "pub.leaflet.blocks.text",
40 "pub.leaflet.blocks.header",
41 "pub.leaflet.blocks.blockquote",
42 "pub.leaflet.blocks.code",
43}
44
45
46def log(message: str) -> None:
47 print(f"[{dt.datetime.now(dt.timezone.utc).isoformat(timespec='seconds')}] {message}", flush=True)
48
49
50def connect(path: Path) -> sqlite3.Connection:
51 db = sqlite3.connect(path)
52 db.execute("PRAGMA journal_mode=WAL")
53 db.execute("PRAGMA synchronous=NORMAL")
54 db.executescript(
55 """
56 CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
57 CREATE TABLE IF NOT EXISTS repos (
58 did TEXT PRIMARY KEY,
59 in_relay INTEGER NOT NULL DEFAULT 1,
60 pds TEXT,
61 resolution_status TEXT,
62 policy TEXT,
63 audit_status TEXT,
64 source_count INTEGER,
65 error TEXT
66 );
67 CREATE TABLE IF NOT EXISTS source_documents (
68 uri TEXT PRIMARY KEY,
69 did TEXT NOT NULL,
70 rkey TEXT,
71 cid TEXT,
72 title TEXT,
73 publication_uri TEXT,
74 path TEXT,
75 published_at TEXT,
76 eligibility TEXT NOT NULL,
77 content_fingerprint TEXT
78 );
79 CREATE INDEX IF NOT EXISTS source_documents_did ON source_documents(did);
80 CREATE INDEX IF NOT EXISTS source_documents_fingerprint
81 ON source_documents(did, content_fingerprint);
82 CREATE TABLE IF NOT EXISTS turso_documents (
83 uri TEXT PRIMARY KEY,
84 source_rowid INTEGER,
85 did TEXT NOT NULL,
86 rkey TEXT,
87 title TEXT,
88 content_length INTEGER,
89 created_at TEXT,
90 publication_uri TEXT,
91 path TEXT,
92 platform TEXT,
93 source_collection TEXT,
94 indexed_at TEXT,
95 embedded_at TEXT,
96 verified_at TEXT,
97 source_cid TEXT,
98 is_bridgyfed INTEGER,
99 url_dead INTEGER
100 );
101 CREATE INDEX IF NOT EXISTS turso_documents_did ON turso_documents(did);
102 CREATE TABLE IF NOT EXISTS snapshot_documents (
103 uri TEXT PRIMARY KEY,
104 source_rowid INTEGER,
105 did TEXT NOT NULL,
106 source_collection TEXT,
107 indexed_at TEXT
108 );
109 CREATE TABLE IF NOT EXISTS vectors (
110 id TEXT PRIMARY KEY,
111 uri TEXT,
112 did TEXT
113 );
114 """
115 )
116 # Checkpoints created before CID-aware reconciliation need the additive
117 # column too; CREATE TABLE IF NOT EXISTS does not evolve an existing DB.
118 columns = {row[1] for row in db.execute("PRAGMA table_info(turso_documents)")}
119 if "source_cid" not in columns:
120 db.execute("ALTER TABLE turso_documents ADD COLUMN source_cid TEXT")
121 db.commit()
122 if "source_rowid" not in columns:
123 db.execute("ALTER TABLE turso_documents ADD COLUMN source_rowid INTEGER")
124 db.commit()
125 snapshot_columns = {row[1] for row in db.execute("PRAGMA table_info(snapshot_documents)")}
126 if "source_rowid" not in snapshot_columns:
127 db.execute("ALTER TABLE snapshot_documents ADD COLUMN source_rowid INTEGER")
128 db.commit()
129 return db
130
131
132def env_file_value(name: str) -> str | None:
133 value = os.environ.get(name)
134 if value:
135 return value
136 path = Path(".env")
137 if not path.exists():
138 return None
139 for line in path.read_text().splitlines():
140 if line.startswith(name + "="):
141 return line.split("=", 1)[1].strip().strip("'\"")
142 return None
143
144
145def backend_machine() -> str:
146 global _BACKEND_MACHINE
147 if _BACKEND_MACHINE:
148 return _BACKEND_MACHINE
149 proc = subprocess.run(
150 ["fly", "machine", "list", "-a", "leaflet-search-backend", "--json"],
151 capture_output=True,
152 text=True,
153 timeout=30,
154 check=True,
155 )
156 machines = json.loads(proc.stdout)
157 for machine in machines:
158 metadata = machine.get("config", {}).get("metadata", {})
159 if machine.get("state") == "started" and metadata.get("fly_process_group") == "app":
160 _BACKEND_MACHINE = machine["id"]
161 return _BACKEND_MACHINE
162 raise RuntimeError("no started backend app machine found")
163
164
165async def request_json(
166 client: httpx.AsyncClient,
167 url: str,
168 *,
169 params: dict[str, str | int] | None = None,
170 attempts: int = 4,
171) -> tuple[int | None, Any | None, str | None]:
172 error = None
173 for attempt in range(attempts):
174 try:
175 response = await client.get(url, params=params, timeout=30)
176 if response.status_code == 200:
177 return 200, response.json(), None
178 if response.status_code in {400, 404}:
179 return response.status_code, None, response.text[:300]
180 error = f"HTTP {response.status_code}: {response.text[:200]}"
181 except Exception as exc: # network inventory: retain error, retry
182 error = repr(exc)
183 await asyncio.sleep(0.5 * 2**attempt)
184 return None, None, error
185
186
187async def inventory_relay(db: sqlite3.Connection, client: httpx.AsyncClient) -> None:
188 if db.execute("SELECT 1 FROM meta WHERE key='relay_inventory_at'").fetchone():
189 log("relay inventory already checkpointed")
190 return
191 # Relay pagination cursors are not durable in older checkpoints. If a
192 # prior run stopped mid-inventory, restart this inexpensive phase rather
193 # than accepting a partial repo universe.
194 if db.execute("SELECT COUNT(*) FROM repos WHERE in_relay=1").fetchone()[0]:
195 db.execute("DELETE FROM repos WHERE in_relay=1")
196 db.commit()
197 cursor = None
198 total = 0
199 while True:
200 params: dict[str, str | int] = {"collection": COLLECTION, "limit": 2000}
201 if cursor:
202 params["cursor"] = cursor
203 status, body, error = await request_json(
204 client,
205 f"{RELAY}/xrpc/com.atproto.sync.listReposByCollection",
206 params=params,
207 )
208 if status != 200:
209 raise RuntimeError(f"relay inventory failed: {error}")
210 repos = body.get("repos") or body.get("dids") or []
211 dids = [item if isinstance(item, str) else item["did"] for item in repos]
212 db.executemany("INSERT OR IGNORE INTO repos(did,in_relay) VALUES (?,1)", ((did,) for did in dids))
213 db.commit()
214 total += len(dids)
215 cursor = body.get("cursor")
216 log(f"relay inventory: {total} repos")
217 if not cursor:
218 break
219 db.execute(
220 "INSERT OR REPLACE INTO meta(key,value) VALUES('relay_inventory_at',?)",
221 (dt.datetime.now(dt.timezone.utc).isoformat(),),
222 )
223 db.commit()
224
225
226def did_document_url(did: str) -> str | None:
227 if did.startswith("did:plc:"):
228 return f"https://plc.directory/{did}"
229 if did.startswith("did:web:"):
230 parts = did[len("did:web:") :].split(":")
231 host = parts[0]
232 path = "/".join(parts[1:])
233 return f"https://{host}/{path + '/' if path else '.well-known/'}did.json"
234 return None
235
236
237def pds_from_doc(body: dict[str, Any]) -> str | None:
238 for service in body.get("service", []):
239 if service.get("type") == "AtprotoPersonalDataServer" or str(service.get("id", "")).endswith("#atproto_pds"):
240 endpoint = service.get("serviceEndpoint")
241 if isinstance(endpoint, str):
242 return endpoint.rstrip("/")
243 return None
244
245
246async def resolve_repos(
247 db: sqlite3.Connection,
248 client: httpx.AsyncClient,
249 banned: set[str],
250 workers: int,
251) -> None:
252 pending = [row[0] for row in db.execute("SELECT did FROM repos WHERE resolution_status IS NULL")]
253 log(f"resolving {len(pending)} repository identities")
254 queue: asyncio.Queue[str] = asyncio.Queue()
255 for did in pending:
256 queue.put_nowait(did)
257 lock = asyncio.Lock()
258 completed = 0
259
260 async def worker() -> None:
261 nonlocal completed
262 while not queue.empty():
263 try:
264 did = queue.get_nowait()
265 except asyncio.QueueEmpty:
266 return
267 url = did_document_url(did)
268 if not url:
269 result = (did, None, "unsupported_did", "unresolved", "unsupported DID method")
270 else:
271 status, body, error = await request_json(client, url)
272 pds = pds_from_doc(body) if status == 200 and isinstance(body, dict) else None
273 resolution = "resolved" if pds else ("gone" if status in {400, 404} else "unresolved")
274 host = (urlparse(pds).hostname or "") if pds else ""
275 policy = "banned" if did in banned else ("bridgy" if host.endswith("brid.gy") or "bridgy" in host else "eligible")
276 result = (did, pds, resolution, policy, error)
277 async with lock:
278 db.execute(
279 "UPDATE repos SET pds=?, resolution_status=?, policy=?, error=? WHERE did=?",
280 (result[1], result[2], result[3], result[4], result[0]),
281 )
282 completed += 1
283 if completed % 250 == 0:
284 db.commit()
285 log(f"identity resolution: {completed}/{len(pending)}")
286 queue.task_done()
287
288 await asyncio.gather(*(worker() for _ in range(workers)))
289 db.commit()
290
291
292def list_item_text(item: Any) -> list[str]:
293 if not isinstance(item, dict):
294 return []
295 parts: list[str] = []
296 content = item.get("content")
297 if isinstance(content, dict) and isinstance(content.get("plaintext"), str) and content["plaintext"]:
298 parts.append(content["plaintext"])
299 children = item.get("children")
300 if isinstance(children, list):
301 for child in children:
302 parts.extend(list_item_text(child))
303 return parts
304
305
306def page_text(pages: Any) -> list[str]:
307 if not isinstance(pages, list):
308 return []
309 parts: list[str] = []
310 for page in pages:
311 blocks = page.get("blocks") if isinstance(page, dict) else None
312 if not isinstance(blocks, list):
313 continue
314 for wrapper in blocks:
315 block = wrapper.get("block") if isinstance(wrapper, dict) else None
316 if not isinstance(block, dict):
317 continue
318 block_type = block.get("$type")
319 if block_type in PLAIN_BLOCKS and isinstance(block.get("plaintext"), str) and block["plaintext"]:
320 parts.append(block["plaintext"])
321 elif block_type == "pub.leaflet.blocks.button" and isinstance(block.get("text"), str) and block["text"]:
322 parts.append(block["text"])
323 elif block_type == "pub.leaflet.blocks.unorderedList":
324 for item in block.get("children", []) if isinstance(block.get("children"), list) else []:
325 parts.extend(list_item_text(item))
326 return parts
327
328
329def extract_record(value: Any) -> tuple[str, str | None]:
330 if not isinstance(value, dict) or not isinstance(value.get("title"), str):
331 return "missing_title", None
332 if isinstance(value.get("textContent"), str):
333 content = value["textContent"]
334 elif isinstance(value.get("content"), str):
335 content = value["content"]
336 else:
337 parts: list[str] = []
338 if isinstance(value.get("description"), str) and value["description"]:
339 parts.append(value["description"])
340 pages = value.get("pages")
341 if pages is None and isinstance(value.get("content"), dict):
342 pages = value["content"].get("pages")
343 parts.extend(page_text(pages))
344 if not parts:
345 return "no_content", None
346 content = " ".join(parts)
347 digest = hashlib.sha256((value["title"] + "\0" + content).encode()).hexdigest()
348 return "eligible", digest
349
350
351def record_row(did: str, record: dict[str, Any]) -> tuple[Any, ...]:
352 value = record.get("value") or {}
353 site = value.get("publication") or value.get("site")
354 if isinstance(site, dict):
355 site = site.get("uri")
356 if not isinstance(site, str):
357 site = None
358 eligibility, fingerprint = extract_record(value)
359 return (
360 record.get("uri"),
361 did,
362 str(record.get("uri", "")).rsplit("/", 1)[-1],
363 record.get("cid"),
364 value.get("title") if isinstance(value.get("title"), str) else None,
365 site,
366 value.get("path") if isinstance(value.get("path"), str) else None,
367 value.get("publishedAt") or value.get("createdAt"),
368 eligibility,
369 fingerprint,
370 )
371
372
373async def audit_source_repos(
374 db: sqlite3.Connection,
375 client: httpx.AsyncClient,
376 workers: int,
377) -> None:
378 excluded = db.execute(
379 "UPDATE repos SET audit_status='policy_excluded' WHERE audit_status IS NULL AND policy IN ('banned','bridgy')"
380 ).rowcount
381 db.commit()
382 pending = list(
383 db.execute(
384 "SELECT did,pds FROM repos WHERE policy='eligible' AND resolution_status='resolved' AND audit_status IS NULL"
385 )
386 )
387 log(f"enumerating {len(pending)} eligible PDS repositories ({excluded} policy-excluded)")
388 queue: asyncio.Queue[tuple[str, str]] = asyncio.Queue()
389 for row in pending:
390 queue.put_nowait((row[0], row[1]))
391 lock = asyncio.Lock()
392 host_locks: dict[str, asyncio.Semaphore] = {}
393 completed = 0
394 source_total = 0
395
396 async def one_repo(did: str, pds: str) -> tuple[list[tuple[Any, ...]] | None, str | None]:
397 records: list[tuple[Any, ...]] = []
398 cursor = None
399 host = urlparse(pds).hostname or pds
400 semaphore = host_locks.setdefault(host, asyncio.Semaphore(8))
401 while True:
402 params: dict[str, str | int] = {"repo": did, "collection": COLLECTION, "limit": 100}
403 if cursor:
404 params["cursor"] = cursor
405 async with semaphore:
406 status, body, error = await request_json(
407 client,
408 f"{pds}/xrpc/com.atproto.repo.listRecords",
409 params=params,
410 )
411 if status != 200:
412 return None, f"{status}: {error}"
413 records.extend(record_row(did, record) for record in body.get("records", []))
414 cursor = body.get("cursor")
415 if not cursor:
416 return records, None
417
418 async def worker() -> None:
419 nonlocal completed, source_total
420 while not queue.empty():
421 try:
422 did, pds = queue.get_nowait()
423 except asyncio.QueueEmpty:
424 return
425 records, error = await one_repo(did, pds)
426 async with lock:
427 if records is None:
428 db.execute("UPDATE repos SET audit_status='error', error=? WHERE did=?", (error, did))
429 else:
430 db.executemany(
431 """INSERT OR REPLACE INTO source_documents
432 (uri,did,rkey,cid,title,publication_uri,path,published_at,eligibility,content_fingerprint)
433 VALUES (?,?,?,?,?,?,?,?,?,?)""",
434 records,
435 )
436 db.execute(
437 "UPDATE repos SET audit_status='complete', source_count=?, error=NULL WHERE did=?",
438 (len(records), did),
439 )
440 source_total += len(records)
441 completed += 1
442 if completed % 100 == 0:
443 db.commit()
444 log(f"PDS enumeration: {completed}/{len(pending)} repos, {source_total} records this run")
445 queue.task_done()
446
447 await asyncio.gather(*(worker() for _ in range(workers)))
448 db.commit()
449
450
451def remote_turso_query(sql: str, timeout: int = 180) -> list[dict[str, Any]]:
452 global _BACKEND_MACHINE
453 body = json.dumps(
454 {"requests": [{"type": "execute", "stmt": {"sql": sql}}, {"type": "close"}]},
455 separators=(",", ":"),
456 )
457 body64 = base64.b64encode(body.encode()).decode()
458 script = f"""body=$(printf %s {body64} | base64 -d)
459host=${{TURSO_URL#libsql://}}
460len=${{#body}}
461printf 'POST /v2/pipeline HTTP/1.1\\r\\nHost: %s\\r\\nAuthorization: Bearer %s\\r\\nContent-Type: application/json\\r\\nConnection: close\\r\\nContent-Length: %s\\r\\n\\r\\n%s' "$host" "$TURSO_TOKEN" "$len" "$body" | openssl s_client -quiet -connect "$host:443" -servername "$host" 2>/dev/null
462"""
463 encoded = base64.b64encode(script.encode()).decode()
464 command = f"sh -lc 'printf %s {encoded} | base64 -d | sh'"
465 last_error: Exception | None = None
466 for attempt in range(3):
467 try:
468 proc = subprocess.run(
469 ["fly", "ssh", "console", "-a", "leaflet-search-backend", "--machine", backend_machine(), "-C", command],
470 capture_output=True,
471 text=True,
472 timeout=timeout,
473 check=True,
474 )
475 break
476 except (subprocess.SubprocessError, OSError) as exc:
477 last_error = exc
478 _BACKEND_MACHINE = None
479 if attempt < 2:
480 time.sleep(2**attempt)
481 else:
482 raise RuntimeError("Turso admin query failed after 3 attempts") from last_error
483 payload = json.loads(proc.stdout[proc.stdout.index('{"baton"') :])
484 result = payload["results"][0]
485 if result["type"] != "ok":
486 raise RuntimeError(result)
487 result = result["response"]["result"]
488 columns = [col["name"] for col in result["cols"]]
489 return [
490 {key: (cell.get("value") if cell.get("type") != "null" else None) for key, cell in zip(columns, row)}
491 for row in result["rows"]
492 ]
493
494
495def import_turso(db: sqlite3.Connection) -> None:
496 completed = db.execute("SELECT 1 FROM meta WHERE key='turso_inventory_at'").fetchone()
497 if completed:
498 log("Turso inventory already checkpointed")
499 return
500 page_size = 2000
501 existing = db.execute("SELECT COUNT(*),MAX(source_rowid) FROM turso_documents").fetchone()
502 # Checkpoints made before source_rowid was stored cannot prove where a
503 # partial export stopped. Restart that one table rather than silently
504 # treating a truncated inventory as complete.
505 if existing[0] and existing[1] is None:
506 db.execute("DELETE FROM turso_documents")
507 db.commit()
508 existing = (0, None)
509 log("exporting Turso document inventory")
510 last_rowid = int(existing[1] or 0)
511 total = int(existing[0])
512 while True:
513 rows = remote_turso_query(
514 f"""SELECT rowid AS source_rowid,uri,did,rkey,title,LENGTH(content) AS content_length,created_at,
515 publication_uri,path,platform,source_collection,indexed_at,embedded_at,verified_at,
516 COALESCE(source_cid,'') AS source_cid,
517 COALESCE(is_bridgyfed,0) AS is_bridgyfed,
518 COALESCE(url_dead,0) AS url_dead
519 FROM documents
520 WHERE rowid > {last_rowid}
521 ORDER BY rowid
522 LIMIT {page_size}""",
523 timeout=180,
524 )
525 if not rows:
526 break
527 db.executemany(
528 """INSERT INTO turso_documents
529 (source_rowid,uri,did,rkey,title,content_length,created_at,publication_uri,path,platform,source_collection,
530 indexed_at,embedded_at,verified_at,source_cid,is_bridgyfed,url_dead)
531 VALUES (:source_rowid,:uri,:did,:rkey,:title,:content_length,:created_at,:publication_uri,:path,:platform,
532 :source_collection,:indexed_at,:embedded_at,:verified_at,:source_cid,:is_bridgyfed,:url_dead)""",
533 rows,
534 )
535 db.commit()
536 total += len(rows)
537 last_rowid = int(rows[-1]["source_rowid"])
538 log(f"Turso inventory: {total}")
539 if len(rows) < page_size:
540 break
541 db.execute(
542 "INSERT OR REPLACE INTO meta(key,value) VALUES('turso_inventory_at',?)",
543 (dt.datetime.now(dt.timezone.utc).isoformat(),),
544 )
545 db.commit()
546
547
548def import_snapshot(db: sqlite3.Connection) -> None:
549 global _BACKEND_MACHINE
550 if db.execute("SELECT 1 FROM meta WHERE key='snapshot_inventory_at'").fetchone():
551 log("snapshot inventory already checkpointed")
552 return
553 page_size = 2000
554 existing = db.execute("SELECT COUNT(*),MAX(source_rowid) FROM snapshot_documents").fetchone()
555 if existing[0] and existing[1] is None:
556 db.execute("DELETE FROM snapshot_documents")
557 db.commit()
558 existing = (0, None)
559 # Fly SSH can time out while buffering a full-table JSON response. Page the
560 # read and durably checkpoint each page so retries remain cheap and bounded.
561 log("exporting adopted serving snapshot inventory")
562 last_rowid = int(existing[1] or 0)
563 total = int(existing[0])
564 while True:
565 sql = (
566 "SELECT rowid AS source_rowid,uri,did,source_collection,indexed_at FROM documents "
567 f"WHERE rowid>{last_rowid} ORDER BY rowid LIMIT {page_size};"
568 )
569 last_error: Exception | None = None
570 for attempt in range(3):
571 try:
572 proc = subprocess.run(
573 [
574 "fly", "ssh", "console", "-a", "leaflet-search-backend",
575 "--machine", backend_machine(), "-C",
576 f'sqlite3 -readonly -json /data/local.db "{sql}"',
577 ],
578 capture_output=True,
579 text=True,
580 timeout=60,
581 check=True,
582 )
583 break
584 except (subprocess.SubprocessError, OSError) as exc:
585 last_error = exc
586 _BACKEND_MACHINE = None
587 if attempt < 2:
588 time.sleep(2**attempt)
589 else:
590 raise RuntimeError("snapshot admin query failed after 3 attempts") from last_error
591 start = proc.stdout.find("[")
592 rows = json.loads(proc.stdout[start:]) if start >= 0 else []
593 if not rows:
594 break
595 db.executemany(
596 """INSERT INTO snapshot_documents(source_rowid,uri,did,source_collection,indexed_at)
597 VALUES (:source_rowid,:uri,:did,:source_collection,:indexed_at)""",
598 rows,
599 )
600 db.commit()
601 total += len(rows)
602 last_rowid = int(rows[-1]["source_rowid"])
603 log(f"serving snapshot inventory: {total}")
604 if len(rows) < page_size:
605 break
606 db.execute(
607 "INSERT OR REPLACE INTO meta(key,value) VALUES('snapshot_inventory_at',?)",
608 (dt.datetime.now(dt.timezone.utc).isoformat(),),
609 )
610 db.commit()
611
612
613def import_vectors(db: sqlite3.Connection) -> None:
614 if db.execute("SELECT COUNT(*) FROM vectors").fetchone()[0]:
615 log("vector inventory already checkpointed")
616 return
617 api_key = env_file_value("TURBOPUFFER_API_KEY")
618 if not api_key:
619 raise RuntimeError("TURBOPUFFER_API_KEY is unavailable")
620 log("exporting complete turbopuffer namespace")
621 url = f"https://api.turbopuffer.com/v2/namespaces/{TPUF_NAMESPACE}/query"
622 headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
623 last_id = None
624 total = 0
625 with httpx.Client(timeout=120) as client:
626 while True:
627 body: dict[str, Any] = {
628 "rank_by": ["id", "asc"],
629 "limit": 10000,
630 "include_attributes": ["uri", "did"],
631 }
632 if last_id is not None:
633 body["filters"] = ["id", "Gt", last_id]
634 response = client.post(url, headers=headers, json=body)
635 response.raise_for_status()
636 rows = response.json().get("rows", [])
637 if not rows:
638 break
639 db.executemany(
640 "INSERT OR REPLACE INTO vectors(id,uri,did) VALUES (:id,:uri,:did)", rows
641 )
642 db.commit()
643 total += len(rows)
644 last_id = rows[-1]["id"]
645 log(f"vector inventory: {total}")
646 if len(rows) < 10000:
647 break
648
649
650def scalar(db: sqlite3.Connection, sql: str) -> int:
651 return int(db.execute(sql).fetchone()[0])
652
653
654def build_report(db: sqlite3.Connection) -> dict[str, Any]:
655 # Permanent repo disappearance is an audited outcome, not a transient
656 # failure. Keep unreachable/5xx hosts separate so the report's blind spots
657 # remain explicit.
658 db.execute(
659 """UPDATE repos SET audit_status='gone'
660 WHERE audit_status='error' AND (
661 error LIKE '400:%Could not find repo%'
662 OR error LIKE '400:%target repository not found%'
663 OR error LIKE '400:%Repo not found%'
664 OR error LIKE '404:%Domain not found%'
665 OR error LIKE '404:%<html%'
666 )"""
667 )
668 # This classification is part of the resumable checkpoint, not merely a
669 # presentation detail. Without the commit, process exit rolls it back and
670 # a later reconciliation run cannot distinguish gone from unreachable.
671 db.commit()
672 report: dict[str, Any] = {
673 "generated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
674 "relay_inventory_at": db.execute("SELECT value FROM meta WHERE key='relay_inventory_at'").fetchone()[0],
675 "repos": {
676 "total_audit_universe": scalar(db, "SELECT COUNT(*) FROM repos"),
677 "relay_total": scalar(db, "SELECT COUNT(*) FROM repos WHERE in_relay=1"),
678 "turso_only_repos": scalar(db, "SELECT COUNT(*) FROM repos WHERE in_relay=0"),
679 "resolved": scalar(db, "SELECT COUNT(*) FROM repos WHERE resolution_status='resolved'"),
680 "unresolved": scalar(db, "SELECT COUNT(*) FROM repos WHERE resolution_status!='resolved' OR resolution_status IS NULL"),
681 "eligible": scalar(db, "SELECT COUNT(*) FROM repos WHERE policy='eligible'"),
682 "bridgy": scalar(db, "SELECT COUNT(*) FROM repos WHERE policy='bridgy'"),
683 "banned": scalar(db, "SELECT COUNT(*) FROM repos WHERE policy='banned'"),
684 "audited_complete": scalar(db, "SELECT COUNT(*) FROM repos WHERE audit_status='complete'"),
685 "source_repo_gone": scalar(db, "SELECT COUNT(*) FROM repos WHERE audit_status='gone'"),
686 "audit_errors": scalar(db, "SELECT COUNT(*) FROM repos WHERE audit_status='error'"),
687 },
688 "source": {
689 "active_records": scalar(db, "SELECT COUNT(*) FROM source_documents"),
690 "eligible_records": scalar(db, "SELECT COUNT(*) FROM source_documents WHERE eligibility='eligible'"),
691 "metadata_only": scalar(db, "SELECT COUNT(*) FROM source_documents WHERE eligibility='no_content'"),
692 "missing_title": scalar(db, "SELECT COUNT(*) FROM source_documents WHERE eligibility='missing_title'"),
693 },
694 "layers": {
695 "turso_documents": scalar(db, "SELECT COUNT(*) FROM turso_documents"),
696 "snapshot_documents": scalar(db, "SELECT COUNT(*) FROM snapshot_documents"),
697 "vectors": scalar(db, "SELECT COUNT(*) FROM vectors"),
698 },
699 }
700 # A source URI absent from Turso can still be intentionally content-deduped:
701 # another live source record from the same DID and identical title+content is present.
702 db.execute("DROP TABLE IF EXISTS source_missing_classified")
703 db.execute(
704 """CREATE TEMP TABLE source_missing_classified AS
705 SELECT s.*,
706 CASE
707 WHEN s.eligibility != 'eligible' THEN s.eligibility
708 WHEN EXISTS (
709 SELECT 1 FROM turso_documents t
710 WHERE t.did=s.did AND t.rkey=s.rkey
711 ) THEN 'cross_collection_same_rkey'
712 WHEN EXISTS (
713 SELECT 1 FROM source_documents d
714 JOIN turso_documents t ON t.uri=d.uri
715 WHERE d.did=s.did AND d.content_fingerprint=s.content_fingerprint
716 AND d.uri != s.uri
717 ) THEN 'content_duplicate_present'
718 WHEN COALESCE(s.path,'') != '' AND EXISTS (
719 SELECT 1 FROM turso_documents t
720 WHERE t.did=s.did
721 AND COALESCE(t.publication_uri,'')=COALESCE(s.publication_uri,'')
722 AND COALESCE(t.path,'')=COALESCE(s.path,'')
723 ) THEN 'canonical_duplicate_present'
724 ELSE 'genuine_missing'
725 END AS missing_class
726 FROM source_documents s
727 LEFT JOIN turso_documents t ON t.uri=s.uri
728 WHERE t.uri IS NULL"""
729 )
730 report["source_to_turso"] = {
731 row[0]: row[1]
732 for row in db.execute(
733 "SELECT missing_class,COUNT(*) FROM source_missing_classified GROUP BY missing_class ORDER BY missing_class"
734 )
735 }
736 report["turso_to_snapshot"] = {
737 "missing_from_snapshot": scalar(
738 db,
739 "SELECT COUNT(*) FROM turso_documents t LEFT JOIN snapshot_documents s ON s.uri=t.uri WHERE s.uri IS NULL AND COALESCE(t.is_bridgyfed,0)=0",
740 ),
741 "snapshot_only": scalar(
742 db,
743 "SELECT COUNT(*) FROM snapshot_documents s LEFT JOIN turso_documents t ON t.uri=s.uri WHERE t.uri IS NULL",
744 ),
745 }
746 report["read_model_drift"] = {
747 "stale_site_standard_rows": scalar(
748 db,
749 """SELECT COUNT(*) FROM turso_documents t
750 JOIN repos r ON r.did=t.did
751 LEFT JOIN source_documents s ON s.uri=t.uri
752 WHERE t.source_collection='site.standard.document'
753 AND r.audit_status='complete' AND s.uri IS NULL""",
754 ),
755 "title_mismatches": scalar(
756 db,
757 "SELECT COUNT(*) FROM source_documents s JOIN turso_documents t ON t.uri=s.uri WHERE s.title<>t.title",
758 ),
759 "path_mismatches": scalar(
760 db,
761 "SELECT COUNT(*) FROM source_documents s JOIN turso_documents t ON t.uri=s.uri WHERE COALESCE(s.path,'')<>COALESCE(t.path,'')",
762 ),
763 "publication_mismatches": scalar(
764 db,
765 "SELECT COUNT(*) FROM source_documents s JOIN turso_documents t ON t.uri=s.uri WHERE COALESCE(s.publication_uri,'')<>COALESCE(t.publication_uri,'')",
766 ),
767 "indexed_rows_now_metadata_only": scalar(
768 db,
769 "SELECT COUNT(*) FROM source_documents s JOIN turso_documents t ON t.uri=s.uri WHERE s.eligibility='no_content'",
770 ),
771 "stale_rows_never_verified": scalar(
772 db,
773 """SELECT COUNT(*) FROM turso_documents t
774 JOIN repos r ON r.did=t.did LEFT JOIN source_documents s ON s.uri=t.uri
775 WHERE t.source_collection='site.standard.document'
776 AND r.audit_status='complete' AND s.uri IS NULL AND t.verified_at IS NULL""",
777 ),
778 }
779 report["vectors"] = {
780 "turso_marked_embedded_without_vector": scalar(
781 db,
782 "SELECT COUNT(*) FROM turso_documents t LEFT JOIN vectors v ON v.uri=t.uri WHERE t.embedded_at IS NOT NULL AND v.uri IS NULL AND COALESCE(t.is_bridgyfed,0)=0",
783 ),
784 "eligible_turso_without_vector": scalar(
785 db,
786 """SELECT COUNT(*) FROM turso_documents t LEFT JOIN vectors v ON v.uri=t.uri
787 WHERE v.uri IS NULL AND COALESCE(t.is_bridgyfed,0)=0
788 AND t.content_length > 50
789 AND t.title NOT IN ('test','testing','Test','Testing','Untitled')""",
790 ),
791 "embedding_pending_without_vector": scalar(
792 db,
793 """SELECT COUNT(*) FROM turso_documents t LEFT JOIN vectors v ON v.uri=t.uri
794 WHERE t.embedded_at IS NULL AND v.uri IS NULL AND COALESCE(t.is_bridgyfed,0)=0
795 AND t.content_length > 50
796 AND t.title NOT IN ('test','testing','Test','Testing','Untitled')""",
797 ),
798 "vector_without_turso": scalar(
799 db,
800 "SELECT COUNT(*) FROM vectors v LEFT JOIN turso_documents t ON t.uri=v.uri WHERE t.uri IS NULL",
801 ),
802 }
803 report["top_genuine_missing_repos"] = [
804 {"did": row[0], "missing": row[1], "source": row[2], "pds": row[3]}
805 for row in db.execute(
806 """SELECT m.did,COUNT(*) AS missing,r.source_count,r.pds
807 FROM source_missing_classified m JOIN repos r ON r.did=m.did
808 WHERE m.missing_class='genuine_missing'
809 GROUP BY m.did ORDER BY missing DESC,m.did LIMIT 100"""
810 )
811 ]
812 report["audit_errors"] = [
813 {"did": row[0], "pds": row[1], "error": row[2]}
814 for row in db.execute("SELECT did,pds,error FROM repos WHERE audit_status='error' ORDER BY did")
815 ]
816 return report
817
818
819async def async_main(args: argparse.Namespace, db: sqlite3.Connection) -> None:
820 limits = httpx.Limits(max_connections=args.workers * 2, max_keepalive_connections=args.workers)
821 async with httpx.AsyncClient(
822 headers={"User-Agent": "pub-search-corpus-audit/1.0"},
823 follow_redirects=True,
824 limits=limits,
825 ) as client:
826 await inventory_relay(db, client)
827 db.execute(
828 """INSERT OR IGNORE INTO repos(did,in_relay)
829 SELECT DISTINCT did,0 FROM turso_documents
830 WHERE source_collection='site.standard.document'"""
831 )
832 db.commit()
833 banned = {
834 line.split("#", 1)[0].strip()
835 for line in Path("banned-dids.txt").read_text().splitlines()
836 if line.split("#", 1)[0].strip()
837 }
838 await resolve_repos(db, client, banned, args.workers)
839 await audit_source_repos(db, client, args.workers)
840
841
842def main() -> None:
843 parser = argparse.ArgumentParser()
844 parser.add_argument("--db", type=Path, default=Path("/tmp/pub-search-corpus-audit.sqlite"))
845 parser.add_argument("--report", type=Path, default=Path("/tmp/pub-search-corpus-audit.json"))
846 parser.add_argument("--workers", type=int, default=40)
847 args = parser.parse_args()
848 db = connect(args.db)
849 import_turso(db)
850 asyncio.run(async_main(args, db))
851 import_snapshot(db)
852 import_vectors(db)
853 report = build_report(db)
854 args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
855 print(json.dumps(report, ensure_ascii=False, indent=2))
856 log(f"checkpoint: {args.db}")
857 log(f"report: {args.report}")
858
859
860if __name__ == "__main__":
861 try:
862 main()
863 except KeyboardInterrupt:
864 sys.exit(130)