Graham McIntire
3060eddfb2
Two stacked wins for the form-submit hot path: 1) create_mission now hands path-matrix population to the same async reconcile worker update_mission uses, so the request returns right after the Mission row insert. For a mission with hundreds of (rover x station x band) tuples this used to block the LiveView for many seconds. 2) Inside reconcile/enqueue_paths_for, the per-pair Multi.insert loop is replaced with one Repo.insert_all for the Path rows and one Oban.insert_all for the worker jobs. N round-trips collapse to two regardless of pair count.
292 lines
9.3 KiB
Elixir
292 lines
9.3 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 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])
|
||
# 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 """
|
||
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
|
||
# 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
|