Begin trying to work on CD rtc publishing

this is going to be meaningfully tricky to respect auth policies and the complexity of the underlying data

lib/console/deployments/pubsub/rtc.ex

@@ -1,21 +1,100 @@
+defmodule Console.Deployments.Rtc.Utils do
+  alias Console.Repo
+  alias Console.Schema.{Pipeline, Service, Cluster, DeploymentSettings}
+  alias Console.Deployments.Settings
+
+  def audience(%Pipeline{} = pipe) do
+    %{read_bindings: bindings} = Repo.preload(pipe, [:read_bindings])
+    settings = Settings.fetch()
+    collect(bindings, settings)
+  end
+
+  def audience(%Service{} = service) do
+    %{read_bindings: bindings, cluster: cluster} = Repo.preload(service, [:read_bindings, cluster: :read_bindings])
+    collect(bindings, audience(cluster))
+  end
+
+  def audience(%Cluster{} = cluster) do
+    %{read_bindings: bindings} = Repo.preload(cluster, [:read_bindings])
+    settings = Settings.fetch()
+    collect(bindings, settings)
+  end
+
+  defp collect([_ | _] = bindings, %DeploymentSettings{read_bindings: [_ | _] = globals}),
+    do: do_collect(bindings ++ globals)
+  defp collect([_ | _] = bindings, {_, _} = acc), do: do_collect(bindings, acc)
+  defp collect([_ | _] = bindings, _), do: do_collect(bindings)
+  defp collect(_, _), do: {[], []}
+
+  defp do_collect(bindings, acc \\ {[], []}) do
+    Enum.reduce(bindings, acc, fn
+      %{user_id: uid}, {users, groups} when is_binary(uid) -> {[uid | users], groups}
+      %{group_id: gid}, {users, groups} when is_binary(gid) -> {users, [gid | groups]}
+      _, acc -> acc
+    end)
+  end
+end
+
 ## I need to find a way to provide authorization for this that's not a major perf drag
 
 # defimpl Console.PubSub.Rtc, for: [
 #   Console.PubSub.ServiceUpdated,
 #   Console.PubSub.ClusterUpdated,
-#   Console.PubSub.ProviderUpdated,
-#   Console.PubSub.GitRepositoryUpdated,
+#   Console.PubSub.PipelineUpdated,
+#   Console.PubSub.ServiceDeleted,
+#   Console.PubSub.ServiceComponentsUpdated,
 # ] do
-#   def deliver(%{item: item}), do: {item, :update}
+#   alias Console.Deployments.Rtc.Utils
+#   def deliver(%{item: item}) do
+#     {users, groups} = Utils.audience(item)
+#     channel = channel(item)
+#     topics =
+#       Enum.map(users, & "#{channel}:user:#{&1}") ++
+#       Enum.map(groups, & "#{channel}:group:#{&1}")
+#     {item, topics, :update}
+#   end
+
+#   defp channel(%Console.Schema.Service{}), do: "service"
+#   defp channel(%Console.Schema.Cluster{}), do: "cluster"
+#   defp channel(%Console.Schema.Pipeline{}), do: "pipeline"
 # end
 
 # defimpl Console.PubSub.Rtc, for: [
 #   Console.PubSub.ServiceCreated,
-#   Console.PubSub.ServiceDeleted,
 #   Console.PubSub.ClusterCreated,
-#   Console.PubSub.ClusterDeleted,
-#   Console.PubSub.ProviderCreated,
-#   Console.PubSub.GitRepositoryCreated
+#   Console.PubSub.PipelineCreated,
+# ] do
+#   alias Console.Deployments.Rtc.Utils
+#   def deliver(%{item: item}) do
+#     {users, groups} = Utils.audience(item)
+#     channel = channel(item)
+#     topics =
+#       Enum.map(users, & "#{channel}:user:#{&1}") ++
+#       Enum.map(groups, & "#{channel}:group:#{&1}")
+#     {item, topics, :create}
+#   end
+
+#   defp channel(%Console.Schema.Service{}), do: "service"
+#   defp channel(%Console.Schema.Cluster{}), do: "cluster"
+#   defp channel(%Console.Schema.Pipeline{}), do: "pipeline"
+# end
+
+# defimpl Console.PubSub.Rtc, for: [
+#   Console.PubSub.ServiceHardDeleted,
+#   Console.PubSub.ClusterCreated,
+#   Console.PubSub.PipelineCreated
 # ] do
-#   def deliver(%{item: item}), do: {item, :create}
+#   alias Console.Deployments.Rtc.Utils
+#   def deliver(%{item: item}) do
+#     {users, groups} = Utils.audience(item)
+#     channel = channel(item)
+#     topics =
+#       Enum.map(users, & "#{channel}:user:#{&1}") ++
+#       Enum.map(groups, & "#{channel}:group:#{&1}")
+#     {item, topics, :create}
+#   end
+
+#   defp channel(%Console.Schema.Service{}), do: "service"
+#   defp channel(%Console.Schema.Cluster{}), do: "cluster"
+#   defp channel(%Console.Schema.Pipeline{}), do: "pipeline"
 # end