feat: add DNS service discovery for upstream groups (Phase 11.2)

Add DNS SRV-based service discovery that automatically resolves and
reconciles upstream targets. Includes discovery source CRUD, periodic
sync via Oban worker, manual refresh, auto-sync toggle, LiveView UI
section on upstream group show page, and 5 REST API endpoints.

681 tests passing (33 new).

Author: raffaelschneider

config/test.exs

@@ -51,6 +51,9 @@ config :sentinel_cp, :github_webhook, secret: "test_webhook_secret"
 # Use mock GitHub client in tests
 config :sentinel_cp, :github_client, SentinelCp.Webhooks.GitHubClient.Mock
 
+# Use mock DNS resolver in tests
+config :sentinel_cp, :dns_resolver, SentinelCp.Services.DnsResolver.Mock
+
 # Wallaby E2E test configuration
 config :wallaby,
   otp_app: :sentinel_cp,

lib/sentinel_cp/application.ex

@@ -30,6 +30,7 @@ defmodule SentinelCp.Application do
     SentinelCp.Nodes.DriftWorker.ensure_started()
     SentinelCp.Services.CertificateExpiryWorker.ensure_started()
     SentinelCp.Analytics.PruneWorker.ensure_started()
+    SentinelCp.Services.DiscoverySyncWorker.ensure_started()
 
     result
   end

lib/sentinel_cp/services.ex

@@ -8,7 +8,7 @@ defmodule SentinelCp.Services do
 
   import Ecto.Query, warn: false
   alias SentinelCp.Repo
-  alias SentinelCp.Services.{Service, ServiceTemplate, ProjectConfig, UpstreamGroup, UpstreamTarget, Certificate, AuthPolicy, OpenApiSpec}
+  alias SentinelCp.Services.{Service, ServiceTemplate, ProjectConfig, UpstreamGroup, UpstreamTarget, Certificate, AuthPolicy, OpenApiSpec, DiscoverySource, DiscoverySync}
 
   ## Services
 
@@ -483,6 +483,140 @@ defmodule SentinelCp.Services do
     end)
   end
 
+  ## Discovery Sources
+
+  @doc """
+  Lists discovery sources for a project, preloading upstream_group.
+  """
+  def list_discovery_sources(project_id) do
+    from(d in DiscoverySource,
+      where: d.project_id == ^project_id,
+      preload: [:upstream_group]
+    )
+    |> Repo.all()
+  end
+
+  @doc """
+  Gets a single discovery source by ID.
+  """
+  def get_discovery_source(id), do: Repo.get(DiscoverySource, id)
+
+  @doc """
+  Gets a single discovery source by ID, raises if not found.
+  """
+  def get_discovery_source!(id), do: Repo.get!(DiscoverySource, id)
+
+  @doc """
+  Gets the discovery source for an upstream group, if any.
+  """
+  def get_discovery_source_for_group(upstream_group_id) do
+    from(d in DiscoverySource, where: d.upstream_group_id == ^upstream_group_id)
+    |> Repo.one()
+  end
+
+  @doc """
+  Creates a discovery source.
+  """
+  def create_discovery_source(attrs) do
+    %DiscoverySource{}
+    |> DiscoverySource.changeset(attrs)
+    |> Repo.insert()
+  end
+
+  @doc """
+  Updates a discovery source (hostname, interval, auto_sync).
+  """
+  def update_discovery_source(%DiscoverySource{} = source, attrs) do
+    source
+    |> DiscoverySource.update_changeset(attrs)
+    |> Repo.update()
+  end
+
+  @doc """
+  Deletes a discovery source.
+  """
+  def delete_discovery_source(%DiscoverySource{} = source) do
+    Repo.delete(source)
+  end
+
+  @doc """
+  Lists all discovery sources with auto_sync enabled (across all projects).
+  """
+  def list_auto_sync_sources do
+    from(d in DiscoverySource, where: d.auto_sync == true)
+    |> Repo.all()
+  end
+
+  @doc """
+  Synchronizes a discovery source by resolving DNS SRV records and reconciling targets.
+
+  Returns `{:ok, %{added: N, removed: N, kept: N}}` on success or `{:error, reason}` on failure.
+  """
+  def sync_discovery_source(%DiscoverySource{} = source) do
+    # Mark as syncing
+    {:ok, source} =
+      source
+      |> DiscoverySource.sync_changeset(%{last_sync_status: "syncing"})
+      |> Repo.update()
+
+    case dns_resolver().resolve_srv(source.hostname) do
+      {:ok, records} ->
+        group = get_upstream_group!(source.upstream_group_id)
+        current_targets = group.targets || []
+
+        result = DiscoverySync.reconcile(current_targets, records)
+
+        # Apply additions
+        for target_attrs <- result.add do
+          add_upstream_target(
+            Map.merge(target_attrs, %{upstream_group_id: source.upstream_group_id})
+          )
+        end
+
+        # Apply removals
+        for target <- result.remove do
+          remove_upstream_target(target)
+        end
+
+        total_count = length(result.add) + length(result.keep)
+
+        {:ok, source} =
+          source
+          |> DiscoverySource.sync_changeset(%{
+            last_synced_at: DateTime.utc_now() |> DateTime.truncate(:second),
+            last_sync_status: "synced",
+            last_sync_error: nil,
+            last_sync_targets_count: total_count
+          })
+          |> Repo.update()
+
+        Phoenix.PubSub.broadcast(
+          SentinelCp.PubSub,
+          "discovery:#{source.id}",
+          {:discovery_synced, source.id}
+        )
+
+        {:ok, %{added: length(result.add), removed: length(result.remove), kept: length(result.keep)}}
+
+      {:error, reason} ->
+        error_msg = if is_binary(reason), do: reason, else: inspect(reason)
+
+        source
+        |> DiscoverySource.sync_changeset(%{
+          last_synced_at: DateTime.utc_now() |> DateTime.truncate(:second),
+          last_sync_status: "error",
+          last_sync_error: error_msg
+        })
+        |> Repo.update()
+
+        {:error, error_msg}
+    end
+  end
+
+  defp dns_resolver do
+    Application.get_env(:sentinel_cp, :dns_resolver, SentinelCp.Services.DnsResolver.Inet)
+  end
+
   defp resolve_auth_policy_id(%{security_refs: refs}, policy_map) when is_list(refs) do
     Enum.find_value(refs, fn ref -> Map.get(policy_map, ref) end)
   end

