chg: [cruds] decoupling redisinstance - moving to connections

lib/cocktailparty/catalog.ex

@@ -141,7 +141,7 @@ defmodule Cocktailparty.Catalog do
 
   """
   def create_source(attrs \\ %{}) do
-    Cocktailparty.Input.get_connection!(attrs["connection_id"])
+    Cocktailparty.Input.get_connection_map!(attrs["connection_id"])
     |> Ecto.build_assoc(:sources)
     |> change_source(attrs)
     |> Repo.insert()

lib/cocktailparty/dynamic_supervisor_boot.ex

@@ -1,9 +1,9 @@
 defmodule Cocktailparty.DynamicSupervisorBoot do
-  alias Cocktailparty.Input.Connection
+  # alias Cocktailparty.Input.Connection
   use Supervisor
 
   require Logger
-  import Cocktailparty.Input
+  # import Cocktailparty.Input
 
   def start_link(init_arg) do
     Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
@@ -20,8 +20,8 @@ defmodule Cocktailparty.DynamicSupervisorBoot do
          Logger.info("Connection Dynamic Supervisor Task started")
          # When starting we check what inputs are available
          # for each instance, we start a redix connection along with a broker gen_server
-         list_connections()
-         |> Enum.each(fn x -> Connection.start(x) end)
+        #  list_connections()
+        #  |> Enum.each(fn x -> Connection.start(x) end)
        end}
     ]
 

lib/cocktailparty/input.ex

@@ -66,7 +66,7 @@ defmodule Cocktailparty.Input do
   end
 
   @doc """
-  Gets a single connection.
+  Gets a single connection -- with config as a text field
 
   Raises `Ecto.NoResultsError` if the connection does not exist.
 
@@ -79,9 +79,37 @@ defmodule Cocktailparty.Input do
       ** (Ecto.NoResultsError)
 
   """
-  def get_connection!(id) do
+  def get_connection_text!(id) do
     instance = Repo.get!(Connection, id)
-    Map.put(instance, :connected, connected?(instance))
+
+    instance
+    |> Map.put(:connected, connected?(instance))
+    |> Map.put(:config, config_to_yaml(instance.config))
+  end
+
+  @doc """
+  Gets a single connection -- with config as a map
+
+  Raises `Ecto.NoResultsError` if the connection does not exist.
+
+  ## Examples
+
+      iex> get_connection!(123)
+      %Connection{}
+
+      iex> get_connection!(456)
+      ** (Ecto.NoResultsError)
+
+  """
+  def get_connection_map!(id) do
+    instance = Repo.get!(Connection, id)
+
+    instance
+    |> Map.put(:connected, connected?(instance))
+  end
+
+  def config_to_yaml(config) do
+    Ymlr.document!(config)
   end
 
   @doc """

lib/cocktailparty/input/connection.ex

@@ -39,8 +39,8 @@ defmodule Cocktailparty.Input.Connection do
 
         spec_redix =
           {Redix,
-           host: rc.config.hostname,
-           port: rc.config.port,
+           host: Map.get(rc.config, "hostname"),
+           port: Map.get(rc.config, "port"),
            name: {:global, "redix_" <> Integer.to_string(rc.id)}}
 
         # TODO remove the nested name (impacts catalog.ex)

lib/cocktailparty/input/connection_behaviour.ex

@@ -0,0 +1,4 @@
+defmodule Cocktailparty.Input.ConnectionBehavior do
+  @callback start_link(config :: map) :: {:ok, pid} | {:error, term}
+  @callback validate_config(config :: map) :: :ok | {:error, String.t()}
+end

lib/cocktailparty/input/connection_type.ex

@@ -0,0 +1,9 @@
+defmodule Cocktailparty.Input.ConnectionTypes do
+  @connection_types [
+    {"Redis", "redis"},
+    {"STOMP", "stomp"}
+    # Add other connection types here
+  ]
+
+  def all, do: @connection_types
+end

lib/cocktailparty/input/redis.ex

@@ -0,0 +1,16 @@
+defmodule Cocktailparty.Input.Redis do
+  @behaviour Cocktailparty.Input.ConnectionBehavior
+
+  def start_link(_config) do
+    # Implement Redis connection logic
+  end
+
+  def validate_config(config) do
+    required_keys = ["hostname", "port"]
+
+    case Enum.all?(required_keys, &Map.has_key?(config, &1)) do
+      true -> :ok
+      false -> {:error, "Missing required keys in config"}
+    end
+  end
+end

lib/cocktailparty/input/stomp.ex

@@ -0,0 +1,16 @@
+defmodule Cocktailparty.Input.Stomp do
+  @behaviour Cocktailparty.Input.ConnectionBehavior
+
+  def start_link(_config) do
+    # Implement rabbitmq stomp connection logic
+  end
+
+  def validate_config(config) do
+    required_keys = ["hostname", "port", "username", "password"]
+
+    case Enum.all?(required_keys, &Map.has_key?(config, &1)) do
+      true -> :ok
+      false -> {:error, "Missing required keys in config"}
+    end
+  end
+end

