chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/node - Node.js HTTP server adapter
import { createHash } from 'node:crypto';
import { createServer as createHttpServer } from 'node:http';
import { createConnection } from 'node:net';
import { secp256k1 } from '@noble/curves/secp256k1';
import { createFsBackupTarget, createFsBlobs } from '@pdsjs/blobs-fs';
import { PersonalDataServer } from '@pdsjs/core';
import { LexiconResolver } from '@pdsjs/lexicon-resolver';
import { createActorStorage, createSharedStorage } from '@pdsjs/storage-sqlite';
import { WebSocketServer } from 'ws';
/** How often extension background work is checked for anything due. */
const EXTENSION_TICK_MS = 10 * 1000;
/**
* node:crypto behind the core Hasher interface, for streamed blob uploads.
* @returns {import('@pdsjs/core/ports').Hasher}
*/
function createNodeHasher() {
const hash = createHash('sha256');
return {
/** @param {Uint8Array} chunk */
update(chunk) {
hash.update(chunk);
},
digest() {
return new Uint8Array(hash.digest());
},
};
}
/**
* Verify a secp256k1 signature the way the reference stack does: SHA-256 the
* data, then require a compact, low-S signature. Malformed input returns false
* rather than throwing. This is the injected half of the signature verifier —
* P-256 lives in @pdsjs/core/verify; secp256k1 needs @noble/curves, which core
* deliberately does not depend on.
*
* @param {Uint8Array} publicKey - compressed 33-byte point from a did:key
* @param {Uint8Array} data - the signed bytes
* @param {Uint8Array} sig - 64-byte compact r||s
* @returns {Promise<boolean>}
*/
export async function secp256k1Verify(publicKey, data, sig) {
try {
const hash = new Uint8Array(
await crypto.subtle.digest('SHA-256', /** @type {BufferSource} */ (data)),
);
return secp256k1.verify(sig, hash, publicKey, {
format: 'compact',
lowS: true,
});
} catch {
return false;
}
}
/**
* Sign with secp256k1 account keys — the curve the reference PDS gives every
* account, and the one some third-party services (Go feed generators) are
* alone able to verify. WebCrypto cannot sign this curve, so core takes the
* implementation as a port, mirroring the verifier above.
*
* @type {import('@pdsjs/core/ports').Secp256k1SignerPort}
*/
export const k256Signer = {
generatePrivateKey: () => secp256k1.utils.randomPrivateKey(),
getPublicKey: (privateKey, compressed = true) =>
secp256k1.getPublicKey(privateKey, compressed),
async sign(privateKey, data) {
const hash = new Uint8Array(
await crypto.subtle.digest('SHA-256', /** @type {BufferSource} */ (data)),
);
// Low-S, compact r||s: what atproto verifies and what secp256k1Verify
// above requires.
return secp256k1.sign(hash, privateKey, { lowS: true }).toCompactRawBytes();
},
};
/**
* @typedef {Object} PdsServer
* @property {PersonalDataServer} pds - The PDS instance
* @property {import('node:http').Server} server - HTTP server
* @property {Object} db - SQLite database instance
* @property {() => Promise<import('node:http').Server>} listen - Start server
* @property {() => Promise<void>} close - Close server
*/
/**
* Create WebSocket port for Node.js
* @param {WeakMap<Request, import('ws').WebSocket>} upgradeMap
* @param {import('ws').WebSocketServer} wss - WebSocket server for broadcast
* @returns {import('@pdsjs/core/ports').WebSocketPort}
*/
function createWebSocket(upgradeMap, wss) {
return {
isUpgrade(request) {
return upgradeMap.has(request);
},
upgrade(request, onConnect) {
const ws = upgradeMap.get(request);
if (!ws) {
return new Response('WebSocket upgrade failed', { status: 500 });
}
onConnect({
send(data) {
if (ws.readyState === ws.OPEN) {
ws.send(data);
}
},
close() {
ws.close();
},
});
// Return 200 - actual upgrade handled by server's 'upgrade' event
// Node.js Response doesn't support status 101
return new Response(null, {
status: 200,
headers: { 'X-WebSocket-Upgraded': 'true' },
});
},
broadcast(data) {
// Send to all connected WebSocket clients
for (const client of wss.clients) {
if (client.readyState === client.OPEN) {
try {
client.send(data);
} catch {
// Client may have disconnected
}
}
}
},
};
}
/**
* A mail sender that writes the message to stderr instead of delivering it.
*
* For local development: it makes the verification code visible without a
* mail provider. Never point a deployment at it — every code it "sends"
* lands in the server log, where anyone with log access can read it.
* @returns {import('@pdsjs/core/ports').EmailPort}
*/
export function createLogEmailer() {
return {
async send({ to, subject, text }) {
console.error(
`\n--- email (not sent) ---\nTo: ${to}\nSubject: ${subject}\n\n${text}\n------------------------\n`,
);
},
};
}
/**
* A mail sender that delivers over SMTP, for pointing local development at a
* catch-all inbox such as Mailpit so sent mail is visible in a browser. It
* speaks only enough of the protocol for an unauthenticated, non-TLS relay,
* which is what those dev inboxes expose. Never point it at a real provider.
* @param {Object} [options]
* @param {string} [options.host='127.0.0.1'] - SMTP host
* @param {number} [options.port=1025] - SMTP port
* @param {string} [options.from='no-reply@localhost'] - Envelope and From address
* @returns {import('@pdsjs/core/ports').EmailPort}
*/
export function createSmtpEmailer({
host = '127.0.0.1',
port = 1025,
from = 'no-reply@localhost',
} = {}) {
return {
send({ to, subject, text, html }) {
return deliverOverSmtp({ host, port, from, to, subject, text, html });
},
};
}
/**
* Build an RFC 5322 message, with CRLF line endings and leading dots doubled
* so a body line of "." cannot end the DATA stage early. With `html` present
* the two bodies are carried as multipart/alternative, so a client shows the
* HTML and falls back to the text.
* @param {{from: string, to: string, subject: string, text: string, html?: string}} message
*/
function buildMessage({ from, to, subject, text, html }) {
const headers = [
`From: ${from}`,
`To: ${to}`,
`Subject: ${subject}`,
`Date: ${new Date().toUTCString()}`,
'MIME-Version: 1.0',
];
let body;
if (html) {
const boundary = `=_pdsjs_${crypto.randomUUID()}`;
headers.push(`Content-Type: multipart/alternative; boundary="${boundary}"`);
body = [
'',
`--${boundary}`,
'Content-Type: text/plain; charset=utf-8',
'',
...text.split(/\r?\n/),
`--${boundary}`,
'Content-Type: text/html; charset=utf-8',
'',
...html.split(/\r?\n/),
`--${boundary}--`,
];
} else {
headers.push('Content-Type: text/plain; charset=utf-8', '');
body = text.split(/\r?\n/);
}
return [...headers, ...body]
.map((line) => (line.startsWith('.') ? `.${line}` : line))
.join('\r\n');
}
/**
* @param {{host: string, port: number, from: string, to: string, subject: string, text: string, html?: string}} options
* @returns {Promise<void>}
*/
function deliverOverSmtp({ host, port, from, to, subject, text, html }) {
return new Promise((resolve, reject) => {
const socket = createConnection({ host, port });
socket.setEncoding('utf8');
socket.setTimeout(10000);
// Each step waits for a reply carrying `expect`, then sends the next line.
// The opening step has no line to send: it just awaits the 220 greeting.
const script = [
{ expect: 220 },
{ send: 'EHLO localhost', expect: 250 },
{ send: `MAIL FROM:<${from}>`, expect: 250 },
{ send: `RCPT TO:<${to}>`, expect: 250 },
{ send: 'DATA', expect: 354 },
{
send: `${buildMessage({ from, to, subject, text, html })}\r\n.`,
expect: 250,
},
{ send: 'QUIT', expect: 221 },
];
let step = 0;
let buffer = '';
const fail = (/** @type {unknown} */ error) => {
socket.destroy();
reject(error instanceof Error ? error : new Error(String(error)));
};
socket.on('error', fail);
socket.on('timeout', () => fail(new Error('SMTP connection timed out')));
socket.on('data', (/** @type {string} */ chunk) => {
buffer += chunk;
// A reply may span several lines (`250-...` then `250 ...`); the line
// with a space after the code is the last one.
let match = buffer.match(/^(?:\d{3}-[^\n]*\n)*(\d{3}) [^\n]*\n/);
while (match) {
const code = Number(match[1]);
buffer = buffer.slice(match[0].length);
if (code !== script[step].expect)
return fail(
new Error(`SMTP replied ${code}, expected ${script[step].expect}`),
);
step += 1;
if (step >= script.length) {
socket.end();
return resolve(undefined);
}
socket.write(`${script[step].send}\r\n`);
match = buffer.match(/^(?:\d{3}-[^\n]*\n)*(\d{3}) [^\n]*\n/);
}
});
});
}
/**
* Create Node.js PDS server
* @param {Object} options
* @param {string} options.dbPath - Path to SQLite database
* @param {string} [options.blobsDir] - Directory for blob storage (used if blobs not provided)
* @param {import('@pdsjs/core/ports').BlobPort} [options.blobs] - Custom blob storage adapter
* @param {string} options.jwtSecret - JWT signing secret
* @param {number} [options.blobUploadLimit] - Max blob upload size in bytes (default 50 MiB). Bluesky video posting needs room for the processed video the video service writes back, up to its 300MB cap.
* @param {number} [options.port=3000] - Server port
* @param {string} [options.hostname] - PDS hostname
* @param {string} [options.appviewUrl] - AppView URL for proxying
* @param {string} [options.appviewDid] - AppView DID for service auth
* @param {string} [options.relayUrl] - Relay URL for firehose notifications (e.g., http://localhost:2470)
* @param {string} [options.plcUrl] - PLC directory URL for identity operations
* @param {string} [options.password] - Password for createSession
* @param {import('@pdsjs/core/ports').LexiconResolverPort} [options.lexiconResolver] - Lexicon resolver for record validation
* @param {import('@pdsjs/core/ports').EmailPort} [options.emailer] - Sends account mail; without one an address cannot be confirmed
* @param {boolean} [options.accountUi] - Serve the account application at /account. Needs the account-ui package (an optional peer dependency); without it the server-rendered pages are served instead.
* @param {string} [options.backupDir] - Directory scheduled backups land in, ideally on a different disk than the database and blobs
* @param {import('@pdsjs/core/ports').BackupTargetPort} [options.backupTarget] - Custom backup target (createS3BackupTarget from the blobs-s3 package, say); takes precedence over backupDir. Neither given, the backups page reports backups as unavailable.
* @param {boolean} [options.spaces] - Enable permissioned data (proposal 0016).
* Requires the spaces package to be installed; it is an optional peer dependency.
* @param {{git?: {http?: boolean, browse?: boolean, external?: boolean, guardOnly?: boolean}, sites?: {domain?: string, apexName?: string, queries?: boolean}, drive?: boolean, oci?: boolean, npm?: boolean}} [options.experimental] - Feature
* experiments, exempt from semver: anything in this group may change shape
* or disappear in any release, including patches. A feature graduates by
* moving to a top-level option.
*
* `git` (requires the git package, an optional peer dependency): `http`
* serves read-only git smart HTTP under /git/ for dev.pdsjs.git.repo
* records (default off); `browse` backs the account page's file tree
* (default on when the package is installed). `external: true` turns the
* embedded hosting off so a separate git appview serves the reads while
* this PDS stays the data store; `guardOnly: true` keeps the one piece
* that cannot move, the branch-protection write guard, and drops the
* routes. `external` implies `guardOnly`.
*
* `sites` (requires the sites package, an optional peer dependency):
* `domain` is the apex static sites serve under (dev.pdsjs.site.deploy
* records + blobs) and setting it enables serving; the reverse proxy must
* route the apex and *.domain here, with a wildcard certificate (DNS-01).
* `apexName` is the site served at the apex (default "home"). `queries`
* serves dev.pdsjs.query.run over this repo's dev.pdsjs.query.def records.
*
* `drive` backs the account page's Drive section over dev.pdsjs.drive.file
* records (default on when the drive package, an optional peer dependency,
* is installed).
*
* `oci` (requires the oci package, an optional peer dependency) serves an
* OCI image registry under /v2/ for dev.pdsjs.oci.repo records (default
* off). Pulls are anonymous; pushes authenticate with Basic credentials
* createSession accepts.
*
* `npm` (requires the npm package, an optional peer dependency) serves an
* npm registry under /npm/ for dev.pdsjs.npm.package records (default
* off). Installs are anonymous; publishes authenticate with credentials
* createSession accepts, stored by npm login as the auth token.
* @returns {Promise<PdsServer>}
*/
export async function createServer({
dbPath,
blobsDir,
blobs: blobsArg,
jwtSecret,
blobUploadLimit,
port = 3000,
hostname,
appviewUrl,
appviewDid,
relayUrl,
plcUrl,
password,
lexiconResolver,
emailer,
accountUi = false,
spaces = false,
experimental = {},
backupDir,
backupTarget: backupTargetArg,
}) {
const {
git = {},
sites = {},
drive = true,
oci = false,
npm = false,
} = experimental;
const gitHttp = Boolean(git.http);
const gitBrowse = git.browse !== false;
const {
domain: siteDomain,
apexName: siteApexName,
queries: enableQueries,
} = sites;
// Dynamic import for better-sqlite3 (optional peer dependency)
const Database = (await import('better-sqlite3')).default;
const db = new Database(dbPath);
// Create actor storage (per-DID data)
const actorStorage = createActorStorage(db);
// Create shared storage (global data) - can use same db for single-tenant
const sharedStorage = createSharedStorage(db);
// Signature verification, shared by createAccount (migration service tokens)
// and, when enabled, spaces (credentials). P-256 is built into core;
// secp256k1 — the curve the reference PDS gives every account — is supplied
// here via @noble/curves. The DID resolver returns full documents and honours
// this server's PLC directory rather than the public default.
const { createVerifier } = await import('@pdsjs/core/verify');
const { createCachedDidResolver } = await import('@pdsjs/lexicon-resolver');
const verifier = createVerifier({ secp256k1: secp256k1Verify });
// One resolver for the whole server, built once so its cache spans requests.
// Every space request verifies a credential, which reads its issuer's key.
const didResolver = createCachedDidResolver({ plcUrl });
if (!blobsArg && !blobsDir) {
throw new Error('Either blobs or blobsDir must be provided');
}
const blobs = blobsArg ?? createFsBlobs(/** @type {string} */ (blobsDir));
/** @type {import('@pdsjs/core/ports').SpaceStoragePort|undefined} */
let spaceStorageRef;
// Feature packages on the extension seam, each an optional peer imported
// only when its option is on. Git is on whenever the package is installed,
// unless `git.external` hands the reads to a separate git appview: then
// only the branch-protection write guard stays (guardOnly), and with
// neither option the full extension serves the routes itself.
/** @type {import('@pdsjs/core/ports').ExtensionFactory[]} */
const extensions = [];
// Static sites and apps: the installs listing needs no option, a site
// domain turns serving on.
try {
const { createSitesExtension } = await import('@pdsjs/sites/extension');
extensions.push(
createSitesExtension({
domain: siteDomain,
apexName: siteApexName,
queries: enableQueries,
}),
);
} catch {
// Optional peer not installed
}
try {
if (git.external || git.guardOnly) {
// The reads are served elsewhere; only the write-time enforcement
// stays, and the routes drop with it.
const { createGitGuardExtension } = await import('@pdsjs/git/extension');
extensions.push(createGitGuardExtension());
} else {
const { createGitExtension } = await import('@pdsjs/git/extension');
extensions.push(createGitExtension({ http: gitHttp, browse: gitBrowse }));
}
} catch {
// Optional peer not installed
}
// Permissioned data (proposal 0016), off by default. The storage is this
// platform's; the space endpoints, the admin port and the account page's
// space routes come from @pdsjs/spaces, an optional peer imported only
// here. With this off the space endpoints do not exist and requests fall
// through to the normal 404, which matters because space-host discovery
// falls back to #atproto_pds: a space-unaware PDS still receives space
// requests from optimistic clients.
if (spaces) {
const [{ createSpaceStorage }, { createSpacesExtension }] =
await Promise.all([
import('@pdsjs/storage-sqlite'),
import('@pdsjs/spaces/extension'),
]);
spaceStorageRef = createSpaceStorage(db);
extensions.push(createSpacesExtension());
}
// The built application arrives as a module rather than a directory, so it
// loads the same way on every platform. Optional: absent, /account keeps
// serving the server-rendered pages.
let accountApp;
if (accountUi) {
({ accountApp } = await import('@pdsjs/account-ui'));
}
const backupTarget =
backupTargetArg ??
(backupDir ? createFsBackupTarget(backupDir) : undefined);
// WebSocket infrastructure
const upgradeMap = new WeakMap();
const wss = new WebSocketServer({ noServer: true });
const webSocket = createWebSocket(upgradeMap, wss);
// Create default lexicon resolver if not provided
const resolver = lexiconResolver ?? new LexiconResolver();
// Files stored in the repo, for the account page's Drive section: on
// whenever the drive package (an optional peer) is installed, unless
// refused.
if (drive) {
try {
const { createDriveExtension } = await import('@pdsjs/drive/extension');
extensions.push(createDriveExtension());
} catch {
// Optional peer not installed
}
}
if (oci) {
const { createOciExtension } = await import('@pdsjs/oci/extension');
extensions.push(createOciExtension());
}
if (npm) {
const { createNpmExtension } = await import('@pdsjs/npm/extension');
extensions.push(createNpmExtension());
}
// Create PDS with both storages
const pds = new PersonalDataServer({
actorStorage,
sharedStorage,
blobs,
webSocket,
jwtSecret,
blobUploadLimit,
hasher: createNodeHasher,
hostname,
appviewUrl,
appviewDid,
relayUrl,
plcUrl,
password,
lexiconResolver: resolver,
emailer,
spaceStorage: spaceStorageRef,
accountApp,
verifier,
didResolver,
k256Signer,
backupTarget,
extensions,
});
const server = createHttpServer(async (req, res) => {
try {
const request = nodeToWebRequest(req);
const response = await pds.fetch(request);
// Skip normal response for WebSocket upgrades
if (
response.headers.get('X-WebSocket-Upgraded') &&
upgradeMap.has(request)
) {
return; // WebSocket handled by upgrade event
}
await webToNodeResponse(response, res);
} catch (error) {
console.error('Server error:', error);
res.statusCode = 500;
res.setHeader('Content-Type', 'application/json');
res.end(
JSON.stringify({
error: 'InternalServerError',
message: error.message,
}),
);
}
});
// Handle WebSocket upgrades
server.on('upgrade', (req, socket, head) => {
if (req.url?.startsWith('/xrpc/com.atproto.sync.subscribeRepos')) {
wss.handleUpgrade(req, socket, head, async (ws) => {
const request = nodeToWebRequest(req);
upgradeMap.set(request, ws);
try {
await pds.fetch(request);
} finally {
upgradeMap.delete(request);
}
});
} else {
socket.destroy();
}
});
// Start maintenance interval (every hour): cleanup, then a backup if due
const cleanupInterval = setInterval(
async () => {
try {
await pds.runBlobCleanup();
await sharedStorage.cleanupExpiredDpopJtis();
await pds.maybeAnnounceIdentity();
await pds.maybeBackfillKemKey();
await pds.maybeRunScheduledBackup();
} catch (e) {
console.error('Maintenance error:', e);
}
},
60 * 60 * 1000,
);
// An account created before KEM keys gets one now rather than in an hour.
pds.maybeBackfillKemKey();
// Extension background work, such as retrying space write notices. A tick
// with nothing due is one indexed query.
const tickInterval = setInterval(() => {
pds.runExtensionTicks().catch((e) => console.error('Tick error:', e));
}, EXTENSION_TICK_MS);
return {
pds,
server,
db,
/**
* Start listening on configured port
* @returns {Promise<import('node:http').Server>}
*/
listen() {
return new Promise((resolve) => {
server.listen(port, () => {
console.log(`PDS listening on http://localhost:${port}`);
resolve(server);
});
});
},
/**
* Close server and cleanup
* @returns {Promise<void>}
*/
close() {
clearInterval(cleanupInterval);
clearInterval(tickInterval);
// Close all WebSocket connections first
wss.clients.forEach((client) => {
client.terminate();
});
wss.close();
db.close();
return new Promise((resolve, reject) => {
server.close((err) => {
if (err) reject(err);
else resolve();
});
});
},
};
}
/**
* Convert Node.js IncomingMessage to Web Request
* @param {import('node:http').IncomingMessage} req
* @returns {Request}
*/
function nodeToWebRequest(req) {
// Check for encrypted socket (HTTPS)
const socket = /** @type {import('node:tls').TLSSocket} */ (req.socket);
const protocol = socket.encrypted ? 'https' : 'http';
const host = req.headers.host || 'localhost';
const url = new URL(req.url || '/', `${protocol}://${host}`);
const headers = new Headers();
for (const [key, value] of Object.entries(req.headers)) {
if (value) {
headers.set(key, Array.isArray(value) ? value.join(', ') : value);
}
}
const hasBody = req.method !== 'GET' && req.method !== 'HEAD';
return new Request(
url.toString(),
/** @type {RequestInit} */ ({
method: req.method,
headers,
body: hasBody ? req : null,
duplex: hasBody ? 'half' : undefined,
}),
);
}
/**
* Write Web Response to Node.js ServerResponse
* @param {Response} response
* @param {import('node:http').ServerResponse} res
*/
async function webToNodeResponse(response, res) {
res.statusCode = response.status;
res.statusMessage = response.statusText || '';
for (const [key, value] of response.headers) {
res.setHeader(key, value);
}
if (response.body) {
const reader = response.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
// A false return means the socket's buffer is full. Waiting for drain
// holds streamed blobs to one chunk in memory instead of the whole
// body.
if (!res.write(value)) {
await new Promise((resolve) => res.once('drain', resolve));
}
}
}
res.end();
}