// @pdsjs/spaces/star - permissioned repos as STAR archives.
//
//   [ magic | len | meta ] [ 1 | len | key | len | val ] ... [ 0 ]
//
// A control byte comes before each pair (1) and ends the archive (0). Each
// `len` is an unsigned varint in its shortest form. A space repo is the
// `atproto-space-repo` profile: the metadata is the repo's commit, typed by
// `$type`, and each pair is a record path and the record's DAG-CBOR encoding,
// in strict lexicographic byte order of the path.
//
// See https://dasl.ing/star.html

import {
  cborDecode,
  cborEncodeDagCbor,
  cidToString,
  createCid,
  jsonToLex,
} from '@pdsjs/core/repo';
import { RepoCommit, verifyCommit } from './commit.js';
import { bytesEqual, verifyRecordMac } from './mac.js';
import { formatRecordPath, parseRecordPath } from './path.js';
import { isSpaceUri } from './uri.js';

export const STAR_MAGIC = new Uint8Array([0x2a, 0x6c, 0x01, 0x00]);
export const REPO_STAR_TYPE = 'atproto-space-repo';

/** The largest DAG-CBOR encoding a space record may have. */
export const MAX_RECORD_SIZE = 1024 * 1024;

const CONTROL_PAIR = 0x01;
const CONTROL_END = 0x00;
const MAX_VARINT_BYTES = 9;
const MAX_META_LENGTH = 64 * 1024;
const MAX_KEY_LENGTH = 4 * 1024;

export class RepoVerificationError extends Error {
  /** @param {string} message */
  constructor(message) {
    super(message);
    this.name = 'RepoVerificationError';
  }
}

/**
 * @typedef {Object} SerializedRecord
 * @property {string} collection
 * @property {string} rkey
 * @property {Uint8Array} bytes - DAG-CBOR encoded record
 */

/**
 * Lexicographic byte order: the first differing byte decides, and a prefix
 * sorts before any longer key it starts.
 * @param {Uint8Array} a
 * @param {Uint8Array} b
 * @returns {number}
 */
export function compareBytes(a, b) {
  const length = Math.min(a.length, b.length);
  for (let i = 0; i < length; i++) {
    if (a[i] !== b[i]) return a[i] - b[i];
  }
  return a.length - b.length;
}

/**
 * Serialize a repo: the commit as the metadata, then one path/record pair per
 * record. Blobs are excluded. Records may arrive in any order; the archive
 * holds them in path order.
 *
 * @param {import('./mac.js').Commit} commit
 * @param {Iterable<SerializedRecord>} records
 * @returns {Uint8Array}
 */
export function serializeRepo(commit, records) {
  const encoder = new TextEncoder();
  const pairs = [...records]
    .map((record) => ({
      key: encoder.encode(formatRecordPath(record.collection, record.rkey)),
      value: record.bytes,
    }))
    .sort((a, b) => compareBytes(a.key, b.key));
  for (let i = 1; i < pairs.length; i++) {
    if (compareBytes(pairs[i - 1].key, pairs[i].key) === 0) {
      throw new Error('duplicate record path');
    }
  }
  const meta = cborEncodeDagCbor({
    $type: REPO_STAR_TYPE,
    ver: commit.ver,
    repo: commit.repo,
    space: commit.space,
    rev: commit.rev,
    data: commit.data,
  });

  /** @type {Uint8Array[]} */
  const chunks = [STAR_MAGIC, encodeVarint(meta.length), meta];
  const pair = Uint8Array.of(CONTROL_PAIR);
  for (const { key, value } of pairs) {
    chunks.push(pair, encodeVarint(key.length), key);
    chunks.push(encodeVarint(value.length), value);
  }
  chunks.push(Uint8Array.of(CONTROL_END));
  let size = 0;
  for (const chunk of chunks) size += chunk.length;
  const out = new Uint8Array(size);
  let offset = 0;
  for (const chunk of chunks) {
    out.set(chunk, offset);
    offset += chunk.length;
  }
  return out;
}

