Documentation improvements

Author: v0idpwn

README.md

@@ -26,58 +26,4 @@ end
 
 ## Getting started
 
-First, you need to migrate your database to create the outbox. Create a
-migration with Ecto and add the following:
-
-```elixir
-defmodule MyApp.Repo.Migrations.AddKafkaesque
-  def up do
-    Kafkaesque.Migrations.up()
-  end
-
-  def down do
-    Kafkaesque.Migrations.down()
-  end
-end
-```
-
-Then, you must define a publisher module, associated with the Repo you
-ran the Kafkaesque migrations on. For example:
-
-```elixir
-defmodule MyApp.KafkaPublisher do
-  use Kafkaesque, repo: MyApp.Repo
-
-  # Optional, sets the partition for the message. Should return an integer
-  # Defaults to returning 0.
-  def partition(_topic, _body) do
-    Enum.random(0..1)
-  end
-
-  # Optional, encodes the body of the message, should return a String.
-  # Defaults to the identity function.
-  def encode(body) do
-    Jason.encode!(body)
-  end
-end
-```
-
-Finally, you should start a `Kafkaesque` instance in your supervision tree.
-For example:
-
-```elixir
-defmodule MyApp.Application do
-  def start(_type, _args) do
-    children = [
-      MyApp.Repo,
-      {Kafkaesque, [repo: MyApp.Repo, client_opts: [client_id: :my_client, brokers: [{"localhost, 9092"}]]]},
-    ]
-  end
-end
-```
-
-Done! Now you can publish messages. Try it out with:
-
-```elixir
-MyApp.KafkaPublisher.publish("topic", %{hello: :kafka})`
-```
+Check the [Getting started](https://hexdocs.pm/kafkaesque/getting-started.html) guide in Hexdocs.

guides/getting-started.md

@@ -0,0 +1,57 @@
+# Getting started
+
+First, you need to migrate your database to create the outbox. Create a
+migration with Ecto and add the following:
+
+```elixir
+defmodule MyApp.Repo.Migrations.AddKafkaesque
+  def up do
+    Kafkaesque.Migrations.up()
+  end
+
+  def down do
+    Kafkaesque.Migrations.down()
+  end
+end
+```
+
+Then, you must define an outbox module, associated with the Repo you
+ran the Kafkaesque migrations on. For example:
+
+```elixir
+defmodule MyApp.Outbox do
+  use Kafkaesque, repo: MyApp.Repo
+
+  # Optional, sets the partition for the message. Should return an integer
+  # Defaults to returning 0.
+  def partition(_topic, _body) do
+    Enum.random(0..1)
+  end
+
+  # Optional, encodes the body of the message, should return a String.
+  # Defaults to the identity function.
+  def encode(body) do
+    Jason.encode!(body)
+  end
+end
+```
+
+Finally, you should start a `Kafkaesque` instance in your supervision tree.
+For example:
+
+```elixir
+defmodule MyApp.Application do
+  def start(_type, _args) do
+    children = [
+      MyApp.Repo,
+      {MyApp.Outbox, [client_opts: [client_id: :my_client, brokers: [{"localhost, 9092"}]]]},
+    ]
+  end
+end
+```
+
+Done! Now you can publish messages. Try it out with:
+
+```elixir
+MyApp.Outbox.publish("topic", %{hello: :kafka})`
+```

lib/kafkaesque.ex

@@ -1,33 +1,31 @@
 defmodule Kafkaesque do
   @moduledoc """
-  This module provides a message publishing API through `Kafkaesque.publish/3`.
+  This module provides the main APIs for Kafkaesque
 
-  It also allows you to `use` it and have a streamlined publishing function,
-  possibly providing encodin function:
+  ### Introduction
 
+  Kafkaesque is an outbox library built for PostgreSQL and designed, primarily,
+  for Kafka, though usage with other software is possible and encouraged.
 
-      defmodule MyApp.Kafka do
-        use Kafkaesque, repo: MyApp.Repo
+  - Transactional safety for messages: if they were created, they **will** be
+  eventually published. They're only created if the transaction commits.
+  - Ordering: messages are published sequentially for topic/partition combinations
+  - Shutdown safety: has graceful shutdown and rescue for cases where it doesn't
+  happen.
+  - Observability: all operations publish telemetry events.
+  - Garbage collection: outbox table is periodically cleaned.
+  - Multi-node safe: safety powered by PostgreSQL.
 
-        # Optional, defaults to `body`
-        def encode(body) do
-          Jason.encode!(body)
-        end
+  For a comprehensive installation guide, check the [Getting started]("getting-started.md")
+  guide.
 
-        def partition(_topic, %MyMessage{id: id}) do
-          id
-        end
-      end
-
-      MyApp.Kafka.publish("my_topic", %{hello: :kafka})
-
-  `encode/1` should return a string, and defaults to the identity function.
+  ![Basic diagram](http://www.plantuml.com/plantuml/proxy?cache=no&src=https://raw.githubusercontent.com/v0idpwn/kafkaesque/master/diagrams/basic.iuml)
 
-  `partition/2` should return an integer, and defaults to 0. It will be used
-  as the partition for the message.
+  #### Note
 
-  See the documentation for `Kafkaesque.start_link/1` to learn about starting
-  Kafkaesque in your application.
+  For most cases, this module shouldn't be called directly. Instead, you want to
+  `use` it to define an outbox for your application, and call the outbox, as you
+  do with Ecto repos.
   """
 
   alias Kafkaesque.Message
