Triggers

A trigger waits for a signal from anywhere — a poll of a database, a message on a broker, a call to a web API — and hands what it found to Fluxify, which runs the workflow the trigger is attached to. Write one in src/triggers as a file that default-exports trigger({...}):

import { param } from "@fluxify/sdk/blocks";
import { poll, trigger } from "@fluxify/sdk/triggers";

export default trigger({
	name: "acme/shop/orders/new",
	label: "New orders",
	icon: "./icons/orders.svg",
	permissions: { integrations: ["database"] },
	params: {
		database: param.integration({ kind: "database", variants: ["PostgreSQL"], required: true }),
		table: param.string({ label: "Table", required: true, default: "orders" }),
	},
	async start(ctx) {
		const db = await ctx.integrations.connection(ctx.params.database);
		return poll(ctx, {
			every: 5_000,
			async tick() {
				const rows = await readNewOrders(db, ctx.params.table);
				if (rows.length) await ctx.emit(rows);
			},
		});
	},
});

trigger() checks the definition when the file loads and throws an error that names the trigger, so a mistake shows up in bun test and in the build.

No webhooks

A trigger cannot receive an HTTP request. Routes already own the project's URLs. A route that needs to start a workflow uses the Send Message or Trigger Workflow block.

Fields

Field	Required	Default	What it is
`name`	Yes		The trigger's kind: `scope/name/folder/trigger`
`label`	Yes		The name shown in the trigger wizard
`description`	No		One line under the label
`icon`	Yes		An image inside the plugin
`permissions`	No	None	What `ctx` may reach
`params`	No	None	The settings the user fills in
`tabs`	No	By first use	The order of the wizard's steps
`retry`	No	`false`	Let Fluxify retry a failed run
`start`	Yes		Opens the listener or poll loop

name

The trigger's kind: the plugin name plus the path of the file under src/triggers. It must be lowercase and match the path exactly, as a block's type must. The build stops if they differ.

A project's trigger stores its kind as plugin:<name>, in a column that holds 255 characters, so a name may be at most 248 characters. The build refuses a longer one.

name: "acme/shop/orders/new", // src/triggers/orders/new.ts in acme/shop

Renaming or moving a trigger breaks every trigger the project created from it; add a new kind and remove the old one later instead.

label

What the user sees on the wizard's Source step and in the trigger list, next to a Plugin badge.

label: "New orders",

description

A short line under the label on the Source step.

description: "Starts the workflow for each order a poll finds.",

icon

A path to an image inside the plugin: .svg, .png, .jpg, .jpeg, .gif or .webp. The build copies an icon from outside icons/ into icons/ so it ships, and stops if the image is missing or too large (about 50 KB). The icon is shown on the wizard's Source step and on the trigger list.

icon: "./icons/orders.svg",

permissions

What ctx may reach, beyond what every trigger gets. A creator sees it when approving the plugin, and the approval covers exactly what you ask for.

Declare	You get	Use
`integrations: ["database"]`	`ctx.integrations`	The project's integrations of that kind, by id (below)
`appconfig: true`	`ctx.getConfig`	Any App config key. Without it, `getConfig` reads only the keys your `param.secret()` fields name

A param.integration needs its kind in permissions.integrations, and a param.appconfig needs permissions.appconfig: true. The build refuses a trigger with the param but no permission.

permissions: { integrations: ["database"], appconfig: true },

params

The settings the user fills in, typed from the declaration so params in start has the right types. A trigger takes the same param.* builders as a block, with two differences:

param.secret() is allowed. The user picks an App config key, and only the key is stored, as cfg:<KEY>.
js, schema, conditions and fieldMap are refused. They describe something evaluated per request, and a trigger has no request.

params: {
	database: param.integration({ kind: "database", required: true }),
	table: param.string({ label: "Table", required: true }),
	everySeconds: param.number({
		label: "Poll every (seconds)",
		default: 30,
		min: 1,
		max: 86_400,
		integer: true,
	}),
},

An integration param can name the variants it accepts with variants (below). all, any and names starting with __ are reserved. Params lists every type and option.

tabs

The wizard's steps, in order. A trigger's tabs sit between Basics and Workflow, and params with no tab land in a Settings step before them. Without tabs, the steps appear in order of first use, with Settings first. A param's tab that tabs leaves out is added after the listed ones, and the build warns about it.

