/**
 * Jetstream replay: the archive, paged, downloaded and folded.
 *
 * Jetstream v2 keeps the whole network in sealed segment files and serves
 * it through one plan-then-download surface. `planSnapshot` returns the
 * segment files or block ranges that may hold matching events, `getBlock`
 * and `getSegment` stream them, and the fold converges to network truth
 * the same way the live tail does: creates apply, deletes and account
 * deletions remove. The archive is at-least-once, so every write here is
 * idempotent.
 *
 * On Bluesky's public instances the replay HTTP endpoints are metered and
 * take a Bearer API key; a 429 names a Retry-After in seconds. The block
 * and segment payloads are exact on-disk bytes, so the columnar block
 * format is decoded here and a zstd decompressor rides in on the same
 * injection the live tail uses.
 */

import { cborDecode } from '@pdsjs/core/repo';

/** The kind column values the on-disk format names. */
const KIND_CREATE = 1;
const KIND_UPDATE = 2;
const KIND_DELETE = 3;
const KIND_IDENTITY = 4;
const KIND_ACCOUNT = 5;
const KIND_SYNC = 6;
const KIND_CREATE_RESYNC = 7;

/** The sealed segment's fixed header, at file offset zero. */
const HEADER_BLOCK_COUNT = 14;
const HEADER_BLOCK_INDEX = 58;

/** One block index entry, from the block index at the footer offset. */
const INDEX_ENTRY_OFFSET = 0;
const INDEX_ENTRY_SIZE = 52;
const INDEX_COMPRESSED_SIZE = 8;

/** The length prefix before a block's frame body. */
const BLOCK_LENGTH_PREFIX = 8;

/**
 * What carries an archive request; the global fetch unless one is injected.
 * @typedef {(input: string|URL|Request, init?: RequestInit) => Promise<Response>} FetchLike
 */

/**
 * @param {number} seconds
 * @param {AbortSignal} [signal]
 * @returns {Promise<void>}
 */
function sleep(seconds, signal) {
  return new Promise((resolve) => {
    const timer = setTimeout(resolve, seconds * 1000);
    signal?.addEventListener('abort', () => {
      clearTimeout(timer);
      resolve();
    });
  });
}

/**
 * The headers the metered archive endpoints want. A 429 answer names a
 * Retry-After in seconds; the caller backs off rather than failing.
 * @param {string|undefined} apiKey
 * @returns {{'content-type': string, authorization?: string}}
 */
function archiveHeaders(apiKey) {
  return {
    'content-type': 'application/json',
    ...(apiKey ? { authorization: `Bearer ${apiKey}` } : {}),
  };
}

/**
 * One page of the download plan for a data pattern. `afterSeq` is
 * exclusive; `beforeSeq` caps the archive tip so pagination never floats.
 * @param {Object} options
 * @param {string} options.service - the Jetstream base URL
 * @param {string[]} [options.collections]
 * @param {string[]} [options.dids]
 * @param {string[]} [options.kinds]
 * @param {number} [options.afterSeq]
 * @param {number} [options.beforeSeq]
 * @param {string} [options.apiKey]
 * @param {FetchLike} [options.fetch]
 * @param {AbortSignal} [options.signal]
 * @returns {Promise<PlanPage>}
 */
export async function planSnapshot(options) {
  const {
    service,
    collections = [],
    dids = [],
    kinds = [],
    apiKey,
    signal,
  } = options;
  /** @type {Record<string, unknown>} */
  const body = { collections, dids, kinds };
  if (options.afterSeq !== undefined) body.afterSeq = options.afterSeq;
  if (options.beforeSeq !== undefined) body.beforeSeq = options.beforeSeq;
  const send = options.fetch ?? globalThis.fetch;
  const url = `${service.replace(/\/+$/, '')}/xrpc/network.bsky.jetstream.planSnapshot`;

  for (;;) {
    const res = await send(url, {
      method: 'POST',
      headers: archiveHeaders(apiKey),
      body: JSON.stringify(body),
      signal,
    });
    if (res.status === 429) {
      const retry = Number(res.headers.get('retry-after'));
      if (Number.isFinite(retry) && retry > 0) await sleep(retry, signal);
      continue;
    }
    if (!res.ok) {
      throw new Error(
        `replay plan refused (${res.status}): ${await res.text()}`,
      );
    }
    return /** @type {PlanPage} */ (await res.json());
  }
}

/**
 * One block of a sealed segment, exactly as the archive stored it: a zstd
 * frame over the columnar body. The plan names it by index.
 * @param {Object} options
 * @param {string} options.service
 * @param {string} options.segment - the segment file name
 * @param {number} options.blockIndex
 * @param {string} [options.apiKey]
 * @param {FetchLike} [options.fetch]
 * @param {AbortSignal} [options.signal]
 * @returns {Promise<Uint8Array>}
 */
