chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
import { writeFileSync } from 'node:fs';
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { createWriteHandler, watch } from '../src/daemon.js';
import { createStateStore, statePath } from '../src/state.js';
const SERVICE = 'https://pds.example.com';
const DID = 'did:plc:abc';
const REPOS = [{ did: DID, repoName: 'proj', service: SERVICE }];
/** @type {string} */
let dir;
beforeEach(async () => {
dir = await mkdtemp(join(tmpdir(), 'git-ci-state-'));
});
afterEach(async () => {
await rm(dir, { recursive: true, force: true });
});
const MAIN = 'a'.repeat(40);
/** The repository's record key, which is not what it is called. */
const RKEY = '3lprojkey';
/**
* @param {Array<{name: string, sha: string}>} refs
* @param {string} [name]
* @param {string} [rkey]
*/
function write(refs, name = 'proj', rkey = RKEY) {
return {
did: DID,
collection: 'dev.pdsjs.git.repo',
rkey,
cid: 'bafyreiaaa',
uri: `at://${DID}/dev.pdsjs.git.repo/${rkey}`,
seq: 1,
value: {
name,
refs,
bundles: [],
createdAt: '2026-08-18T00:00:00.000Z',
},
};
}
/**
* @param {Record<string, unknown>} value
* @param {Object} [opts]
* @param {string} [opts.rkey]
* @param {string} [opts.cid]
* @param {string} [opts.did]
*/
function requestWrite(value, opts = {}) {
const rkey = opts.rkey ?? 'proj:refs_heads_main';
const did = opts.did ?? DID;
return {
did,
collection: 'dev.pdsjs.git.checkRequest',
rkey,
cid: opts.cid ?? 'bafyreireq1',
uri: `at://${did}/dev.pdsjs.git.checkRequest/${rkey}`,
seq: 2,
value,
};
}
/** @param {{refs?: Array<{name: string, sha: string}>}} [opts] */
function harness(opts = {}) {
const state = createStateStore(join(dir, 'state.json'));
/** @type {Array<Record<string, *>>} */
const runs = [];
/** @type {string[]} */
const notices = [];
/** @type {Array<Record<string, *>>} */
const reads = [];
const onWrite = createWriteHandler({
service: SERVICE,
repos: REPOS,
state,
runCheck: async (/** @type {Record<string, *>} */ request) => {
runs.push(request);
return 'success';
},
readRepoRecord: async (/** @type {Record<string, *>} */ target) => {
reads.push(target);
if (target.rkey && target.rkey !== RKEY) {
throw new Error(`no repository at ${target.rkey}`);
}
return {
uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
cid: 'bafyreicurrent',
name: 'proj',
rkey: RKEY,
refs: opts.refs ?? [{ name: 'refs/heads/main', sha: MAIN }],
};
},
onNotice: (/** @type {string} */ message) => notices.push(message),
});
return { onWrite, runs, state, notices, reads };
}
describe('createWriteHandler', () => {
it('runs the ref that moved', async () => {
const { onWrite, runs } = harness();
await onWrite(write([{ name: 'refs/heads/main', sha: 'a'.repeat(40) }]));
expect(runs).toHaveLength(1);
expect(runs[0].ref).toBe('refs/heads/main');
expect(runs[0].sha).toBe('a'.repeat(40));
expect(runs[0].repoName).toBe('proj');
expect(runs[0].service).toBe(SERVICE);
expect(runs[0].subject).toEqual({
uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
cid: 'bafyreiaaa',
});
});
it('ignores a repository it does not watch', async () => {
const { onWrite, runs } = harness();
await onWrite(
write([{ name: 'refs/heads/main', sha: 'a'.repeat(40) }], 'other'),
);
expect(runs).toEqual([]);
});
it('does not re-run a ref that did not move', async () => {
const { onWrite, runs } = harness();
const event = write([{ name: 'refs/heads/main', sha: 'a'.repeat(40) }]);
await onWrite(event);
await onWrite(event);
expect(runs).toHaveLength(1);
});
it('runs once per ref that moved in the same push', async () => {
const { onWrite, runs } = harness();
await onWrite(
write([
{ name: 'refs/heads/main', sha: 'a'.repeat(40) },
{ name: 'refs/heads/topic', sha: 'b'.repeat(40) },
]),
);
expect(runs.map((run) => run.ref)).toEqual([
'refs/heads/main',
'refs/heads/topic',
]);
});
it('leaves the ref map behind when a run throws, so the event replays', async () => {
const state = createStateStore(join(dir, 'state.json'));
let attempts = 0;
const onWrite = createWriteHandler({
service: SERVICE,
repos: REPOS,
state,
runCheck: async () => {
attempts += 1;
if (attempts === 1) throw new Error('runner died');
return 'success';
},
});
const event = write([{ name: 'refs/heads/main', sha: 'a'.repeat(40) }]);
await expect(onWrite(event)).rejects.toThrow('runner died');
await onWrite(event);
expect(attempts).toBe(2);
});
});
describe('createWriteHandler, on a check request', () => {
const REQUEST = {
repo: RKEY,
ref: 'refs/heads/main',
requestedAt: '2026-08-19T00:00:00.000Z',
};
it('runs the ref at the commit the repository record names now', async () => {
const { onWrite, runs, reads } = harness();
await onWrite(requestWrite(REQUEST));
expect(reads).toEqual([{ service: SERVICE, did: DID, rkey: RKEY }]);
expect(runs).toHaveLength(1);
expect(runs[0].ref).toBe('refs/heads/main');
expect(runs[0].sha).toBe(MAIN);
expect(runs[0].subject).toEqual({
uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
cid: 'bafyreicurrent',
});
});
it('runs the commit the request names, not the ref tip', async () => {
const { onWrite, runs } = harness();
const sha = 'b'.repeat(40);
await onWrite(requestWrite({ ...REQUEST, sha }));
expect(runs[0].sha).toBe(sha);
});
it('ignores a request naming a repository it does not watch', async () => {
const { onWrite, runs, reads } = harness();
await onWrite(requestWrite({ ...REQUEST, repo: '3lotherkey' }));
await onWrite(requestWrite(REQUEST, { did: 'did:plc:someoneelse' }));
expect(runs).toEqual([]);
// An account nothing is watched of costs no read at all; a key this
// account does not hold costs the one read that says so.
expect(reads).toEqual([{ service: SERVICE, did: DID, rkey: '3lotherkey' }]);
});
it('ignores a request for a repository of a name it does not watch', async () => {
/** @type {Array<Record<string, *>>} */
const runs = [];
const onWrite = createWriteHandler({
service: SERVICE,
repos: REPOS,
state: createStateStore(join(dir, 'state.json')),
runCheck: async (/** @type {Record<string, *>} */ request) => {
runs.push(request);
return 'success';
},
// The account holds this repository; the runner watches another of
// its repositories, and the name is what says so.
readRepoRecord: async () => ({
uri: `at://${DID}/dev.pdsjs.git.repo/3lotherkey`,
cid: 'bafyreicurrent',
name: 'unwatched',
rkey: '3lotherkey',
refs: [{ name: 'refs/heads/main', sha: MAIN }],
}),
});
await onWrite(requestWrite({ ...REQUEST, repo: '3lotherkey' }));
expect(runs).toEqual([]);
});
it('ignores a request missing the repository or the ref', async () => {
const { onWrite, runs } = harness();
await onWrite(requestWrite({ ref: 'refs/heads/main' }));
await onWrite(requestWrite({ repo: RKEY }));
await onWrite(requestWrite({ repo: RKEY, ref: '' }));
expect(runs).toEqual([]);
});
it('runs the same request version only once', async () => {
const { onWrite, runs } = harness();
await onWrite(requestWrite(REQUEST));
await onWrite(requestWrite(REQUEST));
expect(runs).toHaveLength(1);
});
it('runs again when the request is made afresh', async () => {
const { onWrite, runs } = harness();
await onWrite(requestWrite(REQUEST));
await onWrite(
requestWrite(
{ ...REQUEST, requestedAt: '2026-08-19T01:00:00.000Z' },
{ cid: 'bafyreireq2' },
),
);
expect(runs).toHaveLength(2);
});
it('leaves the ref map alone, so the next push still starts a run', async () => {
const { onWrite, runs, state } = harness();
await onWrite(requestWrite(REQUEST));
expect(state.refs('did:plc:abc/proj')).toEqual({});
await onWrite(write([{ name: 'refs/heads/main', sha: MAIN }]));
expect(runs).toHaveLength(2);
});
it('leaves the request unrecorded when the run throws, so it replays', async () => {
const state = createStateStore(join(dir, 'state.json'));
let attempts = 0;
const onWrite = createWriteHandler({
service: SERVICE,
repos: REPOS,
state,
runCheck: async () => {
attempts += 1;
if (attempts === 1) throw new Error('runner died');
return 'success';
},
readRepoRecord: async () => ({
uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
cid: 'bafyreicurrent',
name: 'proj',
rkey: RKEY,
refs: [{ name: 'refs/heads/main', sha: MAIN }],
}),
});
await expect(onWrite(requestWrite(REQUEST))).rejects.toThrow('runner died');
await onWrite(requestWrite(REQUEST));
expect(attempts).toBe(2);
});
it('says so and gives up on a ref the repository does not have', async () => {
const { onWrite, runs, notices } = harness();
await onWrite(requestWrite({ ...REQUEST, ref: 'refs/heads/gone' }));
expect(runs).toEqual([]);
expect(notices.at(-1)).toMatch(/refs\/heads\/gone is not a ref/);
// Marked handled, so the same request replaying costs no second read.
await onWrite(requestWrite({ ...REQUEST, ref: 'refs/heads/gone' }));
expect(notices).toHaveLength(1);
});
it('reads the repository over XRPC when the context names no reader', async () => {
const state = createStateStore(join(dir, 'state.json'));
/** @type {string[]} */
const asked = [];
const original = globalThis.fetch;
globalThis.fetch = async (/** @type {any} */ url) => {
asked.push(String(url));
return Response.json({
uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
cid: 'bafyreicurrent',
value: {
name: 'proj',
refs: [],
bundles: [],
createdAt: '2026-08-18T00:00:00.000Z',
},
});
};
try {
// No onNotice either: the ref is absent, which is the branch that
// reports, and the default callback has to receive it silently.
const onWrite = createWriteHandler({
service: SERVICE,
repos: REPOS,
state,
runCheck: async () => 'success',
});
await onWrite(requestWrite(REQUEST));
} finally {
globalThis.fetch = original;
}
expect(asked[0]).toContain('/xrpc/com.atproto.repo.getRecord');
expect(asked[0]).toContain('collection=dev.pdsjs.git.repo');
});
});
describe('createStateStore', () => {
it('round-trips the cursor, the ref map and the requests through the file', () => {
const path = join(dir, 'state.json');
const first = createStateStore(path);
first.setCursor(SERVICE, 99);
first.setRefs('did:plc:abc/proj', { 'refs/heads/main': 'aa' });
first.setRequest('did:plc:abc/proj', 'proj:refs_heads_main', 'bafyreireq');
const second = createStateStore(path);
expect(second.cursor(SERVICE)).toBe(99);
expect(second.refs('did:plc:abc/proj')).toEqual({
'refs/heads/main': 'aa',
});
expect(second.request('did:plc:abc/proj', 'proj:refs_heads_main')).toBe(
'bafyreireq',
);
});
it("keeps a repository's requests apart by ref", () => {
const store = createStateStore(join(dir, 'state.json'));
store.setRequest('did:plc:abc/proj', 'proj:refs_heads_main', 'one');
store.setRequest('did:plc:abc/proj', 'proj:refs_heads_topic', 'two');
expect(store.request('did:plc:abc/proj', 'proj:refs_heads_main')).toBe(
'one',
);
expect(store.request('did:plc:abc/proj', 'proj:refs_heads_topic')).toBe(
'two',
);
});
// A regular file where a directory must go fails the write for every
// uid, root included, so these run in the root containers CI uses. A
// directory mode cannot stop root.
it('keeps working in memory when the file cannot be written', () => {
// A bind mount the daemon does not own reads this way, and losing
// durability must not stop the events reaching the runner.
const blocker = join(dir, 'blocker');
writeFileSync(blocker, '');
/** @type {string[]} */
const warnings = [];
const store = createStateStore(
join(blocker, 'sub', 'state.json'),
(message) => warnings.push(message),
);
expect(() => store.setCursor(SERVICE, 7)).not.toThrow();
expect(() => store.setRefs('did:plc:abc/proj', { a: 'b' })).not.toThrow();
expect(store.cursor(SERVICE)).toBe(7);
expect(store.refs('did:plc:abc/proj')).toEqual({ a: 'b' });
// Warned once, not once per write.
expect(warnings).toHaveLength(1);
expect(warnings[0]).toMatch(/not writable/);
});
it('carries on with no warning callback at all', () => {
const blocker = join(dir, 'blocker-quiet');
writeFileSync(blocker, '');
const store = createStateStore(join(blocker, 'sub', 'state.json'));
expect(() => store.setCursor(SERVICE, 7)).not.toThrow();
expect(store.cursor(SERVICE)).toBe(7);
});
it('starts empty when the file is absent or unreadable', () => {
const store = createStateStore(join(dir, 'missing', 'state.json'));
expect(store.cursor(SERVICE)).toBeUndefined();
expect(store.refs('did:plc:abc/proj')).toEqual({});
expect(store.request('did:plc:abc/proj', 'anything')).toBeUndefined();
});
});
describe('watch startup', () => {
/** A firehose that connects, stays quiet, and ends when aborted. */
const quietSignal = () => {
const controller = new AbortController();
queueMicrotask(() => controller.abort());
return controller.signal;
};
it('adopts a repository refs before the first event', async () => {
const store = createStateStore(join(dir, 'state.json'));
/** @type {string[]} */
const ran = [];
await watch({
repos: REPOS,
runCheck: async (/** @type {{ref: string}} */ request) => {
ran.push(request.ref);
return 'success';
},
state: store,
signal: quietSignal(),
readRepoRecord: async () => ({
uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
cid: 'bafyreiaaa',
name: 'proj',
rkey: RKEY,
refs: [
{ name: 'refs/heads/main', sha: MAIN },
{ name: 'refs/tags/v1.0.0', sha: MAIN },
],
}),
});
// Without this the next event reads both refs as new and reruns the tag,
// which for a release workflow means publishing again.
expect(store.refs(`${DID}/proj`)).toEqual({
'refs/heads/main': MAIN,
'refs/tags/v1.0.0': MAIN,
});
expect(ran).toEqual([]);
});
it('leaves a ref whose run was cut off, so it runs again', async () => {
const store = createStateStore(join(dir, 'state.json'));
/** @type {string[]} */
const notices = [];
await watch({
repos: REPOS,
runCheck: async () => 'success',
state: store,
signal: quietSignal(),
onNotice: (message) => notices.push(message),
readRepoRecord: async () => ({
uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
cid: 'bafyreiaaa',
name: 'proj',
rkey: RKEY,
refs: [
{ name: 'refs/heads/main', sha: MAIN },
{ name: 'refs/tags/v1.0.0', sha: MAIN },
],
}),
unfinishedRefs: async () => ['refs/heads/main'],
});
// Adopting main would leave its half-written check record running for
// good. Unadopted, the next event for the repository runs it again.
expect(store.refs(`${DID}/proj`)).toEqual({ 'refs/tags/v1.0.0': MAIN });
expect(notices.some((n) => n.includes('was left mid-run'))).toBe(true);
});
it('keeps the refs it already knows', async () => {
const store = createStateStore(join(dir, 'state.json'));
store.setRefs(`${DID}/proj`, { 'refs/heads/main': MAIN });
let reads = 0;
await watch({
repos: REPOS,
runCheck: async () => 'success',
state: store,
signal: quietSignal(),
readRepoRecord: async () => {
reads += 1;
return {
uri: 'at://x',
cid: 'bafyreiaaa',
name: 'proj',
rkey: RKEY,
refs: [],
};
},
});
expect(reads).toBe(0);
expect(store.refs(`${DID}/proj`)).toEqual({ 'refs/heads/main': MAIN });
});
it('watches a repository whose record it cannot read', async () => {
const store = createStateStore(join(dir, 'state.json'));
/** @type {string[]} */
const notices = [];
await watch({
repos: REPOS,
runCheck: async () => 'success',
state: store,
signal: quietSignal(),
onNotice: (message) => notices.push(message),
readRepoRecord: async () => {
throw new Error('502');
},
});
expect(store.refs(`${DID}/proj`)).toEqual({});
expect(notices.some((n) => n.includes('could not read its refs'))).toBe(
true,
);
});
});
describe('statePath', () => {
it('takes an explicit path first', () => {
expect(statePath({ PDSJS_CI_STATE: '/tmp/x.json' })).toBe('/tmp/x.json');
});
it('falls back to the XDG state directory', () => {
expect(statePath({ XDG_STATE_HOME: '/state' })).toBe(
'/state/pdsjs-git-ci/state.json',
);
});
});