/**
 * Read a STAR archive: its metadata, then its pairs.
 *
 * Lengths are checked before a frame is read, so an oversized frame fails
 * without being sliced out.
 *
 * @param {Uint8Array} bytes
 * @param {{maxValueLength?: number}} [opts]
 * @returns {{meta: Record<string, unknown>, pairs: Array<{key: Uint8Array, value: Uint8Array}>}}
 * @throws {RepoVerificationError}
 */
export function readStar(bytes, opts = {}) {
  const maxValueLength = opts.maxValueLength ?? Number.MAX_SAFE_INTEGER;
  const reader = { bytes, offset: 0 };
  if (!bytesEqual(bytes.subarray(0, STAR_MAGIC.length), STAR_MAGIC)) {
    throw new RepoVerificationError('invalid STAR magic bytes');
  }
  reader.offset = STAR_MAGIC.length;
  const metaBytes = readFrame(reader, MAX_META_LENGTH);

  /** @type {unknown} */
  let meta;
  try {
    meta = cborDecode(metaBytes);
  } catch {
    throw new RepoVerificationError('invalid metadata');
  }
  if (!isPlainObject(meta) || typeof meta.$type !== 'string') {
    throw new RepoVerificationError('metadata must be a map with a $type');
  }
  if (!bytesEqual(cborEncodeDagCbor(meta), metaBytes)) {
    throw new RepoVerificationError('metadata is not canonical DAG-CBOR');
  }

  const pairs = [];
  for (;;) {
    const control = bytes[reader.offset++];
    if (control === undefined) {
      throw new RepoVerificationError('unexpected end of archive');
    }
    if (control === CONTROL_END) break;
    if (control !== CONTROL_PAIR) {
      throw new RepoVerificationError(`invalid control byte: ${control}`);
    }
    const key = readFrame(reader, MAX_KEY_LENGTH);
    const value = readFrame(reader, maxValueLength);
    pairs.push({ key, value });
  }
  if (reader.offset !== bytes.length) {
    throw new RepoVerificationError('unexpected bytes after end of archive');
  }
  return { meta, pairs };
}

/**
 * @typedef {Object} VerifiedRecord
 * @property {string} collection
 * @property {string} rkey
 * @property {string} cid
 * @property {Record<string, unknown>} record
 */

/**
 * Verify a serialized repo and collect its records.
 *
 * The commit in the metadata is verified against the response's MAC, each
 * record is folded into a set hash, and the result must match the commit's
 * data. Only that match shows the repo arrived complete.
 *
 * @param {Uint8Array} star
 * @param {import('./commit.js').VerifyCommitParams} params
 * @returns {Promise<{commit: import('./mac.js').Commit, records: VerifiedRecord[]}>}
 * @throws {RepoVerificationError}
 */
export async function verifyRepoStar(star, params) {
  const { meta, pairs } = readStar(star, { maxValueLength: MAX_RECORD_SIZE });
  const commit = parseRepoMeta(meta);
  if (!(await verifyCommit(commit, params))) {
    throw new RepoVerificationError('commit failed verification');
  }

  const decoder = new TextDecoder();
  const repo = new RepoCommit();
  /** @type {VerifiedRecord[]} */
  const records = [];
  /** @type {Uint8Array|undefined} */
  let prevKey;
  for (const { key, value } of pairs) {
    const path = decoder.decode(key);
    /** @type {{collection: string, rkey: string}} */
    let parts;
    try {
      parts = parseRecordPath(path);
    } catch {
      throw new RepoVerificationError(`invalid record path: ${path}`);
    }
    if (prevKey !== undefined && compareBytes(key, prevKey) <= 0) {
      throw new RepoVerificationError(`records out of order at ${path}`);
    }
    prevKey = key;

    /** @type {unknown} */
    let record;
    try {
      record = cborDecode(value);
    } catch {
      throw new RepoVerificationError(`invalid record cbor at ${path}`);
    }
    if (!isPlainObject(record)) {
      throw new RepoVerificationError(`invalid record at ${path}`);
    }
    const cid = cidToString(await createCid(value));
    repo.add(parts.collection, parts.rkey, cid);
    records.push({ ...parts, cid, record });
  }

  if (!(await repo.matches(commit))) {
    throw new RepoVerificationError('records do not match the commit data');
  }
  return { commit, records };
}

