cozy cozy sql

lib/elixir_ai/conversation_manager.ex

@@ -20,7 +20,7 @@ defmodule ElixirAi.ConversationManager do
   def init(_) do
     Logger.info("ConversationManager initializing...")
     send(self(), :load_conversations)
-    {:ok, :loading_conversations}
+    {:ok, %{conversations: :loading, subscriptions: MapSet.new()}}
   end
 
   def create_conversation(name, ai_provider_id) do
@@ -39,58 +39,77 @@ defmodule ElixirAi.ConversationManager do
     GenServer.call(@name, {:get_messages, name})
   end
 
-  def handle_call(message, from, :loading_conversations) do
+  def handle_call(message, from, %{conversations: :loading} = state) do
     Logger.warning(
       "Received call #{inspect(message)} from #{inspect(from)} while loading conversations. Retrying after delay."
     )
 
     Process.send_after(self(), {:retry_call, message, from}, 100)
-    {:noreply, :loading_conversations}
+    {:noreply, state}
   end
 
-  def handle_call({:create, name, ai_provider_id}, _from, conversations) do
+  def handle_call(
+        {:create, name, ai_provider_id},
+        _from,
+        %{conversations: conversations, subscriptions: subscriptions} = state
+      ) do
     if Map.has_key?(conversations, name) do
-      {:reply, {:error, :already_exists}, conversations}
+      {:reply, {:error, :already_exists}, state}
     else
       case Conversation.create(name, ai_provider_id) do
         :ok ->
-          case start_and_subscribe(name) do
-            {:ok, _pid} = ok -> {:reply, ok, Map.put(conversations, name, [])}
-            error -> {:reply, error, conversations}
+          case start_and_subscribe(name, subscriptions) do
+            {:ok, pid, new_subscriptions} ->
+              {:reply, {:ok, pid},
+               %{
+                 state
+                 | conversations: Map.put(conversations, name, []),
+                   subscriptions: new_subscriptions
+               }}
+
+            {:error, _reason} = error ->
+              {:reply, error, state}
           end
 
         {:error, _} = error ->
-          {:reply, error, conversations}
+          {:reply, error, state}
       end
     end
   end
 
-  def handle_call({:open, name}, _from, conversations) do
+  def handle_call(
+        {:open, name},
+        _from,
+        %{conversations: conversations, subscriptions: subscriptions} = state
+      ) do
     if Map.has_key?(conversations, name) do
-      case start_and_subscribe(name) do
-        {:ok, _pid} = ok -> {:reply, ok, conversations}
-        error -> {:reply, error, conversations}
+      case start_and_subscribe(name, subscriptions) do
+        {:ok, pid, new_subscriptions} ->
+          {:reply, {:ok, pid}, %{state | subscriptions: new_subscriptions}}
+
+        {:error, _reason} = error ->
+          {:reply, error, state}
       end
     else
-      {:reply, {:error, :not_found}, conversations}
+      {:reply, {:error, :not_found}, state}
     end
   end
 
-  def handle_call(:list, _from, conversations) do
+  def handle_call(:list, _from, %{conversations: conversations} = state) do
     keys = Map.keys(conversations)
 
     Logger.debug(
       "list_conversations returning: #{inspect(keys, limit: :infinity, printable_limit: :infinity, binaries: :as_binaries)}"
     )
 
-    {:reply, keys, conversations}
+    {:reply, keys, state}
   end
 
-  def handle_call({:get_messages, name}, _from, conversations) do
-    {:reply, Map.get(conversations, name, []), conversations}
+  def handle_call({:get_messages, name}, _from, %{conversations: conversations} = state) do
+    {:reply, Map.get(conversations, name, []), state}
   end
 
-  def handle_info({:store_message, name, message}, conversations) do
+  def handle_info({:store_message, name, message}, %{conversations: conversations} = state) do
     case Conversation.find_id(name) do
       {:ok, conv_id} ->
         Message.insert(conv_id, message, topic: ElixirAi.ChatRunner.message_topic(name))
@@ -99,17 +118,17 @@ defmodule ElixirAi.ConversationManager do
         :ok
     end
 
-    {:noreply, Map.update(conversations, name, [message], &(&1 ++ [message]))}
+    {:noreply,
+     %{state | conversations: Map.update(conversations, name, [message], &(&1 ++ [message]))}}
   end
 
