feat: publish events over amqp

Signed-off-by: Zomatree <me@zomatree.live>
fix: openapi using old naming (#777 )
2026-07-06 03:06:04 +00:00 · 2026-06-19 21:05:22 +01:00 · 2026-06-07 16:56:00 -07:00 · 2026-06-03 11:06:38 -07:00 · 2026-06-02 17:42:04 -07:00 · 2026-06-01 13:28:29 +00:00
68 changed files with 1703 additions and 1603 deletions
--- a/.github/workflows/docker-cleanup.yaml
+++ b/.github/workflows/docker-cleanup.yaml
@@ -0,0 +1,46 @@
+name: Docker PR Image Cleanup
+
+on:
+  pull_request:
+    types:
+      - closed
+
+permissions:
+  contents: read
+  packages: write
+
+concurrency:
+  group: docker-cleanup-${{ github.event.pull_request.number }}
+  cancel-in-progress: false
+
+jobs:
+  cleanup:
+    runs-on: ubuntu-latest
+    if: ${{ !github.event.pull_request.head.repo.fork }}
+    strategy:
+      fail-fast: false
+      matrix:
+        package:
+          - base
+          - api
+          - events
+          - file-server
+          - proxy
+          - gifbox
+          - crond
+          - pushd
+          - voice-ingress
+    steps:
+      - env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          ORG: stoatchat
+          PACKAGE: ${{ matrix.package }}
+          TAG: pr-${{ github.event.pull_request.number }}
+        run: |
+          set -euo pipefail
+          gh api --paginate \
+            "/orgs/${ORG}/packages/container/${PACKAGE}/versions" \
+            --jq ".[] | select(.metadata.container.tags | index(\"${TAG}\")) | .id" \
+          | while read -r id; do
+              gh api -X DELETE "/orgs/${ORG}/packages/container/${PACKAGE}/versions/${id}"
+            done
--- a/.github/workflows/docker.yaml
+++ b/.github/workflows/docker.yaml
@@ -5,8 +5,6 @@ on:
    tags:
      - "*"
  pull_request:
-    paths:
-      - "Dockerfile"
  workflow_dispatch:

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

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

  publish:
    runs-on: arc-runner-set
-    if: github.event_name != 'pull_request'
+    if: ${{ github.event_name != 'pull_request' || !github.event.pull_request.head.repo.fork }}
    name: Publish Docker images
    steps:
      # Configure build environment
@@ -59,6 +57,15 @@ jobs:
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

+      - name: Determine base image tag
+        id: base
+        run: |
+          if [ "${{ github.event_name }}" = "pull_request" ]; then
+            echo "tag=pr-${{ github.event.number }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tag=latest" >> "$GITHUB_OUTPUT"
+          fi
+
      # Build the image
      - name: Build base image
        uses: docker/build-push-action@v4
@@ -66,7 +73,9 @@ jobs:
          context: .
          push: true
          platforms: linux/amd64,linux/arm64
-          tags: ghcr.io/${{ github.repository_owner }}/base:latest
+          tags: ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
+          cache-from: type=gha,scope=buildx-base-multi-arch
+          cache-to: type=gha,scope=buildx-base-multi-arch,mode=max

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

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

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

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

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

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

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

      # stoatchat/voice-ingress
@@ -217,5 +226,5 @@ jobs:
          file: crates/daemons/voice-ingress/Dockerfile
          tags: ${{ steps.meta-voice-ingress.outputs.tags }}
          build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
+            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:${{ steps.base.outputs.tag }}
          labels: ${{ steps.meta-voice-ingress.outputs.labels }}
--- a/.github/workflows/publish-crates.yml
+++ b/.github/workflows/publish-crates.yml
@@ -21,4 +21,6 @@ jobs:
          github-token: ${{ secrets.GITHUB_TOKEN }}

      - name: Publish
+        env:
+          CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
        run: mise publish --workspace
--- a/.github/workflows/renovate.yml
+++ b/.github/workflows/renovate.yml
@@ -0,0 +1,30 @@
+# DO NOT EDIT DIRECTLY IN REPOSITORY
+# Managed in Terraform templates
+
+name: Renovate
+on:
+  workflow_dispatch:
+  schedule:
+    - cron: '0/15 * * * *'
+jobs:
+  renovate:
+    runs-on: ubuntu-latest
+    steps:
+      - id: app-token
+        uses: actions/create-github-app-token@v2
+        with:
+          app-id: ${{ secrets.GH_STOAT_RELEASE_APP_ID }}
+          private-key: ${{ secrets.GH_STOAT_RELEASE_APP_PRIVATE_KEY }}
+
+      - name: Setup Mise
+        uses: immich-app/devtools/actions/use-mise@7b8610a904d57da241e4ddba17fa62b62b15aed4 # use-mise-action-v2.0.2
+        with:
+          github_token: ${{ steps.app-token.outputs.token }}
+
+      - name: Self-hosted Renovate
+        uses: renovatebot/github-action@v46.1.14
+        with:
+          token: '${{ steps.app-token.outputs.token }}'
+        env:
+          RENOVATE_PLATFORM_COMMIT: 'enabled'
+          RENOVATE_REPOSITORIES: '${{ github.repository }}'
--- a/.github/workflows/rust.yaml
+++ b/.github/workflows/rust.yaml
@@ -37,6 +37,7 @@ jobs:
      - name: Reference Test
        env:
          TEST_DB: REFERENCE
+        continue-on-error: ${{ github.ref_name == 'main' }}
        run: |
          mise test

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

--- a/.release-please-manifest.json
+++ b/.release-please-manifest.json
@@ -1,3 +1,3 @@
 {
-    ".": "0.12.1"
+    ".": "0.13.7"
 }
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,97 @@
 # 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,31 +174,6 @@ dependencies = [
 "url",
 ]

-[[package]]
-name = "amqp_serde"
-version = "0.4.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f5f450f572a1ec4cdb4af7af09cbd0c7c3e1b9da2bfc7414c059a780993a8e16"
-dependencies = [
- "bytes 1.11.1",
- "serde",
- "serde_bytes",
-]
-
-[[package]]
-name = "amqprs"
-version = "1.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3f1b4afcbd862e16c272b7625b6b057930b052d63c720bc90f6afab0d9abe8a8"
-dependencies = [
- "amqp_serde",
- "async-trait",
- "bytes 1.11.1",
- "serde",
- "serde_bytes_ng",
- "tokio 1.51.0",
-]
-
 [[package]]
 name = "android_system_properties"
 version = "0.1.5"
@@ -570,7 +545,7 @@ dependencies = [
 "futures-util",
 "log",
 "pin-project-lite 0.2.17",
- "tungstenite",
+ "tungstenite 0.17.3",
 ]

 [[package]]
@@ -5048,11 +5023,14 @@ 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",
 ]

@@ -7419,16 +7397,22 @@ 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",
@@ -7504,7 +7488,7 @@ dependencies = [

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

 [[package]]
 name = "revolt-voice-ingress"
-version = "0.12.1"
+version = "0.13.7"
 dependencies = [
- "amqprs",
 "async-std",
 "chrono",
 "futures",
@@ -8978,15 +8961,6 @@ dependencies = [
 "serde_core",
 ]

-[[package]]
-name = "serde_bytes_ng"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bdb0ebce8684e2253f964e8b6ce51f0ccc6666bbb448fb4a6788088bda6544b6"
-dependencies = [
- "serde",
-]
-
 [[package]]
 name = "serde_core"
 version = "1.0.228"
@@ -9891,6 +9865,21 @@ 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"
@@ -10146,6 +10135,26 @@ 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,7 +159,6 @@ opentelemetry-appender-tracing = "0.31.1"
 authifier = "1.0.16"

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

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

 # Local packages
-revolt-coalesced = { version = "0.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" }
+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" }
--- a/compose.yml
+++ b/compose.yml
@@ -34,6 +34,7 @@ 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.12.1"
+version = "0.13.7"
 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,6 +1,7 @@
 use std::collections::{HashMap, HashSet};

 use futures::future::join_all;
+use redis_kiss::AsyncCommands;
 use revolt_database::{
    events::client::{EventV1, ReadyPayloadFields},
    util::permissions::DatabasePermissionQuery,
--- 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::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
+use redis_kiss::{get_connection, AsyncCommands, PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
 use revolt_config::report_internal_error;
 use revolt_database::{
    events::{client::EventV1, server::ClientMessage},
@@ -32,6 +32,7 @@ use sentry::Level;

 use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
 use crate::events::state::{State, SubscriptionStateChange};
+use revolt_models::v0;

 type WsReader = SplitStream<WebSocketStream<TcpStream>>;
 type WsWriter = SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>;
@@ -128,6 +129,14 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
        return;
    }

+    let slowmodes = fetch_user_slowmodes(&user_id).await.unwrap_or_default();
+    if !slowmodes.is_empty() {
+        let event = EventV1::UserSlowmodes { slowmodes };
+        if report_internal_error!(write.send(config.encode(&event)).await).is_err() {
+            return;
+        }
+    }
+
    // Create presence session.
    let (first_session, session_id) = create_session(&user_id, 0).await;

@@ -527,4 +536,43 @@ 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.12.1"
+version = "0.13.7"
 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.12.1"
+version = "0.13.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
--- a/crates/core/config/Revolt.toml
+++ b/crates/core/config/Revolt.toml
@@ -30,10 +30,11 @@ host = "rabbit"
 port = 5672
 username = "rabbituser"
 password = "rabbitpass"
-default_exchange = "revolt"
+default_exchange = "revolt.default"

 [rabbit.queues]
 acks = "internal.ack"
+events = "internal.event"

 [api]

--- a/crates/core/config/src/lib.rs
+++ b/crates/core/config/src/lib.rs
@@ -125,6 +125,7 @@ pub struct Database {
 #[derive(Deserialize, Debug, Clone)]
 pub struct RabbitQueues {
    pub acks: String,
+    pub events: String,
 }

 #[derive(Deserialize, Debug, Clone)]
--- a/crates/core/database/Cargo.toml
+++ b/crates/core/database/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-database"
-version = "0.12.1"
+version = "0.13.7"
 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
-amqprs = { workspace = true }
+lapin = { workspace = true, features = ["tokio"] }

 # Voice
-livekit-api = { workspace = true, optional = true }
+livekit-api = { workspace = true, features = ["rustls-tls-native-roots"], 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,98 +1,92 @@
-use std::collections::HashSet;
-
-use crate::events::rabbit::*;
-use crate::User;
-use amqprs::channel::{
-    BasicPublishArguments, ExchangeDeclareArguments, ExchangeType, QueueBindArguments,
-    QueueDeclareArguments,
+use std::{
+    collections::HashSet,
+    sync::{Arc, OnceLock},
 };
-use amqprs::connection::OpenConnectionArguments;
-use amqprs::{channel::Channel, connection::Connection, error::Error as AMQPError};
-use amqprs::{BasicProperties, FieldTable};
+
+use crate::events::{client::EventV1, rabbit::*};
+use crate::User;
+use lapin::{
+    options::BasicPublishOptions,
+    protocol::basic::AMQPProperties,
+    types::{AMQPValue, FieldTable},
+    BasicProperties, Channel, Connection, ConnectionProperties, Error as AMQPError,
+};
+use revolt_config::config;
 use revolt_models::v0::PushNotification;
 use revolt_presence::filter_online;
+use revolt_result::Result;

 use serde_json::to_string;

+static AMQP_INSTANCE: OnceLock<AMQP> = OnceLock::new();
+
+pub fn get_amqp() -> &'static AMQP {
+    AMQP_INSTANCE.get().expect("No AMQP instance set.")
+}
+
 #[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>,
+    publish_event: Arc<Channel>,
    #[allow(unused)]
-    connection: Connection,
-    channel: Channel,
+    connection: Arc<Connection>,
 }

 impl AMQP {
-    pub fn new(connection: Connection, channel: Channel) -> AMQP {
-        AMQP {
+    pub async fn new(connection: Arc<Connection>) -> Self {
+        let this = 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,
+            publish_event: Self::create_channel(&connection).await,
            connection,
-            channel,
-        }
+        };
+
+        let _ = AMQP_INSTANCE.set(this.clone());
+
+        this
    }

-    pub async fn new_auto() -> AMQP {
+    pub async fn new_auto() -> Self {
        let config = revolt_config::config().await;

-        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(),
+        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 declare exchange");
+            .expect("Failed to connect to RabbitMQ"),
+        );

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

-    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(())
+    async fn create_channel(connection: &Connection) -> Arc<Channel> {
+        Arc::new(
+            connection
+                .create_channel()
+                .await
+                .expect("Failed to create channel"),
+        )
    }

    pub async fn friend_request_accepted(
@@ -112,19 +106,20 @@ impl AMQP {
            config.pushd.get_fr_accepted_routing_key(),
            payload
        );
-        self.channel
+
+        self.friend_request_accepted
            .basic_publish(
-                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(),
-                ),
+                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),
            )
-            .await
+            .await?;
+
+        Ok(())
    }

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

-        self.channel
+        self.friend_request_received
            .basic_publish(
-                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(),
-                ),
+                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),
            )
-            .await
+            .await?;
+
+        Ok(())
    }

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

-        self.channel
+        self.generic_message
            .basic_publish(
-                BasicProperties::default()
-                    .with_content_type("application/json")
-                    .with_persistence(true)
-                    .finish(),
-                payload.into(),
-                BasicPublishArguments::new(
-                    &config.pushd.exchange,
-                    &config.pushd.get_generic_routing_key(),
-                ),
+                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),
            )
-            .await
+            .await?;
+
+        Ok(())
    }

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