tabs: ["Connection", "Polling"],

retry

retry: true opts into Fluxify's retry system. When a run fails, Fluxify runs it again as many times as the trigger's Max attempts setting says (at most 5), waiting the trigger's Retry delay before each one, which doubles after each failure. emit rejects only after the last attempt fails. The wizard shows Max attempts and Retry delay only for a kind that declares retry: true.

With retry off — the default — one emit call is one run, and retrying is your code's job.

retry: true,

There is never an automatic commit, acknowledgement or dead-letter. Retrying a run is not the same as settling the source event; that stays with you (below).

start

async start(ctx) {
	// open the listener, or start the poll loop in the background
	return async () => {
		// close what this call opened; must return
	};
},

start is called once per running trigger instance. It opens the connection or poll loop and resolves once it is running — keep the loop going in the background and do not wait for the first event. A start that does not resolve within 30 seconds is treated as a failure.

It returns that instance's stop function. Fluxify aborts ctx.signal first, then calls the returned function, and expects it to return within 10 seconds. One definition can run as several triggers at once, each with its own ctx, so the cleanup belongs in the closure — not in a top-level field.

async start(ctx) {
	const client = new MyClient(ctx.getConfig(ctx.params.token));
	return poll(ctx, {
		every: 5_000,
		async tick() {
			const items = await client.read();
			if (items.length) await ctx.emit(items);
		},
	});
},

Nothing to close is async () => {}.

There is no top-level stop field. trigger() refuses one, because one definition can run as several instances and each one's cleanup belongs to its own start call.

Polling and waiting

@fluxify/sdk/triggers ships the two helpers almost every trigger needs, so the loop does not have to be written by hand. Both are browser- and Node-safe and add no dependency.

`poll(ctx, { every, tick, onError? })`

poll starts tick at once, waits every milliseconds after it returns, and repeats until ctx.signal aborts. It returns the stop function start resolves to.

import { poll, trigger } from "@fluxify/sdk/triggers";

async start(ctx) {
	return poll(ctx, {
		every: ctx.params.pollIntervalMs,
		async tick() {
			const items = await readBatch(ctx.params.queue);
			if (items.length) await ctx.emit(items.map(decode));
			// a full batch means more may be waiting: do not sleep before the next read
			return items.length === ctx.params.batchSize ? "again" : undefined;
		},
	});
},

It does	It does not
Run `tick` until `ctx.signal` aborts	Back off on its own — the restart-with-backoff and turn-off after 5 failures are the runtime's
Wait `every` between ticks, unless `tick` returned `"again"`	Commit, acknowledge or dead-letter anything — delivery stays yours
Wait for the tick in progress when it stops, so the runtime's 10-second stop limit still applies	Report an error raised after the trigger stopped
Log a failed tick as a warning and keep going by default	Swallow an error you call fatal through `onError`

Errors. When tick throws, poll calls onError(error). Without one it logs a warning and waits every, which suits a transient connection error. Return "fail" from onError — or throw from it — to report the error through ctx.fail and end the loop, for something that will never fix itself. Returning "fail" reports the tick's error; throwing reports the error you threw.

onError(error) {
	// a wrong key type never fixes itself: stop and let Fluxify report it
	if (String(error).includes("WRONGTYPE")) return "fail";
	ctx.logger.logWarn(error);
},

A failed emit is a failed run, not a failed loop: catch it inside tick if the source should try the same items again, or let tick throw and have onError decide.

Stopping. Fluxify aborts ctx.signal first, then calls the stop function poll returned. That function waits for the current tick to finish; a tick that never returns is a stop that times out.

`sleep(ms, signal)`

sleep resolves after ms, or at once when signal aborts. It resolves rather than rejects on abort, so a loop only has to re-check signal.aborted, and it always removes its listener — a loop that waits a thousand times leaves none behind.

import { sleep } from "@fluxify/sdk/triggers";

while (!ctx.signal.aborted) {
	await work();
	await sleep(1_000, ctx.signal);
}

Check ctx.signal.aborted before waiting, not just inside the listener: stop can arrive while the work is still running, and a listener added to an already-aborted signal never fires. poll already does both, which is why most triggers need only the one call.

