feat(commuter_rail_occupancies): read commuter rail occupancies from S3 instead of firebase (#858)

Problem:
Keolis needs to decommission their Firebase feed

Solution:
We're having them move the contents of that Firebase feed to S3.

Author: rudiejd

apps/api_accounts/config/dev.exs

@@ -2,8 +2,4 @@ import Config
 
 config :api_accounts, table_prefix: "DEV"
 
-config :ex_aws,
-  access_key_id: "DevAccessKey",
-  secret_access_key: "DevSecretKey"
-
 config :api_accounts, ApiAccounts.Mailer, adapter: Bamboo.LocalAdapter

apps/parse/lib/parse/commuter_rail_occupancies.ex

@@ -13,20 +13,38 @@ defmodule Parse.CommuterRailOccupancies do
       {:ok, %{"data" => data}} when is_list(data) ->
         Enum.flat_map(data, &parse_record/1)
 
+      {:ok, data} when is_list(data) ->
+        Enum.flat_map(data, &parse_record/1)
+
       e ->
         Logger.warning("#{__MODULE__} decode_error e=#{inspect(e)}")
         []
     end
   end
 
-  defp parse_record(
-         %{
-           "MedianDensity" => density,
-           "MedianDensityFlag" => flag,
-           "cTrainNo" => train
-         } = record
-       ) do
-    with {:ok, flag} <- density_flag(flag),
+  defp parse_density_fields(%{
+         "MedianDensity" => density,
+         "MedianDensityFlag" => flag,
+         "cTrainNo" => train
+       }),
+       do: {:ok, {density, flag, train}}
+
+  # new format keolis started providing when they switched this data over to S3
+  defp parse_density_fields(%{
+         "Median Density" => density,
+         "Median Density Flag" => flag,
+         "Trip Name" => train
+       }),
+       do: {:ok, {density, flag, train}}
+
+  defp parse_density_fields(record) do
+    Logger.warning("#{__MODULE__} parse_error error=missing_fields #{inspect(record)}")
+    {:error, :missing_fields}
+  end
+
+  defp parse_record(record) do
+    with {:ok, {density, flag, train}} <- parse_density_fields(record),
+         {:ok, flag} <- density_flag(flag),
          {:ok, percentage} <- percentage(density),
          {:ok, name} <- trip_name(train) do
       [
@@ -43,11 +61,6 @@ defmodule Parse.CommuterRailOccupancies do
     end
   end
 
-  defp parse_record(record) do
-    Logger.warning("#{__MODULE__} parse_error error=missing_fields #{inspect(record)}")
-    []
-  end
-
   defp density_flag(0), do: {:ok, :many_seats_available}
   defp density_flag(1), do: {:ok, :few_seats_available}
   defp density_flag(2), do: {:ok, :full}

apps/state_mediator/config/config.exs

@@ -10,7 +10,10 @@ config :state_mediator, :commuter_rail_crowding,
     "CR_CROWDING_BASE_URL",
     "https://keolis-api-development.firebaseio.com/p-kcs-trms-firebase-7dayloading.json"
   },
-  enabled: {:system, "CR_CROWDING_ENABLED", "false"}
+  enabled: {:system, "CR_CROWDING_ENABLED", "true"},
+  s3_bucket: {:system, "CR_CROWDING_S3_BUCKET"},
+  s3_object: {:system, "CR_CROWDING_S3_OBJECT"},
+  source: {:system, "CR_CROWING_SOURCE", "s3"}
 
 config :state_mediator, Realtime,
   gtfs_url: {:system, "MBTA_GTFS_URL", "https://cdn.mbta.com/MBTA_GTFS.zip"},

apps/state_mediator/lib/state_mediator.ex

@@ -7,9 +7,14 @@ defmodule StateMediator do
   # See http://elixir-lang.org/docs/stable/elixir/Application.html
   # for more information on OTP Applications
   def start(_type, _args) do
+    crowding_source = app_value(:commuter_rail_crowding, :source)
+
     children =
       children(Application.get_env(:state_mediator, :start)) ++
-        crowding_children(app_value(:commuter_rail_crowding, :enabled) == "true")
+        crowding_children(
+          app_value(:commuter_rail_crowding, :enabled) == "true",
+          crowding_source
+        )
 
     # See http://elixir-lang.org/docs/stable/elixir/Supervisor.html
     # for other strategies and supported options
@@ -94,9 +99,30 @@ defmodule StateMediator do
     }
   end
 
