Workflow builder

The workflow-editor block stores workflows as graphs that members edit on a canvas and publish per tenant. A published version runs on the engine you choose. The Workflow SDK adapter compiles each version to a workflow and records the status of each node, so the canvas shows a run as it happens.

better-supabase add workflow-editor   # adds workflows, tenant and access as well

Grant the keys to your roles in blocks.permissions.roles.

GraphsCopy Anchor Link

A graph is a list of nodes and the edges between them. It has at least one trigger node. A run starts at one of them, and the other nodes run in dependency order once that trigger reaches them; the triggers it didn't start at never run. A trigger picks its node with config.node, and a graph with one trigger node starts there.

import type { WorkflowGraph } from "better-supabase/blocks/workflow-editor";

const graph: WorkflowGraph = {
  nodes: [
    { id: "start", kind: "trigger" },
    {
      id: "lookup",
      kind: "step",
      step: "crm.lookup",
      config: { field: "email" },
    },
    {
      id: "isPaid",
      kind: "condition",
      config: { path: "results.lookup.plan", op: "equals", value: "pro" },
    },
    { id: "review", kind: "approval" },
    { id: "wait", kind: "sleep", config: { duration: "1d" } },
    {
      id: "welcome",
      kind: "step",
      step: "email.send",
      config: { template: "welcome" },
    },
  ],
  edges: [
    { id: "e1", source: "start", target: "lookup" },
    { id: "e2", source: "lookup", target: "isPaid" },
    { id: "e3", source: "isPaid", target: "review", branch: "true" },
    { id: "e4", source: "isPaid", target: "wait", branch: "false" },
    { id: "e5", source: "review", target: "welcome", branch: "true" },
    { id: "e6", source: "wait", target: "welcome" },
  ],
};

An edge without a branch runs its target when its source ran. From a condition or an approval, branch: "true" or "false" picks the outcome it follows. A node runs when one of its incoming edges is active, so two branches can join again.

validateGraph(graph, steps) returns the errors: no trigger, a cycle, an edge to an unknown node, a step outside the library or an invalid condition. The database runs the same checks when it saves a draft and when it publishes one. diffGraphs(from, to) lists the nodes and edges that were added, removed or changed.

Filter conditionsCopy Anchor Link

A condition node can test a whole tree instead of one path: the FilterQuery format of list filters and the rules block, with And and Or groups and every operator of its field types. config.fields declares each field the rules name; its id is the path it reads under input, results or subject. A date field compares instants unless its columnType is date.

const node = {
  id: "large",
  kind: "condition",
  config: {
    fields: [
      { id: "input.amount", type: "number" },
      { id: "subject.type", type: "select" },
    ],
    filter: {
      id: "root",
      type: "group",
      combinator: "and",
      rules: [
        {
          id: "amount",
          type: "rule",
          path: ["input", "amount"],
          operator: "gt",
          value: 500,
        },
        {
          id: "type",
          type: "rule",
          path: ["subject", "type"],
          operator: "is",
          value: "expense",
        },
      ],
    },
  },
};

conditionOf checks the tree against the fields with the list limits (depth 4, 50 rules), and validateGraph reports a tree without a finished rule as an invalid condition. The single-path form keeps working. Relative dates such as "7 days ago" resolve against Date.now(), which the workflow runtime replays deterministically.

Editing and publishingCopy Anchor Link

import "server-only";

import {
  createWorkflowEditor,
  sqlTransport,
} from "better-supabase/blocks/workflow-editor";
import {
  compileGraph,
  graphStarter,
} from "better-supabase/workflow-sdk/builder";

import { steps, stepLibrary } from "@/workflows/steps";
import { graphExecutor } from "@/workflows/graph";

export const builderFor = (sql: Sql) =>
  createWorkflowEditor({
    transport: sqlTransport(sql),
    service: sqlTransport(postgres.admin),
    steps: stepLibrary,
    compile: compileGraph,
    start: graphStarter({ steps, executor: graphExecutor }),
  });

const builder = builderFor(await bsb.sql());

const definition = await builder.definitions
  .upsert({ tenant: organizationId, slug: "onboarding", name: "Onboarding" })
  .orThrow();
