Merge branch 'main' into fix/adjust-bit-manipulation

Signed-off-by: ispik <ispik@ispik.dev>

.github/workflows/docker-cleanup.yaml

@@ -0,0 +1,46 @@
+name: Docker PR Image Cleanup
+
+on:
+  pull_request:
+    types:
+      - closed
+
+permissions:
+  contents: read
+  packages: write
+
+concurrency:
+  group: docker-cleanup-${{ github.event.pull_request.number }}
+  cancel-in-progress: false
+
+jobs:
+  cleanup:
+    runs-on: ubuntu-latest
+    if: ${{ !github.event.pull_request.head.repo.fork }}
+    strategy:
+      fail-fast: false
+      matrix:
+        package:
+          - base
+          - api
+          - events
+          - file-server
+          - proxy
+          - gifbox
+          - crond
+          - pushd
+          - voice-ingress
+    steps:
+      - env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          ORG: stoatchat
+          PACKAGE: ${{ matrix.package }}
+          TAG: pr-${{ github.event.pull_request.number }}
+        run: |
+          set -euo pipefail
+          gh api --paginate \
+            "/orgs/${ORG}/packages/container/${PACKAGE}/versions" \
+            --jq ".[] | select(.metadata.container.tags | index(\"${TAG}\")) | .id" \
+          | while read -r id; do
+              gh api -X DELETE "/orgs/${ORG}/packages/container/${PACKAGE}/versions/${id}"
+            done

.github/workflows/docker.yaml

@@ -5,8 +5,6 @@ on:
     tags:
       - "*"
   pull_request:
-    paths:
-      - "Dockerfile"
   workflow_dispatch:
 
 permissions:
@@ -19,9 +17,9 @@ concurrency:
 
 jobs:
   base:
-    name: Test base image build
+    name: Test base image build (fork)
     runs-on: arc-runner-set
-    if: github.event_name == 'pull_request'
+    if: github.event_name == 'pull_request' && github.event.pull_request.head.repo.fork
     steps:
       # Configure build environment
       - name: Checkout
@@ -42,7 +40,7 @@ jobs:
 
   publish:
     runs-on: arc-runner-set
-    if: github.event_name != 'pull_request'
+    if: ${{ github.event_name != 'pull_request' || !github.event.pull_request.head.repo.fork }}
     name: Publish Docker images
     steps:
       # Configure build environment
@@ -59,6 +57,15 @@ jobs:
           username: ${{ github.actor }}
           password: ${{ secrets.GITHUB_TOKEN }}
 
+      - name: Determine base image tag
+        id: base
+        run: |
+          if [ "${{ github.event_name }}" = "pull_request" ]; then
+            echo "tag=pr-${{ github.event.number }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tag=latest" >> "$GITHUB_OUTPUT"
+          fi
+
       # Build the image
       - name: Build base image
         uses: docker/build-push-action@v4
@@ -66,7 +73,9 @@ jobs:
           context: .
           push: true
           platforms: linux/amd64,linux/arm64
-          tags: ghcr.io/${{ github.repository_owner }}/base:latest
+          tags: ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
+          cache-from: type=gha,scope=buildx-base-multi-arch
+          cache-to: type=gha,scope=buildx-base-multi-arch,mode=max
 
       # stoatchat/api
       - name: Docker meta
@@ -84,7 +93,7 @@ jobs:
           file: crates/delta/Dockerfile
           tags: ${{ steps.meta-delta.outputs.tags }}
           build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
           labels: ${{ steps.meta-delta.outputs.labels }}
 
       # stoatchat/events
@@ -103,7 +112,7 @@ jobs:
           file: crates/bonfire/Dockerfile
           tags: ${{ steps.meta-bonfire.outputs.tags }}
           build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
           labels: ${{ steps.meta-bonfire.outputs.labels }}
 
       # stoatchat/file-server
@@ -122,7 +131,7 @@ jobs:
           file: crates/services/autumn/Dockerfile
           tags: ${{ steps.meta-autumn.outputs.tags }}
           build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
           labels: ${{ steps.meta-autumn.outputs.labels }}
 
       # stoatchat/proxy
@@ -141,7 +150,7 @@ jobs:
           file: crates/services/january/Dockerfile
           tags: ${{ steps.meta-january.outputs.tags }}
           build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
           labels: ${{ steps.meta-january.outputs.labels }}
 
       # stoatchat/gifbox
@@ -160,7 +169,7 @@ jobs:
           file: crates/services/gifbox/Dockerfile
           tags: ${{ steps.meta-gifbox.outputs.tags }}
           build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
           labels: ${{ steps.meta-gifbox.outputs.labels }}
 
       # stoatchat/crond
@@ -179,7 +188,7 @@ jobs:
           file: crates/daemons/crond/Dockerfile
           tags: ${{ steps.meta-crond.outputs.tags }}
           build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
           labels: ${{ steps.meta-crond.outputs.labels }}
 
       # stoatchat/pushd
@@ -198,7 +207,7 @@ jobs:
           file: crates/daemons/pushd/Dockerfile
           tags: ${{ steps.meta-pushd.outputs.tags }}
           build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
           labels: ${{ steps.meta-pushd.outputs.labels }}
 
       # stoatchat/voice-ingress
@@ -217,5 +226,5 @@ jobs:
           file: crates/daemons/voice-ingress/Dockerfile
           tags: ${{ steps.meta-voice-ingress.outputs.tags }}
           build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
           labels: ${{ steps.meta-voice-ingress.outputs.labels }}

.github/workflows/publish-crates.yml

@@ -21,4 +21,6 @@ jobs:
           github-token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Publish
+        env:
+          CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
         run: mise publish --workspace

.github/workflows/rust.yaml

@@ -37,6 +37,7 @@ jobs:
       - name: Reference Test
         env:
           TEST_DB: REFERENCE
+        continue-on-error: ${{ github.ref_name == 'main' }}
         run: |
           mise test
 
@@ -44,6 +45,7 @@ jobs:
         env:
           TEST_DB: MONGODB
           MONGODB: mongodb://localhost
+        continue-on-error: ${{ github.ref_name == 'main' }}
         run: |
           mise test
 

.release-please-manifest.json

@@ -1,3 +1,3 @@
 {
-    ".": "0.13.4"
+    ".": "0.13.7"
 }

CHANGELOG.md

@@ -1,5 +1,41 @@
 # Changelog
 