The context

Member	What it is
`ctx.params`	The settings, defaulted and checked, as a block's are
`ctx.integrations`	`get(id)` and `connection(id)` for the integrations the trigger was granted (below)
`ctx.getConfig(key)`	An App config value (below)
`ctx.logger`	`logInfo`, `logWarn` and `logError`, tagged with the trigger's id and kind
`ctx.signal`	An `AbortSignal`, aborted when Fluxify stops the trigger, before your stop function runs
`ctx.emit(events, options?)`	Runs the attached workflow with the events
`ctx.fail(error)`	Reports a fatal error from your own loop

Integrations

ctx.integrations.get(id) returns the integration a param.integration param picked, and ctx.integrations.connection(id) returns the same client plus its variant and the variant's own client. Both work for the built-in groups and for an integration your plugin provides. The kind is checked against the trigger's permissions, and the errors are the ones a block sees — no integration was selected, was not granted … integrations, an unknown id, a disabled plugin.

A trigger is long-lived, not a request: each id is opened once per instance, and the trigger holds that client until it stops. A poll loop calling get every round reuses one connection instead of opening one per round. Stopping the trigger releases the clients.

permissions: { integrations: ["kv"] },
params: { store: param.integration({ kind: "kv", required: true }) },
async start(ctx) {
	const kv = await ctx.integrations.get<"kv">(ctx.params.store);
	// one client for the life of this trigger instance
},

Use connection when the trigger needs the variant's own client, such as a Redis XREADGROUP or a database's raw query interface. Variant-specific integrations covers variants and narrowing.

getConfig

ctx.getConfig(key) reads an App config value. With permissions.appconfig it reads any key; without it, only the keys the trigger's own param.secret() fields name. It takes the key or a secret param's stored value, so this works:

params: { dsn: param.secret({ label: "Connection URL", required: true }) },
async start(ctx) {
	const url = ctx.getConfig(ctx.params.dsn); // ctx.params.dsn is "cfg:DATABASE_URL"
},

An error names the missing key, never a value. A secret param whose key has no value in this environment keeps the trigger from starting, the way a missing value stops a queue trigger.

emit

ctx.emit(events, options?) runs the workflow attached to this trigger, in the same process, on the worker running the trigger. It takes an array, one item per event; each item becomes one entry in the workflow's trigger.data, with its data as you passed it.

await ctx.emit([
	{ orderId: "A-1024", total: 42 },
	{ orderId: "A-1025", total: 17 },
]);

When exactly one event arrives, input is that event's payload. All of them are in trigger.data:

trigger.data[0].data.orderId // "A-1024"
trigger.meta.size            // 2
trigger.meta.attempt         // 1, then 2 on a retry

emit:

resolves when the run succeeds, and rejects when it fails;
with retry: true, rejects only after the Max attempts setting is used up, with the last error;
runs one workflow run per call, with every event of that call in it;
allows concurrent calls — two emits in flight are two runs;
rejects without running when the trigger is stopping, or when the workflow is not loaded on this worker.

An empty array is a run with no events; call it only when you mean that.

The connection

emit takes an optional connection, which the workflow sees as trigger.connection:

await ctx.emit(rows, {
	connection: {
		raw: transaction,
		commit: async () => {
			await markDone(rows);
		},
		moveToDLQ: async (error) => {
			await park(rows, error);
		},
		lag: async () => countWaiting(),
	},
});

Member	What it is
`commit()`	Marks this batch processed. Called by a workflow that commits manually
`moveToDLQ(error?)`	Parks this batch somewhere else
`lag()`	Messages waiting behind this batch, or `null` when unknown
`raw`	Whatever you put here: the source's own client, or a transaction handle

All three functions are required; raw is not. This is the same API the built-in queue triggers expose in manual commit mode, and it works because the run is in the same process: await trigger.connection.commit() in a JS Runner block calls your function.

Nothing calls commit for you. A workflow only commits if it is written to.

fail

ctx.fail(error) reports a fatal error from your own background loop — one Fluxify cannot see otherwise. The instance stops (ctx.signal aborts, then your stop function runs) and starts again after a backoff. After 5 failures in a row the trigger is turned off, with the error's message as the reason on the trigger row.

