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, handle_change/2, which is invoked once per changed document, in order, and one extra way of returning from 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 init/1 and 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 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 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 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 handle_error/2.

Errors

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