-  @spec crowding_children(boolean()) :: [:supervisor.child_spec() | {module(), term()} | module()]
-  defp crowding_children(true) do
-    Logger.info("#{__MODULE__} CR_CROWDING_ENABLED=true")
+  @spec crowding_children(boolean(), String.t()) :: [
+          :supervisor.child_spec() | {module(), term()} | module()
+        ]
+  defp crowding_children(true, "s3") do
+    Logger.info("#{__MODULE__} CR_CROWDING_ENABLED=true, source=s3")
+
+    [
+      {
+        StateMediator.S3Mediator,
+        [
+          spec_id: :cr_s3_crowding_mediator,
+          bucket_arn: "mbta-gtfs-commuter-rail-staging",
+          object: "crowding-trends.json",
+          spec_id: :s3_mediator,
+          interval: 5 * 60 * 1_000,
+          sync_timeout: 30_000,
+          state: State.CommuterRailOccupancy
+        ]
+      }
+    ]
+  end
+
+  defp crowding_children(true, "firebase") do
+    Logger.info("#{__MODULE__} CR_CROWDING_ENABLED=true, source=firebase")
 
     credentials = :commuter_rail_crowding |> app_value(:firebase_credentials) |> Jason.decode!()
 
@@ -124,7 +150,7 @@ defmodule StateMediator do
     ]
   end
 
-  defp crowding_children(false) do
+  defp crowding_children(false, _) do
     Logger.info("#{__MODULE__} CR_CROWDING_ENABLED=false")
     []
   end

apps/state_mediator/lib/state_mediator/s3_mediator.ex

@@ -0,0 +1,114 @@
+defmodule StateMediator.S3Mediator do
+  @moduledoc """
+
+  S3Mediator is responsible for reading files from an S3 bucket and
+  sending messages to the state module.
+
+  """
+
+  defstruct [
+    :module,
+    :bucket_arn,
+    :object,
+    :sync_timeout,
+    :interval
+  ]
+
+  @opaque t :: %__MODULE__{
+            module: module,
+            bucket_arn: String.t(),
+            object: String.t(),
+            sync_timeout: pos_integer()
+          }
+
+  use GenServer
+  require Logger
+  alias ExAws.S3
+
+  def child_spec(opts) do
+    {spec_id, opts} = Keyword.pop!(opts, :spec_id)
+
+    %{
+      id: spec_id,
+      start: {__MODULE__, :start_link, [opts]}
+    }
+  end
+
+  @spec start_link(Keyword.t()) :: {:ok, pid}
+  def start_link(options) do
+    GenServer.start_link(__MODULE__, options)
+  end
+
+  @spec stop(pid) :: :ok
+  def stop(pid) do
+    GenServer.stop(pid)
+  end
+
+  @spec init(Keyword.t()) :: {:ok, __MODULE__.t(), {:continue, any()}}
+  def init(options) do
+    state_module = Keyword.fetch!(options, :state)
+
+    bucket_arn = Keyword.fetch!(options, :bucket_arn)
+    object = Keyword.fetch!(options, :object)
+    sync_timeout = options |> Keyword.get(:sync_timeout, 5000)
+    interval = options |> Keyword.get(:interval, 5000)
+
+    state = %__MODULE__{
+      interval: interval,
+      module: state_module,
+      bucket_arn: bucket_arn,
+      object: object,
+      sync_timeout: sync_timeout
+    }
+
+    {:ok, state, {:continue, nil}}
+  end
+
+  @spec handle_continue(any, t) :: {:noreply, t} | {:noreply, t, :hibernate}
+  def handle_continue(_, %{module: state_module} = state) do
+    _ = Logger.debug(fn -> "#{__MODULE__} #{state_module} initial sync starting" end)
+    fetch(state)
+  end
+
+  def handle_info(:timeout, %{module: state_module} = state) do
+    _ = Logger.debug(fn -> "#{__MODULE__} #{state_module} timeout sync starting" end)
+    fetch(state)
+  end
+
+  defp fetch(%{bucket_arn: bucket_arn, object: object} = state) do
+    bucket_arn
+    |> S3.get_object(object)
+    |> ExAws.request()
+    |> handle_response(state)
+  end
+
+  defp handle_response(
+         {:ok, %{body: body}},
+         %{sync_timeout: sync_timeout, module: state_module} = state
+       ) do
+    debug_time("#{state_module} new state", fn -> state_module.new_state(body, sync_timeout) end)
+
+    schedule_update(state)
+  end
+
+  defp handle_response(
+         response,
+         state
+       ) do
+    Logger.warning(
+      "Received unknown response when getting commuter rail occupancies from S3: #{inspect(response)}"
+    )
+
+    schedule_update(state)
+  end
+
+  defp schedule_update(%{interval: interval} = state) when interval != nil do
+    {:noreply, state, interval}
+  end
+
+  defp debug_time(description, func) do
+    State.Logger.debug_time(func, fn milliseconds ->
+      "#{__MODULE__} #{description} took #{milliseconds}ms"
+    end)
+  end
+end

