Inngest Infrastructure

Inngest handles background jobs and event-driven workflows for the FSA Platform.

Key Files

Type	Location
Client	`packages/core/src/infrastructure/inngest/client.ts`
Events	`packages/core/src/domain/events.ts`
Jobs	`packages/core/src/application/jobs/`

Event Definition

// packages/core/src/domain/events.ts
export type InngestEvents = {
  'organisation.create.provisional': {
    data: { accountId: string; name: string; recordId?: number };
  };
  'transaction.create': {
    data: { accountId: string; name: string; organisationId: string; segmentId?: string };
  };
};

Job Function Template

// packages/core/src/application/jobs/organisation/create-provisional.ts
import { inngest } from '../../../infrastructure/inngest/client';

export const createProvisionalOrganisation = inngest.createFunction(
  {
    id: 'organisation-create-provisional',
    concurrency: { limit: 100 },
    throttle: { limit: 1000, period: '1m', burst: 100 },
  },
  { event: 'organisation.create.provisional' },
  async ({ event, step }) => {
    const { accountId, name, recordId } = event.data;

    // Step 1: Create organisation
    const org = await step.run('create-provisional', async () => {
      return new CreateProvisionalOrganisation().execute({ name });
    });

    // Step 2: Link to account
    if (accountId) {
      await step.run('link-to-account', async () => {
        await AccountRepository.addOrganisation(accountId, { organisationId: org.id, name: org.name, recordId });
      });
    }

    return { organisationId: org.id, name: org.name };
  }
);

Step Patterns

Sequential Steps

const result1 = await step.run('step-1', () => doFirst());
const result2 = await step.run('step-2', () => doSecond(result1));

Parallel Steps

const [a, b] = await Promise.all([
  step.run('parallel-1', () => doA()),
  step.run('parallel-2', () => doB()),
]);

Wait for Event

const childEvent = await step.waitForEvent('wait-for-child', {
  event: 'child.completed',
  match: 'data.parentId',
  timeout: '1h',
});

Send Events

await step.sendEvent('emit-event', {
  name: 'something.happened',
  data: { id: '123' },
});

Error Handling

Non-Fatal Errors

await step.run('optional-step', async () => {
  try {
    return await riskyOperation();
  } catch (error) {
    console.error('Non-fatal:', error);
    return null;  // Continue execution
  }
});

Failure Handler

inngest.createFunction(
  {
    id: 'my-function',
    retries: 3,
    onFailure: async ({ error, event }) => {
      await notifyFailure(error, event);
    },
  },
  { event: 'my.event' },
  async ({ event, step }) => { ... }
);

Concurrency & Throttling

{
  id: 'my-function',
  concurrency: {
    limit: 100,
    key: 'event.data.accountId',  // Per account
  },
  throttle: {
    limit: 1000,
    period: '1m',
    burst: 100,
  },
}

Prioritising interactive work over bulk (shared pool)

