chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
#!/usr/bin/env node
/**
* Move every git repository record off the repository's name and onto a
* minted key, then re-point everything that named the old address.
*
* A repository's record key was its name, so renaming one changed its
* AT-URI and broke every statement about it. The key is now minted and the
* name is a field. This is the one-time move.
*
* Two passes, because statements about a repository live in their authors'
* own accounts rather than the repository's:
*
* keys moves the repository records and their configs on one account,
* and writes the old-address to new-address map.
* remap rewrites, on any one account, every record naming an old address:
* the issues, reviews, pull requests, checks and forks.
*
* Run `keys` once per account holding repositories, then `remap` on every
* account that has spoken about them, including each runner's.
*
* Usage: node scripts/migrate-git-repo-keys.mjs keys --service <url> [--out map.json] [--dry-run]
* node scripts/migrate-git-repo-keys.mjs remap --service <url> --map map.json [--dry-run]
* Env: PDS_PASSWORD the account password for --service
*/
import { readFile, writeFile } from 'node:fs/promises';
const REPO = 'dev.pdsjs.git.repo';
const CONFIG = 'dev.pdsjs.git.config';
/** Every collection a repository's address can appear in, and where. */
const SPEAKERS = [
{ collection: 'dev.pdsjs.git.issue', at: 'subject' },
{ collection: 'dev.pdsjs.git.review', at: 'subject' },
{ collection: 'dev.pdsjs.git.pull', at: 'subject', keyed: 'pull' },
{ collection: 'dev.pdsjs.git.check', at: 'subject' },
{ collection: 'dev.pdsjs.git.latestCheck', at: 'checks', keyed: 'latest' },
{ collection: 'dev.pdsjs.git.checkRequest', at: 'repo', keyed: 'request' },
{ collection: REPO, at: 'upstream' },
];
/** Characters an atproto record key permits. */
const RKEY_SAFE = /[^A-Za-z0-9._~:-]/g;
/** Writes per commit, so one repository's move is one commit. */
const BATCH = 50;
const TID_CHARS = '234567abcdefghijklmnopqrstuvwxyz';
const clockId = Math.floor(Math.random() * 1024);
let lastTid = 0;
/** A minted record key, in the atproto TID form. */
function createRepoKey() {
let micros = Date.now() * 1000;
if (micros <= lastTid) micros = lastTid + 1;
lastTid = micros;
let tid = '';
let left = micros;
for (let i = 0; i < 11; i += 1) {
tid = TID_CHARS[left & 31] + tid;
left = Math.floor(left / 32);
}
return tid + TID_CHARS[(clockId >> 5) & 31] + TID_CHARS[clockId & 31];
}
/** @param {string[]} argv */
function parseArgs(argv) {
const [command, ...rest] = argv;
/** @type {Record<string, string|boolean>} */
const flags = {};
for (let i = 0; i < rest.length; i += 1) {
const flag = rest[i];
if (!flag.startsWith('--')) continue;
const name = flag.slice(2);
const next = rest[i + 1];
if (next && !next.startsWith('--')) {
flags[name] = next;
i += 1;
} else {
flags[name] = true;
}
}
return { command, flags };
}
/** The DID and an authenticated caller for one service. */
async function signIn(service, password) {
const did = (await (await fetch(`${service}/.well-known/atproto-did`)).text())
.trim()
.replace(/^"|"$/g, '');
const res = await fetch(`${service}/xrpc/com.atproto.server.createSession`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ identifier: did, password }),
});
if (!res.ok) throw new Error(`sign-in failed (${res.status})`);
const { accessJwt } = await res.json();
return { did, auth: { Authorization: `Bearer ${accessJwt}` } };
}
/** Every record of one collection, following the cursor. */
async function listAll(service, did, collection) {
const rows = [];
let cursor = '';
for (;;) {
const params = new URLSearchParams({ repo: did, collection, limit: '100' });
if (cursor) params.set('cursor', cursor);
const res = await fetch(
`${service}/xrpc/com.atproto.repo.listRecords?${params}`,
);
if (!res.ok) return rows;
const body = await res.json();
rows.push(...(body.records ?? []));
if (!body.cursor) return rows;
cursor = body.cursor;
}
}
/** One applyWrites batch, or a line saying what it would have written. */
async function apply(ctx, writes, label) {
if (writes.length === 0) return;
for (const write of writes) {
const verb = write.$type.split('#')[1];
console.log(` ${verb} ${write.collection}/${write.rkey}`);
}
if (ctx.dryRun) return;
for (let at = 0; at < writes.length; at += BATCH) {
const res = await fetch(
`${ctx.service}/xrpc/com.atproto.repo.applyWrites`,
{
method: 'POST',
headers: { 'Content-Type': 'application/json', ...ctx.auth },
body: JSON.stringify({
repo: ctx.did,
validate: false,
writes: writes.slice(at, at + BATCH),
}),
},
);
if (!res.ok) {
throw new Error(`${label} failed: ${res.status} ${await res.text()}`);
}
}
}
const create = (collection, rkey, value) => ({
$type: 'com.atproto.repo.applyWrites#create',
collection,
rkey,
value,
});
const remove = (collection, rkey) => ({
$type: 'com.atproto.repo.applyWrites#delete',
collection,
rkey,
});
const rkeyOf = (uri) => uri.split('/').pop() ?? '';
/**
* Move each repository record, and the config beside it, onto a minted key.
*
* The config moves in its own commit, and first. The write guard reads a
* repository's config for the branches it protects, and refuses to delete a
* record that holds one; with the config already moved, the delete of the
* old repository record reads no config and stands.
*/
async function migrateKeys(ctx, out) {
const rows = await listAll(ctx.service, ctx.did, REPO);
const configs = new Map(
(await listAll(ctx.service, ctx.did, CONFIG)).map((row) => [
rkeyOf(row.uri),
row.value,
]),
);
/** @type {Record<string, {uri: string, rkey: string, oldRkey: string, name: string}>} */
const map = {};
for (const row of rows) {
const oldRkey = rkeyOf(row.uri);
const name = typeof row.value?.name === 'string' ? row.value.name : oldRkey;
if (oldRkey !== name) {
console.log(`${name}: already at ${oldRkey}`);
continue;
}
const rkey = createRepoKey();
console.log(`${name}: ${oldRkey} -> ${rkey}`);
const config = configs.get(oldRkey);
if (config) {
await apply(
ctx,
[create(CONFIG, rkey, config), remove(CONFIG, oldRkey)],
`${name} config`,
);
}
await apply(
ctx,
[create(REPO, rkey, row.value), remove(REPO, oldRkey)],
`${name} record`,
);
map[row.uri] = {
uri: `at://${ctx.did}/${REPO}/${rkey}`,
rkey,
oldRkey,
name,
};
}
await writeFile(out, `${JSON.stringify(map, null, 2)}\n`);
console.log(`\n${Object.keys(map).length} moved; map written to ${out}`);
}
/** The value with its subject re-pointed, or null where it names nothing moved. */
function repoint(collection, at, value, map) {
if (at === 'subject') {
const moved = map[value?.subject?.uri];
if (!moved) return null;
return { ...value, subject: { ...value.subject, uri: moved.uri } };
}
if (at === 'upstream') {
const moved = map[value?.upstream];
if (!moved) return null;
return { ...value, upstream: moved.uri };
}
if (at === 'repo') {
const moved = Object.values(map).find((one) => one.oldRkey === value?.repo);
if (!moved) return null;
return { ...value, repo: moved.rkey };
}
// A latest-check record holds one entry per workflow, each with its own
// subject. A record written before entries existed carries one workflow's
// fields at the top level, so both shapes are read.
const checks = Array.isArray(value?.checks) ? value.checks : [];
const point = (/** @type {Object} */ one) =>
map[one?.subject?.uri]
? { ...one, subject: { ...one.subject, uri: map[one.subject.uri].uri } }
: one;
const moved =
checks.map((entry) => map[entry?.subject?.uri]).find(Boolean) ??
map[value?.subject?.uri];
if (!moved) return null;
return {
...point(value),
...(checks.length > 0 ? { checks: checks.map(point) } : {}),
};
}
/**
* The key a record moves to, where its own key names the repository. Each is
* derived from the record's own fields, the way its writer derives it.
*/
function rekey(keyed, oldRkey, value, map) {
const safe = (text) => text.replace(RKEY_SAFE, '_');
const ref = typeof value?.ref === 'string' ? value.ref : '';
if (!keyed || !ref) return oldRkey;
if (keyed === 'pull') {
const moved = map[value?.subject?.uri];
return moved ? safe(`${moved.rkey}:${ref}`) : oldRkey;
}
if (keyed === 'request') {
const moved = Object.values(map).find((one) => one.oldRkey === value?.repo);
return moved ? safe(`${moved.rkey}:${ref}`) : oldRkey;
}
// A latest-check record is keyed by the repository's owner as well, which
// the subject of any of its entries names. A record from before entries
// existed names it at the top level.
const checks = Array.isArray(value?.checks) ? value.checks : [];
const named =
checks.find((entry) => map[entry?.subject?.uri]) ??
(map[value?.subject?.uri] ? value : null);
if (!named) return oldRkey;
const moved = map[named.subject.uri];
return safe(`${named.subject.uri.split('/')[2]}:${moved.rkey}:${ref}`);
}
/**
* Rewrite every record on this account that named a moved repository.
*
* A derived key names a repository and a ref, so several records for one
* branch collapse onto one key. The newest is the one to keep: that is the
* word a reader already takes, and writing the others after it would put a
* superseded title back on the branch.
*/
async function remapAddresses(ctx, map) {
for (const { collection, at, keyed } of SPEAKERS) {
const rows = await listAll(ctx.service, ctx.did, collection);
/** @type {Map<string, {value: Object, at: string}>} */
const byKey = new Map();
/** @type {string[]} */
const stale = [];
let touched = 0;
for (const row of rows) {
const oldRkey = rkeyOf(row.uri);
const value = repoint(collection, at, row.value, map);
if (!value) continue;
touched += 1;
const rkey = rekey(keyed, oldRkey, row.value, map);
const when = String(row.value?.createdAt ?? '');
const held = byKey.get(rkey);
if (!held || when >= held.at) byKey.set(rkey, { value, at: when });
if (rkey !== oldRkey) stale.push(oldRkey);
}
/** @type {unknown[]} */
const writes = [
...[...byKey].map(([rkey, one]) => create(collection, rkey, one.value)),
...stale
.filter((rkey) => !byKey.has(rkey))
.map((rkey) => remove(collection, rkey)),
];
if (writes.length === 0) continue;
const collapsed = touched - byKey.size;
console.log(
`${collection}: ${writes.length} writes` +
(collapsed > 0 ? ` (${collapsed} duplicates dropped)` : ''),
);
await apply(ctx, writes, collection);
}
}
async function main() {
const { command, flags } = parseArgs(process.argv.slice(2));
const service = String(flags.service ?? '');
if (!service || (command !== 'keys' && command !== 'remap')) {
console.error(
'Usage: migrate-git-repo-keys.mjs keys|remap --service <url> [--map map.json] [--out map.json] [--dry-run]',
);
process.exit(1);
}
const password = process.env.PDS_PASSWORD;
if (!password) throw new Error('PDS_PASSWORD is required');
const { did, auth } = await signIn(service, password);
const ctx = { service, did, auth, dryRun: Boolean(flags['dry-run']) };
if (ctx.dryRun) console.log('dry run: nothing is written\n');
console.log(`${service} is ${did}\n`);
if (command === 'keys') {
await migrateKeys(ctx, String(flags.out ?? 'git-repo-keys.json'));
return;
}
const map = JSON.parse(await readFile(String(flags.map ?? ''), 'utf8'));
await remapAddresses(ctx, map);
}
main().catch((err) => {
console.error(err instanceof Error ? err.message : err);
process.exit(1);
});