Build event workflows with Pathways

@flowcore/pathways is the application-facing library for defining type-safe Flowcore event workflows. It connects runtime schemas, event publishing, handlers, resource provisioning, and optional in-process pumping through one builder.

Use it when you want the compiler and runtime validator to agree about which event paths exist and what their payloads contain.

The PathwaysBuilder mental model

A builder has three main responsibilities:

1. Register event contracts.
2. Publish writable events.
3. Route received events to handlers.

The builder does not make every registered handler run immediately after a write. Publishing and consuming are separate operations. A managed Data Pathway or virtual pump must deliver the stored event back to the processing runtime.

Create a builder

import { PathwaysBuilder } from "@flowcore/pathways"
 
const pathways = new PathwaysBuilder({
  baseUrl: "https://webhook.api.flowcore.io",
  tenant: process.env.FLOWCORE_TENANT!,
  dataCore: "commerce",
  apiKey: process.env.FLOWCORE_API_KEY!,
  dataCoreDescription: "Durable commerce events",
  dataCoreAccessControl: "private",
  dataCoreDeleteProtection: true,
})

The required connection fields identify the Flowcore endpoint, tenant, data core, and API key. Optional data-core fields declare provisioning ownership and production safeguards.

For a dedicated Flowcore installation, use the service endpoint supplied by the platform operator.

Register runtime-validated contracts

import { z } from "zod"
 
const orderPlacedSchema = z.object({
  orderId: z.string().uuid(),
  customerId: z.string().uuid(),
  currency: z.string().length(3),
  totalMinor: z.number().int().nonnegative(),
  placedAt: z.string().datetime(),
})
 
const app = pathways.register({
  flowType: "order.0",
  eventType: "order.placed.0",
  schema: orderPlacedSchema,
  writable: true,
  flowTypeDescription: "Order lifecycle events",
  description: "An order was accepted for processing",
})

Registration adds the literal path order.0/order.placed.0 to the builder's inferred type map. Writers and handlers then receive the corresponding payload type.

writable: true allows publishing through write(). A read-only registration can describe an event that this service consumes but must not produce.

Register several events fluently

Registration returns the typed builder, so contracts and handlers can be chained:

const app = pathways
  .register({
    flowType: "order.0",
    eventType: "order.placed.0",
    schema: orderPlacedSchema,
    writable: true,
    flowTypeDescription: "Order lifecycle events",
    description: "An order was placed",
  })
  .handle("order.0/order.placed.0", handleOrderPlaced)
  .register({
    flowType: "order.0",
    eventType: "order.cancelled.0",
    schema: orderCancelledSchema,
    writable: true,
    description: "An order was cancelled",
  })
  .handle("order.0/order.cancelled.0", handleOrderCancelled)

A flow-type description only needs to be declared once for the same flow type. Keep all registrations for one service discoverable rather than scattering hidden builder mutations across unrelated modules.

Publish events

const eventId = await app.write("order.0/order.placed.0", {
  data: {
    orderId,
    customerId,
    currency: "EUR",
    totalMinor: 12900,
    placedAt: new Date().toISOString(),
  },
  metadata: {
    source: "checkout-api",
    correlationId,
  },
})

The schema validates data before ingestion. The returned event ID identifies the stored event.

Metadata can carry correlation, causation, source, and audit context. It should not hide required business fields or contain credentials.

Publish batches

Pathways supports object-style batch writes:

const eventIds = await app.write("order.0/order.placed.0", {
  batch: true,
  data: orders.map(toOrderPlacedPayload),
  metadata: {
    source: "order-import",
    importBatchId,
  },
})

The SDK ingestion batch limit still applies. Chunk larger imports, keep stable source identities, and reconcile accepted and failed chunks.

Batch publication does not make all downstream handlers one transaction.

Handle delivered events

A handler receives the complete Flowcore envelope with a schema-validated payload:

app.handle("order.0/order.placed.0", async (event) => {
  await projectOrder({
    sourceEventId: event.eventId,
    orderId: event.payload.orderId,
    customerId: event.payload.customerId,
    totalMinor: event.payload.totalMinor,
    validTime: event.validTime,
  })
})

Use event.eventId for durable duplicate detection. Use the envelope's flow type, event type, metadata, valid time, and time bucket when the projection or audit record needs them.

