// @pdsjs/core/handlers/xrpc-sync - com.atproto.sync.* repo reads.
//
// This server has one account. If a `did` parameter gives a different account,
// the endpoint sends a RepoNotFound error. It does not do a search.
//
// getBlob and listBlobs are in the same namespace, but they read blob storage
// and not the repo. They are with the other blob endpoints.

import { mstGet } from '../mst.js';
import {
  buildCarFile,
  carBlock,
  carHeader,
  cborDecode,
  cidToString,
} from '../repo.js';

/**
 * @typedef {Object} SyncContext
 * @property {import('../ports.js').ActorStoragePort} actorStorage
 * @property {() => Promise<string|null>} getDid
 * @property {() => Promise<boolean>} isDeactivated
 * @property {<T>(fn: () => Promise<T>) => Promise<T>} [withRepoLock] - Runs
 *   `fn` after the repo's writes queued before it; the sweep before getRepo
 *   holds it. Absent, `fn` runs at once.
 */

/** Blocks read between two waits on a timer, or in one page of the table. */
const BLOCKS_PER_WAIT = 500;

/** @returns {Response} */
function repoNotFound() {
  return Response.json(
    { error: 'RepoNotFound', message: 'Repository not found' },
    { status: 404 },
  );
}

/**
 * @param {SyncContext} ctx
 * @returns {{
 *   buildFullRepoCar: (commitCid: string) => Promise<Uint8Array>,
 *   routes: import('../pds.js').Routes,
 * }}
 */
