Cloudflare Workers Durable Objects:分散協調、WebSockets、エッジステート

.
├── src
│   ├── index.ts          # Worker entry point, routes requests to DOs
│   ├── chatRoom.ts       # Durable Object for a chat room
│   └── types.ts          # Shared types
├── wrangler.toml         # Cloudflare Workers configuration
└── package.json

name = "chat-service"
main = "src/index.ts"
compatibility_date = "2024-01-01"

[[durable_objects.bindings]]
name = "CHAT_ROOM"
class_name = "ChatRoom"
script_name = "chat-service" # Optional: if DO is in a different worker

export interface ChatMessage {
  sender: string;
  message: string;
  timestamp: number;
}

export interface WebSocketMessage {
  type: 'chat' | 'join' | 'leave';
  payload: ChatMessage | { user: string };
}

import { ChatMessage, WebSocketMessage } from './types';

// Maximum number of messages to store in memory and persist
const MAX_MESSAGES = 100;
// Key for storing messages in Durable Object storage
const MESSAGES_STORAGE_KEY = 'messages';

export class ChatRoom {
  private state: DurableObjectState;
  private env: Env; // Environment variables, if any
  private sessions: WebSocket[] = []; // Active WebSocket connections
  private messages: ChatMessage[] = []; // In-memory message buffer

  constructor(state: DurableObjectState, env: Env) {
    this.state = state;
    this.env = env;

    // Restore messages from storage on DO activation
    this.state.blockConcurrencyWhile(async () => {
      const storedMessages = await this.state.storage.get<ChatMessage[]>(MESSAGES_STORAGE_KEY);
      if (storedMessages) {
        this.messages = storedMessages;
      }
      console.log(`ChatRoom ${this.state.id.toString()} initialized with ${this.messages.length} messages.`);
    });
  }

  // Handles incoming HTTP requests, primarily WebSocket upgrades
  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);

    if (url.pathname === '/websocket') {
      // Upgrade the request to a WebSocket connection
      const { 0: client, 1: server } = new WebSocketPair();

      this.handleWebSocket(server);

      return new Response(null, { status: 101, webSocket: client });
    } else if (url.pathname === '/messages') {
      // Return recent messages via HTTP for initial load or REST clients
      return new Response(JSON.stringify(this.messages), {
        headers: { 'Content-Type': 'application/json' },
      });
    }

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

  // Handles the lifecycle of a single WebSocket connection
  private async handleWebSocket(server: WebSocket) {
    server.accept();
    this.sessions.push(server);

    // Send historical messages to the new client
    server.send(JSON.stringify({ type: 'history', payload: this.messages }));

    server.addEventListener('message', async (event) => {
      try {
        const message: WebSocketMessage = JSON.parse(event.data as string);
        if (message.type === 'chat') {
          const chatMessage: ChatMessage = {
            ...message.payload as ChatMessage,
            timestamp: Date.now(),
          };
          this.broadcastMessage(chatMessage);
          await this.storeMessage(chatMessage);
        } else if (message.type === 'join') {
          const user = (message.payload as { user: string }).user;
          this.broadcastSystemMessage(`${user} joined the room.`);
        }
      } catch (err) {
        console.error('WebSocket message parsing error:', err);
        server.send(JSON.stringify({ type: 'error', payload: 'Invalid message format.' }));
      }
    });

    server.addEventListener('close', async (event) => {
      console.log(`WebSocket closed: ${event.code} ${event.reason}`);
      this.sessions = this.sessions.filter(s => s !== server);
      // Optionally broadcast leave message
    });

    server.addEventListener('error', (err) => {
      console.error('WebSocket error:', err);
      server.close(1011, 'Internal Error');
    });
  }

  // Broadcasts a message to all connected clients
  private broadcastMessage(message: ChatMessage) {
    const wsMessage: WebSocketMessage = { type: 'chat', payload: message };
    const serializedMessage = JSON.stringify(wsMessage);
    this.sessions.forEach(session => {
      try {
        session.send(serializedMessage);
      } catch (err) {
        console.error('Failed to send message to session:', err);
        // Consider removing broken sessions here
      }
    });
  }

  // Broadcasts a system message (e.g., join/leave)
  private broadcastSystemMessage(text: string) {
    const systemMessage: ChatMessage = { sender: 'System', message: text, timestamp: Date.now() };
    const wsMessage: WebSocketMessage = { type: 'chat', payload: systemMessage };
    const serializedMessage = JSON.stringify(wsMessage);
    this.sessions.forEach(session => {
      try {
        session.send(serializedMessage);
      } catch (err) {
        console.error('Failed to send system message to session:', err);
      }
    });
  }

  // Stores a message in memory and persists it to Durable Object storage
  private async storeMessage(message: ChatMessage) {
    this.messages.push(message);
    if (this.messages.length > MAX_MESSAGES) {
      this.messages.shift(); // Keep only the latest MAX_MESSAGES
    }

    // Persist messages transactionally
    await this.state.storage.transaction(async (txn) => {
      txn.put(MESSAGES_STORAGE_KEY, this.messages);
    });

    // Set an alarm to prune old messages or perform other background tasks
    // This demonstrates using alarms for cron-like functionality
    const currentAlarm = await this.state.storage.getAlarm();
    if (currentAlarm === null || currentAlarm < Date.now() + 60 * 1000) { // Set alarm for 1 minute from now
      await this.state.storage.setAlarm(Date.now() + 60 * 1000);
    }
  }

  // Alarm handler for background tasks
  async alarm() {
    console.log(`Alarm triggered for ChatRoom ${this.state.id.toString()}. Pruning messages.`);
    // Example: Prune messages older than 24 hours
    const twentyFourHoursAgo = Date.now() - (24 * 60 * 60 * 1000);
    const initialLength = this.messages.length;
    this.messages = this.messages.filter(msg => msg.timestamp > twentyFourHoursAgo);

    if (this.messages.length < initialLength) {
      // Only persist if messages were actually pruned
      await this.state.storage.transaction(async (txn) => {
        txn.put(MESSAGES_STORAGE_KEY, this.messages);
      });
      console.log(`Pruned ${initialLength - this.messages.length} messages.`);
    }

    // Optionally reschedule the alarm for the next pruning cycle
    await this.state.storage.setAlarm(Date.now() + 24 * 60 * 60 * 1000); // Reschedule for 24 hours later
  }
}

// Define the environment interface for type safety
interface Env {
  CHAT_ROOM: DurableObjectNamespace;
}

import { ChatRoom } from './chatRoom';

interface Env {
  CHAT_ROOM: DurableObjectNamespace;
}

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    const url = new URL(request.url);
    const roomId = url.pathname.split('/')[1] || 'default-room'; // Extract room ID from path

    // Get a Durable Object ID for the room
    const id = env.CHAT_ROOM.idFromName(roomId);

    // Get the Durable Object stub
    const stub = env.CHAT_ROOM.get(id);

    // Forward the request to the Durable Object
    return stub.fetch(request);
  },
};

// Export the Durable Object class for Wrangler to discover
export { ChatRoom };

await this.state.storage.transaction(async (txn) => {
  txn.put(MESSAGES_STORAGE_KEY, this.messages);
});

// Setting an alarm
await this.state.storage.setAlarm(Date.now() + 60 * 1000); // 1 minute from now

// Alarm handler
async alarm() {
  // ... background task logic ...
  await this.state.storage.setAlarm(Date.now() + 24 * 60 * 60 * 1000); // Reschedule
}