-        self.channel
+        self.message_sent
            .basic_publish(
-                BasicProperties::default()
-                    .with_content_type("application/json")
-                    .with_persistence(true)
-                    .finish(),
-                payload.into(),
-                BasicPublishArguments::new(
-                    &config.pushd.exchange,
-                    &config.pushd.get_message_routing_key(),
-                ),
+                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),
            )
-            .await
+            .await?;
+
+        Ok(())
    }

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

-        self.channel
+        self.mass_mention_message_sent
            .basic_publish(
-                BasicProperties::default()
-                    .with_content_type("application/json")
-                    .with_persistence(true)
-                    .finish(),
-                payload.into(),
-                BasicPublishArguments::new(&config.pushd.exchange, routing_key.as_str()),
+                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),
            )
-            .await
+            .await?;
+
+        Ok(())
    }

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

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

-        self.channel
+        self.ack_notification_message
            .basic_publish(
-                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),
+                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),
            )
-            .await
+            .await?;
+
+        Ok(())
    }

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

-        self.channel
+        self.dm_call_updated
            .basic_publish(
-                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(),
-                ),
+                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),
            )
-            .await
+            .await?;
+
+        Ok(())
    }

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

-        self.channel
+        self.process_ack
            .basic_publish(
-                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,
-                ),
+                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),
            )
-            .await
+            .await?;
+
+        Ok(())
+    }
+
+    pub async fn publish_event(&self, channel: String, event: &EventV1) -> Result<(), AMQPError> {
+        let mut headers = FieldTable::default();
+        headers.insert("c".into(), AMQPValue::LongString(channel.into()));
+
+        let config = config().await;
+
+        self.publish_event
+            .basic_publish(
+                config.rabbit.default_exchange.clone().into(),
+                config.rabbit.queues.events.into(),
+                BasicPublishOptions::default(),
+                &serde_json::to_vec(event).unwrap(),
+                BasicProperties::default().with_headers(headers),
+            )
+            .await?;
+
+        Ok(())
    }
 }
--- a/crates/core/database/src/amqp/mod.rs
+++ b/crates/core/database/src/amqp/mod.rs
@@ -1,2 +1,4 @@
 #[allow(clippy::module_inception)]
 pub mod amqp;
+
+pub use amqp::{AMQP, get_amqp};
--- a/crates/core/database/src/events/client.rs
+++ b/crates/core/database/src/events/client.rs
@@ -3,10 +3,15 @@ use revolt_result::Error;
 use serde::{Deserialize, Serialize};

 use revolt_models::v0::{
-    AppendMessage, Channel, ChannelUnread, ChannelVoiceState, Emoji, FieldsChannel, FieldsMember, FieldsMessage, FieldsRole, FieldsServer, FieldsUser, FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialEmoji, PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialUserVoiceState, PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings, UserVoiceState, Webhook
+    AppendMessage, Channel, ChannelSlowmode, ChannelUnread, ChannelVoiceState, Emoji,
+    FieldsChannel, FieldsMember, FieldsMessage, FieldsRole, FieldsServer, FieldsUser,
+    FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialEmoji,
+    PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialUserVoiceState,
+    PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings,
+    UserVoiceState, Webhook,
 };