lib/sentinel_cp/services/discovery_source.ex

@@ -0,0 +1,68 @@
+defmodule SentinelCp.Services.DiscoverySource do
+  @moduledoc """
+  Schema for DNS-based service discovery sources.
+
+  A discovery source attaches to an upstream group and periodically resolves
+  SRV records to reconcile upstream targets automatically.
+  """
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  @primary_key {:id, :binary_id, autogenerate: true}
+  @foreign_key_type :binary_id
+
+  @source_types ~w(dns_srv)
+  @sync_statuses ~w(pending syncing synced error)
+
+  schema "discovery_sources" do
+    field :source_type, :string, default: "dns_srv"
+    field :hostname, :string
+    field :sync_interval_seconds, :integer, default: 60
+    field :auto_sync, :boolean, default: true
+    field :last_synced_at, :utc_datetime
+    field :last_sync_status, :string, default: "pending"
+    field :last_sync_error, :string
+    field :last_sync_targets_count, :integer, default: 0
+
+    belongs_to :upstream_group, SentinelCp.Services.UpstreamGroup
+    belongs_to :project, SentinelCp.Projects.Project
+
+    timestamps(type: :utc_datetime)
+  end
+
+  def changeset(source, attrs) do
+    source
+    |> cast(attrs, [
+      :source_type,
+      :hostname,
+      :sync_interval_seconds,
+      :auto_sync,
+      :upstream_group_id,
+      :project_id
+    ])
+    |> validate_required([:hostname, :upstream_group_id, :project_id])
+    |> validate_inclusion(:source_type, @source_types)
+    |> validate_number(:sync_interval_seconds, greater_than_or_equal_to: 10)
+    |> unique_constraint(:upstream_group_id)
+    |> foreign_key_constraint(:upstream_group_id)
+    |> foreign_key_constraint(:project_id)
+  end
+
+  def update_changeset(source, attrs) do
+    source
+    |> cast(attrs, [:hostname, :sync_interval_seconds, :auto_sync])
+    |> validate_required([:hostname])
+    |> validate_number(:sync_interval_seconds, greater_than_or_equal_to: 10)
+  end
+
+  def sync_changeset(source, attrs) do
+    source
+    |> cast(attrs, [
+      :last_synced_at,
+      :last_sync_status,
+      :last_sync_error,
+      :last_sync_targets_count
+    ])
+    |> validate_inclusion(:last_sync_status, @sync_statuses)
+  end
+end

lib/sentinel_cp/services/discovery_sync.ex

