[#106] Verbose logging and updated_at column for push notifications

Changelog: enhancement

Author: pfuto21454

services/notifications/entity/src/push_token_model.rs

@@ -15,6 +15,7 @@ pub struct Model {
     pub token: String,
 
     pub created_at: TimeDateTime,
+    pub updated_at: TimeDateTime,
 }
 
 impl ActiveModelBehavior for ActiveModel {}

services/notifications/migration/src/lib.rs

@@ -1,14 +1,16 @@
 pub use sea_orm_migration::prelude::*;
 
 mod m20260504_000001_create_push_token_table;
+mod m20260616_000001_add_updated_at_to_push_token;
 
 pub struct Migrator;
 
 #[async_trait::async_trait]
 impl MigratorTrait for Migrator {
     fn migrations() -> Vec<Box<dyn MigrationTrait>> {
-        vec![Box::new(
-            m20260504_000001_create_push_token_table::Migration,
-        )]
+        vec![
+            Box::new(m20260504_000001_create_push_token_table::Migration),
+            Box::new(m20260616_000001_add_updated_at_to_push_token::Migration),
+        ]
     }
 }

services/notifications/migration/src/m20260616_000001_add_updated_at_to_push_token.rs

@@ -0,0 +1,68 @@
+use notifications_entity::push_token_model as push_token;
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        if manager.has_column("push_token", "updated_at").await? {
+            return Ok(());
+        }
+
+        // Add the column nullable first so the existing rows are accepted.
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(push_token::Entity)
+                    .add_column(
+                        ColumnDef::new(push_token::Column::UpdatedAt)
+                            .date_time()
+                            .null(),
+                    )
+                    .to_owned(),
+            )
+            .await?;
+
+        // Backfill: a token's last-updated time starts as its creation time.
+        let conn = manager.get_connection();
+        let backfill = Query::update()
+            .table(push_token::Entity)
+            .value(
+                push_token::Column::UpdatedAt,
+                Expr::col(push_token::Column::CreatedAt),
+            )
+            .to_owned();
+        conn.execute(&backfill).await?;
+
+        // Now enforce NOT NULL to match the entity definition.
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(push_token::Entity)
+                    .modify_column(
+                        ColumnDef::new(push_token::Column::UpdatedAt)
+                            .date_time()
+                            .not_null(),
+                    )
+                    .to_owned(),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(push_token::Entity)
+                    .drop_column(push_token::Column::UpdatedAt)
+                    .to_owned(),
+            )
+            .await?;
+
+        Ok(())
+    }
+}

services/notifications/src/manager.rs

@@ -1,6 +1,6 @@
 use std::{error::Error, fmt};
 
-use log::warn;
+use log::{debug, warn};
 use sea_orm::{DbConn, DbErr, EnumIter};
 
 use super::repository as token_repository;
@@ -202,18 +202,27 @@ impl NotificationManager {
 
         let rich_content = Self::avatar_rich_content(ctx, profile.avatar).await;
 
-        self.send_to_identity(
-            ctx,
-            &reply_recipient,
-            NotificationData {
-                collapse_id: collapse_id.to_owned(),
-                title,
-                body,
-                data,
-                rich_content,
-            },
-        )
-        .await?;
+        debug!("Firing reply notification: from={author} to={reply_recipient} key={key:?}");
+
+        let response = self
+            .send_to_identity(
+                ctx,
+                &reply_recipient,
+                NotificationData {
+                    collapse_id: collapse_id.to_owned(),
+                    title,
+                    body,
+                    data,
+                    rich_content,
+                },
+            )
+            .await?;
+
+        if let Some(errors) = Self::ticket_errors(&response) {
+            warn!(
+                "Reply notification errors: from={author} to={reply_recipient} key={key:?} errors=[{errors}]"
+            );
+        }
 
         Ok(())
     }
@@ -247,18 +256,31 @@ impl NotificationManager {
 
         let rich_content = Self::avatar_rich_content(ctx, profile.avatar).await;
 
-        self.send_to_identity(
-            ctx,
-            &follow.identity,
-            NotificationData {
-                collapse_id: collapse_id.to_owned(),
-                title,
-                body: "Followed you".to_string(),
-                data,
-                rich_content,
-            },
-        )
-        .await?;
+        debug!(
+            "Firing follow notification: from={author} to={} key={key:?}",
+            follow.identity
+        );
+
+        let response = self
+            .send_to_identity(
+                ctx,
+                &follow.identity,
+                NotificationData {
+                    collapse_id: collapse_id.to_owned(),
+                    title,
+                    body: "Followed you".to_string(),
+                    data,
+                    rich_content,
+                },
+            )
+            .await?;
+
+        if let Some(errors) = Self::ticket_errors(&response) {
+            warn!(
+                "Follow notification errors: from={author} to={} key={key:?} errors=[{errors}]",
+                follow.identity
+            );
+        }
 
         Ok(())
     }
