Functions and steps

A function is code the engine runs for you when its trigger fires. Its body is made of steps: the engine saves each step's result, retries a failed step, and can pause the function between steps. If these words are new, read Core ideas first.

All the code on this page is in examples/ts/steps.ts and examples/go/steps/main.go.

Define a function

TypeScript: flow.createFunction(options, trigger, handler). List the result in serve({ client: flow, functions: [...] }).

Go: flow.CreateFunction(client, opts, trigger, handler). It adds the function to the client, and client.Serve() serves every function on the client. Make all functions before you call Serve. The type parameter (flow.Input[Order]) is the shape of the event's data; the SDK decodes the JSON into it.

TypeScript

const fulfilOrder = flow.createFunction(
  { id: "fulfil-order" },
  { event: "order/placed" },
  async ({ event, step }) => {
    const orderId: string = event.data.orderId;

    // run: do some work once, and keep its result.
    const charge = await step.run("charge-card", ({ idempotencyKey }) =>
      // The same key on every try of this step: the payment API charges once.
      chargeCard(orderId, event.data.total, idempotencyKey),
    );

    // waitForEvent: pause until order/shipped comes for this order, or 14 days pass.
    const shipped = await step.waitForEvent("wait-for-shipment", {
      event: "order/shipped",
      match: "data.orderId", // the new event's data.orderId must equal this one's
      timeout: "14d",
    });
    if (shipped === null) {
      await step.run("alert-support", () => alertSupport(`order ${orderId} did not ship in 14 days`));
      return { orderId, shipped: false };
    }

    // sendEvent: send events from a function, once.
    await step.sendEvent("ask-for-review", {
      name: "review/requested",
      data: { orderId, carrier: shipped.data.carrier },
    });

    return { orderId, chargeId: charge.id, shipped: true };
  },
);

Go

// Order is the data of order/placed and order/shipped events.
type Order struct {
	OrderID string `json:"orderId"`
	Total   int    `json:"total"`
	Carrier string `json:"carrier"`
}

func addFulfilOrder(client *flow.Client) error {
	_, err := flow.CreateFunction(client,
		flow.FunctionOpts{ID: "fulfil-order"},
		flow.EventTrigger("order/placed", nil),
		func(ctx context.Context, in flow.Input[Order]) (any, error) {
			orderID := in.Event.Data.OrderID

			// Run: do some work once, and keep its result.
			chargeID, err := step.Run(ctx, "charge-card", func(ctx context.Context) (string, error) {
				// The same key on every try of this step: the payment API charges once.
				return chargeCard(ctx, orderID, in.Event.Data.Total, step.IdempotencyKey(ctx))
			})
			if err != nil {
				return nil, err
			}

			// WaitForEvent: pause until order/shipped comes for this order, or 14 days pass.
			shipped, err := step.WaitForEvent[flow.GenericEvent[Order]](ctx, "wait-for-shipment", step.WaitForEventOpts{
				Event:   "order/shipped",
				Match:   flow.StrPtr("data.orderId"), // the new event's data.orderId must equal this one's
				Timeout: 14 * 24 * time.Hour,
			})
			if errors.Is(err, step.ErrEventNotReceived) {
				_, err := step.Run(ctx, "alert-support", func(ctx context.Context) (bool, error) {
					return true, alertSupport(ctx, "order "+orderID+" did not ship in 14 days")
				})
				return map[string]any{"orderId": orderID, "shipped": false}, err
			}
			if err != nil {
				return nil, err
			}

			// Send: send events from a function, once.
			if _, err := step.Send(ctx, "ask-for-review", flow.Event{
				Name: "review/requested",
				Data: map[string]string{"orderId": orderID, "carrier": shipped.Data.Carrier},
			}); err != nil {
				return nil, err
			}

			return map[string]any{"orderId": orderID, "chargeId": chargeID, "shipped": true}, nil
		},
	)
	return err
}

IDs. App and function IDs use letters, digits, - and _, start with a letter or digit, and are at most 64 characters. The engine knows a function by <appId>-<functionId>; here orders-fulfil-order. A new ID is a new function: the old one is archived on the next sync.

Options. The engine checks every option when you sync, and refuses an unknown one with function <id>: unknown option "retry". A change of options applies to new runs; runs that already started keep the options they started with.

The return value is the run's output. It is saved as JSON and shown in the dashboard, and step.invoke gives it to the caller.

Triggers

