@spec archive_messages(Ecto.Repo.t(), String.t(), [pos_integer()]) ::
  {:ok, [integer()]} | {:error, term()}

Archives messages out of a flow's live queue.

Returns {:ok, archived_msg_ids} — pgmq reports only the ids it actually archived, so ids already absent from the live queue are missing from the result rather than raising.

@spec compile_flow(Ecto.Repo.t(), String.t(), map(), [map()]) ::
  {:ok, term()} | {:error, term()}

Compiles and upserts a flow definition.

Parameters

• repo - The Ecto repository
• slug - The flow identifier slug
• opts - Flow options map (e.g., %{max_retries: 3})
• steps - List of step definitions as maps

@spec complete_task(
  Ecto.Repo.t(),
  String.t(),
  String.t(),
  non_neg_integer(),
  map() | list()
) ::
  {:ok, term()} | {:error, term()}

Marks a task as completed with output data.

Parameters

• repo - The Ecto repository
• run_id - The flow run UUID
• step_slug - The step identifier slug
• task_index - The task index (0-based)
• output - Output data as an Elixir term (will be encoded as JSONB)

Returns

• {:ok, result} - Success result from the database
• {:error, reason} - Error details if the operation fails

@spec delay_run(Ecto.Repo.t(), String.t(), String.t(), non_neg_integer()) ::
  :ok | {:error, term()}

Delays the first queued task for a flow run by moving its pgmq visibility time.

This is a lower-level helper for public APIs such as PgFlow.enqueue_in/3 and PgFlow.enqueue_at/3. Call it in the same repository transaction as start_flow/3 when callers need to ensure workers cannot see the task before the delay is applied.

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

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

Parameters

• repo - The Ecto repository
• slug - The flow identifier slug

Returns

