chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/core/handlers/xrpc-repo - com.atproto.repo.* record reads and writes.
//
// A write goes through the same three steps every time: check the scope allows
// the collection, put the record, then commit. applyWrites does all of its
// operations against one commit, so a batch either lands whole or not at all.
import {
cborDecode,
cborEncodeDagCbor,
cidToString,
createCid,
createTid,
findBlobRefs,
recordTimeOf,
toDisplayValue,
} from '../repo.js';
import { ScopePermissions } from '../scope.js';
/** What createRecord's locked write returns when the key is taken. */
const EXISTS = Symbol('exists');
/**
* @typedef {Object} RepoContext
* @property {import('../ports.js').ActorStoragePort} actorStorage
* @property {import('../ports.js').LexiconResolverPort} [lexiconResolver]
* @property {boolean} readOnly
* @property {() => Promise<string|null>} getDid
* @property {() => Response} readOnlyError
* @property {<T>(fn: () => Promise<T>) => Promise<T>} withRepoLock - Runs `fn` with no other repo write in progress
* @property {(did: string, ops: Array<{action: string, uri: string, cid?: string}>) => Promise<Object>} createCommit - Expects the caller to hold the repo lock
* @property {() => Promise<Response|null>} checkAccountActive
* @property {(request: Request) => Promise<any|Response>} readJsonBounded
* @property {(request: Request) => Promise<Response|null>} handleProxyHeader
* @property {(did: string, olderThan: number, only?: string[]) => Promise<void>} reapOrphanedBlobs
* @property {import('../ports.js').RecordWriteGuardPort|null} [writeGuard]
*/
/**
* @param {RepoContext} ctx
* @returns {{ routes: import('../pds.js').Routes }}
*/
export function createRepoHandlers(ctx) {
const {
actorStorage,
lexiconResolver,
readOnly,
getDid,
readOnlyError,
withRepoLock,
createCommit,
checkAccountActive,
readJsonBounded,
handleProxyHeader,
reapOrphanedBlobs,
writeGuard,
} = ctx;
/**
* The answer to a create at a key the repo already holds.
* @param {string} uri
* @returns {Response}
*/
function recordExists(uri) {
return Response.json(
{ error: 'InvalidRequest', message: `Record already exists: ${uri}` },
{ status: 400 },
);
}
/**
* Consult the write guard for one write in a collection it covers. Null to
* proceed, the refusal Response otherwise.
* @param {string} collection
* @param {string} rkey
* @param {{value: Uint8Array|ArrayBuffer}|null|undefined} existing
* @param {unknown|null} nextValue - null for a deletion
* @returns {Promise<Response|null>}
*/
async function guardWrite(collection, rkey, existing, nextValue) {
if (!writeGuard || !writeGuard.collections.includes(collection)) {
return null;
}
try {
await writeGuard.check({
space: null,
collection,
rkey,
prevValue: existing
? cborDecode(
existing.value instanceof Uint8Array
? existing.value
: new Uint8Array(existing.value),
)
: null,
nextValue,
});
return null;
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
const code =
err instanceof Error && 'code' in err
? String(/** @type {{code: unknown}} */ (err).code)
: 'InvalidRequest';
return Response.json({ error: code, message }, { status: 400 });
}
}
/**
* com.atproto.repo.describeRepo
* @param {Request} _request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleDescribeRepo(_request, url) {
const did = await getDid();
const repo = url.searchParams.get('repo');
// If repo specified and doesn't match, return 404
if (repo && repo !== did) {
return Response.json(
{ error: 'RepoNotFound', message: 'Repository not found on this PDS' },
{ status: 404 },
);
}
const handle = await actorStorage.getHandle();
// Get unique collections from records
const allRecords = await actorStorage.listAllRecords();
const collections = [
...new Set(allRecords.map((r) => r.key.split('/')[0])),
];
return Response.json({
handle: handle || did,
did,
didDoc: {},
collections,
handleIsCorrect: !!handle,
});
}
/**
* com.atproto.repo.createRecord
* @param {Request} request
* @param {URL} _url
* @param {import('../pds.js').Auth|null} auth
* @returns {Promise<Response>}
*/
async function handleCreateRecord(request, _url, auth) {
if (readOnly) return readOnlyError();
if (!auth) {
return Response.json(
{ error: 'AuthenticationRequired', message: 'Authentication required' },
{ status: 401 },
);
}
const body = await readJsonBounded(request);
if (body instanceof Response) return body;
const inactive = await checkAccountActive();
if (inactive) return inactive;
const {
repo,
collection,
rkey: providedRkey,
record,
validate: _validate,
} = body;
const did = await getDid();
if (!did) {
return Response.json(
{ error: 'InvalidRequest', message: 'PDS not initialized' },
{ status: 400 },
);
}
if (repo !== did && repo !== auth.did) {
return Response.json(
{ error: 'Forbidden', message: 'Not authorized for this repository' },
{ status: 403 },
);
}
// Check scope permissions for create action
const permissions = new ScopePermissions(auth.scope);
if (!permissions.allowsRepo(collection, 'create')) {
return Response.json(
{
error: 'Forbidden',
message: `Missing required scope for create on ${collection}`,
},
{ status: 403 },
);
}
const rkey = providedRkey || createTid();
const uri = `at://${did}/${collection}/${rkey}`;
// Ensure $type is set
const recordWithType = { $type: collection, ...record };
// Validate record against schema if resolver provided
const shouldValidate = _validate !== false;
let validationStatus;
if (shouldValidate && lexiconResolver) {
try {
validationStatus = await lexiconResolver.assertValid(
collection,
recordWithType,
{ strict: _validate === true },
);
} catch (err) {
return Response.json(
{
error: 'InvalidRecord',
message: `Invalid ${collection} record: ${err.message}`,
},
{ status: 400 },
);
}
} else if (shouldValidate && !lexiconResolver) {
validationStatus = 'unknown';
}
// If _validate === false, validationStatus remains undefined
// Encode record and create CID
const encoded = cborEncodeDagCbor(recordWithType);
const cid = cidToString(await createCid(encoded));
const refused = await guardWrite(collection, rkey, null, recordWithType);
if (refused) return refused;
const latestCommit = await withRepoLock(async () => {
// Clients claim a key by creating at it, so a create never replaces.
if (await actorStorage.getRecord(uri)) return EXISTS;
await actorStorage.putBlock(cid, encoded);
await actorStorage.putRecord(uri, cid, collection, rkey, encoded);
const blobRefs = findBlobRefs(record);
for (const blobCid of blobRefs) {
await actorStorage.linkBlobToRecord(blobCid, uri, recordTimeOf(record));
}
await createCommit(did, [{ action: 'create', uri, cid }]);
return actorStorage.getLatestCommit();
});
if (latestCommit === EXISTS) return recordExists(uri);
return Response.json({
uri,
cid,
commit: latestCommit
? { cid: latestCommit.cid, rev: latestCommit.rev }
: null,
validationStatus,
});
}
/**
* com.atproto.repo.getRecord
* @param {Request} request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleGetRecord(request, url) {
// Check for proxy header first
const proxyResponse = await handleProxyHeader(request);
if (proxyResponse) return proxyResponse;
const repo = url.searchParams.get('repo');
const collection = url.searchParams.get('collection');
const rkey = url.searchParams.get('rkey');
if (!repo || !collection || !rkey) {
return Response.json(
{ error: 'InvalidRequest', message: 'Missing required parameters' },
{ status: 400 },
);
}
const did = await getDid();
// Check if repo matches this PDS
if (repo !== did) {
return Response.json(
{ error: 'RepoNotFound', message: 'Repository not found on this PDS' },
{ status: 404 },
);
}
const uri = `at://${did}/${collection}/${rkey}`;
const result = await actorStorage.getRecord(uri);
if (!result) {
return Response.json(
{ error: 'RecordNotFound', message: 'Record not found' },
{ status: 404 },
);
}
// toDisplayValue turns tag-42 CID bytes (as an imported record carries
// them) back into `{$link}` — without it a blob ref serializes as an object
// full of numbers.
const value = toDisplayValue(cborDecode(result.value));
return Response.json({
uri,
cid: result.cid,
value,
});
}
/**
* com.atproto.repo.listRecords
* @param {Request} _request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleListRecords(_request, url) {
const repo = url.searchParams.get('repo');
const collection = url.searchParams.get('collection');
const limit = Math.min(
parseInt(url.searchParams.get('limit') || '50', 10),
100,
);
const cursor = url.searchParams.get('cursor');
// The reference PDS lists newest rkey first and `reverse=true` flips to
// oldest first. The storage flag means the opposite: it is the
// newest-first switch. Hence the negation.
const reverse = url.searchParams.get('reverse') === 'true';
if (!repo || !collection) {
return Response.json(
{ error: 'InvalidRequest', message: 'Missing required parameters' },
{ status: 400 },
);
}
const _did = await getDid();
const result = await actorStorage.listRecords(
collection,
cursor,
limit,
!reverse,
);
const records = result.records.map((r) => ({
uri: r.uri,
cid: r.cid,
// toDisplayValue: imported records carry CID links as raw bytes; render
// them back as `{$link}` rather than an object full of numbers.
value: toDisplayValue(
cborDecode(
r.value instanceof ArrayBuffer ? new Uint8Array(r.value) : r.value,
),
),
}));
return Response.json({
records,
// Match the reference PDS: cursor is an optional string, so omit it on a
// final page rather than emitting `cursor: null`, which strict lexicon
// validators (@atproto/lexicon) reject.
...(result.cursor ? { cursor: result.cursor } : {}),
});
}
/**
* com.atproto.repo.putRecord
* @param {Request} request
* @param {URL} _url
* @param {import('../pds.js').Auth|null} auth
* @returns {Promise<Response>}
*/
async function handlePutRecord(request, _url, auth) {
if (readOnly) return readOnlyError();
if (!auth) {
return Response.json(
{ error: 'AuthenticationRequired', message: 'Authentication required' },
{ status: 401 },
);
}
const body = await readJsonBounded(request);
if (body instanceof Response) return body;
const inactive = await checkAccountActive();
if (inactive) return inactive;
const {
repo,
collection,
rkey,
record,
validate: _validate,
swapRecord,
swapCommit: _swapCommit,
} = body;
const did = await getDid();
if (!did) {
return Response.json(
{ error: 'InvalidRequest', message: 'PDS not initialized' },
{ status: 400 },
);
}
if (repo !== did && repo !== auth.did) {
return Response.json(
{ error: 'Forbidden', message: 'Not authorized for this repository' },
{ status: 403 },
);
}
// An upsert can add a record or overwrite one, so it needs both actions.
const permissions = new ScopePermissions(auth.scope);
for (const action of ['create', 'update']) {
if (!permissions.allowsRepo(collection, action)) {
return Response.json(
{
error: 'Forbidden',
message: `Missing required scope for ${action} on ${collection}`,
},
{ status: 403 },
);
}
}
if (!rkey) {
return Response.json(
{ error: 'InvalidRequest', message: 'rkey required for putRecord' },
{ status: 400 },
);
}
const uri = `at://${did}/${collection}/${rkey}`;
// Ensure $type is set
const recordWithType = { $type: collection, ...record };
// Validate record against schema if resolver provided
const shouldValidate = _validate !== false;
let validationStatus;
if (shouldValidate && lexiconResolver) {
try {
validationStatus = await lexiconResolver.assertValid(
collection,
recordWithType,
{ strict: _validate === true },
);
} catch (err) {
return Response.json(
{
error: 'InvalidRecord',
message: `Invalid ${collection} record: ${err.message}`,
},
{ status: 400 },
);
}
} else if (shouldValidate && !lexiconResolver) {
validationStatus = 'unknown';
}
// If _validate === false, validationStatus remains undefined
// Encode record and create CID
const encoded = cborEncodeDagCbor(recordWithType);
const cid = cidToString(await createCid(encoded));
const written = await withRepoLock(async () => {
const existing = await actorStorage.getRecord(uri);
if (swapRecord !== undefined) {
const existingCid = existing ? existing.cid : null;
if (swapRecord !== existingCid) {
return Response.json(
{
error: 'InvalidSwap',
message: 'Record CID does not match swapRecord',
},
{ status: 400 },
);
}
}
const refused = await guardWrite(
collection,
rkey,
existing,
recordWithType,
);
if (refused) return refused;
if (existing) {
await actorStorage.unlinkBlobsFromRecord(uri);
}
await actorStorage.putBlock(cid, encoded);
await actorStorage.putRecord(uri, cid, collection, rkey, encoded);
const blobRefs = findBlobRefs(record);
for (const blobCid of blobRefs) {
await actorStorage.linkBlobToRecord(blobCid, uri, recordTimeOf(record));
}
await createCommit(did, [
{ action: existing ? 'update' : 'create', uri, cid },
]);
return actorStorage.getLatestCommit();
});
if (written instanceof Response) return written;
const latestCommit = written;
return Response.json({
uri,
cid,
commit: latestCommit
? { cid: latestCommit.cid, rev: latestCommit.rev }
: null,
validationStatus,
});
}
/**
* com.atproto.repo.applyWrites
* @param {Request} request
* @param {URL} _url
* @param {import('../pds.js').Auth|null} auth
* @returns {Promise<Response>}
*/
async function handleApplyWrites(request, _url, auth) {
if (readOnly) return readOnlyError();
if (!auth) {
return Response.json(
{ error: 'AuthenticationRequired', message: 'Authentication required' },
{ status: 401 },
);
}
const body = await readJsonBounded(request);
if (body instanceof Response) return body;
const inactive = await checkAccountActive();
if (inactive) return inactive;
const { repo, writes, validate: _validate, swapCommit: _swapCommit } = body;
const did = await getDid();
if (!did) {
return Response.json(
{ error: 'InvalidRequest', message: 'PDS not initialized' },
{ status: 400 },
);
}
if (repo !== did && repo !== auth.did) {
return Response.json(
{ error: 'Forbidden', message: 'Not authorized for this repository' },
{ status: 403 },
);
}
if (!writes || !Array.isArray(writes)) {
return Response.json(
{ error: 'InvalidRequest', message: 'writes must be an array' },
{ status: 400 },
);
}
/** @type {Array<{action: string, uri: string, cid?: string}>} */
const ops = [];
/** @type {Array<Record<string, unknown>>} */
const results = [];
const permissions = new ScopePermissions(auth.scope);
// Each operation carries its own action, and the scope has to allow that
// action on that collection. Checked before anything is written, so a batch
// the scope only partly covers changes nothing.
for (const write of writes) {
const action = String(write?.$type ?? '').split('#')[1];
if (
(action === 'create' || action === 'update' || action === 'delete') &&
!permissions.allowsRepo(write.collection, action)
) {
return Response.json(
{
error: 'Forbidden',
message: `Missing required scope for ${action} on ${write.collection}`,
},
{ status: 403 },
);
}
}
// The batch reads, checks and writes its records as one unit. The lock
// also covers validation, which can fetch a lexicon.
return withRepoLock(async () => {
// The write guard, before anything lands, for the same reason as the
// scope pre-pass. A create with no rkey mints one below; a minted TID
// never names a config record, so the guard sees the empty rkey.
// The keys this batch creates, so two creates of one key are refused.
const claimed = new Set();
for (const write of writes) {
const action = String(write?.$type ?? '').split('#')[1];
if (action !== 'create' && action !== 'update' && action !== 'delete') {
continue;
}
const rkey = write.rkey || '';
const existing = rkey
? await actorStorage.getRecord(
`at://${did}/${write.collection}/${rkey}`,
)
: null;
const uri = `at://${did}/${write.collection}/${rkey}`;
if (action === 'create' && rkey) {
if (existing || claimed.has(uri)) return recordExists(uri);
claimed.add(uri);
}
const refused = await guardWrite(
write.collection,
rkey,
existing,
action === 'delete'
? null
: { $type: write.collection, ...write.value },
);
if (refused) return refused;
}
for (const write of writes) {
const { $type, collection, rkey: providedRkey, value } = write;
if ($type === 'com.atproto.repo.applyWrites#create') {
const rkey = providedRkey || createTid();
const uri = `at://${did}/${collection}/${rkey}`;
// Ensure $type is set
const recordWithType = { $type: collection, ...value };
// Validate record against schema if resolver provided
const shouldValidate = _validate !== false;
let validationStatus;
if (shouldValidate && lexiconResolver) {
try {
validationStatus = await lexiconResolver.assertValid(
collection,
recordWithType,
{ strict: _validate === true },
);
} catch (err) {
return Response.json(
{
error: 'InvalidRecord',
message: `Invalid ${collection} record: ${err.message}`,
},
{ status: 400 },
);
}
} else if (shouldValidate && !lexiconResolver) {
validationStatus = 'unknown';
}
// Encode record and create CID
const encoded = cborEncodeDagCbor(recordWithType);
const cid = cidToString(await createCid(encoded));
// Store block
await actorStorage.putBlock(cid, encoded);
// Store record
await actorStorage.putRecord(uri, cid, collection, rkey, encoded);
// Link blobs
const blobRefs = findBlobRefs(value);
for (const blobCid of blobRefs) {
await actorStorage.linkBlobToRecord(
blobCid,
uri,
recordTimeOf(value),
);
}
ops.push({ action: 'create', uri, cid });
results.push({
$type: 'com.atproto.repo.applyWrites#createResult',
uri,
cid,
validationStatus,
});
} else if ($type === 'com.atproto.repo.applyWrites#update') {
if (!providedRkey) {
return Response.json(
{ error: 'InvalidRequest', message: 'rkey required for update' },
{ status: 400 },
);
}
const uri = `at://${did}/${collection}/${providedRkey}`;
// An update of a key the repo does not hold adds it, so the tree
// performs a create. Announcing an update for a key absent from the
// previous tree describes a diff that never happened.
const present = await actorStorage.getRecord(uri);
// Ensure $type is set
const recordWithType = { $type: collection, ...value };
// Validate record against schema if resolver provided
const shouldValidate = _validate !== false;
let validationStatus;
if (shouldValidate && lexiconResolver) {
try {
validationStatus = await lexiconResolver.assertValid(
collection,
recordWithType,
{ strict: _validate === true },
);
} catch (err) {
return Response.json(
{
error: 'InvalidRecord',
message: `Invalid ${collection} record: ${err.message}`,
},
{ status: 400 },
);
}
} else if (shouldValidate && !lexiconResolver) {
validationStatus = 'unknown';
}
// Unlink old blobs
await actorStorage.unlinkBlobsFromRecord(uri);
// Encode record and create CID
const encoded = cborEncodeDagCbor(recordWithType);
const cid = cidToString(await createCid(encoded));
// Store block
await actorStorage.putBlock(cid, encoded);
// Store record
await actorStorage.putRecord(
uri,
cid,
collection,
providedRkey,
encoded,
);
// Link blobs
const blobRefs = findBlobRefs(value);
for (const blobCid of blobRefs) {
await actorStorage.linkBlobToRecord(
blobCid,
uri,
recordTimeOf(value),
);
}
ops.push({ action: present ? 'update' : 'create', uri, cid });
results.push({
$type: 'com.atproto.repo.applyWrites#updateResult',
uri,
cid,
validationStatus,
});
} else if ($type === 'com.atproto.repo.applyWrites#delete') {
if (!providedRkey) {
return Response.json(
{ error: 'InvalidRequest', message: 'rkey required for delete' },
{ status: 400 },
);
}
const uri = `at://${did}/${collection}/${providedRkey}`;
// A delete of a record that is not there contributes no op, for the
// same reason as deleteRecord: the tree does not change, so there is
// nothing to announce. The result below still stands — one per write —
// and a batch whose deletes are all no-ops commits nothing.
if (await actorStorage.getRecord(uri)) {
// Unlink blobs
await actorStorage.unlinkBlobsFromRecord(uri);
// Delete record
await actorStorage.deleteRecord(uri);
ops.push({ action: 'delete', uri });
}
results.push({
$type: 'com.atproto.repo.applyWrites#deleteResult',
});
} else {
return Response.json(
{
error: 'InvalidRequest',
message: `Unknown write operation type: ${$type}`,
},
{ status: 400 },
);
}
}
// Create single commit for all operations
if (ops.length > 0) {
await createCommit(did, ops);
}
// Get commit info
const latestCommit = await actorStorage.getLatestCommit();
return Response.json({
commit: latestCommit
? { cid: latestCommit.cid, rev: latestCommit.rev }
: null,
results,
});
});
}
/**
* com.atproto.repo.deleteRecord
* @param {Request} request
* @param {URL} _url
* @param {import('../pds.js').Auth|null} auth
* @returns {Promise<Response>}
*/
async function handleDeleteRecord(request, _url, auth) {
if (readOnly) return readOnlyError();
if (!auth) {
return Response.json(
{ error: 'AuthenticationRequired', message: 'Authentication required' },
{ status: 401 },
);
}
const { repo, collection, rkey } = await request.json();
const inactive = await checkAccountActive();
if (inactive) return inactive;
const did = await getDid();
if (!did) {
return Response.json(
{ error: 'InvalidRequest', message: 'PDS not initialized' },
{ status: 400 },
);
}
if (repo !== did && repo !== auth.did) {
return Response.json(
{ error: 'Forbidden', message: 'Not authorized for this repository' },
{ status: 403 },
);
}
// Check scope permissions for delete action
const permissions = new ScopePermissions(auth.scope);
if (!permissions.allowsRepo(collection, 'delete')) {
return Response.json(
{
error: 'Forbidden',
message: `Missing required scope for delete on ${collection}`,
},
{ status: 403 },
);
}
const uri = `at://${did}/${collection}/${rkey}`;
const startedAt = Date.now();
const deleted = await withRepoLock(async () => {
const existing = await actorStorage.getRecord(uri);
const refused = await guardWrite(collection, rkey, existing, null);
if (refused) return refused;
// Deleting what is not there is a no-op, and a no-op is not a commit.
// Committing anyway announces an op the tree does not contain: the MST
// is unchanged, so a consumer checking ops against the diff sees a
// delete for a record the repo does not have. Clients retry, and each
// retry would otherwise burn a revision and put another impossible op
// on the firehose.
if (!existing) return null;
const blobs = findBlobRefs(
cborDecode(
existing.value instanceof Uint8Array
? existing.value
: new Uint8Array(existing.value),
),
);
await actorStorage.unlinkBlobsFromRecord(uri);
await actorStorage.deleteRecord(uri);
await createCommit(did, [{ action: 'delete', uri }]);
return blobs;
});
if (deleted instanceof Response) return deleted;
// Only this record's blobs are candidates. Another orphan can be a blob
// that a client uploaded for a record it has not written yet. A blob
// uploaded again after this delete started is newer than the threshold.
if (deleted?.length) {
await reapOrphanedBlobs(did, startedAt + 1, deleted);
}
return Response.json({});
}
return {
routes: {
'/xrpc/com.atproto.repo.describeRepo': {
handler: handleDescribeRepo,
},
'/xrpc/com.atproto.repo.createRecord': {
method: 'POST',
auth: 'required',
handler: handleCreateRecord,
},
'/xrpc/com.atproto.repo.getRecord': {
handler: handleGetRecord,
},
'/xrpc/com.atproto.repo.listRecords': {
handler: handleListRecords,
},
'/xrpc/com.atproto.repo.putRecord': {
method: 'POST',
auth: 'required',
handler: handlePutRecord,
},
'/xrpc/com.atproto.repo.applyWrites': {
method: 'POST',
auth: 'required',
handler: handleApplyWrites,
},
'/xrpc/com.atproto.repo.deleteRecord': {
method: 'POST',
auth: 'required',
handler: handleDeleteRecord,
},
},
};
}