/**
 * End-to-end test of the atproto git remote helper against a real PDS:
 * a Node PDS on a local port, real git subprocesses, and the helper wired
 * in via a PATH shim so atproto:// remotes resolve to our CLI.
 *
 * Git must run asynchronously: the PDS lives on this process's event loop,
 * and the helper subprocess calls back into it, so a synchronous git spawn
 * would deadlock the whole arrangement.
 */

import { execFile } from 'node:child_process';
import { randomBytes } from 'node:crypto';
import {
  chmodSync,
  mkdirSync,
  mkdtempSync,
  rmSync,
  writeFileSync,
} from 'node:fs';
import { tmpdir } from 'node:os';
import { dirname, join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { promisify } from 'node:util';
import { defineLexicon } from '@bigmoves/lexicon';
import { LexiconResolver } from '@pdsjs/lexicon-resolver';
import { createServer } from '@pdsjs/node';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { parseBundle } from '../src/bundle.js';
import { GIT_REPO_COLLECTION, gitRepoLexicon } from '../src/index.js';
import { mergePacks } from '../src/pack.js';
import { decodePackIndex } from '../src/pack-index.js';
import { concatChunks, indexPartsOf } from '../src/record.js';

const execFileAsync = promisify(execFile);

const PORT = 2474;
const BASE = `http://127.0.0.1:${PORT}`;
const PASSWORD = 'test-password';

/** @param {number} length */
function randomBase32(length) {
  const chars = 'abcdefghijklmnopqrstuvwxyz234567';
  return Array.from(randomBytes(length))
    .map((b) => chars[b % 32])
    .join('');
}

const DID = `did:plc:${randomBase32(24)}`;

/** @type {Awaited<ReturnType<typeof createServer>>|null} */
let server = null;
/** @type {string} */
let workDir = '';
/** @type {string} */
let srcDir = '';
/** @type {Record<string, string>} */
let gitEnv = {};

/**
 * @param {string} cwd
 * @param {string[]} args
 * @param {Record<string, string>} [env] - overrides merged over the shared env
 * @returns {Promise<string>} stdout
 */
async function runGit(cwd, args, env = {}) {
  const { stdout } = await execFileAsync('git', args, {
    cwd,
    env: { ...gitEnv, ...env },
    encoding: 'utf8',
  });
  return stdout;
}

/**
 * Download and decode a bundle's object index blob.
 * @param {import('../src/record.js').BundleEntry} bundle
 * @returns {Promise<import('../src/pack-index.js').PackIndex>}
 */
async function fetchBundleIndex(bundle) {
  const parts = indexPartsOf(bundle);
  expect(parts.length).toBeGreaterThan(0);
  /** @type {Uint8Array[]} */
  const chunks = [];
  for (const part of parts) {
    const params = new URLSearchParams({ did: DID, cid: part.ref.$link });
    const res = await fetch(`${BASE}/xrpc/com.atproto.sync.getBlob?${params}`);
    expect(res.ok).toBe(true);
    chunks.push(new Uint8Array(await res.arrayBuffer()));
  }
  return decodePackIndex(concatChunks(chunks));
}

/**
 * The record one repository name addresses. The key is minted, so the name
 * is matched against the listing the way every reader matches it.
 * @param {string} [name]
 * @returns {Promise<import('../src/record.js').GitRepoRecord>}
 */
async function fetchRepoRecord(name = 'testrepo') {
  const params = new URLSearchParams({
    repo: DID,
    collection: GIT_REPO_COLLECTION,
    limit: '100',
  });
  const res = await fetch(
    `${BASE}/xrpc/com.atproto.repo.listRecords?${params}`,
  );
  expect(res.ok).toBe(true);
  const body =
    /** @type {{records: Array<{value: import('../src/record.js').GitRepoRecord}>}} */ (
      await res.json()
    );
  const found = body.records.find((row) => row.value.name === name);
  expect(found, `no repository named ${name}`).toBeDefined();
  return /** @type {NonNullable<typeof found>} */ (found).value;
}

beforeAll(async () => {
  workDir = mkdtempSync(join(tmpdir(), 'pdsjs-git-e2e-'));

  // The lexicon ships as a plain literal without a dependency on the
  // validator's types, so its fields are inferred wide; narrow via unknown
  const lexiconSource = /** @type {Parameters<typeof defineLexicon>[0]} */ (
    /** @type {unknown} */ (gitRepoLexicon)
  );
  const lexiconResolver = new LexiconResolver({
    schemas: [defineLexicon(lexiconSource)],
  });
  server = await createServer({
    port: PORT,
    dbPath: join(workDir, 'pds.db'),
    blobsDir: join(workDir, 'blobs'),
    jwtSecret: 'test-secret-for-git-e2e',
    hostname: `127.0.0.1:${PORT}`,
    password: PASSWORD,
    lexiconResolver,
  });
  await server.listen();

  const init = await fetch(`${BASE}/init?did=${DID}`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      did: DID,
      privateKey: randomBytes(32).toString('hex'),
      handle: 'git.test',
      password: PASSWORD,
    }),
  });
  expect(init.ok).toBe(true);

  // PATH shim so git's atproto:// scheme resolves to our helper CLI
  const binDir = join(workDir, 'bin');
  mkdirSync(binDir);
  const cliPath = join(
    dirname(fileURLToPath(import.meta.url)),
    '..',
    'src',
    'cli.js',
  );
  const shim = join(binDir, 'git-remote-atproto');
  writeFileSync(shim, `#!/bin/sh\nexec node ${cliPath} "$@"\n`);
  chmodSync(shim, 0o755);

  gitEnv = {
    .../** @type {Record<string, string>} */ (process.env),
    PATH: `${binDir}:${process.env.PATH}`,
    ATPROTO_GIT_SERVICE: BASE,
    ATPROTO_GIT_IDENTIFIER: DID,
    ATPROTO_GIT_PASSWORD: PASSWORD,
    // Tiny chunks and a low repack threshold so both paths get exercised;
    // tail compaction off so the threshold alone sets the bundle counts
    ATPROTO_GIT_CHUNK_SIZE: '1024',
    ATPROTO_GIT_REPACK_THRESHOLD: '3',
    ATPROTO_GIT_COMPACT_FACTOR: '0',
    GIT_TERMINAL_PROMPT: '0',
    GIT_AUTHOR_NAME: 'Test',
    GIT_AUTHOR_EMAIL: 'test@example.com',
    GIT_COMMITTER_NAME: 'Test',
    GIT_COMMITTER_EMAIL: 'test@example.com',
  };

  srcDir = join(workDir, 'src-repo');
  mkdirSync(srcDir);
  await runGit(srcDir, ['init', '-b', 'main']);
  writeFileSync(join(srcDir, 'README.md'), '# hello from atproto git\n');
  writeFileSync(join(srcDir, 'data.bin'), randomBytes(4096));
  await runGit(srcDir, ['add', '.']);
  await runGit(srcDir, ['commit', '-m', 'initial commit']);
  await runGit(srcDir, [
    'remote',
    'add',
    'origin',
    `atproto://${DID}/testrepo`,
  ]);
}, 60_000);

