Stream contact detail sections in as each enrichment loads
The contact detail page used to gather every enrichment source in one parallel_hydrate/2 bundle and await all five tasks before updating any assigns, so users watched the whole right-hand column flash in at the speed of the slowest query (usually hrrr_profiles_for_path against a 42M-row table). Each source now runs in its own start_async task: :weather, :solar, :hrrr_path, :terrain, :iemre. Matching handle_async/3 clauses assign the slot and call refresh_computed/1, which rebuilds elevation_profile + propagation_analysis + data_sources from whatever's currently in assigns. Sections appear independently as their data arrives, and the existing PubSub handlers (:hrrr_ready, :weather_ready, :terrain_ready, :solar_ready) still pick up later backfill completions the same way they did before.
This commit is contained in:
parent
480b6decdf
commit
7ed429364a
1 changed files with 152 additions and 98 deletions
|
|
@ -30,48 +30,61 @@ defmodule MicrowavepropWeb.ContactLive.Show do
|
|||
contact = id |> Radio.get_contact!() |> Radio.ensure_positions!()
|
||||
can_enqueue = internal_network?(session)
|
||||
|
||||
if connected?(socket) do
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:#{contact.id}")
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:hrrr")
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:weather")
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:solar")
|
||||
socket =
|
||||
assign(socket,
|
||||
page_title: "#{contact.station1} / #{contact.station2}",
|
||||
contact: contact,
|
||||
can_enqueue: can_enqueue,
|
||||
surface_observations: [],
|
||||
soundings: [],
|
||||
solar: nil,
|
||||
hrrr: nil,
|
||||
hrrr_path: [],
|
||||
iemre: nil,
|
||||
terrain: nil,
|
||||
elevation_profile: nil,
|
||||
propagation_analysis: nil,
|
||||
data_sources: nil,
|
||||
terrain_expanded: false,
|
||||
hrrr_profile_expanded: false,
|
||||
obs_sort_by: "station_name",
|
||||
obs_sort_order: "asc",
|
||||
sounding_sort_by: "station_name",
|
||||
sounding_sort_order: "asc",
|
||||
queue_counts: fetch_queue_counts(),
|
||||
expanded_soundings: MapSet.new(),
|
||||
editing: false,
|
||||
edit_form: nil,
|
||||
pending_edit: load_pending_edit(contact, socket),
|
||||
band_options: @band_options,
|
||||
mode_options: @mode_options
|
||||
)
|
||||
|
||||
# Defer the expensive enrichment loads (HRRR profiles, terrain profile,
|
||||
# IEMRE, elevation profile, propagation analysis — each touching large
|
||||
# tables or doing ITU-R math) until after the initial shell has been
|
||||
# sent. Users see the contact basics in ~20ms instead of waiting the
|
||||
# full ~500ms-2s mount.
|
||||
send(self(), {:hydrate, can_enqueue})
|
||||
end
|
||||
socket =
|
||||
if connected?(socket) do
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:#{contact.id}")
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:hrrr")
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:weather")
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:solar")
|
||||
|
||||
{:ok,
|
||||
assign(socket,
|
||||
page_title: "#{contact.station1} / #{contact.station2}",
|
||||
contact: contact,
|
||||
surface_observations: [],
|
||||
soundings: [],
|
||||
solar: nil,
|
||||
hrrr: nil,
|
||||
iemre: nil,
|
||||
terrain: nil,
|
||||
elevation_profile: nil,
|
||||
propagation_analysis: nil,
|
||||
data_sources: nil,
|
||||
hydrated: false,
|
||||
terrain_expanded: false,
|
||||
hrrr_profile_expanded: false,
|
||||
obs_sort_by: "station_name",
|
||||
obs_sort_order: "asc",
|
||||
sounding_sort_by: "station_name",
|
||||
sounding_sort_order: "asc",
|
||||
queue_counts: fetch_queue_counts(),
|
||||
expanded_soundings: MapSet.new(),
|
||||
editing: false,
|
||||
edit_form: nil,
|
||||
pending_edit: load_pending_edit(contact, socket),
|
||||
band_options: @band_options,
|
||||
mode_options: @mode_options
|
||||
)}
|
||||
kickoff_hydration(socket, contact)
|
||||
else
|
||||
socket
|
||||
end
|
||||
|
||||
{:ok, socket}
|
||||
end
|
||||
|
||||
# Each enrichment source loads in its own task so users see sections appear
|
||||
# independently instead of waiting for the slowest one. handle_async/3
|
||||
# clauses below update the matching slot and re-run the derived analysis.
|
||||
defp kickoff_hydration(socket, contact) do
|
||||
socket
|
||||
|> start_async(:weather, fn -> load_weather(contact) end)
|
||||
|> start_async(:solar, fn -> load_solar(contact) end)
|
||||
|> start_async(:hrrr_path, fn -> Weather.hrrr_profiles_for_path(contact) end)
|
||||
|> start_async(:terrain, fn -> Terrain.get_terrain_profile(contact.id) end)
|
||||
|> start_async(:iemre, fn -> load_iemre(contact) end)
|
||||
end
|
||||
|
||||
defp load_pending_edit(contact, socket) do
|
||||
|
|
@ -226,48 +239,76 @@ defmodule MicrowavepropWeb.ContactLive.Show do
|
|||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:hydrate, can_enqueue}, socket) do
|
||||
def handle_async(:weather, {:ok, weather}, socket) do
|
||||
contact = socket.assigns.contact
|
||||
|
||||
# Five independent DB loads run in parallel — total latency drops from
|
||||
# sum(each) to max(each). The heaviest is `hrrr_profiles_for_path`
|
||||
# against the 42M-row partitioned table.
|
||||
%{
|
||||
weather: weather,
|
||||
solar: solar,
|
||||
hrrr_path: hrrr_path,
|
||||
terrain: terrain,
|
||||
iemre: iemre
|
||||
} = parallel_hydrate(contact, can_enqueue)
|
||||
contact =
|
||||
if socket.assigns.can_enqueue, do: maybe_enqueue_weather(weather, contact), else: contact
|
||||
|
||||
hrrr = List.first(hrrr_path)
|
||||
{hrrr, contact} = if can_enqueue, do: maybe_enqueue_hrrr(hrrr, contact), else: {hrrr, contact}
|
||||
contact = if can_enqueue, do: maybe_enqueue_terrain(terrain, contact), else: contact
|
||||
contact = if can_enqueue, do: maybe_enqueue_weather(weather, contact), else: contact
|
||||
socket =
|
||||
socket
|
||||
|> assign(
|
||||
contact: contact,
|
||||
surface_observations: weather.surface_observations,
|
||||
soundings: weather.soundings
|
||||
)
|
||||
|> refresh_computed()
|
||||
|
||||
elevation_profile = compute_elevation_profile(contact, hrrr_path, weather.soundings)
|
||||
|
||||
propagation_analysis =
|
||||
build_propagation_analysis(contact, hrrr_path, terrain, elevation_profile, weather.soundings)
|
||||
|
||||
data_sources = build_data_sources(hrrr, hrrr_path, terrain, weather, elevation_profile)
|
||||
|
||||
{:noreply,
|
||||
assign(socket,
|
||||
contact: contact,
|
||||
surface_observations: weather.surface_observations,
|
||||
soundings: weather.soundings,
|
||||
solar: solar,
|
||||
hrrr: hrrr,
|
||||
iemre: iemre,
|
||||
terrain: terrain,
|
||||
elevation_profile: elevation_profile,
|
||||
propagation_analysis: propagation_analysis,
|
||||
data_sources: data_sources,
|
||||
hydrated: true
|
||||
)}
|
||||
{:noreply, socket}
|
||||
end
|
||||
|
||||
def handle_async(:solar, {:ok, solar}, socket) do
|
||||
solar =
|
||||
if socket.assigns.can_enqueue and is_nil(solar) do
|
||||
enqueue_missing_solar(socket.assigns.contact)
|
||||
nil
|
||||
else
|
||||
solar
|
||||
end
|
||||
|
||||
{:noreply, assign(socket, :solar, solar)}
|
||||
end
|
||||
|
||||
def handle_async(:hrrr_path, {:ok, hrrr_path}, socket) do
|
||||
contact = socket.assigns.contact
|
||||
hrrr = List.first(hrrr_path)
|
||||
|
||||
{hrrr, contact} =
|
||||
if socket.assigns.can_enqueue, do: maybe_enqueue_hrrr(hrrr, contact), else: {hrrr, contact}
|
||||
|
||||
socket =
|
||||
socket
|
||||
|> assign(contact: contact, hrrr: hrrr, hrrr_path: hrrr_path)
|
||||
|> refresh_computed()
|
||||
|
||||
{:noreply, socket}
|
||||
end
|
||||
|
||||
def handle_async(:terrain, {:ok, terrain}, socket) do
|
||||
contact = socket.assigns.contact
|
||||
|
||||
contact =
|
||||
if socket.assigns.can_enqueue, do: maybe_enqueue_terrain(terrain, contact), else: contact
|
||||
|
||||
socket =
|
||||
socket
|
||||
|> assign(contact: contact, terrain: terrain)
|
||||
|> refresh_computed()
|
||||
|
||||
{:noreply, socket}
|
||||
end
|
||||
|
||||
def handle_async(:iemre, {:ok, iemre}, socket) do
|
||||
{:noreply, assign(socket, :iemre, iemre)}
|
||||
end
|
||||
|
||||
def handle_async(slot, {:exit, reason}, socket)
|
||||
when slot in [:weather, :solar, :hrrr_path, :terrain, :iemre] do
|
||||
Logger.warning("contact hydrate task #{slot} exited: #{inspect(reason)}")
|
||||
{:noreply, socket}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:terrain_ready, _contact_id}, socket) do
|
||||
contact = %{socket.assigns.contact | terrain_status: :complete}
|
||||
terrain = Terrain.get_terrain_profile(contact.id)
|
||||
|
|
@ -401,19 +442,39 @@ defmodule MicrowavepropWeb.ContactLive.Show do
|
|||
defp sounding_sort_key("lifted_index"), do: &(&1.lifted_index || 0)
|
||||
defp sounding_sort_key(_), do: fn s -> s.station.name || s.station.station_code end
|
||||
|
||||
defp parallel_hydrate(contact, can_enqueue) do
|
||||
tasks = %{
|
||||
weather: Task.async(fn -> load_weather(contact) end),
|
||||
solar:
|
||||
Task.async(fn ->
|
||||
if can_enqueue, do: maybe_enqueue_solar(contact), else: load_solar(contact)
|
||||
end),
|
||||
hrrr_path: Task.async(fn -> Weather.hrrr_profiles_for_path(contact) end),
|
||||
terrain: Task.async(fn -> Terrain.get_terrain_profile(contact.id) end),
|
||||
iemre: Task.async(fn -> load_iemre(contact) end)
|
||||
}
|
||||
# Rebuild derived analysis from whatever's currently in assigns. Called from
|
||||
# each handle_async clause that updates a source slot, so sections that
|
||||
# depend on multiple sources (elevation_profile, propagation_analysis,
|
||||
# data_sources) refresh as each upstream source arrives.
|
||||
defp refresh_computed(socket) do
|
||||
%{
|
||||
contact: contact,
|
||||
hrrr: hrrr,
|
||||
hrrr_path: hrrr_path,
|
||||
terrain: terrain,
|
||||
surface_observations: surface_observations,
|
||||
soundings: soundings
|
||||
} = socket.assigns
|
||||
|
||||
Map.new(tasks, fn {key, task} -> {key, Task.await(task, 30_000)} end)
|
||||
elevation_profile = compute_elevation_profile(contact, hrrr_path, soundings)
|
||||
|
||||
propagation_analysis =
|
||||
build_propagation_analysis(contact, hrrr_path, terrain, elevation_profile, soundings)
|
||||
|
||||
data_sources =
|
||||
build_data_sources(
|
||||
hrrr,
|
||||
hrrr_path,
|
||||
terrain,
|
||||
%{surface_observations: surface_observations, soundings: soundings},
|
||||
elevation_profile
|
||||
)
|
||||
|
||||
assign(socket,
|
||||
elevation_profile: elevation_profile,
|
||||
propagation_analysis: propagation_analysis,
|
||||
data_sources: data_sources
|
||||
)
|
||||
end
|
||||
|
||||
defp load_iemre(contact) do
|
||||
|
|
@ -590,16 +651,9 @@ defmodule MicrowavepropWeb.ContactLive.Show do
|
|||
|> Weather.get_solar_index()
|
||||
end
|
||||
|
||||
defp maybe_enqueue_solar(contact) do
|
||||
solar = load_solar(contact)
|
||||
|
||||
if solar do
|
||||
solar
|
||||
else
|
||||
date = DateTime.to_date(contact.qso_timestamp)
|
||||
Oban.insert(SolarIndexWorker.new(%{"date" => Date.to_iso8601(date)}))
|
||||
nil
|
||||
end
|
||||
defp enqueue_missing_solar(contact) do
|
||||
date = DateTime.to_date(contact.qso_timestamp)
|
||||
Oban.insert(SolarIndexWorker.new(%{"date" => Date.to_iso8601(date)}))
|
||||
end
|
||||
|
||||
@impl true
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue