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