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