chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/spaces/handlers/write - com.atproto.space write endpoints.
//
// createRecord, putRecord, deleteRecord, applyWrites and clearRecords are thin
// shells over applyWrites() in ../writer.js: they validate the request,
// translate it into write descriptors, and shape the response. All the repo
// logic lives there.
//
// Authorization is two checks: the caller may only write its own repo, and the
// token's `space:` scope must cover the target space, action and collection.
import {
cborDecodeJson,
createTid,
findBlobRefs,
recordTimeOf,
} from '@pdsjs/core/repo';
import { ScopePermissions } from '@pdsjs/core/scope';
import { createWriteNotifier } from '../notify.js';
import { createNotifyOutbox } from '../outbox.js';
import {
assertNotPolicyRecord,
PUBLISHED_COLLECTIONS,
SpacePolicyError,
} from '../policy.js';
import { parseSpaceUri } from '../uri.js';
import {
applyWrites,
MAX_WRITES_PER_COMMIT,
SpaceWriteError,
} from '../writer.js';
const MAX_JSON_BODY = 512 * 1024;
/**
* @param {string} error
* @param {string} message
* @param {number} [status]
* @returns {Response}
*/
function errorResponse(error, message, status = 400) {
return Response.json({ error, message }, { status });
}
/**
* Parse a JSON body, rejecting oversized or malformed input.
* @param {Request} request
* @returns {Promise<{ok: true, body: any}|{ok: false, response: Response}>}
*/
async function readJson(request) {
const text = await request.text();
if (text.length > MAX_JSON_BODY) {
return {
ok: false,
response: errorResponse('InvalidRequest', 'Request body too large', 413),
};
}
try {
return { ok: true, body: JSON.parse(text) };
} catch {
return {
ok: false,
response: errorResponse('InvalidRequest', 'Invalid JSON body'),
};
}
}
/**
* Space URIs arrive typed only as `at-uri`, so a malformed one is a request
* error rather than a 500.
* @param {string} space
* @returns {Response|null}
*/
function validateSpaceUri(space) {
if (typeof space !== 'string') {
return errorResponse('InvalidRequest', 'space is required');
}
try {
parseSpaceUri(space);
return null;
} catch {
return errorResponse('InvalidSpaceUri', `Not a space uri: ${space}`);
}
}
/**
* Assert the token's `space:` scope covers this operation.
*
* A grant that names no `collection` should fall back to the space type's
* declared collections. Resolving a space-type declaration means fetching its
* lexicon, which this implementation does not do yet — so such a grant
* currently authorizes no write targets. That fails closed, and a grant that
* names its collections explicitly works as specified.
*
* @param {{did: string, scope?: string}|null} auth
* @param {string} space
* @param {{action: string, collection?: string}} op
* @returns {Response|null}
*/
function checkSpaceScope(auth, space, op) {
const { spaceDid, spaceType, skey } = parseSpaceUri(space);
const permissions = new ScopePermissions(auth?.scope);
const allowed = permissions.allowsSpace({
spaceType,
authority: spaceDid,
skey,
action: op.action,
collection: op.collection,
userDid: auth?.did,
});
if (!allowed) {
return errorResponse(
'InvalidToken',
`Scope does not permit ${op.action} in ${space}`,
403,
);
}
return null;
}
/**
* A writer may only write its own repo.
* @param {string} repo
* @param {string} did
* @returns {Response|null}
*/
function validateRepo(repo, did) {
if (repo !== did) {
return errorResponse(
'Forbidden',
'repo must match authenticated user',
403,
);
}
return null;
}
/**
* @param {unknown} err
* @returns {Response}
*/
function writeErrorResponse(err) {
if (err instanceof SpaceWriteError || err instanceof SpacePolicyError) {
return errorResponse(err.code, err.message);
}
throw err;
}
/**
* @param {string} space
* @param {string} did
* @param {string} collection
* @param {string} rkey
* @returns {string}
*/
function recordUri(space, did, collection, rkey) {
return `${space}/${did}/${collection}/${rkey}`;
}
const WRITE_ACTIONS = ['create', 'update', 'delete'];
/**
* Which of the three operations a write in an applyWrites batch is.
*
* The proposal made these a closed union discriminated by `$type` after clients
* had shipped against an `action` string. Both say the same thing, so both are
* read; anything else is a write this server cannot place.
*
* @param {any} write
* @returns {'create'|'update'|'delete'|null}
*/
function writeAction(write) {
const fromType = String(write?.$type ?? '').match(
/^com\.atproto\.space\.applyWrites#(create|update|delete)$/,
);
if (fromType) return /** @type {any} */ (fromType[1]);
if (WRITE_ACTIONS.includes(write?.action)) return write.action;
return null;
}
/**
* @param {Object} ctx
* @param {import('@pdsjs/core/ports').SpaceStoragePort} ctx.spaceStorage
* @param {(blobCid: string, recordUri: string, recordTime: number|null) => Promise<void>} ctx.linkBlob
* @param {(recordUri: string) => Promise<void>} ctx.unlinkBlobs
* @param {() => Promise<string|null>} ctx.getDid
* @param {() => Promise<import('../token.js').SpaceSigner>} ctx.getSigner
* @param {import('@pdsjs/core/ports').DidResolverPort} ctx.resolveDid
* @param {{collections: string[], check: (write: {space: string|null, collection: string, rkey: string, prevValue: unknown|null, nextValue: unknown|null}) => Promise<void>}} [ctx.recordGuard]
* @param {typeof fetch} [ctx.fetch]
* @param {(work: Promise<unknown>) => void} [ctx.background] - runs the
* syncers' notices after the response; absent, they run before it
* @param {ReturnType<typeof createNotifyOutbox>} [ctx.outbox] - built here
* from the rest of ctx when absent
* @returns {import('@pdsjs/core/pds').Routes}
*/
export function createWriteRoutes(ctx) {
const { spaceStorage, linkBlob, unlinkBlobs, getDid, recordGuard } = ctx;
const outbox = ctx.outbox ?? createNotifyOutbox(ctx);
/**
* Consult the write guard for one write in a collection it covers. Null to
* proceed, the refusal Response otherwise. Read-then-check races the write
* queue; the single account that writes here makes that a non-event.
* @param {string} space
* @param {import('../writer.js').SpaceWriteInput} write
* @returns {Promise<Response|null>}
*/
async function guardWrite(space, write) {
if (!recordGuard || !recordGuard.collections.includes(write.collection)) {
return null;
}
const prev = await spaceStorage.getSpaceRecord(
space,
write.collection,
write.rkey,
);
try {
await recordGuard.check({
space,
collection: write.collection,
rkey: write.rkey,
prevValue: prev ? cborDecodeJson(prev.value) : null,
nextValue: write.action === 'delete' ? null : (write.record ?? null),
});
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 errorResponse(code, message);
}
}
/**
* Record which blobs a space write leaves referenced.
*
* The links go in the account's one blob link table, beside those of public
* records. Orphan reaping asks that table whether anything still points at a
* blob, so a space record protects its blobs the same way a public record
* does, with no second table for the reaper to be taught about.
*
* @param {string} space
* @param {string} did
* @param {import('../writer.js').SpaceWriteInput[]} writes
* @returns {Promise<void>}
*/
async function indexBlobs(space, did, writes) {
for (const write of writes) {
const uri = recordUri(space, did, write.collection, write.rkey);
// An update replaces the record's references rather than adding to them.
await unlinkBlobs(uri);
if (write.action === 'delete') continue;
for (const cid of findBlobRefs(write.record)) {
await linkBlob(cid, uri, recordTimeOf(write.record));
}
}
}
const fireNotifyWrite = createWriteNotifier(ctx, outbox);
/**
* Shared shell for the three single-write endpoints.
* @param {Request} request
* @param {{did: string, scope?: string}|null} auth
* @param {(body: any) => {action: 'create'|'put'|'delete', collection: string, rkey: string, record?: any}} toWrite
* @param {(body: any, result: import('../writer.js').SpaceWriteResult, did: string) => any} toBody
* @returns {Promise<Response>}
*/
async function singleWrite(request, auth, toWrite, toBody) {
const parsed = await readJson(request);
if (!parsed.ok) return parsed.response;
const body = parsed.body;
const did = /** @type {{did: string}} */ (auth).did;
const uriError = validateSpaceUri(body.space);
if (uriError) return uriError;
const repoError = validateRepo(body.repo, did);
if (repoError) return repoError;
if (typeof body.collection !== 'string') {
return errorResponse('InvalidRequest', 'collection is required');
}
let write;
try {
write = toWrite(body);
assertNotPolicyRecord(write.collection);
} catch (err) {
return writeErrorResponse(err);
}
// A put may create or update, so resolve which before checking scope, the
// way the reference implementation does.
/** @type {string} */
let action = write.action;
if (action === 'put') {
const existing = await spaceStorage.getSpaceRecordCid(
body.space,
write.collection,
write.rkey,
);
action = existing ? 'update' : 'create';
}
const scopeError = checkSpaceScope(auth, body.space, {
action,
collection: write.collection,
});
if (scopeError) return scopeError;
const refused = await guardWrite(body.space, write);
if (refused) return refused;
try {
const commit = await applyWrites(spaceStorage, {
space: body.space,
writes: [write],
owner: did,
});
// After the commit: the write is what makes the references real, and a
// link to a record that failed to write would keep a blob alive forever.
await indexBlobs(body.space, did, [
{ ...write, rkey: commit.results[0].rkey },
]);
await fireNotifyWrite(body.space, commit);
return Response.json(toBody(body, commit.results[0], did));
} catch (err) {
return writeErrorResponse(err);
}
}
return {
'/xrpc/com.atproto.space.createRecord': {
method: 'POST',
auth: 'required',
handler: (request, _url, auth) =>
singleWrite(
request,
auth,
(body) => {
if (body.record === undefined) {
throw new SpaceWriteError('record is required', 'InvalidRequest');
}
return {
action: 'create',
collection: body.collection,
rkey: body.rkey ?? createTid(),
record: body.record,
};
},
(body, result, did) => ({
uri: recordUri(body.space, did, result.collection, result.rkey),
cid: result.cid,
}),
),
},
'/xrpc/com.atproto.space.putRecord': {
method: 'POST',
auth: 'required',
handler: (request, _url, auth) =>
singleWrite(
request,
auth,
(body) => {
if (typeof body.rkey !== 'string') {
throw new SpaceWriteError('rkey is required', 'InvalidRequest');
}
if (body.record === undefined) {
throw new SpaceWriteError('record is required', 'InvalidRequest');
}
if (
body.swapRecord !== undefined &&
body.swapRecord !== null &&
typeof body.swapRecord !== 'string'
) {
throw new SpaceWriteError(
'swapRecord must be a CID string or null',
'InvalidRequest',
);
}
return {
action: 'put',
collection: body.collection,
rkey: body.rkey,
record: body.record,
// A pds.js extension the draft proposal does not have: null
// requires that the record does not exist, a CID requires that
// exact version, absent skips the check. A server without it
// ignores the field, so a client falls back to last-write-wins.
...(body.swapRecord !== undefined
? { swapCid: body.swapRecord }
: {}),
};
},
(body, result, did) => ({
uri: recordUri(body.space, did, result.collection, result.rkey),
cid: result.cid,
}),
),
},
'/xrpc/com.atproto.space.deleteRecord': {
method: 'POST',
auth: 'required',
handler: (request, _url, auth) =>
singleWrite(
request,
auth,
(body) => {
if (typeof body.rkey !== 'string') {
throw new SpaceWriteError('rkey is required', 'InvalidRequest');
}
return {
action: 'delete',
collection: body.collection,
rkey: body.rkey,
};
},
() => ({}),
),
},
'/xrpc/com.atproto.space.applyWrites': {
method: 'POST',
auth: 'required',
handler: async (request, _url, auth) => {
const parsed = await readJson(request);
if (!parsed.ok) return parsed.response;
const body = parsed.body;
const did = /** @type {{did: string}} */ (auth).did;
const uriError = validateSpaceUri(body.space);
if (uriError) return uriError;
const repoError = validateRepo(body.repo, did);
if (repoError) return repoError;
if (!Array.isArray(body.writes)) {
return errorResponse('InvalidRequest', 'writes must be an array');
}
/** @type {import('../writer.js').SpaceWriteInput[]} */
const writes = [];
for (const w of body.writes) {
const collection = w?.collection;
if (typeof collection !== 'string') {
return errorResponse(
'InvalidRequest',
'each write needs a collection',
);
}
try {
assertNotPolicyRecord(collection);
} catch (err) {
return writeErrorResponse(err);
}
const action = writeAction(w);
if (!action) {
return errorResponse(
'InvalidRequest',
`Unsupported write type: ${w?.$type ?? w?.action ?? '(none)'}`,
);
}
if (action === 'delete') {
writes.push({ action, collection, rkey: w.rkey });
} else {
writes.push({
action,
collection,
// A create may leave the server to mint the key. An empty string
// is how the earlier clients say so.
rkey: action === 'create' ? w.rkey || createTid() : w.rkey,
record: w.value,
});
}
}
for (const w of writes) {
const scopeError = checkSpaceScope(auth, body.space, {
action: w.action,
collection: w.collection,
});
if (scopeError) return scopeError;
}
for (const w of writes) {
const refused = await guardWrite(body.space, w);
if (refused) return refused;
}
try {
const commit = await applyWrites(spaceStorage, {
space: body.space,
writes,
owner: did,
});
await indexBlobs(
body.space,
did,
writes.map((w, i) => ({ ...w, rkey: commit.results[i].rkey })),
);
await fireNotifyWrite(body.space, commit);
return Response.json({
results: commit.results.map((r) =>
r.action === 'delete'
? { $type: 'com.atproto.space.applyWrites#deleteResult' }
: {
$type: `com.atproto.space.applyWrites#${r.action}Result`,
uri: recordUri(body.space, did, r.collection, r.rkey),
cid: r.cid,
},
),
});
} catch (err) {
return writeErrorResponse(err);
}
},
},
'/xrpc/com.atproto.space.clearRecords': {
method: 'POST',
auth: 'required',
handler: async (request, _url, auth) => {
const parsed = await readJson(request);
if (!parsed.ok) return parsed.response;
const body = parsed.body;
const did = /** @type {{did: string}} */ (auth).did;
const uriError = validateSpaceUri(body.space);
if (uriError) return uriError;
const repoError = validateRepo(body.repo, did);
if (repoError) return repoError;
// In the authority's own repo, the published policy and member
// records are the PDS's copy of the space's configuration, so they
// stay, and keeping them takes no grant.
const keep =
parseSpaceUri(body.space).spaceDid === did
? PUBLISHED_COLLECTIONS
: [];
const records = (
await spaceStorage.listAllSpaceRecords(body.space)
).filter((r) => !keep.includes(r.collection));
for (const collection of new Set(records.map((r) => r.collection))) {
const scopeError = checkSpaceScope(auth, body.space, {
action: 'delete',
collection,
});
if (scopeError) return scopeError;
}
/** @type {import('../writer.js').SpaceWriteInput[]} */
const writes = records.map((r) => ({
action: 'delete',
collection: r.collection,
rkey: r.rkey,
}));
for (const w of writes) {
const refused = await guardWrite(body.space, w);
if (refused) return refused;
}
try {
/** @type {{rev: string, setHash: Uint8Array}|null} */
let commit = null;
for (let i = 0; i < writes.length; i += MAX_WRITES_PER_COMMIT) {
const chunk = writes.slice(i, i + MAX_WRITES_PER_COMMIT);
commit = await applyWrites(spaceStorage, {
space: body.space,
writes: chunk,
owner: did,
});
await indexBlobs(body.space, did, chunk);
}
if (commit) await fireNotifyWrite(body.space, commit);
return Response.json({});
} catch (err) {
return writeErrorResponse(err);
}
},
},
'/xrpc/com.atproto.space.listSpaces': {
auth: 'required',
handler: async (_request, url) => {
const params = url.searchParams;
const limitParam = params.get('limit');
let limit = limitParam === null ? 50 : Number(limitParam);
if (!Number.isInteger(limit) || limit < 1 || limit > 100) {
if (limitParam !== null) {
return errorResponse(
'InvalidRequest',
'limit must be an integer between 1 and 100',
);
}
limit = 50;
}
const { spaces, cursor } = await spaceStorage.listSpaces({
type: params.get('spaceType') ?? params.get('type'),
did: params.get('did'),
cursor: params.get('cursor'),
limit,
excludeDeleted: true,
});
return Response.json({
...(cursor ? { cursor } : {}),
spaces: spaces.map((s) => ({ uri: s.uri, isOwner: s.isOwner })),
});
},
},
};
}