# Job Monitoring Dashboard Implementation Plan > **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. **Goal:** Build a live monitoring dashboard at `/admin/monitoring` showing all polling and discovery job activity across all organizations with real-time updates, health metrics, and diagnostic drill-down. **Architecture:** LiveView dashboard with hybrid update strategy (PubSub for job lifecycle events + periodic polling for metrics). Context layer queries Oban jobs with device/agent associations. Workers broadcast lifecycle events to PubSub topic. **Tech Stack:** Phoenix LiveView, Oban, PubSub, Ecto, TailwindCSS --- ## Task 1: Create JobMonitoring Context (Foundation) **Files:** - Create: `lib/towerops/job_monitoring.ex` - Test: `test/towerops/job_monitoring_test.exs` **Step 1: Write failing test for list_active_jobs** ```elixir # test/towerops/job_monitoring_test.exs defmodule Towerops.JobMonitoringTest do use Towerops.DataCase, async: true alias Towerops.JobMonitoring alias Towerops.Workers.{DevicePollerWorker, DiscoveryWorker} describe "list_active_jobs/0" do test "returns jobs in executing state for polling and discovery workers" do device = insert(:device) # Create executing polling job insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "executing", args: %{"device_id" => device.id}, attempted_at: DateTime.utc_now() ) # Create executing discovery job insert(:oban_job, worker: "Towerops.Workers.DiscoveryWorker", state: "executing", args: %{"device_id" => device.id}, attempted_at: DateTime.utc_now() ) # Create completed job (should not be returned) insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "completed", args: %{"device_id" => device.id} ) jobs = JobMonitoring.list_active_jobs() assert length(jobs) == 2 assert Enum.all?(jobs, fn j -> j.state == "executing" end) end end end ``` **Step 2: Run test to verify it fails** Run: `mix test test/towerops/job_monitoring_test.exs` Expected: FAIL with "module JobMonitoring not defined" **Step 3: Create Oban job factory** ```elixir # test/support/factory.ex (add to existing factory) def oban_job_factory do %Oban.Job{ worker: "Towerops.Workers.DevicePollerWorker", queue: "pollers", args: %{"device_id" => 1}, state: "available", attempt: 0, max_attempts: 3, inserted_at: DateTime.utc_now(), scheduled_at: DateTime.utc_now() } end ``` **Step 4: Write minimal implementation** ```elixir # lib/towerops/job_monitoring.ex defmodule Towerops.JobMonitoring do @moduledoc """ Context for monitoring Oban jobs, specifically polling and discovery operations. Provides queries for active, stuck, failed, and completed jobs with device and agent context for operational monitoring. """ import Ecto.Query alias Towerops.Repo alias Oban.Job @worker_names [ "Towerops.Workers.DevicePollerWorker", "Towerops.Workers.DiscoveryWorker" ] @doc """ Lists all currently executing polling and discovery jobs. """ def list_active_jobs do from(j in Job, where: j.state == "executing", where: j.worker in ^@worker_names, order_by: [asc: j.attempted_at] ) |> Repo.all() end end ``` **Step 5: Run test to verify it passes** Run: `mix test test/towerops/job_monitoring_test.exs` Expected: PASS **Step 6: Commit** ```bash git add lib/towerops/job_monitoring.ex test/towerops/job_monitoring_test.exs test/support/factory.ex git commit -m "feat: add JobMonitoring context with list_active_jobs query" ``` --- ## Task 2: Add Stuck Jobs Detection **Files:** - Modify: `lib/towerops/job_monitoring.ex` - Test: `test/towerops/job_monitoring_test.exs` **Step 1: Write failing test for list_stuck_jobs** ```elixir # test/towerops/job_monitoring_test.exs (add to describe block) describe "list_stuck_jobs/0" do test "returns polling jobs executing longer than 2 minutes" do device = insert(:device) # Stuck polling job (3 minutes ago) stuck_job = insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "executing", args: %{"device_id" => device.id}, attempted_at: DateTime.add(DateTime.utc_now(), -180, :second) ) # Recent polling job (30 seconds ago) - not stuck insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "executing", args: %{"device_id" => device.id}, attempted_at: DateTime.add(DateTime.utc_now(), -30, :second) ) stuck = JobMonitoring.list_stuck_jobs() assert length(stuck) == 1 assert hd(stuck).id == stuck_job.id end test "returns discovery jobs executing longer than 5 minutes" do device = insert(:device) # Stuck discovery job (6 minutes ago) stuck_job = insert(:oban_job, worker: "Towerops.Workers.DiscoveryWorker", state: "executing", args: %{"device_id" => device.id}, attempted_at: DateTime.add(DateTime.utc_now(), -360, :second) ) # Recent discovery job (3 minutes ago) - not stuck insert(:oban_job, worker: "Towerops.Workers.DiscoveryWorker", state: "executing", args: %{"device_id" => device.id}, attempted_at: DateTime.add(DateTime.utc_now(), -180, :second) ) stuck = JobMonitoring.list_stuck_jobs() assert length(stuck) == 1 assert hd(stuck).id == stuck_job.id end end ``` **Step 2: Run test to verify it fails** Run: `mix test test/towerops/job_monitoring_test.exs::list_stuck_jobs` Expected: FAIL with "function list_stuck_jobs/0 undefined" **Step 3: Write minimal implementation** ```elixir # lib/towerops/job_monitoring.ex (add to module) @polling_threshold_seconds 120 # 2 minutes @discovery_threshold_seconds 300 # 5 minutes @doc """ Lists jobs that are executing longer than expected thresholds. Thresholds: - Polling: 2 minutes - Discovery: 5 minutes """ def list_stuck_jobs do now = DateTime.utc_now() polling_threshold = DateTime.add(now, -@polling_threshold_seconds, :second) discovery_threshold = DateTime.add(now, -@discovery_threshold_seconds, :second) from(j in Job, where: j.state == "executing", where: (j.worker == "Towerops.Workers.DevicePollerWorker" and j.attempted_at < ^polling_threshold) or (j.worker == "Towerops.Workers.DiscoveryWorker" and j.attempted_at < ^discovery_threshold), order_by: [asc: j.attempted_at] ) |> Repo.all() end ``` **Step 4: Run test to verify it passes** Run: `mix test test/towerops/job_monitoring_test.exs` Expected: PASS (all tests) **Step 5: Commit** ```bash git add lib/towerops/job_monitoring.ex test/towerops/job_monitoring_test.exs git commit -m "feat: add stuck job detection with per-worker thresholds" ``` --- ## Task 3: Add Failed and Recent Jobs Queries **Files:** - Modify: `lib/towerops/job_monitoring.ex` - Test: `test/towerops/job_monitoring_test.exs` **Step 1: Write failing tests** ```elixir # test/towerops/job_monitoring_test.exs (add describe blocks) describe "list_failed_jobs/0" do test "returns retryable and cancelled jobs" do device = insert(:device) insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "retryable", args: %{"device_id" => device.id} ) insert(:oban_job, worker: "Towerops.Workers.DiscoveryWorker", state: "cancelled", args: %{"device_id" => device.id} ) # Should not include completed insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "completed", args: %{"device_id" => device.id} ) failed = JobMonitoring.list_failed_jobs() assert length(failed) == 2 assert Enum.all?(failed, fn j -> j.state in ["retryable", "cancelled"] end) end end describe "list_recent_completions/1" do test "returns completed, cancelled, and discarded jobs up to limit" do device = insert(:device) # Create 5 completed jobs for _ <- 1..5 do insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "completed", args: %{"device_id" => device.id}, completed_at: DateTime.utc_now() ) end # Should not include executing insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "executing", args: %{"device_id" => device.id} ) recent = JobMonitoring.list_recent_completions(3) assert length(recent) == 3 assert Enum.all?(recent, fn j -> j.state == "completed" end) end end ``` **Step 2: Run test to verify it fails** Run: `mix test test/towerops/job_monitoring_test.exs` Expected: FAIL with "function list_failed_jobs/0 undefined" **Step 3: Write minimal implementation** ```elixir # lib/towerops/job_monitoring.ex (add to module) @doc """ Lists jobs that have failed and are retrying or have been cancelled. Limited to last 100 jobs. """ def list_failed_jobs do from(j in Job, where: j.state in ["retryable", "cancelled"], where: j.worker in ^@worker_names, order_by: [desc: j.updated_at], limit: 100 ) |> Repo.all() end @doc """ Lists recently completed jobs (completed, cancelled, or discarded). ## Options - `:limit` - Maximum number of jobs to return (default: 100) """ def list_recent_completions(limit \\ 100) do from(j in Job, where: j.state in ["completed", "cancelled", "discarded"], where: j.worker in ^@worker_names, order_by: [desc: j.completed_at], limit: ^limit ) |> Repo.all() end ``` **Step 4: Run test to verify it passes** Run: `mix test test/towerops/job_monitoring_test.exs` Expected: PASS (all tests) **Step 5: Commit** ```bash git add lib/towerops/job_monitoring.ex test/towerops/job_monitoring_test.exs git commit -m "feat: add failed jobs and recent completions queries" ``` --- ## Task 4: Add Health Metrics Module **Files:** - Create: `lib/towerops/job_monitoring/metrics.ex` - Test: `test/towerops/job_monitoring/metrics_test.exs` **Step 1: Write failing test for calculate_all** ```elixir # test/towerops/job_monitoring/metrics_test.exs defmodule Towerops.JobMonitoring.MetricsTest do use Towerops.DataCase, async: true alias Towerops.JobMonitoring.Metrics describe "calculate_all/0" do test "returns comprehensive metrics structure" do device = insert(:device) # Executing jobs insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "executing", args: %{"device_id" => device.id} ) # Queued jobs insert(:oban_job, worker: "Towerops.Workers.DiscoveryWorker", state: "scheduled", args: %{"device_id" => device.id} ) # Completed jobs (last hour) now = DateTime.utc_now() insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "completed", args: %{"device_id" => device.id}, completed_at: DateTime.add(now, -1800, :second), attempted_at: DateTime.add(now, -1810, :second) ) metrics = Metrics.calculate_all() assert metrics.executing_count == 1 assert metrics.queued_count == 1 assert metrics.completed_last_hour == 1 assert is_float(metrics.polling_success_rate_1h) assert is_map(metrics.queue_depths) end end end ``` **Step 2: Run test to verify it fails** Run: `mix test test/towerops/job_monitoring/metrics_test.exs` Expected: FAIL with "module Metrics not defined" **Step 3: Write minimal implementation** ```elixir # lib/towerops/job_monitoring/metrics.ex defmodule Towerops.JobMonitoring.Metrics do @moduledoc """ Calculates health metrics for job monitoring dashboard. """ import Ecto.Query alias Towerops.Repo alias Oban.Job @worker_names [ "Towerops.Workers.DevicePollerWorker", "Towerops.Workers.DiscoveryWorker" ] @doc """ Calculates all metrics for the monitoring dashboard. Returns a map with: - executing_count: Number of currently executing jobs - queued_count: Number of scheduled/available jobs - completed_last_hour: Completed jobs in last hour - failed_last_hour: Failed jobs in last hour - polling_success_rate_1h: % successful polls (last hour) - polling_success_rate_24h: % successful polls (last 24 hours) - discovery_success_rate_1h: % successful discoveries (last hour) - discovery_success_rate_24h: % successful discoveries (last 24 hours) - avg_poll_duration: Average poll duration in seconds - avg_discovery_duration: Average discovery duration in seconds - jobs_per_minute: Jobs per minute (last 10 minutes) - queue_depths: Map of queue name to depth """ def calculate_all do now = DateTime.utc_now() one_hour_ago = DateTime.add(now, -3600, :second) twenty_four_hours_ago = DateTime.add(now, -86400, :second) ten_minutes_ago = DateTime.add(now, -600, :second) %{ executing_count: count_by_state("executing"), queued_count: count_by_state(["scheduled", "available"]), completed_last_hour: count_completed_since(one_hour_ago), failed_last_hour: count_failed_since(one_hour_ago), polling_success_rate_1h: success_rate("Towerops.Workers.DevicePollerWorker", one_hour_ago), polling_success_rate_24h: success_rate("Towerops.Workers.DevicePollerWorker", twenty_four_hours_ago), discovery_success_rate_1h: success_rate("Towerops.Workers.DiscoveryWorker", one_hour_ago), discovery_success_rate_24h: success_rate("Towerops.Workers.DiscoveryWorker", twenty_four_hours_ago), avg_poll_duration: avg_duration("Towerops.Workers.DevicePollerWorker", one_hour_ago), avg_discovery_duration: avg_duration("Towerops.Workers.DiscoveryWorker", one_hour_ago), jobs_per_minute: jobs_per_minute(ten_minutes_ago), queue_depths: queue_depths() } end defp count_by_state(state) when is_binary(state) do from(j in Job, where: j.state == ^state, where: j.worker in ^@worker_names, select: count(j.id) ) |> Repo.one() || 0 end defp count_by_state(states) when is_list(states) do from(j in Job, where: j.state in ^states, where: j.worker in ^@worker_names, select: count(j.id) ) |> Repo.one() || 0 end defp count_completed_since(since) do from(j in Job, where: j.state == "completed", where: j.completed_at > ^since, where: j.worker in ^@worker_names, select: count(j.id) ) |> Repo.one() || 0 end defp count_failed_since(since) do from(j in Job, where: j.state in ["retryable", "cancelled", "discarded"], where: j.updated_at > ^since, where: j.worker in ^@worker_names, select: count(j.id) ) |> Repo.one() || 0 end defp success_rate(worker, since) do total = from(j in Job, where: j.worker == ^worker, where: j.completed_at > ^since or j.updated_at > ^since, select: count(j.id) ) |> Repo.one() || 0 successful = from(j in Job, where: j.worker == ^worker, where: j.state == "completed", where: j.completed_at > ^since, select: count(j.id) ) |> Repo.one() || 0 case total do 0 -> 0.0 _ -> (successful / total) * 100 end end defp avg_duration(worker, since) do result = from(j in Job, where: j.worker == ^worker, where: j.state == "completed", where: j.completed_at > ^since, select: avg(fragment("EXTRACT(EPOCH FROM (? - ?))", j.completed_at, j.attempted_at)) ) |> Repo.one() case result do nil -> 0.0 duration -> Float.round(duration, 2) end end defp jobs_per_minute(since) do completed = count_completed_since(since) minutes_elapsed = 10.0 Float.round(completed / minutes_elapsed, 2) end defp queue_depths do queues = ["pollers", "discovery", "maintenance"] Enum.into(queues, %{}, fn queue -> depth = from(j in Job, where: j.queue == ^queue, where: j.state in ["scheduled", "available"], select: count(j.id) ) |> Repo.one() || 0 {queue, depth} end) end end ``` **Step 4: Run test to verify it passes** Run: `mix test test/towerops/job_monitoring/metrics_test.exs` Expected: PASS **Step 5: Commit** ```bash git add lib/towerops/job_monitoring/metrics.ex test/towerops/job_monitoring/metrics_test.exs git commit -m "feat: add health metrics calculation module" ``` --- ## Task 5: Add PubSub Events Module **Files:** - Create: `lib/towerops/job_monitoring/events.ex` - Test: `test/towerops/job_monitoring/events_test.exs` **Step 1: Write failing test** ```elixir # test/towerops/job_monitoring/events_test.exs defmodule Towerops.JobMonitoring.EventsTest do use Towerops.DataCase, async: true alias Towerops.JobMonitoring.Events describe "broadcast_job_event/3" do test "broadcasts job lifecycle event to PubSub" do Phoenix.PubSub.subscribe(Towerops.PubSub, "job:lifecycle") job = %Oban.Job{ id: 123, worker: "Towerops.Workers.DevicePollerWorker", args: %{"device_id" => 456} } Events.broadcast_job_event(job, :started, %{}) assert_receive %{ job_id: 123, worker: "Towerops.Workers.DevicePollerWorker", device_id: 456, event: :started, metadata: %{}, timestamp: %DateTime{} } end end end ``` **Step 2: Run test to verify it fails** Run: `mix test test/towerops/job_monitoring/events_test.exs` Expected: FAIL with "module Events not defined" **Step 3: Write minimal implementation** ```elixir # lib/towerops/job_monitoring/events.ex defmodule Towerops.JobMonitoring.Events do @moduledoc """ PubSub event broadcasting for job lifecycle events. """ @topic "job:lifecycle" @doc """ Broadcasts a job lifecycle event to PubSub. ## Events - `:started` - Job began executing - `:completed` - Job finished successfully - `:failed` - Job failed with error ## Metadata Optional map with additional context: - `duration`: Execution time in seconds - `error`: Error message/reason - `items_collected`: Number of items discovered/polled """ def broadcast_job_event(%Oban.Job{} = job, event, metadata \\ %{}) do Phoenix.PubSub.broadcast( Towerops.PubSub, @topic, %{ job_id: job.id, worker: job.worker, device_id: get_in(job.args, ["device_id"]), event: event, metadata: metadata, timestamp: DateTime.utc_now() } ) end end ``` **Step 4: Run test to verify it passes** Run: `mix test test/towerops/job_monitoring/events_test.exs` Expected: PASS **Step 5: Commit** ```bash git add lib/towerops/job_monitoring/events.ex test/towerops/job_monitoring/events_test.exs git commit -m "feat: add PubSub event broadcasting for job lifecycle" ``` --- ## Task 6: Integrate Events into DevicePollerWorker **Files:** - Modify: `lib/towerops/workers/device_poller_worker.ex` **Step 1: Add event broadcasting to perform/1** ```elixir # lib/towerops/workers/device_poller_worker.ex # Add alias at top of module alias Towerops.JobMonitoring.Events # Modify perform/1 function to broadcast events @impl Oban.Worker def perform(%Oban.Job{args: %{"device_id" => device_id}} = job) do Events.broadcast_job_event(job, :started) start_time = System.monotonic_time(:second) result = case Devices.get_device(device_id) do nil -> Logger.debug("Device #{device_id} no longer exists, skipping poll") :ok device -> maybe_poll_device(device) schedule_next_poll_with_error_handling(device_id, device) :ok end duration = System.monotonic_time(:second) - start_time case result do :ok -> Events.broadcast_job_event(job, :completed, %{duration: duration}) {:error, reason} -> Events.broadcast_job_event(job, :failed, %{error: inspect(reason), duration: duration}) end result end ``` **Step 2: Manually test (no automated test for worker integration)** Run Phoenix server and verify events are broadcast by subscribing in iex: ```elixir iex> Phoenix.PubSub.subscribe(Towerops.PubSub, "job:lifecycle") iex> # Trigger a poll job iex> # Should receive event messages ``` **Step 3: Commit** ```bash git add lib/towerops/workers/device_poller_worker.ex git commit -m "feat: integrate lifecycle event broadcasting into DevicePollerWorker" ``` --- ## Task 7: Integrate Events into DiscoveryWorker **Files:** - Modify: `lib/towerops/workers/discovery_worker.ex` **Step 1: Add event broadcasting to perform/1** ```elixir # lib/towerops/workers/discovery_worker.ex # Add alias at top of module alias Towerops.JobMonitoring.Events # Modify perform/1 to broadcast events (find existing perform/1 and wrap it) @impl Oban.Worker def perform(%Oban.Job{args: %{"device_id" => device_id}} = job) do Events.broadcast_job_event(job, :started) start_time = System.monotonic_time(:second) result = case Devices.get_device_with_details(device_id) do nil -> Logger.warning("Device #{device_id} not found, discarding discovery job") :discard device -> # Existing discovery logic wrapped in Task.async/await task = Task.async(fn -> if device.agent_id && agent_online?(device.agent_id) do perform_agent_discovery(device) else perform_direct_discovery(device) end end) case Task.await(task, @job_timeout_ms) do :ok -> :discard {:error, reason} -> Logger.error("Discovery failed for device #{device_id}: #{inspect(reason)}") :discard end end duration = System.monotonic_time(:second) - start_time case result do :discard -> # Could be success or failure - check if it's an error case Events.broadcast_job_event(job, :completed, %{duration: duration}) {:error, reason} -> Events.broadcast_job_event(job, :failed, %{error: inspect(reason), duration: duration}) end result end ``` **Step 2: Manually test** Same as previous task - verify in iex that discovery events are broadcast. **Step 3: Commit** ```bash git add lib/towerops/workers/discovery_worker.ex git commit -m "feat: integrate lifecycle event broadcasting into DiscoveryWorker" ``` --- ## Task 8: Create MonitoringLive Base Structure **Files:** - Create: `lib/towerops_web/live/admin/monitoring_live.ex` - Create: `lib/towerops_web/live/admin/monitoring_live.html.heex` - Test: `test/towerops_web/live/admin/monitoring_live_test.exs` **Step 1: Write failing mount test** ```elixir # test/towerops_web/live/admin/monitoring_live_test.exs defmodule ToweropsWeb.Admin.MonitoringLiveTest do use ToweropsWeb.ConnCase, async: true import Phoenix.LiveViewTest setup do superuser = insert(:user, superuser: true) %{superuser: superuser} end describe "mount" do test "renders monitoring dashboard for superuser", %{conn: conn, superuser: superuser} do conn = log_in_user(conn, superuser) {:ok, _view, html} = live(conn, ~p"/admin/monitoring") assert html =~ "Job Monitoring" assert html =~ "Active Operations" assert html =~ "Health Metrics" end test "requires superuser authentication", %{conn: conn} do regular_user = insert(:user, superuser: false) conn = log_in_user(conn, regular_user) assert {:error, {:redirect, %{to: "/"}}} = live(conn, ~p"/admin/monitoring") end end end ``` **Step 2: Run test to verify it fails** Run: `mix test test/towerops_web/live/admin/monitoring_live_test.exs` Expected: FAIL with "no route found for GET /admin/monitoring" **Step 3: Add route** ```elixir # lib/towerops_web/router.ex (find admin scope and add) scope "/admin", ToweropsWeb.Admin, as: :admin do pipe_through [:browser, :require_authenticated_user, :require_superadmin] live_session :admin, on_mount: [{ToweropsWeb.UserAuth, :ensure_authenticated}, {ToweropsWeb.UserAuth, :ensure_superadmin}] do live "/", DashboardLive, :index live "/monitoring", MonitoringLive, :index # Add this line # ... existing routes end end ``` **Step 4: Create minimal LiveView** ```elixir # lib/towerops_web/live/admin/monitoring_live.ex defmodule ToweropsWeb.Admin.MonitoringLive do @moduledoc """ Live monitoring dashboard for all polling and discovery jobs. Shows active operations, health metrics, recent activity, and problem detection across all organizations. """ use ToweropsWeb, :live_view alias Towerops.JobMonitoring alias Towerops.JobMonitoring.Metrics @impl true def mount(_params, _session, socket) do if connected?(socket) do Phoenix.PubSub.subscribe(Towerops.PubSub, "job:lifecycle") schedule_metrics_refresh() end {:ok, socket |> assign(:page_title, "Job Monitoring") |> load_initial_data()} end @impl true def handle_params(_params, _url, socket) do {:noreply, socket} end defp load_initial_data(socket) do socket |> assign(:executing_jobs, JobMonitoring.list_active_jobs()) |> assign(:stuck_jobs, JobMonitoring.list_stuck_jobs()) |> assign(:failed_jobs, JobMonitoring.list_failed_jobs()) |> assign(:recent_events, JobMonitoring.list_recent_completions(100)) |> assign(:metrics, Metrics.calculate_all()) |> assign(:diagnostic_modal_open, false) |> assign(:diagnostic_job, nil) end defp schedule_metrics_refresh do Process.send_after(self(), :refresh_metrics, 10_000) end @impl true def handle_info(:refresh_metrics, socket) do schedule_metrics_refresh() {:noreply, assign(socket, :metrics, Metrics.calculate_all())} end @impl true def handle_info(%{event: :started} = event, socket) do # Job started - add to executing jobs (reload from DB to get full context) job = Towerops.Repo.get(Oban.Job, event.job_id) socket = if job do update(socket, :executing_jobs, fn jobs -> [job | jobs] end) else socket end {:noreply, socket} end @impl true def handle_info(%{event: :completed} = event, socket) do # Job completed - remove from executing, add to recent events socket = socket |> update(:executing_jobs, fn jobs -> Enum.reject(jobs, fn j -> j.id == event.job_id end) end) |> update(:recent_events, fn events -> job = Towerops.Repo.get(Oban.Job, event.job_id) if job, do: [job | Enum.take(events, 99)], else: events end) {:noreply, socket} end @impl true def handle_info(%{event: :failed} = event, socket) do # Job failed - remove from executing, add to failed and recent job = Towerops.Repo.get(Oban.Job, event.job_id) socket = if job do socket |> update(:executing_jobs, fn jobs -> Enum.reject(jobs, fn j -> j.id == event.job_id end) end) |> update(:failed_jobs, fn jobs -> [job | jobs] end) |> update(:recent_events, fn events -> [job | Enum.take(events, 99)] end) else socket end {:noreply, socket} end end ``` **Step 5: Create minimal template** ```heex

