ordu

ordu

Execute task functions in a configurable order, sharing a context and mutating a shared data structure. Used by the Seneca framework as its extension-point engine.

A Go port is also available — see go/README.md.

---

Tutorial: your first ordu

Install:

npm install ordu

Build a small pipeline. Each task receives { ctx, data } and may modify data directly, or return a { op, out } instruction:

const { Ordu } = require('ordu')

const w = new Ordu()

w.add(function first(spec) {
  if (null == spec.data.foo) {
    return { op: 'stop', err: new Error('no foo') }
  }
  spec.data.foo = spec.data.foo.substring(0, spec.ctx.len)
})

w.add(function second(spec) {
  spec.data.foo = spec.data.foo.toUpperCase()
})

const ctx  = { len: 3 }
const data = { foo: 'green' }

const res = await w.exec(ctx, data)
console.log(res.data.foo)  // → 'GRE'

Tasks run in the order added. data is the shared, mutable structure; ctx is read-mostly context that travels alongside it. The return value of exec is a structured result containing the final data, the task log, and any error.

For synchronous pipelines, use execSync and tasks that don't return promises.

---

How-to guides

Choose what happens after a task: operators

Each task can return { op, out, err, why }. The op field names an operator that decides whether to continue and how to fold out back into data.

op	Effect
next	Continue to the next task. out is ignored.
merge	Merge out into data (reference-preserving via nua), then continue.
stop	Merge out into data, then stop. err is propagated to the caller.
skip	Continue, marking the task as skipped (no merge).
custom	Any operator registered with w.operator(name, fn).

w.add(() => ({ op: 'merge', out: { stage: 'parsed' } }))
w.add(() => ({ op: 'stop',  err: new Error('halt'), why: 'unauthorized' }))

Returning null or undefined is equivalent to { op: 'next' }.

Register a custom operator

w.operator('retry', (taskResult, ctx, data) => {
  data._retries = (data._retries || 0) + 1
  return { stop: data._retries > 3 }
})

w.add(() => ({ op: 'retry' }))

Operators receive the TaskResult, the ctx, and the data object, and return { stop: boolean, err?: Error }. They may be async.

Order tasks with before / after

Tasks are inserted at the position implied by their before/after references. If neither is given they are appended:

w.add({ name: 'parse',    exec: parse })
w.add({ name: 'validate', exec: validate })
w.add({ name: 'log',      after: 'parse', exec: log })
// order: parse, log, validate

Run a task only when data matches a pattern

w.add({
  if: { stage: 'parsed', kind: 'request' },
  exec: handleRequest,
})

Each if key is a dot-separated path into data; values match by deep equality (objects/arrays match if they are deeply contained).

Restrict execution to a subset (active flag)

A task with active: false is skipped on every run, but stays in the list (so tooling can still see it):

w.add({ name: 'debug-dump', active: false, exec: dump })

Tag-based filtering

if plus a tag-style key on data lets you flip subsets on per-run:

w.add({ if: { upper: true }, exec: (s) => { s.data.foo = s.data.foo.toUpperCase() }})

await w.exec({}, { foo: 'green', upper: true })   // → 'GREEN'
await w.exec({}, { foo: 'green' })                // → 'green'

Walk into nested data with select + apply

select picks a child collection out of data (by dot-path or function); apply is a sub-pipeline run once per child. The sub-pipeline's task receives the child as spec.node:

w.add({
  select: 'items',
  apply: [
    function tally(s) { s.data.total = (s.data.total || 0) + s.node.val.qty },
  ],
})

await w.exec({}, { items: [{ qty: 1 }, { qty: 2 }, { qty: 3 }], total: 0 })
// → data.total === 6

select may also be a function (source, spec) => any, useful when the child set isn't directly addressable by path.

Observe progress with events

w.on('task-result', (result) => console.log(result.name, result.op))
w.on('task-end',    (entry)  => console.log('done', entry.name))

The Ordu class extends EventEmitter. Events fire synchronously during execution.

Sort selected children

When applying to a collection, sort children alphabetically by key:

await w.exec({}, data, { select: { sort: 1 } })   // ascending
await w.exec({}, data, { select: { sort: -1 } })  // descending

---

Reference

new Ordu(opts?)

Construct an ordu. Options:

Option	Type	Default	Description
debug	boolean	false	Capture call sites for add() and snapshot data per task.
select.sort	number | null	null	Default sort direction for select+apply children.

Instance methods

Method	Description
add(taskDef | taskExec | array)	Register one or more tasks. Returns this.
tasks()	Return the ordered task list.
task[name]	Map of registered tasks by name.
operator(name, fn) / operator(fn)	Register an operator (uses fn.name if name is omitted).
operators()	Return the operator map.
exec(ctx?, data?, opts?)	Run the pipeline, returning a Promise<ExecResult>.
execSync(ctx?, data?, opts?)	Synchronous variant; throws if any task returns a Promise.

TaskDef

{
  id?:     string                // stable identifier (default: random)
  name?:   string                // default: `task<N>`
  before?: string                // insert before this task name
  after?:  string                // insert after this task name
  exec?:   (spec: TaskSpec) => any
  if?:     { [path: string]: any }
  active?: boolean               // default: true
  meta?:   any
  select?: string | ((source, spec) => any)
  apply?:  TaskDef | TaskDef[]
}

TaskSpec (passed to exec)

{ ctx, data, task, async, opts?, node? }

node is set inside select+apply sub-tasks to { key, val } for the current child.

TaskExec return shape

{ op?: string, out?: object, err?: Error, why?: string } | void

ExecResult

{
  tasklog:    TaskLogEntry[]
  task?:      Task                // the task that errored, if any
  taskcount:  number              // tasks actually executed
  tasktotal:  number              // tasks considered
  start:      number              // ms since epoch
  end:        number
  err?:       Error
  data:       any                 // the (mutated) data object
}

Built-in operators

next, merge, stop, skip (see the How-to guide above).

Events

Event	Payload
task-result	TaskResult
task-end	{ name, op, task, result, operate, data? }

---

Explanation: why an ordered task list?

A lot of frameworks need an extension point that is more than "fire a hook" but less than a full plugin system. You want:

• Determinism. The order tasks run in is the order you registered them, with explicit before/after overrides — not the order modules happened to load.
• Inspection. tasks() lists exactly what will run; tasklog records what did run, with timings and operator decisions.
• In-place data. A shared data object is mutated as it flows through the pipeline, with nua-backed merges so external references to inner objects stay live across merge operations.
• Early exit without throwing. Returning { op: 'stop', err } ends the run cleanly and surfaces the error in ExecResult.err, with the failing task in ExecResult.task.

The trade-off is that data is mutable and shared. If you want immutability, copy data before calling exec, or use an operator that produces a fresh object instead of merging.

select + apply exist for the (common) case where one stage of the pipeline needs to fan out over a child collection. Doing it as nested ordus keeps the same operator/event machinery available at each level.

---

Notes

From the Irish ordú ("instruction"). Pronounced or-doo.

License

Copyright (c) 2014-2025, Richard Rodger and other contributors. Licensed under MIT.