chore: migrate authifier into codebase (#658)

Co-authored-by: izzy <me@insrt.uk>
Signed-off-by: Zomatree <me@zomatree.live>
Signed-off-by: izzy <me@insrt.uk>

crates/daemons/crond/Cargo.toml

@@ -15,6 +15,7 @@ log = { workspace = true }
 
 # Async
 tokio = { workspace = true }
+futures = { workspace = true }
 
 # Redis
 redis-kiss = { workspace = true }

crates/daemons/crond/src/main.rs

@@ -1,23 +1,51 @@
-use revolt_config::configure;
-use revolt_database::{DatabaseInfo, AMQP};
+use std::{future::Future, panic::AssertUnwindSafe, time::Duration};
+
+use futures::FutureExt;
+use revolt_config::{capture_error, configure};
+use revolt_database::{Database, DatabaseInfo, AMQP};
 use revolt_result::Result;
-use tasks::{acks, file_deletion, prune_dangling_files, prune_members};
-use tokio::try_join;
+use tasks::*;
+use tokio::{join, time::sleep};
 
 pub mod tasks;
 
+pub async fn cron_task_wrapper<Fut: Future<Output = Result<()>>>(
+    func: fn(Database, AMQP) -> Fut,
+    db: Database,
+    amqp: AMQP,
+) {
+    loop {
+        let wrapper = AssertUnwindSafe(func(db.clone(), amqp.clone()));
+
+        match wrapper.catch_unwind().await {
+            Ok(Ok(())) => {
+                log::error!("cron unexpectedly finshed, Retrying after 60s");
+            }
+            Ok(Err(error)) => {
+                log::error!("cron task failed unexpectedly: {error:?}\nRetrying after 60s");
+                capture_error(&error);
+            }
+            _ => {
+                log::error!("cron task failed unexpectedly\nRetrying after 60s");
+            }
+        }
+
+        sleep(Duration::from_secs(60)).await;
+    }
+}
+
 #[tokio::main]
-async fn main() -> Result<()> {
+async fn main() {
     configure!(crond);
 
     let db = DatabaseInfo::Auto.connect().await.expect("database");
     let amqp = AMQP::new_auto().await;
 
-    try_join!(
-        file_deletion::task(db.clone()),
-        prune_dangling_files::task(db.clone()),
-        prune_members::task(db.clone()),
-        acks::task(db.clone(), amqp.clone()),
-    )
-    .map(|_| ())
+    join!(
+        cron_task_wrapper(file_deletion::task, db.clone(), amqp.clone()),
+        cron_task_wrapper(prune_dangling_files::task, db.clone(), amqp.clone()),
+        cron_task_wrapper(prune_members::task, db.clone(), amqp.clone()),
+        cron_task_wrapper(delete_accounts::task, db.clone(), amqp.clone()),
+        cron_task_wrapper(acks::task, db.clone(), amqp.clone()),
+    );
 }

crates/daemons/crond/src/tasks/delete_accounts.rs

@@ -0,0 +1,23 @@
+use std::time::Duration;
+
+use revolt_database::Database;
+use revolt_result::Result;
+use tokio::time::sleep;
+
+pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
+    loop {
+        let accounts = db.fetch_accounts_due_for_deletion().await?;
+        let count = accounts.len();
+
+        for mut account in accounts {
+            let mut user = db.fetch_user(&account.id).await?;
+
+            user.delete(&db).await?;
+            account.mark_deleted(&db).await?;
+        }
+
+        log::info!("Deleted {count} accounts.");
+
+        sleep(Duration::from_hours(1)).await
+    }
+}

crates/daemons/crond/src/tasks/file_deletion.rs

@@ -6,21 +6,17 @@ use revolt_files::delete_from_s3;
 use revolt_result::Result;
 use tokio::time::sleep;
 
-pub async fn task(db: Database) -> Result<()> {
+pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
     loop {
         let files = db.fetch_deleted_attachments().await?;
 
         for file in files {
             if let Some(hash) = &file.hash {
-                let count = db
-                    .count_file_hash_references(hash)
-                    .await?;
+                let count = db.count_file_hash_references(hash).await?;
 
                 // No other files reference this file on disk anymore
                 if count <= 1 {
-                    let file_hash = db
-                        .fetch_attachment_hash(hash)
-                        .await?;
+                    let file_hash = db.fetch_attachment_hash(hash).await?;
 
                     // Delete from S3
                     delete_from_s3(&file_hash.bucket_id, &file_hash.path).await?;

crates/daemons/crond/src/tasks/mod.rs

@@ -1,4 +1,6 @@
+pub mod delete_accounts;
 pub mod acks;
 pub mod file_deletion;
 pub mod prune_dangling_files;
 pub mod prune_members;
+pub mod prune_mfa_tickets;

crates/daemons/crond/src/tasks/prune_dangling_files.rs

@@ -6,7 +6,7 @@ use tokio::time::sleep;
 
 use log::info;
 
-pub async fn task(db: Database) -> Result<()> {
+pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
     loop {
         // This could just be a single database query
         // ... but timestamps are inconsistently serialised

crates/daemons/crond/src/tasks/prune_members.rs

@@ -5,7 +5,7 @@ use revolt_database::Database;
 use revolt_result::Result;
 use tokio::time::sleep;
 
-pub async fn task(db: Database) -> Result<()> {
+pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
     loop {
         let success = db.remove_dangling_members().await;
         if let Err(s) = success {

crates/daemons/crond/src/tasks/prune_mfa_tickets.rs

@@ -0,0 +1,14 @@
+use std::time::Duration;
+
+use revolt_database::Database;
+use revolt_result::Result;
+use tokio::time::sleep;
+
+pub async fn task(db: Database) -> Result<()> {
+    loop {
+        let count = db.delete_expired_tickets().await?;
+        log::info!("Pruned {count} expired MFA tickets");
+
+        sleep(Duration::from_mins(5)).await
+    }
+}

crates/daemons/pushd/Cargo.toml

@@ -25,8 +25,6 @@ tokio = { workspace = true }
 async-trait = { workspace = true }
 ulid = { workspace = true }
 
-authifier = { workspace = true }
-
 log = { workspace = true }
 pretty_env_logger = { workspace = true }
 

crates/daemons/pushd/src/consumers/inbound/ack.rs

@@ -10,7 +10,6 @@ use revolt_database::{events::rabbit::*, Database};
 #[allow(unused)]
 pub struct AckConsumer {
     db: Database,
-    authifier_db: authifier::Database,
     connection: Arc<Connection>,
     channel: Arc<Channel>,
 }
@@ -19,13 +18,11 @@ pub struct AckConsumer {
 impl Consumer for AckConsumer {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<Channel>,
     ) -> Self {
         Self {
             db,
-            authifier_db,
             connection,
             channel,
         }
@@ -58,7 +55,7 @@ impl Consumer for AckConsumer {
             return Ok(());
         };
 
-        if let Ok(sessions) = self.authifier_db.find_sessions(&payload.user_id).await {
+        if let Ok(sessions) = self.db.fetch_sessions(&payload.user_id).await {
             let config = revolt_config::config().await;
             // Step 2: find any apple sessions, since we don't need to calculate this for anything else.
             // If there's no apple sessions, we can return early

crates/daemons/pushd/src/consumers/inbound/dm_call.rs

@@ -11,7 +11,6 @@ use revolt_database::{events::rabbit::*, Database};
 #[allow(unused)]
 pub struct DmCallConsumer {
     db: Database,
-    authifier_db: authifier::Database,
     connection: Arc<Connection>,
     channel: Arc<Channel>,
 }
@@ -20,13 +19,11 @@ pub struct DmCallConsumer {
 impl Consumer for DmCallConsumer {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<Channel>,
     ) -> Self {
         Self {
             db,
-            authifier_db,
             connection,
             channel,
         }
@@ -70,7 +67,7 @@ impl Consumer for DmCallConsumer {
         let config = revolt_config::config().await;
 
         for user_id in call_recipients {
-            if let Ok(sessions) = self.authifier_db.find_sessions(&user_id).await {
+            if let Ok(sessions) = self.db.fetch_sessions(&user_id).await {
                 for session in sessions {
                     if let Some(sub) = session.subscription {
                         let mut sendable = PayloadToService {

crates/daemons/pushd/src/consumers/inbound/fr_accepted.rs

@@ -11,7 +11,6 @@ use revolt_database::{events::rabbit::*, Database};
 #[allow(unused)]
 pub struct FRAcceptedConsumer {
     db: Database,
-    authifier_db: authifier::Database,
     connection: Arc<Connection>,
     channel: Arc<Channel>,
 }
@@ -20,13 +19,11 @@ pub struct FRAcceptedConsumer {
 impl Consumer for FRAcceptedConsumer {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<Channel>,
     ) -> Self {
         Self {
             db,
-            authifier_db,
             connection,
             channel,
         }
@@ -42,7 +39,7 @@ impl Consumer for FRAcceptedConsumer {
 
         debug!("Received FR accept event");
 
-        if let Ok(sessions) = self.authifier_db.find_sessions(&payload.user).await {
+        if let Ok(sessions) = self.db.fetch_sessions(&payload.user).await {
             let config = revolt_config::config().await;
             for session in sessions {
                 if let Some(sub) = session.subscription {

crates/daemons/pushd/src/consumers/inbound/fr_received.rs

@@ -11,7 +11,6 @@ use revolt_database::{events::rabbit::*, Database};
 #[allow(unused)]
 pub struct FRReceivedConsumer {
     db: Database,
-    authifier_db: authifier::Database,
     connection: Arc<Connection>,
     channel: Arc<Channel>,
 }
@@ -20,13 +19,11 @@ pub struct FRReceivedConsumer {
 impl Consumer for FRReceivedConsumer {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<Channel>,
     ) -> Self {
         Self {
             db,
-            authifier_db,
             connection,
             channel,
         }
@@ -42,7 +39,7 @@ impl Consumer for FRReceivedConsumer {
 
         debug!("Received FR received event");
 
-        if let Ok(sessions) = self.authifier_db.find_sessions(&payload.user).await {
+        if let Ok(sessions) = self.db.fetch_sessions(&payload.user).await {
             let config = revolt_config::config().await;
             for session in sessions {
                 if let Some(sub) = session.subscription {

crates/daemons/pushd/src/consumers/inbound/generic.rs

@@ -11,7 +11,6 @@ use revolt_database::{events::rabbit::*, Database};
 #[allow(unused)]
 pub struct GenericConsumer {
     db: Database,
-    authifier_db: authifier::Database,
     connection: Arc<Connection>,
     channel: Arc<Channel>,
 }
@@ -20,13 +19,11 @@ pub struct GenericConsumer {
 impl Consumer for GenericConsumer {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<Channel>,
     ) -> Self {
         Self {
             db,
-            authifier_db,
             connection,
             channel,
         }
@@ -43,8 +40,8 @@ impl Consumer for GenericConsumer {
         debug!("Received message event on origin");
 
         if let Ok(sessions) = self
-            .authifier_db
-            .find_sessions_with_subscription(&payload.users)
+            .db
+            .fetch_sessions_with_subscription(&payload.users)
             .await
         {
             let config = revolt_config::config().await;

crates/daemons/pushd/src/consumers/inbound/mass_mention.rs

@@ -19,7 +19,6 @@ use revolt_result::ToRevoltError;
 #[allow(unused)]
 pub struct MassMessageConsumer {
     db: Database,
-    authifier_db: authifier::Database,
     connection: Arc<Connection>,
     channel: Arc<Channel>,
 }
@@ -31,8 +30,8 @@ impl MassMessageConsumer {
         users: &[String],
     ) -> Result<()> {
         if let Ok(sessions) = self
-            .authifier_db
-            .find_sessions_with_subscription(users)
+            .db
+            .fetch_sessions_with_subscription(users)
             .await
         {
             let config = revolt_config::config().await;
@@ -75,13 +74,11 @@ impl MassMessageConsumer {
 impl Consumer for MassMessageConsumer {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<Channel>,
     ) -> Self {
         Self {
             db,
-            authifier_db,
             connection,
             channel,
         }

crates/daemons/pushd/src/consumers/inbound/message.rs

@@ -11,7 +11,6 @@ use revolt_database::{events::rabbit::*, Database};
 #[allow(unused)]
 pub struct MessageConsumer {
     db: Database,
-    authifier_db: authifier::Database,
     connection: Arc<Connection>,
     channel: Arc<Channel>,
 }
@@ -20,13 +19,11 @@ pub struct MessageConsumer {
 impl Consumer for MessageConsumer {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<Channel>,
     ) -> Self {
         Self {
             db,
-            authifier_db,
             connection,
             channel,
         }
@@ -48,8 +45,8 @@ impl Consumer for MessageConsumer {
         debug!("Received message event on origin");
 
         if let Ok(sessions) = self
-            .authifier_db
-            .find_sessions_with_subscription(&payload.users)
+            .db
+            .fetch_sessions_with_subscription(&payload.users)
             .await
         {
             let config = revolt_config::config().await;

crates/daemons/pushd/src/consumers/outbound/apn.rs

@@ -80,7 +80,6 @@ impl<'a> PayloadLike for CallStartStopPayload<'a> {
 #[allow(unused)]
 pub struct ApnsOutboundConsumer {
     db: Database,
-    authifier_db: authifier::Database,
     connection: Arc<Connection>,
     channel: Arc<AMQPChannel>,
     client: Client,
@@ -126,7 +125,6 @@ impl ApnsOutboundConsumer {
 impl Consumer for ApnsOutboundConsumer {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<AMQPChannel>,
     ) -> Self {
@@ -161,7 +159,6 @@ impl Consumer for ApnsOutboundConsumer {
 
         Self {
             db,
-            authifier_db,
             connection,
             channel,
             client,

crates/daemons/pushd/src/consumers/outbound/fcm.rs

@@ -119,7 +119,6 @@ impl NotificationData {
 #[allow(unused)]
 pub struct FcmOutboundConsumer {
     db: Database,
-    authifier_db: authifier::Database,
     connection: Arc<Connection>,
     channel: Arc<AMQPChannel>,
     client: Client,
@@ -129,7 +128,6 @@ pub struct FcmOutboundConsumer {
 impl Consumer for FcmOutboundConsumer {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<AMQPChannel>,
     ) -> Self {
@@ -137,7 +135,6 @@ impl Consumer for FcmOutboundConsumer {
 
         Self {
             db,
-            authifier_db,
             connection,
             channel,
             client: Client::new(

crates/daemons/pushd/src/consumers/outbound/vapid.rs

@@ -19,7 +19,6 @@ use web_push::{
 #[allow(unused)]
 pub struct VapidOutboundConsumer {
     db: Database,
-    authifier_db: authifier::Database,
     connection: Arc<Connection>,
     channel: Arc<AMQPChannel>,
     client: IsahcWebPushClient,
@@ -30,7 +29,6 @@ pub struct VapidOutboundConsumer {
 impl Consumer for VapidOutboundConsumer {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<AMQPChannel>,
     ) -> Self {
@@ -48,7 +46,6 @@ impl Consumer for VapidOutboundConsumer {
 
         Self {
             db,
-            authifier_db,
             connection,
             channel,
             client: IsahcWebPushClient::new().unwrap(),

crates/daemons/pushd/src/main.rs

@@ -32,17 +32,6 @@ async fn main() {
 
     // Setup database
     let db = revolt_database::DatabaseInfo::Auto.connect().await.unwrap();
-    let authifier: authifier::Database;
-
-    if let Some(client) = match &db {
-        revolt_database::Database::Reference(_) => None,
-        revolt_database::Database::MongoDb(mongo) => Some(mongo),
-    } {
-        authifier =
-            authifier::Database::MongoDb(authifier::database::MongoDb(client.database("revolt")));
-    } else {
-        panic!("Mongo is not in use, can't connect via authifier!")
-    }
 
     let config = config().await;
 
@@ -75,7 +64,6 @@ async fn main() {
     channels.push(
         make_queue_and_consume::<GenericConsumer>(
             &db,
-            &authifier,
             &connection,
             &config,
             &config.pushd.generic_queue,
@@ -88,7 +76,6 @@ async fn main() {
     channels.push(
         make_queue_and_consume::<MessageConsumer>(
             &db,
-            &authifier,
             &connection,
             &config,
             &config.pushd.message_queue,
@@ -101,7 +88,6 @@ async fn main() {
     channels.push(
         make_queue_and_consume::<FRReceivedConsumer>(
             &db,
-            &authifier,
             &connection,
             &config,
             &config.pushd.fr_received_queue,
@@ -114,7 +100,6 @@ async fn main() {
     channels.push(
         make_queue_and_consume::<FRAcceptedConsumer>(
             &db,
-            &authifier,
             &connection,
             &config,
             &config.pushd.fr_accepted_queue,
@@ -127,7 +112,6 @@ async fn main() {
     channels.push(
         make_queue_and_consume::<MassMessageConsumer>(
             &db,
-            &authifier,
             &connection,
             &config,
             &config.pushd.mass_mention_queue,
@@ -140,7 +124,6 @@ async fn main() {
     channels.push(
         make_queue_and_consume::<DmCallConsumer>(
             &db,
-            &authifier,
             &connection,
             &config,
             &config.pushd.dm_call_queue,
@@ -154,7 +137,6 @@ async fn main() {
         channels.push(
             make_queue_and_consume::<ApnsOutboundConsumer>(
                 &db,
-                &authifier,
                 &connection,
                 &config,
                 &config.pushd.apn.queue,
@@ -170,7 +152,6 @@ async fn main() {
         channels.push(
             make_queue_and_consume::<AckConsumer>(
                 &db,
-                &authifier,
                 &connection,
                 &config,
                 &config.pushd.ack_queue,
@@ -185,7 +166,6 @@ async fn main() {
         channels.push(
             make_queue_and_consume::<FcmOutboundConsumer>(
                 &db,
-                &authifier,
                 &connection,
                 &config,
                 &config.pushd.fcm.queue,
@@ -200,7 +180,6 @@ async fn main() {
         channels.push(
             make_queue_and_consume::<VapidOutboundConsumer>(
                 &db,
-                &authifier,
                 &connection,
                 &config,
                 &config.pushd.vapid.queue,
@@ -220,7 +199,6 @@ async fn main() {
 
 async fn make_queue_and_consume<F>(
     db: &Database,
-    authifier_db: &authifier::Database,
     connection: &Arc<Connection>,
     config: &Settings,
     queue_name: &str,
@@ -300,7 +278,6 @@ where
     let delegate = Delegate(
         F::create(
             db.clone(),
-            authifier_db.clone(),
             connection.clone(),
             channel.clone(),
         )

crates/daemons/pushd/src/utils/consumer.rs

@@ -18,7 +18,6 @@ use revolt_database::Database;
 pub trait Consumer: Clone + Send + Sync + 'static {
     async fn create(
         db: Database,
-        authifier_db: authifier::Database,
         connection: Arc<Connection>,
         channel: Arc<Channel>,
     ) -> Self;