GITGUD

neiam neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

Filter arrivals to today's service, and read realtime from a table

af6ed7e · Gabriel Morell · 2026-09-01 20:10

17 files +685 -28
Message 
{commit_body(@commit)}

Files changed

modified apps/room_gtfs/lib/room_gtfs/application.ex
+4 −2
@@ -16,7 +16,9 @@ defmodule RoomGtfs.Application do
16 16 boot rather than lazily at the first decode.
17 17
18 18 It also owns `RoomGtfs.FeedCache`, so that several sources pointing at one
19 − realtime URL fetch it once between them. The per-source workers are not
19 + realtime URL fetch it once between them, and `RoomGtfs.RTIndex`, the table
20 + the realtime feeds are read out of -- owned here rather than by a worker so
21 + that a worker restarting does not take its source's data with it. The per-source workers are not
20 22 started here -- room_zeus does that, one per configured source.
21 23 """
22 24
@@ -29,7 +31,7 @@ defmodule RoomGtfs.Application do
29 31 # start callback, it is.
30 32 :ok = Protobuf.load_extensions()
31 33
32 − children = [RoomGtfs.FeedCache]
34 + children = [RoomGtfs.FeedCache, RoomGtfs.RTIndex]
33 35
34 36 Supervisor.start_link(children, strategy: :one_for_one, name: RoomGtfs.Supervisor)
35 37 end
modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+100 −21
@@ -55,11 +55,76 @@ defmodule RoomGtfs.Worker do
55 55 |> GenServer.cast(:update_realtime)
56 56 end
57 57
58 − def get_current_realtime(name, trips, stop) do
59 −# IO.inspect({name, trips, stop})
60 − "gtfs-rt#{name}"
61 − |> via_tuple
62 − |> GenServer.call({:query_realtime, trips, stop}, 30_000)
58 + @doc """
59 + The realtime trip updates for a stop, read rather than asked for.
60 +
61 + This used to be a `GenServer.call` into the source's RT worker -- the same
62 + process that fetches the feeds over HTTP -- so it queued behind those fetches
63 + and behind every other caller. A stop lookup could take tens of seconds, and
64 + the vision that asked for it gives up at thirty.
65 +
66 + The table is written on each poll and read here directly. The call remains
67 + for the window before a source has stored its first feed.
68 + """
69 + def get_current_realtime(name, trips, stop, config \\ nil) do
70 + # The caller usually holds the source already. Reading it again here would
71 + # put back one of the queries that taking the timezone as an argument
72 + # removed.
73 + config = config || source_config(name)
74 +
75 + indexed =
76 + case config do
77 + %{rt_trip_id_suffix: true} -> RoomGtfs.RTIndex.trip_updates_by_suffix(name, trips, stop)
78 + _otherwise -> RoomGtfs.RTIndex.trip_updates(name, trips, stop)
79 + end
80 +
81 + case indexed do
82 + {:ok, updates} ->
83 + updates
84 +
85 + :miss ->
86 + try do
87 + "gtfs-rt#{name}"
88 + |> via_tuple()
89 + |> GenServer.call({:query_realtime, trips, stop}, 5_000)
90 + |> Enum.map(&slim_entity/1)
91 + catch
92 + :exit, _ -> []
93 + end
94 + end
95 + end
96 +
97 + # The worker still replies with protobuf entities; the index speaks a smaller
98 + # shape, and callers should only have to know one of them.
99 + defp slim_entity(%{trip_update: tu}) do
100 + stu = tu.stop_time_update
101 +
102 + %{
103 + trip_id: tu.trip.trip_id,
104 + schedule_relationship: tu.trip.schedule_relationship,
105 + timestamp: tu.timestamp,
106 + stops: %{},
107 + stop:
108 + stu &&
109 + %{
110 + arrival: stu.arrival,
111 + departure: stu.departure,
112 + schedule_relationship: stu.schedule_relationship,
113 + stop_time_properties: stu.stop_time_properties,
114 + departure_occupancy_status: stu.departure_occupancy_status
115 + }
116 + }
117 + end
118 +
119 + # The suffix rule is a property of the source, and reading it here keeps the
120 + # lookup out of the worker entirely.
121 + defp source_config(name) do
122 + case Configuration.get_source!(:bare, name) do
123 + %{config: config} -> config
124 + _otherwise -> %{}
125 + end
126 + rescue
127 + _ -> %{}
63 128 end
64 129
65 130 def query_alerts(name, stop, route_ids) do
@@ -90,15 +155,27 @@ defmodule RoomGtfs.Worker do
90 155 end
91 156
92 157 def get_current_vehicle_positions(name) do
93 − "gtfs-rt#{name}"
94 − |> via_tuple
95 − |> GenServer.call(:query_vehicle_positions, 30_000)
158 + case RoomGtfs.RTIndex.vehicles(name) do
159 + {:ok, vehicles} ->
160 + vehicles
161 +
162 + :miss ->
163 + try do
164 + "gtfs-rt#{name}" |> via_tuple() |> GenServer.call(:query_vehicle_positions, 5_000)
165 + catch
166 + :exit, _ -> []
167 + end
168 + end
96 169 end
97 170
98 − def get_current_vehicle_positions(name, trips, timeout \\ 30_000) do
99 − "gtfs-rt#{name}"
100 − |> via_tuple
101 − |> GenServer.call({:query_vehicle_positions, trips}, timeout)
171 + def get_current_vehicle_positions(name, trips, timeout \\ 5_000) do
172 + case RoomGtfs.RTIndex.vehicles(name, trips) do
173 + {:ok, vehicles} ->
174 + vehicles
175 +
176 + :miss ->
177 + "gtfs-rt#{name}" |> via_tuple() |> GenServer.call({:query_vehicle_positions, trips}, timeout)
178 + end
102 179 end
103 180
104 181 @doc """
@@ -157,7 +234,7 @@ defmodule RoomGtfs.Worker do
157 234 _val ->
158 235 trips = res |> Enum.map(fn x -> x.trip_id end)
159 236
160 − case get_current_realtime(id, trips, query.stop) do
237 + case get_current_realtime(id, trips, query.stop, inst.config) do
161 238 [] ->
162 239 res
163 240
@@ -170,19 +247,19 @@ defmodule RoomGtfs.Worker do
170 247 res
171 248 |> Enum.map(fn x ->
172 249 case Enum.find(rtvals, fn v ->
173 − trip_id_match?(inst.config, x.trip_id, v.trip_update.trip.trip_id)
250 + trip_id_match?(inst.config, x.trip_id, v.trip_id)
174 251 end) do
175 252 nil ->
176 253 x
177 254
178 255 val ->
179 − stu = val.trip_update.stop_time_update
256 + stu = val.stop
180 257 time_event = stu && (stu.arrival || stu.departure)
181 258
182 259 x =
183 260 x
184 261 |> merge_occupancy(occupancy_for(inst.config, x.trip_id, stu, vehicles))
185 − |> merge_schedule_relationship(val.trip_update, stu)
262 + |> merge_schedule_relationship(val.schedule_relationship, stu)
186 263 |> merge_stop_properties(stu)
187 264
188 265 case time_event do
@@ -322,11 +399,9 @@ defmodule RoomGtfs.Worker do
322 399 #
323 400 # SCHEDULED is the proto default and means "as published", so it is left off
324 401 # rather than recorded -- only a departure from the schedule is news.
325 − def merge_schedule_relationship(arrival, trip_update, stu) do
326 − trip = trip_update && trip_update.trip
327 −
402 + def merge_schedule_relationship(arrival, trip_relationship, stu) do
328 403 arrival
329 − |> maybe_put(:trip_status, relationship(trip && trip.schedule_relationship))
404 + |> maybe_put(:trip_status, relationship(trip_relationship))
330 405 |> maybe_put(:stop_status, relationship(stu && stu.schedule_relationship))
331 406 end
332 407
@@ -1230,11 +1305,15 @@ defmodule RoomGtfs.Worker.RT do
1230 1305 # nothing they see.
1231 1306 defp store_feed(state, :sa, feed), do: Map.put(state, :rt_sa, feed)
1232 1307
1233 − defp store_feed(state, :tu, feed), do: Map.put(state, :rt_tu, feed)
1308 + defp store_feed(state, :tu, feed) do
1309 + RoomGtfs.RTIndex.put_trip_updates(state.id, feed.entity)
1310 + Map.put(state, :rt_tu, feed)
1311 + end
1234 1312
1235 1313 defp store_feed(state, :vp, feed) do
1236 1314 state = state |> Map.put(:rt_vp, feed) |> ensure_stops()
1237 1315 vehicles = vehicle_positions_from(feed, state.stops)
1316 + RoomGtfs.RTIndex.put_vehicles(state.id, vehicles)
1238 1317
1239 1318 # The source page watches its own channel; the query pages watch the shared
1240 1319 # one and sort out which vehicles are theirs.
added apps/room_gtfs/lib/room_gtfs/rt_index.ex
+184 −0
@@ -0,0 +1,184 @@
1 +defmodule RoomGtfs.RTIndex do
2 + @moduledoc """
3 + The realtime feeds, indexed by trip and readable without asking anyone.
4 +
5 + Arrivals used to be answered by two `GenServer.call`s into the source's RT
6 + worker -- one for trip updates, one for vehicle positions. That worker also
7 + fetches the feeds over HTTP, so the calls queued behind each other and behind
8 + the fetches, and a stop lookup could take tens of seconds. The vision that
9 + asked for it gives up after 30 and kills its own task, so a busy source meant
10 + a board that showed nothing at all.
11 +
12 + A realtime feed is immutable data replaced on a timer, which is the shape ETS
13 + is for. Readers take what they need straight out of the table; nothing
14 + serialises behind the worker's own work.
15 +
16 + Indexed by trip rather than stored whole, deliberately. A `:tu` feed is ~1600
17 + entities, and handing a reader all of them to find its sixteen would copy far
18 + more than the GenServer ever did -- the call at least filtered before
19 + replying. One row per trip, holding only the fields anything reads, means a
20 + lookup copies sixteen small maps.
21 +
22 + The table is public so each source's worker writes its own rows, and owned by
23 + this app's supervisor rather than by a worker, so a worker restarting does
24 + not take the data with it.
25 + """
26 +
27 + @table :gtfs_rt_index
28 +
29 + def table, do: @table
30 +
31 + @doc false
32 + def child_spec(_opts) do
33 + %{id: __MODULE__, start: {__MODULE__, :start_link, []}}
34 + end
35 +
36 + @doc false
37 + def start_link do
38 + Task.start_link(fn ->
39 + :ets.new(@table, [:named_table, :public, :set, read_concurrency: true])
40 + Process.hibernate(Process, :sleep, [:infinity])
41 + end)
42 + end
43 +
44 + @doc """
45 + Replace a source's trip updates with what just arrived.
46 +
47 + Written as one row per trip, carrying only what a stop lookup reads: the
48 + trip's own identity and relationship, and its stop times keyed by stop id.
49 + """
50 + def put_trip_updates(source_id, entities) do
51 + rows =
52 + entities
53 + |> Enum.filter(&(&1.trip_update && &1.trip_update.trip && &1.trip_update.trip.trip_id))
54 + |> Enum.map(fn %{trip_update: tu} ->
55 + {{source_id, :tu, tu.trip.trip_id},
56 + %{
57 + trip_id: tu.trip.trip_id,
58 + schedule_relationship: tu.trip.schedule_relationship,
59 + timestamp: tu.timestamp,
60 + stops: stops_by_id(tu.stop_time_update)
61 + }}
62 + end)
63 +
64 + replace(source_id, :tu, rows)
65 + end
66 +
67 + @doc """
68 + Replace a source's vehicle positions.
69 +
70 + Kept twice over: a row per trip, for "the vehicle on this trip", and one row
71 + holding the lot, for the map. The second is a large copy on read and is only
72 + taken by callers that genuinely want every vehicle.
73 + """
74 + def put_vehicles(source_id, vehicles) do
75 + rows =
76 + vehicles
77 + |> Enum.filter(& &1.trip_id)
78 + |> Enum.map(&{{source_id, :vp, &1.trip_id}, &1})
79 +
80 + replace(source_id, :vp, rows)
81 + :ets.insert(@table, {{source_id, :vp_all}, vehicles})
82 + :ok
83 + end
84 +
85 + @doc """
86 + The trip updates for a set of scheduled trip ids, at one stop.
87 +
88 + Answers `{:ok, list}` when the source has a feed indexed, and `:miss` when it
89 + has none -- so a caller can tell "nothing is running" from "ask the worker
90 + instead", which is what a source whose feed has never been stored needs.
91 + """
92 + def trip_updates(source_id, trip_ids, stop_id) do
93 + case indexed?(source_id, :tu) do
94 + false ->
95 + :miss
96 +
97 + true ->
98 + found =
99 + trip_ids
100 + |> Enum.flat_map(&:ets.lookup(@table, {source_id, :tu, &1}))
101 + |> Enum.map(fn {_k, tu} -> Map.put(tu, :stop, Map.get(tu.stops, stop_id)) end)
102 +
103 + {:ok, found}
104 + end
105 + end
106 +
107 + @doc """
108 + The trip updates for a set of scheduled trip ids where the feed names trips
109 + by a suffix of the scheduled id -- NYCT's shape, which no key lookup matches.
110 +
111 + A scan of one source's rows, which is still cheaper than a call into a busy
112 + worker, and only sources that declare the suffix take this path.
113 + """
114 + def trip_updates_by_suffix(source_id, trip_ids, stop_id) do
115 + case indexed?(source_id, :tu) do
116 + false ->
117 + :miss
118 +
119 + true ->
120 + found =
121 + @table
122 + |> :ets.match_object({{source_id, :tu, :_}, :_})
123 + |> Enum.filter(fn {{_, _, rt_id}, _} ->
124 + Enum.any?(trip_ids, &String.ends_with?(&1, rt_id))
125 + end)
126 + |> Enum.map(fn {_k, tu} -> Map.put(tu, :stop, Map.get(tu.stops, stop_id)) end)
127 +
128 + {:ok, found}
129 + end
130 + end
131 +
132 + @doc """
133 + The vehicles on a set of trips, or every vehicle the source is reporting.
134 + """
135 + def vehicles(source_id, trip_ids) do
136 + case indexed?(source_id, :vp) do
137 + false -> :miss
138 + true -> {:ok, Enum.flat_map(trip_ids, &(:ets.lookup(@table, {source_id, :vp, &1}) |> Enum.map(fn {_k, v} -> v end)))}
139 + end
140 + end
141 +
142 + def vehicles(source_id) do
143 + case :ets.lookup(@table, {source_id, :vp_all}) do
144 + [{_k, vehicles}] -> {:ok, vehicles}
145 + [] -> :miss
146 + end
147 + end
148 +
149 + @doc false
150 + # Whether this source has ever stored a feed of this kind. Distinct from
151 + # having no rows: a feed can legitimately arrive carrying nothing.
152 + def indexed?(source_id, kind) do
153 + :ets.member(@table, {source_id, {:present, kind}})
154 + end
155 +
156 + # The marker is a two-element key on purpose: the per-trip rows are scanned
157 + # with a three-element pattern for suffix matching, and a marker shaped like
158 + # one of them comes back from that scan as a trip whose id is an atom.
159 + defp replace(source_id, kind, rows) do
160 + :ets.match_delete(@table, {{source_id, kind, :_}, :_})
161 + :ets.insert(@table, rows)
162 + :ets.insert(@table, {{source_id, {:present, kind}}, true})
163 + :ok
164 + end
165 +
166 + # A trip calls at a stop once in the ordinary case; a loop route can call
167 + # twice, and the earlier call is the one an arrivals board wants.
168 + defp stops_by_id(stop_time_updates) do
169 + stop_time_updates
170 + |> List.wrap()
171 + |> Enum.reject(&(&1.stop_id in [nil, ""]))
172 + |> Enum.reverse()
173 + |> Map.new(fn stu ->
174 + {stu.stop_id,
175 + %{
176 + arrival: stu.arrival,
177 + departure: stu.departure,
178 + schedule_relationship: stu.schedule_relationship,
179 + stop_time_properties: stu.stop_time_properties,
180 + departure_occupancy_status: stu.departure_occupancy_status
181 + }}
182 + end)
183 + end
184 +end
modified apps/room_gtfs/test/occupancy_test.exs
+1 −5
@@ -109,11 +109,7 @@ defmodule RoomGtfs.OccupancyTest do
109 109
110 110 describe "schedule relationship" do
111 111 defp merged(trip_rel, stop_rel) do
112 − Worker.merge_schedule_relationship(
113 − %{},
114 − %{trip: %{schedule_relationship: trip_rel}},
115 − %{schedule_relationship: stop_rel}
116 − )
112 + Worker.merge_schedule_relationship(%{}, trip_rel, %{schedule_relationship: stop_rel})
117 113 end
118 114
119 115 test "as published is not news" do
added apps/room_gtfs/test/rt_index_test.exs
+153 −0
@@ -0,0 +1,153 @@
1 +defmodule RoomGtfs.RTIndexTest do
2 + @moduledoc """
3 + Reading the realtime feeds out of a table instead of asking a worker for them.
4 +
5 + The worker that held these also fetches the feeds over HTTP, so every stop
6 + lookup queued behind those fetches and behind every other caller. Indexed by
7 + trip, because handing a reader the whole feed to find its sixteen trips would
8 + copy more than the call it replaced.
9 + """
10 + use ExUnit.Case, async: false
11 +
12 + alias RoomGtfs.RTIndex
13 + alias TransitRealtime, as: T
14 +
15 + setup do
16 + if :ets.whereis(RTIndex.table()) == :undefined do
17 + {:ok, _} = RTIndex.start_link()
18 + Process.sleep(20)
19 + end
20 +
21 + id = System.unique_integer([:positive])
22 + on_exit(fn -> :ets.match_delete(RTIndex.table(), {{id, :_, :_}, :_}) end)
23 + %{id: id}
24 + end
25 +
26 + defp entity(trip_id, stop_ids, attrs \\ []) do
27 + %T.FeedEntity{
28 + id: trip_id,
29 + trip_update: %T.TripUpdate{
30 + trip: struct(%T.TripDescriptor{trip_id: trip_id}, attrs),
31 + stop_time_update:
32 + Enum.map(stop_ids, fn s ->
33 + %T.TripUpdate.StopTimeUpdate{
34 + stop_id: s,
35 + arrival: %T.TripUpdate.StopTimeEvent{time: 123, delay: 60}
36 + }
37 + end)
38 + }
39 + }
40 + end
41 +
42 + describe "trip updates" do
43 + test "a source with nothing stored is a miss, not an empty answer", %{id: id} do
44 + # The distinction matters: a caller can fall back to the worker rather
45 + # than reporting that no trains are running.
46 + assert RTIndex.trip_updates(id, ["t1"], "s1") == :miss
47 + end
48 +
49 + test "a stored feed with no entities is an answer", %{id: id} do
50 + RTIndex.put_trip_updates(id, [])
51 +
52 + assert {:ok, []} = RTIndex.trip_updates(id, ["t1"], "s1")
53 + end
54 +
55 + test "only the trips asked for come back", %{id: id} do
56 + RTIndex.put_trip_updates(id, [entity("t1", ["s1"]), entity("t2", ["s1"]), entity("t3", ["s1"])])
57 +
58 + assert {:ok, found} = RTIndex.trip_updates(id, ["t1", "t3"], "s1")
59 + assert Enum.map(found, & &1.trip_id) |> Enum.sort() == ["t1", "t3"]
60 + end
61 +
62 + test "the stop asked for is picked out of the trip's calls", %{id: id} do
63 + RTIndex.put_trip_updates(id, [entity("t1", ["s1", "s2", "s3"])])
64 +
65 + assert {:ok, [tu]} = RTIndex.trip_updates(id, ["t1"], "s2")
66 + assert tu.stop.arrival.time == 123
67 + end
68 +
69 + test "a trip that does not call at the stop still answers, with no stop", %{id: id} do
70 + RTIndex.put_trip_updates(id, [entity("t1", ["s1"])])
71 +
72 + assert {:ok, [tu]} = RTIndex.trip_updates(id, ["t1"], "elsewhere")
73 + assert tu.stop == nil
74 + end
75 +
76 + test "the trip's own relationship rides along", %{id: id} do
77 + RTIndex.put_trip_updates(id, [entity("t1", ["s1"], schedule_relationship: :CANCELED)])
78 +
79 + assert {:ok, [tu]} = RTIndex.trip_updates(id, ["t1"], "s1")
80 + assert tu.schedule_relationship == :CANCELED
81 + end
82 +
83 + test "a later feed replaces the earlier one rather than adding to it", %{id: id} do
84 + RTIndex.put_trip_updates(id, [entity("t1", ["s1"]), entity("gone", ["s1"])])
85 + RTIndex.put_trip_updates(id, [entity("t1", ["s1"])])
86 +
87 + assert {:ok, [_]} = RTIndex.trip_updates(id, ["t1", "gone"], "s1")
88 + end
89 +
90 + test "one source's feed is not another's", %{id: id} do
91 + other = System.unique_integer([:positive])
92 + RTIndex.put_trip_updates(id, [entity("t1", ["s1"])])
93 +
94 + assert RTIndex.trip_updates(other, ["t1"], "s1") == :miss
95 + end
96 +
97 + test "a stop update with no stop id is not indexed by one", %{id: id} do
98 + RTIndex.put_trip_updates(id, [entity("t1", [nil, ""])])
99 +
100 + assert {:ok, [tu]} = RTIndex.trip_updates(id, ["t1"], "s1")
101 + assert tu.stops == %{}
102 + end
103 + end
104 +
105 + describe "suffix matching" do
106 + test "a feed naming the tail of a scheduled trip id still matches", %{id: id} do
107 + # NYCT: realtime "098600_5..S03R", schedule "ASP26GEN-1038-Sunday-00_098600_5..S03R"
108 + RTIndex.put_trip_updates(id, [entity("098600_5..S03R", ["s1"])])
109 +
110 + assert {:ok, [tu]} =
111 + RTIndex.trip_updates_by_suffix(
112 + id,
113 + ["ASP26GEN-1038-Sunday-00_098600_5..S03R"],
114 + "s1"
115 + )
116 +
117 + assert tu.trip_id == "098600_5..S03R"
118 + end
119 +
120 + test "an unrelated trip does not match on a suffix", %{id: id} do
121 + RTIndex.put_trip_updates(id, [entity("something-else", ["s1"])])
122 +
123 + assert {:ok, []} = RTIndex.trip_updates_by_suffix(id, ["ASP26_098600"], "s1")
124 + end
125 + end
126 +
127 + describe "vehicles" do
128 + test "by trip, and all at once", %{id: id} do
129 + vehicles = [
130 + %{vehicle_id: "v1", trip_id: "t1", occupancy_status: :FULL},
131 + %{vehicle_id: "v2", trip_id: "t2", occupancy_status: :EMPTY}
132 + ]
133 +
134 + RTIndex.put_vehicles(id, vehicles)
135 +
136 + assert {:ok, [%{vehicle_id: "v1"}]} = RTIndex.vehicles(id, ["t1"])
137 + assert {:ok, both} = RTIndex.vehicles(id)
138 + assert length(both) == 2
139 + end
140 +
141 + test "a vehicle on no trip is still in the whole-feed answer", %{id: id} do
142 + RTIndex.put_vehicles(id, [%{vehicle_id: "v1", trip_id: nil}])
143 +
144 + assert {:ok, [_]} = RTIndex.vehicles(id)
145 + assert {:ok, []} = RTIndex.vehicles(id, ["t1"])
146 + end
147 +
148 + test "nothing stored is a miss", %{id: id} do
149 + assert RTIndex.vehicles(id) == :miss
150 + assert RTIndex.vehicles(id, ["t1"]) == :miss
151 + end
152 + end
153 +end
modified apps/room_sanctum/lib/room_sanctum/storage.ex
+70 −0
@@ -1318,6 +1318,75 @@ defmodule RoomSanctum.Storage do
1318 1318 val
1319 1319 end
1320 1320
1321 + @doc false
1322 + # Keep only the trips whose service actually runs on `date`, and be careful
1323 + # about what "actually" can be known.
1324 + #
1325 + # A stop's timetable holds every service pattern the feed publishes --
1326 + # weekday, Saturday, Sunday, and on MBTA a dozen dated variants of each -- so
1327 + # without this the board is mostly departures that will not happen, and the
1328 + # limit is spent on them. One stop there mixes twelve patterns.
1329 + #
1330 + # Two things stop this from being a filter that empties boards, and both are
1331 + # load-bearing rather than defensive dressing. Measured on the dev database,
1332 + # two of eight sources have no usable service data at all: one has 33,163
1333 + # trips and not a single calendar row, another has calendars that expired
1334 + # days ago.
1335 + #
1336 + # * A service nothing says anything about is kept. Neither file mentioning
1337 + # it means we do not know when it runs, and hiding it asserts something
1338 + # we have not been told.
1339 + #
1340 + # * A source with nothing running today is not filtered at all. That is a
1341 + # feed whose service data has gone stale, and a board of possibly-wrong
1342 + # times beats a blank one.
1343 + defp running_on(queryable, source_id, date) do
1344 + weekday = weekday_column(Date.day_of_week(date))
1345 +
1346 + scheduled =
1347 + "gtfs_calendars"
1348 + |> where([c], c.source_id == ^source_id and c.start_date <= ^date and c.end_date >= ^date)
1349 + |> where([c], field(c, ^weekday) == 1)
1350 +
1351 + exception = fn type ->
1352 + "gtfs_calendar_dates"
1353 + |> where([cd], cd.source_id == ^source_id and cd.date == ^date and cd.exception_type == ^type)
1354 + end
1355 +
1356 + from([st, trip: t] in queryable,
1357 + where:
1358 + (exists(from(c in scheduled, where: c.service_id == parent_as(:trip).service_id, select: 1)) or
1359 + exists(
1360 + from(cd in exception.(1), where: cd.service_id == parent_as(:trip).service_id, select: 1)
1361 + ) or
1362 + not exists(
1363 + from(c in "gtfs_calendars",
1364 + where: c.source_id == ^source_id and c.service_id == parent_as(:trip).service_id,
1365 + select: 1
1366 + )
1367 + ) and
1368 + not exists(
1369 + from(cd in "gtfs_calendar_dates",
1370 + where: cd.source_id == ^source_id and cd.service_id == parent_as(:trip).service_id,
1371 + select: 1
1372 + )
1373 + ) or
1374 + not exists(from(c in scheduled, select: 1)) and
1375 + not exists(from(cd in exception.(1), select: 1))) and
1376 + not exists(
1377 + from(cd in exception.(2), where: cd.service_id == parent_as(:trip).service_id, select: 1)
1378 + )
1379 + )
1380 + end
1381 +
1382 + defp weekday_column(1), do: :monday
1383 + defp weekday_column(2), do: :tuesday
1384 + defp weekday_column(3), do: :wednesday
1385 + defp weekday_column(4), do: :thursday
1386 + defp weekday_column(5), do: :friday
1387 + defp weekday_column(6), do: :saturday
1388 + defp weekday_column(7), do: :sunday
1389 +
1321 1390 defp conditional_where(queryable, source_id, stop_id, timestamp_time_hour, timestamp_time) do
1322 1391 # case timestamp_time < timestamp_time_hour do
1323 1392 # true ->
@@ -1430,6 +1499,7 @@ defmodule RoomSanctum.Storage do
1430 1499 as: :trip,
1431 1500 on: t.trip_id == st.trip_id and t.source_id == st.source_id
1432 1501 )
1502 + |> running_on(source_id, DateTime.to_date(timestamp))
1433 1503 |> join(:left, [st, trip: t], r in Route,
1434 1504 as: :route,
1435 1505 on: t.route_id == r.route_id and r.source_id == st.source_id
added apps/room_sanctum/test/room_sanctum/arrival_service_day_test.exs
+173 −0
@@ -0,0 +1,173 @@
1 +defmodule RoomSanctum.ArrivalServiceDayTest do
2 + @moduledoc """
3 + The arrival query keeps only trips whose service runs today -- and knows when
4 + it is not entitled to an opinion.
5 +
6 + A stop's timetable holds every service pattern the feed publishes. On the
7 + MBTA bus feed in the dev database one stop mixed twelve, and 80% of an hour's
8 + arrivals were departures that will not happen. Filtering them is most of what
9 + makes a board correct.
10 +
11 + The two escapes matter as much as the filter. Two of eight sources there have
12 + no usable service data at all -- one with 33,163 trips and not a single
13 + calendar row -- and a filter without them would show those sources nothing.
14 + """
15 + use RoomSanctum.DataCase
16 +
17 + alias RoomSanctum.{Accounts, Configuration, Repo, Storage}
18 + alias RoomSanctum.Storage.GTFS.{Calendar, CalendarDate, StopTime, Trip}
19 +
20 + # 2026-09-01 is a Tuesday.
21 + @today ~D[2026-09-01]
22 + @at DateTime.new!(~D[2026-09-01], ~T[08:00:00], "Etc/UTC")
23 +
24 + setup do
25 + {:ok, user} =
26 + Accounts.register_user(%{
27 + email: "sd#{System.unique_integer([:positive])}@example.com",
28 + password: "hello world!hello world!"
29 + })
30 +
31 + {:ok, source} =
32 + Configuration.create_source(%{
33 + name: "T", notes: "", type: :gtfs, enabled: true, user_id: user.id,
34 + config: %{"__type__" => "gtfs", "url" => "https://e.test/g.zip", "tz" => "Etc/UTC"}
35 + })
36 +
37 + %{source: source}
38 + end
39 +
40 + defp calendar(source, service_id, days, range \\ {~D[2026-08-31], ~D[2026-09-05]}) do
41 + {start_date, end_date} = range
42 +
43 + attrs =
44 + %{source_id: source.id, service_id: service_id, start_date: start_date, end_date: end_date}
45 + |> Map.merge(Map.new(~w(monday tuesday wednesday thursday friday saturday sunday)a, &{&1, 0}))
46 + |> Map.merge(Map.new(days, &{&1, 1}))
47 +
48 + Repo.insert!(Calendar.changeset(%Calendar{}, attrs))
49 + end
50 +
51 + defp exception(source, service_id, type) do
52 + Repo.insert!(
53 + CalendarDate.changeset(%CalendarDate{}, %{
54 + source_id: source.id, service_id: service_id, date: @today, exception_type: type
55 + })
56 + )
57 + end
58 +
59 + # One trip on a service, calling at "s1" at 08:30.
60 + defp trip_calling(source, service_id) do
61 + trip_id = "trip-#{service_id}-#{System.unique_integer([:positive])}"
62 +
63 + Repo.insert!(%Trip{
64 + source_id: source.id, trip_id: trip_id, route_id: "r1",
65 + service_id: service_id, direction_id: 0, trip_headsign: "Downtown"
66 + })
67 +
68 + Repo.insert!(%StopTime{
69 + source_id: source.id, trip_id: trip_id, stop_id: "s1", stop_sequence: 1,
70 + arrival_time: %Postgrex.Interval{secs: 8 * 3600 + 1800},
71 + departure_time: %Postgrex.Interval{secs: 8 * 3600 + 1800}
72 + })
73 +
74 + trip_id
75 + end
76 +
77 + defp arrivals(source) do
78 + Storage.get_upcoming_arrivals_for_stop(source.id, "s1", 50, @at, "Etc/UTC")
79 + |> Enum.map(& &1.trip_id)
80 + end
81 +
82 + test "a service running today is kept", %{source: source} do
83 + calendar(source, "weekday", [:tuesday])
84 + trip = trip_calling(source, "weekday")
85 +
86 + assert arrivals(source) == [trip]
87 + end
88 +
89 + test "a service for another day is dropped", %{source: source} do
90 + calendar(source, "weekday", [:tuesday])
91 + calendar(source, "saturday", [:saturday])
92 + keep = trip_calling(source, "weekday")
93 + _drop = trip_calling(source, "saturday")
94 +
95 + assert arrivals(source) == [keep]
96 + end
97 +
98 + test "a service whose date range has passed is dropped", %{source: source} do
99 + calendar(source, "weekday", [:tuesday])
100 + calendar(source, "expired", [:tuesday], {~D[2026-07-01], ~D[2026-08-28]})
101 + keep = trip_calling(source, "weekday")
102 + _drop = trip_calling(source, "expired")
103 +
104 + assert arrivals(source) == [keep]
105 + end
106 +
107 + test "a service added for today by exception is kept", %{source: source} do
108 + calendar(source, "weekday", [:tuesday])
109 + # Every weekday flag zero -- the MBTA shape, which runs only by exception.
110 + calendar(source, "by-exception", [], {@today, @today})
111 + exception(source, "by-exception", 1)
112 +
113 + keep = trip_calling(source, "weekday")
114 + also = trip_calling(source, "by-exception")
115 +
116 + assert Enum.sort(arrivals(source)) == Enum.sort([keep, also])
117 + end
118 +
119 + test "a service removed for today by exception is dropped", %{source: source} do
120 + calendar(source, "weekday", [:tuesday])
121 + calendar(source, "holiday", [:tuesday])
122 + exception(source, "holiday", 2)
123 +
124 + keep = trip_calling(source, "weekday")
125 + _drop = trip_calling(source, "holiday")
126 +
127 + assert arrivals(source) == [keep]
128 + end
129 +
130 + describe "the escapes" do
131 + test "a service nothing says anything about is kept", %{source: source} do
132 + calendar(source, "weekday", [:tuesday])
133 + keep = trip_calling(source, "weekday")
134 + # No calendar row, no exception: we have not been told when this runs, so
135 + # hiding it would assert something we do not know.
136 + unknown = trip_calling(source, "no-info")
137 +
138 + assert Enum.sort(arrivals(source)) == Enum.sort([keep, unknown])
139 + end
140 +
141 + test "a source with nothing running today is not filtered at all", %{source: source} do
142 + # Stale service data -- the state two of eight dev sources are actually in.
143 + calendar(source, "expired-a", [:tuesday], {~D[2026-07-01], ~D[2026-08-28]})
144 + calendar(source, "expired-b", [:saturday], {~D[2026-07-01], ~D[2026-08-28]})
145 +
146 + a = trip_calling(source, "expired-a")
147 + b = trip_calling(source, "expired-b")
148 +
149 + # A board of possibly-wrong times beats a blank one.
150 + assert Enum.sort(arrivals(source)) == Enum.sort([a, b])
151 + end
152 +
153 + test "one service running today is enough to switch filtering on", %{source: source} do
154 + calendar(source, "weekday", [:tuesday])
155 + calendar(source, "saturday", [:saturday])
156 +
157 + keep = trip_calling(source, "weekday")
158 + _drop = trip_calling(source, "saturday")
159 +
160 + assert arrivals(source) == [keep]
161 + end
162 +
163 + test "an exception alone is enough to switch filtering on", %{source: source} do
164 + calendar(source, "saturday", [:saturday])
165 + exception(source, "added", 1)
166 +
167 + added = trip_calling(source, "added")
168 + _drop = trip_calling(source, "saturday")
169 +
170 + assert arrivals(source) == [added]
171 + end
172 + end
173 +end

Parents: 27a12aa