lib/cocktailparty/sink_catalog.ex

@@ -154,7 +154,7 @@ defmodule Cocktailparty.SinkCatalog do
 
   """
   def create_sink(attrs \\ %{}) do
-    Cocktailparty.Input.get_connection!(attrs["connection_id"])
+    Cocktailparty.Input.get_connection_map!(attrs["connection_id"])
     |> Ecto.build_assoc(:sinks)
     |> change_sink(attrs)
     |> Repo.insert()

lib/cocktailparty_web/controllers/admin/connection_controller.ex

@@ -3,6 +3,7 @@ defmodule CocktailpartyWeb.Admin.ConnectionController do
 
   alias Cocktailparty.Input
   alias Cocktailparty.Input.Connection
+  alias Cocktailparty.Input.ConnectionTypes
 
   def index(conn, _params) do
     connections = Input.list_connections()
@@ -11,51 +12,97 @@ defmodule CocktailpartyWeb.Admin.ConnectionController do
 
   def new(conn, _params) do
     changeset = Input.change_connection(%Connection{})
-    render(conn, :new, changeset: changeset)
+    connection_types = ConnectionTypes.all()
+    render(conn, :new, changeset: changeset, connection_types: connection_types)
   end
 
   def create(conn, %{"connection" => connection_params}) do
-    case Input.create_connection(connection_params) do
+    {:ok, config_str} = Map.fetch(connection_params, "config")
+    config = YamlElixir.read_from_string!(config_str)
+
+    case Input.create_connection(Map.put(connection_params, "config", config)) do
       {:ok, connection} ->
         # TODO: handle errors
-        Cocktailparty.Input.Connection.start(connection)
+        # Cocktailparty.Input.Connection.start(connection)
 
         conn
         |> put_flash(:info, "Connection created successfully.")
         |> redirect(to: ~p"/admin/connections/#{connection}")
 
       {:error, %Ecto.Changeset{} = changeset} ->
-        render(conn, :new, changeset: changeset)
+        txt_config = Ymlr.document!(changeset.changes.config)
+        new_changes = Map.put(changeset.changes, :config, txt_config)
+        new_changeset = Map.put(changeset, :changes, new_changes)
+
+        connection_types = ConnectionTypes.all()
+
+        render(conn, :new, changeset: new_changeset, connection_types: connection_types)
     end
   end
 
   def show(conn, %{"id" => id}) do
-    connection = Input.get_connection!(id)
+    connection = Input.get_connection_map!(id)
     render(conn, :show, connection: connection)
   end
 
   def edit(conn, %{"id" => id}) do
-    connection = Input.get_connection!(id)
+    connection = Input.get_connection_text!(id)
     changeset = Input.change_connection(connection)
-    render(conn, :edit, connection: connection, changeset: changeset)
+    connection_types = ConnectionTypes.all()
+
+    render(conn, :edit,
+      connection: connection,
+      changeset: changeset,
+      connection_types: connection_types
+    )
   end
 
   def update(conn, %{"id" => id, "connection" => connection_params}) do
-    connection = Input.get_connection!(id)
+    connection = Input.get_connection_text!(id)
 
-    case Input.update_connection(connection, connection_params) do
-      {:ok, connection} ->
-        conn
-        |> put_flash(:info, "Connection updated successfully.")
-        |> redirect(to: ~p"/admin/connections/#{connection}")
+    # TODO wip
+    yaml_config = connection_params["config"]
 
-      {:error, %Ecto.Changeset{} = changeset} ->
-        render(conn, :edit, connection: connection, changeset: changeset)
+    case YamlElixir.read_from_string(yaml_config) do
+      {:ok, config_map} ->
+        connection_params = Map.put(connection_params, "config", config_map)
+        # dbg(config_map)
+        # dbg(connection_params)
+
+        #     case Catalog.update_connection(connection, connection_params) do
+        #       {:ok, _connection} ->
+        #         conn
+        #         |> put_flash(:info, "Connection updated successfully.")
+        #         |> redirect(to: Routes.connection_path(conn, :index))
+        #       {:error, changeset} ->
+        #         render(conn, "edit.html", connection: connection, changeset: changeset)
+        #     end
+
+        #   {:error, reason} ->
+        #     conn
+        #     |> put_flash(:error, "Failed to parse YAML: #{reason}")
+        #     |> render("edit.html", connection: connection, changeset: Connection.changeset(connection, connection_params))
+
+        case Input.update_connection(connection, connection_params) do
+          {:ok, connection} ->
+            conn
+            |> put_flash(:info, "Connection updated successfully.")
+            |> redirect(to: ~p"/admin/connections/#{connection}")
+
+          {:error, %Ecto.Changeset{} = changeset} ->
+            connection_types = ConnectionTypes.all()
+
+            render(conn, :edit,
+              connection: connection,
+              changeset: changeset,
+              connection_types: connection_types
+            )
+        end
     end
   end
 
   def delete(conn, %{"id" => id}) do
-    connection = Input.get_connection!(id)
+    connection = Input.get_connection_map!(id)
     {:ok, _connection} = Input.delete_connection(connection)
     Input.Connection.terminate(connection)
 

lib/cocktailparty_web/controllers/admin/connection_html.ex

@@ -7,6 +7,7 @@ defmodule CocktailpartyWeb.Admin.ConnectionHTML do
   Renders a connection form.
   """
   attr :changeset, Ecto.Changeset, required: true
