Using a unique GenServer per customer server

lib/query_canary/application.ex

@@ -13,10 +13,10 @@ defmodule QueryCanary.Application do
       {DNSCluster, query: Application.get_env(:query_canary, :dns_cluster_query) || :ignore},
       {Phoenix.PubSub, name: QueryCanary.PubSub},
       {Oban, Application.fetch_env!(:query_canary, Oban)},
-      # Moved to Oban Cron to prevent multiple nodes from running this
-      # QueryCanary.CheckScheduler,
-      # Start a worker by calling: QueryCanary.Worker.start_link(arg)
-      # {QueryCanary.Worker, arg},
+      # Registry for connection servers
+      {Registry, keys: :unique, name: QueryCanary.ConnectionRegistry},
+      # Dynamic supervisor for per-server connection GenServers
+      {DynamicSupervisor, strategy: :one_for_one, name: QueryCanary.ConnectionSupervisor},
       # Start to serve requests, typically the last entry
       QueryCanaryWeb.Endpoint
     ]

lib/query_canary/connections/adapter.ex

@@ -0,0 +1,17 @@
+defmodule QueryCanary.Connections.Adapter do
+  @moduledoc """
+  Behaviour for database connection adapters.
+
+  Adapters wrap a single logical connection (or pooled abstraction) and expose
+  a small uniform API used by ConnectionServer.
+  """
+
+  @callback connect(map) :: {:ok, term} | {:error, term}
+  @callback query(term, String.t(), list) :: {:ok, term} | {:error, term}
+  @callback list_tables(term) :: {:ok, list} | {:error, term}
+  @callback get_table_schema(term, String.t()) :: {:ok, term} | {:error, term}
+  @callback get_database_schema(term, String.t()) :: {:ok, term} | {:error, term}
+  @callback disconnect(term) :: :ok | {:error, term}
+
+  @optional_callbacks disconnect: 1
+end

lib/query_canary/connections/clickhouse.ex → lib/query_canary/connections/adapters/clickhouse.ex

@@ -3,6 +3,8 @@ defmodule QueryCanary.Connections.Adapters.ClickHouse do
   ClickHouse adapter for database connections using the `ch` library.
   """
 
+  @behaviour QueryCanary.Connections.Adapter
+
   @doc """
   Connects to a ClickHouse database using the `ch` library.
 
@@ -106,4 +108,22 @@ defmodule QueryCanary.Connections.Adapters.ClickHouse do
         {:error, reason}
     end
   end
+
+  @doc """
+  Disconnects from a ClickHouse database.
+
+  ## Parameters
+    * pid - The process ID of the connection
+
+  ## Returns
+    * :ok - Disconnection successful
+  """
+  def disconnect(pid) when is_pid(pid) do
+    try do
+      GenServer.stop(pid, :normal)
+      :ok
+    catch
+      _, _ -> :ok
+    end
+  end
 end

lib/query_canary/connections/mysql.ex → lib/query_canary/connections/adapters/mysql.ex

@@ -7,6 +7,8 @@ defmodule QueryCanary.Connections.Adapters.MySQL do
 
   require Logger
 
+  @behaviour QueryCanary.Connections.Adapter
+
   @doc """
   Connects to a MySQL database.
 
@@ -167,6 +169,25 @@ defmodule QueryCanary.Connections.Adapters.MySQL do
     end
   end
 
+  @doc """
+  Disconnects from a MySQL database.
+
+  ## Parameters
+    * pid - Process ID of the connection
+
+  ## Returns
+    * :ok - Disconnection successful
+    * {:error, reason} - Disconnection failed
+  """
+  def disconnect(pid) when is_pid(pid) do
+    try do
+      GenServer.stop(pid, :normal)
+      :ok
+    catch
+      _, _ -> :ok
+    end
+  end
+
   # Formats query results into a more usable structure
   defp format_results(%MyXQL.Result{} = result) do
     columns = Enum.map(result.columns || [], &String.to_atom/1)

lib/query_canary/connections/postgresql.ex → lib/query_canary/connections/adapters/postgresql.ex

@@ -7,6 +7,8 @@ defmodule QueryCanary.Connections.Adapters.PostgreSQL do
 
   require Logger
 
+  @behaviour QueryCanary.Connections.Adapter
+
   @doc """
   Connects to a PostgreSQL database.
 
