chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/core/handlers/xrpc-sync - com.atproto.sync.* repo reads.
//
// This server has one account. If a `did` parameter gives a different account,
// the endpoint sends a RepoNotFound error. It does not do a search.
//
// getBlob and listBlobs are in the same namespace, but they read blob storage
// and not the repo. They are with the other blob endpoints.
import { mstGet } from '../mst.js';
import {
buildCarFile,
carBlock,
carHeader,
cborDecode,
cidToString,
} from '../repo.js';
/**
* @typedef {Object} SyncContext
* @property {import('../ports.js').ActorStoragePort} actorStorage
* @property {() => Promise<string|null>} getDid
* @property {() => Promise<boolean>} isDeactivated
* @property {<T>(fn: () => Promise<T>) => Promise<T>} [withRepoLock] - Runs
* `fn` after the repo's writes queued before it; the sweep before getRepo
* holds it. Absent, `fn` runs at once.
*/
/** Blocks read between two waits on a timer, or in one page of the table. */
const BLOCKS_PER_WAIT = 500;
/** @returns {Response} */
function repoNotFound() {
return Response.json(
{ error: 'RepoNotFound', message: 'Repository not found' },
{ status: 404 },
);
}
/**
* @param {SyncContext} ctx
* @returns {{
* buildFullRepoCar: (commitCid: string) => Promise<Uint8Array>,
* routes: import('../pds.js').Routes,
* }}
*/
export function createSyncHandlers(ctx) {
const { actorStorage, getDid, isDeactivated } = ctx;
/**
* com.atproto.sync.listRepos
* @returns {Promise<Response>}
*/
async function handleListRepos() {
const did = await getDid();
const commit = await actorStorage.getLatestCommit();
if (!did) {
return Response.json({ repos: [] });
}
const deactivated = await isDeactivated();
return Response.json({
repos: [
{
did,
head: commit?.cid || null,
rev: commit?.rev || null,
active: !deactivated,
...(deactivated ? { status: 'deactivated' } : {}),
},
],
});
}
/**
* com.atproto.sync.getRepoStatus
* @param {Request} _request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleGetRepoStatus(_request, url) {
const didParam = url.searchParams.get('did');
const did = await getDid();
if (!did || (didParam && didParam !== did)) {
return repoNotFound();
}
const commit = await actorStorage.getLatestCommit();
if (!commit) {
return repoNotFound();
}
const deactivated = await isDeactivated();
return Response.json({
did,
active: !deactivated,
status: deactivated ? 'deactivated' : 'active',
rev: commit.rev,
});
}
/**
* com.atproto.sync.getRepo
* @param {Request} _request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleGetRepo(_request, url) {
const didParam = url.searchParams.get('did');
const did = await getDid();
if (!did || (didParam && didParam !== did)) {
return repoNotFound();
}
const commit = await actorStorage.getLatestCommit();
if (!commit) {
return repoNotFound();
}
if (await blocksCollected()) {
// Swept, the head may have moved on while this waited for the lock.
const head = (await actorStorage.getLatestCommit()) ?? commit;
return new Response(streamRepoCar(head.cid), {
headers: { 'Content-Type': 'application/vnd.ipld.car' },
});
}
let car;
try {
car = await buildFullRepoCar(commit.cid);
} catch (err) {
return Response.json(
{
error: 'InternalServerError',
message:
err instanceof Error ? err.message : 'Failed to collect blocks',
},
{ status: 500 },
);
}
return new Response(/** @type {BodyInit} */ (car), {
headers: { 'Content-Type': 'application/vnd.ipld.car' },
});
}
/**
* The whole repo as one CAR: the commit, every MST node, and every record.
* Serves com.atproto.sync.getRepo from a store that keeps every block it
* was given, and is the repo half of a backup snapshot.
* @param {string} commitCid
* @returns {Promise<Uint8Array>}
*/
async function buildFullRepoCar(commitCid) {
return buildCarFile(commitCid, await repoBlocks(commitCid));
}
/**
* Every block the commit reaches, by walking its tree.
*
* Throws if any of those blocks is absent. A caller must not answer with a
* part of a repo, because a relay accepts it as the whole one.
*
* The walk waits on a timer after every BLOCKS_PER_WAIT blocks. A Durable
* Object delivers no other request until its code waits on something other
* than a promise already settled, and a whole repo is thousands of blocks.
* @param {string} commitCid
* @returns {Promise<Array<{cid: string, data: Uint8Array}>>}
*/
async function repoBlocks(commitCid) {
/** @type {Array<{cid: string, data: Uint8Array}>} */
const blocks = [];
const taken = new Set();
/**
* @param {string} cidStr
* @returns {Promise<Uint8Array>}
*/
const take = async (cidStr) => {
const data = await actorStorage.getBlock(cidStr);
if (!data) {
throw new Error(`Missing block: ${cidStr}`);
}
if (!taken.has(cidStr)) {
taken.add(cidStr);
blocks.push({ cid: cidStr, data });
if (blocks.length % BLOCKS_PER_WAIT === 0) {
await new Promise((resolve) => setTimeout(resolve, 0));
}
}
return data;
};
// An MST node keeps the left subtree in `l` and one entry per key in `e`.
// Each entry has a record CID in `v` and a right subtree in `t`.
//
// A record is a leaf. This takes the block and does not read its fields, so
// a record with a field named `data`, `l` or `e` cannot direct the walk at
// something that is not a block.
//
// The walk needs no guard against a repeat visit. Every CID is the hash of
// its own block, so the tree cannot hold a cycle, and one node has one
// parent.
/** @param {string} cidStr */
const collectNode = async (cidStr) => {
const node = cborDecode(await take(cidStr));
if (node?.l instanceof Uint8Array) {
await collectNode(cidToString(node.l));
}
for (const entry of Array.isArray(node?.e) ? node.e : []) {
if (entry.v instanceof Uint8Array) {
await take(cidToString(entry.v));
}
if (entry.t instanceof Uint8Array) {
await collectNode(cidToString(entry.t));
}
}
};
const commit = cborDecode(await take(commitCid));
if (commit?.data instanceof Uint8Array) {
await collectNode(cidToString(commit.data));
}
return blocks;
}
/**
* Whether the block table holds only what the current commit reaches, so
* that it can be read as the repo; sweeping it first, once, if not.
*
* Each commit drops what it leaves unreachable, but blocks stored before
* commits did, or left by a rebuilt tree or an imported CAR, stay until
* this sweeps them. It holds the repo lock, so no commit stores a block it
* has not counted, and waits on a timer between pages, as the walk does.
* @returns {Promise<boolean>}
*/
async function blocksCollected() {
const store = actorStorage;
if (
!store.listBlocks ||
!store.deleteBlocks ||
!store.getBlocksCollected ||
!store.setBlocksCollected
) {
return false;
}
if (await store.getBlocksCollected()) return true;
const listBlocks = store.listBlocks.bind(store);
const deleteBlocks = store.deleteBlocks.bind(store);
const setBlocksCollected = store.setBlocksCollected.bind(store);
const sweep = async () => {
// Another request may have swept while this one waited for the lock.
if (await store.getBlocksCollected?.()) return true;
const commit = await actorStorage.getLatestCommit();
if (!commit) return false;
const reached = new Set(
(await repoBlocks(commit.cid)).map((block) => block.cid),
);
for (const record of await actorStorage.listAllRecords()) {
reached.add(record.cid);
}
/** @type {string|null} */
let cursor = null;
do {
const page = await listBlocks(cursor, BLOCKS_PER_WAIT);
const unreached = page.blocks
.map((block) => block.cid)
.filter((cid) => !reached.has(cid));
await deleteBlocks(unreached);
cursor = page.cursor;
await new Promise((resolve) => setTimeout(resolve, 0));
} while (cursor);
await setBlocksCollected(true);
return true;
};
return ctx.withRepoLock ? ctx.withRepoLock(sweep) : sweep();
}
/**
* The repo as a CAR read from the block table a page at a time, as the
* reference PDS reads its own, and sent as it is read: the next page is
* read only once the last has gone out, so other requests run between.
*
* A commit that lands meanwhile drops blocks the stream may not have sent
* yet. The reference's stream does the same; a consumer finds the CAR does
* not hold its root's tree and asks again.
* @param {string} commitCid
* @returns {ReadableStream<Uint8Array>}
*/
function streamRepoCar(commitCid) {
const listBlocks =
/** @type {NonNullable<typeof actorStorage.listBlocks>} */ (
actorStorage.listBlocks
).bind(actorStorage);
/** @type {string|null} */
let cursor = null;
let started = false;
return new ReadableStream({
async pull(controller) {
if (!started) {
started = true;
controller.enqueue(carHeader(commitCid));
return;
}
const page = await listBlocks(cursor, BLOCKS_PER_WAIT);
for (const block of page.blocks) controller.enqueue(carBlock(block));
cursor = page.cursor;
if (!cursor) controller.close();
},
});
}
/**
* com.atproto.sync.getRecord (returns CAR with proof)
* @param {Request} _request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleSyncGetRecord(_request, url) {
const didParam = url.searchParams.get('did');
const collection = url.searchParams.get('collection');
const rkey = url.searchParams.get('rkey');
if (!collection || !rkey) {
return Response.json(
{ error: 'InvalidRequest', message: 'Missing required parameters' },
{ status: 400 },
);
}
const did = await getDid();
if (!did || (didParam && didParam !== did)) {
return repoNotFound();
}
const uri = `at://${did}/${collection}/${rkey}`;
const record = await actorStorage.getRecord(uri);
if (!record) {
return Response.json(
{ error: 'RecordNotFound', message: 'Record not found' },
{ status: 404 },
);
}
const commit = await actorStorage.getLatestCommit();
if (!commit) {
return repoNotFound();
}
// Build CAR with commit, MST proof path, and record
/** @type {Array<{cid: string, data: Uint8Array}>} */
const blocks = [];
const included = new Set();
/** @param {string} cidStr */
const addBlock = async (cidStr) => {
if (included.has(cidStr)) return;
included.add(cidStr);
const data = await actorStorage.getBlock(cidStr);
if (data) {
blocks.push({ cid: cidStr, data });
}
};
// Add commit block
await addBlock(commit.cid);
// Get commit data to find MST root
const commitData = await actorStorage.getBlock(commit.cid);
if (commitData) {
const decoded = cborDecode(commitData);
if (decoded.data) {
const mstRootCid = cidToString(decoded.data);
// The proof is the path from the root to the key: the nodes a
// lookup reads.
/** @type {string[]} */
const path = [];
/** @param {string} cid */
const getBlock = (cid) => {
path.push(cid);
return actorStorage.getBlock(cid);
};
// A lookup only reads: it never calls putBlock.
const reader = /** @type {import('../mst.js').MstStore} */ ({
getBlock,
});
await mstGet(mstRootCid, `${collection}/${rkey}`, reader);
for (const cid of path) {
await addBlock(cid);
}
}
}
// Add record block
await addBlock(record.cid);
const car = buildCarFile(commit.cid, blocks);
return new Response(/** @type {BodyInit} */ (car), {
headers: { 'Content-Type': 'application/vnd.ipld.car' },
});
}
/**
* com.atproto.sync.getLatestCommit
* @returns {Promise<Response>}
*/
async function handleGetLatestCommit() {
// The response does not use this value. The call fills the DID cache.
const _did = await getDid();
const commit = await actorStorage.getLatestCommit();
if (!commit) {
return repoNotFound();
}
return Response.json({
cid: commit.cid,
rev: commit.rev,
});
}
return {
buildFullRepoCar,
routes: {
'/xrpc/com.atproto.sync.listRepos': {
handler: handleListRepos,
},
'/xrpc/com.atproto.sync.getRepoStatus': {
handler: handleGetRepoStatus,
},
'/xrpc/com.atproto.sync.getRepo': {
handler: handleGetRepo,
},
'/xrpc/com.atproto.sync.getRecord': {
handler: handleSyncGetRecord,
},
'/xrpc/com.atproto.sync.getLatestCommit': {
handler: handleGetLatestCommit,
},
},
};
}