inngest

Inngest 集成

Inngest 专家，专注无服务器优先的后台任务、事件驱动工作流， 以及无需管理队列或 worker 的持久化执行。

原则

• 事件是基本单元 - 一切由事件触发，而非队列
• 步骤是你的检查点 - 每个步骤的结果都会被持久化存储
• 休眠不是权宜之计 - Inngest 的休眠是真实的，不会阻塞线程
• 重试是自动的 - 但策略由你掌控
• 函数只是 HTTP 处理器 - 可部署到任何提供 HTTP 服务的地方
• 并发是一等关注点 - 保护下游服务
• 幂等键可防止重复 - 关键操作请使用它们
• 扇出是内置的 - 一个事件可以触发多个函数

能力

• inngest-functions
• event-driven-workflows
• step-functions
• serverless-background-jobs
• durable-sleep
• fan-out-patterns
• concurrency-control
• scheduled-functions

范围

• redis-queues -> bullmq-specialist
• serverless-queues -> upstash-qstash
• workflow-orchestration -> temporal-craftsman
• message-streaming -> event-architect
• infrastructure -> infra-architect

工具

核心

• inngest
• inngest-cli

框架

• nextjs
• express
• hono
• remix
• sveltekit

部署

• vercel
• cloudflare-workers
• netlify
• railway
• fly-io

模式

• step-functions
• event-fan-out
• scheduled-cron
• webhook-handling

模式

基本函数设置

在 Next.js 中使用带类型事件的 Inngest 函数

何时使用：在任何 Next.js 项目中开始使用 Inngest 时

// lib/inngest/client.ts import { Inngest } from 'inngest';

export const inngest = new Inngest({ id: 'my-app', schemas: new EventSchemas().fromRecord(), });

// Define your events with types type Events = { 'user/signed.up': { data: { userId: string; email: string } }; 'order/placed': { data: { orderId: string; total: number } }; };

// lib/inngest/functions.ts import { inngest } from './client';

export const sendWelcomeEmail = inngest.createFunction( { id: 'send-welcome-email' }, { event: 'user/signed.up' }, async ({ event, step }) => { // Step 1: Get user details const user = await step.run('get-user', async () => { return await db.users.findUnique({ where: { id: event.data.userId } }); });

// Step 2: Send welcome email
await step.run('send-email', async () => {
  await resend.emails.send({
    to: user.email,
    subject: 'Welcome!',
    template: 'welcome',
  });
});

// Step 3: Wait 24 hours, then send tips
await step.sleep('wait-for-tips', '24h');

await step.run('send-tips', async () => {
  await resend.emails.send({
    to: user.email,
    subject: 'Getting Started Tips',
    template: 'tips',
  });
});

} );

// app/api/inngest/route.ts (Next.js App Router) import { serve } from 'inngest/next'; import { inngest } from '@/lib/inngest/client'; import { sendWelcomeEmail } from '@/lib/inngest/functions';

export const { GET, POST, PUT } = serve({ client: inngest, functions: [sendWelcomeEmail], });

多步骤工作流

包含并行步骤和错误处理的复杂工作流

何时使用：涉及多个服务或长时间等待的处理

export const processOrder = inngest.createFunction( { id: 'process-order', retries: 3, concurrency: { limit: 10 }, // 最多同时处理 10 个订单 }, { event: 'order/placed' }, async ({ event, step }) => { const { orderId } = event.data;

// 并行步骤 - 两者同时运行
const [inventory, payment] = await Promise.all([
  step.run('check-inventory', () => checkInventory(orderId)),
  step.run('validate-payment', () => validatePayment(orderId)),
]);

if (!inventory.available) {
  // 发送事件而不是直接调用（扇出模式）
  await step.sendEvent('notify-backorder', {
    name: 'order/backordered',
    data: { orderId, items: inventory.missing },
  });
  return { status: 'backordered' };
}

// 处理付款
const charge = await step.run('charge-payment', async () => {
  return await stripe.charges.create({
    amount: event.data.total,
    customer: payment.customerId,
  });
});

// 发货
await step.run('ship-order', () => fulfillment.ship(orderId));

return { status: 'completed', chargeId: charge.id };

} );