const draft = await builder.versions.upsert(definition.id, graph).orThrow();
const published = await builder.versions.publish(draft.id).orThrow();
const runId = await builder
  .run({ definition: definition.id, input: { email } })
  .orThrow();

versions.upsert updates the open draft, or opens the next version when the last one is published. versions.publish runs compile on the graph (a throw fails the publish), publishes the version as the caller and archives the version that was published before. The compiled form is stored through the service transport; without one it isn't stored. publish_workflow_version refuses a non-null compiled from anyone but the service role (WORKFLOW_COMPILED_FORBIDDEN), so a member with workflow.publish can't store code for the server to run. run starts the published version through start with a fresh idempotency key unless you pass one, and returns the engine's run id. The run appears in workflow_runs like any other run.

versions.upsert(definition, graph, { inputSchema }) stores the JSON Schema of the input a run takes on the version (input_schema). A later save without it keeps the schema, and a fork copies it. The builder doesn't check run inputs against it; a template checks the inputs it attaches against the published version's schema.

Runs on a recordCopy Anchor Link

A definition lists the record types its runs can belong to in subjectTypes. Pass a subject to run, and the run is stored with it, so the record's page can list its runs (workflows.runs.list({ subject })) and conditions can read subject.type and subject.id:

await builder.definitions
  .upsert({
    tenant: organizationId,
    slug: "deal-review",
    name: "Deal review",
    subjectTypes: ["deal"],
  })
  .orThrow();

const runId = await builder
  .run({ definition: id, subject: { type: "deal", id: dealId } })
  .orThrow();
const forDeals = await builder.definitions
  .list(organizationId, { subjectType: "deal" })
  .orThrow();

workflow_subject_start_target refuses a type the definition doesn't list (WORKFLOW_SUBJECT_TYPE). With the subjects module installed, the caller must also be able to read the record (can_subject); it runs as the caller, so the record's RLS applies. An event trigger attaches the event's record (subjectType and subjectId on the event or in its payload) when the definition lists its type. Each step gets the subject in call.subject.

Test runsCopy Anchor Link

test starts a run of the open draft (or the version you pass) without running any step: each step node returns its stub, sleeps are skipped, and an approval node takes its stub as the decision (approved without one). It needs workflow.edit. A step without a stub returns the example in its library outputSchema (its first examples entry, else default), else { stubbed: true }.

const runId = await builder
  .test({
    definition: id,
    triggerNode: "start",
    input: { email: "a@example.com" },
    stubs: { lookup: { plan: "pro" }, review: { approved: false } },
  })
  .orThrow();

The run records mode: "test": it writes no outbox events, fires no alerts, and run lists leave it out by default. With dynamic workflows the compiled source handles test runs itself. A static executor must spread graphRunFields(meta) into its runtime and you pass executorRunsTests: true to graphStarter; otherwise graphStarter refuses test runs, because an executor that ignores the stubs would run real steps.

System workflows, forks and opt-outsCopy Anchor Link

A definition with scope: "system" belongs to the app, not to one organization. Only the service role creates one, with no tenant, and the scope can't change afterwards. Every organization whose members hold workflow.read sees it in definitions.list(tenant), and its runs belong to the organization they run in: pass tenant to run. A system version is organization-owned, so its steps run as that organization's service principal (versions.runAs(call.version, call.tenant)).

const mine = await builder.definitions
  .fork({ source: systemId, tenant: organizationId, slug: "onboarding" })
  .orThrow();
await builder.definitions
  .optOut({ tenant: organizationId, definition: systemId })
  .orThrow();

fork copies the source's published graph into a new definition in the tenant as its draft, and records forkedFrom and forkedFromVersion. Lists then return upstreamVersion and behind, so the canvas can say when the source published something newer. A fork doesn't opt the organization out of the source; call optOut for that. An opted-out organization's starts fail with WORKFLOW_OPTED_OUT, its event triggers skip the definition, and a schedule tick moves the schedule on without a run. optOut with optedOut: false opts back in.

Restricting a definitionCopy Anchor Link

