Cloudflare D1과 Durable Objects: 분산 SQLite로 엣지에서 강한 일관성 경계를 설계하는 기준

// D1 기본 쿼리 — 가장 가까운 복제본에서 읽기
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const products = await env.DB.prepare(
      "SELECT id, name, price FROM products WHERE active = 1"
    ).all();
 
    return Response.json(products.results);
  },
};

// 세션 일관성이 필요한 경우 — 쓰기 직후 즉시 읽기
const session = env.DB.withSession("first-primary");
 
await session
  .prepare("INSERT INTO cart_items (user_id, product_id) VALUES (?, ?)")
  .bind(userId, productId)
  .run();
 
// 동일 세션 → 방금 쓴 데이터가 반드시 읽힘
const cart = await session
  .prepare("SELECT * FROM cart_items WHERE user_id = ?")
  .bind(userId)
  .all();

// src/routes/products.ts
import { Hono } from "hono";
import { drizzle } from "drizzle-orm/d1";
import { products, inventory, cartItems } from "../schema";
import { eq } from "drizzle-orm";
 
const app = new Hono<{ Bindings: Env }>();
 
// 상품 목록 — 가장 가까운 복제본에서 읽기 (빠름)
app.get("/products", async (c) => {
  const db = drizzle(c.env.DB);
  const items = await db
    .select()
    .from(products)
    .where(eq(products.active, true));
  return c.json(items);
});
 
// 장바구니 추가 후 즉시 확인 — 세션 일관성 필요
app.post("/cart", async (c) => {
  const { productId, userId } = await c.req.json();
 
  // "first-primary": 쓰기와 동일한 Primary 읽기 일관성 보장
  const session = c.env.DB.withSession("first-primary");
  const db = drizzle(session);
 
  // 최신 재고 확인 (복제본 지연 허용 불가)
  const [item] = await db
    .select()
    .from(inventory)
    .where(eq(inventory.productId, productId));
 
  if (item.stock < 1) {
    return c.json({ error: "재고 없음" }, 409);
  }
 
  // 재고 차감 + 장바구니 추가 원자적 실행
  await db.batch([
    db
      .update(inventory)
      .set({ stock: item.stock - 1 })
      .where(eq(inventory.productId, productId)),
    db.insert(cartItems).values({ userId, productId }),
  ]);
 
  // 동일 세션 → 방금 추가한 항목이 반드시 조회됨
  const cart = await db
    .select()
    .from(cartItems)
    .where(eq(cartItems.userId, userId));
 
  return c.json(cart);
});
 
export default app;

// src/durable-objects/document.ts
import { DurableObject } from "cloudflare:workers";
 
interface EditOperation {
  userId: string;
  position: number;
  text: string;
  type: "insert" | "delete";
  timestamp: number;
}
 
interface Operation {
  op_type: string;
  position: number;
  content: string | null;
}
 
export class DocumentDO extends DurableObject {
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
 
    this.ctx.blockConcurrencyWhile(async () => {
      this.ctx.storage.sql.exec(`
        CREATE TABLE IF NOT EXISTS operations (
          id        INTEGER PRIMARY KEY AUTOINCREMENT,
          user_id   TEXT    NOT NULL,
          op_type   TEXT    NOT NULL,
          position  INTEGER NOT NULL,
          content   TEXT,
          applied_at INTEGER NOT NULL
        )
      `);
    });
  }
 
  async fetch(request: Request): Promise<Response> {
    if (request.headers.get("Upgrade") === "websocket") {
      const [client, server] = Object.values(new WebSocketPair());
      // acceptWebSocket + getWebSockets() → Hibernation API 패턴
      this.ctx.acceptWebSocket(server);
      return new Response(null, { status: 101, webSocket: client });
    }
    return new Response("Not found", { status: 404 });
  }
 
  // Hibernation API: WebSocket 메시지는 이 메서드로 라우팅됨
  async webSocketMessage(ws: WebSocket, message: string): Promise<void> {
    const op: EditOperation = JSON.parse(message);
 
    // 단일 스레드 — 동시 편집 요청이 와도 순서대로 처리됨
    this.ctx.storage.sql.exec(
      `INSERT INTO operations (user_id, op_type, position, content, applied_at)
       VALUES (?, ?, ?, ?, ?)`,
      op.userId,
      op.type,
      op.position,
      op.text,
      op.timestamp
    );
 
    const ops = [
      ...this.ctx.storage.sql.exec<Operation>(
        `SELECT op_type, position, content FROM operations ORDER BY applied_at ASC`
      ),
    ];
 
    // ⚠️ 단순화된 데모 구현 (O(n) 재적용)
    // 실제 프로덕션에서는 OT(Operational Transformation)나 CRDT가 필요합니다
    const content = this.reconstructDocument(ops);
 
    for (const socket of this.ctx.getWebSockets()) {
      if (socket !== ws) {
        socket.send(JSON.stringify({ type: "op_applied", op, content }));
      }
    }
  }
 
  private reconstructDocument(ops: Operation[]): string {
    let doc = "";
    for (const op of ops) {
      if (op.op_type === "insert") {
        doc = doc.slice(0, op.position) + op.content + doc.slice(op.position);
      } else {
        doc =
          doc.slice(0, op.position) +
          doc.slice(op.position + (op.content?.length ?? 0));
      }
    }
    return doc;
  }
}
 
