feat: Support Keycloak auth

.envrc.example

@@ -2,6 +2,7 @@ export ALERT_API_URL=http://s3.amazonaws.com/mbta-realtime-test/alerts_enhanced.
 export API_KEY=
 export API_URL=https://api-dev.mbtace.com/
 export APP_COOKIE=
+export AUTHENTICATION_SOURCE=
 export AWS_ACCESS_KEY_ID=
 export AWS_SECRET_ACCESS_KEY=
 export DATABASE_URL_DEV=postgresql://__username__:password@localhost:5432/alert_concierge_dev

apps/alert_processor/config/config.exs

@@ -40,6 +40,8 @@ config :alert_processor,
 config :alert_processor, api_url: {:system, "API_URL", "https://api-dev.mbtace.com/"}
 config :alert_processor, api_key: {:system, "API_KEY", nil}
 
+config :alert_processor, user_update_sqs_queue_url: {:system, "USER_UPDATE_SQS_QUEUE_URL", ""}
+
 # Number of workers for sending notifications
 config :alert_processor, notification_workers: 40
 

apps/alert_processor/config/test.exs

@@ -14,6 +14,9 @@ config :alert_processor, :ex_aws, ExAws.Mock
 # Don't run periodic background alert processing
 config :alert_processor, :process_alerts?, false
 
+# Don't run periodic background user update polling
+config :alert_processor, :poll_for_user_updates?, false
+
 config :alert_processor, :mailer, AlertProcessor.MailerMock
 config :alert_processor, :mailer_email, AlertProcessor.EmailMock
 

apps/alert_processor/lib/alert_processor.ex

@@ -21,21 +21,28 @@ defmodule AlertProcessor do
         do: [],
         else: [check_interval: nil]
 
-    children = [
-      AlertProcessor.Repo,
-      {ConCache,
-       [
-         name: AlertProcessor.CachedApiClient.cache_name(),
-         global_ttl: :timer.minutes(60),
-         ttl_check_interval: :timer.seconds(60)
-       ]},
-      AlertProcessor.ServiceInfoCache,
-      AlertProcessor.SendingQueue,
-      AlertProcessor.Reminders,
-      {AlertProcessor.AlertWorker, alert_worker_config},
-      AlertProcessor.SmsOptOutWorker,
-      :poolboy.child_spec(:message_worker, message_worker_config, [])
-    ]
+    user_update_children =
+      if Application.get_env(:alert_processor, :poll_for_user_updates?, true),
+        do: [AlertProcessor.UserUpdateWorker],
+        else: []
+
+    children =
+      [
+        AlertProcessor.Repo,
+        {ConCache,
+         [
+           name: AlertProcessor.CachedApiClient.cache_name(),
+           global_ttl: :timer.minutes(60),
+           ttl_check_interval: :timer.seconds(60)
+         ]},
+        AlertProcessor.ServiceInfoCache,
+        AlertProcessor.SendingQueue,
+        AlertProcessor.Reminders,
+        {AlertProcessor.AlertWorker, alert_worker_config},
+        AlertProcessor.SmsOptOutWorker,
+        :poolboy.child_spec(:message_worker, message_worker_config, [])
+      ] ++
+        user_update_children
 
     Supervisor.start_link(children, strategy: :one_for_one)
   end

apps/alert_processor/lib/helpers/phone_number.ex

@@ -0,0 +1,9 @@
+defmodule AlertProcessor.Helpers.PhoneNumber do
+  @moduledoc """
+  Helper functions for parsing and formatting phone numbers
+  """
+
+  @spec strip_us_country_code(String.t() | nil) :: String.t() | nil
+  def strip_us_country_code("+1" <> phone_number), do: phone_number
+  def strip_us_country_code(phone_number), do: phone_number
+end

apps/alert_processor/lib/model/user.ex

@@ -121,7 +121,6 @@ defmodule AlertProcessor.Model.User do
     |> update_change(:email, &lowercase_email/1)
     |> validate_email()
     |> validate_password()
-    |> clear_phone_if_mode_is_email(params)
     |> update_change(:phone_number, &clean_phone_number/1)
     |> validate_phone_number()
     |> hash_password()
@@ -135,7 +134,11 @@ defmodule AlertProcessor.Model.User do
         %{"communication_mode" => "sms", "email" => _email} = params
       ) do
     struct
-    |> changeset(params, [:phone_number, :email])
+    |> changeset(params, [:email])
+    # Validate phone number as required separately for the custom error message.
+    |> validate_required([:phone_number],
+      message: "Please click the link above to add your phone number to your account."
+    )
     |> update_change(:phone_number, &clean_phone_number/1)
     |> validate_phone_number()
     |> validate_accept_tnc(params)
@@ -149,7 +152,11 @@ defmodule AlertProcessor.Model.User do
         %{"communication_mode" => "sms"} = params
       ) do
     struct
-    |> changeset(params, [:phone_number])
+    |> changeset(params)
+    # Validate phone number as required separately for the custom error message.
+    |> validate_required([:phone_number],
+      message: "Please click the link above to add your phone number to your account."
+    )
     |> update_change(:phone_number, &clean_phone_number/1)
     |> validate_phone_number()
     |> validate_accept_tnc(params)
@@ -177,12 +184,6 @@ defmodule AlertProcessor.Model.User do
     |> unique_constraint(:email, message: "Sorry, that email has already been taken.")
   end
 
-  defp clear_phone_if_mode_is_email(changeset, %{"communication_mode" => "email"}) do
-    update_change(changeset, :phone_number, &clear_value/1)
-  end
-
-  defp clear_phone_if_mode_is_email(changeset, _), do: changeset
-
   defp clear_value(_), do: nil
 
   defp validate_phone_number(changeset) do
@@ -331,8 +332,8 @@ defmodule AlertProcessor.Model.User do
     |> Repo.transaction()
   end
 
-  defp normalize_papertrail_result({:ok, %{model: user}}), do: {:ok, user}
-  defp normalize_papertrail_result(result), do: result
+  @spec get(id()) :: t() | nil
+  def get(id), do: Repo.get(__MODULE__, id)
 
   @spec for_email(String.t()) :: t | nil
   def for_email(email) do
@@ -364,4 +365,29 @@ defmodule AlertProcessor.Model.User do
 
   def inside_opt_out_freeze_window?(%__MODULE__{sms_opted_out_at: sms_opted_out_at}),
     do: Date.diff(Date.utc_today(), DateTime.to_date(sms_opted_out_at)) <= 30
+
+  @doc """
+  Returns the email address for the given user.
+
+  ## Examples
+
+      iex> User.email(%User{email: "[email protected]"})
+      "[email protected]"
+  """
+  @spec email(t()) :: String.t()
+  def email(%__MODULE__{email: email}), do: email
+
+  @doc """
+  Returns the phone number for the given user.
+
+  ## Examples
+
+      iex> User.phone_number(%User{phone_number: "5551234567"})
+      "5551234567"
+  """
+  @spec phone_number(t()) :: String.t() | nil
+  def phone_number(%__MODULE__{phone_number: phone_number}), do: phone_number
+
+  defp normalize_papertrail_result({:ok, %{model: user}}), do: {:ok, user}
+  defp normalize_papertrail_result(result), do: result
 end

apps/alert_processor/lib/user_update_worker.ex