Job Monitoring

Real-time monitoring of polling and discovery operations

<%= if length(@stuck_jobs) > 0 or length(@failed_jobs) > 0 do %>

⚠️ Problems Detected

<%= if length(@stuck_jobs) > 0 do %>

{length(@stuck_jobs)} stuck jobs

<% end %> <%= if length(@failed_jobs) > 0 do %>

{length(@failed_jobs)} failed jobs

<% end %>
<% end %>

Active Operations

{length(@executing_jobs)} jobs currently executing

Health Metrics

Executing
{@metrics.executing_count}
Queued
{@metrics.queued_count}

Recent Activity

Last {length(@recent_events)} events

``` **Step 6: Run test to verify it passes** Run: `mix test test/towerops_web/live/admin/monitoring_live_test.exs` Expected: PASS **Step 7: Commit** ```bash git add lib/towerops_web/router.ex lib/towerops_web/live/admin/monitoring_live.ex lib/towerops_web/live/admin/monitoring_live.html.heex test/towerops_web/live/admin/monitoring_live_test.exs git commit -m "feat: create MonitoringLive base structure with PubSub subscription" ``` --- ## Task 9: Add Link to Monitoring Dashboard from Admin Dashboard **Files:** - Modify: `lib/towerops_web/live/admin/dashboard_live.html.heex` **Step 1: Add monitoring card to dashboard** ```heex

