Cloudflare Durable Objects

import { DurableObject } from 'cloudflare:workers';

export class AgentSocket extends DurableObject<Env> {
  async fetch(request: Request): Promise<Response> {
    // WebSocket 업그레이드 처리
  }

  async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): Promise<void> {
    // 메시지 수신 처리
  }

  async webSocketClose(ws: WebSocket, code: number, reason: string, wasClean: boolean): Promise<void> {
    // 연결 종료 처리
  }

  async webSocketError(ws: WebSocket, error: unknown): Promise<void> {
    // 에러 처리
  }
}

async fetch(request: Request): Promise<Response> {
  const upgradeHeader = request.headers.get('Upgrade');
  if (!upgradeHeader || upgradeHeader.toLowerCase() !== 'websocket') {
    return new Response('Expected WebSocket upgrade', { status: 426 });
  }

  // WebSocketPair 생성 (client ↔ server)
  const pair = new WebSocketPair();
  const [client, server] = Object.values(pair);

  // Hibernation API로 수락 (태그로 식별자 부여)
  this.ctx.acceptWebSocket(server, ['my-tag']);

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

async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): Promise<void> {
  if (typeof message === 'string') {
    try {
      const data = JSON.parse(message) as { type?: string };

      if (data.type === 'ping') {
        ws.send(JSON.stringify({ type: 'pong' }));
        return;
      }

      // 비즈니스 로직 처리
      ws.send(JSON.stringify({ type: 'ack', received: data }));
    } catch {
      // JSON이 아닌 메시지
    }
  }
}

async webSocketClose(ws: WebSocket, code: number, reason: string, _wasClean: boolean): Promise<void> {
  const tags = this.ctx.getTags(ws);
  const id = tags[0];
  // 정리 로직 (DB 상태 업데이트 등)
  ws.close(code, reason);
}

async webSocketError(ws: WebSocket, _error: unknown): Promise<void> {
  const tags = this.ctx.getTags(ws);
  const id = tags[0];
  // 에러 처리 로직
  ws.close(1011, 'Unexpected error');
}

import { Hono } from 'hono';

export { AgentSocket } from './agent-socket';

const app = new Hono<{ Bindings: Env }>();

app.get('/ws/:id', async (c) => {
  const id = c.req.param('id');

  // DO 인스턴스 획득 (같은 id → 같은 인스턴스)
  const doId = c.env.AGENT_SOCKET.idFromName(id);
  const stub = c.env.AGENT_SOCKET.get(doId);

  // 원본 요청을 DO로 전달 (헤더 포함)
  return stub.fetch(
    new Request(c.req.url, {
      headers: c.req.raw.headers,
    })
  );
});

export default app;

const TICKET_TTL_MS = 30_000; // 30초

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

  // 티켓 저장 엔드포인트
  if (url.pathname.endsWith('/ticket')) {
    const { ticket, agentId } = await request.json();
    await this.ctx.storage.put(`ticket:${ticket}`, agentId);
    await this.ctx.storage.setAlarm(Date.now() + TICKET_TTL_MS);
    return new Response('ok');
  }

  // WebSocket 업그레이드 시 티켓 검증
  const ticket = url.searchParams.get('ticket');
  if (!ticket) return new Response('Missing ticket', { status: 400 });

  const agentId = await this.ctx.storage.get<string>(`ticket:${ticket}`);
  if (!agentId) return new Response('Invalid or expired ticket', { status: 401 });

  // 일회용: 즉시 삭제
  await this.ctx.storage.delete(`ticket:${ticket}`);

  // ... WebSocket 업그레이드 진행
}

// 만료된 티켓 정리
async alarm(): Promise<void> {
  const entries = await this.ctx.storage.list({ prefix: 'ticket:' });
  if (entries.size > 0) {
    await this.ctx.storage.delete([...entries.keys()]);
  }
}

broadcast(message: string, exclude?: WebSocket): void {
  const sockets = this.ctx.getWebSockets();
  for (const ws of sockets) {
    if (ws !== exclude) {
      ws.send(message);
    }
  }
}

import { UserProfile } from "./generated/user"; // ts-proto로 생성된 파일

export class UserDO extends DurableObject {
  async saveUser(data: UserProfile) {
    // 1. 데이터를 바이너리로 인코딩
    const encoded = UserProfile.encode(data).finish();
    // 2. DO Storage에 직접 저장 (바이너리 저장이 JSON보다 저렴)
    await this.ctx.storage.put("user_data", encoded);
  }

  async getUser(): Promise<UserProfile | undefined> {
    const stored = await this.ctx.storage.get<Uint8Array>("user_data");
    if (!stored) return undefined;
    // 3. 디코딩하여 객체로 복원
    return UserProfile.decode(stored);
  }
}

export class MyChatRoom extends DurableObject {
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
  }

  async fetch(request: Request) {
    // 1. WebSocket 업그레이드 요청 확인
    const webSocketPair = new WebSocketPair();
    const [client, server] = Object.values(webSocketPair);

    // 2. 서버 측 소켓 수락 및 'Hibernation' 모드 활성화
    this.ctx.acceptWebSocket(server);

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

  // 3. 메시지가 도착했을 때만 DO가 깨어나서 이 메서드를 실행합니다.
  async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
    // 메시지 처리 및 브로드캐스트
    this.ctx.getWebSockets().forEach(peer => {
      if (peer !== ws) peer.send(`상대방: ${message}`);
    });
  }

  // 4. 연결이 끊겼을 때 실행
  async webSocketClose(ws: WebSocket, code: number, reason: string, wasClean: boolean) {
    console.log("연결 종료");
  }
}