chore: ignore WS protocol errors

refactor: redo the bonfire client structure

.github/workflows/book.yml

@@ -0,0 +1,37 @@
+name: Build documentation
+on:
+  push:
+    branches:
+      - main
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write # To push a branch
+      pages: write # To push to a GitHub Pages site
+      id-token: write # To update the deployment status
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - name: Install latest mdbook
+        run: |
+          tag=$(curl 'https://api.github.com/repos/rust-lang/mdbook/releases/latest' | jq -r '.tag_name')
+          url="https://github.com/rust-lang/mdbook/releases/download/${tag}/mdbook-${tag}-x86_64-unknown-linux-gnu.tar.gz"
+          mkdir mdbook
+          curl -sSL $url | tar -xz --directory=./mdbook
+          echo `pwd`/mdbook >> $GITHUB_PATH
+      - name: Build Book
+        run: |
+          cd doc
+          mdbook build
+      - name: Setup Pages
+        uses: actions/configure-pages@v4
+      - name: Upload artifact
+        uses: actions/upload-pages-artifact@v3
+        with:
+          path: "doc/book"
+      - name: Deploy to GitHub Pages
+        id: deployment
+        uses: actions/deploy-pages@v4

.github/workflows/docker.yaml

@@ -5,6 +5,8 @@ on:
     tags:
       - "*"
   pull_request:
+    branches:
+      - "main"
     paths:
       - "Dockerfile"
 
@@ -12,10 +14,6 @@ permissions:
   contents: read
   packages: write
 
-concurrency: 
-  group: ${{ github.head_ref }}
-  cancel-in-progress: true
-
 jobs:
   base:
     name: Test base image build
@@ -51,6 +49,13 @@ jobs:
         uses: docker/setup-buildx-action@v2
 
       # Authenticate with Docker Hub and GHCR
+      - name: Login to DockerHub
+        uses: docker/login-action@v2
+        with:
+          registry: docker.io
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
       - name: Login to Github Container Registry
         uses: docker/login-action@v2
         with:
@@ -67,13 +72,14 @@ jobs:
           platforms: linux/amd64,linux/arm64
           tags: ghcr.io/${{ github.repository_owner }}/base:latest
 
-      # stoatchat/api
+      # revoltchat/server
       - name: Docker meta
         id: meta-delta
         uses: docker/metadata-action@v4
         with:
           images: |
-            ghcr.io/stoatchat/api
+            docker.io/revoltchat/server
+            ghcr.io/revoltchat/server
       - name: Publish
         uses: docker/build-push-action@v4
         with:
@@ -86,13 +92,14 @@ jobs:
             BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
           labels: ${{ steps.meta-delta.outputs.labels }}
 
-      # stoatchat/events
+      # revoltchat/bonfire
       - name: Docker meta
         id: meta-bonfire
         uses: docker/metadata-action@v4
         with:
           images: |
-            ghcr.io/stoatchat/events
+            docker.io/revoltchat/bonfire
+            ghcr.io/revoltchat/bonfire
       - name: Publish
         uses: docker/build-push-action@v4
         with:
@@ -105,13 +112,14 @@ jobs:
             BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
           labels: ${{ steps.meta-bonfire.outputs.labels }}
 
-      # stoatchat/file-server
+      # revoltchat/autumn
       - name: Docker meta
         id: meta-autumn
         uses: docker/metadata-action@v4
         with:
           images: |
-            ghcr.io/stoatchat/file-server
+            docker.io/revoltchat/autumn
+            ghcr.io/revoltchat/autumn
       - name: Publish
         uses: docker/build-push-action@v4
         with:
@@ -124,13 +132,14 @@ jobs:
             BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
           labels: ${{ steps.meta-autumn.outputs.labels }}
 
-      # stoatchat/proxy
+      # revoltchat/january
       - name: Docker meta
         id: meta-january
         uses: docker/metadata-action@v4
         with:
           images: |
-            ghcr.io/stoatchat/proxy
+            docker.io/revoltchat/january
+            ghcr.io/revoltchat/january
       - name: Publish
         uses: docker/build-push-action@v4
         with:
@@ -143,32 +152,14 @@ jobs:
             BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
           labels: ${{ steps.meta-january.outputs.labels }}
 
-      # stoatchat/gifbox
-      - name: Docker meta
-        id: meta-gifbox
-        uses: docker/metadata-action@v4
-        with:
-          images: |
-            ghcr.io/stoatchat/gifbox
-      - name: Publish
-        uses: docker/build-push-action@v4
-        with:
-          context: .
-          push: true
-          platforms: linux/amd64,linux/arm64
-          file: crates/services/gifbox/Dockerfile
-          tags: ${{ steps.meta-gifbox.outputs.tags }}
-          build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
-          labels: ${{ steps.meta-gifbox.outputs.labels }}
-
-      # stoatchat/crond
+      # revoltchat/crond
       - name: Docker meta
         id: meta-crond
         uses: docker/metadata-action@v4
         with:
           images: |
-            ghcr.io/stoatchat/crond
+            docker.io/revoltchat/crond
+            ghcr.io/revoltchat/crond
       - name: Publish
         uses: docker/build-push-action@v4
         with:
@@ -181,13 +172,14 @@ jobs:
             BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
           labels: ${{ steps.meta-crond.outputs.labels }}
 
-      # stoatchat/pushd
+      # revoltchat/pushd
       - name: Docker meta
         id: meta-pushd
         uses: docker/metadata-action@v4
         with:
           images: |
-            ghcr.io/stoatchat/pushd
+            docker.io/revoltchat/pushd
+            ghcr.io/revoltchat/pushd
       - name: Publish
         uses: docker/build-push-action@v4
         with:
@@ -199,22 +191,3 @@ jobs:
           build-args: |
             BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
           labels: ${{ steps.meta-pushd.outputs.labels }}
-
-      # stoatchat/voice-ingress
-      - name: Docker meta
-        id: meta-voice-ingress
-        uses: docker/metadata-action@v4
-        with:
-          images: |
-            ghcr.io/stoatchat/voice-ingress
-      - name: Publish
-        uses: docker/build-push-action@v4
-        with:
-          context: .
-          push: true
-          platforms: linux/amd64,linux/arm64
-          file: crates/daemons/voice-ingress/Dockerfile
-          tags: ${{ steps.meta-voice-ingress.outputs.tags }}
-          build-args: |
-            BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
-          labels: ${{ steps.meta-voice-ingress.outputs.labels }}

.github/workflows/docs-test.yml