/**
 * Verify a getRecord response against the MAC it carried: it must be the
 * record that was requested, the MAC must cover its uri and cid, and its value
 * must encode to that cid.
 *
 * @param {{uri: string, cid: string, value: unknown}} record
 * @param {{uri: string, sharedSecret: Uint8Array, mac: Uint8Array}} params
 * @returns {Promise<boolean>}
 */
export async function verifyRecord(record, params) {
  if (record.uri !== params.uri) return false;
  try {
    if (
      !(await verifyRecordMac(
        params.sharedSecret,
        record.uri,
        record.cid,
        params.mac,
      ))
    ) {
      return false;
    }
  } catch {
    return false;
  }
  const valueCid = cidToString(
    await createCid(cborEncodeDagCbor(jsonToLex(record.value))),
  );
  return valueCid === record.cid;
}

/**
 * @param {Record<string, unknown>} meta
 * @returns {import('./mac.js').Commit}
 */
function parseRepoMeta(meta) {
  const { $type, ver, repo, space, rev, data } = meta;
  if (
    $type !== REPO_STAR_TYPE ||
    typeof ver !== 'number' ||
    typeof repo !== 'string' ||
    !repo.startsWith('did:') ||
    typeof space !== 'string' ||
    !isSpaceUri(space) ||
    typeof rev !== 'string' ||
    !(data instanceof Uint8Array)
  ) {
    throw new RepoVerificationError('invalid repo metadata');
  }
  return { ver, repo, space, rev, data };
}

/**
 * @param {unknown} value
 * @returns {value is Record<string, unknown>}
 */
function isPlainObject(value) {
  return (
    typeof value === 'object' &&
    value !== null &&
    !Array.isArray(value) &&
    !(value instanceof Uint8Array)
  );
}

/**
 * @param {number} n
 * @returns {Uint8Array}
 */
export function encodeVarint(n) {
  const out = [];
  let v = n;
  while (v >= 0x80) {
    out.push((v % 0x80) | 0x80);
    v = Math.floor(v / 0x80);
  }
  out.push(v);
  return new Uint8Array(out);
}

/**
 * The next frame.
 * @param {{bytes: Uint8Array, offset: number}} reader
 * @param {number} maxLength
 * @returns {Uint8Array}
 */
function readFrame(reader, maxLength) {
  const length = readVarint(reader);
  if (length > maxLength) {
    throw new RepoVerificationError(
      `length ${length} exceeds limit of ${maxLength}`,
    );
  }
  if (reader.offset + length > reader.bytes.length) {
    throw new RepoVerificationError('unexpected end of archive');
  }
  const frame = reader.bytes.subarray(reader.offset, reader.offset + length);
  reader.offset += length;
  return frame;
}

/**
 * Unsigned LEB128, least significant group first. Arithmetic rather than
 * bitwise operators, since lengths can exceed 32 bits.
 * @param {{bytes: Uint8Array, offset: number}} reader
 * @returns {number}
 */
function readVarint(reader) {
  let value = 0;
  for (let i = 0; i < MAX_VARINT_BYTES; i++) {
    if (reader.offset >= reader.bytes.length) {
      throw new RepoVerificationError('unexpected end of archive');
    }
    const byte = reader.bytes[reader.offset++];
    value += (byte & 0x7f) * 2 ** (7 * i);
    if (byte < 0x80) {
      // A final zero byte only pads the encoding, which STAR forbids.
      if (byte === 0 && i > 0) {
        throw new RepoVerificationError('non-canonical varint');
      }
      return value;
    }
  }
  throw new RepoVerificationError('varint too long');
}