-use crate::Database;
+use crate::{Database, amqp::get_amqp};

 /// Ping Packet
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -51,9 +56,13 @@ impl Default for ReadyPayloadFields {
 #[serde(tag = "type")]
 pub enum EventV1 {
    /// Multiple events
-    Bulk { v: Vec<EventV1> },
+    Bulk {
+        v: Vec<EventV1>,
+    },
    /// Error event
-    Error { data: Error },
+    Error {
+        data: Error,
+    },

    /// Successfully authenticated
    Authenticated,
@@ -84,7 +93,9 @@ pub enum EventV1 {
    },

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

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

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

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

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

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

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

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

    /// Update existing server member
    ServerMemberUpdate {
@@ -187,10 +206,16 @@ pub enum EventV1 {
    },

    /// Server role deleted
-    ServerRoleDelete { id: String, role_id: String },
+    ServerRoleDelete {
+        id: String,
+        role_id: String,
+    },

    /// Server roles ranks updated
-    ServerRoleRanksUpdate { id: String, ranks: Vec<String> },
+    ServerRoleRanksUpdate {
+        id: String,
+        ranks: Vec<String>,
+    },

    /// Update existing user
    UserUpdate {
@@ -202,9 +227,15 @@ pub enum EventV1 {
    },

    /// Relationship with another user changed
-    UserRelationship { id: String, user: User },
+    UserRelationship {
+        id: String,
+        user: User,
+    },
    /// Settings updated remotely
-    UserSettingsUpdate { id: String, update: UserSettings },
+    UserSettingsUpdate {
+        id: String,
+        update: UserSettings,
+    },

    /// User has been platform banned or deleted their account
    ///
@@ -215,7 +246,10 @@ pub enum EventV1 {
    /// - Server Memberships
    ///
    /// User flags are specified to explain why a wipe is occurring though not all reasons will necessarily ever appear.
-    UserPlatformWipe { user_id: String, flags: i32 },
+    UserPlatformWipe {
+        user_id: String,
+        flags: i32,
+    },
    /// New emoji
    EmojiCreate(Emoji),

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

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

    /// New report
    ReportCreate(Report),
@@ -242,19 +278,33 @@ pub enum EventV1 {
    },

    /// Delete channel
-    ChannelDelete { id: String },
+    ChannelDelete {
+        id: String,
+    },

    /// User joins a group
-    ChannelGroupJoin { id: String, user: String },
+    ChannelGroupJoin {
+        id: String,
+        user: String,
+    },

    /// User leaves a group
-    ChannelGroupLeave { id: String, user: String },
+    ChannelGroupLeave {
+        id: String,
+        user: String,
+    },

    /// User started typing in a channel
-    ChannelStartTyping { id: String, user: String },
+    ChannelStartTyping {
+        id: String,
+        user: String,
+    },

    /// User stopped typing in a channel
-    ChannelStopTyping { id: String, user: String },
+    ChannelStopTyping {
+        id: String,
+        user: String,
+    },

    /// User acknowledged message in channel
    ChannelAck {
@@ -274,7 +324,9 @@ pub enum EventV1 {
    },

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

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

 impl EventV1 {
@@ -316,8 +372,16 @@ impl EventV1 {
        #[cfg(debug_assertions)]
        info!("Publishing event to {channel}: {self:?}");

-        #[cfg(debug_assertions)]
-        redis_kiss::publish(channel, self).await.unwrap();
+        // #[cfg(debug_assertions)]
+        // redis_kiss::publish(channel, self).await.unwrap();
+
+        if let Err(e) = get_amqp().publish_event(channel, &self).await {
+            if cfg!(debug_assertions) {
+                panic!("{e:?}");
+            } else {
+                log::error!("{e:?}");
+            };
+        };
    }

    /// Publish user event
--- 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.into())
+        .map(|x| x.replace('\u{FE0F}', ""))
        .collect()
 });

@@ -108,7 +108,8 @@ impl Emoji {
            db.fetch_emoji(emoji).await?;
            Ok(true)
        } else {
-            Ok(PERMISSIBLE_EMOJIS.contains(emoji))
+            let sanitized_emoji = emoji.replace('\u{FE0F}', "");
+            Ok(PERMISSIBLE_EMOJIS.contains(&sanitized_emoji))
        }
    }
 }
--- a/crates/core/database/src/models/emojis/unicode_emoji.txt
+++ b/crates/core/database/src/models/emojis/unicode_emoji.txt
@@ -1,4 +1,3 @@
-
 *️⃣
 0️⃣
 1️⃣
@@ -242,6 +241,7 @@
 🆘
 🆙
 🆚
+🇦
 🇦🇨
 🇦🇩
 🇦🇪
@@ -259,6 +259,7 @@
 🇦🇼
 🇦🇽
 🇦🇿
+🇧
 🇧🇦
 🇧🇧
 🇧🇩
@@ -280,6 +281,7 @@
 🇧🇼
 🇧🇾
 🇧🇿
+🇨
 🇨🇦
 🇨🇨
 🇨🇩
@@ -301,6 +303,7 @@
 🇨🇽
 🇨🇾
 🇨🇿
+🇩
 🇩🇪
 🇩🇬
 🇩🇯
@@ -308,6 +311,7 @@
 🇩🇲
 🇩🇴
 🇩🇿
+🇪
 🇪🇦
 🇪🇨
 🇪🇪
@@ -317,12 +321,14 @@
 🇪🇸
 🇪🇹
 🇪🇺
+🇫
 🇫🇮
 🇫🇯
 🇫🇰
 🇫🇲
 🇫🇴
 🇫🇷
+🇬
 🇬🇦
 🇬🇧
 🇬🇩
@@ -342,12 +348,14 @@
 🇬🇺
 🇬🇼
 🇬🇾
+🇭
 🇭🇰
 🇭🇲
 🇭🇳
 🇭🇷
 🇭🇹
 🇭🇺
+🇮
 🇮🇨
 🇮🇩
 🇮🇪
@@ -359,10 +367,12 @@
 🇮🇷
 🇮🇸
 🇮🇹
+🇯
 🇯🇪
 🇯🇲
 🇯🇴
 🇯🇵
+🇰
 🇰🇪
 🇰🇬
 🇰🇭
@@ -374,6 +384,7 @@
 🇰🇼
 🇰🇾
 🇰🇿
+🇱
 🇱🇦
 🇱🇧
 🇱🇨
@@ -385,6 +396,7 @@
 🇱🇺
 🇱🇻
 🇱🇾
+🇲
 🇲🇦
 🇲🇨
 🇲🇩
@@ -408,6 +420,7 @@
 🇲🇽
 🇲🇾
 🇲🇿
+🇳
 🇳🇦
 🇳🇨
 🇳🇪
@@ -420,7 +433,9 @@
 🇳🇷
 🇳🇺
 🇳🇿
+🇴
 🇴🇲
+🇵
 🇵🇦
 🇵🇪
 🇵🇫
@@ -435,12 +450,15 @@
 🇵🇹
 🇵🇼
 🇵🇾
+🇶
 🇶🇦
+🇷
 🇷🇪
 🇷🇴
 🇷🇸
 🇷🇺
 🇷🇼
+🇸
 🇸🇦
 🇸🇧
 🇸🇨
@@ -462,6 +480,7 @@
 🇸🇽
 🇸🇾
 🇸🇿
+🇹
 🇹🇦
 🇹🇨
 🇹🇩
@@ -479,6 +498,7 @@
 🇹🇻
 🇹🇼
 🇹🇿
+🇺
 🇺🇦
 🇺🇬
 🇺🇲
@@ -486,6 +506,7 @@
 🇺🇸
 🇺🇾
 🇺🇿
+🇻
 🇻🇦
 🇻🇨
 🇻🇪
@@ -493,11 +514,15 @@
 🇻🇮
 🇻🇳
 🇻🇺
+🇼
 🇼🇫
 🇼🇸
+🇽
 🇽🇰
+🇾
 🇾🇪
 🇾🇹
+🇿
 🇿🇦
 🇿🇲
 🇿🇼
--- a/crates/core/files/Cargo.toml
+++ b/crates/core/files/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-files"
-version = "0.12.1"
+version = "0.13.7"
 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.12.1"
+version = "0.13.7"
 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,6 +314,12 @@ auto_derived!(
        /// Only used when the user is the first one connected.
        pub recipients: Option<Vec<String>>,
    }
+
+    pub struct ChannelSlowmode {
+        pub channel_id: String,
+        pub duration: u64,
+        pub retry_after: u64,
+    }
 );

 impl Channel {
--- a/crates/core/parser/Cargo.toml
+++ b/crates/core/parser/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-parser"
-version = "0.12.1"
+version = "0.13.7"
 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.12.1"
+version = "0.13.7"
 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.12.1"
+version = "0.13.7"
 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.12.1"
+version = "0.13.7"
 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
-        .remote()
-        .map(|x| x.ip().to_string())
+        .client_ip()
+        .map(|r| r.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.12.1"
+version = "0.13.7"
 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.12.1"
+version = "0.13.7"
 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;
+use revolt_database::{DatabaseInfo, AMQP};
 use revolt_result::Result;
 use tasks::{acks, file_deletion, prune_dangling_files, prune_members};
 use tokio::try_join;
@@ -11,11 +11,13 @@ async fn main() -> Result<()> {
    configure!(crond);

    let db = DatabaseInfo::Auto.connect().await.expect("database");
+    let amqp = AMQP::new_auto().await;
+
    try_join!(
        file_deletion::task(db.clone()),
        prune_dangling_files::task(db.clone()),
        prune_members::task(db.clone()),
-        acks::task(db.clone())
+        acks::task(db.clone(), amqp.clone()),
    )
    .map(|_| ())
 }
--- 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,
+    ConnectionBuilder, ConnectionProperties, ExchangeKind,
 };
-use log::info;
+use log::{debug, info};
 use redis_kiss::{get_connection, AsyncCommands, Conn as RedisConnection};
 use revolt_config::config;
-use revolt_database::{events::rabbit::AckEventPayload, Database};
+use revolt_database::{events::rabbit::AckEventPayload, Database, AMQP};
 use revolt_result::{Result, ToRevoltError};
 use serde_json;

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

    let mut redis = get_connection()
@@ -46,6 +46,42 @@ pub async fn task(db: Database) -> Result<()> {
        .await
        .expect("Failed to create channel");

+    reader_channel
+        .exchange_declare(
+            config.rabbit.default_exchange.clone().into(),
+            ExchangeKind::Topic,
+            ExchangeDeclareOptions {
+                durable: true,
+                ..Default::default()
+            },
+            FieldTable::default(),
+        )
+        .await
+        .expect("Failed to declare exchange");
+
+    reader_channel
+        .queue_declare(
+            config.rabbit.queues.acks.clone().into(),
+            QueueDeclareOptions {
+                durable: true,
+                ..Default::default()
+            },
+            FieldTable::default(),
+        )
+        .await
+        .expect("Failed to bind queue");
+
+    reader_channel
+        .queue_bind(
+            config.rabbit.queues.acks.clone().into(),
+            config.rabbit.default_exchange.into(),
+            config.rabbit.queues.acks.clone().into(),
+            QueueBindOptions::default(),
+            FieldTable::default(),
+        )
+        .await
+        .expect("Failed to bind channel");
+
    let mut consumer = reader_channel
        .basic_consume(
            config.rabbit.queues.acks.into(),
@@ -58,12 +94,14 @@ pub async fn task(db: Database) -> Result<()> {

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

    if let Some(message_id) = message_id {
-        // This will be uncommented eventually, but we need to sort out the transition to lapin first. For now we'll simply disable the badge update logic.
-        // 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?;

-        //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();

-        //     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);
-        //         }
-        //     };
-        // }
+            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);
+                }
+            };
+        }

        Ok(())
    } else {
--- a/crates/daemons/crond/src/tasks/file_deletion.rs
+++ b/crates/daemons/crond/src/tasks/file_deletion.rs
@@ -11,22 +11,24 @@ pub async fn task(db: Database) -> Result<()> {
        let files = db.fetch_deleted_attachments().await?;

        for file in files {
-            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"))
+            if let Some(hash) = &file.hash {
+                let count = db
+                    .count_file_hash_references(hash)
                    .await?;

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

-                // Delete the hash
-                db.delete_attachment_hash(&file_hash.id).await?;
-                info!("Deleted file hash {}", file_hash.id);
+                    // 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 file
--- a/crates/daemons/pushd/Cargo.toml
+++ b/crates/daemons/pushd/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-pushd"
-version = "0.12.1"
+version = "0.13.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 publish = false
@@ -15,7 +15,7 @@ revolt-parser = { workspace = true }

 anyhow = { workspace = true }

-amqprs = { workspace = true }
+lapin = { workspace = true }
 fcm_v1 = { workspace = true }
 web-push = { workspace = true }
 isahc = { workspace = true, features = ["json"], optional = true }
--- a/crates/daemons/pushd/src/consumers/inbound/ack.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/ack.rs
@@ -1,96 +1,69 @@
-use crate::consumers::inbound::internal::*;
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
+use std::sync::Arc;
+
+use crate::utils::Consumer;
+use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use revolt_database::{events::rabbit::*, Database};

+#[derive(Clone)]
+#[allow(unused)]
 pub struct AckConsumer {
-    #[allow(dead_code)]
    db: Database,
    authifier_db: authifier::Database,
-    conn: Option<Connection>,
-    channel: Option<Channel>,
+    connection: Arc<Connection>,
+    channel: Arc<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 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 {
+#[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,
-            conn: None,
-            channel: None,
+            connection,
+            channel,
        }
    }
-}

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

        // Step 1: fetch unreads and don't continue if there's no unreads
-        #[allow(clippy::disallowed_methods)]
-        let unreads = self.db.fetch_unread_mentions(&payload.user_id).await;
+        // #[allow(clippy::disallowed_methods)]

        debug!("Processing unreads for {:}", &payload.user_id);

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

        if let Ok(sessions) = self.authifier_db.find_sessions(&payload.user_id).await {
            let config = revolt_config::config().await;
            // Step 2: find any apple sessions, since we don't need to calculate this for anything else.
            // If there's no apple sessions, we can return early
-            let apple_sessions: Vec<&authifier::models::Session> = sessions
-                .iter()
+            let mut apple_sessions = sessions
+                .into_iter()
                .filter(|session| {
                    if let Some(sub) = &session.subscription {
                        sub.endpoint == "apn"
@@ -98,19 +71,19 @@ impl AsyncConsumer for AckConsumer {
                        false
                    }
                })
-                .collect();
+                .peekable();

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

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

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

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

-                    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());
-                }
+                self.publish_message(
+                    payload.as_bytes(),
+                    &config.pushd.exchange,
+                    &config.pushd.apn.queue,
+                )
+                .await?;
            }
        }
+
+        Ok(())
    }
 }
--- a/crates/daemons/pushd/src/consumers/inbound/dm_call.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/dm_call.rs
@@ -1,70 +1,44 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};

-use crate::consumers::inbound::internal::*;
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
+use crate::utils::Consumer;
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
 use revolt_database::{events::rabbit::*, Database};

+#[derive(Clone)]
+#[allow(unused)]
 pub struct DmCallConsumer {
-    #[allow(dead_code)]
    db: Database,
    authifier_db: authifier::Database,
-    conn: Option<Connection>,
-    channel: Option<Channel>,
+    connection: Arc<Connection>,
+    channel: Arc<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 get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl DmCallConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> DmCallConsumer {
-        DmCallConsumer {
+#[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,
-            conn: None,
-            channel: None,
+            connection,
+            channel,
        }
    }

-    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())?;
+    fn channel(&self) -> &Arc<Channel> {
+        &self.channel
+    }
+
+    /// 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)?;
        let payload = _p.payload;

        debug!("Received dm call start/stop event");
@@ -107,36 +81,27 @@ impl DmCallConsumer {
                            extras: HashMap::new(),
                        };

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

-                        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());
-                        }
+                                &config.pushd.vapid.queue
+                            }
+                        };

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

-                        publish_message(self, payload.into(), args).await;
+                        self.publish_message(
+                            payload.as_bytes(),
+                            &config.pushd.exchange,
+                            routing_key,
+                        )
+                        .await?;
                    }
                }
            }
