Cloudflare Workers + D1 + Queues로 단일 파일에서 팬아웃과 재시도 구성하기

name = "my-worker"
main = "src/worker.ts"
compatibility_date = "2024-09-23"
 
[[queues.producers]]
binding = "MY_QUEUE"
queue = "my-job-queue"
 
[[queues.consumers]]
queue = "my-job-queue"
max_batch_size = 10
max_batch_timeout = 5
max_retries = 3
dead_letter_queue = "my-job-dlq"
 
[[queues.consumers]]
queue = "my-job-dlq"
max_batch_size = 5
max_batch_timeout = 30
 
[[d1_databases]]
binding = "DB"
database_name = "jobs-db"
database_id = "<your-database-id>"

-- migrations/0001_create_jobs.sql
CREATE TABLE IF NOT EXISTS jobs (
  id TEXT PRIMARY KEY,
  type TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending',
  payload TEXT,
  error TEXT,
  created_at INTEGER NOT NULL DEFAULT (unixepoch()),
  updated_at INTEGER NOT NULL DEFAULT (unixepoch())
);
 
CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status);
CREATE INDEX IF NOT EXISTS idx_jobs_status_updated ON jobs(status, updated_at);
 
CREATE TABLE IF NOT EXISTS webhook_deliveries (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  endpoint_id TEXT NOT NULL,
  event_type TEXT NOT NULL,
  status TEXT NOT NULL,
  error TEXT,
  created_at INTEGER NOT NULL DEFAULT (unixepoch())
);

// src/worker.ts
export interface Env {
  MY_QUEUE: Queue<JobMessage>;
  DB: D1Database;
}
 
export interface JobMessage {
  id: string;
  type: 'send_email' | 'webhook_fanout' | 'process_file';
  payload: unknown;
}
 
export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    if (req.method !== 'POST') {
      return new Response('Method Not Allowed', { status: 405 });
    }
 
    const body = (await req.json()) as {
      type: JobMessage['type'];
      payload: unknown;
    };
    const job: JobMessage = {
      id: crypto.randomUUID(),
      type: body.type,
      payload: body.payload,
    };
 
    // D1과 Queues는 단일 트랜잭션으로 묶이지 않으므로 D1을 먼저 기록
    await env.DB.prepare(
      'INSERT INTO jobs (id, type, status, payload) VALUES (?, ?, ?, ?)'
    )
      .bind(job.id, job.type, 'pending', JSON.stringify(job.payload))
      .run();
 
    await env.MY_QUEUE.send(job);
 
    return Response.json({ jobId: job.id }, { status: 202 });
  },
 
  async queue(
    batch: MessageBatch<JobMessage>,
    env: Env
  ): Promise<void> {
    if (batch.queue === 'my-job-dlq') {
      await handleDlqBatch(batch, env);
      return;
    }
    await handleMainBatch(batch, env);
  },
};

async function handleMainBatch(
  batch: MessageBatch<JobMessage>,
  env: Env
): Promise<void> {
  for (const message of batch.messages) {
    const job = message.body;
 
    // 낙관적 잠금: pending 상태일 때만 processing으로 전환
    const claim = await env.DB.prepare(
      `UPDATE jobs
         SET status = 'processing', updated_at = unixepoch()
       WHERE id = ? AND status = 'pending'`
    )
      .bind(job.id)
      .run();
 
    // 다른 인스턴스가 이미 잡았거나 완료된 경우 — 중복 처리 방지
    if (claim.meta.changes === 0) {
      message.ack();
      continue;
    }
 
    let processed = false;
    try {
      await processJob(job, env);
      processed = true;
    } catch (err) {
      const errorMsg = err instanceof Error ? err.message : String(err);
      await safeUpdateStatus(env, job.id, 'pending', errorMsg);
      // 지수 백오프 지연 재시도 — attempts는 메시지 메타에서 제공
      message.retry({ delaySeconds: backoffSeconds(message.attempts) });
      continue;
    }
 
    try {
      await env.DB.prepare(
        `UPDATE jobs
            SET status = 'done', error = NULL, updated_at = unixepoch()
          WHERE id = ?`
      )
        .bind(job.id)
        .run();
      message.ack();
    } catch (dbErr) {
      // 작업은 성공했지만 상태 기록만 실패 — 메시지는 ack해서 재실행을 막고,
      // 상태 복구는 뒤의 보상 로직이 스캔으로 처리
      console.error('status update failed after success', job.id, dbErr);
      message.ack();
    }
 
    void processed;
  }
}
 
