Hono 生产实战 05：队列、定时任务和后台作业

import { Hono } from 'hono'
import { z } from 'zod'
import { zValidator } from '@hono/zod-validator'

type Bindings = {
  REPORT_QUEUE: Queue
  DB: D1Database
}

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

const createReportSchema = z.object({
  startDate: z.string(),
  endDate: z.string(),
})

app.post(
  '/api/reports/export',
  zValidator('json', createReportSchema),
  async (c) => {
    const user = c.get('user')
    const input = c.req.valid('json')
    const jobId = crypto.randomUUID()
    const now = new Date().toISOString()

    await c.env.DB.prepare(`
      insert into jobs (id, user_id, type, status, payload, created_at, updated_at)
      values (?, ?, ?, ?, ?, ?, ?)
    `).bind(
      jobId,
      user.id,
      'report.export',
      'pending',
      JSON.stringify({ ...input, userId: user.id }),
      now,
      now
    ).run()

    await c.env.REPORT_QUEUE.send({
      jobId,
      type: 'report.export',
    })

    return c.json({
      data: {
        jobId,
        status: 'pending',
      },
    }, 202)
  }
)

await c.env.REPORT_QUEUE.send({
  jobId,
  type: 'report.export',
})

const job = await env.DB.prepare(`
  select * from jobs where id = ?
`).bind(message.body.jobId).first()

import { Hono } from 'hono'

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

app.get('/health', (c) => c.json({ ok: true }))

export default {
  fetch: app.fetch,

  async queue(batch: MessageBatch, env: Env, ctx: ExecutionContext) {
    for (const message of batch.messages) {
      await processJob(message.body, env)
    }
  },
}

async function processJob(body: unknown, env: Env) {
  const message = body as {
    jobId: string
    type: string
  }

  const now = new Date().toISOString()

  await env.DB.prepare(`
    update jobs
    set status = ?, attempts = attempts + 1, updated_at = ?
    where id = ? and status in ('pending', 'failed')
  `).bind('processing', now, message.jobId).run()

  try {
    if (message.type === 'report.export') {
      await exportReport(message.jobId, env)
    }

    await env.DB.prepare(`
      update jobs
      set status = ?, updated_at = ?, finished_at = ?
      where id = ?
    `).bind('succeeded', now, now, message.jobId).run()
  } catch (error) {
    await env.DB.prepare(`
      update jobs
      set status = ?, error = ?, updated_at = ?
      where id = ?
    `).bind(
      'failed',
      error instanceof Error ? error.message : String(error),
      now,
      message.jobId
    ).run()

    throw error
  }
}

async function exportReport(jobId: string, env: Env) {
  const job = await env.DB.prepare(`
    select * from jobs where id = ?
  `).bind(jobId).first<{
    id: string
    status: string
    payload: string
    result: string | null
  }>()

  if (!job) {
    throw new Error(`job not found: ${jobId}`)
  }

  if (job.status === 'succeeded' && job.result) {
    return
  }

  const outputKey = `reports/${job.id}.csv`
  const existing = await env.R2_BUCKET.head(outputKey)

  if (existing) {
    await markJobSucceeded(job.id, outputKey, env)
    return
  }

  const payload = JSON.parse(job.payload)
  const csv = await buildReportCsv(payload, env)

  await env.R2_BUCKET.put(outputKey, csv, {
    httpMetadata: {
      contentType: 'text/csv; charset=utf-8',
    },
  })

  await markJobSucceeded(job.id, outputKey, env)
}

export default {
  fetch: app.fetch,

  async scheduled(
    event: ScheduledEvent,
    env: Env,
    ctx: ExecutionContext
  ) {
    ctx.waitUntil(runScheduledJobs(event, env))
  },
}

async function cleanupExpiredJobs(env: Env) {
  const expiredJobs = await env.DB.prepare(`
    select id
    from jobs
    where status in ('pending', 'processing')
      and created_at < ?
    order by created_at asc
    limit 100
  `).bind(
    new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString()
  ).all<{ id: string }>()

  for (const job of expiredJobs.results) {
    await env.DB.prepare(`
      update jobs
      set status = ?, updated_at = ?
      where id = ?
    `).bind('failed', new Date().toISOString(), job.id).run()
  }
}

app.get('/api/jobs/:id', async (c) => {
  const user = c.get('user')
  const jobId = c.req.param('id')

  const job = await c.env.DB.prepare(`
    select id, type, status, result, error, created_at, updated_at, finished_at
    from jobs
    where id = ? and user_id = ?
  `).bind(jobId, user.id).first()

  if (!job) {
    return c.json({
      error: {
        code: 'JOB_NOT_FOUND',
        message: '任务不存在',
      },
    }, 404)
  }

  return c.json({ data: job })
})

console.error('job_failed', {
  jobId,
  type,
  attempts,
  durationMs,
  message: error instanceof Error ? error.message : String(error),
})