defmodule Solvent do
  use TelemetryRegistry

  telemetry_event %{
    event: [:solvent, :event, :published],
    description: "Emitted when an event is published",
    measurements: "%{}",
    metadata: "%{event_source: String.t(), event_id: String.t(), event_type: String.t(), subscriber_count: non_neg_integer()}"
  }

  telemetry_event %{
    event: [:solvent, :subscriber, :processing, :start],
    description: "Emitted when a subscriber begins processing an event",
    measurements: "%{}",
    metadata: "%{subscriber_id: String.t(), event_source: String.t(), event_id: String.t(), event_type: String.t()}"
  }

  telemetry_event %{
    event: [:solvent, :subscriber, :processing, :stop],
    description: "Emitted when a subscriber finishes processing an event",
    measurements: "%{duration: non_neg_integer()}",
    metadata: "%{subscriber_id: String.t(), event_source: String.t(), event_id: String.t(), event_type: String.t()}"
  }

  telemetry_event %{
    event: [:solvent, :subscriber, :subscribing, :start],
    description: "Emitted when a subscriber begins subscribing to the event stream",
    measurements: "%{}",
    metadata: "%{subscriber_id: String.t(), filter: String.t()}"
  }

  telemetry_event %{
    event: [:solvent, :subscriber, :subscribing, :stop],
    description: "Emitted when a subscriber is finished subscribing to the event stream",
    measurements: "%{duration: non_neg_integer()}",
    metadata: "%{subscriber_id: String.t(), filter: String.t()}"
  }

  @moduledoc """
  Solvent is an event bus built to be fast and easy-to-use,
  and takes a lot of inspiration from the [CloudEvents](https://cloudevents.io) spec for the best interoperability with other event systems.

  In the CloudEvents specification, every event is required to have an ID, a source, and a type.
  The `source` field identifies the system that sent the event, and the ID must identify the event and be unique in the scope of the source.
  Lastly, the `type` field, also called a "topic," identifies the kind of event that took place.

  In Solvent, only the `type` field is required, but it is strongly recommended to provide a `source` field as well.
  Solvent generates a version 7 UUID for event ID fields and this rarely needs to be overridden.
  See the `Solvent.Event` docs for details on what defaults are provided.

  Subscribe to the event stream with `subscribe/2`, and publish events using `publish/2`.
  You can also create a `Solvent.Subscription` or a `Solvent.Event` yourself and pass those to `subscribe/1` and `publish/1`, respectively.
  The functions here share a signature with their corresponding struct functions, but have the additional benefit of interacting with the event bus.

  ## Telemetry

  #{telemetry_docs()}

  """

  alias Solvent.Subscription

  require Logger

  @doc """
  Subscribe to the event stream with a pre-made `Solvent.Subscription` struct.
  """
  def subscribe(%Subscription{} = sub) do
    :telemetry.span(
      [:solvent, :subscriber, :subscribing],
      %{subscriber_id: sub.id, filter: sub.filters},
      fn ->
        :ok = Solvent.SubscriberStore.insert(sub)
        {:ok, %{}}
      end
    )

    {:ok, sub.id}
  end

  @doc """
  Subscribe to the event stream.

  The sink is what receives your events, and can be anything that implements the `Solvent.Sink` protocol,
  which includes anonymous functions, module-function-args tuples, and PIDs.
  Here's an example with a module-function-args tuple that subscribes to a single event type
  and identifies the subscription as `"My first subscriber"`.

      iex> Solvent.subscribe(
      ...>  {Solvent.MessengerHandler, :handle_event, []},
      ...>  types: ["com.example.event.published"],
      ...>  id: "My first subscriber"
      ...> )
      {:ok, "My first subscriber"}

  This function shares its signature with `Solvent.Subscription.new/2`,
  and creates a `Solvent.Subscription` in the same way,
  while also inserting the subscription into the `Solvent.SubscriberStore` so that the sink will begin to receive events.

  Sinks are given an event _identifier_, and _not_ the event itself.
  Your sink must fetch the full event from the `Solvent.EventStore`, so that extra data is not copied between processes.
  You must also call `Solvent.EventStore.ack/2` once you are done with the event,
  unless you want the event to stay in the event store forever.

  > #### Tip {: .tip}
  >
  > Use the `Solvent.Subscriber` module to make a subscriber that automatically acknowledges events,
  > along with lots of other nice features.

  You can also create a `Solvent.Subscription` struct yourself, and pass it to `subscribe/1`.
  """
  def subscribe(sink, opts \\ [])

  def subscribe(module, opts) when is_atom(module) and is_list(opts) do
    subscription = apply(module, :subscription, [opts])

    subscribe(subscription)
  end

  def subscribe(sink, opts) do
    sub = Subscription.new(sink, opts)

    subscribe(sub)
  end

  @doc """
  Remove a subscriber.
  """
  def unsubscribe(id) do
    Solvent.SubscriberStore.delete(id)
  end

  @doc """
  Publish an event to the event bus, triggering all subscriber functions.

  Only a type (AKA "topic") is required. All other fields can be supplied using the options.
  See `Solvent.Event` for details on what that struct contains.
  All values given as options are inserted into the event struct.

  ID values are version 7 UUIDs by default, and you don't need to provide them.

  I would recommend using the CloudEvents format for your event's `type` field, which starts with a reversed DNS name, and is dot-separated.
  This will help avoid collisions with events from other applications.

      Solvent.publish("io.github.cantido.myevent.published")
      {:ok, "0b06bdb7-06a7-4df9-a825-1fd225ceea43"}

  It is also recommended to provide a `source` option to identify your application, especially if your subscription did not specify a `source`.
  Otherwise, all events will be published with a default source, and subscriptions will be triggered for every event traveling through Solvent (including events from other applications).
  A good `source` value is either a fully-qualified domain name, or a UUID in URN format.

      iex> Solvent.publish(
      ...>   "io.github.cantido.documentation.read",
      ...>   id: "read-docs-id",
      ...>   source: "myapp"
      ...> )
      {:ok, {"myapp", "read-docs-id"}}

  You can also build an event yourself with `Solvent.Event.new/1` and publish it with this function.
  """
  def publish(event, opts \\ [])

  def publish(type, opts) when is_binary(type) do
    event = Solvent.Event.new(type, opts)
    publish(event, opts)
  end

  def publish(event, _opts) do
    Task.Supervisor.start_child(Solvent.TaskSupervisor, fn ->
      subscribers = Solvent.SubscriberStore.listeners_for(event)
      subscriber_ids = Enum.map(subscribers, &elem(&1, 1)) |> Enum.uniq()

      Logger.debug("Publishing event #{event.id}, (#{event.type}). Subscribers are: #{inspect subscriber_ids, pretty: true}")

      :telemetry.execute(
        [:solvent, :event, :published],
        %{},
        %{event_source: event.source, event_id: event.id, event_type: event.type, subscriber_count: Enum.count(subscribers)}
      )

      if Enum.count(subscribers) > 0 do
          :ok = Solvent.EventStore.insert(event, subscriber_ids)

        notifier_fun = fn {subscriber_id, subscription} ->
          Task.Supervisor.start_child(Solvent.TaskSupervisor, fn ->
            :telemetry.span(
              [:solvent, :subscriber, :processing],
              %{subscriber_id: subscriber_id, event_source: event.source, event_id: event.id, event_type: event.type},
              fn ->
                Logger.metadata(
                  solvent_subscriber_id: subscriber_id,
                  solvent_event_source: event.source,
                  solvent_event_id: event.id,
                  solvent_event_type: event.type
                )
                Solvent.Sink.deliver(subscription.sink, event)
                {:ok, %{}}
              end
            )

            :ok
          end)
        end

        Task.Supervisor.async_stream(Solvent.TaskSupervisor, subscribers, notifier_fun,
          timeout: :infinity
        )
        |> Stream.run()
      else
        Logger.warn("No subscribers matched event type #{event.type}. Solvent will not insert the event into the event store.")
      end
    end)

    {:ok, {event.source, event.id}}
  end

  def build_filters(filters) when is_list(filters) do
    Enum.map(filters, &build_filter/1)
  end

  defp build_filter({:exact, props}), do: %Solvent.Filter.Exact{properties: props}
  defp build_filter({:prefix, props}), do: %Solvent.Filter.Prefix{properties: props}
  defp build_filter({:suffix, props}), do: %Solvent.Filter.Suffix{properties: props}

  defp build_filter({:any, subs}) do
    %Solvent.Filter.Any{subfilters: build_filters(subs)}
  end

  defp build_filter({:all, subs}) do
    %Solvent.Filter.All{subfilters: build_filters(subs)}
  end

  defp build_filter({:not, subfilter}) do
    %Solvent.Filter.Not{subfilter: build_filter(subfilter)}
  end
end