+ Query historical sensor readings, interface traffic rates, and monitoring check results.
+ All time-series queries support a
+
+ timeRange
+
+ argument (e.g. "1h", "24h", "72h").
+
+ List all SNMP sensors discovered on a device. +
+| + Argument + | +Type | ++ Description + | +
|---|---|---|
| deviceId | +ID! | +Required device ID | +
<%= raw(~S[{
+ deviceSensors(deviceId: "device-uuid") {
+ id
+ sensorType
+ sensorUnit
+ sensorDescr
+ currentValue
+ lastCheckedAt
+ monitored
+ }
+}]) %>+ Get historical readings for a specific sensor. +
+| + Argument + | +Type | ++ Description + | +
|---|---|---|
| sensorId | +ID! | +Required sensor ID | +
| timeRange | +String | +e.g. "1h", "24h", "72h" (default: "24h") | +
<%= raw(~S[{
+ sensorReadings(sensorId: "sensor-uuid", timeRange: "24h") {
+ sensorId
+ value
+ status
+ checkedAt
+ }
+}]) %>+ Get traffic rates (bits per second) computed from SNMP counter deltas for a network interface. +
+| + Argument + | +Type | ++ Description + | +
|---|---|---|
| interfaceId | +ID! | +Required interface ID | +
| timeRange | +String | +e.g. "1h", "24h", "72h" (default: "24h") | +
<%= raw(~S[{
+ interfaceTraffic(interfaceId: "interface-uuid", timeRange: "24h") {
+ timestamp
+ inBps
+ outBps
+ inErrors
+ outErrors
+ }
+}]) %>+ Get historical results for a monitoring check (ping, HTTP, etc.). +
+| + Argument + | +Type | ++ Description + | +
|---|---|---|
| checkId | +ID! | +Required check ID | +
| timeRange | +String | +e.g. "1h", "24h" (default: "1h") | +
<%= raw(~S[{
+ checkResults(checkId: "check-uuid", timeRange: "1h") {
+ checkId
+ value
+ status
+ checkedAt
+ responseTimeMs
+ }
+}]) %>
+ Subscribe to real-time events via WebSocket. Connect to
+
+ wss://app.towerops.net/socket/graphql
+
+ with your API token as a connection parameter.
+
+ Pass your API token in the connection parameters when establishing the WebSocket: +
+<%= raw(~S[// JavaScript example using @absinthe/socket
+import { Socket as PhoenixSocket } from "phoenix";
+import * as AbsintheSocket from "@absinthe/socket";
+
+const phoenixSocket = new PhoenixSocket(
+ "wss://app.towerops.net/socket/graphql",
+ { params: { token: "your-api-token" } }
+);
+const absintheSocket = AbsintheSocket.create(phoenixSocket);]) %>+ Subscribe to device status changes. Optionally filter by a specific device ID, or omit to receive all device events in your organization. +
+| + Argument + | +Type | ++ Description + | +
|---|---|---|
| deviceId | +ID | +Optional — filter to a single device | +
<%= raw(~S[subscription {
+ deviceStatusChanged(deviceId: "device-uuid") {
+ deviceId
+ deviceName
+ status
+ changedAt
+ }
+}
+
+# Or subscribe to all devices in your organization:
+subscription {
+ deviceStatusChanged {
+ deviceId
+ deviceName
+ status
+ changedAt
+ }
+}]) %>+ Subscribe to alert lifecycle events (created, acknowledged, resolved) for your organization. +
+<%= raw(~S[subscription {
+ alertEvent {
+ alertId
+ alertType
+ severity
+ message
+ deviceId
+ eventType
+ triggeredAt
+ }
+}]) %>+ Subscribe to live sensor reading updates for a specific device. Delivers batched readings as they are polled. +
+| + Argument + | +Type | ++ Description + | +
|---|---|---|
| deviceId | +ID! | +Required device ID | +
<%= raw(~S[subscription {
+ sensorReadingsUpdated(deviceId: "device-uuid") {
+ deviceId
+ deviceName
+ readings {
+ sensorId
+ sensorType
+ sensorDescr
+ sensorUnit
+ value
+ checkedAt
+ }
+ }
+}]) %>+ id, sensorType, sensorUnit, sensorDescr, currentValue, lastCheckedAt, monitored +
+ ++ sensorId, value, status, checkedAt +
+ ++ timestamp, inBps (Float), outBps (Float), inErrors (Int), outErrors (Int) +
+ ++ checkId, value, status, checkedAt, responseTimeMs +
+ ++ deviceId, deviceName, status, changedAt +
+ ++ alertId, alertType, severity, message, deviceId, eventType, triggeredAt +
+ ++ deviceId, deviceName, readings (list of SensorReadingSnapshot: sensorId, sensorType, sensorDescr, sensorUnit, value, checkedAt) +
+DeleteResult: diff --git a/lib/towerops_web/endpoint.ex b/lib/towerops_web/endpoint.ex index 074deb32..36c3530d 100644 --- a/lib/towerops_web/endpoint.ex +++ b/lib/towerops_web/endpoint.ex @@ -1,5 +1,6 @@ defmodule ToweropsWeb.Endpoint do use Phoenix.Endpoint, otp_app: :towerops + use Absinthe.Phoenix.Endpoint # The session will be stored in the cookie and signed, # this means its contents can be read but not tampered with. @@ -20,6 +21,10 @@ defmodule ToweropsWeb.Endpoint do websocket: [connect_info: [:peer_data]], longpoll: false + socket "/socket/graphql", ToweropsWeb.GraphQLSocket, + websocket: true, + longpoll: false + # Serve at "/" the static files from "priv/static" directory. # # When code reloading is disabled (e.g., in production), diff --git a/lib/towerops_web/graphql/resolvers/time_series.ex b/lib/towerops_web/graphql/resolvers/time_series.ex new file mode 100644 index 00000000..8751de17 --- /dev/null +++ b/lib/towerops_web/graphql/resolvers/time_series.ex @@ -0,0 +1,172 @@ +defmodule ToweropsWeb.GraphQL.Resolvers.TimeSeries do + @moduledoc "GraphQL resolvers for time-series queries." + + import Ecto.Query, warn: false + + alias Towerops.Devices.Device + alias Towerops.Monitoring + alias Towerops.Monitoring.Check + alias Towerops.Repo + alias Towerops.Snmp + alias Towerops.Snmp.Device, as: SnmpDevice + alias Towerops.Snmp.Interface + alias Towerops.Snmp.Sensor + alias ToweropsWeb.GraphQL.Resolvers.Helpers + alias ToweropsWeb.ScopedResource + + def device_sensors(_parent, %{device_id: device_id}, %{context: %{organization_id: org_id}}) do + with {:ok, device} <- fetch_org_device(device_id, org_id) do + device = Repo.preload(device, :snmp_device) + + sensors = + case device.snmp_device do + nil -> [] + snmp_device -> Snmp.list_sensors(snmp_device.id) + end + + {:ok, sensors} + end + end + + def device_sensors(_parent, _args, _resolution), do: Helpers.authentication_error() + + def sensor_readings(_parent, %{sensor_id: sensor_id} = args, %{context: %{organization_id: org_id}}) do + with {:ok, _sensor} <- verify_sensor_access(sensor_id, org_id) do + hours = parse_time_range(Map.get(args, :time_range, "24h")) + since = DateTime.add(DateTime.utc_now(), -hours * 3600, :second) + readings = Snmp.get_sensor_readings(sensor_id, since: since) + {:ok, readings} + end + end + + def sensor_readings(_parent, _args, _resolution), do: Helpers.authentication_error() + + def interface_traffic(_parent, %{interface_id: interface_id} = args, %{context: %{organization_id: org_id}}) do + with {:ok, _interface} <- verify_interface_access(interface_id, org_id) do + hours = parse_time_range(Map.get(args, :time_range, "24h")) + since = DateTime.add(DateTime.utc_now(), -hours * 3600, :second) + stats = Snmp.get_interface_stats(interface_id, since: since) + {:ok, compute_traffic_rates(stats)} + end + end + + def interface_traffic(_parent, _args, _resolution), do: Helpers.authentication_error() + + def check_results(_parent, %{check_id: check_id} = args, %{context: %{organization_id: org_id}}) do + with {:ok, _check} <- verify_check_access(check_id, org_id) do + hours = parse_time_range(Map.get(args, :time_range, "1h")) + from_time = DateTime.add(DateTime.utc_now(), -hours * 3600, :second) + results = Monitoring.get_check_results(check_id, from: from_time) + + formatted = + Enum.map(results, fn r -> + %{ + check_id: r.check_id, + value: r.value, + status: r.status, + checked_at: to_string(r.checked_at), + response_time_ms: r.response_time_ms + } + end) + + {:ok, formatted} + end + end + + def check_results(_parent, _args, _resolution), do: Helpers.authentication_error() + + # Org scoping helpers + + defp fetch_org_device(id, org_id) do + case ScopedResource.fetch(Device, id, org_id) do + {:ok, device} -> {:ok, device} + {:error, _reason} -> {:error, "Device not found"} + end + end + + defp verify_sensor_access(sensor_id, org_id) do + query = + from(s in Sensor, + join: sd in SnmpDevice, + on: s.snmp_device_id == sd.id, + join: d in Device, + on: sd.device_id == d.id, + where: s.id == ^sensor_id and d.organization_id == ^org_id + ) + + case Repo.one(query) do + nil -> {:error, "Sensor not found"} + sensor -> {:ok, sensor} + end + end + + defp verify_interface_access(interface_id, org_id) do + query = + from(i in Interface, + join: sd in SnmpDevice, + on: i.snmp_device_id == sd.id, + join: d in Device, + on: sd.device_id == d.id, + where: i.id == ^interface_id and d.organization_id == ^org_id + ) + + case Repo.one(query) do + nil -> {:error, "Interface not found"} + interface -> {:ok, interface} + end + end + + defp verify_check_access(check_id, org_id) do + query = + from(c in Check, + where: c.id == ^check_id and c.organization_id == ^org_id + ) + + case Repo.one(query) do + nil -> {:error, "Check not found"} + check -> {:ok, check} + end + end + + # Compute bps rates from counter deltas (same pattern as GraphLive) + defp compute_traffic_rates(stats) do + # Stats come back ordered desc by checked_at, reverse for chronological order + sorted = Enum.reverse(stats) + + sorted + |> Enum.chunk_every(2, 1, :discard) + |> Enum.map(fn [stat1, stat2] -> + time_diff = stat2.checked_at |> DateTime.diff(stat1.checked_at, :second) |> max(1) + + in_bps = compute_bps(stat1.if_in_octets, stat2.if_in_octets, time_diff) + out_bps = compute_bps(stat1.if_out_octets, stat2.if_out_octets, time_diff) + + in_errors = (stat2.if_in_errors || 0) - (stat1.if_in_errors || 0) + out_errors = (stat2.if_out_errors || 0) - (stat1.if_out_errors || 0) + + %{ + timestamp: to_string(stat2.checked_at), + in_bps: in_bps, + out_bps: out_bps, + in_errors: max(in_errors, 0), + out_errors: max(out_errors, 0) + } + end) + end + + defp compute_bps(nil, _, _time_diff), do: 0.0 + defp compute_bps(_, nil, _time_diff), do: 0.0 + + defp compute_bps(octets1, octets2, time_diff) do + ((octets2 - octets1) * 8 / time_diff) + |> max(0) + |> Float.round(2) + end + + defp parse_time_range(range) do + case Integer.parse(String.replace(range, "h", "")) do + {hours, _} -> hours + :error -> 24 + end + end +end diff --git a/lib/towerops_web/graphql/schema.ex b/lib/towerops_web/graphql/schema.ex index f53ddd9a..5178fc18 100644 --- a/lib/towerops_web/graphql/schema.ex +++ b/lib/towerops_web/graphql/schema.ex @@ -19,6 +19,7 @@ defmodule ToweropsWeb.GraphQL.Schema do import_types(ToweropsWeb.GraphQL.Types.Schedule) import_types(ToweropsWeb.GraphQL.Types.EscalationPolicy) import_types(ToweropsWeb.GraphQL.Types.MaintenanceWindow) + import_types(ToweropsWeb.GraphQL.Types.TimeSeries) query do # Devices @@ -103,6 +104,30 @@ defmodule ToweropsWeb.GraphQL.Schema do resolve(&ToweropsWeb.GraphQL.Resolvers.Device.interfaces/3) end + # Time-series queries + field :device_sensors, list_of(:sensor) do + arg(:device_id, non_null(:id)) + resolve(&ToweropsWeb.GraphQL.Resolvers.TimeSeries.device_sensors/3) + end + + field :sensor_readings, list_of(:sensor_reading) do + arg(:sensor_id, non_null(:id)) + arg(:time_range, :string, default_value: "24h") + resolve(&ToweropsWeb.GraphQL.Resolvers.TimeSeries.sensor_readings/3) + end + + field :interface_traffic, list_of(:interface_traffic_point) do + arg(:interface_id, non_null(:id)) + arg(:time_range, :string, default_value: "24h") + resolve(&ToweropsWeb.GraphQL.Resolvers.TimeSeries.interface_traffic/3) + end + + field :check_results, list_of(:check_result_point) do + arg(:check_id, non_null(:id)) + arg(:time_range, :string, default_value: "1h") + resolve(&ToweropsWeb.GraphQL.Resolvers.TimeSeries.check_results/3) + end + # Schedules field :schedules, list_of(:schedule) do resolve(&ToweropsWeb.GraphQL.Resolvers.Schedule.list/3) @@ -372,4 +397,33 @@ defmodule ToweropsWeb.GraphQL.Schema do resolve(&ToweropsWeb.GraphQL.Resolvers.MaintenanceWindow.delete/3) end end + + subscription do + field :device_status_changed, :device_status_event do + arg(:device_id, :id) + + config(fn args, %{context: %{organization_id: org_id}} -> + topic = + if args[:device_id], + do: "gql:device_status:#{org_id}:#{args.device_id}", + else: "gql:device_status:#{org_id}" + + {:ok, topic: topic} + end) + end + + field :alert_event, :alert_event do + config(fn _args, %{context: %{organization_id: org_id}} -> + {:ok, topic: "gql:alerts:#{org_id}"} + end) + end + + field :sensor_readings_updated, :sensor_readings_event do + arg(:device_id, non_null(:id)) + + config(fn %{device_id: device_id}, %{context: %{organization_id: org_id}} -> + {:ok, topic: "gql:sensor_readings:#{org_id}:#{device_id}"} + end) + end + end end diff --git a/lib/towerops_web/graphql/subscriptions.ex b/lib/towerops_web/graphql/subscriptions.ex new file mode 100644 index 00000000..53d145e0 --- /dev/null +++ b/lib/towerops_web/graphql/subscriptions.ex @@ -0,0 +1,71 @@ +defmodule ToweropsWeb.GraphQL.Subscriptions do + @moduledoc """ + Publishes events to GraphQL subscriptions. + + Each function builds the subscription payload and calls + `Absinthe.Subscription.publish/3` to push to connected subscribers. + """ + + @doc """ + Publishes a device status change event. + + Publishes to both org-wide and device-specific subscription topics. + """ + def publish_device_status(device_id, org_id, %{device_name: device_name, status: status}) do + payload = %{ + device_id: device_id, + device_name: device_name, + status: status, + changed_at: to_string(DateTime.utc_now()) + } + + Absinthe.Subscription.publish( + ToweropsWeb.Endpoint, + payload, + device_status_changed: "gql:device_status:#{org_id}", + device_status_changed: "gql:device_status:#{org_id}:#{device_id}" + ) + end + + @doc """ + Publishes an alert lifecycle event (created, acknowledged, resolved). + + Accepts both Alert structs and plain maps. + """ + def publish_alert_event(alert, event_type) do + payload = %{ + alert_id: Map.get(alert, :id), + alert_type: Map.get(alert, :alert_type), + severity: Map.get(alert, :severity), + message: Map.get(alert, :message), + device_id: Map.get(alert, :device_id), + event_type: event_type, + triggered_at: to_string(Map.get(alert, :triggered_at)) + } + + org_id = Map.get(alert, :organization_id) + + Absinthe.Subscription.publish( + ToweropsWeb.Endpoint, + payload, + alert_event: "gql:alerts:#{org_id}" + ) + end + + @doc """ + Publishes a batch of sensor readings for a device. + """ + def publish_sensor_readings(device_id, org_id, device_name, readings) do + payload = %{ + device_id: device_id, + device_name: device_name, + readings: readings + } + + Absinthe.Subscription.publish( + ToweropsWeb.Endpoint, + payload, + sensor_readings_updated: "gql:sensor_readings:#{org_id}:#{device_id}" + ) + end +end diff --git a/lib/towerops_web/graphql/types/time_series.ex b/lib/towerops_web/graphql/types/time_series.ex new file mode 100644 index 00000000..bde465bb --- /dev/null +++ b/lib/towerops_web/graphql/types/time_series.ex @@ -0,0 +1,69 @@ +defmodule ToweropsWeb.GraphQL.Types.TimeSeries do + @moduledoc "GraphQL types for time-series data: sensors, readings, interface traffic, and check results." + use Absinthe.Schema.Notation + + object :sensor do + field :id, :id + field :sensor_type, :string + field :sensor_unit, :string + field :sensor_descr, :string + field :current_value, :float, resolve: fn sensor, _, _ -> {:ok, sensor.last_value} end + field :last_checked_at, :string, resolve: fn sensor, _, _ -> {:ok, to_string(sensor.last_checked_at)} end + field :monitored, :boolean + end + + object :sensor_reading do + field :sensor_id, :id + field :value, :float + field :status, :string + field :checked_at, :string, resolve: fn reading, _, _ -> {:ok, to_string(reading.checked_at)} end + end + + object :interface_traffic_point do + field :timestamp, :string + field :in_bps, :float + field :out_bps, :float + field :in_errors, :integer + field :out_errors, :integer + end + + object :check_result_point do + field :check_id, :id + field :value, :float + field :status, :integer + field :checked_at, :string + field :response_time_ms, :float + end + + object :device_status_event do + field :device_id, :id + field :device_name, :string + field :status, :string + field :changed_at, :string + end + + object :alert_event do + field :alert_id, :id + field :alert_type, :string + field :severity, :integer + field :message, :string + field :device_id, :id + field :event_type, :string + field :triggered_at, :string + end + + object :sensor_readings_event do + field :device_id, :id + field :device_name, :string + field :readings, list_of(:sensor_reading_snapshot) + end + + object :sensor_reading_snapshot do + field :sensor_id, :id + field :sensor_type, :string + field :sensor_descr, :string + field :sensor_unit, :string + field :value, :float + field :checked_at, :string + end +end diff --git a/lib/towerops_web/graphql_socket.ex b/lib/towerops_web/graphql_socket.ex new file mode 100644 index 00000000..af979b1b --- /dev/null +++ b/lib/towerops_web/graphql_socket.ex @@ -0,0 +1,32 @@ +defmodule ToweropsWeb.GraphQLSocket do + @moduledoc """ + WebSocket endpoint for GraphQL subscriptions. + + Authenticates via API token passed as a connection parameter. + """ + + use Phoenix.Socket + use Absinthe.Phoenix.Socket, schema: ToweropsWeb.GraphQL.Schema + + @impl true + def connect(%{"token" => token}, socket, _connect_info) when is_binary(token) do + case Towerops.ApiTokens.verify_token(token) do + {:ok, org_id, user} -> + socket = + socket + |> assign(:organization_id, org_id) + |> assign(:user, user) + |> Absinthe.Phoenix.Socket.put_options(context: %{organization_id: org_id, user: user}) + + {:ok, socket} + + {:error, :invalid_token} -> + :error + end + end + + def connect(_params, _socket, _connect_info), do: :error + + @impl true + def id(socket), do: "graphql_socket:#{socket.assigns.organization_id}" +end diff --git a/mix.exs b/mix.exs index 337b62f6..a780a71f 100644 --- a/mix.exs +++ b/mix.exs @@ -80,6 +80,7 @@ defmodule Towerops.MixProject do {:mox, "~> 1.0", only: :test}, {:absinthe, "~> 1.7"}, {:absinthe_plug, "~> 1.5"}, + {:absinthe_phoenix, "~> 2.0"}, {:honeybadger, "~> 0.24"}, {:error_tracker, "~> 0.7"}, {:error_tracker_notifier, "~> 0.2"}, diff --git a/mix.lock b/mix.lock index 47825efb..589b7e0c 100644 --- a/mix.lock +++ b/mix.lock @@ -1,5 +1,6 @@ %{ "absinthe": {:hex, :absinthe, "1.9.0", "28f11753d01c0e8b6cb6e764a23cf4081e0e6cae88f53f4c9e4320912aee9c07", [:mix], [{:dataloader, "~> 1.0.0 or ~> 2.0", [hex: :dataloader, repo: "hexpm", optional: true]}, {:decimal, "~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}, {:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}, {:opentelemetry_process_propagator, "~> 0.2.1 or ~> 0.3", [hex: :opentelemetry_process_propagator, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "db65993420944ad90e932827663d4ab704262b007d4e3900cd69615f14ccc8ce"}, + "absinthe_phoenix": {:hex, :absinthe_phoenix, "2.0.4", "f36999412fbd6a2339abb5b7e24a4cc9492bbc7909d5806deeef83b06f55c508", [:mix], [{:absinthe, "~> 1.5", [hex: :absinthe, repo: "hexpm", optional: false]}, {:absinthe_plug, "~> 1.5", [hex: :absinthe_plug, repo: "hexpm", optional: false]}, {:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.5", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.13 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}], "hexpm", "66617ee63b725256ca16264364148b10b19e2ecb177488cd6353584f2e6c1cf3"}, "absinthe_plug": {:hex, :absinthe_plug, "1.5.9", "4f66fd46aecf969b349dd94853e6132db6d832ae6a4b951312b6926ad4ee7ca3", [:mix], [{:absinthe, "~> 1.7", [hex: :absinthe, repo: "hexpm", optional: false]}, {:plug, "~> 1.4", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "dcdc84334b0e9e2cd439bd2653678a822623f212c71088edf0a4a7d03f1fa225"}, "argon2_elixir": {:hex, :argon2_elixir, "4.1.3", "4f28318286f89453364d7fbb53e03d4563fd7ed2438a60237eba5e426e97785f", [:make, :mix], [{:comeonin, "~> 5.3", [hex: :comeonin, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.6", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "7c295b8d8e0eaf6f43641698f962526cdf87c6feb7d14bd21e599271b510608c"}, "bandit": {:hex, :bandit, "1.10.3", "1e5d168fa79ec8de2860d1b4d878d97d4fbbe2fdbe7b0a7d9315a4359d1d4bb9", [:mix], [{:hpax, "~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.18", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.0", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "99a52d909c48db65ca598e1962797659e3c0f1d06e825a50c3d75b74a5e2db18"}, diff --git a/priv/static/changelog.txt b/priv/static/changelog.txt index 99fc89bf..22e54bf6 100644 --- a/priv/static/changelog.txt +++ b/priv/static/changelog.txt @@ -1,3 +1,8 @@ +2026-03-12 — Real-Time GraphQL API +* New time-series queries for sensor readings, interface traffic, and check results +* GraphQL subscriptions for live device status, alerts, and sensor reading updates +* WebSocket support for real-time data streaming to custom dashboards and integrations + 2026-03-09 — User Interface Improvements * Improved button visibility across the application * Sign up and account creation actions now use prominent button styling diff --git a/test/towerops_web/graphql/resolvers/time_series_test.exs b/test/towerops_web/graphql/resolvers/time_series_test.exs new file mode 100644 index 00000000..57aa1f0c --- /dev/null +++ b/test/towerops_web/graphql/resolvers/time_series_test.exs @@ -0,0 +1,615 @@ +defmodule ToweropsWeb.GraphQL.Resolvers.TimeSeriesTest do + use ToweropsWeb.ConnCase, async: true + + import Towerops.AccountsFixtures + import Towerops.DevicesFixtures + import Towerops.OrganizationsFixtures + import Towerops.SnmpFixtures + + alias Towerops.ApiTokens + alias Towerops.Monitoring + alias Towerops.Repo + alias Towerops.Snmp + alias Towerops.Snmp.Interface + alias Towerops.Snmp.Sensor + + setup do + user = user_fixture() + organization = organization_fixture(user.id) + {:ok, site} = Towerops.Sites.create_site(%{name: "Test Site", organization_id: organization.id}) + device = device_fixture(organization.id, site.id, %{name: "Test Router"}) + snmp_device = snmp_device_fixture(%{device: device}) + + {:ok, {_token, raw_token}} = + ApiTokens.create_api_token(%{ + organization_id: organization.id, + user_id: user.id, + name: "GraphQL Test Token" + }) + + %{ + user: user, + organization: organization, + device: device, + snmp_device: snmp_device, + raw_token: raw_token + } + end + + defp graphql_query(conn, query, variables) do + conn + |> put_req_header("content-type", "application/json") + |> post("/api/graphql", Jason.encode!(%{query: query, variables: variables})) + |> json_response(200) + end + + defp auth_conn(conn, raw_token) do + put_req_header(conn, "authorization", "Bearer #{raw_token}") + end + + describe "device_sensors query" do + test "returns sensors for a device", %{ + conn: conn, + device: device, + snmp_device: snmp_device, + raw_token: raw_token + } do + _sensor = + %Sensor{} + |> Sensor.changeset(%{ + snmp_device_id: snmp_device.id, + sensor_type: "temperature", + sensor_index: "1", + sensor_oid: "1.3.6.1.4.1.9.1.1", + sensor_descr: "CPU Temp", + sensor_unit: "°C", + monitored: true + }) + |> Repo.insert!() + + query = """ + query($deviceId: ID!) { + deviceSensors(deviceId: $deviceId) { + id + sensorType + sensorUnit + sensorDescr + monitored + } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"deviceId" => device.id}) + + assert %{"data" => %{"deviceSensors" => sensors}} = result + assert length(sensors) == 1 + [sensor] = sensors + assert sensor["sensorType"] == "temperature" + assert sensor["sensorUnit"] == "°C" + assert sensor["sensorDescr"] == "CPU Temp" + assert sensor["monitored"] == true + end + + test "requires authentication", %{conn: conn, device: device} do + query = """ + query($deviceId: ID!) { + deviceSensors(deviceId: $deviceId) { id } + } + """ + + conn = + conn + |> put_req_header("content-type", "application/json") + |> post("/api/graphql", Jason.encode!(%{query: query, variables: %{"deviceId" => device.id}})) + + assert json_response(conn, 401)["error"] =~ "Authorization" + end + + test "returns error for device in different org", %{ + conn: conn, + raw_token: raw_token + } do + other_user = user_fixture() + other_org = organization_fixture(other_user.id) + {:ok, other_site} = Towerops.Sites.create_site(%{name: "Other Site", organization_id: other_org.id}) + other_device = device_fixture(other_org.id, other_site.id) + + query = """ + query($deviceId: ID!) { + deviceSensors(deviceId: $deviceId) { id } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"deviceId" => other_device.id}) + + assert %{"errors" => [%{"message" => "Device not found"}]} = result + end + end + + describe "sensor_readings query" do + test "returns sensor readings", %{ + conn: conn, + snmp_device: snmp_device, + raw_token: raw_token + } do + sensor = + %Sensor{} + |> Sensor.changeset(%{ + snmp_device_id: snmp_device.id, + sensor_type: "temperature", + sensor_index: "1", + sensor_oid: "1.3.6.1.4.1.9.1.2", + sensor_descr: "CPU Temp", + sensor_unit: "°C" + }) + |> Repo.insert!() + + now = DateTime.truncate(DateTime.utc_now(), :second) + + {:ok, _} = + Snmp.create_sensor_reading(%{ + sensor_id: sensor.id, + value: 42.5, + status: "ok", + checked_at: now + }) + + {:ok, _} = + Snmp.create_sensor_reading(%{ + sensor_id: sensor.id, + value: 43.0, + status: "ok", + checked_at: DateTime.add(now, -60, :second) + }) + + query = """ + query($sensorId: ID!, $timeRange: String) { + sensorReadings(sensorId: $sensorId, timeRange: $timeRange) { + sensorId + value + status + checkedAt + } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"sensorId" => sensor.id, "timeRange" => "24h"}) + + assert %{"data" => %{"sensorReadings" => readings}} = result + assert length(readings) == 2 + [first | _] = readings + assert first["value"] == 42.5 + assert first["status"] == "ok" + end + + test "returns empty list when no readings exist", %{ + conn: conn, + snmp_device: snmp_device, + raw_token: raw_token + } do + sensor = + %Sensor{} + |> Sensor.changeset(%{ + snmp_device_id: snmp_device.id, + sensor_type: "temperature", + sensor_index: "1", + sensor_oid: "1.3.6.1.4.1.9.1.3", + sensor_descr: "CPU Temp", + sensor_unit: "°C" + }) + |> Repo.insert!() + + query = """ + query($sensorId: ID!, $timeRange: String) { + sensorReadings(sensorId: $sensorId, timeRange: $timeRange) { + value + } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"sensorId" => sensor.id, "timeRange" => "24h"}) + + assert %{"data" => %{"sensorReadings" => []}} = result + end + + test "returns error for sensor in different org", %{ + conn: conn, + raw_token: raw_token + } do + other_user = user_fixture() + other_org = organization_fixture(other_user.id) + {:ok, other_site} = Towerops.Sites.create_site(%{name: "Other Site", organization_id: other_org.id}) + other_device = device_fixture(other_org.id, other_site.id) + other_snmp_device = snmp_device_fixture(%{device: other_device}) + + other_sensor = + %Sensor{} + |> Sensor.changeset(%{ + snmp_device_id: other_snmp_device.id, + sensor_type: "temperature", + sensor_index: "1", + sensor_oid: "1.3.6.1.4.1.9.1.4", + sensor_descr: "Other Sensor", + sensor_unit: "°C" + }) + |> Repo.insert!() + + query = """ + query($sensorId: ID!) { + sensorReadings(sensorId: $sensorId) { value } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"sensorId" => other_sensor.id}) + + assert %{"errors" => [%{"message" => "Sensor not found"}]} = result + end + + test "requires authentication", %{conn: conn} do + query = """ + query($sensorId: ID!) { + sensorReadings(sensorId: $sensorId) { value } + } + """ + + conn = + conn + |> put_req_header("content-type", "application/json") + |> post("/api/graphql", Jason.encode!(%{query: query, variables: %{"sensorId" => Ecto.UUID.generate()}})) + + assert json_response(conn, 401)["error"] =~ "Authorization" + end + end + + describe "interface_traffic query" do + test "returns traffic data computed from counter deltas", %{ + conn: conn, + snmp_device: snmp_device, + raw_token: raw_token + } do + interface = + %Interface{} + |> Interface.changeset(%{ + snmp_device_id: snmp_device.id, + if_index: 1, + if_name: "eth0", + if_descr: "Ethernet 0" + }) + |> Repo.insert!() + + now = DateTime.truncate(DateTime.utc_now(), :second) + + # Two data points 60 seconds apart: + # Stat 1: 1000 bytes in, 2000 bytes out + # Stat 2: 2000 bytes in, 4000 bytes out + # => in_bps = (1000 * 8) / 60 = 133.33, out_bps = (2000 * 8) / 60 = 266.67 + {:ok, _} = + Snmp.create_interface_stat(%{ + interface_id: interface.id, + if_in_octets: 1000, + if_out_octets: 2000, + if_in_errors: 0, + if_out_errors: 0, + checked_at: DateTime.add(now, -60, :second) + }) + + {:ok, _} = + Snmp.create_interface_stat(%{ + interface_id: interface.id, + if_in_octets: 2000, + if_out_octets: 4000, + if_in_errors: 1, + if_out_errors: 0, + checked_at: now + }) + + query = """ + query($interfaceId: ID!, $timeRange: String) { + interfaceTraffic(interfaceId: $interfaceId, timeRange: $timeRange) { + timestamp + inBps + outBps + inErrors + outErrors + } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"interfaceId" => interface.id, "timeRange" => "24h"}) + + assert %{"data" => %{"interfaceTraffic" => points}} = result + assert length(points) == 1 + [point] = points + assert_in_delta point["inBps"], 133.33, 0.1 + assert_in_delta point["outBps"], 266.67, 0.1 + assert point["inErrors"] == 1 + assert point["outErrors"] == 0 + end + + test "returns empty list with single data point", %{ + conn: conn, + snmp_device: snmp_device, + raw_token: raw_token + } do + interface = + %Interface{} + |> Interface.changeset(%{ + snmp_device_id: snmp_device.id, + if_index: 2, + if_name: "eth1", + if_descr: "Ethernet 1" + }) + |> Repo.insert!() + + now = DateTime.truncate(DateTime.utc_now(), :second) + + {:ok, _} = + Snmp.create_interface_stat(%{ + interface_id: interface.id, + if_in_octets: 1000, + if_out_octets: 2000, + if_in_errors: 0, + if_out_errors: 0, + checked_at: now + }) + + query = """ + query($interfaceId: ID!, $timeRange: String) { + interfaceTraffic(interfaceId: $interfaceId, timeRange: $timeRange) { + inBps + } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"interfaceId" => interface.id, "timeRange" => "24h"}) + + assert %{"data" => %{"interfaceTraffic" => []}} = result + end + + test "returns error for interface in different org", %{ + conn: conn, + raw_token: raw_token + } do + other_user = user_fixture() + other_org = organization_fixture(other_user.id) + {:ok, other_site} = Towerops.Sites.create_site(%{name: "Other Site", organization_id: other_org.id}) + other_device = device_fixture(other_org.id, other_site.id) + other_snmp_device = snmp_device_fixture(%{device: other_device}) + + other_interface = + %Interface{} + |> Interface.changeset(%{ + snmp_device_id: other_snmp_device.id, + if_index: 1, + if_name: "eth0", + if_descr: "Ethernet 0" + }) + |> Repo.insert!() + + query = """ + query($interfaceId: ID!) { + interfaceTraffic(interfaceId: $interfaceId) { inBps } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"interfaceId" => other_interface.id}) + + assert %{"errors" => [%{"message" => "Interface not found"}]} = result + end + + test "requires authentication", %{conn: conn} do + query = """ + query($interfaceId: ID!) { + interfaceTraffic(interfaceId: $interfaceId) { inBps } + } + """ + + conn = + conn + |> put_req_header("content-type", "application/json") + |> post("/api/graphql", Jason.encode!(%{query: query, variables: %{"interfaceId" => Ecto.UUID.generate()}})) + + assert json_response(conn, 401)["error"] =~ "Authorization" + end + end + + describe "check_results query" do + test "returns check results", %{ + conn: conn, + device: device, + organization: organization, + raw_token: raw_token + } do + {:ok, check} = + Monitoring.create_check(%{ + name: "Ping Check", + check_type: "ping", + organization_id: organization.id, + device_id: device.id, + config: %{"host" => "10.0.0.1"} + }) + + now = DateTime.truncate(DateTime.utc_now(), :second) + + {:ok, _} = + Monitoring.create_check_result(%{ + check_id: check.id, + organization_id: organization.id, + checked_at: now, + status: 0, + value: 10.5, + response_time_ms: 10.5 + }) + + query = """ + query($checkId: ID!, $timeRange: String) { + checkResults(checkId: $checkId, timeRange: $timeRange) { + checkId + value + status + checkedAt + responseTimeMs + } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"checkId" => check.id, "timeRange" => "1h"}) + + assert %{"data" => %{"checkResults" => results}} = result + assert length(results) == 1 + [res] = results + assert res["checkId"] == check.id + assert res["value"] == 10.5 + assert res["status"] == 0 + assert res["responseTimeMs"] == 10.5 + end + + test "returns empty list when no results exist", %{ + conn: conn, + device: device, + organization: organization, + raw_token: raw_token + } do + {:ok, check} = + Monitoring.create_check(%{ + name: "Empty Check", + check_type: "ping", + organization_id: organization.id, + device_id: device.id, + config: %{"host" => "10.0.0.2"} + }) + + query = """ + query($checkId: ID!, $timeRange: String) { + checkResults(checkId: $checkId, timeRange: $timeRange) { + value + } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"checkId" => check.id, "timeRange" => "1h"}) + + assert %{"data" => %{"checkResults" => []}} = result + end + + test "returns error for check in different org", %{ + conn: conn, + raw_token: raw_token + } do + other_user = user_fixture() + other_org = organization_fixture(other_user.id) + {:ok, other_site} = Towerops.Sites.create_site(%{name: "Other Site", organization_id: other_org.id}) + other_device = device_fixture(other_org.id, other_site.id) + + {:ok, other_check} = + Monitoring.create_check(%{ + name: "Other Check", + check_type: "ping", + organization_id: other_org.id, + device_id: other_device.id, + config: %{"host" => "10.0.0.99"} + }) + + query = """ + query($checkId: ID!) { + checkResults(checkId: $checkId) { value } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"checkId" => other_check.id}) + + assert %{"errors" => [%{"message" => "Check not found"}]} = result + end + + test "requires authentication", %{conn: conn} do + query = """ + query($checkId: ID!) { + checkResults(checkId: $checkId) { value } + } + """ + + conn = + conn + |> put_req_header("content-type", "application/json") + |> post("/api/graphql", Jason.encode!(%{query: query, variables: %{"checkId" => Ecto.UUID.generate()}})) + + assert json_response(conn, 401)["error"] =~ "Authorization" + end + end + + describe "device_sensors edge cases" do + test "returns empty list for device without SNMP device", %{ + conn: conn, + organization: organization, + raw_token: raw_token + } do + {:ok, site} = Towerops.Sites.create_site(%{name: "No SNMP Site", organization_id: organization.id}) + bare_device = device_fixture(organization.id, site.id, %{name: "No SNMP Router"}) + + query = """ + query($deviceId: ID!) { + deviceSensors(deviceId: $deviceId) { id } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"deviceId" => bare_device.id}) + + assert %{"data" => %{"deviceSensors" => []}} = result + end + + test "returns empty list for device with SNMP device but no sensors", %{ + conn: conn, + device: device, + raw_token: raw_token + } do + query = """ + query($deviceId: ID!) { + deviceSensors(deviceId: $deviceId) { id } + } + """ + + result = + conn + |> auth_conn(raw_token) + |> graphql_query(query, %{"deviceId" => device.id}) + + assert %{"data" => %{"deviceSensors" => []}} = result + end + end +end diff --git a/test/towerops_web/graphql/subscriptions_test.exs b/test/towerops_web/graphql/subscriptions_test.exs new file mode 100644 index 00000000..06492b0a --- /dev/null +++ b/test/towerops_web/graphql/subscriptions_test.exs @@ -0,0 +1,68 @@ +defmodule ToweropsWeb.GraphQL.SubscriptionsTest do + use Towerops.DataCase, async: true + + alias ToweropsWeb.GraphQL.Subscriptions + + describe "publish_device_status/3" do + test "publishes to both org-wide and device-specific topics" do + device_id = Ecto.UUID.generate() + org_id = Ecto.UUID.generate() + + # Subscribe to org-wide topic + Phoenix.PubSub.subscribe(Towerops.PubSub, "__absinthe__:doc:gql:device_status:#{org_id}") + # Subscribe to device-specific topic + Phoenix.PubSub.subscribe(Towerops.PubSub, "__absinthe__:doc:gql:device_status:#{org_id}:#{device_id}") + + Subscriptions.publish_device_status(device_id, org_id, %{ + device_name: "Router 1", + status: "up" + }) + + # The publish call itself should succeed without error + # Actual subscription delivery requires a full Absinthe subscription setup + # which is tested in integration tests. Here we verify the function runs + # without crashing. + assert true + end + end + + describe "publish_alert_event/2" do + test "publishes alert events without error" do + alert = %{ + id: Ecto.UUID.generate(), + alert_type: "device_down", + severity: 2, + message: "Device is unreachable", + device_id: Ecto.UUID.generate(), + organization_id: Ecto.UUID.generate(), + triggered_at: DateTime.utc_now() + } + + # Should not raise + Subscriptions.publish_alert_event(alert, "created") + assert true + end + end + + describe "publish_sensor_readings/4" do + test "publishes sensor reading events without error" do + device_id = Ecto.UUID.generate() + org_id = Ecto.UUID.generate() + + readings = [ + %{ + sensor_id: Ecto.UUID.generate(), + sensor_type: "temperature", + sensor_descr: "CPU Temp", + sensor_unit: "°C", + value: 42.5, + checked_at: to_string(DateTime.utc_now()) + } + ] + + # Should not raise + Subscriptions.publish_sensor_readings(device_id, org_id, "Router 1", readings) + assert true + end + end +end diff --git a/test/towerops_web/graphql_socket_test.exs b/test/towerops_web/graphql_socket_test.exs new file mode 100644 index 00000000..fafb7568 --- /dev/null +++ b/test/towerops_web/graphql_socket_test.exs @@ -0,0 +1,66 @@ +defmodule ToweropsWeb.GraphQLSocketTest do + use Towerops.DataCase, async: true + + import Towerops.AccountsFixtures + import Towerops.OrganizationsFixtures + + alias Towerops.ApiTokens + alias ToweropsWeb.GraphQLSocket + + describe "connect/3" do + setup do + user = user_fixture() + organization = organization_fixture(user.id) + + {:ok, {_token, raw_token}} = + ApiTokens.create_api_token(%{ + organization_id: organization.id, + user_id: user.id, + name: "GraphQL Test Token" + }) + + %{user: user, organization: organization, raw_token: raw_token} + end + + test "connects with valid API token", %{ + organization: organization, + raw_token: raw_token + } do + socket = %Phoenix.Socket{ + assigns: %{}, + transport: :websocket + } + + assert {:ok, socket} = GraphQLSocket.connect(%{"token" => raw_token}, socket, %{}) + assert socket.assigns.organization_id == organization.id + end + + test "rejects invalid token" do + socket = %Phoenix.Socket{ + assigns: %{}, + transport: :websocket + } + + assert :error = GraphQLSocket.connect(%{"token" => "invalid_token"}, socket, %{}) + end + + test "rejects missing token" do + socket = %Phoenix.Socket{ + assigns: %{}, + transport: :websocket + } + + assert :error = GraphQLSocket.connect(%{}, socket, %{}) + end + end + + describe "id/1" do + test "returns socket ID based on organization_id" do + socket = %Phoenix.Socket{ + assigns: %{organization_id: "some-org-id"} + } + + assert GraphQLSocket.id(socket) == "graphql_socket:some-org-id" + end + end +end