@@ -1,29 +0,0 @@
-name: Documentation (test)
-
-on:
-  pull_request:
-
-jobs:
-  test-deploy:
-    name: Test deployment
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: ./docs
-    steps:
-      - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-      - uses: pnpm/action-setup@v4
-        with:
-          package_json_file: ./docs/package.json
-      - uses: actions/setup-node@v4
-        with:
-          node-version: lts/*
-          cache: pnpm
-          cache-dependency-path: ./docs/pnpm-lock.yaml
-
-      - name: Install dependencies
-        run: pnpm install
-      - name: Test build website
-        run: pnpm run build

.github/workflows/docs.yml

@@ -1,54 +0,0 @@
-name: Documentation
-
-on:
-  push:
-    branches:
-      - main
-
-jobs:
-  build:
-    name: Build Docusaurus
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: ./docs
-    steps:
-      - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-      - uses: pnpm/action-setup@v4
-        with:
-          package_json_file: ./docs/package.json
-      - uses: actions/setup-node@v4
-        with:
-          node-version: lts/*
-          cache: pnpm
-          cache-dependency-path: ./docs/pnpm-lock.yaml
-
-      - name: Install dependencies
-        run: pnpm install
-      - name: Build website
-        run: pnpm run build
-
-      - name: Upload Build Artifact
-        uses: actions/upload-pages-artifact@v3
-        with:
-          path: ./docs/build
-
-  deploy:
-    name: Deploy to GitHub Pages
-    needs: build
-
-    permissions:
-      pages: write # to deploy to Pages
-      id-token: write # to verify the deployment originates from an appropriate source
-
-    environment:
-      name: github-pages
-      url: ${{ steps.deployment.outputs.page_url }}
-
-    runs-on: ubuntu-latest
-    steps:
-      - name: Deploy to GitHub Pages
-        id: deployment
-        uses: actions/deploy-pages@v4

.github/workflows/rust.yaml

@@ -2,12 +2,16 @@ name: Rust build, test, and generate specification
 
 on:
   push:
-    branches: [main]
+    paths-ignore:
+      - ".github/**"
+      - "!.github/workflows/rust.yaml"
+      - ".vscode/**"
+      - "doc/**"
+      - ".gitignore"
+      - "LICENSE"
+      - "README"
   pull_request:
-
-concurrency: 
-  group: ${{ github.head_ref }}
-  cancel-in-progress: true
+    branches: [main]
 
 env:
   CARGO_TERM_COLOR: always
@@ -18,10 +22,6 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
-      - name: Free up disk space
-        run: |
-          sudo rm -rf /usr/local/lib/android /usr/share/dotnet /opt/ghc
-
       - name: Install latest stable
         uses: actions-rs/toolchain@v1
         with:
@@ -32,7 +32,7 @@ jobs:
         uses: baptiste0928/cargo-install@v1
         with:
           crate: cargo-nextest
-          args: --locked
+          locked: true
 
       - name: Run cargo build
         uses: actions-rs/cargo@v1
@@ -73,9 +73,9 @@ jobs:
         if: github.event_name != 'pull_request' && github.ref_name == 'main'
         uses: actions/checkout@v3
         with:
-          repository: stoatchat/javascript-client-api
+          repository: revoltchat/api
           path: api
-          ssh-key: ${{ secrets.DEPLOY_KEY_JAVASCRIPT_CLIENT_API }}
+          token: ${{ secrets.PAT }}
 
       - name: Download OpenAPI specification
         if: github.event_name != 'pull_request' && github.ref_name == 'main'

.github/workflows/triage_issue.yml

@@ -0,0 +1,54 @@
+name: Add Issue to Board
+
+on:
+  issues:
+    types: [opened]
+
+jobs:
+  track_issue:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Get project data
+        env:
+          GITHUB_TOKEN: ${{ secrets.PAT }}
+        run: |
+          gh api graphql -f query='
+            query {
+              organization(login: "revoltchat"){
+                projectV2(number: 3) {
+                  id
+                  fields(first:20) {
+                    nodes {
+                      ... on ProjectV2SingleSelectField {
+                        id
+                        name
+                        options {
+                          id
+                          name
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            }' > project_data.json
+
+          echo 'PROJECT_ID='$(jq '.data.organization.projectV2.id' project_data.json) >> $GITHUB_ENV
+          echo 'STATUS_FIELD_ID='$(jq '.data.organization.projectV2.fields.nodes[] | select(.name== "Status") | .id' project_data.json) >> $GITHUB_ENV
+          echo 'TODO_OPTION_ID='$(jq '.data.organization.projectV2.fields.nodes[] | select(.name== "Status") | .options[] | select(.name=="Todo") |.id' project_data.json) >> $GITHUB_ENV
+
+      - name: Add issue to project
+        env:
+          GITHUB_TOKEN: ${{ secrets.PAT }}
+          ISSUE_ID: ${{ github.event.issue.node_id }}
+        run: |
+          item_id="$( gh api graphql -f query='
+            mutation($project:ID!, $issue:ID!) {
+              addProjectV2ItemById(input: {projectId: $project, contentId: $issue}) {
+                item {
+                  id
+                }
+              }
+            }' -f project=$PROJECT_ID -f issue=$ISSUE_ID --jq '.data.addProjectV2ItemById.item.id')"
+
+          echo 'ITEM_ID='$item_id >> $GITHUB_ENV

.github/workflows/triage_pr.yml

@@ -0,0 +1,79 @@
+name: Add PR to Board
+
+on:
+  pull_request_target:
+    types: [opened, synchronize, ready_for_review, review_requested]
+
+jobs:
+  track_pr:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Get project data
+        env:
+          GITHUB_TOKEN: ${{ secrets.PAT }}
+        run: |
+          gh api graphql -f query='
+            query {
+              organization(login: "revoltchat"){
+                projectV2(number: 5) {
+                  id
+                  fields(first:20) {
+                    nodes {
+                      ... on ProjectV2SingleSelectField {
+                        id
+                        name
+                        options {
+                          id
+                          name
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            }' > project_data.json
+
+          echo 'PROJECT_ID='$(jq '.data.organization.projectV2.id' project_data.json) >> $GITHUB_ENV
+          echo 'STATUS_FIELD_ID='$(jq '.data.organization.projectV2.fields.nodes[] | select(.name== "Status") | .id' project_data.json) >> $GITHUB_ENV
+          echo 'INCOMING_OPTION_ID='$(jq '.data.organization.projectV2.fields.nodes[] | select(.name== "Status") | .options[] | select(.name=="🆕 Untriaged") |.id' project_data.json) >> $GITHUB_ENV
+
+      - name: Add PR to project
+        env:
+          GITHUB_TOKEN: ${{ secrets.PAT }}
+          PR_ID: ${{ github.event.pull_request.node_id }}
+        run: |
+          item_id="$( gh api graphql -f query='
+            mutation($project:ID!, $pr:ID!) {
+              addProjectV2ItemById(input: {projectId: $project, contentId: $pr}) {
+                item {
+                  id
+                }
+              }
+            }' -f project=$PROJECT_ID -f pr=$PR_ID --jq '.data.addProjectV2ItemById.item.id')"
+
+          echo 'ITEM_ID='$item_id >> $GITHUB_ENV
+
+      - name: Set fields
+        env:
+          GITHUB_TOKEN: ${{ secrets.PAT }}
+        run: |
+          gh api graphql -f query='
+            mutation (
+              $project: ID!
+              $item: ID!
+              $status_field: ID!
+              $status_value: String!
+            ) {
+              set_status: updateProjectV2ItemFieldValue(input: {
+                projectId: $project
+                itemId: $item
+                fieldId: $status_field
+                value: {
+                  singleSelectOptionId: $status_value
+                }
+              }) {
+                projectV2Item {
+                  id
+                }
+              }
+            }' -f project=$PROJECT_ID -f item=$ITEM_ID -f status_field=$STATUS_FIELD_ID -f status_value=${{ env.INCOMING_OPTION_ID }} --silent

.github/workflows/validate-pr-title.yml

@@ -1,20 +0,0 @@
-name: "Lint PR"
-
-on:
-  pull_request_target:
-    types:
-      - opened
-      - reopened
-      - edited
-      - synchronize
-
-jobs:
-  main:
-    name: Validate PR title
-    runs-on: ubuntu-latest
-    permissions:
-      pull-requests: read
-    steps:
-      - uses: amannn/action-semantic-pull-request@v6
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.gitignore

@@ -11,5 +11,4 @@ venv/
 .vercel
 .DS_Store
 
-livekit.yml
 .idea

Cargo.toml

@@ -11,9 +11,8 @@ members = [
 
 [patch.crates-io]
 redis23 = { package = "redis", version = "0.23.3", git = "https://github.com/revoltchat/redis-rs", rev = "523b2937367e17bd0073722bf6e23d06042cb4e4" }
-authifier = { package = "authifier", version = "1.0.10", path = "../authifier/crates/authifier" }
-rocket_authifier = { package = "rocket_authifier", version = "1.0.10", path = "../authifier/crates/rocket_authifier" }
-iso8601-timestamp = { path = "../iso8601-timestamp" }
+#authifier = { package = "authifier", version = "1.0.10", path = "../authifier/crates/authifier" }
+#rocket_authifier = { package = "rocket_authifier", version = "1.0.10", path = "../authifier/crates/rocket_authifier" }
 
 # I'm 99% sure this is overloading the GitHub worker
 # hence builds have been failing since, let's just

Dockerfile

@@ -27,14 +27,10 @@ COPY crates/core/parser/Cargo.toml ./crates/core/parser/
 COPY crates/core/permissions/Cargo.toml ./crates/core/permissions/
 COPY crates/core/presence/Cargo.toml ./crates/core/presence/
 COPY crates/core/result/Cargo.toml ./crates/core/result/
-COPY crates/core/coalesced/Cargo.toml ./crates/core/coalesced/
-COPY crates/core/ratelimits/Cargo.toml ./crates/core/ratelimits/
 COPY crates/services/autumn/Cargo.toml ./crates/services/autumn/
 COPY crates/services/january/Cargo.toml ./crates/services/january/
-COPY crates/services/gifbox/Cargo.toml ./crates/services/gifbox/
 COPY crates/daemons/crond/Cargo.toml ./crates/daemons/crond/
 COPY crates/daemons/pushd/Cargo.toml ./crates/daemons/pushd/
-COPY crates/daemons/voice-ingress/Cargo.toml ./crates/daemons/voice-ingress/
 RUN sh /tmp/build-image-layer.sh deps
 
 # Build all apps

Dockerfile.useCurrentArch

@@ -23,14 +23,10 @@ COPY crates/core/parser/Cargo.toml ./crates/core/parser/
 COPY crates/core/permissions/Cargo.toml ./crates/core/permissions/
 COPY crates/core/presence/Cargo.toml ./crates/core/presence/
 COPY crates/core/result/Cargo.toml ./crates/core/result/
-COPY crates/core/coalesced/Cargo.toml ./crates/core/coalesced/
-COPY crates/core/ratelimits/Cargo.toml ./crates/core/ratelimits/
 COPY crates/services/autumn/Cargo.toml ./crates/services/autumn/
 COPY crates/services/january/Cargo.toml ./crates/services/january/
-COPY crates/services/gifbox/Cargo.toml ./crates/services/gifbox/
 COPY crates/daemons/crond/Cargo.toml ./crates/daemons/crond/
 COPY crates/daemons/pushd/Cargo.toml ./crates/daemons/pushd/
-COPY crates/daemons/voice-ingress/Cargo.toml ./crates/daemons/voice-ingress/
 RUN sh /tmp/build-image-layer.sh deps
 
 # Build all apps

README.md

@@ -21,11 +21,9 @@ The services and libraries that power the Revolt service.<br/>
 | `core/permissions` | [crates/core/permissions](crates/core/permissions) | Core: Permission Logic              | ![Crates.io Version](https://img.shields.io/crates/v/revolt-permissions) ![Crates.io Version](https://img.shields.io/crates/msrv/revolt-permissions) ![Crates.io Version](https://img.shields.io/crates/size/revolt-permissions) ![Crates.io License](https://img.shields.io/crates/l/revolt-permissions) |
 | `core/presence`    | [crates/core/presence](crates/core/presence)       | Core: User Presence                 | ![Crates.io Version](https://img.shields.io/crates/v/revolt-presence) ![Crates.io Version](https://img.shields.io/crates/msrv/revolt-presence) ![Crates.io Version](https://img.shields.io/crates/size/revolt-presence) ![Crates.io License](https://img.shields.io/crates/l/revolt-presence)             |
 | `core/result`      | [crates/core/result](crates/core/result)           | Core: Result and Error types        | ![Crates.io Version](https://img.shields.io/crates/v/revolt-result) ![Crates.io Version](https://img.shields.io/crates/msrv/revolt-result) ![Crates.io Version](https://img.shields.io/crates/size/revolt-result) ![Crates.io License](https://img.shields.io/crates/l/revolt-result)                     |
-| `core/coalesced`   | [crates/core/coalesced](crates/core/coalesced)     | Core: Coalescion service            | ![Crates.io Version](https://img.shields.io/crates/v/revolt-coalesced) ![Crates.io Version](https://img.shields.io/crates/msrv/revolt-coalesced) ![Crates.io Version](https://img.shields.io/crates/size/revolt-coalesced) ![Crates.io License](https://img.shields.io/crates/l/revolt-coalesced)         |
 | `delta`            | [crates/delta](crates/delta)                       | REST API server                     | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue)                                                                                                                                                                                                                                |
 | `bonfire`          | [crates/bonfire](crates/bonfire)                   | WebSocket events server             | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue)                                                                                                                                                                                                                                |
 | `services/january` | [crates/services/january](crates/services/january) | Proxy server                        | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue)                                                                                                                                                                                                                                |
-| `services/gifbox`  | [crates/services/gifbox](crates/services/gifbox)   | Tenor proxy server                  | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue)                                                                                                                                                                                                                                |
 | `services/autumn`  | [crates/services/autumn](crates/services/autumn)   | File server                         | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue)                                                                                                                                                                                                                                |
 | `daemons/crond`    | [crates/daemons/crond](crates/daemons/crond)       | Timed data clean up daemon server   | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue)                                                                                                                                                                                                                                |
 | `daemons/pushd`    | [crates/daemons/pushd](crates/daemons/pushd)       | Push notification daemon server     | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue)                                                                                                                                                                                                                                |
@@ -68,7 +66,6 @@ As a heads-up, the development environment uses the following ports:
 | `crates/bonfire`          |     14703      |
 | `crates/services/autumn`  |     14704      |
 | `crates/services/january` |     14705      |
-| `crates/services/gifbox`  |     14706      |
 
 Now you can clone and build the project:
 
@@ -144,8 +141,6 @@ cargo run --bin revolt-bonfire
 cargo run --bin revolt-autumn
 # run the proxy server
 cargo run --bin revolt-january
-# run the tenor proxy
-cargo run --bin revolt-gifbox
 # run the push daemon (not usually needed in regular development)
 cargo run --bin revolt-pushd
 

Revolt.toml

@@ -26,11 +26,6 @@ january = "http://local.revolt.chat:14705"
 voso_legacy = ""
 voso_legacy_ws = ""
 
-# Public urls for livekit nodes
-# each entry here should have a corresponding entry under `api.livekit.nodes`
-[hosts.livekit]
-worldwide = "ws://local.revolt.chat:14706"
-
 [api]
 
 [api.smtp]
@@ -45,18 +40,6 @@ port = 14025
 use_tls = false
 use_starttls = false
 
-[api.livekit]
-
-# Config for livekit nodes
-# Make sure to change the secret when deploying
-# The key and secret should match the values livekit is using
-[api.livekit.nodes.worldwide]
-url = "http://livekit"
-lat = 0.0
-lon = 0.0
-key = "worldwide"
-secret = "ZjCofRlfm6GGtjlifmNpCDkcQbEIIVC0"
-
 [files.s3]
 # S3 protocol endpoint
 endpoint = "http://127.0.0.1:14009"

STYLE_GUIDE.md

@@ -0,0 +1,73 @@
+# Code Style Guide
+
+Beyond using Cargo format and Clippy, there are some specific code style guidelines laid out in this document for different parts of the project.
+
+## Writing Style
+
+- Shorten "identifier" to "Id" with that exact casing, i.e. Server Id.
+
+## `core/database` crate
+
+w.r.t. `model.rs` files
+
+- All struct definitions must be commented.
+  ```rust
+  /// Server
+  pub struct Server {
+    /// Name of the server
+    pub name: String,
+  ```
+- Struct definitions should not include derives unless necessary (if additional traits such as Hash are required) and instead use `auto_derived!` and `auto_derived_partial!`.
+  ```rust
+  auto_derived_partial!(
+    /// Server
+    pub struct Server { .. },
+    "PartialServer"
+  );
+  ```
+- `auto_derived!` macro accepts multiple entries and should be used as such:
+
+  ```rust
+  auto_derived!(
+    /// Optional fields on server object
+    pub enum FieldsServer { .. }
+
+    /// Optional fields on server object
+    pub enum FieldsRole { .. }
+  );
+  ```
+
+- If special serialisation conditions are required, such as checking if a boolean is false, use the existing definitions for these functions from the crate root:
+  ```rust
+  #[serde(skip_serializing_if = "crate::if_false", default)]
+  ```
+- `impl` blocks may be defined below the struct definitions and should be ordered in the same order of definition. Methods in the block must follow the same guidelines as traits where-in: methods are ordered in terms of CRUD, there are empty line breaks, and methods are commented.
+
+w.r.t. `ops` module for models
+
+- All traits must use a the name format `AbstractPlural` where Plural is the plural form of the collection. e.g. Servers
+- Traits defined must follow these guidelines:
+
+  - Methods are ordered in terms of CRUD, create-read-update-delete ordering.
+
+    ```rust
+    #[async_trait]
+    pub trait AbstractServerMembers: Sync + Send {
+      /// Insert a new server member into the database
+      async fn insert_member(&self, member: &Member) -> Result<()>;
+
+      /// Fetch a server member by their id
+      async fn fetch_member(&self, server_id: &str, user_id: &str) -> Result<Member>;
+
+      /// Update information for a server member
+      async fn update_member(&self, .. ) -> Result<()>;
+
+      /// Delete a server member by their id
+      async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
+    }
+    ```
+
+  - There should be an empty line break between each method declaration.
+  - All methods must have an appropriate comment.
+
+- When implementing the trait defined in `ops.rs` with each driver, the method declaration style should be the same for ease of searching: same ordering, same comments, same line breaks.

compose.yml

@@ -34,13 +34,13 @@ services:
       - minio
     entrypoint: >
       /bin/sh -c "while ! /usr/bin/mc ready minio; do
-        /usr/bin/mc alias set minio http://minio:9000 minioautumn minioautumn;
+        /usr/bin/mc config host add minio http://minio:9000 minioautumn minioautumn;
         echo 'Waiting minio...' && sleep 1;
       done; /usr/bin/mc mb minio/revolt-uploads; exit 0;"
 
   # Rabbit
   rabbit:
-    image: rabbitmq:4-management
+    image: rabbitmq:3-management
     environment:
       RABBITMQ_DEFAULT_USER: rabbituser
       RABBITMQ_DEFAULT_PASS: rabbitpass
@@ -64,10 +64,3 @@ services:
       MAILDEV_WEB_PORT: 8080
       MAILDEV_INCOMING_USER: smtp
       MAILDEV_INCOMING_PASS: smtp
-
-  livekit:
-    image: ghcr.io/stoatchat/livekit-server:v1.9.9
-    command: --config /etc/livekit.yml
-    network_mode: "host"
-    volumes:
-      - ./livekit.yml:/etc/livekit.yml

crates/bonfire/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-bonfire"
-version = "0.8.9"
+version = "0.8.8"
 license = "AGPL-3.0-or-later"
 edition = "2021"
 
@@ -9,6 +9,7 @@ edition = "2021"
 [dependencies]
 # util
 log = "*"
+rand = "*"
 sentry = "0.31.5"
 lru = "0.7.6"
 ulid = "0.5.0"
@@ -19,7 +20,6 @@ async-channel = "2.3.1"
 
 # parsing
 querystring = "1.1.0"
-regex = "1.11.1"
 
 # serde
 bincode = "1.3.3"
@@ -39,11 +39,15 @@ async-std = { version = "1.8.0", features = [
 # core
 authifier = { version = "1.0.15" }
 revolt-result = { path = "../core/result" }
+revolt-broker = { path = "../core/broker" }
 revolt-models = { path = "../core/models" }
 revolt-config = { path = "../core/config" }
-revolt-database = { path = "../core/database", features = ["voice"] }
-revolt-permissions = { version = "0.8.9", path = "../core/permissions" }
+revolt-database = { path = "../core/database" }
+revolt-permissions = { version = "0.8.8", path = "../core/permissions" }
 revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
 
 # redis
 fred = { version = "8.0.1", features = ["subscriber-client"] }
+
+# rabbit
+lapin = { version = "3.0.0" }

crates/bonfire/Dockerfile

@@ -1,5 +1,5 @@
 # Build Stage
-FROM ghcr.io/stoatchat/base:latest AS builder
+FROM ghcr.io/revoltchat/base:latest AS builder
 FROM debian:12 AS debian
 
 # Bundle Stage

crates/bonfire/src/client/core.rs

@@ -0,0 +1,129 @@
+use async_std::{net::TcpStream, sync::Mutex};
+use async_tungstenite::WebSocketStream;
+use futures::{join, SinkExt, StreamExt, TryStreamExt};
+use revolt_config::report_internal_error;
+use revolt_database::{
+    events::{client::EventV1, server::ClientMessage},
+    iso8601_timestamp::Timestamp,
+    Database, User, UserHint,
+};
+use revolt_presence::{create_session, delete_session};
+use revolt_result::create_error;
+
+use crate::{
+    client::{
+        subscriber::client_subscriber,
+        worker::{client_worker, WorkerRef},
+    },
+    config::ProtocolConfiguration,
+    events::state::State,
+};
+
+/// Core event loop of gateway clients
+pub async fn client_core(
+    db: &'static Database,
+    ws: WebSocketStream<TcpStream>,
+    mut config: ProtocolConfiguration,
+) {
+    // Split the socket for simultaneously read and write.
+    let (mut write, mut read) = ws.split();
+
+    // If the user has not provided authentication, request information.
+    if config.get_session_token().is_none() {
+        while let Ok(Some(message)) = read.try_next().await {
+            if let Ok(ClientMessage::Authenticate { token }) = config.decode(&message) {
+                config.set_session_token(token);
+                break;
+            }
+        }
+    }
+
+    // Try to authenticate the user.
+    let Some(token) = config.get_session_token().as_ref() else {
+        write
+            .send(config.encode(&EventV1::Error {
+                data: create_error!(InvalidSession),
+            }))
+            .await
+            .ok();
+        return;
+    };
+
+    let (user, session_id) = match User::from_token(db, token, UserHint::Any).await {
+        Ok(user) => user,
+        Err(err) => {
+            write
+                .send(config.encode(&EventV1::Error { data: err }))
+                .await
+                .ok();
+            return;
+        }
+    };
+
+    info!(
+        "Authenticated user {}#{}",
+        user.username, user.discriminator
+    );
+
+    db.update_session_last_seen(&session_id, Timestamp::now_utc())
+        .await
+        .ok();
+
+    // Create local state.
+    let mut state = State::from(user, session_id);
+    let user_id = state.cache.user_id.clone();
+
+    // Notify socket we have authenticated.
+    if report_internal_error!(write.send(config.encode(&EventV1::Authenticated)).await).is_err() {
+        return;
+    }
+
+    // Download required data to local cache and send Ready payload.
+    let ready_payload = match report_internal_error!(
+        state
+            .generate_ready_payload(db, config.get_ready_payload_fields())
+            .await
+    ) {
+        Ok(ready_payload) => ready_payload,
+        Err(_) => return,
+    };
+
+    if report_internal_error!(write.send(config.encode(&ready_payload)).await).is_err() {
+        return;
+    }
+
+    // Create presence session.
+    let (first_session, session_id) = create_session(&user_id, 0).await;
+
+    // If this was the first session, notify other users that we just went online.
+    if first_session {
+        state.broadcast_presence_change(true).await;
+    }
+
+    {
+        let worker_ref = WorkerRef::from(&state);
+        let write = Mutex::new(write);
+        let (reload, reloaded) = async_channel::bounded(1);
+        let (cancel_1, cancelled_1) = async_channel::bounded(1);
+        let (cancel_2, cancelled_2) = async_channel::bounded(1);
+
+        join!(
+            async {
+                client_subscriber(&write, cancelled_1, reloaded, &config, db, &mut state).await;
+                cancel_2.send(()).await.ok();
+            },
+            async {
+                client_worker(read, &write, cancelled_2, reload, &config, worker_ref).await;
+                cancel_1.send(()).await.ok();
+            }
+        );
+    }
+
+    // Clean up presence session.
+    let last_session = delete_session(&user_id, session_id).await;
+
+    // If this was the last session, notify other users that we just went offline.
+    if last_session {
+        state.broadcast_presence_change(false).await;
+    }
+}

crates/bonfire/src/client/mod.rs

@@ -0,0 +1,3 @@
+pub mod core;
+pub mod subscriber;
+pub mod worker;

crates/bonfire/src/client/subscriber.rs

@@ -0,0 +1,112 @@
+use async_channel::Receiver;
+use async_std::{net::TcpStream, sync::Mutex};
+use async_tungstenite::WebSocketStream;
+use authifier::AuthifierEvent;
+use futures::{pin_mut, select, stream::SplitSink, FutureExt, SinkExt};
+use revolt_broker::event_stream;
+use revolt_database::{events::client::EventV1, Database};
+use sentry::Level;
+
+use crate::{
+    config::ProtocolConfiguration,
+    events::state::{State, SubscriptionStateChange},
+};
+
+/// Event subscriber loop
+pub async fn client_subscriber(
+    write: &Mutex<SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>>,
+    cancelled: Receiver<()>,
+    reloaded: Receiver<()>,
+    protocol_config: &ProtocolConfiguration,
+    db: &'static Database,
+    state: &mut State,
+) {
+    let mut consumer = event_stream::Consumer::new().await;
+    consumer.set_topics(state.subscribed.read().await.clone());
+
+    let mut cancel = false;
+
+    loop {
+        // Reload consumer if subscriptions change
+        if !matches!(state.apply_state().await, SubscriptionStateChange::None) {
+            consumer.set_topics(state.subscribed.read().await.clone());
+        }
+
+        // Read incoming events
+        loop {
+            let reloaded = reloaded.recv().fuse();
+            let cancelled = cancelled.recv().fuse();
+            let delivery = consumer.next().fuse();
+            pin_mut!(delivery, reloaded, cancelled);
+
+            select! {
+                _ = reloaded => {
+                    break;
+                }
+                _ = cancelled => {
+                    cancel = true;
+                    break;
+                }
+                event = delivery => {
+                    if let Some(mut event) = event {
+                        // Handle the event
+                        if let EventV1::Auth(auth) = &event {
+                            if let AuthifierEvent::DeleteSession { session_id, .. } = auth {
+                                if &state.session_id == session_id {
+                                    event = EventV1::Logout;
+                                }
+                            } else if let AuthifierEvent::DeleteAllSessions {
+                                exclude_session_id, ..
+                            } = auth
+                            {
+                                if let Some(excluded) = exclude_session_id {
+                                    if &state.session_id != excluded {
+                                        event = EventV1::Logout;
+                                    }
+                                } else {
+                                    event = EventV1::Logout;
+                                }
+                            }
+                        } else {
+                            let should_send = state.handle_incoming_event_v1(db, &mut event).await;
+                            if !should_send {
+                                continue;
+                            }
+                        }
+
+                        let result = write.lock().await.send(protocol_config.encode(&event)).await;
+                        if let Err(e) = result {
+                            use async_tungstenite::tungstenite::Error;
+                            if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
+                                let err = format!("Error while sending an event: {e:?}");
+                                warn!("{}", err);
+                                sentry::capture_message(&err, Level::Warning);
+                            }
+
+                            cancel = true;
+                            break;
+                        }
+
+                        if let EventV1::Logout = event {
+                            info!("User {} received log out event!", state.user_id);
+                            cancel = true;
+                            break;
+                        }
+
+                        break;
+                    } else {
+                        cancel = true;
+                        break;
+                    }
+                }
+            }
+        }
+
+        // Break out if cancelled
+        if cancel {
+            break;
+        }
+    }
+
+    consumer.dispose_channel().await;
+}

crates/bonfire/src/client/worker.rs

@@ -0,0 +1,124 @@
+use std::{collections::HashSet, sync::Arc};
+
+use async_channel::{Receiver, Sender};
+use async_std::{
+    net::TcpStream,
+    sync::{Mutex, RwLock},
+};
+use async_tungstenite::WebSocketStream;
+use futures::{
+    pin_mut, select,
+    stream::{SplitSink, SplitStream},
+    FutureExt, SinkExt, TryStreamExt,
+};
+use revolt_database::events::{client::EventV1, server::ClientMessage};
+use sentry::Level;
+
+use crate::{config::ProtocolConfiguration, events::state::State};
+
+pub struct WorkerRef {
+    user_id: String,
+    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
+    subscribed: Arc<RwLock<HashSet<String>>>,
+}
+
+impl WorkerRef {
+    pub fn from(state: &State) -> WorkerRef {
+        WorkerRef {
+            user_id: state.user_id.clone(),
+            active_servers: state.active_servers.clone(),
+            subscribed: state.subscribed.clone(),
+        }
+    }
+}
+
+/// Incoming message handling
+pub async fn client_worker(
+    mut read: SplitStream<WebSocketStream<TcpStream>>,
+    write: &Mutex<SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>>,
+    cancelled: Receiver<()>,
+    reload: Sender<()>,
+    config: &ProtocolConfiguration,
+    state: WorkerRef,
+) {
+    loop {
+        let read = read.try_next().fuse();
+        let cancelled = cancelled.recv().fuse();
+        pin_mut!(read, cancelled);
+
+        select! {
+            _ = cancelled => { return; },
+            msg = read => {
+                let msg = match msg {
+                    Ok(Some(msg)) => msg,
+                    Ok(None) => {
+                        warn!("Received a None message!");
+                        return;
+                    }
+                    Err(e) => {
+                        use async_tungstenite::tungstenite::Error;
+                        if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed | Error::Protocol(_)) {
+                            let err = format!("Error while reading an event: {e:?}");
+                            warn!("{}", err);
+                            sentry::capture_message(&err, Level::Warning);
+                        }
+
+                        return;
+                    }
+                };
+
+                let Ok(payload) = config.decode(&msg) else {
+                    continue;
+                };
+
+                match payload {
+                    ClientMessage::BeginTyping { channel } => {
+                        if !state.subscribed.read().await.contains(&channel) {
+                            continue;
+                        }
+
+                        EventV1::ChannelStartTyping {
+                            id: channel.clone(),
+                            user: state.user_id.clone(),
+                        }
+                        .p(channel.clone())
+                        .await;
+                    }
+                    ClientMessage::EndTyping { channel } => {
+                        if !state.subscribed.read().await.contains(&channel) {
+                            continue;
+                        }
+
+                        EventV1::ChannelStopTyping {
+                            id: channel.clone(),
+                            user: state.user_id.clone(),
+                        }
+                        .p(channel.clone())
+                        .await;
+                    }
+                    ClientMessage::Subscribe { server_id } => {
+                        let mut servers = state.active_servers.lock().await;
+                        let has_item = servers.contains_key(&server_id);
+                        servers.insert(server_id, ());
+
+                        if !has_item {
+                            // Poke the listener to adjust subscriptions
+                            reload.send(()).await.ok();
+                        }
+                    }
+                    ClientMessage::Ping { data, responded } => {
+                        if responded.is_none() {
+                            write
+                                .lock()
+                                .await
+                                .send(config.encode(&EventV1::Pong { data }))
+                                .await
+                                .ok();
+                        }
+                    }
+                    _ => {}
+                }
+            }
+        }
+    }
+}

crates/bonfire/src/config.rs

@@ -1,15 +1,9 @@
 use async_tungstenite::tungstenite::{handshake, Message};
 use futures::channel::oneshot::Sender;
-use once_cell::sync::Lazy;
-use regex::Regex;
 use revolt_database::events::client::ReadyPayloadFields;
 use revolt_result::{create_error, Result};
 use serde::{Deserialize, Serialize};
 
-/// matches either a single word ie "users" or a key and value ie "settings[notifications]"
-static READY_PAYLOAD_FIELD_REGEX: Lazy<Regex> =
-    Lazy::new(|| Regex::new(r#"^(\w+)(?:\[(\S+)\])?$"#).unwrap());
-
 /// Enumeration of supported protocol formats
 #[derive(Debug)]
 pub enum ProtocolFormat {
@@ -23,7 +17,6 @@ pub struct ProtocolConfiguration {
     protocol_version: i32,
     format: ProtocolFormat,
     session_token: Option<String>,
-    ready_payload_fields: ReadyPayloadFields,
 }
 
 impl ProtocolConfiguration {
@@ -32,13 +25,11 @@ impl ProtocolConfiguration {
         protocol_version: i32,
         format: ProtocolFormat,
         session_token: Option<String>,
-        ready_payload_fields: ReadyPayloadFields,
     ) -> Self {
         Self {
             protocol_version,
             format,
             session_token,
-            ready_payload_fields,
         }
     }
 
@@ -95,8 +86,14 @@ impl ProtocolConfiguration {
     }
 
     /// Get ready payload fields
-    pub fn get_ready_payload_fields(&self) -> &ReadyPayloadFields {
-        &self.ready_payload_fields
+    pub fn get_ready_payload_fields(&self) -> Vec<ReadyPayloadFields> {
+        vec![
+            ReadyPayloadFields::Users,
+            ReadyPayloadFields::Servers,
+            ReadyPayloadFields::Channels,
+            ReadyPayloadFields::Members,
+            ReadyPayloadFields::Emoji,
+        ]
     }
 }
 
@@ -127,23 +124,6 @@ impl handshake::server::Callback for WebsocketHandshakeCallback {
         let mut protocol_version = 1;
         let mut format = ProtocolFormat::Json;
         let mut session_token = None;
-        let mut ready_payload_fields = if params.iter().any(|(k, _)| *k == "ready") {
-            // If they pass the ready field, set all fields to false
-
-            ReadyPayloadFields {
-                users: false,
-                servers: false,
-                channels: false,
-                members: false,
-                emojis: false,
-                voice_states: false,
-                user_settings: Vec::new(),
-                channel_unreads: false,
-                policy_changes: false,
-            }
-        } else {
-            ReadyPayloadFields::default()
-        };
 
         // Parse and map parameters from key-value to known variables.
         for (key, value) in params {
@@ -159,31 +139,6 @@ impl handshake::server::Callback for WebsocketHandshakeCallback {
                     _ => {}
                 },
                 "token" => session_token = Some(value.into()),
-                "ready" => {
-                    // Re-enable all the fields the client specifies
-                    if let Some(captures) = READY_PAYLOAD_FIELD_REGEX.captures(value) {
-                        if let Some(field) = captures.get(0) {
-                            match field.as_str() {
-                                "users" => ready_payload_fields.users = true,
-                                "servers" => ready_payload_fields.servers = true,
-                                "channels" => ready_payload_fields.channels = true,
-                                "members" => ready_payload_fields.members = true,
-                                "emojis" => ready_payload_fields.emojis = true,
-                                "voice_states" => ready_payload_fields.voice_states = true,
-                                "channel_unreads" => ready_payload_fields.channel_unreads = true,
-                                "user_settings" => {
-                                    if let Some(subkey) = captures.get(1) {
-                                        ready_payload_fields
-                                            .user_settings
-                                            .push(subkey.as_str().to_string());
-                                    }
-                                }
-                                "policy_changes" => ready_payload_fields.policy_changes = true,
-                                _ => {}
-                            }
-                        }
-                    }
-                }
                 _ => {}
             }
         }
@@ -196,7 +151,6 @@ impl handshake::server::Callback for WebsocketHandshakeCallback {
                 protocol_version,
                 format,
                 session_token,
-                ready_payload_fields,
             })
             .is_ok()
         {

crates/bonfire/src/events/impl.rs

@@ -1,10 +1,9 @@
-use std::collections::{HashMap, HashSet};
+use std::collections::HashSet;
 
 use futures::future::join_all;
 use revolt_database::{
     events::client::{EventV1, ReadyPayloadFields},
     util::permissions::DatabasePermissionQuery,
-    voice::get_channel_voice_state,
     Channel, Database, Member, MemberCompositeKey, Presence, RelationshipStatus,
 };
 use revolt_models::v0;
@@ -18,9 +17,8 @@ use super::state::{Cache, State};
 impl Cache {
     /// Check whether the current user can view a channel
     pub async fn can_view_channel(&self, db: &Database, channel: &Channel) -> bool {
-        #[allow(deprecated)]
         match &channel {
-            Channel::TextChannel { server, .. } => {
+            Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => {
                 let member = self.members.get(server);
                 let server = self.servers.get(server);
                 let mut query =
@@ -97,23 +95,21 @@ impl State {
     pub async fn generate_ready_payload(
         &mut self,
         db: &Database,
-        fields: &ReadyPayloadFields,
+        fields: Vec<ReadyPayloadFields>,
     ) -> Result<EventV1> {
         let user = self.clone_user();
         self.cache.is_bot = user.bot.is_some();
 
         // Fetch pending policy changes.
-        let policy_changes = if user.bot.is_some() || !fields.policy_changes {
-            None
+        let policy_changes = if user.bot.is_some() {
+            vec![]
         } else {
-            Some(
-                db.fetch_policy_changes()
-                    .await?
-                    .into_iter()
-                    .filter(|policy| policy.created_time > user.last_acknowledged_policy_change)
-                    .map(Into::into)
-                    .collect(),
-            )
+            db.fetch_policy_changes()
+                .await?
+                .into_iter()
+                .filter(|policy| policy.created_time > user.last_acknowledged_policy_change)
+                .map(Into::into)
+                .collect()
         };
 
         // Find all relationships to the user.
@@ -124,7 +120,12 @@ impl State {
             .unwrap_or_default();
 
         // Fetch all memberships with their corresponding servers.
-        let mut members: Vec<Member> = db.fetch_all_memberships(&user.id).await?;
+        let members: Vec<Member> = db.fetch_all_memberships(&user.id).await?;
+        self.cache.members = members
+            .iter()
+            .cloned()
+            .map(|x| (x.id.server.clone(), x))
+            .collect();
 
         let server_ids: Vec<String> = members.iter().map(|x| x.id.server.clone()).collect();
         let servers = db.fetch_servers(&server_ids).await?;
@@ -153,51 +154,6 @@ impl State {
             }
         }
 
-        let voice_states = if fields.voice_states {
-            let mut voice_state_server_members: HashMap<String, HashSet<String>> = HashMap::new();
-
-            // fetch voice states for all the channels we can see
-            let mut voice_states = Vec::new();
-
-            for channel in channels.iter().filter(|c| {
-                matches!(
-                    c,
-                    Channel::DirectMessage { .. }
-                        | Channel::Group { .. }
-                        | Channel::TextChannel { voice: Some(_), .. }
-                )
-            }) {
-                if let Ok(Some(voice_state)) = get_channel_voice_state(channel).await {
-                    if let Some(server) = channel.server() {
-                        let set = voice_state_server_members.entry(server.to_string()).or_default();
-
-                        for participant in &voice_state.participants {
-                            user_ids.insert(participant.id.clone());
-                            set.insert(participant.id.clone());
-                        }
-                    } else {
-                        for participant in &voice_state.participants {
-                            user_ids.insert(participant.id.clone());
-                        }
-                    }
-
-                    voice_states.push(voice_state);
-                }
-            }
-
-            // Fetch all the members for for the participants who are in a server
-            for (server, user_ids) in voice_state_server_members {
-                let user_ids = user_ids.into_iter().collect::<Vec<_>>();
-                let voice_members = db.fetch_members(&server, &user_ids).await?;
-
-                members.extend(voice_members);
-            }
-
-            Some(voice_states)
-        } else {
-            None
-        };
-
         // Fetch presence data for known users.
         let online_ids = filter_online(&user_ids.iter().cloned().collect::<Vec<String>>()).await;
 
@@ -211,14 +167,8 @@ impl State {
             )
             .await?;
 
-        self.cache.members = members
-            .iter()
-            .cloned()
-            .map(|x| (x.id.server.clone(), x))
-            .collect();
-
         // Fetch customisations.
-        let emojis = if fields.emojis {
+        let emojis = if fields.contains(&ReadyPayloadFields::Emoji) {
             Some(
                 db.fetch_emoji_by_parent_ids(
                     &servers
@@ -226,34 +176,25 @@ impl State {
                         .map(|x| x.id.to_string())
                         .collect::<Vec<String>>(),
                 )
-                .await?
-                .into_iter()
-                .map(|emoji| emoji.into())
-                .collect(),
+                .await?,
             )
         } else {
             None
         };
 
         // Fetch user settings
-        let user_settings = if !fields.user_settings.is_empty() {
-            Some(
-                db.fetch_user_settings(&user.id, &fields.user_settings)
-                    .await?,
-            )
+        let user_settings = if let Some(ReadyPayloadFields::UserSettings(keys)) = fields
+            .iter()
+            .find(|e| matches!(e, ReadyPayloadFields::UserSettings(_)))
+        {
+            Some(db.fetch_user_settings(&user.id, keys).await?)
         } else {
             None
         };
 
         // Fetch channel unreads
-        let channel_unreads = if fields.channel_unreads {
-            Some(
-                db.fetch_unreads(&user.id)
-                    .await?
-                    .into_iter()
-                    .map(|unread| unread.into())
-                    .collect(),
-            )
+        let channel_unreads = if fields.contains(&ReadyPayloadFields::ChannelUnreads) {
+            Some(db.fetch_unreads(&user.id).await?)
         } else {
             None
         };
@@ -300,27 +241,30 @@ impl State {
         }
 
         Ok(EventV1::Ready {
-            users: if fields.users { Some(users) } else { None },
-            servers: if fields.servers {
+            users: if fields.contains(&ReadyPayloadFields::Users) {
+                Some(users)
+            } else {
+                None
+            },
+            servers: if fields.contains(&ReadyPayloadFields::Servers) {
                 Some(servers.into_iter().map(Into::into).collect())
             } else {
                 None
             },
-            channels: if fields.channels {
+            channels: if fields.contains(&ReadyPayloadFields::Channels) {
                 Some(channels.into_iter().map(Into::into).collect())
             } else {
                 None
             },
-            members: if fields.members {
+            members: if fields.contains(&ReadyPayloadFields::Members) {
                 Some(members.into_iter().map(Into::into).collect())
             } else {
                 None
             },
-            voice_states,
+            emojis: emojis.map(|vec| vec.into_iter().map(Into::into).collect()),
 
-            emojis,
             user_settings,
-            channel_unreads,
+            channel_unreads: channel_unreads.map(|vec| vec.into_iter().map(Into::into).collect()),
 
             policy_changes,
         })
@@ -335,14 +279,19 @@ impl State {
 
             let id = &id.to_string();
             for (channel_id, channel) in &self.cache.channels {
-                if channel.server() == Some(id) {
-                    channel_ids.insert(channel_id.clone());
+                match channel {
+                    Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => {
+                        if server == id {
+                            channel_ids.insert(channel_id.clone());
 
-                    if self.cache.can_view_channel(db, channel).await {
-                        added_channels.push(channel_id.clone());
-                    } else {
-                        removed_channels.push(channel_id.clone());
+                            if self.cache.can_view_channel(db, channel).await {
+                                added_channels.push(channel_id.clone());
+                            } else {
+                                removed_channels.push(channel_id.clone());
+                            }
+                        }
                     }
+                    _ => {}
                 }
             }
 
@@ -510,7 +459,6 @@ impl State {
                 server,
                 channels,
                 emojis: _,
-                voice_states: _,
             } => {
                 self.insert_subscription(id.clone()).await;
 

crates/bonfire/src/events/state.rs

@@ -66,6 +66,7 @@ impl Default for Cache {
 pub struct State {
     pub cache: Cache,
 
+    pub user_id: String,
     pub session_id: String,
     pub private_topic: String,
     pub state: SubscriptionStateChange,
@@ -87,6 +88,7 @@ impl State {
             ..Default::default()
         };
 
+        let user_id = user.id.clone();
         cache.users.insert(user.id.clone(), user);
 
         State {
@@ -96,6 +98,7 @@ impl State {
                 Duration::from_secs(900),
                 5,
             ))),
+            user_id,
             session_id,
             private_topic,
             state: SubscriptionStateChange::Reset,

crates/bonfire/src/main.rs

@@ -6,10 +6,11 @@ use revolt_presence::clear_region;
 #[macro_use]
 extern crate log;
 
-pub mod config;
-pub mod events;
+pub mod client;
 
+mod config;
 mod database;
+mod events;
 mod websocket;
 
 #[async_std::main]

crates/bonfire/src/websocket.rs

@@ -1,40 +1,11 @@
-use std::{collections::HashSet, net::SocketAddr, sync::Arc};
+use std::net::SocketAddr;
 
-use async_tungstenite::WebSocketStream;
-use authifier::AuthifierEvent;
-use fred::{
-    error::RedisErrorKind,
-    interfaces::{ClientLike, EventInterface, PubsubInterface},
-    types::RedisConfig,
-};
-use futures::{
-    channel::oneshot,
-    join, pin_mut, select,
-    stream::{SplitSink, SplitStream},
-    FutureExt, SinkExt, StreamExt, TryStreamExt,
-};
-use redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
-use revolt_config::report_internal_error;
-use revolt_database::{
-    events::{client::EventV1, server::ClientMessage},
-    iso8601_timestamp::Timestamp,
-    Database, User, UserHint,
-};
-use revolt_presence::{create_session, delete_session};
+use async_std::net::TcpStream;
+use futures::channel::oneshot;
+use revolt_database::Database;
 
-use async_std::{
-    net::TcpStream,
-    sync::{Mutex, RwLock},
-    task::spawn,
-};
-use revolt_result::create_error;
-use sentry::Level;
-
-use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
-use crate::events::state::{State, SubscriptionStateChange};
-
-type WsReader = SplitStream<WebSocketStream<TcpStream>>;
-type WsWriter = SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>;
+use crate::client::core::client_core;
+use crate::config::WebsocketHandshakeCallback;
 
 /// Start a new WebSocket client worker given access to the database,
 /// the relevant TCP stream and the remote address of the client.
@@ -54,7 +25,7 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
     };
 
     // Verify we've received a valid config, otherwise we should just drop the connection.
-    let Ok(mut config) = receiver.await else {
+    let Ok(config) = receiver.await else {
         return;
     };
 
@@ -64,446 +35,5 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
         config.get_protocol_format()
     );
 
-    // Split the socket for simultaneously read and write.
-    let (mut write, mut read) = ws.split();
-
-    // If the user has not provided authentication, request information.
-    if config.get_session_token().is_none() {
-        while let Ok(Some(message)) = read.try_next().await {
-            if let Ok(ClientMessage::Authenticate { token }) = config.decode(&message) {
-                config.set_session_token(token);
-                break;
-            }
-        }
-    }
-
-    // Try to authenticate the user.
-    let Some(token) = config.get_session_token().as_ref() else {
-        write
-            .send(config.encode(&EventV1::Error {
-                data: create_error!(InvalidSession),
-            }))
-            .await
-            .ok();
-        return;
-    };
-
-    let (user, session_id) = match User::from_token(db, token, UserHint::Any).await {
-        Ok(user) => user,
-        Err(err) => {
-            write
-                .send(config.encode(&EventV1::Error { data: err }))
-                .await
-                .ok();
-            return;
-        }
-    };
-
-    info!("User {addr:?} authenticated as @{}", user.username);
-
-    db.update_session_last_seen(&session_id, Timestamp::now_utc())
-        .await
-        .ok();
-
-    // Create local state.
-    let mut state = State::from(user, session_id);
-    let user_id = state.cache.user_id.clone();
-
-    // Notify socket we have authenticated.
-    if report_internal_error!(write.send(config.encode(&EventV1::Authenticated)).await).is_err() {
-        return;
-    }
-
-    // Download required data to local cache and send Ready payload.
-    let ready_payload = match report_internal_error!(
-        state
-            .generate_ready_payload(db, config.get_ready_payload_fields())
-            .await
-    ) {
-        Ok(ready_payload) => ready_payload,
-        Err(_) => return,
-    };
-
-    if report_internal_error!(write.send(config.encode(&ready_payload)).await).is_err() {
-        return;
-    }
-
-    // Create presence session.
-    let (first_session, session_id) = create_session(&user_id, 0).await;
-
-    // If this was the first session, notify other users that we just went online.
-    if first_session {
-        state.broadcast_presence_change(true).await;
-    }
-
-    {
-        // Setup channels and mutexes
-        let write = Mutex::new(write);
-        let subscribed = state.subscribed.clone();
-        let active_servers = state.active_servers.clone();
-        let (topic_signal_s, topic_signal_r) = async_channel::unbounded();
-
-        // TODO: this needs to be rewritten
-        // Create channels through which the tasks can signal to each other they need to clean up
-        let (kill_signal_1_s, kill_signal_1_r) = async_channel::bounded(1);
-        let (kill_signal_2_s, kill_signal_2_r) = async_channel::bounded(1);
-
-        // Create a PubSub connection to poll on.
-        let listener = listener_with_kill_signal(
-            db,
-            &mut state,
-            addr,
-            &config,
-            topic_signal_r,
-            kill_signal_1_r,
-            &write,
-            kill_signal_2_s,
-        );
-
-        // Read from WebSocket stream.
-        let worker = worker_with_kill_signal(
-            addr,
-            subscribed,
-            active_servers,
-            user_id.clone(),
-            &config,
-            topic_signal_s,
-            kill_signal_2_r,
-            read,
-            &write,
-            kill_signal_1_s,
-        );
-
-        join!(listener, worker);
-    }
-    // Clean up presence session.
-    let last_session = delete_session(&user_id, session_id).await;
-
-    // If this was the last session, notify other users that we just went offline.
-    if last_session {
-        state.broadcast_presence_change(false).await;
-    }
-}
-
-#[allow(clippy::too_many_arguments)]
-async fn listener_with_kill_signal(
-    db: &'static Database,
-    state: &mut State,
-    addr: SocketAddr,
-    config: &ProtocolConfiguration,
-    topic_signal_r: async_channel::Receiver<()>,
-    kill_signal_r: async_channel::Receiver<()>,
-    write: &Mutex<WsWriter>,
-    kill_signal_s: async_channel::Sender<()>,
-) {
-    listener(
-        db,
-        state,
-        addr,
-        config,
-        topic_signal_r,
-        kill_signal_r,
-        write,
-    )
-    .await;
-    kill_signal_s.send(()).await.ok();
-}
-
-async fn listener(
-    db: &'static Database,
-    state: &mut State,
-    addr: SocketAddr,
-    config: &ProtocolConfiguration,
-    topic_signal_r: async_channel::Receiver<()>,
-    kill_signal_r: async_channel::Receiver<()>,
-    write: &Mutex<WsWriter>,
-) {
-    let redis_config = RedisConfig::from_url(&REDIS_URI).unwrap();
-    let subscriber = match report_internal_error!(
-        fred::types::Builder::from_config(redis_config).build_subscriber_client()
-    ) {
-        Ok(subscriber) => subscriber,
-        Err(_) => return,
-    };
-
-    if report_internal_error!(subscriber.init().await).is_err() {
-        return;
-    }
-
-    // Handle Redis connection dropping
-    let (clean_up_s, clean_up_r) = async_channel::bounded(1);
-    let clean_up_s = Arc::new(Mutex::new(clean_up_s));
-    subscriber.on_error(move |err| {
-        if let RedisErrorKind::Canceled = err.kind() {
-            let clean_up_s = clean_up_s.clone();
-            spawn(async move {
-                clean_up_s.lock().await.send(()).await.ok();
-            });
-        }
-
-        Ok(())
-    });
-
-    let mut message_rx = subscriber.message_rx();
-    'out: loop {
-        // Check for state changes for subscriptions.
-        match state.apply_state().await {
-            SubscriptionStateChange::Reset => {
-                if report_internal_error!(subscriber.unsubscribe_all().await).is_err() {
-                    break 'out;
-                }
-
-                let subscribed = state.subscribed.read().await;
-                for id in subscribed.iter() {
-                    if report_internal_error!(subscriber.subscribe(id).await).is_err() {
-                        break 'out;
-                    }
-                }
-
-                #[cfg(debug_assertions)]
-                info!("{addr:?} has reset their subscriptions");
-            }
-            SubscriptionStateChange::Change { add, remove } => {
-                for id in remove {
-                    #[cfg(debug_assertions)]
-                    info!("{addr:?} unsubscribing from {id}");
-
-                    if report_internal_error!(subscriber.unsubscribe(id).await).is_err() {
-                        break 'out;
-                    }
-                }
-
-                for id in add {
-                    #[cfg(debug_assertions)]
-                    info!("{addr:?} subscribing to {id}");
-
-                    if report_internal_error!(subscriber.subscribe(id).await).is_err() {
-                        break 'out;
-                    }
-                }
-            }
-            SubscriptionStateChange::None => {}
-        }
-
-        let t1 = message_rx.recv().fuse();
-        let t2 = topic_signal_r.recv().fuse();
-        let t3 = kill_signal_r.recv().fuse();
-        let t4 = clean_up_r.recv().fuse();
-
-        pin_mut!(t1, t2, t3, t4);
-
-        select! {
-            _ = t4 => {
-                break 'out;
-            },
-            _ = t3 => {
-                break 'out;
-            },
-            _ = t2 => {},
-            message = t1 => {
-                // Handle incoming events.
-                let message = match report_internal_error!(message) {
-                    Ok(message) => message,
-                    Err(_) => break 'out
-                };
-
-                let event = match *REDIS_PAYLOAD_TYPE {
-                    PayloadType::Json => message
-                        .value
-                        .as_str()
-                        .and_then(|s| report_internal_error!(serde_json::from_str::<EventV1>(s.as_ref())).ok()),
-                    PayloadType::Msgpack => message
-                        .value
-                        .as_bytes()
-                        .and_then(|b| report_internal_error!(rmp_serde::from_slice::<EventV1>(b)).ok()),
-                    PayloadType::Bincode => message
-                        .value
-                        .as_bytes()
-                        .and_then(|b| report_internal_error!(bincode::deserialize::<EventV1>(b)).ok()),
-                };
-
-                let Some(mut event) = event else {
-                    let err = format!(
-                        "Failed to deserialise event for {}: `{:?}`",
-                        message.channel,
-                        message
-                            .value
-                    );
-
-                    error!("{}", err);
-                    sentry::capture_message(&err, Level::Error);
-                    break 'out;
-                };
-
-                if let EventV1::Auth(auth) = &event {
-                    if let AuthifierEvent::DeleteSession { session_id, .. } = auth {
-                        if &state.session_id == session_id {
-                            event = EventV1::Logout;
-                        }
-                    } else if let AuthifierEvent::DeleteAllSessions {
-                        exclude_session_id, ..
-                    } = auth
-                    {
-                        if let Some(excluded) = exclude_session_id {
-                            if &state.session_id != excluded {
-                                event = EventV1::Logout;
-                            }
-                        } else {
-                            event = EventV1::Logout;
-                        }
-                    }
-                } else {
-                    let should_send = state.handle_incoming_event_v1(db, &mut event).await;
-                    if !should_send {
-                        continue;
-                    }
-                }
-
-                let result = write.lock().await.send(config.encode(&event)).await;
-                if let Err(e) = result {
-                    use async_tungstenite::tungstenite::Error;
-                    if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
-                        let err = format!("Error while sending an event to {addr:?}: {e:?}");
-                        warn!("{}", err);
-                        sentry::capture_message(&err, Level::Warning);
-                    }
-
-                    break 'out;
-                }
-
-                if let EventV1::Logout = event {
-                    info!("User {addr:?} received log out event!");
-                    break 'out;
-                }
-            }
-        }
-    }
-
-    report_internal_error!(subscriber.quit().await).ok();
-}
-
-#[allow(clippy::too_many_arguments)]
-async fn worker_with_kill_signal(
-    addr: SocketAddr,
-    subscribed: Arc<RwLock<HashSet<String>>>,
-    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
-    user_id: String,
-    config: &ProtocolConfiguration,
-    topic_signal_s: async_channel::Sender<()>,
-    kill_signal_r: async_channel::Receiver<()>,
-    read: WsReader,
-    write: &Mutex<WsWriter>,
-    kill_signal_s: async_channel::Sender<()>,
-) {
-    worker(
-        addr,
-        subscribed,
-        active_servers,
-        user_id,
-        config,
-        topic_signal_s,
-        kill_signal_r,
-        read,
-        write,
-    )
-    .await;
-    kill_signal_s.send(()).await.ok();
-}
-
-#[allow(clippy::too_many_arguments)]
-async fn worker(
-    addr: SocketAddr,
-    subscribed: Arc<RwLock<HashSet<String>>>,
-    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
-    user_id: String,
-    config: &ProtocolConfiguration,
-    topic_signal_s: async_channel::Sender<()>,
-    kill_signal_r: async_channel::Receiver<()>,
-    mut read: WsReader,
-    write: &Mutex<WsWriter>,
-) {
-    loop {
-        let t1 = read.try_next().fuse();
-        let t2 = kill_signal_r.recv().fuse();
-
-        pin_mut!(t1, t2);
-
-        select! {
-            _ = t2 => {
-                return;
-            },
-            result = t1 => {
-                let msg = match result {
-                    Ok(Some(msg)) => msg,
-                    Ok(None) => {
-                        warn!("Received a None message!");
-                        sentry::capture_message("Received a None message!", Level::Warning);
-                        return;
-                    }
-                    Err(e) => {
-                        use async_tungstenite::tungstenite::Error;
-                        if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
-                            let err = format!("Error while reading an event from {addr:?}: {e:?}");
-                            warn!("{}", err);
-                            sentry::capture_message(&err, Level::Warning);
-                        }
-
-                        return;
-                    }
-                };
-
-                let Ok(payload) = config.decode(&msg) else {
-                    continue;
-                };
-
-                match payload {
-                    ClientMessage::BeginTyping { channel } => {
-                        if !subscribed.read().await.contains(&channel) {
-                            continue;
-                        }
-
-                        EventV1::ChannelStartTyping {
-                            id: channel.clone(),
-                            user: user_id.clone(),
-                        }
-                        .p(channel.clone())
-                        .await;
-                    }
-                    ClientMessage::EndTyping { channel } => {
-                        if !subscribed.read().await.contains(&channel) {
-                            continue;
-                        }
-
-                        EventV1::ChannelStopTyping {
-                            id: channel.clone(),
-                            user: user_id.clone(),
-                        }
-                        .p(channel.clone())
-                        .await;
-                    }
-                    ClientMessage::Subscribe { server_id } => {
-                        let mut servers = active_servers.lock().await;
-                        let has_item = servers.contains_key(&server_id);
-                        servers.insert(server_id, ());
-
-                        if !has_item {
-                            // Poke the listener to adjust subscriptions
-                            topic_signal_s.send(()).await.ok();
-                        }
-                    }
-                    ClientMessage::Ping { data, responded } => {
-                        if responded.is_none() {
-                            write
-                                .lock()
-                                .await
-                                .send(config.encode(&EventV1::Pong { data }))
-                                .await
-                                .ok();
-                        }
-                    }
-                    _ => {}
-                }
-            }
-        }
-    }
+    client_core(db, ws, config).await;
 }

crates/core/broker/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "revolt-broker"
+version = "0.8.8"
+edition = "2024"
+license = "AGPL-3.0-or-later"
+authors = ["Paul Makles <me@insrt.uk>"]
+description = "Revolt Backend: Event Broker"
+
+[dependencies]
+# Utility
+log = "0.4"
+rand = "0.9.1"
+
+# RabbitMQ/AMQP client
+lapin = "3.0.0"
+
+# Async runtime
+async-std = { version = "1.8.0" }
+
+# Serialisation
+serde = "1"
+rmp-serde = "1.3.0"
+
+# Core
+revolt-config = { version = "0.8.8", path = "../config" }

crates/core/broker/src/event_stream/consumer.rs

@@ -0,0 +1,180 @@
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
+
+use async_std::stream::StreamExt;
+use lapin::{
+    Channel, Connection,
+    options::BasicAckOptions,
+    types::{AMQPValue, FieldArray, FieldTable, LongLongInt},
+};
+use log::info;
+use rand::Rng;
+use revolt_config::{capture_internal_error, config};
+use serde::de::DeserializeOwned;
+
+use crate::event_stream::{create_channel, get_connection};
+
+pub struct Consumer {
+    #[allow(dead_code)]
+    conn: Arc<Connection>,
+    channel: Channel,
+    tag: String,
+    topics: HashSet<String>,
+    topics_changed: bool,
+    consumer: Option<lapin::Consumer>,
+    offset: Option<LongLongInt>,
+}
+
+impl Consumer {
+    /// Create a new event stream consumer
+    pub async fn new() -> Consumer {
+        let config = config().await;
+        let conn = get_connection().await;
+        let channel = create_channel(&conn, config.rabbit.event_stream).await;
+
+        Consumer {
+            conn,
+            channel,
+            tag: rand::rng()
+                .sample_iter::<char, _>(&rand::distr::StandardUniform)
+                .take(32)
+                .collect(),
+            topics: HashSet::new(),
+            topics_changed: false,
+            consumer: None,
+            offset: None,
+        }
+    }
+
+    /// Update the set of topics
+    pub fn set_topics(&mut self, topics: HashSet<String>) {
+        self.topics = topics;
+        self.topics_changed = true;
+    }
+
+    /// Get the current consumer
+    pub async fn ensure_consumer(&mut self) {
+        if self.topics_changed {
+            info!("Topics changed, disposing the consumer.");
+            self.dispose_consumer().await;
+            self.topics_changed = false;
+        }
+
+        if self.consumer.is_none() {
+            info!("Creating a new consumer, tag={}", self.tag);
+            let config = config().await;
+
+            // Build arguments for consumer
+            let mut args: FieldTable = Default::default();
+
+            // Configure stream filter to select topics we are listening for
+            {
+                let mut filter: FieldArray = Default::default();
+                for topic in &self.topics {
+                    filter.push(AMQPValue::LongString(topic.as_str().into()));
+                }
+
+                args.insert("x-stream-filter".into(), AMQPValue::FieldArray(filter));
+            }
+
+            // Set stream offset if applicable
+            if let Some(offset) = self.offset {
+                args.insert("x-stream-offset".into(), AMQPValue::LongLongInt(offset));
+            }
+
+            // Create the consumer
+            self.consumer = Some(
+                self.channel
+                    .basic_consume(
+                        &config.rabbit.event_stream.queue,
+                        &self.tag,
+                        Default::default(),
+                        args,
+                    )
+                    .await
+                    .unwrap(),
+            );
+        }
+    }
+
+    /// Close the active consumer if one exists
+    pub async fn dispose_consumer(&mut self) {
+        if let Some(consumer) = self.consumer.as_ref() {
+            if consumer.state().is_active() {
+                if let Err(err) = self
+                    .channel
+                    .basic_cancel(&self.tag, Default::default())
+                    .await
+                {
+                    eprintln!("Failed to close consumer! {:?}", err);
+                }
+
+                // is this necessary?
+                // else {
+                // Read the consumer to the end
+                //     while let Some(delivery) = consumer.next().await {
+                //         let delivery = delivery.expect("error in consumer");
+                //         delivery.ack(BasicAckOptions::default()).await.expect("ack");
+                //     }
+                // }
+            }
+
+            self.consumer = None;
+        }
+    }
+
+    /// Close the active channel
+    pub async fn dispose_channel(&mut self) {
+        // Close the channel -- don't do this actually
+        capture_internal_error!(self.channel.close(0, "closing channel").await);
+    }
+
+    /// Get the next item
+    pub async fn next<T: DeserializeOwned>(&mut self) -> Option<T> {
+        self.ensure_consumer().await;
+
+        let consumer = self.consumer.as_mut().unwrap();
+
+        while let Some(Ok(delivery)) = consumer.next().await {
+            // Acknowledgement is required
+            delivery.ack(BasicAckOptions::default()).await.expect("ack");
+
+            // Parse the delivery headers
+            let headers: HashMap<String, AMQPValue> = delivery
+                .properties
+                .headers()
+                .as_ref()
+                .map(|table| {
+                    table
+                        .into_iter()
+                        .map(|(k, v)| (k.to_string(), v.clone()))
+                        .collect()
+                })
+                .unwrap_or_default();
+
+            // Keep track of the current offset
+            let stream_offset = headers
+                .get("x-stream-offset")
+                .expect("`x-stream-offset` not present in message!");
+
+            self.offset = Some(stream_offset.as_long_long_int().unwrap() + 1);
+
+            // Client-side topic filtering (broker uses Bloom filter so may have false-positives)
+            let filter_value = headers
+                .get("x-stream-filter-value")
+                .expect("`x-stream-filter-value` not present in message!")
+                .as_long_string()
+                .expect("`string`")
+                .to_string();
+
+            if self.topics.contains(&filter_value) {
+                // Deserialise the data
+                return Some(rmp_serde::from_slice(&delivery.data).expect("`data`"));
+            }
+        }
+
+        None
+    }
+}

crates/core/broker/src/event_stream/mod.rs

@@ -0,0 +1,7 @@
+mod consumer;
+mod pool;
+mod publish;
+
+pub use consumer::Consumer;
+pub use pool::{create_channel, get_connection};
+pub use publish::publish_event;

crates/core/broker/src/event_stream/pool.rs

@@ -0,0 +1,99 @@
+use async_std::sync::Mutex;
+use lapin::{
+    Connection,
+    options::QueueDeclareOptions,
+    types::{AMQPValue, FieldTable},
+};
+use log::{debug, warn};
+use revolt_config::{RabbitEventStream, config};
+use std::sync::Arc;
+
+use crate::create_client;
+
+/// Get a handle to the event stream
+pub async fn get_connection() -> Arc<Connection> {
+    let config = config().await;
+
+    static CONNECTIONS: Mutex<Vec<Arc<Connection>>> = Mutex::new(Vec::new());
+
+    let mut connections = CONNECTIONS.lock().await;
+    connections.retain(|item| {
+        if item.status().connected() {
+            true
+        } else {
+            warn!(
+                "Dropping connection with status {:?}",
+                item.status().state()
+            );
+
+            false
+        }
+    });
+
+    debug!(
+        "Connections: {}, Clients: {:?}",
+        connections.len(),
+        connections
+            .iter()
+            .map(Arc::strong_count)
+            .collect::<Vec<usize>>()
+    );
+
+    for conn in connections.iter() {
+        if Arc::strong_count(conn) < config.rabbit.event_stream.channels_per_conn {
+            return conn.clone();
+        }
+    }
+
+    let conn = Arc::new(create_client().await);
+
+    connections.push(conn.clone());
+    conn
+}
+
+/// Create a channel
+pub async fn create_channel(
+    conn: &lapin::Connection,
+    event_stream: RabbitEventStream,
+) -> lapin::Channel {
+    let channel = conn.create_channel().await.unwrap();
+
+    let mut args: FieldTable = Default::default();
+
+    args.insert(
+        // set queue type to stream
+        "x-queue-type".into(),
+        AMQPValue::LongString("stream".into()),
+    );
+
+    args.insert(
+        // max. size of the stream
+        "x-max-length-bytes".into(),
+        AMQPValue::LongLongInt(event_stream.stream_max_length_bytes),
+    );
+
+    args.insert(
+        // size of the Bloom filter
+        "x-stream-filter-size-bytes".into(),
+        AMQPValue::LongLongInt(event_stream.filter_size_bytes),
+    );
+
+    channel
+        .queue_declare(
+            &event_stream.queue,
+            QueueDeclareOptions {
+                durable: true,
+                ..Default::default()
+            },
+            args,
+        )
+        .await
+        .unwrap();
+
+    channel
+        .basic_qos(event_stream.qos_prefetch, Default::default())
+        .await
+        .unwrap();
+
+    channel
+}

crates/core/broker/src/event_stream/publish.rs

@@ -0,0 +1,37 @@
+use lapin::{
+    Error,
+    protocol::basic::AMQPProperties,
+    publisher_confirm::PublisherConfirm,
+    types::{AMQPValue, FieldTable},
+};
+use revolt_config::config;
+use serde::Serialize;
+
+use crate::event_stream::{create_channel, get_connection};
+
+/// Publish an event to the message broker
+pub async fn publish_event<T: Serialize>(
+    channel: &str,
+    data: &T,
+) -> Result<PublisherConfirm, Error> {
+    let config = config().await;
+
+    let mut headers: FieldTable = Default::default();
+    headers.insert(
+        "x-stream-filter-value".into(),
+        AMQPValue::LongString(channel.into()),
+    );
+
+    let conn = get_connection().await;
+
+    create_channel(&conn, config.rabbit.event_stream.clone())
+        .await
+        .basic_publish(
+            &config.rabbit.event_stream.exchange,
+            &config.rabbit.event_stream.queue,
+            Default::default(),
+            &rmp_serde::to_vec_named(data).unwrap(),
+            AMQPProperties::default().with_headers(headers),
+        )
+        .await
+}

crates/core/broker/src/lib.rs

@@ -0,0 +1,18 @@
+use revolt_config::config;
+
+pub mod event_stream;
+
+/// Create a lapin client
+pub async fn create_client() -> lapin::Connection {
+    let config = config().await;
+
+    lapin::Connection::connect(
+        &format!(
+            "amqp://{}:{}@{}:{}/%2f",
+            config.rabbit.username, config.rabbit.password, config.rabbit.host, config.rabbit.port
+        ),
+        Default::default(),
+    )
+    .await
+    .unwrap()
+}

crates/core/coalesced/Cargo.toml

@@ -1,22 +0,0 @@
-[package]
-name = "revolt-coalesced"
-version = "0.8.9"
-edition = "2021"
-license = "MIT"
-authors = ["Paul Makles <me@insrt.uk>", "Zomatree <me@zomatree.live>"]
-description = "Revolt Backend: Coalescion service"
-
-[features]
-tokio = ["dep:tokio"]
-queue = ["dep:indexmap"]
-cache = ["dep:lru"]
-
-default = ["tokio"]
-
-[dependencies]
-tokio = { version = "1.47.0", features = ["sync"], optional = true }
-indexmap = { version = "*", optional = true }
-lru = { version = "*", optional = true }
-
-[dev-dependencies]
-tokio = { version = "1.47.0", features = ["rt", "rt-multi-thread", "macros", "time"] }

crates/core/coalesced/LICENSE

@@ -1,9 +0,0 @@
-MIT License
-
-Copyright (c) 2024 Pawel Makles
-
-Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

crates/core/coalesced/src/config.rs

@@ -1,24 +0,0 @@
-#[derive(Clone, PartialEq, Eq, Debug)]
-/// Config values for [`CoalescionService`].
-pub struct CoalescionServiceConfig {
-    /// How many tasks are running at once
-    pub max_concurrent: Option<usize>,
-    /// Whether to queue tasks once `max_concurrent` is reached
-    #[cfg(feature = "queue")]
-    pub queue_requests: bool,
-    /// Max amount of tasks in the buffer queue
-    #[cfg(feature = "queue")]
-    pub max_queue: Option<usize>,
-}
-
-impl Default for CoalescionServiceConfig {
-    fn default() -> Self {
-        Self {
-            max_concurrent: Some(100),
-            #[cfg(feature = "queue")]
-            queue_requests: true,
-            #[cfg(feature = "queue")]
-            max_queue: Some(100)
-        }
-    }
-}

crates/core/coalesced/src/error.rs

@@ -1,27 +0,0 @@
-use std::fmt;
-
-#[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)]
-/// Coalescion service error.
-pub enum Error {
-    /// Failed to receive the actions return from the channel for unknown reason
-    RecvError,
-    /// Reached the `max_concurrent` amount of actions running at once and could not queue the action
-    MaxConcurrent,
-    /// Reached the `max_queue` amount of actions in the queue
-    MaxQueue,
-    /// Failed to downcast the type to the current type being returned, this will be most likely an ID collision
-    DowncastError,
-}
-
-impl fmt::Display for Error {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            Error::RecvError => write!(f, "Unable to receive data from the channel"),
-            Error::MaxConcurrent => write!(f, "Max number of tasks running at once"),
-            Error::MaxQueue => write!(f, "Max number of tasks in queue"),
-            Error::DowncastError => write!(f, "Failed to downcast type, possible key collision with different types")
-        }
-    }
-}
-
-impl std::error::Error for Error {}

crates/core/coalesced/src/lib.rs

@@ -1,39 +0,0 @@
-//! # Coalesced
-//!
-//! Coalescion service to group, caching and queue duplicate actions.
-//! useful for deduplicating web requests, database lookups and other similar resource
-//! intensive or rate-limited actions.
-//!
-//! ## Features
-//! - `tokio`: Uses tokio for the async backend, this is currently the only backend.
-//! - `queue`: Whether to support queueing requests to only allow X amount of actions running at once.
-//! - `cache`: Whether to cache the actions results for future actions with the same id, uses an LRU cache internally.
-//!
-//! [`CoalescionService`] uses both [`Arc`] and [`RwLock`] internally and can be cheaply cloned to
-//! use in your codebase.
-//!
-//! It is common practice to wrap the service and in your own which delegates the executions to ensure all ids are tracked in one location across your codebase.
-//!
-//! All values are stored using [`Any`] and must be [`'static`] + [`Send`] + [`Sync`], if there is an id mismatch
-//! and a type is wrong the library will return an error, values returned from the service are also
-//! wrapped in an [`Arc`] as they are shared to each duplicate action.
-//!
-//! ## Example:
-//! ```rs
-//! use revolt_coalesced::CoalescionService;
-//!
-//! let service = CoalescionService::new();
-//!
-//! let user_id = "my_user_id";
-//! let user = service.execute(user_id, || async move {
-//!     database.fetch_user(user_id).await.unwrap()
-//! }).await;
-//! ```
-
-mod config;
-mod error;
-mod service;
-
-pub use config::CoalescionServiceConfig;
-pub use error::Error;
-pub use service::CoalescionService;

crates/core/coalesced/src/service.rs

@@ -1,208 +0,0 @@
-use std::{any::Any, collections::HashMap, fmt::Debug, future::Future, hash::Hash, sync::Arc};
-
-use tokio::sync::{
-    watch::{channel as watch_channel, Receiver},
-    RwLock,
-};
-
-#[cfg(feature = "cache")]
-use lru::LruCache;
-
-#[cfg(feature = "queue")]
-use indexmap::IndexMap;
-
-use crate::{CoalescionServiceConfig, Error};
-
-#[derive(Debug, Clone)]
-#[allow(clippy::type_complexity)]
-/// Coalescion service
-///
-/// See module description for example usage.
-pub struct CoalescionService<Id: Hash + Clone + Eq> {
-    config: Arc<CoalescionServiceConfig>,
-    watchers: Arc<RwLock<HashMap<Id, Receiver<Option<Result<Arc<dyn Any + Send + Sync>, Error>>>>>>,
-    #[cfg(feature = "queue")]
-    queue: Arc<RwLock<IndexMap<Id, Receiver<Option<Result<Arc<dyn Any + Send + Sync>, Error>>>>>>,
-    #[cfg(feature = "cache")]
-    cache: Option<Arc<tokio::sync::Mutex<LruCache<Id, Arc<dyn Any + Send + Sync>>>>>,
-}
-
-impl<Id: Hash + Clone + Eq> CoalescionService<Id> {
-    pub fn new() -> Self {
-        Default::default()
-    }
-
-    pub fn from_config(config: CoalescionServiceConfig) -> Self {
-        Self {
-            config: Arc::new(config),
-            watchers: Arc::new(RwLock::new(HashMap::new())),
-            #[cfg(feature = "queue")]
-            queue: Arc::new(RwLock::new(IndexMap::new())),
-            #[cfg(feature = "cache")]
-            cache: None,
-        }
-    }
-
-    #[cfg(feature = "cache")]
-    pub fn from_cache(
-        config: CoalescionServiceConfig,
-        cache: LruCache<Id, Arc<dyn Any + Send + Sync>>,
-    ) -> Self {
-        Self {
-            cache: Some(Arc::new(Mutex::new(cache))),
-            ..Self::from_config(config)
-        }
-    }
-
-    async fn wait_for<Value: Any + Send + Sync>(
-        &self,
-        mut receiver: Receiver<Option<Result<Arc<dyn Any + Send + Sync>, Error>>>,
-    ) -> Result<Arc<Value>, Error> {
-        receiver
-            .wait_for(|v| v.is_some())
-            .await
-            .map_err(|_| Error::RecvError)
-            .and_then(|r| r.clone().unwrap())
-            .and_then(|arc| Arc::downcast(arc).map_err(|_| Error::DowncastError))
-    }
-
-    async fn insert_and_execute<
-        Value: Send + Sync + 'static,
-        F: FnOnce() -> Fut,
-        Fut: Future<Output = Value>,
-    >(
-        &self,
-        id: Id,
-        func: F,
-    ) -> Result<Arc<Value>, Error> {
-        let (send, recv) = watch_channel(None);
-
-        self.watchers.write().await.insert(id.clone(), recv);
-
-        let value = Ok(Arc::new(func().await));
-
-        send.send_modify(|opt| {
-            opt.replace(value.clone().map(|v| v as Arc<dyn Any + Send + Sync>));
-        });
-
-        #[cfg(feature = "cache")]
-        if let Some(cache) = self.cache.as_ref() {
-            if let Ok(value) = &value {
-                cache.lock().await.push(id.clone(), value.clone());
-            }
-        };
-
-        self.watchers.write().await.remove(&id);
-
-        value
-    }
-
-    /// Coalesces an function, the actual function may not run if one with the same id is already running,
-    /// queued to be ran, or cached, the id should be globally unique for this specific action.
-    pub async fn execute<
-        Value: Send + Sync + 'static,
-        F: FnOnce() -> Fut,
-        Fut: Future<Output = Value>,
-    >(
-        &self,
-        id: Id,
-        func: F,
-    ) -> Result<Arc<Value>, Error> {
-        #[cfg(feature = "cache")]
-        if let Some(cache) = self.cache.as_ref() {
-            if let Some(value) = cache.lock().await.get(&id) {
-                return Arc::downcast::<Value>(value.clone()).map_err(|_| Error::DowncastError);
-            }
-        };
-
-        let (receiver, length) = {
-            let watchers = self.watchers.read().await;
-            let length = watchers.len();
-
-            (watchers.get(&id).cloned(), length)
-        };
-
-        if let Some(receiver) = receiver {
-            self.wait_for(receiver).await
-        } else {
-            match self.config.max_concurrent {
-                Some(max_concurrent) if length >= max_concurrent => {
-                    #[cfg(feature = "queue")]
-                    if self.config.queue_requests {
-                        let (receiver, length) = {
-                            let queue = self.queue.read().await;
-
-                            (queue.get(&id).cloned(), queue.len())
-                        };
-
-                        if let Some(receiver) = receiver {
-                            return self.wait_for(receiver).await;
-                        } else {
-                            if self
-                                .config
-                                .max_queue
-                                .is_some_and(|max_queue| max_queue >= length)
-                            {
-                                return Err(Error::MaxQueue);
-                            };
-
-                            let (send, recv) = watch_channel(None);
-
-                            self.queue.write().await.insert(id.clone(), recv);
-
-                            loop {
-                                let length = self.watchers.read().await.len();
-
-                                if length < max_concurrent {
-                                    let first_key = {
-                                        let queue = self.queue.read().await;
-                                        queue.first().map(|v| v.0).cloned()
-                                    };
-
-                                    if first_key == Some(id.clone()) {
-                                        self.queue.write().await.shift_remove(&id);
-
-                                        let response = self.insert_and_execute(id, func).await;
-
-                                        send.send_modify(|opt| {
-                                            opt.replace(
-                                                response
-                                                    .clone()
-                                                    .map(|v| v as Arc<dyn Any + Send + Sync>),
-                                            );
-                                        });
-
-                                        return response;
-                                    }
-                                }
-                            }
-                        }
-                    } else {
-                        Err(Error::MaxConcurrent)
-                    }
-
-                    #[cfg(not(feature = "queue"))]
-                    Err(Error::MaxConcurrent)
-                }
-                _ => self.insert_and_execute(id, func).await,
-            }
-        }
-    }
-
-    /// Fetches the amount of currently running tasks
-    pub async fn current_task_count(&self) -> usize {
-        self.watchers.read().await.len()
-    }
-
-    #[cfg(feature = "queue")]
-    /// Fetches the current length of the queue
-    pub async fn current_queue_len(&self) -> usize {
-        self.queue.read().await.len()
-    }
-}
-
-impl<Id: Hash + Clone + Eq> Default for CoalescionService<Id> {
-    fn default() -> Self {
-        Self::from_config(CoalescionServiceConfig::default())
-    }
-}

crates/core/config/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-config"
-version = "0.8.9"
+version = "0.8.8"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -11,9 +11,8 @@ description = "Revolt Backend: Configuration"
 [features]
 anyhow = ["dep:sentry-anyhow"]
 report-macros = ["revolt-result"]
-sentry = ["dep:sentry"]
 test = ["async-std"]
-default = ["test", "sentry"]
+default = ["test", "anyhow"]
 
 [dependencies]
 # Utility
@@ -33,8 +32,8 @@ log = "0.4.14"
 pretty_env_logger = "0.4.0"
 
 # Sentry
-sentry = { version = "0.31.5", optional = true }
+sentry = "0.31.5"
 sentry-anyhow = { version = "0.38.1", optional = true }
 
 # Core
-revolt-result = { version = "0.8.9", path = "../result", optional = true }
+revolt-result = { version = "0.8.8", path = "../result", optional = true }

crates/core/config/Revolt.toml

@@ -22,14 +22,26 @@ january = "http://local.revolt.chat/january"
 voso_legacy = ""
 voso_legacy_ws = ""
 
-[hosts.livekit]
-
 [rabbit]
 host = "rabbit"
 port = 5672
 username = "rabbituser"
 password = "rabbitpass"
 
+[rabbit.event_stream]
+# Configuration for event brokerage
+# Using default/direct exchange
+exchange = ""
+queue = "revolt.events"
+# Number of channels that can be opened per single TCP connection
+channels_per_conn = 128
+# Maximum size of the stream
+stream_max_length_bytes = 5_000_000_000
+# Size of the Bloom filter
+filter_size_bytes = 26
+# Number of messages to prefetch
+qos_prefetch = 100
+
 [api]
 
 [api.registration]
@@ -58,8 +70,6 @@ voso_legacy_token = ""
 trust_cloudflare = false
 # easypwned endpoint
 easypwned = ""
-# Tenor API Key
-tenor_key = ""
 
 [api.security.captcha]
 # hCaptcha configuration
@@ -70,14 +80,9 @@ hcaptcha_sitekey = ""
 # Maximum concurrent connections (to proxy server)
 max_concurrent_connections = 50
 
-[api.livekit]
-# How long to ring devices for when calling in dms/groups, in seconds
-call_ring_duration = 30
-
-[api.livekit.nodes]
-
 [api.users]
 
+
 [pushd]
 # this changes the names of the queues to not overlap 
 # prod/beta if they happen to be on the same exchange/instance.
@@ -94,7 +99,6 @@ message_queue = "notifications.origin.message"
 mass_mention_queue = "notifications.origin.mass_mention" # handles messages that contain role or everyone mentions
 fr_accepted_queue = "notifications.ingest.fr_accepted"   # friend request accepted
 fr_received_queue = "notifications.ingest.fr_received"   # friend request received
-dm_call_queue = "notifications.ingest.dm_call"   # direct message voice call
 generic_queue = "notifications.ingest.generic"           # generic messages (title + body)
 ack_queue = "notifications.process.ack"                  # updates badges for apple devices
 
@@ -238,18 +242,6 @@ message_attachments = 5
 # Maximum number of servers the user can create/join
 servers = 50
 
-# Maximum audio frequency (Hz) in voice calls
-voice_quality = 16000
-
-# Whether the user can use video streams in voice calls
-video = true
-
-# Mamimum resolution (width, height) of video streams in voice calls
-video_resolution = [1080, 720]
-
-# Minimum and maximum aspect ratio of video streams in voice calls
-video_aspect_ratio = [0.3, 2.5]
-
 [features.limits.new_user.file_upload_size_limit]
 # Maximum file size limits (in bytes)
 attachments = 20_000_000
@@ -277,18 +269,6 @@ message_attachments = 5
 # Maximum number of servers the user can create/join
 servers = 100
 
-# Maximum audio frequency (Hz) in voice calls
-voice_quality = 16000
-
-# Whether the user can use video streams in voice calls
-video = true
-
-# Mamimum resolution (width, height) of video streams in voice calls
-video_resolution = [1080, 720]
-
-# Minimum and maximum aspect ratio of video streams in voice calls
-video_aspect_ratio = [0.3, 2.5]
-
 [features.limits.default.file_upload_size_limit]
 # Maximum file size limits (in bytes)
 attachments = 20_000_000
@@ -307,9 +287,7 @@ process_message_delay_limit = 5
 # Configuration for Sentry error reporting
 api = ""
 events = ""
-voice_ingress = ""
 files = ""
 proxy = ""
 pushd = ""
 crond = ""
-gifbox = ""

crates/core/config/src/lib.rs

@@ -6,12 +6,10 @@ use futures_locks::RwLock;
 use once_cell::sync::Lazy;
 use serde::Deserialize;
 
-#[cfg(feature = "sentry")]
 pub use sentry::{capture_error, capture_message, Level};
-#[cfg(feature = "anyhow")]
 pub use sentry_anyhow::capture_anyhow;
 
-#[cfg(all(feature = "report-macros", feature = "sentry"))]
+#[cfg(feature = "report-macros")]
 #[macro_export]
 macro_rules! report_error {
     ( $expr: expr, $error: ident $( $tt:tt )? ) => {
@@ -26,7 +24,7 @@ macro_rules! report_error {
     };
 }
 
-#[cfg(all(feature = "report-macros", feature = "sentry"))]
+#[cfg(feature = "report-macros")]
 #[macro_export]
 macro_rules! capture_internal_error {
     ( $expr: expr ) => {
@@ -37,7 +35,7 @@ macro_rules! capture_internal_error {
     };
 }
 
-#[cfg(all(feature = "report-macros", feature = "sentry"))]
+#[cfg(feature = "report-macros")]
 #[macro_export]
 macro_rules! report_internal_error {
     ( $expr: expr ) => {
@@ -110,12 +108,24 @@ pub struct Database {
     pub redis: String,
 }
 
+#[derive(Deserialize, Debug, Clone)]
+pub struct RabbitEventStream {
+    pub exchange: String,
+    pub queue: String,
+    pub channels_per_conn: usize,
+    pub stream_max_length_bytes: i64,
+    pub filter_size_bytes: i64,
+    pub qos_prefetch: u16,
+}
+
 #[derive(Deserialize, Debug, Clone)]
 pub struct Rabbit {
     pub host: String,
     pub port: u16,
     pub username: String,
     pub password: String,
+
+    pub event_stream: RabbitEventStream,
 }
 
 #[derive(Deserialize, Debug, Clone)]
@@ -125,7 +135,8 @@ pub struct Hosts {
     pub events: String,
     pub autumn: String,
     pub january: String,
-    pub livekit: HashMap<String, String>,
+    pub voso_legacy: String,
+    pub voso_legacy_ws: String,
 }
 
 #[derive(Deserialize, Debug, Clone)]
@@ -189,7 +200,6 @@ pub struct ApiSecurity {
     pub captcha: ApiSecurityCaptcha,
     pub trust_cloudflare: bool,
     pub easypwned: String,
-    pub tenor_key: String,
 }
 
 #[derive(Deserialize, Debug, Clone)]
@@ -197,25 +207,6 @@ pub struct ApiWorkers {
     pub max_concurrent_connections: usize,
 }
 
-#[derive(Deserialize, Debug, Clone)]
-pub struct ApiLiveKit {
-    pub call_ring_duration: usize,
-    pub nodes: HashMap<String, LiveKitNode>,
-}
-
-#[derive(Deserialize, Debug, Clone)]
-pub struct LiveKitNode {
-    pub url: String,
-    pub lat: f64,
-    pub lon: f64,
-    pub key: String,
-    pub secret: String,
-
-    // whether to hide the node in the nodes list
-    #[serde(default)]
-    pub private: bool,
-}
-
 #[derive(Deserialize, Debug, Clone)]
 pub struct ApiUsers {
     pub early_adopter_cutoff: Option<u64>,
@@ -227,7 +218,6 @@ pub struct Api {
     pub smtp: ApiSmtp,
     pub security: ApiSecurity,
     pub workers: ApiWorkers,
-    pub livekit: ApiLiveKit,
     pub users: ApiUsers,
 }
 
@@ -240,7 +230,6 @@ pub struct Pushd {
     // Queues
     pub message_queue: String,
     pub mass_mention_queue: String,
-    pub dm_call_queue: String,
     pub fr_accepted_queue: String,
     pub fr_received_queue: String,
     pub generic_queue: String,
@@ -271,10 +260,6 @@ impl Pushd {
         self.get_routing_key(self.mass_mention_queue.clone())
     }
 
-    pub fn get_dm_call_routing_key(&self) -> String {
-        self.get_routing_key(self.dm_call_queue.clone())
-    }
-
     pub fn get_fr_accepted_routing_key(&self) -> String {
         self.get_routing_key(self.fr_accepted_queue.clone())
     }
@@ -342,10 +327,6 @@ pub struct FeaturesLimits {
     pub message_length: usize,
     pub message_attachments: usize,
     pub servers: usize,
-    pub voice_quality: u32,
-    pub video: bool,
-    pub video_resolution: [u32; 2],
-    pub video_aspect_ratio: [f32; 2],
 
     pub file_upload_size_limit: HashMap<String, usize>,
 }
@@ -390,12 +371,10 @@ pub struct Features {
 pub struct Sentry {
     pub api: String,
     pub events: String,
-    pub voice_ingress: String,
     pub files: String,
     pub proxy: String,
     pub pushd: String,
     pub crond: String,
-    pub gifbox: String,
 }
 
 #[derive(Deserialize, Debug, Clone)]
@@ -452,7 +431,6 @@ pub async fn config() -> Settings {
 }
 
 /// Configure logging and common Rust variables
-#[cfg(feature = "sentry")]
 pub async fn setup_logging(release: &'static str, dsn: String) -> Option<sentry::ClientInitGuard> {
     if std::env::var("RUST_LOG").is_err() {
         std::env::set_var("RUST_LOG", "info");
@@ -478,7 +456,6 @@ pub async fn setup_logging(release: &'static str, dsn: String) -> Option<sentry:
     }
 }
 
-#[cfg(feature = "sentry")]
 #[macro_export]
 macro_rules! configure {
     ($application: ident) => {

crates/core/database/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-database"
-version = "0.8.9"
+version = "0.8.8"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -10,34 +10,34 @@ description = "Revolt Backend: Database Implementation"
 
 [features]
 # Databases
-mongodb = ["dep:mongodb", "bson", "authifier/database-mongodb"]
+mongodb = ["dep:mongodb", "bson"]
 
 # ... Other
 tasks = ["isahc", "linkify", "url-escape"]
-async-std-runtime = ["async-std", "authifier/async-std-runtime"]
-rocket-impl = ["rocket"]
+async-std-runtime = ["async-std"]
+rocket-impl = ["rocket", "schemars", "revolt_okapi", "revolt_rocket_okapi"]
 axum-impl = ["axum"]
 redis-is-patched = ["revolt-presence/redis-is-patched"]
-voice = ["livekit-api", "livekit-protocol", "livekit-runtime"]
 
 # Default Features
 default = ["mongodb", "async-std-runtime", "tasks"]
 
 [dependencies]
 # Core
-revolt-config = { version = "0.8.9", path = "../config", features = [
+revolt-config = { version = "0.8.8", path = "../config", features = [
     "report-macros",
 ] }
-revolt-result = { version = "0.8.9", path = "../result" }
-revolt-models = { version = "0.8.9", path = "../models", features = [
+revolt-result = { version = "0.8.8", path = "../result" }
+revolt-models = { version = "0.8.8", path = "../models", features = [
     "validator",
 ] }
-revolt-presence = { version = "0.8.9", path = "../presence" }
-revolt-permissions = { version = "0.8.9", path = "../permissions", features = [
+revolt-presence = { version = "0.8.8", path = "../presence" }
+revolt-permissions = { version = "0.8.8", path = "../permissions", features = [
     "serde",
     "bson",
 ] }
-revolt-parser = { version = "0.8.9", path = "../parser" }
+revolt-parser = { version = "0.8.8", path = "../parser" }
+revolt-broker = { version = "0.8.8", path = "../broker" }
 
 # Utility
 log = "0.4"
@@ -47,7 +47,7 @@ ulid = "1.0.0"
 nanoid = "0.4.0"
 base64 = "0.21.3"
 once_cell = "1.17"
-indexmap = "2.12.0"
+indexmap = "1.9.1"
 decancer = "1.6.2"
 deadqueue = "0.2.4"
 linkify = { optional = true, version = "0.8.1" }
@@ -57,9 +57,10 @@ isahc = { optional = true, version = "1.7", features = ["json"] }
 
 # Serialisation
 serde_json = "1"
+rmp-serde = "1.0.0"
 revolt_optional_struct = "0.2.0"
 serde = { version = "1", features = ["derive"] }
-iso8601-timestamp = { version = "0.4.0", features = ["serde", "bson"] }
+iso8601-timestamp = { version = "0.2.10", features = ["serde", "bson"] }
 
 # Events
 redis-kiss = { version = "0.1.4" }
@@ -82,23 +83,24 @@ async-recursion = "1.0.4"
 async-std = { version = "1.8.0", features = ["attributes"], optional = true }
 
 # Axum Impl
-axum = { version = "0.8.6", optional = true }
+axum = { version = "0.7.5", optional = true }
 
 # Rocket Impl
+schemars = { version = "0.8.8", optional = true }
 rocket = { version = "0.5.1", default-features = false, features = [
     "json",
 ], optional = true }
+revolt_okapi = { version = "0.9.1", optional = true }
+revolt_rocket_okapi = { version = "0.10.0", optional = true }
 
-# Openapi Schema
-utoipa = { version = "5.4.0", optional = true }
+# Notifications
+fcm_v1 = "0.3.0"
+web-push = "0.10.0"
+revolt_a2 = { version = "0.10", default-features = false, features = ["ring"] }
 
 # Authifier
-authifier = { version = "1.0.15" }
+authifier = { version = "1.0.15", features = ["rocket_impl"] }
 
 # RabbitMQ
 amqprs = { version = "1.7.0" }
-
-# Voice
-livekit-api = { version = "=0.4.4", optional = true}
-livekit-protocol = { version = "=0.4.0", optional = true }
-livekit-runtime = { version = "=0.3.1", features = ["tokio"], optional = true }
+lapin = { version = "3.0.0" }

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

@@ -2,8 +2,7 @@ use std::collections::HashSet;
 
 use crate::events::rabbit::*;
 use crate::User;
-use amqprs::channel::{BasicPublishArguments, ExchangeDeclareArguments};
-use amqprs::connection::OpenConnectionArguments;
+use amqprs::channel::BasicPublishArguments;
 use amqprs::{channel::Channel, connection::Connection, error::Error as AMQPError};
 use amqprs::{BasicProperties, FieldTable};
 use revolt_models::v0::PushNotification;
@@ -26,35 +25,6 @@ impl AMQP {
         }
     }
 
-    pub async fn new_auto() -> AMQP {
-        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(),
-            )
-            .await
-            .expect("Failed to declare exchange");
-
-        AMQP::new(connection, channel)
-    }
-
     pub async fn friend_request_accepted(
         &self,
         accepted_request_user: &User,
@@ -270,50 +240,4 @@ impl AMQP {
             )
             .await
     }
-
-    /// # DM Call Update
-    /// Used to send an update about a DM call, eg. start or end of a call.
-    /// Recipients can be used to narrow the scope of recipients, otherwise all recipients will be notified.
-    /// `ended` refers to the ringing period, not necessarily the call itself.
-    pub async fn dm_call_updated(
-        &self,
-        initiator_id: &str,
-        channel_id: &str,
-        started_at: Option<&str>,
-        ended: bool,
-        recipients: Option<Vec<String>>,
-    ) -> Result<(), AMQPError> {
-        let config = revolt_config::config().await;
-
-        let payload = InternalDmCallPayload {
-            payload: DmCallPayload {
-                initiator_id: initiator_id.to_string(),
-                channel_id: channel_id.to_string(),
-                started_at: started_at.map(|f| f.to_string()),
-                ended,
-            },
-            recipients,
-        };
-        let payload = to_string(&payload).unwrap();
-
-        debug!(
-            "Sending dm call update payload on channel {}: {}",
-            config.pushd.get_dm_call_routing_key(),
-            payload
-        );
-
-        self.channel
-            .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(),
-                ),
-            )
-            .await
-    }
 }

crates/core/database/src/drivers/mod.rs

@@ -1,4 +1,3 @@
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 
@@ -14,7 +13,6 @@ use authifier::Authifier;
 use rand::Rng;
 use revolt_config::config;
 
-#[cfg(feature = "mongodb")]
 pub use self::mongodb::*;
 pub use self::reference::*;
 
@@ -27,20 +25,17 @@ pub enum DatabaseInfo {
     /// Use the mock database
     Reference,
     /// Connect to MongoDB
-    #[cfg(feature = "mongodb")]
     MongoDb { uri: String, database_name: String },
     /// Use existing MongoDB connection
-    #[cfg(feature = "mongodb")]
     MongoDbFromClient(::mongodb::Client, String),
 }
 
 /// Database
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub enum Database {
     /// Mock database
     Reference(ReferenceDb),
     /// MongoDB database
-    #[cfg(feature = "mongodb")]
     MongoDb(MongoDb),
 }
 
@@ -50,7 +45,7 @@ impl DatabaseInfo {
     pub async fn connect(self) -> Result<Database, String> {
         let config = config().await;
 
-        match self {
+        Ok(match self {
             DatabaseInfo::Auto => {
                 if std::env::var("TEST_DB").is_ok() {
                     DatabaseInfo::Test(format!(
@@ -58,20 +53,16 @@ impl DatabaseInfo {
                         rand::thread_rng().gen_range(1_000_000..10_000_000)
                     ))
                     .connect()
-                    .await
+                    .await?
                 } else if !config.database.mongodb.is_empty() {
-                    #[cfg(feature = "mongodb")]
-                    return DatabaseInfo::MongoDb {
+                    DatabaseInfo::MongoDb {
                         uri: config.database.mongodb,
                         database_name: "revolt".to_string(),
                     }
                     .connect()
-                    .await;
-
-                    #[cfg(not(feature = "mongodb"))]
-                    return Err("MongoDB not enabled.".to_string())
+                    .await?
                 } else {
-                    DatabaseInfo::Reference.connect().await
+                    DatabaseInfo::Reference.connect().await?
                 }
             }
             DatabaseInfo::Test(database_name) => {
@@ -79,36 +70,30 @@ impl DatabaseInfo {
                     .expect("`TEST_DB` environment variable should be set to REFERENCE or MONGODB")
                     .as_str()
                 {
-                    "REFERENCE" => DatabaseInfo::Reference.connect().await,
+                    "REFERENCE" => DatabaseInfo::Reference.connect().await?,
                     "MONGODB" => {
-                        #[cfg(feature = "mongodb")]
-                        return DatabaseInfo::MongoDb {
+                        DatabaseInfo::MongoDb {
                             uri: config.database.mongodb,
                             database_name,
                         }
                         .connect()
-                        .await;
-
-                        #[cfg(not(feature = "mongodb"))]
-                        return Err("MongoDB not enabled.".to_string())
+                        .await?
                     }
                     _ => unreachable!("must specify REFERENCE or MONGODB"),
                 }
             }
-            DatabaseInfo::Reference => Ok(Database::Reference(Default::default())),
-            #[cfg(feature = "mongodb")]
+            DatabaseInfo::Reference => Database::Reference(Default::default()),
             DatabaseInfo::MongoDb { uri, database_name } => {
                 let client = ::mongodb::Client::with_uri_str(uri)
                     .await
                     .map_err(|_| "Failed to init db connection.".to_string())?;
 
-                Ok(Database::MongoDb(MongoDb(client, database_name)))
+                Database::MongoDb(MongoDb(client, database_name))
             }
-            #[cfg(feature = "mongodb")]
             DatabaseInfo::MongoDbFromClient(client, database_name) => {
-                Ok(Database::MongoDb(MongoDb(client, database_name)))
+                Database::MongoDb(MongoDb(client, database_name))
             }
-        }
+        })
     }
 }
 
@@ -234,16 +219,12 @@ impl Database {
         Authifier {
             database: match self {
                 Database::Reference(_) => Default::default(),
-                #[cfg(feature = "mongodb")]
                 Database::MongoDb(MongoDb(client, _)) => authifier::Database::MongoDb(
                     authifier::database::MongoDb(client.database("revolt")),
                 ),
             },
             config: auth_config,
-            #[cfg(feature = "tasks")]
             event_channel: Some(crate::tasks::authifier_relay::sender()),
-            #[cfg(not(feature = "tasks"))]
-            event_channel: None,
         }
     }
 }

crates/core/database/src/drivers/mongodb.rs

@@ -10,8 +10,8 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 
 database_derived!(
+    #[cfg(feature = "mongodb")]
     /// MongoDB implementation
-    #[derive(Debug)]
     pub struct MongoDb(pub ::mongodb::Client, pub String);
 );
 

crates/core/database/src/drivers/reference.rs

@@ -10,7 +10,7 @@ use crate::{
 
 database_derived!(
     /// Reference implementation
-    #[derive(Default, Debug)]
+    #[derive(Default)]
     pub struct ReferenceDb {
         pub bots: Arc<Mutex<HashMap<String, Bot>>>,
         pub channels: Arc<Mutex<HashMap<String, Channel>>>,

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

@@ -1,9 +1,13 @@
 use authifier::AuthifierEvent;
+use revolt_broker::event_stream;
 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, PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialUserVoiceState, PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings, UserVoiceState, Webhook
+    AppendMessage, Channel, ChannelUnread, Emoji, FieldsChannel, FieldsMember, FieldsMessage,
+    FieldsRole, FieldsServer, FieldsUser, FieldsWebhook, Member, MemberCompositeKey, Message,
+    PartialChannel, PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser,
+    PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings, Webhook,
 };
 
 use crate::Database;
@@ -17,33 +21,16 @@ pub enum Ping {
 }
 
 /// Fields provided in Ready payload
-#[derive(PartialEq, Debug, Clone, Deserialize)]
-pub struct ReadyPayloadFields {
-    pub users: bool,
-    pub servers: bool,
-    pub channels: bool,
-    pub members: bool,
-    pub emojis: bool,
-    pub voice_states: bool,
-    pub user_settings: Vec<String>,
-    pub channel_unreads: bool,
-    pub policy_changes: bool,
-}
+#[derive(PartialEq)]
+pub enum ReadyPayloadFields {
+    Users,
+    Servers,
+    Channels,
+    Members,
+    Emoji,
 
-impl Default for ReadyPayloadFields {
-    fn default() -> Self {
-        Self {
-            users: true,
-            servers: true,
-            channels: true,
-            members: true,
-            emojis: true,
-            voice_states: true,
-            user_settings: Vec::new(),
-            channel_unreads: false,
-            policy_changes: true,
-        }
-    }
+    UserSettings(Vec<String>),
+    ChannelUnreads,
 }
 
 /// Protocol Events
@@ -71,16 +58,13 @@ pub enum EventV1 {
         members: Option<Vec<Member>>,
         #[serde(skip_serializing_if = "Option::is_none")]
         emojis: Option<Vec<Emoji>>,
-        #[serde(skip_serializing_if = "Option::is_none")]
-        voice_states: Option<Vec<ChannelVoiceState>>,
 
         #[serde(skip_serializing_if = "Option::is_none")]
         user_settings: Option<UserSettings>,
         #[serde(skip_serializing_if = "Option::is_none")]
         channel_unreads: Option<Vec<ChannelUnread>>,
 
-        #[serde(skip_serializing_if = "Option::is_none")]
-        policy_changes: Option<Vec<PolicyChange>>,
+        policy_changes: Vec<PolicyChange>,
     },
 
     /// Ping response
@@ -139,7 +123,6 @@ pub enum EventV1 {
         server: Server,
         channels: Vec<Channel>,
         emojis: Vec<Emoji>,
-        voice_states: Vec<ChannelVoiceState>
     },
 
     /// Update existing server
@@ -162,13 +145,7 @@ pub enum EventV1 {
     },
 
     /// User joins server
-    ServerMemberJoin {
-        id: String,
-        // Deprecated: use member.id.user
-        #[deprecated = "Use member.id.user instead"]
-        user: String,
-        member: Member,
-    },
+    ServerMemberJoin { id: String, user: String },
 
     /// User left server
     ServerMemberLeave {
@@ -272,46 +249,21 @@ pub enum EventV1 {
 
     /// Auth events
     Auth(AuthifierEvent),
-
-    /// Voice events
-    VoiceChannelJoin {
-        id: String,
-        state: UserVoiceState,
-    },
-    VoiceChannelLeave {
-        id: String,
-        user: String,
-    },
-    VoiceChannelMove {
-        user: String,
-        from: String,
-        to: String,
-        state: UserVoiceState
-    },
-    UserVoiceStateUpdate {
-        id: String,
-        channel_id: String,
-        data: PartialUserVoiceState,
-    },
-    UserMoveVoiceChannel {
-        node: String,
-        from: String,
-        to: String,
-        token: String,
-    }
 }
 
 impl EventV1 {
     /// Publish helper wrapper
     pub async fn p(self, channel: String) {
-        #[cfg(not(debug_assertions))]
-        redis_kiss::p(channel, self).await;
-
         #[cfg(debug_assertions)]
         info!("Publishing event to {channel}: {self:?}");
 
+        let result = event_stream::publish_event(&channel, &self).await;
+
+        #[cfg(not(debug_assertions))]
+        result.ok();
+
         #[cfg(debug_assertions)]
-        redis_kiss::publish(channel, self).await.unwrap();
+        result.unwrap();
     }
 
     /// Publish user event

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

@@ -37,20 +37,6 @@ pub struct GenericPayload {
     pub user: User,
 }
 
-#[derive(Serialize, Deserialize, Clone)]
-pub struct DmCallPayload {
-    pub initiator_id: String,
-    pub channel_id: String,
-    pub started_at: Option<String>,
-    pub ended: bool,
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct InternalDmCallPayload {
-    pub payload: DmCallPayload,
-    pub recipients: Option<Vec<String>>,
-}
-
 #[derive(Serialize, Deserialize)]
 #[serde(tag = "type", content = "data")]
 #[allow(clippy::large_enum_variant)]
@@ -60,7 +46,6 @@ pub enum PayloadKind {
     FRReceived(FRReceivedPayload),
     BadgeUpdate(usize),
     Generic(GenericPayload),
-    DmCallStartEnd(DmCallPayload),
 }
 
 #[derive(Serialize, Deserialize)]

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

@@ -1,8 +1,8 @@
-use serde::{Serialize, Deserialize};
+use serde::Deserialize;
 
 use super::client::Ping;
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Deserialize, Debug)]
 #[serde(tag = "type")]
 pub enum ClientMessage {
     Authenticate { token: String },

crates/core/database/src/lib.rs

@@ -25,13 +25,6 @@ pub use mongodb;
 #[macro_use]
 extern crate bson;
 
-#[cfg(feature = "utoipa")]
-#[macro_use]
-extern crate utoipa;
-
-#[cfg(not(feature = "async-std-runtime"))]
-compile_error!("async-std-runtime feature must be enabled.");
-
 #[macro_export]
 #[cfg(debug_assertions)]
 macro_rules! query {
@@ -64,7 +57,7 @@ macro_rules! database_derived {
 macro_rules! auto_derived {
     ( $( $item:item )+ ) => {
         $(
-            #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
+            #[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)]
             $item
         )+
     };
@@ -72,8 +65,8 @@ macro_rules! auto_derived {
 
 macro_rules! auto_derived_partial {
     ( $item:item, $name:expr ) => {
-        #[derive(OptionalStruct, Serialize, Deserialize, Debug, Clone, PartialEq)]
-        #[optional_derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq)]
+        #[derive(OptionalStruct, Serialize, Deserialize, Debug, Clone, Eq, PartialEq)]
+        #[optional_derive(Serialize, Deserialize, Debug, Clone, Default, Eq, PartialEq)]
         #[optional_name = $name]
         #[opt_skip_serializing_none]
         #[opt_some_priority]
@@ -110,16 +103,11 @@ pub mod util;
 pub use models::*;
 
 pub mod events;
-#[cfg(feature = "tasks")]
 pub mod tasks;
 
 mod amqp;
 pub use amqp::amqp::AMQP;
 
-#[cfg(feature = "voice")]
-pub mod voice;
-
-
 /// Utility function to check if a boolean value is false
 pub fn if_false(t: &bool) -> bool {
     !t

crates/core/database/src/models/admin_migrations/ops.rs

@@ -1,4 +1,3 @@
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/admin_migrations/ops/mongodb/scripts.rs

@@ -9,13 +9,14 @@ use crate::{
         bson::{doc, from_bson, from_document, to_document, Bson, DateTime, Document},
         options::FindOptions,
     },
-    AbstractServers, Invite, MongoDb, User, DISCRIMINATOR_SEARCH_SPACE,
+    AbstractChannels, AbstractServers, Channel, Invite, MongoDb, User, DISCRIMINATOR_SEARCH_SPACE,
 };
 use bson::{oid::ObjectId, to_bson};
 use futures::StreamExt;
 use iso8601_timestamp::Timestamp;
 use rand::seq::SliceRandom;
-use revolt_permissions::{ChannelPermission, DEFAULT_WEBHOOK_PERMISSIONS};
+use revolt_permissions::DEFAULT_WEBHOOK_PERMISSIONS;
+use revolt_result::{Error, ErrorType};
 use serde::{Deserialize, Serialize};
 use unicode_segmentation::UnicodeSegmentation;
 
@@ -25,7 +26,7 @@ struct MigrationInfo {
     revision: i32,
 }
 
-pub const LATEST_REVISION: i32 = 49; // MUST BE +1 to last migration
+pub const LATEST_REVISION: i32 = 42; // MUST BE +1 to last migration
 
 pub async fn migrate_database(db: &MongoDb) {
     let migrations = db.col::<Document>("migrations");
@@ -913,7 +914,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
     }
 
     if revision <= 26 {
-        // Need to migrate fields on attachments, change `user_id`, `object_id`, etc to `parent`.
         info!("Running migration [revision 26 / 15-05-2024]: fix invites being incorrectly serialized with wrong enum tagging.");
 
         auto_derived!(
@@ -1080,14 +1080,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
             channel_id: String,
         }
 
-        #[allow(clippy::enum_variant_names)]
-        #[derive(serde::Serialize, serde::Deserialize)]
-        enum Channel {
-            Group { owner: String },
-            TextChannel { server: String },
-            VoiceChannel { server: String }
-        }
-
         let webhooks = db
             .db()
             .collection::<WebhookShell>("channel_webhooks")
@@ -1099,8 +1091,8 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
             .await;
 
         for webhook in webhooks {
-            match db.col::<Channel>("channels").find_one(doc! { "_id": &webhook.channel_id }).await.unwrap() {
-                Some(channel) => {
+            match db.fetch_channel(&webhook.channel_id).await {
+                Ok(channel) => {
                     let creator_id = match channel {
                         Channel::Group { owner, .. } => owner,
                         Channel::TextChannel { server, .. }
@@ -1108,6 +1100,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
                             let server = db.fetch_server(&server).await.expect("server");
                             server.owner
                         }
+                        _ => unreachable!("not server or group channel!"),
                     };
 
                     db.db()
@@ -1125,13 +1118,17 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
                         .await
                         .expect("update webhook");
                 }
-                None => {
+                Err(Error {
+                    error_type: ErrorType::NotFound,
+                    ..
+                }) => {
                     db.db()
                         .collection::<WebhookShell>("channel_webhooks")
                         .delete_one(doc! { "_id": webhook._id })
                         .await
                         .expect("failed to delete invalid webhook");
                 }
+                Err(err) => panic!("{err:?}"),
             }
         }
     }
@@ -1172,9 +1169,9 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
             .expect("failed to update users");
     }
 
-    if revision <= 43 {
+    if revision <= 41 {
         info!(
-            "Running migration [revision 43 / 05-06-2025]: convert role ranks to uniform numbers."
+            "Running migration [revision 41 / 05-06-2025]: convert role ranks to uniform numbers."
         );
 
         #[derive(Serialize, Deserialize, Clone)]
@@ -1229,41 +1226,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
         }
     }
 
-    if revision <= 46 {
-        info!("Running migration [revision 46 / 29-04-2025]: Convert all `VoiceChannel`'s into `TextChannel`");
-
-        db.col::<Document>("channels")
-            .update_many(
-                doc! { "channel_type": "VoiceChannel" },
-                doc! {
-                    "$set": {
-                        "channel_type": "TextChannel",
-                        "voice": {}
-                    }
-                }
-            )
-            .await
-            .expect("Failed to update voice channels");
-    };
-
-        if revision <= 48 {
-        info!("Running migration [revision 48 / 22-10-2025]: Add Video + Listen to default permissions");
-
-        db.col::<Document>("servers")
-            .update_many(
-                doc! { },
-                doc! {
-                    "$bit": {
-                        "default_permissions": {
-                            "or": (ChannelPermission::Video + ChannelPermission::Speak + ChannelPermission::Listen) as i64
-                        },
-                    }
-                }
-            )
-            .await
-            .expect("Failed to update default_permissions");
-    };
-
     // Reminder to update LATEST_REVISION when adding new migrations.
     LATEST_REVISION.max(revision)
 }

crates/core/database/src/models/bots/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::{Bot, FieldsBot, PartialBot};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

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

@@ -69,7 +69,7 @@ impl Invite {
                 creator: creator.id.clone(),
                 channel: id.clone(),
             }),
-            Channel::TextChannel { id, server, .. } => {
+            Channel::TextChannel { id, server, .. } | Channel::VoiceChannel { id, server, .. } => {
                 Ok(Invite::Server {
                     code,
                     creator: creator.id.clone(),

crates/core/database/src/models/channel_invites/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::Invite;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

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

@@ -14,7 +14,7 @@ auto_derived!(
     }
 
     /// Composite primary key consisting of channel and user id
-    #[derive(Hash, Eq)]
+    #[derive(Hash)]
     pub struct ChannelCompositeKey {
         /// Channel Id
         pub channel: String,

crates/core/database/src/models/channel_unreads/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::ChannelUnread;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/channel_webhooks/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::{FieldsWebhook, PartialWebhook, Webhook};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

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

@@ -1,5 +1,4 @@
-#![allow(deprecated)]
-use std::{borrow::Cow, collections::HashMap};
+use std::collections::HashMap;
 
 use revolt_config::config;
 use revolt_models::v0::{self, MessageAuthor};
@@ -9,12 +8,10 @@ use serde::{Deserialize, Serialize};
 use ulid::Ulid;
 
 use crate::{
-    events::client::EventV1, Database, File, PartialServer, Server, SystemMessage, User, AMQP,
+    events::client::EventV1, tasks::ack::AckEvent, Database, File, IntoDocumentPath, PartialServer,
+    Server, SystemMessage, User, AMQP,
 };
 
-#[cfg(feature = "mongodb")]
-use crate::IntoDocumentPath;
-
 auto_derived!(
     #[serde(tag = "channel_type")]
     pub enum Channel {
@@ -106,18 +103,38 @@ auto_derived!(
             /// Whether this channel is marked as not safe for work
             #[serde(skip_serializing_if = "crate::if_false", default)]
             nsfw: bool,
-
-            /// Voice Information for when this channel is also a voice channel
-            #[serde(skip_serializing_if = "Option::is_none")]
-            voice: Option<VoiceInformation>,
         },
-    }
+        /// Voice channel belonging to a server
+        VoiceChannel {
+            /// Unique Id
+            #[serde(rename = "_id")]
+            id: String,
+            /// Id of the server this channel belongs to
+            server: String,
 
-    #[derive(Default)]
-    pub struct VoiceInformation {
-        /// Maximium amount of users allowed in the voice channel at once
-        #[serde(skip_serializing_if = "Option::is_none")]
-        pub max_users: Option<usize>,
+            /// Display name of the channel
+            name: String,
+            #[serde(skip_serializing_if = "Option::is_none")]
+            /// Channel description
+            description: Option<String>,
+            /// Custom icon attachment
+            #[serde(skip_serializing_if = "Option::is_none")]
+            icon: Option<File>,
+
+            /// Default permissions assigned to users in this channel
+            #[serde(skip_serializing_if = "Option::is_none")]
+            default_permissions: Option<OverrideField>,
+            /// Permissions assigned based on role to this channel
+            #[serde(
+                default = "HashMap::<String, OverrideField>::new",
+                skip_serializing_if = "HashMap::<String, OverrideField>::is_empty"
+            )]
+            role_permissions: HashMap<String, OverrideField>,
+
+            /// Whether this channel is marked as not safe for work
+            #[serde(skip_serializing_if = "crate::if_false", default)]
+            nsfw: bool,
+        },
     }
 );
 
@@ -144,8 +161,6 @@ auto_derived!(
         pub default_permissions: Option<OverrideField>,
         #[serde(skip_serializing_if = "Option::is_none")]
         pub last_message_id: Option<String>,
-        #[serde(skip_serializing_if = "Option::is_none")]
-        pub voice: Option<VoiceInformation>,
     }
 
     /// Optional fields on channel object
@@ -153,7 +168,6 @@ auto_derived!(
         Description,
         Icon,
         DefaultPermissions,
-        Voice,
     }
 );
 
@@ -205,19 +219,16 @@ impl Channel {
                 default_permissions: None,
                 role_permissions: HashMap::new(),
                 nsfw: data.nsfw.unwrap_or(false),
-                voice: data.voice.map(|voice| voice.into()),
             },
-            v0::LegacyServerChannelType::Voice => Channel::TextChannel {
+            v0::LegacyServerChannelType::Voice => Channel::VoiceChannel {
                 id: id.clone(),
                 server: server.id.to_owned(),
                 name: data.name,
                 description: data.description,
                 icon: None,
-                last_message_id: None,
                 default_permissions: None,
                 role_permissions: HashMap::new(),
                 nsfw: data.nsfw.unwrap_or(false),
-                voice: Some(data.voice.unwrap_or_default().into()),
             },
         };
 
@@ -418,28 +429,8 @@ impl Channel {
             Channel::DirectMessage { id, .. }
             | Channel::Group { id, .. }
             | Channel::SavedMessages { id, .. }
-            | Channel::TextChannel { id, .. } => id,
-        }
-    }
-
-    /// Clone this channel's server id
-    pub fn server(&self) -> Option<&str> {
-        match self {
-            Channel::TextChannel { server, .. } => Some(server),
-            _ => None,
-        }
-    }
-
-    /// Gets this channel's voice information
-    pub fn voice(&self) -> Option<Cow<VoiceInformation>> {
-        match self {
-            Self::DirectMessage { .. } | Self::Group { .. } => {
-                Some(Cow::Owned(VoiceInformation::default()))
-            }
-            Self::TextChannel {
-                voice: Some(voice), ..
-            } => Some(Cow::Borrowed(voice)),
-            _ => None,
+            | Channel::TextChannel { id, .. }
+            | Channel::VoiceChannel { id, .. } => id,
         }
     }
 
@@ -456,6 +447,12 @@ impl Channel {
                 server,
                 role_permissions,
                 ..
+            }
+            | Channel::VoiceChannel {
+                id,
+                server,
+                role_permissions,
+                ..
             } => {
                 db.set_channel_role_permission(id, role_id, permissions)
                     .await?;
@@ -502,7 +499,7 @@ impl Channel {
             clear: remove.into_iter().map(|v| v.into()).collect(),
         }
         .p(match self {
-            Self::TextChannel { server, .. } => server.clone(),
+            Self::TextChannel { server, .. } | Self::VoiceChannel { server, .. } => server.clone(),
             _ => id,
         })
         .await;
@@ -514,13 +511,17 @@ impl Channel {
     pub fn remove_field(&mut self, field: &FieldsChannel) {
         match field {
             FieldsChannel::Description => match self {
-                Self::Group { description, .. } | Self::TextChannel { description, .. } => {
+                Self::Group { description, .. }
+                | Self::TextChannel { description, .. }
+                | Self::VoiceChannel { description, .. } => {
                     description.take();
                 }
                 _ => {}
             },
             FieldsChannel::Icon => match self {
-                Self::Group { icon, .. } | Self::TextChannel { icon, .. } => {
+                Self::Group { icon, .. }
+                | Self::TextChannel { icon, .. }
+                | Self::VoiceChannel { icon, .. } => {
                     icon.take();
                 }
                 _ => {}
@@ -529,17 +530,15 @@ impl Channel {
                 Self::TextChannel {
                     default_permissions,
                     ..
+                }
+                | Self::VoiceChannel {
+                    default_permissions,
+                    ..
                 } => {
                     default_permissions.take();
                 }
                 _ => {}
             },
-            FieldsChannel::Voice => match self {
-                Self::TextChannel { voice, .. } => {
-                    voice.take();
-                }
-                _ => {}
-            },
         }
     }
 
@@ -551,7 +550,6 @@ impl Channel {
     }
 
     /// Apply partial channel to channel
-    #[allow(deprecated)]
     pub fn apply_options(&mut self, partial: PartialChannel) {
         match self {
             Self::SavedMessages { .. } => {}
@@ -600,7 +598,15 @@ impl Channel {
                 nsfw,
                 default_permissions,
                 role_permissions,
-                voice,
+                ..
+            }
+            | Self::VoiceChannel {
+                name,
+                description,
+                icon,
+                nsfw,
+                default_permissions,
+                role_permissions,
                 ..
             } => {
                 if let Some(v) = partial.name {
@@ -626,10 +632,6 @@ impl Channel {
                 if let Some(v) = partial.default_permissions {
                     default_permissions.replace(v);
                 }
-
-                if let Some(v) = partial.voice {
-                    voice.replace(v);
-                }
             }
         }
     }
@@ -644,11 +646,10 @@ impl Channel {
         .private(user.to_string())
         .await;
 
-        #[cfg(feature = "tasks")]
         crate::tasks::ack::queue_ack(
             self.id().to_string(),
             user.to_string(),
-            crate::tasks::ack::AckEvent::AckMessage {
+            AckEvent::AckMessage {
                 id: message.to_string(),
             },
         )
@@ -765,14 +766,12 @@ impl Channel {
     }
 }
 
-#[cfg(feature = "mongodb")]
 impl IntoDocumentPath for FieldsChannel {
     fn as_path(&self) -> Option<&'static str> {
         Some(match self {
             FieldsChannel::Description => "description",
             FieldsChannel::Icon => "icon",
             FieldsChannel::DefaultPermissions => "default_permissions",
-            FieldsChannel::Voice => "voice",
         })
     }
 }

crates/core/database/src/models/channels/ops.rs

@@ -1,7 +1,5 @@
 use crate::{revolt_result::Result, Channel, FieldsChannel, PartialChannel};
 use revolt_permissions::OverrideField;
-
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/channels/ops/mongodb.rs

@@ -184,7 +184,7 @@ impl AbstractChannels for MongoDb {
     async fn delete_channel(&self, channel: &Channel) -> Result<()> {
         let id = channel.id().to_string();
         let server_id = match channel {
-            Channel::TextChannel { server, .. } => {
+            Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => {
                 Some(server)
             }
             _ => None,

crates/core/database/src/models/channels/ops/reference.rs

@@ -94,6 +94,9 @@ impl AbstractChannels for ReferenceDb {
             match &mut channel {
                 Channel::TextChannel {
                     role_permissions, ..
+                }
+                | Channel::VoiceChannel {
+                    role_permissions, ..
                 } => {
                     if role_permissions.get(role_id).is_some() {
                         role_permissions.remove(role_id);

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

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::Emoji;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/file_hashes/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::FileHash;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/files/ops.rs

@@ -4,7 +4,6 @@ use crate::File;
 
 use super::FileUsedFor;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

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

@@ -14,6 +14,7 @@ use validator::Validate;
 
 use crate::{
     events::client::EventV1,
+    tasks::{self, ack::AckEvent},
     util::{
         bulk_permissions::BulkDatabasePermissionQuery, idempotency::IdempotencyKey,
         permissions::DatabasePermissionQuery,
@@ -21,9 +22,6 @@ use crate::{
     Channel, Database, Emoji, File, User, AMQP,
 };
 
-#[cfg(feature = "tasks")]
-use crate::tasks::{self, ack::AckEvent};
-
 auto_derived_partial!(
     /// Message
     pub struct Message {
@@ -114,11 +112,6 @@ auto_derived!(
         MessagePinned { id: String, by: String },
         #[serde(rename = "message_unpinned")]
         MessageUnpinned { id: String, by: String },
-        #[serde(rename = "call_started")]
-        CallStarted {
-            by: String,
-            finished_at: Option<Timestamp>,
-        },
     }
 
     /// Name and / or avatar override information
@@ -331,7 +324,9 @@ impl Message {
         }
 
         let server_id = match channel {
-            Channel::TextChannel { ref server, .. } => Some(server.clone()),
+            Channel::TextChannel { ref server, .. } | Channel::VoiceChannel { ref server, .. } => {
+                Some(server.clone())
+            }
             _ => None,
         };
 
@@ -443,8 +438,7 @@ impl Message {
         }
 
         // Verify replies are valid.
-        let mut replies = Vec::new();
-
+        let mut replies = HashSet::new();
         if let Some(entries) = data.replies {
             if entries.len() > config.features.limits.global.message_replies {
                 return Err(create_error!(TooManyReplies {
@@ -452,8 +446,6 @@ impl Message {
                 }));
             }
 
-            replies.reserve(entries.len());
-
             for ReplyIntent {
                 id,
                 mention,
@@ -467,12 +459,7 @@ impl Message {
                             user_mentions.insert(message.author.to_owned());
                         }
 
-                        // This is O(n^2), but this is faster than a HashSet
-                        // when n < 20; as long as the message_replies limit
-                        // is reasonable, this will be fast.
-                        if !replies.contains(&message.id) {
-                            replies.push(message.id);
-                        }
+                        replies.insert(message.id);
                     }
                     // If the referenced message doesn't exist and fail_if_not_exists
                     // is set to false, send the message without the reply.
@@ -489,7 +476,6 @@ impl Message {
 
         // Validate the mentions go to users in the channel/server
         if !user_mentions.is_empty() {
-            #[allow(deprecated)]
             match channel {
                 Channel::DirectMessage { ref recipients, .. }
                 | Channel::Group { ref recipients, .. } => {
@@ -497,7 +483,8 @@ impl Message {
                     user_mentions.retain(|m| recipients_hash.contains(m));
                     role_mentions.clear();
                 }
-                Channel::TextChannel { ref server, .. }=> {
+                Channel::TextChannel { ref server, .. }
+                | Channel::VoiceChannel { ref server, .. } => {
                     let mentions_vec = Vec::from_iter(user_mentions.iter().cloned());
 
                     let valid_members = db.fetch_members(server.as_str(), &mentions_vec[..]).await;
@@ -545,7 +532,9 @@ impl Message {
         }
 
         if !replies.is_empty() {
-            message.replies.replace(replies);
+            message
+                .replies
+                .replace(replies.into_iter().collect::<Vec<String>>());
         }
 
         // Calculate final message flags
@@ -627,11 +616,9 @@ impl Message {
             .await;
 
         // Update last_message_id
-        #[cfg(feature = "tasks")]
         tasks::last_message_id::queue(self.channel.to_string(), self.id.to_string(), is_dm).await;
 
         // Add mentions for affected users
-        #[cfg(feature = "tasks")]
         if !mentions_elsewhere {
             if let Some(mentions) = &self.mentions {
                 tasks::ack::queue_message(
@@ -650,7 +637,6 @@ impl Message {
         }
 
         // Generate embeds
-        #[cfg(feature = "tasks")]
         if generate_embeds {
             if let Some(content) = &self.content {
                 tasks::process_embeds::queue(
@@ -691,7 +677,6 @@ impl Message {
             && (self.mentions.is_some() || self.contains_mass_push_mention())
         {
             // send Push notifications
-            #[cfg(feature = "tasks")]
             tasks::ack::queue_message(
                 self.channel.to_string(),
                 AckEvent::ProcessMessage {
@@ -802,7 +787,7 @@ impl Message {
         query: MessageQuery,
         perspective: &User,
         include_users: Option<bool>,
-        server_id: Option<&str>,
+        server_id: Option<String>,
     ) -> Result<BulkMessageResponse> {
         let messages: Vec<v0::Message> = db
             .fetch_messages(query)
@@ -845,7 +830,6 @@ impl Message {
                             v0::SystemMessage::MessageUnpinned { by, .. } => {
                                 users.push(by.clone());
                             }
-                            v0::SystemMessage::CallStarted { by, .. } => users.push(by.clone()),
                         }
                     }
                     users
@@ -860,7 +844,7 @@ impl Message {
                 users,
                 members: if let Some(server_id) = server_id {
                     Some(
-                        db.fetch_members(server_id, &user_ids)
+                        db.fetch_members(&server_id, &user_ids)
                             .await?
                             .into_iter()
                             .map(Into::into)

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

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/messages/ops/reference.rs

@@ -247,7 +247,7 @@ impl AbstractMessages for ReferenceDb {
         let mut messages = self.messages.lock().await;
         if let Some(message) = messages.get_mut(id) {
             if let Some(users) = message.reactions.get_mut(emoji) {
-                users.swap_remove(&user.to_string());
+                users.remove(&user.to_string());
             }
 
             Ok(())
@@ -260,7 +260,7 @@ impl AbstractMessages for ReferenceDb {
     async fn clear_reaction(&self, id: &str, emoji: &str) -> Result<()> {
         let mut messages = self.messages.lock().await;
         if let Some(message) = messages.get_mut(id) {
-            message.reactions.swap_remove(emoji);
+            message.reactions.remove(emoji);
             Ok(())
         } else {
             Err(create_error!(NotFound))

crates/core/database/src/models/mod.rs

@@ -38,10 +38,7 @@ pub use servers::*;
 pub use user_settings::*;
 pub use users::*;
 
-use crate::{Database, ReferenceDb};
-
-#[cfg(feature = "mongodb")]
-use crate::MongoDb;
+use crate::{Database, MongoDb, ReferenceDb};
 
 pub trait AbstractDatabase:
     Sync
@@ -69,8 +66,6 @@ pub trait AbstractDatabase:
 }
 
 impl AbstractDatabase for ReferenceDb {}
-
-#[cfg(feature = "mongodb")]
 impl AbstractDatabase for MongoDb {}
 
 impl std::ops::Deref for Database {
@@ -79,7 +74,6 @@ impl std::ops::Deref for Database {
     fn deref(&self) -> &Self::Target {
         match &self {
             Database::Reference(dummy) => dummy,
-            #[cfg(feature = "mongodb")]
             Database::MongoDb(mongo) => mongo,
         }
     }

crates/core/database/src/models/policy_changes/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::PolicyChange;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/ratelimit_events/ops.rs

@@ -1,8 +1,6 @@
 use std::time::Duration;
 
 use crate::{revolt_result::Result, RatelimitEvent, RatelimitEventType};
-
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/safety_reports/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::Report;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/safety_snapshots/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::Snapshot;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/server_bans/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::{MemberCompositeKey, ServerBan};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

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

@@ -3,18 +3,10 @@ use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
 
 use crate::{
-    events::client::EventV1, util::permissions::DatabasePermissionQuery, Channel,
-    Database, File, Server, SystemMessage, User,
+    events::client::EventV1, util::permissions::DatabasePermissionQuery, Channel, Database, File,
+    Server, SystemMessage, User,
 };
 
-fn default_true() -> bool {
-    true
-}
-
-fn is_true(x: &bool) -> bool {
-    *x
-}
-
 auto_derived_partial!(
     /// Server Member
     pub struct Member {
@@ -38,24 +30,13 @@ auto_derived_partial!(
         /// Timestamp this member is timed out until
         #[serde(skip_serializing_if = "Option::is_none")]
         pub timeout: Option<Timestamp>,
-
-        /// Whether the member is server-wide voice muted
-        #[serde(skip_serializing_if = "is_true", default = "default_true")]
-        pub can_publish: bool,
-        /// Whether the member is server-wide voice deafened
-        #[serde(skip_serializing_if = "is_true", default = "default_true")]
-        pub can_receive: bool,
-        
-        // This value only exists in the database, not the models.
-        // If it is not-None, the database layer should return None to member fetching queries.
-        // pub pending_deletion_at: Option<Timestamp>
     },
     "PartialMember"
 );
 
 auto_derived!(
     /// Composite primary key consisting of server and user id
-    #[derive(Hash, Default, Eq)]
+    #[derive(Hash, Default)]
     pub struct MemberCompositeKey {
         /// Server Id
         pub server: String,
@@ -69,9 +50,6 @@ auto_derived!(
         Avatar,
         Roles,
         Timeout,
-        CanReceive,
-        CanPublish,
-        JoinedAt,
     }
 
     /// Member removal intention
@@ -91,8 +69,6 @@ impl Default for Member {
             avatar: None,
             roles: vec![],
             timeout: None,
-            can_publish: true,
-            can_receive: true,
         }
     }
 }
@@ -114,7 +90,7 @@ impl Member {
             return Err(create_error!(AlreadyInServer));
         }
 
-        let mut member = Member {
+        let member = Member {
             id: MemberCompositeKey {
                 server: server.id.to_string(),
                 user: user.id.to_string(),
@@ -122,9 +98,7 @@ impl Member {
             ..Default::default()
         };
 
-        if let Some(updated) = db.insert_or_merge_member(&member).await? {
-            member = updated;
-        }
+        db.insert_member(&member).await?;
 
         let should_fetch = channels.is_none();
         let mut channels = channels.unwrap_or_default();
@@ -147,20 +121,9 @@ impl Member {
 
         let emojis = db.fetch_emoji_by_parent_id(&server.id).await?;
 
-        #[allow(unused_mut)]
-        let mut voice_states = Vec::new();
-
-        #[cfg(feature = "voice")]
-        for channel in &channels {
-            if let Ok(Some(voice_state)) = crate::voice::get_channel_voice_state(channel).await {
-                voice_states.push(voice_state)
-            }
-        }
-
         EventV1::ServerMemberJoin {
             id: server.id.clone(),
             user: user.id.clone(),
-            member: member.clone().into(),
         }
         .p(server.id.clone())
         .await;
@@ -174,7 +137,6 @@ impl Member {
                 .map(|channel| channel.into())
                 .collect(),
             emojis: emojis.into_iter().map(|emoji| emoji.into()).collect(),
-            voice_states
         }
         .private(user.id.clone())
         .await;
@@ -224,13 +186,10 @@ impl Member {
 
     pub fn remove_field(&mut self, field: &FieldsMember) {
         match field {
-            FieldsMember::JoinedAt => (),
             FieldsMember::Avatar => self.avatar = None,
             FieldsMember::Nickname => self.nickname = None,
             FieldsMember::Roles => self.roles.clear(),
             FieldsMember::Timeout => self.timeout = None,
-            FieldsMember::CanReceive => self.can_receive = true,
-            FieldsMember::CanPublish => self.can_publish = true,
         }
     }
 
@@ -265,7 +224,7 @@ impl Member {
         intention: RemovalIntention,
         silent: bool,
     ) -> Result<()> {
-        db.soft_delete_member(&self.id).await?;
+        db.delete_member(&self.id).await?;
 
         EventV1::ServerMemberLeave {
             id: self.id.server.to_string(),
@@ -301,74 +260,3 @@ impl Member {
         Ok(())
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use iso8601_timestamp::{Duration, Timestamp};
-    use revolt_models::v0::DataCreateServer;
-
-    use crate::{Member, PartialMember, RemovalIntention, Server, User};
-
-    #[async_std::test]
-    async fn muted_member_rejoin() {
-        database_test!(|db| async move {
-            match db {
-                crate::Database::Reference(_) => return,
-                crate::Database::MongoDb(_) => (),
-            }
-            let owner = User::create(&db, "Server Owner".to_string(), None, None)
-                .await
-                .unwrap();
-
-            let kickable_user = User::create(&db, "Member".to_string(), None, None)
-                .await
-                .unwrap();
-
-            let server = Server::create(
-                &db,
-                DataCreateServer {
-                    name: "Server".to_string(),
-                    description: None,
-                    nsfw: None,
-                },
-                &owner,
-                false,
-            )
-            .await
-            .unwrap()
-            .0;
-
-            Member::create(&db, &server, &owner, None).await.unwrap();
-            let mut kickable_member = Member::create(&db, &server, &kickable_user, None)
-                .await
-                .unwrap()
-                .0;
-
-            kickable_member
-                .update(
-                    &db,
-                    PartialMember {
-                        timeout: Some(Timestamp::now_utc() + Duration::minutes(5)),
-                        ..Default::default()
-                    },
-                    vec![],
-                )
-                .await
-                .unwrap();
-
-            assert!(kickable_member.in_timeout());
-
-            kickable_member
-                .remove(&db, &server, RemovalIntention::Kick, false)
-                .await
-                .unwrap();
-
-            let kickable_member = Member::create(&db, &server, &kickable_user, None)
-                .await
-                .unwrap()
-                .0;
-
-            assert!(kickable_member.in_timeout())
-        });
-    }
-}

crates/core/database/src/models/server_members/ops.rs

@@ -1,20 +1,16 @@
-#[cfg(feature = "mongodb")]
-use ::mongodb::{ClientSession, SessionCursor};
-
+use ::mongodb::SessionCursor;
 use revolt_result::Result;
 
 use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 
 #[derive(Debug)]
-#[allow(clippy::large_enum_variant)]
 pub enum ChunkedServerMembersGenerator {
     #[cfg(feature = "mongodb")]
     MongoDb {
-        session: ClientSession,
+        session: ::mongodb::ClientSession,
         cursor: Option<SessionCursor<Member>>,
     },
 
@@ -26,7 +22,7 @@ pub enum ChunkedServerMembersGenerator {
 
 impl ChunkedServerMembersGenerator {
     #[cfg(feature = "mongodb")]
-    pub fn new_mongo(session: ClientSession, cursor: SessionCursor<Member>) -> Self {
+    pub fn new_mongo(session: ::mongodb::ClientSession, cursor: SessionCursor<Member>) -> Self {
         ChunkedServerMembersGenerator::MongoDb {
             session,
             cursor: Some(cursor),
@@ -73,13 +69,13 @@ impl ChunkedServerMembersGenerator {
 #[async_trait]
 pub trait AbstractServerMembers: Sync + Send {
     /// Insert a new server member into the database
-    async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>>;
+    async fn insert_member(&self, member: &Member) -> Result<()>;
 
     /// Fetch a server member by their id
     async fn fetch_member(&self, server_id: &str, user_id: &str) -> Result<Member>;
 
     /// Fetch all members in a server
-    async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>>;
+    async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>>;
 
     /// Fetch all members in a server as an iterator
     async fn fetch_all_members_chunked(
@@ -100,10 +96,10 @@ pub trait AbstractServerMembers: Sync + Send {
     ) -> Result<ChunkedServerMembersGenerator>;
 
     /// Fetch all memberships for a user
-    async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>>;
+    async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>>;
 
     /// Fetch multiple members by their ids
-    async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>>;
+    async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>>;
 
     /// Fetch member count of a server
     async fn fetch_member_count(&self, server_id: &str) -> Result<usize>;
@@ -119,14 +115,6 @@ pub trait AbstractServerMembers: Sync + Send {
         remove: Vec<FieldsMember>,
     ) -> Result<()>;
 
-    /// Marks a user as no longer a member of a server, while retaining the database value.
-    /// This is used to keep information such as timeouts in place, but will remove information such as join date and applied roles.
-    async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
-
-    /// Forcibly delete a server member by their id.
-    /// This will cancel any pending timeouts or other longer term actions, and they will not be reapplied on rejoin.
-    async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
-
-    /// Fetch all members who have been marked for deletion.
-    async fn remove_dangling_members(&self) -> Result<()>;
+    /// Delete a server member by their id
+    async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
 }

crates/core/database/src/models/server_members/ops/mongodb.rs

@@ -1,6 +1,4 @@
-use bson::Document;
 use futures::StreamExt;
-use iso8601_timestamp::Timestamp;
 use mongodb::options::ReadConcern;
 use revolt_result::Result;
 
@@ -13,42 +11,9 @@ static COL: &str = "server_members";
 
 #[async_trait]
 impl AbstractServerMembers for MongoDb {
-    /// Insert a new server member (or use the existing member if one is found)
-    async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>> {
-        let existing: Result<Option<Document>> = query!(
-            self,
-            find_one,
-            COL,
-            doc! {
-                "_id.server": &member.id.server,
-                "_id.user": &member.id.user,
-                "pending_deletion_at": {"$exists": true}
-            }
-        );
-        // Update the existing record if it exist, otherwise make a new record
-        if existing.is_ok_and(|x| x.is_some()) {
-            self.col::<Member>(COL)
-                .find_one_and_update(
-                    doc! {
-                        "_id.server": &member.id.server,
-                        "_id.user": &member.id.user,
-                    },
-                    doc! {
-                        "$set": {
-                            "joined_at": member.joined_at.duration_since(Timestamp::UNIX_EPOCH).whole_seconds(),
-                        },
-                        "$unset": {
-                            "pending_deletion_at": ""
-                        }
-                    },
-                )
-                .return_document(mongodb::options::ReturnDocument::After)
-                .await
-                .map_err(|_| create_database_error!("update_one", COL))
-        } else {
-            query!(self, insert_one, COL, &member).map(|_| ())?;
-            Ok(None)
-        }
+    /// Insert a new server member into the database
+    async fn insert_member(&self, member: &Member) -> Result<()> {
+        query!(self, insert_one, COL, &member).map(|_| ())
     }
 
     /// Fetch a server member by their id
@@ -59,20 +24,18 @@ impl AbstractServerMembers for MongoDb {
             COL,
             doc! {
                 "_id.server": server_id,
-                "_id.user": user_id,
-                "pending_deletion_at": {"$exists": false}
+                "_id.user": user_id
             }
         )?
         .ok_or_else(|| create_error!(NotFound))
     }
 
     /// Fetch all members in a server
-    async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>> {
         Ok(self
             .col::<Member>(COL)
             .find(doc! {
-                "_id.server": server_id,
-                "pending_deletion_at": {"$exists": false}
+                "_id.server": server_id
             })
             .await
             .map_err(|_| create_database_error!("find", COL))?
@@ -176,12 +139,11 @@ impl AbstractServerMembers for MongoDb {
     }
 
     /// Fetch all memberships for a user
-    async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> {
         Ok(self
             .col::<Member>(COL)
             .find(doc! {
-                "_id.user": user_id,
-                "pending_deletion_at": {"$exists": false}
+                "_id.user": user_id
             })
             .await
             .map_err(|_| create_database_error!("find", COL))?
@@ -197,12 +159,11 @@ impl AbstractServerMembers for MongoDb {
     }
 
     /// Fetch multiple members by their ids
-    async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>> {
+    async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>> {
         Ok(self
             .col::<Member>(COL)
             .find(doc! {
                 "_id.server": server_id,
-                "pending_deletion_at": {"$exists": false},
                 "_id.user": {
                     "$in": ids
                 }
@@ -224,8 +185,7 @@ impl AbstractServerMembers for MongoDb {
     async fn fetch_member_count(&self, server_id: &str) -> Result<usize> {
         self.col::<Member>(COL)
             .count_documents(doc! {
-                "_id.server": server_id,
-                "pending_deletion_at": {"$exists": false}
+                "_id.server": server_id
             })
             .await
             .map(|c| c as usize)
@@ -236,8 +196,7 @@ impl AbstractServerMembers for MongoDb {
     async fn fetch_server_count(&self, user_id: &str) -> Result<usize> {
         self.col::<Member>(COL)
             .count_documents(doc! {
-                "_id.user": user_id,
-                "pending_deletion_at": {"$exists": false}
+                "_id.user": user_id
             })
             .await
             .map(|c| c as usize)
@@ -266,42 +225,8 @@ impl AbstractServerMembers for MongoDb {
         .map(|_| ())
     }
 
-    /// Marks a member for deletion.
-    /// This will remove the record if the user has no pending actions (eg. timeout),
-    /// otherwise will slate the record for deletion by revolt_crond once the actions expire.
-    async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
-        let member = self.fetch_member(&id.server, &id.user).await;
-        if let Ok(member) = member {
-            if member.in_timeout() {
-                self.col::<Document>(COL)
-                    .update_many(
-                        doc! {
-                            "_id.server": &id.server,
-                            "_id.user": &id.user,
-                        },
-                        doc! {
-                            "$set": {"pending_deletion_at": format!("{}", member.timeout.unwrap().format())},
-                            "$unset": {
-                                "joined_at": "",
-                                "avatar": "",
-                                "nickname": "",
-                                "roles": ""
-                            }
-                        },
-                    )
-                    .await
-                    .map(|_| ())
-                    .map_err(|_| create_database_error!("update_many", COL))
-            } else {
-                self.force_delete_member(id).await
-            }
-        } else {
-            Err(create_database_error!("fetch_member", COL))
-        }
-    }
-
     /// Delete a server member by their id
-    async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
+    async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
         query!(
             self,
             delete_one,
@@ -313,31 +238,15 @@ impl AbstractServerMembers for MongoDb {
         )
         .map(|_| ())
     }
-
-    async fn remove_dangling_members(&self) -> Result<()> {
-        let now = Timestamp::now_utc();
-        let date = bson::to_bson(&now).expect("Failed to serialize timestamp");
-
-        self.col::<Document>(COL)
-            .delete_many(doc! {
-                "pending_deletion_at": {"$lt": date}
-            })
-            .await
-            .map(|_| ())
-            .map_err(|_| create_database_error!("count_documents", COL))
-    }
 }
 
 impl IntoDocumentPath for FieldsMember {
     fn as_path(&self) -> Option<&'static str> {
         Some(match self {
-            FieldsMember::JoinedAt => "joined_at",
             FieldsMember::Avatar => "avatar",
             FieldsMember::Nickname => "nickname",
             FieldsMember::Roles => "roles",
             FieldsMember::Timeout => "timeout",
-            FieldsMember::CanPublish => "can_publish",
-            FieldsMember::CanReceive => "can_receive",
         })
     }
 }

crates/core/database/src/models/server_members/ops/reference.rs

@@ -8,13 +8,13 @@ use super::{AbstractServerMembers, ChunkedServerMembersGenerator};
 #[async_trait]
 impl AbstractServerMembers for ReferenceDb {
     /// Insert a new server member into the database
-    async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>> {
+    async fn insert_member(&self, member: &Member) -> Result<()> {
         let mut server_members = self.server_members.lock().await;
         if server_members.contains_key(&member.id) {
             Err(create_database_error!("insert", "member"))
         } else {
             server_members.insert(member.id.clone(), member.clone());
-            Ok(None)
+            Ok(())
         }
     }
 
@@ -31,7 +31,7 @@ impl AbstractServerMembers for ReferenceDb {
     }
 
     /// Fetch all members in a server
-    async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>> {
         let server_members = self.server_members.lock().await;
         Ok(server_members
             .values()
@@ -105,7 +105,7 @@ impl AbstractServerMembers for ReferenceDb {
     }
 
     /// Fetch all memberships for a user
-    async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> {
         let server_members = self.server_members.lock().await;
         Ok(server_members
             .values()
@@ -115,7 +115,7 @@ impl AbstractServerMembers for ReferenceDb {
     }
 
     /// Fetch multiple members by their ids
-    async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>> {
+    async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>> {
         let server_members = self.server_members.lock().await;
         Ok(ids
             .iter()
@@ -169,26 +169,8 @@ impl AbstractServerMembers for ReferenceDb {
         }
     }
 
-    /// Soft delete a member
-    async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
-        let mut server_members = self.server_members.lock().await;
-
-        let member = server_members.get_mut(id);
-        if let Some(member) = member {
-            if member.in_timeout() {
-                panic!("Soft deletion is not implemented.")
-            } else if server_members.remove(id).is_some() {
-                Ok(())
-            } else {
-                Err(create_error!(NotFound))
-            }
-        } else {
-            Err(create_error!(NotFound))
-        }
-    }
-
     /// Delete a server member by their id
-    async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
+    async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
         let mut server_members = self.server_members.lock().await;
         if server_members.remove(id).is_some() {
             Ok(())
@@ -196,8 +178,4 @@ impl AbstractServerMembers for ReferenceDb {
             Err(create_error!(NotFound))
         }
     }
-
-    async fn remove_dangling_members(&self) -> Result<()> {
-        todo!()
-    }
 }

crates/core/database/src/models/servers/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::{FieldsRole, FieldsServer, PartialRole, PartialServer, Role, Server};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/user_settings/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::UserSettings;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/users/axum.rs

@@ -1,19 +1,14 @@
-use axum::{extract::{FromRef, FromRequestParts}, http::request::Parts};
+use axum::{extract::FromRequestParts, http::request::Parts};
 
 use revolt_result::{create_error, Error, Result};
 
 use crate::{Database, User};
 
-impl<S> FromRequestParts<S> for User
-where
-    Database: FromRef<S>,
-    S: Send + Sync
-{
+#[async_trait::async_trait]
+impl FromRequestParts<Database> for User {
     type Rejection = Error;
 
-    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<User> {
-        let db = Database::from_ref(state);
-
+    async fn from_request_parts(parts: &mut Parts, db: &Database) -> Result<User> {
         if let Some(Ok(bot_token)) = parts.headers.get("x-bot-token").map(|v| v.to_str()) {
             let bot = db.fetch_bot_by_token(bot_token).await?;
             db.fetch_user(&bot.id).await

crates/core/database/src/models/users/mod.rs

@@ -4,6 +4,8 @@ mod model;
 mod ops;
 #[cfg(feature = "rocket-impl")]
 mod rocket;
+#[cfg(feature = "rocket-impl")]
+mod schema;
 
 pub use model::*;
 pub use ops::*;

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

@@ -15,7 +15,7 @@ use serde_json::json;
 use ulid::Ulid;
 
 auto_derived_partial!(
-    /// User
+    /// # User
     pub struct User {
         /// Unique Id
         #[serde(rename = "_id")]

crates/core/database/src/models/users/ops.rs

@@ -4,7 +4,6 @@ use revolt_result::Result;
 
 use crate::{FieldsUser, PartialUser, RelationshipStatus, User};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/users/ops/mongodb.rs

@@ -212,34 +212,16 @@ impl AbstractUsers for MongoDb {
         partial: &PartialUser,
         remove: Vec<FieldsUser>,
     ) -> Result<()> {
-        if remove.contains(&FieldsUser::StatusText) && partial.status.is_some() {
-            // stupid-ass workaround to fix mongo conflicting the same item
-            let _: Result<()> = query!(
-                self,
-                update_one_by_id,
-                COL,
-                id,
-                PartialUser {
-                    ..Default::default()
-                },
-                remove.iter().map(|x| x as &dyn IntoDocumentPath).collect(),
-                None
-            )
-            .map(|_| ());
-
-            query!(self, update_one_by_id, COL, id, partial, vec![], None).map(|_| ())
-        } else {
-            query!(
-                self,
-                update_one_by_id,
-                COL,
-                id,
-                partial,
-                remove.iter().map(|x| x as &dyn IntoDocumentPath).collect(),
-                None
-            )
-            .map(|_| ())
-        }
+        query!(
+            self,
+            update_one_by_id,
+            COL,
+            id,
+            partial,
+            remove.iter().map(|x| x as &dyn IntoDocumentPath).collect(),
+            None
+        )
+        .map(|_| ())
     }
 
     /// Set relationship with another user

crates/core/database/src/models/users/schema.rs

@@ -0,0 +1,31 @@
+use revolt_okapi::openapi3::{SecurityScheme, SecuritySchemeData};
+use revolt_rocket_okapi::{
+    gen::OpenApiGenerator,
+    request::{OpenApiFromRequest, RequestHeaderInput},
+};
+
+use crate::User;
+
+impl OpenApiFromRequest<'_> for User {
+    fn from_request_input(
+        _gen: &mut OpenApiGenerator,
+        _name: String,
+        _required: bool,
+    ) -> revolt_rocket_okapi::Result<RequestHeaderInput> {
+        let mut requirements = schemars::Map::new();
+        requirements.insert("Session Token".to_owned(), vec![]);
+
+        Ok(RequestHeaderInput::Security(
+            "Session Token".to_owned(),
+            SecurityScheme {
+                data: SecuritySchemeData::ApiKey {
+                    name: "x-session-token".to_owned(),
+                    location: "header".to_owned(),
+                },
+                description: Some("Used to authenticate as a user.".to_owned()),
+                extensions: schemars::Map::new(),
+            },
+            requirements,
+        ))
+    }
+}

crates/core/database/src/tasks/ack.rs

@@ -14,10 +14,10 @@ use validator::HasLen;
 use revolt_result::Result;
 
 use super::DelayedTask;
-use crate::Channel::TextChannel;
+use crate::Channel::{TextChannel, VoiceChannel};
 
 /// Enumeration of possible events
-#[derive(Debug, PartialEq)]
+#[derive(Debug, Eq, PartialEq)]
 pub enum AckEvent {
     /// Add mentions for a channel
     ProcessMessage {
@@ -191,14 +191,17 @@ pub async fn handle_ack_event(
                     .await
                     .expect("Failed to fetch channel from db");
 
-                if let TextChannel { server, .. } = channel {
-                    if let Err(err) =
-                        amqp.mass_mention_message_sent(server, mass_mentions).await
-                    {
-                        revolt_config::capture_error(&err);
+                match channel {
+                    TextChannel { server, .. } | VoiceChannel { server, .. } => {
+                        if let Err(err) =
+                            amqp.mass_mention_message_sent(server, mass_mentions).await
+                        {
+                            revolt_config::capture_error(&err);
+                        }
+                    }
+                    _ => {
+                        panic!("Unknown channel type when sending mass mention event");
                     }
-                } else {
-                    panic!("Unknown channel type when sending mass mention event");
                 }
             }
         }

crates/core/database/src/util/bridge/v0.rs

@@ -143,7 +143,6 @@ impl From<crate::FieldsWebhook> for FieldsWebhook {
 }
 
 impl From<crate::Channel> for Channel {
-    #[allow(deprecated)]
     fn from(value: crate::Channel) -> Self {
         match value {
             crate::Channel::SavedMessages { id, user } => Channel::SavedMessages { id, user },
@@ -189,7 +188,6 @@ impl From<crate::Channel> for Channel {
                 default_permissions,
                 role_permissions,
                 nsfw,
-                voice,
             } => Channel::TextChannel {
                 id,
                 server,
@@ -200,14 +198,31 @@ impl From<crate::Channel> for Channel {
                 default_permissions,
                 role_permissions,
                 nsfw,
-                voice: voice.map(|voice| voice.into()),
+            },
+            crate::Channel::VoiceChannel {
+                id,
+                server,
+                name,
+                description,
+                icon,
+                default_permissions,
+                role_permissions,
+                nsfw,
+            } => Channel::VoiceChannel {
+                id,
+                server,
+                name,
+                description,
+                icon: icon.map(|file| file.into()),
+                default_permissions,
+                role_permissions,
+                nsfw,
             },
         }
     }
 }
 
 impl From<Channel> for crate::Channel {
-    #[allow(deprecated)]
     fn from(value: Channel) -> crate::Channel {
         match value {
             Channel::SavedMessages { id, user } => crate::Channel::SavedMessages { id, user },
@@ -253,7 +268,6 @@ impl From<Channel> for crate::Channel {
                 default_permissions,
                 role_permissions,
                 nsfw,
-                voice,
             } => crate::Channel::TextChannel {
                 id,
                 server,
@@ -264,7 +278,25 @@ impl From<Channel> for crate::Channel {
                 default_permissions,
                 role_permissions,
                 nsfw,
-                voice: voice.map(|voice| voice.into()),
+            },
+            Channel::VoiceChannel {
+                id,
+                server,
+                name,
+                description,
+                icon,
+                default_permissions,
+                role_permissions,
+                nsfw,
+            } => crate::Channel::VoiceChannel {
+                id,
+                server,
+                name,
+                description,
+                icon: icon.map(|file| file.into()),
+                default_permissions,
+                role_permissions,
+                nsfw,
             },
         }
     }
@@ -283,7 +315,6 @@ impl From<crate::PartialChannel> for PartialChannel {
             role_permissions: value.role_permissions,
             default_permissions: value.default_permissions,
             last_message_id: value.last_message_id,
-            voice: value.voice.map(|voice| voice.into())
         }
     }
 }
@@ -301,7 +332,6 @@ impl From<PartialChannel> for crate::PartialChannel {
             role_permissions: value.role_permissions,
             default_permissions: value.default_permissions,
             last_message_id: value.last_message_id,
-            voice: value.voice.map(|voice| voice.into())
         }
     }
 }
@@ -312,7 +342,6 @@ impl From<FieldsChannel> for crate::FieldsChannel {
             FieldsChannel::Description => crate::FieldsChannel::Description,
             FieldsChannel::Icon => crate::FieldsChannel::Icon,
             FieldsChannel::DefaultPermissions => crate::FieldsChannel::DefaultPermissions,
-            FieldsChannel::Voice => crate::FieldsChannel::Voice,
         }
     }
 }
@@ -323,7 +352,6 @@ impl From<crate::FieldsChannel> for FieldsChannel {
             crate::FieldsChannel::Description => FieldsChannel::Description,
             crate::FieldsChannel::Icon => FieldsChannel::Icon,
             crate::FieldsChannel::DefaultPermissions => FieldsChannel::DefaultPermissions,
-            crate::FieldsChannel::Voice => FieldsChannel::Voice,
         }
     }
 }
@@ -515,7 +543,6 @@ impl From<crate::SystemMessage> for SystemMessage {
             crate::SystemMessage::UserRemove { id, by } => Self::UserRemove { id, by },
             crate::SystemMessage::MessagePinned { id, by } => Self::MessagePinned { id, by },
             crate::SystemMessage::MessageUnpinned { id, by } => Self::MessageUnpinned { id, by },
-            crate::SystemMessage::CallStarted { by, finished_at } => Self::CallStarted { by, finished_at }
         }
     }
 }
@@ -612,8 +639,6 @@ impl From<crate::Member> for Member {
             avatar: value.avatar.map(|f| f.into()),
             roles: value.roles,
             timeout: value.timeout,
-            can_publish: value.can_publish,
-            can_receive: value.can_receive,
         }
     }
 }
@@ -627,8 +652,6 @@ impl From<Member> for crate::Member {
             avatar: value.avatar.map(|f| f.into()),
             roles: value.roles,
             timeout: value.timeout,
-            can_publish: value.can_publish,
-            can_receive: value.can_receive,
         }
     }
 }
@@ -642,8 +665,6 @@ impl From<crate::PartialMember> for PartialMember {
             avatar: value.avatar.map(|f| f.into()),
             roles: value.roles,
             timeout: value.timeout,
-            can_publish: value.can_publish,
-            can_receive: value.can_receive,
         }
     }
 }
@@ -657,8 +678,6 @@ impl From<PartialMember> for crate::PartialMember {
             avatar: value.avatar.map(|f| f.into()),
             roles: value.roles,
             timeout: value.timeout,
-            can_publish: value.can_publish,
-            can_receive: value.can_receive,
         }
     }
 }
@@ -688,9 +707,6 @@ impl From<crate::FieldsMember> for FieldsMember {
             crate::FieldsMember::Nickname => FieldsMember::Nickname,
             crate::FieldsMember::Roles => FieldsMember::Roles,
             crate::FieldsMember::Timeout => FieldsMember::Timeout,
-            crate::FieldsMember::CanReceive => FieldsMember::CanReceive,
-            crate::FieldsMember::CanPublish => FieldsMember::CanPublish,
-            crate::FieldsMember::JoinedAt => FieldsMember::JoinedAt,
         }
     }
 }
@@ -702,9 +718,6 @@ impl From<FieldsMember> for crate::FieldsMember {
             FieldsMember::Nickname => crate::FieldsMember::Nickname,
             FieldsMember::Roles => crate::FieldsMember::Roles,
             FieldsMember::Timeout => crate::FieldsMember::Timeout,
-            FieldsMember::CanReceive => crate::FieldsMember::CanReceive,
-            FieldsMember::CanPublish => crate::FieldsMember::CanPublish,
-            FieldsMember::JoinedAt => crate::FieldsMember::JoinedAt,
         }
     }
 }
@@ -1372,19 +1385,3 @@ impl From<FieldsMessage> for crate::FieldsMessage {
         }
     }
 }
-
-impl From<VoiceInformation> for crate::VoiceInformation {
-    fn from(value: VoiceInformation) -> Self {
-        crate::VoiceInformation {
-            max_users: value.max_users
-        }
-    }
-}
-
-impl From<crate::VoiceInformation> for VoiceInformation {
-    fn from(value: crate::VoiceInformation) -> Self {
-        VoiceInformation {
-            max_users: value.max_users
-        }
-    }
-}

crates/core/database/src/util/bulk_permissions.rs

@@ -144,6 +144,10 @@ impl<'z> BulkDatabasePermissionQuery<'z> {
                 Channel::TextChannel {
                     default_permissions,
                     ..
+                }
+                | Channel::VoiceChannel {
+                    default_permissions,
+                    ..
                 } => default_permissions.unwrap_or_default().into(),
                 _ => Default::default(),
             }
@@ -152,14 +156,16 @@ impl<'z> BulkDatabasePermissionQuery<'z> {
         }
     }
 
-    #[allow(dead_code, deprecated)]
+    #[allow(dead_code)]
     fn get_channel_type(&mut self) -> ChannelType {
         if let Some(channel) = &self.channel {
             match channel {
                 Channel::DirectMessage { .. } => ChannelType::DirectMessage,
                 Channel::Group { .. } => ChannelType::Group,
                 Channel::SavedMessages { .. } => ChannelType::SavedMessages,
-                Channel::TextChannel { .. } => ChannelType::ServerChannel,
+                Channel::TextChannel { .. } | Channel::VoiceChannel { .. } => {
+                    ChannelType::ServerChannel
+                }
             }
         } else {
             ChannelType::Unknown
@@ -173,6 +179,9 @@ impl<'z> BulkDatabasePermissionQuery<'z> {
             match channel {
                 Channel::TextChannel {
                     role_permissions, ..
+                }
+                | Channel::VoiceChannel {
+                    role_permissions, ..
                 } => role_permissions,
                 _ => panic!("Not supported for non-server channels"),
             }
@@ -199,6 +208,12 @@ async fn calculate_members_permissions<'a>(
             role_permissions,
             default_permissions,
             ..
+        }
+        | Channel::VoiceChannel {
+            id,
+            role_permissions,
+            default_permissions,
+            ..
         } => (id, role_permissions, default_permissions),
         _ => panic!("Calculation of member permissions must be done on a server channel"),
     };

crates/core/database/src/util/funcs.rs

@@ -1,24 +0,0 @@
-use crate::Database;
-use revolt_result::Result;
-
-/// Formats a user's name depending on their optional features and location.
-/// Factors in server display names and user display names before falling back to username#discriminator.
-/// Passing a server in which the user is not a member will result in an Err.
-pub async fn format_display_name(
-    db: &Database,
-    user_id: &str,
-    server_id: Option<&str>,
-) -> Result<String> {
-    if let Some(server_id) = server_id {
-        let member = db.fetch_member(server_id, user_id).await?;
-        if let Some(nick) = member.nickname {
-            return Ok(nick);
-        }
-    }
-
-    let user = db.fetch_user(user_id).await?;
-    if let Some(display) = user.display_name {
-        return Ok(display);
-    }
-    Ok(format!("{}#{}", user.username, user.discriminator))
-}

crates/core/database/src/util/idempotency.rs

@@ -10,16 +10,16 @@ use once_cell::sync::Lazy;
 use serde::{Deserialize, Serialize};
 
 #[derive(Serialize, Deserialize)]
-#[cfg_attr(feature = "utoipa", derive(IntoParams))]
-#[cfg_attr(feature = "utoipa", into_params(names("Idempotency-Key"), parameter_in=Header))]
-pub struct IdempotencyKey(String);
+pub struct IdempotencyKey {
+    key: String,
+}
 
 static TOKEN_CACHE: Lazy<Mutex<lru::LruCache<String, ()>>> =
     Lazy::new(|| Mutex::new(lru::LruCache::new(NonZeroUsize::new(1000).unwrap())));
 
 impl IdempotencyKey {
     pub fn unchecked_from_string(key: String) -> Self {
-        Self(key)
+        Self { key }
     }
 
     // Backwards compatibility.
@@ -32,56 +32,56 @@ impl IdempotencyKey {
             }
 
             cache.put(v.clone(), ());
-            self.0 = v;
+            self.key = v;
         }
 
         Ok(())
     }
 
     pub fn into_key(self) -> String {
-        self.0
+        self.key
     }
 }
 
-// #[cfg(feature = "rocket-impl")]
-// use revolt_rocket_okapi::{
-//     gen::OpenApiGenerator,
-//     request::{OpenApiFromRequest, RequestHeaderInput},
-//     revolt_okapi::openapi3::{Parameter, ParameterValue},
-// };
+#[cfg(feature = "rocket-impl")]
+use revolt_rocket_okapi::{
+    gen::OpenApiGenerator,
+    request::{OpenApiFromRequest, RequestHeaderInput},
+    revolt_okapi::openapi3::{Parameter, ParameterValue},
+};
 
-// #[cfg(feature = "rocket-impl")]
-// use schemars::schema::{InstanceType, SchemaObject, SingleOrVec};
+#[cfg(feature = "rocket-impl")]
+use schemars::schema::{InstanceType, SchemaObject, SingleOrVec};
 
-// #[cfg(feature = "rocket-impl")]
-// impl OpenApiFromRequest<'_> for IdempotencyKey {
-//     fn from_request_input(
-//         _gen: &mut OpenApiGenerator,
-//         _name: String,
-//         _required: bool,
-//     ) -> revolt_rocket_okapi::Result<RequestHeaderInput> {
-//         Ok(RequestHeaderInput::Parameter(Parameter {
-//             name: "Idempotency-Key".to_string(),
-//             description: Some("Unique key to prevent duplicate requests".to_string()),
-//             allow_empty_value: false,
-//             required: false,
-//             deprecated: false,
-//             extensions: schemars::Map::new(),
-//             location: "header".to_string(),
-//             value: ParameterValue::Schema {
-//                 allow_reserved: false,
-//                 example: None,
-//                 examples: None,
-//                 explode: None,
-//                 style: None,
-//                 schema: SchemaObject {
-//                     instance_type: Some(SingleOrVec::Single(Box::new(InstanceType::String))),
-//                     ..Default::default()
-//                 },
-//             },
-//         }))
-//     }
-// }
+#[cfg(feature = "rocket-impl")]
+impl OpenApiFromRequest<'_> for IdempotencyKey {
+    fn from_request_input(
+        _gen: &mut OpenApiGenerator,
+        _name: String,
+        _required: bool,
+    ) -> revolt_rocket_okapi::Result<RequestHeaderInput> {
+        Ok(RequestHeaderInput::Parameter(Parameter {
+            name: "Idempotency-Key".to_string(),
+            description: Some("Unique key to prevent duplicate requests".to_string()),
+            allow_empty_value: false,
+            required: false,
+            deprecated: false,
+            extensions: schemars::Map::new(),
+            location: "header".to_string(),
+            value: ParameterValue::Schema {
+                allow_reserved: false,
+                example: None,
+                examples: None,
+                explode: None,
+                style: None,
+                schema: SchemaObject {
+                    instance_type: Some(SingleOrVec::Single(Box::new(InstanceType::String))),
+                    ..Default::default()
+                },
+            },
+        }))
+    }
+}
 
 #[cfg(feature = "rocket-impl")]
 use rocket::{
@@ -110,16 +110,18 @@ impl<'r> FromRequest<'r> for IdempotencyKey {
                 ));
             }
 
-            let idempotency = IdempotencyKey(key);
+            let idempotency = IdempotencyKey { key };
             let mut cache = TOKEN_CACHE.lock().await;
-            if cache.get(&idempotency.0).is_some() {
+            if cache.get(&idempotency.key).is_some() {
                 return Outcome::Error((Status::Conflict, create_error!(DuplicateNonce)));
             }
 
-            cache.put(idempotency.0.clone(), ());
+            cache.put(idempotency.key.clone(), ());
             return Outcome::Success(idempotency);
         }
 
-        Outcome::Success(IdempotencyKey(ulid::Ulid::new().to_string()))
+        Outcome::Success(IdempotencyKey {
+            key: ulid::Ulid::new().to_string(),
+        })
     }
 }

crates/core/database/src/util/mod.rs

@@ -1,11 +1,6 @@
 pub mod bridge;
 pub mod bulk_permissions;
-mod funcs;
 pub mod idempotency;
 pub mod permissions;
 pub mod reference;
 pub mod test_fixtures;
-#[cfg(feature = "utoipa")]
-pub mod utoipa;
-
-pub use funcs::*;

crates/core/database/src/util/permissions.rs

@@ -185,26 +185,9 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
         }
     }
 
-    async fn do_we_have_publish_overwrites(&mut self) -> bool {
-        if let Some(member) = &self.member {
-            member.can_publish
-        } else {
-            true
-        }
-    }
-
-    async fn do_we_have_receive_overwrites(&mut self) -> bool {
-        if let Some(member) = &self.member {
-            member.can_receive
-        } else {
-            true
-        }
-    }
-
     // * For calculating channel permission
 
     /// Get the type of the channel
-    #[allow(deprecated)]
     async fn get_channel_type(&mut self) -> ChannelType {
         if let Some(channel) = &self.channel {
             match channel {
@@ -216,7 +199,9 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
                 Cow::Borrowed(Channel::SavedMessages { .. })
                 | Cow::Owned(Channel::SavedMessages { .. }) => ChannelType::SavedMessages,
                 Cow::Borrowed(Channel::TextChannel { .. })
-                | Cow::Owned(Channel::TextChannel { .. }) => ChannelType::ServerChannel,
+                | Cow::Owned(Channel::TextChannel { .. })
+                | Cow::Borrowed(Channel::VoiceChannel { .. })
+                | Cow::Owned(Channel::VoiceChannel { .. }) => ChannelType::ServerChannel,
             }
         } else {
             ChannelType::Unknown
@@ -240,6 +225,14 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
                 | Cow::Owned(Channel::TextChannel {
                     default_permissions,
                     ..
+                })
+                | Cow::Borrowed(Channel::VoiceChannel {
+                    default_permissions,
+                    ..
+                })
+                | Cow::Owned(Channel::VoiceChannel {
+                    default_permissions,
+                    ..
                 }) => default_permissions.unwrap_or_default().into(),
                 _ => Default::default(),
             }
@@ -257,6 +250,12 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
                 })
                 | Cow::Owned(Channel::TextChannel {
                     role_permissions, ..
+                })
+                | Cow::Borrowed(Channel::VoiceChannel {
+                    role_permissions, ..
+                })
+                | Cow::Owned(Channel::VoiceChannel {
+                    role_permissions, ..
                 }) => {
                     if let Some(server) = &self.server {
                         let member_roles = self
@@ -344,10 +343,11 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
     /// (this will only ever be called for server channels, use unimplemented!() for other code paths)
     async fn set_server_from_channel(&mut self) {
         if let Some(channel) = &self.channel {
-            #[allow(deprecated)]
             match channel {
                 Cow::Borrowed(Channel::TextChannel { server, .. })
-                | Cow::Owned(Channel::TextChannel { server, .. }) => {
+                | Cow::Owned(Channel::TextChannel { server, .. })
+                | Cow::Borrowed(Channel::VoiceChannel { server, .. })
+                | Cow::Owned(Channel::VoiceChannel { server, .. }) => {
                     if let Some(known_server) =
                         // I'm not sure why I can't just pattern match both at once here?
                         // It throws some weird error and the provided fix doesn't work :/

crates/core/database/src/util/reference.rs

@@ -3,46 +3,52 @@ use std::str::FromStr;
 use revolt_result::Result;
 #[cfg(feature = "rocket-impl")]
 use rocket::request::FromParam;
+#[cfg(feature = "rocket-impl")]
+use schemars::{
+    schema::{InstanceType, Schema, SchemaObject, SingleOrVec},
+    JsonSchema,
+};
 
 use crate::{
     Bot, Channel, Database, Emoji, Invite, Member, Message, Server, ServerBan, User, Webhook,
 };
 
 /// Reference to some object in the database
-pub struct Reference<'a> {
+#[derive(Serialize, Deserialize)]
+pub struct Reference {
     /// Id of object
-    pub id: &'a str,
+    pub id: String,
 }
 
-impl<'a> Reference<'a> {
+impl Reference {
     /// Create a Ref from an unchecked string
-    pub fn from_unchecked(id: &'a str) -> Reference<'a> {
+    pub fn from_unchecked(id: String) -> Reference {
         Reference { id }
     }
 
     /// Fetch ban from Ref
     pub async fn as_ban(&self, db: &Database, server: &str) -> Result<ServerBan> {
-        db.fetch_ban(server, self.id).await
+        db.fetch_ban(server, &self.id).await
     }
 
     /// Fetch bot from Ref
     pub async fn as_bot(&self, db: &Database) -> Result<Bot> {
-        db.fetch_bot(self.id).await
+        db.fetch_bot(&self.id).await
     }
 
     /// Fetch emoji from Ref
     pub async fn as_emoji(&self, db: &Database) -> Result<Emoji> {
-        db.fetch_emoji(self.id).await
+        db.fetch_emoji(&self.id).await
     }
 
     /// Fetch channel from Ref
     pub async fn as_channel(&self, db: &Database) -> Result<Channel> {
-        db.fetch_channel(self.id).await
+        db.fetch_channel(&self.id).await
     }
 
     /// Fetch invite from Ref or create invite to server if discoverable
     pub async fn as_invite(&self, db: &Database) -> Result<Invite> {
-        if ulid::Ulid::from_str(self.id).is_ok() {
+        if ulid::Ulid::from_str(&self.id).is_ok() {
             let server = self.as_server(db).await?;
             if !server.discoverable {
                 return Err(create_error!(NotFound));
@@ -59,18 +65,18 @@ impl<'a> Reference<'a> {
                     .ok_or(create_error!(NotFound))?,
             })
         } else {
-            db.fetch_invite(self.id).await
+            db.fetch_invite(&self.id).await
         }
     }
 
     /// Fetch message from Ref
     pub async fn as_message(&self, db: &Database) -> Result<Message> {
-        db.fetch_message(self.id).await
+        db.fetch_message(&self.id).await
     }
 
     /// Fetch message from Ref and validate channel
     pub async fn as_message_in_channel(&self, db: &Database, channel: &str) -> Result<Message> {
-        let msg = db.fetch_message(self.id).await?;
+        let msg = db.fetch_message(&self.id).await?;
         if msg.channel != channel {
             return Err(create_error!(NotFound));
         }
@@ -80,30 +86,44 @@ impl<'a> Reference<'a> {
 
     /// Fetch member from Ref
     pub async fn as_member(&self, db: &Database, server: &str) -> Result<Member> {
-        db.fetch_member(server, self.id).await
+        db.fetch_member(server, &self.id).await
     }
 
     /// Fetch server from Ref
     pub async fn as_server(&self, db: &Database) -> Result<Server> {
-        db.fetch_server(self.id).await
+        db.fetch_server(&self.id).await
     }
 
     /// Fetch user from Ref
     pub async fn as_user(&self, db: &Database) -> Result<User> {
-        db.fetch_user(self.id).await
+        db.fetch_user(&self.id).await
     }
 
     /// Fetch webhook from Ref
     pub async fn as_webhook(&self, db: &Database) -> Result<Webhook> {
-        db.fetch_webhook(self.id).await
+        db.fetch_webhook(&self.id).await
     }
 }
 
 #[cfg(feature = "rocket-impl")]
-impl<'r> FromParam<'r> for Reference<'r> {
+impl<'r> FromParam<'r> for Reference {
     type Error = &'r str;
 
     fn from_param(param: &'r str) -> Result<Self, Self::Error> {
-        Ok(Reference::from_unchecked(param))
+        Ok(Reference::from_unchecked(param.into()))
+    }
+}
+
+#[cfg(feature = "rocket-impl")]
+impl JsonSchema for Reference {
+    fn schema_name() -> String {
+        "Id".to_string()
+    }
+
+    fn json_schema(_gen: &mut schemars::gen::SchemaGenerator) -> Schema {
+        Schema::Object(SchemaObject {
+            instance_type: Some(SingleOrVec::Single(Box::new(InstanceType::String))),
+            ..Default::default()
+        })
     }
 }

crates/core/database/src/util/test_fixtures.rs

@@ -39,9 +39,7 @@ pub async fn load_fixture(db: &Database, input: &str) -> HashMap<String, String>
             LoadedFixture::User(user) => db.insert_user(&user).await.unwrap(),
             LoadedFixture::Channel(channel) => db.insert_channel(&channel).await.unwrap(),
             LoadedFixture::Server(server) => db.insert_server(&server).await.unwrap(),
-            LoadedFixture::ServerMember(member) => {
-                db.insert_or_merge_member(&member).await.unwrap();
-            }
+            LoadedFixture::ServerMember(member) => db.insert_member(&member).await.unwrap(),
         }
     }
 

crates/core/database/src/util/utoipa.rs

@@ -1,49 +0,0 @@
-use utoipa::{
-    openapi::{
-        schema::SchemaType,
-        security::{ApiKey, ApiKeyValue, SecurityScheme},
-        ObjectBuilder, OpenApi, RefOr, Schema, Type,
-    },
-    Modify, PartialSchema, ToSchema,
-};
-
-use crate::util::reference::Reference;
-
-pub struct TokenSecurity;
-
-impl Modify for TokenSecurity {
-    fn modify(&self, openapi: &mut OpenApi) {
-        let components = openapi.components.get_or_insert_default();
-
-        components.add_security_scheme(
-            "Session-Token",
-            SecurityScheme::ApiKey(ApiKey::Header(ApiKeyValue::new(
-                "X-Session-Token".to_string(),
-            ))),
-        );
-
-        components.add_security_scheme(
-            "Bot-Token",
-            SecurityScheme::ApiKey(ApiKey::Header(ApiKeyValue::new("X-Bot-Ticket".to_string()))),
-        );
-    }
-}
-
-impl ToSchema for Reference<'_> {
-    fn name() -> std::borrow::Cow<'static, str> {
-        std::borrow::Cow::Borrowed("Reference")
-    }
-}
-
-impl PartialSchema for Reference<'_> {
-    fn schema() -> RefOr<Schema> {
-        RefOr::T(
-            ObjectBuilder::new()
-                .description(Some("An id referencing a stoat model."))
-                .schema_type(SchemaType::Type(Type::String))
-                .examples(["01FD58YK5W7QRV5H3D64KTQYX3"])
-                .build()
-                .into(),
-        )
-    }
-}

crates/core/database/src/voice/mod.rs

@@ -1,598 +0,0 @@
-use crate::{
-    events::client::EventV1,
-    models::{Channel, User},
-    util::{permissions::DatabasePermissionQuery, reference::Reference},
-    Database, Server,
-};
-use iso8601_timestamp::{Duration, Timestamp};
-use livekit_protocol::ParticipantPermission;
-use redis_kiss::{get_connection as _get_connection, redis::Pipeline, AsyncCommands, Conn};
-use revolt_config::FeaturesLimits;
-use revolt_models::v0::{self, PartialUserVoiceState, UserVoiceState};
-use revolt_permissions::{calculate_channel_permissions, ChannelPermission, PermissionValue};
-use revolt_result::{create_error, Result, ToRevoltError};
-
-mod voice_client;
-pub use voice_client::VoiceClient;
-
-async fn get_connection() -> Result<Conn> {
-    _get_connection().await.map_err(|_| create_error!(InternalError))
-}
-
-pub async fn raise_if_in_voice(user: &User, channel_id: &str) -> Result<()> {
-    let mut conn = get_connection().await?;
-
-    if user.bot.is_some()
-    // bots can be in as many voice channels as it wants so we just check if its already connected to the one its trying to connect to
-        && conn.sismember(format!("vc:{}", &user.id), channel_id)
-            .await
-            .to_internal_error()?
-    {
-        Err(create_error!(AlreadyConnected))
-    } else if conn
-        .scard::<_, u32>(format!("vc:{}", &user.id)) // check if the current vc set is empty
-        .await
-        .to_internal_error()?
-        > 0
-    {
-        Err(create_error!(AlreadyConnected))
-    } else {
-        Ok(())
-    }
-}
-
-pub async fn set_channel_node(channel: &str, node: &str) -> Result<()> {
-    get_connection()
-        .await?
-        .set(format!("node:{channel}"), node)
-        .await
-        .to_internal_error()
-}
-
-pub async fn get_channel_node(channel: &str) -> Result<Option<String>> {
-    get_connection()
-        .await?
-        .get(format!("node:{channel}"))
-        .await
-        .to_internal_error()
-}
-
-pub async fn get_user_voice_channels(user_id: &str) -> Result<Vec<String>> {
-    get_connection()
-        .await?
-        .smembers(format!("vc:{user_id}"))
-        .await
-        .to_internal_error()
-}
-
-pub async fn set_user_moved_from_voice(
-    old_channel: &str,
-    new_channel: &str,
-    user_id: &str,
-) -> Result<()> {
-    get_connection()
-        .await?
-        .set_ex(
-            format!("moved_from:{user_id}:{old_channel}"),
-            new_channel,
-            10,
-        )
-        .await
-        .to_internal_error()
-}
-
-pub async fn get_user_moved_from_voice(channel_id: &str, user_id: &str) -> Result<Option<String>> {
-    get_connection()
-        .await?
-        .get_del(format!("moved_from:{user_id}:{channel_id}"))
-        .await
-        .to_internal_error()
-}
-
-pub async fn set_user_moved_to_voice(
-    new_channel: &str,
-    old_channel: &str,
-    user_id: &str,
-) -> Result<()> {
-    get_connection()
-        .await?
-        .set_ex(format!("moved_to:{user_id}:{new_channel}"), old_channel, 10)
-        .await
-        .to_internal_error()
-}
-
-pub async fn get_user_moved_to_voice(channel_id: &str, user_id: &str) -> Result<Option<String>> {
-    get_connection()
-        .await?
-        .get_del(format!("moved_to:{user_id}:{channel_id}"))
-        .await
-        .to_internal_error()
-}
-
-pub async fn is_in_voice_channel(user_id: &str, channel_id: &str) -> Result<bool> {
-    get_connection()
-        .await?
-        .sismember(format!("vc:{user_id}"), channel_id)
-        .await
-        .to_internal_error()
-}
-
-pub async fn get_user_voice_channel_in_server(
-    user_id: &str,
-    server_id: &str,
-) -> Result<Option<String>> {
-    let mut conn = get_connection().await?;
-
-    let unique_key = format!("{user_id}:{server_id}");
-
-    conn.get(&unique_key).await.to_internal_error()
-}
-
-pub fn get_allowed_sources(
-    limits: &FeaturesLimits,
-    permissions: PermissionValue,
-) -> Vec<&'static str> {
-    let mut allowed_sources = Vec::new();
-
-    if permissions.has(ChannelPermission::Speak as u64) {
-        allowed_sources.push("microphone")
-    };
-
-    if permissions.has(ChannelPermission::Video as u64) && limits.video {
-        allowed_sources.extend(["camera", "screen_share", "screen_share_audio"]);
-    };
-
-    allowed_sources
-}
-
-pub async fn create_voice_state(
-    channel_id: &str,
-    server_id: Option<&str>,
-    user_id: &str,
-    joined_at: Timestamp,
-) -> Result<UserVoiceState> {
-    let unique_key = format!("{}:{}", &user_id, server_id.unwrap_or(channel_id));
-
-    let voice_state = UserVoiceState {
-        joined_at,
-        id: user_id.to_string(),
-        is_receiving: true,
-        is_publishing: false,
-        screensharing: false,
-        camera: false,
-    };
-
-    Pipeline::new()
-        .sadd(format!("vc_members:{channel_id}"), user_id)
-        .sadd(format!("vc:{user_id}"), channel_id)
-        .set(&unique_key, channel_id)
-        .set(
-            format!("joined_at:{unique_key}"),
-            joined_at
-                .duration_since(Timestamp::UNIX_EPOCH)
-                .whole_milliseconds() as i64,
-        )
-        .set(
-            format!("is_publishing:{unique_key}"),
-            voice_state.is_publishing,
-        )
-        .set(
-            format!("is_receiving:{unique_key}"),
-            voice_state.is_receiving,
-        )
-        .set(
-            format!("screensharing:{unique_key}"),
-            voice_state.screensharing,
-        )
-        .set(format!("camera:{unique_key}"), voice_state.camera)
-        .query_async::<_, ()>(&mut get_connection().await?.into_inner())
-        .await
-        .to_internal_error()?;
-
-    Ok(voice_state)
-}
-
-pub async fn delete_voice_state(
-    channel_id: &str,
-    server_id: Option<&str>,
-    user_id: &str,
-) -> Result<()> {
-    let unique_key = format!("{}:{}", &user_id, server_id.unwrap_or(channel_id));
-
-    Pipeline::new()
-        .srem(format!("vc_members:{channel_id}"), user_id)
-        .srem(format!("vc:{user_id}"), channel_id)
-        .del(&[
-            format!("joined_at:{unique_key}"),
-            format!("is_publishing:{unique_key}"),
-            format!("is_receiving:{unique_key}"),
-            format!("screensharing:{unique_key}"),
-            format!("camera:{unique_key}"),
-            unique_key.clone(),
-        ])
-        .query_async(&mut get_connection().await?.into_inner())
-        .await
-        .to_internal_error()
-}
-
-pub async fn delete_channel_voice_state(
-    channel_id: &str,
-    server_id: Option<&str>,
-    user_ids: &[String],
-) -> Result<()> {
-    let parent_id = server_id.unwrap_or(channel_id);
-
-    let mut pipeline = Pipeline::new();
-    pipeline.del(format!("vc_members:{channel_id}"));
-
-    for user_id in user_ids {
-        let unique_key = format!("{user_id}:{parent_id}");
-
-        pipeline.srem(format!("vc:{user_id}"), channel_id).del(&[
-            format!("joined_at:{unique_key}"),
-            format!("is_publishing:{unique_key}"),
-            format!("is_receiving:{unique_key}"),
-            format!("screensharing:{unique_key}"),
-            format!("camera:{unique_key}"),
-            unique_key.clone(),
-        ]);
-    }
-
-    pipeline
-        .query_async(&mut get_connection().await?.into_inner())
-        .await
-        .to_internal_error()
-}
-
-pub async fn update_voice_state_tracks(
-    channel_id: &str,
-    server_id: Option<&str>,
-    user_id: &str,
-    added: bool,
-    track: i32,
-) -> Result<PartialUserVoiceState> {
-    let partial = match track {
-        /* TrackSource::Unknown */ 0 => PartialUserVoiceState::default(),
-        /* TrackSource::Camera */
-        1 => PartialUserVoiceState {
-            camera: Some(added),
-            ..Default::default()
-        },
-        /* TrackSource::Microphone */
-        2 => PartialUserVoiceState {
-            is_publishing: Some(added),
-            ..Default::default()
-        },
-        /* TrackSource::ScreenShare | TrackSource::ScreenShareAudio */
-        3 | 4 => PartialUserVoiceState {
-            screensharing: Some(added),
-            ..Default::default()
-        },
-        _ => unreachable!(),
-    };
-
-    update_voice_state(channel_id, server_id, user_id, &partial).await?;
-
-    Ok(partial)
-}
-
-pub async fn update_voice_state(
-    channel_id: &str,
-    server_id: Option<&str>,
-    user_id: &str,
-    partial: &PartialUserVoiceState,
-) -> Result<()> {
-    let unique_key = format!("{}:{}", &user_id, server_id.unwrap_or(channel_id));
-
-    let mut pipeline = Pipeline::new();
-
-    if let Some(camera) = &partial.camera {
-        pipeline.set(format!("camera:{unique_key}"), camera);
-    };
-
-    if let Some(is_publishing) = &partial.is_publishing {
-        pipeline.set(format!("is_publishing:{unique_key}"), is_publishing);
-    }
-
-    if let Some(is_receiving) = &partial.is_receiving {
-        pipeline.set(format!("is_receiving:{unique_key}"), is_receiving);
-    }
-
-    if let Some(screensharing) = &partial.screensharing {
-        pipeline.set(format!("screensharing:{unique_key}"), screensharing);
-    }
-
-    pipeline
-        .query_async(&mut get_connection().await?.into_inner())
-        .await
-        .to_internal_error()
-}
-
-pub async fn get_voice_channel_members(channel_id: &str) -> Result<Option<Vec<String>>> {
-    get_connection()
-        .await?
-        .smembers::<_, Option<Vec<String>>>(format!("vc_members:{channel_id}"))
-        .await
-        .to_internal_error()
-        .map(|opt| opt.and_then(|v| if v.is_empty() { None } else { Some(v) }))
-}
-
-pub async fn get_voice_state(
-    channel_id: &str,
-    server_id: Option<&str>,
-    user_id: &str,
-) -> Result<Option<UserVoiceState>> {
-    let unique_key = format!("{}:{}", user_id, server_id.unwrap_or(channel_id));
-
-    let (joined_at, is_publishing, is_receiving, screensharing, camera) = get_connection()
-        .await?
-        .mget(&[
-            format!("joined_at:{unique_key}"),
-            format!("is_publishing:{unique_key}"),
-            format!("is_receiving:{unique_key}"),
-            format!("screensharing:{unique_key}"),
-            format!("camera:{unique_key}"),
-        ])
-        .await
-        .to_internal_error()?;
-
-    match (
-        joined_at,
-        is_publishing,
-        is_receiving,
-        screensharing,
-        camera,
-    ) {
-        (
-            Some(joined_at),
-            Some(is_publishing),
-            Some(is_receiving),
-            Some(screensharing),
-            Some(camera),
-        ) => Ok(Some(v0::UserVoiceState {
-            joined_at: Timestamp::UNIX_EPOCH
-                .checked_add(Duration::milliseconds(joined_at))
-                .unwrap(),
-            id: user_id.to_string(),
-            is_receiving,
-            is_publishing,
-            screensharing,
-            camera,
-        })),
-        _ => Ok(None),
-    }
-}
-
-pub async fn get_channel_voice_state(channel: &Channel) -> Result<Option<v0::ChannelVoiceState>> {
-    let members = get_voice_channel_members(channel.id()).await?;
-
-    let server = channel.server();
-
-    if let Some(members) = members {
-        let mut participants = Vec::with_capacity(members.len());
-
-        for user_id in members {
-            if let Some(voice_state) = get_voice_state(channel.id(), server, &user_id).await? {
-                participants.push(voice_state);
-            } else {
-                log::info!("Voice state not found but member in voice channel members, removing.");
-
-                delete_voice_state(channel.id(), server, &user_id).await?;
-            }
-        }
-
-        // In case a user voice state failed to be fetched, the vec's capacity will be larger than the length, shrink it
-        participants.shrink_to_fit();
-
-        Ok(Some(v0::ChannelVoiceState {
-            id: channel.id().to_string(),
-            participants,
-        }))
-    } else {
-        Ok(None)
-    }
-}
-
-pub async fn move_user(user: &str, from: &str, to: &str) -> Result<()> {
-    get_connection()
-        .await?
-        .smove(
-            format!("vc-members-{from}"),
-            format!("vc-members-{to}"),
-            user,
-        )
-        .await
-        .to_internal_error()
-}
-
-pub async fn sync_voice_permissions(
-    db: &Database,
-    voice_client: &VoiceClient,
-    channel: &Channel,
-    server: Option<&Server>,
-    role_id: Option<&str>,
-) -> Result<()> {
-    let Some(node) = get_channel_node(channel.id()).await? else {
-        return Ok(());
-    };
-
-    for user_id in get_voice_channel_members(channel.id())
-        .await?
-        .iter()
-        .flatten()
-    {
-        let user = Reference::from_unchecked(user_id).as_user(db).await?;
-
-        sync_user_voice_permissions(db, voice_client, &node, &user, channel, server, role_id)
-            .await?;
-    }
-
-    Ok(())
-}
-
-pub async fn sync_user_voice_permissions(
-    db: &Database,
-    voice_client: &VoiceClient,
-    node: &str,
-    user: &User,
-    channel: &Channel,
-    server: Option<&Server>,
-    role_id: Option<&str>,
-) -> Result<()> {
-    let channel_id = channel.id();
-    let server_id = server.as_ref().map(|s| s.id.as_str());
-
-    let member = match server_id {
-        Some(server_id) => Some(
-            Reference::from_unchecked(&user.id)
-                .as_member(db, server_id)
-                .await?,
-        ),
-        None => None,
-    };
-
-    if role_id.is_none_or(|role_id| {
-        member
-            .as_ref()
-            .is_none_or(|member| member.roles.iter().any(|r| r == role_id))
-    }) {
-        let Some(voice_state) = get_voice_state(channel_id, server_id, &user.id).await? else {
-            return Ok(());
-        };
-
-        let mut query = DatabasePermissionQuery::new(db, user)
-            .channel(channel)
-            .user(user);
-
-        if let (Some(server), Some(member)) = (server, member.as_ref()) {
-            query = query.member(member).server(server)
-        }
-
-        let permissions = calculate_channel_permissions(&mut query).await;
-        let limits = user.limits().await;
-
-        let mut update_event = PartialUserVoiceState {
-            id: Some(user.id.clone()),
-            ..Default::default()
-        };
-
-        let before = update_event.clone();
-
-        let can_video =
-            limits.video && permissions.has_channel_permission(ChannelPermission::Video);
-        let can_speak = permissions.has_channel_permission(ChannelPermission::Speak);
-        let can_listen = permissions.has_channel_permission(ChannelPermission::Listen);
-
-        update_event.camera = voice_state.camera.then_some(can_video);
-        update_event.screensharing = voice_state.screensharing.then_some(can_video);
-        update_event.is_publishing = voice_state.is_publishing.then_some(can_speak);
-
-        update_voice_state(channel_id, server_id, &user.id, &update_event).await?;
-
-        voice_client
-            .update_permissions(
-                node,
-                user,
-                channel_id,
-                ParticipantPermission {
-                    can_subscribe: can_listen,
-                    can_publish: can_speak,
-                    can_publish_data: can_speak,
-                    ..Default::default()
-                },
-            )
-            .await?;
-
-        if update_event != before {
-            EventV1::UserVoiceStateUpdate {
-                id: user.id.clone(),
-                channel_id: channel_id.to_string(),
-                data: update_event,
-            }
-            .p(channel_id.to_string())
-            .await;
-        };
-    };
-
-    Ok(())
-}
-
-pub async fn set_channel_call_started_system_message(
-    channel_id: &str,
-    message_id: &str,
-) -> Result<()> {
-    get_connection()
-        .await?
-        .set(format!("call_started_message:{channel_id}"), message_id)
-        .await
-        .to_internal_error()
-}
-
-pub async fn take_channel_call_started_system_message(channel_id: &str) -> Result<Option<String>> {
-    get_connection()
-        .await?
-        .get_del(format!("call_started_message:{channel_id}"))
-        .await
-        .to_internal_error()
-}
-
-pub async fn set_call_notification_recipients(
-    channel_id: &str,
-    user_id: &str,
-    recipients: &[String],
-) -> Result<()> {
-    get_connection()
-        .await?
-        .set_ex(
-            format!("call_notification_recipients:{channel_id}-{user_id}"),
-            recipients,
-            10,
-        )
-        .await
-        .to_internal_error()
-}
-
-pub async fn get_call_notification_recipients(
-    channel_id: &str,
-    user_id: &str,
-) -> Result<Option<Vec<String>>> {
-    get_connection()
-        .await?
-        .get_del(format!(
-            "call_notification_recipients:{channel_id}-{user_id}"
-        ))
-        .await
-        .to_internal_error()
-}
-
-pub async fn remove_user_from_voice_channels(db: &Database, voice_client: &VoiceClient, user_id: &str) -> Result<()> {
-    for channel_id in get_user_voice_channels(user_id).await? {
-        remove_user_from_voice_channel(db, voice_client, &channel_id, user_id).await?;
-    };
-
-    Ok(())
-}
-
-pub async fn remove_user_from_voice_channel(db: &Database, voice_client: &VoiceClient, channel_id: &str, user_id: &str) -> Result<()> {
-    if let Some(node) = get_channel_node(channel_id).await? {
-        let _ = voice_client.remove_user(&node, user_id, channel_id).await;
-    }
-
-    let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
-
-    delete_voice_state(channel_id, channel.server(), user_id).await?;
-
-    Ok(())
-}
-
-pub async fn delete_voice_channel(voice_client: &VoiceClient, channel_id: &str, server_id: Option<&str>) -> Result<()> {
-    if let Some(users) = get_voice_channel_members(channel_id).await? {
-        let node = get_channel_node(channel_id).await?.unwrap();
-
-        voice_client.delete_room(&node, channel_id).await?;
-
-        delete_channel_voice_state(channel_id, server_id, &users).await?;
-    };
-
-    Ok(())
-}

