// @pdsjs/core/handlers/migration - The account-migration flow and the active
// state it pivots on: checkAccountStatus, importRepo and listMissingBlobs for
// an account arriving here, activateAccount and deactivateAccount for the
// switch-over on either end.

import { loadRepositoryFromCar } from '../loader.js';
import { ScopePermissions } from '../scope.js';

/**
 * These are structural types. Thus a member does not name the module that owns
 * the function you give it.
 *
 * @typedef {Object} MigrationContext
 * @property {import('../ports.js').ActorStoragePort} actorStorage
 * @property {boolean} readOnly
 * @property {() => Promise<string|null>} getDid
 * @property {() => Promise<import('../crypto.js').SigningKey|null>} getSigningKey
 * @property {() => Response} readOnlyError
 * @property {(type: 'account'|'identity'|'sync', body: Object) => Promise<void>} emitEvent
 * @property {(commit: {cid: string, rev: string}) => Promise<void>} emitSyncEvent
 * @property {(did: string) => Promise<void>} notifyRelay
 * @property {() => Promise<Array<{cid: string}>>} collectRepoBlocks
 * @property {() => Promise<Array<{cid: string, recordUri: string}>>} collectReferencedBlobs
 * @property {() => Promise<number>} relinkBlobRecords
 */

/**
 * @param {MigrationContext} ctx
 * @returns {{
 *   isDeactivated: () => Promise<boolean>,
 *   checkAccountActive: () => Promise<Response|null>,
 *   setAccountActive: (active: boolean) => Promise<void>,
 *   routes: import('../pds.js').Routes,
 * }}
 */
