> Discover all available pages from the documentation index: https://mastra.zisheng.pro/ja/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 は Step で構成される Function にロジックを整理し、`createWorkflow()` と `createStep()` で定義された Mastra Workflow はこの構造に直接対応します。各 Mastra Workflow は一意の識別子を持つ Inngest Function になり、Workflow 内の各 Step は Inngest Step に対応します。

`serve()` 関数は Mastra Workflow を Inngest Function として登録し、実行と監視に必要な Event Handler を設定して 2 つのシステムを接続します。

イベントが Workflow を開始すると、Inngest は Step ごとに実行し、各結果を Memoize します。再試行または再開時には、保存された結果に基づいて完了済みの Step を省略します。ループ、条件分岐、ネストされた Workflow などの Mastra 制御フロープリミティブは、構成、分岐、中断を維持したまま、同じ Inngest Function と Step モデルに対応します。

Inngest の Publish-subscribe システムとダッシュボードにより、リアルタイム監視、中断と再開、Step レベルの Observability が有効になります。各 Step の実行時に、その状態と出力が 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 を構築する

このガイドでは、値が 10 に達するまで増分するカウンターアプリケーションを例に、Inngest と Mastra を使用した Workflow の作成方法を説明します。

### Inngest を初期化する

Inngest 統合を初期化し、Mastra 互換の Workflow Helper を取得します。`createWorkflow()` と `createStep()` 関数を使用して、Mastra と Inngest に対応する Workflow と Step のオブジェクトを作成します。

開発環境では、次のように設定します。

```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` で Dev Mode または Cloud Mode を選択する