<.icon name="hero-cpu-chip" class="w-5 h-5 inline mr-1" /> Job Monitoring

Real-time polling and discovery job monitoring

<.link navigate={~p"/admin/monitoring"} class="text-blue-600 dark:text-blue-400 hover:underline text-sm" > Open Monitoring →
``` **Step 2: Manual test** Start Phoenix server: `mix phx.server` Navigate to: `http://localhost:4000/admin` Verify: "Job Monitoring" card appears and link works **Step 3: Commit** ```bash git add lib/towerops_web/live/admin/dashboard_live.html.heex git commit -m "feat: add job monitoring link to admin dashboard" ``` --- ## Task 10: Enhance Active Operations Display **Files:** - Modify: `lib/towerops_web/live/admin/monitoring_live.html.heex` - Modify: `lib/towerops/job_monitoring.ex` **Step 1: Add device preloading to queries** ```elixir # lib/towerops/job_monitoring.ex # Update list_active_jobs to preload device def list_active_jobs do from(j in Job, where: j.state == "executing", where: j.worker in ^@worker_names, order_by: [asc: j.attempted_at] ) |> Repo.all() |> preload_device_context() end # Add helper to fetch device context defp preload_device_context(jobs) do Enum.map(jobs, fn job -> device_id = get_in(job.args, ["device_id"]) device = device_id && Towerops.Repo.get(Towerops.Devices.Device, device_id) Map.put(job, :device, device) end) end ``` **Step 2: Enhance active operations template** ```heex

