chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
import { PassThrough } from 'node:stream';
import { computeJwkThumbprint } from '@pdsjs/core/crypto';
import { parseDpopProof } from '@pdsjs/core/oauth';
import { describe, expect, it } from 'vitest';
import { createDpopProof, generateDpopKey } from '../src/dpop.js';
import { AtprotoRemoteHelper } from '../src/helper.js';
import { GIT_SPACE_TYPE, gitSpaceLexicon } from '../src/index.js';
import {
compactionStart,
concatChunks,
indexPartsOf,
parseRepoRecord,
splitIntoChunks,
} from '../src/record.js';
import { parseAtprotoUrl } from '../src/url.js';
import { XrpcClient, XrpcError } from '../src/xrpc.js';
describe('gitSpaceLexicon', () => {
it('declares the Git collections under the canonical space type', () => {
expect(GIT_SPACE_TYPE).toBe('dev.pdsjs.git.space');
expect(gitSpaceLexicon).toEqual({
lexicon: 1,
id: GIT_SPACE_TYPE,
defs: {
main: {
type: 'space',
key: 'any',
name: 'Git repository',
collections: ['dev.pdsjs.git.repo', 'dev.pdsjs.git.config'],
},
},
});
});
});
describe('parseAtprotoUrl', () => {
it('parses a handle and a repository', () => {
expect(parseAtprotoUrl('atproto://chadtmiller.com/pds.js')).toEqual({
authority: 'chadtmiller.com',
repoName: 'pds.js',
});
});
it('parses a DID and a repository', () => {
expect(parseAtprotoUrl('atproto://did:plc:abc123/my-repo')).toEqual({
authority: 'did:plc:abc123',
repoName: 'my-repo',
});
});
it('drops a trailing slash', () => {
expect(parseAtprotoUrl('atproto://a.com/repo/').repoName).toBe('repo');
});
it('is one form, public or private', () => {
// A private repository is addressed by its name like any other; the
// helper finds the space it keeps. The space form is not an address.
expect(() =>
parseAtprotoUrl('atproto://a.com/space/dev.pdsjs.git.space/repo/repo'),
).toThrow(/expected atproto:/);
});
it('rejects other schemes and shapes', () => {
expect(() => parseAtprotoUrl('https://a.com/repo')).toThrow(
/expected atproto:/,
);
expect(() => parseAtprotoUrl('atproto://a.com')).toThrow(
/expected atproto:/,
);
expect(() => parseAtprotoUrl('atproto://a.com/re/po/x')).toThrow(
/expected atproto:/,
);
expect(() => parseAtprotoUrl('atproto://a.com//repo')).toThrow(
/expected atproto:/,
);
expect(() => parseAtprotoUrl('atproto://a.com/re%20po')).toThrow(
/invalid repository name/,
);
});
});
// The server side is @pdsjs/core's parseDpopProof, so the round-trip below is
// against the verifier the helper will actually meet.
describe('dpop', () => {
it('produces a proof the core verifier accepts', async () => {
const key = await generateDpopKey();
const url = 'https://pds.example/xrpc/com.atproto.space.getRecord?x=1';
const proof = await createDpopProof({
key,
method: 'GET',
url,
credential: 'credential-jwt',
});
const parsed = await parseDpopProof(
proof,
'GET',
'https://pds.example/xrpc/com.atproto.space.getRecord',
key.jkt,
'credential-jwt',
);
expect(parsed.jkt).toBe(key.jkt);
});
it('computes the RFC 7638 thumbprint core computes', async () => {
const key = await generateDpopKey();
expect(key.jkt).toBe(await computeJwkThumbprint(key.jwk));
});
it('binds the proof to the credential and method', async () => {
const key = await generateDpopKey();
const url = 'https://pds.example/xrpc/com.atproto.space.getBlob';
const proof = await createDpopProof({
key,
method: 'GET',
url,
credential: 'credential-jwt',
});
await expect(
parseDpopProof(proof, 'GET', url, key.jkt, 'another-token'),
).rejects.toThrow(/ath/);
await expect(
parseDpopProof(proof, 'POST', url, key.jkt, 'credential-jwt'),
).rejects.toThrow(/htm/);
});
});
describe('record creation', () => {
it('posts a record without choosing its key', async () => {
const originalFetch = globalThis.fetch;
/** @type {Array<{url: string, init: RequestInit}>} */
const calls = [];
globalThis.fetch = /** @type {typeof fetch} */ (
async (input, init) => {
calls.push({ url: String(input), init: init ?? {} });
return Response.json({ uri: 'at://did:plc:alice/c/rkey', cid: 'cid' });
}
);
try {
const client = new XrpcClient('https://pds.example');
client.accessJwt = 'access';
await expect(
client.createRecord('did:plc:alice', 'dev.pdsjs.git.pull', {
ref: 'refs/heads/change',
}),
).resolves.toEqual({ uri: 'at://did:plc:alice/c/rkey', cid: 'cid' });
expect(calls[0].url).toBe(
'https://pds.example/xrpc/com.atproto.repo.createRecord',
);
expect(calls[0].init.method).toBe('POST');
expect(calls[0].init.body).toBe(
JSON.stringify({
repo: 'did:plc:alice',
collection: 'dev.pdsjs.git.pull',
record: { ref: 'refs/heads/change' },
}),
);
} finally {
globalThis.fetch = originalFetch;
}
});
});
describe('chunking', () => {
it('splits and reassembles byte-for-byte', () => {
const bytes = new Uint8Array(10_000).map((_, i) => i % 251);
const chunks = splitIntoChunks(bytes, 1024);
expect(chunks.length).toBe(10);
expect(chunks[0].length).toBe(1024);
expect(chunks[9].length).toBe(10_000 - 9 * 1024);
expect(concatChunks(chunks)).toEqual(bytes);
});
it('keeps a single chunk for small payloads and empty input', () => {
expect(splitIntoChunks(new Uint8Array(10), 1024).length).toBe(1);
expect(splitIntoChunks(new Uint8Array(0), 1024).length).toBe(1);
});
});
describe('compactionStart', () => {
it('keeps a chain whose sizes already decay geometrically', () => {
expect(compactionStart([1000, 10], 2)).toBe(1);
expect(compactionStart([30, 20, 10], 2)).toBe(2);
});
it('selects the tail that fell under the progression', () => {
// 10 < 2 * 10, and 1000 covers the merged 20
expect(compactionStart([1000, 10, 10], 2)).toBe(1);
});
it('merges the whole chain of equal sizes', () => {
expect(compactionStart([10, 10, 10], 2)).toBe(0);
});
it('selects the last index for a single bundle and for factor 0', () => {
expect(compactionStart([100], 2)).toBe(0);
expect(compactionStart([10, 10, 10], 0)).toBe(2);
});
});
describe('parseRepoRecord', () => {
const blob = {
$type: 'blob',
ref: { $link: 'bafkreihexample' },
mimeType: 'application/octet-stream',
size: 42,
};
const sha = 'a'.repeat(40);
it('accepts a well-formed record', () => {
const record = parseRepoRecord({
name: 'repo',
defaultBranch: 'main',
refs: [{ name: 'refs/heads/main', sha }],
bundles: [{ heads: [sha], parts: [blob] }],
createdAt: '2026-08-11T00:00:00.000Z',
});
expect(record.refs[0].sha).toBe(sha);
expect(record.bundles[0].parts[0].ref.$link).toBe('bafkreihexample');
});
it('gives a chunked and a single index as one list', () => {
const second = { ...blob, ref: { $link: 'bafkreisecond' } };
const record = parseRepoRecord({
name: 'repo',
refs: [{ name: 'refs/heads/main', sha }],
bundles: [
{ heads: [sha], parts: [blob], index: blob },
{ heads: [sha], parts: [blob], indexParts: [blob, second] },
{ heads: [sha], parts: [blob] },
],
createdAt: '2026-08-11T00:00:00.000Z',
});
expect(indexPartsOf(record.bundles[0])).toEqual([blob]);
expect(indexPartsOf(record.bundles[1])).toEqual([blob, second]);
expect(indexPartsOf(record.bundles[2])).toEqual([]);
});
it('reads a malformed index as absent', () => {
const record = parseRepoRecord({
name: 'repo',
refs: [],
bundles: [
{ heads: [sha], parts: [blob], index: { ref: 'wrong' } },
{ heads: [sha], parts: [blob], indexParts: [blob, { ref: 'wrong' }] },
],
createdAt: '2026-08-11T00:00:00.000Z',
});
expect(indexPartsOf(record.bundles[0])).toEqual([]);
expect(indexPartsOf(record.bundles[1])).toEqual([]);
});
it('rejects malformed refs and bundles', () => {
const base = { name: 'r', createdAt: 'now', refs: [], bundles: [] };
expect(() =>
parseRepoRecord({ ...base, refs: [{ name: 'x', sha: 'not-hex' }] }),
).toThrow(/malformed ref/);
expect(() =>
parseRepoRecord({ ...base, bundles: [{ heads: [sha], parts: [{}] }] }),
).toThrow(/malformed bundle/);
expect(() => parseRepoRecord(null)).toThrow(/not an object/);
expect(() => parseRepoRecord({ name: 'r' })).toThrow(/missing/);
});
});
describe('chunked upload', () => {
/** @param {number} limit - the largest blob the fake server takes */
function fakeClient(limit) {
/** @type {number[]} */
const taken = [];
return {
taken,
async uploadBlob(/** @type {Uint8Array} */ bytes) {
if (bytes.length > limit) {
throw new XrpcError(413, 'PayloadTooLarge', 'too large');
}
taken.push(bytes.length);
return {
$type: 'blob',
ref: { $link: `bafy${taken.length}` },
mimeType: 'application/octet-stream',
size: bytes.length,
};
},
};
}
/** @param {Record<string, string>} env */
const helper = (env, /** @type {string[]} */ logged) =>
new AtprotoRemoteHelper({
url: 'atproto://a.com/repo',
input: new PassThrough(),
write: () => {},
log: (text) => logged.push(text),
env,
});
it('halves the chunk until the server takes one, and keeps the size', async () => {
/** @type {string[]} */
const logged = [];
const push = helper(
{ ATPROTO_GIT_CHUNK_SIZE: String(64 * 1024 * 4) },
logged,
);
const client = fakeClient(64 * 1024);
const bytes = new Uint8Array(64 * 1024 * 5 + 10);
const parts = await push.uploadChunked(/** @type {any} */ (client), bytes);
// 256 KiB refused, 128 KiB refused, 64 KiB taken: six chunks.
expect(parts).toHaveLength(6);
expect(client.taken).toEqual([65536, 65536, 65536, 65536, 65536, 10]);
expect(logged.filter((line) => line.includes('refused'))).toHaveLength(2);
expect(push.chunk).toBe(64 * 1024);
// The next upload starts at the size the server took.
await push.uploadChunked(/** @type {any} */ (client), new Uint8Array(100));
expect(logged.filter((line) => line.includes('refused'))).toHaveLength(2);
});
it('gives up below the smallest chunk', async () => {
const push = helper({ ATPROTO_GIT_CHUNK_SIZE: String(64 * 1024) }, []);
const client = fakeClient(1024);
await expect(
push.uploadChunked(/** @type {any} */ (client), new Uint8Array(4096)),
).rejects.toMatchObject({ status: 413 });
});
it('passes any other refusal through', async () => {
const push = helper({}, []);
const client = {
async uploadBlob() {
throw new XrpcError(401, 'AuthenticationRequired', 'no');
},
};
await expect(
push.uploadChunked(/** @type {any} */ (client), new Uint8Array(10)),
).rejects.toMatchObject({ status: 401 });
});
});
describe('push refspec parsing', () => {
const helper = new AtprotoRemoteHelper({
url: 'atproto://a.com/repo',
input: new PassThrough(),
write: () => {},
log: () => {},
env: {},
});
it('parses plain, force, and delete refspecs', () => {
expect(helper.parsePushSpec('refs/heads/main:refs/heads/main')).toEqual({
force: false,
src: 'refs/heads/main',
dst: 'refs/heads/main',
});
expect(helper.parsePushSpec('+refs/heads/a:refs/heads/b')).toEqual({
force: true,
src: 'refs/heads/a',
dst: 'refs/heads/b',
});
expect(helper.parsePushSpec(':refs/heads/gone')).toEqual({
force: false,
src: '',
dst: 'refs/heads/gone',
});
expect(() => helper.parsePushSpec('nocolon')).toThrow(/malformed/);
});
});
describe('password session re-login', () => {
/**
* A fetch standing in for the PDS: createSession issues a numbered token,
* and putRecord accepts only the newest one. The daemon scenario is an
* expired token, which answers the same 401.
* @returns {{fetch: typeof fetch, log: string[]}}
*/
function fakePds() {
/** @type {string[]} */
const log = [];
let sessions = 0;
const impl = /** @type {typeof fetch} */ (
async (/** @type {*} */ url, /** @type {*} */ init) => {
if (String(url).includes('createSession')) {
sessions += 1;
log.push('createSession');
return new Response(
JSON.stringify({
did: 'did:plc:runner',
accessJwt: `t${sessions}`,
}),
{ headers: { 'Content-Type': 'application/json' } },
);
}
const auth = init?.headers?.Authorization ?? '';
if (auth !== `Bearer t${sessions}`) {
log.push(`putRecord 401 (${auth})`);
return new Response(
JSON.stringify({ error: 'AuthenticationRequired' }),
{ status: 401 },
);
}
log.push('putRecord ok');
return new Response(JSON.stringify({ uri: 'at://x', cid: 'c' }), {
headers: { 'Content-Type': 'application/json' },
});
}
);
return { fetch: impl, log };
}
it('signs in again on a 401 and retries the write', async () => {
const pds = fakePds();
const originalFetch = globalThis.fetch;
globalThis.fetch = pds.fetch;
try {
const client = new XrpcClient('https://pds.example');
await client.createSession('runner.test', 'app-password');
// The server stops honoring the token, as it does at expiry.
client.accessJwt = 'stale';
const result = await client.putRecord('did:plc:runner', 'c', 'r', {});
expect(result.cid).toBe('c');
expect(pds.log).toEqual([
'createSession',
'putRecord 401 (Bearer stale)',
'createSession',
'putRecord ok',
]);
} finally {
globalThis.fetch = originalFetch;
}
});
it('does not retry without stored credentials', async () => {
const pds = fakePds();
const originalFetch = globalThis.fetch;
globalThis.fetch = pds.fetch;
try {
const client = new XrpcClient('https://pds.example');
client.accessJwt = 'stale';
await expect(
client.putRecord('did:plc:runner', 'c', 'r', {}),
).rejects.toMatchObject({ status: 401 });
expect(pds.log).toEqual(['putRecord 401 (Bearer stale)']);
} finally {
globalThis.fetch = originalFetch;
}
});
});
describe('session store', async () => {
const { loadSession, saveSession, deleteSession, findSessionByDid } =
await import('../src/session-store.js');
const { mkdtempSync, rmSync, statSync } = await import('node:fs');
const { tmpdir } = await import('node:os');
const { join } = await import('node:path');
const session = /** @type {any} */ ({
did: 'did:plc:abc',
identifier: 'alice.test',
service: 'https://pds.example',
tokenEndpoint: 'https://pds.example/oauth/token',
clientId: 'http://127.0.0.1:1/',
scope: 'atproto',
accessToken: 'a',
refreshToken: 'r',
expiresAt: 1,
privateJwk: {},
});
it('round-trips a session under its identifier, owner-only', () => {
const dir = mkdtempSync(join(tmpdir(), 'pdsjs-git-store-'));
const env = { ATPROTO_GIT_CONFIG_DIR: dir };
try {
saveSession(env, 'alice.test', session);
expect(loadSession(env, 'alice.test')).toEqual(session);
expect(loadSession(env, 'bob.test')).toBeNull();
const mode = statSync(join(dir, 'alice.test.json')).mode & 0o777;
expect(mode).toBe(0o600);
expect(findSessionByDid(env, 'did:plc:abc')).toEqual(session);
expect(findSessionByDid(env, 'did:plc:other')).toBeNull();
deleteSession(env, 'alice.test');
expect(loadSession(env, 'alice.test')).toBeNull();
} finally {
rmSync(dir, { recursive: true, force: true });
}
});
it('answers null rather than throwing when nothing is stored', () => {
const env = { ATPROTO_GIT_CONFIG_DIR: '/nonexistent/atproto-git' };
expect(loadSession(env, 'alice.test')).toBeNull();
expect(findSessionByDid(env, 'did:plc:abc')).toBeNull();
});
});
describe('parseCommit', async () => {
const { parseCommit } = await import('../src/objects.js');
const encode = (/** @type {string} */ text) => new TextEncoder().encode(text);
it('reads headers, idents, and the message', () => {
const commit = parseCommit(
encode(
[
'tree aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa',
'parent bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb',
'parent cccccccccccccccccccccccccccccccccccccccc',
'author Alice <alice@example.com> 1700000000 +0100',
'committer Bob <bob@example.com> 1700000100 -0500',
'',
'merge the thing',
'',
'longer body here',
].join('\n'),
),
);
expect(commit.tree).toBe('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa');
expect(commit.parents).toHaveLength(2);
expect(commit.author).toBe('Alice <alice@example.com>');
expect(commit.authoredAt).toBe(1700000000000);
expect(commit.committer).toBe('Bob <bob@example.com>');
expect(commit.committedAt).toBe(1700000100000);
expect(commit.message).toBe('merge the thing\n\nlonger body here');
});
it('skips gpgsig continuation lines and survives a bare header', () => {
const commit = parseCommit(
encode(
[
'tree aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa',
'gpgsig -----BEGIN PGP SIGNATURE-----',
' abcdefabcdef',
' -----END PGP SIGNATURE-----',
'author A <a@b> 1700000000 +0000',
'committer A <a@b> 1700000000 +0000',
].join('\n'),
),
);
expect(commit.parents).toEqual([]);
expect(commit.message).toBe('');
expect(commit.author).toBe('A <a@b>');
});
});