// @pdsjs/spaces/handlers/read - com.atproto.space read and sync endpoints.
//
// These are the repo-host role: an application syncing a space pulls each
// member's repo from its PDS through here.
//
// Two ways in: the hosted account's own session, or a space credential issued
// by the space's authority. The credential is what lets a syncer read a repo it
// does not own. It is signed by the authority, so it verifies without
// contacting them.
//
// A read that returns a commit or a record is MACed for the syncer that asked,
// under a secret derived from its Atproto-Space-Want-Mac key share and this
// account's KEM key. A credential read must ask. The account's own session may.

import { cborDecode, cborDecodeJson, findBlobRefs } from '@pdsjs/core/repo';
import { RepoCommit } from '../commit.js';
import { decapsulate } from '../kem.js';
import {
  COMMIT_MAC_TYPE,
  computeCommitMac,
  computeRecordMac,
  formatMacHeader,
  MAC_HEADER,
  parseWantMacHeader,
  RECORD_MAC_TYPE,
  SpaceMacError,
  WANT_MAC_HEADER,
} from '../mac.js';
import { serializeRepo } from '../star.js';
import { parseSpaceUri } from '../uri.js';
import { createReadAuthorizer } from './auth.js';

/**
 * @param {string} error
 * @param {string} message
 * @param {number} [status]
 * @returns {Response}
 */
function errorResponse(error, message, status = 400) {
  return Response.json({ error, message }, { status });
}

/**
 * @param {URL} url
 * @returns {{space: string, repo: string|null}|Response}
 */
function readTarget(url) {
  const space = url.searchParams.get('space');
  if (!space) return errorResponse('InvalidRequest', 'space is required');
  try {
    parseSpaceUri(space);
  } catch {
    return errorResponse('InvalidSpaceUri', `Not a space uri: ${space}`);
  }
  return { space, repo: url.searchParams.get('repo') };
}

/**
 * @param {URL} url
 * @param {string} name
 * @param {number} def
 * @param {number} max
 * @returns {number|Response}
 */
function readLimit(url, name, def, max) {
  const raw = url.searchParams.get(name);
  if (raw === null) return def;
  const value = Number(raw);
  if (!Number.isInteger(value) || value < 1 || value > max) {
    return errorResponse(
      'InvalidRequest',
      `${name} must be an integer between 1 and ${max}`,
    );
  }
  return value;
}

/**
 * The storage sort key a listRecords cursor names. A cursor is a record's
 * AT-URI, as the reference gives it. Every row here shares `prefix`, so a
 * cursor outside it puts either every row or no row past it.
 *
 * @param {string|null} cursor
 * @param {string} prefix - `<space>/<did>/`
 * @param {boolean} reverse
 * @returns {string|null|false} false when no row is past the cursor
 */
function storageCursor(cursor, prefix, reverse) {
  if (!cursor) return null;
  if (cursor.startsWith(prefix)) return cursor.slice(prefix.length);
  return cursor < prefix === reverse ? null : false;
}

/**
 * atproto's JSON representation of a byte string.
 * @param {Uint8Array} bytes
 * @returns {{$bytes: string}}
 */
function toJsonBytes(bytes) {
  let binary = '';
  for (const b of bytes) binary += String.fromCharCode(b);
  return { $bytes: btoa(binary).replace(/=+$/, '') };
}

/**
 * @param {import('../mac.js').Commit} commit
 * @returns {Object}
 */
function toJsonCommit(commit) {
  return {
    ver: commit.ver,
    repo: commit.repo,
    space: commit.space,
    rev: commit.rev,
    data: toJsonBytes(commit.data),
  };
}

/**
 * A commit over the repo's current state. Undefined when the repo has never
 * been written to.
 *
 * @param {import('@pdsjs/core/ports').SpaceStoragePort} spaceStorage
 * @param {string} space
 * @param {string} repo
 * @returns {Promise<import('../mac.js').Commit|undefined>}
 */
