수집 Workflow

섭취인프라를 관리하지 않고도 백그라운드 Workflow를 구축하고 실행하기 위한 개발자 플랫폼입니다.

고급 흐름 제어 기능의 전체 예는 Inngest Workflow 예제를 참조하세요.

Ingest가 Mastra와 작동하는 방식

Inngest와 Mastra는 Workflow Model을 일치시키는 방식으로 통합됩니다. Inngest는 로직을 단계로 구성된 함수로 구성하며, createWorkflow() 및 createStep()을 사용해 정의한 Mastra Workflow는 이 구조에 직접 매핑됩니다. 각 Mastra Workflow는 고유 식별자를 가진 Inngest 함수가 되고, Workflow 내의 각 단계는 Inngest 단계에 매핑됩니다. serve() 함수는 Mastra Workflow를 Inngest 함수로 등록하고 실행 및 모니터링에 필요한 이벤트 핸들러를 설정하여 두 시스템을 연결합니다. 이벤트가 Workflow를 트리거하면 Ingest는 이를 단계별로 실행하고 각 결과를 메모합니다. 재시도 또는 재개 시 Ingest는 저장된 결과를 기반으로 완료된 단계를 건너뜁니다. 루프, 조건부 및 중첩된 Workflow와 같은 Mastra 제어 흐름 기본 요소는 구성, 분기 및 정지를 유지하면서 동일한 Ingest 함수 및 단계 Model에 매핑됩니다.

실시간 모니터링, 일시 중지/재개 및 단계 수준 관찰 기능은 Ingest의 게시-구독 시스템 및 대시보드를 통해 활성화됩니다. 각 단계가 실행될 때 해당 상태와 출력은 Mastra 스토리지를 사용하여 추적되며 필요에 따라 재개될 수 있습니다.

설정

필수 패키지를 설치합니다:

npm

npm install @mastra/inngest@latest inngest

pnpm

pnpm add @mastra/inngest@latest inngest

Yarn

yarn add @mastra/inngest@latest inngest

Bun

bun add @mastra/inngest@latest inngest

:::참고 inngest@^4 및 Inngest Dev Server v1.18.0 이상이 필요합니다. v4에서는 Realtime이 SDK에 내장되어 있으므로 @inngest/realtime 및 realtimeMiddleware는 더 이상 사용하지 않습니다. :::

Ingest Workflow 구축

이 가이드는 Ingest 및 Mastra를 사용하여 Workflow를 생성하는 과정을 안내하며 10에 도달할 때까지 값을 증가시키는 카운터 애플리케이션을 보여줍니다.

수집 초기화

Mastra 호환 Workflow 도우미를 가져오려면 Inngest 통합을 초기화하세요. createWorkflow() 및 createStep() 함수는 Mastra 및 Inngest와 호환되는 Workflow 및 단계 객체를 생성하는 데 사용됩니다. 개발 중:

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

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

생산 중:

src/mastra/inngest/index.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 환경 변수를 설정하면 환경이 달라도 동일한 클라이언트 코드를 사용할 수 있습니다.

• 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을 생략할 수 있습니다.

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

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

경고

프로덕션에서는 INNGEST_DEV를 설정하지 마세요. 이 변수는 Inngest Cloud에서 이벤트를 안전하게 수신하는 데 필요한 서명 검증을 비활성화합니다.

둘 다 설정된 경우 new Inngest()의 isDev 옵션이 INNGEST_DEV보다 우선합니다.

단계 만들기

Workflow를 구성할 개별 단계를 정의하십시오.

src/mastra/workflows/index.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 서버에 호출 가능한 함수를 생성합니다.

src/mastra/workflows/index.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를 등록하고 Ingest API 엔드포인트를 구성합니다.

src/mastra/index.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는 기본 제공 경로(Agent, Workflow, Memory)에 /api 접두사를 예약합니다. 서버의 apiPrefix(기본값 /api)로 시작하는 사용자 지정 apiRoutes 경로는 시작 시 오류를 발생시킵니다. 자세한 배경은 #15743을 참조하세요. /api/inngest를 유지해야 한다면 사용자 지정 apiPrefix 사용으로 이동하세요. :::

Workflow 실행

로컬에서 실행

1. npx mastra dev를 실행하여 포트 4111에서 Mastra 서버를 로컬로 시작하세요

2. Ingest 개발 서버를 시작합니다. 새 터미널에서 다음을 실행합니다.

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

   :::참고 URL 앞의 -u는 Mastra의 /inngest/api 엔드포인트를 찾을 위치를 Inngest Dev Server에 알려 줍니다 :::

3. http://localhost:8288에서 Inngest 대시보드를 열고 사이드바의 Apps 섹션으로 이동하여 Mastra Workflow가 등록되었는지 확인하세요

