Use Postgres notify/listen for event queue

pbzweihander · Aug 16, 2023 · 4f9f47e · 4f9f47e
1 parent 183324d
commit 4f9f47e
Show file tree

Hide file tree

Showing 20 changed files with 115 additions and 273 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/backend/Cargo.toml b/backend/Cargo.toml
@@ -11,7 +11,6 @@ async-trait = "0.1.72"
 axum = { version = "0.6.20", features = ["headers"] }
 axum-extra = { version = "0.7.7", features = ["async-read-body"] }
 bcrypt = "0.15.0"
-cached = { version = "0.44.0", features = ["async", "redis_store", "redis_tokio"] }
 chrono = { version = "0.4.26", features = ["serde"] }
 derivative = "2.2.0"
 dotenvy = "0.15.7"
@@ -23,11 +22,12 @@ mime = "0.3.17"
 mime_serde_shim = "0.2.2"
 object_store = { version = "0.6.1", features = ["aws"] }
 once_cell = "1.18.0"
-redis = { version = "0.23.2", features = ["aio", "tokio-comp", "connection-manager"] }
 reqwest = { version = "0.11.18", features = ["json"] }
 sea-orm = { version = "0.12.2", features = ["sqlx-postgres", "runtime-tokio-native-tls", "macros", "sea-orm-internal"] }
 serde = { version = "1.0.182", features = ["derive"] }
 serde_json = "1.0.104"
+sqlx = { version = "0.7.1", features = ["postgres"] }
+sqlx-postgres = "0.7.1"
 stopper = "0.2.0"
 tokio = { version = "1.29.1", features = ["rt-multi-thread", "macros", "signal", "time", "sync"] }
 tower-http = { version = "0.4.3", features = ["trace"] }

diff --git a/backend/src/ap/announce.rs b/backend/src/ap/announce.rs
@@ -68,7 +68,7 @@ impl ActivityHandler for Announce {
         let event = Event::Update(Update::CreatePost {
             post_id: post.id.into(),
         });
-        event.send(&*data.db, &mut data.redis()).await?;
+        event.send(&*data.db).await?;
 
         if let Some(repost_id) = post.repost_id {
             let local_person_reposted_count = post::Entity::find_by_id(repost_id)
@@ -84,7 +84,7 @@ impl ActivityHandler for Announce {
                         .into(),
                     post_id: repost_id.into(),
                 }));
-                event.send(&*data.db, &mut data.redis()).await?;
+                event.send(&*data.db).await?;
             }
         }
 

diff --git a/backend/src/ap/delete.rs b/backend/src/ap/delete.rs
@@ -111,7 +111,7 @@ impl ActivityHandler for Delete {
             let event = Event::Update(Update::DeletePost {
                 post_id: post_id.into(),
             });
-            event.send(&*data.db, &mut data.redis()).await?;
+            event.send(&*data.db).await?;
 
             return Ok(());
         }
@@ -134,7 +134,7 @@ impl ActivityHandler for Delete {
             let event = Event::Update(Update::DeleteUser {
                 user_id: user_id.into(),
             });
-            event.send(&*data.db, &mut data.redis()).await?;
+            event.send(&*data.db).await?;
 
             return Ok(());
         }

diff --git a/backend/src/ap/flag.rs b/backend/src/ap/flag.rs
@@ -92,7 +92,7 @@ impl ActivityHandler for Flag {
         let event = Event::Notification(Notification::new(NotificationType::CreateReport {
             report_id: report.id.into(),
         }));
-        event.send(&*data.db, &mut data.redis()).await?;
+        event.send(&*data.db).await?;
 
         Ok(())
     }

diff --git a/backend/src/ap/follow.rs b/backend/src/ap/follow.rs
@@ -85,7 +85,7 @@ impl ActivityHandler for Follow {
         let event = Event::Notification(Notification::new(NotificationType::CreateFollower {
             user_id: follower.from_id.into(),
         }));
-        event.send(&*data.db, &mut data.redis()).await?;
+        event.send(&*data.db).await?;
 
         Ok(())
     }
@@ -149,7 +149,7 @@ impl ActivityHandler for FollowAccept {
         let event = Event::Notification(Notification::new(NotificationType::AcceptFollow {
             user_id: follow.to_id.into(),
         }));
-        event.send(&*data.db, &mut data.redis()).await?;
+        event.send(&*data.db).await?;
 
         Ok(())
     }
@@ -247,7 +247,7 @@ impl ActivityHandler for FollowReject {
             let event = Event::Notification(Notification::new(NotificationType::RejectFollow {
                 user_id: follow_id.into(),
             }));
-            event.send(&*data.db, &mut data.redis()).await?;
+            event.send(&*data.db).await?;
 
             Ok(())
         } else {

diff --git a/backend/src/ap/like.rs b/backend/src/ap/like.rs
@@ -83,7 +83,7 @@ impl ActivityHandler for Like {
         let event = Event::Update(Update::CreateReaction {
             post_id: reaction.post_id.into(),
         });
-        event.send(&*data.db, &mut data.redis()).await?;
+        event.send(&*data.db).await?;
 
         let local_person_reacted_count = reaction
             .find_related(post::Entity)
@@ -96,7 +96,7 @@ impl ActivityHandler for Like {
                 post_id: reaction.post_id.into(),
                 reaction_id: reaction.id.into(),
             }));
-            event.send(&*data.db, &mut data.redis()).await?;
+            event.send(&*data.db).await?;
         }
 
         Ok(())

diff --git a/backend/src/ap/note.rs b/backend/src/ap/note.rs
@@ -129,7 +129,7 @@ impl ActivityHandler for CreateNote {
         let event = Event::Update(Update::CreatePost {
             post_id: post.id.into(),
         });
-        event.send(&*data.db, &mut data.redis()).await?;
+        event.send(&*data.db).await?;
 
         let local_person_mentioned_count = post
             .find_related(mention::Entity)
@@ -141,7 +141,7 @@ impl ActivityHandler for CreateNote {
             let event = Event::Notification(Notification::new(NotificationType::Mentioned {
                 post_id: post.id.into(),
             }));
-            event.send(&*data.db, &mut data.redis()).await?;
+            event.send(&*data.db).await?;
         }
 
         if let Some(repost_id) = post.repost_id {
@@ -154,7 +154,7 @@ impl ActivityHandler for CreateNote {
                 let event = Event::Notification(Notification::new(NotificationType::Quoted {
                     post_id: post.id.into(),
                 }));
-                event.send(&*data.db, &mut data.redis()).await?;
+                event.send(&*data.db).await?;
             }
         }
 

diff --git a/backend/src/ap/person.rs b/backend/src/ap/person.rs
@@ -289,7 +289,7 @@ impl ActivityHandler for PersonUpdate {
         let event = Event::Update(Update::UpdateUser {
             user_id: user.id.into(),
         });
-        event.send(&*data.db, &mut data.redis()).await?;
+        event.send(&*data.db).await?;
         Ok(())
     }
 }
diff --git a/backend/src/ap/undo.rs b/backend/src/ap/undo.rs
@@ -117,7 +117,7 @@ impl ActivityHandler for Undo<Follow> {
         let event = Event::Notification(Notification::new(NotificationType::DeleteFollower {
             user_id: follower_id.into(),
         }));
-        event.send(&*data.db, &mut data.redis()).await?;
+        event.send(&*data.db).await?;
 
         Ok(())
     }