Redis와 WebSocket 서버 없이 엣지에서 채팅·분산 락 구현하기 — Cloudflare Durable Objects 실전 가이드

# wrangler.toml
[[durable_objects.bindings]]
name = "CHAT_ROOM"
class_name = "ChatRoom"
 
[durable_objects]
new_sqlite_classes = ["ChatRoom"]

export class ChatRoom extends DurableObject {
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    this.ctx.storage.sql.exec(`
      CREATE TABLE IF NOT EXISTS messages (
        id    INTEGER PRIMARY KEY AUTOINCREMENT,
        user  TEXT    NOT NULL,
        body  TEXT    NOT NULL,
        ts    INTEGER NOT NULL
      )
    `);
  }
}

export class ChatRoom extends DurableObject {
  // Hibernation API — 클래스 메서드 방식으로만 활성화됩니다
  async webSocketOpen(ws: WebSocket) { /* 새 연결 수립 시 최초 1회 */ }
  async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) { /* 메시지 수신 */ }
  async webSocketClose(ws: WebSocket, code: number, reason: string) { /* 연결 종료 */ }
  async webSocketError(ws: WebSocket, error: unknown) { /* 오류 발생 */ }
 
  // 이렇게 하면 Hibernation이 꺼집니다
  // server.addEventListener('message', handler)  ← 사용하지 마세요
}

// chat-room.ts
import { DurableObject } from "cloudflare:workers";
 
interface Env {
  CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
 
interface SessionMeta {
  userId: string;
  username: string;
}
 
export class ChatRoom extends DurableObject {
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    this.ctx.storage.sql.exec(`
      CREATE TABLE IF NOT EXISTS messages (
        id       INTEGER PRIMARY KEY AUTOINCREMENT,
        user     TEXT    NOT NULL,
        body     TEXT    NOT NULL,
        ts       INTEGER NOT NULL
      )
    `);
  }
 
  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
 
    if (url.pathname === "/websocket") {
      if (request.headers.get("Upgrade") !== "websocket") {
        return new Response("WebSocket 업그레이드 필요", { status: 426 });
      }
 
      const userId = url.searchParams.get("userId") ?? crypto.randomUUID();
      const username = url.searchParams.get("username") ?? "익명";
 
      const [client, server] = Object.values(new WebSocketPair()) as [WebSocket, WebSocket];
 
      // Hibernation 활성화 — acceptWebSocket 사용
      this.ctx.acceptWebSocket(server);
 
      // Hibernation 사이클 사이에도 세션 정보 보존
      server.serializeAttachment({ userId, username } satisfies SessionMeta);
 
      return new Response(null, { status: 101, webSocket: client });
    }
 
    return new Response("Not Found", { status: 404 });
  }
 
  // 연결 수립 직후 최초 1회 호출 — 히스토리 전송에 적합한 시점
  async webSocketOpen(ws: WebSocket) {
    const history = [...this.ctx.storage.sql.exec(
      "SELECT user, body, ts FROM messages ORDER BY id DESC LIMIT 50"
    )].reverse();
    ws.send(JSON.stringify({ type: "history", messages: history }));
  }
 
  async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
    const meta = ws.deserializeAttachment() as SessionMeta;
    const { body } = JSON.parse(message as string);
    const ts = Date.now();
 
    this.ctx.storage.sql.exec(
      "INSERT INTO messages (user, body, ts) VALUES (?, ?, ?)",
      meta.username,
      body,
      ts
    );
 
    const payload = JSON.stringify({ type: "message", user: meta.username, body, ts });
    for (const peer of this.ctx.getWebSockets()) {
      try {
        peer.send(payload);
      } catch {
        // 이미 닫힌 연결은 무시
      }
    }
  }
 
  async webSocketClose(ws: WebSocket, code: number) {
    ws.close(code, "연결 종료");
  }
 
  async webSocketError(ws: WebSocket, _error: unknown) {
    ws.close(1011, "서버 오류");
  }
}

// worker.ts — 라우터 역할
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    const roomId = url.searchParams.get("room") ?? "general";
 
    // 같은 roomId → 전 세계 어디서 호출해도 항상 같은 DO 인스턴스
    const id = env.CHAT_ROOM.idFromName(roomId);
    const stub = env.CHAT_ROOM.get(id);
 
    return stub.fetch(request);
  },
};

# wrangler.toml에 추가
[[durable_objects.bindings]]
name = "DISTRIBUTED_LOCK"
class_name = "DistributedLock"
 
[durable_objects]
new_sqlite_classes = ["ChatRoom", "DistributedLock"]

// distributed-lock.ts
import { DurableObject } from "cloudflare:workers";
 
interface Env {
  DISTRIBUTED_LOCK: DurableObjectNamespace<DistributedLock>;
}
 
