# `AgentDb.JobQueue`
[🔗](https://github.com/manhvu/agent_db/blob/v0.1.0/lib/agent_db/job_queue.ex#L1)

The durable job queue: the SQL behind `AgentDb.Core.Storage`'s work
operations.

Every mutation runs on the single writer connection, so a claim is a
serialized read-then-update rather than a race. Jobs survive a restart
because they are rows, and a job left `running` by a process that died is
returned to `pending` on the next boot.

# `conn`

```elixir
@type conn() :: Exqlite.Sqlite3.db()
```

# `job`

```elixir
@type job() :: %{
  id: integer(),
  kind: kind(),
  payload: payload(),
  attempts: integer(),
  max_attempts: integer()
}
```

# `kind`

```elixir
@type kind() :: :embed | :summarize_abstract | :summarize_overview
```

# `payload`

```elixir
@type payload() :: map()
```

# `status`

```elixir
@type status() :: :pending | :running | :done | :failed
```

# `all_kinds`

```elixir
@spec all_kinds() :: [kind()]
```

# `cancel_for_uri`

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

# `cancel_jobs`

```elixir
@spec cancel_jobs(String.t()) :: :ok | {:error, term()}
```

# `complete`

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

Marks a job as completed successfully.

# `complete`

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

# `count_for_kind`

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

# `count_for_uri`

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

# `defer`

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

# `dequeue`

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

# `detail`

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

# `enqueue`

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

# `enqueue`

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

# `enqueue_many`

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

# `fail`

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

# `fail`

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

# `requeue`

```elixir
@spec requeue(conn(), integer(), non_neg_integer()) :: :ok | {:error, term()}
```

# `reset_running_jobs`

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

Resets jobs left running by a previous run to pending.

# `reset_running_jobs`

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

# `stats`

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

Returns a map of job counts by status.

---

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