// src/index.ts — Worker에서 DO 라우팅
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    const docId = url.searchParams.get("docId");
    if (!docId) return new Response("docId required", { status: 400 });
 
    // 동일 docId → 항상 동일 DO 인스턴스로 라우팅
    const id = env.DOCUMENT_DO.idFromName(`doc:${docId}`);
    return env.DOCUMENT_DO.get(id).fetch(request);
  },
};

┌─────────────────────────────────────────────────────┐
│                  Cloudflare Workers                 │
│                                                     │
│   요청 라우팅                                        │
│      │                                              │
│      ├─── 공유 데이터 ──────▶ D1 글로벌 DB          │
│      │    (회원, 플랜, 청구)   (읽기 복제본 분산)    │
│      │                                              │
│      ├─── 테넌트 격리 데이터 ▶ DO per Tenant        │
│      │    (테넌트별 SQLite)    (독립 인스턴스)       │
│      │                                              │
│      └─── API 레이트 리밋 ──▶ DO per API Key        │
│           (슬라이딩 윈도우)   (단일 스레드 카운터)   │
└─────────────────────────────────────────────────────┘

// src/durable-objects/rate-limiter.ts
import { DurableObject } from "cloudflare:workers";
 
export class RateLimiterDO extends DurableObject {
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    this.ctx.blockConcurrencyWhile(async () => {
      this.ctx.storage.sql.exec(`
        CREATE TABLE IF NOT EXISTS requests (
          id        INTEGER PRIMARY KEY AUTOINCREMENT,
          timestamp INTEGER NOT NULL
        )
      `);
    });
  }
 
  // DO 인스턴스 자체가 API 키 단위로 격리되므로 파라미터로 받을 필요 없음
  async checkLimit(limit: number, windowMs: number): Promise<boolean> {
    const now = Date.now();
    const windowStart = now - windowMs;
 
    this.ctx.storage.sql.exec(
      `DELETE FROM requests WHERE timestamp < ?`,
      windowStart
    );
 
    const result = this.ctx.storage.sql
      .exec<{ cnt: number }>(`SELECT COUNT(*) as cnt FROM requests`)
      .one();
 
    if (result.cnt >= limit) return false;
 
    this.ctx.storage.sql.exec(
      `INSERT INTO requests (timestamp) VALUES (?)`,
      now
    );
 
    return true;
  }
}
 
// src/middleware/rate-limiter.ts
import { Hono } from "hono";
import { drizzle } from "drizzle-orm/d1";
import { tenants } from "../schema";
import { eq } from "drizzle-orm";
 
const app = new Hono<{ Bindings: Env }>();
 
app.use("*", async (c, next) => {
  const apiKey = c.req.header("X-API-Key");
  if (!apiKey) return c.json({ error: "API key required" }, 401);
 
  // API 키당 DO 인스턴스 — 강한 일관성으로 정확한 카운팅
  const id = c.env.RATE_LIMITER.idFromName(`key:${apiKey}`);
  const limiter = c.env.RATE_LIMITER.get(id);
 
  const allowed = await limiter.checkLimit(100, 60_000);
  if (!allowed) return c.json({ error: "Rate limit exceeded" }, 429);
 
  // 테넌트 정보는 D1에서 읽기 (복제본 OK — 쓰기 충돌 없음)
  const db = drizzle(c.env.DB);
  const [tenant] = await db
    .select()
    .from(tenants)
    .where(eq(tenants.apiKey, apiKey));
 
  c.set("tenant", tenant);
  await next();
});
 
export default app;