Merge branch 'develop'

Author: ku1ik

.github/workflows/release.yml

@@ -30,12 +30,6 @@ jobs:
           echo IMAGE_ID=$IMAGE_ID >> $GITHUB_OUTPUT
           echo IMAGE_TAG=$IMAGE_TAG >> $GITHUB_OUTPUT
 
-      - name: Create the release
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        run: |
-          gh release create ${{ github.ref_name }} --draft --verify-tag --title ${{ steps.vars.outputs.VERSION }}
-
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
 
@@ -58,3 +52,9 @@ jobs:
           platforms: linux/amd64,linux/arm64
           build-args: |
             VERSION=${{ steps.vars.outputs.VERSION }}
+
+      - name: Create the release
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          gh release create ${{ github.ref_name }} --draft --verify-tag --title ${{ steps.vars.outputs.VERSION }}

lib/asciinema/streaming/parser.ex

@@ -9,5 +9,6 @@ defmodule Asciinema.Streaming.Parser do
   def get("raw"), do: %{impl: Parser.Raw, state: Parser.Raw.init()}
   def get("v0.alis"), do: %{impl: Parser.AlisV0, state: Parser.AlisV0.init()}
   def get("v1.alis"), do: %{impl: Parser.AlisV1, state: Parser.AlisV1.init()}
-  def get("v2.asciicast"), do: %{impl: Parser.Json, state: Parser.Json.init()}
+  def get("v2.asciicast"), do: %{impl: Parser.AsciicastV2, state: Parser.AsciicastV2.init()}
+  def get("v3.asciicast"), do: %{impl: Parser.AsciicastV3, state: Parser.AsciicastV3.init()}
 end

lib/asciinema/streaming/parser/json.ex renamed to lib/asciinema/streaming/parser/asciicast_v2.ex

@@ -1,4 +1,4 @@
-defmodule Asciinema.Streaming.Parser.Json do
+defmodule Asciinema.Streaming.Parser.AsciicastV2 do
   @moduledoc """
   asciicast v2 compatible stream protocol parser.
   """

lib/asciinema/streaming/parser/asciicast_v3.ex

@@ -0,0 +1,119 @@
+defmodule Asciinema.Streaming.Parser.AsciicastV3 do
+  @moduledoc """
+  asciicast v3 compatible stream protocol parser.
+  """
+
+  alias Asciinema.Colors
+
+  @behaviour Asciinema.Streaming.Parser
+
+  def name, do: "v3.asciicast"
+
+  def init, do: %{first: true, last_event_id: 0, time_offset: 0}
+
+  def parse({:text, "\n"}, state), do: {:ok, [], state}
+  def parse({:text, "#" <> _}, state), do: {:ok, [], state}
+
+  def parse({:text, payload}, state) do
+    case Jason.decode(payload) do
+      {:ok, message} ->
+        handle_message(message, state)
+
+      {:error, %Jason.DecodeError{} = reason} ->
+        {:error, "JSON decode error: #{Jason.DecodeError.message(reason)}"}
+    end
+  end
+
+  def handle_message(%{"term" => %{"cols" => cols, "rows" => rows}} = header, state)
+      when is_integer(cols) and is_integer(rows) do
+    commands = [
+      init: %{
+        last_id: state.last_event_id,
+        time: 0,
+        term_size: {cols, rows},
+        term_theme: parse_theme(get_in(header, ["term", "theme"]))
+      }
+    ]
+
+    {:ok, commands, %{state | first: false}}
+  end
+
+  def handle_message(_message, %{first: true}) do
+    {:error, :init_expected}
+  end
+
+  def handle_message([time, "o", text], state) when is_number(time) and is_binary(text) do
+    {id, state} = get_next_id(state)
+    time = state.time_offset + time_as_micros(time)
+
+    {:ok, [output: %{id: id, time: time, text: text}], %{state | time_offset: time}}
+  end
+
+  def handle_message([time, "i", text], state) when is_number(time) and is_binary(text) do
+    {id, state} = get_next_id(state)
+    time = state.time_offset + time_as_micros(time)
+
+    {:ok, [input: %{id: id, time: time, text: text}], %{state | time_offset: time}}
+  end
+
+  def handle_message([time, "r", data], state) when is_number(time) and is_binary(data) do
+    {id, state} = get_next_id(state)
+    time = state.time_offset + time_as_micros(time)
+    [cols, rows] = String.split(data, "x")
+    cols = String.to_integer(cols)
+    rows = String.to_integer(rows)
+
+    {:ok, [resize: %{id: id, time: time, term_size: {cols, rows}}], %{state | time_offset: time}}
+  end
+
+  def handle_message([time, "m", label], state) when is_number(time) and is_binary(label) do
+    {id, state} = get_next_id(state)
+    time = state.time_offset + time_as_micros(time)
+
+    {:ok, [marker: %{id: id, time: time, label: label}], %{state | time_offset: time}}
+  end
+
+  def handle_message([time, "x", status], state) when is_number(time) and is_binary(status) do
+    {id, state} = get_next_id(state)
+    time = state.time_offset + time_as_micros(time)
+    status = String.to_integer(status)
+
+    {:ok, [exit: %{id: id, time: time, status: status}], %{state | time_offset: time}}
+  end
+
+  def handle_message([time, type, data], state)
+      when is_number(time) and is_binary(type) and is_binary(data) do
+    time = state.time_offset + time
+
+    {:ok, [], %{state | time_offset: time}}
+  end
+
+  def handle_message(_message, _state) do
+    {:error, :message_invalid}
+  end
+
+  defp parse_theme(nil), do: nil
+
+  defp parse_theme(%{"fg" => fg, "bg" => bg, "palette" => palette}) do
+    palette =
+      palette
+      |> String.split(":")
+      |> Enum.map(&Colors.parse/1)
+
+    true = length(palette) in [8, 16]
+
+    %{
+      fg: Colors.parse(fg),
+      bg: Colors.parse(bg),
+      palette: palette
+    }
+  end
+
+  defp get_next_id(state) do
+    id = state.last_event_id + 1
+
+    {id, %{state | last_event_id: id}}
+  end
+
+  defp time_as_micros(time), do: round(time * 1_000_000)
+end