A handler should throw when required work fails. Swallowing an error can let the delivery runtime advance even though state was not updated.

Add processing observers

Pathways supports subscriptions around processing and error hooks:

app.subscribe(
  "order.0/order.placed.0",
  (event) => metrics.increment("orders.received", {
    eventType: event.eventType,
  }),
  "before",
)
 
app.onError("order.0/order.placed.0", (error, event) => {
  logger.error("Order projection failed", {
    error,
    eventId: event.eventId,
  })
})
 
app.onAnyError((error, event, pathway) => {
  alertProcessingFailure({ error, eventId: event.eventId, pathway })
})

Use observers for metrics, tracing, and diagnostics. Do not put required state transitions only in a best-effort observer.

Avoid logging complete payloads. Include event identity and path, then retrieve protected details through an authorized diagnostic workflow.

Use session-scoped publishing when actor context matters

SessionPathwayBuilder associates a session with writes and can resolve an acting user or key:

import { SessionPathwayBuilder } from "@flowcore/pathways"
 
const session = new SessionPathwayBuilder(app, requestId)
 
session.withUserResolver(async () => ({
  entityId: authenticatedUser.id,
  entityType: "user",
}))
 
await session.write("order.0/order.placed.0", {
  data: payload,
})

Session context is useful for audit metadata across a request or workflow. It does not replace application authorization; validate the caller before publishing.

Choose explicit resource ownership

Pathways can create or update shared Flowcore resources when provisioning is enabled.

Descriptions signal ownership:

• dataCoreDescription owns the data core description.
• flowTypeDescription owns the flow type description.
• description owns the event type description.

If a description is absent, Pathways expects the resource to exist and fails clearly when it is missing.

Provisioning is additive. It does not delete resources that disappear from code.

Provision without starting a pump

Use provision() in a bootstrap job, CI step, or managed-pathway application that must not start local processing:

await app.provision()

This separates platform setup from runtime handling. It can also support least privilege: a bootstrap identity provisions resources, while the service runtime uses a narrower key.

Control auto-provisioning by stage

await app.startPump({
  stateManagerFactory,
  autoProvision: {
    dataCore: false,
    flowType: false,
    eventType: false,
    pathway: true,
  },
})

The granular configuration controls shared resources and pathway-instance provisioning independently.

The older boolean form remains available, but explicit stage settings make production ownership easier to review. The deprecated defaultAutoProvision option should not be used for new code.

Understand development, managed, and virtual behavior

Pathways uses environment-aware runtime behavior. Production defaults toward managed mode, while development and test default toward virtual mode.

Set runtimeEnv and pathwayMode explicitly in production. A deployment should not switch from a local pump to managed delivery merely because an environment variable changed unexpectedly.

Managed production mode provisions or uses a Flowcore-operated Data Pathway and does not start a local pump. Virtual mode runs the pump in your application and requires durable state and production coordination.

Use optional payload encryption carefully

Current Pathways versions support optional symmetric encryption for registered non-file pathways:

const securePathways = new PathwaysBuilder({
  baseUrl: "https://webhook.api.flowcore.io",
  tenant,
  dataCore: "protected-content",
  apiKey,
  encryption: {
    mode: "symmetric",
    key: process.env.PATHWAYS_ENCRYPTION_KEY!,
  },
})
 
securePathways.register({
  flowType: "secret.0",
  eventType: "secret.created.0",
  schema: secretSchema,
  encrypted: true,
})

The plaintext is validated before encryption. Flowcore receives an object envelope with encryption metadata, and registered processing decrypts before handler validation.

Encryption requires key lifecycle management. Losing the key makes retained payloads unreadable; exposing it defeats the control. It does not replace IAM, data minimization, or secure logging.

Keep builder setup maintainable

For larger applications, separate concerns without hiding the final event map:

• contracts/ exports schemas and event path constants.
• handlers/ contains projection and workflow handlers.
• pathways.ts creates the builder and registers every contract.
• runtime.ts starts managed endpoint routing, a virtual pump, or provisioning.

This structure keeps contract review centralized while allowing handler implementations to remain focused.

Continue with Run managed and virtual Data Pathways to choose and operate the delivery runtime.