分布式 Workers 的解耦设计

// API Gateway 创建 trace
const traceManager = new TraceManager();
const { traceId } = traceManager.createTrace('ai_generation');

// 入队时传递 traceId
await env.AI_QUEUE.send({
  jobId,
  payload,
  traceId  // 追踪上下文
});

// AI 处理器继承 trace
const { traceId: incomingTraceId } = message.body;
const span = traceManager.createSpan(incomingTraceId, 'fal_api_call');

aiLogger.taskStatusChange(jobId, 'queued', 'processing', {
  trace_id: traceId,
  model: payload.model,
  worker: 'ai-processor'
});

const maxRetries = 5;
const retryCount = job.retry_count || 0;
const backoffDelay = Math.pow(2, retryCount) * 1000; // 1s, 2s, 4s, 8s, 16s

if (retryCount >= maxRetries) {
  await env.DLQ.send({ jobId, error, originalMessage });
  return;
}

// 每 15 分钟运行
const zombieJobs = await supabase.rpc('get_zombie_jobs', {
  p_timeout_minutes: 30
});

for (const job of zombieJobs) {
  // 向厂商查询真实状态
  const status = await vendorAdapter.getJobStatus(job.vendor_task_id);
  await syncJobStatus(job.id, status);
}

async function verifyFalSignature(body, signature, secret) {
  const expectedSignature = await computeHmacSha256(body, secret);
  return safeCompare(signature, expectedSignature);
}

const timestamp = headers['X-Timestamp'];
const now = Date.now();
if (Math.abs(now - timestamp) > 300000) { // 5 分钟
  return { error: 'Request expired' };
}

const existing = await supabase
  .from('ai_generations')
  .select('status')
  .eq('id', jobId)
  .single();

if (existing.status === 'completed') {
  return { message: 'Already processed' };
}

# 主 Worker 配置
name = "api-gateway"
main = "src/api-gateway.js"

# 其他 Workers 通过 [[workers]] 节点定义
[[workers]]
name = "ai-processor"
main = "src/ai-processor.js"

# 全局 KV 和 Queue 绑定
[[kv_namespaces]]
binding = "MODEL_CACHE_KV"

[[queues.producers]]
queue = "ai-task-queue"
binding = "AI_QUEUE"

export default {
  async queue(batch, env) {
    const promises = batch.messages.map(msg =>
      this.handleMessage(msg, env)
    );
    await Promise.all(promises);
  }
}