fix: minio wants the default region

fix: avoid stack overflow in database tests
Signed-off-by: Infiland <ljubica.citydesign@gmail.com>
2026-07-07 03:35:59 +00:00 · 2026-05-08 16:47:20 -07:00 · 2026-05-08 16:00:27 -07:00
62 changed files with 1501 additions and 1538 deletions
--- a/.github/workflows/docker-cleanup.yaml
+++ b/.github/workflows/docker-cleanup.yaml
@@ -1,46 +0,0 @@
-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
--- a/.github/workflows/docker.yaml
+++ b/.github/workflows/docker.yaml
@@ -5,6 +5,8 @@ on:
    tags:
      - "*"
  pull_request:
+    paths:
+      - "Dockerfile"
  workflow_dispatch:

 permissions:
@@ -17,9 +19,9 @@ concurrency:

 jobs:
  base:
-    name: Test base image build (fork)
+    name: Test base image build
    runs-on: arc-runner-set
-    if: github.event_name == 'pull_request' && github.event.pull_request.head.repo.fork
+    if: github.event_name == 'pull_request'
    steps:
      # Configure build environment
      - name: Checkout
@@ -40,7 +42,7 @@ jobs:

  publish:
    runs-on: arc-runner-set
-    if: ${{ github.event_name != 'pull_request' || !github.event.pull_request.head.repo.fork }}
+    if: github.event_name != 'pull_request'
    name: Publish Docker images
    steps:
      # Configure build environment
@@ -57,15 +59,6 @@ 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
@@ -73,9 +66,7 @@ jobs:
          context: .
          push: true
          platforms: linux/amd64,linux/arm64
-          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
+          tags: ghcr.io/${{ github.repository_owner }}/base:latest

      # stoatchat/api
      - name: Docker meta
@@ -93,7 +84,7 @@ jobs:
          file: crates/delta/Dockerfile
          tags: ${{ steps.meta-delta.outputs.tags }}
          build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
          labels: ${{ steps.meta-delta.outputs.labels }}

      # stoatchat/events
@@ -112,7 +103,7 @@ jobs:
          file: crates/bonfire/Dockerfile
          tags: ${{ steps.meta-bonfire.outputs.tags }}
          build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
          labels: ${{ steps.meta-bonfire.outputs.labels }}

      # stoatchat/file-server
@@ -131,7 +122,7 @@ jobs:
          file: crates/services/autumn/Dockerfile
          tags: ${{ steps.meta-autumn.outputs.tags }}
          build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
          labels: ${{ steps.meta-autumn.outputs.labels }}

      # stoatchat/proxy
@@ -150,7 +141,7 @@ jobs:
          file: crates/services/january/Dockerfile
          tags: ${{ steps.meta-january.outputs.tags }}
          build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
          labels: ${{ steps.meta-january.outputs.labels }}

      # stoatchat/gifbox
@@ -169,7 +160,7 @@ jobs:
          file: crates/services/gifbox/Dockerfile
          tags: ${{ steps.meta-gifbox.outputs.tags }}
          build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
          labels: ${{ steps.meta-gifbox.outputs.labels }}

      # stoatchat/crond
@@ -188,7 +179,7 @@ jobs:
          file: crates/daemons/crond/Dockerfile
          tags: ${{ steps.meta-crond.outputs.tags }}
          build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
          labels: ${{ steps.meta-crond.outputs.labels }}

      # stoatchat/pushd
@@ -207,7 +198,7 @@ jobs:
          file: crates/daemons/pushd/Dockerfile
          tags: ${{ steps.meta-pushd.outputs.tags }}
          build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
          labels: ${{ steps.meta-pushd.outputs.labels }}

      # stoatchat/voice-ingress
@@ -226,5 +217,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:${{ steps.base.outputs.tag }}
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
          labels: ${{ steps.meta-voice-ingress.outputs.labels }}
--- a/.github/workflows/publish-crates.yml
+++ b/.github/workflows/publish-crates.yml
@@ -21,6 +21,4 @@ jobs:
          github-token: ${{ secrets.GITHUB_TOKEN }}

      - name: Publish
-        env:
-          CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
        run: mise publish --workspace
--- a/.github/workflows/rust.yaml
+++ b/.github/workflows/rust.yaml
@@ -37,7 +37,6 @@ jobs:
      - name: Reference Test
        env:
          TEST_DB: REFERENCE
-        continue-on-error: ${{ github.ref_name == 'main' }}
        run: |
          mise test

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

--- a/.release-please-manifest.json
+++ b/.release-please-manifest.json
@@ -1,3 +1,3 @@
 {
-    ".": "0.13.7"
+    ".": "0.12.1"
 }
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,97 +1,5 @@
 # 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)