When one event class is user-waiting (interactive) and another is latency-tolerant (bulk), cap bulk so a slice is always free for interactive, let interactive spill into idle bulk slots, and give it queue priority. The enrichment pipeline (ADR-0007, issue #1360) shares one bulk pool across all three of its functions via an env-scoped key:

import { ENRICHMENT_CONCURRENCY, INTERACTIVE_PRIORITY_RUN } from '@repo/core/infrastructure/inngest/concurrency';

inngest.createFunction(
  {
    id: 'organisation-enrich',                    // same on load-statements + generate-issues
    concurrency: ENRICHMENT_CONCURRENCY,          // [ shared env-scoped bulk pool, per-org ]
    priority: { run: INTERACTIVE_PRIORITY_RUN },  // interactive backdated 600s → dequeues first
    triggers: [organisationEnrich],
  },
  handler,
);

ENRICHMENT_CONCURRENCY is two constraints (Inngest's max):

[
  // scope:'env' + the same 'enrichment-bulk' key on all three functions → ONE shared bulk pool.
  // Bulk/unset events share it (capped); interactive keys on the unique event.id → uncapped → spills.
  { scope: 'env', key: "event.data.priority == 'interactive' ? event.id : 'enrichment-bulk'", limit: bulk },
  { limit: 1, key: 'event.data.organisationId' }, // function-scoped: one run per org per phase
]

Events carry data.priority: 'interactive' | 'bulk' (unset ⇒ bulk). One env knob sizes the shared bulk pool; interactive is the prioritised remainder up to your Inngest account/env ceiling:

Var	Default	Meaning
`INNGEST_ENRICHMENT_BULK_CONCURRENCY`	40	Max concurrent bulk runs across enrich + statements + issues combined

Notes:

Set the Inngest account/env concurrency to your budget; the interactive reserve is ceiling − bulk, always free.
scope: 'env' + a shared key is how a concurrency limit spans multiple functions (the per-org lock stays function-scoped).
priority is a separate function-config field — it does not count against the 2-concurrency-constraint limit.
Concurrency is enforced on Cloud and the dev server; priority ordering is only visible under a saturated backlog.
Concurrency is read at registration time, so changing an env var needs a redeploy / re-sync.

Reserving production capacity (prod vs non-prod)

Inngest's account plan concurrency is one ceiling shared by every environment (production, staging, every preview branch) — scope:'env' limits cap within an environment but never the cross-environment total, and Inngest's native deprioritisation of non-prod environments only reorders latency, it doesn't reserve capacity. To keep preview/staging from drawing down what production needs, every function carries an account-scoped reservation that — in non-production only — buckets its runs into shared pools, each capped at INNGEST_NONPROD_CONCURRENCY:

'nonprod-interactive' — interactive enrichment, so a bulk upload can't starve it;
'nonprod-bulk' — all other non-prod work.

Total non-prod ≤ 2 × INNGEST_NONPROD_CONCURRENCY, leaving the rest of the ceiling for production. In production it is a no-op — every function keeps its existing config. The branch is baked from process.env.VERCEL_ENV at sync time (prod and non-prod sync independently). Use the helpers from @repo/core/inngest/concurrency so new functions opt in:

import { overallConcurrency, withNonProdReservation } from '@repo/core/inngest/concurrency';

// Function with ONE existing constraint — unchanged in prod, joins the non-prod pool otherwise:
concurrency: withNonProdReservation({ limit: 100 }),

// Function already at 2 constraints (overall cap + a per-entity lock) — swap the overall cap:
concurrency: [overallConcurrency(50), { limit: 1, key: 'event.data.organisationId' }],

Var	Default	Meaning
`INNGEST_NONPROD_CONCURRENCY`	10	Per-pool cap for non-prod; total non-prod ≤ `2×` this

Notes:

Set INNGEST_NONPROD_CONCURRENCY on Vercel's Preview scope; keep 2× value below the account ceiling.
A function added without the reservation leaks non-prod budget — always use the helpers.
Cross-environment (account-scope) behaviour is only observable on Cloud — validate by saturating staging plus a preview and confirming their combined running count caps at 2×INNGEST_NONPROD_CONCURRENCY.
See ADR-0020 for the full rationale (including why a separate Inngest account was rejected).

Triggering Jobs

import { inngest } from '@repo/core/infrastructure/inngest/client';

// From use case or server action
await inngest.send({
  name: 'organisation.create.provisional',
  data: { accountId, name },
});

Best Practices

Use steps for durability — each step is checkpointed
Idempotent operations — steps may be retried
Meaningful step names — for debugging in the Inngest dashboard
Fire-and-forget fan-out — use step.sendEvent to dispatch child work, avoid waitForEvent at scale
Use step.invoke — to call child functions and get their return values (1 step, no event matching)
Concurrency limits — prevent overwhelming resources
Stay under 1000 steps — Inngest limit per function run; use fan-out to child functions instead

Type	Location
Client	`packages/core/src/infrastructure/inngest/client.ts`
Events	`packages/core/src/domain/events.ts`
Jobs	`packages/core/src/application/jobs/`
API Route	`apps/web/src/app/api/inngest/route.ts`

On this page