定时/Cron 函数

按计划运行的函数

何时使用：每日报告或清理任务等重复性任务

export const dailyDigest = inngest.createFunction( { id: 'daily-digest' }, { cron: '0 9 * * *' }, // 每天 UTC 时间上午 9 点 async ({ step }) => { // 获取所有想要摘要的用户 const users = await step.run('get-users', async () => { return await db.users.findMany({ where: { digestEnabled: true }, }); });

// 发送给每个用户（创建子事件）
await step.sendEvent(
  'send-digests',
  users.map(user => ({
    name: 'digest/send',
    data: { userId: user.id },
  }))
);

return { sent: users.length };

} );

// 单独的函数处理单个摘要的发送 export const sendDigest = inngest.createFunction( { id: 'send-digest', concurrency: { limit: 50 } }, { event: 'digest/send' }, async ({ event, step }) => { // ... 发送单个摘要 } );

带幂等性的 Webhook 处理程序

通过去重安全地处理 webhook

何时使用：处理 Stripe、GitHub 或其他 webhook

export const handleStripeWebhook = inngest.createFunction( { id: 'stripe-webhook', // 按 Stripe 事件 ID 去重 idempotency: 'event.data.stripeEventId', }, { event: 'stripe/webhook.received' }, async ({ event, step }) => { const { type, data } = event.data;

switch (type) {
  case 'checkout.session.completed':
    await step.run('fulfill-order', async () => {
      await fulfillOrder(data.session.id);
    });
    break;

  case 'customer.subscription.deleted':
    await step.run('cancel-subscription', async () => {
      await cancelSubscription(data.subscription.id);
    });
    break;
}

} );

长时间处理的 AI 流水线

多步骤 AI 处理，分块执行工作

何时使用：可能需要数分钟才能完成的 AI 工作流

export const processDocument = inngest.createFunction( { id: 'process-document', retries: 2, concurrency: { limit: 5 }, // Limit API usage }, { event: 'document/uploaded' }, async ({ event, step }) => { // Step 1: Extract text (may take a while) const text = await step.run('extract-text', async () => { return await extractTextFromPDF(event.data.fileUrl); });

// Step 2: Chunk for embedding
const chunks = await step.run('chunk-text', async () => {
  return chunkText(text, { maxTokens: 500 });
});

// Step 3: Generate embeddings (API rate limited)
const embeddings = await step.run('generate-embeddings', async () => {
  return await openai.embeddings.create({
    model: 'text-embedding-3-small',
    input: chunks,
  });
});

// Step 4: Store in vector DB
await step.run('store-vectors', async () => {
  await vectorDb.upsert({
    vectors: embeddings.data.map((e, i) => ({
      id: `${event.data.documentId}-${i}`,
      values: e.embedding,
      metadata: { chunk: chunks[i] },
    })),
  });
});

return { chunks: chunks.length, status: 'indexed' };

} );

校验检查

存在 Inngest serve 处理程序

严重程度：CRITICAL

消息：Inngest 需要一个 serve 处理程序来接收事件

修复操作：创建 app/api/inngest/route.ts 并导出 serve()

函数已通过 serve 注册

严重程度：ERROR

消息：确保所有 Inngest 函数都在 serve() 调用中注册

修复操作：将函数添加到 serve() 中的 functions 数组

Step.run 具有描述性名称

严重程度：WARNING

消息：步骤名称应为 kebab-case 且具有描述性

修复操作：使用描述性步骤名称，如 'fetch-user' 或 'send-email'

waitForEvent 具有超时设置

严重程度：ERROR

消息：waitForEvent 应设置超时，以防止无限等待

