Build your first event pipeline

In this tutorial, you will build a small event pipeline for a to-do application. The application will:

1. Define a runtime-validated todo.created.0 contract.
2. Provision the Flowcore resources used by the contract.
3. Start a local development pump.
4. Publish an event.
5. Process the event with a handler.

The example uses @flowcore/pathways because it keeps the event contract, publisher, and handler type-safe in one place.

Prerequisites

Complete Set up Flowcore v2 first.

You need these environment variables:

FLOWCORE_TENANT=your-development-tenant
FLOWCORE_API_KEY=fc_replace_with_your_key

The API key must be allowed to use the tenant and create or access the tutorial's data core, flow type, and event type.

1. Create a small TypeScript project

Create a directory and install the dependencies:

mkdir flowcore-todo
cd flowcore-todo
npm init -y
npm install @flowcore/pathways zod
npm install --save-dev typescript tsx @types/node

Add a script to package.json:

{
  "scripts": {
    "dev": "tsx src/index.ts"
  }
}

Create src/index.ts.

2. Define the event payload

Start with the data that describes one completed business fact:

import { PathwaysBuilder } from "@flowcore/pathways"
import { z } from "zod"
 
const todoCreatedSchema = z.object({
  id: z.string().uuid(),
  title: z.string().min(1),
  createdAt: z.string().datetime(),
})

The Zod schema validates data at runtime and lets Pathways infer the TypeScript type used by writers and handlers.

The event is named todo.created.0 because it records a fact in the past tense. The .0 suffix gives the contract an explicit version. If its meaning changes incompatibly later, introduce a new event type rather than reinterpreting existing events.

3. Create the Pathways builder

Add the builder below the schema:

const tenant = process.env.FLOWCORE_TENANT
const apiKey = process.env.FLOWCORE_API_KEY
 
if (!tenant || !apiKey) {
  throw new Error("FLOWCORE_TENANT and FLOWCORE_API_KEY are required")
}
 
const pathways = new PathwaysBuilder({
  baseUrl: "https://webhook.api.flowcore.io",
  tenant,
  dataCore: "todo-tutorial",
  apiKey,
  runtimeEnv: "development",
  pathwayMode: "virtual",
  dataCoreDescription: "Events for the Flowcore getting-started tutorial",
  dataCoreAccessControl: "private",
  dataCoreDeleteProtection: false,
  autoProvision: {
    dataCore: true,
    flowType: true,
    eventType: true,
    pathway: false,
  },
})

Important choices in this configuration:

• runtimeEnv: "development" makes the runtime intent explicit.
• pathwayMode: "virtual" allows the application to run the local pump.
• The data core is private.
• A description declares that this code owns provisioning for the data core.
• Shared resources can be provisioned, but no production pathway instance is created.

For production, delete protection should normally be enabled. It is disabled here so a development administrator can remove the tutorial resources later.

4. Register the event contract

Register the flow type and event type:

const app = pathways.register({
  flowType: "todo.0",
  eventType: "todo.created.0",
  schema: todoCreatedSchema,
  writable: true,
  flowTypeDescription: "Lifecycle events for tutorial to-do items",
  description: "A to-do item was created",
})

Descriptions are significant during Pathways provisioning:

• dataCoreDescription allows the data core to be created or updated.
• flowTypeDescription allows the flow type to be created or updated.
• description allows the event type to be created or updated.

If a description is omitted, Pathways expects that resource to already exist. Provisioning is additive: it creates missing declared resources and updates owned descriptions, but does not delete resources that disappear from code.

5. Add a handler

For the tutorial, keep an in-memory read model:

const todos = new Map<string, z.infer<typeof todoCreatedSchema>>()
 
app.handle("todo.0/todo.created.0", async (event) => {
  if (todos.has(event.eventId)) {
    return
  }
 
  todos.set(event.eventId, event.payload)
  console.log("Projected todo:", event.payload)
})

The map is only suitable for demonstrating the handler. A production projection should store the event ID and its state change durably, preferably in one database transaction.