afterAll(async () => {
  if (server) await server.close();
  rmSync(workDir, { recursive: true, force: true });
});

describe('git-remote-atproto e2e', () => {
  it('pushes a new repository and writes the repo record', async () => {
    await runGit(srcDir, ['push', 'origin', 'main']);

    const record = await fetchRepoRecord();
    expect(record.name).toBe('testrepo');
    expect(record.defaultBranch).toBe('main');
    expect(record.refs).toHaveLength(1);
    expect(record.refs[0].name).toBe('refs/heads/main');
    expect(record.refs[0].sha).toBe(
      (await runGit(srcDir, ['rev-parse', 'main'])).trim(),
    );
    expect(record.bundles).toHaveLength(1);
    // 4 KiB of random data with 1 KiB chunks must span multiple blobs
    expect(record.bundles[0].parts.length).toBeGreaterThan(1);

    // The push also wrote the bundle's object index, and it names every
    // object the bundle holds.
    const decoded = await fetchBundleIndex(record.bundles[0]);
    const listed = (await runGit(srcDir, ['rev-list', '--objects', 'main']))
      .trim()
      .split('\n')
      .map((line) => line.split(' ')[0]);
    expect(new Set(decoded.entries.map((e) => e.sha))).toEqual(new Set(listed));
  });

  it('clones the repository', async () => {
    const cloneDir = join(workDir, 'clone');
    await runGit(workDir, ['clone', `atproto://${DID}/testrepo`, cloneDir]);
    const readme = await runGit(cloneDir, ['show', 'HEAD:README.md']);
    expect(readme).toContain('hello from atproto git');
    expect((await runGit(cloneDir, ['branch', '--show-current'])).trim()).toBe(
      'main',
    );
  });

  it('pushes incrementally and pulls into the clone', async () => {
    writeFileSync(join(srcDir, 'second.txt'), 'second commit\n');
    await runGit(srcDir, ['add', '.']);
    await runGit(srcDir, ['commit', '-m', 'second']);
    await runGit(srcDir, ['push', 'origin', 'main']);

    const record = await fetchRepoRecord();
    expect(record.bundles).toHaveLength(2);

    // The incremental bundle deltas against the first push's objects, so
    // its index build exercised the local-repository base resolver.
    const decoded = await fetchBundleIndex(record.bundles[1]);
    expect(decoded.entries.length).toBeGreaterThan(0);

    const cloneDir = join(workDir, 'clone');
    await runGit(cloneDir, ['pull', 'origin', 'main']);
    expect(await runGit(cloneDir, ['show', 'HEAD:second.txt'])).toBe(
      'second commit\n',
    );
  });

  it('rejects a non-fast-forward push and accepts a forced one', async () => {
    await runGit(srcDir, ['reset', '--hard', 'HEAD~1']);
    writeFileSync(join(srcDir, 'rewritten.txt'), 'rewritten\n');
    await runGit(srcDir, ['add', '.']);
    await runGit(srcDir, ['commit', '-m', 'rewritten history']);

    await expect(runGit(srcDir, ['push', 'origin', 'main'])).rejects.toThrow();

    await runGit(srcDir, ['push', '--force', 'origin', 'main']);
    const record = await fetchRepoRecord();
    expect(record.refs.find((r) => r.name === 'refs/heads/main')?.sha).toBe(
      (await runGit(srcDir, ['rev-parse', 'main'])).trim(),
    );
  });

  it('repacks the bundle chain once it crosses the threshold', async () => {
    // The force push above already crossed ATPROTO_GIT_REPACK_THRESHOLD=3
    // and collapsed the chain, so it starts this test at one bundle
    expect((await fetchRepoRecord()).bundles).toHaveLength(1);

    writeFileSync(join(srcDir, 'third.txt'), 'third\n');
    await runGit(srcDir, ['add', '.']);
    await runGit(srcDir, ['commit', '-m', 'third']);
    await runGit(srcDir, ['push', 'origin', 'main']);
    expect((await fetchRepoRecord()).bundles).toHaveLength(2);

    writeFileSync(join(srcDir, 'fourth.txt'), 'fourth\n');
    await runGit(srcDir, ['add', '.']);
    await runGit(srcDir, ['commit', '-m', 'fourth']);
    await runGit(srcDir, ['push', 'origin', 'main']);

    const record = await fetchRepoRecord();
    expect(record.bundles).toHaveLength(1);

    const cloneDir = join(workDir, 'clone-after-repack');
    await runGit(workDir, ['clone', `atproto://${DID}/testrepo`, cloneDir]);
    expect(await runGit(cloneDir, ['show', 'HEAD:fourth.txt'])).toBe(
      'fourth\n',
    );
    expect(
      (await runGit(cloneDir, ['rev-list', '--count', 'HEAD'])).trim(),
    ).toBe((await runGit(srcDir, ['rev-list', '--count', 'HEAD'])).trim());
  });

  it('pushes branches and tags, then deletes the branch', async () => {
    await runGit(srcDir, ['branch', 'feature']);
    await runGit(srcDir, ['tag', '-a', 'v1', '-m', 'release v1']);
    await runGit(srcDir, ['push', 'origin', 'feature', 'v1']);

    let record = await fetchRepoRecord();
    const names = record.refs.map((r) => r.name);
    expect(names).toContain('refs/heads/feature');
    expect(names).toContain('refs/tags/v1');

    const cloneDir = join(workDir, 'clone-tags');
    await runGit(workDir, ['clone', `atproto://${DID}/testrepo`, cloneDir]);
    expect((await runGit(cloneDir, ['tag', '-l'])).trim()).toBe('v1');

    await runGit(srcDir, ['push', 'origin', ':feature']);
    record = await fetchRepoRecord();
    expect(record.refs.map((r) => r.name)).not.toContain('refs/heads/feature');
  });

  it('splits an index too large for one blob and clones it back', async () => {
    const wideDir = join(workDir, 'wide-repo');
    mkdirSync(wideDir);
    await runGit(wideDir, ['init', '-b', 'main']);
    // 1 KiB chunks hold 31 index rows, so 60 files index past one blob.
    for (let i = 0; i < 60; i++) {
      writeFileSync(join(wideDir, `file${i}.txt`), `contents of file ${i}\n`);
    }
    await runGit(wideDir, ['add', '.']);
    await runGit(wideDir, ['commit', '-m', 'many files']);
    await runGit(wideDir, ['push', `atproto://${DID}/widerepo`, 'main']);

    const record = await fetchRepoRecord('widerepo');
    expect(record.bundles[0].index).toBeUndefined();
    expect(indexPartsOf(record.bundles[0]).length).toBeGreaterThan(1);

    // The joined blobs decode, and name every object git lists.
    const decoded = await fetchBundleIndex(record.bundles[0]);
    const listed = (await runGit(wideDir, ['rev-list', '--objects', 'main']))
      .trim()
      .split('\n')
      .map((line) => line.split(' ')[0]);
    expect(new Set(decoded.entries.map((e) => e.sha))).toEqual(new Set(listed));

    const cloneDir = join(workDir, 'wide-clone');
    await runGit(workDir, ['clone', `atproto://${DID}/widerepo`, cloneDir]);
    expect((await runGit(cloneDir, ['show', 'HEAD:file59.txt'])).trim()).toBe(
      'contents of file 59',
    );
  });
});

