raven_integration_kafka

defmodule Integrations.Kafka.Display do
  @moduledoc """
  Dashboard panel for `Integrations.Kafka` (same package).

  Nested under the monitor's own namespace deliberately — the package's
  top-level module name never changes, and this module's name is
  guaranteed distinct from any separately-published standalone display
  package. `Display.BundledDefault` auto-hooks this whenever
  `Integrations.Kafka` starts, the same as if they were still one module.
  """

  use CodeNameRaven.Display, category: :infrastructure, size_hint: :large

  @default_port 9092

  @impl true
  def display_name, do: "Kafka"

  @impl true
  def compatible_monitors, do: [Integrations.Kafka]

  @impl true
  def render(assigns) do
    samples  = CodeNameRaven.Runtime.recent_samples(assigns.config.id, limit: 120)
    local_id = CodeNameRaven.Runtime.instance_id()
    result   = assigns[:result]

    assigns = Map.merge(assigns, %{
      samples:  samples,
      local_id: local_id,
      result:   result
    })

    ~H"""
    <div class="space-y-4">
      <%!-- Header --%>
      <div class="flex items-start justify-between gap-2">
        <div class="min-w-0">
          <h2 class="text-sm font-semibold text-base-content truncate">
            <%= @config.name || kafka_title(@config.params) %>
          </h2>
          <p class="text-xs text-base-content/40 font-mono truncate">
            <%= kafka_title(@config.params) %>
          </p>
        </div>
        <CodeNameRaven.MonitorComponents.status_chip status={@status} />
      </div>

      <%!-- Cluster stats --%>
      <div :if={@result} class="flex gap-2 flex-wrap">
        <.stat label="Latency"    value={"#{@result.latency_ms} ms"} />
        <.stat label="Brokers"    value={"#{@result[:broker_count] || "–"}"} />
        <.stat label="Topics"     value={"#{@result[:topic_count] || "–"}"} />
        <.stat label="Partitions" value={"#{@result[:partition_count] || "–"}"} />
        <.stat :if={@result[:controller_id]}
               label="Controller"
               value={"##{@result[:controller_id]}"} />
      </div>

      <%!-- Health warnings --%>
      <div :if={@result && (has_offline?(@result) or has_under_replicated?(@result))}
           class="flex gap-2 flex-wrap">
        <.warn_stat :if={has_offline?(@result)}
                    label="Offline partitions"
                    value={"#{@result[:offline_partitions]}"} />
        <.warn_stat :if={has_under_replicated?(@result)}
                    label="Under-replicated"
                    value={"#{@result[:under_replicated_partitions]}"} />
      </div>

      <%!-- Broker list --%>
      <div :if={@result && @result[:brokers] not in [nil, []]}
           class="text-xs font-mono text-base-content/40 leading-relaxed">
        <span :for={b <- @result[:brokers] || []}>
          <span class={if b.node_id == @result[:controller_id], do: "text-base-content/70", else: ""}>
            <%= b.host %>:<%= b.port %><%= if b.node_id == @result[:controller_id], do: "✦", else: "" %>
          </span>
          <span class="mr-3"></span>
        </span>
      </div>

      <%!-- Charts --%>
      <div class="grid grid-cols-1 sm:grid-cols-2 gap-3">
        <div>
          <p class="text-xs text-base-content/40 mb-1">Latency</p>
          <CodeNameRaven.MonitorComponents.line_chart
            id={"chart-#{@config.id}-lat"}
            samples={@samples}
            local_id={@local_id} />
        </div>
        <div>
          <p class="text-xs text-base-content/40 mb-1">Under-replicated partitions</p>
          <CodeNameRaven.MonitorComponents.metric_chart
            id={"chart-#{@config.id}-ur"}
            samples={@samples}
            local_id={@local_id}
            metric_name="under_replicated_partitions" />
        </div>
      </div>

      <CodeNameRaven.MonitorComponents.error_message message={@last_error} />
      <p class="text-xs text-base-content/30">
        Checked <%= CodeNameRaven.Timezone.format_datetime(@last_checked_at, __MODULE__) %>
      </p>
    </div>
    """
  end

  # ---------------------------------------------------------------------------
  # Private components
  # ---------------------------------------------------------------------------

  defp stat(assigns) do
    ~H"""
    <div class="bg-base-200 rounded px-3 py-2 text-center min-w-[70px]">
      <p class="text-xs text-base-content/50"><%= @label %></p>
      <p class="text-sm font-semibold text-base-content"><%= @value %></p>
    </div>
    """
  end

  defp warn_stat(assigns) do
    ~H"""
    <div class="flex items-center gap-1.5 bg-warning/10 border border-warning/30 rounded px-3 py-2">
      <span class="text-xs text-warning/80"><%= @label %>:</span>
      <span class="text-sm font-bold text-warning"><%= @value %></span>
    </div>
    """
  end

  # ---------------------------------------------------------------------------
  # Render helpers
  # ---------------------------------------------------------------------------

  defp kafka_title(params) do
    host = params[:host] || params["host"] || "?"
    port = parse_int(params[:port] || params["port"], @default_port)
    "#{host}:#{port}"
  end

  defp has_offline?(%{offline_partitions: n}) when is_integer(n) and n > 0, do: true
  defp has_offline?(_), do: false

  defp has_under_replicated?(%{under_replicated_partitions: n}) when is_integer(n) and n > 0, do: true
  defp has_under_replicated?(_), do: false

  defp parse_int(nil, default), do: default
  defp parse_int(v, _) when is_integer(v), do: v
  defp parse_int(v, default) when is_binary(v) do
    case Integer.parse(v) do
      {n, _} -> n
      :error -> default
    end
  end
  defp parse_int(_, default), do: default
end