@@ -60,7 +58,9 @@ defmodule Kafkaesque do
   end
 
   @doc """
-  Starts a Kafkaesque instance. Accepts the following opts:
+  Starts a Kafkaesque instance.
+
+  Accepts the following opts:
 
   - `:repo`: the repo where messages will be read from. Usually should be the
   same repo that you're writing to.
@@ -89,11 +89,16 @@ defmodule Kafkaesque do
   the table. Defaults to 72 hours.
   - `query_opts`: Options to pass to Ecto queries. Defaults to [log: false]
   """
+  @spec start_link(Keyword.t()) :: {:ok, pid()}
   def start_link(opts) do
     opts = Keyword.validate!(opts, [:repo] ++ @default_opts)
     Kafkaesque.Supervisor.start_link(opts)
   end
 
+  @doc """
+  Child spec for a Kafkaesque instance
+  """
+  @spec child_spec(Keyword.t()) :: Supervisor.child_spec()
   def child_spec(opts) do
     opts = Keyword.validate!(opts, [:repo] ++ @default_opts)
     Kafkaesque.Supervisor.child_spec(opts)
@@ -120,6 +125,18 @@ defmodule Kafkaesque do
         0
       end
 
+      @spec start_link(Keyword.t()) :: {:ok, pid}
+      def start_link(opts) do
+        opts = Keyword.merge(unquote(opts), opts)
+        Kafkaesque.start_link(opts)
+      end
+
+      @spec child_spec(Keyword.t()) :: Supervisor.child_spec()
+      def child_spec(opts) do
+        opts = Keyword.merge(unquote(opts), opts)
+        Kafkaesque.child_spec(opts)
+      end
+
       defoverridable encode: 1, partition: 2
     end
   end

lib/kafkaesque/kafka_client.ex

@@ -1,4 +1,8 @@
 defmodule Kafkaesque.Client do
+  @moduledoc """
+  Behaviour for clients
+  """
+
   alias Kafkaesque.Message
 
   @type client() :: term()

lib/kafkaesque/query.ex

@@ -34,30 +34,33 @@ defmodule Kafkaesque.Query do
       |> order_by([m], m.id)
       |> limit(^demand)
 
-    repo.transaction(fn ->
-      repo.query!("SELECT pg_advisory_xact_lock($1)", [@xact_lock_key], query_opts)
-
-      Message
-      |> where([m], m.id in subquery(subset))
-      |> select([m, _], m)
-      |> update([m],
-        set: [
-          state: :publishing,
-          attempted_at: fragment("CURRENT_TIMESTAMP"),
-          attempted_by: ^inspect(node())
-        ],
-        inc: [attempt: 1]
-      )
-      |> repo.update_all([], query_opts)
-      |> case do
-        {0, nil} ->
-          {0, []}
-
-        {count, messages} ->
-          sorted = Enum.sort(messages, fn m1, m2 -> m1.id <= m2.id end)
-          {count, sorted}
-      end
-    end, query_opts)
+    repo.transaction(
+      fn ->
+        repo.query!("SELECT pg_advisory_xact_lock($1)", [@xact_lock_key], query_opts)
+
+        Message
+        |> where([m], m.id in subquery(subset))
+        |> select([m, _], m)
+        |> update([m],
+          set: [
+            state: :publishing,
+            attempted_at: fragment("CURRENT_TIMESTAMP"),
+            attempted_by: ^inspect(node())
+          ],
+          inc: [attempt: 1]
+        )
+        |> repo.update_all([], query_opts)
+        |> case do
+          {0, nil} ->
+            {0, []}
+
+          {count, messages} ->
+            sorted = Enum.sort(messages, fn m1, m2 -> m1.id <= m2.id end)
+            {count, sorted}
+        end
+      end,
+      query_opts
+    )
   end
 
   @spec update_success_batch(Ecto.Repo.t(), [pos_integer()], Keyword.t()) :: :ok