let verifyCount = 0;

/**
 * Merge the chain's packfiles the way the smart HTTP endpoint does and
 * have real git index the result the way a clone does. A chain that
 * repeats an object fails here: "The same object appears twice in the
 * pack". A plain fetch tolerates the repeat, so the clone flag matters.
 * @param {string} rkey
 * @returns {Promise<import('../src/record.js').GitRepoRecord>}
 */
async function indexMergedChain(rkey) {
  const record = await fetchRepoRecord(rkey);
  /** @type {Uint8Array[]} */
  const packs = [];
  for (const bundle of record.bundles) {
    /** @type {Uint8Array[]} */
    const chunks = [];
    for (const part of bundle.parts) {
      const params = new URLSearchParams({ did: DID, cid: part.ref.$link });
      const res = await fetch(
        `${BASE}/xrpc/com.atproto.sync.getBlob?${params}`,
      );
      expect(res.ok).toBe(true);
      chunks.push(new Uint8Array(await res.arrayBuffer()));
    }
    packs.push(parseBundle(concatChunks(chunks)).pack);
  }
  verifyCount += 1;
  const scratch = join(workDir, `verify-${rkey}-${verifyCount}`);
  mkdirSync(scratch);
  await runGit(scratch, ['init', '-b', 'main']);
  const packPath = join(scratch, 'merged.pack');
  writeFileSync(packPath, mergePacks(packs));
  await runGit(scratch, [
    'index-pack',
    '--check-self-contained-and-connected',
    packPath,
  ]);
  return record;
}

