Something went wrong. Try again.
An AT Protocol Personal Data Server written in JavaScript pdsjs.dev
pds atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456#!/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`);