Active Operations ({length(@executing_jobs)})

<%= if length(@executing_jobs) == 0 do %>

No jobs currently executing

<% else %>
<%= for job <- @executing_jobs do %>

<%= if job.device do %> <%= job.device.name %> <% else %> Device #{get_in(job.args, ["device_id"])} <% end %>

<%= worker_name(job.worker) %>

Started <%= ToweropsWeb.TimeHelpers.time_ago(job.attempted_at) %>

<.icon name="hero-arrow-path" class="w-4 h-4 animate-spin text-blue-600" />
<% end %>
<% end %>
``` **Step 3: Add helper function to LiveView** ```elixir # lib/towerops_web/live/admin/monitoring_live.ex (add to module) defp worker_name("Towerops.Workers.DevicePollerWorker"), do: "Device Poll" defp worker_name("Towerops.Workers.DiscoveryWorker"), do: "SNMP Discovery" defp worker_name(worker), do: worker ``` **Step 4: Manual test** Start Phoenix server and trigger some polling jobs. Verify they appear in Active Operations. **Step 5: Commit** ```bash git add lib/towerops/job_monitoring.ex lib/towerops_web/live/admin/monitoring_live.ex lib/towerops_web/live/admin/monitoring_live.html.heex git commit -m "feat: enhance active operations display with device context" ``` --- ## Task 11: Build Problems Section UI **Files:** - Modify: `lib/towerops_web/live/admin/monitoring_live.html.heex` **Step 1: Replace problems section with detailed view** ```heex <%= if length(@stuck_jobs) > 0 or length(@failed_jobs) > 0 do %>

