Handling Mux Video Uploads in LiveView with Process Messaging

I recently faced a challenge where multiple users needed to upload video files as part of a form submission. We integrated Mux to handle these uploads and had to ensure that the form data was updated once the upload was complete. Initially, I attempted to use Phoenix PubSub for broadcasting updates, but I encountered difficulties in restricting events to only the relevant users.

The approach that ultimately worked was to send a direct message to the LiveView process that initiated the upload once Mux’s webhook event was received. Here’s how I implemented it:

1. Registering the LiveView Process

At the start of a direct upload, I register the responsible LiveView process in a registry using the upload_id as its name. This allows me to retrieve the LiveView process by the upload_id later and send it a message when the upload is complete.

# Initiating the video upload and registering the LiveView process

def create_video_upload(socket, filename, pid) do
  with {:ok, upload_url, upload_id} <- Client.create_direct_upload(),
       {:ok, _pid} <- WebhookHandler.register(upload_id, pid) do
    ...
  end
end

# Webhook Handler: Register the PID in a registry

def register(upload_id, pid) do
  Registry.register(@registry_name, upload_id, pid)
end

2. Handling the Webhook Event

When Mux sends a webhook event indicating that the video processing is complete, I retrieve the registered LiveView process using the upload_id from the registry and send it a message containing the relevant video data.

# Controller: Handling the webhook event from Mux

def handle(
      conn,
      %{"type" => "video.asset.ready", "data" => %{"id" => asset_id, "upload_id" => upload_id}} = event
    ) do
  playback_id = get_playback(event)

  WebhookHandler.handle_webhook_response(%{
    asset_id: asset_id,
    playback_id: playback_id,
    upload_id: upload_id
  })
  ...
end

# Webhook Handler: Lookup the process and send a message

def handle_webhook_response(response) do
  case Registry.lookup(@registry_name, response.upload_id) do
    [{_pid, pid}] ->
      send(pid, {:video_ready, response})

    [] ->
      Logger.error("No PID found for upload_id: #{response.upload_id}")
  end
end

3. Updating the LiveView Component

Finally, in the LiveView module, I handle the incoming message and update the form accordingly.

# LiveView: Handling the incoming video_ready message

def handle_info({:video_ready, %{asset_id: asset_id, playback_id: playback_id}}, socket) do
  ... # Update the form state
end

This approach ensures that each user receives only the updates related to their video uploads without unnecessary broadcasting. By leveraging Elixir’s process messaging and the Registry module, we efficiently manage LiveView state updates in response to external events like Mux webhooks.