defmodule Microwaveprop.RoverPlanning do @moduledoc """ Context for the `/rover-planning` feature: planned rover missions with stationary endpoints and per-rover-location terrain path profiles. Mutations are scoped to a user (or admin); reads are global. After a successful `create_mission/2` or `replace_mission_paths/1`, the context enqueues one `RoverPathProfileWorker` job per (rover-location, station) pair so the show page can render results as they land. """ import Ecto.Query alias Microwaveprop.Accounts.User alias Microwaveprop.Repo alias Microwaveprop.Rover.Location alias Microwaveprop.RoverPlanning.Mission alias Microwaveprop.RoverPlanning.Path alias Microwaveprop.RoverPlanning.Station alias Microwaveprop.Workers.RoverMissionReconcileWorker alias Microwaveprop.Workers.RoverPathProfileWorker @spec list_missions() :: [Mission.t()] def list_missions do Mission |> order_by([m], desc: m.inserted_at) |> preload([:user, :stations]) |> Repo.all() end @spec get_mission(Ecto.UUID.t()) :: Mission.t() | nil | [%{atom() => any()}] def get_mission(id) when is_binary(id) do case Ecto.UUID.cast(id) do {:ok, uuid} -> Mission |> Repo.get(uuid) |> Repo.preload([:user, stations: from(s in Station, order_by: s.position)]) :error -> nil end end @spec get_mission_with_paths(Ecto.UUID.t()) :: Mission.t() | nil | [%{atom() => any()}] def get_mission_with_paths(id) do case Ecto.UUID.cast(id) do {:ok, uuid} -> Mission |> Repo.get(uuid) |> Repo.preload([:user, stations: from(s in Station, order_by: s.position), paths: [:rover_location, :station]]) :error -> nil end end @spec create_mission(User.t(), map()) :: {:ok, Mission.t()} | {:error, Ecto.Changeset.t()} def create_mission(%User{} = user, attrs) do changeset = Mission.changeset(%Mission{user_id: user.id}, attrs) case Repo.insert(changeset) do {:ok, mission} -> mission = Repo.preload(mission, [:user, :stations]) # Async: a brand-new mission can have hundreds of (rover x # station x band) tuples. Hand the matrix population off to # the reconcile worker so the form-submit request returns # immediately instead of blocking on N inserts + N Oban # enqueues per pair. The worker calls reconcile_mission_paths, # which is identical to enqueue_paths_for for an empty matrix. :ok = enqueue_reconcile(mission) {:ok, mission} {:error, _} = error -> error end end @spec update_mission(User.t(), Ecto.UUID.t(), map()) :: {:ok, Mission.t()} | {:error, :not_found | Ecto.Changeset.t()} def update_mission(%User{} = user, id, attrs) do case fetch_owned(user, id) do {:ok, mission} -> mission |> Repo.preload(stations: from(s in Station, order_by: s.position)) |> Mission.changeset(attrs) |> Repo.update() |> case do {:ok, updated} -> updated = Repo.preload(updated, [:user, :stations], force: true) :ok = enqueue_reconcile(updated) {:ok, updated} other -> other end {:error, :not_found} -> {:error, :not_found} end end @spec delete_mission(User.t(), Ecto.UUID.t()) :: {:ok, Mission.t()} | {:error, :not_found} def delete_mission(%User{} = user, id) do case fetch_owned(user, id) do {:ok, mission} -> Repo.delete(mission) {:error, :not_found} -> {:error, :not_found} end end @spec list_paths(Mission.t()) :: [Path.t()] def list_paths(%Mission{id: mission_id}) do Repo.all( from(p in Path, where: p.mission_id == ^mission_id, preload: [:rover_location, :station], order_by: [asc: p.station_id, asc: p.rover_location_id] ) ) end @doc "Fetches a single path by id or raises Ecto.NoResultsError." @spec get_path!(Ecto.UUID.t()) :: Path.t() def get_path!(id), do: Repo.get!(Path, id) @doc """ Reconciles the path matrix for a mission against the current set of (rover_location × station × band) tuples. Deletes paths for tuples that no longer apply, leaves :complete paths in place when they're still relevant, and enqueues new pending paths for any tuple missing from the table. Called after a mission edit, or after the user adds/removes a rover site. Aliased as `replace_mission_paths/1` for backwards compatibility with callers that expected the previous destructive behaviour; internally both go through the same reconcile path. """ @spec reconcile_mission_paths(Mission.t()) :: :ok def reconcile_mission_paths(%Mission{} = mission) do desired = MapSet.new(desired_tuples(mission)) actual_paths = Repo.all(from(p in Path, where: p.mission_id == ^mission.id)) actual = MapSet.new(actual_paths, &{&1.rover_location_id, &1.station_id, &1.band_mhz}) stale_ids = actual_paths |> Enum.reject(&MapSet.member?(desired, {&1.rover_location_id, &1.station_id, &1.band_mhz})) |> Enum.map(& &1.id) Repo.transaction(fn -> if stale_ids != [] do Repo.delete_all(from(p in Path, where: p.id in ^stale_ids)) end desired |> MapSet.difference(actual) |> Enum.to_list() |> insert_pending_paths_and_jobs(mission) end) :ok end @spec replace_mission_paths(Mission.t()) :: :ok defdelegate replace_mission_paths(mission), to: __MODULE__, as: :reconcile_mission_paths @doc """ Async entry point for path-matrix reconciliation. Enqueues a `RoverMissionReconcileWorker` job and returns immediately so callers (form save, rover-site add/remove) don't pay the per-tuple insert cost on the request path. Oban inline mode in tests still runs this synchronously, so test expectations don't have to change. """ @spec enqueue_reconcile(Mission.t()) :: :ok def enqueue_reconcile(%Mission{id: mission_id}) do _ = %{"mission_id" => mission_id} |> RoverMissionReconcileWorker.new() |> Oban.insert() :ok end @doc """ Walks every mission and re-enqueues any (rover_location × station × band) pair whose `Path` is missing, not `:complete`, OR whose cached `result` lacks the `"term"` blob (paths persisted before PathCompute extraction shipped — the row click on /rover-planning needs that blob to render `/path` from cache instead of recomputing). Idempotent: the path-profile worker short-circuits on `:complete`, and we only nudge :complete rows back to :pending when their term is missing. Safe to run on a cron. """ @spec backfill_paths() :: :ok def backfill_paths do reset_paths_missing_term() Mission |> Repo.all() |> Repo.preload(stations: from(s in Station, order_by: s.position)) |> Enum.each(&enqueue_paths_for/1) :ok end # Flip :complete paths whose result map has no "term" key back to # :pending so the next reconcile pass picks them up. JSONB # `key_exists` is `?` in pg, but Ecto's fragment uses `?` as a # placeholder — use the function form `jsonb_exists/2` instead. defp reset_paths_missing_term do Repo.update_all( from(p in Path, where: p.status == :complete, where: not fragment("jsonb_exists(?, 'term')", p.result) ), set: [status: :pending] ) end defp enqueue_paths_for(%Mission{} = mission) do mission |> desired_tuples() |> insert_pending_paths_and_jobs(mission) end # Cross product of (rover_location_id, station_id, band_mhz) for the # mission's current scope. The reconcile + initial-create paths both # go through this same source of truth. @spec desired_tuples(Mission.t()) :: [{Ecto.UUID.t(), Ecto.UUID.t(), integer()}] defp desired_tuples(%Mission{} = mission) do rovers = candidate_rover_locations(mission) stations = mission.stations || [] bands = Mission.bands(mission) for rover <- rovers, station <- stations, band <- bands do {rover.id, station.id, band} end end @doc """ Rover-locations the matrix actually scores against, given the mission's `only_known_good` flag. Public so the show LiveView can render the same list the worker enqueues against. """ @spec candidate_rover_locations(Mission.t()) :: [Location.t()] def candidate_rover_locations(%Mission{only_known_good: true}) do Repo.all(from(l in Location, where: l.status == :good, order_by: [desc: l.inserted_at], limit: 500)) end def candidate_rover_locations(%Mission{only_known_good: false}) do Repo.all(from(l in Location, order_by: [desc: l.inserted_at], limit: 500)) end defp insert_pending_paths_and_jobs([], _mission), do: :ok defp insert_pending_paths_and_jobs(tuples, mission) when is_list(tuples) do # Batch the path-row inserts and the Oban-job inserts into a # single SQL statement each. The previous Multi-of-N-inserts + # N x Oban.insert call took ~1.5s per pair × 900 pairs in the # worst-case mission — `insert_all` collapses both to one # roundtrip per kind regardless of pair count. now = DateTime.truncate(DateTime.utc_now(), :second) path_rows = Enum.map(tuples, fn {rover_id, station_id, band_mhz} -> %{ id: Ecto.UUID.generate(), mission_id: mission.id, rover_location_id: rover_id, station_id: station_id, band_mhz: band_mhz, status: :pending, inserted_at: now, updated_at: now } end) _ = Repo.insert_all(Path, path_rows, on_conflict: :nothing, conflict_target: [:mission_id, :rover_location_id, :station_id, :band_mhz] ) enqueue_jobs(tuples, mission) :ok end defp enqueue_jobs(tuples, mission) do changesets = Enum.map(tuples, fn {rover_id, station_id, band_mhz} -> RoverPathProfileWorker.new(%{ "mission_id" => mission.id, "rover_location_id" => rover_id, "station_id" => station_id, "band_mhz" => band_mhz }) end) _ = Oban.insert_all(changesets) :ok end defp fetch_owned(%User{is_admin: true}, id) do case get_mission(id) do nil -> {:error, :not_found} mission -> {:ok, mission} end end defp fetch_owned(%User{id: user_id}, id) do case get_mission(id) do %Mission{user_id: ^user_id} = mission -> {:ok, mission} _ -> {:error, :not_found} end end end