feat(state,parse): Perform partial updates on StopEvent

apps/model/lib/model/stop_event.ex

@@ -14,7 +14,8 @@ defmodule Model.StopEvent do
     :stop_id,
     :stop_sequence,
     :arrived,
-    :departed
+    :departed,
+    :timestamp
   ]
 
   @typedoc """
@@ -35,6 +36,7 @@ defmodule Model.StopEvent do
       See [GTFS `stop_times.txt` `stop_sequence`](https://gtfs.org/documentation/schedule/reference/#stop_timestxt).
   * `:arrived` - When the vehicle (`vehicle_id`) arrived at the stop (`stop_id`) as a time-zone aware [RFC 3339 datetime](https://datatracker.ietf.org/doc/html/rfc3339#page-10). `nil` if the stop is the first stop on the trip (`trip_id`).
   * `:departed` - When the vehicle (`vehicle_id`) departed from the stop (`stop_id`) as time-zone aware [RFC 3339 datetime](https://datatracker.ietf.org/doc/html/rfc3339#page-10). `nil` if the last stop (`stop_id`) on the trip (`trip_id`).
+  * `:timestamp` - Unix timestamp representing when this stop event record was last updated in the source system.
   """
 
   @type t :: %__MODULE__{
@@ -48,6 +50,7 @@ defmodule Model.StopEvent do
           stop_sequence: Model.Schedule.stop_sequence(),
           revenue: :REVENUE | :NON_REVENUE,
           arrived: DateTime.t() | nil,
-          departed: DateTime.t() | nil
+          departed: DateTime.t() | nil,
+          timestamp: integer() | nil
         }
 end

apps/parse/lib/parse/stop_events.ex

@@ -10,13 +10,28 @@ defmodule Parse.StopEvents do
   @behaviour Parse
 
   @impl Parse
-  @spec parse(binary()) :: [Model.StopEvent.t()]
-  def parse(body) do
-    body
-    |> decompress()
-    |> String.split("\n", trim: true)
-    |> Enum.map(&parse_line/1)
-    |> Enum.reject(&is_nil/1)
+  def parse(binary) when is_binary(binary) do
+    parse(binary, [])
+  end
+
+  @spec parse(binary(), keyword()) :: [Model.StopEvent.t()] | {:partial, [Model.StopEvent.t()]}
+  def parse(body, opts) when is_binary(body) and is_list(opts) do
+    newer_than = Keyword.get(opts, :newer_than)
+
+    events =
+      body
+      |> decompress()
+      |> String.split("\n", trim: true)
+      |> Enum.map(&parse_line(&1, newer_than))
+      |> Enum.reject(&is_nil/1)
+
+    # When filtering by timestamp, always return {:partial, events} tuple
+    # to distinguish filtered updates from full state replacements
+    if newer_than do
+      {:partial, events}
+    else
+      events
+    end
   end
 
   defp decompress(body) do
@@ -25,17 +40,32 @@ defmodule Parse.StopEvents do
     _ -> body
   end
 
-  defp parse_line(line) do
+  defp parse_line(line, newer_than) do
     case Jason.decode(line) do
       {:ok, record} ->
-        parse_record(record)
+        if should_include_record?(record, newer_than) do
+          parse_record(record)
+        else
+          nil
+        end
 
       e ->
         Logger.error("#{__MODULE__} decode_error error=#{inspect(e)}")
         nil
     end
   end
 
