Cloudflare Durable Objects + Hono로 실시간 협업 기능 구현하기

name = "realtime-collab"
main = "src/index.ts"
compatibility_date = "2024-09-23"
 
[[durable_objects.bindings]]
name = "ROOM"
class_name = "Room"
 
[[migrations]]
tag = "v1"
new_sqlite_classes = ["Room"]

// src/index.ts
import { Hono } from 'hono';
import { Room } from './room';
 
export { Room };
 
type Env = {
  ROOM: DurableObjectNamespace;
};
 
const app = new Hono<{ Bindings: Env }>();
 
app.use('/room/*', async (c, next) => {
  const token = c.req.header('Authorization')?.replace('Bearer ', '');
  if (!token) return c.json({ error: 'Unauthorized' }, 401);
 
  const userId = verifyToken(token).sub;
  c.set('userId', userId);
  await next();
});
 
app.get('/room/:roomId/ws', async (c) => {
  const roomId = c.req.param('roomId');
  const userId = c.get('userId');
 
  const id = c.env.ROOM.idFromName(roomId);
  const stub = c.env.ROOM.get(id);
 
  // 검증된 userId를 헤더에 주입 — .set()으로 클라이언트 제공 값을 덮어씀
  const headers = new Headers(c.req.raw.headers);
  headers.set('X-User-Id', userId);
 
  return stub.fetch(new Request(c.req.raw, { headers }));
});
 
export default app;

// src/room.ts
import { DurableObject } from 'cloudflare:workers';
 
interface SessionMeta {
  userId: string;
  username: string;
}
 
interface Env {}
 
export class Room extends DurableObject<Env> {
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
 
    // 인스턴스 = 단일 방이므로 room_id 컬럼 불필요
    this.ctx.blockConcurrencyWhile(async () => {
      this.ctx.storage.sql.exec(`
        CREATE TABLE IF NOT EXISTS messages (
          id         INTEGER PRIMARY KEY AUTOINCREMENT,
          user_id    TEXT    NOT NULL,
          username   TEXT    NOT NULL,
          content    TEXT    NOT NULL,
          created_at INTEGER NOT NULL
        )
      `);
    });
  }
 
  async fetch(request: Request): Promise<Response> {
    if (request.headers.get('Upgrade') !== 'websocket') {
      return new Response('WebSocket expected', { status: 426 });
    }
 
    // Worker가 주입한 헤더에서만 userId를 읽음 — 쿼리 파라미터는 신뢰하지 않음
    const userId = request.headers.get('X-User-Id');
    if (!userId) {
      return new Response('Missing user identity', { status: 400 });
    }
 
    const url = new URL(request.url);
    const username = url.searchParams.get('username') ?? 'Guest';
 
    const { 0: client, 1: server } = new WebSocketPair();
 
    // Hibernation 활성화 — 이 한 줄이 없으면 유휴 연결도 과금됨
    this.ctx.acceptWebSocket(server);
 
    // Worker가 검증한 userId 기반이므로 클라이언트 위조 불가
    const meta: SessionMeta = { userId, username };
    server.serializeAttachment(meta);
 
    // 신규 연결자에게 최근 메시지 히스토리 전송
    this.sendHistory(server);
 
    this.broadcast(
      { type: 'join', userId, username, timestamp: Date.now() },
      server,
    );
 
    return new Response(null, { status: 101, webSocket: client });
  }
 
  async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
    // 바이너리 메시지는 이 구현에서 지원하지 않음
    if (typeof message !== 'string') {
      ws.send(JSON.stringify({ type: 'error', message: 'Binary messages not supported' }));
      return;
    }
 
    const meta = ws.deserializeAttachment() as SessionMeta;
 
    let parsed: { content?: unknown };
    try {
      parsed = JSON.parse(message);
    } catch {
      ws.send(JSON.stringify({ type: 'error', message: 'Invalid JSON' }));
      return;
    }
 
    const content = parsed.content;
    if (typeof content !== 'string' || content.trim() === '') return;
 
    // storage.sql.exec()는 동기 API — await 불필요
    this.ctx.storage.sql.exec(
      `INSERT INTO messages (user_id, username, content, created_at)
       VALUES (?, ?, ?, ?)`,
      meta.userId,
      meta.username,
      content,
      Date.now(),
    );
 
    this.broadcast({
      type: 'message',
      userId: meta.userId,
      username: meta.username,
      content,
      timestamp: Date.now(),
    });
  }
 
  async webSocketClose(ws: WebSocket, _code: number) {
    const meta = ws.deserializeAttachment() as SessionMeta;
    this.broadcast(
      { type: 'leave', userId: meta.userId, username: meta.username, timestamp: Date.now() },
      ws,
    );
  }
 
  async webSocketError(ws: WebSocket, _error: unknown) {
    // webSocketClose가 자동 호출되지 않는 경우가 있으므로 별도 처리
    const meta = ws.deserializeAttachment() as SessionMeta | null;
    if (meta) {
      this.broadcast(
        { type: 'leave', userId: meta.userId, username: meta.username, timestamp: Date.now() },
        ws,
      );
    }
  }
 
  private sendHistory(ws: WebSocket) {
    const rows = this.ctx.storage.sql
      .exec(
        `SELECT user_id, username, content, created_at
         FROM messages
         ORDER BY created_at DESC
         LIMIT 50`,
      )
      .toArray();
 
    ws.send(JSON.stringify({ type: 'history', messages: rows.reverse() }));
  }
 
  private broadcast(data: unknown, exclude?: WebSocket) {
    const payload = JSON.stringify(data);
    for (const conn of this.ctx.getWebSockets()) {
      if (conn !== exclude) {
        try {
          conn.send(payload);
        } catch {
          // 이미 닫힌 연결은 무시
        }
      }
    }
  }
}

// src/collab-document.ts
import { DurableObject } from 'cloudflare:workers';
import { YDurableObjects } from 'y-durableobjects';
 
export class CollabDocument extends YDurableObjects(DurableObject) {
  // fetch, webSocketMessage, webSocketClose 모두 믹스인이 처리
  // 충돌 해소는 Yjs CRDT가 자동으로 담당
}

// src/index.ts
app.get('/doc/:docId/ws', async (c) => {
  const id = c.env.COLLAB_DOC.idFromName(c.req.param('docId'));
  return c.env.COLLAB_DOC.get(id).fetch(c.req.raw);
});

// 위험한 패턴 — 휴면 후 재생성 시 초기화됨
private sessions = new Map<WebSocket, SessionMeta>();
 
async webSocketMessage(ws: WebSocket, msg: string) {
  const meta = this.sessions.get(ws); // 재생성 후에는 undefined
}

[[durable_objects.bindings]]
name = "ROOM"
class_name = "RoomHandler"  # 실제 export 이름과 다름

// src/room.ts
export class Room extends DurableObject {}  // class_name과 불일치