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 {}:

/**
 * @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

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):

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):

/**
 * 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:

/**
 * 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):

'/xrpc/com.atproto.sync.subscribeRepos': {
  handler: PersonalDataServer.prototype.handleSubscribeRepos,
},

Step 5: Commit

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:

/**
 * 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:

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

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

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:

import { WebSocketServer } from 'ws';

Add factory function before createServer:

/**
 * 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:

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

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:

import WebSocket from 'ws';

Step 2: Add subscribeRepos test suite

Add new describe block (after Foreign DID proxying tests):

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

npm install -D ws

Step 4: Run tests

npm run test:e2e:node
npm run test:e2e:cloudflare

Expected: All tests pass including new WebSocket tests.

Step 5: Commit

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

npx wrangler deploy

Step 2: Request crawl from relay

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

# 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)