activemq_client

defmodule ActiveMQClient do
  @moduledoc """
  Generic ActiveMQ client using STOMP protocol for Artemis ActiveMQ.

  This library provides a high-level interface for connecting to and interacting with
  Apache ActiveMQ Artemis using the STOMP protocol. It handles message publishing, 
  consuming, and subscription management with configurable message filtering.

  ## Features

  - STOMP-based connection to Artemis ActiveMQ
  - Message publishing and consuming
  - Topic and queue subscription management
  - Configurable message filtering with selectors
  - Automatic reconnection and error handling
  - Callback-based message processing

  ## Usage

      # Start the client
      {:ok, _} = ActiveMQClient.start_link([
        host: "localhost", 
        port: 61613,
        username: "artemis",
        password: "artemis",
        message_handler: &MyApp.handle_message/1
      ])
      
      # Publish a message
      ActiveMQClient.publish_message("Hello World!", "/queue/test")
      
      # Subscribe to a queue
      ActiveMQClient.subscribe_to_queue("/queue/test")
      
      # Subscribe to topic with message filter
      ActiveMQClient.subscribe_to_topic_with_selector(
        "/topic/events", 
        "type = 'MY_EVENT'", 
        "my_subscription"
      )
  """

  use GenServer

  require Logger
  alias Bitwise

  # Default STOMP connection settings
  @default_host "localhost"
  @default_port 61613
  @default_username "artemis"
  @default_password "artemis"
  @default_queue "/queue/artemis_queue"
  @default_topic "/topic/artemis_topic"

  # Message headers and configuration

  defstruct [
    :stomp_client,
    :host,
    :port,
    :username,
    :password,
    :queue,
    :topic,
    :connected,
    :message_handler,
    :subscriptions
  ]

  # Client API

  @doc """
  Starts the ActiveMQ client process.

  ## Options

  - `:host` - ActiveMQ server hostname (default: "localhost")
  - `:port` - ActiveMQ STOMP port (default: 61613) 
  - `:username` - Authentication username (default: "artemis")
  - `:password` - Authentication password (default: "artemis")
  - `:queue` - Default queue destination (default: "/queue/artemis_queue")
  - `:topic` - Default topic destination (default: "/topic/artemis_topic")
  - `:message_handler` - Custom message handler function
  """
  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @doc """
  Publishes a message to a destination.

  ## Parameters

  - `message` - Message content as string
  - `destination` - Destination queue or topic (optional, uses default if nil)

  ## Returns

  - `:ok` on success
  - `{:error, reason}` on failure
  """
  def publish_message(message, destination \\ nil) do
    GenServer.call(__MODULE__, {:publish, message, destination})
  end

  @doc """
  Subscribes to a queue destination.

  ## Parameters

  - `queue` - Queue destination (optional, uses default if nil)

  ## Returns

  - `:ok` on success
  - `{:error, reason}` on failure
  """
  def subscribe_to_queue(queue \\ nil) do
    destination = queue || @default_queue
    GenServer.call(__MODULE__, {:subscribe, destination})
  end

  @doc """
  Subscribes to a topic destination.

  ## Parameters

  - `topic` - Topic destination (optional, uses default if nil)

  ## Returns

  - `:ok` on success
  - `{:error, reason}` on failure
  """
  def subscribe_to_topic(topic \\ nil) do
    destination = topic || @default_topic
    GenServer.call(__MODULE__, {:subscribe, destination})
  end

  @doc """
  Subscribes to a topic with a message selector.

  ## Parameters

  - `topic` - Topic destination
  - `selector` - Message selector for filtering
  - `subscription_id` - Unique subscription identifier

  ## Returns

  - `:ok` on success
  - `{:error, reason}` on failure
  """
  def subscribe_to_topic_with_selector(topic, selector, subscription_id) do
    GenServer.call(__MODULE__, {:subscribe_with_selector, topic, selector, subscription_id})
  end

  @doc """
  Subscribes to multiple topics/queues with different selectors.

  ## Parameters

  - `subscriptions` - List of {destination, selector, subscription_id} tuples

  ## Returns

  - `:ok` on success
  - `{:error, reason}` on failure
  """
  def subscribe_multiple(subscriptions) do
    GenServer.call(__MODULE__, {:subscribe_multiple, subscriptions})
  end

  @doc """
  Unsubscribes from a destination.

  ## Parameters

  - `destination` - Destination to unsubscribe from

  ## Returns

  - `:ok` on success
  - `{:error, reason}` on failure
  """
  def unsubscribe(destination) do
    GenServer.call(__MODULE__, {:unsubscribe, destination})
  end

  @doc """
  Gets the current client status.

  ## Returns

  Map containing connection status, host, port, subscriptions, etc.
  """
  def get_status do
    GenServer.call(__MODULE__, :status)
  end

  @doc """
  Triggers a manual reconnection attempt.
  """
  def reconnect do
    GenServer.cast(__MODULE__, :reconnect)
  end

  @doc """
  Sets a custom message handler function.

  ## Parameters

  - `handler` - Function to handle incoming messages
  """
  def set_message_handler(handler) do
    GenServer.cast(__MODULE__, {:set_handler, handler})
  end

  @doc """
  Sends a reply message to a destination with message type.

  ## Parameters

  - `reply_json` - JSON reply content
  - `reply_destination` - Destination to send reply to
  - `message_type` - Message type header

  ## Returns

  - `:ok` on success
  """
  def send_reply(reply_json, reply_destination, message_type) do
    GenServer.cast(__MODULE__, {:send_reply, reply_json, reply_destination, reply_destination, message_type})
  end

  # GenServer Callbacks

  @impl true
  def init(opts) do
    state = %__MODULE__{
      host: Keyword.get(opts, :host, @default_host),
      port: Keyword.get(opts, :port, @default_port),
      username: Keyword.get(opts, :username, @default_username),
      password: Keyword.get(opts, :password, @default_password),
      queue: Keyword.get(opts, :queue, @default_queue),
      topic: Keyword.get(opts, :topic, @default_topic),
      message_handler: Keyword.get(opts, :message_handler, &default_message_handler/1),
      connected: false,
      subscriptions: MapSet.new()
    }

    # Try to connect immediately
    send(self(), :connect)
    {:ok, state}
  end

  @impl true
  def handle_call({:publish, message, destination}, _from, state) do
    dest = destination || state.queue

    case publish_to_destination(state, message, dest) do
      :ok ->
        Logger.info("Message published successfully to #{dest}: #{inspect(message)}")
        {:reply, :ok, state}

      {:error, reason} ->
        Logger.error("Failed to publish message: #{inspect(reason)}")
        {:reply, {:error, reason}, state}
    end
  end

  @impl true
  def handle_call({:subscribe, destination}, _from, state) do
    case subscribe_to_destination(state, destination) do
      :ok ->
        Logger.info("Successfully subscribed to #{destination}")
        new_subscriptions = MapSet.put(state.subscriptions, destination)
        {:reply, :ok, %{state | subscriptions: new_subscriptions}}

      {:error, reason} ->
        Logger.error("Failed to subscribe to #{destination}: #{inspect(reason)}")
        {:reply, {:error, reason}, state}
    end
  end

  @impl true
  def handle_call({:unsubscribe, destination}, _from, state) do
    case unsubscribe_from_destination(state, destination) do
      :ok ->
        Logger.info("Successfully unsubscribed from #{destination}")
        new_subscriptions = MapSet.delete(state.subscriptions, destination)
        {:reply, :ok, %{state | subscriptions: new_subscriptions}}

      {:error, reason} ->
        Logger.error("Failed to unsubscribe from #{destination}: #{inspect(reason)}")
        {:reply, {:error, reason}, state}
    end
  end

  @impl true
  def handle_call({:subscribe_with_selector, topic, selector, subscription_id}, _from, state) do
    case subscribe_to_topic_with_selector_internal(state, topic, selector, subscription_id) do
      :ok ->
        Logger.info("Successfully subscribed to #{topic} with selector: #{selector}")
        new_subscriptions = MapSet.put(state.subscriptions, {topic, selector, subscription_id})
        {:reply, :ok, %{state | subscriptions: new_subscriptions}}

      {:error, reason} ->
        Logger.error("Failed to subscribe to #{topic} with selector: #{inspect(reason)}")
        {:reply, {:error, reason}, state}
    end
  end

  @impl true
  def handle_call({:subscribe_multiple, subscriptions}, _from, state) do
    results =
      Enum.map(subscriptions, fn {destination, selector, subscription_id} ->
        subscribe_to_topic_with_selector_internal(state, destination, selector, subscription_id)
      end)

    if Enum.all?(results, &(&1 == :ok)) do
      Logger.info("Successfully subscribed to #{length(subscriptions)} destinations")

      new_subscriptions =
        Enum.reduce(subscriptions, state.subscriptions, fn {dest, selector, sub_id}, acc ->
          MapSet.put(acc, {dest, selector, sub_id})
        end)

      {:reply, :ok, %{state | subscriptions: new_subscriptions}}
    else
      Logger.error("Failed to subscribe to multiple destinations: #{inspect(results)}")
      {:reply, {:error, results}, state}
    end
  end

  @impl true
  def handle_call(:status, _from, state) do
    status = %{
      connected: state.connected,
      host: state.host,
      port: state.port,
      queue: state.queue,
      topic: state.topic,
      subscriptions: MapSet.to_list(state.subscriptions)
    }

    {:reply, status, state}
  end

  @impl true
  def handle_cast(:reconnect, state) do
    Logger.info("Manual reconnection triggered")

    if state.stomp_client do
      StompClient.disconnect(state.stomp_client)
    end

    send(self(), :connect)
    {:noreply, %{state | connected: false, stomp_client: nil}}
  end

  @impl true
  def handle_info({:stomp_client, :on_send, message}, state) do
    Logger.info("STOMP client send message: #{inspect(message)}")
    {:noreply, state}
  end

  @impl true
  def handle_cast({:set_handler, handler}, state) do
    {:noreply, %{state | message_handler: handler}}
  end

  @impl true
  def handle_cast({:send_reply, reply_json, formatted_reply_to, reply_to_address, message_type}, state) do
    case publish_to_destination_with_type(
           state,
           reply_json,
           formatted_reply_to,
           message_type
         ) do
      :ok ->
        Logger.info("Successfully sent #{message_type} reply to #{reply_to_address}")

      {:error, reason} ->
        Logger.error(
          "Failed to send #{message_type} reply to #{reply_to_address}: #{inspect(reason)}"
        )
    end

    {:noreply, state}
  end

  @impl true
  def handle_info(:connect, state) do
    case establish_connection(state) do
      {:ok, new_state} ->
        Logger.info("Successfully connected to Artemis ActiveMQ via STOMP")
        # Resubscribe to previously subscribed destinations
        resubscribe_all(new_state)
        {:noreply, new_state}

      {:error, reason} ->
        Logger.error("Failed to connect to Artemis ActiveMQ: #{inspect(reason)}")
        # Retry connection after 5 seconds
        Process.send_after(self(), :connect, 5000)
        {:noreply, state}
    end
  end


  @impl true
  def handle_info({:stomp_client, :on_message, message}, state) do
    # The message comes as a map with string keys
    destination = Map.get(message, "destination", "unknown")
    body = Map.get(message, "body", "")
    headers = Map.delete(message, "body") |> Map.delete("destination")

    Logger.info("Received STOMP message from #{destination}: #{body}")

    # Process the message using the handler
    try do
      state.message_handler.({destination, headers, body})
    rescue
      exception ->
        Logger.error("Error processing message: #{inspect(exception)}")
    end

    {:noreply, state}
  end

  @impl true
  def handle_info({:stomp_client, :on_connect, _message}, state) do
    Logger.info("STOMP client connected successfully")
    {:noreply, state}
  end

  @impl true
  def handle_info({:stomp_client, :on_connect_error, message}, state) do
    Logger.error("STOMP client connection error: #{inspect(message)}")
    {:noreply, %{state | connected: false}}
  end

  @impl true
  def handle_info({:stomp_client, :on_disconnect, _acknowledged}, state) do
    Logger.info("STOMP client disconnected")
    {:noreply, %{state | connected: false, stomp_client: nil}}
  end

  @impl true
  def handle_info(msg, state) do
    Logger.debug("Unhandled message: #{inspect(msg)}")
    {:noreply, state}
  end

  # Private Functions

  defp establish_connection(state) do
    connect_opts = build_connection_options(state)

    try do
      {:ok, client_pid} = StompClient.connect(connect_opts, callback_handler: self())
      Logger.info("STOMP connection established to #{state.host}:#{state.port}")

      new_state = %{state | stomp_client: client_pid, connected: true}
      {:ok, new_state}
    rescue
      error ->
        Logger.error("Failed to establish STOMP connection: #{inspect(error)}")
        {:error, error}
    end
  end

  defp build_connection_options(state) do
    [
      host: state.host,
      port: state.port,
      login: state.username,
      passcode: state.password
    ]
  end

  defp publish_to_destination(state, message, destination) do
    if state.connected and state.stomp_client do
      {dest_type, clean_destination} = parse_destination(destination)
      headers = build_message_headers(dest_type)

      case StompClient.send(state.stomp_client, clean_destination, message, headers) do
        :ok -> :ok
        error -> {:error, error}
      end
    else
      {:error, :not_connected}
    end
  end

  defp parse_destination(destination) do
    case destination do
      "/topic/" <> topic_name -> {"MULTICAST", topic_name}
      "/queue/" <> queue_name -> {"ANYCAST", queue_name}
      # Default to queue semantics
      _ -> {"ANYCAST", destination}
    end
  end

  defp build_message_headers(dest_type) do
    [
      {"destination-type", dest_type},
      {"persistent", "true"},
      {"JMSMessageType", "TextMessage"},
      {"JMSType", "TextMessage"}
    ]
  end

  defp build_message_headers_with_type(dest_type, message_type) do
    [
      {"destination-type", dest_type},
      {"persistent", "true"},
      {"JMSMessageType", "TextMessage"},
      {"JMSType", "TextMessage"},
      {"type", message_type}
    ]
  end

  defp publish_to_destination_with_type(state, message, destination, message_type) do
    if state.connected and state.stomp_client do
      {dest_type, clean_destination} = parse_destination(destination)
      headers = build_message_headers_with_type(dest_type, message_type)

      case StompClient.send(state.stomp_client, clean_destination, message, headers) do
        :ok -> :ok
        error -> {:error, error}
      end
    else
      {:error, :not_connected}
    end
  end

  defp subscribe_to_destination(state, destination) do
    if state.connected and state.stomp_client do
      # Generate a unique subscription ID based on destination
      sub_id = "sub-#{:erlang.phash2(destination)}"

      case StompClient.subscribe(state.stomp_client, destination, id: sub_id) do
        :ok -> :ok
        error -> {:error, error}
      end
    else
      {:error, :not_connected}
    end
  end

  defp subscribe_to_topic_with_selector_internal(state, topic, selector, subscription_id) do
    if state.connected and state.stomp_client do
      case StompClient.subscribe(state.stomp_client, topic,
             id: subscription_id,
             selector: selector
           ) do
        :ok -> :ok
        error -> {:error, error}
      end
    else
      {:error, :not_connected}
    end
  end

  defp unsubscribe_from_destination(state, destination) do
    if state.connected and state.stomp_client do
      case StompClient.unsubscribe(state.stomp_client, destination) do
        :ok -> :ok
        error -> {:error, error}
      end
    else
      {:error, :not_connected}
    end
  end

  defp resubscribe_all(state) do
    Enum.each(state.subscriptions, fn destination ->
      case subscribe_to_destination(state, destination) do
        :ok ->
          Logger.info("Resubscribed to #{destination}")

        {:error, reason} ->
          Logger.error("Failed to resubscribe to #{destination}: #{inspect(reason)}")
      end
    end)
  end

  defp default_message_handler({destination, headers, body}) do
    Logger.info("Received message from #{destination}: #{body}")
    Logger.debug("Message headers: #{inspect(headers)}")
  end




















end