-
-
-### Bug Fixes
-
-* add TLS feature to livekit-api crate ([#753](https://github.com/stoatchat/stoatchat/issues/753)) ([6cfee1f](https://github.com/stoatchat/stoatchat/commit/6cfee1f601c1e084df7c8f1e7a5e8a560d1dd514))
-
-## [0.13.3](https://github.com/stoatchat/stoatchat/compare/v0.13.2...v0.13.3) (2026-05-15)
-
-
-### Bug Fixes
-
-* don't automatically set up rabbitmq in delta ([#749](https://github.com/stoatchat/stoatchat/issues/749)) ([7647cfc](https://github.com/stoatchat/stoatchat/commit/7647cfc8d93aba99f5faef13eb3d970097540d76))
-* don't declare queues which seem to cause the backend to crash in prod ([7647cfc](https://github.com/stoatchat/stoatchat/commit/7647cfc8d93aba99f5faef13eb3d970097540d76))
-
-## [0.13.2](https://github.com/stoatchat/stoatchat/compare/v0.13.1...v0.13.2) (2026-05-11)
-
-
-### Bug Fixes
-
-* update default exchange to `revolt.default` ([#746](https://github.com/stoatchat/stoatchat/issues/746)) ([fcb8091](https://github.com/stoatchat/stoatchat/commit/fcb8091cd7a00d7f26c798daa33aae4b923b2a8b))
-
-## [0.13.1](https://github.com/stoatchat/stoatchat/compare/v0.13.0...v0.13.1) (2026-05-10)
-
-
-### Bug Fixes
-
-* amqprs startup bug ([#744](https://github.com/stoatchat/stoatchat/issues/744)) ([1100eaf](https://github.com/stoatchat/stoatchat/commit/1100eaf46f849f2509ae01ac497556ca33bde778))
-
-## [0.13.0](https://github.com/stoatchat/stoatchat/compare/v0.12.1...v0.13.0) (2026-05-08)
-
-
-### Features
-
-* add embed support for YouTube Shorts ([#734](https://github.com/stoatchat/stoatchat/issues/734)) ([d46c7f7](https://github.com/stoatchat/stoatchat/commit/d46c7f7f3c04524c0639c3e0a122626f8e0b3bf7))
-* add emoji rename endpoint ([#714](https://github.com/stoatchat/stoatchat/issues/714)) ([23ad135](https://github.com/stoatchat/stoatchat/commit/23ad1359834bb7d07a460b8678d6a6ebffc73eb0))
-* add legal links to root payload ([#733](https://github.com/stoatchat/stoatchat/issues/733)) ([21d8201](https://github.com/stoatchat/stoatchat/commit/21d82018cf84ab0fdd10613d254b9562aea8eea3))
-* add role icon support ([#724](https://github.com/stoatchat/stoatchat/issues/724)) ([841985d](https://github.com/stoatchat/stoatchat/commit/841985d3b994df1c6eefab2fc7ecbd77ab22c493))
-* Add webhook endpoints for editing and deleting messages ([#682](https://github.com/stoatchat/stoatchat/issues/682)) ([6f3441c](https://github.com/stoatchat/stoatchat/commit/6f3441cf4acac2a8e6e1bf07a279a153b80f7956))
-* automatically sanitise usernames on create/update ([#689](https://github.com/stoatchat/stoatchat/issues/689)) ([e937697](https://github.com/stoatchat/stoatchat/commit/e93769786c7669485a659ee471630740d3cea702))
-* blacklist private ip ranges and add january domain blocklist ([#731](https://github.com/stoatchat/stoatchat/issues/731)) ([6b41db9](https://github.com/stoatchat/stoatchat/commit/6b41db984bb491b2e58324309cc70d8c14e0b814))
-* Rewrite acks ([#741](https://github.com/stoatchat/stoatchat/issues/741)) ([ab5bd47](https://github.com/stoatchat/stoatchat/commit/ab5bd47a39ee889de0b5ae6e7b560620853daead))
-
-
-### Bug Fixes
-
-* add new_user_hours to configuration limits ([#729](https://github.com/stoatchat/stoatchat/issues/729)) ([279f5d5](https://github.com/stoatchat/stoatchat/commit/279f5d5fd7af2df55902c706859ec07f569cdb1e))
-* add reconnection policy to Redis subscriber to prevent ghost state ([#708](https://github.com/stoatchat/stoatchat/issues/708)) ([057f2bb](https://github.com/stoatchat/stoatchat/commit/057f2bb8b359f8b942741a30ff54eeb8fbe3e0b1))
-* docker compose file had personal url in it ([#742](https://github.com/stoatchat/stoatchat/issues/742)) ([0719985](https://github.com/stoatchat/stoatchat/commit/0719985ac5636590f91e6f9ec4b68f3eded70c13))
-* don't strip ICC from exif ([#735](https://github.com/stoatchat/stoatchat/issues/735)) ([d76a711](https://github.com/stoatchat/stoatchat/commit/d76a71141f3e508f6308ba52fa28eaeb56fb3438))
-* dont send notification in fcm ([#721](https://github.com/stoatchat/stoatchat/issues/721)) ([89171e9](https://github.com/stoatchat/stoatchat/commit/89171e9bd0f15711157e78c6eec0fe7b480de93a))
-* encode filenames in redirects ([#737](https://github.com/stoatchat/stoatchat/issues/737)) ([9fd7128](https://github.com/stoatchat/stoatchat/commit/9fd7128f800badbd184baf943d4f799e601201e4))
-* january ip redirects & domain resolver ([#738](https://github.com/stoatchat/stoatchat/issues/738)) ([356491e](https://github.com/stoatchat/stoatchat/commit/356491e934b274f9e895df883dd63ef0b3123510))
-* update message length validation to remove upper limit ([#723](https://github.com/stoatchat/stoatchat/issues/723)) ([ed4fd5e](https://github.com/stoatchat/stoatchat/commit/ed4fd5ebfe6d0ea534a0898da4afdc1f4e2cd6c5))
-* use correct response for NoEffect errors ([#732](https://github.com/stoatchat/stoatchat/issues/732)) ([5378cd2](https://github.com/stoatchat/stoatchat/commit/5378cd22b4c7d85f44c31a6af0dda00941b80d5c))
-
 ## [0.12.1](https://github.com/stoatchat/stoatchat/compare/v0.12.0...v0.12.1) (2026-04-10)


--- a/Cargo.lock
+++ b/Cargo.lock
@@ -174,6 +174,31 @@ 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"
@@ -545,7 +570,7 @@ dependencies = [
 "futures-util",
 "log",
 "pin-project-lite 0.2.17",
- "tungstenite 0.17.3",
+ "tungstenite",
 ]

 [[package]]
@@ -5023,14 +5048,11 @@ dependencies = [
 "prost",
 "rand 0.9.2",
 "reqwest 0.12.28",
- "rustls-native-certs 0.6.3",
 "scopeguard",
 "serde",
 "serde_json",
 "sha2",
 "thiserror 2.0.18",
- "tokio-rustls 0.24.1",
- "tokio-tungstenite",
 "url",
 ]

@@ -7397,22 +7419,16 @@ dependencies = [
 "http-body 1.0.1",
 "http-body-util",
 "hyper 1.9.0",
- "hyper-rustls 0.27.7",
 "hyper-util",
 "js-sys",
 "log",
 "percent-encoding",
 "pin-project-lite 0.2.17",
- "quinn",
- "rustls 0.23.37",
- "rustls-native-certs 0.8.3",
- "rustls-pki-types",
 "serde",
 "serde_json",
 "serde_urlencoded",
 "sync_wrapper 1.0.2",
 "tokio 1.51.0",
- "tokio-rustls 0.26.4",
 "tower",
 "tower-http 0.6.8",
 "tower-service",
@@ -7488,7 +7504,7 @@ dependencies = [

 [[package]]
 name = "revolt-autumn"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "axum",
 "axum-macros",
@@ -7529,7 +7545,7 @@ dependencies = [

 [[package]]
 name = "revolt-bonfire"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "async-channel 2.5.0",
 "async-std",
@@ -7560,7 +7576,7 @@ dependencies = [

 [[package]]
 name = "revolt-coalesced"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "indexmap 2.13.1",
 "lru",
@@ -7569,7 +7585,7 @@ dependencies = [

 [[package]]
 name = "revolt-config"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "async-std",
 "cached",
@@ -7586,7 +7602,7 @@ dependencies = [

 [[package]]
 name = "revolt-crond"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "futures-lite",
 "iso8601-timestamp",
@@ -7606,8 +7622,9 @@ dependencies = [

 [[package]]
 name = "revolt-database"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
+ "amqprs",
 "async-lock 2.8.0",
 "async-recursion",
 "async-std",
@@ -7622,7 +7639,6 @@ dependencies = [
 "indexmap 2.13.1",
 "isahc",
 "iso8601-timestamp",
- "lapin",
 "linkify",
 "livekit-api",
 "livekit-protocol",
@@ -7657,8 +7673,9 @@ dependencies = [

 [[package]]
 name = "revolt-delta"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
+ "amqprs",
 "async-channel 2.5.0",
 "async-std",
 "authifier",
@@ -7668,7 +7685,6 @@ dependencies = [
 "futures",
 "impl_ops",
 "iso8601-timestamp",
- "lapin",
 "lettre",
 "linkify",
 "livekit-api",
@@ -7706,7 +7722,7 @@ dependencies = [

 [[package]]
 name = "revolt-files"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "aes-gcm",
 "anyhow",
@@ -7734,7 +7750,7 @@ dependencies = [

 [[package]]
 name = "revolt-gifbox"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "axum",
 "axum-extra",
@@ -7757,7 +7773,7 @@ dependencies = [

 [[package]]
 name = "revolt-january"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "async-recursion",
 "axum",
@@ -7787,7 +7803,7 @@ dependencies = [

 [[package]]
 name = "revolt-models"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "indexmap 2.13.1",
 "iso8601-timestamp",
@@ -7806,14 +7822,14 @@ dependencies = [

 [[package]]
 name = "revolt-parser"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "logos",
 ]

 [[package]]
 name = "revolt-permissions"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "async-std",
 "async-trait",
@@ -7828,7 +7844,7 @@ dependencies = [

 [[package]]
 name = "revolt-presence"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "async-std",
 "log",
@@ -7840,8 +7856,9 @@ dependencies = [

 [[package]]
 name = "revolt-pushd"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
+ "amqprs",
 "anyhow",
 "async-trait",
 "authifier",
@@ -7849,7 +7866,6 @@ dependencies = [
 "fcm_v1",
 "isahc",
 "iso8601-timestamp",
- "lapin",
 "log",
 "pretty_env_logger",
 "redis-kiss",
@@ -7871,7 +7887,7 @@ dependencies = [

 [[package]]
 name = "revolt-ratelimits"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "async-trait",
 "authifier",
@@ -7888,7 +7904,7 @@ dependencies = [

 [[package]]
 name = "revolt-result"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
 "axum",
 "log",
@@ -7904,8 +7920,9 @@ dependencies = [

 [[package]]
 name = "revolt-voice-ingress"
-version = "0.13.7"
+version = "0.12.1"
 dependencies = [
+ "amqprs",
 "async-std",
 "chrono",
 "futures",
@@ -8961,6 +8978,15 @@ 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"
@@ -9865,21 +9891,6 @@ dependencies = [
 "tokio 1.51.0",
 ]

-[[package]]
-name = "tokio-tungstenite"
-version = "0.20.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "212d5dcb2a1ce06d81107c3d0ffa3121fe974b73f068c8282cb1c32328113b6c"
-dependencies = [
- "futures-util",
- "log",
- "rustls 0.21.12",
- "rustls-native-certs 0.6.3",
- "tokio 1.51.0",
- "tokio-rustls 0.24.1",
- "tungstenite 0.20.1",
-]
-
 [[package]]
 name = "tokio-util"
 version = "0.7.18"
@@ -10135,26 +10146,6 @@ dependencies = [
 "utf-8",
 ]

-[[package]]
-name = "tungstenite"
-version = "0.20.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9e3dac10fd62eaf6617d3a904ae222845979aec67c615d1c842b4002c7666fb9"
-dependencies = [
- "byteorder",
- "bytes 1.11.1",
- "data-encoding",
- "http 0.2.12",
- "httparse",
- "log",
- "rand 0.8.5",
- "rustls 0.21.12",
- "sha1",
- "thiserror 1.0.69",
- "url",
- "utf-8",
-]
-
 [[package]]
 name = "typed-builder"
 version = "0.22.0"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -159,6 +159,7 @@ opentelemetry-appender-tracing = "0.31.1"
 authifier = "1.0.16"

 # RabbitMQ
+amqprs = "1.7.0"
 lapin = "4.7.1"

 # Voice
@@ -191,13 +192,13 @@ futures-lite = "2.6.1"
 vergen = "7.5.0"

 # Local packages
-revolt-coalesced = { version = "0.13.7", path = "crates/core/coalesced" }
-revolt-config = { version = "0.13.7", path = "crates/core/config" }
-revolt-database = { version = "0.13.7", path = "crates/core/database" }
-revolt-files = { version = "0.13.7", path = "crates/core/files" }
-revolt-models = { version = "0.13.7", path = "crates/core/models" }
-revolt-parser = { version = "0.13.7", path = "crates/core/parser" }
-revolt-permissions = { version = "0.13.7", path = "crates/core/permissions" }
-revolt-presence = { version = "0.13.7", path = "crates/core/presence" }
-revolt-ratelimits = { version = "0.13.7", path = "crates/core/ratelimits" }
-revolt-result = { version = "0.13.7", path = "crates/core/result" }
+revolt-coalesced = { version = "0.12.0", path = "crates/core/coalesced" }
+revolt-config = { version = "0.12.0", path = "crates/core/config" }
+revolt-database = { version = "0.12.0", path = "crates/core/database" }
+revolt-files = { version = "0.12.0", path = "crates/core/files" }
+revolt-models = { version = "0.12.0", path = "crates/core/models" }
+revolt-parser = { version = "0.12.0", path = "crates/core/parser" }
+revolt-permissions = { version = "0.12.0", path = "crates/core/permissions" }
+revolt-presence = { version = "0.12.0", path = "crates/core/presence" }
+revolt-ratelimits = { version = "0.12.0", path = "crates/core/ratelimits" }
+revolt-result = { version = "0.12.0", path = "crates/core/result" }
--- a/Revolt.toml
+++ b/Revolt.toml
@@ -61,7 +61,7 @@ secret = "ZjCofRlfm6GGtjlifmNpCDkcQbEIIVC0"
 # S3 protocol endpoint
 endpoint = "http://127.0.0.1:14009"
 # S3 region name
-region = "minio"
+region = "us-east-1"
 # S3 protocol key ID
 access_key_id = "minioautumn"
 # S3 protocol access key
--- a/compose.yml
+++ b/compose.yml
@@ -34,7 +34,6 @@ services:
    environment:
      MINIO_ROOT_USER: minioautumn
      MINIO_ROOT_PASSWORD: minioautumn
-      MINIO_REGION: minio
    volumes:
      - ./.data/minio:/data
    ports:
--- a/crates/bonfire/Cargo.toml
+++ b/crates/bonfire/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-bonfire"
-version = "0.13.7"
+version = "0.12.1"
 license = "AGPL-3.0-or-later"
 edition = "2021"
 publish = false
--- a/crates/bonfire/src/events/impl.rs
+++ b/crates/bonfire/src/events/impl.rs
@@ -1,7 +1,6 @@
 use std::collections::{HashMap, HashSet};

 use futures::future::join_all;
-use redis_kiss::AsyncCommands;
 use revolt_database::{
    events::client::{EventV1, ReadyPayloadFields},
    util::permissions::DatabasePermissionQuery,
--- a/crates/bonfire/src/websocket.rs
+++ b/crates/bonfire/src/websocket.rs
@@ -13,7 +13,7 @@ use futures::{
    stream::{SplitSink, SplitStream},
    FutureExt, SinkExt, StreamExt, TryStreamExt,
 };
-use redis_kiss::{get_connection, AsyncCommands, PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
+use redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
 use revolt_config::report_internal_error;
 use revolt_database::{
    events::{client::EventV1, server::ClientMessage},
@@ -32,7 +32,6 @@ 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>;
@@ -129,14 +128,6 @@ 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;

@@ -536,43 +527,4 @@ 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)
-}
+}
--- a/crates/core/coalesced/Cargo.toml
+++ b/crates/core/coalesced/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-coalesced"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>", "Zomatree <me@zomatree.live>"]
--- a/crates/core/config/Cargo.toml
+++ b/crates/core/config/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-config"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
--- a/crates/core/config/Revolt.toml
+++ b/crates/core/config/Revolt.toml
@@ -30,7 +30,7 @@ host = "rabbit"
 port = 5672
 username = "rabbituser"
 password = "rabbitpass"
-default_exchange = "revolt.default"
+default_exchange = "revolt"

 [rabbit.queues]
 acks = "internal.ack"
--- a/crates/core/database/Cargo.toml
+++ b/crates/core/database/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-database"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -95,9 +95,9 @@ revolt_rocket_okapi = { workspace = true, optional = true }
 authifier = { workspace = true }

 # RabbitMQ
-lapin = { workspace = true, features = ["tokio"] }
+amqprs = { workspace = true }

 # Voice
-livekit-api = { workspace = true, features = ["rustls-tls-native-roots"], optional = true }
+livekit-api = { workspace = true, optional = true }
 livekit-protocol = { workspace = true, optional = true }
 livekit-runtime = { workspace = true, features = ["tokio"], optional = true }
--- a/crates/core/database/src/amqp/amqp.rs
+++ b/crates/core/database/src/amqp/amqp.rs
@@ -1,77 +1,98 @@
 use std::collections::HashSet;
-use std::sync::Arc;

 use crate::events::rabbit::*;
 use crate::User;
-use lapin::{
-    options::BasicPublishOptions,
-    protocol::basic::AMQPProperties,
-    types::{AMQPValue, FieldTable},
-    Channel, Connection, ConnectionProperties, Error as AMQPError,
+use amqprs::channel::{
+    BasicPublishArguments, ExchangeDeclareArguments, ExchangeType, QueueBindArguments,
+    QueueDeclareArguments,
 };
+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: Arc<Connection>,
+    connection: Connection,
+    channel: Channel,
 }

 impl AMQP {
-    pub async fn new(connection: Arc<Connection>) -> Self {
-        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,
+    pub fn new(connection: Connection, channel: Channel) -> AMQP {
+        AMQP {
            connection,
+            channel,
        }
    }

-    pub async fn new_auto() -> Self {
+    pub async fn new_auto() -> AMQP {
        let config = revolt_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(),
+        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
+            .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 connect to RabbitMQ"),
-        );
+            .expect("Failed to declare exchange");

-        Self::new(connection).await
+        AMQP::new(connection, channel)
    }

-    async fn create_channel(connection: &Connection) -> Arc<Channel> {
-        Arc::new(
-            connection
-                .create_channel()
-                .await
-                .expect("Failed to create channel"),
-        )
+    pub async fn configure_channels(&self) -> revolt_result::Result<()> {
+        let config = revolt_config::config().await;
+
+        self.channel
+            .exchange_declare(
+                ExchangeDeclareArguments::new(
+                    &config.rabbit.default_exchange,
+                    &ExchangeType::Topic.to_string(),
+                )
+                .durable(true)
+                .finish(),
+            )
+            .await
+            .expect("Failed to declare exchange");
+
+        // Configure acks channel & routing
+        self.channel
+            .queue_declare(
+                QueueDeclareArguments::new(&config.rabbit.queues.acks)
+                    .durable(true)
+                    .no_wait(true)
+                    .finish(),
+            )
+            .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(
@@ -91,20 +112,19 @@ impl AMQP {
            config.pushd.get_fr_accepted_routing_key(),
            payload
        );
-
-        self.friend_request_accepted
+        self.channel
            .basic_publish(
-                config.pushd.exchange.clone().into(),
-                config.pushd.get_fr_accepted_routing_key().into(),
-                BasicPublishOptions::default(),
-                payload.as_bytes(),
-                AMQPProperties::default()
-                    .with_content_type("application/json".into())
-                    .with_delivery_mode(2),
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(
+                    &config.pushd.exchange,
+                    &config.pushd.get_fr_accepted_routing_key(),
+                ),
            )
-            .await?;
-
-        Ok(())
+            .await
    }

    pub async fn friend_request_received(
@@ -125,19 +145,19 @@ impl AMQP {
            payload
        );

-        self.friend_request_received
+        self.channel
            .basic_publish(
-                config.pushd.exchange.clone().into(),
-                config.pushd.get_fr_received_routing_key().into(),
-                BasicPublishOptions::default(),
-                payload.as_bytes(),
-                AMQPProperties::default()
-                    .with_content_type("application/json".into())
-                    .with_delivery_mode(2),
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(
+                    &config.pushd.exchange,
+                    &config.pushd.get_fr_received_routing_key(),
+                ),
            )
-            .await?;
-
-        Ok(())
+            .await
    }

    pub async fn generic_message(
@@ -162,19 +182,19 @@ impl AMQP {
            payload
        );

-        self.generic_message
+        self.channel
            .basic_publish(
-                config.pushd.exchange.clone().into(),
-                config.pushd.get_generic_routing_key().into(),
-                BasicPublishOptions::default(),
-                payload.as_bytes(),
-                AMQPProperties::default()
-                    .with_content_type("application/json".into())
-                    .with_delivery_mode(2),
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(
+                    &config.pushd.exchange,
+                    &config.pushd.get_generic_routing_key(),
+                ),
            )
-            .await?;
-
-        Ok(())
+            .await
    }

    pub async fn message_sent(
@@ -205,19 +225,19 @@ impl AMQP {
            payload
        );

-        self.message_sent
+        self.channel
            .basic_publish(
-                config.pushd.exchange.clone().into(),
-                config.pushd.get_message_routing_key().into(),
-                BasicPublishOptions::default(),
-                payload.as_bytes(),
-                AMQPProperties::default()
-                    .with_content_type("application/json".into())
-                    .with_delivery_mode(2),
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(
+                    &config.pushd.exchange,
+                    &config.pushd.get_message_routing_key(),
+                ),
            )
-            .await?;
-
-        Ok(())
+            .await
    }

    pub async fn mass_mention_message_sent(
@@ -240,19 +260,16 @@ impl AMQP {
            routing_key, payload
        );

-        self.mass_mention_message_sent
+        self.channel
            .basic_publish(
-                config.pushd.exchange.clone().into(),
-                routing_key.into(),
-                BasicPublishOptions::default(),
-                payload.as_bytes(),
-                AMQPProperties::default()
-                    .with_content_type("application/json".into())
-                    .with_delivery_mode(2),
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(&config.pushd.exchange, routing_key.as_str()),
            )
-            .await?;
-
-        Ok(())
+            .await
    }

    /// # Sends an ack to pushd to update badges on iPhones.
@@ -277,25 +294,23 @@ impl AMQP {
            config.pushd.ack_queue, payload
        );

-        let mut headers = FieldTable::default();
+        let mut headers = FieldTable::new();
        headers.insert(
-            "x-deduplication-header".into(),
-            AMQPValue::LongString(format!("{}-{}", &user_id, &channel_id).into()),
+            "x-deduplication-header".try_into().unwrap(),
+            format!("{}-{}", &user_id, &channel_id).into(),
        );

-        self.ack_notification_message
+        self.channel
            .basic_publish(
-                config.pushd.exchange.clone().into(),
-                config.pushd.ack_queue.into(),
-                BasicPublishOptions::default(),
-                payload.as_bytes(),
-                AMQPProperties::default()
-                    .with_content_type("application/json".into())
-                    .with_delivery_mode(2),
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    //.with_headers(headers)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(&config.pushd.exchange, &config.pushd.ack_queue),
            )
-            .await?;
-
-        Ok(())
+            .await
    }

    /// # DM Call Update
@@ -329,19 +344,19 @@ impl AMQP {
            payload
        );

-        self.dm_call_updated
+        self.channel
            .basic_publish(
-                config.pushd.exchange.clone().into(),
-                config.pushd.get_dm_call_routing_key().into(),
-                BasicPublishOptions::default(),
-                payload.as_bytes(),
-                AMQPProperties::default()
-                    .with_content_type("application/json".into())
-                    .with_delivery_mode(2),
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(
+                    &config.pushd.exchange,
+                    &config.pushd.get_dm_call_routing_key(),
+                ),
            )
-            .await?;
-
-        Ok(())
+            .await
    }

    /// # Send an ack to crond for processing
@@ -365,18 +380,19 @@ impl AMQP {
            config.rabbit.default_exchange, config.rabbit.queues.acks, payload
        );

-        self.process_ack
+        self.channel
            .basic_publish(
-                config.rabbit.default_exchange.clone().into(),
-                config.rabbit.queues.acks.into(),
-                BasicPublishOptions::default(),
-                payload.as_bytes(),
-                AMQPProperties::default()
-                    .with_content_type("application/json".into())
-                    .with_delivery_mode(2),
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    //.with_headers(headers)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(
+                    &config.rabbit.default_exchange,
+                    &config.rabbit.queues.acks,
+                ),
            )
-            .await?;
-
-        Ok(())
+            .await
    }
 }
--- a/crates/core/database/src/events/client.rs
+++ b/crates/core/database/src/events/client.rs
@@ -3,12 +3,7 @@ use revolt_result::Error;
 use serde::{Deserialize, Serialize};

 use revolt_models::v0::{
-    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,
+    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
 };

 use crate::Database;
@@ -56,13 +51,9 @@ 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,
@@ -93,9 +84,7 @@ pub enum EventV1 {
    },

    /// Ping response
-    Pong {
-        data: Ping,
-    },
+    Pong { data: Ping },
    /// New message
    Message(Message),

@@ -116,10 +105,7 @@ pub enum EventV1 {
    },

    /// Delete message
-    MessageDelete {
-        id: String,
-        channel: String,
-    },
+    MessageDelete { id: String, channel: String },

    /// New reaction to a message
    MessageReact {
@@ -145,10 +131,7 @@ pub enum EventV1 {
    },

    /// Bulk delete messages
-    BulkMessageDelete {
-        channel: String,
-        ids: Vec<String>,
-    },
+    BulkMessageDelete { channel: String, ids: Vec<String> },

    /// New server
    ServerCreate {
@@ -156,7 +139,7 @@ pub enum EventV1 {
        server: Server,
        channels: Vec<Channel>,
        emojis: Vec<Emoji>,
-        voice_states: Vec<ChannelVoiceState>,
+        voice_states: Vec<ChannelVoiceState>
    },

    /// Update existing server
@@ -168,9 +151,7 @@ pub enum EventV1 {
    },

    /// Delete server
-    ServerDelete {
-        id: String,
-    },
+    ServerDelete { id: String },

    /// Update existing server member
    ServerMemberUpdate {
@@ -206,16 +187,10 @@ 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 {
@@ -227,15 +202,9 @@ 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
    ///
@@ -246,10 +215,7 @@ 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),

@@ -260,9 +226,7 @@ pub enum EventV1 {
    },

    /// Delete emoji
-    EmojiDelete {
-        id: String,
-    },
+    EmojiDelete { id: String },

    /// New report
    ReportCreate(Report),
@@ -278,33 +242,19 @@ 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 {
@@ -324,9 +274,7 @@ pub enum EventV1 {
    },

    /// Delete webhook
-    WebhookDelete {
-        id: String,
-    },
+    WebhookDelete { id: String },

    /// Auth events
    Auth(AuthifierEvent),
@@ -344,7 +292,7 @@ pub enum EventV1 {
        user: String,
        from: String,
        to: String,
-        state: UserVoiceState,
+        state: UserVoiceState
    },
    UserVoiceStateUpdate {
        id: String,
@@ -356,11 +304,7 @@ pub enum EventV1 {
        from: String,
        to: String,
        token: String,
-    },
-    /// User's active slowmodes
-    UserSlowmodes {
-        slowmodes: Vec<ChannelSlowmode>,
-    },
+    }
 }

 impl EventV1 {
--- a/crates/core/database/src/lib.rs
+++ b/crates/core/database/src/lib.rs
@@ -95,7 +95,7 @@ macro_rules! database_test {
        db.drop_database().await;

        #[allow(clippy::redundant_closure_call)]
-        (|$db: $crate::Database| $test)(db.clone()).await;
+        std::boxed::Box::pin((|$db: $crate::Database| $test)(db.clone())).await;

        db.drop_database().await
    };
--- a/crates/core/database/src/models/emojis/model.rs
+++ b/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.replace('\u{FE0F}', ""))
+        .map(|x| x.into())
        .collect()
 });

@@ -108,8 +108,7 @@ impl Emoji {
            db.fetch_emoji(emoji).await?;
            Ok(true)
        } else {
-            let sanitized_emoji = emoji.replace('\u{FE0F}', "");
-            Ok(PERMISSIBLE_EMOJIS.contains(&sanitized_emoji))
+            Ok(PERMISSIBLE_EMOJIS.contains(emoji))
        }
    }
 }
--- a/crates/core/database/src/models/emojis/unicode_emoji.txt
+++ b/crates/core/database/src/models/emojis/unicode_emoji.txt
@@ -1,3 +1,4 @@
+
 *️⃣
 0️⃣
 1️⃣
@@ -241,7 +242,6 @@
 🆘
 🆙
 🆚
-🇦
 🇦🇨
 🇦🇩
 🇦🇪
@@ -259,7 +259,6 @@
 🇦🇼
 🇦🇽
 🇦🇿
-🇧
 🇧🇦
 🇧🇧
 🇧🇩
@@ -281,7 +280,6 @@
 🇧🇼
 🇧🇾
 🇧🇿
-🇨
 🇨🇦
 🇨🇨
 🇨🇩
@@ -303,7 +301,6 @@
 🇨🇽
 🇨🇾
 🇨🇿
-🇩
 🇩🇪
 🇩🇬
 🇩🇯
@@ -311,7 +308,6 @@
 🇩🇲
 🇩🇴
 🇩🇿
-🇪
 🇪🇦
 🇪🇨
 🇪🇪
@@ -321,14 +317,12 @@
 🇪🇸
 🇪🇹
 🇪🇺
-🇫
 🇫🇮
 🇫🇯
 🇫🇰
 🇫🇲
 🇫🇴
 🇫🇷
-🇬
 🇬🇦
 🇬🇧
 🇬🇩
@@ -348,14 +342,12 @@
 🇬🇺
 🇬🇼
 🇬🇾
-🇭
 🇭🇰
 🇭🇲
 🇭🇳
 🇭🇷
 🇭🇹
 🇭🇺
-🇮
 🇮🇨
 🇮🇩
 🇮🇪
@@ -367,12 +359,10 @@
 🇮🇷
 🇮🇸
 🇮🇹
-🇯
 🇯🇪
 🇯🇲
 🇯🇴
 🇯🇵
-🇰
 🇰🇪
 🇰🇬
 🇰🇭
@@ -384,7 +374,6 @@
 🇰🇼
 🇰🇾
 🇰🇿
-🇱
 🇱🇦
 🇱🇧
 🇱🇨
@@ -396,7 +385,6 @@
 🇱🇺
 🇱🇻
 🇱🇾
-🇲
 🇲🇦
 🇲🇨
 🇲🇩
@@ -420,7 +408,6 @@
 🇲🇽
 🇲🇾
 🇲🇿
-🇳
 🇳🇦
 🇳🇨
 🇳🇪
@@ -433,9 +420,7 @@
 🇳🇷
 🇳🇺
 🇳🇿
-🇴
 🇴🇲
-🇵
 🇵🇦
 🇵🇪
 🇵🇫
@@ -450,15 +435,12 @@
 🇵🇹
 🇵🇼
 🇵🇾
-🇶
 🇶🇦
-🇷
 🇷🇪
 🇷🇴
 🇷🇸
 🇷🇺
 🇷🇼
-🇸
 🇸🇦
 🇸🇧
 🇸🇨
@@ -480,7 +462,6 @@
 🇸🇽
 🇸🇾
 🇸🇿
-🇹
 🇹🇦
 🇹🇨
 🇹🇩
@@ -498,7 +479,6 @@
 🇹🇻
 🇹🇼
 🇹🇿
-🇺
 🇺🇦
 🇺🇬
 🇺🇲
@@ -506,7 +486,6 @@
 🇺🇸
 🇺🇾
 🇺🇿
-🇻
 🇻🇦
 🇻🇨
 🇻🇪
@@ -514,15 +493,11 @@
 🇻🇮
 🇻🇳
 🇻🇺
-🇼
 🇼🇫
 🇼🇸
-🇽
 🇽🇰
-🇾
 🇾🇪
 🇾🇹
-🇿
 🇿🇦
 🇿🇲
 🇿🇼
--- a/crates/core/database/src/models/messages/model.rs
+++ b/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 - 1);
+        let mask = 1 << bit;
        self.0 & mask == mask
    }

@@ -227,11 +227,10 @@ 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 |= mask;
+            self.0 |= 1 << bit;
        } else {
-            self.0 &= !mask;
+            self.0 &= !(1 << bit);
        }
        self
    }
--- a/crates/core/files/Cargo.toml
+++ b/crates/core/files/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-files"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
--- a/crates/core/models/Cargo.toml
+++ b/crates/core/models/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-models"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
--- a/crates/core/models/src/v0/channels.rs
+++ b/crates/core/models/src/v0/channels.rs
@@ -314,12 +314,6 @@ 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 {
--- a/crates/core/parser/Cargo.toml
+++ b/crates/core/parser/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-parser"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "MIT"
 authors = ["Zomatree <me@zomatree.live>", "Paul Makles <me@insrt.uk>"]
--- a/crates/core/permissions/Cargo.toml
+++ b/crates/core/permissions/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-permissions"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
--- a/crates/core/presence/Cargo.toml
+++ b/crates/core/presence/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-presence"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
--- a/crates/core/ratelimits/Cargo.toml
+++ b/crates/core/ratelimits/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-ratelimits"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2024"
 license = "MIT"
 authors = ["Zomatree <me@zomatree.live>", "Paul Makles <me@insrt.uk>"]
--- a/crates/core/ratelimits/src/rocket.rs
+++ b/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
-        .client_ip()
-        .map(|r| r.to_string())
+        .remote()
+        .map(|x| x.ip().to_string())
        .unwrap_or_default()
 }

--- a/crates/core/result/Cargo.toml
+++ b/crates/core/result/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-result"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
--- a/crates/daemons/crond/Cargo.toml
+++ b/crates/daemons/crond/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-crond"
-version = "0.13.7"
+version = "0.12.1"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
 edition = "2021"
--- a/crates/daemons/crond/src/main.rs
+++ b/crates/daemons/crond/src/main.rs
@@ -1,5 +1,5 @@
 use revolt_config::configure;
-use revolt_database::{DatabaseInfo, AMQP};
+use revolt_database::DatabaseInfo;
 use revolt_result::Result;
 use tasks::{acks, file_deletion, prune_dangling_files, prune_members};
 use tokio::try_join;
@@ -11,13 +11,11 @@ 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(), amqp.clone()),
+        acks::task(db.clone())
    )
    .map(|_| ())
 }
--- a/crates/daemons/crond/src/tasks/acks.rs
+++ b/crates/daemons/crond/src/tasks/acks.rs
@@ -3,16 +3,16 @@ use lapin::{
    options::*,
    types::FieldTable,
    uri::{AMQPAuthority, AMQPQueryString, AMQPUri, AMQPUserInfo},
-    ConnectionBuilder, ConnectionProperties, ExchangeKind,
+    ConnectionBuilder, ConnectionProperties,
 };
-use log::{debug, info};
+use log::info;
 use redis_kiss::{get_connection, AsyncCommands, Conn as RedisConnection};
 use revolt_config::config;
-use revolt_database::{events::rabbit::AckEventPayload, Database, AMQP};
+use revolt_database::{events::rabbit::AckEventPayload, Database};
 use revolt_result::{Result, ToRevoltError};
 use serde_json;

-pub async fn task(db: Database, amqp: AMQP) -> Result<()> {
+pub async fn task(db: Database) -> Result<()> {
    let config = config().await;

    let mut redis = get_connection()
@@ -46,42 +46,6 @@ pub async fn task(db: Database, amqp: AMQP) -> 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(),
@@ -94,14 +58,12 @@ pub async fn task(db: Database, amqp: AMQP) -> Result<()> {

    while let Some(delivery) = consumer.next().await {
        if let Ok(delivery) = delivery {
-            let payload = serde_json::from_slice::<AckEventPayload>(&delivery.data);
-
+            let payload: std::result::Result<AckEventPayload, _> =
+                serde_json::from_slice(&delivery.data);
            if let Ok(payload) = payload {
-                debug!("Received ack event: {payload:?}");
-
+                info!("{:?}", payload);
                if let Err(e) = process_channel_ack(
                    &db,
-                    &amqp,
                    payload.user_id,
                    payload.channel_id.unwrap(),
                    &mut redis,
@@ -127,7 +89,6 @@ pub async fn task(db: Database, amqp: AMQP) -> Result<()> {
 #[allow(clippy::disallowed_methods)]
 async fn process_channel_ack(
    db: &Database,
-    amqp: &AMQP,
    user: String,
    channel: String,
    redis: &mut RedisConnection,
@@ -138,24 +99,28 @@ async fn process_channel_ack(
        .to_internal_error()?;

    if let Some(message_id) = message_id {
-        let unread = db.fetch_unread(&user, &channel).await?;
-        let updated = db.acknowledge_message(&channel, &user, &message_id).await?;
+        // 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.
+        // We also drop a db request as a bonus.

+        //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) {
-            let before_mentions = before.mentions.unwrap_or_default().len();
-            let after_mentions = after.mentions.unwrap_or_default().len();
+        // if let (Some(before), Some(after)) = (unread, updated) {
+        //     let before_mentions = before.mentions.unwrap_or_default().len();
+        //     let after_mentions = after.mentions.unwrap_or_default().len();

-            if after_mentions < before_mentions {
-                if let Err(err) = amqp
-                    .ack_notification_message(user.to_string(), channel.to_string(), message_id)
-                    .await
-                {
-                    revolt_config::capture_error(&err);
-                }
-            };
-        }
+        //     let mentions_acked = before_mentions - after_mentions;
+
+        //     if mentions_acked > 0 {
+        //         if let Err(err) = amqp
+        //             .ack_message(user.to_string(), channel.to_string(), payload.message_id)
+        //             .await
+        //         {
+        //             revolt_config::capture_error(&err);
+        //         }
+        //     };
+        // }

        Ok(())
    } else {
--- a/crates/daemons/crond/src/tasks/file_deletion.rs
+++ b/crates/daemons/crond/src/tasks/file_deletion.rs
@@ -11,24 +11,22 @@ 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(hash)
+            let count = db
+                .count_file_hash_references(file.hash.as_ref().expect("no `hash` present"))
+                .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"))
                    .await?;

-                // No other files reference this file on disk anymore
-                if count <= 1 {
-                    let file_hash = db
-                        .fetch_attachment_hash(hash)
-                        .await?;
+                // Delete from S3
+                delete_from_s3(&file_hash.bucket_id, &file_hash.path).await?;

-                    // Delete from S3
-                    delete_from_s3(&file_hash.bucket_id, &file_hash.path).await?;
-
-                    // Delete the hash
-                    db.delete_attachment_hash(&file_hash.id).await?;
-                    info!("Deleted file hash {}", file_hash.id);
-                }
+                // Delete the hash
+                db.delete_attachment_hash(&file_hash.id).await?;
+                info!("Deleted file hash {}", file_hash.id);
            }

            // Delete the file
--- a/crates/daemons/pushd/Cargo.toml
+++ b/crates/daemons/pushd/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-pushd"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 publish = false
@@ -15,7 +15,7 @@ revolt-parser = { workspace = true }

 anyhow = { workspace = true }

-lapin = { workspace = true }
+amqprs = { workspace = true }
 fcm_v1 = { workspace = true }
 web-push = { workspace = true }
 isahc = { workspace = true, features = ["json"], optional = true }
--- a/crates/daemons/pushd/src/consumers/inbound/ack.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/ack.rs
@@ -1,69 +1,96 @@
-use std::sync::Arc;
-
-use crate::utils::Consumer;
-use anyhow::Result;
+use crate::consumers::inbound::internal::*;
+use amqprs::{
+    channel::{BasicPublishArguments, Channel},
+    connection::Connection,
+    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,
-    connection: Arc<Connection>,
-    channel: Arc<Channel>,
+    conn: Option<Connection>,
+    channel: Option<Channel>,
 }

-#[async_trait]
-impl Consumer for AckConsumer {
-    async fn create(
-        db: Database,
-        authifier_db: authifier::Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
-        Self {
-            db,
-            authifier_db,
-            connection,
-            channel,
+impl Channeled for AckConsumer {
+    fn get_connection(&self) -> Option<&Connection> {
+        if self.conn.is_none() {
+            None
+        } else {
+            Some(self.conn.as_ref().unwrap())
        }
    }

-    fn channel(&self) -> &Arc<Channel> {
-        &self.channel
+    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,
+            channel: None,
+        }
+    }
+}
+
+#[allow(unused_variables)]
+#[async_trait]
+impl AsyncConsumer for AckConsumer {
    /// This consumer processes all acks the platform receives, and sends relevant badge updates to apple platforms.
-    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        let payload: AckPayload = serde_json::from_slice(&delivery.data)?;
+    async fn consume(
+        &mut self,
+        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);

-        let unreads = if let Ok(u) = self.db.fetch_unread_mentions(&payload.user_id).await {
+        if let Ok(u) = &unreads {
            if u.is_empty() {
                debug!(
                    "Discarding unread task (no mentions found) for {:}",
                    &payload.user_id
                );
-                return Ok(());
-            };
-
-            u
+                return;
+            }
        } else {
-            return Ok(());
-        };
+            return;
+        }

        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 mut apple_sessions = sessions
-                .into_iter()
+            let apple_sessions: Vec<&authifier::models::Session> = sessions
+                .iter()
                .filter(|session| {
                    if let Some(sub) = &session.subscription {
                        sub.endpoint == "apn"
@@ -71,19 +98,19 @@ impl Consumer for AckConsumer {
                        false
                    }
                })
-                .peekable();
+                .collect();

-            if apple_sessions.peek().is_none() {
+            if apple_sessions.is_empty() {
                debug!(
                    "Discarding unread task (no apn sessions found) for {:}",
                    &payload.user_id
                );
-                return Ok(());
+                return;
            }

            // Step 3: calculate the actual mention count, since we have to send it out
            let mut mention_count = 0;
-            for u in &unreads {
+            for u in &unreads.unwrap() {
                mention_count += u.mentions.as_ref().unwrap().len()
            }

@@ -96,22 +123,26 @@ impl Consumer for AckConsumer {
                    token: session.subscription.as_ref().unwrap().auth.clone(),
                    extras: Default::default(),
                };
-                let payload = serde_json::to_string(&service_payload)?;
+                let raw_service_payload = serde_json::to_string(&service_payload);

-                log::debug!(
-                    "Publishing ack to apn session {}",
-                    session.subscription.as_ref().unwrap().auth
-                );
+                if let Ok(p) = raw_service_payload {
+                    let args = BasicPublishArguments::new(
+                        config.pushd.exchange.as_str(),
+                        config.pushd.apn.queue.as_str(),
+                    )
+                    .finish();

-                self.publish_message(
-                    payload.as_bytes(),
-                    &config.pushd.exchange,
-                    &config.pushd.apn.queue,
-                )
-                .await?;
+                    log::debug!(
+                        "Publishing ack to apn session {}",
+                        session.subscription.as_ref().unwrap().auth
+                    );
+
+                    publish_message(self, p.into(), args).await;
+                } else {
+                    log::warn!("Failed to serialize ack badge update payload!");
+                    revolt_config::capture_error(&raw_service_payload.unwrap_err());
+                }
            }
        }
-
-        Ok(())
    }
 }
--- a/crates/daemons/pushd/src/consumers/inbound/dm_call.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/dm_call.rs
@@ -1,44 +1,70 @@
-use std::{collections::HashMap, sync::Arc};
+use std::collections::HashMap;

-use crate::utils::Consumer;
+use crate::consumers::inbound::internal::*;
+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,
-    connection: Arc<Connection>,
-    channel: Arc<Channel>,
+    conn: Option<Connection>,
+    channel: Option<Channel>,
 }

-#[async_trait]
-impl Consumer for DmCallConsumer {
-    async fn create(
-        db: Database,
-        authifier_db: authifier::Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
-        Self {
-            db,
-            authifier_db,
-            connection,
-            channel,
+impl Channeled for DmCallConsumer {
+    fn get_connection(&self) -> Option<&Connection> {
+        if self.conn.is_none() {
+            None
+        } else {
+            Some(self.conn.as_ref().unwrap())
        }
    }

-    fn channel(&self) -> &Arc<Channel> {
-        &self.channel
+    fn get_channel(&self) -> Option<&Channel> {
+        if self.channel.is_none() {
+            None
+        } else {
+            Some(self.channel.as_ref().unwrap())
+        }
    }

-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        let _p: InternalDmCallPayload = serde_json::from_slice(&delivery.data)?;
+    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,
+            channel: None,
+        }
+    }
+
+    async fn consume_event(
+        &mut self,
+        _channel: &Channel,
+        _deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) -> Result<()> {
+        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");
@@ -81,27 +107,36 @@ impl Consumer for DmCallConsumer {
                            extras: HashMap::new(),
                        };

-                        let routing_key = match sub.endpoint.as_str() {
-                            "apn" => &config.pushd.apn.queue,
-                            "fcm" => &config.pushd.fcm.queue,
-                            endpoint => {
-                                sendable.extras.insert("p256dh".to_string(), sub.p256dh);
-                                sendable
-                                    .extras
-                                    .insert("endpoint".to_string(), endpoint.to_string());
+                        let args: BasicPublishArguments;

-                                &config.pushd.vapid.queue
-                            }
-                        };
+                        if sub.endpoint == "apn" {
+                            args = BasicPublishArguments::new(
+                                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());
+                        }

                        let payload = serde_json::to_string(&sendable)?;

-                        self.publish_message(
-                            payload.as_bytes(),
-                            &config.pushd.exchange,
-                            routing_key,
-                        )
-                        .await?;
+                        publish_message(self, payload.into(), args).await;
                    }
                }
            }
@@ -110,3 +145,24 @@ impl Consumer for 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:?}");
+        }
+    }
+}
--- a/crates/daemons/pushd/src/consumers/inbound/fr_accepted.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/fr_accepted.rs
@@ -1,44 +1,70 @@
-use std::{collections::HashMap, sync::Arc};
+use std::collections::HashMap;

-use crate::utils::Consumer;
+use crate::consumers::inbound::internal::*;
+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,
-    connection: Arc<Connection>,
-    channel: Arc<Channel>,
+    conn: Option<Connection>,
+    channel: Option<Channel>,
 }

-#[async_trait]
-impl Consumer for FRAcceptedConsumer {
-    async fn create(
-        db: Database,
-        authifier_db: authifier::Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
-        Self {
-            db,
-            authifier_db,
-            connection,
-            channel,
+impl Channeled for FRAcceptedConsumer {
+    fn get_connection(&self) -> Option<&Connection> {
+        if self.conn.is_none() {
+            None
+        } else {
+            Some(self.conn.as_ref().unwrap())
        }
    }

-    fn channel(&self) -> &Arc<Channel> {
-        &self.channel
+    fn get_channel(&self) -> Option<&Channel> {
+        if self.channel.is_none() {
+            None
+        } else {
+            Some(self.channel.as_ref().unwrap())
+        }
    }

-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        let payload: FRAcceptedPayload = serde_json::from_slice(&delivery.data)?;
+    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,
+            channel: None,
+        }
+    }
+
+    async fn consume_event(
+        &mut self,
+        _channel: &Channel,
+        _deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) -> Result<()> {
+        let content = String::from_utf8(content)?;
+        let payload: FRAcceptedPayload = serde_json::from_str(content.as_str())?;

        debug!("Received FR accept event");

@@ -54,23 +80,36 @@ impl Consumer for FRAcceptedConsumer {
                        extras: HashMap::new(),
                    };

-                    let routing_key = match sub.endpoint.as_str() {
-                        "apn" => &config.pushd.apn.queue,
-                        "fcm" => &config.pushd.fcm.queue,
-                        endpoint => {
-                            sendable.extras.insert("p256dh".to_string(), sub.p256dh);
-                            sendable
-                                .extras
-                                .insert("endpoint".to_string(), endpoint.to_string());
+                    let args: BasicPublishArguments;

-                            &config.pushd.vapid.queue
-                        }
-                    };
+                    if sub.endpoint == "apn" {
+                        args = BasicPublishArguments::new(
+                            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());
+                    }

                    let payload = serde_json::to_string(&sendable)?;

-                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
-                        .await?;
+                    publish_message(self, payload.into(), args).await;
                }
            }
        }
@@ -78,3 +117,24 @@ impl Consumer for 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:?}");
+        }
+    }
+}
--- a/crates/daemons/pushd/src/consumers/inbound/fr_received.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/fr_received.rs
@@ -1,44 +1,70 @@
-use std::{collections::HashMap, sync::Arc};
+use std::collections::HashMap;

-use crate::utils::Consumer;
+use crate::consumers::inbound::internal::*;
+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,
-    connection: Arc<Connection>,
-    channel: Arc<Channel>,
+    conn: Option<Connection>,
+    channel: Option<Channel>,
 }

-#[async_trait]
-impl Consumer for FRReceivedConsumer {
-    async fn create(
-        db: Database,
-        authifier_db: authifier::Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
-        Self {
-            db,
-            authifier_db,
-            connection,
-            channel,
+impl Channeled for FRReceivedConsumer {
+    fn get_connection(&self) -> Option<&Connection> {
+        if self.conn.is_none() {
+            None
+        } else {
+            Some(self.conn.as_ref().unwrap())
        }
    }

-    fn channel(&self) -> &Arc<Channel> {
-        &self.channel
+    fn get_channel(&self) -> Option<&Channel> {
+        if self.channel.is_none() {
+            None
+        } else {
+            Some(self.channel.as_ref().unwrap())
+        }
    }

-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        let payload: FRReceivedPayload = serde_json::from_slice(&delivery.data)?;
+    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,
+            channel: None,
+        }
+    }
+
+    async fn consume_event(
+        &mut self,
+        _channel: &Channel,
+        _deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) -> Result<()> {
+        let content = String::from_utf8(content)?;
+        let payload: FRReceivedPayload = serde_json::from_str(content.as_str())?;

        debug!("Received FR received event");

@@ -54,23 +80,36 @@ impl Consumer for FRReceivedConsumer {
                        extras: HashMap::new(),
                    };

-                    let routing_key = match sub.endpoint.as_str() {
-                        "apn" => &config.pushd.apn.queue,
-                        "fcm" => &config.pushd.fcm.queue,
-                        endpoint => {
-                            sendable.extras.insert("p256dh".to_string(), sub.p256dh);
-                            sendable
-                                .extras
-                                .insert("endpoint".to_string(), endpoint.to_string());
+                    let args: BasicPublishArguments;

-                            &config.pushd.vapid.queue
-                        }
-                    };
+                    if sub.endpoint == "apn" {
+                        args = BasicPublishArguments::new(
+                            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());
+                    }

                    let payload = serde_json::to_string(&sendable)?;

-                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
-                        .await?;
+                    publish_message(self, payload.into(), args).await;
                }
            }
        }
@@ -78,3 +117,24 @@ impl Consumer for 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:?}");
+        }
+    }
+}
--- a/crates/daemons/pushd/src/consumers/inbound/generic.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/generic.rs
@@ -1,44 +1,70 @@
-use std::{collections::HashMap, sync::Arc};
+use std::collections::HashMap;

-use crate::utils::Consumer;
+use crate::consumers::inbound::internal::*;
+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,
-    connection: Arc<Connection>,
-    channel: Arc<Channel>,
+    conn: Option<Connection>,
+    channel: Option<Channel>,
 }

-#[async_trait]
-impl Consumer for GenericConsumer {
-    async fn create(
-        db: Database,
-        authifier_db: authifier::Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
-        Self {
-            db,
-            authifier_db,
-            connection,
-            channel,
+impl Channeled for GenericConsumer {
+    fn get_connection(&self) -> Option<&Connection> {
+        if self.conn.is_none() {
+            None
+        } else {
+            Some(self.conn.as_ref().unwrap())
        }
    }

-    fn channel(&self) -> &Arc<Channel> {
-        &self.channel
+    fn get_channel(&self) -> Option<&Channel> {
+        if self.channel.is_none() {
+            None
+        } else {
+            Some(self.channel.as_ref().unwrap())
+        }
    }

-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        let payload: MessageSentPayload = serde_json::from_slice(&delivery.data)?;
+    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,
+            channel: None,
+        }
+    }
+
+    async fn consume_event(
+        &mut self,
+        _channel: &Channel,
+        _deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) -> Result<()> {
+        let content = String::from_utf8(content)?;
+        let payload: MessageSentPayload = serde_json::from_str(content.as_str())?;

        debug!("Received message event on origin");

@@ -60,23 +86,36 @@ impl Consumer for GenericConsumer {
                        extras: HashMap::new(),
                    };

-                    let routing_key = match sub.endpoint.as_str() {
-                        "apn" => &config.pushd.apn.queue,
-                        "fcm" => &config.pushd.fcm.queue,
-                        endpoint => {
-                            sendable.extras.insert("p256dh".to_string(), sub.p256dh);
-                            sendable
-                                .extras
-                                .insert("endpoint".to_string(), endpoint.to_string());
+                    let args: BasicPublishArguments;

-                            &config.pushd.vapid.queue
-                        }
-                    };
+                    if sub.endpoint == "apn" {
+                        args = BasicPublishArguments::new(
+                            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());
+                    }

                    let payload = serde_json::to_string(&sendable)?;

-                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
-                        .await?;
+                    publish_message(self, payload.into(), args).await;
                }
            }
        }
@@ -84,3 +123,24 @@ impl Consumer for 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:?}");
+        }
+    }
+}
--- a/crates/daemons/pushd/src/consumers/inbound/internal.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/internal.rs
@@ -0,0 +1,53 @@
+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)!")
+    }
+}
--- a/crates/daemons/pushd/src/consumers/inbound/mass_mention.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/mass_mention.rs
@@ -1,13 +1,17 @@
 use std::{
    collections::{HashMap, HashSet},
    hash::RandomState,
-    sync::Arc,
 };

-use crate::utils::{render_notification_content, Consumer};
+use crate::{consumers::inbound::internal::*, utils};
+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,
@@ -15,18 +19,52 @@ 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,
-    connection: Arc<Connection>,
-    channel: Arc<Channel>,
+    conn: Option<Connection>,
+    channel: Option<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(
-        &self,
+        &mut self,
        push: &PushNotification,
        users: &[String],
    ) -> Result<()> {
@@ -46,58 +84,56 @@ impl MassMessageConsumer {
                        extras: HashMap::new(),
                    };

-                    let routing_key = match sub.endpoint.as_str() {
-                        "apn" => &config.pushd.apn.queue,
-                        "fcm" => &config.pushd.fcm.queue,
-                        endpoint => {
-                            sendable.extras.insert("p256dh".to_string(), sub.p256dh);
-                            sendable
-                                .extras
-                                .insert("endpoint".to_string(), endpoint.to_string());
+                    let args: BasicPublishArguments;

-                            &config.pushd.vapid.queue
-                        }
-                    };
+                    if sub.endpoint == "apn" {
+                        args = BasicPublishArguments::new(
+                            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());
+                    }

                    let payload = serde_json::to_string(&sendable)?;

-                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
-                        .await?;
+                    publish_message(self, payload.into(), args).await;
                }
            }
        }

        Ok(())
    }
-}

-#[async_trait]
-impl Consumer for MassMessageConsumer {
-    async fn create(
-        db: Database,
-        authifier_db: authifier::Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
-        Self {
-            db,
-            authifier_db,
-            connection,
-            channel,
-        }
-    }
-
-    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)?;
+    async fn consume_event(
+        &mut self,
+        _channel: &Channel,
+        _deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) -> Result<()> {
        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) = render_notification_content(push, &self.db)
+            if let Ok(body) = utils::render_notification_content(push, &self.db)
                .await
                .to_internal_error()
            {
@@ -244,3 +280,24 @@ impl Consumer for 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:?}");
+        }
+    }
+}
--- a/crates/daemons/pushd/src/consumers/inbound/message.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/message.rs
@@ -1,46 +1,76 @@
-use std::{collections::HashMap, sync::Arc};
+use std::collections::HashMap;

-use crate::utils::{render_notification_content, Consumer};
+use crate::{consumers::inbound::internal::*, utils};
+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,
-    connection: Arc<Connection>,
-    channel: Arc<Channel>,
+    conn: Option<Connection>,
+    channel: Option<Channel>,
 }

-#[async_trait]
-impl Consumer for MessageConsumer {
-    async fn create(
-        db: Database,
-        authifier_db: authifier::Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
-        Self {
-            db,
-            authifier_db,
-            connection,
-            channel,
+impl Channeled for MessageConsumer {
+    fn get_connection(&self) -> Option<&Connection> {
+        if self.conn.is_none() {
+            None
+        } else {
+            Some(self.conn.as_ref().unwrap())
        }
    }

-    fn channel(&self) -> &Arc<Channel> {
-        &self.channel
+    fn get_channel(&self) -> Option<&Channel> {
+        if self.channel.is_none() {
+            None
+        } else {
+            Some(self.channel.as_ref().unwrap())
+        }
    }

-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        let mut payload: MessageSentPayload = serde_json::from_slice(&delivery.data)?;
+    fn set_connection(&mut self, conn: Connection) {
+        self.conn = Some(conn);
+    }

-        if let Ok(body) = render_notification_content(&payload.notification, &self.db).await {
+    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,
+            channel: None,
+        }
+    }
+
+    async fn consume_event(
+        &mut self,
+        _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)
+            .await
+            .to_internal_error()
+        {
            payload.notification.raw_body = Some(payload.notification.body);
            payload.notification.body = body;
        }
@@ -65,22 +95,36 @@ impl Consumer for MessageConsumer {
                        extras: HashMap::new(),
                    };

-                    let routing_key = match sub.endpoint.as_str() {
-                        "apn" => &config.pushd.apn.queue,
-                        "fcm" => &config.pushd.fcm.queue,
-                        endpoint => {
-                            sendable.extras.insert("p256dh".to_string(), sub.p256dh);
-                            sendable
-                                .extras
-                                .insert("endpoint".to_string(), endpoint.to_string());
+                    let args: BasicPublishArguments;

-                            &config.pushd.vapid.queue
-                        }
-                    };
+                    if sub.endpoint == "apn" {
+                        args = BasicPublishArguments::new(
+                            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());
+                    }

                    let payload = serde_json::to_string(&sendable)?;
-                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
-                        .await?;
+
+                    publish_message(self, payload.into(), args).await;
                }
            }
        }
@@ -88,3 +132,24 @@ impl Consumer for 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:?}");
+        }
+    }
+}
--- a/crates/daemons/pushd/src/consumers/inbound/mod.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/mod.rs
@@ -3,5 +3,6 @@ pub mod dm_call;
 pub mod fr_accepted;
 pub mod fr_received;
 pub mod generic;
+mod internal;
 pub mod mass_mention;
 pub mod message;
--- a/crates/daemons/pushd/src/consumers/outbound/apn.rs
+++ b/crates/daemons/pushd/src/consumers/outbound/apn.rs
@@ -1,13 +1,12 @@
-use std::{borrow::Cow, collections::BTreeMap, io::Cursor, sync::Arc};
+use std::{borrow::Cow, collections::BTreeMap, io::Cursor};

-use crate::utils::Consumer;
-use anyhow::Result;
+use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};
+use anyhow::{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},
@@ -43,7 +42,7 @@ impl<'a> PayloadLike for MessagePayload<'a> {
    fn get_device_token(&self) -> &'a str {
        self.device_token
    }
-    fn get_options(&self) -> &NotificationOptions<'a> {
+    fn get_options(&self) -> &NotificationOptions {
        &self.options
    }
 }
@@ -69,20 +68,16 @@ impl<'a> PayloadLike for CallStartStopPayload<'a> {
    fn get_device_token(&self) -> &'a str {
        self.device_token
    }
-    fn get_options(&self) -> &NotificationOptions<'a> {
+    fn get_options(&self) -> &NotificationOptions {
        &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,
 }

@@ -122,21 +117,15 @@ impl ApnsOutboundConsumer {
    }
 }

-#[async_trait]
-impl Consumer for ApnsOutboundConsumer {
-    async fn create(
-        db: Database,
-        authifier_db: authifier::Database,
-        connection: Arc<Connection>,
-        channel: Arc<AMQPChannel>,
-    ) -> Self {
+impl ApnsOutboundConsumer {
+    pub async fn new(db: Database) -> Result<ApnsOutboundConsumer, &'static str> {
        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()
        {
-            panic!("Missing APN keys.");
+            return Err("Missing APN keys.");
        }

        let endpoint = if config.pushd.apn.sandbox {
@@ -159,21 +148,18 @@ impl Consumer for ApnsOutboundConsumer {
        )
        .expect("could not create APN client");

-        Self {
-            db,
-            authifier_db,
-            connection,
-            channel,
-            client,
-        }
+        Ok(ApnsOutboundConsumer { db, client })
    }

-    fn channel(&self) -> &Arc<AMQPChannel> {
-        &self.channel
-    }
-
-    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;
+    async fn consume_event(
+        &mut self,
+        _channel: &AmqpChannel,
+        _deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) -> Result<()> {
+        let content = String::from_utf8(content)?;
+        let payload: PayloadToService = serde_json::from_str(content.as_str())?;

        let payload_options = NotificationOptions {
            apns_id: None,
@@ -184,15 +170,20 @@ impl Consumer for ApnsOutboundConsumer {
            apns_collapse_id: None,
        };

-        let resp = match payload.notification {
+        let resp: Result<Response, Error>;
+
+        match payload.notification {
            PayloadKind::FRReceived(alert) => {
                let loc_args = vec![Cow::from(
-                    alert.from_user.display_name.clone().unwrap_or_else(|| {
-                        format!(
+                    alert
+                        .from_user
+                        .display_name
+                        .or(Some(format!(
                            "{}#{}",
                            alert.from_user.username, alert.from_user.discriminator
-                        )
-                    }),
+                        )))
+                        .clone()
+                        .ok_or_else(|| anyhow!("missing name"))?,
                )];

                let apn_payload = Payload {
@@ -225,17 +216,20 @@ impl Consumer for ApnsOutboundConsumer {
                    "Sending friend request received for user: {:}",
                    &payload.user_id
                );
-                self.client.send(apn_payload).await
+                resp = self.client.send(apn_payload).await;
            }

            PayloadKind::FRAccepted(alert) => {
                let loc_args = vec![Cow::from(
-                    alert.accepted_user.display_name.clone().unwrap_or_else(|| {
-                        format!(
+                    alert
+                        .accepted_user
+                        .display_name
+                        .or(Some(format!(
                            "{}#{}",
                            alert.accepted_user.username, alert.accepted_user.discriminator
-                        )
-                    }),
+                        )))
+                        .clone()
+                        .ok_or_else(|| anyhow!("missing name"))?,
                )];

                let apn_payload = Payload {
@@ -268,7 +262,7 @@ impl Consumer for ApnsOutboundConsumer {
                    "Sending friend request accept for user: {:}",
                    &payload.user_id
                );
-                self.client.send(apn_payload).await
+                resp = self.client.send(apn_payload).await;
            }
            PayloadKind::Generic(alert) => {
                let apn_payload = Payload {
@@ -301,7 +295,7 @@ impl Consumer for ApnsOutboundConsumer {
                    "Sending generic notification for user: {:}",
                    &payload.user_id
                );
-                self.client.send(apn_payload).await
+                resp = self.client.send(apn_payload).await;
            }

            PayloadKind::MessageNotification(alert) => {
@@ -340,7 +334,7 @@ impl Consumer for ApnsOutboundConsumer {
                    "Sending message notification for user: {:}",
                    &payload.user_id
                );
-                self.client.send(apn_payload).await
+                resp = self.client.send(apn_payload).await;
            }

            PayloadKind::BadgeUpdate(badge) => {
@@ -355,7 +349,7 @@ impl Consumer for ApnsOutboundConsumer {
                };

                debug!("Sending badge update for user: {:}", &payload.user_id);
-                self.client.send(apn_payload).await
+                resp = self.client.send(apn_payload).await;
            }

            PayloadKind::DmCallStartEnd(alert) => {
@@ -384,37 +378,58 @@ impl Consumer for ApnsOutboundConsumer {
                    "Sending call start/stop notification for user: {:}",
                    &payload.user_id
                );
-                self.client.send(apn_payload).await
+                resp = self.client.send(apn_payload).await;
            }
-        };
+        }

-        match resp {
-            Err(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)
-                    .await
-                {
+        if let Err(err) = resp {
+            match err {
+                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)
+                        .await
+                    {
+                        revolt_config::capture_error(&err);
+                    }
+                }
+                err => {
                    revolt_config::capture_error(&err);
                }
            }
-            resp => {
-                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:?}");
+        }
+    }
+}
--- a/crates/daemons/pushd/src/consumers/outbound/fcm.rs
+++ b/crates/daemons/pushd/src/consumers/outbound/fcm.rs
@@ -1,16 +1,17 @@
-use std::{collections::HashMap, sync::Arc, time::Duration};
+use std::{collections::HashMap, time::Duration};

-use crate::utils::Consumer;
-use anyhow::{bail, Result};
+use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};
+
+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
@@ -30,12 +31,10 @@ pub enum NotificationData {
        image: Option<String>,
    },
    Message {
-        message: String,
+        title: String,
        body: String,
        image: String,
-        channel: String,
-        author: String,
-        author_name: String,
+        tag: String,
    },
    DmCallStartEnd {
        initiator_id: String,
@@ -82,19 +81,15 @@ impl NotificationData {
                }
            }
            NotificationData::Message {
-                message,
+                title,
                body,
                image,
-                channel,
-                author,
-                author_name,
+                tag,
            } => {
-                data.insert("message".to_string(), Value::String(message));
+                data.insert("title".to_string(), Value::String(title));
                data.insert("body".to_string(), Value::String(body));
                data.insert("image".to_string(), Value::String(image));
-                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));
+                data.insert("tag".to_string(), Value::String(tag));
            }
            NotificationData::DmCallStartEnd {
                initiator_id,
@@ -115,31 +110,37 @@ impl NotificationData {
    }
 }

-#[derive(Clone)]
-#[allow(unused)]
 pub struct FcmOutboundConsumer {
    db: Database,
-    authifier_db: authifier::Database,
-    connection: Arc<Connection>,
-    channel: Arc<AMQPChannel>,
    client: Client,
 }

-#[async_trait]
-impl Consumer for FcmOutboundConsumer {
-    async fn create(
-        db: Database,
-        authifier_db: authifier::Database,
-        connection: Arc<Connection>,
-        channel: Arc<AMQPChannel>,
-    ) -> Self {
+impl FcmOutboundConsumer {
+    fn format_title(&self, notification: &PushNotification) -> String {
+        // ideally this changes depending on context
+        // in a server, it would look like "Sendername, #channelname in servername"
+        // in a group, it would look like "Sendername in groupname"
+        // in a dm it should just be "Sendername".
+        // not sure how feasible all those are given the PushNotification object as it currently stands.
+
+        #[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;

-        Self {
+        Ok(FcmOutboundConsumer {
            db,
-            authifier_db,
-            connection,
-            channel,
            client: Client::new(
                Authenticator::service_account::<&str>(ServiceAccountKey {
                    key_type: Some(config.pushd.fcm.key_type),
@@ -159,27 +160,33 @@ impl Consumer for FcmOutboundConsumer {
                false,
                Duration::from_secs(5),
            ),
-        }
+        })
    }

-    fn channel(&self) -> &Arc<AMQPChannel> {
-        &self.channel
-    }
-
-    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;
+    async fn consume_event(
+        &mut self,
+        _channel: &AmqpChannel,
+        _deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) -> 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.from_user.display_name.clone().unwrap_or_else(|| {
-                    format!(
+                let name = alert
+                    .from_user
+                    .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,
@@ -196,12 +203,15 @@ impl Consumer for FcmOutboundConsumer {
            }

            PayloadKind::FRAccepted(alert) => {
-                let name = alert.accepted_user.display_name.clone().unwrap_or_else(|| {
-                    format!(
+                let name = alert
+                    .accepted_user
+                    .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,
@@ -234,12 +244,10 @@ impl Consumer for FcmOutboundConsumer {

            PayloadKind::MessageNotification(alert) => {
                let data = NotificationData::Message {
-                    message: alert.message.id,
+                    title: self.format_title(&alert),
                    body: alert.body,
                    image: alert.icon,
-                    channel: alert.message.channel,
-                    author: alert.message.author,
-                    author_name: alert.author,
+                    tag: alert.tag,
                };

                let msg = Message {
@@ -274,21 +282,43 @@ impl Consumer for FcmOutboundConsumer {
            }
        }

-        match resp {
-            Err(FcmError::Auth) => {
-                if let Err(err) = self
-                    .db
-                    .remove_push_subscription_by_session_id(&payload.session_id)
-                    .await
-                {
+        if let Err(err) = resp {
+            match err {
+                FcmError::Auth => {
+                    if let Err(err) = self
+                        .db
+                        .remove_push_subscription_by_session_id(&payload.session_id)
+                        .await
+                    {
+                        revolt_config::capture_error(&err);
+                    }
+                }
+                err => {
                    revolt_config::capture_error(&err);
                }
            }
-            res => {
-                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:?}");
+        }
+    }
+}
--- a/crates/daemons/pushd/src/consumers/outbound/vapid.rs
+++ b/crates/daemons/pushd/src/consumers/outbound/vapid.rs
@@ -1,6 +1,6 @@
-use std::{collections::HashMap, sync::Arc};
+use std::collections::HashMap;

-use crate::utils::Consumer;
+use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};

 use anyhow::{anyhow, bail, Result};
 use async_trait::async_trait;
@@ -8,60 +8,46 @@ 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: Arc<Vec<u8>>,
+    pkey: Vec<u8>,
 }

-#[async_trait]
-impl Consumer for VapidOutboundConsumer {
-    async fn create(
-        db: Database,
-        authifier_db: authifier::Database,
-        connection: Arc<Connection>,
-        channel: Arc<AMQPChannel>,
-    ) -> Self {
+impl VapidOutboundConsumer {
+    pub async fn new(db: Database) -> Result<VapidOutboundConsumer> {
        let config = revolt_config::config().await;

-        if config.pushd.vapid.private_key.is_empty() || config.pushd.vapid.public_key.is_empty() {
-            panic!("no Vapid keys present");
+        if config.pushd.vapid.private_key.is_empty() | config.pushd.vapid.public_key.is_empty() {
+            bail!("no Vapid keys present");
        }

-        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`"),
-        );
+        let web_push_private_key = engine::general_purpose::URL_SAFE_NO_PAD
+            .decode(config.pushd.vapid.private_key)
+            .expect("valid `VAPID_PRIVATE_KEY`");

-        Self {
+        Ok(VapidOutboundConsumer {
            db,
-            authifier_db,
-            connection,
-            channel,
            client: IsahcWebPushClient::new().unwrap(),
            pkey: web_push_private_key,
-        }
+        })
    }

-    fn channel(&self) -> &Arc<AMQPChannel> {
-        &self.channel
-    }
-
-    async fn consume(&self, delivery: Delivery) -> Result<()> {
-        let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;
+    async fn consume_event(
+        &mut self,
+        _channel: &AmqpChannel,
+        _deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) -> Result<()> {
+        let content = String::from_utf8(content)?;
+        let payload: PayloadToService = serde_json::from_str(content.as_str())?;

        let subscription = SubscriptionInfo {
            endpoint: payload
@@ -79,7 +65,10 @@ impl Consumer for VapidOutboundConsumer {
            },
        };

-        let payload_body = match payload.notification {
+        #[allow(clippy::needless_late_init)]
+        let payload_body: String;
+
+        match payload.notification {
            PayloadKind::FRReceived(alert) => {
                let name = alert
                    .from_user
@@ -94,7 +83,7 @@ impl Consumer for VapidOutboundConsumer {
                let mut body = HashMap::new();
                body.insert("body", format!("{} sent you a friend request", name));

-                serde_json::to_string(&body)?
+                payload_body = serde_json::to_string(&body)?;
            }
            PayloadKind::FRAccepted(alert) => {
                let name = alert
@@ -110,10 +99,14 @@ impl Consumer for VapidOutboundConsumer {
                let mut body = HashMap::new();
                body.insert("body", format!("{} accepted your friend request", name));

-                serde_json::to_string(&body)?
+                payload_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()
@@ -139,41 +132,59 @@ impl Consumer for VapidOutboundConsumer {
                    _ => bail!("Invalid DmCallStart/End channel type"),
                }

-                serde_json::to_string(&body)?
+                payload_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);

-        let mut builder = WebPushMessageBuilder::new(&subscription);
-        builder.set_vapid_signature(signature);
+                    builder.set_payload(ContentEncoding::AesGcm, payload_body.as_bytes());

-        builder.set_payload(ContentEncoding::AesGcm, payload_body.as_bytes());
+                    match builder.build() {
+                        Ok(msg) => {
+                            if let Err(err) = self.client.send(msg).await {
+                                if err == WebPushError::Unauthorized {
+                                    self.db
+                                        .remove_push_subscription_by_session_id(&payload.session_id)
+                                        .await?;
+                                }
+                            }

-        let msg = builder.build()?;
-
-        match self.client.send(msg).await {
-            Err(WebPushError::Unauthorized) => {
-                if let Err(err) = self
-                    .db
-                    .remove_push_subscription_by_session_id(&payload.session_id)
-                    .await
-                {
-                    revolt_config::capture_error(&err);
+                            Ok(())
+                        }
+                        Err(err) => Err(err.into()),
+                    }
                }
-            }
-            res => {
-                res?;
-            }
-        };
-
-        Ok(())
+                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:?}");
+        }
    }
 }
--- a/crates/daemons/pushd/src/main.rs
+++ b/crates/daemons/pushd/src/main.rs
@@ -1,16 +1,17 @@
 #[macro_use]
 extern crate log;

-use std::sync::Arc;
-
-use lapin::{
-    options::{BasicConsumeOptions, ExchangeDeclareOptions, QueueBindOptions, QueueDeclareOptions},
-    types::{AMQPValue, FieldTable},
-    Channel, Connection, ConnectionProperties,
+use amqprs::{
+    channel::{
+        BasicConsumeArguments, Channel, ExchangeDeclareArguments, QueueBindArguments,
+        QueueDeclareArguments,
+    },
+    connection::{Connection, OpenConnectionArguments},
+    consumer::AsyncConsumer,
+    FieldTable,
 };
 use revolt_config::{config, Settings};
-use revolt_database::Database;
-use tokio::signal::ctrl_c;
+use tokio::sync::Notify;

 mod consumers;
 mod utils;
@@ -23,8 +24,6 @@ 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
@@ -44,24 +43,7 @@ async fn main() {
        panic!("Mongo is not in use, can't connect via authifier!")
    }

-    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();
+    let mut connections: Vec<(Channel, Connection)> = Vec::new();

    // An explainer of how this works:
    // The inbound connections are on separate routing keys, such that they only receive the proper payload
@@ -72,178 +54,171 @@ 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.

-    channels.push(
-        make_queue_and_consume::<GenericConsumer>(
-            &db,
-            &authifier,
-            &connection,
+    let config = config().await;
+
+    // inbound: generic
+    connections.push(
+        make_queue_and_consume(
            &config,
            &config.pushd.generic_queue,
-            &config.pushd.get_generic_routing_key(),
+            config.pushd.get_generic_routing_key().as_str(),
            None,
+            GenericConsumer::new(db.clone(), authifier.clone()),
        )
        .await,
    );

-    channels.push(
-        make_queue_and_consume::<MessageConsumer>(
-            &db,
-            &authifier,
-            &connection,
+    // inbound: messages
+    connections.push(
+        make_queue_and_consume(
            &config,
            &config.pushd.message_queue,
-            &config.pushd.get_message_routing_key(),
+            config.pushd.get_message_routing_key().as_str(),
            None,
+            MessageConsumer::new(db.clone(), authifier.clone()),
        )
        .await,
    );

-    channels.push(
-        make_queue_and_consume::<FRReceivedConsumer>(
-            &db,
-            &authifier,
-            &connection,
+    // inbound: FR received
+    connections.push(
+        make_queue_and_consume(
            &config,
            &config.pushd.fr_received_queue,
-            &config.pushd.get_fr_received_routing_key(),
+            config.pushd.get_fr_received_routing_key().as_str(),
            None,
+            FRReceivedConsumer::new(db.clone(), authifier.clone()),
        )
        .await,
    );

-    channels.push(
-        make_queue_and_consume::<FRAcceptedConsumer>(
-            &db,
-            &authifier,
-            &connection,
+    // inbound: FR accepted
+    connections.push(
+        make_queue_and_consume(
            &config,
            &config.pushd.fr_accepted_queue,
-            &config.pushd.get_fr_accepted_routing_key(),
+            config.pushd.get_fr_accepted_routing_key().as_str(),
            None,
+            FRAcceptedConsumer::new(db.clone(), authifier.clone()),
        )
        .await,
    );

-    channels.push(
-        make_queue_and_consume::<MassMessageConsumer>(
-            &db,
-            &authifier,
-            &connection,
+    // inbound: Mass Mentions
+    connections.push(
+        make_queue_and_consume(
            &config,
            &config.pushd.mass_mention_queue,
-            &config.pushd.get_mass_mention_routing_key(),
+            config.pushd.get_mass_mention_routing_key().as_str(),
            None,
+            MassMessageConsumer::new(db.clone(), authifier.clone()),
        )
        .await,
    );

-    channels.push(
-        make_queue_and_consume::<DmCallConsumer>(
-            &db,
-            &authifier,
-            &connection,
+    // inbound: Dm Calls
+    connections.push(
+        make_queue_and_consume(
            &config,
            &config.pushd.dm_call_queue,
-            &config.pushd.get_dm_call_routing_key(),
+            config.pushd.get_dm_call_routing_key().as_str(),
            None,
+            DmCallConsumer::new(db.clone(), authifier.clone()),
        )
        .await,
    );

    if !config.pushd.apn.pkcs8.is_empty() {
-        channels.push(
-            make_queue_and_consume::<ApnsOutboundConsumer>(
-                &db,
-                &authifier,
-                &connection,
+        connections.push(
+            make_queue_and_consume(
                &config,
                &config.pushd.apn.queue,
                &config.pushd.apn.queue,
                None,
+                ApnsOutboundConsumer::new(db.clone()).await.unwrap(),
            )
            .await,
        );

-        let mut table = FieldTable::default();
-        table.insert("x-message-deduplication".into(), AMQPValue::Boolean(true));
+        let mut table = FieldTable::new();
+        table.insert("x-message-deduplication".try_into().unwrap(), "true".into());

-        channels.push(
-            make_queue_and_consume::<AckConsumer>(
-                &db,
-                &authifier,
-                &connection,
+        connections.push(
+            make_queue_and_consume(
                &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() {
-        channels.push(
-            make_queue_and_consume::<FcmOutboundConsumer>(
-                &db,
-                &authifier,
-                &connection,
+        connections.push(
+            make_queue_and_consume(
                &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() {
-        channels.push(
-            make_queue_and_consume::<VapidOutboundConsumer>(
-                &db,
-                &authifier,
-                &connection,
+        connections.push(
+            make_queue_and_consume(
                &config,
                &config.pushd.vapid.queue,
                &config.pushd.vapid.queue,
                None,
+                VapidOutboundConsumer::new(db.clone()).await.unwrap(),
            )
            .await,
-        );
+        )
    }

-    ctrl_c().await.unwrap();
+    let guard = Notify::new();
+    guard.notified().await;

-    for channel in channels {
-        let _ = channel.close(0, "close".into()).await;
+    for (channel, conn) in connections {
+        channel.close().await.expect("Unable to close channel");
+        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>,
-) -> Arc<Channel>
+    consumer: F,
+) -> (Channel, Connection)
 where
-    F: Consumer,
+    F: AsyncConsumer + Send + 'static,
 {
-    let channel = Arc::new(connection.create_channel().await.unwrap());
+    let connection = Connection::open(&OpenConnectionArguments::new(
+        &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(
-            config.pushd.exchange.clone().into(),
-            lapin::ExchangeKind::Direct,
-            ExchangeDeclareOptions {
-                durable: true,
-                ..Default::default()
-            },
-            FieldTable::default(),
+            ExchangeDeclareArguments::new(&config.pushd.exchange, "direct")
+                .durable(true)
+                .finish(),
        )
        .await
-        .expect("Failed to declare exchange");
+        .expect("Failed to declare pushd exchange");

    let mut queue_name = queue_name.to_string();

@@ -255,59 +230,35 @@ where

    let queue_name = queue_name.as_str();

-    let args = QueueDeclareOptions {
-        durable: true,
-        ..Default::default()
-    };
+    let mut args = QueueDeclareArguments::new(queue_name);
+    args.durable(true);
+
+    if let Some(arg) = queue_args {
+        args.arguments(arg);
+    }
+
+    let args = args.finish();
+    _ = channel.queue_declare(args).await.unwrap().unwrap();

    channel
-        .queue_declare(queue_name.into(), args, queue_args.unwrap_or_default())
-        .await
-        .unwrap();
-
-    channel
-        .queue_bind(
-            queue_name.into(),
-            config.pushd.exchange.clone().into(),
-            routing_key.into(),
-            QueueBindOptions::default(),
-            FieldTable::default(),
-        )
+        .queue_bind(QueueBindArguments::new(
+            queue_name,
+            &config.pushd.exchange,
+            routing_key,
+        ))
        .await
        .expect(
            "This probably means the revolt.notifications exchange does not exist in rabbitmq!",
        );

-    let consumer = channel
-        .basic_consume(
-            queue_name.into(),
-            "".into(),
-            BasicConsumeOptions {
-                no_ack: true,
-                ..Default::default()
-            },
-            FieldTable::default(),
-        )
-        .await
-        .unwrap();
+    let args = BasicConsumeArguments::new(queue_name, "")
+        .manual_ack(false)
+        .finish();
+
+    let routing_key = channel.basic_consume(consumer, args).await.unwrap();
    info!(
        "Consuming routing key {} as queue {}, tag {}",
-        routing_key,
-        queue_name,
-        consumer.tag()
+        routing_key, queue_name, routing_key
    );
-
-    let delegate = Delegate(
-        F::create(
-            db.clone(),
-            authifier_db.clone(),
-            connection.clone(),
-            channel.clone(),
-        )
-        .await,
-    );
-
-    consumer.set_delegate(delegate);
-
-    channel
+    (channel, connection)
 }
--- a/crates/daemons/pushd/src/utils/consumer.rs
+++ b/crates/daemons/pushd/src/utils/consumer.rs
@@ -1,91 +0,0 @@
-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:?}") }),
-        }
-    }
-}
--- a/crates/daemons/pushd/src/utils/mod.rs
+++ b/crates/daemons/pushd/src/utils/mod.rs
@@ -1,5 +1,2 @@
 mod renderer;
-mod consumer;
-
 pub use renderer::render_notification_content;
-pub use consumer::{Consumer, Delegate};
--- a/crates/daemons/voice-ingress/Cargo.toml
+++ b/crates/daemons/voice-ingress/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-voice-ingress"
-version = "0.13.7"
+version = "0.12.1"
 license = "AGPL-3.0-or-later"
 edition = "2021"
 publish = false
@@ -44,3 +44,6 @@ revolt-permissions = { workspace = true }
 livekit-api = { workspace = true }
 livekit-protocol = { workspace = true }
 livekit-runtime = { workspace = true, features = ["tokio"] }
+
+# RabbitMQ
+amqprs = { workspace = true }
--- a/crates/delta/Cargo.toml
+++ b/crates/delta/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-delta"
-version = "0.13.7"
+version = "0.12.1"
 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
-lapin = { workspace = true, features = ["tokio"] }
+amqprs = { workspace = true }

 # core
 authifier = { workspace = true }
--- a/crates/delta/src/main.rs
+++ b/crates/delta/src/main.rs
@@ -9,7 +9,8 @@ pub mod routes;
 pub mod util;

 use revolt_config::config;
-use revolt_database::{AMQP, events::client::EventV1};
+use revolt_database::events::client::EventV1;
+use revolt_database::AMQP;
 use revolt_ratelimits::rocket as ratelimiter;
 use rocket::{Build, Rocket};
 use rocket_cors::{AllowedOrigins, CorsOptions};
@@ -17,6 +18,10 @@ 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;
@@ -31,6 +36,7 @@ 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
@@ -90,8 +96,33 @@ 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 amqp = AMQP::new_auto().await;
+    let channel = connection
+        .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 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());
@@ -121,7 +152,6 @@ 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()
        })
 }
--- a/crates/delta/src/routes/channels/message_send.rs
+++ b/crates/delta/src/routes/channels/message_send.rs
@@ -1,14 +1,12 @@
 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};
@@ -86,16 +84,6 @@ 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() {
@@ -104,29 +92,10 @@ 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
--- a/crates/delta/src/util/test.rs
+++ b/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,6 +25,8 @@ 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");
@@ -47,7 +49,19 @@ impl TestHarness {
            .expect("`Authifier`")
            .clone();

-        let amqp = AMQP::new_auto().await;
+        let connection = amqprs::connection::Connection::open(
+            &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,
--- a/crates/services/autumn/Cargo.toml
+++ b/crates/services/autumn/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-autumn"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 publish = false
--- a/crates/services/gifbox/Cargo.toml
+++ b/crates/services/gifbox/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-gifbox"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 publish = false
--- a/crates/services/january/Cargo.toml
+++ b/crates/services/january/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-january"
-version = "0.13.7"
+version = "0.12.1"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 publish = false
--- a/crates/services/january/src/requests.rs
+++ b/crates/services/january/src/requests.rs
@@ -4,7 +4,6 @@ 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,
 };
@@ -12,7 +11,6 @@ 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,
@@ -23,7 +21,6 @@ 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())
@@ -61,73 +58,18 @@ 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,
@@ -333,12 +275,7 @@ impl Request {
        let mut url = url;
        let url_host_str = url.host_str().ok_or(create_error!(ProxyError))?.to_string();

-        let mut blocker = Request::url_is_blacklisted(&url).await?;
-
-        if blocker.blocked {
-            return Err(create_error!(InvalidOperation));
-        }
-
+        Request::url_is_blacklisted(&url).await?;
        let mut redirect_count = 0;

        loop {
@@ -367,13 +304,9 @@ impl Request {
                    let location = location.to_str().map_err(|_| create_error!(ProxyError))?;
                    url = Url::from_str(location).to_internal_error()?;

-                    blocker = Request::url_is_blacklisted(&url).await?;
-
-                    if blocker.blocked {
-                        return Err(create_error!(InvalidOperation));
+                    if !Request::url_is_blacklisted(&url).await? {
+                        continue;
                    }
-
-                    continue;
                } else {
                    return Err(create_error!(ProxyError));
                }
@@ -418,25 +351,17 @@ impl Request {
        Ok(Request::exists(proper_url).await)
    }

-    pub async fn url_is_blacklisted(url: &Url) -> Result<IPRequest> {
-        let resolved_address: IpAddr;
-
+    pub async fn url_is_blacklisted(url: &Url) -> Result<bool> {
        if let Some(host) = url.host() {
            match host {
                Host::Ipv4(ipv4) => {
-                    resolved_address = ipv4.into();
-                    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()) {
+                    let url_str = ipv4.to_string();
+                    if !IP_BLOCKLIST.is_allowed(&url_str) {
                        return Err(create_error!(InvalidOperation));
                    }
                }
                Host::Domain(domain) => {
-                    let domain = domain.to_string();
+                    let mut domain = domain.to_string();

                    let config = config().await;

@@ -447,22 +372,14 @@ impl Request {
                        return Err(create_error!(InvalidOperation));
                    }

+                    if !domain.contains(":") {
+                        domain += ":80";
+                    }
+
                    // Second step: resolve the IP and check the blocklist
-                    let resolver = CachedDnsResolver {};
-                    if let Ok(mut resolved_ip) = resolver
-                        .resolve(
-                            Name::from_str(&domain)
-                                .map_err(|_| create_error!(ProxyError))
-                                .unwrap(),
-                        )
-                        .await
-                    {
+                    if let Ok(mut resolved_ip) = tokio::net::lookup_host(domain.clone()).await {
                        if let Some(resolved_ip) = resolved_ip.next() {
-                            resolved_address = resolved_ip.ip();
-                            let resolved_string = resolved_address.to_string();
-                            if !IP_BLOCKLIST.is_allowed(&resolved_string)
-                                || resolved_string.contains("::ffff:")
-                            {
+                            if !IP_BLOCKLIST.is_allowed(&resolved_ip.ip().to_string()) {
                                return Err(create_error!(InvalidOperation));
                            }
                        } else {
@@ -472,15 +389,10 @@ impl Request {
                        return Err(create_error!(ProxyError));
                    }
                }
+                _ => (),
            }
-        } else {
-            return Err(create_error!(ProxyError));
        };

-        Ok(IPRequest {
-            url: url.clone(),
-            ip: resolved_address,
-            blocked: false,
-        })
+        Ok(false)
    }
 }
--- a/version.txt
+++ b/version.txt
@@ -1 +1 @@
-0.13.7
+0.12.1
Author	SHA1	Message	Date
IAmTomahawkx	173a3effda	fix: minio wants the default region	2026-05-08 16:47:20 -07:00
Infiland	34f05f4b2f	fix: avoid stack overflow in database tests Signed-off-by: Infiland <ljubica.citydesign@gmail.com>	2026-05-08 16:00:27 -07:00
@@ -1 +1 @@
 .13.7
 .12.1