> Discover all available pages from the documentation index: https://mastra.zisheng.pro/zh-TW/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 會根據已儲存的結果略過完成的步驟。Mastra 的迴圈、條件判斷和巢狀 Workflow 等流程控制基本元素，都會對應至相同的 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 起，Realtime 已內建於 SDK，因此不再使用 `@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、記憶體）使用。若自訂 `apiRoutes` 路徑以伺服器的 `apiPrefix`（預設為 `/api`）開頭，啟動時便會擲回錯誤。背景資訊請參閱 [#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 Dev Server 要去哪裡尋找 Mastra 的 `/inngest/api` 端點

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

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

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

5. 在 **Runs** 分頁中監控 Workflow 執行情形，查看各步驟的執行進度

### 在正式環境執行

開始之前，請確認你具備：

- Vercel 帳號，且已安裝 Vercel CLI（`npm i -g vercel`）
- Inngest 帳號
- Vercel 權杖

1. 在環境中設定 Vercel 權杖：

   ```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 儀表板](https://app.inngest.com/env/production/apps)同步

   > **警告:** Inngest 的自動探索慣例預設使用 `/api/inngest`。由於本指南使用 `/inngest/api`，請將 Inngest 應用程式的 **URL** 欄位設為已部署的來源網址加上 `/inngest/api`（例如 `https://your-app.vercel.app/inngest/api`）。若保留預設值，Inngest 儀表板將找不到應用程式的函式。

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 會自動轉換為識別碼格式如 `workflow.${workflowId}` 的 Inngest 函式
2. 自訂函式會保留指定的識別碼（例如 `send-welcome-email`、`process-webhook`）
3. 所有函式都會透過同一個 `/inngest/api` 端點一併提供

如此即可將 Mastra 的 Workflow 協調功能與現有 Inngest 函式結合使用。

## 搭配其他框架使用

預設的 `serve` 函式內部使用 Hono。若使用 Express、Fastify 或 Koa 等其他 Web 框架，請搭配適當的 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 目前處於公開測試階段。Connect worker 不支援 Vercel 和 AWS Lambda 等無伺服器執行環境。

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

- worker 處理程序在僅允許對外連線的私人網路中執行。
- 你希望將繁重的 Workflow 執行工作，從處理使用者 HTTP 流量的處理程序中分離出來。
- 你希望 worker 能獨立於 Mastra HTTP 伺服器進行擴縮，並為每個 worker 設定並行限制。

`serve()` 與 `connect()` 可使用相同的 Mastra Workflow 定義。標準 Workflow、巢狀 Workflow、cron Workflow 和其他 Inngest 函式的收集行為皆相同。

### 必要條件

- `inngest@^4`
- Node.js `22.13.0` 或更新版本
- 供 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` 設為部署識別碼，例如提交 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 儀表板。預設為機器主機名稱。
- `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`（例如在不變更儀表板 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-TW/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. 系統會自動建立另一個識別碼為 `workflow.${workflowId}.cron` 的 Inngest 函式
2. 此函式會向 Inngest 註冊，並依指定排程觸發
3. 每次排程執行都會使用提供的 `inputData` 與 `initialState` 建立新的 Workflow 執行
4. 呼叫 `serve()` 時，cron 函式與主要 Workflow 函式會一併提供

你可以在 Inngest 儀表板的 **Functions** 與 **Runs** 區段中監控排程執行情形。cron 函式會以獨立函式的形式，與主要 Workflow 函式並列顯示。

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