-  def handle_info(:load_conversations, _conversations) do
+  def handle_info(:load_conversations, state) do
     conversation_list = Conversation.all_names()
     Logger.info("Loaded #{length(conversation_list)} conversations from DB")
 
     conversations = Map.new(conversation_list, fn %{name: name} -> {name, []} end)
     Logger.info("Conversation map keys: #{inspect(Map.keys(conversations))}")
-    # {:ok, conversations}
-    {:noreply, conversations}
+    {:noreply, %{state | conversations: conversations}}
   end
 
   def handle_info({:retry_call, message, from}, state) do
@@ -123,7 +142,7 @@ defmodule ElixirAi.ConversationManager do
     end
   end
 
-  defp start_and_subscribe(name) do
+  defp start_and_subscribe(name, subscriptions) do
     result =
       case Horde.DynamicSupervisor.start_child(
              ElixirAi.ChatRunnerSupervisor,
@@ -135,12 +154,19 @@ defmodule ElixirAi.ConversationManager do
       end
 
     case result do
-      {:ok, _pid} ->
-        Phoenix.PubSub.subscribe(ElixirAi.PubSub, ElixirAi.ChatRunner.message_topic(name))
-        result
+      {:ok, pid} ->
+        new_subscriptions =
+          if MapSet.member?(subscriptions, name) do
+            subscriptions
+          else
+            Phoenix.PubSub.subscribe(ElixirAi.PubSub, ElixirAi.ChatRunner.message_topic(name))
+            MapSet.put(subscriptions, name)
+          end
 
-      _ ->
-        result
+        {:ok, pid, new_subscriptions}
+
+      error ->
+        error
     end
   end
 end

lib/elixir_ai/data/ai_provider.ex

@@ -28,44 +28,44 @@ defmodule ElixirAi.AiProvider do
     sql = "SELECT id, name, model_name FROM ai_providers"
     params = %{}
 
-    case DbHelpers.run_sql(sql, params, "ai_providers") do
-      {:error, :db_error} ->
+    case DbHelpers.run_sql(sql, params, "ai_providers", AiProviderSchema.partial_schema()) do
+      {:error, _} ->
         []
 
-      result ->
-        results =
-          Enum.map(result.rows, fn [id, name, model_name] ->
-            attrs = %{id: id, name: name, model_name: model_name} |> convert_id_to_string()
-
-            case Zoi.parse(AiProviderSchema.partial_schema(), attrs) do
-              {:ok, valid} ->
-                struct(AiProviderSchema, valid)
-
-              {:error, errors} ->
-                Logger.error("Invalid provider data from DB: #{inspect(errors)}")
-                raise ArgumentError, "Invalid provider data: #{inspect(errors)}"
-            end
-          end)
-
-        Logger.debug("AiProvider.all() returning: #{inspect(results)}")
-
-        results
+      rows ->
+        rows
+        |> Enum.map(fn row ->
+          row |> convert_uuid_to_string() |> then(&struct(AiProviderSchema, &1))
+        end)
+        |> tap(&Logger.debug("AiProvider.all() returning: #{inspect(&1)}"))
     end
   end
 
-  # Convert binary UUID to string for frontend
-  defp convert_id_to_string(%{id: id} = provider) when is_binary(id) do
+  defp convert_uuid_to_string(%{id: id} = provider) when is_binary(id) do
     %{provider | id: Ecto.UUID.cast!(id)}
   end
 
-  defp convert_id_to_string(provider), do: provider
+  defp convert_uuid_to_string(provider), do: provider
 
   def create(attrs) do
     now = DateTime.truncate(DateTime.utc_now(), :second)
 
     sql = """
-    INSERT INTO ai_providers (name, model_name, api_token, completions_url, inserted_at, updated_at)
-    VALUES ($(name), $(model_name), $(api_token), $(completions_url), $(inserted_at), $(updated_at))
+    INSERT INTO ai_providers (
+      name,
+      model_name,
+      api_token,
+      completions_url,
+      inserted_at,
+      updated_at
+    ) VALUES (
+      $(name),
+      $(model_name),
+      $(api_token),
+      $(completions_url),
+      $(inserted_at),
+      $(updated_at)
+    )
     """
 
     params = %{
@@ -102,32 +102,10 @@ defmodule ElixirAi.AiProvider do
 
     params = %{"name" => name}
 
-    case DbHelpers.run_sql(sql, params, "ai_providers") do
-      {:error, :db_error} ->
-        {:error, :db_error}
-
-      %{rows: []} ->
-        {:error, :not_found}
-
-      %{rows: [[id, name, model_name, api_token, completions_url] | _]} ->
-        attrs =
-          %{
-            id: id,
-            name: name,
-            model_name: model_name,
-            api_token: api_token,
-            completions_url: completions_url
-          }
-          |> convert_id_to_string()
-
-        case Zoi.parse(AiProviderSchema.schema(), attrs) do
-          {:ok, valid} ->
-            {:ok, struct(AiProviderSchema, valid)}
-
-          {:error, errors} ->
-            Logger.error("Invalid provider data from DB: #{inspect(errors)}")
-            {:error, :invalid_data}
-        end
+    case DbHelpers.run_sql(sql, params, "ai_providers", AiProviderSchema.schema()) do
+      {:error, _} -> {:error, :db_error}
+      [] -> {:error, :not_found}
+      [row | _] -> {:ok, row |> convert_uuid_to_string() |> then(&struct(AiProviderSchema, &1))}
     end
   end
 
@@ -139,9 +117,9 @@ defmodule ElixirAi.AiProvider do
       {:error, :db_error} ->
         {:error, :db_error}
 
-      result ->
-        case result.rows do
-          [[0]] ->
+      rows ->
+        case rows do
+          [%{"count" => 0}] ->
             attrs = %{
               name: "default",
               model_name: Application.fetch_env!(:elixir_ai, :ai_model),

lib/elixir_ai/data/conversation.ex

@@ -34,40 +34,26 @@ defmodule ElixirAi.Conversation do
 
   def all_names do
     sql = """
-    SELECT c.name, p.name, p.model_name, p.api_token, p.completions_url
+    SELECT c.name,
+      json_build_object(
+        'name', p.name,
+        'model_name', p.model_name,
+        'api_token', p.api_token,
+        'completions_url', p.completions_url
+      ) as provider
     FROM conversations c
     LEFT JOIN ai_providers p ON c.ai_provider_id = p.id
     """
 
     params = %{}
 
-    case DbHelpers.run_sql(sql, params, "conversations") do
-      {:error, :db_error} ->
+    case DbHelpers.run_sql(sql, params, "conversations", ConversationInfo.schema()) do
+      {:error, _} ->
         []
 
-      result ->
-        Enum.map(result.rows, fn [name, provider_name, model_name, api_token, completions_url] ->
-          attrs = %{
-            name: name,
-            provider: %{
-              name: provider_name,
-              model_name: model_name,
-              api_token: api_token,
-              completions_url: completions_url
-            }
-          }
-
-          case Zoi.parse(ConversationInfo.schema(), attrs) do
-            {:ok, valid} ->
-              struct(
-                ConversationInfo,
-                Map.put(valid, :provider, struct(Provider, valid.provider))
-              )
-
-            {:error, errors} ->
-              Logger.error("Invalid conversation data: #{inspect(errors)}")
-              raise ArgumentError, "Invalid conversation data: #{inspect(errors)}"
-          end
+      rows ->
+        Enum.map(rows, fn row ->
+          struct(ConversationInfo, Map.put(row, :provider, struct(Provider, row.provider)))
         end)
     end
   end
@@ -76,8 +62,17 @@ defmodule ElixirAi.Conversation do
     case Ecto.UUID.dump(ai_provider_id) do
       {:ok, binary_id} ->
         sql = """
-        INSERT INTO conversations (name, ai_provider_id, inserted_at, updated_at)
-        VALUES ($(name), $(ai_provider_id), $(inserted_at), $(updated_at))
+        INSERT INTO conversations (
+          name,
+          ai_provider_id,
+          inserted_at,
+          updated_at)
+        VALUES (
+          $(name),
+          $(ai_provider_id),
+          $(inserted_at),
+          $(updated_at)
+        )
         """
 
         timestamp = now()
@@ -110,11 +105,11 @@ defmodule ElixirAi.Conversation do
       {:error, :db_error} ->
         {:error, :db_error}
 
-      %{rows: []} ->
+      [] ->
         {:error, :not_found}
 
-      %{rows: [[id] | _]} ->
-        {:ok, id}
+      [row | _] ->
+        {:ok, row["id"]}
     end
   end
 

lib/elixir_ai/data/db_helpers.ex

@@ -2,11 +2,21 @@ defmodule ElixirAi.Data.DbHelpers do
   require Logger
   @get_named_param ~r/\$\((\w+)\)/
 
+  def run_sql(sql, params, topic, schema) do
+    run_sql(sql, params, topic) |> validate_rows(schema, topic)
+  end
+
   def run_sql(sql, params, topic) do
     {sql, params} = named_params_to_positional_params(sql, params)
 
     try do
-      Ecto.Adapters.SQL.query!(ElixirAi.Repo, sql, params)
+      result = Ecto.Adapters.SQL.query!(ElixirAi.Repo, sql, params)
+
+      # Transform rows to maps with column names as keys
+      Enum.map(result.rows, fn row ->
+        Enum.zip(result.columns, row)
+        |> Enum.into(%{})
+      end)
     rescue
       exception ->
         Logger.error("Database error: #{Exception.message(exception)}")
@@ -21,6 +31,31 @@ defmodule ElixirAi.Data.DbHelpers do
     end
   end
 
+  defp validate_rows({:error, :db_error}, _schema, _topic), do: {:error, :db_error}
+
+  defp validate_rows(rows, schema, topic) do
+    rows
+    |> Enum.reduce_while({:ok, []}, fn row, {:ok, acc} ->
+      case Zoi.parse(schema, row, coerce: true) do
+        {:ok, valid} ->
+          {:cont, {:ok, [valid | acc]}}
+
+        {:error, errors} ->
+          Logger.error("Schema validation error: #{inspect(errors)}")
+          {:halt, {:error, :validation_error}}
+      end
+    end)
+    |> then(fn
+      {:ok, valid_rows} ->
+        Enum.reverse(valid_rows)
+
+      error ->
+        Logger.error("Validation error: #{inspect(error)}")
+        Phoenix.PubSub.broadcast(ElixirAi.PubSub, topic, {:sql_result_validation_error, error})
+        error
+    end)
+  end
+
   def named_params_to_positional_params(query, params) do
     param_occurrences = Regex.scan(@get_named_param, query)
 

lib/elixir_ai/data/message.ex

@@ -31,17 +31,9 @@ defmodule ElixirAi.Message do
       {:error, :db_error} ->
         []
 
-      result ->
-        Enum.map(result.rows, fn row ->
-          raw = %{
-            role: Enum.at(row, 0),
-            content: Enum.at(row, 1),
-            reasoning_content: Enum.at(row, 2),
-            tool_calls: Enum.at(row, 3),
-            tool_call_id: Enum.at(row, 4)
-          }
-
-          decoded = decode_message(raw)
+      rows ->
+        Enum.map(rows, fn row ->
+          decoded = decode_message(row)
 
           case Zoi.parse(MessageSchema.schema(), decoded) do
             {:ok, _valid} ->
@@ -69,9 +61,22 @@ defmodule ElixirAi.Message do
       when is_binary(conversation_id) and byte_size(conversation_id) == 16 do
     sql = """
     INSERT INTO messages (
-      conversation_id, role, content, reasoning_content,
-      tool_calls, tool_call_id, inserted_at
-    ) VALUES ($(conversation_id), $(role), $(content), $(reasoning_content), $(tool_calls), $(tool_call_id), $(inserted_at))
+      conversation_id,
+      role,
+      content,
+      reasoning_content,
+      tool_calls,
+      tool_call_id,
+      inserted_at
+    ) VALUES (
+      $(conversation_id),
+      $(role),
+      $(content),
+      $(reasoning_content),
+      $(tool_calls),
+      $(tool_call_id),
+      $(inserted_at)
+    )
     """
 
     params = %{
@@ -114,6 +119,7 @@ defmodule ElixirAi.Message do
 
   defp decode_message(row) do
     row
+    |> Map.new(fn {k, v} -> {String.to_existing_atom(k), v} end)
     |> Map.update!(:role, &String.to_existing_atom/1)
     |> Map.update(:tool_calls, nil, fn
       nil ->