chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
import { execFile, execFileSync } from 'node:child_process';
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
import { createServer } from 'node:http';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { promisify } from 'node:util';
import { afterEach, describe, expect, it } from 'vitest';
import { generateDpopKey } from '../src/dpop.js';
import { saveSession } from '../src/session-store.js';
const execFileAsync = promisify(execFile);
const cliPath = fileURLToPath(new URL('../src/cli.js', import.meta.url));
/** @type {Array<{close: () => Promise<unknown>}>} */
const servers = [];
/** The key this account's own copy is at, which is not its name. */
const OWN_KEY = '3lforkkey';
/** The canonical repository this copy was taken from. */
const CANONICAL = 'at://did:plc:canonical/dev.pdsjs.git.repo/3lcanonkey';
afterEach(async () => {
for (const server of servers.splice(0)) await server.close();
});
describe('git-remote-atproto pr create', () => {
it('writes the pull record at the branch key, requiring none exists', async () => {
/** @type {Record<string, unknown>|null} */
let put = null;
const server = createServer((request, response) => {
const url = new URL(request.url || '/', 'http://127.0.0.1');
// A copy is its holder's own record under their own key, so the CLI
// finds it by the name it carries.
if (
request.method === 'GET' &&
url.pathname.endsWith('repo.listRecords')
) {
const owner = url.searchParams.get('repo');
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
records: [
{
uri: `at://${owner}/dev.pdsjs.git.repo/${OWN_KEY}`,
cid: 'fork-cid',
value: {
name: 'project',
...(CANONICAL ? { upstream: CANONICAL } : {}),
},
},
],
}),
);
return;
}
if (request.method === 'GET' && url.pathname.endsWith('repo.getRecord')) {
const owner = url.searchParams.get('repo');
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
uri: CANONICAL,
cid: 'canonical-cid',
value: { name: 'project' },
}),
);
return;
}
if (
request.method === 'POST' &&
url.pathname.endsWith('repo.putRecord')
) {
let body = '';
request.on('data', (chunk) => {
body += chunk;
});
request.on('end', () => {
put = JSON.parse(body);
const rkey = /** @type {string} */ (put?.rkey);
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
uri: `at://did:plc:alice/dev.pdsjs.git.pull/${rkey}`,
cid: 'pull-cid',
}),
);
});
return;
}
response.writeHead(404).end();
});
await new Promise((resolve) =>
server.listen({ port: 0, host: '127.0.0.1' }, () => resolve(undefined)),
);
servers.push({
close: () =>
new Promise((resolve) => {
server.close(() => resolve(undefined));
}),
});
const address = /** @type {import('node:net').AddressInfo} */ (
server.address()
);
const service = `http://127.0.0.1:${address.port}`;
const root = mkdtempSync(join(tmpdir(), 'pdsjs-git-pr-'));
const repo = join(root, 'repo');
const config = join(root, 'config');
mkdirSync(repo);
try {
await execFileAsync('git', ['init', '-b', 'main'], { cwd: repo });
await execFileAsync('git', ['config', 'user.name', 'Test'], {
cwd: repo,
});
await execFileAsync('git', ['config', 'user.email', 'test@example.com'], {
cwd: repo,
});
writeFileSync(join(repo, 'README.md'), '# project\n');
await execFileAsync('git', ['add', 'README.md'], { cwd: repo });
await execFileAsync('git', ['commit', '-m', 'initial'], { cwd: repo });
await execFileAsync('git', ['switch', '-c', 'feature/cli'], {
cwd: repo,
});
writeFileSync(join(repo, 'change.txt'), 'change\n');
await execFileAsync('git', ['add', 'change.txt'], { cwd: repo });
await execFileAsync('git', ['commit', '-m', 'change'], { cwd: repo });
await execFileAsync(
'git',
['remote', 'add', 'origin', 'atproto://did:plc:alice/project'],
{ cwd: repo },
);
const key = await generateDpopKey({ extractable: true });
const privateJwk = await crypto.subtle.exportKey('jwk', key.privateKey);
saveSession({ ATPROTO_GIT_CONFIG_DIR: config }, 'alice.test', {
did: 'did:plc:alice',
identifier: 'alice.test',
service,
tokenEndpoint: `${service}/oauth/token`,
clientId: 'http://127.0.0.1:1/',
scope: 'atproto repo:dev.pdsjs.git.pull',
accessToken: 'access',
refreshToken: 'refresh',
expiresAt: Date.now() + 60_000,
privateJwk,
});
const { stdout } = await execFileAsync(
process.execPath,
[
cliPath,
'pr',
'create',
'--title',
'CLI change',
'--body',
'Created from the checkout.',
'--draft',
],
{
cwd: repo,
env: {
...process.env,
ATPROTO_GIT_CONFIG_DIR: config,
ATPROTO_GIT_SERVICE: service,
},
},
);
const sha = (
await execFileAsync('git', ['rev-parse', 'HEAD'], { cwd: repo })
).stdout.trim();
// Keyed by the canonical repository's key, which both sides hold.
expect(stdout.trim()).toBe(
'at://did:plc:alice/dev.pdsjs.git.pull/3lcanonkey:refs_heads_feature_cli',
);
expect(put).toMatchObject({
collection: 'dev.pdsjs.git.pull',
repo: 'did:plc:alice',
rkey: '3lcanonkey:refs_heads_feature_cli',
swapRecord: null,
record: {
$type: 'dev.pdsjs.git.pull',
subject: {
uri: CANONICAL,
cid: 'canonical-cid',
},
// The copy holding the branch, under this account's own key.
repo: OWN_KEY,
ref: 'refs/heads/feature/cli',
sha,
title: 'CLI change',
status: 'draft',
note: 'Created from the checkout.',
},
});
expect(
/** @type {Record<string, unknown>} */ (/** @type {unknown} */ (put))
.record,
).toMatchObject({
createdAt: expect.any(String),
});
} finally {
rmSync(root, { recursive: true, force: true });
}
});
});
describe('git-remote-atproto pr update', () => {
/**
* A checkout on a feature branch with a remote into did:plc:alice's copy.
* @returns {{root: string, repo: string, config: string}}
*/
function checkout() {
const root = mkdtempSync(join(tmpdir(), 'pdsjs-git-pr-update-'));
const repo = join(root, 'repo');
const config = join(root, 'config');
mkdirSync(repo);
execFileSync('git', ['init', '-b', 'main'], { cwd: repo, stdio: 'ignore' });
execFileSync('git', ['config', 'user.name', 'Test'], { cwd: repo });
execFileSync('git', ['config', 'user.email', 'test@example.com'], {
cwd: repo,
});
writeFileSync(join(repo, 'README.md'), '# project\n');
execFileSync('git', ['add', 'README.md'], { cwd: repo });
execFileSync('git', ['commit', '-m', 'initial'], { cwd: repo });
execFileSync('git', ['switch', '-c', 'feature/cli'], { cwd: repo });
writeFileSync(join(repo, 'change.txt'), 'change\n');
execFileSync('git', ['add', 'change.txt'], { cwd: repo });
execFileSync('git', ['commit', '-m', 'change'], { cwd: repo });
execFileSync(
'git',
['remote', 'add', 'origin', 'atproto://did:plc:alice/project'],
{ cwd: repo },
);
return { root, repo, config };
}
/** @param {string} config @param {string} service */
async function session(config, service) {
const key = await generateDpopKey({ extractable: true });
const privateJwk = await crypto.subtle.exportKey('jwk', key.privateKey);
saveSession({ ATPROTO_GIT_CONFIG_DIR: config }, 'alice.test', {
did: 'did:plc:alice',
identifier: 'alice.test',
service,
tokenEndpoint: `${service}/oauth/token`,
clientId: 'http://127.0.0.1:1/',
scope: 'atproto repo:dev.pdsjs.git.pull',
accessToken: 'access',
refreshToken: 'refresh',
expiresAt: Date.now() + 60_000,
privateJwk,
});
}
it('rewrites the record at the branch key, keeping what a flag did not set', async () => {
/** @type {Array<{swapRecord: string|null, record: Record<string, unknown>}>} */
const puts = [];
const server = createServer((request, response) => {
const url = new URL(request.url || '/', 'http://127.0.0.1');
// A copy is its holder's own record under their own key, so the CLI
// finds it by the name it carries.
if (
request.method === 'GET' &&
url.pathname.endsWith('repo.listRecords')
) {
const owner = url.searchParams.get('repo');
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
records: [
{
uri: `at://${owner}/dev.pdsjs.git.repo/${OWN_KEY}`,
cid: 'fork-cid',
value: {
name: 'project',
...(CANONICAL ? { upstream: CANONICAL } : {}),
},
},
],
}),
);
return;
}
if (request.method === 'GET' && url.pathname.endsWith('repo.getRecord')) {
const owner = url.searchParams.get('repo');
const rkey = url.searchParams.get('rkey');
// The pull record at the branch's derived key, or the canonical
// repository the fork names.
if (rkey === `3lcanonkey:refs_heads_feature_cli`) {
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
uri: `at://${owner}/dev.pdsjs.git.pull/${rkey}`,
cid: 'pull-cid-1',
value: {
subject: { uri: CANONICAL },
repo: OWN_KEY,
ref: 'refs/heads/feature/cli',
sha: 'oldsha',
title: 'First title',
status: 'draft',
note: 'First note',
createdAt: '2026-08-01T00:00:00Z',
},
}),
);
return;
}
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
uri: CANONICAL,
cid: 'canonical-cid',
value: { name: 'project' },
}),
);
return;
}
if (
request.method === 'POST' &&
url.pathname.endsWith('repo.putRecord')
) {
let body = '';
request.on('data', (chunk) => {
body += chunk;
});
request.on('end', () => {
puts.push(JSON.parse(body));
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
uri: `at://did:plc:alice/dev.pdsjs.git.pull/${JSON.parse(body).rkey}`,
cid: 'pull-cid-2',
}),
);
});
return;
}
response.writeHead(404).end();
});
await new Promise((resolve) =>
server.listen({ port: 0, host: '127.0.0.1' }, () => resolve(undefined)),
);
servers.push({
close: () =>
new Promise((resolve) => {
server.close(() => resolve(undefined));
}),
});
const address = /** @type {import('node:net').AddressInfo} */ (
server.address()
);
const service = `http://127.0.0.1:${address.port}`;
const { root, repo, config } = checkout();
try {
await session(config, service);
const env = {
...process.env,
ATPROTO_GIT_CONFIG_DIR: config,
ATPROTO_GIT_SERVICE: service,
};
writeFileSync(join(repo, 'change.txt'), 'change again\n');
execFileSync('git', ['add', 'change.txt'], { cwd: repo });
execFileSync('git', ['commit', '-m', 'change again'], { cwd: repo });
const sha = execFileSync('git', ['rev-parse', 'HEAD'], {
cwd: repo,
encoding: 'utf8',
}).trim();
const { stdout } = await execFileAsync(
process.execPath,
[
cliPath,
'pr',
'update',
'--title',
'Second title',
'--body',
'Second body',
],
{ cwd: repo, env },
);
expect(stdout.trim()).toBe(
'at://did:plc:alice/dev.pdsjs.git.pull/3lcanonkey:refs_heads_feature_cli',
);
expect(puts).toHaveLength(1);
expect(puts[0]).toMatchObject({
collection: 'dev.pdsjs.git.pull',
repo: 'did:plc:alice',
rkey: '3lcanonkey:refs_heads_feature_cli',
swapRecord: 'pull-cid-1',
record: {
$type: 'dev.pdsjs.git.pull',
subject: { uri: CANONICAL, cid: 'canonical-cid' },
repo: OWN_KEY,
ref: 'refs/heads/feature/cli',
sha,
title: 'Second title',
status: 'draft',
note: 'Second body',
},
});
} finally {
rmSync(root, { recursive: true, force: true });
}
});
it('refuses to update when nothing would change', async () => {
const server = createServer((request, response) => {
const url = new URL(request.url || '/', 'http://127.0.0.1');
// A copy is its holder's own record under their own key, so the CLI
// finds it by the name it carries.
if (
request.method === 'GET' &&
url.pathname.endsWith('repo.listRecords')
) {
const owner = url.searchParams.get('repo');
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
records: [
{
uri: `at://${owner}/dev.pdsjs.git.repo/${OWN_KEY}`,
cid: 'fork-cid',
value: { name: 'project' },
},
],
}),
);
return;
}
// Nothing at the branch's key: the copy is listed, and the pull
// record the update would rewrite is not there.
if (request.method === 'GET' && url.pathname.endsWith('repo.getRecord')) {
response.writeHead(404).end();
return;
}
response.writeHead(404).end();
});
await new Promise((resolve) =>
server.listen({ port: 0, host: '127.0.0.1' }, () => resolve(undefined)),
);
servers.push({
close: () =>
new Promise((resolve) => {
server.close(() => resolve(undefined));
}),
});
const address = /** @type {import('node:net').AddressInfo} */ (
server.address()
);
const { root, repo, config } = checkout();
try {
await session(config, `http://127.0.0.1:${address.port}`);
await expect(
execFileAsync(process.execPath, [cliPath, 'pr', 'update'], {
cwd: repo,
env: {
...process.env,
ATPROTO_GIT_CONFIG_DIR: config,
ATPROTO_GIT_SERVICE: `http://127.0.0.1:${address.port}`,
},
}),
).rejects.toThrow(/at least one of --title, --body or --draft/);
} finally {
rmSync(root, { recursive: true, force: true });
}
});
it('refuses to update a branch with no pull request', async () => {
const server = createServer((request, response) => {
const url = new URL(request.url || '/', 'http://127.0.0.1');
// A copy is its holder's own record under their own key, so the CLI
// finds it by the name it carries.
if (
request.method === 'GET' &&
url.pathname.endsWith('repo.listRecords')
) {
const owner = url.searchParams.get('repo');
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
records: [
{
uri: `at://${owner}/dev.pdsjs.git.repo/${OWN_KEY}`,
cid: 'fork-cid',
value: { name: 'project' },
},
],
}),
);
return;
}
if (request.method === 'GET' && url.pathname.endsWith('repo.getRecord')) {
const owner = url.searchParams.get('repo');
const rkey = url.searchParams.get('rkey');
// The branch's pull record does not exist yet; the copy's repo record
// does.
if (rkey === `${OWN_KEY}:refs_heads_feature_cli`) {
response.writeHead(404).end();
return;
}
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
uri: `at://${owner}/dev.pdsjs.git.repo/${OWN_KEY}`,
cid: 'fork-cid',
value: { name: 'project' },
}),
);
return;
}
response.writeHead(404).end();
});
await new Promise((resolve) =>
server.listen({ port: 0, host: '127.0.0.1' }, () => resolve(undefined)),
);
servers.push({
close: () =>
new Promise((resolve) => {
server.close(() => resolve(undefined));
}),
});
const address = /** @type {import('node:net').AddressInfo} */ (
server.address()
);
const service = `http://127.0.0.1:${address.port}`;
const { root, repo, config } = checkout();
try {
await session(config, service);
await expect(
execFileAsync(
process.execPath,
[cliPath, 'pr', 'update', '--title', 'No prior pull'],
{
cwd: repo,
env: {
...process.env,
ATPROTO_GIT_CONFIG_DIR: config,
ATPROTO_GIT_SERVICE: service,
},
},
),
).rejects.toThrow(/no pull request for feature\/cli/);
} finally {
rmSync(root, { recursive: true, force: true });
}
});
});
describe('git-remote-atproto pr merge', () => {
/**
* A checkout on a feature branch with a remote into did:plc:alice's copy.
* @returns {{root: string, repo: string, config: string}}
*/
function checkout() {
const root = mkdtempSync(join(tmpdir(), 'pdsjs-git-pr-merge-'));
const repo = join(root, 'repo');
const config = join(root, 'config');
mkdirSync(repo);
execFileSync('git', ['init', '-b', 'main'], { cwd: repo, stdio: 'ignore' });
execFileSync('git', ['config', 'user.name', 'Test'], { cwd: repo });
execFileSync('git', ['config', 'user.email', 'test@example.com'], {
cwd: repo,
});
writeFileSync(join(repo, 'README.md'), '# project\n');
execFileSync('git', ['add', 'README.md'], { cwd: repo });
execFileSync('git', ['commit', '-m', 'initial'], { cwd: repo });
execFileSync('git', ['switch', '-c', 'feature/cli'], { cwd: repo });
writeFileSync(join(repo, 'change.txt'), 'change\n');
execFileSync('git', ['add', 'change.txt'], { cwd: repo });
execFileSync('git', ['commit', '-m', 'change'], { cwd: repo });
execFileSync(
'git',
['remote', 'add', 'origin', 'atproto://did:plc:alice/project'],
{ cwd: repo },
);
return { root, repo, config };
}
/** @param {string} config @param {string} service @param {string} did */
async function session(config, service, did) {
const key = await generateDpopKey({ extractable: true });
const privateJwk = await crypto.subtle.exportKey('jwk', key.privateKey);
saveSession({ ATPROTO_GIT_CONFIG_DIR: config }, `${did}-test`, {
did,
identifier: `${did}-test`,
service,
tokenEndpoint: `${service}/oauth/token`,
clientId: 'http://127.0.0.1:1/',
scope: 'atproto repo:dev.pdsjs.git.repo',
accessToken: did.endsWith('canonical') ? 'owner-access' : 'access',
refreshToken: 'refresh',
expiresAt: Date.now() + 60_000,
privateJwk,
});
}
/**
* A server for one merge: the copy's repo record, the canonical record it
* names when given, and the mergePull procedure.
* @param {{upstream?: boolean, owner?: boolean}} [fork] - upstream names the
* canonical record; owner answers the canonical copy's record
* @returns {{server: ReturnType<typeof createServer>, puts: Array<{body: Record<string, unknown>, auth: string|null}>}}
*/
function mergeServer(fork) {
/** @type {Array<{body: Record<string, unknown>, auth: string|null}>} */
const puts = [];
const server = createServer((request, response) => {
const url = new URL(request.url || '/', 'http://127.0.0.1');
// A copy is its holder's own record under their own key, so the CLI
// finds it by the name it carries.
if (
request.method === 'GET' &&
url.pathname.endsWith('repo.listRecords')
) {
const owner = url.searchParams.get('repo');
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
records: [
{
uri: `at://${owner}/dev.pdsjs.git.repo/${OWN_KEY}`,
cid: 'fork-cid',
value: {
name: 'project',
...((fork?.upstream ? CANONICAL : '')
? { upstream: fork?.upstream ? CANONICAL : '' }
: {}),
},
},
],
}),
);
return;
}
if (request.method === 'GET' && url.pathname.endsWith('repo.getRecord')) {
const owner = url.searchParams.get('repo');
const rkey = url.searchParams.get('rkey');
// The canonical record a fork names, or the copy's own.
if (owner === 'did:plc:canonical') {
if (!fork?.owner) {
response.writeHead(404).end();
return;
}
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
uri: 'at://did:plc:canonical/dev.pdsjs.git.repo/project',
cid: 'canonical-cid',
value: { name: 'project' },
}),
);
return;
}
if (rkey === 'project') {
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
uri: `at://${owner}/dev.pdsjs.git.repo/project`,
cid: 'fork-cid',
value: {
name: 'project',
...(fork?.upstream
? {
upstream:
'at://did:plc:canonical/dev.pdsjs.git.repo/project',
}
: {}),
},
}),
);
return;
}
response.writeHead(404).end();
return;
}
if (
request.method === 'POST' &&
url.pathname.endsWith('dev.pdsjs.git.mergePull')
) {
let body = '';
request.on('data', (chunk) => {
body += chunk;
});
request.on('end', () => {
puts.push({
body: JSON.parse(body),
auth: /** @type {string|null} */ (
request.headers.authorization ?? null
),
});
response.setHeader('content-type', 'application/json').end(
JSON.stringify({
sha: 'merge-sha',
ref: 'refs/heads/main',
mode: 'merge',
objects: 1,
}),
);
});
return;
}
response.writeHead(404).end();
});
return { server, puts };
}
it('merges a branch of the remote copy with the branch session', async () => {
const { server, puts } = mergeServer();
await new Promise((resolve) =>
server.listen({ port: 0, host: '127.0.0.1' }, () => resolve(undefined)),
);
servers.push({
close: () =>
new Promise((resolve) => {
server.close(() => resolve(undefined));
}),
});
const address = /** @type {import('node:net').AddressInfo} */ (
server.address()
);
const service = `http://127.0.0.1:${address.port}`;
const { root, repo, config } = checkout();
try {
await session(config, service, 'did:plc:alice');
const { stdout } = await execFileAsync(
process.execPath,
[cliPath, 'pr', 'merge'],
{
cwd: repo,
env: {
...process.env,
ATPROTO_GIT_CONFIG_DIR: config,
ATPROTO_GIT_SERVICE: service,
},
},
);
expect(stdout.trim()).toBe('merge-sha');
expect(puts).toHaveLength(1);
expect(puts[0].body).toMatchObject({
name: 'project',
did: 'did:plc:alice',
ref: 'refs/heads/feature/cli',
mode: 'merge',
handle: 'did:plc:alice-test',
});
// The merge rides the branch's own session, which is the owner's.
expect(puts[0].auth).toBe('DPoP access');
} finally {
rmSync(root, { recursive: true, force: true });
}
});
it('refuses to merge a fork without the owner session', async () => {
const { server } = mergeServer({ upstream: true, owner: true });
await new Promise((resolve) =>
server.listen({ port: 0, host: '127.0.0.1' }, () => resolve(undefined)),
);
servers.push({
close: () =>
new Promise((resolve) => {
server.close(() => resolve(undefined));
}),
});
const address = /** @type {import('node:net').AddressInfo} */ (
server.address()
);
const service = `http://127.0.0.1:${address.port}`;
const { root, repo, config } = checkout();
try {
await session(config, service, 'did:plc:alice');
await expect(
execFileAsync(process.execPath, [cliPath, 'pr', 'merge'], {
cwd: repo,
env: {
...process.env,
ATPROTO_GIT_CONFIG_DIR: config,
ATPROTO_GIT_SERVICE: service,
},
}),
).rejects.toThrow(/only the repository's owner can merge/);
} finally {
rmSync(root, { recursive: true, force: true });
}
});
it('merges a fork with the owner session, as a squash', async () => {
const { server, puts } = mergeServer({ upstream: true, owner: true });
await new Promise((resolve) =>
server.listen({ port: 0, host: '127.0.0.1' }, () => resolve(undefined)),
);
servers.push({
close: () =>
new Promise((resolve) => {
server.close(() => resolve(undefined));
}),
});
const address = /** @type {import('node:net').AddressInfo} */ (
server.address()
);
const service = `http://127.0.0.1:${address.port}`;
const { root, repo, config } = checkout();
try {
await session(config, service, 'did:plc:alice');
await session(config, service, 'did:plc:canonical');
const { stdout } = await execFileAsync(
process.execPath,
[
cliPath,
'pr',
'merge',
'--mode',
'squash',
'--subject',
'My squashed change',
],
{
cwd: repo,
env: {
...process.env,
ATPROTO_GIT_CONFIG_DIR: config,
ATPROTO_GIT_SERVICE: service,
},
},
);
expect(stdout.trim()).toBe('merge-sha');
expect(puts).toHaveLength(1);
expect(puts[0].body).toMatchObject({
name: 'project',
did: 'did:plc:alice',
ref: 'refs/heads/feature/cli',
mode: 'squash',
handle: 'did:plc:alice-test',
subject: 'My squashed change',
});
// The merge rides the owner's session, not the branch author's.
expect(puts[0].auth).toBe('DPoP owner-access');
} finally {
rmSync(root, { recursive: true, force: true });
}
});
});