#!/usr/bin/env node
/**
 * The work screen against a repository the size of this one.
 *
 * `dev-git-collab-demo.mjs` is a four-commit toy: it says whether the model
 * reads, not whether it holds. This pushes pds.js itself into alice's account
 * and opens pull requests the size an agent writes: tens of files over a
 * dozen commits, two of them overlapping heavily, one far behind main.
 *
 * Every push here uploads a full bundle chain, because a collaborator's copy is
 * self-contained. Three copies of a 65 MB repository is the cost of the model,
 * and this script is where it is paid in front of you: it prints the seconds
 * each step took.
 *
 * Usage: node scripts/dev-git-scale-demo.mjs [source-repo]
 * Env:   PDS_DEV_URL, PDS_DEV_BOB_URL, PDS_DEV_CAROL_URL, PDS_DEV_NOVA_URL
 *        PDS_DEV_PASSWORD, PDS_DEV_PLC_URL, PDS_DEV_WORK_DIR
 */
import { spawnSync } from 'node:child_process';
import {
  mkdirSync,
  readFileSync,
  rmSync,
  symlinkSync,
  writeFileSync,
} from 'node:fs';
import { dirname, resolve } from 'node:path';
import { fileURLToPath } from 'node:url';

const PASSWORD = process.env.PDS_DEV_PASSWORD || 'test-password';
const PLC_URL = process.env.PDS_DEV_PLC_URL || 'http://localhost:2582';
const WORK = resolve(process.env.PDS_DEV_WORK_DIR || '.dev-pds/scale');
const ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '..');
const SOURCE = resolve(process.argv[2] || ROOT);
const NAME = 'pds.js';

const REPO_COLLECTION = 'dev.pdsjs.git.repo';
const CONFIG_COLLECTION = 'dev.pdsjs.git.config';
const CHECK_COLLECTION = 'dev.pdsjs.git.check';
const REVIEW_COLLECTION = 'dev.pdsjs.git.review';

const PEOPLE = {
  alice: {
    base: process.env.PDS_DEV_URL || 'http://localhost:2471',
    name: 'Alice',
    email: 'alice@localhost',
  },
  bob: {
    base: process.env.PDS_DEV_BOB_URL || 'http://localhost:2472',
    name: 'Bob',
    email: 'bob@localhost',
  },
  carol: {
    base: process.env.PDS_DEV_CAROL_URL || 'http://localhost:2473',
    name: 'Carol',
    email: 'carol@localhost',
  },
  nova: {
    base: process.env.PDS_DEV_NOVA_URL || 'http://localhost:2482',
    name: 'Nova',
    email: 'nova@localhost',
  },
};

/** @param {string} command @param {string[]} args @param {object} [options] */
function run(command, args, options = {}) {
  const result = spawnSync(command, args, {
    stdio: ['ignore', 'pipe', 'pipe'],
    encoding: 'utf8',
    maxBuffer: 64 * 1024 * 1024,
    ...options,
    env: {
      ...process.env,
      PATH: `${WORK}/bin:${process.env.PATH}`,
      ATPROTO_GIT_PASSWORD: PASSWORD,
      ATPROTO_GIT_PLC_URL: PLC_URL,
      ...(options.env ?? {}),
    },
  });
  if (result.status !== 0) {
    throw new Error(
      `${command} ${args.join(' ')} failed:\n${(result.stderr || result.stdout).slice(-2000)}`,
    );
  }
  return result.stdout?.trim() ?? '';
}

const git = (who, ...args) => run('git', args, { cwd: `${WORK}/${who}` });

/** Time one step, so the cost of the model is on screen rather than implied. */
async function step(label, work) {
  const started = Date.now();
  const answer = await work();
  console.log(`${label} — ${((Date.now() - started) / 1000).toFixed(1)}s`);
  return answer;
}

/** Commit times march forward, so a log reads in the order things happened. */
let clock = Date.now() - 8 * 60 * 60 * 1000;
function nextTime() {
  clock += 7 * 60 * 1000;
  return new Date(clock);
}

/** @param {string} who @param {string} subject */
function commit(who, subject) {
  const person = PEOPLE[who];
  const when = nextTime().toISOString();
  git(who, 'add', '-A');
  run(
    'git',
    [
      '-c',
      `user.email=${person.email}`,
      '-c',
      `user.name=${person.name}`,
      'commit',
      '-q',
      '-m',
      subject,
    ],
    {
      cwd: `${WORK}/${who}`,
      env: { GIT_AUTHOR_DATE: when, GIT_COMMITTER_DATE: when },
    },
  );
}

/** @param {string} base @param {string} path @param {RequestInit} [init] */
async function xrpc(base, path, init) {
  const res = await fetch(`${base}/xrpc/${path}`, init);
  if (!res.ok) {
    throw new Error(
      `${path.split('?')[0]} ${res.status}: ${(await res.text()).slice(0, 300)}`,
    );
  }
  return res.json();
}

/** @type {Record<string, {did: string, token: string}>} */
const sessions = {};

