fix: Drop unknown FCM users

autoendpoint/src/db/client.rs

@@ -49,6 +49,15 @@ pub trait DbClient: Send + Sync {
     /// Save a message to the message table
     async fn save_message(&self, uaid: Uuid, message: Notification) -> DbResult<()>;
 
+    /// Delete a notification
+    async fn remove_message(&self, uaid: Uuid, sort_key: String) -> DbResult<()>;
+
+    /// Check if the router table exists
+    async fn router_table_exists(&self) -> DbResult<bool>;
+
+    /// Check if the message table exists
+    async fn message_table_exists(&self) -> DbResult<bool>;
+
     /// Get the message table name
     fn message_table(&self) -> &str;
 
@@ -95,6 +104,32 @@ impl DbClientImpl {
             message_table,
         })
     }
+
+    /// Check if a table exists
+    async fn table_exists(&self, table_name: String) -> DbResult<bool> {
+        let input = DescribeTableInput { table_name };
+
+        let output = match retry_policy()
+            .retry_if(
+                || self.ddb.describe_table(input.clone()),
+                retryable_describe_table_error(self.metrics.clone()),
+            )
+            .await
+        {
+            Ok(output) => output,
+            Err(RusotoError::Service(DescribeTableError::ResourceNotFound(_))) => {
+                return Ok(false);
+            }
+            Err(e) => return Err(e.into()),
+        };
+
+        let status = output
+            .table
+            .and_then(|table| table.table_status)
+            .ok_or(DbError::TableStatusUnknown)?;
+
+        Ok(["CREATING", "UPDATING", "ACTIVE"].contains(&status.as_str()))
+    }
 }
 
 #[async_trait]
@@ -256,6 +291,33 @@ impl DbClient for DbClientImpl {
         Ok(())
     }
 
+    async fn remove_message(&self, uaid: Uuid, sort_key: String) -> DbResult<()> {
+        let input = DeleteItemInput {
+            table_name: self.message_table.clone(),
+            key: ddb_item! {
+               uaid: s => uaid.to_simple().to_string(),
+               chidmessageid: s => sort_key
+            },
+            ..Default::default()
+        };
+
+        retry_policy()
+            .retry_if(
+                || self.ddb.delete_item(input.clone()),
+                retryable_delete_error(self.metrics.clone()),
+            )
+            .await?;
+        Ok(())
+    }
+
+    async fn router_table_exists(&self) -> DbResult<bool> {
+        self.table_exists(self.router_table.clone()).await
+    }
+
+    async fn message_table_exists(&self) -> DbResult<bool> {
+        self.table_exists(self.message_table.clone()).await
+    }
+
     fn message_table(&self) -> &str {
         &self.message_table
     }

autoendpoint/src/db/mock.rs

@@ -29,6 +29,12 @@ mockall::mock! {
 
         fn save_message(&self, uaid: Uuid, message: Notification) -> DbResult<()>;
 
+        fn remove_message(&self, uaid: Uuid, sort_key: String) -> DbResult<()>;
+
+        fn router_table_exists(&self) -> DbResult<bool>;
+
+        fn message_table_exists(&self) -> DbResult<bool>;
+
         fn message_table(&self) -> &str;
 
         fn box_clone(&self) -> Box<dyn DbClient>;
@@ -65,6 +71,18 @@ impl DbClient for Arc<MockDbClient> {
         Arc::as_ref(self).save_message(uaid, message)
     }
 
+    async fn remove_message(&self, uaid: Uuid, sort_key: String) -> DbResult<()> {
+        Arc::as_ref(self).remove_message(uaid, sort_key)
+    }
+
+    async fn router_table_exists(&self) -> DbResult<bool> {
+        Arc::as_ref(self).router_table_exists()
+    }
+
+    async fn message_table_exists(&self) -> DbResult<bool> {
+        Arc::as_ref(self).message_table_exists()
+    }
+
     fn message_table(&self) -> &str {
         Arc::as_ref(self).message_table()
     }

autoendpoint/src/routes/webpush.rs

@@ -28,7 +28,7 @@ pub async fn delete_notification_route(
     trace!("message_id = {:?}", message_id);
     state
         .ddb
-        .delete_message(message_id.uaid(), sort_key)
+        .remove_message(message_id.uaid(), sort_key)
         .await?;
 
     Ok(HttpResponse::NoContent().finish())