4. ~ Functions에서 Workflow를 여세요. Invoke를 선택하고 다음 입력을 제공하세요.

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

5. Runs 탭에서 Workflow 실행을 모니터링하여 단계별 실행 진행 상황을 확인하세요

프로덕션에서 실행 중

시작하기 전에 다음 사항을 확인하세요.

• Vercel 계정 및 Vercel CLI 설치(npm i -g vercel)
• 수집 계정
• Vercel 토큰

1. 사용자 환경에서 Vercel 토큰을 설정하십시오.

   .env

   export VERCEL_TOKEN=your_vercel_token

2. Mastra 인스턴스에 VercelDeployer를 추가하세요

   src/mastra/index.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. 마스터라 인스턴스 빌드

   npx mastra build

4. Vercel에 배포

   cd .mastra/output
   vercel login
   vercel --prod

5. Inngest 대시보드에서 Sync new app with Vercel을 선택하고 안내에 따라 동기화하세요

   경고

   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를 선택하고 다음 입력을 제공하세요.

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

7. Runs 탭에서 실행을 모니터링하여 단계별 진행 상황을 확인하세요

사용자 정의 Ingest 함수 추가

serve()의 functions 매개변수를 사용하면 Mastra Workflow와 함께 추가 Inngest 함수를 제공할 수 있습니다.

사용자 정의 함수 만들기

먼저 사용자 지정 Ingest 함수를 만듭니다.

src/inngest/custom-functions.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 구성을 업데이트하십시오. 강조 표시된 줄은 추가 사항을 보여줍니다.

src/mastra/index.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를 사용하여 Ingest 함수로 자동 변환됩니다.workflow.${workflowId}
2. 맞춤 함수는 지정된 ID를 유지합니다(예:send-welcome-email, process-webhook)
3. 모든 기능은 동일하게 함께 제공됩니다./inngest/api endpoint

이를 통해 Mastra의 Workflow 조정을 기존 Ingest 기능과 결합할 수 있습니다.

다른 프레임워크와 함께 사용

기본 serve 함수는 내부적으로 Hono를 사용합니다. Express, Fastify 또는 Koa와 같은 다른 웹 프레임워크를 사용한다면 적절한 Inngest 어댑터와 함께 createServe 팩터리 함수를 사용하세요.

표현하다

src/server.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)

고정하다

src/server.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 })

코아

src/server.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

app/inngest/api/route.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 문서를 참조하세요.

Connect 작업자로 실행

serve()는 Inngest가 호출하는 HTTP 엔드포인트를 노출합니다. 반면 connect()는 작업자에서 Inngest로 장기 실행 아웃바운드 연결을 열기 때문에 작업자에 공개적으로 접근 가능한 엔드포인트가 필요하지 않습니다. Kubernetes, Docker, ECS, Fly.io 또는 Render와 같은 런타임의 장기 실행 작업자 프로세스에 사용하세요. :::참고 Ingest Connect는 공개 베타 버전입니다. Vercel 및 AWS Lambda와 같은 서버리스 런타임은 Connect 작업자에 대해 지원되지 않습니다. :::

serve() 대신 connect()를 사용해야 하는 경우

• 작업자 프로세스는 아웃바운드 연결만 허용하는 개인 네트워크에서 실행됩니다.
• 사용자 대상 HTTP 트래픽을 처리하는 동일한 프로세스에서 과도한 Workflow 실행을 방지하려고 합니다.
• 작업자당 동시성 제한을 사용하여 Mastra HTTP 서버와 독립적으로 작업자를 확장 및 축소하려고 합니다.

serve() 또는 connect()에 동일한 Mastra Workflow 정의를 사용하세요. 표준 Workflow, 중첩 Workflow, cron Workflow 및 추가 Inngest 함수의 수집 동작은 동일합니다.

요구사항

• inngest@^4
• Node.js 22.13.0 이상
• 작업자를 위한 장기 실행 프로세스

설정

Mastra 서버와 Connect 작업자를 두 개의 프로세스로 실행합니다. 이 설정의 Mastra 서버는 노출할 필요가 없습니다./inngest/api:

src/mastra/index.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' }),
})

src/worker.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로 작업자를 시작하세요. Inngest Dev Server가 아웃바운드 연결을 통해 작업자를 검색하므로 -u 플래그가 필요하지 않습니다. 프로덕션에서는 INNGEST_EVENT_KEY 및 INNGEST_SIGNING_KEY를 설정하세요. Inngest가 롤링 배포를 관리할 수 있도록 Inngest 클라이언트의 appVersion을 커밋 SHA 또는 이미지 태그와 같은 배포 식별자로 설정하세요. 기존 프로덕션 앱을 serve()에서 connect()로 마이그레이션한다면 트래픽을 전환하기 전에 별도의 Inngest 앱에서 작업자를 테스트하세요.

옵션

