// Commit events must carry a diff since the previous repo state, not the
// entire MST. See com.atproto.sync.subscribeRepos #commit `blocks`.

import { PersonalDataServer } from '@pdsjs/core';
import { parseCarFile } from '@pdsjs/core/car';
import { mstInvert } from '@pdsjs/core/mst';
import {
  cborDecode,
  cborDecodeFirst,
  cidToString,
  getKeyDepth,
} from '@pdsjs/core/repo';
import { beforeEach, describe, expect, test } from 'vitest';
import {
  createMemorySharedStorage,
  createMemoryStorage,
} from './helpers/memory-storage.js';

const DID = 'did:plc:committestaaaaaaaaaaaaaa';
const PASSWORD = 'test-password';

/** Count the blocks carried in a stored commit event's CAR. */
function commitBlockCount(
  /** @type {any} */ storage,
  /** @type {number} */ index,
) {
  const evt = cborDecode(storage._events[index].evt);
  return parseCarFile(evt.blocks).blocks.size;
}

describe('commit event blocks', () => {
  /** @type {ReturnType<typeof createMemoryStorage>} */
  let storage;
  /** @type {PersonalDataServer} */
  let pds;
  /** @type {string} */
  let token;

  beforeEach(async () => {
    storage = createMemoryStorage();
    pds = new PersonalDataServer({
      actorStorage: /** @type {any} */ (storage),
      sharedStorage: /** @type {any} */ (createMemorySharedStorage()),
      blobs: {
        async get() {
          return null;
        },
        async put() {},
        async delete() {},
      },
      jwtSecret: 'test-secret',
      hostname: 'pds.example.com',
      password: PASSWORD,
    });

    await pds.fetch(
      new Request(`https://pds.example.com/init?did=${DID}`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          did: DID,
          privateKey: Buffer.from(
            crypto.getRandomValues(new Uint8Array(32)),
          ).toString('hex'),
          handle: 'commit.test',
          password: PASSWORD,
        }),
      }),
    );

    const sessionRes = await pds.fetch(
      new Request(
        'https://pds.example.com/xrpc/com.atproto.server.createSession',
        {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({ identifier: DID, password: PASSWORD }),
        },
      ),
    );
    const sessionBody = await sessionRes.json();
    expect(sessionBody.accessJwt, 'session setup failed').toBeTruthy();
    token = sessionBody.accessJwt;
  });

  /** @param {number} i */
  async function createRecord(i) {
    const res = await pds.fetch(
      new Request(
        'https://pds.example.com/xrpc/com.atproto.repo.createRecord',
        {
          method: 'POST',
          headers: {
            'Content-Type': 'application/json',
            Authorization: `Bearer ${token}`,
          },
          body: JSON.stringify({
            repo: DID,
            collection: 'app.bsky.feed.post',
            record: { text: `post ${i}`, createdAt: '2026-01-01T00:00:00Z' },
          }),
        },
      ),
    );
    expect(res.status).toBe(200);
  }

  test('carries only blocks the commit created, not the whole MST', async () => {
    for (let i = 0; i < 120; i++) await createRecord(i);

    const commitEvents = storage._events.filter((e) => e.commit_cid !== '');
    expect(commitEvents.length).toBe(120);

    // A single-record write carries the commit, the new record, the MST
    // nodes along one root-to-key path, and the nodes beside that path that
    // undoing the write reads. So every commit stays bounded by tree depth
    // no matter how many records exist. Shipping the whole tree instead
    // would grow with the record count (~38 blocks by the 120th commit here).
    // Record keys are TIDs with a random component, so the largest commit
    // varies between runs: 8 to 16 blocks over 100 runs.
    const counts = commitEvents.map((evt) =>
      commitBlockCount(storage, storage._events.indexOf(evt)),
    );
    expect(Math.max(...counts)).toBeLessThan(24);
  });

  test('includes the commit block and the newly written record', async () => {
    await createRecord(0);
    await createRecord(1);

    const commitEvents = storage._events.filter((e) => e.commit_cid !== '');
    const lastIdx = storage._events.indexOf(
      commitEvents[commitEvents.length - 1],
    );
    const evt = cborDecode(storage._events[lastIdx].evt);
    const { roots, blocks } = parseCarFile(evt.blocks);

    // Commit block is the CAR root and is present
    expect(roots.length).toBe(1);
    expect(blocks.has(roots[0])).toBe(true);

    // The record written by this commit travels with it
    const recordCid = evt.ops[0].cid.$link;
    expect(blocks.has(recordCid)).toBe(true);
  });

  // Sync 1.1 consumers invert each commit against prevData, which needs the
  // CID an update or delete replaced. Without it jetstream's verifier fails
  // the commit and never delivers the op.
  describe('op prev', () => {
    /**
     * @param {string} method
     * @param {object} body
     */
    async function xrpc(method, body) {
      const res = await pds.fetch(
        new Request(`https://pds.example.com/xrpc/com.atproto.repo.${method}`, {
          method: 'POST',
          headers: {
            'Content-Type': 'application/json',
            Authorization: `Bearer ${token}`,
          },
          body: JSON.stringify({ repo: DID, ...body }),
        }),
      );
      expect(res.status).toBe(200);
      return res.json();
    }

    /** @param {string} title */
    function putGallery(title) {
      return xrpc('putRecord', {
        collection: 'social.grain.gallery',
        rkey: 'g1',
        record: { title, createdAt: '2026-01-01T00:00:00Z' },
      });
    }

    /** The ops of the most recent commit event, as stored. */
    function lastOps() {
      const commits = storage._events.filter((e) => e.commit_cid !== '');
      return cborDecode(commits[commits.length - 1].evt).ops;
    }

    test('create carries no prev', async () => {
      await putGallery('original');
      expect(lastOps()[0]).not.toHaveProperty('prev');
    });

    test('update carries the CID it replaced', async () => {
      const original = await putGallery('original');
      const edited = await putGallery('edited');

      const [op] = lastOps();
      expect(op.action).toBe('update');
      expect(op.cid.$link).toBe(edited.cid);
      expect(op.prev.$link).toBe(original.cid);
    });

    test('delete carries the CID it removed', async () => {
      const original = await putGallery('original');
      await xrpc('deleteRecord', {
        collection: 'social.grain.gallery',
        rkey: 'g1',
      });

      const [op] = lastOps();
      expect(op.action).toBe('delete');
      expect(op.prev.$link).toBe(original.cid);
    });

    test('the firehose frame encodes prev as a CID link', async () => {
      const original = await putGallery('original');
      await putGallery('edited');

      const commits = storage._events.filter((e) => e.commit_cid !== '');
      const frame = pds._firehose.formatFirehoseEvent(
        commits[commits.length - 1],
      );
      const [, headerLength] = cborDecodeFirst(frame);
      const body = cborDecode(frame.subarray(headerLength));

      expect(cidToString(body.ops[0].prev)).toBe(original.cid);
    });
  });
  // Sync 1.1 consumers (relays, tap, jetstream) check each commit by undoing
  // its ops on the new tree with only the blocks the commit carries, back to
  // prevData. One they cannot check makes each of them fetch the whole repo
  // again, all at once. Undoing a delete reads the unchanged nodes beside the
  // key it put back, so a delete's commit carries those too.
  describe('a consumer can check every commit from its own blocks', () => {
    /**
     * @param {string} method
     * @param {object} body
     */
    async function xrpc(method, body) {
      const res = await pds.fetch(
        new Request(`https://pds.example.com/xrpc/com.atproto.repo.${method}`, {
          method: 'POST',
          headers: {
            'Content-Type': 'application/json',
            Authorization: `Bearer ${token}`,
          },
          body: JSON.stringify({ repo: DID, ...body }),
        }),
      );
      expect(res.status, await res.clone().text()).toBe(200);
      return res.json();
    }

    const COLLECTION = 'dev.knotless.check';

    /** Creates `n` records keyed `<prefix>0`..., in batches. */
    async function seed(/** @type {string} */ prefix, /** @type {number} */ n) {
      for (let at = 0; at < n; at += 100) {
        const writes = [];
        for (let i = at; i < Math.min(n, at + 100); i++) {
          writes.push({
            $type: 'com.atproto.repo.applyWrites#create',
            collection: COLLECTION,
            rkey: `${prefix}${String(i).padStart(5, '0')}`,
            value: { status: 'running', n: i },
          });
        }
        await xrpc('applyWrites', { writes });
      }
    }

    /** The commit events since event `from`, decoded. */
    function commitsSince(/** @type {number} */ from) {
      return storage._events
        .slice(from)
        .filter((e) => e.commit_cid !== '')
        .map((e) => cborDecode(e.evt));
    }

    /**
     * Undoes `evt`'s ops with only its own blocks, as a consumer does: throws
     * on a node the commit does not carry, and returns the root reached.
     */
    async function undo(/** @type {any} */ evt) {
      const { roots, blocks } = parseCarFile(evt.blocks);
      const commit = cborDecode(
        /** @type {Uint8Array} */ (blocks.get(roots[0])),
      );
      const ops = evt.ops.map((/** @type {any} */ op) => ({
        action: op.action,
        key: op.path,
        prev: op.prev?.$link,
      }));
      const { root } = await mstInvert(
        cidToString(commit.data),
        ops,
        async (cid) => blocks.get(cid) ?? null,
      );
      return root;
    }

    /** Asserts each of `evts` undoes to its prevData from its own blocks. */
    async function checkable(/** @type {any[]} */ evts) {
      expect(evts.length).toBeGreaterThan(0);
      for (const evt of evts) {
        const ops = evt.ops.map((/** @type {any} */ o) => o.action).join(',');
        await expect(undo(evt), `${evt.rev}: ${ops}`).resolves.toBe(
          evt.prevData,
        );
      }
    }

    /** The keys of `prefix`, with the deepest MST layer among them. */
    async function deepest(/** @type {string[]} */ keys) {
      let max = 0;
      for (const key of keys) max = Math.max(max, await getKeyDepth(key));
      return max;
    }

    test('a delete that merges the nodes beside its key', async () => {
      await seed('a', 300);
      const from = storage._events.length;
      // Every key in turn, so the deletes include keys above the leaves,
      // whose removal merges the subtrees on either side of them.
      const keys = [];
      for (let i = 0; i < 300; i += 7) {
        const rkey = `a${String(i).padStart(5, '0')}`;
        keys.push(`${COLLECTION}/${rkey}`);
        await xrpc('deleteRecord', { collection: COLLECTION, rkey });
      }
      expect(await deepest(keys)).toBeGreaterThan(0);
      await checkable(commitsSince(from));
    });

    test('many deletes in one write, as a compaction makes', async () => {
      await seed('p', 400);
      const from = storage._events.length;
      /** @type {Array<{$type: string, collection: string, rkey: string, value?: object}>} */
      const writes = [
        {
          $type: 'com.atproto.repo.applyWrites#create',
          collection: COLLECTION,
          rkey: 'p99999',
          value: { status: 'success', n: -1 },
        },
      ];
      for (let i = 0; i < 47 * 8; i += 8) {
        writes.push({
          $type: 'com.atproto.repo.applyWrites#delete',
          collection: COLLECTION,
          rkey: `p${String(i).padStart(5, '0')}`,
        });
      }
      await xrpc('applyWrites', { writes });
      const [evt] = commitsSince(from);
      expect(evt.ops.length).toBe(48);
      await checkable([evt]);
    });

    test('a delete deep in a tree of several layers', async () => {
      await seed('d', 2000);
      const from = storage._events.length;
      // The keys at the deepest layer there is, and one beside each.
      const keys = [];
      for (let i = 0; i < 2000; i++) {
        keys.push(`${COLLECTION}/d${String(i).padStart(5, '0')}`);
      }
      const top = await deepest(keys);
      expect(top).toBeGreaterThanOrEqual(2);
      const chosen = [];
      for (let i = 0; i < keys.length; i++) {
        if ((await getKeyDepth(keys[i])) === top) chosen.push(i, i + 1);
      }
      for (const i of chosen.filter((i) => i < keys.length)) {
        const rkey = keys[i].split('/')[1];
        await xrpc('deleteRecord', { collection: COLLECTION, rkey });
      }
      await checkable(commitsSince(from));
    });

    test('creates and updates as well', async () => {
      await seed('c', 200);
      const from = storage._events.length;
      for (let i = 0; i < 200; i += 9) {
        await xrpc('putRecord', {
          collection: COLLECTION,
          rkey: `c${String(i).padStart(5, '0')}`,
          record: { status: 'success', n: i },
        });
        await xrpc('putRecord', {
          collection: COLLECTION,
          rkey: `c${String(i).padStart(5, '0')}x`,
          record: { status: 'running', n: i },
        });
      }
      await checkable(commitsSince(from));
    });
  });
});