function putRecord(who, collection, value, rkey) {
  const { did, token } = sessions[who];
  return xrpc(
    PEOPLE[who].base,
    rkey ? 'com.atproto.repo.putRecord' : 'com.atproto.repo.createRecord',
    {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        Authorization: `Bearer ${token}`,
      },
      body: JSON.stringify({
        repo: did,
        collection,
        ...(rkey ? { rkey } : {}),
        record: { $type: collection, ...value },
      }),
    },
  );
}

async function repoRef(who) {
  const params = new URLSearchParams({
    repo: sessions[who].did,
    collection: REPO_COLLECTION,
    rkey: NAME,
  });
  const body = await xrpc(
    PEOPLE[who].base,
    `com.atproto.repo.getRecord?${params}`,
  );
  return { uri: body.uri, cid: body.cid };
}

// ---- sign in, and clear what an earlier run left ------------------------

for (const [who, person] of Object.entries(PEOPLE)) {
  const did = (
    await (await fetch(`${person.base}/.well-known/atproto-did`)).text()
  )
    .trim()
    .replace(/^"|"$/g, '');
  if (!did.startsWith('did:')) {
    throw new Error(`${person.base} serves no account yet`);
  }
  const session = await xrpc(person.base, 'com.atproto.server.createSession', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ identifier: did, password: PASSWORD }),
  });
  sessions[who] = { did, token: session.accessJwt };
}

for (const who of Object.keys(PEOPLE)) {
  const { did, token } = sessions[who];
  const headers = {
    'Content-Type': 'application/json',
    Authorization: `Bearer ${token}`,
  };
  const remove = (collection, rkey) =>
    fetch(`${PEOPLE[who].base}/xrpc/com.atproto.repo.deleteRecord`, {
      method: 'POST',
      headers,
      body: JSON.stringify({ repo: did, collection, rkey }),
    });

  // The config goes before the repository: one holding a protected branch
  // refuses deletion.
  await remove(CONFIG_COLLECTION, NAME);
  await remove(REPO_COLLECTION, NAME);

  // A check or a review from an earlier run names a commit no branch points
  // at any more, which reads as an earlier version of a branch that was never
  // rewritten. Only the ones about this repository go.
  for (const collection of [CHECK_COLLECTION, REVIEW_COLLECTION]) {
    const params = new URLSearchParams({ repo: did, collection, limit: '100' });
    const listed = await fetch(
      `${PEOPLE[who].base}/xrpc/com.atproto.repo.listRecords?${params}`,
    );
    if (!listed.ok) continue;
    for (const row of (await listed.json()).records ?? []) {
      if (!String(row.value?.subject?.uri ?? '').endsWith(`/${NAME}`)) continue;
      await remove(collection, row.uri.split('/').pop());
    }
  }
}

const url = (who) => `atproto://${sessions[who].did}/${NAME}`;

rmSync(WORK, { recursive: true, force: true });
mkdirSync(`${WORK}/bin`, { recursive: true });
symlinkSync(
  `${ROOT}/packages/git/src/cli.js`,
  `${WORK}/bin/git-remote-atproto`,
);

// ---- 1. the repository itself, pushed whole ------------------------------

await step('cloned the source repository', async () =>
  run('git', ['clone', '-q', SOURCE, `${WORK}/alice`], { cwd: WORK }),
);
git('alice', 'checkout', '-q', '-B', 'main');
const commits = git('alice', 'rev-list', '--count', 'HEAD');
const tracked = git('alice', 'ls-files').split('\n').length;
console.log(`${NAME}: ${commits} commits, ${tracked} files`);

await step(`alice pushed ${NAME}`, async () =>
  git('alice', 'push', '-q', url('alice'), 'main'),
);

/**
 * A pull request of the size an agent writes: a block inserted in each of many
 * files, spread over several commits.
 * @param {string} who
 * @param {string[]} paths
 * @param {number} perCommit
 * @param {string} branch
 * @param {string} subject
 */
