DO 實戰:多人協作聊天室從零打造 | Cloudflare 完整教學

// wrangler.jsonc
{
  "name": "chat-app",
  "main": "src/index.ts",
  "compatibility_date": "2024-04-03",
  "durable_objects": {
    "bindings": [
      { "name": "CHAT_ROOM", "class_name": "ChatRoom" }
    ]
  },
  "migrations": [
    { "tag": "v1", "new_sqlite_classes": ["ChatRoom"] }
  ]
}

// src/index.ts
import { DurableObject } from "cloudflare:workers";

export interface Env {
  // ★ 泛型參數讓 stub 帶有 ChatRoom 的 RPC 方法型別
  CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}

export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    const url = new URL(req.url);

    // 路由:/room/:roomId/ws → WebSocket 升級
    const wsMatch = url.pathname.match(/^\/room\/([^/]+)\/ws$/);
    if (wsMatch) {
      const roomId = wsMatch[1];
      // ★ 房間路由:相同 roomId 永遠指向同一個 DO 實例
      const stub = env.CHAT_ROOM.getByName(`room:${roomId}`);
      // 把原始 Request(含 Upgrade header)直接轉交給 DO
      return stub.fetch(req);
    }

    // 路由:/room/:roomId/history → 用 RPC 取得歷史訊息
    const histMatch = url.pathname.match(/^\/room\/([^/]+)\/history$/);
    if (histMatch) {
      const stub = env.CHAT_ROOM.getByName(`room:${histMatch[1]}`);
      const before = Number(url.searchParams.get("before") ?? 0) || undefined;
      const messages = await stub.getHistory(50, before); // 型別安全的 RPC 呼叫
      return Response.json(messages);
    }

    return new Response("Not Found", { status: 404 });
  },
} satisfies ExportedHandler<Env>;

export class ChatRoom extends DurableObject<Env> {
  private sql = this.ctx.storage.sql;

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);

    // ★ 只在初始化時 block:建表在第一個請求進來前完成
    ctx.blockConcurrencyWhile(async () => {
      this.sql.exec(`
        CREATE TABLE IF NOT EXISTS messages (
          id   INTEGER PRIMARY KEY AUTOINCREMENT,
          who  TEXT NOT NULL,
          text TEXT NOT NULL,
          ts   INTEGER NOT NULL DEFAULT (unixepoch() * 1000)
        );
      `);
    });

    // ★ ping/pong 心跳自動回應,不需喚醒物件(省計費)
    ctx.setWebSocketAutoResponse(
      new WebSocketRequestResponsePair("ping", "pong")
    );
  }
}

export class ChatRoom extends DurableObject<Env> {
  // ...(承上)

  async fetch(req: Request): Promise<Response> {
    if (req.headers.get("Upgrade") !== "websocket") {
      return new Response("Expected WebSocket", { status: 426 });
    }

    const url = new URL(req.url);
    const userId = url.searchParams.get("userId") ?? crypto.randomUUID();
    const username = url.searchParams.get("username") ?? "訪客";

    const { 0: client, 1: server } = new WebSocketPair();

    // ★ 關鍵:用 acceptWebSocket(而非 server.accept())啟用 Hibernation
    //   第二參數是 tags,之後可用 getWebSockets(tag) 篩選
    this.ctx.acceptWebSocket(server, [userId]);

    // ★ 連線資料存進 attachment,跨 Hibernation 存活(上限 16 KB)
    server.serializeAttachment({ userId, username });

    // ★ 廣播「加入」給其他人,並把目前在線名單發給新連上的人
    this.broadcast({ type: "join", userId, username }, server);
    server.send(JSON.stringify({ type: "presence", users: this.listPresence() }));

    return new Response(null, { status: 101, webSocket: client });
  }

  // ★ 目前在線名單:一律從 runtime 維護的連線清單推導,不自己存 Set
  private listPresence(): { userId: string; username: string }[] {
    return this.ctx.getWebSockets().map((ws) => {
      const meta = ws.deserializeAttachment() as { userId: string; username: string };
      return { userId: meta.userId, username: meta.username };
    });
  }
}

export class ChatRoom extends DurableObject<Env> {
  // ...(承上)

  async webSocketMessage(ws: WebSocket, raw: string | ArrayBuffer): Promise<void> {
    // Hibernation 後 in-memory 狀態已清空,一律從 attachment 讀回身分
    const { username } = ws.deserializeAttachment() as { username: string };

    let data: { type: string; text?: string };
    try {
      data = JSON.parse(typeof raw === "string" ? raw : new TextDecoder().decode(raw));
    } catch {
      ws.send(JSON.stringify({ type: "error", message: "格式錯誤" }));
      return;
    }

    if (data.type === "message" && data.text) {
      if (data.text.length > 500) {
        ws.send(JSON.stringify({ type: "error", message: "訊息過長(上限 500 字)" }));
        return;
      }

      // ★ 先持久化:INSERT ... RETURNING 直接拿回自增 id 與時間
      const row = this.sql
        .exec<{ id: number; ts: number }>(
          "INSERT INTO messages (who, text) VALUES (?, ?) RETURNING id, ts",
          username,
          data.text
        )
        .one();

      // ★ 再廣播:Output Gate 保證 SQLite 寫入完成後,訊息才對外送出
      this.broadcast({
        type: "message",
        id: row.id,
        who: username,
        text: data.text,
        ts: row.ts,
      });

      // ★ 確保有一個每日清理鬧鐘(冪等:已有就不重設)
      await this.ensureCleanupAlarm();
    }
  }

