Typed operational reads and cleanup for persisted PgFlow workers.
Every function receives the consumer repository explicitly. Worker summaries
combine a persisted worker with its current health and queue load, while task
reads return complete PgFlow.Schema.StepTask records.
Summary
Functions
Gets the adjacent worker UUID in newest-heartbeat-first navigation order.
Counts persisted workers matching the flow and health filters.
Deletes a persisted worker without deleting its historical task or run rows.
Gets a worker with its calculated health and queue load.
Lists worker summaries in newest-heartbeat-first order.
Lists complete persisted task rows last owned by a worker.
Functions
@spec adjacent(module(), Ecto.UUID.t(), :next | :prev) :: {:ok, Ecto.UUID.t()} | {:error, :invalid_id | :invalid_direction | :not_found}
Gets the adjacent worker UUID in newest-heartbeat-first navigation order.
:next selects the next older worker and :prev selects the previous newer
worker. Timestamp ties are resolved by UUID.
@spec count( module(), keyword() ) :: {:ok, non_neg_integer()}
Counts persisted workers matching the flow and health filters.
Pagination options do not affect the count.
@spec delete(module(), Ecto.UUID.t()) :: :ok | {:error, :invalid_id | term()}
Deletes a persisted worker without deleting its historical task or run rows.
PostgreSQL clears StepTask.last_worker_id through the table's
ON DELETE SET NULL foreign key, so the task remains without worker
attribution.
Deleting an absent worker is a successful no-op.
@spec get(module(), Ecto.UUID.t()) :: {:ok, PgFlow.WorkerSummary.t()} | {:error, :invalid_id | :not_found}
Gets a worker with its calculated health and queue load.
@spec list( module(), keyword() ) :: {:ok, [PgFlow.WorkerSummary.t()]} | {:error, :invalid_id}
Lists worker summaries in newest-heartbeat-first order.
Supported options are :flow_slug, :health_status, :cursor, and :limit.
Health is healthy within 30 seconds, stale through 60 seconds, and dead
after 60 seconds. Stopped or deprecated workers are always dead. Queue-load
aggregates are calculated only after pagination.
@spec list_tasks(module(), Ecto.UUID.t(), keyword()) :: {:ok, [PgFlow.Schema.StepTask.t()]} | {:error, :invalid_id}
Lists complete persisted task rows last owned by a worker.