Broadcasts

A broadcast chains multiple signals into a directed acyclic graph (DAG). Each node in the graph is a signal. Nodes execute when all their upstream dependencies complete. Outputs from upstream nodes are passed to downstream nodes automatically. The broadcast runner orchestrates the entire graph, handling fan-out, fan-in, conditional execution, and failure propagation.

broadcast(name)

Creates a named broadcast definition. The name must be unique, start with a letter, and contain only letters, digits, hyphens, and underscores. Returns a builder that connects signals into a dependency graph.

import { broadcast } from "station-broadcast";import { signal, z } from "station-signal"; const validate = signal("validate")  .input(z.object({ orderId: z.string() }))  .output(z.object({ orderId: z.string(), total: z.number() }))  .run(async (input) => {    const order = await db.orders.findById(input.orderId);    return { orderId: order.id, total: order.total };  }); const charge = signal("charge")  .input(z.object({ orderId: z.string(), total: z.number() }))  .output(z.object({ chargeId: z.string() }))  .run(async (input) => {    const result = await paymentGateway.charge(input.total);    return { chargeId: result.id };  }); const notify = signal("notify")  .input(z.object({ chargeId: z.string() }))  .run(async (input) => {    await emailService.sendReceipt(input.chargeId);  }); export const orderFlow = broadcast("orderFlow")  .input(validate)  .then(charge)  .then(notify)  .build();

Builder methods

.input(signal)

Sets the entry signal -- the root node of the DAG. This is required and must be called first. The entry signal's input type becomes the broadcast's input type: when you call broadcast.trigger(data), that data is passed to this signal.

const pipeline = broadcast("pipeline")  .input(validate)  // validate's input schema defines what .trigger() accepts  // ...

.then(signal, opts?)

Adds one or more downstream nodes. By default, each node depends on the most recently added tier (the "last tier"). Multiple signals passed to a single .then() call create parallel fan-out -- they all depend on the same upstream tier and run concurrently.

// Single signal with options:.then(charge, {  as: "payment",  after: ["validate"],  map: (upstream) => ({ total: upstream.validate.total }),  when: (upstream) => upstream.validate.total > 0,}) // Fan-out (multiple signals, no options):.then(emailReceipt, smsNotification, slackAlert)

.every(interval)

Makes the broadcast recurring. Accepts the same duration strings as signals: "30s", "5m", "1h", "1d". The broadcast runner enqueues a new run after each interval elapses. If a pending or running instance already exists, that tick is skipped to prevent overlap.

const hourlySync = broadcast("hourlySync")  .input(fetchData)  .then(transformData)  .then(loadData)  .every("1h")  .withInput({ source: "production" })  .build();

.withInput(data)

Default input data for recurring broadcasts. Without this, recurring broadcasts run with {} as input. The data is passed to the entry signal.

.onFailure(policy)

Sets the failure policy that determines what happens when a node fails. Default: "fail-fast".

const resilientPipeline = broadcast("resilientPipeline")  .input(fetchData)  .then(sendEmail)       // non-critical: ok if this fails  .then(updateDatabase)  // critical  .onFailure("continue")  .build();

.timeout(ms)

Maximum total time for the entire broadcast execution in milliseconds. If the broadcast runs longer than this, all active nodes are cancelled and the broadcast is marked as failed with a timeout error. This is separate from per-signal timeouts, which still apply individually.

const timeBoundPipeline = broadcast("timeBoundPipeline")  .input(validate)  .then(processA)  .then(processB)  .timeout(60_000) // entire broadcast must finish within 60 seconds  .build();

.build()

Finalizes the broadcast definition. Validates the DAG: checks for duplicate node names, missing dependencies, and cycles. Throws BroadcastValidationError or BroadcastCycleError if the graph is invalid. Returns a BroadcastDefinition object that can be registered with the runner and triggered.

Patterns

Linear chain

A → B → C. Each node depends on the previous. The output of each node is passed directly as input to the next.

const linear = broadcast("linear")  .input(validate)   // A  .then(charge)      // B: receives validate's output  .then(notify)      // C: receives charge's output  .build();

Fan-out

A → [B, C, D]. Multiple nodes run in parallel after A completes. Each receives A's output.

const fanOut = broadcast("fanOut")  .input(validate)  .then(emailReceipt, smsNotification, slackAlert)  .build(); // Equivalent to:const fanOutExplicit = broadcast("fanOutExplicit")  .input(validate)  .then(emailReceipt, { after: ["validate"] })  .then(smsNotification, { after: ["validate"] })  .then(slackAlert, { after: ["validate"] })  .build();