lib/asciinema/streaming/parser/raw.ex

@@ -7,7 +7,7 @@ defmodule Asciinema.Streaming.Parser.Raw do
 
   def init, do: %{first: true, start_time: nil, last_event_id: 0}
 
-  def parse({:binary, text}, %{first: true} = state) do
+  def parse({_type, text}, %{first: true} = state) do
     size = size_from_resize_seq(text) || size_from_script_start_message(text) || @default_size
 
     commands = [
@@ -17,7 +17,7 @@ defmodule Asciinema.Streaming.Parser.Raw do
     {:ok, commands, %{state | first: false, start_time: Timex.now()}}
   end
 
-  def parse({:binary, text}, state) do
+  def parse({_type, text}, state) do
     {id, state} = get_next_id(state)
     time = stream_time(state)
 

lib/asciinema_web/controllers/error_json.ex

@@ -10,4 +10,27 @@ defmodule AsciinemaWeb.ErrorJSON do
   def render("404.json", _assigns) do
     %{type: "not_found", message: "Resource not found"}
   end
+
+  def render("413.json", _assigns) do
+    {_, limit, _, _} = AsciinemaWeb.Plug.Parsers.MULTIPART.init([])
+    limit = format_byte_size(limit)
+
+    %{
+      type: "content_too_large",
+      message: "The recording exceeds the server-configured size limit (#{limit})"
+    }
+  end
+
+  defp format_byte_size(size) do
+    cond do
+      rem(size, 1024 * 1024) == 0 ->
+        "#{div(size, 1024 * 1024)} MiB"
+
+      rem(size, 1000 * 1000) == 0 ->
+        "#{div(size, 1000 * 1000)} MB"
+
+      true ->
+        to_string(size)
+    end
+  end
 end

lib/asciinema_web/plug/parsers/multipart.ex

@@ -1,12 +1,13 @@
 defmodule AsciinemaWeb.Plug.Parsers.MULTIPART do
   @multipart Plug.Parsers.MULTIPART
 
-  def init(opts), do: opts
-
-  def parse(conn, "multipart", subtype, headers, opts) do
+  def init(opts) do
     opts = if length = config(:length), do: [{:length, length} | opts], else: opts
-    opts = @multipart.init(opts)
 
+    @multipart.init(opts)
+  end
+
+  def parse(conn, "multipart", subtype, headers, opts) do
     @multipart.parse(conn, "multipart", subtype, headers, opts)
   end
 

lib/asciinema_web/stream_consumer_socket.ex

@@ -41,7 +41,7 @@ defmodule AsciinemaWeb.StreamConsumerSocket do
 
   def websocket_init(%{token: token, user_id: user_id}) do
     with {:ok, stream} <- fetch_stream(token),
-         :ok <- authorize(stream, user_id) do
+         :ok <- authorize(stream, user_id, token) do
       Logger.info("consumer/#{stream.id}: connected")
       state = %{stream_id: stream.id, init: false, last_event_time: 0.0}
       StreamServer.subscribe(stream.id, [:output, :input, :resize, :marker, :end, :reset])
@@ -134,9 +134,9 @@ defmodule AsciinemaWeb.StreamConsumerSocket do
   end
 
   def websocket_info(%StreamServer.Update{event: :marker} = update, state) do
-    %{time: time, label: label} = update.data
+    %{id: id, time: time, label: label} = update.data
     rel_time = time - state.last_event_time
-    msg = serialize_marker(rel_time, label)
+    msg = serialize_marker(id, rel_time, label)
     state = %{state | last_event_time: time}
 
     {:reply, msg, state}
@@ -190,9 +190,11 @@ defmodule AsciinemaWeb.StreamConsumerSocket do
 
   defp fetch_stream(token), do: OK.required(Streaming.lookup_stream(token), :stream_not_found)
 
-  defp authorize(stream, user_id) do
-    if Authorization.can?(nil, :show, stream) ||
-         Authorization.can?(Accounts.get_user(user_id), :show, stream) do
+  defp authorize(stream, user_id, token) do
+    resource = Map.put(stream, :id, token)
+
+    if Authorization.can?(nil, :show, resource) ||
+         (user_id && Authorization.can?(Accounts.get_user(user_id), :show, resource)) do
       :ok
     else
       {:error, :forbidden}
@@ -239,8 +241,8 @@ defmodule AsciinemaWeb.StreamConsumerSocket do
     {:binary, msg}
   end
 
-  defp serialize_marker(time, label) do
-    msg = <<?m>> <> encode_varint(time) <> serialize_string(label)
+  defp serialize_marker(id, time, label) do
+    msg = <<?m>> <> encode_varint(id) <> encode_varint(time) <> serialize_string(label)
 
     {:binary, msg}
   end

lib/asciinema_web/stream_producer_socket.ex

@@ -60,7 +60,7 @@ defmodule AsciinemaWeb.StreamProducerSocket do
           Logger.info("producer/#{stream.id}: negotiated #{parser.impl.name()} protocol")
           save_protocol(state.stream_id, parser.impl.name())
         else
-          Logger.info("producer/#{stream.id}: no protocol negotiated (legacy client)")
+          Logger.info("producer/#{stream.id}: no protocol negotiated, will try to auto-detect")
         end
 
         {:ok, state}
@@ -70,8 +70,6 @@ defmodule AsciinemaWeb.StreamProducerSocket do
   @impl true
   def websocket_handle(frame, state)
 
-  # legacy clause for CLI 3.0 RC 3 and earlier, which doesn't do protocol negotiation
-  # TODO: remove after release of the final CLI 3.0
   def websocket_handle(message, %{parser: nil} = state) do
     parser = Parser.get(detect_protocol(message))
     Logger.info("producer/#{state.stream_id}: detected #{parser.impl.name()} protocol")
@@ -293,7 +291,7 @@ defmodule AsciinemaWeb.StreamProducerSocket do
     end
   end
 
-  @protos ~w(v1.alis v2.asciicast raw)
+  @protos ~w(v1.alis v2.asciicast v3.asciicast raw)
 
   defp select_protocol(protos) do
     # Choose common protos between the client and the server using client preferred order.
@@ -302,16 +300,21 @@ defmodule AsciinemaWeb.StreamProducerSocket do
     List.first(common)
   end
 
-  defp detect_protocol({:binary, "ALiS" <> _}), do: "v0.alis"
-  defp detect_protocol({:binary, _}), do: "raw"
-  defp detect_protocol({:text, _}), do: "v2.asciicast"
+  def detect_protocol({:binary, "ALiS" <> _}), do: "v0.alis"
+  def detect_protocol({:binary, _}), do: "raw"
+
+  def detect_protocol({:text, header}) do
+    case Jason.decode(header) do
+      {:ok, %{"version" => 2}} -> "v2.asciicast"
+      {:ok, %{"version" => 3}} -> "v3.asciicast"
+      _otherwise -> "raw"
+    end
+  end
 
   defp save_protocol(stream_id, protocol) do
-    Task.Supervisor.start_child(Asciinema.TaskSupervisor, fn ->
-      stream_id
-      |> Streaming.get_stream()
-      |> Streaming.update_stream(protocol: protocol)
-    end)
+    stream_id
+    |> Streaming.get_stream()
+    |> Streaming.update_stream(protocol: protocol)
   end
 
   defp config(key, default) do