+## [0.13.7](https://github.com/stoatchat/stoatchat/compare/v0.13.6...v0.13.7) (2026-05-21)
+
+
+### Bug Fixes
+
+* sanitize emoji input to handle variation selectors ([#774](https://github.com/stoatchat/stoatchat/issues/774)) ([2d308e0](https://github.com/stoatchat/stoatchat/commit/2d308e03d58c19f27b5b4d65dc2a15ef20b56190))
+* update mention count badge for channel acks ([#769](https://github.com/stoatchat/stoatchat/issues/769)) ([0d9ae50](https://github.com/stoatchat/stoatchat/commit/0d9ae508d9d2199f0e408b8ca634d20489be6f61))
+
+## [0.13.6](https://github.com/stoatchat/stoatchat/compare/v0.13.5...v0.13.6) (2026-05-18)
+
+
+### Features
+
+* Update FCM payload for android notifications ([#766](https://github.com/stoatchat/stoatchat/issues/766)) ([acbc087](https://github.com/stoatchat/stoatchat/commit/acbc087982e9aeb05cabc5ab4c9b1291f67490ad))
+* user slowmode events ([#760](https://github.com/stoatchat/stoatchat/issues/760)) ([af0d8aa](https://github.com/stoatchat/stoatchat/commit/af0d8aad14dc68d88159d0e1c714077d362e21e4))
+
+
+### Bug Fixes
+
+* include `minio` region as tests need it ([#761](https://github.com/stoatchat/stoatchat/issues/761)) ([298742d](https://github.com/stoatchat/stoatchat/commit/298742dbad4eafae356f976c56b9db23904b0c3a))
+* set env var for publishing crates ([#768](https://github.com/stoatchat/stoatchat/issues/768)) ([018afaf](https://github.com/stoatchat/stoatchat/commit/018afaf38f6330d92dad2a68b640c0cb3f6b639a))
+* Use proper headers to determine IP when not behind cloudflare ([#764](https://github.com/stoatchat/stoatchat/issues/764)) ([494c8b7](https://github.com/stoatchat/stoatchat/commit/494c8b7cabaae2a51039a7a5b559d5e2e5279554))
+* voice ingress crashing due to new Result in AMQP::new_auto() ([#765](https://github.com/stoatchat/stoatchat/issues/765)) ([2871632](https://github.com/stoatchat/stoatchat/commit/2871632382395cb20cbe0047c542d3ac31ff3f03))
+
+
+### Miscellaneous Chores
+
+* switch to lapin ([#767](https://github.com/stoatchat/stoatchat/issues/767)) ([5b19853](https://github.com/stoatchat/stoatchat/commit/5b1985381ae829a92c80a19e91a414cd9dc4de93))
+
+## [0.13.5](https://github.com/stoatchat/stoatchat/compare/v0.13.4...v0.13.5) (2026-05-17)
+
+
+### Bug Fixes
+
+* dont panic on hash missing when deleting files ([#755](https://github.com/stoatchat/stoatchat/issues/755)) ([c902077](https://github.com/stoatchat/stoatchat/commit/c902077cf51076fee11712eb732dc8a8f786fc4b))
+
 ## [0.13.4](https://github.com/stoatchat/stoatchat/compare/v0.13.3...v0.13.4) (2026-05-16)
 
 

Cargo.lock

@@ -174,31 +174,6 @@ dependencies = [
  "url",
 ]
 
-[[package]]
-name = "amqp_serde"
-version = "0.4.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f5f450f572a1ec4cdb4af7af09cbd0c7c3e1b9da2bfc7414c059a780993a8e16"
-dependencies = [
- "bytes 1.11.1",
- "serde",
- "serde_bytes",
-]
-
-[[package]]
-name = "amqprs"
-version = "1.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3f1b4afcbd862e16c272b7625b6b057930b052d63c720bc90f6afab0d9abe8a8"
-dependencies = [
- "amqp_serde",
- "async-trait",
- "bytes 1.11.1",
- "serde",
- "serde_bytes_ng",
- "tokio 1.51.0",
-]
-
 [[package]]
 name = "android_system_properties"
 version = "0.1.5"
@@ -7513,7 +7488,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-autumn"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "axum",
  "axum-macros",
@@ -7554,7 +7529,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-bonfire"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "async-channel 2.5.0",
  "async-std",
@@ -7585,7 +7560,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-coalesced"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "indexmap 2.13.1",
  "lru",
@@ -7594,7 +7569,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-config"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "async-std",
  "cached",
@@ -7611,7 +7586,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-crond"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "futures-lite",
  "iso8601-timestamp",
@@ -7631,9 +7606,8 @@ dependencies = [
 
 [[package]]
 name = "revolt-database"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
- "amqprs",
  "async-lock 2.8.0",
  "async-recursion",
  "async-std",
@@ -7648,6 +7622,7 @@ dependencies = [
  "indexmap 2.13.1",
  "isahc",
  "iso8601-timestamp",
+ "lapin",
  "linkify",
  "livekit-api",
  "livekit-protocol",
@@ -7682,9 +7657,8 @@ dependencies = [
 
 [[package]]
 name = "revolt-delta"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
- "amqprs",
  "async-channel 2.5.0",
  "async-std",
  "authifier",
@@ -7694,6 +7668,7 @@ dependencies = [
  "futures",
  "impl_ops",
  "iso8601-timestamp",
+ "lapin",
  "lettre",
  "linkify",
  "livekit-api",
@@ -7731,7 +7706,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-files"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "aes-gcm",
  "anyhow",
@@ -7759,7 +7734,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-gifbox"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "axum",
  "axum-extra",
@@ -7782,7 +7757,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-january"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "async-recursion",
  "axum",
@@ -7812,7 +7787,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-models"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "indexmap 2.13.1",
  "iso8601-timestamp",
@@ -7831,14 +7806,14 @@ dependencies = [
 
 [[package]]
 name = "revolt-parser"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "logos",
 ]
 
 [[package]]
 name = "revolt-permissions"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "async-std",
  "async-trait",
@@ -7853,7 +7828,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-presence"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "async-std",
  "log",
@@ -7865,9 +7840,8 @@ dependencies = [
 
 [[package]]
 name = "revolt-pushd"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
- "amqprs",
  "anyhow",
  "async-trait",
  "authifier",
@@ -7875,6 +7849,7 @@ dependencies = [
  "fcm_v1",
  "isahc",
  "iso8601-timestamp",
+ "lapin",
  "log",
  "pretty_env_logger",
  "redis-kiss",
@@ -7896,7 +7871,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-ratelimits"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "async-trait",
  "authifier",
@@ -7913,7 +7888,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-result"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
  "axum",
  "log",
@@ -7929,9 +7904,8 @@ dependencies = [
 
 [[package]]
 name = "revolt-voice-ingress"
-version = "0.13.4"
+version = "0.13.7"
 dependencies = [
- "amqprs",
  "async-std",
  "chrono",
  "futures",
@@ -8987,15 +8961,6 @@ dependencies = [
  "serde_core",
 ]
 
-[[package]]
-name = "serde_bytes_ng"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bdb0ebce8684e2253f964e8b6ce51f0ccc6666bbb448fb4a6788088bda6544b6"
-dependencies = [
- "serde",
-]
-
 [[package]]
 name = "serde_core"
 version = "1.0.228"

Cargo.toml

@@ -159,7 +159,6 @@ opentelemetry-appender-tracing = "0.31.1"
 authifier = "1.0.16"
 
 # RabbitMQ
-amqprs = "1.7.0"
 lapin = "4.7.1"
 
 # Voice
@@ -192,13 +191,13 @@ futures-lite = "2.6.1"
 vergen = "7.5.0"
 
 # Local packages
-revolt-coalesced = { version = "0.13.4", path = "crates/core/coalesced" }
+revolt-coalesced = { version = "0.13.7", path = "crates/core/coalesced" }
-revolt-config = { version = "0.13.4", path = "crates/core/config" }
+revolt-config = { version = "0.13.7", path = "crates/core/config" }
-revolt-database = { version = "0.13.4", path = "crates/core/database" }
+revolt-database = { version = "0.13.7", path = "crates/core/database" }
-revolt-files = { version = "0.13.4", path = "crates/core/files" }
+revolt-files = { version = "0.13.7", path = "crates/core/files" }
-revolt-models = { version = "0.13.4", path = "crates/core/models" }
+revolt-models = { version = "0.13.7", path = "crates/core/models" }
-revolt-parser = { version = "0.13.4", path = "crates/core/parser" }
+revolt-parser = { version = "0.13.7", path = "crates/core/parser" }
-revolt-permissions = { version = "0.13.4", path = "crates/core/permissions" }
+revolt-permissions = { version = "0.13.7", path = "crates/core/permissions" }
-revolt-presence = { version = "0.13.4", path = "crates/core/presence" }
+revolt-presence = { version = "0.13.7", path = "crates/core/presence" }
-revolt-ratelimits = { version = "0.13.4", path = "crates/core/ratelimits" }
+revolt-ratelimits = { version = "0.13.7", path = "crates/core/ratelimits" }
-revolt-result = { version = "0.13.4", path = "crates/core/result" }
+revolt-result = { version = "0.13.7", path = "crates/core/result" }

compose.yml

@@ -34,6 +34,7 @@ services:
     environment:
       MINIO_ROOT_USER: minioautumn
       MINIO_ROOT_PASSWORD: minioautumn
+      MINIO_REGION: minio
     volumes:
       - ./.data/minio:/data
     ports:

crates/bonfire/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-bonfire"
-version = "0.13.4"
+version = "0.13.7"
 license = "AGPL-3.0-or-later"
 edition = "2021"
 publish = false

crates/bonfire/src/events/impl.rs

@@ -1,6 +1,7 @@
 use std::collections::{HashMap, HashSet};
 
 use futures::future::join_all;
+use redis_kiss::AsyncCommands;
 use revolt_database::{
     events::client::{EventV1, ReadyPayloadFields},
     util::permissions::DatabasePermissionQuery,

crates/bonfire/src/websocket.rs

@@ -13,7 +13,7 @@ use futures::{
     stream::{SplitSink, SplitStream},
     FutureExt, SinkExt, StreamExt, TryStreamExt,
 };
-use redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
+use redis_kiss::{get_connection, AsyncCommands, PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
 use revolt_config::report_internal_error;
 use revolt_database::{
     events::{client::EventV1, server::ClientMessage},
@@ -32,6 +32,7 @@ use sentry::Level;
 
 use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
 use crate::events::state::{State, SubscriptionStateChange};
+use revolt_models::v0;
 
 type WsReader = SplitStream<WebSocketStream<TcpStream>>;
 type WsWriter = SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>;
@@ -128,6 +129,14 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
         return;
     }
 
+    let slowmodes = fetch_user_slowmodes(&user_id).await.unwrap_or_default();
+    if !slowmodes.is_empty() {
+        let event = EventV1::UserSlowmodes { slowmodes };
+        if report_internal_error!(write.send(config.encode(&event)).await).is_err() {
+            return;
+        }
+    }
+
     // Create presence session.
     let (first_session, session_id) = create_session(&user_id, 0).await;
 
@@ -528,3 +537,42 @@ async fn worker(
         }
     }
 }
+
+async fn fetch_user_slowmodes(user_id: &str) -> Option<Vec<v0::ChannelSlowmode>> {
+    let mut conn = get_connection().await.ok()?.into_inner();
+    let idx_key = format!("slowmode_idx:{}", user_id);
+
+    let channel_ids: Vec<String> = conn.smembers(&idx_key).await.unwrap_or_default();
+    if channel_ids.is_empty() {
+        return Some(vec![]);
+    }
+
+    // Bulk fetch all TTLs in one round trip
+    let mut pipe = redis_kiss::redis::pipe();
+    for channel_id in &channel_ids {
+        pipe.ttl(format!("slowmode:{}:{}", user_id, channel_id));
+    }
+    let ttls: Vec<i64> = pipe.query_async(&mut conn).await.unwrap_or_default();
+
+    // Partition into alive/expired in one pass
+    let mut slowmodes = vec![];
+    let mut expired = vec![];
+    for (channel_id, ttl) in channel_ids.iter().zip(ttls.iter()) {
+        if *ttl > 0 {
+            slowmodes.push(v0::ChannelSlowmode {
+                channel_id: channel_id.clone(),
+                duration: *ttl as u64,
+                retry_after: *ttl as u64,
+            });
+        } else {
+            expired.push(channel_id.as_str());
+        }
+    }
+
+    // Bulk remove all expired members in one SREM call
+    if !expired.is_empty() {
+        conn.srem::<_, _, ()>(&idx_key, expired).await.ok();
+    }
+
+    Some(slowmodes)
+}

crates/core/coalesced/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-coalesced"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>", "Zomatree <me@zomatree.live>"]

crates/core/config/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-config"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/core/database/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-database"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -95,7 +95,7 @@ revolt_rocket_okapi = { workspace = true, optional = true }
 authifier = { workspace = true }
 
 # RabbitMQ
-amqprs = { workspace = true }
+lapin = { workspace = true, features = ["tokio"] }
 
 # Voice
 livekit-api = { workspace = true, features = ["rustls-tls-native-roots"], optional = true }

crates/core/database/src/amqp/amqp.rs

@@ -1,103 +1,77 @@
 use std::collections::HashSet;
+use std::sync::Arc;
 
 use crate::events::rabbit::*;
 use crate::User;
-use amqprs::channel::{
+use lapin::{
-    BasicPublishArguments, ExchangeDeclareArguments, ExchangeType, QueueBindArguments,
+    options::BasicPublishOptions,
-    QueueDeclareArguments,
+    protocol::basic::AMQPProperties,
+    types::{AMQPValue, FieldTable},
+    Channel, Connection, ConnectionProperties, Error as AMQPError,
 };
-use amqprs::connection::OpenConnectionArguments;
-use amqprs::{channel::Channel, connection::Connection, error::Error as AMQPError};
-use amqprs::{BasicProperties, FieldTable};
 use revolt_models::v0::PushNotification;
 use revolt_presence::filter_online;
+use revolt_result::Result;
 
 use serde_json::to_string;
 
 #[derive(Clone)]
 pub struct AMQP {
+    friend_request_accepted: Arc<Channel>,
+    friend_request_received: Arc<Channel>,
+    generic_message: Arc<Channel>,
+    message_sent: Arc<Channel>,
+    mass_mention_message_sent: Arc<Channel>,
+    ack_notification_message: Arc<Channel>,
+    dm_call_updated: Arc<Channel>,
+    process_ack: Arc<Channel>,
     #[allow(unused)]
-    connection: Connection,
+    connection: Arc<Connection>,
-    channel: Channel,
 }
 
 impl AMQP {
-    pub fn new(connection: Connection, channel: Channel) -> AMQP {
+    pub async fn new(connection: Arc<Connection>) -> Self {
-        AMQP {
+        Self {
+            friend_request_accepted: Self::create_channel(&connection).await,
+            friend_request_received: Self::create_channel(&connection).await,
+            generic_message: Self::create_channel(&connection).await,
+            message_sent: Self::create_channel(&connection).await,
+            mass_mention_message_sent: Self::create_channel(&connection).await,
+            ack_notification_message: Self::create_channel(&connection).await,
+            dm_call_updated: Self::create_channel(&connection).await,
+            process_ack: Self::create_channel(&connection).await,
             connection,
-            channel,
         }
     }
 
-    pub async fn new_auto() -> revolt_result::Result<AMQP> {
+    pub async fn new_auto() -> Self {
         let config = revolt_config::config().await;
 
-        let connection = Connection::open(&OpenConnectionArguments::new(
+        let connection = Arc::new(
-            &config.rabbit.host,
+            Connection::connect(
-            config.rabbit.port,
+                &format!(
+                    "amqp://{}:{}@{}:{}",
                     &config.rabbit.username,
                     &config.rabbit.password,
-        ))
+                    &config.rabbit.host,
-        .await
+                    &config.rabbit.port,
-        .expect("Failed to connect to RabbitMQ");
+                ),
-
+                ConnectionProperties::default(),
-        let channel = connection
-            .open_channel(None)
-            .await
-            .expect("Failed to open RabbitMQ channel");
-
-        let mut resp = AMQP::new(connection, channel);
-        //resp.configure_channels().await?;
-        Ok(resp)
-    }
-
-    pub async fn repoen_channel(&mut self) {
-        self.channel = self
-            .connection
-            .open_channel(None)
-            .await
-            .expect("Failed to open RabbitMQ channel");
-    }
-
-    pub async fn configure_channels(&mut self) -> revolt_result::Result<()> {
-        let config = revolt_config::config().await;
-
-        if !self.channel.is_open() {
-            self.repoen_channel().await;
-        }
-
-        self.channel
-            .exchange_declare(
-                ExchangeDeclareArguments::new(
-                    &config.rabbit.default_exchange,
-                    &ExchangeType::Topic.to_string(),
-                )
-                .durable(true)
-                .finish(),
             )
             .await
-            .expect("Failed to declare exchange");
+            .expect("Failed to connect to RabbitMQ"),
+        );
 
-        // Configure acks channel & routing
+        Self::new(connection).await
-        self.channel
+    }
-            .queue_declare(
+
-                QueueDeclareArguments::new(&config.rabbit.queues.acks)
+    async fn create_channel(connection: &Connection) -> Arc<Channel> {
-                    .durable(true)
+        Arc::new(
-                    .no_wait(true)
+            connection
-                    .finish(),
+                .create_channel()
+                .await
+                .expect("Failed to create channel"),
         )
-            .await
-            .expect("Failed to bind queue");
-
-        self.channel
-            .queue_bind(QueueBindArguments::new(
-                &config.rabbit.queues.acks,
-                &config.rabbit.default_exchange,
-                &config.rabbit.queues.acks,
-            ))
-            .await
-            .expect("Failed to bind channel");
-        Ok(())
     }
 
     pub async fn friend_request_accepted(
@@ -117,19 +91,20 @@ impl AMQP {
             config.pushd.get_fr_accepted_routing_key(),
             payload
         );
-        self.channel
+
+        self.friend_request_accepted
             .basic_publish(
-                BasicProperties::default()
+                config.pushd.exchange.clone().into(),
-                    .with_content_type("application/json")
+                config.pushd.get_fr_accepted_routing_key().into(),
-                    .with_persistence(true)
+                BasicPublishOptions::default(),
-                    .finish(),
+                payload.as_bytes(),
-                payload.into(),
+                AMQPProperties::default()
-                BasicPublishArguments::new(
+                    .with_content_type("application/json".into())
-                    &config.pushd.exchange,
+                    .with_delivery_mode(2),
-                    &config.pushd.get_fr_accepted_routing_key(),
-                ),
             )
-            .await
+            .await?;
+
+        Ok(())
     }
 
     pub async fn friend_request_received(
@@ -150,19 +125,19 @@ impl AMQP {
             payload
         );
 
-        self.channel
+        self.friend_request_received
             .basic_publish(
-                BasicProperties::default()
+                config.pushd.exchange.clone().into(),
-                    .with_content_type("application/json")
+                config.pushd.get_fr_received_routing_key().into(),
-                    .with_persistence(true)
+                BasicPublishOptions::default(),
-                    .finish(),
+                payload.as_bytes(),
-                payload.into(),
+                AMQPProperties::default()
-                BasicPublishArguments::new(
+                    .with_content_type("application/json".into())
-                    &config.pushd.exchange,
+                    .with_delivery_mode(2),
-                    &config.pushd.get_fr_received_routing_key(),
-                ),
             )
-            .await
+            .await?;
+
+        Ok(())
     }
 
     pub async fn generic_message(
@@ -187,19 +162,19 @@ impl AMQP {
             payload
         );
 
-        self.channel
+        self.generic_message
             .basic_publish(
-                BasicProperties::default()
+                config.pushd.exchange.clone().into(),
-                    .with_content_type("application/json")
+                config.pushd.get_generic_routing_key().into(),
-                    .with_persistence(true)
+                BasicPublishOptions::default(),
-                    .finish(),
+                payload.as_bytes(),
-                payload.into(),
+                AMQPProperties::default()
-                BasicPublishArguments::new(
+                    .with_content_type("application/json".into())
-                    &config.pushd.exchange,
+                    .with_delivery_mode(2),
-                    &config.pushd.get_generic_routing_key(),
-                ),
             )
-            .await
+            .await?;
+
+        Ok(())
     }
 
     pub async fn message_sent(
@@ -230,19 +205,19 @@ impl AMQP {
             payload
         );
 
-        self.channel
+        self.message_sent
             .basic_publish(
-                BasicProperties::default()
+                config.pushd.exchange.clone().into(),
-                    .with_content_type("application/json")
+                config.pushd.get_message_routing_key().into(),
-                    .with_persistence(true)
+                BasicPublishOptions::default(),
-                    .finish(),
+                payload.as_bytes(),
-                payload.into(),
+                AMQPProperties::default()
-                BasicPublishArguments::new(
+                    .with_content_type("application/json".into())
-                    &config.pushd.exchange,
+                    .with_delivery_mode(2),
-                    &config.pushd.get_message_routing_key(),
-                ),
             )
-            .await
+            .await?;
+
+        Ok(())
     }
 
     pub async fn mass_mention_message_sent(
@@ -265,16 +240,19 @@ impl AMQP {
             routing_key, payload
         );
 
-        self.channel
+        self.mass_mention_message_sent
             .basic_publish(
-                BasicProperties::default()
+                config.pushd.exchange.clone().into(),
-                    .with_content_type("application/json")
+                routing_key.into(),
-                    .with_persistence(true)
+                BasicPublishOptions::default(),
-                    .finish(),
+                payload.as_bytes(),
-                payload.into(),
+                AMQPProperties::default()
-                BasicPublishArguments::new(&config.pushd.exchange, routing_key.as_str()),
+                    .with_content_type("application/json".into())
+                    .with_delivery_mode(2),
             )
-            .await
+            .await?;
+
+        Ok(())
     }
 
     /// # Sends an ack to pushd to update badges on iPhones.
@@ -299,23 +277,25 @@ impl AMQP {
             config.pushd.ack_queue, payload
         );
 
-        let mut headers = FieldTable::new();
+        let mut headers = FieldTable::default();
         headers.insert(
-            "x-deduplication-header".try_into().unwrap(),
+            "x-deduplication-header".into(),
-            format!("{}-{}", &user_id, &channel_id).into(),
+            AMQPValue::LongString(format!("{}-{}", &user_id, &channel_id).into()),
         );
 
-        self.channel
+        self.ack_notification_message
             .basic_publish(
-                BasicProperties::default()
+                config.pushd.exchange.clone().into(),
-                    .with_content_type("application/json")
+                config.pushd.ack_queue.into(),
-                    .with_persistence(true)
+                BasicPublishOptions::default(),
-                    //.with_headers(headers)
+                payload.as_bytes(),
-                    .finish(),
+                AMQPProperties::default()
-                payload.into(),
+                    .with_content_type("application/json".into())
-                BasicPublishArguments::new(&config.pushd.exchange, &config.pushd.ack_queue),
+                    .with_delivery_mode(2),
             )
-            .await
+            .await?;
+
+        Ok(())
     }
 
     /// # DM Call Update
@@ -349,19 +329,19 @@ impl AMQP {
             payload
         );
 
-        self.channel
+        self.dm_call_updated
             .basic_publish(
-                BasicProperties::default()
+                config.pushd.exchange.clone().into(),
-                    .with_content_type("application/json")
+                config.pushd.get_dm_call_routing_key().into(),
-                    .with_persistence(true)
+                BasicPublishOptions::default(),
-                    .finish(),
+                payload.as_bytes(),
-                payload.into(),
+                AMQPProperties::default()
-                BasicPublishArguments::new(
+                    .with_content_type("application/json".into())
-                    &config.pushd.exchange,
+                    .with_delivery_mode(2),
-                    &config.pushd.get_dm_call_routing_key(),
-                ),
             )
-            .await
+            .await?;
+
+        Ok(())
     }
 
     /// # Send an ack to crond for processing
@@ -385,19 +365,18 @@ impl AMQP {
             config.rabbit.default_exchange, config.rabbit.queues.acks, payload
         );
 
-        self.channel
+        self.process_ack
             .basic_publish(
-                BasicProperties::default()
+                config.rabbit.default_exchange.clone().into(),
-                    .with_content_type("application/json")
+                config.rabbit.queues.acks.into(),
-                    .with_persistence(true)
+                BasicPublishOptions::default(),
-                    //.with_headers(headers)
+                payload.as_bytes(),
-                    .finish(),
+                AMQPProperties::default()
-                payload.into(),
+                    .with_content_type("application/json".into())
-                BasicPublishArguments::new(
+                    .with_delivery_mode(2),
-                    &config.rabbit.default_exchange,
-                    &config.rabbit.queues.acks,
-                ),
             )
-            .await
+            .await?;
+
+        Ok(())
     }
 }

crates/core/database/src/events/client.rs

@@ -3,7 +3,12 @@ use revolt_result::Error;
 use serde::{Deserialize, Serialize};
 
 use revolt_models::v0::{
-    AppendMessage, Channel, ChannelUnread, ChannelVoiceState, Emoji, FieldsChannel, FieldsMember, FieldsMessage, FieldsRole, FieldsServer, FieldsUser, FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialEmoji, PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialUserVoiceState, PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings, UserVoiceState, Webhook
+    AppendMessage, Channel, ChannelSlowmode, ChannelUnread, ChannelVoiceState, Emoji,
+    FieldsChannel, FieldsMember, FieldsMessage, FieldsRole, FieldsServer, FieldsUser,
+    FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialEmoji,
+    PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialUserVoiceState,
+    PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings,
+    UserVoiceState, Webhook,
 };
 
 use crate::Database;
@@ -51,9 +56,13 @@ impl Default for ReadyPayloadFields {
 #[serde(tag = "type")]
 pub enum EventV1 {
     /// Multiple events
-    Bulk { v: Vec<EventV1> },
+    Bulk {
+        v: Vec<EventV1>,
+    },
     /// Error event
-    Error { data: Error },
+    Error {
+        data: Error,
+    },
 
     /// Successfully authenticated
     Authenticated,
@@ -84,7 +93,9 @@ pub enum EventV1 {
     },
 
     /// Ping response
-    Pong { data: Ping },
+    Pong {
+        data: Ping,
+    },
     /// New message
     Message(Message),
 
@@ -105,7 +116,10 @@ pub enum EventV1 {
     },
 
     /// Delete message
-    MessageDelete { id: String, channel: String },
+    MessageDelete {
+        id: String,
+        channel: String,
+    },
 
     /// New reaction to a message
     MessageReact {
@@ -131,7 +145,10 @@ pub enum EventV1 {
     },
 
     /// Bulk delete messages
-    BulkMessageDelete { channel: String, ids: Vec<String> },
+    BulkMessageDelete {
+        channel: String,
+        ids: Vec<String>,
+    },
 
     /// New server
     ServerCreate {
@@ -139,7 +156,7 @@ pub enum EventV1 {
         server: Server,
         channels: Vec<Channel>,
         emojis: Vec<Emoji>,
-        voice_states: Vec<ChannelVoiceState>
+        voice_states: Vec<ChannelVoiceState>,
     },
 
     /// Update existing server
@@ -151,7 +168,9 @@ pub enum EventV1 {
     },
 
     /// Delete server
-    ServerDelete { id: String },
+    ServerDelete {
+        id: String,
+    },
 
     /// Update existing server member
     ServerMemberUpdate {
@@ -187,10 +206,16 @@ pub enum EventV1 {
     },
 
     /// Server role deleted
-    ServerRoleDelete { id: String, role_id: String },
+    ServerRoleDelete {
+        id: String,
+        role_id: String,
+    },
 
     /// Server roles ranks updated
-    ServerRoleRanksUpdate { id: String, ranks: Vec<String> },
+    ServerRoleRanksUpdate {
+        id: String,
+        ranks: Vec<String>,
+    },
 
     /// Update existing user
     UserUpdate {
@@ -202,9 +227,15 @@ pub enum EventV1 {
     },
 
     /// Relationship with another user changed
-    UserRelationship { id: String, user: User },
+    UserRelationship {
+        id: String,
+        user: User,
+    },
     /// Settings updated remotely
-    UserSettingsUpdate { id: String, update: UserSettings },
+    UserSettingsUpdate {
+        id: String,
+        update: UserSettings,
+    },
 
     /// User has been platform banned or deleted their account
     ///
@@ -215,7 +246,10 @@ pub enum EventV1 {
     /// - Server Memberships
     ///
     /// User flags are specified to explain why a wipe is occurring though not all reasons will necessarily ever appear.
-    UserPlatformWipe { user_id: String, flags: i32 },
+    UserPlatformWipe {
+        user_id: String,
+        flags: i32,
+    },
     /// New emoji
     EmojiCreate(Emoji),
 
@@ -226,7 +260,9 @@ pub enum EventV1 {
     },
 
     /// Delete emoji
-    EmojiDelete { id: String },
+    EmojiDelete {
+        id: String,
+    },
 
     /// New report
     ReportCreate(Report),
@@ -242,19 +278,33 @@ pub enum EventV1 {
     },
 
     /// Delete channel
-    ChannelDelete { id: String },
+    ChannelDelete {
+        id: String,
+    },
 
     /// User joins a group
-    ChannelGroupJoin { id: String, user: String },
+    ChannelGroupJoin {
+        id: String,
+        user: String,
+    },
 
     /// User leaves a group
-    ChannelGroupLeave { id: String, user: String },
+    ChannelGroupLeave {
+        id: String,
+        user: String,
+    },
 
     /// User started typing in a channel
-    ChannelStartTyping { id: String, user: String },
+    ChannelStartTyping {
+        id: String,
+        user: String,
+    },
 
     /// User stopped typing in a channel
-    ChannelStopTyping { id: String, user: String },
+    ChannelStopTyping {
+        id: String,
+        user: String,
+    },
 
     /// User acknowledged message in channel
     ChannelAck {
@@ -274,7 +324,9 @@ pub enum EventV1 {
     },
 
     /// Delete webhook
-    WebhookDelete { id: String },
+    WebhookDelete {
+        id: String,
+    },
 
     /// Auth events
     Auth(AuthifierEvent),
@@ -292,7 +344,7 @@ pub enum EventV1 {
         user: String,
         from: String,
         to: String,
-        state: UserVoiceState
+        state: UserVoiceState,
     },
     UserVoiceStateUpdate {
         id: String,
@@ -304,7 +356,11 @@ pub enum EventV1 {
         from: String,
         to: String,
         token: String,
-    }
+    },
+    /// User's active slowmodes
+    UserSlowmodes {
+        slowmodes: Vec<ChannelSlowmode>,
+    },
 }
 
 impl EventV1 {

crates/core/database/src/models/emojis/model.rs

@@ -12,7 +12,7 @@ use crate::Database;
 static PERMISSIBLE_EMOJIS: Lazy<HashSet<String>> = Lazy::new(|| {
     include_str!("unicode_emoji.txt")
         .split('\n')
-        .map(|x| x.into())
+        .map(|x| x.replace('\u{FE0F}', ""))
         .collect()
 });
 
@@ -108,7 +108,8 @@ impl Emoji {
             db.fetch_emoji(emoji).await?;
             Ok(true)
         } else {
-            Ok(PERMISSIBLE_EMOJIS.contains(emoji))
+            let sanitized_emoji = emoji.replace('\u{FE0F}', "");
+            Ok(PERMISSIBLE_EMOJIS.contains(&sanitized_emoji))
         }
     }
 }

crates/core/database/src/models/emojis/unicode_emoji.txt

@@ -1,4 +1,3 @@
-
 *️⃣
 0️⃣
 1️⃣
@@ -242,6 +241,7 @@
 🆘
 🆙
 🆚
+🇦
 🇦🇨
 🇦🇩
 🇦🇪
@@ -259,6 +259,7 @@
 🇦🇼
 🇦🇽
 🇦🇿
+🇧
 🇧🇦
 🇧🇧
 🇧🇩
@@ -280,6 +281,7 @@
 🇧🇼
 🇧🇾
 🇧🇿
+🇨
 🇨🇦
 🇨🇨
 🇨🇩
@@ -301,6 +303,7 @@
 🇨🇽
 🇨🇾
 🇨🇿
+🇩
 🇩🇪
 🇩🇬
 🇩🇯
@@ -308,6 +311,7 @@
 🇩🇲
 🇩🇴
 🇩🇿
+🇪
 🇪🇦
 🇪🇨
 🇪🇪
@@ -317,12 +321,14 @@
 🇪🇸
 🇪🇹
 🇪🇺
+🇫
 🇫🇮
 🇫🇯
 🇫🇰
 🇫🇲
 🇫🇴
 🇫🇷
+🇬
 🇬🇦
 🇬🇧
 🇬🇩
@@ -342,12 +348,14 @@
 🇬🇺
 🇬🇼
 🇬🇾
+🇭
 🇭🇰
 🇭🇲
 🇭🇳
 🇭🇷
 🇭🇹
 🇭🇺
+🇮
 🇮🇨
 🇮🇩
 🇮🇪
@@ -359,10 +367,12 @@
 🇮🇷
 🇮🇸
 🇮🇹
+🇯
 🇯🇪
 🇯🇲
 🇯🇴
 🇯🇵
+🇰
 🇰🇪
 🇰🇬
 🇰🇭
@@ -374,6 +384,7 @@
 🇰🇼
 🇰🇾
 🇰🇿
+🇱
 🇱🇦
 🇱🇧
 🇱🇨
@@ -385,6 +396,7 @@
 🇱🇺
 🇱🇻
 🇱🇾
+🇲
 🇲🇦
 🇲🇨
 🇲🇩
@@ -408,6 +420,7 @@
 🇲🇽
 🇲🇾
 🇲🇿
+🇳
 🇳🇦
 🇳🇨
 🇳🇪
@@ -420,7 +433,9 @@
 🇳🇷
 🇳🇺
 🇳🇿
+🇴
 🇴🇲
+🇵
 🇵🇦
 🇵🇪
 🇵🇫
@@ -435,12 +450,15 @@
 🇵🇹
 🇵🇼
 🇵🇾
+🇶
 🇶🇦
+🇷
 🇷🇪
 🇷🇴
 🇷🇸
 🇷🇺
 🇷🇼
+🇸
 🇸🇦
 🇸🇧
 🇸🇨
@@ -462,6 +480,7 @@
 🇸🇽
 🇸🇾
 🇸🇿
+🇹
 🇹🇦
 🇹🇨
 🇹🇩
@@ -479,6 +498,7 @@
 🇹🇻
 🇹🇼
 🇹🇿
+🇺
 🇺🇦
 🇺🇬
 🇺🇲
@@ -486,6 +506,7 @@
 🇺🇸
 🇺🇾
 🇺🇿
+🇻
 🇻🇦
 🇻🇨
 🇻🇪
@@ -493,11 +514,15 @@
 🇻🇮
 🇻🇳
 🇻🇺
+🇼
 🇼🇫
 🇼🇸
+🇽
 🇽🇰
+🇾
 🇾🇪
 🇾🇹
+🇿
 🇿🇦
 🇿🇲
 🇿🇼

crates/core/database/src/models/messages/model.rs

@@ -219,7 +219,7 @@ impl MessageFlagsValue {
         self.has_value(flag as u32)
     }
     pub fn has_value(&self, bit: u32) -> bool {
-        let mask = 1 << bit;
+        let mask = 1 << (bit - 1);
         self.0 & mask == mask
     }
 
@@ -227,10 +227,11 @@ impl MessageFlagsValue {
         self.set_value(flag as u32, toggle)
     }
     pub fn set_value(&mut self, bit: u32, toggle: bool) -> &mut Self {
+        let mask = 1 << (bit - 1);
         if toggle {
-            self.0 |= 1 << bit;
+            self.0 |= mask;
         } else {
-            self.0 &= !(1 << bit);
+            self.0 &= !mask;
         }
         self
     }

crates/core/files/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-files"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/core/models/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-models"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/core/models/src/v0/channels.rs

@@ -314,6 +314,12 @@ auto_derived!(
         /// Only used when the user is the first one connected.
         pub recipients: Option<Vec<String>>,
     }
+
+    pub struct ChannelSlowmode {
+        pub channel_id: String,
+        pub duration: u64,
+        pub retry_after: u64,
+    }
 );
 
 impl Channel {

crates/core/parser/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-parser"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "MIT"
 authors = ["Zomatree <me@zomatree.live>", "Paul Makles <me@insrt.uk>"]

crates/core/permissions/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-permissions"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/core/presence/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-presence"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/core/ratelimits/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-ratelimits"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2024"
 license = "MIT"
 authors = ["Zomatree <me@zomatree.live>", "Paul Makles <me@insrt.uk>"]

crates/core/ratelimits/src/rocket.rs

@@ -1,12 +1,12 @@
 use async_trait::async_trait;
 use log::info;
+use revolt_config::config;
 use rocket::fairing::{Fairing, Info, Kind};
 use rocket::http::uri::Origin;
 use rocket::http::{Method, Status};
 use rocket::request::{FromRequest, Outcome};
 use rocket::serde::json::Json;
 use rocket::{Data, Request, Response, State};
-use revolt_config::config;
 
 use revolt_rocket_okapi::r#gen::OpenApiGenerator;
 use revolt_rocket_okapi::request::{OpenApiFromRequest, RequestHeaderInput};
@@ -28,8 +28,8 @@ pub type RatelimitStorage = crate::ratelimiter::RatelimitStorage<RocketRequestKi
 /// Find the remote IP of the client
 fn to_ip(request: &'_ rocket::Request<'_>) -> String {
     request
-        .remote()
+        .client_ip()
-        .map(|x| x.ip().to_string())
+        .map(|r| r.to_string())
         .unwrap_or_default()
 }
 

crates/core/result/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-result"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/daemons/crond/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-crond"
-version = "0.13.4"
+version = "0.13.7"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
 edition = "2021"

crates/daemons/crond/src/main.rs

@@ -1,5 +1,5 @@
 use revolt_config::configure;
-use revolt_database::DatabaseInfo;
+use revolt_database::{DatabaseInfo, AMQP};
 use revolt_result::Result;
 use tasks::{acks, file_deletion, prune_dangling_files, prune_members};
 use tokio::try_join;
@@ -11,11 +11,13 @@ async fn main() -> Result<()> {
     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())
+        acks::task(db.clone(), amqp.clone()),
     )
     .map(|_| ())
 }

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

@@ -3,16 +3,16 @@ use lapin::{
     options::*,
     types::FieldTable,
     uri::{AMQPAuthority, AMQPQueryString, AMQPUri, AMQPUserInfo},
-    ConnectionBuilder, ConnectionProperties,
+    ConnectionBuilder, ConnectionProperties, ExchangeKind,
 };
-use log::info;
+use log::{debug, info};
 use redis_kiss::{get_connection, AsyncCommands, Conn as RedisConnection};
 use revolt_config::config;
-use revolt_database::{events::rabbit::AckEventPayload, Database};
+use revolt_database::{events::rabbit::AckEventPayload, Database, AMQP};
 use revolt_result::{Result, ToRevoltError};
 use serde_json;
 
-pub async fn task(db: Database) -> Result<()> {
+pub async fn task(db: Database, amqp: AMQP) -> Result<()> {
     let config = config().await;
 
     let mut redis = get_connection()
@@ -46,6 +46,42 @@ pub async fn task(db: Database) -> Result<()> {
         .await
         .expect("Failed to create channel");
 
+    reader_channel
+        .exchange_declare(
+            config.rabbit.default_exchange.clone().into(),
+            ExchangeKind::Topic,
+            ExchangeDeclareOptions {
+                durable: true,
+                ..Default::default()
+            },
+            FieldTable::default(),
+        )
+        .await
+        .expect("Failed to declare exchange");
+
+    reader_channel
+        .queue_declare(
+            config.rabbit.queues.acks.clone().into(),
+            QueueDeclareOptions {
+                durable: true,
+                ..Default::default()
+            },
+            FieldTable::default(),
+        )
+        .await
+        .expect("Failed to bind queue");
+
+    reader_channel
+        .queue_bind(
+            config.rabbit.queues.acks.clone().into(),
+            config.rabbit.default_exchange.into(),
+            config.rabbit.queues.acks.clone().into(),
+            QueueBindOptions::default(),
+            FieldTable::default(),
+        )
+        .await
+        .expect("Failed to bind channel");
+
     let mut consumer = reader_channel
         .basic_consume(
             config.rabbit.queues.acks.into(),
@@ -58,12 +94,14 @@ pub async fn task(db: Database) -> Result<()> {
 
     while let Some(delivery) = consumer.next().await {
         if let Ok(delivery) = delivery {
-            let payload: std::result::Result<AckEventPayload, _> =
+            let payload = serde_json::from_slice::<AckEventPayload>(&delivery.data);
-                serde_json::from_slice(&delivery.data);
+
             if let Ok(payload) = payload {
-                info!("{:?}", payload);
+                debug!("Received ack event: {payload:?}");
+
                 if let Err(e) = process_channel_ack(
                     &db,
+                    &amqp,
                     payload.user_id,
                     payload.channel_id.unwrap(),
                     &mut redis,
@@ -89,6 +127,7 @@ pub async fn task(db: Database) -> Result<()> {
 #[allow(clippy::disallowed_methods)]
 async fn process_channel_ack(
     db: &Database,
+    amqp: &AMQP,
     user: String,
     channel: String,
     redis: &mut RedisConnection,
@@ -99,28 +138,24 @@ async fn process_channel_ack(
         .to_internal_error()?;
 
     if let Some(message_id) = message_id {
-        // This will be uncommented eventually, but we need to sort out the transition to lapin first. For now we'll simply disable the badge update logic.
+        let unread = db.fetch_unread(&user, &channel).await?;
-        // We also drop a db request as a bonus.
+        let updated = db.acknowledge_message(&channel, &user, &message_id).await?;
 
-        //let unread = db.fetch_unread(&user, &channel).await?;
-        let _updated = db.acknowledge_message(&channel, &user, &message_id).await?;
         info!("Set new state for ack: {}:{}:{}", channel, user, message_id);
 
-        // if let (Some(before), Some(after)) = (unread, updated) {
+        if let (Some(before), Some(after)) = (unread, updated) {
-        //     let before_mentions = before.mentions.unwrap_or_default().len();
+            let before_mentions = before.mentions.unwrap_or_default().len();
-        //     let after_mentions = after.mentions.unwrap_or_default().len();
+            let after_mentions = after.mentions.unwrap_or_default().len();
 
-        //     let mentions_acked = before_mentions - after_mentions;
+            if after_mentions < before_mentions {
-
+                if let Err(err) = amqp
-        //     if mentions_acked > 0 {
+                    .ack_notification_message(user.to_string(), channel.to_string(), message_id)
-        //         if let Err(err) = amqp
+                    .await
-        //             .ack_message(user.to_string(), channel.to_string(), payload.message_id)
+                {
-        //             .await
+                    revolt_config::capture_error(&err);
-        //         {
+                }
-        //             revolt_config::capture_error(&err);
+            };
-        //         }
+        }
-        //     };
-        // }
 
         Ok(())
     } else {

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

@@ -11,14 +11,15 @@ pub async fn task(db: Database) -> Result<()> {
         let files = db.fetch_deleted_attachments().await?;
 
         for file in files {
+            if let Some(hash) = &file.hash {
                 let count = db
-                .count_file_hash_references(file.hash.as_ref().expect("no `hash` present"))
+                    .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(file.hash.as_ref().expect("no `hash` present"))
+                        .fetch_attachment_hash(hash)
                         .await?;
 
                     // Delete from S3
@@ -28,6 +29,7 @@ pub async fn task(db: Database) -> Result<()> {
                     db.delete_attachment_hash(&file_hash.id).await?;
                     info!("Deleted file hash {}", file_hash.id);
                 }
+            }
 
             // Delete the file
             db.delete_attachment(&file.id).await?;

crates/daemons/pushd/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-pushd"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 publish = false
@@ -15,7 +15,7 @@ revolt-parser = { workspace = true }
 
 anyhow = { workspace = true }
 
-amqprs = { workspace = true }
+lapin = { workspace = true }
 fcm_v1 = { workspace = true }
 web-push = { workspace = true }
 isahc = { workspace = true, features = ["json"], optional = true }

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

@@ -1,96 +1,69 @@
-use crate::consumers::inbound::internal::*;
+use std::sync::Arc;
-use amqprs::{
+
-    channel::{BasicPublishArguments, Channel},
+use crate::utils::Consumer;
-    connection::Connection,
+use anyhow::Result;
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use revolt_database::{events::rabbit::*, Database};
 
+#[derive(Clone)]
+#[allow(unused)]
 pub struct AckConsumer {
-    #[allow(dead_code)]
     db: Database,
     authifier_db: authifier::Database,
-    conn: Option<Connection>,
+    connection: Arc<Connection>,
-    channel: Option<Channel>,
+    channel: Arc<Channel>,
 }
 
-impl Channeled for AckConsumer {
+#[async_trait]
-    fn get_connection(&self) -> Option<&Connection> {
+impl Consumer for AckConsumer {
-        if self.conn.is_none() {
+    async fn create(
-            None
+        db: Database,
-        } else {
+        authifier_db: authifier::Database,
-            Some(self.conn.as_ref().unwrap())
+        connection: Arc<Connection>,
-        }
+        channel: Arc<Channel>,
-    }
+    ) -> Self {
-
+        Self {
-    fn get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl AckConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> AckConsumer {
-        AckConsumer {
             db,
             authifier_db,
-            conn: None,
+            connection,
-            channel: None,
+            channel,
         }
     }
-}
 
-#[allow(unused_variables)]
+    fn channel(&self) -> &Arc<Channel> {
-#[async_trait]
+        &self.channel
-impl AsyncConsumer for AckConsumer {
+    }
+
     /// This consumer processes all acks the platform receives, and sends relevant badge updates to apple platforms.
-    async fn consume(
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        &mut self,
+        let payload: AckPayload = serde_json::from_slice(&delivery.data)?;
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        let content = String::from_utf8(content).unwrap();
-        let payload: AckPayload = serde_json::from_str(content.as_str()).unwrap();
 
         // Step 1: fetch unreads and don't continue if there's no unreads
-        #[allow(clippy::disallowed_methods)]
+        // #[allow(clippy::disallowed_methods)]
-        let unreads = self.db.fetch_unread_mentions(&payload.user_id).await;
 
         debug!("Processing unreads for {:}", &payload.user_id);
 
-        if let Ok(u) = &unreads {
+        let unreads = if let Ok(u) = self.db.fetch_unread_mentions(&payload.user_id).await {
             if u.is_empty() {
                 debug!(
                     "Discarding unread task (no mentions found) for {:}",
                     &payload.user_id
                 );
-                return;
+                return Ok(());
-            }
+            };
+
+            u
         } else {
-            return;
+            return Ok(());
-        }
+        };
 
         if let Ok(sessions) = self.authifier_db.find_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
-            let apple_sessions: Vec<&authifier::models::Session> = sessions
+            let mut apple_sessions = sessions
-                .iter()
+                .into_iter()
                 .filter(|session| {
                     if let Some(sub) = &session.subscription {
                         sub.endpoint == "apn"
@@ -98,19 +71,19 @@ impl AsyncConsumer for AckConsumer {
                         false
                     }
                 })
-                .collect();
+                .peekable();
 
-            if apple_sessions.is_empty() {
+            if apple_sessions.peek().is_none() {
                 debug!(
                     "Discarding unread task (no apn sessions found) for {:}",
                     &payload.user_id
                 );
-                return;
+                return Ok(());
             }
 
             // Step 3: calculate the actual mention count, since we have to send it out
             let mut mention_count = 0;
-            for u in &unreads.unwrap() {
+            for u in &unreads {
                 mention_count += u.mentions.as_ref().unwrap().len()
             }
 
@@ -123,26 +96,22 @@ impl AsyncConsumer for AckConsumer {
                     token: session.subscription.as_ref().unwrap().auth.clone(),
                     extras: Default::default(),
                 };
-                let raw_service_payload = serde_json::to_string(&service_payload);
+                let payload = serde_json::to_string(&service_payload)?;
-
-                if let Ok(p) = raw_service_payload {
-                    let args = BasicPublishArguments::new(
-                        config.pushd.exchange.as_str(),
-                        config.pushd.apn.queue.as_str(),
-                    )
-                    .finish();
 
                 log::debug!(
                     "Publishing ack to apn session {}",
                     session.subscription.as_ref().unwrap().auth
                 );
 
-                    publish_message(self, p.into(), args).await;
+                self.publish_message(
-                } else {
+                    payload.as_bytes(),
-                    log::warn!("Failed to serialize ack badge update payload!");
+                    &config.pushd.exchange,
-                    revolt_config::capture_error(&raw_service_payload.unwrap_err());
+                    &config.pushd.apn.queue,
-                }
+                )
+                .await?;
             }
         }
+
+        Ok(())
     }
 }

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

@@ -1,70 +1,44 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
-use crate::consumers::inbound::internal::*;
+use crate::utils::Consumer;
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
 use revolt_database::{events::rabbit::*, Database};
 
+#[derive(Clone)]
+#[allow(unused)]
 pub struct DmCallConsumer {
-    #[allow(dead_code)]
     db: Database,
     authifier_db: authifier::Database,
-    conn: Option<Connection>,
+    connection: Arc<Connection>,
-    channel: Option<Channel>,
+    channel: Arc<Channel>,
 }
 
-impl Channeled for DmCallConsumer {
+#[async_trait]
-    fn get_connection(&self) -> Option<&Connection> {
+impl Consumer for DmCallConsumer {
-        if self.conn.is_none() {
+    async fn create(
-            None
+        db: Database,
-        } else {
+        authifier_db: authifier::Database,
-            Some(self.conn.as_ref().unwrap())
+        connection: Arc<Connection>,
-        }
+        channel: Arc<Channel>,
-    }
+    ) -> Self {
-
+        Self {
-    fn get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl DmCallConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> DmCallConsumer {
-        DmCallConsumer {
             db,
             authifier_db,
-            conn: None,
+            connection,
-            channel: None,
+            channel,
         }
     }
 
-    async fn consume_event(
+    fn channel(&self) -> &Arc<Channel> {
-        &mut self,
+        &self.channel
-        _channel: &Channel,
+    }
-        _deliver: Deliver,
+
-        _basic_properties: BasicProperties,
+    /// This consumer handles delegating messages into their respective platform queues.
-        content: Vec<u8>,
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
-    ) -> Result<()> {
+        let _p: InternalDmCallPayload = serde_json::from_slice(&delivery.data)?;
-        let content = String::from_utf8(content)?;
-        let _p: InternalDmCallPayload = serde_json::from_str(content.as_str())?;
         let payload = _p.payload;
 
         debug!("Received dm call start/stop event");
@@ -107,36 +81,27 @@ impl DmCallConsumer {
                             extras: HashMap::new(),
                         };
 
-                        let args: BasicPublishArguments;
+                        let routing_key = match sub.endpoint.as_str() {
-
+                            "apn" => &config.pushd.apn.queue,
-                        if sub.endpoint == "apn" {
+                            "fcm" => &config.pushd.fcm.queue,
-                            args = BasicPublishArguments::new(
+                            endpoint => {
-                                config.pushd.exchange.as_str(),
-                                config.pushd.apn.queue.as_str(),
-                            )
-                            .finish();
-                        } else if sub.endpoint == "fcm" {
-                            args = BasicPublishArguments::new(
-                                config.pushd.exchange.as_str(),
-                                config.pushd.fcm.queue.as_str(),
-                            )
-                            .finish();
-                        } else {
-                            // web push (vapid)
-                            args = BasicPublishArguments::new(
-                                config.pushd.exchange.as_str(),
-                                config.pushd.vapid.queue.as_str(),
-                            )
-                            .finish();
                                 sendable.extras.insert("p256dh".to_string(), sub.p256dh);
                                 sendable
                                     .extras
-                                .insert("endpoint".to_string(), sub.endpoint.clone());
+                                    .insert("endpoint".to_string(), endpoint.to_string());
+
+                                &config.pushd.vapid.queue
                             }
+                        };
 
                         let payload = serde_json::to_string(&sendable)?;
 
-                        publish_message(self, payload.into(), args).await;
+                        self.publish_message(
+                            payload.as_bytes(),
+                            &config.pushd.exchange,
+                            routing_key,
+                        )
+                        .await?;
                     }
                 }
             }
@@ -145,24 +110,3 @@ impl DmCallConsumer {
         Ok(())
     }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for DmCallConsumer {
-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            warn!("Failed to process dm call start/stop event: {err:?}");
-        }
-    }
-}

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

@@ -1,70 +1,44 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
-use crate::consumers::inbound::internal::*;
+use crate::utils::Consumer;
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
 use revolt_database::{events::rabbit::*, Database};
 
+#[derive(Clone)]
+#[allow(unused)]
 pub struct FRAcceptedConsumer {
-    #[allow(dead_code)]
     db: Database,
     authifier_db: authifier::Database,
-    conn: Option<Connection>,
+    connection: Arc<Connection>,
-    channel: Option<Channel>,
+    channel: Arc<Channel>,
 }
 
-impl Channeled for FRAcceptedConsumer {
+#[async_trait]
-    fn get_connection(&self) -> Option<&Connection> {
+impl Consumer for FRAcceptedConsumer {
-        if self.conn.is_none() {
+    async fn create(
-            None
+        db: Database,
-        } else {
+        authifier_db: authifier::Database,
-            Some(self.conn.as_ref().unwrap())
+        connection: Arc<Connection>,
-        }
+        channel: Arc<Channel>,
-    }
+    ) -> Self {
-
+        Self {
-    fn get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl FRAcceptedConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> FRAcceptedConsumer {
-        FRAcceptedConsumer {
             db,
             authifier_db,
-            conn: None,
+            connection,
-            channel: None,
+            channel,
         }
     }
 
-    async fn consume_event(
+    fn channel(&self) -> &Arc<Channel> {
-        &mut self,
+        &self.channel
-        _channel: &Channel,
+    }
-        _deliver: Deliver,
+
-        _basic_properties: BasicProperties,
+    /// This consumer handles delegating messages into their respective platform queues.
-        content: Vec<u8>,
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
-    ) -> Result<()> {
+        let payload: FRAcceptedPayload = serde_json::from_slice(&delivery.data)?;
-        let content = String::from_utf8(content)?;
-        let payload: FRAcceptedPayload = serde_json::from_str(content.as_str())?;
 
         debug!("Received FR accept event");
 
@@ -80,36 +54,23 @@ impl FRAcceptedConsumer {
                         extras: HashMap::new(),
                     };
 
-                    let args: BasicPublishArguments;
+                    let routing_key = match sub.endpoint.as_str() {
-
+                        "apn" => &config.pushd.apn.queue,
-                    if sub.endpoint == "apn" {
+                        "fcm" => &config.pushd.fcm.queue,
-                        args = BasicPublishArguments::new(
+                        endpoint => {
-                            config.pushd.exchange.as_str(),
-                            config.pushd.apn.queue.as_str(),
-                        )
-                        .finish();
-                    } else if sub.endpoint == "fcm" {
-                        args = BasicPublishArguments::new(
-                            config.pushd.exchange.as_str(),
-                            config.pushd.fcm.queue.as_str(),
-                        )
-                        .finish();
-                    } else {
-                        // web push (vapid)
-                        args = BasicPublishArguments::new(
-                            config.pushd.exchange.as_str(),
-                            config.pushd.vapid.queue.as_str(),
-                        )
-                        .finish();
                             sendable.extras.insert("p256dh".to_string(), sub.p256dh);
                             sendable
                                 .extras
-                            .insert("endpoint".to_string(), sub.endpoint.clone());
+                                .insert("endpoint".to_string(), endpoint.to_string());
+
+                            &config.pushd.vapid.queue
                         }
+                    };
 
                     let payload = serde_json::to_string(&sendable)?;
 
-                    publish_message(self, payload.into(), args).await;
+                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
+                        .await?;
                 }
             }
         }
@@ -117,24 +78,3 @@ impl FRAcceptedConsumer {
         Ok(())
     }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for FRAcceptedConsumer {
-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process friend request accepted event: {err:?}");
-        }
-    }
-}

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

@@ -1,70 +1,44 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
-use crate::consumers::inbound::internal::*;
+use crate::utils::Consumer;
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
 use revolt_database::{events::rabbit::*, Database};
 
+#[derive(Clone)]
+#[allow(unused)]
 pub struct FRReceivedConsumer {
-    #[allow(dead_code)]
     db: Database,
     authifier_db: authifier::Database,
-    conn: Option<Connection>,
+    connection: Arc<Connection>,
-    channel: Option<Channel>,
+    channel: Arc<Channel>,
 }
 
-impl Channeled for FRReceivedConsumer {
+#[async_trait]
-    fn get_connection(&self) -> Option<&Connection> {
+impl Consumer for FRReceivedConsumer {
-        if self.conn.is_none() {
+    async fn create(
-            None
+        db: Database,
-        } else {
+        authifier_db: authifier::Database,
-            Some(self.conn.as_ref().unwrap())
+        connection: Arc<Connection>,
-        }
+        channel: Arc<Channel>,
-    }
+    ) -> Self {
-
+        Self {
-    fn get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl FRReceivedConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> FRReceivedConsumer {
-        FRReceivedConsumer {
             db,
             authifier_db,
-            conn: None,
+            connection,
-            channel: None,
+            channel,
         }
     }
 
-    async fn consume_event(
+    fn channel(&self) -> &Arc<Channel> {
-        &mut self,
+        &self.channel
-        _channel: &Channel,
+    }
-        _deliver: Deliver,
+
-        _basic_properties: BasicProperties,
+    /// This consumer handles delegating messages into their respective platform queues.
-        content: Vec<u8>,
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
-    ) -> Result<()> {
+        let payload: FRReceivedPayload = serde_json::from_slice(&delivery.data)?;
-        let content = String::from_utf8(content)?;
-        let payload: FRReceivedPayload = serde_json::from_str(content.as_str())?;
 
         debug!("Received FR received event");
 
@@ -80,36 +54,23 @@ impl FRReceivedConsumer {
                         extras: HashMap::new(),
                     };
 
-                    let args: BasicPublishArguments;
+                    let routing_key = match sub.endpoint.as_str() {
-
+                        "apn" => &config.pushd.apn.queue,
-                    if sub.endpoint == "apn" {
+                        "fcm" => &config.pushd.fcm.queue,
-                        args = BasicPublishArguments::new(
+                        endpoint => {
-                            config.pushd.exchange.as_str(),
-                            config.pushd.apn.queue.as_str(),
-                        )
-                        .finish();
-                    } else if sub.endpoint == "fcm" {
-                        args = BasicPublishArguments::new(
-                            config.pushd.exchange.as_str(),
-                            config.pushd.fcm.queue.as_str(),
-                        )
-                        .finish();
-                    } else {
-                        // web push (vapid)
-                        args = BasicPublishArguments::new(
-                            config.pushd.exchange.as_str(),
-                            config.pushd.vapid.queue.as_str(),
-                        )
-                        .finish();
                             sendable.extras.insert("p256dh".to_string(), sub.p256dh);
                             sendable
                                 .extras
-                            .insert("endpoint".to_string(), sub.endpoint.clone());
+                                .insert("endpoint".to_string(), endpoint.to_string());
+
+                            &config.pushd.vapid.queue
                         }
+                    };
 
                     let payload = serde_json::to_string(&sendable)?;
 
-                    publish_message(self, payload.into(), args).await;
+                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
+                        .await?;
                 }
             }
         }
@@ -117,24 +78,3 @@ impl FRReceivedConsumer {
         Ok(())
     }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for FRReceivedConsumer {
-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process friend request received event: {err:?}");
-        }
-    }
-}

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

@@ -1,70 +1,44 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
-use crate::consumers::inbound::internal::*;
+use crate::utils::Consumer;
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
 use revolt_database::{events::rabbit::*, Database};
 
+#[derive(Clone)]
+#[allow(unused)]
 pub struct GenericConsumer {
-    #[allow(dead_code)]
     db: Database,
     authifier_db: authifier::Database,
-    conn: Option<Connection>,
+    connection: Arc<Connection>,
-    channel: Option<Channel>,
+    channel: Arc<Channel>,
 }
 
-impl Channeled for GenericConsumer {
+#[async_trait]
-    fn get_connection(&self) -> Option<&Connection> {
+impl Consumer for GenericConsumer {
-        if self.conn.is_none() {
+    async fn create(
-            None
+        db: Database,
-        } else {
+        authifier_db: authifier::Database,
-            Some(self.conn.as_ref().unwrap())
+        connection: Arc<Connection>,
-        }
+        channel: Arc<Channel>,
-    }
+    ) -> Self {
-
+        Self {
-    fn get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl GenericConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> GenericConsumer {
-        GenericConsumer {
             db,
             authifier_db,
-            conn: None,
+            connection,
-            channel: None,
+            channel,
         }
     }
 
-    async fn consume_event(
+    fn channel(&self) -> &Arc<Channel> {
-        &mut self,
+        &self.channel
-        _channel: &Channel,
+    }
-        _deliver: Deliver,
+
-        _basic_properties: BasicProperties,
+    /// This consumer handles delegating messages into their respective platform queues.
-        content: Vec<u8>,
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
-    ) -> Result<()> {
+        let payload: MessageSentPayload = serde_json::from_slice(&delivery.data)?;
-        let content = String::from_utf8(content)?;
-        let payload: MessageSentPayload = serde_json::from_str(content.as_str())?;
 
         debug!("Received message event on origin");
 
@@ -86,36 +60,23 @@ impl GenericConsumer {
                         extras: HashMap::new(),
                     };
 
-                    let args: BasicPublishArguments;
+                    let routing_key = match sub.endpoint.as_str() {
-
+                        "apn" => &config.pushd.apn.queue,
-                    if sub.endpoint == "apn" {
+                        "fcm" => &config.pushd.fcm.queue,
-                        args = BasicPublishArguments::new(
+                        endpoint => {
-                            config.pushd.exchange.as_str(),
-                            config.pushd.apn.queue.as_str(),
-                        )
-                        .finish();
-                    } else if sub.endpoint == "fcm" {
-                        args = BasicPublishArguments::new(
-                            config.pushd.exchange.as_str(),
-                            config.pushd.fcm.queue.as_str(),
-                        )
-                        .finish();
-                    } else {
-                        // web push (vapid)
-                        args = BasicPublishArguments::new(
-                            config.pushd.exchange.as_str(),
-                            config.pushd.vapid.queue.as_str(),
-                        )
-                        .finish();
                             sendable.extras.insert("p256dh".to_string(), sub.p256dh);
                             sendable
                                 .extras
-                            .insert("endpoint".to_string(), sub.endpoint.clone());
+                                .insert("endpoint".to_string(), endpoint.to_string());
+
+                            &config.pushd.vapid.queue
                         }
+                    };
 
                     let payload = serde_json::to_string(&sendable)?;
 
-                    publish_message(self, payload.into(), args).await;
+                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
+                        .await?;
                 }
             }
         }
@@ -123,24 +84,3 @@ impl GenericConsumer {
         Ok(())
     }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for GenericConsumer {
-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process generic event: {err:?}");
-        }
-    }
-}

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

@@ -1,53 +0,0 @@
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::{Connection, OpenConnectionArguments},
-    BasicProperties,
-};
-use log::{debug, warn};
-
-pub(crate) trait Channeled {
-    #[allow(unused)]
-    fn get_connection(&self) -> Option<&Connection>;
-    fn get_channel(&self) -> Option<&Channel>;
-    fn set_connection(&mut self, conn: Connection);
-    fn set_channel(&mut self, channel: Channel);
-}
-
-pub(crate) async fn make_channel<T: Channeled>(consumer: &mut T) {
-    let config = revolt_config::config().await;
-
-    let args = OpenConnectionArguments::new(
-        &config.rabbit.host,
-        config.rabbit.port,
-        &config.rabbit.username,
-        &config.rabbit.password,
-    );
-    let conn = amqprs::connection::Connection::open(&args).await.unwrap();
-
-    let channel = conn.open_channel(None).await.unwrap();
-
-    consumer.set_connection(conn);
-    consumer.set_channel(channel);
-}
-
-pub(crate) async fn publish_message<T: Channeled>(
-    consumer: &mut T,
-    payload: Vec<u8>,
-    args: BasicPublishArguments,
-) {
-    let routing_key = &args.routing_key.clone();
-    let mut channel = consumer.get_channel();
-    if channel.is_none() {
-        make_channel(consumer).await;
-        channel = consumer.get_channel();
-    }
-
-    if let Some(chnl) = channel {
-        chnl.basic_publish(BasicProperties::default(), payload.clone(), args.clone())
-            .await
-            .unwrap();
-        debug!("Sent message to queue for target {}", routing_key);
-    } else {
-        warn!("Failed to unwrap channel (including attempt to make a channel)!")
-    }
-}

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

@@ -1,17 +1,13 @@
 use std::{
     collections::{HashMap, HashSet},
     hash::RandomState,
+    sync::Arc,
 };
 
-use crate::{consumers::inbound::internal::*, utils};
+use crate::utils::{render_notification_content, Consumer};
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use revolt_database::{
     events::rabbit::*, util::bulk_permissions::BulkDatabasePermissionQuery, Database, Member,
     MessageFlagsValue,
@@ -19,52 +15,18 @@ use revolt_database::{
 use revolt_models::v0::{MessageFlags, PushNotification};
 use revolt_result::ToRevoltError;
 
+#[derive(Clone)]
+#[allow(unused)]
 pub struct MassMessageConsumer {
-    #[allow(dead_code)]
     db: Database,
     authifier_db: authifier::Database,
-    conn: Option<Connection>,
+    connection: Arc<Connection>,
-    channel: Option<Channel>,
+    channel: Arc<Channel>,
-}
-
-impl Channeled for MassMessageConsumer {
-    fn get_connection(&self) -> Option<&Connection> {
-        if self.conn.is_none() {
-            None
-        } else {
-            Some(self.conn.as_ref().unwrap())
-        }
-    }
-
-    fn get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
 }
 
 impl MassMessageConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> MassMessageConsumer {
-        MassMessageConsumer {
-            db,
-            authifier_db,
-            conn: None,
-            channel: None,
-        }
-    }
-
     async fn fire_notification_for_users(
-        &mut self,
+        &self,
         push: &PushNotification,
         users: &[String],
     ) -> Result<()> {
@@ -84,56 +46,58 @@ impl MassMessageConsumer {
                         extras: HashMap::new(),
                     };
 
-                    let args: BasicPublishArguments;
+                    let routing_key = match sub.endpoint.as_str() {
-
+                        "apn" => &config.pushd.apn.queue,
-                    if sub.endpoint == "apn" {
+                        "fcm" => &config.pushd.fcm.queue,
-                        args = BasicPublishArguments::new(
+                        endpoint => {
-                            config.pushd.exchange.as_str(),
-                            config.pushd.apn.queue.as_str(),
-                        )
-                        .finish();
-                    } else if sub.endpoint == "fcm" {
-                        args = BasicPublishArguments::new(
-                            config.pushd.exchange.as_str(),
-                            config.pushd.fcm.queue.as_str(),
-                        )
-                        .finish();
-                    } else {
-                        // web push (vapid)
-                        args = BasicPublishArguments::new(
-                            config.pushd.exchange.as_str(),
-                            config.pushd.vapid.queue.as_str(),
-                        )
-                        .finish();
                             sendable.extras.insert("p256dh".to_string(), sub.p256dh);
                             sendable
                                 .extras
-                            .insert("endpoint".to_string(), sub.endpoint.clone());
+                                .insert("endpoint".to_string(), endpoint.to_string());
+
+                            &config.pushd.vapid.queue
                         }
+                    };
 
                     let payload = serde_json::to_string(&sendable)?;
 
-                    publish_message(self, payload.into(), args).await;
+                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
+                        .await?;
                 }
             }
         }
 
         Ok(())
     }
+}
 
-    async fn consume_event(
+#[async_trait]
-        &mut self,
+impl Consumer for MassMessageConsumer {
-        _channel: &Channel,
+    async fn create(
-        _deliver: Deliver,
+        db: Database,
-        _basic_properties: BasicProperties,
+        authifier_db: authifier::Database,
-        content: Vec<u8>,
+        connection: Arc<Connection>,
-    ) -> Result<()> {
+        channel: Arc<Channel>,
+    ) -> Self {
+        Self {
+            db,
+            authifier_db,
+            connection,
+            channel,
+        }
+    }
+
+    fn channel(&self) -> &Arc<Channel> {
+        &self.channel
+    }
+
+    /// This consumer handles adding mentions for all the users affected by a mass mention ping, and then sends out push notifications.
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
+        let mut payload: MassMessageSentPayload = serde_json::from_slice(&delivery.data)?;
         let config = revolt_config::config().await;
-        let content = String::from_utf8(content)?;
-        let mut payload: MassMessageSentPayload = serde_json::from_str(content.as_str())?;
 
         for push in payload.notifications.iter_mut() {
-            if let Ok(body) = utils::render_notification_content(push, &self.db)
+            if let Ok(body) = render_notification_content(push, &self.db)
                 .await
                 .to_internal_error()
             {
@@ -280,24 +244,3 @@ impl MassMessageConsumer {
         Ok(())
     }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for MassMessageConsumer {
-    /// This consumer handles adding mentions for all the users affected by a mass mention ping, and then sends out push notifications
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process mass message event: {err:?}");
-        }
-    }
-}

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

@@ -1,76 +1,46 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
-use crate::{consumers::inbound::internal::*, utils};
+use crate::utils::{render_notification_content, Consumer};
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
 use revolt_database::{events::rabbit::*, Database};
-use revolt_result::ToRevoltError;
 
+#[derive(Clone)]
+#[allow(unused)]
 pub struct MessageConsumer {
-    #[allow(dead_code)]
     db: Database,
     authifier_db: authifier::Database,
-    conn: Option<Connection>,
+    connection: Arc<Connection>,
-    channel: Option<Channel>,
+    channel: Arc<Channel>,
 }
 
-impl Channeled for MessageConsumer {
+#[async_trait]
-    fn get_connection(&self) -> Option<&Connection> {
+impl Consumer for MessageConsumer {
-        if self.conn.is_none() {
+    async fn create(
-            None
+        db: Database,
-        } else {
+        authifier_db: authifier::Database,
-            Some(self.conn.as_ref().unwrap())
+        connection: Arc<Connection>,
-        }
+        channel: Arc<Channel>,
-    }
+    ) -> Self {
-
+        Self {
-    fn get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl MessageConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> MessageConsumer {
-        MessageConsumer {
             db,
             authifier_db,
-            conn: None,
+            connection,
-            channel: None,
+            channel,
         }
     }
 
-    async fn consume_event(
+    fn channel(&self) -> &Arc<Channel> {
-        &mut self,
+        &self.channel
-        _channel: &Channel,
+    }
-        _deliver: Deliver,
-        _basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) -> Result<()> {
-        let content = String::from_utf8(content)?;
-        let mut payload: MessageSentPayload = serde_json::from_str(content.as_str())?;
 
-        if let Ok(body) = utils::render_notification_content(&payload.notification, &self.db)
+    /// This consumer handles delegating messages into their respective platform queues.
-            .await
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
-            .to_internal_error()
+        let mut payload: MessageSentPayload = serde_json::from_slice(&delivery.data)?;
-        {
+
+        if let Ok(body) = render_notification_content(&payload.notification, &self.db).await {
             payload.notification.raw_body = Some(payload.notification.body);
             payload.notification.body = body;
         }
@@ -95,36 +65,22 @@ impl MessageConsumer {
                         extras: HashMap::new(),
                     };
 
-                    let args: BasicPublishArguments;
+                    let routing_key = match sub.endpoint.as_str() {
-
+                        "apn" => &config.pushd.apn.queue,
-                    if sub.endpoint == "apn" {
+                        "fcm" => &config.pushd.fcm.queue,
-                        args = BasicPublishArguments::new(
+                        endpoint => {
-                            config.pushd.exchange.as_str(),
-                            config.pushd.apn.queue.as_str(),
-                        )
-                        .finish();
-                    } else if sub.endpoint == "fcm" {
-                        args = BasicPublishArguments::new(
-                            config.pushd.exchange.as_str(),
-                            config.pushd.fcm.queue.as_str(),
-                        )
-                        .finish();
-                    } else {
-                        // web push (vapid)
-                        args = BasicPublishArguments::new(
-                            config.pushd.exchange.as_str(),
-                            config.pushd.vapid.queue.as_str(),
-                        )
-                        .finish();
                             sendable.extras.insert("p256dh".to_string(), sub.p256dh);
                             sendable
                                 .extras
-                            .insert("endpoint".to_string(), sub.endpoint.clone());
+                                .insert("endpoint".to_string(), endpoint.to_string());
+
+                            &config.pushd.vapid.queue
                         }
+                    };
 
                     let payload = serde_json::to_string(&sendable)?;
-
+                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
-                    publish_message(self, payload.into(), args).await;
+                        .await?;
                 }
             }
         }
@@ -132,24 +88,3 @@ impl MessageConsumer {
         Ok(())
     }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for MessageConsumer {
-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process message event: {err:?}");
-        }
-    }
-}

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

@@ -3,6 +3,5 @@ pub mod dm_call;
 pub mod fr_accepted;
 pub mod fr_received;
 pub mod generic;
-mod internal;
 pub mod mass_mention;
 pub mod message;

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

@@ -1,12 +1,13 @@
-use std::{borrow::Cow, collections::BTreeMap, io::Cursor};
+use std::{borrow::Cow, collections::BTreeMap, io::Cursor, sync::Arc};
 
-use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};
+use crate::utils::Consumer;
-use anyhow::{anyhow, Result};
+use anyhow::Result;
 use async_trait::async_trait;
 use base64::{
     engine::{self},
     Engine as _,
 };
+use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
 use revolt_a2::{
     request::{
         notification::{DefaultAlert, NotificationOptions},
@@ -42,7 +43,7 @@ impl<'a> PayloadLike for MessagePayload<'a> {
     fn get_device_token(&self) -> &'a str {
         self.device_token
     }
-    fn get_options(&self) -> &NotificationOptions {
+    fn get_options(&self) -> &NotificationOptions<'a> {
         &self.options
     }
 }
@@ -68,16 +69,20 @@ impl<'a> PayloadLike for CallStartStopPayload<'a> {
     fn get_device_token(&self) -> &'a str {
         self.device_token
     }
-    fn get_options(&self) -> &NotificationOptions {
+    fn get_options(&self) -> &NotificationOptions<'a> {
         &self.options
     }
 }
 
 // region: consumer
 
+#[derive(Clone)]
+#[allow(unused)]
 pub struct ApnsOutboundConsumer {
-    #[allow(dead_code)]
     db: Database,
+    authifier_db: authifier::Database,
+    connection: Arc<Connection>,
+    channel: Arc<AMQPChannel>,
     client: Client,
 }
 
@@ -117,15 +122,21 @@ impl ApnsOutboundConsumer {
     }
 }
 
-impl ApnsOutboundConsumer {
+#[async_trait]
-    pub async fn new(db: Database) -> Result<ApnsOutboundConsumer, &'static str> {
+impl Consumer for ApnsOutboundConsumer {
+    async fn create(
+        db: Database,
+        authifier_db: authifier::Database,
+        connection: Arc<Connection>,
+        channel: Arc<AMQPChannel>,
+    ) -> Self {
         let config = revolt_config::config().await;
 
         if config.pushd.apn.pkcs8.is_empty()
             || config.pushd.apn.key_id.is_empty()
             || config.pushd.apn.team_id.is_empty()
         {
-            return Err("Missing APN keys.");
+            panic!("Missing APN keys.");
         }
 
         let endpoint = if config.pushd.apn.sandbox {
@@ -148,18 +159,21 @@ impl ApnsOutboundConsumer {
         )
         .expect("could not create APN client");
 
-        Ok(ApnsOutboundConsumer { db, client })
+        Self {
+            db,
+            authifier_db,
+            connection,
+            channel,
+            client,
+        }
     }
 
-    async fn consume_event(
+    fn channel(&self) -> &Arc<AMQPChannel> {
-        &mut self,
+        &self.channel
-        _channel: &AmqpChannel,
+    }
-        _deliver: Deliver,
+
-        _basic_properties: BasicProperties,
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        content: Vec<u8>,
+        let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;
-    ) -> Result<()> {
-        let content = String::from_utf8(content)?;
-        let payload: PayloadToService = serde_json::from_str(content.as_str())?;
 
         let payload_options = NotificationOptions {
             apns_id: None,
@@ -170,20 +184,15 @@ impl ApnsOutboundConsumer {
             apns_collapse_id: None,
         };
 
-        let resp: Result<Response, Error>;
+        let resp = match payload.notification {
-
-        match payload.notification {
             PayloadKind::FRReceived(alert) => {
                 let loc_args = vec![Cow::from(
-                    alert
+                    alert.from_user.display_name.clone().unwrap_or_else(|| {
-                        .from_user
+                        format!(
-                        .display_name
-                        .or(Some(format!(
                             "{}#{}",
                             alert.from_user.username, alert.from_user.discriminator
-                        )))
+                        )
-                        .clone()
+                    }),
-                        .ok_or_else(|| anyhow!("missing name"))?,
                 )];
 
                 let apn_payload = Payload {
@@ -216,20 +225,17 @@ impl ApnsOutboundConsumer {
                     "Sending friend request received for user: {:}",
                     &payload.user_id
                 );
-                resp = self.client.send(apn_payload).await;
+                self.client.send(apn_payload).await
             }
 
             PayloadKind::FRAccepted(alert) => {
                 let loc_args = vec![Cow::from(
-                    alert
+                    alert.accepted_user.display_name.clone().unwrap_or_else(|| {
-                        .accepted_user
+                        format!(
-                        .display_name
-                        .or(Some(format!(
                             "{}#{}",
                             alert.accepted_user.username, alert.accepted_user.discriminator
-                        )))
+                        )
-                        .clone()
+                    }),
-                        .ok_or_else(|| anyhow!("missing name"))?,
                 )];
 
                 let apn_payload = Payload {
@@ -262,7 +268,7 @@ impl ApnsOutboundConsumer {
                     "Sending friend request accept for user: {:}",
                     &payload.user_id
                 );
-                resp = self.client.send(apn_payload).await;
+                self.client.send(apn_payload).await
             }
             PayloadKind::Generic(alert) => {
                 let apn_payload = Payload {
@@ -295,7 +301,7 @@ impl ApnsOutboundConsumer {
                     "Sending generic notification for user: {:}",
                     &payload.user_id
                 );
-                resp = self.client.send(apn_payload).await;
+                self.client.send(apn_payload).await
             }
 
             PayloadKind::MessageNotification(alert) => {
@@ -334,7 +340,7 @@ impl ApnsOutboundConsumer {
                     "Sending message notification for user: {:}",
                     &payload.user_id
                 );
-                resp = self.client.send(apn_payload).await;
+                self.client.send(apn_payload).await
             }
 
             PayloadKind::BadgeUpdate(badge) => {
@@ -349,7 +355,7 @@ impl ApnsOutboundConsumer {
                 };
 
                 debug!("Sending badge update for user: {:}", &payload.user_id);
-                resp = self.client.send(apn_payload).await;
+                self.client.send(apn_payload).await
             }
 
             PayloadKind::DmCallStartEnd(alert) => {
@@ -378,24 +384,24 @@ impl ApnsOutboundConsumer {
                     "Sending call start/stop notification for user: {:}",
                     &payload.user_id
                 );
-                resp = self.client.send(apn_payload).await;
+                self.client.send(apn_payload).await
-            }
             }
+        };
 
-        if let Err(err) = resp {
+        match resp {
-            match err {
+            Err(Error::ResponseError(Response {
-                Error::ResponseError(Response {
                 error:
                     Some(ErrorBody {
                         reason: ErrorReason::BadDeviceToken | ErrorReason::Unregistered,
                         ..
                     }),
                 ..
-                }) => {
+            })) => {
                 info!(
                     "Removing APNS subscription id {:} (user: {:}) due to invalid token",
                     &payload.session_id, &payload.user_id
                 );
+
                 if let Err(err) = self
                     .db
                     .remove_push_subscription_by_session_id(&payload.session_id)
@@ -404,32 +410,11 @@ impl ApnsOutboundConsumer {
                     revolt_config::capture_error(&err);
                 }
             }
-                err => {
+            resp => {
-                    revolt_config::capture_error(&err);
+                resp?;
-                }
-            }
             }
+        };
 
         Ok(())
     }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for ApnsOutboundConsumer {
-    async fn consume(
-        &mut self,
-        channel: &AmqpChannel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process APN event: {err:?}");
-        }
-    }
-}

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

@@ -1,17 +1,16 @@
-use std::{collections::HashMap, time::Duration};
+use std::{collections::HashMap, sync::Arc, time::Duration};
 
-use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};
+use crate::utils::Consumer;
-
+use anyhow::{bail, Result};
-use anyhow::{anyhow, bail, Result};
 use async_trait::async_trait;
 use fcm_v1::{
     auth::{Authenticator, ServiceAccountKey},
     message::Message,
     Client, Error as FcmError,
 };
+use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
 use revolt_config::config;
 use revolt_database::{events::rabbit::*, Database};
-use revolt_models::v0::{Channel, PushNotification};
 use serde_json::Value;
 
 /// Custom notification data
@@ -31,10 +30,12 @@ pub enum NotificationData {
         image: Option<String>,
     },
     Message {
-        title: String,
+        message: String,
         body: String,
         image: String,
-        tag: String,
+        channel: String,
+        author: String,
+        author_name: String,
     },
     DmCallStartEnd {
         initiator_id: String,
@@ -81,15 +82,19 @@ impl NotificationData {
                 }
             }
             NotificationData::Message {
-                title,
+                message,
                 body,
                 image,
-                tag,
+                channel,
+                author,
+                author_name,
             } => {
-                data.insert("title".to_string(), Value::String(title));
+                data.insert("message".to_string(), Value::String(message));
                 data.insert("body".to_string(), Value::String(body));
                 data.insert("image".to_string(), Value::String(image));
-                data.insert("tag".to_string(), Value::String(tag));
+                data.insert("channel".to_string(), Value::String(channel));
+                data.insert("author".to_string(), Value::String(author));
+                data.insert("author_name".to_string(), Value::String(author_name));
             }
             NotificationData::DmCallStartEnd {
                 initiator_id,
@@ -110,37 +115,31 @@ impl NotificationData {
     }
 }
 
+#[derive(Clone)]
+#[allow(unused)]
 pub struct FcmOutboundConsumer {
     db: Database,
+    authifier_db: authifier::Database,
+    connection: Arc<Connection>,
+    channel: Arc<AMQPChannel>,
     client: Client,
 }
 
-impl FcmOutboundConsumer {
+#[async_trait]
-    fn format_title(&self, notification: &PushNotification) -> String {
+impl Consumer for FcmOutboundConsumer {
-        // ideally this changes depending on context
+    async fn create(
-        // in a server, it would look like "Sendername, #channelname in servername"
+        db: Database,
-        // in a group, it would look like "Sendername in groupname"
+        authifier_db: authifier::Database,
-        // in a dm it should just be "Sendername".
+        connection: Arc<Connection>,
-        // not sure how feasible all those are given the PushNotification object as it currently stands.
+        channel: Arc<AMQPChannel>,
-
+    ) -> Self {
-        #[allow(deprecated)]
-        match &notification.channel {
-            Channel::DirectMessage { .. } => notification.author.clone(),
-            Channel::Group { name, .. } => format!("{}, #{}", notification.author, name),
-            Channel::TextChannel { name, .. } => {
-                format!("{} in #{}", notification.author, name)
-            }
-            _ => "Unknown".to_string(),
-        }
-    }
-}
-
-impl FcmOutboundConsumer {
-    pub async fn new(db: Database) -> Result<FcmOutboundConsumer, &'static str> {
         let config = revolt_config::config().await;
 
-        Ok(FcmOutboundConsumer {
+        Self {
             db,
+            authifier_db,
+            connection,
+            channel,
             client: Client::new(
                 Authenticator::service_account::<&str>(ServiceAccountKey {
                     key_type: Some(config.pushd.fcm.key_type),
@@ -160,33 +159,27 @@ impl FcmOutboundConsumer {
                 false,
                 Duration::from_secs(5),
             ),
-        })
+        }
     }
 
-    async fn consume_event(
+    fn channel(&self) -> &Arc<AMQPChannel> {
-        &mut self,
+        &self.channel
-        _channel: &AmqpChannel,
+    }
-        _deliver: Deliver,
+
-        _basic_properties: BasicProperties,
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        content: Vec<u8>,
+        let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;
-    ) -> Result<()> {
-        let content = String::from_utf8(content)?;
-        let payload: PayloadToService = serde_json::from_str(content.as_str())?;
 
         #[allow(clippy::needless_late_init)]
         let resp: Result<Message, FcmError>;
 
         match payload.notification {
             PayloadKind::FRReceived(alert) => {
-                let name = alert
+                let name = alert.from_user.display_name.clone().unwrap_or_else(|| {
-                    .from_user
+                    format!(
-                    .display_name
-                    .or(Some(format!(
                         "{}#{}",
                         alert.from_user.username, alert.from_user.discriminator
-                    )))
+                    )
-                    .clone()
+                });
-                    .ok_or_else(|| anyhow!("missing name"))?;
 
                 let data = NotificationData::FRReceived {
                     id: alert.from_user.id,
@@ -203,15 +196,12 @@ impl FcmOutboundConsumer {
             }
 
             PayloadKind::FRAccepted(alert) => {
-                let name = alert
+                let name = alert.accepted_user.display_name.clone().unwrap_or_else(|| {
-                    .accepted_user
+                    format!(
-                    .display_name
-                    .or(Some(format!(
                         "{}#{}",
                         alert.accepted_user.username, alert.accepted_user.discriminator
-                    )))
+                    )
-                    .clone()
+                });
-                    .ok_or_else(|| anyhow!("missing name"))?;
 
                 let data = NotificationData::FRAccepted {
                     id: alert.accepted_user.id,
@@ -244,10 +234,12 @@ impl FcmOutboundConsumer {
 
             PayloadKind::MessageNotification(alert) => {
                 let data = NotificationData::Message {
-                    title: self.format_title(&alert),
+                    message: alert.message.id,
                     body: alert.body,
                     image: alert.icon,
-                    tag: alert.tag,
+                    channel: alert.message.channel,
+                    author: alert.message.author,
+                    author_name: alert.author,
                 };
 
                 let msg = Message {
@@ -282,9 +274,8 @@ impl FcmOutboundConsumer {
             }
         }
 
-        if let Err(err) = resp {
+        match resp {
-            match err {
+            Err(FcmError::Auth) => {
-                FcmError::Auth => {
                 if let Err(err) = self
                     .db
                     .remove_push_subscription_by_session_id(&payload.session_id)
@@ -293,32 +284,11 @@ impl FcmOutboundConsumer {
                     revolt_config::capture_error(&err);
                 }
             }
-                err => {
+            res => {
-                    revolt_config::capture_error(&err);
+                res?;
-                }
-            }
             }
+        };
 
         Ok(())
     }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for FcmOutboundConsumer {
-    async fn consume(
-        &mut self,
-        channel: &AmqpChannel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process FCM event: {err:?}");
-        }
-    }
-}

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

@@ -1,6 +1,6 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
-use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};
+use crate::utils::Consumer;
 
 use anyhow::{anyhow, bail, Result};
 use async_trait::async_trait;
@@ -8,46 +8,60 @@ use base64::{
     engine::{self},
     Engine as _,
 };
+use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
 use revolt_database::{events::rabbit::*, util::format_display_name, Database};
 use web_push::{
     ContentEncoding, IsahcWebPushClient, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder,
     WebPushClient, WebPushError, WebPushMessageBuilder,
 };
 
+#[derive(Clone)]
+#[allow(unused)]
 pub struct VapidOutboundConsumer {
     db: Database,
+    authifier_db: authifier::Database,
+    connection: Arc<Connection>,
+    channel: Arc<AMQPChannel>,
     client: IsahcWebPushClient,
-    pkey: Vec<u8>,
+    pkey: Arc<Vec<u8>>,
 }
 
-impl VapidOutboundConsumer {
+#[async_trait]
-    pub async fn new(db: Database) -> Result<VapidOutboundConsumer> {
+impl Consumer for VapidOutboundConsumer {
+    async fn create(
+        db: Database,
+        authifier_db: authifier::Database,
+        connection: Arc<Connection>,
+        channel: Arc<AMQPChannel>,
+    ) -> Self {
         let config = revolt_config::config().await;
 
-        if config.pushd.vapid.private_key.is_empty() | config.pushd.vapid.public_key.is_empty() {
+        if config.pushd.vapid.private_key.is_empty() || config.pushd.vapid.public_key.is_empty() {
-            bail!("no Vapid keys present");
+            panic!("no Vapid keys present");
         }
 
-        let web_push_private_key = engine::general_purpose::URL_SAFE_NO_PAD
+        let web_push_private_key = Arc::new(
+            engine::general_purpose::URL_SAFE_NO_PAD
                 .decode(config.pushd.vapid.private_key)
-            .expect("valid `VAPID_PRIVATE_KEY`");
+                .expect("valid `VAPID_PRIVATE_KEY`"),
+        );
 
-        Ok(VapidOutboundConsumer {
+        Self {
             db,
+            authifier_db,
+            connection,
+            channel,
             client: IsahcWebPushClient::new().unwrap(),
             pkey: web_push_private_key,
-        })
+        }
     }
 
-    async fn consume_event(
+    fn channel(&self) -> &Arc<AMQPChannel> {
-        &mut self,
+        &self.channel
-        _channel: &AmqpChannel,
+    }
-        _deliver: Deliver,
+
-        _basic_properties: BasicProperties,
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        content: Vec<u8>,
+        let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;
-    ) -> Result<()> {
-        let content = String::from_utf8(content)?;
-        let payload: PayloadToService = serde_json::from_str(content.as_str())?;
 
         let subscription = SubscriptionInfo {
             endpoint: payload
@@ -65,10 +79,7 @@ impl VapidOutboundConsumer {
             },
         };
 
-        #[allow(clippy::needless_late_init)]
+        let payload_body = match payload.notification {
-        let payload_body: String;
-
-        match payload.notification {
             PayloadKind::FRReceived(alert) => {
                 let name = alert
                     .from_user
@@ -83,7 +94,7 @@ impl VapidOutboundConsumer {
                 let mut body = HashMap::new();
                 body.insert("body", format!("{} sent you a friend request", name));
 
-                payload_body = serde_json::to_string(&body)?;
+                serde_json::to_string(&body)?
             }
             PayloadKind::FRAccepted(alert) => {
                 let name = alert
@@ -99,14 +110,10 @@ impl VapidOutboundConsumer {
                 let mut body = HashMap::new();
                 body.insert("body", format!("{} accepted your friend request", name));
 
-                payload_body = serde_json::to_string(&body)?;
+                serde_json::to_string(&body)?
-            }
-            PayloadKind::Generic(alert) => {
-                payload_body = serde_json::to_string(&alert)?;
-            }
-            PayloadKind::MessageNotification(alert) => {
-                payload_body = serde_json::to_string(&alert)?;
             }
+            PayloadKind::Generic(alert) => serde_json::to_string(&alert)?,
+            PayloadKind::MessageNotification(alert) => serde_json::to_string(&alert)?,
             PayloadKind::DmCallStartEnd(alert) => {
                 let initiator_name = if let Some(server_id) =
                     self.db.fetch_channel(&alert.channel_id).await?.server()
@@ -132,59 +139,41 @@ impl VapidOutboundConsumer {
                     _ => bail!("Invalid DmCallStart/End channel type"),
                 }
 
-                payload_body = serde_json::to_string(&body)?;
+                serde_json::to_string(&body)?
             }
             PayloadKind::BadgeUpdate(_) => {
                 bail!("Vapid cannot handle badge updates and they should not be sent here.");
             }
-        }
+        };
+
+        let signature = VapidSignatureBuilder::from_pem(
+            std::io::Cursor::new(self.pkey.as_ref()),
+            &subscription,
+        )?
+        .build()?;
 
-        match VapidSignatureBuilder::from_pem(std::io::Cursor::new(&self.pkey), &subscription) {
-            Ok(sig_builder) => match sig_builder.build() {
-                Ok(signature) => {
         let mut builder = WebPushMessageBuilder::new(&subscription);
         builder.set_vapid_signature(signature);
 
         builder.set_payload(ContentEncoding::AesGcm, payload_body.as_bytes());
 
-                    match builder.build() {
+        let msg = builder.build()?;
-                        Ok(msg) => {
+
-                            if let Err(err) = self.client.send(msg).await {
+        match self.client.send(msg).await {
-                                if err == WebPushError::Unauthorized {
+            Err(WebPushError::Unauthorized) => {
-                                    self.db
+                if let Err(err) = self
+                    .db
                     .remove_push_subscription_by_session_id(&payload.session_id)
-                                        .await?;
+                    .await
+                {
+                    revolt_config::capture_error(&err);
                 }
             }
+            res => {
+                res?;
+            }
+        };
 
         Ok(())
     }
-                        Err(err) => Err(err.into()),
-                    }
-                }
-                Err(err) => Err(err.into()),
-            },
-            Err(err) => Err(err.into()),
-        }
-    }
-}
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for VapidOutboundConsumer {
-    async fn consume(
-        &mut self,
-        channel: &AmqpChannel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process Vapid event: {err:?}");
-        }
-    }
 }

crates/daemons/pushd/src/main.rs

@@ -1,17 +1,16 @@
 #[macro_use]
 extern crate log;
 
-use amqprs::{
+use std::sync::Arc;
-    channel::{
+
-        BasicConsumeArguments, Channel, ExchangeDeclareArguments, QueueBindArguments,
+use lapin::{
-        QueueDeclareArguments,
+    options::{BasicConsumeOptions, ExchangeDeclareOptions, QueueBindOptions, QueueDeclareOptions},
-    },
+    types::{AMQPValue, FieldTable},
-    connection::{Connection, OpenConnectionArguments},
+    Channel, Connection, ConnectionProperties,
-    consumer::AsyncConsumer,
-    FieldTable,
 };
 use revolt_config::{config, Settings};
-use tokio::sync::Notify;
+use revolt_database::Database;
+use tokio::signal::ctrl_c;
 
 mod consumers;
 mod utils;
@@ -24,6 +23,8 @@ use consumers::{
     outbound::{apn::ApnsOutboundConsumer, fcm::FcmOutboundConsumer, vapid::VapidOutboundConsumer},
 };
 
+use crate::utils::{Consumer, Delegate};
+
 #[tokio::main(flavor = "multi_thread", worker_threads = 2)]
 async fn main() {
     // Configure logging and environment
@@ -43,7 +44,24 @@ async fn main() {
         panic!("Mongo is not in use, can't connect via authifier!")
     }
 
-    let mut connections: Vec<(Channel, Connection)> = Vec::new();
+    let config = config().await;
+
+    let connection = Arc::new(
+        Connection::connect(
+            &format!(
+                "amqp://{}:{}@{}:{}",
+                &config.rabbit.username,
+                &config.rabbit.password,
+                &config.rabbit.host,
+                &config.rabbit.port,
+            ),
+            ConnectionProperties::default(),
+        )
+        .await
+        .expect("Failed to connect to RabbitMQ"),
+    );
+
+    let mut channels = Vec::new();
 
     // An explainer of how this works:
     // The inbound connections are on separate routing keys, such that they only receive the proper payload
@@ -54,171 +72,178 @@ async fn main() {
     // This'll require some interesting shimming if we need to add more events once this is in prod (different payloads between prod and test),
     // but that sounds like a problem for future us.
 
-    let config = config().await;
+    channels.push(
-
+        make_queue_and_consume::<GenericConsumer>(
-    // inbound: generic
+            &db,
-    connections.push(
+            &authifier,
-        make_queue_and_consume(
+            &connection,
             &config,
             &config.pushd.generic_queue,
-            config.pushd.get_generic_routing_key().as_str(),
+            &config.pushd.get_generic_routing_key(),
             None,
-            GenericConsumer::new(db.clone(), authifier.clone()),
         )
         .await,
     );
 
-    // inbound: messages
+    channels.push(
-    connections.push(
+        make_queue_and_consume::<MessageConsumer>(
-        make_queue_and_consume(
+            &db,
+            &authifier,
+            &connection,
             &config,
             &config.pushd.message_queue,
-            config.pushd.get_message_routing_key().as_str(),
+            &config.pushd.get_message_routing_key(),
             None,
-            MessageConsumer::new(db.clone(), authifier.clone()),
         )
         .await,
     );
 
-    // inbound: FR received
+    channels.push(
-    connections.push(
+        make_queue_and_consume::<FRReceivedConsumer>(
-        make_queue_and_consume(
+            &db,
+            &authifier,
+            &connection,
             &config,
             &config.pushd.fr_received_queue,
-            config.pushd.get_fr_received_routing_key().as_str(),
+            &config.pushd.get_fr_received_routing_key(),
             None,
-            FRReceivedConsumer::new(db.clone(), authifier.clone()),
         )
         .await,
     );
 
-    // inbound: FR accepted
+    channels.push(
-    connections.push(
+        make_queue_and_consume::<FRAcceptedConsumer>(
-        make_queue_and_consume(
+            &db,
+            &authifier,
+            &connection,
             &config,
             &config.pushd.fr_accepted_queue,
-            config.pushd.get_fr_accepted_routing_key().as_str(),
+            &config.pushd.get_fr_accepted_routing_key(),
             None,
-            FRAcceptedConsumer::new(db.clone(), authifier.clone()),
         )
         .await,
     );
 
-    // inbound: Mass Mentions
+    channels.push(
-    connections.push(
+        make_queue_and_consume::<MassMessageConsumer>(
-        make_queue_and_consume(
+            &db,
+            &authifier,
+            &connection,
             &config,
             &config.pushd.mass_mention_queue,
-            config.pushd.get_mass_mention_routing_key().as_str(),
+            &config.pushd.get_mass_mention_routing_key(),
             None,
-            MassMessageConsumer::new(db.clone(), authifier.clone()),
         )
         .await,
     );
 
-    // inbound: Dm Calls
+    channels.push(
-    connections.push(
+        make_queue_and_consume::<DmCallConsumer>(
-        make_queue_and_consume(
+            &db,
+            &authifier,
+            &connection,
             &config,
             &config.pushd.dm_call_queue,
-            config.pushd.get_dm_call_routing_key().as_str(),
+            &config.pushd.get_dm_call_routing_key(),
             None,
-            DmCallConsumer::new(db.clone(), authifier.clone()),
         )
         .await,
     );
 
     if !config.pushd.apn.pkcs8.is_empty() {
-        connections.push(
+        channels.push(
-            make_queue_and_consume(
+            make_queue_and_consume::<ApnsOutboundConsumer>(
+                &db,
+                &authifier,
+                &connection,
                 &config,
                 &config.pushd.apn.queue,
                 &config.pushd.apn.queue,
                 None,
-                ApnsOutboundConsumer::new(db.clone()).await.unwrap(),
             )
             .await,
         );
 
-        let mut table = FieldTable::new();
+        let mut table = FieldTable::default();
-        table.insert("x-message-deduplication".try_into().unwrap(), "true".into());
+        table.insert("x-message-deduplication".into(), AMQPValue::Boolean(true));
 
-        connections.push(
+        channels.push(
-            make_queue_and_consume(
+            make_queue_and_consume::<AckConsumer>(
+                &db,
+                &authifier,
+                &connection,
                 &config,
                 &config.pushd.ack_queue,
                 &config.pushd.ack_queue,
                 Some(table),
-                AckConsumer::new(db.clone(), authifier.clone()),
             )
             .await,
         );
     }
 
     if !config.pushd.fcm.auth_uri.is_empty() {
-        connections.push(
+        channels.push(
-            make_queue_and_consume(
+            make_queue_and_consume::<FcmOutboundConsumer>(
+                &db,
+                &authifier,
+                &connection,
                 &config,
                 &config.pushd.fcm.queue,
                 &config.pushd.fcm.queue,
                 None,
-                FcmOutboundConsumer::new(db.clone()).await.unwrap(),
             )
             .await,
-        )
+        );
     }
 
     if !config.pushd.vapid.public_key.is_empty() {
-        connections.push(
+        channels.push(
-            make_queue_and_consume(
+            make_queue_and_consume::<VapidOutboundConsumer>(
+                &db,
+                &authifier,
+                &connection,
                 &config,
                 &config.pushd.vapid.queue,
                 &config.pushd.vapid.queue,
                 None,
-                VapidOutboundConsumer::new(db.clone()).await.unwrap(),
             )
             .await,
-        )
+        );
     }
 
-    let guard = Notify::new();
+    ctrl_c().await.unwrap();
-    guard.notified().await;
 
-    for (channel, conn) in connections {
+    for channel in channels {
-        channel.close().await.expect("Unable to close channel");
+        let _ = channel.close(0, "close".into()).await;
-        conn.close().await.expect("Unable to close connection");
     }
 }
 
 async fn make_queue_and_consume<F>(
+    db: &Database,
+    authifier_db: &authifier::Database,
+    connection: &Arc<Connection>,
     config: &Settings,
     queue_name: &str,
     routing_key: &str,
     queue_args: Option<FieldTable>,
-    consumer: F,
+) -> Arc<Channel>
-) -> (Channel, Connection)
 where
-    F: AsyncConsumer + Send + 'static,
+    F: Consumer,
 {
-    let connection = Connection::open(&OpenConnectionArguments::new(
+    let channel = Arc::new(connection.create_channel().await.unwrap());
-        &config.rabbit.host,
-        config.rabbit.port,
-        &config.rabbit.username,
-        &config.rabbit.password,
-    ))
-    .await
-    .unwrap();
-
-    let channel = connection.open_channel(None).await.unwrap();
 
     channel
         .exchange_declare(
-            ExchangeDeclareArguments::new(&config.pushd.exchange, "direct")
+            config.pushd.exchange.clone().into(),
-                .durable(true)
+            lapin::ExchangeKind::Direct,
-                .finish(),
+            ExchangeDeclareOptions {
+                durable: true,
+                ..Default::default()
+            },
+            FieldTable::default(),
         )
         .await
-        .expect("Failed to declare pushd exchange");
+        .expect("Failed to declare exchange");
 
     let mut queue_name = queue_name.to_string();
 
@@ -230,35 +255,59 @@ where
 
     let queue_name = queue_name.as_str();
 
-    let mut args = QueueDeclareArguments::new(queue_name);
+    let args = QueueDeclareOptions {
-    args.durable(true);
+        durable: true,
-
+        ..Default::default()
-    if let Some(arg) = queue_args {
+    };
-        args.arguments(arg);
-    }
-
-    let args = args.finish();
-    _ = channel.queue_declare(args).await.unwrap().unwrap();
 
     channel
-        .queue_bind(QueueBindArguments::new(
+        .queue_declare(queue_name.into(), args, queue_args.unwrap_or_default())
-            queue_name,
+        .await
-            &config.pushd.exchange,
+        .unwrap();
-            routing_key,
+
-        ))
+    channel
+        .queue_bind(
+            queue_name.into(),
+            config.pushd.exchange.clone().into(),
+            routing_key.into(),
+            QueueBindOptions::default(),
+            FieldTable::default(),
+        )
         .await
         .expect(
             "This probably means the revolt.notifications exchange does not exist in rabbitmq!",
         );
 
-    let args = BasicConsumeArguments::new(queue_name, "")
+    let consumer = channel
-        .manual_ack(false)
+        .basic_consume(
-        .finish();
+            queue_name.into(),
-
+            "".into(),
-    let routing_key = channel.basic_consume(consumer, args).await.unwrap();
+            BasicConsumeOptions {
+                no_ack: true,
+                ..Default::default()
+            },
+            FieldTable::default(),
+        )
+        .await
+        .unwrap();
     info!(
         "Consuming routing key {} as queue {}, tag {}",
-        routing_key, queue_name, routing_key
+        routing_key,
+        queue_name,
+        consumer.tag()
     );
-    (channel, connection)
+
+    let delegate = Delegate(
+        F::create(
+            db.clone(),
+            authifier_db.clone(),
+            connection.clone(),
+            channel.clone(),
+        )
+        .await,
+    );
+
+    consumer.set_delegate(delegate);
+
+    channel
 }

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

@@ -0,0 +1,91 @@
+use std::{
+    future::{ready, Future},
+    pin::Pin,
+    sync::Arc,
+};
+
+use anyhow::Result;
+use async_trait::async_trait;
+use lapin::{
+    message::{Delivery, DeliveryResult},
+    options::BasicPublishOptions,
+    BasicProperties, Channel, Connection, ConsumerDelegate, Error as AMQPError,
+};
+use log::debug;
+use revolt_database::Database;
+
+#[async_trait]
+pub trait Consumer: Clone + Send + Sync + 'static {
+    async fn create(
+        db: Database,
+        authifier_db: authifier::Database,
+        connection: Arc<Connection>,
+        channel: Arc<Channel>,
+    ) -> Self;
+    fn channel(&self) -> &Arc<Channel>;
+    async fn consume(&self, delivery: Delivery) -> Result<()>;
+
+    async fn publish_message_with_options(
+        &self,
+        payload: &[u8],
+        exchange: &str,
+        routing_key: &str,
+        options: BasicPublishOptions,
+        properties: BasicProperties,
+    ) -> Result<(), AMQPError> {
+        let channel = self.channel();
+
+        channel
+            .basic_publish(
+                exchange.into(),
+                routing_key.into(),
+                options,
+                payload,
+                properties,
+            )
+            .await?;
+        debug!("Sent message to queue for target {}", routing_key);
+
+        Ok(())
+    }
+
+    async fn publish_message(
+        &self,
+        payload: &[u8],
+        exchange: &str,
+        routing_key: &str,
+    ) -> Result<(), AMQPError> {
+        self.publish_message_with_options(
+            payload,
+            exchange,
+            routing_key,
+            BasicPublishOptions::default(),
+            BasicProperties::default(),
+        )
+        .await
+    }
+}
+
+pub struct Delegate<C: Consumer>(pub C);
+
+impl<C: Consumer> ConsumerDelegate for Delegate<C> {
+    fn on_new_delivery(
+        &self,
+        delivery: DeliveryResult,
+    ) -> Pin<Box<dyn Future<Output = ()> + Send>> {
+        match delivery {
+            Ok(Some(delivery)) => {
+                let consumer = self.0.clone();
+
+                Box::pin(async move {
+                    if let Err(e) = consumer.consume(delivery).await {
+                        revolt_config::capture_anyhow(&e);
+                        log::error!("{e:?}");
+                    };
+                })
+            }
+            Ok(None) => Box::pin(ready(())),
+            Err(e) => Box::pin(async move { log::error!("Received bad delivery: {e:?}") }),
+        }
+    }
+}

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

@@ -1,2 +1,5 @@
 mod renderer;
+mod consumer;
+
 pub use renderer::render_notification_content;
+pub use consumer::{Consumer, Delegate};

crates/daemons/voice-ingress/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-voice-ingress"
-version = "0.13.4"
+version = "0.13.7"
 license = "AGPL-3.0-or-later"
 edition = "2021"
 publish = false
@@ -44,6 +44,3 @@ revolt-permissions = { workspace = true }
 livekit-api = { workspace = true }
 livekit-protocol = { workspace = true }
 livekit-runtime = { workspace = true, features = ["tokio"] }
-
-# RabbitMQ
-amqprs = { workspace = true }

crates/delta/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-delta"
-version = "0.13.4"
+version = "0.13.7"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <paulmakles@gmail.com>"]
 edition = "2018"
@@ -64,7 +64,7 @@ schemars = { workspace = true }
 revolt_rocket_okapi = { workspace = true, features = ["swagger"] }
 
 # rabbit
-amqprs = { workspace = true }
+lapin = { workspace = true, features = ["tokio"] }
 
 # core
 authifier = { workspace = true }

crates/delta/src/main.rs

@@ -9,8 +9,7 @@ pub mod routes;
 pub mod util;
 
 use revolt_config::config;
-use revolt_database::events::client::EventV1;
+use revolt_database::{AMQP, events::client::EventV1};
-use revolt_database::AMQP;
 use revolt_ratelimits::rocket as ratelimiter;
 use rocket::{Build, Rocket};
 use rocket_cors::{AllowedOrigins, CorsOptions};
@@ -18,10 +17,6 @@ use rocket_prometheus::PrometheusMetrics;
 use std::net::Ipv4Addr;
 use std::str::FromStr;
 
-use amqprs::{
-    channel::ExchangeDeclareArguments,
-    connection::{Connection, OpenConnectionArguments},
-};
 use async_std::channel::unbounded;
 use authifier::AuthifierEvent;
 use revolt_database::voice::VoiceClient;
@@ -36,7 +31,6 @@ pub async fn web() -> Rocket<Build> {
 
     // Setup database
     let db = revolt_database::DatabaseInfo::Auto.connect().await.unwrap();
-    log::info!("database_here {db:?}");
     db.migrate_database().await.unwrap();
 
     // Setup Authifier event channel
@@ -96,33 +90,8 @@ pub async fn web() -> Rocket<Build> {
     // Voice handler
     let voice_client = VoiceClient::new(config.api.livekit.nodes.clone());
     // Configure Rabbit
-    let connection = Connection::open(&OpenConnectionArguments::new(
-        &config.rabbit.host,
-        config.rabbit.port,
-        &config.rabbit.username,
-        &config.rabbit.password,
-    ))
-    .await
-    .expect("Failed to connect to RabbitMQ");
 
-    let channel = connection
+    let amqp = AMQP::new_auto().await;
-        .open_channel(None)
-        .await
-        .expect("Failed to open RabbitMQ channel");
-
-    channel
-        .exchange_declare(
-            ExchangeDeclareArguments::new(&config.pushd.exchange, "direct")
-                .durable(true)
-                .finish(),
-        )
-        .await
-        .expect("Failed to declare exchange");
-
-    let mut amqp = AMQP::new(connection, channel);
-    // amqp.configure_channels()
-    //     .await
-    //     .expect("Failed to configure channels");
 
     // Launch background task workers
     revolt_database::tasks::start_workers(db.clone(), amqp.clone());
@@ -152,6 +121,7 @@ pub async fn web() -> Rocket<Build> {
             limits: rocket::data::Limits::default().limit("string", 5.megabytes()),
             address: Ipv4Addr::new(0, 0, 0, 0).into(),
             port: 14702,
+            ip_header: Some("X-Forwarded-For".into()),
             ..Default::default()
         })
 }

crates/delta/src/routes/channels/message_send.rs

@@ -1,12 +1,14 @@
 use std::time::Duration;
 
 use redis_kiss::{get_connection, redis, AsyncCommands};
+use revolt_database::events::client::EventV1;
 use revolt_database::util::permissions::DatabasePermissionQuery;
 use revolt_database::{
     util::idempotency::IdempotencyKey, util::reference::Reference, Database, User,
 };
 use revolt_database::{Channel, Interactions, Message, AMQP};
 use revolt_models::v0;
+use revolt_models::v0::ChannelSlowmode;
 use revolt_permissions::PermissionQuery;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
@@ -84,6 +86,16 @@ pub async fn message_send(
                         .await
                         .unwrap_or(None);
 
+                    if set_result.is_some() {
+                        let idx_key = format!("slowmode_idx:{}", user.id);
+                        conn.sadd::<_, _, ()>(&idx_key, channel_id.as_str())
+                            .await
+                            .ok();
+                        conn.expire::<_, ()>(&idx_key, *channel_slowmode as usize)
+                            .await
+                            .ok();
+                    }
+
                     // If `set_result` is None, the `NX` condition failed because the key already exists.
                     // This means the user is currently in slowmode.
                     if set_result.is_none() {
@@ -92,10 +104,29 @@ pub async fn message_send(
 
                         // Redis returns positive integers for valid TTLs
                         if ttl > 0 {
+                            EventV1::UserSlowmodes {
+                                slowmodes: vec![ChannelSlowmode {
+                                    channel_id: channel_id.to_string(),
+                                    duration: *channel_slowmode,
+                                    retry_after: ttl as u64,
+                                }],
+                            }
+                            .private(user.id.clone())
+                            .await;
                             return Err(create_error!(InSlowmode {
                                 retry_after: ttl as u64
                             }));
                         }
+                    } else {
+                        EventV1::UserSlowmodes {
+                            slowmodes: vec![ChannelSlowmode {
+                                channel_id: channel_id.to_string(),
+                                duration: *channel_slowmode,
+                                retry_after: *channel_slowmode,
+                            }],
+                        }
+                        .private(user.id.clone())
+                        .await;
                     }
                 }
                 // If Redis connection fails, just skip the slowmode check

crates/delta/src/util/test.rs

@@ -4,7 +4,7 @@ use authifier::{
 };
 use futures::StreamExt;
 use rand::Rng;
-use redis_kiss::redis::aio::PubSub;
+use redis_kiss::{redis::aio::PubSub};
 use revolt_database::{
     events::client::EventV1, Channel, Database, Member, Message, PartialRole, Server, User, AMQP,
 };
@@ -25,8 +25,6 @@ pub struct TestHarness {
 
 impl TestHarness {
     pub async fn new() -> TestHarness {
-        let config = revolt_config::config().await;
-
         let client = Client::tracked(crate::web().await)
             .await
             .expect("valid rocket instance");
@@ -49,19 +47,7 @@ impl TestHarness {
             .expect("`Authifier`")
             .clone();
 
-        let connection = amqprs::connection::Connection::open(
+        let amqp = AMQP::new_auto().await;
-            &amqprs::connection::OpenConnectionArguments::new(
-                &config.rabbit.host,
-                config.rabbit.port,
-                &config.rabbit.username,
-                &config.rabbit.password,
-            ),
-        )
-        .await
-        .unwrap();
-        let channel = connection.open_channel(None).await.unwrap();
-
-        let amqp = AMQP::new(connection, channel);
 
         TestHarness {
             client,

crates/services/autumn/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-autumn"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 publish = false

crates/services/gifbox/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-gifbox"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 publish = false

crates/services/january/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-january"
-version = "0.13.4"
+version = "0.13.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 publish = false

crates/services/january/src/requests.rs

@@ -4,6 +4,7 @@ use mime::Mime;
 use pdk_ip_filter_lib::IpFilter;
 use regex::Regex;
 use reqwest::{
+    dns::{Addrs, Name, Resolve},
     header::{self, CONTENT_TYPE},
     redirect, Client, Response,
 };
@@ -11,6 +12,7 @@ use revolt_config::{config, report_internal_error};
 use revolt_files::{create_thumbnail, decode_image, image_size_vec, is_valid_image, video_size};
 use revolt_models::v0::{Embed, Image, ImageSize, Video};
 use revolt_result::{create_error, Error, Result, ToRevoltError};
+use std::net::{IpAddr, SocketAddr};
 use std::{
     io::{Cursor, Write},
     str::FromStr,
@@ -21,6 +23,7 @@ use url::{Host, Url};
 lazy_static! {
     /// Request client
     static ref CLIENT: Client = reqwest::Client::builder()
+        .dns_resolver(CachedDnsResolver {})
         .timeout(Duration::from_secs(10)) // TODO config
         .connect_timeout(Duration::from_secs(5)) // TODO config
         .redirect(redirect::Policy::none())
@@ -58,18 +61,73 @@ lazy_static! {
         .time_to_live(Duration::from_secs(60)) // For up to 1 minute
         .build();
 
+    static ref DNS_CACHE: moka::future::Cache<String, Vec<SocketAddr>> = moka::future::Cache::builder()
+        .max_capacity(10_000)
+        .time_to_idle(Duration::from_secs(30))
+        .build();
+
     static ref IP_BLOCKLIST: IpFilter = IpFilter::block(&[
+        "0.0.0.0/8",
         "10.0.0.0/8",
         "192.168.0.0/16",
         "127.0.0.0/8",
         "172.16.0.0/12",
         "169.254.0.0/16",
         "::1",
-        "fc00::/7"
+        "fc00::/7",
         ]
     ).unwrap();
 }
 
+#[derive(Clone)]
+pub struct IPRequest {
+    url: Url,
+    ip: IpAddr,
+    pub blocked: bool,
+}
+
+impl From<IPRequest> for Url {
+    fn from(value: IPRequest) -> Self {
+        let mut url = value.url.clone();
+        url.set_host(Some(&value.ip.to_string()))
+            .map(|_| url)
+            .unwrap_or(value.url)
+    }
+}
+
+struct CachedDnsResolver {}
+
+impl reqwest::dns::Resolve for CachedDnsResolver {
+    fn resolve(&self, name: Name) -> reqwest::dns::Resolving {
+        Box::pin(async move {
+            {
+                if let Some(addrs) = DNS_CACHE.get(&name.as_str().to_string()).await {
+                    let resp: Addrs = Box::new(addrs.clone().into_iter());
+                    return Ok(resp);
+                }
+            }
+
+            let mut lookup = name.as_str().to_string();
+            if !lookup.contains(":") {
+                lookup += ":0";
+            }
+
+            let fallback: Vec<SocketAddr> = tokio::net::lookup_host(&lookup)
+                .await
+                .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) })?
+                .collect();
+
+            {
+                DNS_CACHE
+                    .insert(name.as_str().to_string().clone(), fallback.clone())
+                    .await;
+                let addrs: Addrs = Box::new(fallback.clone().into_iter());
+                Ok(addrs)
+            }
+        })
+    }
+}
+
 /// Information about a successful request
 pub struct Request {
     response: Response,
@@ -275,7 +333,12 @@ impl Request {
         let mut url = url;
         let url_host_str = url.host_str().ok_or(create_error!(ProxyError))?.to_string();
 
-        Request::url_is_blacklisted(&url).await?;
+        let mut blocker = Request::url_is_blacklisted(&url).await?;
+
+        if blocker.blocked {
+            return Err(create_error!(InvalidOperation));
+        }
+
         let mut redirect_count = 0;
 
         loop {
@@ -304,9 +367,13 @@ impl Request {
                     let location = location.to_str().map_err(|_| create_error!(ProxyError))?;
                     url = Url::from_str(location).to_internal_error()?;
 
-                    if !Request::url_is_blacklisted(&url).await? {
+                    blocker = Request::url_is_blacklisted(&url).await?;
-                        continue;
+
+                    if blocker.blocked {
+                        return Err(create_error!(InvalidOperation));
                     }
+
+                    continue;
                 } else {
                     return Err(create_error!(ProxyError));
                 }
@@ -351,17 +418,25 @@ impl Request {
         Ok(Request::exists(proper_url).await)
     }
 
-    pub async fn url_is_blacklisted(url: &Url) -> Result<bool> {
+    pub async fn url_is_blacklisted(url: &Url) -> Result<IPRequest> {
+        let resolved_address: IpAddr;
+
         if let Some(host) = url.host() {
             match host {
                 Host::Ipv4(ipv4) => {
-                    let url_str = ipv4.to_string();
+                    resolved_address = ipv4.into();
-                    if !IP_BLOCKLIST.is_allowed(&url_str) {
+                    if !IP_BLOCKLIST.is_allowed(&ipv4.to_string()) {
+                        return Err(create_error!(InvalidOperation));
+                    }
+                }
+                Host::Ipv6(ipv6) => {
+                    resolved_address = ipv6.into();
+                    if !IP_BLOCKLIST.is_allowed(&ipv6.to_string()) {
                         return Err(create_error!(InvalidOperation));
                     }
                 }
                 Host::Domain(domain) => {
-                    let mut domain = domain.to_string();
+                    let domain = domain.to_string();
 
                     let config = config().await;
 
@@ -372,14 +447,22 @@ impl Request {
                         return Err(create_error!(InvalidOperation));
                     }
 
-                    if !domain.contains(":") {
-                        domain += ":80";
-                    }
-
                     // Second step: resolve the IP and check the blocklist
-                    if let Ok(mut resolved_ip) = tokio::net::lookup_host(domain.clone()).await {
+                    let resolver = CachedDnsResolver {};
+                    if let Ok(mut resolved_ip) = resolver
+                        .resolve(
+                            Name::from_str(&domain)
+                                .map_err(|_| create_error!(ProxyError))
+                                .unwrap(),
+                        )
+                        .await
+                    {
                         if let Some(resolved_ip) = resolved_ip.next() {
-                            if !IP_BLOCKLIST.is_allowed(&resolved_ip.ip().to_string()) {
+                            resolved_address = resolved_ip.ip();
+                            let resolved_string = resolved_address.to_string();
+                            if !IP_BLOCKLIST.is_allowed(&resolved_string)
+                                || resolved_string.contains("::ffff:")
+                            {
                                 return Err(create_error!(InvalidOperation));
                             }
                         } else {
@@ -389,10 +472,15 @@ impl Request {
                         return Err(create_error!(ProxyError));
                     }
                 }
-                _ => (),
             }
+        } else {
+            return Err(create_error!(ProxyError));
         };
 
-        Ok(false)
+        Ok(IPRequest {
+            url: url.clone(),
+            ip: resolved_address,
+            blocked: false,
+        })
     }
 }

version.txt

@@ -1 +1 @@
-0.13.4
+0.13.7