derive

# derive
A small, simple, and yet flexible API for deriving state from an event source, that works out of the box with [Ecto](https://hexdocs.pm/ecto).
It comes ships with all the essential side-effects:
- insert
- update
- merge
- upsert
- delete
If you need more exotic / complex side-effects, you can easily create your own by implementing the `Derive.SideEffect` protocol for your custom side-effect structs.
## show me the code
Here's the most basic example I can thnk of:
```elixir
# priv/repo/migrations/20260127044710_create_derive_tables.exs
defmodule Dummy.Repo.Migrations.CreateDerivedTables do
  use Derive.Migration
  def up do
    Derive.Migration.up(1)
  end
  def down do
    Derive.Migration.down(1)
  end
end
# lib/dummy.ex
defmodule Dummy.Application do
  use Application
  def start(_, _) do
    children = [
      Dummy.Repo,
      Dummy.Consumer # {Dummy.Consumer, filters: [customer_id: 167810624]}
    ]
    Supervisor.start_link(children, strategy: :one_for_one, name: Dummy.Supervisor)
  end
end
# lib/dummy/consumer.ex
defmodule Dummy.Consumer do
  use Derive, otp_app: :dummy
  @impl Derive
  def fetch_events(repo, filters) do
    filters
    |> Dummy.Inbox.query()
    |> repo.all()
    |> Enum.map(&Dummy.Inbox.to_event/1)
  end
  @impl Derive
  def handle_event(%LoginFailed{timestamp: ts} = event) do
    [
      insert(%Dummy.SecurityIncident{
        user_id: event.user_id,
        type: :login_failed,
        occurrencies: 1,
        last_occurred_at: ts
      })
      |> on_conflict(inc: [ocurrencies: 1], set: [last_occurred_at: ts])
    ]
  end
  def handle_event(_), do: :skip
end
```
And here's an example with "director's commentary on":
```elixir
# priv/repo/migrations/20260127044710_create_derive_tables.exs
defmodule Dummy.Repo.Migrations.CreateDerivedTables do
  use Derive.Migration
  # ↑ - be sure to use derive's migration module instead of
  #     Ecto.Migration
  #   - turns out it takes a lot of code to provide an easy-to-use
  #     migration API - I was lazy, so I just put together a couple
  #     macros
  #   - the options were that you either would have to:
  #     a) add a `require Derive.Migration` to the migration file; or
  #     b) `use Derive.Migration`
  #     the latter felt less awkward, just personal preference
  #   - if you don't want to / can't do `use Derive.Migration` that's
  #     ok, just do `require Derive.Migration` instead so that you can
  #     use the `change/1` macro
  def up do
    Derive.Migration.up(1)
  end
  def down do
    Derive.Migration.down(1)
  end
end
# lib/dummy.ex
defmodule Dummy.Application do
  use Application
  def start(_, _) do
    children = [
      Dummy.Repo,
      # ↓ simplest setup
      Dummy.Consumer,
      #   - alternatively you can provide filters to produce different
      #     event streams for the same consumer module
      #   - this is particularly useful when you want to parallelize
      #     processing of events by a sharding key, like :customer_id
      #   - you could have a Supervisor dedicated to managing a bunch
      #     of consumers, or even a DynamicSupervisor if you need to
      # ↓   spawn consumers on demand
      {Dummy.Consumer, filters: [customer_id: 167810624]}
    ]
    Supervisor.start_link(children, strategy: :one_for_one, name: Dummy.Supervisor)
  end
end
# lib/dummy/consumer.ex
defmodule Dummy.Consumer do
  use Derive,
    # ↓ derive figures out the ecto repo from your configs
    otp_app: :dummy
  @impl Derive
  #   - fetches the next batch of events
  #   - store events however you want, wherever you want - it doesn't
  # ↓   have to be in an ecto repo, could be an API call for example
  def fetch_events(repo, filters) do
    filters
    # ↑ - filters is a keyword list containing the filters you might
    #     have provided in the Supervisor children
    #   - derive only makes two assumptions about your application,
    #     one of them is that your event's :id is an integer sequence
    #     in chronological order - that's a big assumption, I know -,
    #     that's because derive uses :id as a cursor - I plan on
    # ↓   supporting sort-safe strings at some point
    |> Dummy.Inbox.query()
    # ↑ - besides the filters, there're two keys your `query/1`
    #     function must handle in this example - that's the second
    #     assumption:
    #     1. :after - the last processed event's :id; and
    #     2. :take - the maximum number of events to fetch
    #     you can override the default :batch_size either here or by
    #     providing :batch_size in the Supervisor children along with
    #     any filters you might have
    #   - that's the only time your code is exposed directly to the
    #     consumer's cursor, that' was unavoidable - for the resto of
    #     the time derive manages it the hood for you under
    |> repo.all()
    #   - best place to transform / normalize / filter out your events
    # ↓   before they hit `handle_event/1`
    |> Enum.map(&Dummy.Inbox.to_event/1)
  end
  @impl Derive
  def handle_event(%LoginFailed{timestamp: ts} = event) do
    [
      #   derive ships with essential side-effects, such as:
      #   - insert (can tweak options :conflict_target and :on_conflict
      #     to get the upsert behaviour);
      #   - update;
      #   - merge;
      #   - upsert (alias to insert); and
      #   - delete
      #   you can easily create your own side-effects by implementing
      #   the `Derive.SideEffect` protocol for your custom side-effect
      # ↓ structs
      insert(%Dummy.SecurityIncident{
        user_id: event.user_id,
        type: :login_failed,
        occurrencies: 1,
        last_occurred_at: ts
      })
      |> on_conflict(inc: [ocurrencies: 1], set: [last_occurred_at: ts])
    ]
  end
  def handle_event(_), do: []
  # ↑ - return an empty list to ignore the event
  # ↓ - returning :skip works too, whatever feels more natural to you
  def handle_event(_), do: :skip
  @impl Derive
  #   this is an optional callback, when implemented it allows you to
  # ↓ override the persistence logic
  def persist(repo, side_effects, multi) do
    multi = into_multi(side_effects, multi)
    # ↑ - by default, derive accumulates the side-effects of a batch
    #     of events into a multi and then persists them in a single
    #     transaction, but you can optionally implement your own
    #     persistence logic in this function
    #   - you can choose how much you want to override, if you just
    #     want to override like the ecto repo to persist to, you don't
    #     have to re-implement the whole thing - the logic for
    #     accumulating side-effects into a multi is available via the
    #     `into_multi/2` function
    #   - beware though that the multi received by this function isn't
    #     empty, it already includes cursor operations, meaning you
    #     need to transact it regardless of your strategy for state
    #     persistence
    case custom_logic(repo, multi) do
      {:ok, _} -> :ok
      # ↑ - by returning `:ok` derive assumes it can move on to the
      #     next batch of events
      {:error, _, reason, _} -> {:error, reason}
      # ↑ - errors will bubble up to some function that will log and
      #     handle it appropriately
      #   - you can find a copy of the error in the consumer's cursor
      #     record in the database, along with the :stuck_since
      #     timestamp
    end
  end
end
```
As promised, the API's surface is small and designed to both:
* **just work** for the intended use case where we source events and persist state changes to an ecto repo;
* be **easily extensible / overridable** for more exotic / complex use-cases.
## scope
* **In Scope** - all the code needed to reduce events into persisted state changes, including:
  * consuming an event source / forming an event streams, easy to override it with your own sourcing logic to supports sources other than an Ecto table
  * essetial side-effects, easy to create your own
  * persisting state changes to an Ecto repo, easy to override it with your own persistence logic
  * parallelizing event processing
  * tracking processing progress via cursors
  * persisting state changes
* **Out of Scope**, as in I have no plans at all - everything related to producing, persisting, broadcasting or synchronizing events.
* **Unclear Scope** - honestly, I'll only pay attention to these if I personally need them:
  * event versioning
  * event schemas / validation
  * event lifecycle hooks (e.g.: processing, succeeded, failed, etc)