• :ok on success (including when flow doesn't exist)
• {:error, term()} on failure

@spec delete_message(Ecto.Repo.t(), String.t(), pos_integer()) ::
  {:ok, boolean()} | {:error, term()}

Deletes a message from a PGMQ queue.

@spec fail_task(Ecto.Repo.t(), String.t(), String.t(), non_neg_integer(), String.t()) ::
  {:ok, term()} | {:error, term()}

Marks a task as failed with an error message.

Parameters

• repo - The Ecto repository
• run_id - The flow run UUID
• step_slug - The step identifier slug
• task_index - The task index (0-based)
• error_message - Error description string

Returns

• {:ok, result} - Success result from the database
• {:error, reason} - Error details if the operation fails

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

Checks if a flow exists in the database.

@spec get_flow_input(Ecto.Repo.t(), String.t()) ::
  {:ok, map() | list()} | {:error, term()}

Retrieves the input data for a flow run.

@spec get_run(Ecto.Repo.t(), String.t()) ::
  {:ok, %{status: String.t(), output: term()}} | {:error, :not_found | term()}

Retrieves a flow run's current state.

Parameters

• repo - The Ecto repository
• run_id - The flow run UUID

Returns

• {:ok, %{status: String.t(), output: term()}} - Run state
• {:error, :not_found} - Run does not exist
• {:error, reason} - Error details if the operation fails

@spec get_step_output(Ecto.Repo.t(), String.t(), String.t()) ::
  {:ok, map() | nil} | {:error, term()}

Retrieves the output for a specific step in a flow run.

@spec list_skipped_steps(Ecto.Repo.t(), String.t()) ::
  {:ok, [%{step_slug: String.t(), skip_reason: String.t()}]} | {:error, term()}

Lists skipped steps for a flow run in dependency order.

Ordered by skip time, ties broken by pgflow.steps.step_index. A cascade stamps every step in the chain with the same skipped_at = now(), so ties are the rule rather than the exception; step_index is assigned in add_step order, which is topological (a step's deps must already exist), and is what core SQL itself uses to walk a cascade. Breaking ties by slug would announce a child before its parent.

Parameters

• repo - The Ecto repository
• run_id - The flow run UUID

Returns

• {:ok, [%{step_slug: String.t(), skip_reason: String.t()}]} - Skipped steps
• {:error, reason} - Error details if the operation fails

@spec orphaned_queue_messages(Ecto.Repo.t(), String.t(), [pos_integer()]) ::
  {:ok,
   [
     %{
       msg_id: integer(),
       step_slug: String.t() | nil,
       step_status: String.t() | nil
     }
   ]}
  | {:error, term()}

Diagnoses messages pgflow.start_tasks declined to convert into tasks.

Under the archive invariant (every SQL path that ends a step — skip, completion, permanent failure — archives that step's queued pgmq messages in the same transaction), a declined message has normally already left the queue. This probe returns the declined messages still present in the live queue, each with the status of the step it belongs to, so the worker can tell a benign race (message already archived) from a broken invariant (message still queued for a terminal step, doomed to redeliver forever).

Each returned orphan is %{msg_id:, step_slug:, step_status:}; the step fields are nil when the message has no matching step_tasks row.

@spec prune_data(Ecto.Repo.t(), pos_integer(), keyword()) ::
  {:ok, prune_result()} | {:error, term()}

Prunes old flow run data older than the specified retention period.

Options

• :flow_slugs - List of flow slugs to prune (default: all flows)

@spec read(Ecto.Repo.t(), String.t(), pos_integer(), pos_integer()) ::
  {:ok, [list()]} | {:error, term()}

Reads messages from a queue without blocking (non-blocking read).

Uses pgmq.read() to fetch available messages. Returns immediately whether or not messages are available. Messages are made invisible for the visibility timeout period to prevent duplicate processing.

Queue poll SQL logging is disabled by default because workers call this frequently. Set config :pgflow, :log_queue_polls, true to enable Ecto query logging for these reads while debugging.

Parameters

• repo - The Ecto repository
• queue_name - The name of the queue to read from (matches flow_slug)
• visibility_timeout - Time in seconds messages remain invisible
• batch_size - Maximum number of messages to retrieve

Returns

• {:ok, messages} - List of message records from pgmq (may be empty)
• {:error, reason} - Error details if the operation fails

@spec recover_stalled_tasks(Ecto.Repo.t(), pos_integer()) ::
  {:ok, non_neg_integer()} | {:error, term()}

Recovers stalled tasks via the pgflow.recover_stalled_tasks helper, returning the number requeued.

A task is stalled once it has been started longer than its effective timeout — coalesce(step.opt_timeout, flow.opt_timeout) — plus stale_threshold seconds of buffer. Stalled tasks are reset to queued; past a requeue cap they are archived and marked permanently_stalled_at.

The deadline is step-aware on purpose: upstream pgflow's requeue_stalled_tasks() deadlines on the flow timeout alone, which would reclaim a healthy long step (e.g. a step with timeout: 120 under a 30s flow default) mid-flight. Step-awareness matches how start_tasks sets each message's pgmq visibility timeout.

@spec start_flow(Ecto.Repo.t(), String.t(), map() | list()) ::
  {:ok, String.t()} | {:error, term()}

Starts a new flow execution.

Parameters

• repo - The Ecto repository
• flow_slug - The flow identifier slug
• input - Input data as an Elixir term (will be encoded as JSONB)

Returns

• {:ok, run_id} - The UUID of the created flow run
• {:error, reason} - Error details if the operation fails

@spec start_flow_with_run(Ecto.Repo.t(), String.t(), map() | list()) ::
  {:ok, String.t(), map()} | {:error, term()}

Starts a new flow execution and returns the run row pgflow.start_flow itself produced, alongside the run id.

pgflow.start_flow runs run creation, condition evaluation (cascade_resolve_conditions), taskless-step completion, and initial task enqueuing all inside the one implicit transaction backing this statement. The row this function returns is read back inside that same transaction, so it is the authoritative snapshot of whatever the statement decided — no external worker can have touched the run yet, because nothing about it is visible to another session until this statement commits. Callers that need to know whether start_flow itself completed or failed the run synchronously (e.g. a root condition with when_unmet: :fail) should use this snapshot instead of issuing a second query, which would race a fast worker and risk mislabeling a genuine handler failure.

This snapshot only covers the run's own status/output — it does not extend to skipped steps. The private emit_post_start client helper still calls PgFlow.Telemetry.emit_skipped_steps/3 as a separate, post-commit query, which can race a worker sweeping the same run; see the delivery contract on PgFlow.Telemetry.emit_skipped_steps/4 for what that guarantees.

Returns

• {:ok, run_id, run} - The run id and a map of the returned run row (:run_id, :flow_slug, :status, :input, :output, :remaining_steps, :started_at, :completed_at, :failed_at)
• {:error, reason} - Error details if the operation fails

@spec start_tasks(Ecto.Repo.t(), String.t(), [pos_integer()], String.t()) ::
  {:ok, [list()]} | {:error, term()}

Starts multiple tasks by marking messages as in-progress.

Parameters

• repo - The Ecto repository
• flow_slug - The flow identifier slug
• msg_ids - List of message IDs from pgmq
• worker_id - The worker UUID string

Returns

• {:ok, task_details} - List of task detail records
• {:error, reason} - Error details if the operation fails

@spec upsert_flow(Ecto.Repo.t(), String.t(), map(), [map()]) ::
  {:ok, map()} | {:error, term()}

Recompiles a flow definition from runtime options.

Uses create_flow + add_step (the proven low-level SQL functions) to register a flow. If the flow already exists, it is dropped and re-created to ensure the definition matches.

This operation is destructive for existing flows: all historical run and task data for the slug is deleted before recompiling.

Parameters

• repo - The Ecto repository
• slug - The flow identifier slug
• opts - Flow-level options map with keys: "max_attempts", "base_delay", "timeout"
• steps - List of step definition maps with keys: "slug", "deps", "step_type", and optional "max_attempts", "base_delay", "timeout", "start_delay", "if", "if_not", "when_unmet", "when_exhausted" (string or atom keys)

Returns

• {:ok, %{"status" => status}} where status is "compiled" or "recompiled"
• {:error, term()} on failure

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

Checks if a slug is valid according to core pgflow rules.