export function createMigrationHandlers(ctx) {
  const {
    actorStorage,
    readOnly,
    getDid,
    getSigningKey,
    readOnlyError,
    emitEvent,
    emitSyncEvent,
    notifyRelay,
    collectRepoBlocks,
    collectReferencedBlobs,
    relinkBlobRecords,
  } = ctx;

  /**
   * com.atproto.server.checkAccountStatus
   * @param {Request} _request
   * @param {URL} _url
   * @param {import('../pds.js').Auth|null} auth
   * @returns {Promise<Response>}
   */
  async function handleCheckAccountStatus(_request, _url, auth) {
    if (!auth) {
      return Response.json(
        { error: 'AuthenticationRequired', message: 'Authentication required' },
        { status: 401 },
      );
    }

    const did = await getDid();
    if (!did) {
      return Response.json(
        { error: 'InvalidRequest', message: 'PDS not initialized' },
        { status: 400 },
      );
    }

    const commit = await actorStorage.getLatestCommit();
    const blocks = await collectRepoBlocks();
    const records = await actorStorage.listAllRecords();
    const referenced = await collectReferencedBlobs();
    const expected = new Set(referenced.map((r) => r.cid));

    let imported = 0;
    for (const cid of expected) {
      if (await actorStorage.getBlob(cid)) imported++;
    }

    return Response.json({
      activated: !(await isDeactivated()),
      validDid: !!(await getSigningKey()),
      repoCommit: commit?.cid || '',
      repoRev: commit?.rev || '',
      repoBlocks: blocks.length,
      indexedRecords: records.length,
      privateStateValues: 0,
      expectedBlobs: expected.size,
      importedBlobs: imported,
    });
  }

  /**
   * com.atproto.repo.listMissingBlobs
   * @param {Request} _request
   * @param {URL} url
   * @param {import('../pds.js').Auth|null} auth
   * @returns {Promise<Response>}
   */
  async function handleListMissingBlobs(_request, url, auth) {
    if (!auth) {
      return Response.json(
        { error: 'AuthenticationRequired', message: 'Authentication required' },
        { status: 401 },
      );
    }

    const limit = Math.min(
      Number.parseInt(url.searchParams.get('limit') || '500', 10) || 500,
      1000,
    );
    const cursor = url.searchParams.get('cursor');

    const referenced = await collectReferencedBlobs();
    /** @type {Array<{cid: string, recordUri: string}>} */
    const missing = [];
    const checked = new Set();

    for (const ref of referenced) {
      if (checked.has(ref.cid)) continue;
      checked.add(ref.cid);
      if (cursor && ref.cid <= cursor) continue;
      if (!(await actorStorage.getBlob(ref.cid))) {
        missing.push(ref);
      }
    }

    missing.sort((a, b) => (a.cid < b.cid ? -1 : a.cid > b.cid ? 1 : 0));
    const page = missing.slice(0, limit);

    return Response.json({
      blobs: page,
      ...(page.length === limit ? { cursor: page[page.length - 1].cid } : {}),
    });
  }

  /**
   * com.atproto.repo.importRepo
   * @param {Request} request
   * @param {URL} _url
   * @param {import('../pds.js').Auth|null} auth
   * @returns {Promise<Response>}
   */
  async function handleImportRepo(request, _url, auth) {
    if (readOnly) return readOnlyError();

    if (!auth) {
      return Response.json(
        { error: 'AuthenticationRequired', message: 'Authentication required' },
        { status: 401 },
      );
    }

    const permissions = new ScopePermissions(auth.scope);
    if (!permissions.hasFullAccess()) {
      return Response.json(
        { error: 'InvalidScope', message: 'Full access required' },
        { status: 403 },
      );
    }

    const did = await getDid();
    if (!did) {
      return Response.json(
        { error: 'InvalidRequest', message: 'PDS not initialized' },
        { status: 400 },
      );
    }

    const car = new Uint8Array(await request.arrayBuffer());
    if (car.length === 0) {
      return Response.json(
        { error: 'InvalidRequest', message: 'Empty CAR body' },
        { status: 400 },
      );
    }

    let result;
    try {
      result = await loadRepositoryFromCar(car, actorStorage);
    } catch (err) {
      return Response.json(
        { error: 'InvalidRequest', message: `Invalid CAR: ${err.message}` },
        { status: 400 },
      );
    }

    if (result.did !== did) {
      return Response.json(
        {
          error: 'InvalidRequest',
          message: `CAR belongs to ${result.did}, not ${did}`,
        },
        { status: 400 },
      );
    }

    // Register blob → record links for the imported records. loadRepositoryFromCar
    // stores records and blocks but not these links; without them the orphan
    // cleanup treats every blob the imported repo references as unreferenced and
    // reaps it, silently destroying a migrated account's blobs.
    await relinkBlobRecords();

    // Imported state is non-inductive for consumers: announce it
    const commit = await actorStorage.getLatestCommit();
    if (commit) await emitSyncEvent(commit);

    return Response.json({});
  }

  /**
   * Return whether the account is deactivated.
   * @returns {Promise<boolean>}
   */
  async function isDeactivated() {
    return (await actorStorage.getAccountStatus()) === 'deactivated';
  }

  /**
   * Reject repo/blob writes while the account is deactivated.
   * @returns {Promise<Response|null>} 409 response when deactivated
   */
  async function checkAccountActive() {
    if (!(await isDeactivated())) return null;
    return Response.json(
      {
        error: 'AccountDeactivated',
        message: 'Account is deactivated',
      },
      { status: 409 },
    );
  }

  /**
   * com.atproto.server.deactivateAccount
   * @param {Request} request
   * @param {URL} _url
   * @param {import('../pds.js').Auth|null} auth
   * @returns {Promise<Response>}
   */
  async function handleDeactivateAccount(request, _url, auth) {
    // Consume the body: responding without draining the request stream
    // raises an uncaught error on Workers runtimes
    await request.text().catch(() => '');

    if (!auth) {
      return Response.json(
        { error: 'AuthenticationRequired', message: 'Authentication required' },
        { status: 401 },
      );
    }
    const permissions = new ScopePermissions(auth.scope);
    if (!permissions.hasFullAccess()) {
      return Response.json(
        { error: 'InvalidScope', message: 'Full access required' },
        { status: 403 },
      );
    }

    // setAccountActive emits the #account (inactive) event, telling the network
    // the account is no longer active here (e.g. it moved away).
    await setAccountActive(false);
    return Response.json({});
  }

  /**
   * Set the account's active state and announce it on the firehose. Shared by
   * the XRPC handlers and the account page, so the network sees the same
   * event whichever triggered it.
   * @param {boolean} active
   * @returns {Promise<void>}
   */
  async function setAccountActive(active) {
    await actorStorage.setAccountStatus(active ? 'active' : 'deactivated');
    await emitEvent(
      'account',
      active ? { active: true } : { active: false, status: 'deactivated' },
    );
  }

  /**
   * com.atproto.server.activateAccount
   * @param {Request} request
   * @param {URL} _url
   * @param {import('../pds.js').Auth|null} auth
   * @returns {Promise<Response>}
   */
  async function handleActivateAccount(request, _url, auth) {
    // Consume the body: responding without draining the request stream
    // raises an uncaught error on Workers runtimes
    await request.text().catch(() => '');

    if (!auth) {
      return Response.json(
        { error: 'AuthenticationRequired', message: 'Authentication required' },
        { status: 401 },
      );
    }
    const permissions = new ScopePermissions(auth.scope);
    if (!permissions.hasFullAccess()) {
      return Response.json(
        { error: 'InvalidScope', message: 'Full access required' },
        { status: 403 },
      );
    }

    // setAccountActive emits the #account (active) event. On top of that we
    // re-announce identity: activation is the last step of an inbound migration,
    // right after the DID was repointed to this server with a fresh signing key,
    // so consumers must re-resolve the DID. Without the #identity event, AppViews
    // keep verifying this account's service tokens against the *old* PDS's key
    // and reject them ("could not verify jwt signature").
    await setAccountActive(true);
    // The crawl request goes first. A relay that has never heard of this server
    // has no subscription, and an event announced before it asks to crawl
    // reaches nobody. maybeAnnounceIdentity repeats the announcement each hour
    // until the relay reports that it reads this server.
    const did = await getDid();
    if (did) await notifyRelay(did);
    const handle = await actorStorage.getHandle();
    await emitEvent('identity', { handle: handle || null });
    return Response.json({});
  }

  return {
    isDeactivated,
    checkAccountActive,
    setAccountActive,
    routes: {
      '/xrpc/com.atproto.server.checkAccountStatus': {
        auth: 'required',
        handler: handleCheckAccountStatus,
      },
      '/xrpc/com.atproto.repo.importRepo': {
        method: 'POST',
        auth: 'required',
        handler: handleImportRepo,
      },
      '/xrpc/com.atproto.repo.listMissingBlobs': {
        auth: 'required',
        handler: handleListMissingBlobs,
      },
      '/xrpc/com.atproto.server.activateAccount': {
        method: 'POST',
        auth: 'required',
        handler: handleActivateAccount,
      },
      '/xrpc/com.atproto.server.deactivateAccount': {
        method: 'POST',
        auth: 'required',
        handler: handleDeactivateAccount,
      },
    },
  };
}