Why use the event ID? Event delivery is normally at least once. A consumer may receive the same event again after a timeout, restart, or retry. Recording processed IDs makes the handler idempotent.

6. Create development checkpoint storage

A data pump needs a durable position for every flow type. For this tutorial, use an in-memory state manager:

const positions = new Map<
  string,
  { timeBucket: string; eventId?: string }
>()
 
const stateManagerFactory = (flowType: string) => ({
  getState: async () => positions.get(flowType) ?? null,
  setState: async (state: { timeBucket: string; eventId?: string }) => {
    positions.set(flowType, state)
  },
})

This state disappears when the process stops, so it is not safe for production. Production virtual pathways should use persistent checkpoint storage such as PostgreSQL.

The checkpoint and the handler's deduplication record solve related but different problems:

• The checkpoint records how far the pump has progressed.
• The processed event ID records whether the projection applied one event.

A robust system needs both.

7. Provision resources and start the pump

Start the development pump:

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

This provisions the shared resources declared by the builder and registrations, then starts local event processing.

If provisioning fails, check:

• The tenant and data core names
• Whether the API key can inspect and create the resources
• Whether a resource exists but is outside the key's permissions
• Whether your dedicated deployment uses a different service base URL

Do not respond to a permission failure by switching to a broad administrator key. Fix the policy for the application's intended resources.

8. Publish an event

Publish a valid to-do event:

const eventId = await app.write("todo.0/todo.created.0", {
  data: {
    id: crypto.randomUUID(),
    title: "Learn Flowcore",
    createdAt: new Date().toISOString(),
  },
  metadata: {
    source: "getting-started-tutorial",
    correlationId: crypto.randomUUID(),
  },
})
 
console.log("Published event:", eventId)

write() validates the payload before sending it. A missing title, invalid UUID, or invalid timestamp is rejected locally by the schema.

A successful write confirms that Flowcore accepted the event. The handler runs asynchronously when the pump receives it. Publishing does not synchronously guarantee that every consumer has finished.

9. Keep the process running and shut down cleanly

Add graceful shutdown handling:

const shutdown = async () => {
  await app.stopPump()
  process.exit(0)
}
 
process.on("SIGINT", shutdown)
process.on("SIGTERM", shutdown)
 
console.log("Waiting for events. Press Ctrl+C to stop.")

Run the application:

npm run dev

You should see a published event ID followed by the projected payload when the pump handles the event.

10. Test validation

Temporarily replace the title with an empty string:

title: ""

Run the application again. The write should fail schema validation instead of storing an invalid event.

Restore the valid title after confirming the behavior.

What you built

Your application now has:

• A private Flowcore data core named todo-tutorial
• A versioned todo.0 flow type
• A versioned todo.created.0 event type
• Runtime payload validation
• An event publisher
• A local development pump
• An idempotency-aware projection handler
• Development-only checkpoint storage

Common problems

The API key is rejected

Confirm that the complete key is available to the process and has not been quoted incorrectly. Verify that it belongs to the intended tenant and has the required resource and ingestion permissions.

Provisioning says a resource is missing

Pathways only creates a resource when the corresponding description is present and auto-provisioning is enabled for that stage. Add the intended description or provision the resource separately.

The event publishes but the handler does not run

A registered handler does not consume history by itself. Confirm that startPump() completed, the process is still running, and the checkpoint and notifier can reach the Flowcore services.

The handler runs more than once

That is expected under at-least-once delivery. Replace the in-memory map with durable duplicate detection before production.

The tutorial reprocesses history after restart

The in-memory checkpoint was lost. Use a persistent state manager when restart continuity matters.

Clean up carefully

Event and resource deletion is an administrative lifecycle operation. Before removing the tutorial data core, confirm that no other test uses it and that your environment permits deletion.

Never automate production deletion merely because a declaration was removed from code. Pathways provisioning is intentionally additive.

Next step

Continue with Prepare a Flowcore service for production before deploying this pattern to a shared environment.