routeros_api

defmodule RouterosApi.Stream do
  @moduledoc """
  Streaming API for continuous RouterOS data.

  This module provides an interface for consuming continuous data streams
  from RouterOS monitoring commands like `/interface/monitor-traffic`.

  ## Example

      # Start a stream connection
      {:ok, stream_conn} = RouterosApi.Stream.connect(%{
        host: "192.168.88.1",
        username: "admin",
        password: "password"
      })

      # Create a stream for interface traffic monitoring
      {:ok, stream} = RouterosApi.Stream.monitor(stream_conn, [
        "/interface/monitor-traffic",
        "=interface=ether1"
      ])

      # Consume the stream
      stream
      |> Stream.take(10)
      |> Enum.each(fn data ->
        IO.puts("RX: \#{data["rx-bits-per-second"]} bps")
      end)

      # Disconnect when done
      RouterosApi.Stream.disconnect(stream_conn)

  ## Stream Behavior

  - Streams are lazy - the command is only sent when the stream is consumed
  - Messages are sent to the **consuming process**, not the process that called `monitor/3`
  - Streams automatically cancel when halted (e.g., via `Stream.take/2`)
  - Each stream requires a dedicated connection (cannot be pooled)
  - Only one stream can be active per connection at a time
  - Streams can be safely passed between processes or spawned in Tasks

  ## Duration-Limited Streams

  RouterOS supports duration-limited monitoring:

      {:ok, stream} = RouterosApi.Stream.monitor(stream_conn, [
        "/interface/monitor-traffic",
        "=interface=ether1",
        "=duration=30s"
      ])

  The stream will automatically complete after 30 seconds.

  ## Error Handling

  Errors during streaming raise `RouterosApi.Error` exceptions:

      try do
        stream |> Enum.to_list()
      rescue
        e in RouterosApi.Error ->
          IO.puts("Stream error: \#{e.message}")
      end

  ## Limitations

  - Not compatible with connection pools
  - One stream per connection (sequential streams supported)
  - No automatic reconnection on connection loss
  """

  alias RouterosApi.{Error, StreamConnection}

  @type stream_opts :: [
          timeout: pos_integer()
        ]

  @doc """
  Creates a streaming connection to a RouterOS device.

  This starts a dedicated connection for streaming operations.
  The connection cannot be used with pools.

  ## Options

  Same as `RouterosApi.Connection.start_link/1`:
  - `:host` - Router hostname or IP address (required)
  - `:port` - Port number (optional, defaults to 8728/8729)
  - `:username` - RouterOS username (required)
  - `:password` - RouterOS password (required)
  - `:ssl` - Use TLS connection (optional)
  - `:ssl_opts` - SSL options (optional)
  - `:timeout` - Connection timeout in ms (default: 5000)

  ## Example

      {:ok, conn} = RouterosApi.Stream.connect(%{
        host: "192.168.88.1",
        username: "admin",
        password: "password"
      })
  """
  @spec connect(map()) :: {:ok, pid()} | {:error, term()}
  def connect(config) when is_map(config) do
    StreamConnection.start_link(config)
  end

  @doc """
  Starts a streaming command and returns an Elixir Stream.

  The stream emits parsed data maps as they arrive from the router.
  The stream is lazy - the command is only sent when the stream is consumed,
  and messages are sent to the consuming process.

  ## Parameters

  - `conn` - StreamConnection pid from `connect/1`
  - `words` - Command words (e.g., `["/interface/monitor-traffic", "=interface=ether1"]`)
  - `opts` - Options (currently unused, reserved for future use)

  ## Returns

  - `{:ok, stream}` - An Elixir Stream that yields data maps
  - `{:error, reason}` - If the connection is invalid

  ## Example

      {:ok, stream} = RouterosApi.Stream.monitor(conn, [
        "/interface/monitor-traffic",
        "=interface=ether1"
      ])

      # Take 10 samples
      samples = stream |> Stream.take(10) |> Enum.to_list()

      # Or process indefinitely
      stream
      |> Stream.each(&IO.inspect/1)
      |> Stream.run()

  ## Process Safety

  The stream can be safely passed between processes. The stream command is
  started when the stream is first consumed, and messages are sent to the
  consuming process (not the process that called `monitor/3`).
  """
  @spec monitor(pid(), [String.t()], stream_opts()) ::
          {:ok, Enumerable.t()} | {:error, term()}
  def monitor(conn, words, opts \\ []) when is_pid(conn) and is_list(words) do
    # Validate connection is alive before returning stream
    if Process.alive?(conn) do
      stream = build_lazy_stream(conn, words, opts)
      {:ok, stream}
    else
      {:error, Error.new(:stream_error, "Stream connection is not alive")}
    end
  end

  @doc """
  Cancels an active stream.

  This sends a cancel command to the router and stops the stream.
  Note: When using `Stream.take/2` or similar, cancellation happens
  automatically when the stream is halted.

  ## Parameters

  - `conn` - StreamConnection pid
  - `ref` - Stream reference from the stream metadata

  ## Example

      {:ok, stream} = RouterosApi.Stream.monitor(conn, [...])
      # Stream ref is embedded in the stream, typically used internally
      RouterosApi.Stream.cancel(conn, ref)
  """
  @spec cancel(pid(), reference()) :: :ok | {:error, term()}
  def cancel(conn, ref) when is_pid(conn) and is_reference(ref) do
    StreamConnection.cancel_stream(conn, ref)
  end

  @doc """
  Disconnects a streaming connection.

  This closes the connection and cleans up resources.
  Any active stream will be terminated.

  ## Example

      RouterosApi.Stream.disconnect(conn)
  """
  @spec disconnect(pid()) :: :ok
  def disconnect(conn) when is_pid(conn) do
    StreamConnection.stop(conn)
  end

  ## Private Functions

  # Build a lazy stream that starts the command when first consumed.
  # This ensures messages are sent to the consuming process, not the
  # process that called monitor/3.
  defp build_lazy_stream(conn, words, opts) do
    Stream.resource(
      # Start function - called when stream is first consumed
      # This runs in the consuming process, so self() is correct
      fn ->
        case StreamConnection.start_stream(conn, words, opts) do
          {:ok, ref} ->
            {conn, ref, :active}

          {:error, reason} ->
            # Return error state to be handled in next function
            {:error, reason}
        end
      end,

      # Next function - get next element
      fn
        {:error, reason} ->
          raise Error.new(:stream_error, "Failed to start stream: #{inspect(reason)}")

        {conn, ref, :active} ->
          receive do
            {:stream_data, ^ref, data} ->
              {[data], {conn, ref, :active}}

            {:stream_done, ^ref} ->
              {:halt, {conn, ref, :done}}

            {:stream_error, ^ref, %Error{} = error} ->
              raise error

            {:stream_error, ^ref, error} ->
              raise Error.new(:stream_error, "Stream error: #{inspect(error)}")
          end

        {_conn, _ref, :done} ->
          {:halt, {nil, nil, :done}}
      end,

      # Cleanup function - called when stream ends
      fn
        {:error, _reason} ->
          :ok

        {conn, ref, :active} when is_pid(conn) and is_reference(ref) ->
          # Stream was halted early, cancel it
          StreamConnection.cancel_stream(conn, ref)

          # Drain any remaining messages
          drain_messages(ref)

        _ ->
          :ok
      end
    )
  end

  defp drain_messages(ref) do
    receive do
      {:stream_data, ^ref, _data} -> drain_messages(ref)
      {:stream_done, ^ref} -> :ok
      {:stream_error, ^ref, _error} -> :ok
    after
      100 -> :ok
    end
  end
end