Initial Commit

lib/river_connect_web/channels/audio_channel.ex

@@ -0,0 +1,79 @@
+defmodule RiverConnectWeb.AudioChannel do
+  use RiverConnectWeb, :channel
+  alias RiverConnect.Audio
+
+  @impl true
+  def join("audio:lobby", _payload, socket) do
+    {:ok, socket}
+  end
+
+  @impl true
+  def handle_in("audio_start", %{"id" => id}, socket) do
+    # Ensure upload directory exists
+    upload_path = Application.app_dir(:river_connect, "priv/static/uploads")
+    File.mkdir_p!(upload_path)
+
+    user_id = socket.assigns[:user_id] || "guest"
+
+    # Create DB record immediately with "recording" status
+    {:ok, message} =
+      Audio.create_message(%{
+        user_id: user_id,
+        file_path: "/uploads/#{id}.webm",
+        status: "recording"
+      })
+
+    # Store temporary file path and message_id in socket assigns
+    temp_file_path = Path.join(upload_path, "#{id}.webm")
+
+    # Broadcast start and new message to other listeners
+    broadcast_from!(socket, "audio_start", %{id: id})
+    broadcast!(socket, "audio_message_created", %{message: message})
+
+    {:noreply,
+     assign(socket, :current_recording, %{id: id, path: temp_file_path, message_id: message.id})}
+  end
+
+  @impl true
+  def handle_in("audio_chunk", {:binary, payload}, socket) do
+    case socket.assigns[:current_recording] do
+      %{path: path} ->
+        # Append binary chunk to file
+        File.write!(path, payload, [:append])
+        # Broadcast chunk to other listeners for live playback
+        broadcast_from!(socket, "audio_chunk", %{data: Base.encode64(payload)})
+        {:noreply, socket}
+
+      _ ->
+        {:noreply, socket}
+    end
+  end
+
+  @impl true
+  def handle_in("audio_end", %{"duration_ms" => duration_ms}, socket) do
+    case socket.assigns[:current_recording] do
+      %{message_id: message_id} ->
+        message = Audio.get_message!(message_id)
+
+        # Update DB record to "processing" status
+        {:ok, message} =
+          Audio.update_message(message, %{
+            duration_ms: duration_ms,
+            status: "processing"
+          })
+
+        # Queue Oban job for processing
+        %{id: message.id}
+        |> RiverConnect.Workers.ProcessRecording.new()
+        |> Oban.insert!()
+
+        # Notify UI that message status changed
+        broadcast!(socket, "audio_message_ready", %{id: message.id})
+
+        {:noreply, assign(socket, :current_recording, nil)}
+
+      _ ->
+        {:noreply, socket}
+    end
+  end
+end

lib/river_connect_web/channels/user_socket.ex

@@ -0,0 +1,37 @@
+defmodule RiverConnectWeb.UserSocket do
+  use Phoenix.Socket
+
+  ## Channels
+  channel "audio:*", RiverConnectWeb.AudioChannel
+
+  # Socket params are passed from the client and can
+  # be used to verify and authenticate a user. After
+  # verification, you can put default assigns into
+  # the socket that will be set for all channels, ie
+  #
+  #     {:ok, assign(socket, :user_id, verified_user_id)}
+  #
+  # To deny connection, return `:error`.
+  #
+  # See `Phoenix.Token` documentation for examples in
+  # performing token verification on connect.
+  @impl true
+  def connect(params, socket, _connect_info) do
+    # Simple user identification using client-provided ID or generating one
+    user_id = params["user_id"] || Ecto.UUID.generate()
+    {:ok, assign(socket, :user_id, user_id)}
+  end
+
+  # Socket id's are topics that allow you to identify all sockets for a given user:
+  #
+  #     def id(socket), do: "user_socket:#{socket.assigns.user_id}"
+  #
+  # Would allow you to broadcast a "disconnect" event and terminate
+  # all active sockets and channels for a given user:
+  #
+  #     Elixir.RiverConnectWeb.Endpoint.broadcast("user_socket:#{user.id}", "disconnect", %{})
+  #
+  # Returning `nil` makes this socket anonymous.
+  @impl true
+  def id(_socket), do: nil
+end