export async function getBlock(options) {
  const { service, segment, blockIndex, apiKey, signal } = options;
  const send = options.fetch ?? globalThis.fetch;
  const url = `${service.replace(/\/+$/, '')}/xrpc/network.bsky.jetstream.getBlock?${new URLSearchParams(
    { segment, blockIndex: String(blockIndex) },
  )}`;

  for (;;) {
    const res = await send(url, {
      headers: archiveHeaders(apiKey),
      signal,
    });
    if (res.status === 429) {
      const retry = Number(res.headers.get('retry-after'));
      if (Number.isFinite(retry) && retry > 0) await sleep(retry, signal);
      continue;
    }
    if (!res.ok) {
      throw new Error(
        `replay block refused (${res.status}): ${await res.text()}`,
      );
    }
    return new Uint8Array(await res.arrayBuffer());
  }
}

/**
 * One sealed segment file, whole. Only a plan entry in `segment` mode
 * fetches this; the header and block index pick the frames out of it.
 * @param {Object} options
 * @param {string} options.service
 * @param {string} options.segment
 * @param {string} [options.apiKey]
 * @param {FetchLike} [options.fetch]
 * @param {AbortSignal} [options.signal]
 * @returns {Promise<Uint8Array>}
 */
export async function getSegment(options) {
  const { service, segment, apiKey, signal } = options;
  const send = options.fetch ?? globalThis.fetch;
  const url = `${service.replace(/\/+$/, '')}/xrpc/network.bsky.jetstream.getSegment?${new URLSearchParams(
    { name: segment },
  )}`;

  for (;;) {
    const res = await send(url, {
      headers: archiveHeaders(apiKey),
      signal,
    });
    if (res.status === 429) {
      const retry = Number(res.headers.get('retry-after'));
      if (Number.isFinite(retry) && retry > 0) await sleep(retry, signal);
      continue;
    }
    if (!res.ok) {
      throw new Error(
        `replay segment refused (${res.status}): ${await res.text()}`,
      );
    }
    return new Uint8Array(await res.arrayBuffer());
  }
}

/**
 * The events one block frame holds, in stored order. The frame is a zstd
 * compression over a columnar body; without an injected `decompress` it is
 * read uncompressed.
 * @param {Uint8Array} frame
 * @param {(bytes: Uint8Array) => Promise<Uint8Array>} [decompress]
 * @returns {Promise<ArchiveRow[]>}
 */
export async function decodeBlock(frame, decompress) {
  const bytes = decompress ? await decompress(frame) : frame;
  return decodeBlockColumns(bytes);
}

/**
 * The columnar block body, exactly as `encodeBlock` writes it: one uint32
 * event count, then each fixed column whole (seq, witnessed_at, indexed_at,
 * kind, and the four length columns), then the concatenated variable
 * columns in the same order. Columns span every event, in stored order.
 * @param {Uint8Array} bytes
 * @returns {ArchiveRow[]}
 */
export function decodeBlockColumns(bytes) {
  const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
  const count = view.getUint32(0, true);
  // seq, witnessed_at, indexed_at, kind, and the four length columns.
  const fixedPerEvent = 8 + 8 + 8 + 1 + 1 + 2 + 1 + 1 + 4;
  if (bytes.byteLength < 4 + count * fixedPerEvent) {
    throw new Error('replay block is shorter than its header claims');
  }
  const str = (/** @type {number} */ offset, /** @type {number} */ length) =>
    new TextDecoder().decode(bytes.subarray(offset, offset + length));

  const seqColumn = 4;
  const witnessedColumn = seqColumn + count * 8;
  const indexedColumn = witnessedColumn + count * 8;
  const kindColumn = indexedColumn + count * 8;
  const collLenColumn = kindColumn + count;
  const didLenColumn = collLenColumn + count;
  const rkeyLenColumn = didLenColumn + count * 2;
  const revLenColumn = rkeyLenColumn + count;
  const payloadLenColumn = revLenColumn + count;

  /** @type {number[]} */
  const collLens = [];
  /** @type {number[]} */
  const didLens = [];
  /** @type {number[]} */
  const rkeyLens = [];
  /** @type {number[]} */
  const revLens = [];
  /** @type {number[]} */
  const payloadLens = [];
  for (let i = 0; i < count; i++) {
    collLens.push(view.getUint8(collLenColumn + i));
    didLens.push(view.getUint16(didLenColumn + i * 2, true));
    rkeyLens.push(view.getUint8(rkeyLenColumn + i));
    revLens.push(view.getUint8(revLenColumn + i));
    payloadLens.push(view.getUint32(payloadLenColumn + i * 4, true));
  }
  const sum = (/** @type {number[]} */ list) =>
    list.reduce((total, length) => total + length, 0);
  // Each variable column is one concatenated blob, in column order.
  const varBase = 4 + count * fixedPerEvent;
  const base = [varBase];
  for (const lens of [collLens, didLens, rkeyLens, revLens, payloadLens]) {
    base.push(base[base.length - 1] + sum(lens));
  }
  if (base[5] !== bytes.byteLength) {
    throw new Error('replay block carries trailing bytes');
  }

  /** @type {number[]} */
  const offset = [0, 0, 0, 0, 0];
  /** @type {ArchiveRow[]} */
  const rows = [];
  for (let i = 0; i < count; i++) {
    const column = (/** @type {number} */ n) => {
      const at = base[n] + offset[n];
      offset[n] += [collLens, didLens, rkeyLens, revLens, payloadLens][n][i];
      return at;
    };
    const payloadStart = column(4);
    rows.push({
      seq: Number(view.getBigUint64(seqColumn + i * 8, true)),
      kind: view.getUint8(kindColumn + i),
      did: str(column(1), didLens[i]),
      collection: str(column(0), collLens[i]),
      rkey: str(column(2), rkeyLens[i]),
      rev: str(column(3), revLens[i]),
      payload: bytes.subarray(payloadStart, payloadStart + payloadLens[i]),
    });
  }
  return rows;
}