  // ★ 廣播核心:一律走 ctx.getWebSockets(),不自己維護 Set
  private broadcast(payload: unknown, exclude?: WebSocket): void {
    const msg = JSON.stringify(payload);
    for (const ws of this.ctx.getWebSockets()) {
      if (ws === exclude) continue;
      try {
        ws.send(msg);
      } catch {
        // 連線已死,忽略(webSocketClose 會處理後續)
      }
    }
  }
}

export class ChatRoom extends DurableObject<Env> {
  // ...(承上)

  async webSocketClose(ws: WebSocket, code: number, reason: string): Promise<void> {
    const meta = ws.deserializeAttachment() as { userId: string; username: string } | null;
    if (meta) {
      this.broadcast({ type: "leave", userId: meta.userId, username: meta.username });
    }
  }

  async webSocketError(ws: WebSocket, error: unknown): Promise<void> {
    console.error("WebSocket error:", error);
    try { ws.close(1011, "internal error"); } catch {}
  }
}

export class ChatRoom extends DurableObject<Env> {
  // ...(承上)

  // ★ RPC:回傳最近 limit 筆(可帶 before 游標往上翻舊訊息)
  async getHistory(
    limit = 50,
    before?: number
  ): Promise<{ id: number; who: string; text: string; ts: number }[]> {
    const rows = before
      ? this.sql.exec<{ id: number; who: string; text: string; ts: number }>(
          "SELECT id, who, text, ts FROM messages WHERE id < ? ORDER BY id DESC LIMIT ?",
          before,
          limit
        )
      : this.sql.exec<{ id: number; who: string; text: string; ts: number }>(
          "SELECT id, who, text, ts FROM messages ORDER BY id DESC LIMIT ?",
          limit
        );
    // DESC 撈出後 reverse 成正序(舊→新)給前端顯示
    return rows.toArray().reverse();
  }
}

export class ChatRoom extends DurableObject<Env> {
  // ...(承上)

  private static readonly RETENTION_MS = 30 * 24 * 60 * 60 * 1000; // 保留 30 天
  private static readonly DAY_MS = 24 * 60 * 60 * 1000;

  // ★ 冪等:已有 alarm 就不重設(避免每則訊息都重排)
  private async ensureCleanupAlarm(): Promise<void> {
    const existing = await this.ctx.storage.getAlarm();
    if (existing === null) {
      await this.ctx.storage.setAlarm(Date.now() + ChatRoom.DAY_MS);
    }
  }

  // ★ Alarm handler:清理 + 重排下一次
  async alarm(): Promise<void> {
    const cutoff = Date.now() - ChatRoom.RETENTION_MS;
    this.sql.exec("DELETE FROM messages WHERE ts < ?", cutoff);

    // 若房間還有連線或還有訊息,續排明天的清理;否則讓它自然沉睡
    const hasSockets = this.ctx.getWebSockets().length > 0;
    const remaining = this.sql
      .exec<{ n: number }>("SELECT COUNT(*) AS n FROM messages")
      .one().n;

    if (hasSockets || remaining > 0) {
      await this.ctx.storage.setAlarm(Date.now() + ChatRoom.DAY_MS);
    }
  }
}

// ❌ 錯誤:全域單一 DO,所有房間共用 → 瓶頸
const stub = env.CHAT_ROOM.getByName("global"); // 全站都撞同一個實例

// ✅ 正確:每房間一個 DO,負載天然分散
const stub = env.CHAT_ROOM.getByName(`room:${roomId}`);

// ❌ 錯誤:自己維護 Set,休眠後清空
private sessions = new Set<WebSocket>();
this.sessions.add(server);            // 休眠後這個 Set 會是空的
for (const ws of this.sessions) ws.send(msg); // 廣播漏人

// ✅ 正確:一律用 runtime 維護的連線清單
for (const ws of this.ctx.getWebSockets()) ws.send(msg);

// ❌ 錯誤:撈全部,房間一大就爆
this.sql.exec("SELECT * FROM messages").toArray();

// ✅ 正確:只撈最近 N 筆,翻頁用 WHERE id < ?
this.sql.exec("SELECT ... FROM messages ORDER BY id DESC LIMIT ?", 50);

// ✅ 正確順序:先寫 SQLite,再廣播
const row = this.sql.exec("INSERT ... RETURNING id, ts", who, text).one();
this.broadcast({ type: "message", id: row.id, ... }); // Output Gate 護航

// ❌ 錯誤:迭代中途 await,快照失效
for (const row of this.sql.exec("SELECT ...")) { await doAsync(row); }

// ✅ 正確:先全取出,再 await
for (const row of this.sql.exec("SELECT ...").toArray()) { await doAsync(row); }