Graham McIntire
fe2024275b
Two fixes: 1) Form save / rover-site add/remove no longer runs the path-matrix reconcile inline. update_mission and the show LiveView now enqueue a single RoverMissionReconcileWorker job and return — for missions with many rover sites x bands the inline path was firing hundreds of Multi inserts + Oban.inserts on the request. 2) RoverPathProfileWorker.load_path/1 now strictly matches a 4-tuple (mission_id, rover_location_id, station_id, band_mhz) with band_mhz as an integer, plus updates the Oban unique key set to include band_mhz so multi-band missions enqueue per-band jobs (not just one). Legacy/malformed args are logged and dropped instead of crashing the queue with MultipleResultsError.
282 lines
8.6 KiB
Elixir
282 lines
8.6 KiB
Elixir
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 Ecto.Multi
|
||
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
|
||
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
|
||
def get_mission_with_paths(id) do
|
||
case get_mission(id) do
|
||
nil -> nil
|
||
mission -> Repo.preload(mission, paths: [:rover_location, :station])
|
||
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])
|
||
:ok = enqueue_paths_for(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 """
|
||
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)
|
||
|
||
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)
|
||
|
||
: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)
|
||
pair whose `Path` is missing or not `:complete`. Existing complete
|
||
rows are left alone — the worker also short-circuits on `:complete`,
|
||
so this is idempotent and safe to run on a cron.
|
||
|
||
Use case: a rolling deploy or a transient elevation-API failure left
|
||
some missions with `:pending` / `:failed` entries; the next backfill
|
||
tick recovers them without operator action.
|
||
"""
|
||
@spec backfill_paths() :: :ok
|
||
def backfill_paths do
|
||
Mission
|
||
|> Repo.all()
|
||
|> Repo.preload(stations: from(s in Station, order_by: s.position))
|
||
|> Enum.each(&enqueue_paths_for/1)
|
||
|
||
:ok
|
||
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]))
|
||
end
|
||
|
||
def candidate_rover_locations(%Mission{only_known_good: false}) do
|
||
Repo.all(from(l in Location, order_by: [desc: l.inserted_at]))
|
||
end
|
||
|
||
defp insert_pending_paths_and_jobs([], _mission), do: :ok
|
||
|
||
defp insert_pending_paths_and_jobs(tuples, mission) when is_list(tuples) do
|
||
{:ok, _} =
|
||
tuples
|
||
|> Enum.reduce(Multi.new(), fn {rover_id, station_id, band_mhz}, multi ->
|
||
attrs = %{
|
||
mission_id: mission.id,
|
||
rover_location_id: rover_id,
|
||
station_id: station_id,
|
||
band_mhz: band_mhz,
|
||
status: :pending
|
||
}
|
||
|
||
Multi.insert(
|
||
multi,
|
||
{:path, rover_id, station_id, band_mhz},
|
||
Path.changeset(%Path{}, attrs),
|
||
on_conflict: :nothing,
|
||
conflict_target: [:mission_id, :rover_location_id, :station_id, :band_mhz]
|
||
)
|
||
end)
|
||
|> Repo.transaction()
|
||
|
||
enqueue_jobs(tuples, mission)
|
||
:ok
|
||
end
|
||
|
||
defp enqueue_jobs(tuples, mission) do
|
||
for {rover_id, station_id, band_mhz} <- tuples do
|
||
%{
|
||
"mission_id" => mission.id,
|
||
"rover_location_id" => rover_id,
|
||
"station_id" => station_id,
|
||
"band_mhz" => band_mhz
|
||
}
|
||
|> RoverPathProfileWorker.new()
|
||
|> Oban.insert()
|
||
end
|
||
|
||
: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
|