> Discover all available pages from the documentation index: https://mastra.zisheng.pro/zh-HK/llms.txt

# Inngest Workflow

[Inngest](https://www.inngest.com/docs) 是一個用於建立及執行背景 Workflow 的開發者平台，毋須管理基礎設施。

如需包含進階流程控制功能的完整範例，請參閱 [Inngest Workflow 範例](https://github.com/mastra-ai/mastra/tree/main/examples/inngest)。

## Inngest 如何與 Mastra 配合運作

Inngest 與 Mastra 透過對齊兩者的 Workflow 模型整合：Inngest 將邏輯組織成由多個步驟組成的函數，而使用 `createWorkflow()` 和 `createStep()` 定義的 Mastra Workflow 會直接對應至此結構。每個 Mastra Workflow 都會成為具有唯一識別碼的 Inngest 函數，而 Workflow 內的每個步驟則對應至一個 Inngest 步驟。

`serve()` 函數會將 Mastra Workflow 註冊為 Inngest 函數，並設定執行及監察所需的事件處理常式，藉此連接兩個系統。

當事件觸發 Workflow 時，Inngest 會逐步執行並記憶每項結果。重試或恢復時，Inngest 會根據這些已儲存的結果略過已完成的步驟。迴圈、條件及巢狀 Workflow 等 Mastra 控制流程基元，會對應至相同的 Inngest 函數及步驟模型，同時保留組合、分支及暫停行為。

Inngest 的發佈／訂閱系統及控制台提供實時監察、暫停／恢復，以及步驟層級的可觀察性。每個步驟執行時，其狀態及輸出都會透過 Mastra 儲存空間追蹤，並可按需要恢復。

## 設定

安裝所需依賴套件：

**npm**:

```bash
npm install @mastra/inngest@latest inngest
```

**pnpm**:

```bash
pnpm add @mastra/inngest@latest inngest
```

**Yarn**:

```bash
yarn add @mastra/inngest@latest inngest
```

**Bun**:

```bash
bun add @mastra/inngest@latest inngest
```

> **備註:** 需要 `inngest@^4` 及 Inngest Dev Server `v1.18.0` 或更新版本。v4 的 SDK 已內置 Realtime，因此不再使用 `@inngest/realtime` 及 `realtimeMiddleware`。

## 建立 Inngest Workflow

本指南會逐步說明如何使用 Inngest 及 Mastra 建立 Workflow，並示範一個遞增數值直至達到 10 的計數器應用程式。

### 初始化 Inngest

初始化 Inngest 整合，以取得與 Mastra 相容的 Workflow 輔助函數。`createWorkflow()` 及 `createStep()` 函數用於建立與 Mastra 和 inngest 相容的 Workflow 及步驟物件。

在開發環境中：

```ts
import { Inngest } from 'inngest'

export const inngest = new Inngest({
  id: 'mastra',
  baseUrl: 'http://localhost:8288',
  isDev: true,
})
```

在正式環境中：

```ts
import { Inngest } from 'inngest'

export const inngest = new Inngest({
  id: 'mastra',
})
```

#### 使用 `INNGEST_DEV` 選擇開發或雲端模式

Inngest SDK 會以開發或雲端兩種模式之一執行。由 v4 起，SDK 預設使用雲端模式，並需要 `INNGEST_EVENT_KEY` 及 `INNGEST_SIGNING_KEY`。上述範例在程式碼中以 `isDev: true` 設定模式。你亦可設定 [`INNGEST_DEV` 環境變數](https://www.inngest.com/docs/sdk/environment-variables#inngest-dev)，讓相同的用戶端程式碼可在不同環境中運作：

- `INNGEST_DEV=1`：強制使用開發模式。SDK 會與本機 Inngest Dev Server 通訊，並停用簽署驗證。
- `INNGEST_DEV=0`：強制使用雲端模式。SDK 會與 Inngest Cloud 通訊，並需要事件及簽署金鑰。
- `INNGEST_DEV=<url>`：強制使用開發模式，並將 SDK 指向位於 `<url>` 的 Dev Server，例如 `http://localhost:8288`。
- 未設定：預設使用雲端模式。

設定 `INNGEST_DEV` 後，你可從用戶端移除 `isDev` 及 `baseUrl`：

```ts
import { Inngest } from 'inngest'

export const inngest = new Inngest({
  id: 'mastra',
})
```

> **注意:** 切勿在正式環境設定 `INNGEST_DEV`。它會停用安全接收 Inngest Cloud 事件所需的簽署驗證。

若兩者同時設定，`new Inngest()` 的 `isDev` 選項會覆寫 `INNGEST_DEV`。

### 建立步驟

定義組成 Workflow 的各個步驟：

```ts
import { z } from 'zod'
import { inngest } from '../inngest'
import { init } from '@mastra/inngest'

// Initialize Inngest with Mastra, pointing to your local Inngest server
const { createWorkflow, createStep } = init(inngest)

// Step: Increment the counter value
const incrementStep = createStep({
  id: 'increment',
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
  execute: async ({ inputData }) => {
    return { value: inputData.value + 1 }
  },
})
```

### 建立 Workflow

使用 `dountil` 迴圈模式將步驟組成 Workflow。`createWorkflow()` 函數會在 Inngest 伺服器上建立可呼叫的函數。

```ts
// workflow that is registered as a function on inngest server
const workflow = createWorkflow({
  id: 'increment-workflow',
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
}).then(incrementStep)

workflow.commit()

export { workflow as incrementWorkflow }
```

### 設定 Mastra 執行個體

向 Mastra 註冊 Workflow，並設定 Inngest API 端點：

```ts
import { Mastra } from '@mastra/core'
import { serve } from '@mastra/inngest'
import { incrementWorkflow } from './workflows'
import { inngest } from './inngest'
import { PinoLogger } from '@mastra/loggers'

export const mastra = new Mastra({
  workflows: { incrementWorkflow },
  server: {
    host: '0.0.0.0',
    apiRoutes: [
      {
        path: '/inngest/api',
        method: 'ALL',
        createHandler: async ({ mastra }) => {
          return serve({ mastra, inngest })
        },
      },
    ],
  },
  logger: new PinoLogger({ name: 'Mastra', level: 'info' }),
})
```

> **備註:** 路徑是 `/inngest/api`，而非 `/api/inngest`。Mastra 保留 `/api` 前綴供內置路由（Agent、Workflow、記憶體）使用。以伺服器 `apiPrefix`（預設為 `/api`）開頭的自訂 `apiRoutes` 路徑會在啟動時擲回錯誤。詳情請參閱 [#15743](https://github.com/mastra-ai/mastra/pull/15743)；如需保留 `/api/inngest`，則可跳至[使用自訂 `apiPrefix`](#using-a-custom-apiprefix)。

## 執行 Workflow

### 在本機執行

1. 執行 `npx mastra dev`，在本機的 4111 連接埠啟動 Mastra 伺服器

2. 啟動 Inngest Dev Server。在新的終端機執行：

   ```bash
   npx inngest-cli@latest dev -u http://localhost:4111/inngest/api
   ```

   > **備註:** `-u` 後的 URL 會告知 Inngest 開發伺服器在何處尋找 Mastra `/inngest/api` 端點

3. 在 <http://localhost:8288> 開啟 Inngest Dashboard，前往側邊欄的 **Apps** 區段，確認 Mastra Workflow 已註冊

4. 在 **Functions** 中開啟 Workflow。選擇 **Invoke** 並提供以下輸入：

   ```json
   {
     "data": {
       "inputData": {
         "value": 5
       }
     }
   }
   ```

5. 在 **Runs** 分頁監察 Workflow 執行，以查看逐步執行進度

### 在正式環境執行

開始前，請確保你具備：

- Vercel 帳戶及已安裝的 Vercel CLI（`npm i -g vercel`）
- Inngest 帳戶
- Vercel token

1. 在環境中設定 Vercel token：

   ```bash
   export VERCEL_TOKEN=your_vercel_token
   ```

2. 將 `VercelDeployer` 加入 Mastra 執行個體

   ```ts
   import { VercelDeployer } from '@mastra/deployer-vercel'

   export const mastra = new Mastra({
     deployer: new VercelDeployer({
       teamSlug: 'your_team_slug',
       projectName: 'your_project_name',
       // you can get your vercel token from the vercel dashboard by clicking on the user icon in the top right corner
       // and then clicking on "Account Settings" and then clicking on "Tokens" on the left sidebar.
       token: process.env.VERCEL_TOKEN,
     }),
   })
   ```

3. 建置 Mastra 執行個體

   ```bash
   npx mastra build
   ```

4. 部署至 Vercel

   ```bash
   cd .mastra/output
   vercel login
   vercel --prod
   ```

5. 選擇 **Sync new app with Vercel** 並依照指示，與 [Inngest dashboard](https://app.inngest.com/env/production/apps) 同步

   > **注意:** Inngest 的自動探索慣例假設路徑為 `/api/inngest`。由於本指南使用 `/inngest/api`，請將 Inngest 應用程式的 **URL** 欄位設為已部署的來源加上 `/inngest/api`（例如 `https://your-app.vercel.app/inngest/api`）。如保留預設值，Inngest dashboard 將找不到應用程式的函數。

6. 在 **Functions** 中開啟 `workflow.increment-workflow`。選擇 **All actions** > **Invoke**，並提供以下輸入：

   ```json
   {
     "data": {
       "inputData": {
         "value": 5
       }
     }
   }
   ```

7. 在 **Runs** 分頁監察執行，以查看逐步進度

## 加入自訂 Inngest 函數

你可使用 `serve()` 的選用 `functions` 參數，在 Mastra Workflow 旁同時提供額外的 Inngest 函數。

### 建立自訂函數

首先，建立自訂 Inngest 函數：

```ts
import { inngest } from '../inngest'

// Define custom Inngest functions
export const customEmailFunction = inngest.createFunction(
  { id: 'send-welcome-email' },
  { event: 'user/registered' },
  async ({ event }) => {
    // Custom email logic here
    console.log(`Sending welcome email to ${event.data.email}`)
    return { status: 'email_sent' }
  },
)

export const customWebhookFunction = inngest.createFunction(
  { id: 'process-webhook' },
  { event: 'webhook/received' },
  async ({ event }) => {
    // Custom webhook processing
    console.log(`Processing webhook: ${event.data.type}`)
    return { processed: true }
  },
)
```

### 與 Workflow 一同提供自訂函數

更新 Mastra 設定，以匯入並加入自訂函數。標示的行顯示新增內容：

```ts
import { Mastra } from '@mastra/core'
import { serve } from '@mastra/inngest'
import { incrementWorkflow } from './workflows'
import { inngest } from './inngest'
import { customEmailFunction, customWebhookFunction } from './inngest/custom-functions'
import { PinoLogger } from '@mastra/loggers'

export const mastra = new Mastra({
  workflows: { incrementWorkflow },
  server: {
    host: '0.0.0.0',
    apiRoutes: [
      {
        path: '/inngest/api',
        method: 'ALL',
        createHandler: async ({ mastra }) => {
          return serve({
            mastra,
            inngest,
            functions: [customEmailFunction, customWebhookFunction],
          })
        },
      },
    ],
  },
  logger: new PinoLogger({ name: 'Mastra', level: 'info' }),
})
```

### 函數註冊

加入自訂函數後：

1. Mastra Workflow 會自動轉換為 ID 類似 `workflow.${workflowId}` 的 Inngest 函數
2. 自訂函數會保留指定的 ID（例如 `send-welcome-email`、`process-webhook`）
3. 所有函數都會在同一個 `/inngest/api` 端點一併提供

這讓你可將 Mastra 的 Workflow 編排與現有 Inngest 函數結合。

## 與其他框架配合使用

預設 `serve` 函數在內部使用 Hono。如你使用 Express、Fastify 或 Koa 等其他網頁框架，請配合適當的 Inngest 適配器使用 `createServe` 工廠函數。

### Express

```ts
import express from 'express'
import { createServe } from '@mastra/inngest'
import { serve as expressAdapter } from 'inngest/express'
import { mastra, inngest } from './mastra'

const app = express()

// Body parsing middleware required for Inngest
app.use(express.json())

const handler = createServe(expressAdapter)({ mastra, inngest })
app.use('/inngest/api', handler)

app.listen(3000)
```

### Fastify

```ts
import Fastify from 'fastify'
import { createServe } from '@mastra/inngest'
import { serve as fastifyAdapter } from 'inngest/fastify'
import { mastra, inngest } from './mastra'

const fastify = Fastify()

// JSON parsing is handled by Fastify's default content-type parser
const handler = createServe(fastifyAdapter)({ mastra, inngest })

fastify.route({
  method: ['GET', 'POST', 'PUT'],
  url: '/inngest/api',
  handler,
})

fastify.listen({ port: 3000 })
```

### Koa

```ts
import Koa from 'koa'
import Router from '@koa/router'
import bodyParser from 'koa-bodyparser'
import { createServe } from '@mastra/inngest'
import { serve as koaAdapter } from 'inngest/koa'
import { mastra, inngest } from './mastra'

const app = new Koa()
const router = new Router()

// Body parsing middleware required for Inngest
app.use(bodyParser())

const handler = createServe(koaAdapter)({ mastra, inngest })
router.all('/inngest/api', handler)

app.use(router.routes())
app.use(router.allowedMethods())

app.listen(3000)
```

### Next.js

```ts
import { createServe } from '@mastra/inngest'
import { serve as nextAdapter } from 'inngest/next'
import { mastra, inngest } from '@/mastra'

const handler = createServe(nextAdapter)({ mastra, inngest })

export { handler as GET, handler as POST, handler as PUT }
```

### 可用的適配器

`createServe` 函數適用於任何 Inngest 適配器。如需包括 AWS Lambda、Cloudflare Workers 等在內的完整可用適配器清單，請參閱 [Inngest serve 文件](https://www.inngest.com/docs/reference/serve)。

## 作為 Connect Worker 執行

`serve()` 會公開一個供 Inngest 呼叫的 HTTP 端點。另一種做法是使用 `connect()`，從 Worker 向 Inngest 開啟長時間維持的輸出連線，讓 Worker 毋須提供可公開連線的端點。這適用於 Kubernetes、Docker、ECS、Fly.io 或 Render 等運行環境中的長時間執行 Worker 程序。

> **備註:** Inngest Connect 正處於公開 beta 階段。Connect Worker 不支援 Vercel 及 AWS Lambda 等無伺服器運行環境。

### 何時使用 `connect()` 而非 `serve()`

- Worker 程序在只允許輸出連線的私人網絡中執行。
- 你想避免在處理面向使用者 HTTP 流量的同一程序中執行繁重的 Workflow。
- 你想獨立於 Mastra HTTP 伺服器擴展或縮減 Worker，並為每個 Worker 設定並行上限。

`serve()` 及 `connect()` 可使用相同的 Mastra Workflow 定義。標準 Workflow、巢狀 Workflow、cron Workflow 及額外 Inngest 函數的收集行為均相同。

### 要求

- `inngest@^4`
- Node.js `22.13.0` or later
- 一個供 Worker 長時間執行的程序

### 設定

以兩個程序分別執行 Mastra 伺服器及 Connect Worker。此設定中的 Mastra 伺服器毋須公開 `/inngest/api`：

```ts
import { Mastra } from '@mastra/core'
import { incrementWorkflow } from './workflows'
import { PinoLogger } from '@mastra/loggers'

export const mastra = new Mastra({
  workflows: { incrementWorkflow },
  logger: new PinoLogger({ name: 'Mastra', level: 'info' }),
})
```

```ts
import { connect } from '@mastra/inngest/connect'
import { mastra } from './mastra'
import { inngest } from './mastra/inngest'

await connect({
  mastra,
  inngest,
  instanceId: process.env.INNGEST_CONNECT_INSTANCE_ID,
  maxWorkerConcurrency: Number(process.env.INNGEST_CONNECT_MAX_WORKER_CONCURRENCY ?? 10),
})
```

本機開發期間，使用 `INNGEST_DEV=1 node --import tsx src/worker.ts` 啟動 Worker。Inngest Dev Server 會透過輸出連線探索 Worker，因此毋須使用 `-u` 標記。

在正式環境設定 `INNGEST_EVENT_KEY` 及 `INNGEST_SIGNING_KEY`。將 `Inngest` 用戶端的 `appVersion` 設為部署識別碼（例如 commit SHA 或映像檔標籤），讓 Inngest 可管理滾動部署。

將現有正式環境應用程式由 `serve()` 遷移至 `connect()` 時，請先使用另一個 Inngest 應用程式測試 Worker，再轉移流量。

### 選項

`connect()` 接受與 Inngest [`connect`](https://www.inngest.com/docs/setup/connect) 相同的選項，以及 Mastra 專用欄位。最常用的包括：

- `mastra`：要公開其 Workflow 的 Mastra 執行個體。
- `inngest`：Inngest 用戶端。使用你會傳給 `serve()` 的同一用戶端。
- `functions`：選用的額外 Inngest 函數陣列，與 Mastra Workflow 一同註冊。
- `instanceId`：Worker 的穩定識別碼，會顯示於 Inngest dashboard。預設為機器主機名稱。
- `maxWorkerConcurrency`：Worker 同時執行的步驟數目上限。預設不設上限。
- `registerOptions`：應用程式註冊期間轉交給 Inngest 的選項（例如 `signingKey`）。若同一欄位在此處及頂層均有設定，以 `registerOptions` 為準。這與 `serve()` 的行為一致。

`connect()` 會回傳 Inngest 的 `WorkerConnection`。Inngest SDK 預設會處理 `SIGINT` 及 `SIGTERM`。只有 Worker 需要自訂關閉控制時，才儲存回傳的連線並呼叫 `.close()`。

如 Mastra 執行個體沒有 `InngestWorkflow`，亦沒有提供額外 `functions`，`connect()` 會記錄警告，因為 Worker 否則會保持連線但沒有任何工作可執行。請在 Mastra 上註冊至少一個 Workflow，或傳入 `functions: [...]`。

## 使用自訂 `apiPrefix`

如需保留 `/api/inngest`（例如在不變更 dashboard URL 的情況下配合 Inngest 的自動探索慣例），請設定 `server.apiPrefix`，以重新定位 Mastra 的內置路由：

```ts
import { Mastra } from '@mastra/core'
import { serve } from '@mastra/inngest'
import { inngest } from './inngest'

export const mastra = new Mastra({
  server: {
    apiPrefix: '/_mastra',
    apiRoutes: [
      {
        path: '/api/inngest',
        method: 'ALL',
        createHandler: async ({ mastra }) => serve({ mastra, inngest }),
      },
    ],
  },
})
```

Mastra 的內置路由現在會解析至 `/_mastra/agents`、`/_mastra/workflows` 等路徑，騰出 `/api/inngest` 路徑供自訂路由使用。

> **注意:** 預設驗證設定會保護 `/api/*`，並將 `/api`、`/api/auth/*` 視為公開路徑。變更 `apiPrefix` 後，這些預設值便不再匹配，內置路由亦會落在受保護模式以外。請更新 `server.auth.protected` 及 `server.auth.public` 以參照新前綴，並更新所有存取 `/api/*` 的用戶端程式碼，包括 [`MastraClient`](https://mastra.zisheng.pro/zh-HK/docs/server/mastra-client) 的 `apiPrefix`。

## 流程控制

Inngest Workflow 支援並行上限、速率限制、節流、防彈跳及優先次序佇列等流程控制功能。這些選項在 `createWorkflow()` 呼叫中設定，有助大規模管理 Workflow 執行。

### 並行

控制可同時執行的 Workflow 執行個體數目：

```ts
const workflow = createWorkflow({
  id: 'user-processing-workflow',
  inputSchema: z.object({ userId: z.string() }),
  outputSchema: z.object({ result: z.string() }),
  steps: [processUserStep],
  // Limit to 10 concurrent executions, scoped by user ID
  concurrency: {
    limit: 10,
    key: 'event.data.userId',
  },
})
```

### 速率限制

限制一段時間內的 Workflow 執行次數：

```ts
const workflow = createWorkflow({
  id: 'api-sync-workflow',
  inputSchema: z.object({ endpoint: z.string() }),
  outputSchema: z.object({ status: z.string() }),
  steps: [apiSyncStep],
  // Maximum 1000 executions per hour
  rateLimit: {
    period: '1h',
    limit: 1000,
  },
})
```

### 節流

確保 Workflow 各次執行之間的最短間隔：

```ts
const workflow = createWorkflow({
  id: 'email-notification-workflow',
  inputSchema: z.object({ organizationId: z.string(), message: z.string() }),
  outputSchema: z.object({ sent: z.boolean() }),
  steps: [sendEmailStep],
  // Only one execution per 10 seconds per organization
  throttle: {
    period: '10s',
    limit: 1,
    key: 'event.data.organizationId',
  },
})
```

### 防彈跳

延遲執行，直至指定時間範圍內沒有新事件到達：

```ts
const workflow = createWorkflow({
  id: 'search-index-workflow',
  inputSchema: z.object({ documentId: z.string() }),
  outputSchema: z.object({ indexed: z.boolean() }),
  steps: [indexDocumentStep],
  // Wait 5 seconds of no updates before indexing
  debounce: {
    period: '5s',
    key: 'event.data.documentId',
  },
})
```

### 優先次序

設定 Workflow 的執行優先次序：

```ts
const workflow = createWorkflow({
  id: 'order-processing-workflow',
  inputSchema: z.object({
    orderId: z.string(),
    priority: z.number().optional(),
  }),
  outputSchema: z.object({ processed: z.boolean() }),
  steps: [processOrderStep],
  // Higher priority orders execute first
  priority: {
    run: 'event.data.priority ?? 50',
  },
})
```

### 組合流程控制選項

單一 Workflow 可組合多個流程控制選項：

```ts
const workflow = createWorkflow({
  id: 'comprehensive-workflow',
  inputSchema: z.object({
    userId: z.string(),
    organizationId: z.string(),
    priority: z.number().optional(),
  }),
  outputSchema: z.object({ result: z.string() }),
  steps: [comprehensiveStep],
  concurrency: {
    limit: 5,
    key: 'event.data.userId',
  },
  rateLimit: {
    period: '1m',
    limit: 100,
  },
  throttle: {
    period: '10s',
    limit: 1,
    key: 'event.data.organizationId',
  },
  priority: {
    run: 'event.data.priority ?? 0',
  },
})
```

所有流程控制選項均為選用。如未指定，Workflow 會按 Inngest 的預設行為執行。詳情請參閱 [Inngest 流程控制文件](https://www.inngest.com/docs/guides/flow-control)。

## Cron 排程

使用 cron 表達式按排程觸發 Inngest Workflow。常見用途包括每日報告、每小時資料同步及維護工作。

### 基本 cron 排程

加入 `cron` 屬性，設定 Workflow 按排程執行：

```ts
const workflow = createWorkflow({
  id: 'daily-report-workflow',
  inputSchema: z.object({ reportType: z.string() }),
  outputSchema: z.object({ generated: z.boolean() }),
  steps: [generateReportStep],
  // Run daily at midnight
  cron: '0 0 * * *',
})
```

### Cron 排程格式

`cron` 屬性接受以下格式的標準 cron 表達式：`minute hour day month dayOfWeek`

- **minute**：0-59
- **hour**：0-23
- **day**：1-31
- **month**：1-12 或 JAN-DEC
- **dayOfWeek**：0-6（星期日 = 0）或 SUN-SAT

常見 cron 模式：

```ts
// Every 15 minutes
cron: '*/15 * * * *'

// Every hour at minute 0
cron: '0 * * * *'

// Every 6 hours
cron: '0 */6 * * *'

// Daily at midnight
cron: '0 0 * * *'

// Daily at 9 AM
cron: '0 9 * * *'

// Every weekday at 9 AM
cron: '0 9 * * 1-5'

// First day of every month at midnight
cron: '0 0 1 * *'

// Every Monday at 8 AM
cron: '0 8 * * 1'
```

### 為排程執行提供輸入資料

你可提供每次排程執行時使用的靜態輸入資料：

```ts
const workflow = createWorkflow({
  id: 'scheduled-data-sync',
  inputSchema: z.object({
    source: z.string(),
    destination: z.string(),
  }),
  outputSchema: z.object({ synced: z.boolean() }),
  steps: [syncDataStep],
  cron: '0 */6 * * *', // Every 6 hours
  // Input data provided to each scheduled run
  inputData: {
    source: 'production-db',
    destination: 'analytics-warehouse',
  },
})
```

### 為排程執行提供初始狀態

你亦可為排程 Workflow 執行設定初始狀態：

```ts
const workflow = createWorkflow({
  id: 'scheduled-aggregation',
  inputSchema: z.object({ date: z.string() }),
  outputSchema: z.object({ aggregated: z.boolean() }),
  stateSchema: z.object({
    processedCount: z.number(),
    lastProcessedDate: z.string(),
  }),
  steps: [aggregateDataStep],
  cron: '0 0 * * *', // Daily at midnight
  inputData: {
    date: new Date().toISOString().split('T')[0], // Today's date
  },
  initialState: {
    processedCount: 0,
    lastProcessedDate: '',
  },
})
```

### 結合 cron 與流程控制

Cron 排程可與流程控制選項結合：

```ts
const workflow = createWorkflow({
  id: 'scheduled-api-sync',
  inputSchema: z.object({ endpoint: z.string() }),
  outputSchema: z.object({ synced: z.boolean() }),
  steps: [syncApiStep],
  cron: '*/30 * * * *', // Every 30 minutes
  inputData: {
    endpoint: 'https://api.example.com/data',
  },
  // Limit concurrent executions even for scheduled runs
  concurrency: {
    limit: 5,
  },
  // Rate limit scheduled executions
  rateLimit: {
    period: '1h',
    limit: 100,
  },
})
```

### Cron 函數的運作方式

使用 `cron` 屬性設定 Workflow 後：

1. 系統會自動建立 ID 為 `workflow.${workflowId}.cron` 的獨立 Inngest 函數
2. 此函數會向 Inngest 註冊，並按指定排程觸發
3. 每次排程執行都會使用所提供的 `inputData` 及 `initialState` 建立新的 Workflow 執行
4. 呼叫 `serve()` 時，cron 函數及主要 Workflow 函數會一併提供

你可在 Inngest dashboard 的 **Functions** 及 **Runs** 區段監察排程執行。Cron 函數會以獨立函數形式顯示於主要 Workflow 函數旁。

如需更多 cron 排程資料，請參閱 [Inngest cron 文件](https://www.inngest.com/docs/guides/scheduled-functions)。