Stream live upload chunks directly to the target file system (#2111)

lib/livebook_web/helpers.ex

@@ -82,4 +82,26 @@ defmodule LivebookWeb.Helpers do
   def format_datetime_relatively(date) do
     date |> DateTime.to_naive() |> Livebook.Utils.Time.time_ago_in_words()
   end
+
+  @doc """
+  Returns a list of human readable messages for all upload and upload
+  entry errors.
+  """
+  @spec upload_error_messages(Phoenix.LiveView.UploadConfig.t()) :: list(String.t())
+  def upload_error_messages(upload) do
+    errors = upload_errors(upload) ++ Enum.flat_map(upload.entries, &upload_errors(upload, &1))
+    Enum.map(errors, &upload_error_to_string/1)
+  end
+
+  @doc """
+  Converts an upload or entry error to string.
+  """
+  @spec upload_error_to_string(term()) :: String.t()
+  def upload_error_to_string(:too_large), do: "Too large"
+  def upload_error_to_string(:too_many_files), do: "You have selected too many files"
+  def upload_error_to_string(:not_accepted), do: "You have selected an unacceptable file type"
+
+  def upload_error_to_string({:writer_failure, message}) when is_binary(message) do
+    Livebook.Utils.upcase_first(message)
+  end
 end

lib/livebook_web/live/file_select_component.ex

@@ -51,7 +51,11 @@ defmodule LivebookWeb.FileSelectComponent do
        accept: :any,
        auto_upload: true,
        max_entries: 1,
-       progress: &handle_progress/3
+       progress: &handle_progress/3,
+       writer: fn _name, entry, socket ->
+         file = FileSystem.File.resolve(socket.assigns.current_dir, entry.client_name)
+         {LivebookWeb.FileSystemWriter, [file: file]}
+       end
      )}
   end
 
@@ -421,19 +425,7 @@ defmodule LivebookWeb.FileSelectComponent do
   end
 
   defp handle_progress(:folder, entry, socket) when entry.done? do
-    consume_uploaded_entries(socket, :folder, fn %{path: file_path}, entry ->
-      content = File.read!(file_path)
-
-      file_path =
-        FileSystem.File.resolve(
-          socket.assigns.current_dir,
-          entry.client_name
-        )
-
-      FileSystem.File.write(file_path, content)
-      {:ok, :ok}
-    end)
-
+    :ok = consume_uploaded_entry(socket, entry, fn %{} -> {:ok, :ok} end)
     {:noreply, update_file_infos(socket, true)}
   end
 

lib/livebook_web/live/file_system_writer.ex

@@ -0,0 +1,53 @@
+defmodule LivebookWeb.FileSystemWriter do
+  @moduledoc false
+
+  # Custom writer for live uploads, uploading directly using the
+  # `Livebook.FileSystem` abstraction.
+  #
+  # ## Options
+  #
+  #   * `:file` (required) - `%Livebook.FileSystem.File{}` to upload
+  #     the contents to
+  #
+
+  @behaviour Phoenix.LiveView.UploadWriter
+
+  @impl true
+  def init(opts) do
+    file = Keyword.fetch!(opts, :file)
+
+    %{file_system: file_system, path: path} = file
+
+    with {:ok, write_state} <- Livebook.FileSystem.write_stream_init(file_system, path, []) do
+      {:ok, %{file: file, write_state: write_state}}
+    end
+  end
+
+  @impl true
+  def meta(state) do
+    %{file: state.file}
+  end
+
+  @impl true
+  def write_chunk(chunk, state) do
+    case Livebook.FileSystem.write_stream_chunk(state.file.file_system, state.write_state, chunk) do
+      {:ok, write_state} -> {:ok, %{state | write_state: write_state}}
+      {:error, message} -> {:error, message, state}
+    end
+  end
+
+  @impl true
+  def close(state, :done) do
+    case Livebook.FileSystem.write_stream_finish(state.file.file_system, state.write_state) do
+      :ok -> {:ok, state}
+      {:error, message} -> {:error, message}
+    end
+  end
+
+  def close(state, _reason) do
+    case Livebook.FileSystem.write_stream_halt(state.file.file_system, state.write_state) do
+      :ok -> {:ok, state}
+      {:error, message} -> {:error, message}
+    end
+  end
+end

lib/livebook_web/live/session_live/add_file_entry_upload_component.ex

