defmodule Drinkup.Jetstream.Event do @moduledoc """ Event handling and dispatch for Jetstream events. Parses incoming JSON events from Jetstream and dispatches them to the configured consumer via Task.Supervisor. """ require Logger alias Drinkup.Jetstream.{Event, Options} @type t() :: Event.Commit.t() | Event.Identity.t() | Event.Account.t() @doc """ Parse a JSON map into an event struct. Jetstream events have a top-level structure with a "kind" field that determines the event type, and a nested object with the event data. ## Example Event Structure %{ "did" => "did:plc:...", "time_us" => 1726880765818347, "kind" => "commit", "commit" => %{...} } Returns the appropriate event struct based on the "kind" field, or `nil` if the event type is not recognized. """ @spec from(map()) :: t() | nil def from(%{"did" => did, "time_us" => time_us, "kind" => kind} = payload) do case kind do "commit" -> case Map.get(payload, "commit") do nil -> Logger.warning("Commit event missing 'commit' field: #{inspect(payload)}") nil commit -> Event.Commit.from(did, time_us, commit) end "identity" -> case Map.get(payload, "identity") do nil -> Logger.warning("Identity event missing 'identity' field: #{inspect(payload)}") nil identity -> Event.Identity.from(did, time_us, identity) end "account" -> case Map.get(payload, "account") do nil -> Logger.warning("Account event missing 'account' field: #{inspect(payload)}") nil account -> Event.Account.from(did, time_us, account) end _ -> Logger.warning("Received unrecognized event kind from Jetstream: #{inspect(kind)}") nil end end def from(payload) do Logger.warning("Received invalid event structure from Jetstream: #{inspect(payload)}") nil end @doc """ Dispatch an event to the consumer via Task.Supervisor. Spawns a task that processes the event via the consumer's `handle_event/1` callback. Unlike Tap, Jetstream does not require acknowledgments. """ @spec dispatch(t(), Options.t()) :: :ok def dispatch(event, %Options{consumer: consumer, name: name}) do supervisor_name = {:via, Registry, {Drinkup.Registry, {name, JetstreamTasks}}} {:ok, _pid} = Task.Supervisor.start_child(supervisor_name, fn -> try do consumer.handle_event(event) rescue e -> Logger.error( "Error in Jetstream event handler: #{Exception.format(:error, e, __STACKTRACE__)}" ) end end) :ok end end