@@ -0,0 +1,193 @@
+defmodule AlertProcessor.UserUpdateWorker do
+  @moduledoc """
+  Fetch user update message sent from Keycloak via an SQS queue.
+  Apply the change to our local record.
+  """
+
+  use GenServer
+
+  require Logger
+
+  alias AlertProcessor.Helpers.{ConfigHelper, PhoneNumber}
+  alias AlertProcessor.Model.User
+  alias ExAws.SQS
+
+  @type message :: %{
+          user_update: user_update(),
+          receipt_handle: String.t()
+        }
+
+  @type user_update :: %{
+          required(:user_id) => String.t(),
+          required(:updates) => %{
+            optional(:email) => String.t(),
+            optional(:phone_number) => String.t()
+          }
+        }
+
+  # Client
+
+  @spec start_link() :: GenServer.on_start()
+  @spec start_link(Keyword.t()) :: GenServer.on_start()
+  def start_link(opts \\ []) do
+    name = Keyword.get(opts, :name, __MODULE__)
+    GenServer.start_link(__MODULE__, opts, name: name)
+  end
+
+  # Server
+
+  @impl GenServer
+  def init(_opts) do
+    send(self(), :fetch_message)
+    {:ok, nil}
+  end
+
+  @impl GenServer
+  def handle_info(:fetch_message, state) do
+    {:ok, messages} = receive_messages()
+
+    for message <- messages do
+      case update_user_record(message.user_update) do
+        :ok ->
+          # Delete the message from SQS once successfully processed
+          Logger.info(
+            "Finished processing and deleting SQS message receipt_handle=#{message.receipt_handle}"
+          )
+
+          delete_message_fn =
+            Application.get_env(:alert_processor, :delete_message_fn, &SQS.delete_message/2)
+
+          request_fn = Application.get_env(:alert_processor, :request_fn, &ExAws.request/2)
+
+          user_update_sqs_queue_url()
+          |> delete_message_fn.(message.receipt_handle)
+          |> request_fn.(region: sqs_aws_region())
+
+          :ok
+
+        :error ->
+          # Leave the message in SQS since it hasn't yet been successfully processed
+          :error
+      end
+    end
+
+    send(self(), :fetch_message)
+    {:noreply, state}
+  end
+
+  @impl GenServer
+  def handle_info(_, state), do: {:noreply, state}
+
+  @spec receive_messages :: {:ok, [message()]}
+  defp receive_messages do
+    receive_message_fn =
+      Application.get_env(:alert_processor, :receive_message_fn, &SQS.receive_message/2)
+
+    request_fn = Application.get_env(:alert_processor, :request_fn, &ExAws.request/2)
+
+    Logger.info("Waiting to receive SQS messages")
+
+    user_update_sqs_queue_url()
+    |> receive_message_fn.(wait_time_seconds: 20)
+    |> request_fn.(region: sqs_aws_region())
+    |> handle_sqs_results()
+  end
+
+  @spec user_update_sqs_queue_url :: String.t()
+  defp user_update_sqs_queue_url, do: ConfigHelper.get_string(:user_update_sqs_queue_url)
+
+  @spec handle_sqs_results({:ok, term()} | {:error, term()}) :: {:ok, [message()]}
+  defp handle_sqs_results(
+         {:error,
+          {:http_error, http_status,
+           %{
+             code: code,
+             message: message,
+             request_id: request_id
+           }}}
+       ) do
+    Logger.error(
+      "SQS request HTTP error: http_status=#{http_status} code=#{code}, message=#{message}, request_id=#{request_id}"
+    )
+
+    {:ok, []}
+  end
+
+  defp handle_sqs_results({:error, error}) do
+    Logger.error("SQS request error: error=#{inspect(error)}")
+    {:ok, []}
+  end
+
+  defp handle_sqs_results({:ok, %{body: %{messages: []}, status_code: 200}}) do
+    Logger.info("Received SQS messages, count=0")
+    {:ok, []}
+  end
+
+  defp handle_sqs_results({:ok, %{body: %{messages: messages}, status_code: 200}})
+       when is_list(messages) do
+    Logger.info("Received SQS messages, count=#{length(messages)}")
+    {:ok, Enum.map(messages, &parse_message/1)}
+  end
+
+  @spec parse_message(map()) :: map()
+  defp parse_message(%{body: body, receipt_handle: receipt_handle}) do
+    user_update =
+      body
+      |> Poison.decode!()
+      |> parse_user_update()
+
+    %{
+      user_update: user_update,
+      receipt_handle: receipt_handle
+    }
+  end
+
+  @spec parse_user_update(map()) :: user_update()
+  defp parse_user_update(%{"mbtaUuid" => user_id, "updates" => updates}) do
+    %{
+      user_id: user_id,
+      updates: %{
+        email: Map.get(updates, "email"),
+        phone_number: updates |> Map.get("phone_number") |> PhoneNumber.strip_us_country_code()
+      }
+    }
+  end
+
+  @spec update_user_record(user_update()) :: :ok | :error
+  # Ignore empty updates. The user changed a property we aren't interested in saving locally.
+  defp update_user_record(%{updates: updates}) when updates == %{}, do: :ok
+
+  defp update_user_record(%{user_id: user_id, updates: updates}) do
+    case User.get(user_id) do
+      nil ->
+        # Ignore updates for a user that has not yet logged in to T-Alerts
+        :ok
+
+      user ->
+        updates =
+          Map.filter(
+            %{
+              "email" => Map.get(updates, :email),
+              "phone_number" => Map.get(updates, :phone_number)
+            },
+            fn {_key, val} -> !is_nil(val) end
+          )
+
+        case User.update_account(user, updates, user) do
+          {:ok, _updated_user} ->
+            Logger.info("Updated user record user_id=#{user_id}")
+            :ok
+
+          {:error, changeset} ->
+            Logger.error(
+              "Unable to update user record user_id=#{user_id} errors: #{inspect(changeset.errors)}"
+            )
+
+            :error
+        end
+    end
+  end
+
+  @spec sqs_aws_region :: String.t()
+  defp sqs_aws_region, do: System.get_env("SQS_AWS_REGION", "us-east-1")
+end

apps/alert_processor/mix.exs

@@ -48,6 +48,7 @@ defmodule AlertProcessor.Mixfile do
       {:ecto_sql, "~> 3.0"},
       {:ex_aws, "~> 2.2.0"},
       {:ex_aws_sns, "~> 2.3.1"},
+      {:ex_aws_sqs, "~> 3.4"},
       {:fast_local_datetime, "~> 1.0.0"},
       {:gettext, "~> 0.11"},
       {:httpoison, "~> 1.8.0"},

apps/alert_processor/test/alert_processor/helpers/phone_number.exs

@@ -0,0 +1,19 @@
+defmodule AlertProcessor.Helpers.PhoneNumberTest do
+  use ExUnit.Case, async: true
+
+  alias AlertProcessor.Helpers.PhoneNumber
+
+  describe "strip_us_country_code/1" do
+    test "strips the US country code from the beginning of a phone number if present" do
+      assert PhoneNumber.strip_us_country_code("+15555555555") == "5555555555"
+    end
+
+    test "doesn't alter a phone number without a US country code" do
+      assert PhoneNumber.strip_us_country_code("5555555555") == "5555555555"
+    end
+
+    test "ignores nil values" do
+      assert PhoneNumber.strip_us_country_code(nil) == nil
+    end
+  end
+end