prop/test/microwaveprop/workers/backfill_enqueue_worker_test.exs

defmodule Microwaveprop.Workers.BackfillEnqueueWorkerTest do
  use Microwaveprop.DataCase, async: true

  alias Microwaveprop.Radio.Contact
  alias Microwaveprop.Terrain.ElevationClient
  alias Microwaveprop.Weather.HrrrClient
  alias Microwaveprop.Weather.IemClient
  alias Microwaveprop.Weather.NarrClient
  alias Microwaveprop.Workers.BackfillEnqueueWorker

  defp create_contact(attrs \\ %{}) do
    default = %{
      station1: "W5XD",
      station2: "K5TR",
      qso_timestamp: ~U[2026-03-28 18:00:00Z],
      mode: "CW",
      band: Decimal.new("1296"),
      grid1: "EM12",
      grid2: "EM00",
      pos1: %{"lat" => 32.9, "lon" => -97.0},
      pos2: %{"lat" => 30.3, "lon" => -97.7},
      distance_km: Decimal.new("295")
    }

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

    contact
  end

  setup do
    Req.Test.stub(IemClient, fn conn ->
      case conn.request_path do
        "/cgi-bin/request/asos.py" ->
          Req.Test.text(conn, "#DEBUG,\nstation,valid,tmpf\n")

        "/iemre/" <> _ ->
          Req.Test.json(conn, %{"data" => [%{"utc_hour" => 0, "p01m_mm" => 0.0}]})

        _ ->
          Req.Test.json(conn, %{"profiles" => []})
      end
    end)

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

    Req.Test.stub(ElevationClient, fn conn ->
      params = Plug.Conn.fetch_query_params(conn).query_params
      lat_count = params["latitude"] |> String.split(",") |> length()
      Req.Test.json(conn, %{"elevation" => List.duplicate(200.0, lat_count)})
    end)

    # NARR fetches land in era5_profiles; stub returns 500 so inline jobs fail
    # gracefully instead of hitting NCEI.
    Req.Test.stub(NarrClient, fn conn ->
      Plug.Conn.send_resp(conn, 500, "stub")
    end)

    :ok
  end

  # NARR fetches (the :era5 type) run inline via testing: :inline — the
  # NarrClient Req.Test stub above keeps them from hitting NCEI. Most tests
  # below target the other pipelines and use this narrowed type list to keep
  # the test surface tight.
  @non_era5_types ["hrrr", "weather", "terrain", "iemre"]

  describe "perform/1" do
    test "enqueues enrichment jobs for pending contacts" do
      create_contact()

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => @non_era5_types}
               })
    end

    test "does not enqueue for already-complete contacts" do
      contact = create_contact()

      contact
      |> Ecto.Changeset.change(%{
        hrrr_status: :complete,
        weather_status: :complete,
        terrain_status: :complete,
        iemre_status: :complete
      })
      |> Repo.update!()

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => @non_era5_types}
               })
    end

    test "respects limit parameter" do
      for i <- 1..5 do
        ts = DateTime.add(~U[2026-01-01 00:00:00Z], i * 3600, :second)
        create_contact(%{qso_timestamp: ts})
      end

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 2, "types" => @non_era5_types}
               })
    end

    test "enqueues all pending contacts when limit is omitted" do
      Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "backfill:enqueue_complete")

      for i <- 1..7 do
        ts = DateTime.add(~U[2026-01-01 00:00:00Z], i * 3600, :second)
        create_contact(%{qso_timestamp: ts})
      end

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"types" => @non_era5_types}
               })

      assert_receive {:enqueue_complete, 7}
    end

    test "broadcasts enqueue_complete with count" do
      Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "backfill:enqueue_complete")

      create_contact()

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => @non_era5_types}
               })

      assert_receive {:enqueue_complete, 1}
    end

    # :unavailable is a terminal state in the sense that workers don't
    # retry it, but the stale-queued reconciler flips :queued → :unavailable
    # after 3 days when no data has landed — which means a contact can end
    # up "Partial" forever even if the upstream archive later starts
    # serving data (new ASOS station in range, re-enabled IEM endpoint).
    # The backfill must re-probe :unavailable contacts so they get another
    # shot at enrichment; otherwise the contact list shows "Partial" for
    # rows that haven't been touched in months.
    test "re-selects :unavailable contacts older than the cooldown for another shot at enrichment" do
      import Ecto.Query

      Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "backfill:enqueue_complete")

      contact = create_contact()

      contact
      |> Ecto.Changeset.change(%{
        hrrr_status: :complete,
        weather_status: :unavailable,
        terrain_status: :complete,
        iemre_status: :complete
      })
      |> Repo.update!()

      # Force updated_at past the 24 h cooldown so the type_filter's
      # :unavailable branch picks this contact up.
      old = DateTime.utc_now() |> DateTime.add(-2 * 86_400, :second) |> DateTime.truncate(:second)
      Repo.update_all(from(c in Contact, where: c.id == ^contact.id), set: [updated_at: old])

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => @non_era5_types}
               })

      assert_receive {:enqueue_complete, 1}
    end

    test "leaves :unavailable contacts inside the cooldown window alone" do
      Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "backfill:enqueue_complete")

      contact = create_contact()

      contact
      |> Ecto.Changeset.change(%{
        hrrr_status: :complete,
        weather_status: :unavailable,
        terrain_status: :complete,
        iemre_status: :complete
      })
      |> Repo.update!()

      # updated_at is fresh (right now) — inside the 24 h cooldown.
      # Contact must not be re-probed yet.
      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => @non_era5_types}
               })

      assert_receive {:enqueue_complete, 0}
    end

    # :narr is a virtual type (no narr_status column) — it targets contacts
    # whose hrrr_status is :unavailable. BackfillEnqueueWorker must not try
    # to reconcile a non-existent narr_status column or use it for priority
    # ordering.
    test "handles virtual :narr type without referencing non-existent narr_status column" do
      # Pre-2014 contact (HRRR unavailable) — NARR candidate.
      contact = create_contact(%{qso_timestamp: ~U[2010-01-01 00:00:00Z]})

      contact
      |> Ecto.Changeset.change(%{hrrr_status: :unavailable})
      |> Repo.update!()

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => ["narr"]}
               })
    end

    # MechanismClassifyWorker sets both propagation_mechanism and
    # mechanism_status = :complete atomically. A prior bug in the
    # re-enqueue path clobbered mechanism_status back to :queued while
    # leaving propagation_mechanism populated. The reconciler flips
    # those rows back to :complete so the loop stops.
    test "reconciles mechanism_queued → complete when propagation_mechanism is already set" do
      contact = create_contact()

      contact
      |> Ecto.Changeset.change(%{
        mechanism_status: :queued,
        propagation_mechanism: "troposcatter",
        hrrr_status: :complete,
        weather_status: :complete,
        terrain_status: :complete,
        iemre_status: :complete,
        radar_status: :complete
      })
      |> Repo.update!()

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => ["mechanism"]}
               })

      assert %Contact{mechanism_status: :complete} = Repo.get!(Contact, contact.id)
    end
  end

  describe "reconcile stuck queued → unavailable" do
    # A contact stuck at :queued for > 3 days is one the cron has already
    # tried ~144 times without the underlying worker landing any data. At that
    # point the data simply doesn't exist in the upstream archive (dead ASOS
    # station, pre-2014 HRRR, out-of-grid IEMRE), so we stop retrying and flip
    # it to :unavailable.
    @stale_cutoff 3 * 86_400

    defp stale_ts do
      DateTime.utc_now()
      |> DateTime.add(-(@stale_cutoff + 3600), :second)
      |> DateTime.truncate(:second)
    end

    defp fresh_ts do
      DateTime.utc_now()
      |> DateTime.add(-3600, :second)
      |> DateTime.truncate(:second)
    end

    # Set the status field AND updated_at directly via update_all so Ecto
    # timestamps can't overwrite what we put there. Also short-circuits the
    # changeset, which ignores the *_status fields entirely.
    defp force_status(contact, field, status, ts) do
      import Ecto.Query

      Repo.update_all(from(c in Contact, where: c.id == ^contact.id), set: [{field, status}, {:updated_at, ts}])
    end

    # Complete the OTHER three statuses so the contact no longer matches the
    # enqueue filter on them — only the one under test remains :queued. That
    # keeps the main enqueue loop from running `mark_status!` on the contact
    # during the same perform/1 call, which would bump updated_at back to now.
    defp only_this_stuck(contact, field) do
      import Ecto.Query

      other_fields = [:weather_status, :hrrr_status, :iemre_status, :terrain_status] -- [field]
      sets = Enum.map(other_fields, &{&1, :complete})

      Repo.update_all(from(c in Contact, where: c.id == ^contact.id), set: sets)
    end

    test "flips weather_status :queued → :unavailable when updated_at is older than the cutoff" do
      contact = create_contact()
      only_this_stuck(contact, :weather_status)
      force_status(contact, :weather_status, :queued, stale_ts())

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => @non_era5_types}
               })

      assert %Contact{weather_status: :unavailable} = Repo.get!(Contact, contact.id)
    end

    test "flips hrrr_status :queued → :unavailable when updated_at is older than the cutoff" do
      contact = create_contact()
      only_this_stuck(contact, :hrrr_status)
      force_status(contact, :hrrr_status, :queued, stale_ts())

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => @non_era5_types}
               })

      assert %Contact{hrrr_status: :unavailable} = Repo.get!(Contact, contact.id)
    end

    test "flips iemre_status :queued → :unavailable when updated_at is older than the cutoff" do
      contact = create_contact()
      only_this_stuck(contact, :iemre_status)
      force_status(contact, :iemre_status, :queued, stale_ts())

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => @non_era5_types}
               })

      assert %Contact{iemre_status: :unavailable} = Repo.get!(Contact, contact.id)
    end

    test "does NOT flip fresh :queued contacts to :unavailable" do
      contact = create_contact()
      only_this_stuck(contact, :weather_status)
      force_status(contact, :weather_status, :queued, fresh_ts())

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => @non_era5_types}
               })

      # Whatever the main enqueue loop does next (mark :queued, :complete,
      # etc), reconcile_stale_queued_to_unavailable must NOT have fired: a
      # contact fresher than the cutoff still has a chance at real data.
      reloaded = Repo.get!(Contact, contact.id)
      refute reloaded.weather_status == :unavailable
    end

    test "only reconciles the types passed in the args — doesn't touch unrelated statuses" do
      # weather is stuck-queued-stale, but the cron is running with only hrrr
      # in its types — weather_status should be left alone.
      contact = create_contact()
      only_this_stuck(contact, :weather_status)
      force_status(contact, :weather_status, :queued, stale_ts())

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => ["hrrr"]}
               })

      assert %Contact{weather_status: :queued} = Repo.get!(Contact, contact.id)
    end
  end

  describe "reconcile hrrr_status against failed hrrr_fetch_tasks" do
    # When the Rust hrrr-point-worker marks a task as `failed` (upstream
    # archive hole, wgrib2 decode error, etc), the matching contacts
    # stay at :queued until the 3-day stale reconcile flips them. For
    # unambiguous upstream data gaps we can do better: if a task is in
    # :failed state, every contact whose rounded HRRR hour matches that
    # task's valid_time has no path to data — flip it to :unavailable
    # immediately so users see an accurate "Partial" marker instead of
    # waiting 3 days.
    import Ecto.Query

    defp insert_failed_task(valid_time) do
      now = DateTime.truncate(DateTime.utc_now(), :microsecond)

      Repo.insert_all(
        "hrrr_fetch_tasks",
        [
          %{
            id: Ecto.UUID.bingenerate(),
            valid_time: DateTime.to_naive(valid_time),
            points: [%{"lat" => 32.9, "lon" => -97.0}],
            status: "failed",
            attempt: 3,
            error: "fetch: idx HTTP 404",
            inserted_at: now,
            updated_at: now
          }
        ]
      )

      :ok
    end

    test "flips hrrr_status :queued → :unavailable for contacts whose rounded hour matches a failed task" do
      # QSO at 17:02 — nearest_hrrr_hour rounds to 17:00 (minute < 30).
      contact = create_contact(%{qso_timestamp: ~U[2018-08-05 17:02:00Z]})
      only_this_stuck(contact, :hrrr_status)
      force_status(contact, :hrrr_status, :queued, fresh_ts())

      insert_failed_task(~U[2018-08-05 17:00:00Z])

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => ["hrrr"]}
               })

      assert %Contact{hrrr_status: :unavailable} = Repo.get!(Contact, contact.id)
    end

    test "rounds UP when qso minute >= 30 before matching failed tasks" do
      # 17:45 rounds to 18:00 (minute >= 30), NOT 17:00.
      contact = create_contact(%{qso_timestamp: ~U[2018-08-05 17:45:00Z]})
      only_this_stuck(contact, :hrrr_status)
      force_status(contact, :hrrr_status, :queued, fresh_ts())

      insert_failed_task(~U[2018-08-05 18:00:00Z])

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => ["hrrr"]}
               })

      assert %Contact{hrrr_status: :unavailable} = Repo.get!(Contact, contact.id)
    end

    test "leaves contacts alone when the matching task is still :queued/:done" do
      contact = create_contact(%{qso_timestamp: ~U[2018-08-05 17:02:00Z]})
      only_this_stuck(contact, :hrrr_status)
      force_status(contact, :hrrr_status, :queued, fresh_ts())

      # No failed task for this hour.
      now = DateTime.truncate(DateTime.utc_now(), :microsecond)

      Repo.insert_all("hrrr_fetch_tasks", [
        %{
          id: Ecto.UUID.bingenerate(),
          valid_time: ~N[2018-08-05 17:00:00],
          points: [%{"lat" => 32.9, "lon" => -97.0}],
          status: "done",
          attempt: 1,
          inserted_at: now,
          updated_at: now
        }
      ])

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => ["hrrr"]}
               })

      refute Repo.get!(Contact, contact.id).hrrr_status == :unavailable
    end

    test "skips the reconcile when :hrrr is not in the types arg" do
      contact = create_contact(%{qso_timestamp: ~U[2018-08-05 17:02:00Z]})
      only_this_stuck(contact, :hrrr_status)
      force_status(contact, :hrrr_status, :queued, fresh_ts())

      insert_failed_task(~U[2018-08-05 17:00:00Z])

      assert :ok =
               BackfillEnqueueWorker.perform(%Oban.Job{
                 args: %{"limit" => 10, "types" => ["weather"]}
               })

      # hrrr_status untouched because the reconcile only runs when :hrrr
      # is in the active type list.
      assert Repo.get!(Contact, contact.id).hrrr_status == :queued
    end
  end
end