Inngest SDK は Dev または Cloud のいずれかの Mode で動作します。v4 以降、SDK のデフォルトは Cloud Mode で、`INNGEST_EVENT_KEY` と `INNGEST_SIGNING_KEY` が必要です。上記の例では、`isDev: true` で Mode をコード内に設定しています。代わりに、[`INNGEST_DEV` 環境変数](https://www.inngest.com/docs/sdk/environment-variables#inngest-dev)を設定すると、同じ Client コードを複数の環境で使用できます。

- `INNGEST_DEV=1`：Dev Mode を強制します。SDK はローカルの Inngest Dev Server と通信し、署名検証を無効にします。
- `INNGEST_DEV=0`：Cloud Mode を強制します。SDK は Inngest Cloud と通信し、Event Key と Signing Key が必要です。
- `INNGEST_DEV=<url>`：Dev Mode を強制し、`<url>`（`http://localhost:8288` など）の Dev Server を参照します。
- 未設定：デフォルトは Cloud Mode です。

`INNGEST_DEV` を設定すると、Client から `isDev` と `baseUrl` を削除できます。

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

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

> **警告:** 本番環境では `INNGEST_DEV` を設定しないでください。Inngest Cloud からイベントを安全に受信するために必要な署名検証が無効になります。

isDev と `INNGEST_DEV` の両方を設定すると、`new Inngest()` の `isDev` オプションが優先されます。

### Step を作成する

Workflow を構成する個別の Step を定義します。

```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` ループパターンを使用して、Step を Workflow に構成します。`createWorkflow()` 関数は、呼び出し可能な Function を Inngest Server 上に作成します。

```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 インスタンスを設定する

Workflow を Mastra に登録し、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' }),
})
```

> **注記:** パスは `/api/inngest` ではなく `/inngest/api` です。Mastra は組み込み Route（Agent、Workflow、Memory）用に `/api` Prefix を予約しています。Server の `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 は、Mastra の `/inngest/api` エンドポイントの場所を Inngest Dev Server に伝えます。

3. <http://localhost:8288> で Inngest Dashboard を開き、サイドバーの **Apps** セクションで Mastra Workflow が登録されていることを確認します。

4. **Functions** で Workflow を開きます。**Invoke** を選択し、次の入力を指定します。

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

5. **Runs** タブで Workflow の実行を監視し、Step ごとの進行状況を確認します。

### 本番環境で実行する

始める前に、次のものを用意してください。

- Vercel アカウントとインストール済みの Vercel CLI（`npm i -g vercel`）
- Inngest アカウント
- Vercel Token

1. 環境に Vercel Token を設定します。

   ```bash
   export VERCEL_TOKEN=your_vercel_token
   ```

2. Mastra インスタンスに `VercelDeployer` を追加します。

   ```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** フィールドに、デプロイ先の Origin と `/inngest/api` を組み合わせた URL（`https://your-app.vercel.app/inngest/api` など）を設定してください。デフォルトのままにすると、Inngest Dashboard はアプリの Function を検出できません。

6. **Functions** で `workflow.increment-workflow` を開きます。**All actions** > **Invoke** を選択し、次の入力を指定します。

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

7. **Runs** タブで実行を監視し、Step ごとの進行状況を確認します。

## カスタム Inngest Function を追加する

`serve()` の任意の `functions` パラメーターを使用すると、Mastra Workflow とともに追加の Inngest Function を提供できます。

### カスタム Function を作成する

最初に、カスタム Inngest Function を作成します。

```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 }
  },
)
```

### カスタム Function を Workflow とともに提供する

Mastra 設定を更新し、カスタム Function を Import して含めます。強調表示された行が追加部分です。

```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' }),
})
```

### Function の登録

カスタム Function を含めると、次のように登録されます。

1. Mastra Workflow は `workflow.${workflowId}` のような ID を持つ Inngest Function に自動変換されます。
2. カスタム Function は指定された ID（`send-welcome-email`、`process-webhook` など）を維持します。
3. すべての Function が同じ `/inngest/api` エンドポイントで提供されます。

これにより、Mastra の Workflow Orchestration と既存の Inngest Function を組み合わせられます。

## その他のフレームワークで使用する

デフォルトの `serve` 関数は内部で Hono を使用します。Express、Fastify、Koa などの別の Web フレームワークを使用する場合は、対応する Inngest Adapter とともに `createServe` Factory 関数を使用します。

### 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 }
```

### 利用可能な Adapter

`createServe` 関数は、すべての Inngest Adapter で動作します。AWS Lambda、Cloudflare Workers など、利用可能な Adapter の完全な一覧については、[Inngest serve ドキュメント](https://www.inngest.com/docs/reference/serve)を参照してください。

## Connect Worker として実行する

`serve()` は Inngest が呼び出す HTTP エンドポイントを公開します。代わりに `connect()` を使用すると、Worker から Inngest への長時間の外向き接続を開くため、Worker に公開エンドポイントは必要ありません。Kubernetes、Docker、ECS、Fly.io、Render などの Runtime で、長時間稼働する Worker プロセスに使用します。

> **注記:** Inngest Connect は Public Beta です。Vercel や AWS Lambda などの Serverless Runtime は Connect Worker をサポートしません。

### `serve()` の代わりに `connect()` を使用するタイミング

- Worker プロセスが、外向き接続のみを許可する Private Network 内で実行される。
- 負荷の高い Workflow の実行を、ユーザー向け HTTP トラフィックを処理するプロセスから分離したい。
- Worker ごとの同時実行制限を設定し、Mastra HTTP サーバーとは独立して Worker をスケールしたい。

同じ Mastra Workflow 定義を `serve()` と `connect()` のどちらでも使用できます。標準 Workflow、ネストされた Workflow、Cron Workflow、追加の Inngest Function の収集動作は同じです。

### 要件

- `inngest@^4`
- Node.js `22.13.0` 以降
- Worker 用の長時間稼働プロセス

### セットアップ

Mastra サーバーと Connect Worker を 2 つのプロセスとして実行します。この構成では、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` Flag は必要ありません。

本番環境では、`INNGEST_EVENT_KEY` と `INNGEST_SIGNING_KEY` を設定します。Inngest が Rolling Deploy を管理できるように、`Inngest` Client の `appVersion` に Commit SHA や Image Tag などのデプロイ識別子を設定します。

既存の本番アプリを `serve()` から `connect()` に移行する場合は、トラフィックを移行する前に、別の Inngest アプリで Worker をテストしてください。

### オプション

`connect()` は Inngest の [`connect`](https://www.inngest.com/docs/setup/connect) と同じオプションに加えて、Mastra 固有のフィールドを受け取ります。主なオプションは次のとおりです。

- `mastra`：公開する Workflow を持つ Mastra インスタンス。
- `inngest`：Inngest Client。`serve()` に渡すものと同じ Client を使用します。
- `functions`：Mastra Workflow とともに登録する追加の Inngest Function の任意の配列。
- `instanceId`：Inngest Dashboard に表示される Worker の安定した識別子。デフォルトはマシンの Hostname です。
- `maxWorkerConcurrency`：Worker が一度に実行する Step の最大数。デフォルトは無制限です。
- `registerOptions`：アプリ登録時に Inngest へ転送されます（`signingKey` など）。同じフィールドがここおよび Top Level に設定されている場合は、`registerOptions` が優先されます。これは `serve()` と同じ動作です。

`connect()` は Inngest の `WorkerConnection` を返します。Inngest SDK はデフォルトで `SIGINT` と `SIGTERM` を処理します。Worker で独自の Shutdown 制御が必要な場合にのみ、返された接続を保存して `.close()` を呼び出してください。

Mastra インスタンスに `InngestWorkflow` がなく、追加の `functions` も指定されていない場合、実行対象がないまま Worker が接続し続けるため、`connect()` は警告を記録します。Mastra に 1 つ以上の Workflow を登録するか、`functions: [...]` を渡してください。

## カスタム `apiPrefix` を使用する

`/api/inngest` を維持する必要がある場合（Dashboard の URL を変更せずに Inngest の自動検出規則に合わせる場合など）は、`server.apiPrefix` を設定して Mastra の組み込み Route を移動します。

```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 の組み込み Route は `/_mastra/agents`、`/_mastra/workflows` などに配置され、カスタム Route で `/api/inngest` パスを使用できるようになります。

> **警告:** デフォルトの認証設定は `/api/*` を保護し、`/api` と `/api/auth/*` を公開として扱います。`apiPrefix` を変更すると、これらのデフォルトと一致しなくなり、組み込み Route が保護パターンの対象外になります。`server.auth.protected` と `server.auth.public` を更新して新しい Prefix を参照し、`/api/*` を呼び出す Client コード（[`MastraClient`](https://mastra.zisheng.pro/ja/docs/server/mastra-client) の `apiPrefix` を含む）も更新してください。

## フロー制御

Inngest Workflow は、同時実行制限、Rate Limiting、Throttling、Debouncing、Priority Queue などのフロー制御機能をサポートします。これらのオプションは `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',
  },
})
```

### Rate Limiting

一定期間内の 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,
  },
})
```

### Throttling

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',
  },
})
```

### Debouncing

一定時間内に新しいイベントが届かなくなるまで、実行を遅延します。

```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',
  },
})
```

### Priority

Workflow の実行 Priority を設定します。

```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',
  },
})
```

### フロー制御オプションを組み合わせる

1 つの 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 Schedule

Cron 式を使用して、Inngest Workflow を Schedule に従って開始します。一般的な用途には、日次レポート、1 時間ごとのデータ同期、メンテナンスタスクなどがあります。

### 基本的な Cron Schedule

`cron` プロパティを追加し、Workflow を Schedule 実行するように設定します。

```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 Schedule の形式

`cron` プロパティは、`minute hour day month dayOfWeek` 形式の標準 Cron 式を受け取ります。

- **minute**：0～59
- **hour**：0～23
- **day**：1～31
- **month**：1～12 または JAN～DEC
- **dayOfWeek**：0～6（Sunday = 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'
```

### Schedule 実行に入力データを渡す

各 Schedule 実行で使用する静的な入力データを指定できます。

```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',
  },
})
```

### Schedule 実行に初期状態を渡す

Schedule 実行される 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 Schedule はフロー制御オプションと組み合わせられます。

```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 Function の仕組み

Workflow に `cron` プロパティを設定すると、次の処理が行われます。

1. ID が `workflow.${workflowId}.cron` の独立した Inngest Function が自動作成されます。
2. この Function が Inngest に登録され、指定した Schedule で開始されます。
3. Schedule 実行ごとに、指定した `inputData` と `initialState` を持つ新しい Workflow の Run が作成されます。
4. `serve()` を呼び出すと、Cron Function とメインの Workflow Function がともに提供されます。

Inngest Dashboard の **Functions** セクションと **Runs** セクションで、Schedule 実行を監視できます。Cron Function はメインの Workflow Function と並んで独立した Function として表示されます。

Cron Schedule について詳しくは、[Inngest Cron ドキュメント](https://www.inngest.com/docs/guides/scheduled-functions)を参照してください。