@@ -0,0 +1,43 @@
+defmodule SentinelCp.Services.DiscoverySync do
+  @moduledoc """
+  Pure-function module for reconciling upstream targets with DNS SRV records.
+
+  No database access — operates on in-memory data structures.
+  """
+
+  @doc """
+  Reconciles current upstream targets against resolved SRV records.
+
+  Returns a map with:
+  - `:add` — list of `%{host, port, weight}` maps for new targets
+  - `:remove` — list of target structs to remove
+  - `:keep` — list of target structs to keep
+  """
+  def reconcile(current_targets, srv_records) do
+    resolved = Enum.map(srv_records, &srv_to_target/1)
+    resolved_set = MapSet.new(resolved, fn t -> {t.host, t.port} end)
+    current_set = MapSet.new(current_targets, fn t -> {t.host, t.port} end)
+
+    add =
+      resolved
+      |> Enum.filter(fn t -> not MapSet.member?(current_set, {t.host, t.port}) end)
+
+    remove =
+      current_targets
+      |> Enum.filter(fn t -> not MapSet.member?(resolved_set, {t.host, t.port}) end)
+
+    keep =
+      current_targets
+      |> Enum.filter(fn t -> MapSet.member?(resolved_set, {t.host, t.port}) end)
+
+    %{add: add, remove: remove, keep: keep}
+  end
+
+  defp srv_to_target({_priority, weight, port, host}) do
+    %{
+      host: to_string(host),
+      port: port,
+      weight: max(weight, 1)
+    }
+  end
+end

lib/sentinel_cp/services/discovery_sync_worker.ex

@@ -0,0 +1,74 @@
+defmodule SentinelCp.Services.DiscoverySyncWorker do
+  @moduledoc """
+  Oban worker that periodically syncs DNS-based discovery sources.
+
+  Checks all auto_sync sources and triggers sync when the configured
+  interval has elapsed since the last sync.
+  """
+  use Oban.Worker,
+    queue: :maintenance,
+    max_attempts: 1,
+    unique: [period: 30]
+
+  require Logger
+
+  alias SentinelCp.Services
+
+  @default_interval_seconds 30
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{}) do
+    Logger.debug("DiscoverySyncWorker: checking discovery sources")
+
+    sources = Services.list_auto_sync_sources()
+
+    for source <- sources do
+      if sync_due?(source) do
+        case Services.sync_discovery_source(source) do
+          {:ok, result} ->
+            Logger.info(
+              "DiscoverySyncWorker: synced #{source.hostname} — " <>
+                "added: #{result.added}, removed: #{result.removed}, kept: #{result.kept}"
+            )
+
+          {:error, reason} ->
+            Logger.warning("DiscoverySyncWorker: failed to sync #{source.hostname}: #{reason}")
+        end
+      end
+    end
+
+    reschedule()
+    :ok
+  end
+
+  defp sync_due?(%{last_synced_at: nil}), do: true
+
+  defp sync_due?(source) do
+    elapsed = DateTime.diff(DateTime.utc_now(), source.last_synced_at, :second)
+    elapsed >= source.sync_interval_seconds
+  end
+
+  defp reschedule do
+    oban_config = Application.get_env(:sentinel_cp, Oban, [])
+
+    unless oban_config[:testing] do
+      %{}
+      |> __MODULE__.new(schedule_in: @default_interval_seconds)
+      |> Oban.insert()
+    end
+  end
+
+  @doc """
+  Starts the discovery sync worker if not already running.
+  Called during application startup.
+  """
+  def ensure_started do
+    oban_config = Application.get_env(:sentinel_cp, Oban, [])
+
+    unless oban_config[:testing] do
+      %{}
+      |> __MODULE__.new()
+      |> Oban.insert()
+    end
+  end
+end

lib/sentinel_cp/services/dns_resolver.ex

@@ -0,0 +1,11 @@
+defmodule SentinelCp.Services.DnsResolver do
+  @moduledoc """
+  Behaviour for DNS SRV record resolution.
+
+  Implementations resolve SRV records for service discovery.
+  """
+
+  @callback resolve_srv(hostname :: String.t()) ::
+              {:ok, [{priority :: integer, weight :: integer, port :: integer, host :: charlist}]}
+              | {:error, term()}
+end

lib/sentinel_cp/services/dns_resolver/inet.ex

@@ -0,0 +1,15 @@
+defmodule SentinelCp.Services.DnsResolver.Inet do
+  @moduledoc """
+  DNS SRV resolver using Erlang's `:inet_res` module.
+  """
+
+  @behaviour SentinelCp.Services.DnsResolver
+
+  @impl true
+  def resolve_srv(hostname) do
+    records = :inet_res.lookup(to_charlist(hostname), :in, :srv)
+    {:ok, records}
+  rescue
+    e -> {:error, Exception.message(e)}
+  end
+end