A definition with restricted: true can be run, edited and published only by members with workflow.admin and by holders of a grant; every member with workflow.read still sees it. An edit grant includes run.

await builder.definitions.upsert({ ...input, restricted: true }).orThrow();
await builder.definitions.grants
  .add({
    definition: id,
    principalKind: "team",
    principalId: teamId,
    action: "edit",
  })
  .orThrow();

A grant names a user id, a team id (a member of the team, with the teams module installed) or an organization role. Changing restricted and managing grants need workflow.admin.

steps.sync() writes steps to the library (service role). Run it at deploy time or on start-up, so the palette and the database's checks know every step the deployment has. Each step has a name, a title, JSON Schemas for its config and output, and the credentialKind it needs.

Publisher rightsCopy Anchor Link

A published version runs with its publisher's rights, never more. publish_workflow_version records three columns on the version:

Each step node needs a permission. The provider's operations table answers first: an entry workflow.node.<step> (such as workflow.node.crm.lookup) names the permission. Next comes the step's own operation, set on the step in steps (such as invoices.void): the provider's entry for it, or the operation name itself as the key. Without either, the step needs the module's run key, after the rename in blocks["workflow-editor"].permissions. Trigger, sleep, approval and condition nodes need none. A user who doesn't hold every one of them in the tenant can't publish the version: the forbidden error has hint: "WORKFLOW_PUBLISHER_FORBIDDEN" and lists the keys they lack in permissions. The operations table reaches SQL only under the provider access model; see authorization providers.

Pass the server's authorization hooks to the builder. automation.mayPublish then decides each publish as well, with the publisher's claims:

export const builderFor = (sql: Sql) =>
  createWorkflowEditor({
    transport: sqlTransport(sql),
    service: sqlTransport(postgres.admin),
    steps: stepLibrary,
    authorization: server.authorization,
  });

await builder.versions.publish(draft.id, { claims: session.claims }).orThrow();

mayPublish gets { publisher, tenant, permissions, claims }. Anything but true refuses the publish, a throw included. With the hook set, a publish without claims is refused too.

Running a step as the publisherCopy Anchor Link

Call builder.versions.runAs(call.version) at the start of every step, from a builder on the service transport. It reads the version's publisher, checks that they still hold workflow.publish and every stored permission, then asks automation.runAs for the claims the step runs with:

export async function crmLookup(call: GraphStepCall) {
  "use step";
  const claims = await serviceBuilder.versions.runAs(call.version).orThrow();
  const db = bsb.actingAs(String(claims["sub"]), {
    tenant_id: claims["tenant_id"],
  });
  return db.contacts.list({ where: { email: call.input.email } }).orThrow();
}

The step fails with WORKFLOW_PUBLISHER_REVOKED when:

• the publisher was demoted, lost their membership or was deleted;
• runAs returns null or throws;
• the version is organization-owned and the organization has no default service principal, or its role no longer holds every stored permission;
• the version was published before publishers were recorded (owner_kind is null). Publish it again.

The check runs per step, so a run stops at the first step after the publisher loses a permission. The trigger that started the run is recorded for audit only and never changes whom a step runs as. Nothing falls back to the service role or admin(); doctor reports admin() in a file that runs builder steps (BS333).

Without a runAs hook, versions.runAs returns the publisher's claims capped to the stored permissions: { role: "authenticated", sub, tenant_id, bsb_permission_cap }. An organization-owned version runs as the organization's default service principal when that module is installed: the SQL check resolves the principal and returns it as the publisher. Without the module, runAs receives publisher: null and must return claims for the tenant, or every step fails. A hook always overrides the default.

TriggersCopy Anchor Link

await builder.triggers.upsert({ definition: id, kind: "webhook" }).orThrow();
const token = await builder.triggers.rotateToken(triggerId).orThrow(); // shown once

export const POST = (request: Request) =>
  serviceBuilder.triggers.webhook(request);

The webhook route answers 202 with { runId }, 404 for an unknown or disabled token and 405 for a method other than POST. A body that is not JSON arrives as { body: "<text>" }. An Idempotency-Key header keys the start, so a retried delivery starts one run. Only the token's SHA-256 is stored; rotating it ends the old one.

