@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(),
  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 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, term()}

Starts a worker for the given flow.

Options

• :repo - Ecto repository (defaults to the configured repository).

Polling and concurrency settings come from the PgFlow supervisor configuration.

Examples

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

@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.

Compatibility is defined against this commit, not against an unreleased npm version such as upstream 0.17.0.