chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
#!/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`);