cyclium

defmodule Cyclium.Bus do
  @moduledoc """
  Event pub/sub that connects actors without coupling.

  Two event layers:
  - Layer A: Domain events (emitted by your application)
  - Layer B: Runtime events (emitted by Cyclium internals)

  Built on Phoenix.PubSub. The consuming app provides the PubSub process
  via `Cyclium.Supervisor` opts.
  """

  @topic "cyclium:events"

  @runtime_events [
    "expectation.due",
    "expectation.triggered",
    "finding.raised",
    "finding.updated",
    "finding.cleared",
    "work.requested",
    "output.proposed",
    "output.delivered",
    "approval.requested",
    "approval.resolved",
    "spec.updated",
    "agent.state_changed",
    "episode.completed",
    "episode.failed",
    "episode.queued",
    "episode.started",
    "episode.dropped",
    "episode.canceled",
    "workflow.started",
    "workflow.completed",
    "workflow.failed"
  ]

  def runtime_events, do: @runtime_events

  @doc """
  Publish an event to the bus. All subscribers receive `{:bus, event_type, payload}`.
  """
  def publish(event_type, payload \\ %{}) when is_binary(event_type) do
    case pubsub() do
      nil -> {:error, :no_pubsub}
      pubsub -> Phoenix.PubSub.broadcast(pubsub, @topic, {:bus, event_type, payload})
    end
  end

  @doc """
  Subscribe the calling process to all bus events.
  Messages arrive as `{:bus, event_type, payload}`.
  """
  def subscribe do
    case pubsub() do
      nil -> {:error, :no_pubsub}
      pubsub -> Phoenix.PubSub.subscribe(pubsub, @topic)
    end
  end

  @doc """
  Subscribe to a specific event type topic.
  Messages arrive as `{:bus, event_type, payload}`.
  """
  def subscribe(event_type) when is_binary(event_type) do
    case pubsub() do
      nil -> {:error, :no_pubsub}
      pubsub -> Phoenix.PubSub.subscribe(pubsub, "#{@topic}:#{event_type}")
    end
  end

  @doc """
  Publish to both the global topic and the event-specific topic,
  so both broad and targeted subscribers receive the event.
  """
  def broadcast(event_type, payload \\ %{}) when is_binary(event_type) do
    case pubsub() do
      nil ->
        {:error, :no_pubsub}

      pubsub ->
        msg = {:bus, event_type, payload}
        Phoenix.PubSub.broadcast(pubsub, @topic, msg)
        Phoenix.PubSub.broadcast(pubsub, "#{@topic}:#{event_type}", msg)
    end
  end

  @cancel_topic_prefix "cyclium:episode_cancel:"

  @doc """
  Subscribe the calling process to cancel requests for one episode.

  A running `EpisodeTask` subscribes so a "stop" reaches it on whatever node
  runs the episode (PubSub is cluster-wide), landing as `{:cyclium_cancel,
  reason}` handled at the next step boundary.
  """
  def subscribe_cancel(episode_id) do
    case pubsub() do
      nil -> {:error, :no_pubsub}
      pubsub -> Phoenix.PubSub.subscribe(pubsub, @cancel_topic_prefix <> to_string(episode_id))
    end
  end

  @doc """
  Broadcast a cancel request for an episode to whichever node runs it. Backs
  `Cyclium.Episodes.request_cancel/2`.
  """
  def publish_cancel(episode_id, reason) do
    case pubsub() do
      nil ->
        {:error, :no_pubsub}

      pubsub ->
        Phoenix.PubSub.broadcast(
          pubsub,
          @cancel_topic_prefix <> to_string(episode_id),
          {:cyclium_cancel, reason}
        )
    end
  end

  defp pubsub do
    Application.get_env(:cyclium, :pubsub)
  end
end