Workflow Inngest

Inngest est une plateforme destinée aux développeurs qui permet de créer et d'exécuter des workflows en arrière-plan sans gérer d'infrastructure.

Pour consulter un exemple complet intégrant des fonctionnalités avancées de contrôle de flux, reportez-vous à l'exemple de workflow Inngest.

Fonctionnement d'Inngest avec Mastra

Inngest et Mastra s'intègrent en faisant correspondre leurs modèles de workflow : Inngest organise la logique en fonctions composées d'étapes, et les workflows Mastra définis avec createWorkflow() et createStep() s'adaptent directement à cette structure. Chaque workflow Mastra devient une fonction Inngest dotée d'un identifiant unique, tandis que chaque étape du workflow correspond à une étape Inngest.

La fonction serve() relie les deux systèmes en enregistrant les workflows Mastra comme fonctions Inngest et en configurant les gestionnaires d'événements nécessaires à leur exécution et à leur suivi.

Lorsqu'un événement déclenche un workflow, Inngest l'exécute étape par étape et mémorise chaque résultat. En cas de nouvelle tentative ou de reprise, Inngest ignore les étapes déjà terminées en s'appuyant sur les résultats enregistrés. Les primitives de contrôle de flux de Mastra, comme les boucles, les conditions et les workflows imbriqués, correspondent au même modèle de fonctions et d'étapes Inngest, tout en préservant la composition, les branchements et la suspension.

Le suivi en temps réel, la suspension et la reprise, ainsi que l'observabilité au niveau des étapes sont assurés par le système de publication-abonnement et le tableau de bord d'Inngest. À mesure que chaque étape s'exécute, son état et sa sortie sont suivis à l'aide du stockage Mastra et peuvent être repris si nécessaire.

Configuration

Installez les packages requis :

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

remarque

Nécessite inngest@^4 et Inngest Dev Server v1.18.0 ou une version ultérieure. Le temps réel est intégré au SDK à partir de la v4 ; @inngest/realtime et realtimeMiddleware ne sont donc plus utilisés.

Créer un workflow Inngest

Ce guide explique comment créer un workflow avec Inngest et Mastra à travers une application de compteur qui incrémente une valeur jusqu'à ce qu'elle atteigne 10.

Initialiser Inngest

Initialisez l'intégration Inngest afin d'obtenir des fonctions auxiliaires de workflow compatibles avec Mastra. Les fonctions createWorkflow() et createStep() servent à créer des objets de workflow et d'étape compatibles avec Mastra et Inngest.

En développement :

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

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

En production :

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

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

Choisir le mode développement ou cloud avec INNGEST_DEV

Le SDK Inngest fonctionne dans l'un des deux modes suivants : développement ou cloud. Depuis la v4, le SDK utilise le mode cloud par défaut et nécessite INNGEST_EVENT_KEY et INNGEST_SIGNING_KEY. Dans les exemples ci-dessus, le mode est défini dans le code avec isDev: true. Vous pouvez également définir la variable d'environnement INNGEST_DEV afin d'utiliser le même code client dans tous les environnements :

• INNGEST_DEV=1 : force le mode développement. Le SDK communique avec un Inngest Dev Server local et désactive la vérification des signatures.
• INNGEST_DEV=0 : force le mode cloud. Le SDK communique avec Inngest Cloud et nécessite des clés d'événement et de signature.
• INNGEST_DEV=<url> : force le mode développement et dirige le SDK vers le Dev Server à l'adresse <url>, par exemple http://localhost:8288.
• Non définie : utilise le mode cloud par défaut.

Lorsque INNGEST_DEV est définie, vous pouvez supprimer isDev et baseUrl du client :

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

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

attention

Ne définissez pas INNGEST_DEV en production. Cette variable désactive la vérification des signatures, indispensable pour recevoir en toute sécurité les événements d'Inngest Cloud.

L'option isDev de new Inngest() prévaut sur INNGEST_DEV lorsque les deux sont définies.

Créer des étapes

Définissez les différentes étapes qui composeront votre 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 }
  },
})

Créer le workflow

Composez les étapes au sein d'un workflow à l'aide du modèle de boucle dountil. La fonction createWorkflow() crée sur le serveur Inngest une fonction qui peut être invoquée.

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 }

Configurer l'instance Mastra

Enregistrez le workflow auprès de Mastra et configurez le point de terminaison de l'API Inngest :

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

remarque

Le chemin est /inngest/api, et non /api/inngest. Mastra réserve le préfixe /api aux routes intégrées (agents, workflows, memory). Les chemins apiRoutes personnalisés qui commencent par l'apiPrefix du serveur (/api par défaut) provoquent une erreur au démarrage. Consultez #15743 pour plus de contexte, ou passez directement à la section Utiliser un apiPrefix personnalisé si vous devez conserver /api/inngest.

Exécuter des workflows

Exécuter localement

1. Exécutez npx mastra dev pour démarrer le serveur Mastra localement sur le port 4111

2. Démarrez Inngest Dev Server. Dans un nouveau terminal, exécutez :

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

   remarque

   L'URL qui suit -u indique au serveur de développement Inngest où trouver votre point de terminaison Mastra /inngest/api

3. Ouvrez le tableau de bord Inngest à l'adresse http://localhost:8288, puis accédez à la section Apps de la barre latérale pour vérifier que votre workflow Mastra est enregistré

4. Dans Functions, ouvrez votre workflow. Sélectionnez Invoke et fournissez l'entrée suivante :

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

5. Suivez l'exécution du workflow dans l'onglet Runs pour consulter sa progression étape par étape

Exécuter en production

Avant de commencer, assurez-vous de disposer des éléments suivants :

• Un compte Vercel et la CLI Vercel installée (npm i -g vercel)
• Un compte Inngest
• Un token Vercel

1. Définissez votre token Vercel dans votre environnement :

   .env

   export VERCEL_TOKEN=your_vercel_token

2. Ajoutez VercelDeployer à l'instance Mastra

   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. Compilez l'instance Mastra

   npx mastra build

4. Déployez sur Vercel

   cd .mastra/output
   vercel login
   vercel --prod

5. Synchronisez l'application avec le tableau de bord Inngest en sélectionnant Sync new app with Vercel, puis en suivant les instructions

   attention

   La convention de découverte automatique d'Inngest suppose le chemin /api/inngest. Comme ce guide utilise /inngest/api, définissez le champ URL de l'application Inngest sur l'origine de votre déploiement suivie de /inngest/api (par exemple https://your-app.vercel.app/inngest/api). Si vous conservez la valeur par défaut, le tableau de bord Inngest ne trouvera pas les fonctions de votre application.

6. Dans Functions, ouvrez workflow.increment-workflow. Sélectionnez All actions > Invoke et fournissez l'entrée suivante :

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

7. Suivez l'exécution dans l'onglet Runs pour consulter sa progression étape par étape

Ajouter des fonctions Inngest personnalisées

Vous pouvez servir des fonctions Inngest supplémentaires aux côtés de vos workflows Mastra à l'aide du paramètre facultatif functions de serve().

Créer des fonctions personnalisées

Commencez par créer vos fonctions Inngest personnalisées :

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

Servir des fonctions personnalisées avec les workflows

Mettez à jour votre configuration Mastra afin d'importer et d'inclure les fonctions personnalisées. Les lignes mises en évidence montrent les ajouts :

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

Enregistrement des fonctions

Lorsque vous incluez des fonctions personnalisées :

1. Les workflows Mastra sont automatiquement convertis en fonctions Inngest avec des ID tels que workflow.${workflowId}
2. Les fonctions personnalisées conservent les ID qui leur ont été attribués (par exemple, send-welcome-email, process-webhook)
3. Toutes les fonctions sont servies ensemble sur le même point de terminaison /inngest/api

Vous pouvez ainsi associer l'orchestration des workflows de Mastra à vos fonctions Inngest existantes.

Utiliser d'autres frameworks

