@spec check_worker(module(), keyword()) :: map() | nil

Returns runtime worker state for a flow on this node, or nil if none is running.

The map includes effective and configured limits, lifecycle, active task count, and the worker's module and pid.

Options

• :timeout - milliseconds (or :infinity) to wait for the worker to reply. Default: 5_000. A worker busy in a slow database call may not reply in time; the caller then exits with {:timeout, {GenServer, :call, _}}.

Raises ArgumentError on unknown or invalid options.

Examples

PgFlow.check_worker(MyApp.Flows.ProcessOrder)
# => nil

{:ok, _} = PgFlow.start_worker(MyApp.Flows.ProcessOrder)
PgFlow.check_worker(MyApp.Flows.ProcessOrder)
# => %{module: MyApp.Flows.ProcessOrder, pid: #PID<...>, lifecycle: :running, ...}

@spec child_spec(keyword()) :: Supervisor.child_spec()

Returns a child specification for starting PgFlow under a supervisor.

Options

See PgFlow.Config for available options.

@spec compatibility_report() :: %{
  core_version_string: String.t(),
  upstream_sha: String.t(),
  upstream_release: String.t(),
  bundled_core_version: non_neg_integer(),
  bundled_helpers_version: non_neg_integer()
}

Returns bundled core/helpers version floors and the pinned upstream SHA.

Use this in release notes, CI evidence, and operator runbooks. The core_version_string field mirrors core_version/0; the integer bundled_core_version and bundled_helpers_version fields are what mix pgflow.check_schema verifies against the database.

@spec core_version() :: String.t()

Returns the legacy pgflow core semantic version string.

This value is kept stable for existing callers. It does not track the upstream npm release number or the bundled EctoEvolver core/helpers integers. Use compatibility_report/0 for the pinned upstream SHA and installed version floors.

Examples

PgFlow.core_version()
#=> "0.5.0"

@spec delete_flow(String.t()) :: :ok | {:error, term()}

Deletes a flow and all associated data (runs, tasks, queue).

Examples

PgFlow.delete_flow("acct_123_hubspot_sync_v1")

@spec enqueue(module(), map()) :: {:ok, String.t()} | {:error, term()}

Enqueues a background job with the given input.

This is the primary API for dispatching jobs. Under the hood, jobs are single-step flows, so this delegates to start_flow/2. Returns {:error, {:flow_not_compiled, flow_slug}} when the job definition has not been compiled into the database.

Examples

{:ok, run_id} = PgFlow.enqueue(MyApp.Jobs.SendEmail, %{"to" => "user@example.com"})

@spec enqueue(module(), map(), keyword()) :: {:ok, String.t()} | {:error, term()}

Enqueues a background job with options.

Supported options:

• :delay_seconds - non-negative integer seconds before the job is available
• :scheduled_at - DateTime when the job should become available

Examples

{:ok, run_id} = PgFlow.enqueue(MyApp.Jobs.SendEmail, %{"to" => "user@example.com"}, delay_seconds: 60)
{:ok, run_id} = PgFlow.enqueue(MyApp.Jobs.SendEmail, %{"to" => "user@example.com"}, scheduled_at: ~U[2026-05-08 12:00:00Z])

@spec enqueue_at(module(), map(), DateTime.t()) ::
  {:ok, String.t()} | {:error, term()}

Enqueues a background job that becomes available at scheduled_at.

Timestamps in the past enqueue the job for immediate execution. Any DateTime time zone is accepted; PgFlow compares the scheduled timestamp as an instant.

Examples

{:ok, run_id} = PgFlow.enqueue_at(MyApp.Jobs.SendEmail, %{"to" => "user@example.com"}, ~U[2026-05-08 12:00:00Z])

@spec enqueue_in(module(), map(), non_neg_integer()) ::
  {:ok, String.t()} | {:error, term()}

Enqueues a background job that becomes available after delay_seconds.

PgFlow persists the run immediately, then delays the initial pgmq task visibility so workers cannot execute it until the delay elapses. Values of 0 enqueue the job for immediate execution.

Examples

{:ok, run_id} = PgFlow.enqueue_in(MyApp.Jobs.SendEmail, %{"to" => "user@example.com"}, 60)

@spec flow_exists?(String.t()) :: {:ok, boolean()} | {:error, term()}

Checks if a flow exists in the database.

Examples

PgFlow.flow_exists?("my_flow")
# => {:ok, true}

@spec get_flow(module() | atom()) :: {:ok, map()} | {:error, :not_found}

Gets the definition for a flow by module or slug.

Examples

{:ok, flow_def} = PgFlow.get_flow(MyApp.Flows.ProcessOrder)
{:ok, flow_def} = PgFlow.get_flow(:process_order)
{:error, :not_found} = PgFlow.get_flow(:unknown)

@spec get_run(String.t()) ::
  {:ok, PgFlow.Schema.Run.t()} | {:error, :invalid_id | :not_found | term()}

Gets a run by ID.

Examples

{:ok, run} = PgFlow.get_run("550e8400-e29b-41d4-a716-446655440000")
{:error, :invalid_id} = PgFlow.get_run("not-a-uuid")
{:error, :not_found} = PgFlow.get_run(Ecto.UUID.generate())

@spec get_run_with_states(String.t()) ::
  {:ok, PgFlow.Schema.Run.t()} | {:error, :invalid_id | :not_found | term()}

Gets a run with all step states preloaded.

Examples

{:ok, run} = PgFlow.get_run_with_states("550e8400-e29b-41d4-a716-446655440000")
run.step_states  # => [%StepState{}, ...]

@spec health_check() :: %{status: :ok, workers: [map()], flows: [map()]}

Returns health check information.

Examples

