Ensure duplicate player states can never occur

lib/ambry/media.ex

@@ -354,33 +354,18 @@ defmodule Ambry.Media do
     {player_states_to_return, player_states != player_states_to_return}
   end
 
-  @doc """
-  Gets or creates a player state for the given user and media.
-  """
-  def get_or_create_player_state!(user_id, media_id) do
-    result =
-      PlayerState
-      |> where([ps], ps.user_id == ^user_id and ps.media_id == ^media_id)
-      |> preload(^@player_state_preload)
-      |> Repo.one()
-
-    case result do
-      nil ->
-        {:ok, player_state} = create_player_state(%{user_id: user_id, media_id: media_id})
-        Repo.preload(player_state, @player_state_preload)
-
-      %PlayerState{} = player_state ->
-        player_state
-    end
-  end
-
   @doc """
   Gets or creates a player state for the given user and media, and marks it as
   the user's loaded player state.
   """
   def load_player_state!(user, media_id) do
-    player_state = get_or_create_player_state!(user.id, media_id)
-    {:ok, _user} = Accounts.update_user_loaded_player_state(user, player_state.id)
+    {:ok, player_state} =
+      Repo.transact(fn ->
+        player_state = get_player_state!(user.id, media_id)
+        {:ok, _user} = Accounts.update_user_loaded_player_state(user, player_state.id)
+
+        {:ok, player_state}
+      end)
 
     player_state
   end
@@ -406,22 +391,18 @@ defmodule Ambry.Media do
   end
 
   @doc """
-  Creates a player_state.
-
-  ## Examples
-
-      iex> create_player_state(%{field: value})
-      {:ok, %PlayerState{}}
-
-      iex> create_player_state(%{field: bad_value})
-      {:error, %Ecto.Changeset{}}
+  Gets a player_state for the given user and media.
 
+  If the player_state does not exist, it will be created.
   """
