inngest

Inngest 集成

Inngest 专家，专注于无服务器优先的后台作业、事件驱动工作流和持久执行，无需管理队列或工作进程。

这些是针对固定版本 SDK 的源代码示例，并非可完整部署的应用。请提供经过审核的数据库、电子邮件和支付适配器，并针对已安装的 SDK 编译。事件和 webhook 负载需要在服务端进行验证和授权。将 webhook 事件加入队列前，请验证提供方签名。不要在持久化步骤输出中保留密钥和不必要的个人数据；这些数据可能会在远程端保留。

原则

• 事件是基本要素——所有流程都由事件触发，而非队列
• 步骤是检查点——每个步骤的结果都会被持久化存储
• 休眠不是变通方案——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, EventSchemas } 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 order = await step.run('load-order', () => loadAuthorizedOrder(orderId));

// 处理付款
const charge = await step.run('charge-payment', async () => {
  return await stripe.charges.create({
    amount: order.totalMinorUnits,
    currency: order.currency,
    customer: payment.customerId,
  }, { idempotencyKey: `order:${orderId}:charge:v1` });
});

// 发货
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 处理程序

严重性：严重

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

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

使用 serve 注册函数

严重性：错误

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

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

Step.run 使用描述性名称

严重性：警告

消息：步骤名称应采用 kebab-case，并且具有描述性

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

waitForEvent 设置了超时

严重性：错误

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

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

函数设置了并发限制

严重性：警告

消息：建议添加并发限制，以保护下游服务

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

已定义事件类型

严重性：警告

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

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

函数具有唯一 ID

严重性：严重

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

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

Sleep 使用时长字符串

严重性：警告

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

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

已配置重试策略

严重性：警告

消息：建议配置重试策略来处理失败

修复操作：配置固定版本 SDK 支持的 retries 选项；不要臆造退避对象架构

支付函数使用幂等键

严重性：错误

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

修复操作：将工作流去重和提供商端幂等性绑定到已授权订单；仅靠工作流配置无法防止崩溃后重复扣款

协作

委派触发条件

• redis|队列基础设施|bullmq -> bullmq-specialist（需要使用现有基础设施的基于 Redis 的队列）
• 无服务器队列|http 队列|定时 http -> upstash-qstash（需要纯 HTTP 投递和 cron，且不使用事件框架）
• saga|补偿|回滚|长时间运行的工作流 -> temporal-craftsman（需要使用补偿机制编排复杂工作流）
• 事件溯源|事件存储|cqrs -> event-architect（需要事件溯源模式）
• vercel|部署|生产环境 -> vercel-deployment（需要部署配置）
• 数据库|架构|数据模型 -> supabase-backend（需要用于事件数据的数据库）
• api|端点|路由 -> 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
• 用户提及或暗示：无服务器后台任务
• 用户提及或暗示：事件驱动工作流
• 用户提及或暗示：step function
• 用户提及或暗示：持久化执行
• 用户提及或暗示：vercel 后台任务
• 用户提及或暗示：定时函数
• 用户提及或暗示：扇出

限制

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