apps/state_mediator/mix.exs

@@ -51,7 +51,8 @@ defmodule StateMediator.Mixfile do
       {:goth, "~> 1.3"},
       {:hackney, "~> 1.18"},
       {:timex, "~> 3.7"},
-      {:emqtt_failover, "~> 0.3"}
+      {:emqtt_failover, "~> 0.3"},
+      {:mox, "~> 1.0", only: :test}
     ]
   end
 end

apps/state_mediator/test/state_mediator/s3_mediator_test.exs

@@ -0,0 +1,58 @@
+defmodule StateMediator.S3MediatorTest do
+  use ExUnit.Case, async: true
+
+  import Mox
+  import StateMediator.S3Mediator
+
+  defmodule StateModule do
+    def size do
+      0
+    end
+
+    def new_state(pid, _timeout) do
+      send(pid, :received_new_state)
+    end
+  end
+
+  @moduletag capture_log: true
+  @opts [
+    bucket_arn: "mbta-gtfs-boom-shakalaka",
+    object: "objection",
+    state: __MODULE__.StateModule,
+    interval: 1_000
+  ]
+
+  describe "init/1" do
+    test "fires a continue" do
+      assert {:ok, _, {:continue, _}} = init(@opts)
+    end
+
+    test "builds an initial state" do
+      assert {:ok, state, {:continue, _}} = init(@opts)
+      assert %StateMediator.S3Mediator{} = state
+      assert state.module == @opts[:state]
+      assert state.bucket_arn == @opts[:bucket_arn]
+      assert state.sync_timeout == 5_000
+      assert state.interval == 1_000
+    end
+  end
+
+  describe "handle_info/2" do
+    test "on body: schedules an update" do
+      {:ok, state, {:continue, _}} = init(@opts)
+      assert {:noreply, ^state, 1_000} = handle_info(:timeout, state)
+    end
+
+    test "on error: schedules an update" do
+      {:ok, state, {:continue, _}} = init(@opts)
+      Mox.defmock(FakeAws, for: ExAws.Behaviour)
+
+      test_pid = self()
+      monitor_pid = GenServer.whereis(StateMediator.S3Mediator)
+      allow(FakeAws, test_pid, monitor_pid)
+      stub(FakeAws, :request, fn _ -> {:error, %{body: "your transit isn't rapid enough"}} end)
+
+      assert {:noreply, ^state, 1_000} = handle_info(:timeout, state)
+    end
+  end
+end

config/runtime.exs

@@ -28,7 +28,9 @@ if is_prod? and is_release? do
       port: "DYNAMO_PORT" |> System.get_env("443") |> String.to_integer(),
       scheme: System.get_env("DYNAMO_SCHEME", "https://"),
       host: System.fetch_env!("DYNAMO_HOST")
-    ]
+    ],
+    json_codec: Jason
+
 
   config :alb_monitor,
     ecs_metadata_uri: System.fetch_env!("ECS_CONTAINER_METADATA_URI"),
@@ -64,7 +66,10 @@ if is_prod? and is_release? do
     secret_key_base: System.fetch_env!("SECRET_KEY_BASE")
 
   config :state_mediator, :commuter_rail_crowding,
-    firebase_credentials: System.fetch_env!("CR_CROWDING_FIREBASE_CREDENTIALS")
+    firebase_credentials: System.fetch_env!("CR_CROWDING_FIREBASE_CREDENTIALS"),
+    s3_bucket: System.fetch_env!("CR_CROWDING_S3_BUCKET"),
+    s3_object: System.fetch_env!("CR_CROWDING_S3_OBJECT"),
+    source: System.fetch_env!("CR_CROWDING_SOURCE")
 
   config :recaptcha,
     enabled: true,