An event trigger's config.when narrows which events start it. It is a filter condition (see Filter conditions) whose field ids start at type, memberships or payload:

await builder.triggers
  .upsert({
    definition: id,
    kind: "event",
    config: {
      type: "expense.submitted",
      when: {
        fields: [{ id: "payload.amount", type: "number" }],
        filter: {
          id: "root",
          type: "group",
          combinator: "and",
          rules: [
            {
              id: "large",
              type: "rule",
              path: ["payload", "amount"],
              operator: "gt",
              value: 500,
            },
          ],
        },
      },
    },
  })
  .orThrow();

onEvent skips a trigger whose when doesn't match the event, and triggers.upsert refuses a when it can't use (invalid_input).

Start from outbox eventsCopy Anchor Link

builder.sink() is an outbox consumer handler that passes each event to triggers.onEvent. App tables with events then start the definitions whose event trigger names task.created or task.updated. A failed start throws, so the outbox retries the batch.

await outbox.consume("workflows", serviceBuilder.sink());

syncSchedule writes through the workflows option, the workflows block built with a transport that may create schedules (the service role); createBlocks passes it as the workflows port when the set has workflowsBlock. Without it, syncSchedule fails with the hint WORKFLOW_BUILDER_WORKFLOWS_CLIENT.

Schedule triggers run on the workflows schedule tick. Pass builder.triggers.starter(fallback) as its start: it starts the builder-trigger: schedules here and hands the others to fallback.

CredentialsCopy Anchor Link

import { vaultCredentials } from "better-supabase/credentials";

const builder = createWorkflowEditor({
  transport,
  service,
  credentials: vaultCredentials(sql),
});

const slack = await builder.credentials
  .create({
    tenant: organizationId,
    kind: "slack",
    name: "Team Slack",
    ref,
    secret,
  })
  .orThrow();
const { token } = await builder.credentials.resolve(slack.id).orThrow(); // inside a step
await builder.credentials.revoke(slack.id);

A credential row holds a credential_ref, never the secret, and the ref carries the row's tenant (tenant refs). create stores secret first when the provider can store one, and authorize returns the URL that connects an account for providers that authorize (such as Vercel Connect). resolve is a service-role call for steps: it returns the token from the provider named in the ref, with the row's scopes. revoke deletes the row and then revokes the secret. Members with workflow.admin manage the tenant's credentials; members with workflow.read list them. Organization exports leave out credential_ref, and the organization purge in data lifecycle revokes each credential before it deletes the row.

Run status on the canvasCopy Anchor Link

nodeRunReporter from better-supabase/workflow-sdk/builder wraps each step so it records running, then completed with its output or failed with its error, in workflow_node_runs. Each record pings workflow-run:<id>.

"use client";

import {
  useWorkflowEditor,
  useWorkflowCanvasRun,
} from "better-supabase/blocks/workflow-editor/react";

export function CanvasStatus({ runId }: { runId: string }) {
  const { run, nodes } = useWorkflowCanvasRun(runId);
  return (
    <p>
      {run?.status}:{" "}
      {Object.values(nodes).filter((n) => n.status === "completed").length}{" "}
      nodes done
    </p>
  );
}

useWorkflowEditor({ tenant }) returns the definitions, the step library and a builder bound to the user's session. useWorkflowCanvasRun(id) returns the run and each node's status by node id, and loads them again on every message on workflow-run:<id>. Both are Client Component hooks; in Server Components, call builder.definitions.list() and builder.nodeRuns.list(run) instead.

VueCopy Anchor Link

better-supabase/blocks/workflow-editor/vue exports useWorkflowEditor and useWorkflowCanvasRun. Each takes the options of the controller it wraps (useWorkflowEditor those of createWorkflowEditorController); auth and ports come from app.use(betterSupabase(bsb)) unless you pass them. state is a shallow ref of the snapshot and actions are the controller's actions. The work starts on mount, and the controller is disposed with the component.