+  defp should_include_record?(_record, nil), do: true
+
+  defp should_include_record?(%{"timestamp" => timestamp}, newer_than)
+       when is_integer(timestamp) and is_integer(newer_than) do
+    timestamp > newer_than
+  end
+
+  # Records without a timestamp field are excluded when filtering is active.
+  # This treats records without timestamps as stale/invalid when doing incremental updates.
+  defp should_include_record?(_record, _newer_than), do: false
+
   defp parse_record(
          %{
            "id" => id,
@@ -64,7 +94,8 @@ defmodule Parse.StopEvents do
         stop_id: stop_id,
         stop_sequence: stop_sequence,
         arrived: arrived,
-        departed: departed
+        departed: departed,
+        timestamp: Map.get(record, "timestamp")
       }
     else
       {:error, reason} ->

apps/parse/test/parse/stop_events_test.exs

@@ -33,7 +33,8 @@ defmodule Parse.StopEventsTest do
           stop_id: "70512",
           stop_sequence: 4,
           arrived: DateTime.from_naive!(~N[2026-02-24T10:18:23], "America/New_York"),
-          departed: DateTime.from_naive!(~N[2026-02-24T10:21:19], "America/New_York")
+          departed: DateTime.from_naive!(~N[2026-02-24T10:21:19], "America/New_York"),
+          timestamp: 1_771_950_045
         },
         # arrival only
         %StopEvent{
@@ -47,7 +48,8 @@ defmodule Parse.StopEventsTest do
           stop_id: "2231",
           stop_sequence: 1,
           arrived: DateTime.from_naive!(~N[2026-02-24T15:54:46], "America/New_York"),
-          departed: nil
+          departed: nil,
+          timestamp: 1_771_968_343
         },
         # departure only
         %StopEvent{
@@ -61,7 +63,8 @@ defmodule Parse.StopEventsTest do
           stop_id: "12232",
           stop_sequence: 2,
           arrived: nil,
-          departed: DateTime.from_naive!(~N[2026-02-24T16:08:53], "America/New_York")
+          departed: DateTime.from_naive!(~N[2026-02-24T16:08:53], "America/New_York"),
+          timestamp: 1_771_968_343
         }
       ]
 
@@ -152,4 +155,71 @@ defmodule Parse.StopEventsTest do
       assert [%StopEvent{id: "test-trip-1"}] = result
     end
   end
+
+  describe "parse with timestamp filtering" do
+    test "returns {:partial, events} when newer_than option filters out old records" do
+      ndjson = """
+      {"id":"old-1","timestamp":1771968300,"start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246}
+      {"id":"old-2","timestamp":1771968340,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246}
+      {"id":"new-1","timestamp":1771968350,"start_date":"20260224","trip_id":"test3","vehicle_id":"v3","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop3","stop_sequence":3,"arrived":1771966486,"departed":1771967246}
+      {"id":"new-2","timestamp":1771968360,"start_date":"20260224","trip_id":"test4","vehicle_id":"v4","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop4","stop_sequence":4,"arrived":1771966486,"departed":1771967246}
+      """
+
+      result = parse(ndjson, newer_than: 1_771_968_343)
+
+      assert {:partial, events} = result
+      assert length(events) == 2
+      assert Enum.all?(events, fn e -> e.id in ["new-1", "new-2"] end)
+    end
+
+    test "returns full list when newer_than option is not provided" do
+      ndjson = """
+      {"id":"event-1","timestamp":1771968300,"start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246}
+      {"id":"event-2","timestamp":1771968350,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246}
+      """
+
+      result = parse(ndjson)
+
+      assert is_list(result)
+      refute match?({:partial, _}, result)
+      assert length(result) == 2
+    end
+
+    test "returns {:partial, []} when all records are older than newer_than" do
+      ndjson = """
+      {"id":"old-1","timestamp":1771968300,"start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246}
+      {"id":"old-2","timestamp":1771968340,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246}
+      """
+
+      result = parse(ndjson, newer_than: 1_771_968_400)
+
+      assert {:partial, []} = result
+    end
+
+    test "excludes records at the boundary (timestamp == newer_than)" do
+      ndjson = """
+      {"id":"at-boundary","timestamp":1771968343,"start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246}
+      {"id":"after-boundary","timestamp":1771968344,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246}
+      """
+
+      result = parse(ndjson, newer_than: 1_771_968_343)
+
+      assert {:partial, events} = result
+      assert length(events) == 1
+      assert [%StopEvent{id: "after-boundary"}] = events
+    end
+
+    test "handles records with missing timestamp field by excluding them" do
+      ndjson = """
+      {"id":"no-timestamp","start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246}
+      {"id":"with-timestamp","timestamp":1771968350,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246}
+      """
+
+      result = parse(ndjson, newer_than: 1_771_968_343)
+
+      assert {:partial, events} = result
+      assert length(events) == 1
+      assert [%StopEvent{id: "with-timestamp"}] = events
+    end
+  end
 end

apps/state/config/config.exs

@@ -867,4 +867,9 @@ config :state, :stops_on_route,
     ]
   }
 
+# Configuration for StopEvent state
+config :state, State.StopEvent,
+  # Retention period in seconds (default: 2 hours = 7200 seconds)
+  retention_seconds: 7200
+
 import_config "#{config_env()}.exs"

apps/state/lib/state/stop_event.ex

@@ -25,6 +25,131 @@ defmodule State.StopEvent do
   # Filter keys ordered by typical selectivity (most selective first)
   @index_keys [:trip_ids, :vehicle_ids, :stop_ids, :route_ids]
 
