prop/test/microwaveprop/workers/radar_frame_worker_test.exs

defmodule Microwaveprop.Workers.RadarFrameWorkerTest do
  use Microwaveprop.DataCase, async: false
  use Oban.Testing, repo: Microwaveprop.Repo

  alias Microwaveprop.Radio.Contact
  alias Microwaveprop.Radio.ContactCommonVolumeRadar
  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.NexradClient
  alias Microwaveprop.Workers.RadarFrameWorker

  setup do
    Req.Test.set_req_test_from_context(%{async: false})
    :ok
  end

  defp insert_contact(attrs) do
    default = %{
      station1: "W5XD",
      station2: "K5TR",
      qso_timestamp: ~U[2024-09-15 18:30:00Z],
      mode: "CW",
      band: Decimal.new("10000"),
      grid1: "EM12",
      grid2: "EM00",
      pos1: %{"lat" => 32.9, "lon" => -97.0},
      pos2: %{"lat" => 33.1, "lon" => -96.8},
      distance_km: Decimal.new("30")
    }

    {:ok, contact} =
      %Contact{}
      |> Contact.changeset(Map.merge(default, attrs))
      |> Repo.insert()

    contact
  end

  describe "process_frame/4" do
    test "writes a ContactCommonVolumeRadar row and marks :complete for every contact in the batch" do
      c1 = insert_contact(%{station1: "A1"})
      c2 = insert_contact(%{station1: "A2"})

      # In-memory pixel buffer: 10x10 empty frame. The aggregator will
      # observe zero echoes but still emit a stats row per contact.
      pixels = :binary.copy(<<0>>, 100)
      width = 10
      frame_ts = ~U[2024-09-15 18:30:00Z]

      assert :ok =
               RadarFrameWorker.process_frame(
                 [Repo.reload!(c1), Repo.reload!(c2)],
                 frame_ts,
                 pixels,
                 width
               )

      assert Repo.get_by(ContactCommonVolumeRadar, contact_id: c1.id)
      assert Repo.get_by(ContactCommonVolumeRadar, contact_id: c2.id)

      assert %Contact{radar_status: :complete} = Repo.get!(Contact, c1.id)
      assert %Contact{radar_status: :complete} = Repo.get!(Contact, c2.id)
    end

    test "re-running process_frame/4 upserts in place (one row per contact, observed_at updated)" do
      # Pins the on_conflict: {:replace, [...]} clause on contact_id. If a
      # column gets added to ContactCommonVolumeRadar and the explicit list
      # isn't updated, or if the list names a non-existent column, this
      # test fails loudly.
      contact = insert_contact(%{station1: "RR1"})

      pixels = :binary.copy(<<0>>, 100)
      width = 10

      :ok =
        RadarFrameWorker.process_frame(
          [Repo.reload!(contact)],
          ~U[2024-09-15 18:30:00Z],
          pixels,
          width
        )

      first = Repo.get_by!(ContactCommonVolumeRadar, contact_id: contact.id)

      :ok =
        RadarFrameWorker.process_frame(
          [Repo.reload!(contact)],
          ~U[2024-09-15 18:35:00Z],
          pixels,
          width
        )

      rows =
        Repo.all(from(r in ContactCommonVolumeRadar, where: r.contact_id == ^contact.id))

      assert length(rows) == 1
      [second] = rows
      assert second.id == first.id
      assert second.observed_at == ~U[2024-09-15 18:35:00Z]
    end
  end

  describe "perform/1" do
    test "marks every contact in the batch :unavailable when the frame 404s" do
      c1 = insert_contact(%{station1: "B1"})
      c2 = insert_contact(%{station1: "B2"})

      Req.Test.stub(NexradClient, fn conn ->
        Plug.Conn.send_resp(conn, 404, "not found")
      end)

      assert :ok =
               perform_job(RadarFrameWorker, %{
                 "frame_ts" => "2024-09-15T18:30:00Z",
                 "contact_ids" => [c1.id, c2.id]
               })

      refute Repo.get_by(ContactCommonVolumeRadar, contact_id: c1.id)
      refute Repo.get_by(ContactCommonVolumeRadar, contact_id: c2.id)
      assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c1.id)
      assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c2.id)
    end

    test "returns {:error, _} and leaves contacts :queued on a 5xx server error (so Oban retries)" do
      c1 = insert_contact(%{station1: "S1"})
      c2 = insert_contact(%{station1: "S2"})

      Req.Test.stub(NexradClient, fn conn ->
        Plug.Conn.send_resp(conn, 503, "service unavailable")
      end)

      assert {:error, _reason} =
               perform_job(RadarFrameWorker, %{
                 "frame_ts" => "2024-09-15T18:30:00Z",
                 "contact_ids" => [c1.id, c2.id]
               })

      # Neither contact should be pinned :unavailable — a retry might succeed.
      assert %Contact{radar_status: status1} = Repo.get!(Contact, c1.id)
      assert %Contact{radar_status: status2} = Repo.get!(Contact, c2.id)
      refute status1 == :unavailable
      refute status2 == :unavailable
    end

    test "returns {:error, _} on a transport error without pinning contacts :unavailable" do
      c1 = insert_contact(%{station1: "T1"})

      Req.Test.stub(NexradClient, fn conn ->
        Req.Test.transport_error(conn, :timeout)
      end)

      assert {:error, _reason} =
               perform_job(RadarFrameWorker, %{
                 "frame_ts" => "2024-09-15T18:30:00Z",
                 "contact_ids" => [c1.id]
               })

      assert %Contact{radar_status: status} = Repo.get!(Contact, c1.id)
      refute status == :unavailable
    end

    test "skips contacts already at a terminal status (idempotent replay)" do
      c_done = insert_contact(%{station1: "DONE"})
      {:ok, _} = c_done |> Ecto.Changeset.change(%{radar_status: :complete}) |> Repo.update()

      c_new = insert_contact(%{station1: "NEW"})

      Req.Test.stub(NexradClient, fn conn ->
        Plug.Conn.send_resp(conn, 404, "not found")
      end)

      assert :ok =
               perform_job(RadarFrameWorker, %{
                 "frame_ts" => "2024-09-15T18:30:00Z",
                 "contact_ids" => [c_done.id, c_new.id]
               })

      # Already-complete contact stays complete; new one is marked unavailable.
      assert %Contact{radar_status: :complete} = Repo.get!(Contact, c_done.id)
      assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c_new.id)
    end

    test "processes every eligible contact in the frame's 5-min bucket, not just contact_ids" do
      # Both contacts land in the 18:30 bucket (18:30 ≤ ts < 18:35).
      c_passed = insert_contact(%{station1: "PASSED", qso_timestamp: ~U[2024-09-15 18:30:00Z]})
      c_omitted = insert_contact(%{station1: "OMITTED", qso_timestamp: ~U[2024-09-15 18:34:00Z]})
      # Outside the 18:30 bucket — must not be touched.
      c_other_bucket = insert_contact(%{station1: "OTHER", qso_timestamp: ~U[2024-09-15 18:35:00Z]})

      Req.Test.stub(NexradClient, fn conn ->
        Plug.Conn.send_resp(conn, 404, "not found")
      end)

      # Only c_passed is explicitly listed, but c_omitted is in the same
      # bucket and still :pending — the worker must pick it up too.
      assert :ok =
               perform_job(RadarFrameWorker, %{
                 "frame_ts" => "2024-09-15T18:30:00Z",
                 "contact_ids" => [c_passed.id]
               })

      assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c_passed.id)
      assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c_omitted.id)
      # Outside-bucket contact stays :pending.
      assert %Contact{radar_status: :pending} = Repo.get!(Contact, c_other_bucket.id)
    end
  end

  describe "NexradClient error contract" do
    # Pins the exact "NEXRAD n0q HTTP <code>" format emitted by
    # NexradClient.fetch_decoded_frame/1 on a non-200 response. RadarFrameWorker
    # pattern-matches on this literal prefix in `permanent_error?/1` to
    # classify 4xx as permanent (stop retrying) and everything else as
    # transient (let Oban retry). A silent format change would quietly turn
    # permanent 404s into infinite retries — fail loudly here instead.
    test "fetch_decoded_frame/1 returns error starting with \"NEXRAD n0q HTTP \" on 4xx" do
      Req.Test.stub(NexradClient, fn conn ->
        Plug.Conn.send_resp(conn, 404, "not found")
      end)

      assert {:error, reason} = NexradClient.fetch_decoded_frame(~U[2024-09-15 18:30:00Z])
      assert byte_size(reason) > 0
      assert String.starts_with?(reason, "NEXRAD n0q HTTP ")
      assert reason == "NEXRAD n0q HTTP 404"
    end
  end

  describe "unique constraint" do
    test "rejects a second insert with the same frame_ts while one is pending" do
      frame_ts = "2024-09-15T18:30:00Z"

      {:ok, first} =
        Oban.insert(RadarFrameWorker.new(%{"frame_ts" => frame_ts, "contact_ids" => ["a"]}))

      # Second insert with a different contact_ids list but same frame_ts
      # must collapse into the existing job — otherwise backfill sweeps
      # duplicate 5 MB PNG fetches per frame.
      {:ok, second} =
        Oban.insert(RadarFrameWorker.new(%{"frame_ts" => frame_ts, "contact_ids" => ["b"]}))

      assert first.id == second.id
    end
  end
end