⚠️ Problems Detected

<%= if length(@stuck_jobs) > 0 do %>

Stuck Jobs ({length(@stuck_jobs)})

<%= for job <- @stuck_jobs do %>

<%= if job.device do %> <%= job.device.name %> <% else %> Device #{get_in(job.args, ["device_id"])} <% end %>

<%= worker_name(job.worker) %>

Running for <%= duration_in_words(job.attempted_at) %>

<% end %>
<% end %> <%= if length(@failed_jobs) > 0 do %>

Failed Jobs ({length(@failed_jobs)})

<%= for job <- Enum.take(@failed_jobs, 5) do %>

<%= if job.device do %> <%= job.device.name %> <% else %> Device #{get_in(job.args, ["device_id"])} <% end %>

<%= worker_name(job.worker) %> - Attempt <%= job.attempt %>/<%= job.max_attempts %>

<%= if job.errors && length(job.errors) > 0 do %>

<%= hd(job.errors)["error"] || "Unknown error" %>

<% end %>
<% end %>
<% end %>
<% end %> ``` **Step 2: Add helper function** ```elixir # lib/towerops_web/live/admin/monitoring_live.ex (add to module) defp duration_in_words(started_at) do seconds = DateTime.diff(DateTime.utc_now(), started_at) minutes = div(seconds, 60) cond do minutes < 1 -> "#{seconds}s" minutes < 60 -> "#{minutes}m" true -> "#{div(minutes, 60)}h #{rem(minutes, 60)}m" end end ``` **Step 3: Manual test** Create a long-running job or modify thresholds temporarily to see stuck jobs. **Step 4: Commit** ```bash git add lib/towerops_web/live/admin/monitoring_live.ex lib/towerops_web/live/admin/monitoring_live.html.heex git commit -m "feat: build detailed problems section UI for stuck and failed jobs" ``` --- ## Task 12: Build Health Metrics Panel **Files:** - Modify: `lib/towerops_web/live/admin/monitoring_live.html.heex` **Step 1: Replace health metrics section with full display** ```heex