@@ -145,24 +110,3 @@ impl DmCallConsumer {
        Ok(())
    }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for DmCallConsumer {
-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            warn!("Failed to process dm call start/stop event: {err:?}");
-        }
-    }
-}
--- a/crates/daemons/pushd/src/consumers/inbound/fr_accepted.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/fr_accepted.rs
@@ -1,70 +1,44 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};

-use crate::consumers::inbound::internal::*;
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
+use crate::utils::Consumer;
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
 use revolt_database::{events::rabbit::*, Database};

+#[derive(Clone)]
+#[allow(unused)]
 pub struct FRAcceptedConsumer {
-    #[allow(dead_code)]
    db: Database,
    authifier_db: authifier::Database,
-    conn: Option<Connection>,
-    channel: Option<Channel>,
+    connection: Arc<Connection>,
+    channel: Arc<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 get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl FRAcceptedConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> FRAcceptedConsumer {
-        FRAcceptedConsumer {
+#[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,
-            conn: None,
-            channel: None,
+            connection,
+            channel,
        }
    }

-    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())?;
+    fn channel(&self) -> &Arc<Channel> {
+        &self.channel
+    }
+
+    /// 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)?;

        debug!("Received FR accept event");

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

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

-                    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());
-                    }
+                            &config.pushd.vapid.queue
+                        }
+                    };

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

-                    publish_message(self, payload.into(), args).await;
+                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
+                        .await?;
                }
            }
        }
@@ -117,24 +78,3 @@ impl FRAcceptedConsumer {
        Ok(())
    }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for FRAcceptedConsumer {
-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process friend request accepted event: {err:?}");
-        }
-    }
-}
--- a/crates/daemons/pushd/src/consumers/inbound/fr_received.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/fr_received.rs
@@ -1,70 +1,44 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};

-use crate::consumers::inbound::internal::*;
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
+use crate::utils::Consumer;
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
 use revolt_database::{events::rabbit::*, Database};

+#[derive(Clone)]
+#[allow(unused)]
 pub struct FRReceivedConsumer {
-    #[allow(dead_code)]
    db: Database,
    authifier_db: authifier::Database,
-    conn: Option<Connection>,
-    channel: Option<Channel>,
+    connection: Arc<Connection>,
+    channel: Arc<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 get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl FRReceivedConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> FRReceivedConsumer {
-        FRReceivedConsumer {
+#[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,
-            conn: None,
-            channel: None,
+            connection,
+            channel,
        }
    }

-    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())?;
+    fn channel(&self) -> &Arc<Channel> {
+        &self.channel
+    }
+
+    /// 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)?;

        debug!("Received FR received event");

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

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

-                    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());
-                    }
+                            &config.pushd.vapid.queue
+                        }
+                    };

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

-                    publish_message(self, payload.into(), args).await;
+                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
+                        .await?;
                }
            }
        }
@@ -117,24 +78,3 @@ impl FRReceivedConsumer {
        Ok(())
    }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for FRReceivedConsumer {
-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process friend request received event: {err:?}");
-        }
-    }
-}
--- a/crates/daemons/pushd/src/consumers/inbound/generic.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/generic.rs
@@ -1,70 +1,44 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};

-use crate::consumers::inbound::internal::*;
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
+use crate::utils::Consumer;
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
 use revolt_database::{events::rabbit::*, Database};

+#[derive(Clone)]
+#[allow(unused)]
 pub struct GenericConsumer {
-    #[allow(dead_code)]
    db: Database,
    authifier_db: authifier::Database,
-    conn: Option<Connection>,
-    channel: Option<Channel>,
+    connection: Arc<Connection>,
+    channel: Arc<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 get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl GenericConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> GenericConsumer {
-        GenericConsumer {
+#[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,
-            conn: None,
-            channel: None,
+            connection,
+            channel,
        }
    }

-    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())?;
+    fn channel(&self) -> &Arc<Channel> {
+        &self.channel
+    }
+
+    /// 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)?;

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

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

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

-                    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());
-                    }
+                            &config.pushd.vapid.queue
+                        }
+                    };

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

-                    publish_message(self, payload.into(), args).await;
+                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
+                        .await?;
                }
            }
        }
@@ -123,24 +84,3 @@ impl GenericConsumer {
        Ok(())
    }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for GenericConsumer {
-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process generic event: {err:?}");
-        }
-    }
-}
--- a/crates/daemons/pushd/src/consumers/inbound/internal.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/internal.rs
@@ -1,53 +0,0 @@
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::{Connection, OpenConnectionArguments},
-    BasicProperties,
-};
-use log::{debug, warn};
-
-pub(crate) trait Channeled {
-    #[allow(unused)]
-    fn get_connection(&self) -> Option<&Connection>;
-    fn get_channel(&self) -> Option<&Channel>;
-    fn set_connection(&mut self, conn: Connection);
-    fn set_channel(&mut self, channel: Channel);
-}
-
-pub(crate) async fn make_channel<T: Channeled>(consumer: &mut T) {
-    let config = revolt_config::config().await;
-
-    let args = OpenConnectionArguments::new(
-        &config.rabbit.host,
-        config.rabbit.port,
-        &config.rabbit.username,
-        &config.rabbit.password,
-    );
-    let conn = amqprs::connection::Connection::open(&args).await.unwrap();
-
-    let channel = conn.open_channel(None).await.unwrap();
-
-    consumer.set_connection(conn);
-    consumer.set_channel(channel);
-}
-
-pub(crate) async fn publish_message<T: Channeled>(
-    consumer: &mut T,
-    payload: Vec<u8>,
-    args: BasicPublishArguments,
-) {
-    let routing_key = &args.routing_key.clone();
-    let mut channel = consumer.get_channel();
-    if channel.is_none() {
-        make_channel(consumer).await;
-        channel = consumer.get_channel();
-    }
-
-    if let Some(chnl) = channel {
-        chnl.basic_publish(BasicProperties::default(), payload.clone(), args.clone())
-            .await
-            .unwrap();
-        debug!("Sent message to queue for target {}", routing_key);
-    } else {
-        warn!("Failed to unwrap channel (including attempt to make a channel)!")
-    }
-}
--- a/crates/daemons/pushd/src/consumers/inbound/mass_mention.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/mass_mention.rs
@@ -1,17 +1,13 @@
 use std::{
    collections::{HashMap, HashSet},
    hash::RandomState,
+    sync::Arc,
 };

-use crate::{consumers::inbound::internal::*, utils};
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
+use crate::utils::{render_notification_content, Consumer};
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use revolt_database::{
    events::rabbit::*, util::bulk_permissions::BulkDatabasePermissionQuery, Database, Member,
    MessageFlagsValue,
@@ -19,52 +15,18 @@ use revolt_database::{
 use revolt_models::v0::{MessageFlags, PushNotification};
 use revolt_result::ToRevoltError;

+#[derive(Clone)]
+#[allow(unused)]
 pub struct MassMessageConsumer {
-    #[allow(dead_code)]
    db: Database,
    authifier_db: authifier::Database,
-    conn: Option<Connection>,
-    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)
-    }
+    connection: Arc<Connection>,
+    channel: Arc<Channel>,
 }

 impl MassMessageConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> MassMessageConsumer {
-        MassMessageConsumer {
-            db,
-            authifier_db,
-            conn: None,
-            channel: None,
-        }
-    }
-
    async fn fire_notification_for_users(
-        &mut self,
+        &self,
        push: &PushNotification,
        users: &[String],
    ) -> Result<()> {
@@ -84,56 +46,58 @@ impl MassMessageConsumer {
                        extras: HashMap::new(),
                    };

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

-                    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());
-                    }
+                            &config.pushd.vapid.queue
+                        }
+                    };

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

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

        Ok(())
    }
+}

-    async fn consume_event(
-        &mut self,
-        _channel: &Channel,
-        _deliver: Deliver,
-        _basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) -> Result<()> {
+#[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)?;
        let config = revolt_config::config().await;
-        let content = String::from_utf8(content)?;
-        let mut payload: MassMessageSentPayload = serde_json::from_str(content.as_str())?;

        for push in payload.notifications.iter_mut() {
-            if let Ok(body) = utils::render_notification_content(push, &self.db)
+            if let Ok(body) = render_notification_content(push, &self.db)
                .await
                .to_internal_error()
            {
@@ -280,24 +244,3 @@ impl MassMessageConsumer {
        Ok(())
    }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for MassMessageConsumer {
-    /// This consumer handles adding mentions for all the users affected by a mass mention ping, and then sends out push notifications
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process mass message event: {err:?}");
-        }
-    }
-}
--- a/crates/daemons/pushd/src/consumers/inbound/message.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/message.rs
@@ -1,76 +1,46 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};

-use crate::{consumers::inbound::internal::*, utils};
-use amqprs::{
-    channel::{BasicPublishArguments, Channel},
-    connection::Connection,
-    consumer::AsyncConsumer,
-    BasicProperties, Deliver,
-};
+use crate::utils::{render_notification_content, Consumer};
 use anyhow::Result;
 use async_trait::async_trait;
+use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
 use revolt_database::{events::rabbit::*, Database};
-use revolt_result::ToRevoltError;

+#[derive(Clone)]
+#[allow(unused)]
 pub struct MessageConsumer {
-    #[allow(dead_code)]
    db: Database,
    authifier_db: authifier::Database,
-    conn: Option<Connection>,
-    channel: Option<Channel>,
+    connection: Arc<Connection>,
+    channel: Arc<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 get_channel(&self) -> Option<&Channel> {
-        if self.channel.is_none() {
-            None
-        } else {
-            Some(self.channel.as_ref().unwrap())
-        }
-    }
-
-    fn set_connection(&mut self, conn: Connection) {
-        self.conn = Some(conn);
-    }
-
-    fn set_channel(&mut self, channel: Channel) {
-        self.channel = Some(channel)
-    }
-}
-
-impl MessageConsumer {
-    pub fn new(db: Database, authifier_db: authifier::Database) -> MessageConsumer {
-        MessageConsumer {
+#[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,
-            conn: None,
-            channel: None,
+            connection,
+            channel,
        }
    }

-    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())?;
+    fn channel(&self) -> &Arc<Channel> {
+        &self.channel
+    }

-        if let Ok(body) = utils::render_notification_content(&payload.notification, &self.db)
-            .await
-            .to_internal_error()
-        {
+    /// 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)?;
+
+        if let Ok(body) = render_notification_content(&payload.notification, &self.db).await {
            payload.notification.raw_body = Some(payload.notification.body);
            payload.notification.body = body;
        }
@@ -95,36 +65,22 @@ impl MessageConsumer {
                        extras: HashMap::new(),
                    };

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

-                    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());
-                    }
+                            &config.pushd.vapid.queue
+                        }
+                    };

                    let payload = serde_json::to_string(&sendable)?;
-
-                    publish_message(self, payload.into(), args).await;
+                    self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
+                        .await?;
                }
            }
        }
@@ -132,24 +88,3 @@ impl MessageConsumer {
        Ok(())
    }
 }
