chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/cloudflare - Cloudflare Workers + Durable Objects adapter
import { secp256k1 } from '@noble/curves/secp256k1';
import { accountApp } from '@pdsjs/account-ui';
import { createS3FetchBackupTarget } from '@pdsjs/blobs-s3';
import { PersonalDataServer } from '@pdsjs/core';
import { APP_PASSWORD_SCHEMA_SQL } from '@pdsjs/core/app-password';
import { SESSION_SCHEMA_SQL } from '@pdsjs/core/auth';
import {
BLOB_FACETS_APPS_SQL,
BLOB_FACETS_KINDS_SQL,
BLOB_FACETS_TOTAL_SQL,
BLOB_FACETS_UNUSED_SQL,
blobListCursor,
buildBlobListQuery,
} from '@pdsjs/core/blob-sql';
import { createVerifier } from '@pdsjs/core/verify';
import { PASSKEY_SCHEMA_SQL } from '@pdsjs/core/webauthn';
import { createDriveExtension } from '@pdsjs/drive/extension';
import { createGitExtension } from '@pdsjs/git/extension';
import {
createCachedDidResolver,
LexiconResolver,
} from '@pdsjs/lexicon-resolver';
import { createNpmExtension } from '@pdsjs/npm/extension';
import { createOciExtension } from '@pdsjs/oci/extension';
import { createSitesExtension } from '@pdsjs/sites/extension';
import { createSpacesExtension } from '@pdsjs/spaces/extension';
import {
CF_API_PROXY_PREFIX,
cfApiProxyEnabled,
handleCfApiProxy,
} from './cf-api-proxy.js';
import { createSpaceStorage } from './space.js';
/** Durable Object storage key holding when hourly maintenance is next due. */
const MAINTENANCE_AT_KEY = 'maintenanceAt';
const MAINTENANCE_INTERVAL_MS = 60 * 60 * 1000;
/**
* 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. Injected into the verifier for secp256k1 keys — the
* curve the reference PDS gives every account — since @pdsjs/core/verify only
* builds in P-256.
*
* @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();
},
};
/**
* @typedef {import('@pdsjs/core/ports').BlobPort} BlobPort
* @typedef {import('@pdsjs/core/ports').ActorStoragePort} ActorStoragePort
* @typedef {import('@pdsjs/core/ports').SharedStoragePort} SharedStoragePort
*/
// ============================================================================
// D1 Storage Adapter (Durable Object SQLite)
// ============================================================================
/**
* @template T
* @typedef {Object} SqlCursor
* @property {() => T[]} toArray - Get all rows as array
* @property {() => T|undefined} one - Get single row (throws if multiple)
* @property {number} rowsRead - Number of rows read
* @property {number} rowsWritten - Number of rows written
*/
/**
* @typedef {Object} D1SqlInterface
* @property {(sql: string, ...params: any[]) => SqlCursor<any>} exec - Execute SQL query
*/
// ============================================================================
// Row type definitions for SQLite queries
// ============================================================================
/** @typedef {{cid: string, value: Uint8Array|ArrayBuffer}} RecordResult */
/** @typedef {{uri: string, cid: string, rkey: string, value: Uint8Array|ArrayBuffer}} RecordRow */
/** @typedef {{collection: string, rkey: string, cid: string}} RecordKeyRow */
/** @typedef {{seq: number, cid: string, rev: string}} CommitRow */
/** @typedef {{seq: number, did: string, commit_cid: string, evt: Uint8Array|ArrayBuffer}} EventRow */
/** @typedef {{mime_type: string, size: number, created_at: number}} BlobMetaRow */
/** @typedef {{cid: string}} CidRow */
/** @typedef {{data: string, expires_at: number}} OAuthRequestRow */
/** @typedef {{data: string}} OAuthTokenRow */
/**
* Safe wrapper for .one() that returns null instead of throwing when no results
* Cloudflare Durable Object SQLite throws "Expected exactly one result" when no rows
* @template T
* @param {SqlCursor<T>} cursor - SQL cursor from sql.exec()
* @returns {T|null}
*/
function safeOne(cursor) {
const rows = cursor.toArray();
return rows.length > 0 ? rows[0] : null;
}
/** Durable Object SQL binds at most 100 parameters to one statement. */
const MAX_BOUND = 100;
/**
* @template T
* @param {T[]} items
* @returns {T[][]}
*/
function chunks(items) {
/** @type {T[][]} */
const out = [];
for (let i = 0; i < items.length; i += MAX_BOUND) {
out.push(items.slice(i, i + MAX_BOUND));
}
return out;
}
/**
* @typedef {Object} BlobLike
* @property {ArrayBuffer} buffer
* @property {number} byteOffset
* @property {number} byteLength
*/
/**
* Normalize BLOB value to Uint8Array
* Cloudflare D1 returns ArrayBuffer for BLOB columns
* @param {ArrayBuffer|Uint8Array|BlobLike|null|undefined} value
* @returns {Uint8Array|null}
*/
function normalizeBlob(value) {
if (!value) return null;
if (value instanceof Uint8Array) return value;
if (value instanceof ArrayBuffer) return new Uint8Array(value);
if ('buffer' in value && value.buffer instanceof ArrayBuffer)
return new Uint8Array(value.buffer, value.byteOffset, value.byteLength);
return /** @type {Uint8Array} */ (value);
}
/**
* Create storage adapter for Cloudflare D1 (Durable Object SQLite). Exported
* for the tests, which run it over better-sqlite3.
* @param {D1SqlInterface} sql - D1 SQL interface from state.storage.sql
* @returns {ActorStoragePort}
*/
export function createActorStorage(sql) {
// Initialize schema
sql.exec(`
CREATE TABLE IF NOT EXISTS metadata (
key TEXT PRIMARY KEY,
value BLOB
);
CREATE TABLE IF NOT EXISTS blocks (
cid TEXT PRIMARY KEY,
data BLOB
);
CREATE TABLE IF NOT EXISTS records (
uri TEXT PRIMARY KEY,
cid TEXT,
collection TEXT,
rkey TEXT,
value BLOB
);
CREATE INDEX IF NOT EXISTS idx_records_collection ON records(collection, rkey);
CREATE INDEX IF NOT EXISTS idx_records_cid ON records(cid);
CREATE TABLE IF NOT EXISTS commits (
seq INTEGER PRIMARY KEY,
cid TEXT,
rev TEXT,
prev TEXT
);
CREATE TABLE IF NOT EXISTS seq_events (
seq INTEGER PRIMARY KEY,
did TEXT,
commit_cid TEXT,
evt BLOB
);
CREATE TABLE IF NOT EXISTS blobs (
cid TEXT PRIMARY KEY,
mime_type TEXT,
size INTEGER,
created_at INTEGER
);
CREATE TABLE IF NOT EXISTS record_blobs (
blob_cid TEXT,
record_uri TEXT,
record_time INTEGER,
PRIMARY KEY (blob_cid, record_uri)
);
`);
// Databases created before record_time existed gain the column in place.
try {
sql.exec('ALTER TABLE record_blobs ADD COLUMN record_time INTEGER');
} catch {
// Already present
}
return {
async getRecord(uri) {
const row = /** @type {RecordResult|null} */ (
safeOne(sql.exec('SELECT cid, value FROM records WHERE uri = ?', uri))
);
if (!row) return null;
const value = normalizeBlob(row.value);
if (!value) return null;
return { cid: row.cid, value };
},
async listRecords(collection, cursor, limit, reverse = false) {
// `order`/`compare` are fixed strings, not input — safe to interpolate.
const order = reverse ? 'DESC' : 'ASC';
const compare = reverse ? '<' : '>';
/** @type {RecordRow[]} */
const rows = cursor
? sql
.exec(
`SELECT uri, cid, rkey, value FROM records WHERE collection = ? AND rkey ${compare} ? ORDER BY rkey ${order} LIMIT ?`,
collection,
cursor,
limit,
)
.toArray()
: sql
.exec(
`SELECT uri, cid, rkey, value FROM records WHERE collection = ? ORDER BY rkey ${order} LIMIT ?`,
collection,
limit,
)
.toArray();
return {
records: rows.map((r) => ({
uri: r.uri,
cid: r.cid,
value: normalizeBlob(r.value) || new Uint8Array(0),
})),
cursor: rows.length === limit ? rows[rows.length - 1].rkey : null,
};
},
async listAllRecords() {
/** @type {RecordKeyRow[]} */
const rows = sql
.exec(
"SELECT collection, rkey, cid FROM records ORDER BY (collection || '/' || rkey)",
)
.toArray();
return rows.map((r) => ({
key: `${r.collection}/${r.rkey}`,
cid: r.cid,
}));
},
async putRecord(uri, cid, collection, rkey, value) {
sql.exec(
'INSERT OR REPLACE INTO records (uri, cid, collection, rkey, value) VALUES (?, ?, ?, ?, ?)',
uri,
cid,
collection,
rkey,
value,
);
},
async deleteRecord(uri) {
sql.exec('DELETE FROM record_blobs WHERE record_uri = ?', uri);
sql.exec('DELETE FROM records WHERE uri = ?', uri);
},
async getBlock(cid) {
const row = safeOne(
sql.exec('SELECT data FROM blocks WHERE cid = ?', cid),
);
return row ? normalizeBlob(row.data) : null;
},
async putBlock(cid, data) {
sql.exec(
'INSERT OR IGNORE INTO blocks (cid, data) VALUES (?, ?)',
cid,
data,
);
},
async deleteBlocks(cids) {
for (const part of chunks(cids)) {
const marks = part.map(() => '?').join(',');
sql.exec(`DELETE FROM blocks WHERE cid IN (${marks})`, ...part);
}
},
async listBlocks(cursor, limit) {
const rows = (
cursor
? sql.exec(
'SELECT cid, data FROM blocks WHERE cid > ? ORDER BY cid LIMIT ?',
cursor,
limit,
)
: sql.exec('SELECT cid, data FROM blocks ORDER BY cid LIMIT ?', limit)
).toArray();
const blocks = rows.map((r) => ({
cid: String(r.cid),
data: /** @type {Uint8Array} */ (normalizeBlob(r.data)),
}));
const last = blocks.length === limit ? blocks[limit - 1].cid : null;
return { blocks, cursor: last };
},
async recordCidsAmong(cids) {
/** @type {string[]} */
const found = [];
for (const part of chunks(cids)) {
const marks = part.map(() => '?').join(',');
const rows = sql
.exec(
`SELECT DISTINCT cid FROM records WHERE cid IN (${marks})`,
...part,
)
.toArray();
found.push(...rows.map((r) => String(r.cid)));
}
return found;
},
async getBlocksCollected() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'blocksCollected'"),
);
return String(row?.value) === '1';
},
async setBlocksCollected(collected) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('blocksCollected', ?)",
collected ? '1' : '0',
);
},
async getLatestCommit() {
const row = /** @type {CommitRow|null} */ (
safeOne(
sql.exec(
'SELECT seq, cid, rev FROM commits ORDER BY seq DESC LIMIT 1',
),
)
);
return row || null;
},
async putCommit(seq, cid, rev, prev) {
sql.exec(
'INSERT INTO commits (seq, cid, rev, prev) VALUES (?, ?, ?, ?)',
seq,
cid,
rev,
prev,
);
},
async getEvents(cursor, limit) {
/** @type {EventRow[]} */
const rows = cursor
? sql
.exec(
'SELECT seq, did, commit_cid, evt FROM seq_events WHERE seq > ? ORDER BY seq LIMIT ?',
cursor,
limit,
)
.toArray()
: sql
.exec(
'SELECT seq, did, commit_cid, evt FROM seq_events ORDER BY seq LIMIT ?',
limit,
)
.toArray();
return {
events: rows.map((r) => ({
seq: r.seq,
commit_cid: r.commit_cid,
evt: normalizeBlob(r.evt) || new Uint8Array(0),
})),
cursor: rows.length > 0 ? rows[rows.length - 1].seq : cursor,
};
},
async putEvent(seq, did, commitCid, evt) {
sql.exec(
'INSERT INTO seq_events (seq, did, commit_cid, evt) VALUES (?, ?, ?, ?)',
seq,
did,
commitCid,
evt,
);
},
async getEventsBefore(before, limit) {
/** @type {EventRow[]} */
const rows = sql
.exec(
'SELECT seq, evt FROM seq_events WHERE seq < ? ORDER BY seq DESC LIMIT ?',
before,
limit,
)
.toArray();
return rows.map((r) => ({
seq: r.seq,
evt: normalizeBlob(r.evt) || new Uint8Array(0),
}));
},
async getLatestSeq() {
// Max across both sequenced tables: a crash between putCommit and
// putEvent leaves a commit whose seq the events table never saw, and
// counting only events would hand that seq out again forever.
const rows = sql
.exec(
'SELECT MAX(s) as seq FROM (SELECT MAX(seq) as s FROM seq_events UNION ALL SELECT MAX(seq) as s FROM commits)',
)
.toArray();
return /** @type {number|null} */ (rows[0]?.seq) ?? 0;
},
async getBlob(cid) {
const row = safeOne(
sql.exec(
'SELECT mime_type, size, created_at FROM blobs WHERE cid = ?',
cid,
),
);
return row
? { mimeType: row.mime_type, size: row.size, createdAt: row.created_at }
: null;
},
async putBlob(cid, mimeType, size) {
sql.exec(
'INSERT OR REPLACE INTO blobs (cid, mime_type, size, created_at) VALUES (?, ?, ?, ?)',
cid,
mimeType,
size,
Date.now(),
);
},
async deleteBlob(cid) {
sql.exec('DELETE FROM blobs WHERE cid = ?', cid);
},
async listBlobs(cursor, limit) {
const rows = cursor
? sql
.exec(
'SELECT cid FROM blobs WHERE cid > ? ORDER BY cid LIMIT ?',
cursor,
limit,
)
.toArray()
: sql
.exec('SELECT cid FROM blobs ORDER BY cid LIMIT ?', limit)
.toArray();
return {
cids: rows.map((r) => r.cid),
cursor: rows.length === limit ? rows[rows.length - 1].cid : null,
};
},
async listBlobDetails(cursor, limit, options = {}) {
const query = buildBlobListQuery(cursor, limit, options);
const rows = sql.exec(query.sql, ...query.params).toArray();
const blobs = rows.map((r) => ({
cid: /** @type {string} */ (r.cid),
mimeType: /** @type {string} */ (r.mime_type),
size: /** @type {number} */ (r.size),
createdAt: /** @type {number} */ (r.created_at),
time: /** @type {number} */ (r.time_ms),
timeFromRecord: r.time_known === 1,
records: r.uris ? String(r.uris).split(',') : [],
}));
return {
blobs,
cursor:
blobs.length === limit
? blobListCursor(blobs[blobs.length - 1], options)
: null,
};
},
async getBlobFacets() {
/** @param {{count?: unknown, bytes?: unknown}|null} row */
const bucket = (row) => ({
count: /** @type {number} */ (row?.count ?? 0),
bytes: /** @type {number} */ (row?.bytes ?? 0),
});
const kinds = sql
.exec(BLOB_FACETS_KINDS_SQL)
.toArray()
.map((r) => ({
kind: /** @type {import('@pdsjs/core/ports').BlobKind} */ (r.kind),
...bucket(r),
}));
const apps = sql
.exec(BLOB_FACETS_APPS_SQL)
.toArray()
.map((r) => ({
authority: /** @type {string} */ (r.authority),
...bucket(r),
}));
return {
total: bucket(safeOne(sql.exec(BLOB_FACETS_TOTAL_SQL))),
unused: bucket(safeOne(sql.exec(BLOB_FACETS_UNUSED_SQL))),
kinds,
apps,
};
},
async getOrphanedBlobs(olderThan) {
const rows = sql
.exec(
'SELECT b.cid FROM blobs b LEFT JOIN record_blobs rb ON b.cid = rb.blob_cid WHERE rb.blob_cid IS NULL AND b.created_at < ?',
olderThan,
)
.toArray();
return rows.map((r) => r.cid);
},
async linkBlobToRecord(blobCid, recordUri, recordTime = null) {
sql.exec(
'INSERT OR REPLACE INTO record_blobs (blob_cid, record_uri, record_time) VALUES (?, ?, ?)',
blobCid,
recordUri,
recordTime,
);
},
async unlinkBlobsFromRecord(recordUri) {
sql.exec('DELETE FROM record_blobs WHERE record_uri = ?', recordUri);
},
async listBlobRecordUris(blobCid) {
const rows = sql
.exec('SELECT record_uri FROM record_blobs WHERE blob_cid = ?', blobCid)
.toArray();
return rows.map((r) => String(r.record_uri));
},
async getDid() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'did'"),
);
if (!row) return null;
if (typeof row.value === 'string') return row.value;
if (row.value instanceof ArrayBuffer)
return new TextDecoder().decode(row.value);
return String(row.value);
},
async setDid(did) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('did', ?)",
did,
);
},
async getPrivateKey() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'privateKey'"),
);
return row ? normalizeBlob(row.value) : null;
},
async setPrivateKey(key) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('privateKey', ?)",
key,
);
},
async getKemKey() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'kemKey'"),
);
return row ? normalizeBlob(row.value) : null;
},
async setKemKey(key) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('kemKey', ?)",
key,
);
},
async getHandle() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'handle'"),
);
if (!row) return null;
if (typeof row.value === 'string') return row.value;
if (row.value instanceof ArrayBuffer)
return new TextDecoder().decode(row.value);
return String(row.value);
},
async getEmail() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'email'"),
);
if (!row) return null;
const raw =
row.value instanceof ArrayBuffer
? new TextDecoder().decode(row.value)
: String(row.value);
try {
return JSON.parse(raw);
} catch {
return null;
}
},
async setEmail(state) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('email', ?)",
JSON.stringify(state),
);
},
async getAccountPassword() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'accountPassword'"),
);
if (!row) return null;
const raw =
row.value instanceof ArrayBuffer
? new TextDecoder().decode(row.value)
: String(row.value);
try {
return JSON.parse(raw);
} catch {
return null;
}
},
async setAccountPassword(state) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('accountPassword', ?)",
JSON.stringify(state),
);
},
async getPlcToken() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'plcToken'"),
);
if (!row) return null;
const raw =
row.value instanceof ArrayBuffer
? new TextDecoder().decode(row.value)
: String(row.value);
try {
return JSON.parse(raw);
} catch {
return null;
}
},
async setPlcToken(token) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('plcToken', ?)",
JSON.stringify(token),
);
},
async getAccountStatus() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'accountStatus'"),
);
if (!row) return null;
if (typeof row.value === 'string') return row.value;
if (row.value instanceof ArrayBuffer)
return new TextDecoder().decode(row.value);
return String(row.value);
},
async setAccountStatus(status) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('accountStatus', ?)",
status,
);
},
async setHandle(handle) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('handle', ?)",
handle,
);
},
// Preferences storage
async getPreferences() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'preferences'"),
);
return row ? JSON.parse(row.value) : [];
},
async setPreferences(preferences) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('preferences', ?)",
JSON.stringify(preferences),
);
},
// Backup settings and run history, one JSON document
async getBackupState() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'backupState'"),
);
if (!row) return null;
const raw =
row.value instanceof ArrayBuffer
? new TextDecoder().decode(row.value)
: String(row.value);
try {
return JSON.parse(raw);
} catch {
return null;
}
},
async setBackupState(state) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('backupState', ?)",
JSON.stringify(state),
);
},
// The account app's own preferences, one JSON document
async getAccountUiState() {
const row = safeOne(
sql.exec("SELECT value FROM metadata WHERE key = 'accountUiState'"),
);
if (!row) return null;
const raw =
row.value instanceof ArrayBuffer
? new TextDecoder().decode(row.value)
: String(row.value);
try {
return JSON.parse(raw);
} catch {
return null;
}
},
async setAccountUiState(state) {
sql.exec(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('accountUiState', ?)",
JSON.stringify(state),
);
},
};
}
// ============================================================================
// R2 Blob Adapter
// ============================================================================
// Every multipart part must be this size except the last: R2 requires the
// parts of one upload to be uniform. R2's minimum is 5 MiB.
const R2_PART_SIZE = 8 * 1024 * 1024;
// R2 accepts about one write per second to one key. A faster write fails with
// error 10058. These delays let the earlier write finish.
const R2_SAME_KEY_RETRY_MS = [250, 500, 1000, 2000];
/**
* Run an R2 write, again after a delay while R2 refuses it with 10058.
* @template T
* @param {() => Promise<T>} write
* @returns {Promise<T>}
*/
async function retrySameKeyLimit(write) {
for (const delay of R2_SAME_KEY_RETRY_MS) {
try {
return await write();
} catch (err) {
if (!String(/** @type {Error} */ (err)?.message).includes('(10058)')) {
throw err;
}
await new Promise((resolve) => setTimeout(resolve, delay));
}
}
return write();
}
/**
* @param {Uint8Array[]} chunks
* @param {number} size
* @returns {Uint8Array}
*/
function joinChunks(chunks, size) {
const out = new Uint8Array(size);
let at = 0;
for (const chunk of chunks) {
out.set(chunk, at);
at += chunk.length;
}
return out;
}
/**
* The runtime's native SHA-256 behind the core Hasher interface.
* @returns {import('@pdsjs/core/ports').Hasher}
*/
function createDigestStreamHasher() {
// DigestStream hangs off the Workers crypto global. The Node Crypto type
// this workspace also loads wins the global merge and lacks it, hence the
// cast.
const workersCrypto = /** @type {{DigestStream: typeof DigestStream}} */ (
/** @type {unknown} */ (crypto)
);
const stream = new workersCrypto.DigestStream('SHA-256');
const writer = stream.getWriter();
return {
/** @param {Uint8Array} chunk */
update(chunk) {
return writer.write(chunk);
},
async digest() {
await writer.close();
return new Uint8Array(await stream.digest);
},
};
}
/**
* Create R2 blob adapter
* @param {R2Bucket} bucket - R2 bucket binding
* @returns {BlobPort}
*/
export function createBlobs(bucket) {
// A blob's key is its CID, so clients that upload the same bytes write the
// same key. R2 refuses concurrent writes to one key. One Durable Object
// serves the account, so this map sees every write to its keys.
/** @type {Map<string, Promise<void>>} */
const keyQueues = new Map();
/**
* Run `write` after the writes to `key` queued before it.
* @param {string} key
* @param {() => Promise<void>} write
* @returns {Promise<void>}
*/
function onKey(key, write) {
const run = (keyQueues.get(key) ?? Promise.resolve()).then(write);
const settled = run.then(
() => {},
() => {},
);
keyQueues.set(key, settled);
settled.then(() => {
if (keyQueues.get(key) === settled) keyQueues.delete(key);
});
return run;
}
/**
* Put the blob at `key` unless R2 holds it. The same CID means the same
* bytes, so a second put changes nothing.
* @param {string} key
* @param {() => Promise<unknown>} put
*/
function putOnce(key, put) {
return onKey(key, async () => {
if (await bucket.head(key)) return;
await retrySameKeyLimit(put);
});
}
return {
/**
* @param {string} did
* @param {string} cid
*/
async get(did, cid) {
const obj = await bucket.get(`${did}/${cid}`);
if (!obj) return null;
const data = new Uint8Array(await obj.arrayBuffer());
const mimeType =
obj.customMetadata?.mime_type || 'application/octet-stream';
return { data, mimeType };
},
/**
* @param {string} did
* @param {string} cid
* @param {Uint8Array} data
* @param {string} mimeType
*/
async put(did, cid, data, mimeType) {
await putOnce(`${did}/${cid}`, () =>
bucket.put(`${did}/${cid}`, data, {
customMetadata: { mime_type: mimeType },
}),
);
},
/**
* @param {string} did
* @param {string} cid
*/
async delete(did, cid) {
const key = `${did}/${cid}`;
await onKey(key, () => retrySameKeyLimit(() => bucket.delete(key)));
},
/**
* @param {string} did
* @param {string} cid
* @param {{start: number, end: number}} [range] - Inclusive byte positions
*/
async getStream(did, cid, range) {
const obj = /** @type {R2ObjectBody|null} */ (
await bucket.get(
`${did}/${cid}`,
range
? {
range: {
offset: range.start,
length: range.end - range.start + 1,
},
}
: undefined,
)
);
if (!obj) return null;
return {
body: obj.body,
mimeType: obj.customMetadata?.mime_type || 'application/octet-stream',
};
},
/**
* Bytes stage under a temporary key and land under the CID at commit. A
* CID contains no slash, so `tmp/` keys are outside the space `get`
* addresses. The R2 binding has no server-side copy, so commit reads the
* staged object back and writes it to the final key as a stream; one part
* buffer is the most this holds in memory. A blob within one part skips
* the staging entirely and lands with a single put.
* @param {string} did
*/
async createWriter(did) {
const tempKey = `${did}/tmp/${crypto.randomUUID()}`;
/** @type {R2MultipartUpload|null} */
let upload = null;
/** @type {R2UploadedPart[]} */
const parts = [];
/** @type {Uint8Array[]} */
let pending = [];
let pendingSize = 0;
async function flushPart() {
if (!upload) upload = await bucket.createMultipartUpload(tempKey);
const buffered = joinChunks(pending, pendingSize);
const body = buffered.subarray(0, R2_PART_SIZE);
const rest = buffered.subarray(R2_PART_SIZE);
parts.push(await upload.uploadPart(parts.length + 1, body));
pending = rest.length ? [rest] : [];
pendingSize = rest.length;
}
return {
/** @param {Uint8Array} chunk */
async write(chunk) {
pending.push(chunk);
pendingSize += chunk.length;
while (pendingSize >= R2_PART_SIZE) await flushPart();
},
/**
* @param {string} cid
* @param {string} mimeType
*/
async commit(cid, mimeType) {
const finalKey = `${did}/${cid}`;
const customMetadata = { mime_type: mimeType };
if (!upload) {
const data = joinChunks(pending, pendingSize);
await putOnce(finalKey, () =>
bucket.put(finalKey, data, { customMetadata }),
);
return;
}
if (pendingSize > 0) {
parts.push(
await upload.uploadPart(
parts.length + 1,
joinChunks(pending, pendingSize),
),
);
pending = [];
pendingSize = 0;
}
await upload.complete(parts);
// A retry reads the staged object again, since a stream is read once.
await putOnce(finalKey, async () => {
const staged = await bucket.get(tempKey);
if (!staged) throw new Error('The staged blob is gone.');
const fixed = new FixedLengthStream(staged.size);
const piped = staged.body.pipeTo(fixed.writable);
await Promise.all([
bucket.put(finalKey, fixed.readable, { customMetadata }),
piped,
]);
});
await bucket.delete(tempKey);
},
async abort() {
if (upload) await upload.abort();
pending = [];
pendingSize = 0;
},
};
},
};
}
/**
* Backups in a second R2 bucket — separate from the live blob bucket, which
* is the point of a backup. Core deletes anything under `snapshots/` and
* `blobs/` that its retention policy no longer wants, so give it a bucket of
* its own.
* @param {R2Bucket} bucket - R2 bucket binding
* @returns {import('@pdsjs/core/ports').BackupTargetPort}
*/
function createBackupTarget(bucket) {
return {
/**
* @param {string} key
* @param {Uint8Array} data
* @param {string} [contentType]
*/
async put(key, data, contentType) {
await bucket.put(
key,
data,
contentType ? { httpMetadata: { contentType } } : undefined,
);
},
/** @param {string} key */
async get(key) {
const obj = await bucket.get(key);
return obj ? new Uint8Array(await obj.arrayBuffer()) : null;
},
/** @param {string} prefix */
async list(prefix) {
/** @type {string[]} */
const keys = [];
/** @type {string|undefined} */
let cursor;
do {
const result = await bucket.list({ prefix, cursor });
for (const object of result.objects) keys.push(object.key);
cursor = result.truncated ? result.cursor : undefined;
} while (cursor);
return keys;
},
/** @param {string} key */
async delete(key) {
await bucket.delete(key);
},
};
}
// ============================================================================
// WebSocket Adapter
// ============================================================================
/**
* Create WebSocket port for Cloudflare Workers with Durable Object hibernation support
* @param {DurableObjectState} state - Durable Object state for hibernation API
* @returns {import('@pdsjs/core/ports').WebSocketPort}
*/
export function createWebSocket(state) {
return {
isUpgrade(request) {
return request.headers.get('Upgrade') === 'websocket';
},
upgrade(_request, onConnect) {
const pair = new WebSocketPair();
const [client, server] = Object.values(pair);
// Use hibernation API - keeps WebSocket alive across DO hibernation
state.acceptWebSocket(server);
// Call onConnect with sender interface
onConnect({
send(data) {
try {
server.send(data);
} catch {
// Client may have disconnected
}
},
close() {
try {
server.close();
} catch {
// Already closed
}
},
});
return new Response(null, { status: 101, webSocket: client });
},
broadcast(data) {
// Use hibernation API to get all connected WebSockets
for (const ws of state.getWebSockets()) {
try {
ws.send(data);
} catch {
// The subscriber is gone. Close rather than swallow: an unclosed
// socket stays in getWebSockets() and keeps absorbing the firehose,
// so the server holds a connection it can no longer deliver on
// while the subscriber, never told to go away, never reconnects.
try {
ws.close(1011, 'send failed');
} catch {
// Already closed.
}
}
}
},
};
}
// ============================================================================
// Durable Object and Worker Entry Point
// ============================================================================
/**
* The Email Service `send_email` binding, narrowed to the fields this adapter
* uses. Cloudflare's own `SendEmail` type is the legacy forwarding envelope and
* does not carry the outbound payload; the runtime takes `from` as a plain
* address string, not an object.
* @typedef {Object} CloudflareEmailer
* @property {(message: {to: string, from: string, subject: string, text: string, html?: string}) => Promise<void>} send
*/
/**
* Adapt the Email Service binding to core's EmailPort: core sends `to`,
* `subject`, `text` and `html` but leaves the sender up to the platform, and
* Cloudflare requires one.
* @param {CloudflareEmailer} binding - The `EMAIL` binding from the worker's env
* @param {string} from - The From address; its domain must be onboarded for Email Sending
* @returns {import('@pdsjs/core/ports').EmailPort}
*/
export function createEmailer(binding, from) {
return {
async send({ to, subject, text, html }) {
await binding.send({ to, from, subject, text, html });
},
};
}
/**
* @typedef {Object} Env
* @property {R2Bucket} [BLOBS] - R2 bucket for blob storage
* @property {R2Bucket} [BLOB_BUCKET] - Alternative R2 bucket binding
* @property {R2Bucket} [BACKUPS] - R2 bucket scheduled backups land in; absent, the backups page reports backups as unavailable
* @property {string} [BACKUP_S3_ENDPOINT] - S3-compatible endpoint for off-Cloudflare backups (Bunny, B2, S3, MinIO); with the four other BACKUP_S3_* values set, this wins over the BACKUPS binding
* @property {string} [BACKUP_S3_BUCKET] - Bucket (or storage zone) name
* @property {string} [BACKUP_S3_REGION] - Signing region; some providers accept anything here
* @property {string} [BACKUP_S3_ACCESS_KEY] - Secret: S3 access key id
* @property {string} [BACKUP_S3_SECRET_KEY] - Secret: S3 secret access key
* @property {string} JWT_SECRET - Secret for JWT signing
* @property {string} [HOSTNAME] - PDS hostname
* @property {string} [APPVIEW_URL] - AppView URL for proxying
* @property {string} [APPVIEW_DID] - AppView DID
* @property {string} [RELAY_URL] - Relay URL for firehose notifications
* @property {string} [PLC_URL] - PLC directory URL for identity operations
* @property {string} [PDS_PASSWORD] - Password for createSession
* @property {string} [PDS_UPDATE_URL] - Where a person updates this deployment; the account page links there when a newer release is published
* @property {string} [PDS_SCRIPT_NAME] - The worker script name this deployment runs as; the update link carries it so the update page can find the worker behind a custom domain
* @property {string} [PDS_EXPERIMENTAL_CF_API_PROXY] - "true" forwards /cf-api/ paths to api.cloudflare.com for the site-hosted setup wizard
* @property {string} [PDS_BLOB_UPLOAD_LIMIT] - Max blob upload size in bytes (default 50 MiB; a Worker accepts a request body of 100 MB on the Free and Pro plans)
* @property {string} [PDS_ENABLE_SPACES] - "true" enables permissioned data
* @property {string} [PDS_EXPERIMENTAL_GIT_HTTP] - "true" serves read-only git smart HTTP under /git/
* @property {string} [PDS_EXPERIMENTAL_OCI] - "true" serves an OCI image registry under /v2/; pulls are anonymous, pushes take Basic credentials createSession accepts
* @property {string} [PDS_EXPERIMENTAL_NPM] - "true" serves an npm registry under /npm/; installs are anonymous, publishes take credentials createSession accepts
* @property {string} [PDS_EXPERIMENTAL_GIT_BROWSE] - "false" turns off file browsing of hosted git repositories on the account page (on by default)
* @property {string} [PDS_EXPERIMENTAL_SITE_DOMAIN] - Apex domain static sites serve under; presence enables the sites feature. Site hostnames must also be routed to this Worker (apex route plus a wildcard zone route).
* @property {string} [PDS_EXPERIMENTAL_SITE_APEX_NAME] - Site name served at the apex (default "home")
* @property {string} [PDS_EXPERIMENTAL_QUERIES] - "true" serves dev.pdsjs.query.run over this repo's dev.pdsjs.query.def records
* @property {string} [PDS_EXPERIMENTAL_DRIVE] - "true" backs the account page's Drive section over dev.pdsjs.drive.file records
* @property {string} [PDS_EMAIL_FROM] - The From address account mail is sent as; its domain must be onboarded for Email Sending. Absent, no-reply@HOSTNAME is used. Neither set, mail cannot be sent and the account page reports email as unavailable.
* @property {CloudflareEmailer} [EMAIL] - Email Service send_email binding. Present only when the binding is configured; without it mail cannot be sent.
* @property {DurableObjectNamespace} PDS - Durable Object namespace
*/
/**
* Durable Object class for per-user PDS instances
*/
export class PDSDurableObject {
/**
* @param {DurableObjectState} state
* @param {Env} env
*/
constructor(state, env) {
this.state = state;
this.env = env;
// Initialize actor storage adapter from Durable Object SQLite
const actorStorage = createActorStorage(state.storage.sql);
// Shared account state — sessions, OAuth tokens, app passwords, passkeys —
// lives in the Durable Object's own SQLite. It is strongly consistent, so a
// read always sees its own writes.
const sharedStorage = createSharedStorageSql(state.storage.sql);
// Initialize blob adapter (R2 bucket)
const blobs = createBlobs(
/** @type {R2Bucket} */ (env.BLOBS || env.BLOB_BUCKET),
);
// Backups: an S3-compatible bucket elsewhere beats an R2 bucket here —
// a copy outside Cloudflare survives what a same-account copy cannot.
// With neither configured the feature reports itself unavailable.
const backupTarget =
env.BACKUP_S3_ENDPOINT &&
env.BACKUP_S3_BUCKET &&
env.BACKUP_S3_ACCESS_KEY &&
env.BACKUP_S3_SECRET_KEY
? createS3FetchBackupTarget({
endpoint: env.BACKUP_S3_ENDPOINT,
bucket: env.BACKUP_S3_BUCKET,
region: env.BACKUP_S3_REGION || 'us-east-1',
accessKeyId: env.BACKUP_S3_ACCESS_KEY,
secretAccessKey: env.BACKUP_S3_SECRET_KEY,
})
: env.BACKUPS
? createBackupTarget(env.BACKUPS)
: undefined;
// Initialize WebSocket adapter with hibernation support
const webSocket = createWebSocket(state);
// Initialize lexicon resolver for record validation
// Wrap fetch to preserve `this` binding in Cloudflare Workers
const lexiconResolver = new LexiconResolver({
fetch: (url, init) => fetch(url, init),
});
// Signature verification, shared by createAccount (migration service
// tokens) and spaces (credentials). P-256 is built into core; secp256k1 —
// the curve the reference PDS gives every account — is supplied here via
// @noble/curves. The resolver returns full DID documents; fetch is wrapped
// to preserve its `this` binding, as elsewhere in this file.
const verifier = createVerifier({ secp256k1: secp256k1Verify });
// Built in the Durable Object constructor, so one cache serves every
// request this object handles. Every space request verifies a credential,
// which reads its issuer's key.
const didResolver = createCachedDidResolver({
plcUrl: env.PLC_URL,
fetch: (url, init) => fetch(url, init),
});
// 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 the spaces extension mounted below. Absent the
// flag the space endpoints do not exist.
/** @type {import('@pdsjs/core/ports').SpaceStoragePort|undefined} */
let spaceStorageRef;
if (env.PDS_ENABLE_SPACES === 'true') {
// state.storage as well as .sql: commitSpaceWrite needs transactionSync,
// which Durable Object SQL exposes on the storage object rather than
// through exec.
spaceStorageRef = createSpaceStorage(state.storage.sql, state.storage);
}
// Account mail. Both a binding and a sender are needed: Cloudflare requires
// a From address whose domain is onboarded for Email Sending. The binding
// alone can store an address but never prove anyone can read it, so mail is
// only enabled when the sender is resolvable.
/** @type {import('@pdsjs/core/ports').EmailPort|undefined} */
const emailer =
env.EMAIL && (env.PDS_EMAIL_FROM || env.HOSTNAME)
? createEmailer(
env.EMAIL,
env.PDS_EMAIL_FROM || `no-reply@${env.HOSTNAME}`,
)
: undefined;
// Create PDS instance with port injection
this.pds = new PersonalDataServer({
// One alarm serves the hourly maintenance and any earlier extension work,
// so a wakeup only ever moves the alarm closer.
scheduleWakeup: async (at) => {
const current = await state.storage.getAlarm();
if (!current || at < current) await state.storage.setAlarm(at);
},
waitUntil: (work) => state.waitUntil(work),
actorStorage,
sharedStorage,
blobs,
webSocket,
jwtSecret: env.JWT_SECRET,
hostname: env.HOSTNAME,
appviewUrl: env.APPVIEW_URL,
appviewDid: env.APPVIEW_DID,
relayUrl: env.RELAY_URL,
plcUrl: env.PLC_URL,
password: env.PDS_PASSWORD,
emailer,
updateUrl: env.PDS_UPDATE_URL,
scriptName: env.PDS_SCRIPT_NAME,
blobUploadLimit: env.PDS_BLOB_UPLOAD_LIMIT
? Number(env.PDS_BLOB_UPLOAD_LIMIT)
: undefined,
hasher: createDigestStreamHasher,
lexiconResolver,
spaceStorage: spaceStorageRef,
accountApp,
verifier,
didResolver,
k256Signer,
backupTarget,
// Feature packages on the extension seam. Static imports, the same as
// spaces above.
extensions: [
...(env.PDS_ENABLE_SPACES === 'true' ? [createSpacesExtension()] : []),
createSitesExtension({
domain: env.PDS_EXPERIMENTAL_SITE_DOMAIN,
apexName: env.PDS_EXPERIMENTAL_SITE_APEX_NAME,
queries: env.PDS_EXPERIMENTAL_QUERIES === 'true',
}),
createGitExtension({
http: env.PDS_EXPERIMENTAL_GIT_HTTP === 'true',
browse: env.PDS_EXPERIMENTAL_GIT_BROWSE !== 'false',
}),
...(env.PDS_EXPERIMENTAL_OCI === 'true' ? [createOciExtension()] : []),
...(env.PDS_EXPERIMENTAL_NPM === 'true' ? [createNpmExtension()] : []),
...(env.PDS_EXPERIMENTAL_DRIVE === 'true'
? [createDriveExtension()]
: []),
],
});
// Arm the alarm if not already set. Maintenance runs hourly — the same
// cadence as the Node and Deno cleanup intervals — and each run decides for
// itself whether cleanup or a scheduled backup is due. Extension work can
// bring the alarm earlier.
state.blockConcurrencyWhile(async () => {
const currentAlarm = await state.storage.getAlarm();
if (!currentAlarm) {
await state.storage.setAlarm(Date.now() + 60 * 60 * 1000);
}
});
// An account that predates KEM keys gets one as soon as this object
// starts, as the Node and Deno servers do at startup, rather than at the
// next hourly maintenance; until then no syncer can read its space repos.
// Once the key exists this is a single storage read.
void this.pds.maybeBackfillKemKey();
}
/**
* Handle incoming request
* @param {Request} request
* @returns {Promise<Response>}
*/
async fetch(request) {
return this.pds.fetch(request);
}
/**
* The alarm: hourly maintenance when it is due (blob cleanup, then a backup
* if one is due), and extension work on every wakeup. Mirrors the intervals
* the Node and Deno servers run.
*/
async alarm() {
const now = Date.now();
let maintenanceAt =
/** @type {number|undefined} */ (
await this.state.storage.get(MAINTENANCE_AT_KEY)
) ?? 0;
/** @type {number|null} */
let next = null;
try {
if (now >= maintenanceAt) {
// Advanced before the run, so a failing run waits an hour like a
// passing one rather than refiring at once.
maintenanceAt = now + MAINTENANCE_INTERVAL_MS;
await this.state.storage.put(MAINTENANCE_AT_KEY, maintenanceAt);
await this.pds.runBlobCleanup();
await this.pds.maybeAnnounceIdentity();
await this.pds.maybeBackfillKemKey();
await this.pds.maybeRunScheduledBackup();
}
} finally {
try {
next = await this.pds.runExtensionTicks();
} finally {
// Re-arm no matter what: a failed run must not end the schedule.
await this.state.storage.setAlarm(
next === null ? maintenanceAt : Math.min(maintenanceAt, next),
);
}
}
}
/**
* Handle WebSocket messages (hibernation API callback)
* @param {WebSocket} ws
* @param {string | ArrayBuffer} message
*/
async webSocketMessage(ws, message) {
// Handle ping/pong keepalive
if (message === 'ping') {
ws.send('pong');
}
}
/**
* Handle WebSocket close (hibernation API callback)
* @param {WebSocket} ws
* @param {number} code
* @param {string} reason
* @param {boolean} _wasClean
*/
async webSocketClose(ws, code, reason, _wasClean) {
// Finish the closing handshake. Until this runs the socket stays in
// getWebSockets(), and a firehose broadcast keeps writing to a subscriber
// that has already gone. (Runtimes on compatibility_date 2026-04-07 or
// later reply automatically; this server's is older.)
try {
ws.close(code, reason);
} catch {
// Already closed.
}
}
/**
* Handle WebSocket errors (hibernation API callback)
* @param {WebSocket} ws
* @param {unknown} _error
*/
async webSocketError(ws, _error) {
// A socket that errored will never deliver again; drop it so it stops
// collecting broadcasts.
try {
ws.close(1011, 'socket error');
} catch {
// Already closed.
}
}
}
// ============================================================================
// Shared storage adapter using Durable Object SQLite
// ============================================================================
/** @typedef {string | number | boolean | null | undefined} SqlBind */
/** @typedef {{did: string, handle: string | null, created_at: number}} ActorRow */
/** @typedef {{jti: string, did: string, label: string, scope: string, user_agent: string | null, created_at: string, refreshed_at: string | null, expires_at: number | null, next_jti: string | null}} SessionRow */
/** @typedef {{credential_id: string, did: string, name: string, public_key: string, algorithm: number, sign_count: number, created_at: string, last_used_at: string | null}} PasskeyRow */
/** @typedef {{name: string, privileged: number, created_at: string}} AppPasswordRow */
/** @typedef {{token_id: string, data: string}} OAuthTokenListRow */
/** @param {SessionRow} row @returns {import('@pdsjs/core/ports').SessionRecord} */
function toSessionRecord(row) {
return {
jti: row.jti,
did: row.did,
label: row.label,
scope: row.scope,
userAgent: row.user_agent,
createdAt: row.created_at,
refreshedAt: row.refreshed_at,
expiresAt: row.expires_at ?? undefined,
nextJti: row.next_jti ?? null,
};
}
/** @param {PasskeyRow} row @returns {import('@pdsjs/core/ports').PasskeyRecord} */
function toPasskeyRecord(row) {
return {
credentialId: row.credential_id,
did: row.did,
name: row.name,
publicKey: row.public_key,
algorithm: row.algorithm,
signCount: row.sign_count,
createdAt: row.created_at,
lastUsedAt: row.last_used_at,
};
}
/** @param {AppPasswordRow} row @returns {import('@pdsjs/core/ports').AppPasswordRecord} */
function toAppPassword(row) {
return {
name: row.name,
privileged: row.privileged === 1,
createdAt: row.created_at,
};
}
/**
* Shared storage on the Durable Object's own SQLite — strongly consistent and
* co-located with the repo, so every account-scoped record (sessions, OAuth,
* app passwords, passkeys) reads back its own writes immediately. The schema
* mirrors the Node adapter's, sharing the same core schema constants.
* @param {D1SqlInterface} sql
* @returns {SharedStoragePort}
*/
export function createSharedStorageSql(sql) {
sql.exec(`
CREATE TABLE IF NOT EXISTS actors (
did TEXT PRIMARY KEY,
handle TEXT,
created_at INTEGER NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS actors_handle_lower_idx ON actors(lower(handle));
CREATE TABLE IF NOT EXISTS oauth_requests (
request_uri TEXT PRIMARY KEY,
data TEXT,
expires_at INTEGER
);
CREATE INDEX IF NOT EXISTS oauth_requests_expires_idx ON oauth_requests(expires_at);
CREATE TABLE IF NOT EXISTS oauth_tokens (
token_id TEXT PRIMARY KEY,
did TEXT NOT NULL,
data TEXT
);
CREATE INDEX IF NOT EXISTS oauth_tokens_did_idx ON oauth_tokens(did);
CREATE TABLE IF NOT EXISTS dpop_jtis (
jti TEXT PRIMARY KEY,
expires_at INTEGER
);
CREATE INDEX IF NOT EXISTS dpop_jtis_expires_idx ON dpop_jtis(expires_at);
`);
sql.exec(APP_PASSWORD_SCHEMA_SQL);
sql.exec(PASSKEY_SCHEMA_SQL);
sql.exec(SESSION_SCHEMA_SQL);
// A session table created before rotations named a successor lacks the
// column, and CREATE TABLE IF NOT EXISTS will not add it. SQLite has no
// ADD COLUMN IF NOT EXISTS, so a re-run just throws.
try {
sql.exec('ALTER TABLE sessions ADD COLUMN next_jti TEXT');
} catch {
// Column already present.
}
// Durable Object SQL rejects `undefined` bindings; map them to null.
/** @param {SqlBind[]} args @returns {(string | number | boolean | null)[]} */
const b = (args) => args.map((a) => (a === undefined ? null : a));
/** @param {string} q @param {...SqlBind} args @returns {unknown} */
const get = (q, ...args) => sql.exec(q, ...b(args)).toArray()[0];
/** @param {string} q @param {...SqlBind} args @returns {unknown[]} */
const all = (q, ...args) => sql.exec(q, ...b(args)).toArray();
/** @param {string} q @param {...SqlBind} args @returns {number} rows written */
const run = (q, ...args) => sql.exec(q, ...b(args)).rowsWritten;
return {
async getActor(did) {
const row = /** @type {ActorRow | undefined} */ (
get('SELECT did, handle, created_at FROM actors WHERE did = ?', did)
);
return row
? { did: row.did, handle: row.handle, createdAt: row.created_at }
: null;
},
async resolveHandle(handle) {
const row = /** @type {{did: string} | undefined} */ (
get('SELECT did FROM actors WHERE lower(handle) = lower(?)', handle)
);
return row ? row.did : null;
},
async putActor(did, handle, createdAt) {
run(
'INSERT OR REPLACE INTO actors (did, handle, created_at) VALUES (?, ?, ?)',
did,
handle,
createdAt,
);
},
async deleteActor(did) {
run('DELETE FROM actors WHERE did = ?', did);
},
async getOAuthRequest(requestUri) {
const row = /** @type {OAuthRequestRow | undefined} */ (
get(
'SELECT data, expires_at FROM oauth_requests WHERE request_uri = ?',
requestUri,
)
);
if (!row) return null;
if (row.expires_at < Date.now()) {
run('DELETE FROM oauth_requests WHERE request_uri = ?', requestUri);
return null;
}
try {
return JSON.parse(row.data);
} catch {
run('DELETE FROM oauth_requests WHERE request_uri = ?', requestUri);
return null;
}
},
async putOAuthRequest(requestUri, data, expiresAt) {
run(
'INSERT OR REPLACE INTO oauth_requests (request_uri, data, expires_at) VALUES (?, ?, ?)',
requestUri,
JSON.stringify(data),
expiresAt,
);
},
async deleteOAuthRequest(requestUri) {
run('DELETE FROM oauth_requests WHERE request_uri = ?', requestUri);
},
async getOAuthToken(tokenId) {
const row = /** @type {OAuthTokenRow | undefined} */ (
get('SELECT data FROM oauth_tokens WHERE token_id = ?', tokenId)
);
if (!row) return null;
try {
return JSON.parse(row.data);
} catch {
run('DELETE FROM oauth_tokens WHERE token_id = ?', tokenId);
return null;
}
},
async putOAuthToken(tokenId, data) {
const tokenData = /** @type {{did: string}} */ (data);
run(
'INSERT OR REPLACE INTO oauth_tokens (token_id, did, data) VALUES (?, ?, ?)',
tokenId,
tokenData.did,
JSON.stringify(data),
);
},
async deleteOAuthToken(tokenId) {
run('DELETE FROM oauth_tokens WHERE token_id = ?', tokenId);
},
async listOAuthTokensByDid(did) {
const rows = /** @type {OAuthTokenListRow[]} */ (
all('SELECT token_id, data FROM oauth_tokens WHERE did = ?', did)
);
return rows
.map((r) => {
try {
return { ...JSON.parse(r.data), tokenId: r.token_id };
} catch {
return null;
}
})
.filter(Boolean);
},
async checkAndStoreDpopJti(jti, expiresAt) {
return (
run(
'INSERT OR IGNORE INTO dpop_jtis (jti, expires_at) VALUES (?, ?)',
jti,
expiresAt,
) > 0
);
},
async cleanupExpiredDpopJtis() {
run('DELETE FROM dpop_jtis WHERE expires_at < ?', Date.now());
},
async putAppPassword(did, name, hash, privileged, createdAt) {
// INSERT OR IGNORE, not REPLACE: reusing a name must not silently
// invalidate the password already issued under it.
return (
run(
'INSERT OR IGNORE INTO app_passwords (did, name, hash, privileged, created_at) VALUES (?, ?, ?, ?, ?)',
did,
name,
hash,
privileged ? 1 : 0,
createdAt,
) > 0
);
},
async getAppPasswordByHash(did, hash) {
const row = /** @type {AppPasswordRow | undefined} */ (
get(
'SELECT name, privileged, created_at FROM app_passwords WHERE did = ? AND hash = ?',
did,
hash,
)
);
return row ? toAppPassword(row) : null;
},
async listAppPasswords(did) {
const rows = /** @type {AppPasswordRow[]} */ (
all(
'SELECT name, privileged, created_at FROM app_passwords WHERE did = ? ORDER BY created_at',
did,
)
);
return rows.map(toAppPassword);
},
async deleteAppPassword(did, name) {
return (
run('DELETE FROM app_passwords WHERE did = ? AND name = ?', did, name) >
0
);
},
async putSession(session) {
run(
'INSERT OR REPLACE INTO sessions (jti, did, label, scope, user_agent, created_at, refreshed_at, expires_at, next_jti) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
session.jti,
session.did,
session.label,
session.scope,
session.userAgent,
session.createdAt,
session.refreshedAt,
session.expiresAt ?? null,
session.nextJti ?? null,
);
},
async getSession(jti) {
const row = /** @type {SessionRow | undefined} */ (
get('SELECT * FROM sessions WHERE jti = ?', jti)
);
return row ? toSessionRecord(row) : null;
},
async listSessions(did) {
const rows = /** @type {SessionRow[]} */ (
all('SELECT * FROM sessions WHERE did = ? ORDER BY created_at', did)
);
return rows.map(toSessionRecord);
},
async deleteSession(jti) {
return run('DELETE FROM sessions WHERE jti = ?', jti) > 0;
},
async spendSession(jti, expiresAt, nextJti) {
// Conditional on the successor, so that of two rotations racing the same
// token only the first takes effect. `sql.exec` runs to completion, so
// nothing interleaves between the test and the write.
return (
run(
'UPDATE sessions SET expires_at = ?, next_jti = ? WHERE jti = ? AND (next_jti IS NULL OR next_jti = ?)',
expiresAt,
nextJti,
jti,
nextJti,
) > 0
);
},
async putPasskey(record) {
return (
run(
'INSERT OR IGNORE INTO passkeys (credential_id, did, name, public_key, algorithm, sign_count, created_at, last_used_at) VALUES (?, ?, ?, ?, ?, 0, ?, NULL)',
record.credentialId,
record.did,
record.name,
record.publicKey,
record.algorithm,
record.createdAt,
) > 0
);
},
async getPasskey(credentialId) {
const row = /** @type {PasskeyRow | undefined} */ (
get('SELECT * FROM passkeys WHERE credential_id = ?', credentialId)
);
return row ? toPasskeyRecord(row) : null;
},
async listPasskeys(did) {
const rows = /** @type {PasskeyRow[]} */ (
all('SELECT * FROM passkeys WHERE did = ? ORDER BY created_at', did)
);
return rows.map(toPasskeyRecord);
},
async deletePasskey(did, credentialId) {
return (
run(
'DELETE FROM passkeys WHERE did = ? AND credential_id = ?',
did,
credentialId,
) > 0
);
},
async touchPasskey(credentialId, signCount, lastUsedAt) {
run(
'UPDATE passkeys SET sign_count = ?, last_used_at = ? WHERE credential_id = ?',
signCount,
lastUsedAt,
credentialId,
);
},
};
}
/**
* Whether this request is for a site hostname: under the configured site
* domain and not the PDS's own hostname. The PDS hostname is excluded even
* when it sits under the site domain, so PDS traffic never takes the site
* path. A `wrangler dev` localhost never matches.
* @param {Request} request
* @param {Env} env
* @returns {boolean}
*/
export function siteHost(request, env) {
if (!env.PDS_EXPERIMENTAL_SITE_DOMAIN) return false;
const host = new URL(request.url).hostname.toLowerCase();
const domain = env.PDS_EXPERIMENTAL_SITE_DOMAIN.toLowerCase();
if (host !== domain && !host.endsWith(`.${domain}`)) return false;
return host !== (env.HOSTNAME || '').split(':')[0].toLowerCase();
}
/**
* Whether this request may be served from and stored in the shared edge
* cache: a GET for a site hostname, never the PDS's own. The Durable Object
* lives in one region while site visitors do not, so this is what keeps
* repeat site traffic off the cross-region round trip. PDS responses carry
* sessions and DPoP nonces and must never enter a shared cache.
* @param {Request} request
* @param {Env} env
* @returns {boolean}
*/
export function siteCacheable(request, env) {
if (request.method !== 'GET' || !siteHost(request, env)) return false;
// The registries answer on every hostname, and their reads are mutable:
// a packument gains versions and a manifest tag moves. A cached copy
// would serve a package that was deleted or miss one just published.
const pathname = new URL(request.url).pathname;
return !pathname.startsWith('/v2/') && !pathname.startsWith('/npm/');
}
/**
* Whether a response may enter the shared edge cache: only complete
* successes, and only ones the origin marked shareable. The site handler's
* 404s say no-store; blob errors and partials fail the status check.
* @param {Response} response
* @returns {boolean}
*/
export function responseStorable(response) {
if (response.status !== 200 || response.headers.has('Set-Cookie')) {
return false;
}
return !/no-store|private/i.test(response.headers.get('Cache-Control') || '');
}
/**
* Whether a request on the PDS's own hostname may be served from the shared
* edge cache: a public blob GET, nothing else. Blob bytes are
* content-addressed and immutable, and the Durable Object lives in one
* region while the video CDN and gallery visitors do not.
* @param {Request} request
* @returns {boolean}
*/
export function blobCacheable(request) {
return (
request.method === 'GET' &&
new URL(request.url).pathname === '/xrpc/com.atproto.sync.getBlob'
);
}
/**
* getBlob through the shared edge cache. Only complete 200s are stored, so
* the space-privacy filter in the Durable Object decides what ever enters
* the cache. The cache answers ranged requests out of a stored 200; a
* ranged request for a cold blob passes through to the Durable Object, and
* a background fetch stores the whole object so later ranges stay at the
* edge. A blob unpublished later can persist in edge caches until eviction,
* like any CDN copy of once-public immutable content.
* @param {Request} request
* @param {{fetch: (request: Request) => Promise<Response>}} stub
* @param {ExecutionContext} ctx
* @returns {Promise<Response>}
*/
async function serveBlobThroughCache(request, stub, ctx) {
const cache = /** @type {CacheStorage & {default: Cache}} */ (caches).default;
const cached = await cache.match(request);
if (cached) {
const hit = new Response(cached.body, cached);
hit.headers.set('X-Blob-Cache', 'hit');
return hit;
}
const response = await stub.fetch(request);
if (responseStorable(response)) {
ctx.waitUntil(cache.put(request, response.clone()));
} else if (response.status === 206) {
const full = new Request(request.url);
ctx.waitUntil(
(async () => {
const whole = await stub.fetch(full);
if (responseStorable(whole)) {
await cache.put(full, whole);
} else {
await whole.body?.cancel();
}
})(),
);
}
return response;
}
/**
* Worker entry point - single-tenant mode
* All requests route to default DO instance
* TODO: Add multi-tenant routing for hosting providers
*/
export default {
/**
* @param {Request} request
* @param {Env} env
* @param {ExecutionContext} ctx
* @returns {Promise<Response>}
*/
async fetch(request, env, ctx) {
// The setup wizard's Cloudflare API calls, when the proxy is enabled.
// Before site dispatch, so a site-hosted wizard reaches it same-origin.
if (
cfApiProxyEnabled(env) &&
new URL(request.url).pathname.startsWith(CF_API_PROXY_PREFIX)
) {
return handleCfApiProxy(request);
}
const stub = env.PDS.get(env.PDS.idFromName('default'));
if (!siteHost(request, env)) {
if (blobCacheable(request)) {
return serveBlobThroughCache(request, stub, ctx);
}
return stub.fetch(request);
}
// Custom domains answer on both schemes, and a plain-HTTP page marks
// the site "not secure" in browsers. Sites are HTTPS only. The URL's own
// scheme reads https even for plain-HTTP visitors; the edge reports the
// real one in X-Forwarded-Proto.
const url = new URL(request.url);
const proto = request.headers.get('X-Forwarded-Proto');
if (proto ? proto === 'http' : url.protocol === 'http:') {
url.protocol = 'https:';
return Response.redirect(url.href, 301);
}
if (!siteCacheable(request, env)) return stub.fetch(request);
const cache = /** @type {CacheStorage & {default: Cache}} */ (caches)
.default;
const cached = await cache.match(request);
if (cached) {
// cache.match ignores conditional headers, so answer If-None-Match
// here; the ETag is the file's CID and survives redeploys unchanged.
const etag = cached.headers.get('ETag');
const conditional = request.headers.get('If-None-Match');
if (etag && conditional?.split(',').some((t) => t.trim() === etag)) {
const headers = new Headers(cached.headers);
headers.delete('Content-Length');
return new Response(null, { status: 304, headers });
}
const hit = new Response(cached.body, cached);
hit.headers.set('X-Sites-Cache', 'hit');
return hit;
}
const response = await stub.fetch(request);
if (responseStorable(response)) {
ctx.waitUntil(cache.put(request, response.clone()));
}
return response;
},
};