后台任务与队列

pnpm add inngest

// inngest/client.ts — 创建 Inngest 客户端
import { Inngest } from 'inngest'
 
export const inngest = new Inngest({
  id: 'ai-chat-app',
  schemas: new EventSchemas(),
})

// app/api/inngest/route.ts — 注册 Inngest 路由
import { serve } from 'inngest/next'
import { inngest } from '@/inngest/client'
import { generateReport, syncUserData, processDocument } from '@/inngest/functions'
 
export const { GET, POST, PUT } = serve({
  client: inngest,
  functions: [generateReport, syncUserData, processDocument],
})

// inngest/functions/generate-report.ts
import { inngest } from '../client'
 
export const generateReport = inngest.createFunction(
  {
    id: 'generate-monthly-report',
    retries: 3,                  // 失败自动重试 3 次
    throttle: { limit: 5, period: '1m' }, // 每分钟最多 5 个并发
  },
  { event: 'report/generate.requested' },
 
  async ({ event, step }) => {
    const { workspaceId, month } = event.data
 
    // Step 1: 查询数据（每个 step 独立可重试）
    const stats = await step.run('fetch-stats', async () => {
      return await db.query.tokenUsages.findMany({
        where: and(
          eq(tokenUsages.workspaceId, workspaceId),
          sql`DATE_TRUNC('month', ${tokenUsages.createdAt}) = ${month}`,
        ),
      })
    })
 
    // Step 2: 生成 PDF
    const pdfUrl = await step.run('generate-pdf', async () => {
      const pdf = await renderReport(stats)
      return await uploadToS3(`reports/${workspaceId}/${month}.pdf`, pdf)
    })
 
    // Step 3: 发送通知邮件
    await step.run('send-email', async () => {
      await sendEmail({
        to: event.data.email,
        subject: `${month} 月度报告`,
        body: `您的报告已生成：${pdfUrl}`,
      })
    })
 
    return { pdfUrl }
  },
)

// actions/report.ts
'use server'
 
export async function requestReport(month: string) {
  const { workspace } = await requireWorkspaceAccess()
 
  await inngest.send({
    name: 'report/generate.requested',
    data: {
      workspaceId: workspace.id,
      month,
      email: (await auth())!.user.email,
    },
  })
 
  return { success: true, message: '报告生成中，完成后将发送到您的邮箱' }
}

export const dailyCleanup = inngest.createFunction(
  { id: 'daily-cleanup' },
  { cron: '0 3 * * *' },  // 每天凌晨 3 点
 
  async ({ step }) => {
    // 清理 30 天前的临时文件
    await step.run('cleanup-temp-files', async () => {
      await db.delete(tempFiles).where(
        lt(tempFiles.createdAt, sql`NOW() - INTERVAL '30 days'`),
      )
    })
 
    // 更新 Token 使用统计
    await step.run('update-daily-stats', async () => {
      await updateDailyTokenStats()
    })
  },
)

// trigger/process-document.ts
import { task } from '@trigger.dev/sdk/v3'
 
export const processDocument = task({
  id: 'process-document',
  maxDuration: 300, // 最长 5 分钟
  retry: { maxAttempts: 3 },
 
  run: async (payload: { documentId: string; workspaceId: string }) => {
    // 1. 下载文档
    const document = await downloadFromS3(payload.documentId)
 
    // 2. 提取文本
    const text = await extractText(document)
 
    // 3. 生成嵌入向量
    const embeddings = await generateEmbeddings(text)
 
    // 4. 存入向量数据库
    await storeEmbeddings(payload.workspaceId, payload.documentId, embeddings)
 
    return { chunks: embeddings.length }
  },
})

import { processDocument } from '@/trigger/process-document'
 
export async function uploadDocument(formData: FormData) {
  // ... 上传文件到 S3
 
  // 触发后台处理
  await processDocument.trigger({
    documentId: doc.id,
    workspaceId: workspace.id,
  })
 
  return { success: true, message: '文档正在处理中' }
}

// vercel.json
{
  "crons": [
    {
      "path": "/api/cron/daily-cleanup",
      "schedule": "0 3 * * *"
    },
    {
      "path": "/api/cron/sync-quotas",
      "schedule": "*/30 * * * *"
    }
  ]
}

// app/api/cron/daily-cleanup/route.ts
export async function GET(request: Request) {
  // 验证是 Vercel Cron 调用（防止外部触发）
  const authHeader = request.headers.get('authorization')
  if (authHeader !== `Bearer ${process.env.CRON_SECRET}`) {
    return Response.json({ error: 'Unauthorized' }, { status: 401 })
  }
 
  await cleanupExpiredSessions()
  await updateDailyStats()
 
  return Response.json({ success: true })
}

// app/api/webhook/stripe/route.ts
import Stripe from 'stripe'
 
const stripe = new Stripe(process.env.STRIPE_SECRET_KEY!)
const webhookSecret = process.env.STRIPE_WEBHOOK_SECRET!
 
export async function POST(request: Request) {
  const body = await request.text()
  const signature = request.headers.get('stripe-signature')!
 
  // 1. 验证签名（防止伪造请求）
  let event: Stripe.Event
  try {
    event = stripe.webhooks.constructEvent(body, signature, webhookSecret)
  } catch {
    return Response.json({ error: 'Invalid signature' }, { status: 400 })
  }
 
  // 2. 根据事件类型处理
  switch (event.type) {
    case 'checkout.session.completed': {
      const session = event.data.object as Stripe.Checkout.Session
      await activateSubscription(session.customer as string, session.subscription as string)
      break
    }
    case 'customer.subscription.deleted': {
      const subscription = event.data.object as Stripe.Subscription
      await deactivateSubscription(subscription.id)
      break
    }
    case 'invoice.payment_failed': {
      const invoice = event.data.object as Stripe.Invoice
      await notifyPaymentFailed(invoice.customer as string)
      break
    }
  }
 
  // 3. 必须返回 200，否则 Stripe 会重试
  return Response.json({ received: true })
}

// 幂等处理：用事件 ID 去重
const processed = await db.query.webhookEvents.findFirst({
  where: eq(webhookEvents.eventId, event.id),
})
if (processed) return Response.json({ received: true }) // 已处理，跳过
 
// 记录事件
await db.insert(webhookEvents).values({ eventId: event.id, type: event.type, payload: body })
 
// 处理业务逻辑...