diff --git a/config/test.exs b/config/test.exs index a279e9ea..1acb5b8c 100644 --- a/config/test.exs +++ b/config/test.exs @@ -34,6 +34,11 @@ config :towerops, Oban, queues: false, plugins: false +# Use mock LLM in tests by default. Tests that exercise the real DeepSeek +# client (e.g. test/towerops/llm/deepseek_test.exs) call the module +# directly and bypass this configuration. +config :towerops, Towerops.LLM, impl: Towerops.LLM.MockClient + # Configure your database # @@ -51,41 +56,21 @@ config :towerops, Towerops.PromEx, grafana: :disabled, metrics_server: :disabled -# MIB upload directory for tests — points at a writable tmp dir so -# -# the MibController upload/delete flows can actually be exercised. -# In CI, DATABASE_URL is set and will be used automatically by Ecto. -# Locally, fall back to individual connection parameters. -config :towerops, :mib_dir, Path.join(System.tmp_dir!(), "towerops-test-mibs") - -if database_url = System.get_env("DATABASE_URL") do - config :towerops, Towerops.Repo, - url: database_url, - pool: Sandbox, - pool_size: System.schedulers_online() * 2, - # Bumped from default 50ms — channel-heavy test files (notably - # AgentChannelTest) intermittently exhaust the pool while waiting on - # background `:proc_lib` processes to release their checked-out - # connections. A longer queue lets the test pool ride out the spike - # rather than crashing with `connection not available`. - queue_target: 500, - queue_interval: 5_000 -else - config :towerops, Towerops.Repo, - username: "postgres", - password: "postgres", - hostname: "localhost", - database: "towerops_test#{System.get_env("MIX_TEST_PARTITION")}", - pool: Sandbox, - pool_size: System.schedulers_online() * 2, - queue_target: 500, - queue_interval: 5_000 -end - -# Use mock LLM in tests by default. Tests that exercise the real DeepSeek -# client (e.g. test/towerops/llm/deepseek_test.exs) call the module -# directly and bypass this configuration. -config :towerops, Towerops.LLM, impl: Towerops.LLM.MockClient +config :towerops, Towerops.Repo, + username: System.get_env("PGUSER", "postgres"), + password: System.get_env("PGPASSWORD", "postgres"), + hostname: System.get_env("PGHOST", "localhost"), + port: String.to_integer(System.get_env("PGPORT", "5432")), + database: "towerops_test#{System.get_env("MIX_TEST_PARTITION")}", + pool: Sandbox, + pool_size: System.schedulers_online() * 2, + # Bumped from default 50ms — channel-heavy test files (notably + # AgentChannelTest) intermittently exhaust the pool while waiting on + # background `:proc_lib` processes to release their checked-out + # connections. A longer queue lets the test pool ride out the spike + # rather than crashing with `connection not available`. + queue_target: 500, + queue_interval: 5_000 # Configure Cloak encryption for testing # Use a fixed test key (same as dev for simplicity) @@ -110,6 +95,10 @@ config :towerops, :agent_webhook_secret, "test-webhook-secret" # Set environment identifier for runtime checks config :towerops, :env, :test + +# MIB upload directory for tests — points at a writable tmp dir so +# the MibController upload/delete flows can actually be exercised. +config :towerops, :mib_dir, Path.join(System.tmp_dir!(), "towerops-test-mibs") config :towerops, :session_encryption_salt, "test-encryption-salt-stable" config :towerops, :session_signing_salt, "test-signing-salt-stable" diff --git a/lib/towerops/snmp/discovery.ex b/lib/towerops/snmp/discovery.ex index a14516f5..ac612cd5 100644 --- a/lib/towerops/snmp/discovery.ex +++ b/lib/towerops/snmp/discovery.ex @@ -14,6 +14,7 @@ defmodule Towerops.Snmp.Discovery do import Ecto.Query + alias Ecto.Adapters.SQL alias Towerops.Devices alias Towerops.Devices.Device, as: DeviceSchema alias Towerops.Devices.Firmware @@ -1545,9 +1546,61 @@ defmodule Towerops.Snmp.Discovery do entities = Repo.all(from(e in EntityPhysical, where: e.snmp_device_id == ^snmp_device_id)) index_to_id = Map.new(entities, &{&1.entity_index, &1.id}) + # Batch-resolve all parent IDs + resolved = + Enum.map(entities, fn entity -> + Map.put( + %{id: entity.id, old_parent_id: entity.parent_id, parent_index: entity.parent_index}, + :new_parent_id, + resolve_parent_id(index_to_id, %{parent_index: entity.parent_index}) + ) + end) + + # Group by old/new parent_id for batch updates + {_unchanged, needs_update} = + Enum.split_with(resolved, &(&1.old_parent_id == &1.new_parent_id)) + + update_entity_parents(needs_update) + + :ok + end + + defp update_entity_parents([]), do: :ok + + defp update_entity_parents(entities) do + query = build_parent_update_query(entities) + + try do + case SQL.query(Repo, query, []) do + {:ok, _result} -> :ok + {:error, _reason} -> update_entity_parents_individually(entities) + end + catch + _kind, _error -> update_entity_parents_individually(entities) + end + end + + defp build_parent_update_query(entities) do + values = + Enum.map_join(entities, ",\n", fn entity -> + parent_id = if entity.new_parent_id, do: "'#{entity.new_parent_id}'::uuid", else: "NULL" + " ('#{entity.id}'::uuid, #{parent_id})" + end) + + """ + UPDATE entity_physical e + SET parent_id = v.new_parent_id::uuid + FROM (VALUES #{values}) AS v(id, new_parent_id) + WHERE e.id = v.id + """ + end + + defp update_entity_parents_individually(entities) do Enum.each(entities, fn entity -> - parent_id = resolve_parent_id(index_to_id, entity) - update_parent_if_changed(entity, parent_id) + Repo.update_all( + from(entity_physical in EntityPhysical, where: entity_physical.id == ^entity.id), + set: [parent_id: entity.new_parent_id] + ) end) end @@ -1555,14 +1608,6 @@ defmodule Towerops.Snmp.Discovery do defp resolve_parent_id(_index_to_id, %{parent_index: "0"}), do: nil defp resolve_parent_id(index_to_id, %{parent_index: idx}), do: Map.get(index_to_id, idx) - defp update_parent_if_changed(%{parent_id: existing} = _entity, existing), do: :ok - - defp update_parent_if_changed(entity, parent_id) do - entity - |> Ecto.Changeset.change(parent_id: parent_id) - |> Repo.update!() - end - @spec update_device_discovery_time(DeviceSchema.t()) :: {:ok, DeviceSchema.t()} | {:error, Ecto.Changeset.t()} defp update_device_discovery_time(device) do @@ -1612,30 +1657,21 @@ defmodule Towerops.Snmp.Discovery do {:ok, device} end - defp update_device_name_from_snmp(device, system_info) do - do_update_device_name(device, system_info, Map.has_key?(system_info, :sys_name)) - end + defp update_device_name_from_snmp(device, %{sys_name: sys_name}) do + if sys_name == "" do + Logger.info("sysName empty or nil, keeping device name unchanged", device_id: device.id) + {:ok, device} + else + Logger.info("Updating device name from SNMP sysName: #{sys_name}", device_id: device.id) - defp do_update_device_name(device, _system_info, false) do - Logger.info("Device name already set or no sysName, skipping update", device_id: device.id) - {:ok, device} - end + result = + device + |> Ecto.Changeset.change(name: sys_name) + |> Repo.update() - defp do_update_device_name(device, %{sys_name: sys_name}, true) when is_nil(sys_name) or sys_name == "" do - Logger.info("sysName empty, keeping device name unchanged", device_id: device.id) - {:ok, device} - end - - defp do_update_device_name(device, %{sys_name: sys_name}, true) do - Logger.info("Updating device name from SNMP sysName: #{sys_name}", device_id: device.id) - - result = - device - |> Ecto.Changeset.change(name: sys_name) - |> Repo.update() - - Logger.info("Device name update completed", device_id: device.id) - result + Logger.info("Device name update completed", device_id: device.id) + result + end end defp log_ip_discovery_results(device, ip_addresses) do diff --git a/lib/towerops/topology.ex b/lib/towerops/topology.ex index ea1f8cc9..88ccc048 100644 --- a/lib/towerops/topology.ex +++ b/lib/towerops/topology.ex @@ -421,13 +421,18 @@ defmodule Towerops.Topology do end defp broadcast_if_changed(true, organization_id) do - Phoenix.PubSub.broadcast( - Towerops.PubSub, - "topology:#{organization_id}", - {:topology_updated, organization_id} - ) + case Phoenix.PubSub.broadcast( + Towerops.PubSub, + "topology:#{organization_id}", + {:topology_updated, organization_id} + ) do + :ok -> + {:ok, :changed} - {:ok, :changed} + {:error, reason} -> + Logger.error("Failed to broadcast topology update", device_id: organization_id, reason: reason) + {:ok, :changed} + end end defp broadcast_if_changed(false, _organization_id), do: {:ok, :unchanged} diff --git a/lib/towerops/workers/wireless_insight_worker.ex b/lib/towerops/workers/wireless_insight_worker.ex index b6bca97f..8166a777 100644 --- a/lib/towerops/workers/wireless_insight_worker.ex +++ b/lib/towerops/workers/wireless_insight_worker.ex @@ -103,7 +103,6 @@ defmodule Towerops.Workers.WirelessInsightWorker do end defp auto_resolve_weak_signals(organization) do - # Get all active weak signal insights active_insights = Repo.all( from(i in Insight, @@ -113,19 +112,33 @@ defmodule Towerops.Workers.WirelessInsightWorker do ) ) - # Check current state and resolve if improved - Enum.each(active_insights, fn insight -> - dedup_key = insight.metadata["dedup_key"] - [device_id, mac_address] = String.split(dedup_key, "_", parts: 2) + # N+1 fix: batch-collect devices+macs then do a single query + device_macs = + Enum.map(active_insights, fn insight -> + [device_id, mac_address] = String.split(insight.metadata["dedup_key"], "_", parts: 2) + {device_id, String.downcase(mac_address)} + end) - # Get current client state - client = - Repo.one( - from(c in WirelessClient, - where: c.device_id == ^device_id, - where: fragment("LOWER(?)", c.mac_address) == ^String.downcase(mac_address) - ) + clients = + if device_macs == [] do + %{} + else + {device_ids, mac_addrs} = Enum.unzip(device_macs) + + from(c in WirelessClient, + where: c.device_id in ^device_ids, + where: fragment("LOWER(?)", c.mac_address) in ^mac_addrs ) + |> Repo.all() + |> Map.new(fn c -> + {{c.device_id, String.downcase(c.mac_address)}, c} + end) + end + + Enum.each(active_insights, fn insight -> + [device_id, mac_address] = String.split(insight.metadata["dedup_key"], "_", parts: 2) + lookup_key = {device_id, String.downcase(mac_address)} + client = Map.get(clients, lookup_key) # Resolve if signal improved (with hysteresis) or client disconnected if is_nil(client) or @@ -188,17 +201,33 @@ defmodule Towerops.Workers.WirelessInsightWorker do ) ) - Enum.each(active_insights, fn insight -> - dedup_key = insight.metadata["dedup_key"] - [device_id, mac_address] = String.split(dedup_key, "_", parts: 2) + # Batch-collect devices+macs then do a single query + device_macs = + Enum.map(active_insights, fn insight -> + [device_id, mac_address] = String.split(insight.metadata["dedup_key"], "_", parts: 2) + {device_id, String.downcase(mac_address)} + end) - client = - Repo.one( - from(c in WirelessClient, - where: c.device_id == ^device_id, - where: fragment("LOWER(?)", c.mac_address) == ^String.downcase(mac_address) - ) + clients = + if device_macs == [] do + %{} + else + {device_ids, mac_addrs} = Enum.unzip(device_macs) + + from(c in WirelessClient, + where: c.device_id in ^device_ids, + where: fragment("LOWER(?)", c.mac_address) in ^mac_addrs ) + |> Repo.all() + |> Map.new(fn c -> + {{c.device_id, String.downcase(c.mac_address)}, c} + end) + end + + Enum.each(active_insights, fn insight -> + [device_id, mac_address] = String.split(insight.metadata["dedup_key"], "_", parts: 2) + lookup_key = {device_id, String.downcase(mac_address)} + client = Map.get(clients, lookup_key) # Resolve if SNR improved (with hysteresis) or client disconnected if is_nil(client) or (client.snr && client.snr >= @snr_recovery_threshold) do diff --git a/lib/towerops_web/live/status_page_live.ex b/lib/towerops_web/live/status_page_live.ex index 988fed7c..6f7b19aa 100644 --- a/lib/towerops_web/live/status_page_live.ex +++ b/lib/towerops_web/live/status_page_live.ex @@ -64,7 +64,7 @@ defmodule ToweropsWeb.StatusPageLive do |> assign(:page_title, "Not Found")} config -> - maybe_subscribe_to_status_page(socket, config) + _ = maybe_subscribe_to_status_page(socket, config) components = StatusPages.list_components(config.id) active_incidents = StatusPages.list_active_incidents(config.id) diff --git a/nix/shell.nix b/nix/shell.nix index b0b34f37..c5432e7d 100644 --- a/nix/shell.nix +++ b/nix/shell.nix @@ -308,7 +308,6 @@ mkShell { export RELEASE_COOKIE="dev_release_cookie_for_distributed_erlang" # Development environment - export MIX_ENV=dev export PHX_HOST=localhost export PORT=4000 diff --git a/test/towerops_web/channels/agent_channel_builders_test.exs b/test/towerops_web/channels/agent_channel_builders_test.exs index 35a794d3..8fde63f6 100644 --- a/test/towerops_web/channels/agent_channel_builders_test.exs +++ b/test/towerops_web/channels/agent_channel_builders_test.exs @@ -156,7 +156,7 @@ defmodule ToweropsWeb.AgentChannelBuildersTest do assert_push "jobs", %{binary: jobs_binary} jobs = decode_jobs(jobs_binary) - job = Enum.find(jobs, &(&1.device_id == v1_device.id)) + job = Enum.find(jobs, &(&1.device_id == v1_device.id and &1.job_type == :DISCOVER)) assert job.snmp_device.version == "1" assert job.snmp_device.community == "public-v1" assert job.snmp_device.port == 161 @@ -173,7 +173,7 @@ defmodule ToweropsWeb.AgentChannelBuildersTest do assert_push "jobs", %{binary: jobs_binary}, 500 jobs = decode_jobs(jobs_binary) - job = Enum.find(jobs, &(&1.device_id == device.id)) + job = Enum.find(jobs, &(&1.device_id == device.id and &1.job_type == :DISCOVER)) assert job.snmp_device.version == "2c" assert job.snmp_device.community == "public" assert job.snmp_device.port == 161 @@ -199,7 +199,7 @@ defmodule ToweropsWeb.AgentChannelBuildersTest do assert_push "jobs", %{binary: jobs_binary} jobs = decode_jobs(jobs_binary) - job = Enum.find(jobs, &(&1.device_id == device.id)) + job = Enum.find(jobs, &(&1.device_id == device.id and &1.job_type == :DISCOVER)) assert job.snmp_device.port == 1161 Process.flag(:trap_exit, true) @@ -235,7 +235,7 @@ defmodule ToweropsWeb.AgentChannelBuildersTest do assert_push "jobs", %{binary: jobs_binary} jobs = decode_jobs(jobs_binary) - job = Enum.find(jobs, &(&1.device_id == device.id)) + job = Enum.find(jobs, &(&1.device_id == device.id and &1.job_type == :DISCOVER)) assert job.snmp_device.version == "3" assert job.snmp_device.v3_security_level == "authPriv" assert job.snmp_device.v3_username == "alice" @@ -270,7 +270,7 @@ defmodule ToweropsWeb.AgentChannelBuildersTest do assert_push "jobs", %{binary: jobs_binary} jobs = decode_jobs(jobs_binary) - job = Enum.find(jobs, &(&1.device_id == device.id)) + job = Enum.find(jobs, &(&1.device_id == device.id and &1.job_type == :DISCOVER)) assert job.snmp_device.version == "3" assert job.snmp_device.v3_username == "minimaluser" assert job.snmp_device.v3_auth_password == "" diff --git a/test/towerops_web/channels/agent_channel_test.exs b/test/towerops_web/channels/agent_channel_test.exs index d89ab311..277e0cf4 100644 --- a/test/towerops_web/channels/agent_channel_test.exs +++ b/test/towerops_web/channels/agent_channel_test.exs @@ -478,12 +478,12 @@ defmodule ToweropsWeb.AgentChannelTest do test "heartbeat rejects messages over @max_message_size", %{socket: socket} do ref = push(socket, "heartbeat", oversized_payload()) - assert_reply ref, :error, %{reason: "Message too large"} + assert_reply ref, :error, %{reason: "Message too large" <> _} end test "error rejects messages over @max_message_size", %{socket: socket} do ref = push(socket, "error", oversized_payload()) - assert_reply ref, :error, %{reason: "Message too large"} + assert_reply ref, :error, %{reason: "Message too large" <> _} end end