Address follow-up Copilot review comments

Author: mjc

lib/reencodarr/media/video_queries.ex

@@ -8,6 +8,7 @@ defmodule Reencodarr.Media.VideoQueries do
 
   import Ecto.Query
   alias Reencodarr.{DbWriter, Media.Video, Media.Vmaf, Repo}
+  require Logger
 
   @doc """
   Gets videos ready for CRF search (state: analyzed).
@@ -103,34 +104,41 @@ defmodule Reencodarr.Media.VideoQueries do
   """
   @spec claim_videos_for_analysis(integer(), keyword()) :: [Video.t()]
   def claim_videos_for_analysis(limit, opts \\ []) do
-    DbWriter.transaction(fn ->
-      candidates =
-        from(v in Video,
-          where: v.state == :needs_analysis,
-          order_by: [desc: v.priority, desc: v.size, desc: v.inserted_at],
-          limit: ^limit,
-          select: v.id
-        )
-        |> Repo.all(opts)
-
-      case candidates do
-        [] ->
-          []
-
-        ids ->
-          {_count, claimed} =
-            from(v in Video,
-              where: v.id in ^ids and v.state == :needs_analysis,
-              select: v
-            )
-            |> Repo.update_all([set: [state: :analyzing, updated_at: DateTime.utc_now()]], opts)
-
-          claimed
-      end
-    end)
+    DbWriter.transaction(
+      fn ->
+        candidates =
+          from(v in Video,
+            where: v.state == :needs_analysis,
+            order_by: [desc: v.priority, desc: v.size, desc: v.inserted_at],
+            limit: ^limit,
+            select: v.id
+          )
+          |> Repo.all(opts)
+
+        case candidates do
+          [] ->
+            []
+
+          ids ->
+            {_count, claimed} =
+              from(v in Video,
+                where: v.id in ^ids and v.state == :needs_analysis,
+                select: v
+              )
+              |> Repo.update_all([set: [state: :analyzing, updated_at: DateTime.utc_now()]], opts)
+
+            claimed
+        end
+      end,
+      label: :video_queries_claim_videos_for_analysis
+    )
     |> case do
-      {:ok, claimed} -> claimed
-      {:error, _reason} -> []
+      {:ok, claimed} ->
+        claimed
+
+      {:error, reason} ->
+        Logger.warning("Failed to claim videos for analysis: #{inspect(reason)}")
+        []
     end
   end
 

lib/reencodarr/media/video_upsert.ex

@@ -21,7 +21,22 @@ defmodule Reencodarr.Media.VideoUpsert do
   """
   @spec upsert(attrs()) :: upsert_result()
   def upsert(attrs) do
-    DbWriter.run(fn -> do_upsert(attrs) end, label: "perform video upsert")
+    DbWriter.transaction(
+      fn ->
+        case do_upsert(attrs) do
+          {:ok, video} ->
+            video
+
+          {:error, reason} ->
+            Repo.rollback(reason)
+        end
+      end,
+      label: "perform video upsert"
+    )
+    |> case do
+      {:ok, video} -> {:ok, video}
+      {:error, reason} -> {:error, reason}
+    end
   end
 
   @doc """