const loop = (async () => {
	while (!ctx.signal.aborted) {
		// ...
	}
})().catch((error) => ctx.fail(error));

With poll you do not call it yourself: return "fail" from onError to report the tick's error, or throw from onError to report the error you threw, and poll calls ctx.fail and ends the loop.

Call ctx.fail from the loop's catch

An error nobody catches does not crash the worker — the execution process logs it and stays up — but it is only blamed on your trigger when Fluxify can prove which trigger it came from. With two instances of one kind running, it cannot, so the error is only logged and your loop is not restarted. Calling ctx.fail from the loop's catch is the reliable path.

A run failure is not a ctx.fail: a failed emit means the workflow run failed, and retrying it is what retry and your code are for. Reserve ctx.fail for the loop itself — a lost connection, a broken decoder, a missing table.

Designing an idempotent, at-least-once trigger

Delivery is at least once. A worker can be killed after the workflow ran but before the source learned about it, and the next poll or delivery runs the same event again. Design for that from the start.

Commit only after `emit` resolves

emit resolves when the run succeeded. Commit, acknowledge or mark the source event done after that, never before:

await ctx.emit(rows);       // the workflow succeeded
await markDone(rows);       // now the source may forget them

If the worker dies between the two, the rows are read again and the workflow runs a second time. That is the cost of at-least-once; losing the event entirely is the alternative. A commit before emit turns a crash into a lost event.

If your source has its own acknowledgement (an XACK, a broker ack), send it in the same place: after emit resolves. With retry on, that is after the last attempt.

Every worker in the group runs start

A trigger runs in the execution process of every worker in its group, and each instance calls start. That is fine for a source you can claim work from, and wrong for one you cannot:

Source	What to do
Queue or broker	Read as a consumer group, so each message goes to one consumer. Kafka consumer groups, NATS durable consumers, RabbitMQ competing consumers and Redis consumer groups all do this
Database	Claim the rows in a transaction: `SELECT … FOR UPDATE SKIP LOCKED`, then `emit`, then mark them done, then commit. A second worker skips the locked rows instead of reading them again
Web API with no way to claim an item	Give the trigger a group that runs one replica. Nothing else stops two workers from both reading the same item

A database poller must lock even for a read. A plain SELECT does not, so on a group with several workers every worker fans the same rows out to its own run.

The claim and the commit belong to the same transaction, and emit happens between them. A worker killed mid-run rolls the transaction back, so the row is free for another worker — and runs again. That is the pattern the integrations example uses.

Groups do not coordinate instances

A group decides where the trigger runs, not which instance does the work. Every worker in the group starts its own copy. Use the broker's consumer groups, row locking, or a single-replica group.

Deduplicate by the source's own id

At-least-once means the same event can arrive twice. Make the workflow safe to repeat:

Put the source's stable id in each event, so the workflow can recognise it.
Store the ids you handled — a processed table, a unique constraint, an idempotency key on the downstream API — and skip or upsert on a repeat.
Do not use a timestamp or a per-delivery number as the key. It changes on a redelivery.

Fluxify does not deduplicate for you: only you know what "already handled" means for your source.

Honour `ctx.signal`

ctx.signal aborts when Fluxify stops the trigger: a withdrawn plugin, a disabled plugin, a group move, a version upgrade, a shutdown. Your loop should check it every round, and every wait should wake when it aborts:

import { sleep } from "@fluxify/sdk/triggers";

while (!ctx.signal.aborted) {
	await work();
	await sleep(5_000, ctx.signal);
}

sleep resolves at once on abort and always removes its listener, so a long-running loop does not leave listeners behind. A hand-written timer must clear it and remove the listener itself; the polling helpers already do this, and poll checks ctx.signal.aborted before every tick and every wait.

Check ctx.signal.aborted before waiting, not just inside the listener: stop can arrive while an emit is still running, and a listener added to an already-aborted signal never fires.

A stop function that does not return within 10 seconds is logged and dropped during a shutdown or a withdrawal. During a restart it counts as a failure and the instance is restarted.

Turn-off, and turning it back on

When the trigger is turned off, the portal shows the reason on the trigger row and on its edit page, as text you wrote or the runtime's summary — for example:

Reason	Cause
`start failed: <message> (failed 5 times in a row)`	`start` threw five times
`start failed: it did not resolve within 30s`	`start` never resolved
`start failed: start() must resolve to its stop function`	`start` returned something else
`ctx.fail: <message> …`	Your loop reported it
`unhandled error: <message> …`	An error traced to this instance
`stop timed out after 10s …`	The stop function did not return

Turning it back on is the normal enable switch: Fluxify re-checks the plugin and the params, starts a fresh instance and resets the failure count. A start error is often a wrong table name or a missing App config value, so fix the cause first — otherwise the count runs out again.

An instance that stayed up for 60 seconds starts the count again, so a trigger that works for a while and then fails once is not five failures into being turned off.

Limits

Limit	Value
`start` must resolve within	30 seconds
The returned stop function must return within	10 seconds
Restart wait after a failure	1 second, doubling each time
Failures in a row before the trigger is turned off	5
Running time that resets the count	60 seconds
Trigger name	248 characters (so `plugin:<name>` fits 255)
Icon size	about 50 KB
Max attempts, with `retry`	1 to 5, default 3
Retry delay, with `retry`	0 to 300000 ms, default 1000, doubling

Scaffolding

bunx fluxify-cli triggers create orders/new

This writes src/triggers/orders/new.ts, a test and a placeholder icon. The template uses poll: it honours ctx.signal, waits between ticks, emits a batch and logs a failed run. The trigger's kind is the plugin name plus the path: acme/shop/orders/new.

Flag	Meaning
`--force`	Not used by this command

bun run build writes the triggers list into fluxify.json, checks the name, params, icon and entry path, and warns about a tab the params do not use.

A complete trigger

The integrations example has a database poller that claims rows with SKIP LOCKED on PostgreSQL or MySQL, emits them, and marks them done only after the run succeeded:

// source: examples/plugins/integrations/src/triggers/orders.ts
// ...
params: {
	database: param.integration({
		// ...
		kind: "database",
		variants: ["PostgreSQL", "MySQL"],
		required: true,
	}),
	// ...
	everySeconds: param.number({
		label: "Poll every (seconds)",
		default: 10,
		description: "From 1 to 3600.",
		min: 1,
		max: 3600,
		integer: true,
	}),
	// ...
	batchSize: param.number({
		label: "Rows per poll",
		// ...
		default: 50,
		min: 1,
		max: 1000,
		integer: true,
	}),
},
async start(ctx) {
	const connection = await ctx.integrations.connection(ctx.params.database);
	const table = ctx.params.table;
	// ...
	const limit = ctx.params.batchSize;
	const emit: Emit = (rows) => ctx.emit(rows);
	const claim =
		connection.variant === "PostgreSQL"
			? () => pollPostgres(connection.raw, table, limit, emit)
			: () => pollMysql(connection.raw, table, limit, emit);
	return poll(ctx, {
		every: ctx.params.everySeconds * 1_000,
		async tick() {
			const handled = await claim();
			// a full batch means there may be more waiting: read again at once
			return handled === limit ? "again" : undefined;
		},
		onError(error) {
			// the claim rolled back, so the rows stay for the next poll
			ctx.logger.logError(error);
		},
	});
},

File	What it shows
`src/triggers/orders.ts`	`variants`, narrowing on `variant`, `FOR UPDATE SKIP LOCKED`, `emit` then commit, `poll` with `"again"` for a full batch, and `onError` logging
`src/triggers/orders.test.ts`	`runTrigger` with a fake connection for each variant, and a real PostgreSQL container

Params: the settings a user fills in
Integrations: narrowing an integration param to the variants you support
Testing: runTrigger and the connection fakes
Upgrading: which changes to a trigger block an upgrade

Triggers ​

Fields ​

name ​

label ​

description ​

icon ​

permissions ​

params ​

tabs ​

retry ​

start ​

Polling and waiting ​

poll(ctx, { every, tick, onError? }) ​

sleep(ms, signal) ​

The context ​

Integrations ​

getConfig ​

emit ​

The connection ​

fail ​

Designing an idempotent, at-least-once trigger ​

Commit only after emit resolves ​

Every worker in the group runs start ​

Deduplicate by the source's own id ​

Honour ctx.signal ​

Turn-off, and turning it back on ​

Limits ​

Scaffolding ​

A complete trigger ​

Next ​