@@ -18,112 +20,48 @@ defmodule QueryCanary.Connections.Adapters.PostgreSQL do
     * {:error, reason} - Connection failed
   """
   def connect(conn_details) do
+    Logger.metadata(db_hostname: conn_details.hostname)
     Logger.info("QueryCanary.Connections: Connecting to #{conn_details.hostname}")
 
-    try do
-      opts = [
-        hostname: conn_details.hostname,
-        port: conn_details.port,
-        username: conn_details.username,
-        password: conn_details.password,
-        database: conn_details.database,
-        socket_options: Map.get(conn_details, :socket_options, []),
-        # Connection pool settings
-        pool_size: 1,
-
-        # Short timeouts
-        connect_timeout: 5000,
-        timeout: 5000,
-
-        # Queue settings to fail fast
-        queue_target: 50,
-        queue_interval: 1000,
-
-        # Misc settings
-        auto_reconnect: false,
-        max_restarts: 1,
-        show_sensitive_data_on_connection_error: true,
-        name: :"db_conn_#{System.unique_integer([:positive])}"
-      ]
-
-      # Build advanced SSL options if present
-      ssl_mode = Map.get(conn_details, :ssl_mode, "allow")
-
-      ssl_opts =
-        [
-          # Map ssl_mode to verify options
-          verify:
-            case ssl_mode do
-              "verify-full" -> :verify_peer
-              "verify-ca" -> :verify_peer
-              "require" -> :verify_none
-              "prefer" -> :verify_none
-              "allow" -> :verify_none
-              _ -> :verify_none
-            end
-        ]
-        |> maybe_add_ssl_cert(conn_details)
-        |> maybe_add_ssl_key(conn_details)
-        |> maybe_add_ssl_ca_cert(conn_details)
-        |> Enum.reject(&is_nil/1)
-
-      opts = opts ++ [ssl: ssl_opts]
-
-      {:ok, pid} = Postgrex.start_link(opts)
-
-      case query(pid, "SELECT 1;") do
-        {:ok, _res} ->
-          Logger.info(
-            "QueryCanary.Connections: Successfully connected to #{conn_details.hostname}"
-          )
-
-          {:ok, pid}
-
-        {:error, _message} when ssl_mode in ["allow", "prefer"] ->
-          GenServer.stop(pid)
-
-          Logger.info(
-            "PostgreSQL connection with SSL failed to #{conn_details.hostname}, retrying without SSL"
-          )
-
-          # Retry without SSL
-          opts_no_ssl = Keyword.delete(opts, :ssl)
-
-          opts_no_ssl =
-            Keyword.put(opts_no_ssl, :name, :"db_conn_#{System.unique_integer([:positive])}")
-
-          # opts_no_ssl = Keyword.delete(opts_no_ssl, :ssl_opts)
-          {:ok, no_ssl_pid} = Postgrex.start_link(opts_no_ssl)
-
-          case query(no_ssl_pid, "SELECT 1;") do
-            {:ok, _res} ->
-              Logger.info(
-                "QueryCanary.Connections: Successfully connected to #{conn_details.hostname}, with non-SSL fallback"
-              )
-
-              {:ok, no_ssl_pid}
-
-            error ->
-              Logger.warning(
-                "QueryCanary.Connections: Failed to connect to #{conn_details.hostname}, even without SSL"
-              )
-
-              GenServer.stop(no_ssl_pid)
-              {:error, "Failed to connect, even without SSL: #{inspect(error)}"}
-          end
-
-        error ->
-          Logger.warning(
-            "QueryCanary.Connections: Failed to connect to #{conn_details.hostname}, no SSL attempted"
-          )
-
-          GenServer.stop(pid)
-          {:error, "Failed to connect: #{inspect(error)}"}
+    opts = [
+      hostname: conn_details.hostname,
+      port: conn_details.port,
+      username: conn_details.username,
+      password: conn_details.password,
+      database: conn_details.database,
+      socket_options: Map.get(conn_details, :socket_options, []),
+      name: :"db_conn_#{System.unique_integer([:positive])}"
+    ]
+
+    # Build advanced SSL options if present
+    ssl_mode = Map.get(conn_details, :ssl_mode)
+
+    opts =
+      if ssl_mode do
+        ssl_opts =
+          [
+            # Map ssl_mode to verify options
+            verify:
+              case ssl_mode do
+                "verify-full" -> :verify_peer
+                "verify-ca" -> :verify_peer
+                "require" -> :verify_none
+                "prefer" -> :verify_none
+                "allow" -> :verify_none
+                _ -> :verify_none
+              end
+          ]
+          |> maybe_add_ssl_cert(conn_details)
+          |> maybe_add_ssl_key(conn_details)
+          |> maybe_add_ssl_ca_cert(conn_details)
+          |> Enum.reject(&is_nil/1)
+
+        opts ++ ssl_opts
+      else
+        opts
       end
-    rescue
-      e ->
-        {:error, "PostgreSQL connection error: #{inspect(e)}"}
-    end
+
+    Postgrex.start_link(opts)
   end
 
   defp maybe_add_ssl_cert(opts, conn_details) do
@@ -338,4 +276,23 @@ defmodule QueryCanary.Connections.Adapters.PostgreSQL do
       raw: result
     }
   end
+
+  @doc """
+  Disconnects from a PostgreSQL database.
+
+  ## Parameters
+    * pid - Process ID of the connection
+
+  ## Returns
+    * :ok - Disconnection successful
+    * :error - Disconnection failed
+  """
+  def disconnect(pid) when is_pid(pid) do
+    try do
+      GenServer.stop(pid, :normal)
+      :ok
+    catch
+      _, _ -> :ok
+    end
+  end
 end

lib/query_canary/connections/sqlite.ex → lib/query_canary/connections/adapters/sqlite.ex

@@ -8,6 +8,8 @@ defmodule QueryCanary.Connections.Adapters.SQLite do
 
   require Logger
 
+  @behaviour QueryCanary.Connections.Adapter
+
   @doc """
   Connects to an SQLite database.
 