@@ -30,12 +45,21 @@ defmodule Reencodarr.Media.VideoUpsert do
   """
   @spec batch_upsert([attrs()]) :: [upsert_result()]
   def batch_upsert(video_attrs_list) when is_list(video_attrs_list) do
+    batch_upsert(video_attrs_list, [])
+  end
+
+  @spec batch_upsert([attrs()], keyword()) :: [upsert_result()]
+  def batch_upsert(video_attrs_list, opts) when is_list(video_attrs_list) and is_list(opts) do
+    writer_opts =
+      [label: :video_batch_upsert, max_attempts: 1]
+      |> Keyword.merge(opts)
+
     DbWriter.run(
       fn ->
         Logger.debug("VideoUpsert batch processing #{length(video_attrs_list)} videos")
         Enum.map(video_attrs_list, &process_single_video_in_batch/1)
       end,
-      label: :video_batch_upsert
+      writer_opts
     )
   end
 

lib/reencodarr/sync.ex

@@ -6,7 +6,7 @@ defmodule Reencodarr.Sync do
   alias Reencodarr.Analyzer.Broadway, as: AnalyzerBroadway
   alias Reencodarr.Core.Parsers
   alias Reencodarr.Dashboard.Events
-  alias Reencodarr.{Media, Repo, Services}
+  alias Reencodarr.{DbWriter, Media, Repo, Services}
 
   alias Reencodarr.Media.{MediaInfoExtractor, VideoFileInfo, VideoUpsert}
   alias Reencodarr.Media.Video.MediaInfoConverter
@@ -335,8 +335,13 @@ defmodule Reencodarr.Sync do
     # Check if video exists and file size hasn't changed
     existing_video = Media.get_video_by_path(info.path)
 
-    # VideoUpsert will automatically set state to needs_analysis for zero bitrate
-    handle_video_upsert(existing_video, info)
+    DbWriter.transaction(
+      fn ->
+        # VideoUpsert will automatically set state to needs_analysis for zero bitrate
+        handle_video_upsert(existing_video, info)
+      end,
+      label: :sync_upsert_video_from_file
+    )
   end
 
   defp handle_video_upsert({:ok, video}, info) do
@@ -419,17 +424,22 @@ defmodule Reencodarr.Sync do
     # Convert directly to MediaInfo format
     mediainfo = MediaInfoConverter.from_service_file(file, service_type)
 
-    # Store in database
-    # VideoUpsert will automatically set state to needs_analysis for missing bitrate
-    VideoUpsert.upsert(%{
-      "path" => file["path"],
-      "size" => file["size"],
-      "service_id" => to_string(file["id"]),
-      "service_type" => to_string(service_type),
-      "mediainfo" => mediainfo,
-      "bitrate" => file["overallBitrate"] || 0,
-      "dateAdded" => file["dateAdded"]
-    })
+    DbWriter.transaction(
+      fn ->
+        # Store in database
+        # VideoUpsert will automatically set state to needs_analysis for missing bitrate
+        VideoUpsert.upsert(%{
+          "path" => file["path"],
+          "size" => file["size"],
+          "service_id" => to_string(file["id"]),
+          "service_type" => to_string(service_type),
+          "mediainfo" => mediainfo,
+          "bitrate" => file["overallBitrate"] || 0,
+          "dateAdded" => file["dateAdded"]
+        })
+      end,
+      label: :sync_upsert_video_from_service_file
+    )
   end
 
   defp build_video_file_info(file, _media_info, service_type) do

test/reencodarr/db_writer_test.exs

@@ -57,27 +57,36 @@ defmodule Reencodarr.DbWriterTest do
     previous_env = Application.get_env(:reencodarr, :env)
     Application.put_env(:reencodarr, :env, :dev)
     on_exit(fn -> Application.put_env(:reencodarr, :env, previous_env) end)
+    test_pid = self()
 
     task =
       Task.async(fn ->
         DbWriter.run(
           fn ->
-            Process.sleep(150)
-            :occupied
+            send(test_pid, :writer_entered)
+
+            receive do
+              :release_writer -> :occupied
+            end
           end,
           writer_timeout: 1_000
         )
       end)
 
-    Process.sleep(25)
+    assert_receive :writer_entered
+
+    available_task =
+      Task.async(fn ->
+        DbWriter.run(
+          fn -> :available end,
+          timeout: 1,
+          writer_timeout: 1_000
+        )
+      end)
 
-    assert :available =
-             DbWriter.run(
-               fn -> :available end,
-               timeout: 1,
-               writer_timeout: 1_000
-             )
+    send(Process.whereis(DbWriter), :release_writer)
 
+    assert Task.await(available_task) == :available
     assert Task.await(task) == :occupied
   end
 
@@ -96,7 +105,10 @@ defmodule Reencodarr.DbWriterTest do
                    label: :failing_job
                  )
 
-        Process.sleep(50)
+        assert :barrier =
+                 DbWriter.run(fn ->
+                   :barrier
+                 end)
       end)
 
     assert log =~ "DbWriter async task failed for :failing_job"