修复操作：添加 timeout 选项：{ timeout: '24h' }

函数具有并发限制

严重程度：WARNING

消息：考虑添加并发限制以保护下游服务

修复操作：在函数配置中添加 concurrency: { limit: 10 }

已定义事件类型

严重程度：WARNING

消息：Inngest 客户端应定义事件 schema 以确保类型安全

修复操作：添加 schemas: new EventSchemas().fromRecord()

函数具有唯一 ID

严重程度：CRITICAL

消息：每个 Inngest 函数都必须具有唯一 ID

修复操作：在函数配置中添加 id: 'my-function-name'

Sleep 使用时长字符串

严重程度：WARNING

消息：step.sleep 应使用 '1h' 或 '30m' 等时长字符串，而不是毫秒

修复操作：使用时长字符串：step.sleep('wait', '1h')

已配置重试策略

严重程度：WARNING

消息：考虑配置重试策略以处理失败

修复操作：添加 retries: 3 或 retries: { attempts: 3, backoff: { ... } }

支付函数的幂等键

严重程度：ERROR

消息：支付相关函数应使用幂等键

修复操作：在函数配置中添加 idempotency: 'event.data.orderId'

协作

委托触发条件

• redis|queue infrastructure|bullmq -> bullmq-specialist (需要基于 Redis 且复用现有基础设施的队列)
• serverless queue|http queue|scheduled http -> upstash-qstash (需要不依赖事件框架的纯 HTTP 投递和定时任务)
• saga|compensation|rollback|long-running workflow -> temporal-craftsman (需要带补偿机制的复杂工作流编排)
• event sourcing|event store|cqrs -> event-architect (需要事件溯源模式)
• vercel|deploy|production -> vercel-deployment (需要部署配置)
• database|schema|data model -> supabase-backend (需要用于事件数据的数据库)
• api|endpoint|route -> backend (需要用于触发事件的 API)

Vercel 后台任务

技能：inngest、nextjs-app-router、vercel-deployment

工作流：

1. Define Inngest functions (inngest)
2. Set up serve handler in Next.js (nextjs-app-router)
3. Configure function timeouts (vercel-deployment)
4. Deploy and test (vercel-deployment)

AI 流水线

技能：inngest、ai-agents-architect、supabase-backend

工作流：

1. Design AI workflow steps (ai-agents-architect)
2. Implement with Inngest durability (inngest)
3. Store results in database (supabase-backend)
4. Handle retries for API failures (inngest)

Webhook 处理

技能：inngest、stripe-integration、backend

工作流：

1. Receive webhook (backend)
2. Send to Inngest with idempotency (inngest)
3. Process payment logic (stripe-integration)
4. Update application state (backend)

邮件自动化

技能：inngest、email-systems、supabase-backend

工作流：

1. Trigger event from user action (inngest)
2. Schedule drip emails with step.sleep (inngest)
3. Send emails with retry (email-systems)
4. Track email status (supabase-backend)

定时任务

技能：inngest、backend、analytics-architecture

工作流：

1. Define cron triggers (inngest)
2. Implement processing logic (backend)
3. Aggregate and report data (analytics-architecture)
4. Handle failures with alerting (inngest)

相关技能

可良好配合：nextjs-app-router、vercel-deployment、supabase-backend、email-systems、ai-agents-architect、stripe-integration

何时使用

• 用户提到或暗示：inngest
• 用户提到或暗示：serverless 后台任务
• 用户提到或暗示：事件驱动工作流
• 用户提到或暗示：步骤函数
• 用户提到或暗示：持久化执行
• 用户提到或暗示：vercel 后台任务
• 用户提到或暗示：定时函数
• 用户提到或暗示：扇出

限制

• 仅当任务明确符合上述范围时才使用此技能。
• 不要将输出视为针对特定环境的验证、测试或专家评审的替代品。
• 如果缺少必需的输入、权限、安全边界或成功标准，请停止并请求澄清。