event_consumer.ex

defmodule UpaEventSourcing.VideoWatchProgress.EventAggregator do
  def start_link(event) do
     Task.start_link(fn ->
       Upa.VideoWatchProgress.save(event) #Saves to the database
     end)
  end
end

defmodule UpaEventSourcing.VideoWatchProgress.EventConsumer do
  use ConsumerSupervisor
  
  alias UpaEventSourcing.VideoWatchProgress.EventStore
  alias UpaEventSourcing.VideoWatchProgress.EventAggregator

  def start_link(%{producer: EventStore, processor: EventAggregator}) do
    children = [
      worker(processor, [], restart: :temporary)
    ]

    ConsumerSupervisor.start_link(children,
      strategy: :one_for_one,
      subscribe_to: [{
        producer,
        min_demand: 1,
        max_demand: 192,
      }])
  end
end