/**
 * Every block frame a sealed segment file holds, decoded in stored order.
 * @param {Uint8Array} file - the .jss bytes
 * @param {(bytes: Uint8Array) => Promise<Uint8Array>} decompress
 * @returns {Promise<ArchiveRow[][]>}
 */
export async function decodeSegment(file, decompress) {
  if (!decompress) {
    throw new Error('replay needs a zstd decompressor');
  }
  const view = new DataView(file.buffer, file.byteOffset, file.byteLength);
  const blockCount = view.getUint32(HEADER_BLOCK_COUNT, true);
  const blockIndex = Number(view.getBigUint64(HEADER_BLOCK_INDEX, true));
  /** @type {ArchiveRow[][]} */
  const blocks = [];
  for (let i = 0; i < blockCount; i++) {
    const entry = blockIndex + i * INDEX_ENTRY_SIZE;
    const offset = Number(view.getBigUint64(entry + INDEX_ENTRY_OFFSET, true));
    const compressed = view.getUint32(entry + INDEX_COMPRESSED_SIZE, true);
    const frame = file.subarray(
      offset + BLOCK_LENGTH_PREFIX,
      offset + BLOCK_LENGTH_PREFIX + compressed,
    );
    blocks.push(await decodeBlock(frame, decompress));
  }
  return blocks;
}

/**
 * Fold one decoded row through the same shapes the live tail hands onEvent.
 * Rows the archive cannot speak to (an identity event with no handle, an
 * account event with no status) fold to nothing rather than a guess.
 * @param {ArchiveRow} row
 * @returns {RecordWrite|RecordDelete|IdentityEvent|AccountEvent|null}
 */
export function rowToEvent(row) {
  const uri =
    row.collection && row.rkey
      ? `at://${row.did}/${row.collection}/${row.rkey}`
      : null;
  switch (row.kind) {
    case KIND_CREATE:
    case KIND_UPDATE:
    case KIND_CREATE_RESYNC: {
      if (!uri) return null;
      let value;
      try {
        value = cborDecode(row.payload);
      } catch {
        return null;
      }
      return {
        did: row.did,
        collection: row.collection,
        rkey: row.rkey,
        cid: null,
        uri,
        seq: row.seq,
        value,
      };
    }
    case KIND_DELETE:
      return uri
        ? {
            did: row.did,
            collection: row.collection,
            rkey: row.rkey,
            uri,
            seq: row.seq,
            deleted: true,
          }
        : null;
    case KIND_IDENTITY: {
      let handle = '';
      try {
        const value = cborDecode(row.payload);
        handle = typeof value?.handle === 'string' ? value.handle : '';
      } catch {
        // A row with no readable payload cannot name a handle.
      }
      return handle
        ? { kind: 'identity', did: row.did, handle, seq: row.seq }
        : null;
    }
    case KIND_ACCOUNT: {
      let status;
      try {
        const value = cborDecode(row.payload);
        status = typeof value?.status === 'string' ? value.status : undefined;
      } catch {
        // A row with no readable payload cannot name a status.
      }
      return status
        ? { kind: 'account', did: row.did, status, seq: row.seq }
        : null;
    }
    default:
      return null;
  }
}