Health Metrics

Current Activity

Executing

{@metrics.executing_count}

Queued

{@metrics.queued_count}

Completed (1h)

{@metrics.completed_last_hour}

Failed (1h)

{@metrics.failed_last_hour}

Success Rates

Polling (1h) <%= Float.round(@metrics.polling_success_rate_1h, 1) %>%
Discovery (1h) <%= Float.round(@metrics.discovery_success_rate_1h, 1) %>%

Performance

Avg Poll Duration
{@metrics.avg_poll_duration}s
Avg Discovery Duration
{@metrics.avg_discovery_duration}s
Jobs/Minute
{@metrics.jobs_per_minute}

Updates every 10 seconds

``` **Step 2: Manual test** View dashboard and verify all metrics display correctly. **Step 3: Commit** ```bash git add lib/towerops_web/live/admin/monitoring_live.html.heex git commit -m "feat: build comprehensive health metrics panel with success rates" ``` --- ## Task 13: Build Recent Activity Timeline **Files:** - Modify: `lib/towerops_web/live/admin/monitoring_live.html.heex` **Step 1: Replace recent activity section** ```heex

Recent Activity

<%= if length(@recent_events) == 0 do %>

No recent activity

<% else %>
<%= for job <- Enum.take(@recent_events, 50) do %>