Trigger	TypeScript	Go
An event	{ event: "order/paid" }	flow.EventTrigger("order/paid", nil)
An event, only some	{ event: "order/paid", if: "event.data.total > 100" }	flow.EventTrigger("order/paid", flow.StrPtr("event.data.total > 100"))
A schedule	{ cron: "0 9 * * MON" }	flow.CronTrigger("0 9 * * MON")
Several	[{ event: "a/one" }, { event: "a/two" }]	flow.MultipleTriggers{flow.EventTrigger("a/one", nil), flow.EventTrigger("a/two", nil)}

if is an expression over event. Schedules are in Crons and scheduled jobs.

What the handler gets

	TypeScript (ctx)	Go (in flow.Input[T])
The event that started the run	event	in.Event
All events (more than one only with batching)	events	in.Events
The run's ID	runId	in.InputCtx.RunID
The try number of the current step, from 0	attempt	in.InputCtx.Attempt
Step tools	step	the step package, with ctx

An event has id, name, data, ts (ms since 1970) and, if the sender set it, user. See Events.

Rules for steps

The function runs from the top on every call from the engine, and steps that are done give back their saved result (replay). So:

1. Put side effects in steps. Code outside a step runs on every call.
2. Step IDs must not change between calls or deploys. The engine finds saved results by ID.
3. Keep the order of steps the same for the same input. Do not choose steps based on the time or a random number, unless you got that value from a step.
4. Step results must be JSON. Return plain data. A Date comes back as a string.
5. Do not swallow the SDK's control flow. In TypeScript, catch only real errors around steps (see Errors and retries). In Go, if you recover() around step calls, call step.Repanic(r) first: the SDK stops the function with a panic when it reaches a new step.
6. Go: no steps in your own goroutines. Use group.Parallel.
7. Up to 1000 steps per run. A run that goes over fails with run exceeded 1000 steps.

The steps

run: do some work, once

Runs your callback, saves what it returns, and gives the saved value back on every later call. If the callback throws (Go: returns an error, or panics), the step fails and the engine retries it. The callback gets the step's idempotencyKey: the same on every try of this step. In Go it is step.IdempotencyKey(ctx).

A step can run more than once (when the app crashes after the work, before the answer). Make it safe to repeat. See Background jobs.

sleep and sleepUntil: pause the run

TypeScript

const followUp = flow.createFunction(
  { id: "follow-up" },
  { event: "order/delivered" },
  async ({ event, step }) => {
    // Pause for a time: "30s", "15m", "2h", "1h30m", "3d", "2w", or ms as a number.
    await step.sleep("wait-3-days", "3d");
    await step.run("ask-how-it-went", () => sendEmail(event.data.email, "How is your order?"));

    // Pause until a moment: a Date, ms since 1970, or an ISO date text.
    await step.sleepUntil("wait-for-new-year", "2027-01-01T09:00:00Z");
    await step.run("send-greetings", () => sendEmail(event.data.email, "Happy new year"));
  },
);

Go

// Delivered is the data of an order/delivered event.
type Delivered struct {
	Email string `json:"email"`
}

func addFollowUp(client *flow.Client) error {
	_, err := flow.CreateFunction(client,
		flow.FunctionOpts{ID: "follow-up"},
		flow.EventTrigger("order/delivered", nil),
		func(ctx context.Context, in flow.Input[Delivered]) (any, error) {
			email := in.Event.Data.Email

			// Pause for a time.
			if err := step.Sleep(ctx, "wait-3-days", 72*time.Hour); err != nil {
				return nil, err
			}
			if _, err := step.Run(ctx, "ask-how-it-went", func(ctx context.Context) (bool, error) {
				return true, sendEmail(ctx, email, "How is your order?")
			}); err != nil {
				return nil, err
			}

			// Pause until a moment.
			newYear := time.Date(2027, time.January, 1, 9, 0, 0, 0, time.UTC)
			if err := step.SleepUntil(ctx, "wait-for-new-year", newYear); err != nil {
				return nil, err
			}
			return step.Run(ctx, "send-greetings", func(ctx context.Context) (bool, error) {
				return true, sendEmail(ctx, email, "Happy new year")
			})
		},
	)
	return err
}

Nothing runs while the run sleeps. A time in the past does not wait: the step ends at once.

A sleep has no upper limit: days, weeks or a year are fine. A sleeping run is never deleted while it sleeps; only ended runs are deleted (see Limits). Two things to know for long sleeps:

• A cancelOn with no timeout listens for at most 366 days. A run that sleeps longer stops hearing its cancel events after that.
• timeouts.finish counts from the run's first call to your app, so time asleep counts toward it.