@@ -133,24 +135,16 @@ defmodule QueryCanary.Connections.Adapters.SQLite do
   @doc """
   Gets complete database schema information.
 
-  ## Parameters
-    * conn - Database connection
-
-  ## Returns
-    * {:ok, schema} - Complete database schema information
-    * {:error, reason} - Operation failed
+  Accepts the database name for behaviour conformity; ignored for SQLite.
   """
-  def get_database_schema(conn) do
+  def get_database_schema(conn, _database_name) do
     case list_tables(conn) do
       {:ok, tables} ->
         schema =
           Enum.reduce(tables, %{}, fn table, acc ->
             case get_table_schema(conn, table) do
-              {:ok, table_schema} ->
-                Map.put(acc, table, table_schema)
-
-              {:error, _reason} ->
-                acc
+              {:ok, table_schema} -> Map.put(acc, table, table_schema)
+              {:error, _} -> acc
             end
           end)
 
@@ -161,6 +155,8 @@ defmodule QueryCanary.Connections.Adapters.SQLite do
     end
   end
 
+  def get_database_schema(conn), do: get_database_schema(conn, nil)
+
   # Formats query results into a more usable structure
   defp format_results(%Exqlite.Result{} = result) do
     columns = Enum.map(result.columns, &String.to_atom/1)