# `PgFlow.Queries.Flows`
[🔗](https://github.com/agoodway/pgflow/blob/v0.3.1/lib/pgflow/queries/flows.ex#L1)

SQL query interface for pgflow flow operations.

Provides functions for starting flows, completing/failing tasks, reading
messages, and managing flow lifecycle. All functions that accept JSON data
expect Elixir terms that will be encoded with Jason.

# `prune_result`

```elixir
@type prune_result() :: %{
  deleted_runs: non_neg_integer(),
  deleted_step_states: non_neg_integer(),
  deleted_step_tasks: non_neg_integer(),
  deleted_workers: non_neg_integer()
}
```

Result of pruning old run data.

# `archive_messages`

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

# `compile_flow`

```elixir
@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

# `complete_task`

```elixir
@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

# `delay_run`

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

# `delete_flow`

```elixir
@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

# `delete_message`

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

Deletes a message from a PGMQ queue.

# `fail_task`

```elixir
@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

# `flow_exists?`

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

Checks if a flow exists in the database.

# `get_flow_input`

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

Retrieves the input data for a flow run.

# `get_run`

```elixir
@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

# `get_step_output`

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

# `list_skipped_steps`

```elixir
@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

# `orphaned_queue_messages`

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

# `prune_data`

```elixir
@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)

# `read`

```elixir
@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

# `recover_stalled_tasks`

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

# `start_flow`

```elixir
@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

# `start_flow_with_run`

```elixir
@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

# `start_tasks`

```elixir
@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

# `upsert_flow`

```elixir
@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

# `valid_slug?`

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

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

---

*Consult [api-reference.md](api-reference.md) for complete listing*