Something went wrong. Try again.
an app to share curated trails sidetrail.app
atproto nextjs react rsc
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251/** * Sync the index with the live network. * * The jetstream ingester only sees records created while it's running, so a * fresh database (or one whose ingester was down longer than jetstream's * replay window) drifts from reality. This script reconciles: after it runs, * the index mirrors the lexicon-valid sidetrail records currently on the * network. * * - Discovers repos via the relay (plus authors already in the DB) * - Fetches their records from each PDS and upserts the lexicon-valid ones, * with the same semantics as ingester/src/handler.ts * - Deletes local rows that are no longer on the network or fail validation * - Never deletes rows belonging to a repo it failed to reach * * Idempotent and safe to rerun anytime. * * Usage: * npm run sync # uses DATABASE_URL from .env * DATABASE_URL=postgres://... npm run sync */import fs from "node:fs";import pg from "pg";import { validateRecord, type IndexedCollection } from "../ingester/src/lexicons.js";
const RELAY = "https://relay1.us-east.bsky.network";const COLLECTIONS: IndexedCollection[] = [ "app.sidetrail.trail", "app.sidetrail.walk", "app.sidetrail.completion",];const TABLE_FOR: Record<IndexedCollection, string> = { "app.sidetrail.trail": "trails", "app.sidetrail.walk": "walks", "app.sidetrail.completion": "completions",};
// PDS errors that definitively mean "this repo has no content to index"// (as opposed to transient failures, which must not trigger pruning)const REPO_GONE = /RepoNotFound|RepoDeactivated|RepoTakendown|RepoSuspended/;
const databaseUrl = process.env.DATABASE_URL ?? fs .readFileSync(new URL("../.env", import.meta.url), "utf8") .match(/^DATABASE_URL="?([^"\n]+)/m)?.[1];
type PdsRecord = { uri: string; cid: string; value: Record<string, unknown> };
async function getJson(url: URL | string): Promise<any> { const res = await fetch(url, { signal: AbortSignal.timeout(15000) }); if (!res.ok) throw new Error(`${res.status} ${await res.text().catch(() => "")} ${url}`); return res.json();}
async function listReposByCollection(collection: string): Promise<Set<string>> { const dids = new Set<string>(); let cursor: string | undefined; do { const url = new URL(`${RELAY}/xrpc/com.atproto.sync.listReposByCollection`); url.searchParams.set("collection", collection); url.searchParams.set("limit", "500"); if (cursor) url.searchParams.set("cursor", cursor); const page = await getJson(url); for (const r of page.repos) dids.add(r.did); cursor = page.cursor; } while (cursor); return dids;}
async function resolvePds(did: string): Promise<string> { const doc = did.startsWith("did:web:") ? await getJson(`https://${did.slice("did:web:".length)}/.well-known/did.json`) : await getJson(`https://plc.directory/${did}`); const endpoint = doc.service?.find((s: any) => s.id === "#atproto_pds")?.serviceEndpoint; if (!endpoint) throw new Error(`no PDS endpoint for ${did}`); return endpoint;}
async function listRecords( pdsEndpoint: string, did: string, collection: string,): Promise<PdsRecord[]> { const records: PdsRecord[] = []; let cursor: string | undefined; do { const url = new URL(`${pdsEndpoint}/xrpc/com.atproto.repo.listRecords`); url.searchParams.set("repo", did); url.searchParams.set("collection", collection); url.searchParams.set("limit", "100"); if (cursor) url.searchParams.set("cursor", cursor); const page = await getJson(url); records.push(...page.records); cursor = page.cursor; } while (cursor); return records;}
async function upsertRecord( db: pg.Client, collection: IndexedCollection, did: string, { uri, cid, value }: PdsRecord,): Promise<void> { const rkey = uri.split("/").pop(); const createdAt = new Date((value.createdAt as string) ?? Date.now()); const table = TABLE_FOR[collection]; if (table === "trails") { await db.query( `INSERT INTO trails (uri, cid, author_did, rkey, record, created_at, indexed_at) VALUES ($1, $2, $3, $4, $5, $6, NOW()) ON CONFLICT (uri) DO UPDATE SET cid = $2, record = $5, indexed_at = NOW()`, [uri, cid, did, rkey, JSON.stringify(value), createdAt], ); } else { await db.query( `INSERT INTO ${table} (uri, cid, author_did, rkey, trail_uri, record, created_at, indexed_at) VALUES ($1, $2, $3, $4, $5, $6, $7, NOW()) ON CONFLICT (uri) DO UPDATE SET cid = $2, record = $6, indexed_at = NOW()`, [ uri, cid, did, rkey, (value.trail as { uri?: string } | undefined)?.uri ?? "", JSON.stringify(value), createdAt, ], ); }}
async function main() { const db = new pg.Client({ connectionString: databaseUrl }); await db.connect();
// Repos to sync: everything the relay knows about, plus authors already // indexed locally (so records deleted from the network get pruned even if // their author no longer appears in the relay's listing) const [relayDidSets, localDids, inactive] = await Promise.all([ Promise.all(COLLECTIONS.map(listReposByCollection)), db.query( `SELECT DISTINCT author_did AS did FROM trails UNION SELECT DISTINCT author_did FROM walks UNION SELECT DISTINCT author_did FROM completions`, ), db.query(`SELECT did FROM accounts WHERE active = 0`), ]); const skip = new Set<string>(inactive.rows.map((r) => r.did)); const dids = [ ...new Set([...relayDidSets.flatMap((s) => [...s]), ...localDids.rows.map((r) => r.did)]), ].filter((did) => !skip.has(did)); console.log(`syncing ${dids.length} repos`);
// uris seen on the network and valid, per collection; only repos in // `fetchedOk` participate in pruning const keep: Record<IndexedCollection, Set<string>> = { "app.sidetrail.trail": new Set(), "app.sidetrail.walk": new Set(), "app.sidetrail.completion": new Set(), }; const fetchedOk = new Set<string>(); const failures: string[] = []; let rejected = 0;
const queue = [...dids]; let done = 0; await Promise.all( Array.from({ length: 8 }, async () => { let did: string | undefined; while ((did = queue.shift())) { done++; try { let pdsEndpoint: string; try { pdsEndpoint = await resolvePds(did); } catch (err) { if (REPO_GONE.test(String(err))) { fetchedOk.add(did); // identity gone: prune everything local console.log(`[${done}/${dids.length}] ${did}: repo gone`); continue; } throw err; } const counts: string[] = []; for (const collection of COLLECTIONS) { let records: PdsRecord[]; try { records = await listRecords(pdsEndpoint, did, collection); } catch (err) { if (REPO_GONE.test(String(err))) { records = []; // repo gone: nothing to keep, prune local rows } else { throw err; } } for (const record of records) { const validation = validateRecord(collection, record.value); if (!validation.success) { rejected++; console.log(` rejecting ${collection} ${record.uri}: ${validation.reason}`); continue; } await upsertRecord(db, collection, did, record); keep[collection].add(record.uri); } if (records.length > 0) counts.push(`${collection.split(".").pop()}=${records.length}`); } fetchedOk.add(did); await db.query( `INSERT INTO accounts (did, active, seq, updated_at) VALUES ($1, 1, 0, NOW()) ON CONFLICT (did) DO NOTHING`, [did], ); console.log(`[${done}/${dids.length}] ${did}:`, counts.join(" ") || "no records"); } catch (err) { failures.push(did); console.error(`[${done}/${dids.length}] ${did}: FAILED - ${(err as Error).message}`); } } }), );
// Prune: rows from successfully-synced repos that aren't in the keep set // (deleted from the network, or no longer lexicon-valid) for (const collection of COLLECTIONS) { const table = TABLE_FOR[collection]; const { rows } = await db.query( `DELETE FROM ${table} WHERE author_did = ANY($1) AND NOT (uri = ANY($2)) RETURNING uri`, [[...fetchedOk], [...keep[collection]]], ); for (const row of rows) console.log(`pruned ${row.uri}`); const { rows: count } = await db.query(`SELECT COUNT(*) FROM ${table}`); console.log(`${table}: ${count[0].count} rows (${rows.length} pruned)`); } console.log(`${rejected} records rejected by lexicon validation`); if (failures.length > 0) { console.error( `\n${failures.length} repos unreachable, left untouched (rerun to retry): ${failures.join(", ")}`, ); process.exitCode = 1; }
await db.end();}
main();