17.01-为什么需要异步任务

// src/routes/tasks.ts
import { Hono } from 'hono'
 
type Bindings = {
  DB: D1Database
  BUCKET: R2Bucket
  TASK_QUEUE: Queue<TaskMessage>
}
 
interface TaskMessage {
  taskId: string
  userId: string
  fileKey: string
  fileType: string
}
 
const app = new Hono<{ Bindings: Bindings }>()
 
app.post('/api/tasks', async (c) => {
  const formData = await c.req.formData()
  const file = formData.get('file') as File
 
  if (!file) return c.json({ error: 'No file' }, 400)
 
  const taskId = crypto.randomUUID()
  const now = new Date().toISOString()
 
  // 1. 文件存 R2
  const fileKey = `uploads/${taskId}/${file.name}`
  await c.env.BUCKET.put(fileKey, file.stream(), {
    httpMetadata: { contentType: file.type },
  })
 
  // 2. 写任务记录到 D1
  await c.env.DB.prepare(`
    INSERT INTO tasks (id, user_id, file_key, file_type, status, progress, created_at, updated_at)
    VALUES (?, ?, ?, ?, 'pending', 0, ?, ?)
  `).bind(taskId, 'anonymous', fileKey, file.type, now, now).run()
 
  // 3. 发消息到队列
  await c.env.TASK_QUEUE.send({
    taskId,
    userId: 'anonymous',
    fileKey,
    fileType: file.type,
  })
 
  // 4. 立刻返回 202
  return c.json({ taskId, status: 'pending' }, 202)
})

app.get('/api/tasks/:taskId', async (c) => {
  const task = await c.env.DB.prepare(`
    SELECT id, status, progress, result_key, error_message, created_at, updated_at
    FROM tasks WHERE id = ?
  `).bind(c.req.param('taskId')).first()
 
  if (!task) return c.json({ error: 'Not found' }, 404)
  return c.json(task)
})

// src/consumers/task-consumer.ts
export async function taskConsumer(
  batch: MessageBatch<TaskMessage>,
  env: { DB: D1Database; BUCKET: R2Bucket; AI: Ai }
) {
  for (const msg of batch.messages) {
    const { taskId, fileKey, fileType } = msg.body
 
    try {
      // 更新状态为 processing
      await env.DB.prepare(
        "UPDATE tasks SET status = 'processing', updated_at = ? WHERE id = ?"
      ).bind(new Date().toISOString(), taskId).run()
 
      // 从 R2 读文件
      const object = await env.BUCKET.get(fileKey)
      if (!object) throw new Error('File not found')
 
      const buffer = await object.arrayBuffer()
 
      // 处理文件（根据类型分发）
      let result: string
      if (fileType === 'application/pdf') {
        result = await parsePdf(buffer)
      } else if (fileType.startsWith('image/')) {
        result = await ocrImage(buffer, env)
      } else {
        result = new TextDecoder().decode(buffer)
      }
 
      // 存结果
      const resultKey = `results/${taskId}/output.txt`
      await env.BUCKET.put(resultKey, result)
 
      // 更新状态为 completed
      await env.DB.prepare(`
        UPDATE tasks SET status = 'completed', progress = 100, result_key = ?, updated_at = ?
        WHERE id = ?
      `).bind(resultKey, new Date().toISOString(), taskId).run()
 
      msg.ack()
    } catch (err) {
      const errMsg = err instanceof Error ? err.message : String(err)
      console.error(`Task ${taskId} failed:`, errMsg)
 
      // 重试或标记失败
      const retryCount = msg.attempts - 1
      if (retryCount < 3) {
        msg.retry({ delaySeconds: 30 * (retryCount + 1) })
      } else {
        await env.DB.prepare(
          "UPDATE tasks SET status = 'failed', error_message = ?, updated_at = ? WHERE id = ?"
        ).bind(errMsg, new Date().toISOString(), taskId).run()
        msg.ack()
      }
    }
  }
}