<script setup lang="ts">
import { useSupabase } from "better-supabase/vue";
import { useWorkflowEditor } from "better-supabase/blocks/workflow-editor/vue";
import {
  createWorkflowEditor,
  rpcTransport,
} from "better-supabase/blocks/workflow-editor";

const { organizationId } = defineProps<{ organizationId: string }>();
const supabase = useSupabase();
const { state } = useWorkflowEditor({
  client: createWorkflowEditor({ transport: rpcTransport(supabase) }),
  tenant: organizationId,
});
</script>

<template>
  <ul>
    <li v-for="definition in state.definitions ?? []" :key="definition.id">
      {{ definition.name }}
    </li>
  </ul>
</template>

SvelteCopy Anchor Link

better-supabase/blocks/workflow-editor/svelte exports useWorkflowEditor and useWorkflowCanvasRun. Each takes the options of the controller it wraps (useWorkflowEditor those of createWorkflowEditorController); auth and ports come from setBetterSupabase(bsb) unless you pass them. The hook returns a readable store of the snapshot, read with $, with the actions on .actions. The work starts on mount, and the controller is disposed with the component.

<script lang="ts">
  import { getBetterSupabase } from "better-supabase/svelte";
  import { useWorkflowEditor } from "better-supabase/blocks/workflow-editor/svelte";
  import { createWorkflowEditor, rpcTransport } from "better-supabase/blocks/workflow-editor";

  let { organizationId }: { organizationId: string } = $props();
  const { supabase } = getBetterSupabase();
  const builder = useWorkflowEditor({
    client: createWorkflowEditor({ transport: rpcTransport(supabase) }),
    tenant: organizationId,
  });
</script>

<ul>
  {#each $builder.definitions ?? [] as definition (definition.id)}
    <li>{definition.name}</li>
  {/each}
</ul>

SolidCopy Anchor Link

better-supabase/blocks/workflow-editor/solid exports useWorkflowEditor and useWorkflowCanvasRun. Each takes the options of the controller it wraps (useWorkflowEditor those of createWorkflowEditorController); auth and ports come from <BetterSupabaseProvider client={bsb}> unless you pass them. state() reads the snapshot through from, and actions are the controller's actions. The work starts on mount, and the controller is disposed with the owner.

import { For } from "solid-js";
import { useSupabase } from "better-supabase/solid";
import { useWorkflowEditor } from "better-supabase/blocks/workflow-editor/solid";
import {
  createWorkflowEditor,
  rpcTransport,
} from "better-supabase/blocks/workflow-editor";

export function Component(props: { organizationId: string }) {
  const supabase = useSupabase();
  const { state } = useWorkflowEditor({
    client: createWorkflowEditor({ transport: rpcTransport(supabase) }),
    tenant: props.organizationId,
  });
  return (
    <ul>
      <For each={state().definitions ?? []}>
        {(definition) => <li>{definition.name}</li>}
      </For>
    </ul>
  );
}

AlertsCopy Anchor Link

await builder.alerts.upsert({
  definition: id,
  onEvent: "failed",
  channel: { type: "email", to },
});
await builder.alerts.upsert({
  definition: id,
  onEvent: "slow",
  threshold: 3600,
  channel,
});
await serviceBuilder.alerts.check(); // from a cron route

A failed alert fires when a run of the definition fails. A slow alert fires from alerts.check() when a run is still unfinished threshold seconds after it started. Each alert fires once per run and emits a workflow_alert.triggered outbox event with the alert, its channel and the run. Your outbox consumer sends it; the module only stores the channel. Runs match a definition on their bsb.definition attribute, which graphStarter sets.

FunctionsCopy Anchor Link

Your own columnsCopy Anchor Link

Add columns to the workflow definition table with blocks["workflow-editor"].options.extraColumns (column name to SQL type), then pass a Standard Schema for them as fields. definitions.upsert validates and writes them, and definitions.upsert, list, get and fork return them parsed under fields. The block also takes hooks around its methods; see extending blocks.

import { createWorkflowEditor } from "better-supabase/blocks/workflow-editor";
import { z } from "zod";

const client = createWorkflowEditor({
  transport,
  fields: z.object({ priority: z.enum(["low", "high"]) }),
});