lib/kafkaesque/rescuer.ex

@@ -32,7 +32,8 @@ defmodule Kafkaesque.Rescuer do
     interval_ms = Keyword.fetch!(opts, :rescuer_interval_ms)
     limit_ms = Keyword.fetch!(opts, :rescuer_limit_ms)
 
-    {:ok, %{repo: repo, query_opts: query_opts, interval_ms: interval_ms, limit_ms: limit_ms}, {:continue, :rescue}}
+    {:ok, %{repo: repo, query_opts: query_opts, interval_ms: interval_ms, limit_ms: limit_ms},
+     {:continue, :rescue}}
   end
 
   @impl GenServer

mix.exs

@@ -1,20 +1,21 @@
 defmodule Kafkaesque.MixProject do
   use Mix.Project
 
-  @version "0.0.1"
+  @version "1.0.0-rc.0"
 
   def project do
     [
       app: :kafkaesque,
       version: @version,
-      elixir: "~> 1.11",
+      elixir: "~> 1.13",
       start_permanent: Mix.env() == :prod,
       aliases: aliases(),
       elixirc_paths: elixirc_paths(Mix.env()),
       deps: deps(),
       description: description(),
       package: package(),
       name: "Kafkaesque",
+      docs: docs(),
       source_url: "https://github.com/v0idpwn/kafkaesque"
     ]
   end
@@ -39,6 +40,21 @@ defmodule Kafkaesque.MixProject do
     ]
   end
 
+  defp docs do
+    [
+      main: "Kafkaesque",
+      source_ref: "#v{@version}",
+      extra_section: "GUIDES",
+      extras: extras()
+    ]
+  end
+
+  defp extras do
+    [
+      "guides/getting-started.md"
+    ]
+  end
+
   defp description do
     "Transactional outbox for kafka"
   end

mix.lock

@@ -5,17 +5,18 @@
   "db_connection": {:hex, :db_connection, "2.4.3", "3b9aac9f27347ec65b271847e6baeb4443d8474289bd18c1d6f4de655b70c94d", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "c127c15b0fa6cfb32eed07465e05da6c815b032508d4ed7c116122871df73c12"},
   "decimal": {:hex, :decimal, "2.0.0", "a78296e617b0f5dd4c6caf57c714431347912ffb1d0842e998e9792b5642d697", [:mix], [], "hexpm", "34666e9c55dea81013e77d9d87370fe6cb6291d1ef32f46a1600230b1d44f577"},
   "dialyxir": {:hex, :dialyxir, "1.2.0", "58344b3e87c2e7095304c81a9ae65cb68b613e28340690dfe1a5597fd08dec37", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "61072136427a851674cab81762be4dbeae7679f85b1272b6d25c3a839aff8463"},
