d7e189c61f
Iterates distinct DIDs in SQLite, resolves each one's PDS, and reindexes every space.remanso.note record into OpenSearch via com.atproto.repo.listRecords. Needed because note content is not persisted to SQLite, so the index would otherwise only cover notes created after deployment.
126 lines
3.5 KiB
TypeScript
126 lines
3.5 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 { db } 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 rows = db.prepare("SELECT DISTINCT did FROM note ORDER BY did").all<
|
|
{ did: string }
|
|
>();
|
|
log(`[backfill] ${rows.length} DID(s) to process`);
|
|
|
|
let totalIndexed = 0;
|
|
let totalFailed = 0;
|
|
for (const { did } of rows) {
|
|
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}`);
|
|
db.close();
|
|
};
|
|
|
|
await main();
|