export function createSyncHandlers(ctx) {
  const { actorStorage, getDid, isDeactivated } = ctx;

  /**
   * com.atproto.sync.listRepos
   * @returns {Promise<Response>}
   */
  async function handleListRepos() {
    const did = await getDid();
    const commit = await actorStorage.getLatestCommit();

    if (!did) {
      return Response.json({ repos: [] });
    }

    const deactivated = await isDeactivated();
    return Response.json({
      repos: [
        {
          did,
          head: commit?.cid || null,
          rev: commit?.rev || null,
          active: !deactivated,
          ...(deactivated ? { status: 'deactivated' } : {}),
        },
      ],
    });
  }

  /**
   * com.atproto.sync.getRepoStatus
   * @param {Request} _request
   * @param {URL} url
   * @returns {Promise<Response>}
   */
  async function handleGetRepoStatus(_request, url) {
    const didParam = url.searchParams.get('did');
    const did = await getDid();

    if (!did || (didParam && didParam !== did)) {
      return repoNotFound();
    }

    const commit = await actorStorage.getLatestCommit();
    if (!commit) {
      return repoNotFound();
    }

    const deactivated = await isDeactivated();
    return Response.json({
      did,
      active: !deactivated,
      status: deactivated ? 'deactivated' : 'active',
      rev: commit.rev,
    });
  }

  /**
   * com.atproto.sync.getRepo
   * @param {Request} _request
   * @param {URL} url
   * @returns {Promise<Response>}
   */
  async function handleGetRepo(_request, url) {
    const didParam = url.searchParams.get('did');
    const did = await getDid();

    if (!did || (didParam && didParam !== did)) {
      return repoNotFound();
    }

    const commit = await actorStorage.getLatestCommit();
    if (!commit) {
      return repoNotFound();
    }

    if (await blocksCollected()) {
      // Swept, the head may have moved on while this waited for the lock.
      const head = (await actorStorage.getLatestCommit()) ?? commit;
      return new Response(streamRepoCar(head.cid), {
        headers: { 'Content-Type': 'application/vnd.ipld.car' },
      });
    }

    let car;
    try {
      car = await buildFullRepoCar(commit.cid);
    } catch (err) {
      return Response.json(
        {
          error: 'InternalServerError',
          message:
            err instanceof Error ? err.message : 'Failed to collect blocks',
        },
        { status: 500 },
      );
    }

    return new Response(/** @type {BodyInit} */ (car), {
      headers: { 'Content-Type': 'application/vnd.ipld.car' },
    });
  }

  /**
   * The whole repo as one CAR: the commit, every MST node, and every record.
   * Serves com.atproto.sync.getRepo from a store that keeps every block it
   * was given, and is the repo half of a backup snapshot.
   * @param {string} commitCid
   * @returns {Promise<Uint8Array>}
   */
  async function buildFullRepoCar(commitCid) {
    return buildCarFile(commitCid, await repoBlocks(commitCid));
  }

  /**
   * Every block the commit reaches, by walking its tree.
   *
   * Throws if any of those blocks is absent. A caller must not answer with a
   * part of a repo, because a relay accepts it as the whole one.
   *
   * The walk waits on a timer after every BLOCKS_PER_WAIT blocks. A Durable
   * Object delivers no other request until its code waits on something other
   * than a promise already settled, and a whole repo is thousands of blocks.
   * @param {string} commitCid
   * @returns {Promise<Array<{cid: string, data: Uint8Array}>>}
   */
  async function repoBlocks(commitCid) {
    /** @type {Array<{cid: string, data: Uint8Array}>} */
    const blocks = [];
    const taken = new Set();

    /**
     * @param {string} cidStr
     * @returns {Promise<Uint8Array>}
     */
    const take = async (cidStr) => {
      const data = await actorStorage.getBlock(cidStr);
      if (!data) {
        throw new Error(`Missing block: ${cidStr}`);
      }
      if (!taken.has(cidStr)) {
        taken.add(cidStr);
        blocks.push({ cid: cidStr, data });
        if (blocks.length % BLOCKS_PER_WAIT === 0) {
          await new Promise((resolve) => setTimeout(resolve, 0));
        }
      }
      return data;
    };

    // An MST node keeps the left subtree in `l` and one entry per key in `e`.
    // Each entry has a record CID in `v` and a right subtree in `t`.
    //
    // A record is a leaf. This takes the block and does not read its fields, so
    // a record with a field named `data`, `l` or `e` cannot direct the walk at
    // something that is not a block.
    //
    // The walk needs no guard against a repeat visit. Every CID is the hash of
    // its own block, so the tree cannot hold a cycle, and one node has one
    // parent.
    /** @param {string} cidStr */
    const collectNode = async (cidStr) => {
      const node = cborDecode(await take(cidStr));
      if (node?.l instanceof Uint8Array) {
        await collectNode(cidToString(node.l));
      }
      for (const entry of Array.isArray(node?.e) ? node.e : []) {
        if (entry.v instanceof Uint8Array) {
          await take(cidToString(entry.v));
        }
        if (entry.t instanceof Uint8Array) {
          await collectNode(cidToString(entry.t));
        }
      }
    };

    const commit = cborDecode(await take(commitCid));
    if (commit?.data instanceof Uint8Array) {
      await collectNode(cidToString(commit.data));
    }
    return blocks;
  }

  /**
   * Whether the block table holds only what the current commit reaches, so
   * that it can be read as the repo; sweeping it first, once, if not.
   *
   * Each commit drops what it leaves unreachable, but blocks stored before
   * commits did, or left by a rebuilt tree or an imported CAR, stay until
   * this sweeps them. It holds the repo lock, so no commit stores a block it
   * has not counted, and waits on a timer between pages, as the walk does.
   * @returns {Promise<boolean>}
   */
  async function blocksCollected() {
    const store = actorStorage;
    if (
      !store.listBlocks ||
      !store.deleteBlocks ||
      !store.getBlocksCollected ||
      !store.setBlocksCollected
    ) {
      return false;
    }
    if (await store.getBlocksCollected()) return true;
    const listBlocks = store.listBlocks.bind(store);
    const deleteBlocks = store.deleteBlocks.bind(store);
    const setBlocksCollected = store.setBlocksCollected.bind(store);
    const sweep = async () => {
      // Another request may have swept while this one waited for the lock.
      if (await store.getBlocksCollected?.()) return true;
      const commit = await actorStorage.getLatestCommit();
      if (!commit) return false;
      const reached = new Set(
        (await repoBlocks(commit.cid)).map((block) => block.cid),
      );
      for (const record of await actorStorage.listAllRecords()) {
        reached.add(record.cid);
      }
      /** @type {string|null} */
      let cursor = null;
      do {
        const page = await listBlocks(cursor, BLOCKS_PER_WAIT);
        const unreached = page.blocks
          .map((block) => block.cid)
          .filter((cid) => !reached.has(cid));
        await deleteBlocks(unreached);
        cursor = page.cursor;
        await new Promise((resolve) => setTimeout(resolve, 0));
      } while (cursor);
      await setBlocksCollected(true);
      return true;
    };
    return ctx.withRepoLock ? ctx.withRepoLock(sweep) : sweep();
  }

  /**
   * The repo as a CAR read from the block table a page at a time, as the
   * reference PDS reads its own, and sent as it is read: the next page is
   * read only once the last has gone out, so other requests run between.
   *
   * A commit that lands meanwhile drops blocks the stream may not have sent
   * yet. The reference's stream does the same; a consumer finds the CAR does
   * not hold its root's tree and asks again.
   * @param {string} commitCid
   * @returns {ReadableStream<Uint8Array>}
   */
  function streamRepoCar(commitCid) {
    const listBlocks =
      /** @type {NonNullable<typeof actorStorage.listBlocks>} */ (
        actorStorage.listBlocks
      ).bind(actorStorage);
    /** @type {string|null} */
    let cursor = null;
    let started = false;
    return new ReadableStream({
      async pull(controller) {
        if (!started) {
          started = true;
          controller.enqueue(carHeader(commitCid));
          return;
        }
        const page = await listBlocks(cursor, BLOCKS_PER_WAIT);
        for (const block of page.blocks) controller.enqueue(carBlock(block));
        cursor = page.cursor;
        if (!cursor) controller.close();
      },
    });
  }

  /**
   * com.atproto.sync.getRecord (returns CAR with proof)
   * @param {Request} _request
   * @param {URL} url
   * @returns {Promise<Response>}
   */
  async function handleSyncGetRecord(_request, url) {
    const didParam = url.searchParams.get('did');
    const collection = url.searchParams.get('collection');
    const rkey = url.searchParams.get('rkey');

    if (!collection || !rkey) {
      return Response.json(
        { error: 'InvalidRequest', message: 'Missing required parameters' },
        { status: 400 },
      );
    }

    const did = await getDid();
    if (!did || (didParam && didParam !== did)) {
      return repoNotFound();
    }

    const uri = `at://${did}/${collection}/${rkey}`;
    const record = await actorStorage.getRecord(uri);

    if (!record) {
      return Response.json(
        { error: 'RecordNotFound', message: 'Record not found' },
        { status: 404 },
      );
    }

    const commit = await actorStorage.getLatestCommit();
    if (!commit) {
      return repoNotFound();
    }

    // Build CAR with commit, MST proof path, and record
    /** @type {Array<{cid: string, data: Uint8Array}>} */
    const blocks = [];
    const included = new Set();

    /** @param {string} cidStr */
    const addBlock = async (cidStr) => {
      if (included.has(cidStr)) return;
      included.add(cidStr);
      const data = await actorStorage.getBlock(cidStr);
      if (data) {
        blocks.push({ cid: cidStr, data });
      }
    };

    // Add commit block
    await addBlock(commit.cid);

    // Get commit data to find MST root
    const commitData = await actorStorage.getBlock(commit.cid);
    if (commitData) {
      const decoded = cborDecode(commitData);
      if (decoded.data) {
        const mstRootCid = cidToString(decoded.data);
        // The proof is the path from the root to the key: the nodes a
        // lookup reads.
        /** @type {string[]} */
        const path = [];
        /** @param {string} cid */
        const getBlock = (cid) => {
          path.push(cid);
          return actorStorage.getBlock(cid);
        };
        // A lookup only reads: it never calls putBlock.
        const reader = /** @type {import('../mst.js').MstStore} */ ({
          getBlock,
        });
        await mstGet(mstRootCid, `${collection}/${rkey}`, reader);
        for (const cid of path) {
          await addBlock(cid);
        }
      }
    }

    // Add record block
    await addBlock(record.cid);

    const car = buildCarFile(commit.cid, blocks);
    return new Response(/** @type {BodyInit} */ (car), {
      headers: { 'Content-Type': 'application/vnd.ipld.car' },
    });
  }

  /**
   * com.atproto.sync.getLatestCommit
   * @returns {Promise<Response>}
   */
  async function handleGetLatestCommit() {
    // The response does not use this value. The call fills the DID cache.
    const _did = await getDid();
    const commit = await actorStorage.getLatestCommit();

    if (!commit) {
      return repoNotFound();
    }

    return Response.json({
      cid: commit.cid,
      rev: commit.rev,
    });
  }

  return {
    buildFullRepoCar,
    routes: {
      '/xrpc/com.atproto.sync.listRepos': {
        handler: handleListRepos,
      },
      '/xrpc/com.atproto.sync.getRepoStatus': {
        handler: handleGetRepoStatus,
      },
      '/xrpc/com.atproto.sync.getRepo': {
        handler: handleGetRepo,
      },
      '/xrpc/com.atproto.sync.getRecord': {
        handler: handleSyncGetRecord,
      },
      '/xrpc/com.atproto.sync.getLatestCommit': {
        handler: handleGetLatestCommit,
      },
    },
  };
}