describe('bundle chain stays disjoint', () => {
  // Repacks and tail merges would collapse the chain and hide a repeated
  // object.
  const NO_REPACK = {
    ATPROTO_GIT_REPACK_THRESHOLD: '99',
    ATPROTO_GIT_COMPACT_FACTOR: '0',
  };
  /** @type {string} */
  let repoDir = '';
  /** @type {string} */
  let staleDir = '';
  /** @type {string} */
  let secondSha = '';

  it('completes the push basis from the chain when a tip is absent locally', async () => {
    repoDir = join(workDir, 'disjoint-src');
    mkdirSync(repoDir);
    await runGit(repoDir, ['init', '-b', 'main']);
    writeFileSync(join(repoDir, 'a.txt'), 'one\n');
    await runGit(repoDir, ['add', '.']);
    await runGit(repoDir, ['commit', '-m', 'one']);
    await runGit(repoDir, [
      'remote',
      'add',
      'origin',
      `atproto://${DID}/disjoint`,
    ]);
    await runGit(repoDir, ['push', 'origin', 'main'], NO_REPACK);

    // A working copy taken now lacks the tip the next push records.
    staleDir = join(workDir, 'disjoint-stale');
    await runGit(workDir, ['clone', repoDir, staleDir]);

    writeFileSync(join(repoDir, 'b.txt'), 'two\n');
    await runGit(repoDir, ['add', '.']);
    await runGit(repoDir, ['commit', '-m', 'two']);
    secondSha = (await runGit(repoDir, ['rev-parse', 'main'])).trim();
    await runGit(repoDir, ['push', 'origin', 'main'], NO_REPACK);

    // Without the remote tip, a basis of local-only tips would re-bundle
    // the shared history and repeat the first bundle's objects.
    await runGit(staleDir, [
      'remote',
      'add',
      'atproto',
      `atproto://${DID}/disjoint`,
    ]);
    await runGit(staleDir, ['checkout', '-b', 'feature']);
    writeFileSync(join(staleDir, 'c.txt'), 'three\n');
    await runGit(staleDir, ['add', '.']);
    await runGit(staleDir, ['commit', '-m', 'three']);
    await runGit(staleDir, ['push', 'atproto', 'feature'], NO_REPACK);

    const record = await indexMergedChain('disjoint');
    expect(record.bundles.length).toBe(3);
  });

  it('excludes history that only a superseded head reaches', async () => {
    // Rewind main. The second commit stays in the chain with no tip on it.
    await runGit(repoDir, ['reset', '--hard', 'HEAD~1']);
    await runGit(repoDir, ['push', '--force', 'origin', 'main'], NO_REPACK);

    // A branch on the rewound commit re-advertises objects the chain holds,
    // so the push updates refs without a new bundle.
    await runGit(repoDir, ['branch', 'keep', secondSha]);
    await runGit(repoDir, ['push', 'origin', 'keep'], NO_REPACK);

    const record = await indexMergedChain('disjoint');
    expect(record.bundles.length).toBe(3);
    const keep = record.refs.find((r) => r.name === 'refs/heads/keep');
    expect(keep?.sha).toBe(secondSha);
  });
});