+  attr :connection_types, :list, required: true
   attr :action, :string, required: true
 
   def connection_form(assigns)

lib/cocktailparty_web/controllers/admin/connection_html/connection_form.html.heex

@@ -3,10 +3,15 @@
     Oops, something went wrong! Please check the errors below.
   </.error>
   <.input field={f[:name]} type="text" label="Name" />
-  <.input field={f[:port]} type="text" label="port" />
+  <.input
+    field={f[:type]}
+    type="select"
+    label="Type"
+    options={Enum.map(@connection_types, fn {label, string} -> {label, string} end)}
+  />
   <.input field={f[:enabled]} type="checkbox" label="Enabled" />
   <.input field={f[:sink]} type="checkbox" label="Sink" />
-  <.input field={f[:config]} type="text" label="config" />
+  <.input field={f[:config]} type="textarea" label="Config" />
   <:actions>
     <.button>Save connection</.button>
   </:actions>

lib/cocktailparty_web/controllers/admin/connection_html/edit.html.heex

@@ -3,6 +3,6 @@
   <:subtitle>Use this form to manage connection records in your database.</:subtitle>
 </.header>
 
-<.connection_form changeset={@changeset} action={~p"/admin/connections/#{@connection}"} />
+<.connection_form changeset={@changeset} action={~p"/admin/connections/#{@connection}"} connection_types={@connection_types} />
 
 <.back navigate={~p"/admin/connections"}>Back to connections</.back>

lib/cocktailparty_web/controllers/admin/connection_html/new.html.heex

@@ -3,6 +3,6 @@
   <:subtitle>Use this form to manage connection records in your database.</:subtitle>
 </.header>
 
-<.connection_form changeset={@changeset} action={~p"/admin/connections"} />
+<.connection_form changeset={@changeset} action={~p"/admin/connections"} connection_types={@connection_types} />
 
 <.back navigate={~p"/admin/connections"}>Back to connections</.back>

lib/cocktailparty_web/controllers/admin/connection_html/show.html.heex

@@ -28,6 +28,33 @@
       offline
     </div>
   </:item>
+  <%!-- <:item title="Config"><%= @connection.config %></:item> --%>
+
+  <:item title="Config">
+    <table class="sm:w-full">
+      <thead class="text-left text-[0.8125rem] leading-6 text-zinc-500">
+        <tr>
+          <th class="p-0 pr-6 font-normal text-center">key</th>
+          <th class="p-0 pr-6 font-normal text-center">value</th>
+        </tr>
+      </thead>
+
+      <%= for {key, value} <- @connection.config do %>
+        <tr>
+          <td>
+            <div class="block pr-6 text-center">
+              <%= key %>
+            </div>
+          </td>
+          <td>
+            <div class="block pr-6 text-center">
+              <%= value %>
+            </div>
+          </td>
+        </tr>
+      <% end %>
+    </table>
+  </:item>
 </.list>
 
 <.back navigate={~p"/admin/connections"}>Back to connections</.back>

lib/cocktailparty_web/controllers/admin/sink_controller.ex

@@ -24,7 +24,7 @@ defmodule CocktailpartyWeb.Admin.SinkController do
         |> redirect(to: ~p"/admin/connections")
 
       _ ->
-        render(conn, :new, changeset: changeset, redis_instances: instances, users: users)
+        render(conn, :new, changeset: changeset, connections: instances, users: users)
     end
   end
 
@@ -41,7 +41,7 @@ defmodule CocktailpartyWeb.Admin.SinkController do
         # get list of users
         users = SinkCatalog.list_authorized_users()
 
-        render(conn, :new, changeset: changeset, redis_instances: instances, users: users)
+        render(conn, :new, changeset: changeset, connections: instances, users: users)
     end
   end
 
@@ -70,7 +70,7 @@ defmodule CocktailpartyWeb.Admin.SinkController do
         render(conn, :edit,
           sink: sink,
           changeset: changeset,
-          redis_instances: instances,
+          connections: instances,
           users: users
         )
     end
@@ -94,7 +94,7 @@ defmodule CocktailpartyWeb.Admin.SinkController do
         render(conn, :edit,
           sink: sink,
           changeset: changeset,
-          redis_instances: instances,
+          connections: instances,
           users: users
         )
     end

lib/cocktailparty_web/controllers/admin/sink_html.ex

@@ -8,7 +8,7 @@ defmodule CocktailpartyWeb.Admin.SinkHTML do
   """
   attr :changeset, Ecto.Changeset, required: true
   attr :action, :string, required: true
-  attr :redis_instances, :list, required: true
+  attr :connections, :list, required: true
   attr :users, :list, required: true
 
   def sink_form(assigns)