6760c434b4
The Database handle is now constructed on first use and held in a module-level slot. New _setDbForTest and _resetDbForTest exports let tests swap in a :memory: handle without rewriting callers. closeDb() replaces the previous db.close() pattern used by short-lived scripts. Adds listDistinctNoteDids() to keep the backfill script from poking the raw connection.
124 lines
3.4 KiB
TypeScript
124 lines
3.4 KiB
TypeScript
// Backfill OpenSearch from each known user's PDS.
|
|
//
|
|
// deno task backfill:search
|
|
//
|
|
// Reads distinct DIDs from the local SQLite database, resolves each DID's PDS
|
|
// via the PLC directory, paginates `com.atproto.repo.listRecords` for
|
|
// `space.remanso.note`, and indexes every record into OpenSearch.
|
|
//
|
|
// Requires OPENSEARCH_URL to be set; falls back to a no-op otherwise.
|
|
// SQLite path follows the same SQLITE_PATH convention as the server/jetstream.
|
|
|
|
import { closeDb, listDistinctNoteDids } from "../src/data/db.ts";
|
|
import { resolvePds } from "../src/auth/verify.ts";
|
|
import { ensureIndex, indexNote } from "../src/search/opensearch.ts";
|
|
import { log } from "../src/log.ts";
|
|
|
|
type ListRecordsResponse = {
|
|
records: { uri: string; value: Record<string, unknown> }[];
|
|
cursor?: string;
|
|
};
|
|
|
|
const COLLECTION = "space.remanso.note";
|
|
const PAGE_SIZE = 100;
|
|
|
|
const rkeyFromUri = (uri: string): string => {
|
|
const parts = uri.split("/");
|
|
return parts[parts.length - 1];
|
|
};
|
|
|
|
const backfillDid = async (
|
|
did: string,
|
|
): Promise<{ indexed: number; failed: number }> => {
|
|
let indexed = 0;
|
|
let failed = 0;
|
|
const pds = await resolvePds(did);
|
|
let cursor: string | undefined = undefined;
|
|
|
|
do {
|
|
const url = new URL(`${pds}/xrpc/com.atproto.repo.listRecords`);
|
|
url.searchParams.set("repo", did);
|
|
url.searchParams.set("collection", COLLECTION);
|
|
url.searchParams.set("limit", String(PAGE_SIZE));
|
|
if (cursor) url.searchParams.set("cursor", cursor);
|
|
|
|
const res = await fetch(url);
|
|
if (!res.ok) {
|
|
throw new Error(`listRecords ${did}: ${res.status} ${await res.text()}`);
|
|
}
|
|
const page = (await res.json()) as ListRecordsResponse;
|
|
|
|
for (const record of page.records) {
|
|
const rkey = rkeyFromUri(record.uri);
|
|
const value = record.value as {
|
|
title?: string;
|
|
content?: string;
|
|
language?: string;
|
|
discoverable?: boolean;
|
|
publishedAt?: string;
|
|
createdAt?: string;
|
|
};
|
|
if (!value.title) {
|
|
failed++;
|
|
log(`[backfill] skip ${did}/${rkey}: missing title`);
|
|
continue;
|
|
}
|
|
try {
|
|
await indexNote({
|
|
did,
|
|
rkey,
|
|
title: value.title,
|
|
content: value.content,
|
|
language: value.language,
|
|
discoverable: value.discoverable,
|
|
publishedAt: value.publishedAt,
|
|
createdAt: value.createdAt,
|
|
});
|
|
indexed++;
|
|
} catch (error) {
|
|
failed++;
|
|
log(`[backfill] index ${did}/${rkey} failed:`, error);
|
|
}
|
|
}
|
|
|
|
cursor = page.cursor;
|
|
} while (cursor);
|
|
|
|
return { indexed, failed };
|
|
};
|
|
|
|
const main = async () => {
|
|
if (!Deno.env.get("OPENSEARCH_URL")) {
|
|
console.error("OPENSEARCH_URL is not set; aborting backfill.");
|
|
Deno.exit(1);
|
|
}
|
|
|
|
try {
|
|
await ensureIndex();
|
|
} catch (error) {
|
|
console.error("Failed to ensure index:", error);
|
|
Deno.exit(1);
|
|
}
|
|
|
|
const dids = listDistinctNoteDids();
|
|
log(`[backfill] ${dids.length} DID(s) to process`);
|
|
|
|
let totalIndexed = 0;
|
|
let totalFailed = 0;
|
|
for (const did of dids) {
|
|
try {
|
|
const { indexed, failed } = await backfillDid(did);
|
|
log(`[backfill] ${did}: indexed=${indexed} failed=${failed}`);
|
|
totalIndexed += indexed;
|
|
totalFailed += failed;
|
|
} catch (error) {
|
|
log(`[backfill] ${did} aborted:`, error);
|
|
totalFailed++;
|
|
}
|
|
}
|
|
log(`[backfill] done. indexed=${totalIndexed} failed=${totalFailed}`);
|
|
closeDb();
|
|
};
|
|
|
|
await main();
|