# 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
Real-time monitoring of polling and discovery operations
{length(@stuck_jobs)} stuck jobs
{length(@failed_jobs)} failed jobs
{length(@executing_jobs)} jobs currently executing
Last {length(@recent_events)} events
Job Monitoring
⚠️ Problems Detected
<%= if length(@stuck_jobs) > 0 do %>
Active Operations
Health Metrics
Recent Activity
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 →No jobs currently executing
<% else %><%= 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) %>
<%= 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) %>
<%= 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 %>Executing
{@metrics.executing_count}
Queued
{@metrics.queued_count}
Completed (1h)
{@metrics.completed_last_hour}
Failed (1h)
{@metrics.failed_last_hour}
Updates every 10 seconds
No recent activity
<% else %><%= if job.device do %> <%= job.device.name %> <% else %> Device #{get_in(job.args, ["device_id"])} <% end %>
<%= worker_name(job.worker) %> · <%= event_outcome(job) %>