PgFlow.health_check()
#=> %{status: :ok, workers: [...], flows: [...]}

@spec list_flows() :: [map()]

Lists all registered flows.

Examples

flows = PgFlow.list_flows()
#=> [%{module: MyApp.Flows.ProcessOrder, slug: :process_order, ...}, ...]

@spec scale_worker(module(), keyword()) ::
  {:ok, map()} | {:error, :not_found | :not_running}

Scales a running worker's concurrency or batch size on this node.

Requires at least one of :max_concurrency or :batch_size; raises ArgumentError on missing or invalid options. :timeout (default 5_000 ms, or :infinity) bounds the call to the worker; when it elapses the caller exits with {:timeout, {GenServer, :call, _}}, and the worker may already have applied (or may still apply) the scale. Scale-down does not interrupt active tasks; any scale that leaves free slots polls for queued work immediately. Runtime limits reset to the configured values when the worker restarts.

Returns {:error, :not_found} when no local worker exists for the module (including the brief window while a crashed worker is being restarted), and {:error, :not_running} when the worker is not in the :running lifecycle state (for example while it is stopping).

Examples

{:ok, pid} = PgFlow.start_worker(MyApp.Flows.Import)
{:ok, %{max_concurrency: 6}} = PgFlow.scale_worker(MyApp.Flows.Import, max_concurrency: 6)

# Scale down while tasks finish; active handlers are not interrupted.
PgFlow.scale_worker(MyApp.Flows.Import, max_concurrency: 1)

# After a crash or restart, limits return to configured values, not the scaled ones.
Process.exit(pid, :kill)
# ... worker supervises back with configured max_concurrency

@spec start_flow(module() | atom() | String.t(), map()) ::
  {:ok, String.t()} | {:error, term()}

Starts a flow run with the given input.

The flow can be specified by module name, slug atom, or slug string. Returns {:error, {:flow_not_compiled, flow_slug}} when the flow definition is not present in pgflow.flows.

Examples

{:ok, run_id} = PgFlow.start_flow(MyApp.Flows.ProcessOrder, %{"order_id" => 123})
{:ok, run_id} = PgFlow.start_flow(:process_order, %{"order_id" => 123})

@spec start_flow_sync(module() | atom() | String.t(), map(), keyword()) ::
  {:ok, PgFlow.Schema.Run.t()}
  | {:error, PgFlow.Schema.Run.t()}
  | {:error, :timeout}
  | {:error, term()}

Starts a flow and waits for completion.

Blocks until the flow completes or the timeout is reached.

Options

• :timeout - Maximum time to wait in milliseconds (default: 60_000)
• :poll_interval - How often to check status in milliseconds (default: 500)

Examples

{:ok, run} = PgFlow.start_flow_sync(MyApp.Flows.ProcessOrder, %{"order_id" => 123})
{:error, run} = PgFlow.start_flow_sync(MyApp.Flows.FailingFlow, %{})

@spec start_link(keyword()) :: Supervisor.on_start()

Starts the PgFlow supervision tree.

Options

See PgFlow.Config for available options.

Examples

PgFlow.start_link(repo: MyApp.Repo, flows: [MyApp.Flows.ProcessOrder])

@spec start_worker(module(), keyword()) ::
  {:ok, pid()}
  | {:error, {:conflicting_worker_options, pid()} | :not_started}
  | {:error, term()}

Starts a worker for the given flow.

Options

• :repo - Ecto repository (defaults to the configured repository).
• :max_concurrency - Override max concurrent tasks for this worker.
• :batch_size - Override batch size for this worker.

Polling and other worker settings come from the PgFlow supervisor configuration. Limits resolve from the global max_concurrency/batch_size, then the module's worker_options entry, then call-time options. worker_options applies to every worker for a module listed in :flows or :jobs, whether PgFlow.FlowStarter or this function starts it; config validation rejects entries for unlisted modules, so a module started only through this function takes its overrides at call time. A worker restarted by its supervisor keeps the options it was started with.

When a worker for the module is already running, call-time options are compared against its configured (start-time) limits, not limits changed later by scale_worker/2: a match returns the running worker and a mismatch returns {:error, {:conflicting_worker_options, pid}}.

Returns {:error, :not_started} when a concurrent supervisor restart started a replacement worker that exited before it could be returned; retrying is safe. Raises ArgumentError for unknown option keys or invalid values.

Examples

{:ok, pid} = PgFlow.start_worker(MyApp.Flows.ProcessOrder)
{:ok, pid} = PgFlow.start_worker(MyApp.Flows.Import, max_concurrency: 4)

@spec stop_worker(module()) :: :ok | {:error, :not_found}

Stops a worker for the given flow.

Examples

:ok = PgFlow.stop_worker(MyApp.Flows.ProcessOrder)

@spec upsert_flow(String.t(), keyword()) :: {:ok, map()} | {:error, term()}

Recompiles a flow definition at runtime.

Worker startup uses PgFlow.Worker.Bootstrap.prepare/2 for shape verification. Use this API for explicit runtime management when destructive recompilation is intended.

Unlike the compile-time DSL (use PgFlow.Flow), this creates flow definitions from plain data - for per-tenant automations and dynamic workflows.

If the flow already exists, this operation is destructive: existing definition and historical run/task data for the slug are deleted first.

Examples

 PgFlow.upsert_flow("acct_123_hubspot_sync_v1",
   max_attempts: 3,
   steps: [
     %{slug: "reshape", deps: []},
     %{slug: "create_contact", deps: ["reshape"]}
   ]
 )

@spec upstream_sha() :: String.t()

Returns the pinned upstream pgflow git SHA this release aligns with.

The schema contract is pgflow 0.17.2 at this commit.