gc cron job to clean up dangling policy_bindings entries

Author: kinjalh

config/prod.exs

@@ -68,6 +68,7 @@ config :console, Console.Cron.Scheduler,
     {"15 * * * *",     {Console.Deployments.Cron, :prune_vuln_reports, []}},
     {"*/15 * * * *",   {Console.Deployments.Cron, :pr_governance, []}},
     {"15 3 * * *",     {Console.Deployments.Cron, :prune_dangling_templates, []}},
+    {"20 3 * * *",     {Console.Deployments.Cron, :prune_dangling_policy_bindings, []}},
     {"30 3 * * *",     {Console.Deployments.Cron, :prune_insight_components, []}},
     {"0 4 * * *",      {Console.Deployments.Cron, :prune_helm_repositories, []}},
     {"0 5 * * *",      {Console.Deployments.Cron, :prune_agent_run_repositories, []}},

lib/console/deployments/cron.ex

@@ -18,6 +18,7 @@ defmodule Console.Deployments.Cron do
     AppNotification,
     Alert,
     ClusterAuditLog,
+    PolicyBinding,
     PolicyConstraint,
     VulnerabilityReport,
     ServiceTemplate,
@@ -298,6 +299,20 @@ defmodule Console.Deployments.Cron do
     |> Stream.run()
   end
 
+  def prune_dangling_policy_bindings() do
+    PolicyBinding.dangling()
+    |> PolicyBinding.ordered(asc: :id)
+    |> Repo.stream(method: :keyset)
+    |> Stream.chunk_every(100)
+    |> Stream.each(fn bindings ->
+      ids = Enum.map(bindings, & &1.id)
+      Logger.info "pruning #{length(ids)} dangling policy bindings"
+      PolicyBinding.for_ids(ids)
+      |> Repo.delete_all(timeout: 10_000)
+    end)
+    |> Stream.run()
+  end
+
   def add_ignore_crds(search) do
     Service.search(search)
     |> Repo.stream(method: :keyset)

lib/console/schema/policy_binding.ex

@@ -10,6 +10,34 @@ defmodule Console.Schema.PolicyBinding do
     timestamps()
   end
 
+  def dangling(query \\ __MODULE__) do
+    from(p in query,
+      where:
+        fragment("NOT EXISTS(SELECT 1 FROM services WHERE write_policy_id = ? OR read_policy_id = ?)", p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM clusters WHERE write_policy_id = ? OR read_policy_id = ?)", p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM projects WHERE write_policy_id = ? OR read_policy_id = ?)", p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM pipelines WHERE write_policy_id = ? OR read_policy_id = ?)", p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM stacks WHERE write_policy_id = ? OR read_policy_id = ?)", p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM catalogs WHERE write_policy_id = ? OR read_policy_id = ? OR create_policy_id = ?)", p.policy_id, p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM deployment_settings WHERE write_policy_id = ? OR read_policy_id = ? OR create_policy_id = ? OR git_policy_id = ?)", p.policy_id, p.policy_id, p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM pr_automations WHERE write_policy_id = ? OR create_policy_id = ?)", p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM flows WHERE write_policy_id = ? OR read_policy_id = ?)", p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM cluster_providers WHERE write_policy_id = ? OR read_policy_id = ?)", p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM mcp_servers WHERE write_policy_id = ? OR read_policy_id = ?)", p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM oidc_providers WHERE bindings_id = ? OR write_policy_id = ?)", p.policy_id, p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM personas WHERE bindings_id = ?)", p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM watchman_users WHERE assume_policy_id = ?)", p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM pull_requests WHERE notifications_policy_id = ?)", p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM cloud_connections WHERE read_policy_id = ?)", p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM compliance_report_generators WHERE read_policy_id = ?)", p.policy_id) and
+        fragment("NOT EXISTS(SELECT 1 FROM agent_runtimes WHERE create_policy_id = ?)", p.policy_id)
+    )
+  end
+
+  def ordered(query \\ __MODULE__, order \\ [asc: :id]) do
+    from(p in query, order_by: ^order)
+  end
+
   @valid ~w(user_id group_id policy_id)a
 
   def changeset(model, attrs \\ %{}) do

test/console/deployments/cron_test.exs

@@ -349,4 +349,31 @@ defmodule Console.Deployments.CronTest do
       assert refetch(keep)
     end
   end
+
+  describe "#prune_dangling_policy_bindings/0" do
+    test "it will prune dangling policy bindings" do
+      user = insert(:user)
+
+      # Create a project with write_bindings - these should be kept
+      project = insert(:project, write_bindings: [%{user_id: user.id}])
+      %{write_bindings: [kept_binding]} = Console.Repo.preload(project, [:write_bindings])
+
+      # Create orphaned policy bindings with random policy_ids that don't exist anywhere
+      orphaned = for _ <- 1..3 do
+        %Console.Schema.PolicyBinding{}
+        |> Console.Schema.PolicyBinding.changeset(%{
+          policy_id: Ecto.UUID.generate(),
+          user_id: user.id
+        })
+        |> Console.Repo.insert!()
+      end
+      :ok = Cron.prune_dangling_policy_bindings()
+
+      # Referenced binding should still exist
+      assert refetch(kept_binding)
+
+      # Orphaned bindings should be deleted
+      for binding <- orphaned, do: refute refetch(binding)
+    end
+  end
 end