Workflow engine

Workflows react to changes in your OData data: when something changes and a condition holds, run a sequence of steps — calling OData, sending email, pausing for human approval, and resuming. Workflows live inside app.json (workflows), so the same schema + versioning applies.

A workflow

{
  "id": "low-stock",
  "name": "Email on low stock",
  "trigger": { "on": "update", "entitySet": "Products", "condition": { "field": "UnitsInStock", "lt": 10 } },
  "steps": [
    { "kind": "approval", "message": "Approve reorder?" },
    { "kind": "email", "to": "buyer@acme.com", "subject": "Reorder", "body": "Stock is low." }
  ]
}

• trigger.on — create · update · delete · manual · schedule (+ entitySet, optional schedule cron).
• trigger.condition — the shared Condition, evaluated against the changed row. A failing condition → the run is skipped.
• steps — run sequentially; the first error stops the run; an approval step pauses it.

Step kinds

kind	Effect
odataAction	Invoke a bound/unbound OData Action.
updateEntity	PATCH an entity.
email	Send mail (via a configured sender).
webhook	POST to a configured target (egress allow‑list — not an arbitrary URL).
delay	Wait.
approval	Pause for a human decision (see below).
aiDecision	Ask the LLM for a { decision, reason } (structured).

Handlers are pluggable: delay/approval/aiDecision are built in; createODataHandlers, createEmailHandler, createWebhookHandler are injected (real OData PATCH/POST are verified against the mock server in tests).

Change detection (the PX0 ladder)

OData isn't an event bus, and no single detection strategy works on every service (we spiked this). Detection is pluggable, picking the highest rung a service supports:

delta links → LastModified polling → ETag + key-set → snapshot+hash → scheduled/manual (baseline)

The shipped universal floor is SnapshotHashDetector: it baselines on the first poll, then emits create/update/delete with an idempotency key entitySet:key:(hash | deleted). The engine dedupes on that key, so overlapping polls never double‑fire.

The product promise is near‑real‑time where the service supports it; scheduled/manual otherwise — not guaranteed real‑time push.

The loop

import { WorkflowEngine, SnapshotHashDetector, MemoryApprovalStore, createEmailHandler } from '@samabaasi/engine';

const engine = new WorkflowEngine({
  workflows: app.workflows,
  detectors: [new SnapshotHashDetector({ baseUrl, entitySet: 'Products', keyField: 'ProductID' })],
  handlers: createEmailHandler(sendMail),
  approvals: new MemoryApprovalStore()
});

await engine.tick();   // poll → dedupe → enqueue → run matching workflows → log

tick() polls detectors, dedupes, enqueues each change, drains the queue (retries → dead‑letter), runs every matching workflow, and appends an ExecutionLogEntry. A scheduler calls tick() on an interval; in production the queue is pg‑boss so execution survives restarts.

Human approval (pause → resume)

When a run hits an approval step it returns awaiting_approval and the engine records a pending approval (the resume state is server‑held; list is metadata‑only):

const [pending] = await approvals.list({ status: 'pending' });
await engine.resolveApproval(pending.id, 'approve', 'manager@acme.com'); // resumes remaining steps
// or 'reject' → the run stops

On approve, the remaining steps run from where it paused (prior outputs seeded); on reject it stops. An Approval Queue UI (reusing the library's DataTable) is the remaining frontend piece.

Safety boundary

Workflow steps are author‑defined — they run as configured (like a CI pipeline). The allow‑list + per‑call approval gating applies to the AI‑chosen path (the agent), covered next.