-
-#[allow(unused_variables)]
-#[async_trait]
-impl AsyncConsumer for MessageConsumer {
-    /// This consumer handles delegating messages into their respective platform queues.
-    async fn consume(
-        &mut self,
-        channel: &Channel,
-        deliver: Deliver,
-        basic_properties: BasicProperties,
-        content: Vec<u8>,
-    ) {
-        if let Err(err) = self
-            .consume_event(channel, deliver, basic_properties, content)
-            .await
-        {
-            revolt_config::capture_anyhow(&err);
-            eprintln!("Failed to process message event: {err:?}");
-        }
-    }
-}
--- a/crates/daemons/pushd/src/consumers/inbound/mod.rs
+++ b/crates/daemons/pushd/src/consumers/inbound/mod.rs
@@ -3,6 +3,5 @@ pub mod dm_call;
 pub mod fr_accepted;
 pub mod fr_received;
 pub mod generic;
-mod internal;
 pub mod mass_mention;
 pub mod message;
--- a/crates/daemons/pushd/src/consumers/outbound/apn.rs
+++ b/crates/daemons/pushd/src/consumers/outbound/apn.rs
@@ -1,12 +1,13 @@
-use std::{borrow::Cow, collections::BTreeMap, io::Cursor};
+use std::{borrow::Cow, collections::BTreeMap, io::Cursor, sync::Arc};

-use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};
-use anyhow::{anyhow, Result};
+use crate::utils::Consumer;
+use anyhow::Result;
 use async_trait::async_trait;
 use base64::{
    engine::{self},
    Engine as _,
 };
+use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
 use revolt_a2::{
    request::{
        notification::{DefaultAlert, NotificationOptions},
@@ -42,7 +43,7 @@ impl<'a> PayloadLike for MessagePayload<'a> {
    fn get_device_token(&self) -> &'a str {
        self.device_token
    }
-    fn get_options(&self) -> &NotificationOptions {
+    fn get_options(&self) -> &NotificationOptions<'a> {
        &self.options
    }
 }
@@ -68,16 +69,20 @@ impl<'a> PayloadLike for CallStartStopPayload<'a> {
    fn get_device_token(&self) -> &'a str {
        self.device_token
    }
-    fn get_options(&self) -> &NotificationOptions {
+    fn get_options(&self) -> &NotificationOptions<'a> {
        &self.options
    }
 }

 // region: consumer

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

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

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

        if config.pushd.apn.pkcs8.is_empty()
            || config.pushd.apn.key_id.is_empty()
            || config.pushd.apn.team_id.is_empty()
        {
-            return Err("Missing APN keys.");
+            panic!("Missing APN keys.");
        }

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

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

-    async fn consume_event(
-        &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())?;
+    fn channel(&self) -> &Arc<AMQPChannel> {
+        &self.channel
+    }
+
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
+        let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;

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

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

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

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

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

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

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

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

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

-        if let Err(err) = resp {
-            match 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 => {
+        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
+                {
                    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,17 +1,16 @@
-use std::{collections::HashMap, time::Duration};
+use std::{collections::HashMap, sync::Arc, time::Duration};

-use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};
-
-use anyhow::{anyhow, bail, Result};
+use crate::utils::Consumer;
+use anyhow::{bail, Result};
 use async_trait::async_trait;
 use fcm_v1::{
    auth::{Authenticator, ServiceAccountKey},
    message::Message,
    Client, Error as FcmError,
 };
+use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
 use revolt_config::config;
 use revolt_database::{events::rabbit::*, Database};
-use revolt_models::v0::{Channel, PushNotification};
 use serde_json::Value;

 /// Custom notification data
@@ -31,10 +30,12 @@ pub enum NotificationData {
        image: Option<String>,
    },
    Message {
-        title: String,
+        message: String,
        body: String,
        image: String,
-        tag: String,
+        channel: String,
+        author: String,
+        author_name: String,
    },
    DmCallStartEnd {
        initiator_id: String,
@@ -81,15 +82,19 @@ impl NotificationData {
                }
            }
            NotificationData::Message {
-                title,
+                message,
                body,
                image,
-                tag,
+                channel,
+                author,
+                author_name,
            } => {
-                data.insert("title".to_string(), Value::String(title));
+                data.insert("message".to_string(), Value::String(message));
                data.insert("body".to_string(), Value::String(body));
                data.insert("image".to_string(), Value::String(image));
-                data.insert("tag".to_string(), Value::String(tag));
+                data.insert("channel".to_string(), Value::String(channel));
+                data.insert("author".to_string(), Value::String(author));
+                data.insert("author_name".to_string(), Value::String(author_name));
            }
            NotificationData::DmCallStartEnd {
                initiator_id,
@@ -110,37 +115,31 @@ impl NotificationData {
    }
 }

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

-impl FcmOutboundConsumer {
-    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> {
+#[async_trait]
+impl Consumer for FcmOutboundConsumer {
+    async fn create(
+        db: Database,
+        authifier_db: authifier::Database,
+        connection: Arc<Connection>,
+        channel: Arc<AMQPChannel>,
+    ) -> Self {
        let config = revolt_config::config().await;

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

-    async fn consume_event(
-        &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())?;
+    fn channel(&self) -> &Arc<AMQPChannel> {
+        &self.channel
+    }
+
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
+        let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;

        #[allow(clippy::needless_late_init)]
        let resp: Result<Message, FcmError>;

        match payload.notification {
            PayloadKind::FRReceived(alert) => {
-                let name = alert
-                    .from_user
-                    .display_name
-                    .or(Some(format!(
+                let name = alert.from_user.display_name.clone().unwrap_or_else(|| {
+                    format!(
                        "{}#{}",
                        alert.from_user.username, alert.from_user.discriminator
-                    )))
-                    .clone()
-                    .ok_or_else(|| anyhow!("missing name"))?;
+                    )
+                });

                let data = NotificationData::FRReceived {
                    id: alert.from_user.id,
@@ -203,15 +196,12 @@ impl FcmOutboundConsumer {
            }

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

                let data = NotificationData::FRAccepted {
                    id: alert.accepted_user.id,
@@ -244,10 +234,12 @@ impl FcmOutboundConsumer {

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

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

-        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 => {
+        match resp {
+            Err(FcmError::Auth) => {
+                if let Err(err) = self
+                    .db
+                    .remove_push_subscription_by_session_id(&payload.session_id)
+                    .await
+                {
                    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;
+use std::{collections::HashMap, sync::Arc};

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

 use anyhow::{anyhow, bail, Result};
 use async_trait::async_trait;
@@ -8,46 +8,60 @@ use base64::{
    engine::{self},
    Engine as _,
 };
+use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
 use revolt_database::{events::rabbit::*, util::format_display_name, Database};
 use web_push::{
    ContentEncoding, IsahcWebPushClient, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder,
    WebPushClient, WebPushError, WebPushMessageBuilder,
 };

+#[derive(Clone)]
+#[allow(unused)]
 pub struct VapidOutboundConsumer {
    db: Database,
+    authifier_db: authifier::Database,
+    connection: Arc<Connection>,
+    channel: Arc<AMQPChannel>,
    client: IsahcWebPushClient,
-    pkey: Vec<u8>,
+    pkey: Arc<Vec<u8>>,
 }

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

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

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

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

-    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())?;
+    fn channel(&self) -> &Arc<AMQPChannel> {
+        &self.channel
+    }
+
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
+        let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;

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

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

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

-                payload_body = serde_json::to_string(&body)?;
-            }
-            PayloadKind::Generic(alert) => {
-                payload_body = serde_json::to_string(&alert)?;
-            }
-            PayloadKind::MessageNotification(alert) => {
-                payload_body = serde_json::to_string(&alert)?;
+                serde_json::to_string(&body)?
            }
+            PayloadKind::Generic(alert) => serde_json::to_string(&alert)?,
+            PayloadKind::MessageNotification(alert) => serde_json::to_string(&alert)?,
            PayloadKind::DmCallStartEnd(alert) => {
                let initiator_name = if let Some(server_id) =
                    self.db.fetch_channel(&alert.channel_id).await?.server()
@@ -132,59 +139,41 @@ impl VapidOutboundConsumer {
                    _ => bail!("Invalid DmCallStart/End channel type"),
                }

-                payload_body = serde_json::to_string(&body)?;
+                serde_json::to_string(&body)?
            }
            PayloadKind::BadgeUpdate(_) => {
                bail!("Vapid cannot handle badge updates and they should not be sent here.");
            }
-        }
+        };

-        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 signature = VapidSignatureBuilder::from_pem(
+            std::io::Cursor::new(self.pkey.as_ref()),
+            &subscription,
+        )?
+        .build()?;

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

-                    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?;
-                                }
-                            }
+        builder.set_payload(ContentEncoding::AesGcm, payload_body.as_bytes());

-                            Ok(())
-                        }
-                        Err(err) => Err(err.into()),
-                    }
+        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);
                }
-                Err(err) => Err(err.into()),
-            },
-            Err(err) => Err(err.into()),
-        }
-    }
-}
+            }
+            res => {
+                res?;
+            }
+        };

-#[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:?}");
-        }
+        Ok(())
    }
 }
--- a/crates/daemons/pushd/src/main.rs
+++ b/crates/daemons/pushd/src/main.rs
@@ -1,17 +1,16 @@
 #[macro_use]
 extern crate log;

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

 mod consumers;
 mod utils;
@@ -24,6 +23,8 @@ use consumers::{
    outbound::{apn::ApnsOutboundConsumer, fcm::FcmOutboundConsumer, vapid::VapidOutboundConsumer},
 };