async function safeUpdateStatus(
  env: Env,
  id: string,
  status: string,
  error: string | null
): Promise<void> {
  try {
    await env.DB.prepare(
      `UPDATE jobs
          SET status = ?, error = ?, updated_at = unixepoch()
        WHERE id = ?`
    )
      .bind(status, error, id)
      .run();
  } catch (e) {
    console.error('status update failed', id, e);
  }
}
 
function backoffSeconds(attempts: number): number {
  // 10s, 20s, 40s ... 최대 300s
  return Math.min(300, 10 * 2 ** Math.max(0, attempts - 1));
}
 
async function processJob(job: JobMessage, env: Env): Promise<void> {
  switch (job.type) {
    case 'send_email':
      await sendEmail(job.payload, env);
      break;
    case 'webhook_fanout':
      await fanoutWebhooks(job.payload as WebhookFanoutPayload, env);
      break;
    case 'process_file':
      await processFile(job.payload, env);
      break;
    default:
      throw new Error(`Unknown job type: ${(job as JobMessage).type}`);
  }
}

interface WebhookEndpoint {
  id: string;
  url: string;
}
 
interface WebhookFanoutPayload {
  eventType: string;
  eventData: unknown;
  endpoints: WebhookEndpoint[];
}
 
async function fanoutWebhooks(
  payload: WebhookFanoutPayload,
  env: Env
): Promise<void> {
  // Promise.allSettled: 하나가 실패해도 나머지는 계속 전송
  const results = await Promise.allSettled(
    payload.endpoints.map((endpoint) =>
      sendWebhook(endpoint, payload.eventType, payload.eventData)
    )
  );
 
  // D1 batch API로 여러 INSERT를 한 번의 왕복으로 처리
  const stmt = env.DB.prepare(
    'INSERT INTO webhook_deliveries (endpoint_id, event_type, status, error) VALUES (?, ?, ?, ?)'
  );
 
  await env.DB.batch(
    results.map((result, i) =>
      stmt.bind(
        payload.endpoints[i].id,
        payload.eventType,
        result.status === 'fulfilled' ? 'delivered' : 'failed',
        result.status === 'rejected' ? String(result.reason) : null
      )
    )
  );
}
 
async function sendWebhook(
  endpoint: WebhookEndpoint,
  eventType: string,
  data: unknown
): Promise<void> {
  const res = await fetch(endpoint.url, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ event: eventType, data, timestamp: Date.now() }),
  });
 
  if (!res.ok) {
    throw new Error(`Webhook failed: ${res.status} ${res.statusText}`);
  }
}

async function handleDlqBatch(
  batch: MessageBatch<JobMessage>,
  env: Env
): Promise<void> {
  for (const message of batch.messages) {
    await env.DB.prepare(
      `UPDATE jobs
          SET status = 'dead', updated_at = unixepoch()
        WHERE id = ?`
    )
      .bind(message.body.id)
      .run();
 
    await notifyOpsTeam(message.body); // 개념적 예시: Slack, PagerDuty 등
 
    // DLQ 메시지도 반드시 ack — 하지 않으면 재처리 루프에 빠짐
    message.ack();
  }
}

# wrangler.toml에 추가
[triggers]
crons = ["*/5 * * * *"]

// 같은 Worker 파일에 scheduled 핸들러 추가
export default {
  async fetch(/* ... */) { /* 위와 동일 */ },
  async queue(/* ... */) { /* 위와 동일 */ },
 
  async scheduled(_event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
    ctx.waitUntil(recoverStuckJobs(env));
  },
};
 
async function recoverStuckJobs(env: Env): Promise<void> {
  // 1) 큐 발행이 유실된 pending — 생성된 지 2분 넘게 pending인 잡
  const pendingStuck = await env.DB.prepare(
    `SELECT id, type, payload FROM jobs
      WHERE status = 'pending'
        AND created_at < unixepoch() - 120
      LIMIT 100`
  ).all<{ id: string; type: JobMessage['type']; payload: string }>();
 
  // 2) processing으로 고착된 잡 — 10분 넘게 갱신 없는 잡을 pending으로 되돌림
  await env.DB.prepare(
    `UPDATE jobs
        SET status = 'pending', updated_at = unixepoch()
      WHERE status = 'processing'
        AND updated_at < unixepoch() - 600`
  ).run();
 
  const stuckProcessing = await env.DB.prepare(
    `SELECT id, type, payload FROM jobs
      WHERE status = 'pending'
        AND updated_at >= unixepoch() - 30
      LIMIT 100`
  ).all<{ id: string; type: JobMessage['type']; payload: string }>();
 
  const candidates = [
    ...(pendingStuck.results ?? []),
    ...(stuckProcessing.results ?? []),
  ];
 
  for (const row of candidates) {
    await env.MY_QUEUE.send({
      id: row.id,
      type: row.type,
      payload: JSON.parse(row.payload),
    });
  }
}