/**
 * 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();