+use crate::utils::{Consumer, Delegate};
+
 #[tokio::main(flavor = "multi_thread", worker_threads = 2)]
 async fn main() {
    // Configure logging and environment
@@ -43,7 +44,24 @@ async fn main() {
        panic!("Mongo is not in use, can't connect via authifier!")
    }

-    let mut connections: Vec<(Channel, Connection)> = Vec::new();
+    let config = config().await;
+
+    let connection = Arc::new(
+        Connection::connect(
+            &format!(
+                "amqp://{}:{}@{}:{}",
+                &config.rabbit.username,
+                &config.rabbit.password,
+                &config.rabbit.host,
+                &config.rabbit.port,
+            ),
+            ConnectionProperties::default(),
+        )
+        .await
+        .expect("Failed to connect to RabbitMQ"),
+    );
+
+    let mut channels = Vec::new();

    // An explainer of how this works:
    // The inbound connections are on separate routing keys, such that they only receive the proper payload
@@ -54,171 +72,178 @@ async fn main() {
    // This'll require some interesting shimming if we need to add more events once this is in prod (different payloads between prod and test),
    // but that sounds like a problem for future us.

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

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

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

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

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

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

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

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

-        connections.push(
-            make_queue_and_consume(
+        channels.push(
+            make_queue_and_consume::<AckConsumer>(
+                &db,
+                &authifier,
+                &connection,
                &config,
                &config.pushd.ack_queue,
                &config.pushd.ack_queue,
                Some(table),
-                AckConsumer::new(db.clone(), authifier.clone()),
            )
            .await,
        );
    }

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

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

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

-    for (channel, conn) in connections {
-        channel.close().await.expect("Unable to close channel");
-        conn.close().await.expect("Unable to close connection");
+    for channel in channels {
+        let _ = channel.close(0, "close".into()).await;
    }
 }

 async fn make_queue_and_consume<F>(
+    db: &Database,
+    authifier_db: &authifier::Database,
+    connection: &Arc<Connection>,
    config: &Settings,
    queue_name: &str,
    routing_key: &str,
    queue_args: Option<FieldTable>,
-    consumer: F,
-) -> (Channel, Connection)
+) -> Arc<Channel>
 where
-    F: AsyncConsumer + Send + 'static,
+    F: Consumer,
 {
-    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();
+    let channel = Arc::new(connection.create_channel().await.unwrap());

    channel
        .exchange_declare(
-            ExchangeDeclareArguments::new(&config.pushd.exchange, "direct")
-                .durable(true)
-                .finish(),
+            config.pushd.exchange.clone().into(),
+            lapin::ExchangeKind::Direct,
+            ExchangeDeclareOptions {
+                durable: true,
+                ..Default::default()
+            },
+            FieldTable::default(),
        )
        .await
-        .expect("Failed to declare pushd exchange");
+        .expect("Failed to declare exchange");

    let mut queue_name = queue_name.to_string();

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

    let queue_name = queue_name.as_str();

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

    channel
-        .queue_bind(QueueBindArguments::new(
-            queue_name,
-            &config.pushd.exchange,
-            routing_key,
-        ))
+        .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(),
+        )
        .await
        .expect(
            "This probably means the revolt.notifications exchange does not exist in rabbitmq!",
        );

-    let args = BasicConsumeArguments::new(queue_name, "")
-        .manual_ack(false)
-        .finish();
-
-    let routing_key = channel.basic_consume(consumer, args).await.unwrap();
+    let consumer = channel
+        .basic_consume(
+            queue_name.into(),
+            "".into(),
+            BasicConsumeOptions {
+                no_ack: true,
+                ..Default::default()
+            },
+            FieldTable::default(),
+        )
+        .await
+        .unwrap();
    info!(
        "Consuming routing key {} as queue {}, tag {}",
-        routing_key, queue_name, routing_key
+        routing_key,
+        queue_name,
+        consumer.tag()
    );
-    (channel, connection)
+
+    let delegate = Delegate(
+        F::create(
+            db.clone(),
+            authifier_db.clone(),
+            connection.clone(),
+            channel.clone(),
+        )
+        .await,
+    );
+
+    consumer.set_delegate(delegate);
+
+    channel
 }
--- a/crates/daemons/pushd/src/utils/consumer.rs
+++ b/crates/daemons/pushd/src/utils/consumer.rs
@@ -0,0 +1,91 @@
+use std::{
+    future::{ready, Future},
+    pin::Pin,
+    sync::Arc,
+};
+
+use anyhow::Result;
+use async_trait::async_trait;
+use lapin::{
+    message::{Delivery, DeliveryResult},
+    options::BasicPublishOptions,
+    BasicProperties, Channel, Connection, ConsumerDelegate, Error as AMQPError,
+};
+use log::debug;
+use revolt_database::Database;
+
+#[async_trait]
+pub trait Consumer: Clone + Send + Sync + 'static {
+    async fn create(
+        db: Database,
+        authifier_db: authifier::Database,
+        connection: Arc<Connection>,
+        channel: Arc<Channel>,
+    ) -> Self;
+    fn channel(&self) -> &Arc<Channel>;
+    async fn consume(&self, delivery: Delivery) -> Result<()>;
+
+    async fn publish_message_with_options(
+        &self,
+        payload: &[u8],
+        exchange: &str,
+        routing_key: &str,
+        options: BasicPublishOptions,
+        properties: BasicProperties,
+    ) -> Result<(), AMQPError> {
+        let channel = self.channel();
+
+        channel
+            .basic_publish(
+                exchange.into(),
+                routing_key.into(),
+                options,
+                payload,
+                properties,
+            )
+            .await?;
+        debug!("Sent message to queue for target {}", routing_key);
+
+        Ok(())
+    }
+
+    async fn publish_message(
+        &self,
+        payload: &[u8],
+        exchange: &str,
+        routing_key: &str,
+    ) -> Result<(), AMQPError> {
+        self.publish_message_with_options(
+            payload,
+            exchange,
+            routing_key,
+            BasicPublishOptions::default(),
+            BasicProperties::default(),
+        )
+        .await
+    }
+}
+
+pub struct Delegate<C: Consumer>(pub C);
+
+impl<C: Consumer> ConsumerDelegate for Delegate<C> {
+    fn on_new_delivery(
+        &self,
+        delivery: DeliveryResult,
+    ) -> Pin<Box<dyn Future<Output = ()> + Send>> {
+        match delivery {
+            Ok(Some(delivery)) => {
+                let consumer = self.0.clone();
+
+                Box::pin(async move {
+                    if let Err(e) = consumer.consume(delivery).await {
+                        revolt_config::capture_anyhow(&e);
+                        log::error!("{e:?}");
+                    };
+                })
+            }
+            Ok(None) => Box::pin(ready(())),
+            Err(e) => Box::pin(async move { log::error!("Received bad delivery: {e:?}") }),
+        }
+    }
+}
--- a/crates/daemons/pushd/src/utils/mod.rs
+++ b/crates/daemons/pushd/src/utils/mod.rs
@@ -1,2 +1,5 @@
 mod renderer;
+mod consumer;
+
 pub use renderer::render_notification_content;
+pub use consumer::{Consumer, Delegate};
--- 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.12.1"
+version = "0.13.7"
 license = "AGPL-3.0-or-later"
 edition = "2021"
 publish = false
@@ -44,6 +44,3 @@ revolt-permissions = { workspace = true }
 livekit-api = { workspace = true }
 livekit-protocol = { workspace = true }
 livekit-runtime = { workspace = true, features = ["tokio"] }
-
-# RabbitMQ
-amqprs = { workspace = true }
--- a/crates/delta/Cargo.toml
+++ b/crates/delta/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-delta"
-version = "0.12.1"
+version = "0.13.7"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <paulmakles@gmail.com>"]
 edition = "2018"
@@ -64,7 +64,7 @@ schemars = { workspace = true }
 revolt_rocket_okapi = { workspace = true, features = ["swagger"] }

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

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

 use revolt_config::config;
-use revolt_database::events::client::EventV1;
-use revolt_database::AMQP;
+use revolt_database::{AMQP, events::client::EventV1};
 use revolt_ratelimits::rocket as ratelimiter;
 use rocket::{Build, Rocket};
 use rocket_cors::{AllowedOrigins, CorsOptions};
@@ -18,10 +17,6 @@ use rocket_prometheus::PrometheusMetrics;
 use std::net::Ipv4Addr;
 use std::str::FromStr;

-use amqprs::{
-    channel::ExchangeDeclareArguments,
-    connection::{Connection, OpenConnectionArguments},
-};
 use async_std::channel::unbounded;
 use authifier::AuthifierEvent;
 use revolt_database::voice::VoiceClient;
@@ -36,7 +31,6 @@ pub async fn web() -> Rocket<Build> {

    // Setup database
    let db = revolt_database::DatabaseInfo::Auto.connect().await.unwrap();
-    log::info!("database_here {db:?}");
    db.migrate_database().await.unwrap();

    // Setup Authifier event channel
@@ -96,33 +90,8 @@ pub async fn web() -> Rocket<Build> {
    // Voice handler
    let voice_client = VoiceClient::new(config.api.livekit.nodes.clone());
    // Configure Rabbit
-    let connection = Connection::open(&OpenConnectionArguments::new(
-        &config.rabbit.host,
-        config.rabbit.port,
-        &config.rabbit.username,
-        &config.rabbit.password,
-    ))
-    .await
-    .expect("Failed to connect to RabbitMQ");

-    let channel = connection
-        .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");
+    let amqp = AMQP::new_auto().await;

    // Launch background task workers
    revolt_database::tasks::start_workers(db.clone(), amqp.clone());
@@ -152,6 +121,7 @@ pub async fn web() -> Rocket<Build> {
            limits: rocket::data::Limits::default().limit("string", 5.megabytes()),
            address: Ipv4Addr::new(0, 0, 0, 0).into(),
            port: 14702,
+            ip_header: Some("X-Forwarded-For".into()),
            ..Default::default()
        })
 }
--- a/crates/delta/src/routes/channels/message_send.rs
+++ b/crates/delta/src/routes/channels/message_send.rs
@@ -1,12 +1,14 @@
 use std::time::Duration;

 use redis_kiss::{get_connection, redis, AsyncCommands};
+use revolt_database::events::client::EventV1;
 use revolt_database::util::permissions::DatabasePermissionQuery;
 use revolt_database::{
    util::idempotency::IdempotencyKey, util::reference::Reference, Database, User,
 };
 use revolt_database::{Channel, Interactions, Message, AMQP};
 use revolt_models::v0;
+use revolt_models::v0::ChannelSlowmode;
 use revolt_permissions::PermissionQuery;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
@@ -84,6 +86,16 @@ pub async fn message_send(
                        .await
                        .unwrap_or(None);

+                    if set_result.is_some() {
+                        let idx_key = format!("slowmode_idx:{}", user.id);
+                        conn.sadd::<_, _, ()>(&idx_key, channel_id.as_str())
+                            .await
+                            .ok();
+                        conn.expire::<_, ()>(&idx_key, *channel_slowmode as usize)
+                            .await
+                            .ok();
+                    }
+
                    // If `set_result` is None, the `NX` condition failed because the key already exists.
                    // This means the user is currently in slowmode.
                    if set_result.is_none() {
@@ -92,10 +104,29 @@ pub async fn message_send(

                        // Redis returns positive integers for valid TTLs
                        if ttl > 0 {
+                            EventV1::UserSlowmodes {
+                                slowmodes: vec![ChannelSlowmode {
+                                    channel_id: channel_id.to_string(),
+                                    duration: *channel_slowmode,
+                                    retry_after: ttl as u64,
+                                }],
+                            }
+                            .private(user.id.clone())
+                            .await;
                            return Err(create_error!(InSlowmode {
                                retry_after: ttl as u64
                            }));
                        }
+                    } else {
+                        EventV1::UserSlowmodes {
+                            slowmodes: vec![ChannelSlowmode {
+                                channel_id: channel_id.to_string(),
+                                duration: *channel_slowmode,
+                                retry_after: *channel_slowmode,
+                            }],
+                        }
+                        .private(user.id.clone())
+                        .await;
                    }
                }
                // If Redis connection fails, just skip the slowmode check
--- a/crates/delta/src/routes/mod.rs
+++ b/crates/delta/src/routes/mod.rs
@@ -64,47 +64,6 @@ pub fn mount(config: Settings, mut rocket: Rocket<Build>) -> Rocket<Build> {
        };
    }