<%= if job.device do %> <%= job.device.name %> <% else %> Device #{get_in(job.args, ["device_id"])} <% end %>

<%= worker_name(job.worker) %> · <%= event_outcome(job) %>

<%= ToweropsWeb.TimeHelpers.time_ago(job.completed_at || job.updated_at) %>
<% end %>
<% end %>
``` **Step 2: Add helper functions** ```elixir # lib/towerops_web/live/admin/monitoring_live.ex (add to module) defp event_border_color("completed"), do: "border-green-500" defp event_border_color("cancelled"), do: "border-red-500" defp event_border_color("discarded"), do: "border-gray-400" defp event_border_color(_), do: "border-gray-300" defp event_outcome(job) do case job.state do "completed" -> "Completed" "cancelled" -> "Failed" "discarded" -> "Discarded" _ -> String.capitalize(job.state) end end ``` **Step 3: Manual test** View dashboard and verify recent activity timeline shows recent job completions. **Step 4: Commit** ```bash git add lib/towerops_web/live/admin/monitoring_live.ex lib/towerops_web/live/admin/monitoring_live.html.heex git commit -m "feat: build recent activity timeline with color-coded events" ``` --- ## Task 14: Add Queue Depth Display **Files:** - Modify: `lib/towerops_web/live/admin/monitoring_live.html.heex` **Step 1: Add queue depth section to metrics panel** ```heex

Queue Depths