waitForEvent: wait for something to happen

Pauses until a matching event arrives, or until the timeout. It gives back the event, or null on timeout (Go: step.ErrEventNotReceived). See the wait-for-shipment step above.

• match is a field path. The wait matches when that field is the same in the event that started this run and in the new event.
• if is an expression for anything more. event is the event that started the run, async is the new one: "async.data.orderId == event.data.orderId && async.data.carrier == 'dhl'".
• timeout is required.
• An event that came a little before the wait started still counts: anything received after the step before the wait started (or after the run was made, if no step came before), up to one hour back.
• One event ends every open wait it matches, in all runs.

sendEvent: send events from a function

Sends one event or a list (up to 100), once, even though the function body runs many times. Use it inside a function instead of flow.send. It gives back the new event IDs. See the ask-for-review step above, and the fan-out pattern.

invoke: call another function and use its result

Starts a run of another function, waits for it, and gives back its return value. The other function can be in another app of the same workspace, in either language.

TypeScript

// The child: any function can be invoked. Its return value goes back to the caller.
const buildInvoice = flow.createFunction(
  { id: "build-invoice" },
  { event: "invoice/requested" },
  async ({ event, step }) => {
    return step.run("make-pdf", () => makeInvoicePdf(event.data.orderId));
  },
);

// The parent: starts a run of buildInvoice and waits for its result.
const closeOrder = flow.createFunction(
  { id: "close-order" },
  { event: "order/closed" },
  async ({ event, step }) => {
    const invoice = await step.invoke<{ url: string }>("build-invoice", {
      function: buildInvoice, // or the full ID: "orders-build-invoice"
      data: { orderId: event.data.orderId }, // becomes the child's event.data
      timeout: "1h", // optional: stop waiting after 1 hour
    });
    await step.run("email-invoice", () => sendEmail(event.data.email, `Your invoice: ${invoice.url}`));
    return invoice;
  },
);

Go

// Invoice is what build-invoice returns.
type Invoice struct {
	URL string `json:"url"`
}

// OrderRef is the data of invoice/requested and order/closed events.
type OrderRef struct {
	OrderID string `json:"orderId"`
	Email   string `json:"email"`
}

func addInvoiceFunctions(client *flow.Client) error {
	// The child: any function can be invoked. Its return value goes back to the caller.
	buildInvoice, err := flow.CreateFunction(client,
		flow.FunctionOpts{ID: "build-invoice"},
		flow.EventTrigger("invoice/requested", nil),
		func(ctx context.Context, in flow.Input[OrderRef]) (any, error) {
			return step.Run(ctx, "make-pdf", func(ctx context.Context) (Invoice, error) {
				return makeInvoicePDF(ctx, in.Event.Data.OrderID)
			})
		},
	)
	if err != nil {
		return err
	}

	// The parent: starts a run of build-invoice and waits for its result.
	_, err = flow.CreateFunction(client,
		flow.FunctionOpts{ID: "close-order"},
		flow.EventTrigger("order/closed", nil),
		func(ctx context.Context, in flow.Input[OrderRef]) (any, error) {
			invoice, err := step.Invoke[Invoice](ctx, "build-invoice", step.InvokeOpts{
				// The full ID, "<appId>-<functionId>": here "orders-build-invoice".
				FunctionID: buildInvoice.FullyQualifiedID(),
				// Becomes the child's event.data.
				Data: map[string]string{"orderId": in.Event.Data.OrderID},
				// Optional: stop waiting after 1 hour.
				Timeout: time.Hour,
			})
			if err != nil {
				return nil, err
			}
			if _, err := step.Run(ctx, "email-invoice", func(ctx context.Context) (bool, error) {
				return true, sendEmail(ctx, in.Event.Data.Email, "Your invoice: "+invoice.URL)
			}); err != nil {
				return nil, err
			}
			return invoice, nil
		},
	)
	return err
}

• The child's event is inngest/function.invoked, with your data (up to 1 MB) as event.data.
• If the child fails or is cancelled, the step fails at once with a StepError (Go: *step.StepError). It is not retried: the child already used its own retries.
• With timeout, the step fails with InvokeTimeout when it runs out. The child keeps running; only the wait stops. Without timeout, it waits as long as the child takes.
• A function that does not exist fails the step at once with FunctionNotFound.
• The child uses its own retries, step timeout, cancelOn, concurrency and priority. Throttle, singleton, rate limit, debounce, batching and idempotency do not apply to it.