describe('geometric compaction', () => {
  // The full-repack threshold stays out of reach; only tail merges may
  // shrink the chain.
  const COMPACT = {
    ATPROTO_GIT_REPACK_THRESHOLD: '99',
    ATPROTO_GIT_COMPACT_FACTOR: '2',
  };
  /** @type {string} */
  let repoDir = '';

  it('bounds the chain and serves a clean merged pack', async () => {
    repoDir = join(workDir, 'geom-src');
    mkdirSync(repoDir);
    await runGit(repoDir, ['init', '-b', 'main']);
    await runGit(repoDir, ['remote', 'add', 'origin', `atproto://${DID}/geom`]);
    for (let i = 0; i < 7; i++) {
      writeFileSync(join(repoDir, `file-${i}.txt`), `content ${i}\n`);
      await runGit(repoDir, ['add', '.']);
      await runGit(repoDir, ['commit', '-m', `commit ${i}`]);
      await runGit(repoDir, ['push', 'origin', 'main'], COMPACT);
    }
    // Seven near-equal pushes fit in three bundles: every kept bundle is
    // at least twice the combined size of the bundles after it.
    const record = await indexMergedChain('geom');
    expect(record.bundles.length).toBeLessThanOrEqual(3);
    // Fast-forward pushes leave one independent head per bundle: a merge
    // prunes the superseded tips its newest head reaches.
    expect(record.bundles.every((b) => b.heads.length === 1)).toBe(true);

    const cloneDir = join(workDir, 'geom-clone');
    await runGit(workDir, ['clone', `atproto://${DID}/geom`, cloneDir]);
    expect(
      (await runGit(cloneDir, ['rev-list', '--count', 'HEAD'])).trim(),
    ).toBe('7');
  });

  it('keeps history a superseded head reaches across a merge', async () => {
    const oldSha = (await runGit(repoDir, ['rev-parse', 'main'])).trim();
    await runGit(repoDir, ['reset', '--hard', 'HEAD~1']);
    writeFileSync(join(repoDir, 'rewritten.txt'), 'rewritten\n');
    await runGit(repoDir, ['add', '.']);
    await runGit(repoDir, ['commit', '-m', 'rewritten']);
    await runGit(repoDir, ['push', '--force', 'origin', 'main'], COMPACT);

    // The superseded tip stays a recorded head, so its history stays in
    // the chain and a branch put back on it is a refs-only push.
    const record = await indexMergedChain('geom');
    expect(record.bundles.flatMap((b) => b.heads)).toContain(oldSha);

    await runGit(repoDir, ['branch', 'keep', oldSha]);
    await runGit(repoDir, ['push', 'origin', 'keep'], COMPACT);
    const after = await indexMergedChain('geom');
    expect(after.bundles.length).toBe(record.bundles.length);

    const cloneDir = join(workDir, 'geom-keep-clone');
    await runGit(workDir, ['clone', `atproto://${DID}/geom`, cloneDir]);
    expect((await runGit(cloneDir, ['rev-parse', 'origin/keep'])).trim()).toBe(
      oldSha,
    );
  });
});