-  def create_player_state(attrs) do
-    %PlayerState{}
-    |> PlayerState.changeset(attrs)
-    |> Repo.insert()
-    |> tap_ok(&PubSub.broadcast_create/1)
+  def get_player_state!(user_id, media_id) when is_integer(user_id) and is_integer(media_id) do
+    %PlayerState{user_id: user_id, media_id: media_id}
+    |> Repo.insert!(
+      on_conflict: {:replace, [:position, :playback_rate, :status]},
+      conflict_target: [:user_id, :media_id],
+      returning: true
+    )
+    |> Repo.preload(@player_state_preload)
   end
 
   @doc """
@@ -443,6 +424,21 @@ defmodule Ambry.Media do
     |> tap_ok(&PubSub.broadcast_update/1)
   end
 
+  @doc """
+  Updates a player state for a user and media.
+
+  If the player state does not exist, it will be created.
+  """
+  def update_player_state(user_id, media_id, position, playback_rate) do
+    %PlayerState{user_id: user_id, media_id: media_id}
+    |> PlayerState.changeset(%{position: position, playback_rate: playback_rate})
+    |> Repo.insert(
+      on_conflict: {:replace, [:position, :playback_rate, :status]},
+      conflict_target: [:user_id, :media_id],
+      returning: true
+    )
+  end
+
   @doc """
   Gets all bookmarks for a media for a user.
   """

lib/ambry/media/player_state.ex

@@ -44,29 +44,34 @@ defmodule Ambry.Media.PlayerState do
   end
 
   defp compute_and_put_status(changeset) do
-    case changeset do
-      %{data: %{media: %{duration: duration}}} ->
-        position = get_field(changeset, :position)
-
-        cond do
-          # 1 minute until it counts as started
-          Decimal.lt?(position, 60) ->
-            put_change(changeset, :status, :not_started)
-
-          # 2 minutes from end it counts as finished
-          duration |> Decimal.sub(position) |> Decimal.lt?(120) ->
-            put_change(changeset, :status, :finished)
-
-          # otherwise it's in progress
-          true ->
-            put_change(changeset, :status, :in_progress)
-        end
-
-      changeset ->
-        changeset
+    duration = get_duration!(changeset)
+    position = get_field(changeset, :position)
+
+    cond do
+      # 1 minute until it counts as started
+      Decimal.lt?(position, 60) ->
+        put_change(changeset, :status, :not_started)
+
+      # 2 minutes from end it counts as finished
+      duration |> Decimal.sub(position) |> Decimal.lt?(120) ->
+        put_change(changeset, :status, :finished)
+
+      # otherwise it's in progress
+      true ->
+        put_change(changeset, :status, :in_progress)
     end
   end
 
+  defp get_duration!(%Ecto.Changeset{data: %__MODULE__{media: %Media{duration: duration}}}) do
+    duration
+  end
+
+  defp get_duration!(changeset) do
+    media_id = get_field(changeset, :media_id)
+    %Media{duration: %Decimal{} = duration} = Ambry.Media.get_media!(media_id)
+    duration
+  end
+
   defimpl Ambry.PubSub.Publishable do
     def topics(player_state) do
       [

lib/ambry_schema/resolvers.ex

@@ -110,7 +110,8 @@ defmodule AmbrySchema.Resolvers do
 
   def load_player_state(%{media_id: media_id}, %{context: %{current_user: %User{} = user}}) do
     with {:ok, %{id: media_id, type: :media}} <- from_global_id(media_id, AmbrySchema) do
-      player_state = Ambry.Media.get_or_create_player_state!(user.id, media_id)
+      media_id = String.to_integer(media_id)
+      player_state = Ambry.Media.get_player_state!(user.id, media_id)
       {:ok, %{player_state: player_state}}
     end
   end
@@ -119,9 +120,10 @@ defmodule AmbrySchema.Resolvers do
         context: %{current_user: %User{} = user}
       }) do
     with {:ok, %{id: media_id, type: :media}} <- from_global_id(media_id, AmbrySchema),
-         player_state = Ambry.Media.get_or_create_player_state!(user.id, media_id),
-         attrs = Map.delete(args, :media_id),
-         {:ok, player_state} <- Ambry.Media.update_player_state(player_state, attrs) do
+         media_id = String.to_integer(media_id),
+         %{position: position, playback_rate: playback_rate} = args,
+         {:ok, player_state} <-
+           Ambry.Media.update_player_state(user.id, media_id, position, playback_rate) do
       {:ok, %{player_state: player_state}}
     end
   end

priv/repo/migrations/20250107194334_fix_duplicate_player_states.exs

@@ -0,0 +1,27 @@
+defmodule Ambry.Repo.Migrations.FixDuplicatePlayerStates do
+  use Ecto.Migration
+
+  def up do
+    # deletes older player_states with duplicate media_id
+    execute """
+    DELETE FROM
+      player_states ps
+    WHERE
+      EXISTS (
+        SELECT
+          *
+        FROM
+          player_states x
+        WHERE
+          x.media_id = ps.media_id
+          AND x.ctid > ps.ctid
+      );
+    """
+
+    create unique_index(:player_states, [:user_id, :media_id])
+  end
+
+  def down do
+    drop unique_index(:player_states, [:user_id, :media_id])
+  end
+end

test/ambry/media_test.exs

@@ -517,26 +517,6 @@ defmodule Ambry.MediaTest do
     end
   end
 
-  describe "get_or_create_player_state!/2" do
-    test "creates a new player state if one doesn't yet exist" do
-      %{id: user_id} = insert(:user)
-      %{id: media_id} = insert(:media)
-
-      player_state = Media.get_or_create_player_state!(user_id, media_id)
-
-      assert %{user_id: ^user_id, media_id: ^media_id} = player_state
-    end
-
-    test "returns an existing player state if one already exists" do
-      %{id: user_id} = insert(:user)
-      %{id: player_state_id, media: %{id: media_id}} = insert(:player_state, user_id: user_id)
-
-      player_state = Media.get_or_create_player_state!(user_id, media_id)
-
-      assert %{id: ^player_state_id} = player_state
-    end
-  end
-
   describe "load_player_state!/2" do
     test "creates a new player state and sets it as the user's loaded player state" do
       user = insert(:user)
@@ -572,24 +552,23 @@ defmodule Ambry.MediaTest do
     end
   end
 
-  describe "create_player_state/1" do
-    test "requires user_id and media_id to be set" do
-      {:error, changeset} = Media.create_player_state(%{})
+  describe "get_player_state!/2" do
+    test "creates a new player state if one doesn't yet exist" do
+      %{id: user_id} = insert(:user)
+      %{id: media_id} = insert(:media)
+
+      player_state = Media.get_player_state!(user_id, media_id)
 
-      assert %{
-               media_id: ["can't be blank"],
-               user_id: ["can't be blank"]
-             } = errors_on(changeset)
+      assert %{user_id: ^user_id, media_id: ^media_id} = player_state
     end
 
-    test "creates a player state when given valid attributes" do
+    test "returns an existing player state if one already exists" do
       %{id: user_id} = insert(:user)
-      %{id: media_id} = insert(:media)
+      %{id: player_state_id, media: %{id: media_id}} = insert(:player_state, user_id: user_id)
 
-      assert {:ok, player_state} =
-               Media.create_player_state(%{user_id: user_id, media_id: media_id})
+      player_state = Media.get_player_state!(user_id, media_id)
 
-      assert %{user_id: ^user_id, media_id: ^media_id} = player_state
+      assert %{id: ^player_state_id} = player_state
     end
   end
 
@@ -638,6 +617,60 @@ defmodule Ambry.MediaTest do
     end
   end
 
+  describe "update_player_state/4" do
+    test "sets position and playback rate if player state doesn't yet exist" do
+      %{id: user_id} = insert(:user)
+      %{id: media_id} = insert(:media)
+
+      position = Decimal.new(300)
+      playback_rate = Decimal.new("1.25")
+
+      {:ok, player_state} = Media.update_player_state(user_id, media_id, position, playback_rate)
+
+      assert %{position: ^position, playback_rate: ^playback_rate} = player_state
+    end
+
+    test "updates position and playback rate if player state already exists" do
+      %{id: user_id} = insert(:user)
+      %{id: player_state_id} = player_state = insert(:player_state, user_id: user_id)
+
+      new_position = Decimal.new(300)
+      new_playback_rate = Decimal.new("1.25")
+
+      {:ok, updated_player_state} =
+        Media.update_player_state(user_id, player_state.media_id, new_position, new_playback_rate)
+
+      assert %{id: ^player_state_id, position: ^new_position, playback_rate: ^new_playback_rate} =
+               updated_player_state
+    end
+
+    test "sets status to `:not_started`" do
+      %{id: user_id} = insert(:user)
+      %{id: media_id} = insert(:media)
+
+      assert {:ok, %{status: :not_started}} =
+               Media.update_player_state(user_id, media_id, Decimal.new(0), Decimal.new(1))
+    end
+
+    test "sets status to `:in_progress`" do
+      %{id: user_id} = insert(:user)
+      %{id: media_id} = insert(:media)
+
+      assert {:ok, %{status: :in_progress}} =
+               Media.update_player_state(user_id, media_id, Decimal.new(60), Decimal.new(1))
+    end
+
+    test "sets status to `:finished`" do
+      %{id: user_id} = insert(:user)
+      media = %{id: media_id} = insert(:media)
+
+      position = Decimal.sub(media.duration, Decimal.new(119))
+
+      assert {:ok, %{status: :finished}} =
+               Media.update_player_state(user_id, media_id, position, Decimal.new(1))
+    end
+  end
+
   describe "list_bookmarks/2" do
     test "returns all bookmarks for the given user and media" do
       user = insert(:user)