chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/deno - Deno HTTP server adapter
/// <reference types="@types/deno" />
import { createHash } from 'node:crypto';
import { DatabaseSync } from 'node:sqlite';
import { secp256k1 } from '@noble/curves/secp256k1';
import { createDenoBlobs } from '@pdsjs/blobs-deno';
import { PersonalDataServer } from '@pdsjs/core';
import { createVerifier } from '@pdsjs/core/verify';
import {
createCachedDidResolver,
LexiconResolver,
} from '@pdsjs/lexicon-resolver';
import { createActorStorage, createSharedStorage } from '@pdsjs/storage-sqlite';
/** 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.js').Hasher}
*/
function createDenoHasher() {
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>}
*/
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}
*/
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();
},
};
/**
* Create WebSocket port for Deno
* @returns {import('@pdsjs/core/ports.js').WebSocketPort}
*/
function createWebSocketPort() {
/** @type {Set<WebSocket>} */
const clients = new Set();
return {
isUpgrade(request) {
return request.headers.get('upgrade')?.toLowerCase() === 'websocket';
},
upgrade(request, onConnect) {
const upgrade = Deno.upgradeWebSocket(request);
const { socket, response } = upgrade;
clients.add(socket);
socket.onclose = () => clients.delete(socket);
socket.onerror = () => clients.delete(socket);
socket.onopen = () => {
onConnect({
send(data) {
if (socket.readyState === WebSocket.OPEN) {
socket.send(data);
}
},
close() {
socket.close();
},
});
};
return response;
},
broadcast(data) {
for (const client of clients) {
if (client.readyState === WebSocket.OPEN) {
try {
client.send(data);
} catch {
// Client may have disconnected
}
}
}
},
};
}
/**
* @typedef {Object} PdsServer
* @property {PersonalDataServer} pds - The PDS instance
* @property {Deno.HttpServer | null} server - HTTP server (null until listen() called)
* @property {DatabaseSync} db - SQLite database instance
* @property {() => Promise<Deno.HttpServer>} listen - Start server
* @property {() => Promise<void>} close - Close server
*/
/**
* Create Deno 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.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
* @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').BackupTargetPort} [options.backupTarget] - Where scheduled backups land (createFsBackupTarget from the blobs-fs package, or createS3BackupTarget from blobs-s3). Absent, the backups page reports backups as unavailable.
* @param {{sites?: {domain?: string, apexName?: string, queries?: boolean}}} [options.experimental] - Feature experiments, exempt from semver: anything in this group may change shape or disappear in any release, including patches. `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; `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.
* @returns {Promise<PdsServer>}
*/
export async function createServer({
dbPath,
blobsDir,
blobs: blobsArg,
jwtSecret,
port = 3000,
hostname,
appviewUrl,
appviewDid,
relayUrl,
plcUrl,
password,
lexiconResolver,
backupTarget,
experimental = {},
}) {
const { sites = {} } = experimental;
const {
domain: siteDomain,
apexName: siteApexName,
queries: enableQueries,
} = sites;
const db = new DatabaseSync(dbPath);
const actorStorage = createActorStorage(db);
const sharedStorage = createSharedStorage(db);
if (!blobsArg && !blobsDir) {
throw new Error('Either blobs or blobsDir must be provided');
}
const blobs = blobsArg ?? createDenoBlobs(/** @type {string} */ (blobsDir));
const webSocket = createWebSocketPort();
const resolver = lexiconResolver ?? new LexiconResolver();
// Signature verification for createAccount (migration service tokens). 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.
const verifier = createVerifier({ secp256k1: secp256k1Verify });
// One resolver for the whole server, built once so its cache spans requests.
const didResolver = createCachedDidResolver({ plcUrl });
// Feature packages on the extension seam. Git is on whenever the package,
// an optional peer, is installed: branch protection and the repositories
// page need no option.
/** @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 {
const { createGitExtension } = await import('@pdsjs/git/extension');
extensions.push(createGitExtension());
} catch {
// Optional peer not installed
}
const pds = new PersonalDataServer({
actorStorage,
sharedStorage,
blobs,
webSocket,
jwtSecret,
hasher: createDenoHasher,
hostname,
appviewUrl,
appviewDid,
relayUrl,
plcUrl,
password,
lexiconResolver: resolver,
verifier,
didResolver,
k256Signer,
backupTarget,
extensions,
});
/** @type {Deno.HttpServer | null} */
let server = null;
/** @type {ReturnType<typeof setInterval> | null} */
let cleanupInterval = null;
/** @type {ReturnType<typeof setInterval> | null} */
let tickInterval = null;
return {
pds,
get server() {
return server;
},
db,
/**
* Start listening on configured port
* @returns {Promise<Deno.HttpServer>}
*/
listen() {
return new Promise((resolve) => {
server = Deno.serve(
{
port,
onListen: () => {
console.log(`PDS listening on http://localhost:${port}`);
resolve(server);
},
},
(request) => pds.fetch(request),
);
// Start maintenance interval (every hour): cleanup, then a backup
// if due
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.
tickInterval = setInterval(() => {
pds.runExtensionTicks().catch((e) => console.error('Tick error:', e));
}, EXTENSION_TICK_MS);
});
},
/**
* Close server and cleanup
* @returns {Promise<void>}
*/
async close() {
if (cleanupInterval) {
clearInterval(cleanupInterval);
}
if (tickInterval) {
clearInterval(tickInterval);
}
if (server) {
await server.shutdown();
}
db.close();
},
};
}