Parallel steps

TypeScript: start the steps without awaiting each one, then Promise.all. Go: group.Parallel; each function runs up to its first new step.

TypeScript

const loadProfile = flow.createFunction(
  { id: "load-profile" },
  { event: "profile/requested" },
  async ({ event, step }) => {
    // Start both steps, then wait for both. The engine runs them at the same time.
    const [user, orders] = await Promise.all([
      step.run("load-user", () => loadUser(event.data.userId)),
      step.run("load-orders", () => loadOrders(event.data.userId)),
    ]);
    return { name: user.name, orders: orders.length };
  },
);

Go

// UserRef is the data of a profile/requested event.
type UserRef struct {
	UserID string `json:"userId"`
}

func addLoadProfile(client *flow.Client) error {
	_, err := flow.CreateFunction(client,
		flow.FunctionOpts{ID: "load-profile"},
		flow.EventTrigger("profile/requested", nil),
		func(ctx context.Context, in flow.Input[UserRef]) (any, error) {
			userID := in.Event.Data.UserID
			// Each function runs up to its first new step. The engine then runs both
			// steps at the same time.
			results := group.Parallel(ctx,
				func(ctx context.Context) (any, error) {
					return step.Run(ctx, "load-user", func(ctx context.Context) (string, error) {
						return loadUserName(ctx, userID)
					})
				},
				func(ctx context.Context) (any, error) {
					return step.Run(ctx, "load-orders", func(ctx context.Context) ([]string, error) {
						return loadOrders(ctx, userID)
					})
				},
			)
			if err := results.AnyError(); err != nil {
				return nil, err
			}
			name, orders := results[0].Value.(string), results[1].Value.([]string)
			return map[string]any{"name": name, "orders": len(orders)}, nil
		},
	)
	return err
}

The engine runs each new run step in its own call to your app, at the same time. Give every step its own ID. Sleeps, waits, invokes and sends can be in parallel too.

The same step in a loop

TypeScript

const importRows = flow.createFunction(
  { id: "import-rows" },
  { event: "import/started" },
  async ({ event, step }) => {
    const rows: string[] = event.data.rows;
    // The same step ID in a loop is fine: the SDK counts each use.
    for (const row of rows) {
      await step.run("import-row", () => saveRow(row));
    }
    return { imported: rows.length };
  },
);

Go

// Rows is the data of an import/started event.
type Rows struct {
	Rows []string `json:"rows"`
}

func addImportRows(client *flow.Client) error {
	_, err := flow.CreateFunction(client,
		flow.FunctionOpts{ID: "import-rows"},
		flow.EventTrigger("import/started", nil),
		func(ctx context.Context, in flow.Input[Rows]) (any, error) {
			// The same step ID in a loop is fine: the SDK counts each use.
			for _, row := range in.Event.Data.Rows {
				if _, err := step.Run(ctx, "import-row", func(ctx context.Context) (bool, error) {
					return true, saveRow(ctx, row)
				}); err != nil {
					return nil, err
				}
			}
			return map[string]int{"imported": len(in.Event.Data.Rows)}, nil
		},
	)
	return err
}

The SDK counts each use of an ID, so the three import-row steps are three steps. Keep the loop over the same list on every call: get the list from the event, or from a step.

Cancel a run with an event

cancelOn stops a run when a matching event arrives:

TypeScript

const onboarding = flow.createFunction(
  {
    id: "onboarding",
    // Stop this run when user/deleted comes for the same user.
    cancelOn: [{ event: "user/deleted", match: "data.userId", timeout: "30d" }],
  },
  { event: "user/signed-up" },
  async ({ event, step }) => {
    await step.run("send-welcome", () => sendEmail(event.data.email, "Welcome!"));
    await step.sleep("wait-a-week", "7d");
    await step.run("send-tips", () => sendEmail(event.data.email, "Tips for your first week"));
  },
);

Go

// SignedUp is the data of a user/signed-up event.
type SignedUp struct {
	UserID string `json:"userId"`
	Email  string `json:"email"`
}