async function buildCommit(spaceStorage, space, repo) {
  const state = await spaceStorage.getSpaceRepo(space);
  if (!state?.setHash || !state.rev) return undefined;
  return RepoCommit.fromState(state.setHash).toCommit({
    space,
    repo,
    rev: state.rev,
  });
}

/**
 * Append a value to a response header that may already hold others.
 * @param {Headers} headers
 * @param {string} name
 * @param {string} value
 */
function appendHeader(headers, name, value) {
  const current = headers.get(name);
  headers.set(name, current ? `${current}, ${value}` : value);
}

/**
 * Every blob CID a space's records reference, paired with the rev of the commit
 * that wrote the referencing record.
 *
 * Read from the records themselves rather than from a link table. A space repo
 * on a personal server is small, and a table that can drift is the one thing a
 * blob's reachability must not rest on.
 *
 * @param {import('@pdsjs/core/ports').SpaceStoragePort} spaceStorage
 * @param {string} space
 * @returns {Promise<Map<string, string>>} blob CID -> earliest referencing rev
 */
async function spaceBlobRefs(spaceStorage, space) {
  /** @type {Map<string, string>} */
  const refs = new Map();
  const paths = await spaceStorage.listAllSpaceRecords(space);
  for (const path of paths) {
    const row = await spaceStorage.getSpaceRecord(
      space,
      path.collection,
      path.rkey,
    );
    if (!row) continue;
    for (const cid of findBlobRefs(cborDecode(row.value))) {
      const seen = refs.get(cid);
      if (seen === undefined || row.repoRev < seen) refs.set(cid, row.repoRev);
    }
  }
  return refs;
}

/**
 * @param {Object} ctx
 * @param {import('@pdsjs/core/ports').SpaceStoragePort} ctx.spaceStorage
 * @param {import('@pdsjs/core/ports').BlobPort} ctx.blobs
 * @param {() => Promise<string|null>} ctx.getDid
 * @param {() => Promise<{privateKey: Uint8Array, didKey: string}|null>} ctx.getKemKey
 * @param {import('@pdsjs/core/ports').DidResolverPort} ctx.resolveDid
 * @param {import('@pdsjs/core/ports').SignatureVerifierPort} ctx.verifier
 * @returns {import('@pdsjs/core/pds').Routes}
 */