La fonction serve utilise Hono en interne par défaut. Si vous utilisez un autre framework web, comme Express, Fastify ou Koa, employez la fonction de fabrique createServe avec l'adaptateur Inngest approprié.

Express

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)

Fastify

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

Koa

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 }

Adaptateurs disponibles

La fonction createServe fonctionne avec tous les adaptateurs Inngest. Consultez la documentation Inngest sur serve pour obtenir la liste complète des adaptateurs disponibles, notamment AWS Lambda, Cloudflare Workers et bien d'autres.

Exécuter en tant que worker Connect

serve() expose un point de terminaison HTTP appelé par Inngest. À la place, connect() peut ouvrir une connexion sortante de longue durée entre votre worker et Inngest ; le worker n'a alors pas besoin d'un point de terminaison accessible publiquement. Utilisez cette approche pour les processus worker de longue durée exécutés dans des environnements tels que Kubernetes, Docker, ECS, Fly.io ou Render.

remarque

Inngest Connect est en bêta publique. Les environnements d'exécution serverless tels que Vercel et AWS Lambda ne prennent pas en charge les workers Connect.

Quand utiliser connect() plutôt que serve()

• Le processus worker s'exécute dans un réseau privé qui n'autorise que les connexions sortantes.
• Vous souhaitez séparer l'exécution intensive des workflows du processus qui sert le trafic HTTP destiné aux utilisateurs.
• Vous souhaitez ajuster le nombre de workers indépendamment du serveur HTTP Mastra, avec des limites de concurrence propres à chaque worker.

Utilisez les mêmes définitions de workflows Mastra avec serve() ou connect(). Le comportement de collecte est identique pour les workflows standard, les workflows imbriqués, les workflows cron et les fonctions Inngest supplémentaires.

Prérequis

• inngest@^4
• Node.js 22.13.0 ou version ultérieure
• Un processus de longue durée pour le worker

Configuration

Exécutez le serveur Mastra et le worker Connect dans deux processus distincts. Dans cette configuration, le serveur Mastra n'a pas besoin d'exposer /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),
})

Pendant le développement local, démarrez le worker avec INNGEST_DEV=1 node --import tsx src/worker.ts. Inngest Dev Server découvre le worker par l'intermédiaire de la connexion sortante ; l'option -u n'est donc pas nécessaire.

En production, définissez INNGEST_EVENT_KEY et INNGEST_SIGNING_KEY. Définissez appVersion sur le client Inngest avec un identifiant de déploiement, tel qu'un SHA de commit ou un tag d'image, afin qu'Inngest puisse gérer les déploiements progressifs.

Lorsque vous faites migrer une application de production existante de serve() vers connect(), testez le worker avec une application Inngest distincte avant de rediriger le trafic.

Options

connect() accepte les mêmes options que la fonction connect d'Inngest, ainsi que des champs propres à Mastra. Voici les plus courants :

• mastra : instance Mastra dont les workflows doivent être exposés.
• inngest : client Inngest. Utilisez le même client que celui que vous transmettriez à serve().
• functions : tableau facultatif de fonctions Inngest supplémentaires à enregistrer aux côtés des workflows Mastra.
• instanceId : identifiant stable du worker, affiché dans le tableau de bord Inngest. Par défaut, il correspond au nom d'hôte de la machine.
• maxWorkerConcurrency : nombre maximal d'étapes exécutées simultanément par le worker. La valeur par défaut est illimitée.
• registerOptions : options transmises à Inngest lors de l'enregistrement de l'application (par exemple signingKey). Lorsqu'un champ est défini à la fois ici et au niveau supérieur, registerOptions prévaut. Ce comportement correspond à celui de serve().

connect() renvoie l'objet WorkerConnection d'Inngest. Le SDK Inngest gère SIGINT et SIGTERM par défaut. Conservez la connexion renvoyée et appelez .close() uniquement si votre worker nécessite un contrôle personnalisé de l'arrêt.

Si l'instance Mastra ne contient aucun InngestWorkflow et qu'aucune functions supplémentaire n'est fournie, connect() consigne un avertissement, car le worker resterait sinon connecté sans rien à exécuter. Enregistrez au moins un workflow dans Mastra ou transmettez functions: [...].