Pollers
{@metrics.queue_depths["pollers"]}
Discovery
{@metrics.queue_depths["discovery"]}
Maintenance
{@metrics.queue_depths["maintenance"]}
``` **Step 2: Manual test** View dashboard and verify queue depths display correctly. **Step 3: Commit** ```bash git add lib/towerops_web/live/admin/monitoring_live.html.heex git commit -m "feat: add queue depth display to health metrics" ``` --- ## Task 15: Final Testing and Polish **Files:** - Test: `test/towerops_web/live/admin/monitoring_live_test.exs` **Step 1: Add comprehensive integration tests** ```elixir # test/towerops_web/live/admin/monitoring_live_test.exs (add tests) describe "real-time updates" do test "updates when job starts via PubSub", %{conn: conn, superuser: superuser} do conn = log_in_user(conn, superuser) {:ok, view, _html} = live(conn, ~p"/admin/monitoring") device = insert(:device) job = insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "executing", args: %{"device_id" => device.id} ) # Simulate PubSub event send(view.pid, %{ event: :started, job_id: job.id, worker: job.worker, device_id: device.id }) assert render(view) =~ device.name end end describe "metrics display" do test "shows current activity metrics", %{conn: conn, superuser: superuser} do device = insert(:device) insert(:oban_job, worker: "Towerops.Workers.DevicePollerWorker", state: "executing", args: %{"device_id" => device.id} ) conn = log_in_user(conn, superuser) {:ok, _view, html} = live(conn, ~p"/admin/monitoring") assert html =~ "Executing" assert html =~ "1" # One executing job end end ``` **Step 2: Run all tests** Run: `mix test test/towerops_web/live/admin/monitoring_live_test.exs` Expected: PASS (all tests) **Step 3: Run full test suite to ensure no regressions** Run: `mix test` Expected: PASS (all tests) **Step 4: Manual end-to-end test** 1. Start Phoenix server: `mix phx.server` 2. Navigate to `/admin/monitoring` 3. Verify all sections render 4. Trigger some polls/discoveries 5. Verify real-time updates work 6. Check metrics refresh after 10 seconds **Step 5: Commit** ```bash git add test/towerops_web/live/admin/monitoring_live_test.exs git commit -m "test: add comprehensive integration tests for monitoring dashboard" ``` --- ## Task 16: Documentation and Cleanup **Files:** - Create: `docs/features/job-monitoring-dashboard.md` **Step 1: Write feature documentation** ```markdown # Job Monitoring Dashboard ## Overview The Job Monitoring Dashboard (`/admin/monitoring`) provides real-time visibility into all polling and discovery operations across all organizations. ## Features ### Real-Time Updates - PubSub broadcasts job lifecycle events (start, complete, fail) - LiveView updates immediately when jobs change state - Metrics refresh every 10 seconds ### Problem Detection - Stuck jobs (executing longer than threshold) - Polling: >2 minutes - Discovery: >5 minutes - Failed/retrying jobs - Discarded jobs ### Health Metrics - Current activity (executing, queued counts) - Success rates (1h and 24h windows) - Performance metrics (avg duration, jobs/minute) - Queue depths per queue ### Recent Activity - Last 100 job completions - Color-coded by outcome - Shows device, operation type, timestamp ## Usage ### Accessing the Dashboard 1. Log in as superuser 2. Navigate to `/admin` dashboard 3. Click "Job Monitoring" card 4. Dashboard loads with current state ### Monitoring Jobs **Active Operations** shows currently executing jobs: - Device name - Operation type (Poll vs Discovery) - Duration so far - Real-time spinner **Problems** section only appears when issues exist: - Stuck jobs with duration - Failed jobs with error messages - Click any job for diagnostic details (future enhancement) **Health Metrics** panel updates every 10 seconds: - Current counts - Success rate progress bars - Performance statistics **Recent Activity** timeline shows last 100 events: - Scrollable list - Color-coded borders (green=success, red=failed) - Relative timestamps ## Technical Details ### Architecture ``` MonitoringLive (LiveView) ├─ Subscribes to "job:lifecycle" PubSub topic ├─ Loads initial data from JobMonitoring context ├─ Handles lifecycle events (started, completed, failed) └─ Refreshes metrics every 10 seconds JobMonitoring (Context) ├─ list_active_jobs/0 - executing jobs ├─ list_stuck_jobs/0 - jobs exceeding threshold ├─ list_failed_jobs/0 - retrying/cancelled └─ list_recent_completions/1 - last N completions Metrics (Module) └─ calculate_all/0 - aggregates all metrics Events (Module) └─ broadcast_job_event/3 - PubSub broadcasting Workers (DevicePollerWorker, DiscoveryWorker) └─ Broadcast lifecycle events on state changes ``` ### Performance Considerations - Queries limited to polling/discovery workers only - Recent events capped at 100 items - Metrics calculation runs every 10s (not per event) - Device context preloaded for active jobs only ## Future Enhancements - Diagnostic modal with full job details - Filtering by organization/device/queue - Job cancellation from UI - Historical trends and sparklines - Alerting when thresholds exceeded ``` **Step 2: Commit** ```bash git add docs/features/job-monitoring-dashboard.md git commit -m "docs: add job monitoring dashboard feature documentation" ``` --- ## Final Checklist Before marking complete, verify: - [ ] All tests pass: `mix test` - [ ] Code is formatted: `mix format` - [ ] No compilation warnings: `mix compile --warnings-as-errors` - [ ] Credo passes: `mix credo --strict` - [ ] Manual testing completed - [ ] Dashboard accessible at `/admin/monitoring` - [ ] Real-time updates work via PubSub - [ ] Metrics refresh every 10 seconds - [ ] Problems section appears/disappears correctly - [ ] All commits follow conventional commit format - [ ] Feature documentation written ## Success Criteria ✅ Dashboard shows all active polling and discovery jobs ✅ Real-time updates via PubSub when jobs start/complete/fail ✅ Health metrics display and refresh automatically ✅ Problems section detects stuck and failed jobs ✅ Recent activity timeline shows last 100 events ✅ Accessible from admin dashboard ✅ Superadmin-only access enforced ✅ All tests passing