export class DistributedLock extends DurableObject {
  private static readonly DEFAULT_TTL_MS = 30_000;
 
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    // 인스턴스당 락 하나 — 단일 행 테이블
    this.ctx.storage.sql.exec(`
      CREATE TABLE IF NOT EXISTS lock_state (
        owner      TEXT    NOT NULL,
        expires_at INTEGER NOT NULL
      )
    `);
  }
 
  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    const action = url.pathname.slice(1);
    const { owner, ttlMs } = await request.json<{ owner: string; ttlMs?: number }>();
 
    switch (action) {
      case "acquire": return Response.json(await this.acquire(owner, ttlMs));
      case "release": return Response.json(await this.release(owner));
      case "renew":   return Response.json(await this.renew(owner, ttlMs));
      default:        return new Response("Not Found", { status: 404 });
    }
  }
 
  async acquire(
    owner: string,
    ttlMs = DistributedLock.DEFAULT_TTL_MS
  ): Promise<{ acquired: boolean }> {
    const expiresAt = Date.now() + ttlMs;
 
    const acquired = await this.ctx.storage.transaction(async () => {
      const existing = [...this.ctx.storage.sql.exec(
        "SELECT expires_at FROM lock_state LIMIT 1"
      )][0] as { expires_at: number } | undefined;
 
      if (existing && existing.expires_at > Date.now()) {
        return false;
      }
 
      // 만료됐거나 없는 경우 — 교체 삽입
      this.ctx.storage.sql.exec("DELETE FROM lock_state");
      this.ctx.storage.sql.exec(
        "INSERT INTO lock_state (owner, expires_at) VALUES (?, ?)",
        owner,
        expiresAt
      );
      return true;
    });
 
    if (acquired) {
      // TTL 만료 시 알람으로 자동 정리 — 외부 cron 불필요
      await this.ctx.storage.setAlarm(expiresAt + 100);
    }
 
    return { acquired };
  }
 
  async release(owner: string): Promise<{ released: boolean }> {
    return await this.ctx.storage.transaction(async () => {
      const existing = [...this.ctx.storage.sql.exec(
        "SELECT owner FROM lock_state LIMIT 1"
      )][0] as { owner: string } | undefined;
 
      if (!existing || existing.owner !== owner) {
        return { released: false };
      }
 
      this.ctx.storage.sql.exec("DELETE FROM lock_state");
      return { released: true };
    });
  }
 
  async renew(
    owner: string,
    ttlMs = DistributedLock.DEFAULT_TTL_MS
  ): Promise<{ renewed: boolean }> {
    const expiresAt = Date.now() + ttlMs;
    const result = [...this.ctx.storage.sql.exec(
      "UPDATE lock_state SET expires_at = ? WHERE owner = ? RETURNING owner",
      expiresAt,
      owner
    )];
 
    if (result.length > 0) {
      // 갱신된 만료 시간에 맞춰 알람도 재설정
      await this.ctx.storage.setAlarm(expiresAt + 100);
      return { renewed: true };
    }
    return { renewed: false };
  }
 
  async alarm() {
    this.ctx.storage.sql.exec(
      "DELETE FROM lock_state WHERE expires_at <= ?",
      Date.now()
    );
  }
}

// lock-service.ts — RPC 노출용 퍼사드
import { WorkerEntrypoint } from "cloudflare:workers";
 
interface Env {
  DISTRIBUTED_LOCK: DurableObjectNamespace<DistributedLock>;
}
 
export class LockService extends WorkerEntrypoint<Env> {
  async acquire(resource: string, owner: string, ttlMs?: number) {
    const stub = this.env.DISTRIBUTED_LOCK.get(
      this.env.DISTRIBUTED_LOCK.idFromName(resource)
    );
    return stub.acquire(owner, ttlMs);
  }
 
  async release(resource: string, owner: string) {
    const stub = this.env.DISTRIBUTED_LOCK.get(
      this.env.DISTRIBUTED_LOCK.idFromName(resource)
    );
    return stub.release(owner);
  }
 
  async renew(resource: string, owner: string, ttlMs?: number) {
    const stub = this.env.DISTRIBUTED_LOCK.get(
      this.env.DISTRIBUTED_LOCK.idFromName(resource)
    );
    return stub.renew(owner, ttlMs);
  }
}

// 호출하는 Worker 쪽
export default {
  async fetch(request: Request, env: Env) {
    // HTTP 직렬화 없이 타입 안전한 호출
    const { acquired } = await env.LOCK_SERVICE.acquire("order-42", "worker-abc");
    if (!acquired) {
      return new Response("락 획득 실패", { status: 409 });
    }
    // ... 임계 구역 처리 ...
    await env.LOCK_SERVICE.release("order-42", "worker-abc");
    return new Response("완료");
  },
};

// 모든 요청이 하나의 인스턴스로 몰림 — 병목
const id = env.CHAT_ROOM.idFromName("global-singleton");
 
// 엔티티를 키로 인스턴스를 분산 — 이렇게 써야 합니다
const id = env.CHAT_ROOM.idFromName(roomId);

// 스토리지 원자성은 transaction()으로
await this.ctx.storage.transaction(async () => {
  const row = [...this.ctx.storage.sql.exec("SELECT count FROM counters")][0];
  this.ctx.storage.sql.exec(
    "UPDATE counters SET count = ?",
    (row?.count as number ?? 0) + 1
  );
});
 
// blockConcurrencyWhile의 올바른 사용처 — 외부 I/O를 포함한 초기화
async initialize() {
  await this.ctx.blockConcurrencyWhile(async () => {
    // 외부 API에서 설정을 받아오는 동안 다른 요청이 진입하지 못하게 차단
    const config = await fetch("https://config.example.com/settings").then(r => r.json());
    this.config = config;
  });
}