# `Sofa.Changes`
[🔗](https://github.com/skunkwerks/sofa/blob/0.5.1/lib/sofa/changes.ex#L1)

A behaviour for following the CouchDB `_changes` feed of a database.

> If the only tool you have is CouchDB, then
> everything looks like `{:ok, :relax}`

`Sofa.Changes` is a `GenServer` with one extra callback,
`c:handle_change/2`, which is invoked once per changed document, in
order, and one extra way of returning from `c:init/1`: alongside the
state it names the database to follow and, optionally, the feed options
described in `Sofa.Changes.Query`.

    defmodule Follower do
      use Sofa.Changes

      def start_link(db), do: Sofa.Changes.start_link(__MODULE__, db)

      @impl true
      def init(db), do: {:ok, db, [include_docs: true], nil}

      @impl true
      def handle_change(%{"id" => id, "deleted" => true}, state) do
        Logger.warning("deleted: #{id}")
        {:ok, state}
      end

      def handle_change(%{"doc" => %Sofa.Doc{id: id, rev: rev, body: body}}, state) do
        Logger.info("doc #{id} rev=#{rev} fields: #{inspect(Map.keys(body))}")
        {:ok, state}
      end

      def handle_change(change, state) do
        Logger.error("unhandled change: #{inspect(change)}")
        {:ok, state}
      end
    end

A ready-to-run script version of this example lives at
`examples/follower.exs`. It prints each document's `id`, `rev` and body
fields, and logs deletions without crashing.

The database is a `%Sofa{}` with its `database` set, as returned by
`Sofa.DB.open!/2`. The follower reuses its Req client and credentials.

`use Sofa.Changes` declares the behaviour and injects overridable
defaults for every callback, so only `c:init/1` and `c:handle_change/2`
usually need writing. The remaining `GenServer` callbacks (`handle_call`,
`handle_cast`, `handle_info`, `terminate`, `code_change`) are delegated to
the callback module with its own state threaded through, and
`c:handle_error/2` decides what to do when a request fails.

## Streaming and reconnecting

Each follower is one process with its own HTTP connection, run by a
linked `Sofa.Changes.Worker`. Chunks of the feed are delivered as
messages to the follower, so `c:handle_change/2` is single threaded and
ordered. Rows are passed through as decoded maps, see
`Sofa.Changes.Parser`, except that with `include_docs: true` the
embedded document becomes a `%Sofa.Doc{}` via `Sofa.Doc.from_map/1`,
with a deletion's tombstone carrying `"_deleted"` in its body.

The follower remembers the `seq` of every delivered row, and the
`last_seq` a response ends with. When the server ends the response, as it
does on its `timeout`, on `limit`, or after each longpoll batch, the
follower reconnects with `since` set to that sequence, after a random
delay of up to 500 ms so that many followers do not reconnect at once.
With `feed: :normal` the single batch is delivered and the follower then
stays idle, which suits a one-off catch-up.
Sequences are opaque and never parsed. Because shards can move, a change
may occasionally be delivered twice. CouchDB documents this, so
`c:handle_change/2` should be idempotent.

Should the server reject an advanced `since` with a 400, the follower
retries once from the last sequence the server did accept, the one it
sent on the previous successful request. If that is rejected too, or
there is no such sequence, the follower stops with reason `:bad_request`.

A watchdog restarts the connection when nothing, not even a heartbeat,
has arrived for twice the `heartbeat` interval, which catches half-open
TCP connections the server will never notice. Client-side receive
timeouts, dropped connections and a crashed connection process restart
the stream the same way, from the last sequence, without involving
`c:handle_error/2`.

## Errors

A response with a status other than 200, or a failed connection, is
reported to `c:handle_error/2` as one of `t:error/0`, after closing the
connection. The reply decides whether and when to reconnect.

# `change`

```elixir
@type change() :: %{required(String.t()) => term() | Sofa.Doc.t()}
```

A changed document, as decoded from the feed, see `t:Sofa.Changes.Parser.row/0`

# `error`

```elixir
@type error() ::
  :bad_request
  | :unauthorized
  | :forbidden
  | :not_found
  | :internal_server_error
  | {:status_code, pos_integer()}
  | {:conn_failed, term()}
```

Why a request failed, as given to `c:handle_error/2`

# `follower`

```elixir
@type follower() ::
  pid()
  | atom()
  | {:global, term()}
  | {:via, module(), term()}
  | {atom(), node()}
```

A follower, as accepted by `GenServer` functions

# `from`

```elixir
@type from() :: {pid(), term()}
```

The reply tag passed to `c:handle_call/3`

# `on_start`

```elixir
@type on_start() ::
  {:ok, pid()} | :ignore | {:error, {:already_started, pid()} | term()}
```

# `t`

```elixir
@type t() :: %Sofa.Changes{
  buffer: binary(),
  connected: boolean(),
  db: Sofa.t(),
  fallback: Sofa.Changes.Query.since(),
  lkg: Sofa.Changes.Query.since(),
  module: module(),
  mstate: term(),
  query: Sofa.Changes.Query.t(),
  ref: reference() | nil,
  retry: reference() | :infinity | nil,
  timeout: timeout() | :hibernate | nil,
  watchdog: reference() | nil,
  worker: pid() | nil,
  workers: %{optional(pid()) => true}
}
```

The follower's internal state, visible via `:sys.get_state/1`

# `code_change`

```elixir
@callback code_change(old_vsn, state :: term(), extra :: term()) ::
  {:ok, new_state :: term()} | {:error, reason :: term()}
when old_vsn: term() | {:down, term()}
```

# `handle_call`

```elixir
@callback handle_call(request :: term(), from(), state :: term()) ::
  {:reply, reply, new_state}
  | {:reply, reply, new_state, timeout() | :hibernate}
  | {:noreply, new_state}
  | {:noreply, new_state, timeout() | :hibernate}
  | {:stop, reason, reply, new_state}
  | {:stop, reason, new_state}
when reply: term(), new_state: term(), reason: term()
```

# `handle_cast`

```elixir
@callback handle_cast(request :: term(), state :: term()) ::
  {:noreply, new_state}
  | {:noreply, new_state, timeout() | :hibernate}
  | {:stop, reason :: term(), new_state}
when new_state: term()
```

# `handle_change`

```elixir
@callback handle_change(change(), state :: term()) ::
  {:ok, new_state}
  | {:ok, new_state, timeout() | :hibernate}
  | {:stop, reason :: term(), new_state}
when new_state: term()
```

Called once for every changed document, in the order received.

A deleted document arrives as a change whose `"deleted"` key is `true`.
With `include_docs: true` its embedded `"doc"` is a tombstone whose body
carries only `"_deleted" => true` and none of the document's former
fields. Callbacks should match on `"deleted"` (or on `"id"`) before
assuming a field such as a document body is present, otherwise a
deletion will raise a `FunctionClauseError` and stop the follower.

`{:stop, reason, state}` stops the follower and calls `c:terminate/2`.

# `handle_error`

```elixir
@callback handle_error(error(), state :: term()) ::
  {:retry, new_state}
  | {:retry, timeout(), new_state}
  | {:stop, reason :: term(), new_state}
when new_state: term()
```

Called when a request fails with a non-200 status or a connection error.

`{:retry, state}` reconnects right away, `{:retry, ms, state}` after
`ms` milliseconds, `{:retry, :infinity, state}` waits for `retry_now/1`,
and `{:stop, reason, state}` stops the follower. Every reconnect adds up
to 500 ms of jitter. The default waits 60 seconds, which also keeps a
follower with bad credentials from tripping CouchDB's account lockout.

# `handle_info`

```elixir
@callback handle_info(msg :: :timeout | term(), state :: term()) ::
  {:noreply, new_state}
  | {:noreply, new_state, timeout() | :hibernate}
  | {:stop, reason :: term(), new_state}
when new_state: term()
```

# `init`

```elixir
@callback init(args :: term()) ::
  {:ok, db :: Sofa.t(), state}
  | {:ok, db :: Sofa.t(), opts :: keyword(), state}
  | {:ok, db :: Sofa.t(), opts :: keyword(), state, timeout() | :hibernate}
  | :ignore
  | {:stop, reason :: term()}
when state: term()
```

Called when the follower starts, inside the `GenServer` init.

Returns the database to follow, optionally the feed options (see
`Sofa.Changes.Query`), the callback state, and optionally a `GenServer`
timeout. `:ignore` and `{:stop, reason}` behave as in `GenServer`.

The first request is only made after `c:init/1` returns, so starting
succeeds even when the server will refuse it. The refusal then reaches
`c:handle_error/2`.

# `terminate`

```elixir
@callback terminate(reason, state :: term()) :: term()
when reason: :normal | :shutdown | {:shutdown, term()} | term()
```

# `call`

```elixir
@spec call(follower(), request :: term(), timeout()) :: term()
```

Calls the follower, handled by its `c:handle_call/3`.

# `cast`

```elixir
@spec cast(follower(), request :: term()) :: :ok
```

Casts to the follower, handled by its `c:handle_cast/2`.

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `reply`

```elixir
@spec reply(from(), term()) :: :ok
```

Replies to a call, see `GenServer.reply/2`.

# `retry_now`

```elixir
@spec retry_now(follower()) :: :ok
```

Reconnects a follower that is waiting after `{:retry, :infinity, state}`
from `c:handle_error/2`. A no-op while a request is in flight.

# `start`

```elixir
@spec start(module(), term(), GenServer.options()) :: on_start()
```

Starts an unlinked follower, see `GenServer.start/3`.

# `start_link`

```elixir
@spec start_link(module(), term(), GenServer.options()) :: on_start()
```

Starts a follower linked to the current process, see `GenServer.start_link/3`.

# `stop`

```elixir
@spec stop(follower(), reason :: term(), timeout()) :: :ok
```

Stops a follower, closing its connection and calling `c:terminate/2`.

# `whereis`

```elixir
@spec whereis(follower()) :: pid() | {atom(), node()} | nil
```

Resolves a registered name to a pid, see `GenServer.whereis/1`.

---

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