+  @impl State.Server
+  def handle_new_state(binary) when is_binary(binary) do
+    # Get the maximum timestamp from existing data
+    max_timestamp = get_max_timestamp()
+
+    # Parse with timestamp filtering if we have existing data
+    opts = if max_timestamp, do: [newer_than: max_timestamp], else: []
+
+    parser = Parse.StopEvents
+
+    parsed_data =
+      try do
+        parser.parse(binary, opts)
+      rescue
+        e ->
+          # log_parse_error returns nil, so we return nil on error to avoid passing
+          # invalid data to super/1. State.Server expects nil to mean
+          # "no data to insert" and will handle it gracefully.
+          State.Server.log_parse_error(__MODULE__, e)
+      end
+
+    # Only proceed with update if parsing succeeded
+    case parsed_data do
+      nil -> :ok
+      data -> super(data)
+    end
+  end
+
+  def handle_new_state(data), do: super(data)
+
+  @impl State.Server
+  def post_commit_hook do
+    evict_old_records()
+    :ok
+  end
+
+  @doc """
+  Evicts records older than the configured retention period.
+
+  ## Parameters
+
+    * `retention_seconds` - Number of seconds to retain records. Defaults to
+       7200 seconds / 2 hours). Records with timestamps older than `now -
+       retention_seconds` will be deleted.
+
+  ## Examples
+
+      # Use default retention period (2 hours)
+      evict_old_records()
+
+      # Use custom retention period (1 hour)
+      evict_old_records(3600)
+
+  """
+  @spec evict_old_records() :: :ok
+  def evict_old_records do
+    evict_old_records(get_retention_seconds())
+  end
+
+  @spec evict_old_records(non_neg_integer()) :: :ok
+  def evict_old_records(retention_seconds) when retention_seconds >= 0 do
+    cutoff = System.system_time(:second) - retention_seconds
+    match_spec = build_eviction_match_spec(cutoff)
+
+    __MODULE__
+    |> :mnesia.dirty_select(match_spec)
+    |> Enum.each(&:mnesia.dirty_delete(__MODULE__, &1))
+
+    :ok
+  end
+
+  defp build_eviction_match_spec(cutoff) do
+    fields = StopEvent.fields()
+    timestamp_pos = Enum.find_index(fields, &(&1 == :timestamp))
+    id_pos = Enum.find_index(fields, &(&1 == :id))
+
+    pattern =
+      :_
+      |> List.duplicate(length(fields))
+      |> List.replace_at(timestamp_pos, :"$1")
+      |> List.replace_at(id_pos, :"$2")
+      |> then(&[StopEvent | &1])
+      |> List.to_tuple()
+
+    # Match spec format: {pattern, guards, result}
+    [{pattern, [{:andalso, {:"/=", :"$1", nil}, {:<, :"$1", cutoff}}], [:"$2"]}]
+  end
+
+  # Get the maximum timestamp from existing data in the table.
+  # Uses a dynamic match spec based on the StopEvent struct to avoid brittleness
+  # from hardcoded field positions.
+  defp get_max_timestamp do
+    # Mnesia records are tuples of {RecordName, field1, field2, ...} where
+    # fields follow Recordable declaration order (from StopEvent.fields/0).
+    fields = StopEvent.fields()
+    timestamp_position = Enum.find_index(fields, &(&1 == :timestamp))
+
+    unless timestamp_position do
+      raise "timestamp field not found in StopEvent struct"
+    end
+
+    # Build tuple pattern: {StopEvent, :_, :_, ..., :"$1"} with $1 at timestamp's position
+    wildcards = List.duplicate(:_, length(fields))
+    pattern_list = [StopEvent | List.replace_at(wildcards, timestamp_position, :"$1")]
+    pattern = List.to_tuple(pattern_list)
+
+    match_spec = [{pattern, [], [:"$1"]}]
+
+    case :mnesia.dirty_select(__MODULE__, match_spec) do
+      [] ->
+        nil
+
+      timestamps ->
+        timestamps
+        |> Enum.reject(&is_nil/1)
+        |> Enum.max(fn -> nil end)
+    end
+  end
+
+  defp get_retention_seconds do
+    :state
+    |> Application.get_env(__MODULE__, [])
+    |> Keyword.get(:retention_seconds, 7200)
+  end
+
   @spec by_id(String.t()) :: StopEvent.t() | nil
   def by_id(id) do
     case super(id) do