feat(notifications): implement email reminder use case (#4216)

* chore(notifications): boilerplate for linking email use case

* chore(notifications): more boilerplate

* chore(notifications): support list_after_id in UserNotificationsSettings

* refactor(notifications): use GaloyUserId::search_begin()

* refactor(notifications): expose created_at for UserNotificationSettings

* chore(notifications): add EmailReminderProjection

* chore(notifications): remove user_transaction_tracker

* feat(notifictions): link email reminders

* refactor(notifications): re-use handle_single_user_event

* refactor(notifications): use kickoff_link_email_reminder to start reminders

* refactor(notifications): more efficient notification update

* refactor(notifications): add email_reminder_projection config

* chore(notifications): link email reminder under security category

* chore(notifications): use minutes in email reminder projection config

* chore(notifications): update schemas

---------

Co-authored-by: bodymindarts <justin@galoy.io>
Co-authored-by: Sam Peters <unclesamtoshi@protonmail.com>

core/notifications/locales/en.yml

@@ -60,3 +60,6 @@ transaction.paid_invoice.body_display_currency: "+%{displayCurrencyAmount} | %{f
 
 price_changed.title: "Bitcoin is on the move!"
 price_changed.body: "Bitcoin is up %{percent_increase}% in the last day to %{price}!"
+
+security.link_email_reminder.title: "Link Email to Secure Account"
+security.link_email_reminder.body: "Link your email to secure your account and receive important updates"

core/notifications/migrations/20240321091909_create_email_reminder_projection.down.sql

@@ -0,0 +1 @@
+-- Add down migration script here

core/notifications/migrations/20240321091909_create_email_reminder_projection.up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE email_reminder_projection (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    galoy_user_id VARCHAR UNIQUE NOT NULL,
+    user_first_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    last_transaction_at TIMESTAMPTZ,
+    last_notified_at TIMESTAMPTZ
+);

core/notifications/notifications.yml

@@ -6,6 +6,12 @@
 # grpc_server:
 #   port: 6685
 app:
+  # jobs:
+  #   link_email_reminder_delay: 5
+  # email_reminder_projection:
+  #   account_liveness_threshold_minutes: 30240
+  #   account_aged_threshold_minutes: 30240
+  #   notification_cool_off_threshold_minutes: 129600
   push_executor:
     fcm:
       google_application_credentials_path: "./config/notifications/fake_service_account.json"

core/notifications/src/app/config.rs

@@ -1,10 +1,17 @@
 use serde::{Deserialize, Serialize};
 
-use crate::{email_executor::EmailExecutorConfig, push_executor::PushExecutorConfig};
+use crate::{
+    email_executor::EmailExecutorConfig, email_reminder_projection::EmailReminderProjectionConfig,
+    job::JobsConfig, push_executor::PushExecutorConfig,
+};
 
 #[derive(Clone, Default, Serialize, Deserialize)]
 pub struct AppConfig {
     pub push_executor: PushExecutorConfig,
     #[serde(default)]
     pub email_executor: EmailExecutorConfig,
+    #[serde(default)]
+    pub jobs: JobsConfig,
+    #[serde(default)]
+    pub email_reminder_projection: EmailReminderProjectionConfig,
 }

core/notifications/src/app/error.rs

@@ -1,7 +1,8 @@
 use thiserror::Error;
 
 use crate::{
-    email_executor::error::EmailExecutorError, job::error::JobError,
+    email_executor::error::EmailExecutorError,
+    email_reminder_projection::error::EmailReminderProjectionError, job::error::JobError,
     notification_cool_off_tracker::NotificationCoolOffTrackerError,
     push_executor::error::PushExecutorError,
     user_notification_settings::error::UserNotificationSettingsError,
@@ -14,6 +15,8 @@ pub enum ApplicationError {
     #[error("{0}")]
     UserNotificationSettingsError(#[from] UserNotificationSettingsError),
     #[error("{0}")]
+    EmailReminderProjectionError(#[from] EmailReminderProjectionError),
+    #[error("{0}")]
     JobError(#[from] JobError),
     #[error("{0}")]
     PushExecutorError(#[from] PushExecutorError),

core/notifications/src/app/mod.rs

@@ -9,6 +9,7 @@ use std::sync::Arc;
 
 use crate::{
     email_executor::EmailExecutor,
+    email_reminder_projection::EmailReminderProjection,
     job::{self},
     notification_cool_off_tracker::*,
     notification_event::*,
@@ -24,6 +25,7 @@ use error::*;
 pub struct NotificationsApp {
     _config: AppConfig,
     settings: UserNotificationSettingsRepo,
+    email_reminder_projection: EmailReminderProjection,
     pool: Pool<Postgres>,
     _runner: Arc<JobRunnerHandle>,
 }
@@ -34,12 +36,26 @@ impl NotificationsApp {
         let push_executor =
             PushExecutor::init(config.push_executor.clone(), settings.clone()).await?;
         let email_executor = EmailExecutor::init(config.email_executor.clone(), settings.clone())?;
-        let runner =
-            job::start_job_runner(&pool, push_executor, email_executor, settings.clone()).await?;
+        let email_reminder_projection =
+            EmailReminderProjection::new(&pool, config.email_reminder_projection.clone());
+        let runner = job::start_job_runner(
+            &pool,
+            push_executor,
+            email_executor,
+            settings.clone(),
+            email_reminder_projection.clone(),
+        )
+        .await?;
+        Self::spawn_kickoff_link_email_reminder(
+            pool.clone(),
+            config.jobs.kickoff_link_email_remainder_delay,
+        )
+        .await?;
         Ok(Self {
             _config: config,
             pool,
             settings,
+            email_reminder_projection,
             _runner: Arc::new(runner),
         })
     }
@@ -164,7 +180,14 @@ impl NotificationsApp {
     ) -> Result<(), ApplicationError> {
         let mut user_settings = self.settings.find_for_user_id(&user_id).await?;
         user_settings.update_email_address(addr);
-        self.settings.persist(&mut user_settings).await?;
+        let mut tx = self.pool.begin().await?;
+        self.settings
+            .persist_in_tx(&mut tx, &mut user_settings)
+            .await?;
+        self.email_reminder_projection
+            .user_added_email(&mut tx, user_id)
+            .await?;
+        tx.commit().await?;
         Ok(())
     }
 
@@ -195,6 +218,22 @@ impl NotificationsApp {
         Ok(())
     }
 
+    #[instrument(name = "app.handle_transaction_occurred_event", skip(self), err)]
+    pub async fn handle_transaction_occurred_event(
+        &self,
+        user_id: GaloyUserId,
+        transaction_occurred: TransactionOccurred,
+    ) -> Result<(), ApplicationError> {
+        let user_settings = self.settings.find_for_user_id(&user_id).await?;
+        if user_settings.email_address().is_none() {
+            self.email_reminder_projection
+                .transaction_occurred_for_user_without_email(&user_id)
+                .await?;
+        }
+        self.handle_single_user_event(user_id, transaction_occurred)
+            .await
+    }
+
     #[instrument(name = "app.handle_price_changed_event", skip(self), err)]
     pub async fn handle_price_changed_event(
         &self,
@@ -239,4 +278,27 @@ impl NotificationsApp {
         tx.commit().await?;
         Ok(())
     }
+
+    #[instrument(
+        name = "app.kickoff_link_email_reminder",
+        level = "trace",
+        skip_all,
+        err
+    )]
+    async fn spawn_kickoff_link_email_reminder(
+        pool: sqlx::PgPool,
+        delay: std::time::Duration,
+    ) -> Result<(), ApplicationError> {
+        tokio::spawn(async move {
+            loop {
+                let _ = job::spawn_kickoff_link_email_reminder(
+                    &pool,
+                    std::time::Duration::from_secs(1),
+                )
+                .await;
+                tokio::time::sleep(delay).await;
+            }
+        });
+        Ok(())
+    }
 }