-    if config.features.webhooks_enabled {
-        mount_endpoints_and_merged_docs! {
-            rocket, "/0.8".to_owned(), settings,
-            "/" => (vec![], custom_openapi_spec()),
-            "" => openapi_get_routes_spec![root::root],
-            "/users" => users::routes(),
-            "/bots" => bots::routes(),
-            "/channels" => channels::routes(),
-            "/servers" => servers::routes(),
-            "/invites" => invites::routes(),
-            "/custom" => customisation::routes(),
-            "/safety" => safety::routes(),
-            "/auth/account" => rocket_authifier::routes::account::routes(),
-            "/auth/session" => rocket_authifier::routes::session::routes(),
-            "/auth/mfa" => rocket_authifier::routes::mfa::routes(),
-            "/onboard" => onboard::routes(),
-            "/push" => push::routes(),
-            "/sync" => sync::routes(),
-            "/webhooks" => webhooks::routes()
-        };
-    } else {
-        mount_endpoints_and_merged_docs! {
-            rocket, "/0.8".to_owned(), settings,
-            "/" => (vec![], custom_openapi_spec()),
-            "" => openapi_get_routes_spec![root::root],
-            "/users" => users::routes(),
-            "/bots" => bots::routes(),
-            "/channels" => channels::routes(),
-            "/servers" => servers::routes(),
-            "/invites" => invites::routes(),
-            "/custom" => customisation::routes(),
-            "/safety" => safety::routes(),
-            "/auth/account" => rocket_authifier::routes::account::routes(),
-            "/auth/session" => rocket_authifier::routes::session::routes(),
-            "/auth/mfa" => rocket_authifier::routes::mfa::routes(),
-            "/onboard" => onboard::routes(),
-            "/push" => push::routes(),
-            "/sync" => sync::routes()
-        };
-    }
-
    rocket
 }