@@ -310,6 +332,29 @@ impl NotificationManager {
         format!("{cdn_url}/blob/{}_{}", digest.r#type, hex)
     }
 
+    /// The error details of any failed tickets in a push response,
+    /// comma-separated, or `None` when every ticket succeeded. Lets callers
+    /// log failures while staying silent on success.
+    fn ticket_errors(response: &ExpoPushResponse) -> Option<String> {
+        let errors: Vec<&str> = response
+            .data
+            .iter()
+            .filter(|ticket| ticket.status == "error")
+            .map(|ticket| {
+                ticket
+                    .details
+                    .as_ref()
+                    .and_then(|details| details.error.as_deref())
+                    .unwrap_or("unknown error")
+            })
+            // DeviceNotRegistered is already handled by
+            // clean_unregistered_push_tokens
+            .filter(|error| *error != "DeviceNotRegistered")
+            .collect();
+
+        (!errors.is_empty()).then(|| errors.join(","))
+    }
+
     /// Sends a push notification to every authorized key of an identity that
     /// has a registered token. Tokens reported as invalid by the push service
     /// are unregistered as part of the send.
@@ -318,7 +363,7 @@ impl NotificationManager {
         ctx: &Context,
         identity: &str,
         notification: NotificationData,
-    ) -> Result<(), NotificationError> {
+    ) -> Result<ExpoPushResponse, NotificationError> {
         let authorized_keys = ctx.polycentric.authorized_keys(identity).await;
 
         let mut rows = vec![];
@@ -341,7 +386,7 @@ impl NotificationManager {
         }
 
         if expo_tokens.is_empty() {
-            return Ok(());
+            return Ok(ExpoPushResponse { data: vec![] });
         }
 
         let expo_tokens_raw: Vec<String> = expo_tokens.iter().map(|item| item.1.clone()).collect();
@@ -360,17 +405,17 @@ impl NotificationManager {
             .post_requests(vec![expo_push_request])
             .await?;
 
-        self.clean_unregistered_push_tokens(ctx, response, expo_tokens)
+        self.clean_unregistered_push_tokens(ctx, &response, expo_tokens)
             .await?;
 
-        Ok(())
+        Ok(response)
     }
 
     /// Removes any tokens from a given batch which are no longer registered from the database
     async fn clean_unregistered_push_tokens(
         &self,
         ctx: &Context,
-        response: ExpoPushResponse,
+        response: &ExpoPushResponse,
         expo_tokens: Vec<(PublicKey, String)>,
     ) -> Result<(), NotificationError> {
         if response.data.len() != expo_tokens.len() {
@@ -639,6 +684,7 @@ mod tests {
             service: PushService::Expo.as_ref().to_string(),
             token: token.to_string(),
             created_at: time::PrimitiveDateTime::MIN,
+            updated_at: time::PrimitiveDateTime::MIN,
         }
     }
 

services/notifications/src/repository.rs

@@ -28,14 +28,15 @@ impl Mutation {
         token: String,
     ) -> Result<(), DbErr> {
         let now = time::OffsetDateTime::now_utc();
-        let created_at = time::PrimitiveDateTime::new(now.date(), now.time());
+        let timestamp = time::PrimitiveDateTime::new(now.date(), now.time());
 
         let active = PushTokenModel::ActiveModel {
             public_key_type: Set(public_key.key_type as i16),
             public_key: Set(public_key.key.clone()),
             service: Set(service),
             token: Set(token),
-            created_at: Set(created_at),
+            created_at: Set(timestamp),
+            updated_at: Set(timestamp),
         };
 
         PushTokenModel::Entity::insert(active)
@@ -47,6 +48,7 @@ impl Mutation {
                 .update_columns([
                     PushTokenModel::Column::Service,
                     PushTokenModel::Column::Token,
+                    PushTokenModel::Column::UpdatedAt,
                 ])
                 .to_owned(),
             )

services/notifications/src/rpc.rs

@@ -144,6 +144,7 @@ mod tests {
                 service: PushService::Expo.as_ref().to_string(),
                 token: "ExponentPushToken[abc123]".to_string(),
                 created_at: synced_at,
+                updated_at: synced_at,
             }]])
             .into_connection();