@@ -9,8 +9,16 @@ defmodule LivebookWeb.SessionLive.AddFileEntryUploadComponent do
   def mount(socket) do
     {:ok,
      socket
-     |> assign(changeset: changeset(), error_message: nil)
-     |> allow_upload(:file, accept: :any, max_entries: 1, max_file_size: 100_000_000_000)}
+     |> assign(changeset: changeset())
+     |> allow_upload(:file,
+       accept: :any,
+       max_entries: 1,
+       max_file_size: 100_000_000_000,
+       writer: fn _name, _entry, socket ->
+         file = file_entry_file(socket)
+         {LivebookWeb.FileSystemWriter, [file: file]}
+       end
+     )}
   end
 
   defp changeset(attrs \\ %{}) do
@@ -26,8 +34,10 @@ defmodule LivebookWeb.SessionLive.AddFileEntryUploadComponent do
   def render(assigns) do
     ~H"""
     <div>
-      <div :if={@error_message} class="mb-6 error-box">
-        <%= @error_message %>
+      <div class="mb-6 flex flex-col gap-2">
+        <div :for={message <- upload_error_messages(@uploads.file)} class="error-box">
+          <%= message %>
+        </div>
       </div>
       <.form
         :let={f}
@@ -96,10 +106,7 @@ defmodule LivebookWeb.SessionLive.AddFileEntryUploadComponent do
   end
 
   def handle_event("clear_file", %{"ref" => ref}, socket) do
-    {:noreply,
-     socket
-     |> cancel_upload(:file, ref)
-     |> assign(error_message: nil)}
+    {:noreply, cancel_upload(socket, :file, ref)}
   end
 
   def handle_event("add", %{"data" => data}, socket) do
@@ -108,29 +115,21 @@ defmodule LivebookWeb.SessionLive.AddFileEntryUploadComponent do
     |> apply_action(:insert)
     |> case do
       {:ok, data} ->
-        %{files_dir: files_dir} = socket.assigns.session
-
-        [upload_result] =
-          consume_uploaded_entries(socket, :file, fn %{path: path}, _entry ->
-            upload_file = FileSystem.File.local(path)
-            destination_file = FileSystem.File.resolve(files_dir, data.name)
-            result = FileSystem.File.copy(upload_file, destination_file)
-            {:ok, result}
-          end)
-
-        case upload_result do
-          :ok ->
-            file_entry = %{name: data.name, type: :attachment}
-            Livebook.Session.add_file_entries(socket.assigns.session.pid, [file_entry])
-            send(self(), {:file_entry_uploaded, file_entry})
-            {:noreply, push_patch(socket, to: ~p"/sessions/#{socket.assigns.session.id}")}
-
-          {:error, message} ->
-            {:noreply, assign(socket, error_message: message)}
-        end
+        [:ok] =
+          consume_uploaded_entries(socket, :file, fn %{}, _entry -> {:ok, :ok} end)
+
+        file_entry = %{name: data.name, type: :attachment}
+        Livebook.Session.add_file_entries(socket.assigns.session.pid, [file_entry])
+        send(self(), {:file_entry_uploaded, file_entry})
+        {:noreply, push_patch(socket, to: ~p"/sessions/#{socket.assigns.session.id}")}
 
       {:error, changeset} ->
         {:noreply, assign(socket, changeset: changeset)}
     end
   end
+
+  defp file_entry_file(socket) do
+    data = apply_changes(socket.assigns.changeset)
+    FileSystem.File.resolve(socket.assigns.session.files_dir, data.name)
+  end
 end

lib/livebook_web/live/session_live/insert_image_component.ex

@@ -9,11 +9,15 @@ defmodule LivebookWeb.SessionLive.InsertImageComponent do
   def mount(socket) do
     {:ok,
      socket
-     |> assign(changeset: changeset(), error_message: nil)
+     |> assign(changeset: changeset())
      |> allow_upload(:image,
        accept: ~w(.jpg .jpeg .png .gif .svg),
        max_entries: 1,
-       max_file_size: 5_000_000
+       max_file_size: 5_000_000,
+       writer: fn _name, _entry, socket ->
+         file = file_entry_file(socket)
+         {LivebookWeb.FileSystemWriter, [file: file]}
+       end
      )}
   end
 
@@ -33,11 +37,10 @@ defmodule LivebookWeb.SessionLive.InsertImageComponent do
       <h3 class="text-2xl font-semibold text-gray-800">
         Insert image
       </h3>
-      <div :if={@uploads.image.errors != []} class="error-box">
-        Invalid image file. The image must be either GIF, JPEG, SVG or PNG and cannot exceed 5MB in size.
-      </div>
-      <div :if={@error_message} class="error-box">
-        <%= @error_message %>
+      <div class="flex flex-col gap-2">
+        <div :for={message <- upload_error_messages(@uploads.image)} class="error-box">
+          <%= message %>
+        </div>
       </div>
       <div :for={entry <- @uploads.image.entries}>
         <.live_img_preview entry={entry} class="max-h-80 m-auto" />
@@ -108,10 +111,7 @@ defmodule LivebookWeb.SessionLive.InsertImageComponent do
   end
 
   def handle_event("clear_file", %{"ref" => ref}, socket) do
-    {:noreply,
-     socket
-     |> cancel_upload(:image, ref)
-     |> assign(error_message: nil)}
+    {:noreply, cancel_upload(socket, :image, ref)}
   end
 
   def handle_event("save", %{"data" => data}, socket) do
@@ -120,35 +120,22 @@ defmodule LivebookWeb.SessionLive.InsertImageComponent do
     |> apply_action(:insert)
     |> case do
       {:ok, data} ->
-        %{files_dir: files_dir} = socket.assigns.session
-
-        [upload_result] =
-          consume_uploaded_entries(socket, :image, fn %{path: path}, _entry ->
-            upload_file = FileSystem.File.local(path)
-            destination_file = FileSystem.File.resolve(files_dir, data.name)
-
-            result =
-              with :ok <- FileSystem.File.copy(upload_file, destination_file) do
-                {:ok, data.name}
-              end
+        [:ok] =
+          consume_uploaded_entries(socket, :image, fn %{}, _entry -> {:ok, :ok} end)
 
-            {:ok, result}
-          end)
-
-        case upload_result do
-          {:ok, filename} ->
-            file_entry = %{name: filename, type: :attachment}
-            Livebook.Session.add_file_entries(socket.assigns.session.pid, [file_entry])
-            url = "files/#{URI.encode(filename, &URI.char_unreserved?/1)}"
-            send(self(), {:insert_image_complete, socket.assigns.insert_image_metadata, url})
-            {:noreply, push_patch(socket, to: socket.assigns.return_to)}
-
-          {:error, message} ->
-            {:noreply, assign(socket, error_message: message)}
-        end
+        file_entry = %{name: data.name, type: :attachment}
+        Livebook.Session.add_file_entries(socket.assigns.session.pid, [file_entry])
+        url = "files/#{URI.encode(data.name, &URI.char_unreserved?/1)}"
+        send(self(), {:insert_image_complete, socket.assigns.insert_image_metadata, url})
+        {:noreply, push_patch(socket, to: socket.assigns.return_to)}
 
       {:error, changeset} ->
         {:noreply, assign(socket, changeset: changeset)}
     end
   end
+
+  defp file_entry_file(socket) do
+    data = apply_changes(socket.assigns.changeset)
+    FileSystem.File.resolve(socket.assigns.session.files_dir, data.name)
+  end
 end

mix.exs

@@ -92,7 +92,8 @@ defmodule Livebook.MixProject do
     [
       {:phoenix, "~> 1.7.0"},
       {:phoenix_html, "~> 3.0"},
-      {:phoenix_live_view, "~> 0.19.0"},
+      # {:phoenix_live_view, "~> 0.19.0"},
+      {:phoenix_live_view, github: "phoenixframework/phoenix_live_view", override: true},
       {:phoenix_live_dashboard, "~> 0.8.0"},
       {:telemetry_metrics, "~> 0.4"},
       {:telemetry_poller, "~> 1.0"},

mix.lock

@@ -24,7 +24,7 @@
   "phoenix_html": {:hex, :phoenix_html, "3.3.1", "4788757e804a30baac6b3fc9695bf5562465dd3f1da8eb8460ad5b404d9a2178", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "bed1906edd4906a15fd7b412b85b05e521e1f67c9a85418c55999277e553d0d3"},
   "phoenix_live_dashboard": {:hex, :phoenix_live_dashboard, "0.8.0", "0b3158b5b198aa444473c91d23d79f52fb077e807ffad80dacf88ce078fa8df2", [:mix], [{:ecto, "~> 3.6.2 or ~> 3.7", [hex: :ecto, repo: "hexpm", optional: true]}, {:ecto_mysql_extras, "~> 0.5", [hex: :ecto_mysql_extras, repo: "hexpm", optional: true]}, {:ecto_psql_extras, "~> 0.7", [hex: :ecto_psql_extras, repo: "hexpm", optional: true]}, {:ecto_sqlite3_extras, "~> 1.1.7", [hex: :ecto_sqlite3_extras, repo: "hexpm", optional: true]}, {:mime, "~> 1.6 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:phoenix_live_view, "~> 0.19.0", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6 or ~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "87785a54474fed91a67a1227a741097eb1a42c2e49d3c0d098b588af65cd410d"},
   "phoenix_live_reload": {:hex, :phoenix_live_reload, "1.4.1", "2aff698f5e47369decde4357ba91fc9c37c6487a512b41732818f2204a8ef1d3", [:mix], [{:file_system, "~> 0.2.1 or ~> 0.3", [hex: :file_system, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.4", [hex: :phoenix, repo: "hexpm", optional: false]}], "hexpm", "9bffb834e7ddf08467fe54ae58b5785507aaba6255568ae22b4d46e2bb3615ab"},
-  "phoenix_live_view": {:hex, :phoenix_live_view, "0.19.4", "dd9ffe3ca0683bdef4f340bcdd2c35a6ee0d581a2696033fc25f52e742618bdc", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix, "~> 1.6.15 or ~> 1.7.0", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 3.3", [hex: :phoenix_html, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.2 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "fd2c666d227476d63af7b8c20e6e61d16f07eb49f924cf4198fca7668156f15b"},
+  "phoenix_live_view": {:git, "https://github.com/phoenixframework/phoenix_live_view.git", "64e22999c2900e2f9266a030ca7a135a042f0645", []},
   "phoenix_pubsub": {:hex, :phoenix_pubsub, "2.1.3", "3168d78ba41835aecad272d5e8cd51aa87a7ac9eb836eabc42f6e57538e3731d", [:mix], [], "hexpm", "bba06bc1dcfd8cb086759f0edc94a8ba2bc8896d5331a1e2c2902bf8e36ee502"},
   "phoenix_template": {:hex, :phoenix_template, "1.0.1", "85f79e3ad1b0180abb43f9725973e3b8c2c3354a87245f91431eec60553ed3ef", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}], "hexpm", "157dc078f6226334c91cb32c1865bf3911686f8bcd6bcff86736f6253e6993ee"},
   "plug": {:hex, :plug, "1.14.2", "cff7d4ec45b4ae176a227acd94a7ab536d9b37b942c8e8fa6dfc0fff98ff4d80", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "842fc50187e13cf4ac3b253d47d9474ed6c296a8732752835ce4a86acdf68d13"},

test/livebook_web/live/session_live_test.exs

@@ -252,6 +252,10 @@ defmodule LivebookWeb.SessionLiveTest do
       )
       |> render_click()
 
+      view
+      |> element(~s/#insert-image-modal form/)
+      |> render_change(%{"data" => %{"name" => "image.jpg"}})
+
       view
       |> file_input(~s/#insert-image-modal form/, :image, [
         %{
@@ -374,6 +378,10 @@ defmodule LivebookWeb.SessionLiveTest do
       view
       |> render_hook("handle_file_drop", %{"section_id" => section_id, "cell_id" => cell_id})
 
+      view
+      |> element(~s{#add-file-entry-form})
+      |> render_change(%{"data" => %{"name" => "image.jpg"}})
+
       view
       |> file_input(~s{#add-file-entry-form}, :file, [
         %{
@@ -1796,6 +1804,16 @@ defmodule LivebookWeb.SessionLiveTest do
 
       {:ok, view, _} = live(conn, ~p"/sessions/#{session.id}/add-file/upload")
 
+      # Validations
+      assert view
+             |> element(~s{#add-file-entry-form})
+             |> render_change(%{"data" => %{"name" => "na me"}}) =~
+               "should contain only alphanumeric characters, dash, underscore and dot"
+
+      assert view
+             |> element(~s{#add-file-entry-form})
+             |> render_change(%{"data" => %{"name" => "image.jpg"}})
+
       view
       |> file_input(~s{#add-file-entry-form}, :file, [
         %{
@@ -1808,12 +1826,6 @@ defmodule LivebookWeb.SessionLiveTest do
       ])
       |> render_upload("image.jpg")
 
-      # Validations
-      assert view
-             |> element(~s{#add-file-entry-form})
-             |> render_change(%{"data" => %{"name" => "na me"}}) =~
-               "should contain only alphanumeric characters, dash, underscore and dot"
-
       view
       |> element(~s{#add-file-entry-form})
       |> render_submit(%{"data" => %{"name" => "image.jpg"}})