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

Deletes every job whose payload targets uri or a descendant, in any state.

Takes a connection rather than acquiring the writer itself, because removal calls this from inside its own transaction on the writer connection. cancel_jobs/1 is the same deletion for a caller that holds no connection.

Matches the payload's uri field rather than the raw JSON text: a substring match would also delete jobs for unrelated URIs whose content happens to mention this one, silently dropping work outside the removed subtree.

@spec complete(integer()) :: :ok | {:error, term()}

Marks a job as completed successfully.

@spec count_for_kind(atom()) :: non_neg_integer()

How many jobs of kind exist, in any status.

A standing job -- one the store schedules itself rather than one a caller asked for -- has no URI to count by, so it is counted by kind. Used to keep a second boot from scheduling a second row of work the first one's deferral cannot cancel.

@spec count_for_uri(String.t(), [String.t()]) :: non_neg_integer()

How many jobs target uri or a descendant and are in one of statuses.

@spec defer(integer(), non_neg_integer()) :: :ok | {:error, term()}

Puts a job back without spending one of its attempts, for work that is waiting rather than failing.

dequeue/0 advances the attempt count when it claims a row, so a job merely waiting for a model would otherwise spend its budget on waiting.

@spec dequeue([kind()]) :: {:ok, job()} | {:error, :empty} | {:error, term()}

Claims the next runnable job of one of kinds and advances its attempt count.

A worker claims only the work it can run. Without the filter, a worker would claim a job belonging to another kind, be unable to interpret it, and fail it -- so idle work would be destroyed by the very pool meant to perform it.

Selection and claim share one writer call, so no two workers can take the same row. The changes() check is a belt-and-braces guard: if the update matched nothing, another writer got there first and the claim is retried.

@spec detail(pos_integer()) :: {:ok, map()} | {:error, term()}

How far behind the queue is and what is failing in it.

oldest_pending_ms is the age of the longest-waiting pending job -- the one number that says whether the workers are keeping up -- and nil when nothing is pending. failed is the newest failures first, bounded by limit, each carrying the classified reason that was persisted when the job gave up.

The URI comes out of the payload rather than a separate column, so it is read where it already is written.

@spec enqueue(kind(), payload()) :: {:ok, integer()} | {:error, term()}

Enqueues a new job.

enqueue/3 is the same insertion on a connection the caller already holds, for a caller that has to record the job in the same transaction as the state the job is about. enqueue/2 is that, on the writer.

@spec enqueue(conn(), kind(), payload()) :: {:ok, integer()} | {:error, term()}

Enqueues a new job on a connection the caller holds.

Takes the connection rather than acquiring the writer itself, because a replacement calls this from inside its own transaction on the writer connection, where a second claim on the writer would deadlock.

@spec enqueue_many(conn(), [{kind(), payload()}]) :: :ok | {:error, term()}

Enqueues several jobs with one INSERT on a connection the caller holds.

A write always enqueues its whole family (embed plus summaries) in one transaction, so one statement replaces one per kind. Row ids are not returned; callers that need them keep using enqueue/3.

@spec fail(integer(), String.t() | nil) :: :ok | {:error, term()}

Records a failure, scheduling a retry while attempts remain and marking the job failed once they are exhausted.

reason is the classified failure the worker saw. It is stored so a failed job can be explained later without waiting for the log line that recorded it to still exist, and it is a classification -- never content, a prompt, or a credential.

@spec reset_running_jobs() :: :ok | {:error, term()}

Resets jobs left running by a previous run to pending.

@spec stats() :: {:ok, map()} | {:error, term()}

Returns a map of job counts by status.