func addOnboarding(client *flow.Client) error {
	_, err := flow.CreateFunction(client,
		flow.FunctionOpts{
			ID: "onboarding",
			// Stop this run when user/deleted comes for the same user.
			CancelOn: []flow.CancelOn{{
				Event:   "user/deleted",
				Match:   flow.StrPtr("data.userId"),
				Timeout: 30 * 24 * time.Hour,
			}},
		},
		flow.EventTrigger("user/signed-up", nil),
		func(ctx context.Context, in flow.Input[SignedUp]) (any, error) {
			email := in.Event.Data.Email
			if _, err := step.Run(ctx, "send-welcome", func(ctx context.Context) (bool, error) {
				return true, sendEmail(ctx, email, "Welcome!")
			}); err != nil {
				return nil, err
			}
			if err := step.Sleep(ctx, "wait-a-week", 7*24*time.Hour); err != nil {
				return nil, err
			}
			return step.Run(ctx, "send-tips", func(ctx context.Context) (bool, error) {
				return true, sendEmail(ctx, email, "Tips for your first week")
			})
		},
	)
	return err
}

• match and if work as in waitForEvent. timeout is optional: after it, the run no longer listens. Without it, the run listens as long as it lives.
• Only events that arrive after the run was made count.
• Up to 5 cancelOn entries per function.
• A cancelled run ends as Cancelled. On a local engine the run above ended with cancelled by event user/deleted (evt_...). If a step is running in your app at that moment, it finishes, but its result is dropped.
• Failure handlers do not run for cancelled runs.

You can also cancel runs in the dashboard: Cancel on a run's page, or Bulk actions on the function's page.

Expressions

if, key, idempotency and priority.run take a small expression, a part of CEL. The variables are event (the event that started the run) and, in waitForEvent and cancelOn, async (the new event).

event.data.plan == 'pro'
event.data.total > 100 && event.data.currency == "EUR"
event.data.country in ['NG', 'GH', 'KE']
has(event.data.coupon) && event.data.coupon.startsWith('VIP')
async.data.orderId == event.data.orderId

You can use
Values	numbers, 'text' or "text", true, false, null, lists [1, 2]
Fields	event.data.x, event.data["x-y"], event.data.items[0]
Compare	== != < <= > >=
Logic	&& || ! and cond ? a : b
Maths and text	+ - * / %, x in list
Functions	size(x), has(event.data.x), .startsWith(), .endsWith(), .contains(), .lowerAscii(), .upperAscii(), .size()

• A missing field counts as null in == and !=. Anywhere else a missing field is an error. Use has(...) first when a field may be absent.
• An if that fails when it runs counts as false. A key that fails is "null", so all those events share one key.
• A bad expression is refused when you sync, with a message that names the function: function f: bad trigger if: unexpected character '='.
• Up to 2048 characters. No regular expressions.

All function options

Option (TypeScript)	Go (flow.FunctionOpts)	What it does
id	ID	Required. The function's ID.
name	Name	A name for the dashboard.
retries	Retries	Retries per step, 0 to 20. Default 4. Errors and retries.
timeouts.step	Timeouts.Step	Longest one call to your app may take. Default 5 minutes, 1 second to 14 minutes.
timeouts.finish	Timeouts.Finish	Longest the whole run may take; then it is cancelled. No limit by default.
cancelOn	CancelOn	Cancel on an event.
onFailure	(a second function)	Runs when a run fails for good.
concurrency	Concurrency	Limit steps at once.
throttle	Throttle	Slow down starts.
rateLimit	RateLimit	Drop extra events.
debounce	Debounce	Wait for quiet.
batchEvents	BatchEvents	Many events, one run.
priority	Priority	Some runs first.
idempotency	Idempotency	One run per key.
singleton	Singleton	One run at a time per key.
tags	Tags	Names to find runs by.

Tags

Tags let you find a function's runs by a value from their event, for example a job or order ID. Then you can list, count and cancel those runs from your app.

const runJob = flow.createFunction(
  {
    id: "run-job",
    // One active run per job key.
    singleton: { key: "event.data.key", mode: "skip" },
    // Tags let you find this job's runs later. Worked out once, when a run starts.
    tags: { jobId: "event.data.jobId" },
  },
  { event: "jobs/run" },
  async ({ event, step }) => {
    await step.run("work", () => doWork(event.data.jobId));
  },
);

• Up to 3 tags. A name is a letter, then letters, digits and _, up to 32. The SDK throws at createFunction for more tags or a bad name.
• A value is an expression over event. It is worked out once, when the run starts, over its first event. Text stays as it is; a number or a boolean becomes text. Anything else (no value, null, an object, a list, an error) leaves the tag out. So does a value over 256 characters.
• Find runs with a tag: flow.runs.list({ function, tags }). Cancel them: flow.runs.cancel({ function, tag }). See the TypeScript reference and the HTTP API.