chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
# SharedStoragePort Implementation Plan
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
**Goal:** Split StoragePort into ActorStoragePort (per-DID data) and SharedStoragePort (global data) to prepare for multi-tenant architecture.
**Architecture:** Create a new SharedStoragePort interface for OAuth requests, OAuth tokens, DPoP JTIs, and actor/handle registry. Implement for both Node.js (SQLite) and Cloudflare (KV). Update PDS to accept both ports.
**Tech Stack:** TypeScript/JSDoc, better-sqlite3, Cloudflare KV, Vitest
---
## Task 1: Define SharedStoragePort Interface
**Files:**
- Modify: `packages/core/ports.js`
**Step 1: Add SharedStoragePort typedef**
Add after the existing StoragePort typedef:
```javascript
/**
* @typedef {Object} ActorRecord
* @property {string} did
* @property {string|null} handle
* @property {number} createdAt
*/
/**
* @typedef {Object} SharedStoragePort
*
* // Actor/handle registry
* @property {(did: string) => Promise<ActorRecord|null>} getActor
* @property {(handle: string) => Promise<string|null>} resolveHandle - Returns DID or null
* @property {(did: string, handle: string|null, createdAt: number) => Promise<void>} putActor
* @property {(did: string) => Promise<void>} deleteActor
*
* // OAuth requests
* @property {(requestUri: string) => Promise<Object|null>} getOAuthRequest
* @property {(requestUri: string, data: Object, expiresAt: number) => Promise<void>} putOAuthRequest
* @property {(requestUri: string) => Promise<void>} deleteOAuthRequest
*
* // OAuth tokens
* @property {(tokenId: string) => Promise<Object|null>} getOAuthToken
* @property {(tokenId: string, data: Object) => Promise<void>} putOAuthToken
* @property {(tokenId: string) => Promise<void>} deleteOAuthToken
* @property {(did: string) => Promise<Object[]>} listOAuthTokensByDid
*
* // DPoP jti tracking
* @property {(jti: string, expiresAt: number) => Promise<boolean>} checkAndStoreDpopJti
* @property {() => Promise<void>} cleanupExpiredDpopJtis
*/
```
**Step 2: Rename StoragePort to ActorStoragePort**
Remove OAuth and DPoP methods from the existing StoragePort, rename to ActorStoragePort:
```javascript
/**
* @typedef {Object} ActorStoragePort
* @property {(uri: string) => Promise<{cid: string, value: Uint8Array}|null>} getRecord
* @property {(collection: string, cursor: string|null, limit: number) => Promise<{records: RecordResult[], cursor: string|null}>} listRecords
* @property {() => Promise<Array<{key: string, cid: string}>>} listAllRecords
* @property {(uri: string, cid: string, collection: string, rkey: string, value: Uint8Array) => Promise<void>} putRecord
* @property {(uri: string) => Promise<void>} deleteRecord
* @property {(cid: string) => Promise<Uint8Array|null>} getBlock
* @property {(cid: string, data: Uint8Array) => Promise<void>} putBlock
* @property {() => Promise<{seq: number, cid: string, rev: string}|null>} getLatestCommit
* @property {(seq: number, cid: string, rev: string, prev: string|null) => Promise<void>} putCommit
* @property {(cursor: number|null, limit: number) => Promise<{events: EventResult[], cursor: number|null}>} getEvents
* @property {(seq: number, did: string, commitCid: string, evt: Uint8Array) => Promise<void>} putEvent
* @property {(cid: string) => Promise<{mimeType: string, size: number, createdAt: number}|null>} getBlob
* @property {(cid: string, mimeType: string, size: number) => Promise<void>} putBlob
* @property {(cid: string) => Promise<void>} deleteBlob
* @property {(cursor: string|null, limit: number) => Promise<{cids: string[], cursor: string|null}>} listBlobs
* @property {(olderThan: number) => Promise<string[]>} getOrphanedBlobs
* @property {(blobCid: string, recordUri: string) => Promise<void>} linkBlobToRecord
* @property {(recordUri: string) => Promise<void>} unlinkBlobsFromRecord
* @property {() => Promise<string|null>} getDid
* @property {(did: string) => Promise<void>} setDid
* @property {() => Promise<Uint8Array|null>} getPrivateKey
* @property {(key: Uint8Array) => Promise<void>} setPrivateKey
* @property {() => Promise<string|null>} getHandle
* @property {(handle: string) => Promise<void>} setHandle
* @property {() => Promise<Object[]>} getPreferences
* @property {(preferences: Object[]) => Promise<void>} setPreferences
*/
```
**Step 3: Run typecheck**
Run: `pnpm typecheck`
Expected: PASS (or type errors to fix in later tasks)
**Step 4: Commit**
```bash
git add packages/core/ports.js
git commit -m "refactor: define SharedStoragePort and rename StoragePort to ActorStoragePort"
```
---
## Task 2: Add SharedStorage to better-sqlite3
**Files:**
- Modify: `packages/storage-better-sqlite3/index.js`
**Step 1: Add the shared storage function to index.js**
Add after the existing `createBetterSqliteStorage` function:
```javascript
// Shared storage adapter for Node.js
/** @typedef {{did: string, handle: string|null, created_at: number}} ActorRow */
/** @typedef {{data: string, expires_at: number}} OAuthRequestRow */
/** @typedef {{data: string}} OAuthTokenRow */
/**
* Create shared storage adapter for better-sqlite3 (Node.js)
* @param {import('better-sqlite3').Database} db
* @returns {import('@pdsjs/core/ports').SharedStoragePort}
*/
export function createSharedStorage(db) {
// Initialize schema
db.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);
`);
const stmts = {
getActor: db.prepare('SELECT did, handle, created_at FROM actors WHERE did = ?'),
resolveHandle: db.prepare('SELECT did FROM actors WHERE lower(handle) = lower(?)'),
putActor: db.prepare('INSERT OR REPLACE INTO actors (did, handle, created_at) VALUES (?, ?, ?)'),
deleteActor: db.prepare('DELETE FROM actors WHERE did = ?'),
getOAuthRequest: db.prepare('SELECT data, expires_at FROM oauth_requests WHERE request_uri = ?'),
putOAuthRequest: db.prepare('INSERT OR REPLACE INTO oauth_requests (request_uri, data, expires_at) VALUES (?, ?, ?)'),
deleteOAuthRequest: db.prepare('DELETE FROM oauth_requests WHERE request_uri = ?'),
getOAuthToken: db.prepare('SELECT data FROM oauth_tokens WHERE token_id = ?'),
putOAuthToken: db.prepare('INSERT OR REPLACE INTO oauth_tokens (token_id, did, data) VALUES (?, ?, ?)'),
deleteOAuthToken: db.prepare('DELETE FROM oauth_tokens WHERE token_id = ?'),
listOAuthTokensByDid: db.prepare('SELECT data FROM oauth_tokens WHERE did = ?'),
storeDpopJti: db.prepare('INSERT OR IGNORE INTO dpop_jtis (jti, expires_at) VALUES (?, ?)'),
cleanupExpiredDpopJtis: db.prepare('DELETE FROM dpop_jtis WHERE expires_at < ?'),
};
return {
async getActor(did) {
const row = /** @type {ActorRow|undefined} */ (stmts.getActor.get(did));
return row ? { did: row.did, handle: row.handle, createdAt: row.created_at } : null;
},
async resolveHandle(handle) {
const row = /** @type {{did: string}|undefined} */ (stmts.resolveHandle.get(handle));
return row ? row.did : null;
},
async putActor(did, handle, createdAt) {
stmts.putActor.run(did, handle, createdAt);
},
async deleteActor(did) {
stmts.deleteActor.run(did);
},
async getOAuthRequest(requestUri) {
const row = /** @type {OAuthRequestRow|undefined} */ (stmts.getOAuthRequest.get(requestUri));
if (!row) return null;
if (row.expires_at < Date.now()) {
stmts.deleteOAuthRequest.run(requestUri);
return null;
}
return JSON.parse(row.data);
},
async putOAuthRequest(requestUri, data, expiresAt) {
stmts.putOAuthRequest.run(requestUri, JSON.stringify(data), expiresAt);
},
async deleteOAuthRequest(requestUri) {
stmts.deleteOAuthRequest.run(requestUri);
},
async getOAuthToken(tokenId) {
const row = /** @type {OAuthTokenRow|undefined} */ (stmts.getOAuthToken.get(tokenId));
return row ? JSON.parse(row.data) : null;
},
async putOAuthToken(tokenId, data) {
const tokenData = /** @type {{did: string}} */ (data);
stmts.putOAuthToken.run(tokenId, tokenData.did, JSON.stringify(data));
},
async deleteOAuthToken(tokenId) {
stmts.deleteOAuthToken.run(tokenId);
},
async listOAuthTokensByDid(did) {
const rows = /** @type {OAuthTokenRow[]} */ (stmts.listOAuthTokensByDid.all(did));
return rows.map((r) => JSON.parse(r.data));
},
async checkAndStoreDpopJti(jti, expiresAt) {
const result = stmts.storeDpopJti.run(jti, expiresAt);
return result.changes > 0;
},
async cleanupExpiredDpopJtis() {
stmts.cleanupExpiredDpopJtis.run(Date.now());
},
};
}
```
**Step 2: Run typecheck**
Run: `pnpm typecheck`
Expected: PASS
**Step 3: Commit**
```bash
git add packages/storage-better-sqlite3/index.js
git commit -m "feat: add createSharedStorage to better-sqlite3 adapter"
```
---
## Task 3: Add SharedStorage to Cloudflare adapter
**Files:**
- Modify: `packages/cloudflare/index.js`
**Step 1: Add the Cloudflare shared storage function to index.js**
Add after the existing PdsDurableObject class:
```javascript
// Shared storage adapter using Cloudflare KV
/**
* @typedef {Object} KVNamespace
* @property {(key: string, options?: {type?: string}) => Promise<any>} get
* @property {(key: string, value: string, options?: {expirationTtl?: number, expiration?: number}) => Promise<void>} put
* @property {(key: string) => Promise<void>} delete
* @property {(options?: {prefix?: string}) => Promise<{keys: {name: string}[]}>} list
*/
/**
* Create shared storage adapter for Cloudflare KV
* @param {KVNamespace} kv
* @returns {import('@pdsjs/core/ports').SharedStoragePort}
*/
export function createCloudflareSharedStorage(kv) {
return {
async getActor(did) {
const data = await kv.get(`actor:${did}`, { type: 'json' });
return data || null;
},
async resolveHandle(handle) {
const did = await kv.get(`handle:${handle.toLowerCase()}`);
return did || null;
},
async putActor(did, handle, createdAt) {
// Store actor record
await kv.put(`actor:${did}`, JSON.stringify({ did, handle, createdAt }));
// Store handle -> did mapping for resolution
if (handle) {
await kv.put(`handle:${handle.toLowerCase()}`, did);
}
},
async deleteActor(did) {
const actor = await this.getActor(did);
if (actor?.handle) {
await kv.delete(`handle:${actor.handle.toLowerCase()}`);
}
await kv.delete(`actor:${did}`);
},
async getOAuthRequest(requestUri) {
const data = await kv.get(`oauth_req:${requestUri}`, { type: 'json' });
return data || null;
},
async putOAuthRequest(requestUri, data, expiresAt) {
const ttl = Math.max(60, Math.floor((expiresAt - Date.now()) / 1000));
await kv.put(`oauth_req:${requestUri}`, JSON.stringify(data), { expirationTtl: ttl });
},
async deleteOAuthRequest(requestUri) {
await kv.delete(`oauth_req:${requestUri}`);
},
async getOAuthToken(tokenId) {
const data = await kv.get(`oauth_tok:${tokenId}`, { type: 'json' });
return data || null;
},
async putOAuthToken(tokenId, data) {
const tokenData = /** @type {{did: string}} */ (data);
await kv.put(`oauth_tok:${tokenId}`, JSON.stringify(data));
// Also maintain a list of tokens per DID (for listOAuthTokensByDid)
const listKey = `oauth_tok_list:${tokenData.did}`;
const existing = await kv.get(listKey, { type: 'json' }) || [];
if (!existing.includes(tokenId)) {
existing.push(tokenId);
await kv.put(listKey, JSON.stringify(existing));
}
},
async deleteOAuthToken(tokenId) {
const token = await this.getOAuthToken(tokenId);
if (token?.did) {
const listKey = `oauth_tok_list:${token.did}`;
const existing = await kv.get(listKey, { type: 'json' }) || [];
const updated = existing.filter((/** @type {string} */ id) => id !== tokenId);
await kv.put(listKey, JSON.stringify(updated));
}
await kv.delete(`oauth_tok:${tokenId}`);
},
async listOAuthTokensByDid(did) {
const listKey = `oauth_tok_list:${did}`;
const tokenIds = await kv.get(listKey, { type: 'json' }) || [];
const tokens = await Promise.all(
tokenIds.map((/** @type {string} */ id) => this.getOAuthToken(id))
);
return tokens.filter(Boolean);
},
async checkAndStoreDpopJti(jti, expiresAt) {
const existing = await kv.get(`dpop:${jti}`);
if (existing) return false;
const ttl = Math.max(60, Math.floor((expiresAt - Date.now()) / 1000));
await kv.put(`dpop:${jti}`, '1', { expirationTtl: ttl });
return true;
},
async cleanupExpiredDpopJtis() {
// KV handles expiration automatically via expirationTtl
},
};
}
```
**Step 2: Run typecheck**
Run: `pnpm typecheck`
Expected: PASS
**Step 3: Commit**
```bash
git add packages/cloudflare/index.js
git commit -m "feat: add createCloudflareSharedStorage to Cloudflare adapter"
```
---
## Task 4: Remove shared methods from ActorStorage adapters
**Files:**
- Modify: `packages/storage-better-sqlite3/index.js`
- Modify: `packages/cloudflare/index.js`
**Step 1: Remove shared tables and methods from better-sqlite3**
In `packages/storage-better-sqlite3/index.js`:
1. Remove from schema creation:
- `oauth_requests` table
- `oauth_tokens` table
- `dpop_jtis` table
- Related indexes
2. Remove prepared statements:
- `getOAuthRequest`, `putOAuthRequest`, `deleteOAuthRequest`
- `getOAuthToken`, `putOAuthToken`, `deleteOAuthToken`, `listOAuthTokensByDid`
- `storeDpopJti`, `cleanupExpiredDpopJtis`
3. Remove methods from returned object:
- `getOAuthRequest`, `putOAuthRequest`, `deleteOAuthRequest`
- `getOAuthToken`, `putOAuthToken`, `deleteOAuthToken`, `listOAuthTokensByDid`
- `checkAndStoreDpopJti`, `cleanupExpiredDpopJtis`
**Step 2: Remove shared tables and methods from Cloudflare**
In `packages/cloudflare/index.js`:
1. Remove from schema creation in `initSchema()`:
- `oauth_requests` table
- `oauth_tokens` table
- `dpop_jtis` table
- Related indexes
2. Remove methods from storage object:
- `getOAuthRequest`, `putOAuthRequest`, `deleteOAuthRequest`
- `getOAuthToken`, `putOAuthToken`, `deleteOAuthToken`, `listOAuthTokensByDid`
- `checkAndStoreDpopJti`, `cleanupExpiredDpopJtis`
**Step 3: Run typecheck**
Run: `pnpm typecheck`
Expected: Errors about missing methods in PDS (fixed in next task)
**Step 4: Commit**
```bash
git add packages/storage-better-sqlite3/index.js packages/cloudflare/index.js
git commit -m "refactor: remove shared storage methods from ActorStorage adapters"
```
---
## Task 5: Update PDS to use both storage ports
**Files:**
- Modify: `packages/core/pds.js`
**Step 1: Update constructor to accept sharedStorage**
Find the constructor and update to accept both storages:
```javascript
/**
* @param {Object} opts
* @param {import('./ports.js').ActorStoragePort} opts.actorStorage
* @param {import('./ports.js').SharedStoragePort} opts.sharedStorage
* @param {import('./ports.js').BlobPort} opts.blobs
* @param {import('./ports.js').WebSocketPort} [opts.websocket]
*/
constructor(opts) {
this.actorStorage = opts.actorStorage;
this.sharedStorage = opts.sharedStorage;
this.blobs = opts.blobs;
this.websocket = opts.websocket;
}
```
**Step 2: Update all storage references**
Replace `this.storage` with appropriate port:
- Actor data operations → `this.actorStorage`
- OAuth request operations → `this.sharedStorage`
- OAuth token operations → `this.sharedStorage`
- DPoP operations → `this.sharedStorage`
Common patterns to find/replace:
- `this.storage.getOAuthRequest` → `this.sharedStorage.getOAuthRequest`
- `this.storage.putOAuthRequest` → `this.sharedStorage.putOAuthRequest`
- `this.storage.getOAuthToken` → `this.sharedStorage.getOAuthToken`
- `this.storage.putOAuthToken` → `this.sharedStorage.putOAuthToken`
- `this.storage.checkAndStoreDpopJti` → `this.sharedStorage.checkAndStoreDpopJti`
- `this.storage.getRecord` → `this.actorStorage.getRecord`
- (etc. for all actor storage methods)
**Step 3: Add resolveHandle support**
Update the `handleResolveHandle` method to use sharedStorage:
```javascript
async handleResolveHandle(handle) {
// First check shared storage for handle -> DID mapping
const did = await this.sharedStorage.resolveHandle(handle);
if (did) {
return { did };
}
// Fall back to checking if this is our handle
const ourHandle = await this.actorStorage.getHandle();
const ourDid = await this.actorStorage.getDid();
if (ourHandle === handle && ourDid) {
return { did: ourDid };
}
throw { error: 'InvalidRequest', message: 'Unable to resolve handle' };
}
```
**Step 4: Run typecheck**
Run: `pnpm typecheck`
Expected: PASS (or errors in entry points, fixed next)
**Step 5: Commit**
```bash
git add packages/core/pds.js
git commit -m "refactor: update PDS to use actorStorage and sharedStorage ports"
```
---
## Task 6: Update Node.js entry point
**Files:**
- Modify: `packages/node/index.js`
**Step 1: Import shared storage**
```javascript
import { createBetterSqliteStorage, createSharedStorage } from '@pdsjs/storage-better-sqlite3';
```
(Update the existing import to include createSharedStorage)
**Step 2: Create both storage instances**
Update the initialization to create both:
```javascript
// Create actor storage (per-DID data)
const actorStorage = createBetterSqliteStorage(db);
// Create shared storage (global data) - can use same db for single-tenant
const sharedStorage = createSharedStorage(db);
// Create PDS with both storages
const pds = new PersonalDataServer({
actorStorage,
sharedStorage,
blobs,
websocket,
});
```
**Step 3: Run typecheck**
Run: `pnpm typecheck`
Expected: PASS
**Step 4: Commit**
```bash
git add packages/node/index.js
git commit -m "refactor: update Node.js entry to use both storage ports"
```
---
## Task 7: Update Cloudflare entry point
**Files:**
- Modify: `packages/cloudflare/index.js`
**Step 1: Update Durable Object to accept KV binding**
The worker needs to pass KV namespace to the Durable Object. Update the fetch handler to extract KV:
```javascript
// In the worker export
export default {
async fetch(request, env) {
// Get Durable Object stub
const id = env.PDS.idFromName('pds');
const stub = env.PDS.get(id);
// Pass KV namespace via header or store in env
return stub.fetch(request);
}
}
```
**Step 2: Create shared storage in DO constructor**
```javascript
import { createCloudflareSharedStorage } from './shared.js';
constructor(state, env) {
this.state = state;
this.env = env;
// ... existing init ...
// Create shared storage using KV
this.sharedStorage = createCloudflareSharedStorage(env.SHARED_KV);
}
```
**Step 3: Pass both storages to PDS**
```javascript
this.pds = new PersonalDataServer({
actorStorage: storage, // DO's SQLite storage
sharedStorage: this.sharedStorage, // KV storage
blobs,
websocket,
});
```
**Step 4: Update wrangler.toml**
Add KV namespace binding:
```toml
[[kv_namespaces]]
binding = "SHARED_KV"
id = "your-kv-namespace-id"
```
**Step 5: Run typecheck**
Run: `pnpm typecheck`
Expected: PASS
**Step 6: Commit**
```bash
git add packages/cloudflare/ wrangler.toml
git commit -m "refactor: update Cloudflare entry to use KV for shared storage"
```
---
## Task 8: Update unit tests
**Files:**
- Modify: `test/pds.test.js`
**Step 1: Update any unit tests that create storage mocks**
If any tests create mock storage objects, update them to match the new ActorStoragePort interface (without OAuth/DPoP methods).
**Step 2: Run unit tests**
Run: `pnpm test -- test/pds.test.js`
Expected: All 126 unit tests PASS
**Step 3: Commit**
```bash
git add test/pds.test.js
git commit -m "test: update unit tests for split storage ports"
```
---
## Task 9: Run e2e tests and verify OAuth flow
**Files:**
- None (verification only)
**Step 1: Run e2e tests**
Run: `pnpm test -- test/e2e.test.js`
Expected: All 84 e2e tests PASS
**Step 2: Verify OAuth endpoints work**
The e2e tests cover:
- OAuth authorization request flow (uses SharedStorage.putOAuthRequest)
- Token exchange (uses SharedStorage.putOAuthToken)
- DPoP proof validation (uses SharedStorage.checkAndStoreDpopJti)
- Token refresh and revocation
If any OAuth tests fail, check that:
1. SharedStorage is being passed correctly to PDS
2. OAuth methods are calling `this.sharedStorage` not `this.actorStorage`
**Step 3: Verify handle resolution works**
Test handle resolution manually or check e2e test coverage:
- `com.atproto.identity.resolveHandle` should use SharedStorage.resolveHandle
---
## Task 10: Register actor on init
**Files:**
- Modify: `packages/core/pds.js`
**Step 1: Update handleInit to register actor in shared storage**
In the `handleInit` method, after setting DID and handle in actor storage, also register in shared storage:
```javascript
async handleInit(did, privateKeyHex, handle) {
// ... existing validation and actor storage setup ...
// Register actor in shared storage for handle resolution
await this.sharedStorage.putActor(did, handle, Date.now());
// ... rest of init ...
}
```
**Step 2: Run all tests**
Run: `pnpm test`
Expected: All 210 tests PASS
**Step 3: Commit**
```bash
git add packages/core/pds.js
git commit -m "feat: register actor in shared storage on init"
```
---
## Summary
After completing all tasks:
- `ActorStoragePort` handles per-DID data (records, blocks, commits, blobs, preferences)
- `SharedStoragePort` handles global data (actors, OAuth, DPoP)
- Node.js uses SQLite for both (same file for now, separate later)
- Cloudflare uses DO SQLite for actor, KV for shared
- Handle resolution uses shared storage
- Ready for multi-tenant when needed