@@ -115,8 +74,8 @@ fn custom_openapi_spec() -> OpenApi {
    extensions.insert(
        "x-logo".to_owned(),
        json!({
-            "url": "https://revolt.chat/header.png",
-            "altText": "Revolt Header"
+            "url": "https://stoat.chat/header.png",
+            "altText": "Stoat Header"
        }),
    );

@@ -124,7 +83,7 @@ fn custom_openapi_spec() -> OpenApi {
        "x-tagGroups".to_owned(),
        json!([
          {
-            "name": "Revolt",
+            "name": "Stoat",
            "tags": [
              "Core"
            ]
@@ -205,18 +164,21 @@ fn custom_openapi_spec() -> OpenApi {
    OpenApi {
        openapi: OpenApi::default_version(),
        info: Info {
-            title: "Revolt API".to_owned(),
+            title: "Stoat API".to_owned(),
            description: Some("Open source user-first chat platform.".to_owned()),
-            terms_of_service: Some("https://revolt.chat/terms".to_owned()),
+            terms_of_service: Some("https://stoat.chat/terms".to_owned()),
            contact: Some(Contact {
-                name: Some("Revolt Support".to_owned()),
-                url: Some("https://revolt.chat".to_owned()),
-                email: Some("contact@revolt.chat".to_owned()),
+                name: Some("Stoat".to_owned()),
+                url: Some("https://stoat.chat".to_owned()),
+                email: Some("contact@stoat.chat".to_owned()),
                ..Default::default()
            }),
            license: Some(License {
                name: "AGPLv3".to_owned(),
-                url: Some("https://github.com/stoatchat/stoatchat/blob/main/crates/delta/LICENSE".to_owned()),
+                url: Some(
+                    "https://github.com/stoatchat/stoatchat/blob/main/crates/delta/LICENSE"
+                        .to_owned(),
+                ),
                ..Default::default()
            }),
            version: env!("CARGO_PKG_VERSION").to_string(),
@@ -224,29 +186,19 @@ fn custom_openapi_spec() -> OpenApi {
        },
        servers: vec![
            Server {
-                url: "https://api.revolt.chat".to_owned(),
-                description: Some("Revolt Production".to_owned()),
+                url: "https://api.stoat.chat".to_owned(),
+                description: Some("Stoat Production".to_owned()),
                ..Default::default()
            },
            Server {
-                url: "https://revolt.chat/api".to_owned(),
-                description: Some("Revolt Staging".to_owned()),
-                ..Default::default()
-            },
-            Server {
-                url: "http://local.revolt.chat:14702".to_owned(),
-                description: Some("Local Revolt Environment".to_owned()),
-                ..Default::default()
-            },
-            Server {
-                url: "http://local.revolt.chat:14702/0.8".to_owned(),
-                description: Some("Local Revolt Environment (v0.8)".to_owned()),
+                url: "https://beta.stoat.chat/api".to_owned(),
+                description: Some("Stoat Beta".to_owned()),
                ..Default::default()
            },
        ],
        external_docs: Some(ExternalDocs {
-            url: "https://developers.revolt.chat".to_owned(),
-            description: Some("Revolt Developer Documentation".to_owned()),
+            url: "https://developers.stoat.chat".to_owned(),
+            description: Some("Stoat Developer Documentation".to_owned()),
            ..Default::default()
        }),
        extensions,
@@ -254,19 +206,19 @@ fn custom_openapi_spec() -> OpenApi {
            Tag {
                name: "Core".to_owned(),
                description: Some(
-                    "Use in your applications to determine information about the Revolt node"
+                    "Use in your applications to determine information about the Stoat node"
                        .to_owned(),
                ),
                ..Default::default()
            },
            Tag {
                name: "User Information".to_owned(),
-                description: Some("Query and fetch users on Revolt".to_owned()),
+                description: Some("Query and fetch users on Stoat".to_owned()),
                ..Default::default()
            },
            Tag {
                name: "Direct Messaging".to_owned(),
-                description: Some("Direct message other users on Revolt".to_owned()),
+                description: Some("Direct message other users on Stoat".to_owned()),
                ..Default::default()
            },
            Tag {
@@ -283,7 +235,7 @@ fn custom_openapi_spec() -> OpenApi {
            },
            Tag {
                name: "Channel Information".to_owned(),
-                description: Some("Query and fetch channels on Revolt".to_owned()),
+                description: Some("Query and fetch channels on Stoat".to_owned()),
                ..Default::default()
            },
            Tag {
@@ -313,7 +265,7 @@ fn custom_openapi_spec() -> OpenApi {
            },
            Tag {
                name: "Server Information".to_owned(),
-                description: Some("Query and fetch servers on Revolt".to_owned()),
+                description: Some("Query and fetch servers on Stoat".to_owned()),
                ..Default::default()
            },
            Tag {
@@ -349,7 +301,7 @@ fn custom_openapi_spec() -> OpenApi {
            Tag {
                name: "Onboarding".to_owned(),
                description: Some(
-                    "After signing up to Revolt, users must pick a unique username".to_owned(),
+                    "After signing up to Stoat, users must pick a unique username".to_owned(),
                ),
                ..Default::default()
            },
@@ -361,7 +313,7 @@ fn custom_openapi_spec() -> OpenApi {
            Tag {
                name: "Web Push".to_owned(),
                description: Some(
-                    "Subscribe to and receive Revolt push notifications while offline".to_owned(),
+                    "Subscribe to and receive Stoat push notifications while offline".to_owned(),
                ),
                ..Default::default()
            },
--- 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,8 +25,6 @@ pub struct TestHarness {

 impl TestHarness {
    pub async fn new() -> TestHarness {
-        let config = revolt_config::config().await;
-
        let client = Client::tracked(crate::web().await)
            .await
            .expect("valid rocket instance");
@@ -49,19 +47,7 @@ impl TestHarness {
            .expect("`Authifier`")
            .clone();

-        let connection = amqprs::connection::Connection::open(
-            &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);
+        let amqp = AMQP::new_auto().await;

        TestHarness {
            client,
--- a/crates/services/autumn/Cargo.toml
+++ b/crates/services/autumn/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "revolt-autumn"
-version = "0.12.1"
+version = "0.13.7"
 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.12.1"
+version = "0.13.7"
 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.12.1"
+version = "0.13.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 publish = false
--- a/crates/services/january/src/main.rs
+++ b/crates/services/january/src/main.rs
@@ -21,32 +21,26 @@ async fn main() -> Result<(), std::io::Error> {
    // Configure API schema
    #[derive(OpenApi)]
    #[openapi(
-        modifiers(&SecurityAddon),
-        paths(
-            api::root,
-            api::proxy,
-            api::embed
-        ),
-        components(
-            schemas(
-                api::RootResponse,
-                revolt_result::Error,
-                revolt_result::ErrorType,
-                revolt_models::v0::ImageSize,
-                revolt_models::v0::Image,
-                revolt_models::v0::Video,
-                revolt_models::v0::TwitchType,
-                revolt_models::v0::LightspeedType,
-                revolt_models::v0::BandcampType,
-                revolt_models::v0::Special,
-                revolt_models::v0::WebsiteMetadata,
-                revolt_models::v0::Text,
-                revolt_models::v0::Embed
-            )
-        )
+        paths(api::root, api::proxy, api::embed),
+        components(schemas(
+            api::RootResponse,
+            revolt_result::Error,
+            revolt_result::ErrorType,
+            revolt_models::v0::ImageSize,
+            revolt_models::v0::Image,
+            revolt_models::v0::Video,
+            revolt_models::v0::TwitchType,
+            revolt_models::v0::LightspeedType,
+            revolt_models::v0::BandcampType,
+            revolt_models::v0::Special,
+            revolt_models::v0::WebsiteMetadata,
+            revolt_models::v0::Text,
+            revolt_models::v0::Embed
+        ))
    )]
    struct ApiDoc;

+    #[allow(dead_code)]
    struct SecurityAddon;

    impl Modify for SecurityAddon {
--- a/crates/services/january/src/requests.rs
+++ b/crates/services/january/src/requests.rs
@@ -4,6 +4,7 @@ use mime::Mime;
 use pdk_ip_filter_lib::IpFilter;
 use regex::Regex;
 use reqwest::{
+    dns::{Addrs, Name, Resolve},
    header::{self, CONTENT_TYPE},
    redirect, Client, Response,
 };
@@ -11,6 +12,7 @@ use revolt_config::{config, report_internal_error};
 use revolt_files::{create_thumbnail, decode_image, image_size_vec, is_valid_image, video_size};
 use revolt_models::v0::{Embed, Image, ImageSize, Video};
 use revolt_result::{create_error, Error, Result, ToRevoltError};
+use std::net::{IpAddr, SocketAddr};
 use std::{
    io::{Cursor, Write},
    str::FromStr,
@@ -21,6 +23,7 @@ use url::{Host, Url};
 lazy_static! {
    /// Request client
    static ref CLIENT: Client = reqwest::Client::builder()
+        .dns_resolver(CachedDnsResolver {})
        .timeout(Duration::from_secs(10)) // TODO config
        .connect_timeout(Duration::from_secs(5)) // TODO config
        .redirect(redirect::Policy::none())
@@ -58,18 +61,73 @@ lazy_static! {
        .time_to_live(Duration::from_secs(60)) // For up to 1 minute
        .build();

+    static ref DNS_CACHE: moka::future::Cache<String, Vec<SocketAddr>> = moka::future::Cache::builder()
+        .max_capacity(10_000)
+        .time_to_idle(Duration::from_secs(30))
+        .build();
+
    static ref IP_BLOCKLIST: IpFilter = IpFilter::block(&[
+        "0.0.0.0/8",
        "10.0.0.0/8",
        "192.168.0.0/16",
        "127.0.0.0/8",
        "172.16.0.0/12",
        "169.254.0.0/16",
        "::1",
-        "fc00::/7"
+        "fc00::/7",
        ]
    ).unwrap();
 }

+#[derive(Clone)]
+pub struct IPRequest {
+    url: Url,
+    ip: IpAddr,
+    pub blocked: bool,
+}
+
+impl From<IPRequest> for Url {
+    fn from(value: IPRequest) -> Self {
+        let mut url = value.url.clone();
+        url.set_host(Some(&value.ip.to_string()))
+            .map(|_| url)
+            .unwrap_or(value.url)
+    }
+}
+
+struct CachedDnsResolver {}
+
+impl reqwest::dns::Resolve for CachedDnsResolver {
+    fn resolve(&self, name: Name) -> reqwest::dns::Resolving {
+        Box::pin(async move {
+            {
+                if let Some(addrs) = DNS_CACHE.get(&name.as_str().to_string()).await {
+                    let resp: Addrs = Box::new(addrs.clone().into_iter());
+                    return Ok(resp);
+                }
+            }
+
+            let mut lookup = name.as_str().to_string();
+            if !lookup.contains(":") {
+                lookup += ":0";
+            }
+
+            let fallback: Vec<SocketAddr> = tokio::net::lookup_host(&lookup)
+                .await
+                .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) })?
+                .collect();
+
+            {
+                DNS_CACHE
+                    .insert(name.as_str().to_string().clone(), fallback.clone())
+                    .await;
+                let addrs: Addrs = Box::new(fallback.clone().into_iter());
+                Ok(addrs)
+            }
+        })
+    }
+}
+
 /// Information about a successful request
 pub struct Request {
    response: Response,
@@ -275,7 +333,12 @@ impl Request {
        let mut url = url;
        let url_host_str = url.host_str().ok_or(create_error!(ProxyError))?.to_string();

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

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

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

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

                    let config = config().await;

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

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

-        Ok(false)
+        Ok(IPRequest {
+            url: url.clone(),
+            ip: resolved_address,
+            blocked: false,
+        })
    }
 }
--- a/docs/docusaurus.config.ts
+++ b/docs/docusaurus.config.ts
@@ -6,7 +6,7 @@ import type { ScalarOptions } from '@scalar/docusaurus';
 const config: Config = {
  title: 'Stoat Developers',
  tagline: 'Developer documentation for Stoat',
-  favicon: 'https://stoat.chat/favicon.svg',
+  favicon: 'https://stoat.chat/favicon-stoat.svg',

  future: {
    v4: true,
@@ -80,10 +80,10 @@ const config: Config = {
      respectPrefersColorScheme: true,
    },
    navbar: {
-      title: 'Stoat Developers',
      logo: {
-        alt: 'Stoat',
-        src: 'https://stoat.chat/favicon.svg',
+        alt: 'Stoat for Developers',
+        src: '/img/navbar.light.svg',
+        srcDark: '/img/navbar.dark.svg'
      },
      items: [
        {
--- a/docs/static/img/navbar.dark.svg
+++ b/docs/static/img/navbar.dark.svg
--- a/docs/static/img/navbar.light.svg
+++ b/docs/static/img/navbar.light.svg
--- a/renovate.json
+++ b/renovate.json
@@ -0,0 +1,11 @@
+{
+  "$schema": "https://docs.renovatebot.com/renovate-schema.json",
+  "extends": [
+    "config:recommended"
+  ],
+  "cloneSubmodules": true,
+  "cloneSubmodulesFilter": [
+    "*",
+    "!packages/client/assets"
+  ]
+}
--- a/version.txt
+++ b/version.txt
@@ -1 +1 @@
-0.12.1
+0.13.7
Author	SHA1	Message	Date
Zomatree	4586e4738a	feat: publish events over amqp Signed-off-by: Zomatree <me@zomatree.live>	2026-06-19 21:05:22 +01:00
Tom	c70459b10c	fix: openapi using old naming (#777 ) * fix: openapi using old naming Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: remove january openapi security header Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix(docs): more Revolt usage Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> --------- Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>	2026-06-07 16:56:00 -07:00
Asraye	bebfe34922	fix: point docs favicon to correct location (#789 ) chore(docs): update favicon Signed-off-by: Asraye <asrayeofficial@gmail.com>	2026-06-03 11:06:38 -07:00
Tom	5b769b60de	chore(docs): update header logo (#796 ) * chore(docs): update header logo Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: don't step on toes Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> --------- Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>	2026-06-02 17:42:04 -07:00
stoat-tofu[bot]	0896e68882	chore: modify renovate.json	2026-06-01 13:28:29 +00:00
stoat-tofu[bot]	65acc64034	chore: modify .github/workflows/renovate.yml	2026-06-01 13:28:27 +00:00
İspik	bd987bf72a	chore: update unicode emoji list (#781 ) Signed-off-by: ispik <ispik@ispik.dev>	2026-05-28 14:55:49 -07:00
stoat-release[bot]	7937179db7	chore(main): release 0.13.7 (#770 ) Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>	2026-05-21 19:33:49 +01:00
İspik	2d308e03d5	fix: sanitize emoji input to handle variation selectors (#774 ) Signed-off-by: ispik <ispik@ispik.dev>	2026-05-21 08:21:43 -07:00
Paul Makles	b38499a05b	ci: hard code packages, gh token limitation [skip ci] (#773 )	2026-05-20 20:10:09 +01:00
Paul Makles	4815429952	ci: create Docker images for PR preview (#772 )	2026-05-20 19:58:18 +01:00
Zomatree	0d9ae508d9	fix: update mention count badge for channel acks (#769 ) Signed-off-by: Zomatree <me@zomatree.live>	2026-05-18 23:38:07 -07:00
stoat-release[bot]	03b52655ff	chore(main): release 0.13.6 (#762 ) * chore(main): release 0.13.6 * chore: update Cargo.lock Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> --------- Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: stoat-release[bot] <245062572+stoat-release[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>	2026-05-18 15:53:42 -07:00
Angelo Kontaxis	5b1985381a	chore: switch to lapin (#767 ) * chore: begin switching to lapin fully Signed-off-by: Zomatree <me@zomatree.live> * chore: update rest of pushd to lapin Signed-off-by: Zomatree <me@zomatree.live> * chore: cleanup code Signed-off-by: Zomatree <me@zomatree.live> * chore: cleanup code Signed-off-by: Zomatree <me@zomatree.live> * fix: github webui sucks Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> --------- Signed-off-by: Zomatree <me@zomatree.live> Signed-off-by: Tom <iamtomahawkx@gmail.com> Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> Co-authored-by: Tom <iamtomahawkx@gmail.com> Release-As: 0.13.6	2026-05-18 15:46:17 -07:00
Tom	018afaf38f	fix: set env var for publishing crates (#768 ) * fix: set env var for publishing crates Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> Release-As: 0.13.6	2026-05-18 15:11:44 -07:00
İspik	af0d8aad14	feat: user slowmode events (#760 ) * feat: user slowmode events Signed-off-by: ispik <ispik@ispik.dev> * fix: remove debug print statement for slowmodes Signed-off-by: ispik <ispik@ispik.dev> * refactor: Send user slowmodes as websocket connects instead of trying to send it in ready payload Signed-off-by: ispik <ispik@ispik.dev> * refactor: optimize user slowmode handling with bulk operations Signed-off-by: ispik <ispik@ispik.dev> * chore: specify release version Release-As: 0.13.6 --------- Signed-off-by: ispik <ispik@ispik.dev>	2026-05-18 14:51:43 -07:00
Tom	acbc087982	feat: Update FCM payload for android notifications (#766 ) * feat: modify fcm payload to jens will Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: add message id Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: rename field Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: whitespace Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> --------- Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>	2026-05-18 10:57:23 -07:00
Tom	2871632382	fix: voice ingress crashing due to new Result in AMQP::new_auto() (#765 ) Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>	2026-05-18 10:56:01 -07:00
Tom	494c8b7cab	fix: Use proper headers to determine IP when not behind cloudflare (#764 ) Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>	2026-05-18 10:51:19 -07:00
Paul Makles	26a8692677	ci: ignore test errors on main (#763 )	2026-05-17 14:54:57 -05:00
Paul Makles	298742dbad	fix: include `minio` region as tests need it (#761 )	2026-05-17 14:41:44 -05:00
stoat-release[bot]	6c920de03a	chore(main): release 0.13.5 (#759 ) * chore(main): release 0.13.5 * chore: update Cargo.lock Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> --------- Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: stoat-release[bot] <245062572+stoat-release[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>	2026-05-17 11:02:21 -07:00
Angelo Kontaxis	c902077cf5	fix: dont panic on hash missing when deleting files (#755 ) Signed-off-by: Zomatree <me@zomatree.live>	2026-05-17 10:59:56 -07:00
Tom	19ee535f45	Merge commit from fork * fix: cache dns & block more ranges * fix: idle time instead of ttl Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> --------- Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>	2026-05-17 10:55:38 -07:00
stoat-release[bot]	ee4575470b	chore(main): release 0.13.4 (#754 ) Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>	2026-05-16 18:27:44 +01:00
Paul Makles	6cfee1f601	fix: add TLS feature to livekit-api crate (#753 )	2026-05-16 18:23:45 +01:00
stoat-release[bot]	ab9b8ccfca	chore(main): release 0.13.3 (#750 ) * chore(main): release 0.13.3 * chore: update Cargo.lock Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> --------- Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: stoat-release[bot] <245062572+stoat-release[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>	2026-05-15 12:22:27 -07:00
Tom	7647cfc8d9	fix: don't automatically set up rabbitmq in delta (#749 ) fix: don't declare queues which seem to cause the backend to crash in prod for now, these exchanges/queues/bindings will need to be declared manually. Hopefully the lapin rewrite will fix this. Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>	2026-05-15 12:17:36 -07:00
stoat-release[bot]	8157e1f6e9	chore(main): release 0.13.2 (#747 ) * chore(main): release 0.13.2 * chore: update Cargo.lock Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> --------- Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: stoat-release[bot] <245062572+stoat-release[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>	2026-05-11 15:26:03 +01:00
Paul Makles	fcb8091cd7	fix: update default exchange to `revolt.default` (#746 ) Signed-off-by: Paul Makles <me@insrt.uk>	2026-05-11 15:21:34 +01:00
stoat-release[bot]	260036488d	chore(main): release 0.13.1 (#745 ) Co-authored-by: stoat-release[bot] <245062572+stoat-release[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>	2026-05-10 15:34:41 +01:00
Tom	1100eaf46f	fix: amqprs startup bug (#744 )	2026-05-10 15:22:26 +01:00
stoat-release[bot]	d52e84c5d3	chore(main): release 0.13.0 (#722 ) Co-authored-by: stoat-release[bot] <245062572+stoat-release[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>	2026-05-09 17:26:38 +01:00
@@ -1 +1 @@
 .12.1
 .13.7