-  "earmark_parser": {:hex, :earmark_parser, "1.4.12", "b245e875ec0a311a342320da0551da407d9d2b65d98f7a9597ae078615af3449", [:mix], [], "hexpm", "711e2cc4d64abb7d566d43f54b78f7dc129308a63bc103fbd88550d2174b3160"},
+  "earmark_parser": {:hex, :earmark_parser, "1.4.31", "a93921cdc6b9b869f519213d5bc79d9e218ba768d7270d46fdcf1c01bacff9e2", [:mix], [], "hexpm", "317d367ee0335ef037a87e46c91a2269fef6306413f731e8ec11fc45a7efd059"},
   "ecto": {:hex, :ecto, "3.9.5", "9f0aa7ae44a1577b651c98791c6988cd1b69b21bc724e3fd67090b97f7604263", [:mix], [{:decimal, "~> 1.6 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "d4f3115d8cbacdc0bfa4b742865459fb1371d0715515842a1fb17fe31920b74c"},
   "ecto_sql": {:hex, :ecto_sql, "3.9.2", "34227501abe92dba10d9c3495ab6770e75e79b836d114c41108a4bf2ce200ad5", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.9.2", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.6.0", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.16.0 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "1eb5eeb4358fdbcd42eac11c1fbd87e3affd7904e639d77903c1358b2abd3f70"},
   "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"},
-  "ex_doc": {:hex, :ex_doc, "0.23.0", "a069bc9b0bf8efe323ecde8c0d62afc13d308b1fa3d228b65bca5cf8703a529d", [:mix], [{:earmark_parser, "~> 1.4.0", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}], "hexpm", "f5e2c4702468b2fd11b10d39416ddadd2fcdd173ba2a0285ebd92c39827a5a16"},
+  "ex_doc": {:hex, :ex_doc, "0.29.4", "6257ecbb20c7396b1fe5accd55b7b0d23f44b6aa18017b415cb4c2b91d997729", [:mix], [{:earmark_parser, "~> 1.4.31", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "2c6699a737ae46cb61e4ed012af931b57b699643b24dabe2400a8168414bc4f5"},
   "gen_stage": {:hex, :gen_stage, "1.2.1", "19d8b5e9a5996d813b8245338a28246307fd8b9c99d1237de199d21efc4c76a1", [:mix], [], "hexpm", "83e8be657fa05b992ffa6ac1e3af6d57aa50aace8f691fcf696ff02f8335b001"},
   "jason": {:hex, :jason, "1.4.0", "e855647bc964a44e2f67df589ccf49105ae039d4179db7f6271dfd3843dc27e6", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "79a3791085b2a0f743ca04cec0f7be26443738779d09302e01318f97bdb82121"},
   "kafka_protocol": {:hex, :kafka_protocol, "4.1.0", "53fac8866969484f783bff204bd4e41e62a97ce9753c83f802a08d5bfc0e0c4c", [:rebar3], [{:crc32cer, "0.1.8", [hex: :crc32cer, repo: "hexpm", optional: false]}], "hexpm", "61cb8b80199bf95122cf8073e0f4c0ad62f82515b4d44c54f946a5972c3f5fa5"},
-  "makeup": {:hex, :makeup, "1.0.5", "d5a830bc42c9800ce07dd97fa94669dfb93d3bf5fcf6ea7a0c67b2e0e4a7f26c", [:mix], [{:nimble_parsec, "~> 0.5 or ~> 1.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cfa158c02d3f5c0c665d0af11512fed3fba0144cf1aadee0f2ce17747fba2ca9"},
-  "makeup_elixir": {:hex, :makeup_elixir, "0.15.1", "b5888c880d17d1cc3e598f05cdb5b5a91b7b17ac4eaf5f297cb697663a1094dd", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.1", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "db68c173234b07ab2a07f645a5acdc117b9f99d69ebf521821d89690ae6c6ec8"},
-  "nimble_parsec": {:hex, :nimble_parsec, "1.1.0", "3a6fca1550363552e54c216debb6a9e95bd8d32348938e13de5eda962c0d7f89", [:mix], [], "hexpm", "08eb32d66b706e913ff748f11694b17981c0b04a33ef470e33e11b3d3ac8f54b"},
+  "makeup": {:hex, :makeup, "1.1.0", "6b67c8bc2882a6b6a445859952a602afc1a41c2e08379ca057c0f525366fc3ca", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "0a45ed501f4a8897f580eabf99a2e5234ea3e75a4373c8a52824f6e873be57a6"},
+  "makeup_elixir": {:hex, :makeup_elixir, "0.16.1", "cc9e3ca312f1cfeccc572b37a09980287e243648108384b97ff2b76e505c3555", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "e127a341ad1b209bd80f7bd1620a15693a9908ed780c3b763bccf7d200c767c6"},
+  "makeup_erlang": {:hex, :makeup_erlang, "0.1.1", "3fcb7f09eb9d98dc4d208f49cc955a34218fc41ff6b84df7c75b3e6e533cc65f", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "174d0809e98a4ef0b3309256cbf97101c6ec01c4ab0b23e926a9e17df2077cbb"},
+  "nimble_parsec": {:hex, :nimble_parsec, "1.3.0", "9e18a119d9efc3370a3ef2a937bf0b24c088d9c4bf0ba9d7c3751d49d347d035", [:mix], [], "hexpm", "7977f183127a7cbe9346981e2f480dc04c55ffddaef746bd58debd566070eef8"},
   "postgrex": {:hex, :postgrex, "0.16.5", "fcc4035cc90e23933c5d69a9cd686e329469446ef7abba2cf70f08e2c4b69810", [:mix], [{:connection, "~> 1.1", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "edead639dc6e882618c01d8fc891214c481ab9a3788dfe38dd5e37fd1d5fb2e8"},
   "snappyer": {:hex, :snappyer, "1.2.8", "201ce9067a33c71a6a5087c0c3a49a010b17112d461e6df696c722dcb6d0934a", [:rebar3], [], "hexpm", "35518e79a28548b56d8fd6aee2f565f12f51c2d3d053f9cfa817c83be88c4f3d"},
   "supervisor3": {:hex, :supervisor3, "1.1.11", "d81cdec31d102fde407423e1d05b569572850deebed86b951d5233c387cba80b", [:rebar3], [], "hexpm", "e6c2dedbcabcba24995a218aca12db5e208b80d3252692b22ef0f1a266104b50"},