connect()는 Inngest의 connect와 동일한 옵션에 Mastra 전용 필드를 더해 지원합니다. 가장 일반적으로 사용하는 필드는 다음과 같습니다.

• mastra: Workflow를 노출할 Mastra 인스턴스입니다.
• inngest: Inngest 클라이언트입니다. serve()에 전달할 때와 동일한 클라이언트를 사용하세요.
• functions: Mastra Workflow와 함께 등록할 추가 Inngest 함수의 선택적 배열입니다.
• instanceId: Inngest 대시보드에 표시되는 작업자의 안정적인 식별자입니다. 기본값은 컴퓨터 호스트 이름입니다.
• maxWorkerConcurrency: 작업자가 동시에 실행하는 최대 단계 수입니다. 기본값은 무제한입니다.
• registerOptions: 앱 등록 중 Inngest에 전달됩니다(예: signingKey). 필드가 여기와 최상위 수준에 모두 설정되면 registerOptions가 우선합니다. 이는 serve()의 동작과 일치합니다. connect()는 Inngest WorkerConnection을 반환합니다. Inngest SDK는 기본적으로 SIGINT 및 SIGTERM을 처리합니다. 작업자에서 종료를 직접 제어해야 할 때만 반환된 연결을 저장하고 .close()를 호출하세요. Mastra 인스턴스에 InngestWorkflow가 없고 추가 functions도 제공되지 않으면 connect()가 경고를 기록합니다. 그렇지 않으면 작업자가 실행할 항목 없이 연결된 상태로 유지되기 때문입니다. Mastra에 Workflow를 하나 이상 등록하거나 functions: [...]를 전달하세요.

사용자 정의 사용apiPrefix

/api/inngest를 유지해야 한다면(예: 대시보드 URL을 변경하지 않고 Inngest의 자동 검색 규칙과 일치시키려면) server.apiPrefix를 설정하여 Mastra의 기본 제공 경로를 다른 위치로 옮기세요.

src/mastra/index.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의 apiPrefix 포함)도 업데이트하세요.

흐름 제어

Inngest Workflow는 동시성 제한, 속도 제한, 스로틀링, 디바운싱 및 우선순위 대기열을 포함한 흐름 제어 기능을 지원합니다. 이러한 옵션은 createWorkflow() 호출에서 구성하며 대규모 Workflow 실행을 관리하는 데 도움이 됩니다.

동시성

동시에 실행할 수 있는 Workflow 인스턴스 수를 제어합니다.

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 실행 수를 제한합니다.

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 실행 사이의 최소 시간을 보장합니다.

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

디바운싱

일정 기간 내에 새 이벤트가 도착하지 않을 때까지 실행을 지연합니다.

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의 실행 우선순위를 설정합니다.

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

흐름 제어 옵션 결합

여러 흐름 제어 옵션을 단일 작업 흐름에 결합할 수 있습니다.

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는 Ingest의 기본 동작으로 실행됩니다. 자세한 내용은 다음을 참조하세요.Inngest flow control documentation.

크론 스케줄링

cron 표현식을 사용하여 일정에 따라 Ingest Workflow를 트리거합니다. 일반적인 용도에는 일일 보고서, 시간별 데이터 동기화, 유지 관리 작업이 포함됩니다.

기본 크론 스케줄링

다음을 추가하여 일정에 따라 실행되도록 Workflow를 구성합니다.cron property:

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 속성은 minute hour day month dayOfWeek 형식의 표준 cron 표현식을 허용합니다.

• 분: 0-59
• 시간: 0-23
• 낮: 1-31
• 월: 112일 또는 1월12월
• 요일: 0-6(일요일 = 0) 또는 SUN-SAT

일반적인 크론 패턴:

// 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'

예약된 실행을 위한 입력 데이터 제공

예약된 각 실행에 사용될 정적 입력 데이터를 제공할 수 있습니다.

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 실행의 초기 상태를 설정할 수도 있습니다.

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 스케줄링은 흐름 제어 옵션과 결합될 수 있습니다.

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

크론 기능의 작동 방식

Workflow를 구성하는 경우cron property:

1. ID가 workflow.${workflowId}.cron인 별도의 Inngest 함수가 자동으로 생성됩니다
2. 이 함수는 Inngest에 등록되고 지정된 일정에 따라 트리거됩니다.
3. 예약된 각 실행은 제공된 inputData 및 initialState를 사용하여 새 Workflow 실행을 생성합니다
4. serve()를 호출하면 cron 함수와 기본 Workflow 함수가 모두 제공됩니다 Inngest 대시보드의 Functions 및 Runs 섹션에서 예약된 실행을 모니터링할 수 있습니다. cron 함수는 기본 Workflow 함수와 별도의 함수로 표시됩니다. 크론 예약에 대한 자세한 내용은 다음을 참조하세요.Inngest cron documentation.