chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
# subscribeRepos WebSocket Implementation Plan
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
**Goal:** Add `com.atproto.sync.subscribeRepos` WebSocket endpoint to the PDS using hexagonal architecture.
**Architecture:** Define a `WebSocketPort` interface in the core that abstracts WebSocket handling. Platform adapters (Cloudflare, Node.js) implement this port. The core PDS calls the port methods without knowing platform specifics.
**Tech Stack:** Cloudflare WebSocketPair + Durable Object hibernation, Node.js `ws` library, Vitest for e2e tests.
---
## Task 1: Add WebSocketPort Types to ports.js
**Files:**
- Modify: `packages/core/ports.js:60-70`
**Step 1: Add WebSocketSender and WebSocketPort typedefs**
Add before the final `export {}`:
```js
/**
* @typedef {Object} WebSocketSender
* @property {(data: Uint8Array) => void} send - Send binary data to client
* @property {() => void} close - Close the connection
*/
/**
* @typedef {Object} WebSocketPort
* @property {(request: Request) => boolean} isUpgrade - Check if request is WebSocket upgrade
* @property {(request: Request, onConnect: (sender: WebSocketSender) => Promise<void>) => Response} upgrade - Upgrade connection and call onConnect with sender
*/
```
**Step 2: Commit**
```bash
git add packages/core/ports.js
git commit -m "feat: add WebSocketPort types to ports.js"
```
---
## Task 2: Add handleSubscribeRepos to Core PDS
**Files:**
- Modify: `packages/core/pds.js:290-310` (constructor)
- Modify: `packages/core/pds.js:2900-2910` (routes)
- Add handler method after sync handlers (~line 2600)
**Step 1: Add webSocket to constructor**
In constructor parameters and assignment (around line 290):
```js
constructor({
storage,
blobs,
jwtSecret,
hostname,
appviewUrl,
appviewDid,
relayUrl,
password,
webSocket, // Add this
}) {
this.storage = storage;
this.blobs = blobs;
this.jwtSecret = jwtSecret;
this.hostname = hostname;
this.appviewUrl = appviewUrl || BSKY_APPVIEW_URL;
this.appviewDid = appviewDid;
this.relayUrl = relayUrl;
this.password = password;
this.webSocket = webSocket; // Add this
this.lastCrawlNotify = 0;
// ... rest
}
```
**Step 2: Add formatFirehoseEvent method**
Add after the existing sync handlers (around line 2600):
```js
/**
* Format event for firehose (AT Protocol frame format)
* @param {{seq: number, evt: Uint8Array}} evt
* @returns {Uint8Array}
*/
formatFirehoseEvent(evt) {
// AT Protocol frame format: header + body
const header = cborEncode({ op: 1, t: '#commit' });
// Decode stored event to rebuild with proper CID encoding
const evtData = cborDecode(evt.evt);
// Build body with blocks included
const body = cborEncodeDagCbor({
seq: evt.seq,
rebase: false,
tooBig: false,
repo: evtData.did,
commit: evtData.commit ? { $link: evtData.commit } : null,
rev: evtData.rev,
since: null,
blocks: evtData.blocks || new Uint8Array(0),
ops: evtData.ops || [],
blobs: [],
time: evtData.time,
});
// Combine header and body
const frame = new Uint8Array(header.length + body.length);
frame.set(header, 0);
frame.set(body, header.length);
return frame;
}
```
**Step 3: Add handleSubscribeRepos method**
Add after formatFirehoseEvent:
```js
/**
* com.atproto.sync.subscribeRepos - WebSocket firehose
* @param {Request} request
* @param {URL} url
* @returns {Response}
*/
handleSubscribeRepos(request, url) {
// Check if WebSocket port is available
if (!this.webSocket) {
return Response.json(
{ error: 'NotImplemented', message: 'WebSocket not supported' },
{ status: 501 },
);
}
// Check for WebSocket upgrade
if (!this.webSocket.isUpgrade(request)) {
return new Response('Expected WebSocket upgrade', { status: 426 });
}
const cursor = url.searchParams.get('cursor');
// Upgrade connection and send historical events
return this.webSocket.upgrade(request, async (sender) => {
if (cursor) {
const cursorNum = parseInt(cursor, 10);
if (!isNaN(cursorNum)) {
const { events } = await this.storage.getEvents(cursorNum, 1000);
for (const evt of events) {
sender.send(this.formatFirehoseEvent(evt));
}
}
}
});
}
```
**Step 4: Add route to ROUTES object**
Add to the Sync section of ROUTES (around line 2905):
```js
'/xrpc/com.atproto.sync.subscribeRepos': {
handler: PersonalDataServer.prototype.handleSubscribeRepos,
},
```
**Step 5: Commit**
```bash
git add packages/core/pds.js
git commit -m "feat: add handleSubscribeRepos WebSocket handler to core PDS"
```
---
## Task 3: Add Cloudflare WebSocket Adapter
**Files:**
- Modify: `packages/cloudflare/index.js:500-540`
**Step 1: Add createCloudflareWebSocket factory function**
Add before the `PDSDurableObject` class:
```js
/**
* Create WebSocket port for Cloudflare Durable Objects
* @param {DurableObjectState} state
* @returns {import('@pdsjs/core/ports.js').WebSocketPort}
*/
function createCloudflareWebSocket(state) {
return {
isUpgrade(request) {
return request.headers.get('Upgrade') === 'websocket';
},
upgrade(request, onConnect) {
const pair = new WebSocketPair();
const [client, server] = Object.values(pair);
// Accept for 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 });
},
};
}
```
**Step 2: Inject webSocket into PDS constructor**
Modify the PDSDurableObject constructor:
```js
constructor(state, env) {
this.state = state;
this.env = env;
const storage = createD1Storage(state.storage.sql);
const blobs = createR2Blobs(env.BLOBS || env.BLOB_BUCKET);
const webSocket = createCloudflareWebSocket(state); // Add this
this.pds = new PersonalDataServer({
storage,
blobs,
webSocket, // Add this
jwtSecret: env.JWT_SECRET,
hostname: env.HOSTNAME,
appviewUrl: env.APPVIEW_URL,
appviewDid: env.APPVIEW_DID,
relayUrl: env.RELAY_URL,
password: env.PDS_PASSWORD,
});
}
```
**Step 3: Commit**
```bash
git add packages/cloudflare/index.js
git commit -m "feat: add Cloudflare WebSocket adapter for subscribeRepos"
```
---
## Task 4: Add Node.js WebSocket Adapter
**Files:**
- Modify: `packages/node/index.js`
- Modify: `packages/node/package.json` (add ws dependency)
**Step 1: Add ws dependency**
```bash
cd packages/node && npm install ws && cd ../..
```
**Step 2: Add createNodeWebSocket factory and server upgrade handling**
Add imports at top of `packages/node/index.js`:
```js
import { WebSocketServer } from 'ws';
```
Add factory function before `createServer`:
```js
/**
* Create WebSocket port for Node.js
* @param {WeakMap<Request, import('ws').WebSocket>} upgradeMap
* @returns {import('@pdsjs/core/ports.js').WebSocketPort}
*/
function createNodeWebSocket(upgradeMap) {
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 101 - actual upgrade handled by server
return new Response(null, { status: 101 });
},
};
}
```
**Step 3: Modify createServer to handle WebSocket upgrades**
Update `createServer` function to create WebSocket infrastructure and handle upgrades:
```js
export async function createServer({
dbPath,
blobsDir,
jwtSecret,
port = 3000,
hostname,
appviewUrl,
appviewDid,
relayUrl,
password,
}) {
const Database = (await import('better-sqlite3')).default;
const db = new Database(dbPath);
const storage = createBetterSqliteStorage(db);
const blobs = createFsBlobs(blobsDir);
// WebSocket infrastructure
const upgradeMap = new WeakMap();
const webSocket = createNodeWebSocket(upgradeMap);
const wss = new WebSocketServer({ noServer: true });
const pds = new PersonalDataServer({
storage,
blobs,
webSocket, // Add this
jwtSecret,
hostname,
appviewUrl,
appviewDid,
relayUrl,
password,
});
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.status === 101 && 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();
}
});
// ... rest of function unchanged
```
**Step 4: Commit**
```bash
git add packages/node/index.js packages/node/package.json
git commit -m "feat: add Node.js WebSocket adapter for subscribeRepos"
```
---
## Task 5: Add E2E Tests for subscribeRepos
**Files:**
- Modify: `test/e2e.test.js`
**Step 1: Add WebSocket import and helper**
Add at top of file with other imports:
```js
import WebSocket from 'ws';
```
**Step 2: Add subscribeRepos test suite**
Add new describe block (after Foreign DID proxying tests):
```js
describe('subscribeRepos WebSocket', () => {
it('rejects non-WebSocket requests with 426', async () => {
const res = await fetch(`${BASE}/xrpc/com.atproto.sync.subscribeRepos`);
expect(res.status).toBe(426);
});
it('connects via WebSocket', async () => {
const wsUrl = BASE.replace('http', 'ws');
const ws = new WebSocket(`${wsUrl}/xrpc/com.atproto.sync.subscribeRepos`);
await new Promise((resolve, reject) => {
ws.on('open', () => {
ws.close();
resolve();
});
ws.on('error', reject);
setTimeout(() => reject(new Error('WebSocket timeout')), 5000);
});
});
it('receives events after cursor', async () => {
// First create a record to generate an event
const { status } = await jsonPost(
'/xrpc/com.atproto.repo.createRecord',
{
repo: DID,
collection: 'app.bsky.feed.post',
record: { text: 'ws test', createdAt: new Date().toISOString() },
},
{ Authorization: `Bearer ${token}` },
);
expect(status).toBe(200);
// Get current commit to use as cursor baseline
const commitRes = await fetch(
`${BASE}/xrpc/com.atproto.sync.getLatestCommit?did=${DID}`,
);
const { rev } = await commitRes.json();
expect(rev).toBeTruthy();
// Connect with cursor=0 to get all events
const wsUrl = BASE.replace('http', 'ws');
const ws = new WebSocket(
`${wsUrl}/xrpc/com.atproto.sync.subscribeRepos?cursor=0`,
);
const events = [];
await new Promise((resolve, reject) => {
ws.on('message', (data) => {
events.push(data);
// Close after receiving at least one event
if (events.length >= 1) {
ws.close();
resolve();
}
});
ws.on('error', reject);
ws.on('open', () => {
// Give it time to receive events, then close
setTimeout(() => {
ws.close();
resolve();
}, 2000);
});
setTimeout(() => reject(new Error('WebSocket timeout')), 10000);
});
// Should have received at least one event
expect(events.length).toBeGreaterThan(0);
// Events should be binary (Uint8Array/Buffer)
expect(Buffer.isBuffer(events[0]) || events[0] instanceof Uint8Array).toBeTruthy();
});
});
```
**Step 3: Add ws as dev dependency to root package.json**
```bash
npm install -D ws
```
**Step 4: Run tests**
```bash
npm run test:e2e:node
npm run test:e2e:cloudflare
```
Expected: All tests pass including new WebSocket tests.
**Step 5: Commit**
```bash
git add test/e2e.test.js package.json package-lock.json
git commit -m "test: add e2e tests for subscribeRepos WebSocket"
```
---
## Task 6: Deploy and Verify Relay Sync
**Step 1: Deploy to Cloudflare**
```bash
npx wrangler deploy
```
**Step 2: Request crawl from relay**
```bash
curl -X POST 'https://bsky.network/xrpc/com.atproto.sync.requestCrawl' \
-H 'Content-Type: application/json' \
-d '{"hostname":"chad-pds.chad-53c.workers.dev"}'
```
**Step 3: Verify relay can sync**
```bash
# Wait a few seconds, then check
curl 'https://bsky.network/xrpc/com.atproto.sync.getLatestCommit?did=did:plc:c6vxslynzebnlk5kw2orx37o'
```
Expected: Returns matching cid and rev from your PDS.
**Step 4: Test posting**
Create a post via bsky.app and verify it appears in your feed.
---
## Summary
| Task | Description | Commit Message |
|------|-------------|----------------|
| 1 | Add WebSocketPort types | `feat: add WebSocketPort types to ports.js` |
| 2 | Add handler to core PDS | `feat: add handleSubscribeRepos WebSocket handler to core PDS` |
| 3 | Cloudflare adapter | `feat: add Cloudflare WebSocket adapter for subscribeRepos` |
| 4 | Node.js adapter | `feat: add Node.js WebSocket adapter for subscribeRepos` |
| 5 | E2E tests | `test: add e2e tests for subscribeRepos WebSocket` |
| 6 | Deploy & verify | (no commit - verification only) |