export function createReadRoutes(ctx) {
  const { spaceStorage, blobs, getDid, getKemKey, resolveDid, verifier } = ctx;

  const authorizeRead = createReadAuthorizer({
    spaceStorage,
    getDid,
    resolveDid,
    verifier,
  });

  /**
   * The repo a request targets. `repo` is optional on the wire; this PDS hosts
   * exactly one account, so anything else is a miss rather than a proxy.
   * @param {string|null} repo
   * @returns {Promise<string|Response>}
   */
  async function resolveRepo(repo) {
    const did = await getDid();
    if (!did) return errorResponse('RepoNotFound', 'Server not initialised');
    if (repo && repo !== did) {
      return errorResponse('RepoNotFound', `Not hosted here: ${repo}`, 404);
    }
    return did;
  }

  /**
   * The secret to MAC this response under, undefined when the account's own
   * session reads without asking for one, or the refusal to return.
   *
   * @param {Request} request
   * @param {{did: string}|null} auth
   * @param {import('../mac.js').MacType} type
   * @returns {Promise<Uint8Array|undefined|Response>}
   */
  async function macSecret(request, auth, type) {
    const header = request.headers.get(WANT_MAC_HEADER);
    const hosted = await getDid();
    const viaSession = Boolean(auth && auth.did === hosted);
    if (header === null && viaSession) return undefined;
    /** @type {string} */
    let keyShare;
    try {
      keyShare = parseWantMacHeader(header, type);
    } catch (err) {
      if (err instanceof SpaceMacError) {
        return errorResponse('BadSpaceMac', err.message);
      }
      throw err;
    }
    const kemKey = await getKemKey();
    if (!kemKey) {
      return errorResponse(
        'InternalServerError',
        'This account has no KEM key yet',
        500,
      );
    }
    try {
      return await decapsulate(kemKey.privateKey, keyShare);
    } catch (err) {
      if (err instanceof SpaceMacError) {
        return errorResponse('BadSpaceMac', err.message);
      }
      throw err;
    }
  }

  /**
   * A response that may carry a MAC. It varies on the request header, and a
   * browser app may read the MAC from a cross-origin response.
   *
   * @param {BodyInit|null} body
   * @param {Record<string, string>} headers
   * @param {string|null} mac - the Atproto-Space-Mac value, when there is one
   * @returns {Response}
   */
  function macResponse(body, headers, mac) {
    const response = new Response(body, { headers });
    appendHeader(response.headers, 'Vary', WANT_MAC_HEADER);
    if (mac) {
      response.headers.set(MAC_HEADER, mac);
      appendHeader(
        response.headers,
        'Access-Control-Expose-Headers',
        MAC_HEADER,
      );
    }
    return response;
  }

  /**
   * @param {unknown} body
   * @returns {[string, Record<string, string>]}
   */
  function json(body) {
    return [JSON.stringify(body), { 'content-type': 'application/json' }];
  }

  /** @type {import('@pdsjs/core/pds').Route} */
  const latestCommitRoute = {
    auth: 'optional',
    handler: async (request, url, auth) => {
      const target = readTarget(url);
      if (target instanceof Response) return target;
      const denied = await authorizeRead(request, target.space, auth);
      if (denied) return denied;
      const did = await resolveRepo(target.repo);
      if (did instanceof Response) return did;
      const secret = await macSecret(request, auth, COMMIT_MAC_TYPE);
      if (secret instanceof Response) return secret;

      const commit = await buildCommit(spaceStorage, target.space, did);
      if (!commit) return macResponse(...json({}), null);
      const mac = secret
        ? formatMacHeader(
            COMMIT_MAC_TYPE,
            await computeCommitMac(secret, commit),
          )
        : null;
      return macResponse(...json({ commit: toJsonCommit(commit) }), mac);
    },
  };

  return {
    '/xrpc/com.atproto.space.getRecord': {
      auth: 'optional',
      handler: async (request, url, auth) => {
        const target = readTarget(url);
        if (target instanceof Response) return target;
        const denied = await authorizeRead(request, target.space, auth);
        if (denied) return denied;
        const did = await resolveRepo(target.repo);
        if (did instanceof Response) return did;

        const collection = url.searchParams.get('collection');
        const rkey = url.searchParams.get('rkey');
        if (!collection || !rkey) {
          return errorResponse(
            'InvalidRequest',
            'collection and rkey are required',
          );
        }
        const secret = await macSecret(request, auth, RECORD_MAC_TYPE);
        if (secret instanceof Response) return secret;

        const row = await spaceStorage.getSpaceRecord(
          target.space,
          collection,
          rkey,
        );
        if (!row) {
          return errorResponse(
            'RecordNotFound',
            `Record not found: ${collection}/${rkey}`,
            404,
          );
        }
        const uri = `${target.space}/${did}/${collection}/${rkey}`;
        const mac = secret
          ? formatMacHeader(
              RECORD_MAC_TYPE,
              await computeRecordMac(secret, uri, row.cid),
            )
          : null;
        return macResponse(
          ...json({ uri, cid: row.cid, value: cborDecodeJson(row.value) }),
          mac,
        );
      },
    },

    '/xrpc/com.atproto.space.listRecords': {
      auth: 'optional',
      handler: async (request, url, auth) => {
        const target = readTarget(url);
        if (target instanceof Response) return target;
        const denied = await authorizeRead(request, target.space, auth);
        if (denied) return denied;
        const did = await resolveRepo(target.repo);
        if (did instanceof Response) return did;

        const limit = readLimit(url, 'limit', 50, 1000);
        if (limit instanceof Response) return limit;

        const prefix = `${target.space}/${did}/`;
        const reverse = url.searchParams.get('reverse') === 'true';
        const after = storageCursor(
          url.searchParams.get('cursor'),
          prefix,
          reverse,
        );
        if (after === false) return Response.json({ records: [] });

        const { records, cursor } = await spaceStorage.listSpaceRecords(
          target.space,
          {
            collection: url.searchParams.get('collection'),
            cursor: after,
            limit,
            reverse,
            excludeValues: url.searchParams.get('excludeValues') === 'true',
          },
        );
        return Response.json({
          ...(cursor ? { cursor: prefix + cursor } : {}),
          records: records.map((r) => ({
            collection: r.collection,
            rkey: r.rkey,
            cid: r.cid,
            ...(r.value === undefined
              ? {}
              : { value: cborDecodeJson(r.value) }),
          })),
        });
      },
    },

    '/xrpc/com.atproto.space.listRepoOps': {
      auth: 'optional',
      handler: async (request, url, auth) => {
        const target = readTarget(url);
        if (target instanceof Response) return target;
        const denied = await authorizeRead(request, target.space, auth);
        if (denied) return denied;
        const did = await resolveRepo(target.repo);
        if (did instanceof Response) return did;

        const limit = readLimit(url, 'limit', 100, 1000);
        if (limit instanceof Response) return limit;
        const excludeValues = url.searchParams.get('excludeValues') === 'true';
        const secret = await macSecret(request, auth, COMMIT_MAC_TYPE);
        if (secret instanceof Response) return secret;

        const ops = await spaceStorage.listSpaceOps(
          target.space,
          url.searchParams.get('since'),
          limit,
        );

        // Inline each op's record value so a syncer advances without a
        // getRecord round-trip per write. A value is omitted for deletes, when
        // excludeValues is set, and when a later op superseded it — the stored
        // record is the current one, so it is only this op's value if the CIDs
        // still agree.
        const entries = [];
        for (const op of ops) {
          /** @type {Record<string, unknown>} */
          const entry = {
            rev: op.rev,
            collection: op.collection,
            rkey: op.rkey,
            cid: op.cid,
            prev: op.prev,
          };
          if (!excludeValues && op.cid) {
            const row = await spaceStorage.getSpaceRecord(
              target.space,
              op.collection,
              op.rkey,
            );
            if (row && row.cid === op.cid) {
              entry.value = cborDecodeJson(row.value);
            }
          }
          entries.push(entry);
        }

        /** @type {Record<string, unknown>} */
        const body = { ops: entries };
        /** @type {string|null} */
        let mac = null;
        // The commit describes the head, so it is only meaningful once the
        // response has reached it. A full page may have more behind it. A
        // page without a commit has nothing to MAC: its ops are checked once
        // the final page's commit is compared against the syncer's set hash.
        if (ops.length < limit) {
          const commit = await buildCommit(spaceStorage, target.space, did);
          if (commit) {
            body.commit = toJsonCommit(commit);
            if (secret) {
              mac = formatMacHeader(
                COMMIT_MAC_TYPE,
                await computeCommitMac(secret, commit),
              );
            }
          }
        } else {
          body.cursor = ops[ops.length - 1].rev;
        }
        return macResponse(...json(body), mac);
      },
    },

    '/xrpc/com.atproto.space.getLatestCommit': latestCommitRoute,

    // The same handler under the name this method shipped as before the draft
    // renamed it. Both are in use, and a client that knows only one reads the
    // whole feature as missing.
    '/xrpc/com.atproto.space.getRepoState': latestCommitRoute,

    // Blobs are uploaded through com.atproto.repo.uploadBlob and referenced
    // from a space record, so a client writing blob-bearing records needs a
    // blob permission alongside its space permission.
    '/xrpc/com.atproto.space.listBlobs': {
      auth: 'optional',
      handler: async (request, url, auth) => {
        const target = readTarget(url);
        if (target instanceof Response) return target;
        const denied = await authorizeRead(request, target.space, auth);
        if (denied) return denied;
        const did = await resolveRepo(target.repo);
        if (did instanceof Response) return did;

        const limit = readLimit(url, 'limit', 500, 1000);
        if (limit instanceof Response) return limit;
        const since = url.searchParams.get('since');
        const cursor = url.searchParams.get('cursor');

        const refs = await spaceBlobRefs(spaceStorage, target.space);
        const cids = [...refs]
          .filter(([, rev]) => (since ? rev > since : true))
          .map(([cid]) => cid)
          .filter((cid) => (cursor ? cid > cursor : true))
          .sort();
        const page = cids.slice(0, limit);

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

    // Scoped to the space: a credential admits its holder to that space, not to
    // the account's whole blob store. The reference serves any blob the account
    // holds once the space read passes.
    '/xrpc/com.atproto.space.getBlob': {
      auth: 'optional',
      handler: async (request, url, auth) => {
        const target = readTarget(url);
        if (target instanceof Response) return target;
        const denied = await authorizeRead(request, target.space, auth);
        if (denied) return denied;
        const did = await resolveRepo(target.repo);
        if (did instanceof Response) return did;

        const cid = url.searchParams.get('cid');
        if (!cid) return errorResponse('InvalidRequest', 'cid is required');

        const refs = await spaceBlobRefs(spaceStorage, target.space);
        if (!refs.has(cid)) {
          return errorResponse(
            'BlobNotFound',
            `No record in ${target.space} references ${cid}`,
            404,
          );
        }

        const blob = await blobs.get(did, cid);
        if (!blob) {
          return errorResponse('BlobNotFound', 'Blob not found', 404);
        }
        return new Response(/** @type {BodyInit} */ (blob.data), {
          headers: {
            'Content-Type': blob.mimeType,
            'X-Content-Type-Options': 'nosniff',
            // Force download and block execution so a hostile blob cannot run
            // in the PDS origin.
            'Content-Disposition': `attachment; filename="${cid}"`,
            'Content-Security-Policy': "default-src 'none'; sandbox",
          },
        });
      },
    },

    '/xrpc/com.atproto.space.getRepo': {
      auth: 'optional',
      handler: async (request, url, auth) => {
        const target = readTarget(url);
        if (target instanceof Response) return target;
        const denied = await authorizeRead(request, target.space, auth);
        if (denied) return denied;
        const did = await resolveRepo(target.repo);
        if (did instanceof Response) return did;
        const secret = await macSecret(request, auth, COMMIT_MAC_TYPE);
        if (secret instanceof Response) return secret;

        const commit = await buildCommit(spaceStorage, target.space, did);
        if (!commit) {
          return errorResponse(
            'RepoNotFound',
            `Could not find repo for space: ${target.space}`,
            404,
          );
        }

        const paths = await spaceStorage.listAllSpaceRecords(target.space);
        /** @type {import('../star.js').SerializedRecord[]} */
        const records = [];
        for (const p of paths) {
          const row = await spaceStorage.getSpaceRecord(
            target.space,
            p.collection,
            p.rkey,
          );
          if (row) {
            records.push({
              collection: p.collection,
              rkey: p.rkey,
              bytes: row.value,
            });
          }
        }

        const star = serializeRepo(commit, records);
        // Copy into a plain ArrayBuffer: a Uint8Array view is not a BodyInit
        // per the DOM lib types, and `.buffer` widens to ArrayBufferLike, which
        // includes SharedArrayBuffer.
        const body = new ArrayBuffer(star.byteLength);
        new Uint8Array(body).set(star);
        const mac = secret
          ? formatMacHeader(
              COMMIT_MAC_TYPE,
              await computeCommitMac(secret, commit),
            )
          : null;
        return macResponse(
          body,
          { 'content-type': 'application/octet-stream' },
          mac,
        );
      },
    },
  };
}