/**
 * Replay the archive from `afterSeq` to the sealed tip, folding each
 * matching event and checkpointing each page's `plannedThroughSeq`. The
 * tip is pinned on the first page, so the range never floats as new data
 * arrives. Resolves with the sealed tip the caller tails live from.
 * @param {Object} options
 * @param {string} options.service
 * @param {string[]} [options.collections]
 * @param {string[]} [options.dids]
 * @param {string[]} [options.kinds]
 * @param {number} [options.afterSeq] - exclusive; the archive starts here
 * @param {string} [options.apiKey]
 * @param {FetchLike} [options.fetch]
 * @param {(bytes: Uint8Array) => Promise<Uint8Array>} [options.decompress]
 * @param {AbortSignal} [options.signal]
 * @param {(event: RecordWrite|RecordDelete|IdentityEvent|AccountEvent) => Promise<void>|void} options.onEvent
 * @param {(seq: number) => Promise<void>|void} [options.onSeq] - each page's
 *   continuation cursor, which a caller checkpoints between pages
 * @param {(message: string) => void} [options.onNotice]
 * @returns {Promise<number>} the sealed tip to resume live from
 */
export async function replayArchive(options) {
  const {
    service,
    collections = [],
    dids = [],
    kinds = [],
    apiKey,
    decompress,
    signal,
    onEvent,
    onSeq = () => {},
    onNotice = () => {},
  } = options;
  const sends = options.fetch ?? globalThis.fetch;
  if (!decompress) {
    throw new Error('replay needs a zstd decompressor');
  }
  let after = options.afterSeq ?? 0;
  let sealed = after;

  /** @param {ArchiveRow} row */
  const foldRow = async (row) => {
    if (row.seq <= after) return;
    if (kinds.length > 0 && !kinds.includes(kindName(row.kind))) return;
    if (dids.length > 0 && !dids.includes(row.did)) return;
    if (
      row.collection &&
      collections.length > 0 &&
      !collections.includes(row.collection)
    ) {
      return;
    }
    const event = rowToEvent(row);
    if (event) await onEvent(event);
  };

  /** @param {Uint8Array} frame */
  const foldFrame = async (frame) => {
    for (const row of await decodeBlock(frame, decompress)) {
      await foldRow(row);
    }
  };

  /** @param {Uint8Array} file */
  const foldFile = async (file) => {
    for (const block of await decodeSegment(file, decompress)) {
      for (const row of block) {
        await foldRow(row);
      }
    }
  };

  for (;;) {
    if (signal?.aborted) return sealed;
    const plan = await planSnapshot({
      service,
      collections,
      dids,
      kinds,
      afterSeq: after,
      beforeSeq: sealed > after ? sealed : undefined,
      apiKey,
      fetch: sends,
      signal,
    });
    sealed = plan.sealedTipSeq;
    onNotice(
      `replay ${after}..${sealed} over ${plan.segments.length} plan entries`,
    );
    for (const segment of plan.segments) {
      if (signal?.aborted) return sealed;
      if (segment.mode === 'segment') {
        await foldFile(
          await getSegment({
            service,
            segment: segment.name,
            apiKey,
            fetch: sends,
            signal,
          }),
        );
      } else {
        for (const range of segment.blocks ?? []) {
          for (let index = range.first; index <= range.last; index++) {
            await foldFrame(
              await getBlock({
                service,
                segment: segment.name,
                blockIndex: index,
                apiKey,
                fetch: sends,
                signal,
              }),
            );
          }
        }
      }
    }
    await onSeq(plan.plannedThroughSeq);
    if (plan.plannedThroughSeq >= sealed) return sealed;
    after = plan.plannedThroughSeq;
  }
}

/**
 * The wire name of a row kind, for the kinds filter a plan accepts.
 * @param {number} kind
 * @returns {string}
 */
function kindName(kind) {
  switch (kind) {
    case KIND_CREATE:
    case KIND_CREATE_RESYNC:
      return 'commit';
    case KIND_UPDATE:
      return 'commit';
    case KIND_DELETE:
      return 'commit';
    case KIND_IDENTITY:
      return 'identity';
    case KIND_ACCOUNT:
      return 'account';
    case KIND_SYNC:
      return 'sync';
    default:
      return 'commit';
  }
}

/** @typedef {import('./store.js').RecordWrite} RecordWrite */
/** @typedef {import('./store.js').RecordDelete} RecordDelete */
/** @typedef {import('./jetstream.js').IdentityEvent} IdentityEvent */
/** @typedef {import('./jetstream.js').AccountEvent} AccountEvent */

/**
 * One decoded row of a block.
 * @typedef {Object} ArchiveRow
 * @property {number} seq
 * @property {number} kind
 * @property {string} did
 * @property {string} collection
 * @property {string} rkey
 * @property {string} rev
 * @property {Uint8Array} payload
 */

/**
 * One page of a download plan.
 * @typedef {Object} PlanPage
 * @property {number} sealedTipSeq
 * @property {number} plannedThroughSeq
 * @property {PlanSegment[]} segments
 */

/**
 * @typedef {Object} PlanSegment
 * @property {string} name
 * @property {number} index
 * @property {string} mode - 'segment' or 'blocks'
 * @property {Array<{first: number, last: number}>} [blocks]
 */