Utiliser un apiPrefix personnalisé

Si vous devez conserver /api/inngest (par exemple pour respecter la convention de découverte automatique d'Inngest sans modifier l'URL du tableau de bord), définissez server.apiPrefix afin de déplacer les routes intégrées de 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 }),
      },
    ],
  },
})

Les routes intégrées de Mastra sont désormais accessibles sous /_mastra/agents, /_mastra/workflows, etc., ce qui libère le chemin /api/inngest pour votre route personnalisée.

attention

La configuration d'authentification par défaut protège /api/* et considère /api et /api/auth/* comme publics. Lorsque vous modifiez apiPrefix, ces valeurs par défaut ne correspondent plus et les routes intégrées sortent du modèle protégé. Mettez à jour server.auth.protected et server.auth.public pour référencer le nouveau préfixe, ainsi que tout code client (notamment MastraClient et son apiPrefix) qui accède à /api/*.

Contrôle de flux

Les workflows Inngest prennent en charge des fonctionnalités de contrôle de flux, notamment les limites de concurrence, la limitation du débit, le throttling, le debounce et la mise en file d'attente prioritaire. Ces options sont configurées dans l'appel à createWorkflow() et facilitent la gestion de l'exécution des workflows à grande échelle.

Concurrence

Contrôlez le nombre d'instances de workflow pouvant s'exécuter simultanément :

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

Limitation du débit

Limitez le nombre d'exécutions du workflow sur une période donnée :

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

Imposez un intervalle minimal entre les exécutions du 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',
  },
})

Debounce

Retardez l'exécution jusqu'à ce qu'aucun nouvel événement n'arrive pendant une période donnée :

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

Priorité

Définissez la priorité d'exécution des workflows :

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

Combiner les options de contrôle de flux

Plusieurs options de contrôle de flux peuvent être combinées dans un même workflow :

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

Toutes les options de contrôle de flux sont facultatives. Si elles ne sont pas indiquées, les workflows s'exécutent selon le comportement par défaut d'Inngest. Pour en savoir plus, consultez la documentation d'Inngest sur le contrôle de flux.

Planification cron

Utilisez des expressions cron pour déclencher des workflows Inngest selon une planification. Les cas d'utilisation courants comprennent les rapports quotidiens, les synchronisations de données horaires et les tâches de maintenance.

Planification cron de base

Configurez l'exécution planifiée d'un workflow en ajoutant une propriété cron :

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

Format de la planification cron

La propriété cron accepte les expressions cron standard au format suivant : minute hour day month dayOfWeek

• minute : 0-59
• hour : 0-23
• day : 1-31
• month : 1-12 ou JAN-DEC
• dayOfWeek : 0-6 (dimanche = 0) ou SUN-SAT

Modèles cron courants :

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

Fournir des données d'entrée aux exécutions planifiées

Vous pouvez fournir des données d'entrée statiques qui seront utilisées à chaque exécution planifiée :

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

Fournir un état initial aux exécutions planifiées

Vous pouvez également définir un état initial pour les exécutions planifiées du 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: '',
  },
})

Combiner cron et le contrôle de flux

La planification cron peut être combinée avec des options de contrôle de flux :

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

Fonctionnement des fonctions cron

Lorsque vous configurez un workflow avec une propriété cron :

1. Une fonction Inngest distincte est automatiquement créée avec l'ID workflow.${workflowId}.cron
2. Cette fonction est enregistrée auprès d'Inngest et se déclenche selon la planification indiquée
3. Chaque exécution planifiée crée une nouvelle exécution de workflow avec les valeurs inputData et initialState fournies
4. La fonction cron et la fonction principale du workflow sont toutes deux servies lorsque vous appelez serve()

Vous pouvez suivre les exécutions planifiées dans les sections Functions et Runs du tableau de bord Inngest. La fonction cron apparaît comme une fonction distincte aux côtés de la fonction principale de votre workflow.

Pour en savoir plus sur la planification cron, consultez la documentation d'Inngest sur cron.