Fan-in

[B, C] → D. Node D waits for both B and C to complete. Use after to declare dependencies on multiple upstream nodes, and map to combine their outputs.

const fanIn = broadcast("fanIn")  .input(validate)  .then(emailReceipt, { as: "email" })  .then(notifyWarehouse, { as: "warehouse" })  .then(generateReport, {    after: ["email", "warehouse"],    map: (upstream) => ({      emailSent: upstream.email.sent,      warehouseAck: upstream.warehouse.ackId,    }),  })  .build();

Conditional execution

Use when to skip nodes based on upstream results. Skipped nodes (reason: "guard") do not count as failures and do not block their dependents.

const conditional = broadcast("conditional")  .input(validate)  .then(chargeCard, {    when: (upstream) => upstream.validate.paymentMethod === "card",    map: (upstream) => ({ amount: upstream.validate.total }),  })  .then(chargeBankTransfer, {    after: ["validate"],    when: (upstream) => upstream.validate.paymentMethod === "bank",    map: (upstream) => ({ amount: upstream.validate.total }),  })  .then(sendConfirmation, {    after: ["chargeCard", "chargeBankTransfer"],  })  .build();

Data mapping

Use map to transform upstream outputs into the downstream signal's expected input format. The upstream argument is keyed by node name (or the as alias).

const mapped = broadcast("mapped")  .input(fetchUser)  .then(enrichProfile, {    map: (upstream) => ({      userId: upstream.fetchUser.id,      email: upstream.fetchUser.email,    }),  })  .build();

Default data when upstream is skipped

When a node is guard-skipped, its output is undefined in downstream map functions. Use nullish coalescing to provide fallback values.

const withFallback = broadcast("withFallback")  .input(validate)  .then(enrichFromCache, {    when: (upstream) => upstream.validate.cacheEnabled,  })  .then(process, {    after: ["validate", "enrichFromCache"],    map: (upstream) => ({      data: upstream.validate.data,      enrichment: upstream.enrichFromCache ?? { source: "none" },    }),  })  .build();

Triggering

Trigger a broadcast to enqueue it for execution. The input is passed to the entry signal. Returns a broadcast run ID.

// Trigger via the definition (uses the global broadcast adapter)const runId = await orderFlow.trigger({ orderId: "ORD-9281" }); // Trigger via the runner (preferred — uses the runner's own adapter)const runId = await broadcastRunner.trigger("orderFlow", { orderId: "ORD-9281" }); // Optionally wait for the broadcast to completeconst result = await broadcastRunner.waitForBroadcastRun(runId, {  timeoutMs: 30_000,  pollMs: 200,}); if (result?.status === "completed") {  console.log("Broadcast finished successfully");} else if (result?.status === "failed") {  console.error("Broadcast failed:", result.error);}

BroadcastRunner

The broadcast runner orchestrates DAG execution. It polls for triggered broadcasts, initializes node run records, triggers root signals, and advances the graph as nodes complete. It coordinates with a SignalRunner instance to execute individual signals and monitor their completion.

import { BroadcastRunner, ConsoleBroadcastSubscriber } from "station-broadcast";import { BroadcastSqliteAdapter } from "station-adapter-sqlite/broadcast"; const broadcastRunner = new BroadcastRunner({  signalRunner: runner,  adapter: new BroadcastSqliteAdapter({ dbPath: "./data/broadcasts.db" }),  subscribers: [new ConsoleBroadcastSubscriber()],  pollIntervalMs: 1000,}); broadcastRunner.register(orderFlow);broadcastRunner.register(hourlySync); // Start the broadcast runner (blocks until stop() is called)await broadcastRunner.start();

await broadcastRunner.stop({ graceful: true });await signalRunner.stop({ graceful: true });

Constructor options

Methods

BroadcastQueueAdapter

The adapter interface for broadcast storage. Manages broadcast runs and their node runs. Two adapters ship with the framework: BroadcastMemoryAdapter (built into station-broadcast) and BroadcastSqliteAdapter (from station-adapter-sqlite).

BroadcastRun

Represents a single execution of a broadcast.

BroadcastNodeRun

Represents a single node's execution within a broadcast run.

BroadcastSubscriber

All methods are optional. Implement only the events you need. Subscriber errors are caught and logged without affecting broadcast execution.

The built-in ConsoleBroadcastSubscriber logs all events to stdout with a [station-broadcast] prefix.