Durable Objects — Workers 上的有状态实时应用

import { DurableObject } from 'cloudflare:workers';

interface Env {
  CHAT_ROOM: DurableObjectNamespace;
}

export class ChatRoom extends DurableObject {
  private sessions = new Map<string, WebSocket>();

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    const name = url.searchParams.get('name') || 'anonymous';

    // WebSocket upgrade
    const pair = new WebSocketPair();
    const [client, server] = Object.values(pair);

    server.accept();
    this.sessions.set(name, server);

    // Broadcast welcome
    this.broadcast(`${name} joined the chat`);

    server.addEventListener('message', (event) => {
      this.broadcast(`${name}: ${event.data}`, server);
    });

    server.addEventListener('close', () => {
      this.sessions.delete(name);
      this.broadcast(`${name} left`);
    });

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

  private broadcast(message: string, sender?: WebSocket) {
    const data = JSON.stringify({ message, timestamp: Date.now() });
    this.sessions.forEach((ws, name) => {
      if (ws !== sender && ws.readyState === WebSocket.OPEN) {
        ws.send(data);
      }
    });
  }
}

// Worker — router 到 DO
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    const roomId = url.searchParams.get('room') || 'default';
    const id = env.CHAT_ROOM.idFromName(roomId);
    const stub = env.CHAT_ROOM.get(id);
    return stub.fetch(request);
  },
};

export class Counter extends DurableObject {
  private storage: DurableObjectStorage;

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    this.storage = ctx.storage;
  }

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);

    switch (url.pathname) {
      case '/increment':
        const count = (await this.storage.get<number>('count')) || 0;
        await this.storage.put('count', count + 1);
        return Response.json({ count: count + 1 });

      case '/get':
        const current = (await this.storage.get<number>('count')) || 0;
        return Response.json({ count: current });

      case '/reset':
        await this.storage.put('count', 0);
        return Response.json({ count: 0 });

      default:
        return new Response('Not found', { status: 404 });
    }
  }
}

export class TodoApp extends DurableObject {
  async fetch(request: Request): Promise<Response> {
    const sql = this.ctx.storage.sql;

    // Create table
    sql.exec(`
      CREATE TABLE IF NOT EXISTS todos (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        completed INTEGER DEFAULT 0
      )
    `);

    const url = new URL(request.url);

    // List
    if (request.method === 'GET') {
      const result = sql.exec('SELECT * FROM todos ORDER BY id DESC');
      return Response.json(result.toArray());
    }

    // Create
    if (request.method === 'POST') {
      const { title } = await request.json();
      const result = sql.exec('INSERT INTO todos (title) VALUES (?)', title);
      return Response.json({ id: result.lastRowId, title, completed: false }, { status: 201 });
    }

    // Update
    if (request.method === 'PUT') {
      const { id, completed } = await request.json();
      sql.exec('UPDATE todos SET completed = ? WHERE id = ?', completed, id);
      return Response.json({ success: true });
    }

    // Delete
    if (request.method === 'DELETE') {
      const id = url.searchParams.get('id');
      sql.exec('DELETE FROM todos WHERE id = ?', id);
      return Response.json({ success: true });
    }

    return new Response('Method not allowed', { status: 405 });
  }
}

export class Reminder extends DurableObject {
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    ctx.blockConcurrencyWhile(async () => {
      const alarm = await ctx.storage.getAlarm();
      if (alarm) {
        console.log('Recovered alarm:', new Date(alarm).toISOString());
      }
    });
  }

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    const delayMs = parseInt(url.searchParams.get('delay') || '5000');

    // Set alarm sau delayMs
    await this.ctx.storage.setAlarm(Date.now() + delayMs);

    return Response.json({ message: `Alarm set for ${delayMs}ms` });
  }

  async alarm() {
    // Được gọi khi alarm hết giờ
    console.log('⏰ Alarm fired!');

    // Gửi notification qua WebSocket nếu có
    // Hoặc call webhook
    await fetch('https://hooks.example.com/notify', {
      method: 'POST',
      body: JSON.stringify({ event: 'alarm_fired', time: Date.now() }),
    });

    // Set alarm lại nếu cần
    await this.ctx.storage.setAlarm(Date.now() + 3600000);
  }
}

interface Player {
  id: string;
  x: number;
  y: number;
  score: number;
}

export class GameRoom extends DurableObject {
  private players = new Map<string, WebSocket>();
  private state: Player[] = [];

  async fetch(request: Request): Promise<Response> {
    const pair = new WebSocketPair();
    const [client, server] = Object.values(pair);

    server.accept();

    server.addEventListener('message', (event) => {
      const msg = JSON.parse(event.data as string);

      switch (msg.type) {
        case 'join':
          this.players.set(msg.playerId, server);
          this.state.push({ id: msg.playerId, x: 0, y: 0, score: 0 });
          this.broadcast({ type: 'players', data: this.state });
          break;

        case 'move':
          const player = this.state.find(p => p.id === msg.playerId);
          if (player) {
            player.x += msg.dx;
            player.y += msg.dy;
            this.broadcast({ type: 'move', playerId: msg.playerId, x: player.x, y: player.y });
          }
          break;

        case 'shoot':
          this.broadcast({
            type: 'shoot',
            playerId: msg.playerId,
            x: msg.x,
            y: msg.y,
            angle: msg.angle,
          });
          break;
      }
    });

    server.addEventListener('close', () => {
      this.state = this.state.filter(p => !this.players.has(p.id));
      this.players.delete(msg.playerId);
      this.broadcast({ type: 'leave', playerId: msg.playerId });
    });

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

  private broadcast(message: object) {
    const data = JSON.stringify(message);
    this.players.forEach((ws) => {
      if (ws.readyState === WebSocket.OPEN) {
        ws.send(data);
      }
    });
  }
}

// Shard user data theo userId
function getUserStub(userId: number, env: Env): DurableObjectStub {
  const shardId = Math.floor(userId / 1000); // 1000 user mỗi shard
  const id = env.USER_STORE.idFromName(`shard-${shardId}`);
  return env.USER_STORE.get(id);
}

// Handler
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    const userId = parseInt(url.searchParams.get('userId') || '0');
    const stub = getUserStub(userId, env);
    return stub.fetch(request);
  },
};