crates/core/database/src/voice/voice_client.rs

@@ -1,156 +0,0 @@
-use crate::{
-    models::{Channel, User},
-    Database,
-};
-use livekit_api::{
-    access_token::{AccessToken, VideoGrants},
-    services::room::{CreateRoomOptions, RoomClient as InnerRoomClient, UpdateParticipantOptions},
-};
-use livekit_protocol::{ParticipantInfo, ParticipantPermission, Room};
-use revolt_config::{config, LiveKitNode};
-use revolt_permissions::{ChannelPermission, PermissionValue};
-use revolt_result::{create_error, Result, ToRevoltError};
-use std::{collections::HashMap, time::Duration};
-
-use super::get_allowed_sources;
-
-#[derive(Debug)]
-pub struct RoomClient {
-    pub client: InnerRoomClient,
-    pub node: LiveKitNode,
-}
-
-#[derive(Debug)]
-pub struct VoiceClient {
-    pub rooms: HashMap<String, RoomClient>,
-}
-
-impl VoiceClient {
-    pub fn new(nodes: HashMap<String, LiveKitNode>) -> Self {
-        Self {
-            rooms: nodes
-                .into_iter()
-                .map(|(name, node)| {
-                    (
-                        name,
-                        RoomClient {
-                            client: InnerRoomClient::with_api_key(
-                                &node.url,
-                                &node.key,
-                                &node.secret,
-                            ),
-                            node,
-                        },
-                    )
-                })
-                .collect(),
-        }
-    }
-
-    pub fn is_enabled(&self) -> bool {
-        !self.rooms.is_empty()
-    }
-
-    pub async fn from_revolt_config() -> Self {
-        let config = config().await;
-
-        Self::new(config.api.livekit.nodes.clone())
-    }
-
-    pub fn get_node(&self, name: &str) -> Result<&RoomClient> {
-        self.rooms
-            .get(name)
-            .ok_or_else(|| create_error!(UnknownNode))
-    }
-
-    pub async fn create_token(
-        &self,
-        node: &str,
-        db: &Database,
-        user: &User,
-        permissions: PermissionValue,
-        channel: &Channel,
-    ) -> Result<String> {
-        let room = self.get_node(node)?;
-
-        let limits = user.limits().await;
-        let allowed_sources = get_allowed_sources(&limits, permissions);
-
-        AccessToken::with_api_key(&room.node.key, &room.node.secret)
-            .with_name(&format!("{}#{}", user.username, user.discriminator))
-            .with_identity(&user.id)
-            .with_metadata(
-                &serde_json::to_string(&user.clone().into(db, None).await).to_internal_error()?,
-            )
-            .with_ttl(Duration::from_secs(10))
-            .with_grants(VideoGrants {
-                room_join: true,
-                can_publish: true,
-                can_publish_data: false,
-                can_publish_sources: allowed_sources
-                    .into_iter()
-                    .map(ToString::to_string)
-                    .collect(),
-                can_subscribe: permissions.has_channel_permission(ChannelPermission::Listen),
-                room: channel.id().to_string(),
-                ..Default::default()
-            })
-            .to_jwt()
-            .to_internal_error()
-    }
-
-    pub async fn create_room(&self, node: &str, channel: &Channel) -> Result<Room> {
-        let room = self.get_node(node)?;
-
-        room.client
-            .create_room(
-                channel.id(),
-                CreateRoomOptions {
-                    empty_timeout: 5 * 60, // 5 minutes,
-                    ..Default::default()
-                },
-            )
-            .await
-            .to_internal_error()
-    }
-
-    pub async fn update_permissions(
-        &self,
-        node: &str,
-        user: &User,
-        channel_id: &str,
-        new_permissions: ParticipantPermission,
-    ) -> Result<ParticipantInfo> {
-        let room = self.get_node(node)?;
-
-        room.client
-            .update_participant(
-                channel_id,
-                &user.id,
-                UpdateParticipantOptions {
-                    permission: Some(new_permissions),
-                    ..Default::default()
-                },
-            )
-            .await
-            .to_internal_error()
-    }
-
-    pub async fn remove_user(&self, node: &str, user_id: &str, channel_id: &str) -> Result<()> {
-        let room = self.get_node(node)?;
-
-        room.client
-            .remove_participant(channel_id, user_id)
-            .await
-            .to_internal_error()
-    }
-
-    pub async fn delete_room(&self, node: &str, channel_id: &str) -> Result<()> {
-        let room = self.get_node(node)?;
-
-        room.client
-            .delete_room(channel_id)
-            .await
-            .to_internal_error()
-    }
-}