function bigPull(who, paths, perCommit, branch, subject) {
  git(who, 'checkout', '-q', '-b', branch);
  for (let i = 0; i < paths.length; i += perCommit) {
    const batch = paths.slice(i, i + perCommit);
    for (const path of batch) {
      const body = readFileSync(`${WORK}/${who}/${path}`, 'utf8');
      writeFileSync(
        `${WORK}/${who}/${path}`,
        `${body}\n// ${subject}\n// ${'-'.repeat(60)}\n${batch
          .map((name, line) => `// ${line}: ${name}\n`)
          .join('')}`,
      );
    }
    // The tip carries the summary, the way a stack's last commit does.
    const last = i + perCommit >= paths.length;
    commit(who, last ? subject : `${subject} (${i / perCommit + 1})`);
  }
}

/** @param {string} who @param {string} glob */
const sources = (who, glob) =>
  git(who, 'ls-files', glob)
    .split('\n')
    .filter((path) => path.endsWith('.js'));

// ---- 2. bob's large pull request ----------------------------------------

await step('bob cloned', async () =>
  run('git', ['clone', '-q', url('alice'), `${WORK}/bob`], { cwd: WORK }),
);
git('bob', 'remote', 'add', 'mine', url('bob'));

const core = sources('bob', 'packages/core/src/*.js').slice(0, 48);
bigPull('bob', core, 4, 'agent/rename-ports', 'rename the port types');

// A second, small pull request from the same account.
git('bob', 'checkout', '-q', 'main');
git('bob', 'checkout', '-q', '-b', 'agent/readme-typo');
writeFileSync(
  `${WORK}/bob/README.md`,
  `${readFileSync(`${WORK}/bob/README.md`, 'utf8')}\nOne line.\n`,
);
commit('bob', 'fix a typo in the readme');

await step('bob pushed his copy', async () =>
  git(
    'bob',
    'push',
    '-q',
    'mine',
    'main',
    'agent/rename-ports',
    'agent/readme-typo',
  ),
);

// ---- 3. carol's pull request, overlapping bob's heavily ----------------

await step('carol cloned', async () =>
  run('git', ['clone', '-q', url('alice'), `${WORK}/carol`], { cwd: WORK }),
);
git('carol', 'remote', 'add', 'mine', url('carol'));

// The first half of the same file list, so the two collide on 24 files rather
// than on one.
bigPull(
  'carol',
  core.slice(0, 24),
  6,
  'agent/structured-logging',
  'log through one writer',
);

await step('carol pushed her copy', async () =>
  git('carol', 'push', '-q', 'mine', 'main', 'agent/structured-logging'),
);

// ---- 4. main moves on under both of them --------------------------------

git('alice', 'checkout', '-q', 'main');
for (let i = 0; i < 3; i++) {
  const path = core[i * 5];
  writeFileSync(
    `${WORK}/alice/${path}`,
    `${readFileSync(`${WORK}/alice/${path}`, 'utf8')}\n// merged into main\n`,
  );
  commit('alice', `tidy ${path.split('/').pop()}`);
}
await step('alice advanced main by 3 commits', async () =>
  git('alice', 'push', '-q', url('alice'), 'main'),
);

// ---- 5. what the runner and the reviewers said --------------------------

const bobRef = await repoRef('bob');
const carolRef = await repoRef('carol');

const check = (sha, ref, status, subject, steps) =>
  putRecord('alice', CHECK_COLLECTION, {
    subject,
    ref,
    sha,
    workflow: 'ci',
    status,
    steps,
    startedAt: nextTime().toISOString(),
    finishedAt: nextTime().toISOString(),
  });

await check(
  git('bob', 'rev-parse', 'agent/rename-ports'),
  'refs/heads/agent/rename-ports',
  'failure',
  bobRef,
  [
    { name: 'install', exitCode: 0, durationMs: 21000 },
    { name: 'typecheck', exitCode: 1, durationMs: 47000 },
  ],
);
await check(
  git('bob', 'rev-parse', 'agent/readme-typo'),
  'refs/heads/agent/readme-typo',
  'success',
  bobRef,
  [{ name: 'install', exitCode: 0, durationMs: 19000 }],
);
await check(
  git('carol', 'rev-parse', 'agent/structured-logging'),
  'refs/heads/agent/structured-logging',
  'success',
  carolRef,
  [
    { name: 'install', exitCode: 0, durationMs: 20000 },
    { name: 'test', exitCode: 0, durationMs: 191000 },
  ],
);

await putRecord('nova', REVIEW_COLLECTION, {
  subject: bobRef,
  sha: git('bob', 'rev-parse', 'agent/rename-ports'),
  author: sessions.bob.did,
  ref: 'refs/heads/agent/rename-ports',
  verdict: 'changesRequested',
  note: 'Typecheck fails on 12 of the 48 files. The rename misses the JSDoc imports.',
  reviewedAt: nextTime().toISOString(),
});
await putRecord('carol', REVIEW_COLLECTION, {
  subject: bobRef,
  sha: git('bob', 'rev-parse', 'agent/readme-typo'),
  author: sessions.bob.did,
  ref: 'refs/heads/agent/readme-typo',
  verdict: 'approve',
  note: '',
  reviewedAt: nextTime().toISOString(),
});

await putRecord(
  'alice',
  CONFIG_COLLECTION,
  {
    collaborators: [sessions.bob.did, sessions.carol.did],
    reviewers: [sessions.carol.did, sessions.nova.did],
    runner: sessions.alice.did,
    protectedBranches: ['main'],
    updatedAt: new Date().toISOString(),
  },
  NAME,
);

const size = async (who) => {
  const record = await xrpc(
    PEOPLE[who].base,
    `com.atproto.repo.getRecord?${new URLSearchParams({
      repo: sessions[who].did,
      collection: REPO_COLLECTION,
      rkey: NAME,
    })}`,
  );
  let bytes = 0;
  let parts = 0;
  for (const bundle of record.value.bundles ?? []) {
    for (const part of bundle.parts ?? []) {
      bytes += part.size ?? 0;
      parts++;
    }
  }
  return `${(bytes / 1024 / 1024).toFixed(1)} MB in ${parts} blobs`;
};

console.log('');
for (const who of ['alice', 'bob', 'carol']) {
  console.log(`  ${who.padEnd(6)} ${await size(who)}`);
}
console.log(`\nOpen /${NAME}/pulls`);