# `Ankusa.Intake`
[🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/intake.ex#L1)

The intake log behind `wal: :dual` (ADR 0002): a replicated log an acked hook
is also confirmed in, so a lost host does not lose it.

An adapter implements this behaviour (`Ankusa.Intake.Memory` is the test
double; `Ankusa.Intake.Kafka` ships in `ankusa_kafka`). Everything else is
core: the writer's dual commit (`Ankusa.Queue.Writer`), the pending set and
breaker (`Ankusa.Intake.Pending`, `Ankusa.Intake.Gate`), the live reconciler
that carries a half-persisted hook to completion
(`Ankusa.Intake.Reconciler`), and boot recovery (`Ankusa.Intake.Recovery`).

A record is `{seq, bin, meta}`: the hook's `hooks` value, written once under
the writer's pre-assigned seq. "Confirmed" is the adapter's promise that the
log's replicas hold it (Kafka: `acks=all` with `min.insync.replicas`); the
adapter makes one attempt per `c:publish_batch/3` call and the reconciler owns
every retry. Nothing here assumes exactly-once publishing: every reader
applies a record through the seq-and-id rule of `Ankusa.Intake.Recovery`, so a
duplicate or reordered record changes nothing.

An adapter's failure is one of the four classes of ground rule 2
(`:not_found | :unavailable | :throttled | :permanent`), with an optional
`retry_after` in milliseconds: `{:throttled, 500}`.

# `class`

```elixir
@type class() :: :not_found | :unavailable | :throttled | :permanent
```

# `failure`

```elixir
@type failure() :: class() | {class(), pos_integer()}
```

# `log_record`

```elixir
@type log_record() :: {seq :: non_neg_integer(), bin :: binary(), meta()}
```

# `meta`

```elixir
@type meta() :: %{
  id: String.t(),
  tenant: String.t(),
  source: String.t(),
  incarnation: String.t()
}
```

# `replayed`

```elixir
@type replayed() :: %{
  position: non_neg_integer(),
  node: String.t(),
  seq: non_neg_integer(),
  id: String.t(),
  kind: String.t(),
  version: String.t(),
  bin: binary()
}
```

# `assign_partition`
*optional* 

```elixir
@callback assign_partition(instance :: atom(), node_id :: String.t()) ::
  {:ok, non_neg_integer()} | :none
```

The partition this node writes, chosen once at its first dual boot.

# `child_spec`

```elixir
@callback child_spec(keyword()) :: Supervisor.child_spec()
```

The adapter's own processes (its client), started before the store.

# `describe`

```elixir
@callback describe(instance :: atom()) :: String.t()
```

The settings the startup sentence quotes, e.g. `acks=all, min.insync.replicas=2, replicas=3`.

# `earliest`

```elixir
@callback earliest(instance :: atom()) ::
  {:ok, non_neg_integer() | nil} | {:error, failure()}
```

The earliest position the log still holds, `nil` for an empty log.

# `publish_batch`

```elixir
@callback publish_batch(instance :: atom(), [log_record()], opts :: keyword()) ::
  {:ok, %{position: non_neg_integer()}} | {:error, failure()}
```

Blocks until the batch is confirmed or `opts[:timeout]` runs out. `position`
is the log position of the batch's last record. `opts` also carries `:node`,
`:topic` and `:partition` (the node's stamped partition, or `nil`).

# `ready?`

```elixir
@callback ready?(instance :: atom()) :: boolean()
```

A cheap local answer (client up, leader known); never a produce.

# `replay`

```elixir
@callback replay(
  instance :: atom(),
  from :: non_neg_integer() | :earliest,
  (replayed() -> :ok | {:error, term()}),
  opts :: keyword()
) :: {:ok, non_neg_integer() | nil} | {:error, failure()}
```

Calls `fun` for every record of this node (`opts[:node]`) from position `from`,
in log order. The adapter skips records of other nodes and of a kind or
version it does not know before calling `fun`, and reports each skipped
unknown one with `skipped/1`. Returns the position of the last record passed.

# `verify`

```elixir
@callback verify(instance :: atom()) :: :ok | {:error, term()}
```

Boot check of the log's topology, retention and record size.

# `assign_partition`

```elixir
@spec assign_partition(atom()) :: {:ok, non_neg_integer()} | :none
```

Asks the adapter for this node's partition; `:none` when it has no such notion.

# `describe`

```elixir
@spec describe(atom()) :: String.t()
```

The settings the adapter quotes in the startup sentence.

# `durability`

```elixir
@spec durability(Ankusa.Config.t()) :: String.t()
```

The startup sentence of `wal: :dual` (ADR 0002, section 1). `describe/1` of the
adapter supplies the verified settings; with `intake.verify_topic: false` the
sentence says they are unverified instead.

# `earliest`

```elixir
@spec earliest(atom()) :: {:ok, non_neg_integer() | nil} | {:error, class()}
```

The earliest position the log holds, `nil` when empty.

# `enabled?`

```elixir
@spec enabled?(Ankusa.Config.t()) :: boolean()
```

Whether the config runs the dual commit.

# `info`

```elixir
@spec info(atom()) :: map()
```

The `intake` object of `GET /v1/wal`.

# `known?`

```elixir
@spec known?(String.t(), String.t()) :: boolean()
```

Whether an adapter should pass a record of this kind and version to the replay function.

# `module`

```elixir
@spec module(Ankusa.Config.t()) :: module()
```

The adapter module `intake.type` names.

# `publish_batch`

```elixir
@spec publish_batch(atom(), [log_record()]) ::
  {:ok, non_neg_integer()} | {:error, class(), pos_integer() | nil}
```

One publish attempt with the configured deadline, measured. Never raises:
`{:ok, position}` or `{:error, class, retry_after_ms | nil}`.

# `ready?`

```elixir
@spec ready?(atom()) :: boolean()
```

The adapter's cheap readiness answer; `false` if it cannot answer.

# `replay`

```elixir
@spec replay(atom(), non_neg_integer() | :earliest, (replayed() ->
                                                 :ok | {:error, term()})) ::
  {:ok, non_neg_integer() | nil} | {:error, class()}
```

Replays this node's records from `from` (see `c:replay/4`).

# `skipped`

```elixir
@spec skipped(atom()) :: :ok
```

A record of a kind or version this release does not know (adapters call it per skip).

# `state`

```elixir
@spec state(atom()) :: :up | :down | :recovering
```

`:recovering` while boot replay runs, `:down` while the breaker is open and the
adapter still says it is not ready (the readiness probe drains the node), `:up`
otherwise: with the breaker open but the adapter ready again, the node is
ready for the one batch that probes it.

# `topic`

```elixir
@spec topic(Ankusa.Config.t()) :: String.t()
```

The log's name: `intake.topic`, else `ankusa.intake.<fleet>`.

# `validate_config!`

```elixir
@spec validate_config!(Ankusa.Config.t()) :: :ok
```

Boot check: the adapter's package is on the code path.

# `verify`

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

The adapter's boot verification.

---

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