chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/core/handlers/migration - The account-migration flow and the active
// state it pivots on: checkAccountStatus, importRepo and listMissingBlobs for
// an account arriving here, activateAccount and deactivateAccount for the
// switch-over on either end.
import { loadRepositoryFromCar } from '../loader.js';
import { ScopePermissions } from '../scope.js';
/**
* These are structural types. Thus a member does not name the module that owns
* the function you give it.
*
* @typedef {Object} MigrationContext
* @property {import('../ports.js').ActorStoragePort} actorStorage
* @property {boolean} readOnly
* @property {() => Promise<string|null>} getDid
* @property {() => Promise<import('../crypto.js').SigningKey|null>} getSigningKey
* @property {() => Response} readOnlyError
* @property {(type: 'account'|'identity'|'sync', body: Object) => Promise<void>} emitEvent
* @property {(commit: {cid: string, rev: string}) => Promise<void>} emitSyncEvent
* @property {(did: string) => Promise<void>} notifyRelay
* @property {() => Promise<Array<{cid: string}>>} collectRepoBlocks
* @property {() => Promise<Array<{cid: string, recordUri: string}>>} collectReferencedBlobs
* @property {() => Promise<number>} relinkBlobRecords
*/
/**
* @param {MigrationContext} ctx
* @returns {{
* isDeactivated: () => Promise<boolean>,
* checkAccountActive: () => Promise<Response|null>,
* setAccountActive: (active: boolean) => Promise<void>,
* routes: import('../pds.js').Routes,
* }}
*/
export function createMigrationHandlers(ctx) {
const {
actorStorage,
readOnly,
getDid,
getSigningKey,
readOnlyError,
emitEvent,
emitSyncEvent,
notifyRelay,
collectRepoBlocks,
collectReferencedBlobs,
relinkBlobRecords,
} = ctx;
/**
* com.atproto.server.checkAccountStatus
* @param {Request} _request
* @param {URL} _url
* @param {import('../pds.js').Auth|null} auth
* @returns {Promise<Response>}
*/
async function handleCheckAccountStatus(_request, _url, auth) {
if (!auth) {
return Response.json(
{ error: 'AuthenticationRequired', message: 'Authentication required' },
{ status: 401 },
);
}
const did = await getDid();
if (!did) {
return Response.json(
{ error: 'InvalidRequest', message: 'PDS not initialized' },
{ status: 400 },
);
}
const commit = await actorStorage.getLatestCommit();
const blocks = await collectRepoBlocks();
const records = await actorStorage.listAllRecords();
const referenced = await collectReferencedBlobs();
const expected = new Set(referenced.map((r) => r.cid));
let imported = 0;
for (const cid of expected) {
if (await actorStorage.getBlob(cid)) imported++;
}
return Response.json({
activated: !(await isDeactivated()),
validDid: !!(await getSigningKey()),
repoCommit: commit?.cid || '',
repoRev: commit?.rev || '',
repoBlocks: blocks.length,
indexedRecords: records.length,
privateStateValues: 0,
expectedBlobs: expected.size,
importedBlobs: imported,
});
}
/**
* com.atproto.repo.listMissingBlobs
* @param {Request} _request
* @param {URL} url
* @param {import('../pds.js').Auth|null} auth
* @returns {Promise<Response>}
*/
async function handleListMissingBlobs(_request, url, auth) {
if (!auth) {
return Response.json(
{ error: 'AuthenticationRequired', message: 'Authentication required' },
{ status: 401 },
);
}
const limit = Math.min(
Number.parseInt(url.searchParams.get('limit') || '500', 10) || 500,
1000,
);
const cursor = url.searchParams.get('cursor');
const referenced = await collectReferencedBlobs();
/** @type {Array<{cid: string, recordUri: string}>} */
const missing = [];
const checked = new Set();
for (const ref of referenced) {
if (checked.has(ref.cid)) continue;
checked.add(ref.cid);
if (cursor && ref.cid <= cursor) continue;
if (!(await actorStorage.getBlob(ref.cid))) {
missing.push(ref);
}
}
missing.sort((a, b) => (a.cid < b.cid ? -1 : a.cid > b.cid ? 1 : 0));
const page = missing.slice(0, limit);
return Response.json({
blobs: page,
...(page.length === limit ? { cursor: page[page.length - 1].cid } : {}),
});
}
/**
* com.atproto.repo.importRepo
* @param {Request} request
* @param {URL} _url
* @param {import('../pds.js').Auth|null} auth
* @returns {Promise<Response>}
*/
async function handleImportRepo(request, _url, auth) {
if (readOnly) return readOnlyError();
if (!auth) {
return Response.json(
{ error: 'AuthenticationRequired', message: 'Authentication required' },
{ status: 401 },
);
}
const permissions = new ScopePermissions(auth.scope);
if (!permissions.hasFullAccess()) {
return Response.json(
{ error: 'InvalidScope', message: 'Full access required' },
{ status: 403 },
);
}
const did = await getDid();
if (!did) {
return Response.json(
{ error: 'InvalidRequest', message: 'PDS not initialized' },
{ status: 400 },
);
}
const car = new Uint8Array(await request.arrayBuffer());
if (car.length === 0) {
return Response.json(
{ error: 'InvalidRequest', message: 'Empty CAR body' },
{ status: 400 },
);
}
let result;
try {
result = await loadRepositoryFromCar(car, actorStorage);
} catch (err) {
return Response.json(
{ error: 'InvalidRequest', message: `Invalid CAR: ${err.message}` },
{ status: 400 },
);
}
if (result.did !== did) {
return Response.json(
{
error: 'InvalidRequest',
message: `CAR belongs to ${result.did}, not ${did}`,
},
{ status: 400 },
);
}
// Register blob → record links for the imported records. loadRepositoryFromCar
// stores records and blocks but not these links; without them the orphan
// cleanup treats every blob the imported repo references as unreferenced and
// reaps it, silently destroying a migrated account's blobs.
await relinkBlobRecords();
// Imported state is non-inductive for consumers: announce it
const commit = await actorStorage.getLatestCommit();
if (commit) await emitSyncEvent(commit);
return Response.json({});
}
/**
* Return whether the account is deactivated.
* @returns {Promise<boolean>}
*/
async function isDeactivated() {
return (await actorStorage.getAccountStatus()) === 'deactivated';
}
/**
* Reject repo/blob writes while the account is deactivated.
* @returns {Promise<Response|null>} 409 response when deactivated
*/
async function checkAccountActive() {
if (!(await isDeactivated())) return null;
return Response.json(
{
error: 'AccountDeactivated',
message: 'Account is deactivated',
},
{ status: 409 },
);
}
/**
* com.atproto.server.deactivateAccount
* @param {Request} request
* @param {URL} _url
* @param {import('../pds.js').Auth|null} auth
* @returns {Promise<Response>}
*/
async function handleDeactivateAccount(request, _url, auth) {
// Consume the body: responding without draining the request stream
// raises an uncaught error on Workers runtimes
await request.text().catch(() => '');
if (!auth) {
return Response.json(
{ error: 'AuthenticationRequired', message: 'Authentication required' },
{ status: 401 },
);
}
const permissions = new ScopePermissions(auth.scope);
if (!permissions.hasFullAccess()) {
return Response.json(
{ error: 'InvalidScope', message: 'Full access required' },
{ status: 403 },
);
}
// setAccountActive emits the #account (inactive) event, telling the network
// the account is no longer active here (e.g. it moved away).
await setAccountActive(false);
return Response.json({});
}
/**
* Set the account's active state and announce it on the firehose. Shared by
* the XRPC handlers and the account page, so the network sees the same
* event whichever triggered it.
* @param {boolean} active
* @returns {Promise<void>}
*/
async function setAccountActive(active) {
await actorStorage.setAccountStatus(active ? 'active' : 'deactivated');
await emitEvent(
'account',
active ? { active: true } : { active: false, status: 'deactivated' },
);
}
/**
* com.atproto.server.activateAccount
* @param {Request} request
* @param {URL} _url
* @param {import('../pds.js').Auth|null} auth
* @returns {Promise<Response>}
*/
async function handleActivateAccount(request, _url, auth) {
// Consume the body: responding without draining the request stream
// raises an uncaught error on Workers runtimes
await request.text().catch(() => '');
if (!auth) {
return Response.json(
{ error: 'AuthenticationRequired', message: 'Authentication required' },
{ status: 401 },
);
}
const permissions = new ScopePermissions(auth.scope);
if (!permissions.hasFullAccess()) {
return Response.json(
{ error: 'InvalidScope', message: 'Full access required' },
{ status: 403 },
);
}
// setAccountActive emits the #account (active) event. On top of that we
// re-announce identity: activation is the last step of an inbound migration,
// right after the DID was repointed to this server with a fresh signing key,
// so consumers must re-resolve the DID. Without the #identity event, AppViews
// keep verifying this account's service tokens against the *old* PDS's key
// and reject them ("could not verify jwt signature").
await setAccountActive(true);
// The crawl request goes first. A relay that has never heard of this server
// has no subscription, and an event announced before it asks to crawl
// reaches nobody. maybeAnnounceIdentity repeats the announcement each hour
// until the relay reports that it reads this server.
const did = await getDid();
if (did) await notifyRelay(did);
const handle = await actorStorage.getHandle();
await emitEvent('identity', { handle: handle || null });
return Response.json({});
}
return {
isDeactivated,
checkAccountActive,
setAccountActive,
routes: {
'/xrpc/com.atproto.server.checkAccountStatus': {
auth: 'required',
handler: handleCheckAccountStatus,
},
'/xrpc/com.atproto.repo.importRepo': {
method: 'POST',
auth: 'required',
handler: handleImportRepo,
},
'/xrpc/com.atproto.repo.listMissingBlobs': {
auth: 'required',
handler: handleListMissingBlobs,
},
'/xrpc/com.atproto.server.activateAccount': {
method: 'POST',
auth: 'required',
handler: handleActivateAccount,
},
'/xrpc/com.atproto.server.deactivateAccount': {
method: 'POST',
auth: 'required',
handler: handleDeactivateAccount,
},
},
};
}