import type { IndexCollection, IndexEvent } from './model.ts';
import { toPutEvent, WANTED_COLLECTIONS } from './jetstream.ts';
import { FETCH_TIMEOUT_MS, isPublicHttpsUrl } from '../net.ts';
import type { Upstream } from '../upstream.ts';

/** Injected rather than `globalThis.fetch` so tests run with no network. */
export interface BackfillDeps {
  relayUrl: string;
  fetch: Upstream;
  /** Resolves a DID to its PDS base URL (normally via `plc.directory` or `did:web`). */
  resolvePds(did: string): Promise<string>;
  /** Per-request timeout. Default `FETCH_TIMEOUT_MS`. */
  timeoutMs?: number;
}

/** Page caps, so a server that never stops handing out cursors can't hold the alarm forever:
 * 1000 relay pages of 1000 repos, and 100 PDS pages of 100 records per repo and collection. */
export const MAX_REPO_PAGES = 1000;
export const MAX_RECORD_PAGES = 100;

/** What a backfill produces. `repo` names every record one repo's PDS still holds in a
 * collection, so the index can drop rows deleted while it wasn't listening. It is emitted only
 * after that repo's listing finished, never after a partial one. */
export type BackfillItem =
  | { kind: 'event'; event: IndexEvent }
  | { kind: 'repo'; collection: IndexCollection; did: string; uris: ReadonlySet<string> };

/** One page of a paged XRPC listing, or `undefined` if the request failed or the body isn't
 * JSON. */
async function getJson(url: URL, deps: BackfillDeps): Promise<unknown> {
  try {
    const res = await deps.fetch(url.toString(), { signal: AbortSignal.timeout(deps.timeoutMs ?? FETCH_TIMEOUT_MS) });
    if (!res.ok) return undefined;
    return await res.json();
  } catch {
    return undefined;
  }
}

type Next = { kind: 'page'; cursor: string } | { kind: 'end' } | { kind: 'repeat' };

/** Where a listing goes next. A missing or empty cursor ends it; a cursor the server already
 * handed out means it would loop forever. */
function nextCursor(body: { cursor?: unknown }, seen: Set<string>): Next {
  const cursor = body.cursor;
  if (typeof cursor !== 'string' || cursor === '') return { kind: 'end' };
  if (seen.has(cursor)) return { kind: 'repeat' };
  seen.add(cursor);
  return { kind: 'page', cursor };
}

/** `com.atproto.sync.listReposByCollection` on the relay, paged until it ends, repeats a
 * cursor, or hits `MAX_REPO_PAGES`. */
async function* listReposByCollection(collection: string, deps: BackfillDeps): AsyncGenerator<string> {
  const seen = new Set<string>();
  let cursor: string | undefined;
  for (let page = 0; page < MAX_REPO_PAGES; page++) {
    const url = new URL('/xrpc/com.atproto.sync.listReposByCollection', deps.relayUrl);
    url.searchParams.set('collection', collection);
    url.searchParams.set('limit', '1000');
    if (cursor) url.searchParams.set('cursor', cursor);
    const body = (await getJson(url, deps)) as { repos?: unknown; cursor?: unknown } | undefined;
    if (!body || !Array.isArray(body.repos)) throw new Error(`listReposByCollection ${collection}: bad response`);
    for (const repo of body.repos as { did?: unknown }[]) {
      if (typeof repo?.did === 'string' && repo.did.startsWith('did:')) yield repo.did;
    }
    const next = nextCursor(body, seen);
    if (next.kind !== 'page') return;
    cursor = next.cursor;
  }
}

/** The record's URI rebuilt from the repo we asked for, or `undefined` when the PDS claims a
 * record in some other repo or collection. A PDS is untrusted: without this, one account's PDS
 * could overwrite index rows that belong to another account. */
export function ownedRecordUri(uri: unknown, did: string, collection: string): string | undefined {
  if (typeof uri !== 'string' || !uri.startsWith('at://')) return undefined;
  const parts = uri.slice('at://'.length).split('/');
  if (parts.length !== 3) return undefined;
  const [authority, recordCollection, rkey] = parts as [string, string, string];
  if (authority !== did || recordCollection !== collection || rkey === '') return undefined;
  return `at://${did}/${collection}/${rkey}`;
}

/** `com.atproto.repo.listRecords` against one repo's own PDS. A repo whose PDS can't be
 * resolved, isn't a public `https:` URL, or answers with an error or a malformed page is skipped
 * rather than failing the whole backfill. Only a listing that ended normally yields a `repo`
 * item, so a skipped, looping, or capped repo has none of its rows pruned. */
async function* listRecords(did: string, collection: IndexCollection, deps: BackfillDeps): AsyncGenerator<BackfillItem> {
  let pds: string;
  try {
    pds = await deps.resolvePds(did);
  } catch {
    return;
  }
  if (!isPublicHttpsUrl(pds)) return;
  const uris = new Set<string>();
  const seen = new Set<string>();
  let cursor: string | undefined;
  for (let page = 0; page < MAX_RECORD_PAGES; page++) {
    const url = new URL('/xrpc/com.atproto.repo.listRecords', pds);
    url.searchParams.set('repo', did);
    url.searchParams.set('collection', collection);
    url.searchParams.set('limit', '100');
    if (cursor) url.searchParams.set('cursor', cursor);
    const body = (await getJson(url, deps)) as { records?: unknown; cursor?: unknown } | undefined;
    if (!body || !Array.isArray(body.records)) return;
    for (const record of body.records as { uri?: unknown; cid?: unknown; value?: unknown }[]) {
      const uri = ownedRecordUri(record?.uri, did, collection);
      if (!uri || typeof record.cid !== 'string') continue;
      uris.add(uri);
      const event = toPutEvent(collection, uri, record.cid, did, record.value);
      if (event) yield { kind: 'event', event };
    }
    const next = nextCursor(body, seen);
    if (next.kind === 'end') yield { kind: 'repo', collection, did, uris };
    if (next.kind !== 'page') return;
    cursor = next.cursor;
  }
}

/** Rebuilds the index for one collection from the relay and each repo's PDS, used when the
 * stored cursor is missing or older than Jetstream's replay window. */
export async function* backfillCollection(collection: IndexCollection, deps: BackfillDeps): AsyncGenerator<BackfillItem> {
  const dids = new Set<string>();
  for await (const did of listReposByCollection(collection, deps)) {
    if (dids.has(did)) continue;
    dids.add(did);
    yield* listRecords(did, collection, deps);
  }
}

/** Backfills every collection the index cares about. */
export async function* backfillAll(deps: BackfillDeps): AsyncGenerator<BackfillItem> {
  for (const collection of WANTED_COLLECTIONS) {
    yield* backfillCollection(collection, deps);
  }
}