chore: Update ruwuma

Also, don't take read receipts into consideration for lazy loading.
Synapse doesn't do this and they're making initial syncs very large.

.forgejo/actions/setup-rust/action.yml

@@ -17,9 +17,9 @@ inputs:
     required: false
     default: ''
   rust-version:
-    description: 'Rust version to install (e.g. nightly). Defaults to 1.87.0'
+    description: 'Rust version to install (e.g. nightly).'
     required: false
-    default: '1.87.0'
+    default: ''
   sccache-cache-limit:
     description: 'Maximum size limit for sccache local cache (e.g. 2G, 500M)'
     required: false
@@ -59,23 +59,7 @@ runs:
         mkdir -p "${{ github.workspace }}/target"
         mkdir -p "${{ github.workspace }}/.rustup"
 
-    - name: Start cache restore group
-      shell: bash
-      run: echo "::group::📦 Restoring caches (registry, toolchain, build artifacts)"
 
-    - name: Cache Cargo registry and git
-      id: registry-cache
-      uses: actions/cache@v4
-      with:
-        path: |
-          .cargo/registry/index
-          .cargo/registry/cache
-          .cargo/git/db
-        # Registry cache saved per workflow, restored from any workflow's cache
-        # Each workflow maintains its own registry that accumulates its needed crates
-        key: continuwuity-cargo-registry-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ github.workflow }}
-        restore-keys: |
-          continuwuity-cargo-registry-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-
 
     - name: Cache toolchain binaries
       id: toolchain-cache
@@ -88,45 +72,6 @@ runs:
         # Shared toolchain cache across all Rust versions
         key: continuwuity-toolchain-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}
 
-
-    - name: Setup sccache
-      uses: https://git.tomfos.tr/tom/sccache-action@v1
-
-    - name: Cache dependencies
-      id: deps-cache
-      uses: actions/cache@v4
-      with:
-        path: |
-          target/**/.fingerprint
-          target/**/deps
-          target/**/*.d
-          target/**/.cargo-lock
-          target/**/CACHEDIR.TAG
-          target/**/.rustc_info.json
-          /timelord/
-        # Dependencies cache - based on Cargo.lock, survives source code changes
-        key: >-
-          continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}
-        restore-keys: |
-          continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
-
-    - name: Cache incremental compilation
-      id: incremental-cache
-      uses: actions/cache@v4
-      with:
-        path: |
-          target/**/incremental
-        # Incremental cache - based on source code changes
-        key: >-
-          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-${{ hashFiles('**/*.rs', '**/Cargo.toml') }}
-        restore-keys: |
-          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-
-          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
-
-    - name: End cache restore group
-      shell: bash
-      run: echo "::endgroup::"
-
     - name: Setup Rust toolchain
       shell: bash
       run: |
@@ -158,6 +103,71 @@ runs:
         fi
         echo "::endgroup::"
 
+    - name: Output Rust version
+      id: rust-setup
+      shell: bash
+      run: |
+        RUST_VERSION=$(rustc --version | cut -d' ' -f2)
+        echo "version=$RUST_VERSION" >> $GITHUB_OUTPUT
+        echo "📋 Rust version: $(rustc --version)"
+
+    - name: Setup sccache
+      uses: https://git.tomfos.tr/tom/sccache-action@v1
+
+
+    - name: Start cache restore group
+      shell: bash
+      run: echo "::group::📦 Restoring caches (registry, toolchain, build artifacts)"
+
+    - name: Cache Cargo registry and git
+      id: registry-cache
+      uses: actions/cache@v4
+      with:
+        path: |
+          .cargo/registry/index
+          .cargo/registry/cache
+          .cargo/git/db
+        # Registry cache saved per workflow, restored from any workflow's cache
+        # Each workflow maintains its own registry that accumulates its needed crates
+        key: continuwuity-cargo-registry-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ github.workflow }}
+        restore-keys: |
+          continuwuity-cargo-registry-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-
+
+    - name: Cache dependencies
+      id: deps-cache
+      uses: actions/cache@v4
+      with:
+        path: |
+          target/**/.fingerprint
+          target/**/deps
+          target/**/*.d
+          target/**/.cargo-lock
+          target/**/CACHEDIR.TAG
+          target/**/.rustc_info.json
+          /timelord/
+        # Dependencies cache - based on Cargo.lock, survives source code changes
+        key: >-
+          continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}
+        restore-keys: |
+          continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
+
+    - name: Cache incremental compilation
+      id: incremental-cache
+      uses: actions/cache@v4
+      with:
+        path: |
+          target/**/incremental
+        # Incremental cache - based on source code changes
+        key: >-
+          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-${{ hashFiles('**/*.rs', '**/Cargo.toml') }}
+        restore-keys: |
+          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-
+          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
+
+    - name: End cache restore group
+      shell: bash
+      run: echo "::endgroup::"
+
     - name: Configure PATH and install tools
       shell: bash
       env:
@@ -225,13 +235,9 @@ runs:
         echo "📦 Installing target: ${{ inputs.rust-target }}"
         rustup target add ${{ inputs.rust-target }}
 
-    - name: Output version and summary
-      id: rust-setup
+    - name: Output summary
       shell: bash
       run: |
-        RUST_VERSION=$(rustc --version | cut -d' ' -f2)
-        echo "version=$RUST_VERSION" >> $GITHUB_OUTPUT
-
         echo "📋 Setup complete:"
         echo "  Rust: $(rustc --version)"
         echo "  Cargo: $(cargo --version)"

.forgejo/regsync/regsync.yml

@@ -40,6 +40,15 @@ creds:
   - registry: registry.gitlab.com
     user: "{{env \"GITLAB_USERNAME\"}}"
     pass: "{{env \"GITLAB_TOKEN\"}}"
+  - registry: git.nexy7574.co.uk
+    user: "{{env \"N7574_GIT_USERNAME\"}}"
+    pass: "{{env \"N7574_GIT_TOKEN\"}}"
+  - registry: ghcr.io
+    user: "{{env \"GH_PACKAGES_USER\"}}"
+    pass: "{{env \"GH_PACKAGES_TOKEN\"}}"
+  - registry: docker.io
+    user: "{{env \"DOCKER_MIRROR_USER\"}}"
+    pass: "{{env \"DOCKER_MIRROR_TOKEN\"}}"
 
 # Global defaults
 defaults:
@@ -53,3 +62,15 @@ sync:
     target: registry.gitlab.com/continuwuity/continuwuity
     type: repository
     <<: *tags-main
+  - source: *source
+    target: git.nexy7574.co.uk/mirrored/continuwuity
+    type: repository
+    <<: *tags-releases
+  - source: *source
+    target: ghcr.io/continuwuity/continuwuity
+    type: repository
+    <<: *tags-main
+  - source: *source
+    target: docker.io/jadedblueeyes/continuwuity
+    type: repository
+    <<: *tags-main

.forgejo/workflows/documentation.yml

@@ -55,7 +55,7 @@ jobs:
 
       - name: Setup Node.js
         if: steps.runner-env.outputs.node_major == '' || steps.runner-env.outputs.node_major < '20'
-        uses: https://github.com/actions/setup-node@v5
+        uses: https://github.com/actions/setup-node@v6
         with:
           node-version: 22
 

.forgejo/workflows/element.yml

@@ -24,7 +24,7 @@ jobs:
 
     steps:
       - name: 📦 Setup Node.js
-        uses: https://github.com/actions/setup-node@v5
+        uses: https://github.com/actions/setup-node@v6
         with:
           node-version: "22"
 

.forgejo/workflows/mirror-images.yml

@@ -11,7 +11,13 @@ on:
         required: false
         default: false
         type: boolean
-
+  push:
+    branches:
+      - main
+    paths:
+      # Re-run when config changes
+      - '.forgejo/regsync/regsync.yml'
+      - '.forgejo/workflows/mirror-images.yml'
 concurrency:
   group: "mirror-images"
   cancel-in-progress: true
@@ -24,12 +30,27 @@ jobs:
       BUILTIN_REGISTRY_PASSWORD: ${{ secrets.BUILTIN_REGISTRY_PASSWORD }}
       GITLAB_USERNAME: ${{ vars.GITLAB_USERNAME }}
       GITLAB_TOKEN: ${{ secrets.GITLAB_TOKEN }}
+      N7574_GIT_USERNAME: ${{ vars.N7574_GIT_USERNAME }}
+      N7574_GIT_TOKEN: ${{ secrets.N7574_GIT_TOKEN }}
+      GH_PACKAGES_USER: ${{ vars.GH_PACKAGES_USER }}
+      GH_PACKAGES_TOKEN: ${{ secrets.GH_PACKAGES_TOKEN }}
+      DOCKER_MIRROR_USER: ${{ vars.DOCKER_MIRROR_USER }}
+      DOCKER_MIRROR_TOKEN: ${{ secrets.DOCKER_MIRROR_TOKEN }}
     steps:
       - name: Checkout repository
         uses: actions/checkout@v5
         with:
           persist-credentials: false
 
+      # - uses: https://github.com/actions/create-github-app-token@v2
+      #   id: app-token
+      #   with:
+      #     app-id: ${{ vars.GH_APP_ID }}
+      #     private-key: ${{ secrets.GH_APP_PRIVATE_KEY }}
+      #     github-api-url: https://api.github.com
+      #     owner: continuwuity
+      #     repositories: continuwuity
+
       - name: Install regctl
         uses: https://forgejo.ellis.link/continuwuation/regclient-actions/regctl-installer@main
         with:

.forgejo/workflows/update-flake-hashes.yml

@@ -7,6 +7,7 @@ on:
       - "Cargo.lock"
       - "Cargo.toml"
       - "rust-toolchain.toml"
+      - "nix/**/*"
       - ".forgejo/workflows/update-flake-hashes.yml"
 
 jobs:
@@ -19,7 +20,8 @@ jobs:
           fetch-tags: false
           fetch-single-branch: true
           submodules: false
-          persist-credentials: false
+          persist-credentials: true
+          token: ${{ secrets.FORGEJO_TOKEN }}
 
       - uses: https://github.com/cachix/install-nix-action@7ab6e7fd29da88e74b1e314a4ae9ac6b5cda3801 # v31.8.0
         with:
@@ -40,13 +42,22 @@ jobs:
           echo "Base: $base"
           echo "HEAD: $(git rev-parse HEAD)"
           git diff --name-only $base HEAD > changed_files.txt
-          echo "files=$(cat changed_files.txt)" >> $FORGEJO_OUTPUT
+          echo "detected changes in $(cat changed_files.txt)"
+          # Join files with commas
+          files=$(paste -sd, changed_files.txt)
+          echo "files=$files" >> $FORGEJO_OUTPUT
+
+      - name: Debug output
+        run: |
+          echo "State of output"
+          echo "Changed files: ${{ steps.changes.outputs.files }}"
 
       - name: Get new toolchain hash
         if: contains(steps.changes.outputs.files, 'Cargo.toml') || contains(steps.changes.outputs.files, 'Cargo.lock') || contains(steps.changes.outputs.files, 'rust-toolchain.toml')
         run: |
           # Set the current sha256 to an empty hash to make `nix build` calculate a new one
-          awk '/fromToolchainFile *\{/{found=1; print; next} found && /sha256 =/{sub(/sha256 = .*/, "sha256 = pkgsHost.lib.fakeSha256;"); found=0} 1' flake.nix > temp.nix && mv temp.nix flake.nix
+          awk '/fromToolchainFile *\{/{found=1; print; next} found && /sha256 =/{sub(/sha256 = .*/, "sha256 = lib.fakeSha256;"); found=0} 1' nix/packages/rust.nix > temp.nix
+          mv temp.nix nix/packages/rust.nix
 
           # Build continuwuity and filter for the new hash
           # We do `|| true` because we want this to fail without stopping the workflow
@@ -54,19 +65,21 @@ jobs:
 
           # Place the new hash in place of the empty hash
           new_hash=$(cat new_toolchain_hash.txt)
-          sed -i "s|pkgsHost.lib.fakeSha256|\"$new_hash\"|" flake.nix
+          sed -i "s|lib.fakeSha256|\"$new_hash\"|" nix/packages/rust.nix
 
           echo "New hash:"
-          awk -F'"' '/fromToolchainFile/{found=1; next} found && /sha256 =/{print $2; found=0}' flake.nix
+          awk -F'"' '/fromToolchainFile/{found=1; next} found && /sha256 =/{print $2; found=0}' nix/packages/rust.nix
           echo "Expected new hash:"
           cat new_toolchain_hash.txt
 
           rm new_toolchain_hash.txt
 
       - name: Get new rocksdb hash
+        if: contains(steps.changes.outputs.files, '.nix') || contains(steps.changes.outputs.files, 'flake.lock')
         run: |
           # Set the current sha256 to an empty hash to make `nix build` calculate a new one
-          awk '/repo = "rocksdb";/{found=1; print; next} found && /sha256 =/{sub(/sha256 = .*/, "sha256 = pkgsHost.lib.fakeSha256;"); found=0} 1' flake.nix > temp.nix && mv temp.nix flake.nix
+          awk '/repo = "rocksdb";/{found=1; print; next} found && /sha256 =/{sub(/sha256 = .*/, "sha256 = lib.fakeSha256;"); found=0} 1' nix/packages/rocksdb/package.nix > temp.nix
+          mv temp.nix nix/packages/rocksdb/package.nix
 
           # Build continuwuity and filter for the new hash
           # We do `|| true` because we want this to fail without stopping the workflow
@@ -74,17 +87,17 @@ jobs:
 
           # Place the new hash in place of the empty hash
           new_hash=$(cat new_rocksdb_hash.txt)
-          sed -i "s|pkgsHost.lib.fakeSha256|\"$new_hash\"|" flake.nix
+          sed -i "s|lib.fakeSha256|\"$new_hash\"|" nix/packages/rocksdb/package.nix
 
           echo "New hash:"
-          awk -F'"' '/repo = "rocksdb";/{found=1; next} found && /sha256 =/{print $2; found=0}' flake.nix
+          awk -F'"' '/repo = "rocksdb";/{found=1; next} found && /sha256 =/{print $2; found=0}' nix/packages/rocksdb/package.nix
           echo "Expected new hash:"
           cat new_rocksdb_hash.txt
 
           rm new_rocksdb_hash.txt
 
       - name: Show diff
-        run: git diff flake.nix
+        run: git diff flake.nix nix
 
       - name: Push changes
         run: |

.pre-commit-config.yaml

@@ -7,7 +7,7 @@ default_stages:
 
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v5.0.0
+    rev: v6.0.0
     hooks:
         - id: fix-byte-order-marker
         - id: check-case-conflict
@@ -23,7 +23,7 @@ repos:
         - id: check-added-large-files
 
   - repo: https://github.com/crate-ci/typos
-    rev: v1.26.0
+    rev: v1.39.0
     hooks:
       - id: typos
       - id: typos

Cargo.toml

@@ -166,8 +166,8 @@ default-features = false
 features = ["raw_value"]
 
 # Used for appservice registration files
-[workspace.dependencies.serde_yml]
-version = "0.0.12"
+[workspace.dependencies.serde-saphyr]
+version = "0.0.7"
 
 # Used to load forbidden room/user regex from config
 [workspace.dependencies.serde_regex]
@@ -210,13 +210,13 @@ default-features = false
 version = "0.1.41"
 default-features = false
 [workspace.dependencies.tracing-subscriber]
-version = "0.3.19"
+version = "0.3.20"
 default-features = false
 features = ["env-filter", "std", "tracing", "tracing-log", "ansi", "fmt"]
 [workspace.dependencies.tracing-journald]
 version = "0.3.1"
 [workspace.dependencies.tracing-core]
-version = "0.1.33"
+version = "0.1.34"
 default-features = false
 
 # for URL previews
@@ -286,7 +286,7 @@ features = [
 ]
 
 [workspace.dependencies.hyper-util]
-version = "0.1.11"
+version = "=0.1.17"
 default-features = false
 features = [
 	"server-auto",
@@ -351,7 +351,7 @@ version = "0.1.2"
 # Used for matrix spec type definitions and helpers
 [workspace.dependencies.ruma]
 git = "https://forgejo.ellis.link/continuwuation/ruwuma"
-rev = "d18823471ab3c09e77ff03eea346d4c07e572654"
+rev = "b96a99ac85264aa6bf102c12e201090c6474c99d"
 features = [
     "compat",
     "rand",
@@ -412,28 +412,27 @@ default-features = false
 
 # optional opentelemetry, performance measurements, flamegraphs, etc for performance measurements and monitoring
 [workspace.dependencies.opentelemetry]
-version = "0.30.0"
+version = "0.31.0"
 
 [workspace.dependencies.tracing-flame]
 version = "0.2.0"
 
 [workspace.dependencies.tracing-opentelemetry]
-version = "0.31.0"
+version = "0.32.0"
 
 [workspace.dependencies.opentelemetry_sdk]
-version = "0.30.0"
+version = "0.31.0"
 features = ["rt-tokio"]
 
 [workspace.dependencies.opentelemetry-otlp]
-version = "0.30.0"
+version = "0.31.0"
 features = ["http", "trace", "logs", "metrics"]
 
-[workspace.dependencies.opentelemetry-jaeger-propagator]
-version = "0.30.0"
+
 
 # optional sentry metrics for crash/panic reporting
 [workspace.dependencies.sentry]
-version = "0.42.0"
+version = "0.45.0"
 default-features = false
 features = [
     "backtrace",
@@ -449,9 +448,9 @@ features = [
 ]
 
 [workspace.dependencies.sentry-tracing]
-version = "0.42.0"
+version = "0.45.0"
 [workspace.dependencies.sentry-tower]
-version = "0.42.0"
+version = "0.45.0"
 
 # jemalloc usage
 [workspace.dependencies.tikv-jemalloc-sys]
@@ -477,7 +476,7 @@ default-features = false
 features = ["use_std"]
 
 [workspace.dependencies.console-subscriber]
-version = "0.4"
+version = "0.5"
 
 [workspace.dependencies.nix]
 version = "0.30.1"
@@ -564,19 +563,7 @@ version = "0.7.5"
 
 # backport of [https://github.com/tokio-rs/tracing/pull/2956] to the 0.1.x branch of tracing.
 # we can switch back to upstream if #2956 is merged and backported in the upstream repo.
-# https://forgejo.ellis.link/continuwuation/tracing/commit/b348dca742af641c47bc390261f60711c2af573c
-[patch.crates-io.tracing-subscriber]
-git = "https://forgejo.ellis.link/continuwuation/tracing"
-rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
-[patch.crates-io.tracing]
-git = "https://forgejo.ellis.link/continuwuation/tracing"
-rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
-[patch.crates-io.tracing-core]
-git = "https://forgejo.ellis.link/continuwuation/tracing"
-rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
-[patch.crates-io.tracing-log]
-git = "https://forgejo.ellis.link/continuwuation/tracing"
-rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
+
 
 # adds a tab completion callback: https://forgejo.ellis.link/continuwuation/rustyline-async/src/branch/main/.patchy/0002-add-tab-completion-callback.patch
 # adds event for CTRL+\: https://forgejo.ellis.link/continuwuation/rustyline-async/src/branch/main/.patchy/0001-add-event-for-ctrl.patch
@@ -600,7 +587,7 @@ rev = "9c8e51510c35077df888ee72a36b4b05637147da"
 # reverts hyperium#148 conflicting with our delicate federation resolver hooks
 [patch.crates-io.hyper-util]
 git = "https://forgejo.ellis.link/continuwuation/hyper-util"
-rev = "e4ae7628fe4fcdacef9788c4c8415317a4489941"
+rev = "5886d5292bf704c246206ad72d010d674a7b77d0"
 
 #
 # Our crates
@@ -960,7 +947,7 @@ semicolon_outside_block = "warn"
 str_to_string = "warn"
 string_lit_chars_any = "warn"
 string_slice = "warn"
-string_to_string = "warn"
+
 suspicious_xor_used_as_pow = "warn"
 tests_outside_test_module = "warn"
 try_err = "warn"

conduwuit-example.toml

@@ -957,6 +957,21 @@
 #
 #rocksdb_bottommost_compression = true
 
+# Compression algorithm for RocksDB's Write-Ahead-Log (WAL).
+#
+# At present, only ZSTD compression is supported by RocksDB for WAL
+# compression. Enabling this can reduce WAL size at the expense of some
+# CPU usage during writes.
+#
+# The options are:
+# - "none" = No compression
+# - "zstd" = ZSTD compression
+#
+# For more information on WAL compression, see:
+# https://github.com/facebook/rocksdb/wiki/WAL-Compression
+#
+#rocksdb_wal_compression = "zstd"
+
 # Database recovery mode (for RocksDB WAL corruption).
 #
 # Use this option when the server reports corruption and refuses to start.
@@ -1497,6 +1512,19 @@
 #
 #block_non_admin_invites = false
 
+# Enable or disable making requests to MSC4284 Policy Servers.
+# It is recommended you keep this enabled unless you experience frequent
+# connectivity issues, such as in a restricted networking environment.
+#
+#enable_msc4284_policy_servers = true
+
+# Enable running locally generated events through configured MSC4284
+# policy servers. You may wish to disable this if your server is
+# single-user for a slight speed benefit in some rooms, but otherwise
+# should leave it enabled.
+#
+#policy_server_check_own_events = true
+
 # Allow admins to enter commands in rooms other than "#admins" (admin
 # room) by prefixing your message with "\!admin" or "\\!admin" followed up
 # a normal continuwuity admin command. The reply will be publicly visible

docker/Dockerfile

@@ -48,7 +48,7 @@ EOF
 
 # Developer tool versions
 # renovate: datasource=github-releases depName=cargo-bins/cargo-binstall
-ENV BINSTALL_VERSION=1.15.7
+ENV BINSTALL_VERSION=1.15.11
 # renovate: datasource=github-releases depName=psastras/sbom-rs
 ENV CARGO_SBOM_VERSION=0.9.1
 # renovate: datasource=crate depName=lddtree

docker/musl.Dockerfile

@@ -18,7 +18,7 @@ RUN --mount=type=cache,target=/etc/apk/cache apk add \
 
 # Developer tool versions
 # renovate: datasource=github-releases depName=cargo-bins/cargo-binstall
-ENV BINSTALL_VERSION=1.15.7
+ENV BINSTALL_VERSION=1.15.11
 # renovate: datasource=github-releases depName=psastras/sbom-rs
 ENV CARGO_SBOM_VERSION=0.9.1
 # renovate: datasource=crate depName=lddtree

docs/admin_reference.md

@@ -1078,7 +1078,10 @@ ###### **Subcommands:**
 
 * `delete` — - Deletes a single media file from our database and on the filesystem via a single MXC URL or event ID (not redacted)
 * `delete-list` — - Deletes a codeblock list of MXC URLs from our database and on the filesystem. This will always ignore errors
-* `delete-past-remote-media` — - Deletes all remote (and optionally local) media created before or after [duration] time using filesystem metadata first created at date, or fallback to last modified date. This will always ignore errors by default
+* `delete-past-remote-media` — Deletes all remote (and optionally local) media created before/after
+[duration] ago, using filesystem metadata first created at date, or
+fallback to last modified date. This will always ignore errors by
+default.
 * `delete-all-from-user` — - Deletes all the local media from a local user on our server. This will always ignore errors by default
 * `delete-all-from-server` — - Deletes all remote media from the specified remote server. This will always ignore errors by default
 * `get-file-info` —
@@ -1110,13 +1113,25 @@ ## `admin media delete-list`
 
 ## `admin media delete-past-remote-media`
 
-- Deletes all remote (and optionally local) media created before or after [duration] time using filesystem metadata first created at date, or fallback to last modified date. This will always ignore errors by default
+Deletes all remote (and optionally local) media created before/after
+[duration] ago, using filesystem metadata first created at date, or
+fallback to last modified date. This will always ignore errors by
+default.
+
+* Examples:
+  * Delete all remote media older than a year:
+
+    `!admin media delete-past-remote-media -b 1y`
+
+  * Delete all remote and local media from 3 days ago, up until now:
+
+    `!admin media delete-past-remote-media -a 3d --yes-i-want-to-delete-local-media`
 
 **Usage:** `admin media delete-past-remote-media [OPTIONS] <DURATION>`
 
 ###### **Arguments:**
 
-* `<DURATION>` — - The relative time (e.g. 30s, 5m, 7d) within which to search
+* `<DURATION>` — - The relative time (e.g. 30s, 5m, 7d) from now within which to search
 
 ###### **Options:**
 

docs/development/code_style.md

@@ -241,7 +241,7 @@ ## Documentation
 ### Code Comments
 
 - Reference related documentation or parts of the specification
-- When a task has multiple ways of being acheved, explain your reasoning for your decision
+- When a task has multiple ways of being achieved, explain your reasoning for your decision
 - Update comments when code changes
 
 ```rs

flake.lock

@@ -1,94 +1,28 @@
 {
   "nodes": {
-    "attic": {
-      "inputs": {
-        "crane": "crane",
-        "flake-compat": "flake-compat",
-        "flake-parts": "flake-parts",
-        "nix-github-actions": "nix-github-actions",
-        "nixpkgs": "nixpkgs",
-        "nixpkgs-stable": "nixpkgs-stable"
-      },
+    "advisory-db": {
+      "flake": false,
       "locked": {
-        "lastModified": 1758711588,
-        "narHash": "sha256-0nZlCCDC5PfndsQJXXtcyrtrfW49I3KadGMDlutzaGU=",
-        "owner": "zhaofengli",
-        "repo": "attic",
-        "rev": "12cbeca141f46e1ade76728bce8adc447f2166c6",
+        "lastModified": 1761112158,
+        "narHash": "sha256-RIXu/7eyKpQHjsPuAUODO81I4ni8f+WYSb7K4mTG6+0=",
+        "owner": "rustsec",
+        "repo": "advisory-db",
+        "rev": "58f3aaec0e1776f4a900737be8cd7cb00972210d",
         "type": "github"
       },
       "original": {
-        "owner": "zhaofengli",
-        "ref": "main",
-        "repo": "attic",
-        "type": "github"
-      }
-    },
-    "cachix": {
-      "inputs": {
-        "devenv": "devenv",
-        "flake-compat": "flake-compat_2",
-        "git-hooks": "git-hooks",
-        "nixpkgs": "nixpkgs_2"
-      },
-      "locked": {
-        "lastModified": 1756385612,
-        "narHash": "sha256-+NU5MMhuPHHRyvZZWNFG7zt+leRSPsJu1MwhOUzkPUk=",
-        "owner": "cachix",
-        "repo": "cachix",
-        "rev": "dc24688cd67518c3711d511fa369c0f5a131063a",
-        "type": "github"
-      },
-      "original": {
-        "owner": "cachix",
-        "ref": "master",
-        "repo": "cachix",
-        "type": "github"
-      }
-    },
-    "cachix_2": {
-      "inputs": {
-        "devenv": [
-          "cachix",
-          "devenv"
-        ],
-        "flake-compat": [
-          "cachix",
-          "devenv"
-        ],
-        "git-hooks": [
-          "cachix",
-          "devenv",
-          "git-hooks"
-        ],
-        "nixpkgs": [
-          "cachix",
-          "devenv",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1748883665,
-        "narHash": "sha256-R0W7uAg+BLoHjMRMQ8+oiSbTq8nkGz5RDpQ+ZfxxP3A=",
-        "owner": "cachix",
-        "repo": "cachix",
-        "rev": "f707778d902af4d62d8dd92c269f8e70de09acbe",
-        "type": "github"
-      },
-      "original": {
-        "owner": "cachix",
-        "ref": "latest",
-        "repo": "cachix",
+        "owner": "rustsec",
+        "repo": "advisory-db",
         "type": "github"
       }
     },
     "crane": {
       "locked": {
-        "lastModified": 1751562746,
-        "narHash": "sha256-smpugNIkmDeicNz301Ll1bD7nFOty97T79m4GUMUczA=",
+        "lastModified": 1760924934,
+        "narHash": "sha256-tuuqY5aU7cUkR71sO2TraVKK2boYrdW3gCSXUkF4i44=",
         "owner": "ipetkov",
         "repo": "crane",
-        "rev": "aed2020fd3dc26e1e857d4107a5a67a33ab6c1fd",
+        "rev": "c6b4d5308293d0d04fcfeee92705017537cad02f",
         "type": "github"
       },
       "original": {
@@ -97,53 +31,6 @@
         "type": "github"
       }
     },
-    "crane_2": {
-      "locked": {
-        "lastModified": 1759893430,
-        "narHash": "sha256-yAy4otLYm9iZ+NtQwTMEbqHwswSFUbhn7x826RR6djw=",
-        "owner": "ipetkov",
-        "repo": "crane",
-        "rev": "1979a2524cb8c801520bd94c38bb3d5692419d93",
-        "type": "github"
-      },
-      "original": {
-        "owner": "ipetkov",
-        "ref": "master",
-        "repo": "crane",
-        "type": "github"
-      }
-    },
-    "devenv": {
-      "inputs": {
-        "cachix": "cachix_2",
-        "flake-compat": [
-          "cachix",
-          "flake-compat"
-        ],
-        "git-hooks": [
-          "cachix",
-          "git-hooks"
-        ],
-        "nix": "nix",
-        "nixpkgs": [
-          "cachix",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1754404745,
-        "narHash": "sha256-BdbW/iTImczgcuATgQIa9sPGuYIBxVq2xqcvICsa2AQ=",
-        "owner": "cachix",
-        "repo": "devenv",
-        "rev": "6563b21105168f90394dfaf58284b078af2d7275",
-        "type": "github"
-      },
-      "original": {
-        "owner": "cachix",
-        "repo": "devenv",
-        "type": "github"
-      }
-    },
     "fenix": {
       "inputs": {
         "nixpkgs": [
@@ -152,53 +39,20 @@
         "rust-analyzer-src": "rust-analyzer-src"
       },
       "locked": {
-        "lastModified": 1760337631,
-        "narHash": "sha256-3nvEN2lEpWtM1x7nfuiwpYHLNDgEUiWeBbyvy4vtVw8=",
+        "lastModified": 1761115517,
+        "narHash": "sha256-Fev/ag/c3Fp3JBwHfup3lpA5FlNXfkoshnQ7dssBgJ0=",
         "owner": "nix-community",
         "repo": "fenix",
-        "rev": "fee7cf67cbd80a74460563388ac358b394014238",
+        "rev": "320433651636186ea32b387cff05d6bbfa30cea7",
         "type": "github"
       },
       "original": {
         "owner": "nix-community",
-        "ref": "main",
         "repo": "fenix",
         "type": "github"
       }
     },
     "flake-compat": {
-      "flake": false,
-      "locked": {
-        "lastModified": 1747046372,
-        "narHash": "sha256-CIVLLkVgvHYbgI2UpXvIIBJ12HWgX+fjA8Xf8PUmqCY=",
-        "owner": "edolstra",
-        "repo": "flake-compat",
-        "rev": "9100a0f413b0c601e0533d1d94ffd501ce2e7885",
-        "type": "github"
-      },
-      "original": {
-        "owner": "edolstra",
-        "repo": "flake-compat",
-        "type": "github"
-      }
-    },
-    "flake-compat_2": {
-      "flake": false,
-      "locked": {
-        "lastModified": 1747046372,
-        "narHash": "sha256-CIVLLkVgvHYbgI2UpXvIIBJ12HWgX+fjA8Xf8PUmqCY=",
-        "owner": "edolstra",
-        "repo": "flake-compat",
-        "rev": "9100a0f413b0c601e0533d1d94ffd501ce2e7885",
-        "type": "github"
-      },
-      "original": {
-        "owner": "edolstra",
-        "repo": "flake-compat",
-        "type": "github"
-      }
-    },
-    "flake-compat_3": {
       "flake": false,
       "locked": {
         "lastModified": 1747046372,
@@ -217,17 +71,14 @@
     },
     "flake-parts": {
       "inputs": {
-        "nixpkgs-lib": [
-          "attic",
-          "nixpkgs"
-        ]
+        "nixpkgs-lib": "nixpkgs-lib"
       },
       "locked": {
-        "lastModified": 1751413152,
-        "narHash": "sha256-Tyw1RjYEsp5scoigs1384gIg6e0GoBVjms4aXFfRssQ=",
+        "lastModified": 1760948891,
+        "narHash": "sha256-TmWcdiUUaWk8J4lpjzu4gCGxWY6/Ok7mOK4fIFfBuU4=",
         "owner": "hercules-ci",
         "repo": "flake-parts",
-        "rev": "77826244401ea9de6e3bac47c2db46005e1f30b5",
+        "rev": "864599284fc7c0ba6357ed89ed5e2cd5040f0c04",
         "type": "github"
       },
       "original": {
@@ -236,214 +87,13 @@
         "type": "github"
       }
     },
-    "flake-parts_2": {
-      "inputs": {
-        "nixpkgs-lib": [
-          "cachix",
-          "devenv",
-          "nix",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1733312601,
-        "narHash": "sha256-4pDvzqnegAfRkPwO3wmwBhVi/Sye1mzps0zHWYnP88c=",
-        "owner": "hercules-ci",
-        "repo": "flake-parts",
-        "rev": "205b12d8b7cd4802fbcb8e8ef6a0f1408781a4f9",
-        "type": "github"
-      },
-      "original": {
-        "owner": "hercules-ci",
-        "repo": "flake-parts",
-        "type": "github"
-      }
-    },
-    "flake-utils": {
-      "inputs": {
-        "systems": "systems"
-      },
-      "locked": {
-        "lastModified": 1731533236,
-        "narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "ref": "main",
-        "repo": "flake-utils",
-        "type": "github"
-      }
-    },
-    "git-hooks": {
-      "inputs": {
-        "flake-compat": [
-          "cachix",
-          "flake-compat"
-        ],
-        "gitignore": "gitignore",
-        "nixpkgs": [
-          "cachix",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1750779888,
-        "narHash": "sha256-wibppH3g/E2lxU43ZQHC5yA/7kIKLGxVEnsnVK1BtRg=",
-        "owner": "cachix",
-        "repo": "git-hooks.nix",
-        "rev": "16ec914f6fb6f599ce988427d9d94efddf25fe6d",
-        "type": "github"
-      },
-      "original": {
-        "owner": "cachix",
-        "repo": "git-hooks.nix",
-        "type": "github"
-      }
-    },
-    "gitignore": {
-      "inputs": {
-        "nixpkgs": [
-          "cachix",
-          "git-hooks",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1709087332,
-        "narHash": "sha256-HG2cCnktfHsKV0s4XW83gU3F57gaTljL9KNSuG6bnQs=",
-        "owner": "hercules-ci",
-        "repo": "gitignore.nix",
-        "rev": "637db329424fd7e46cf4185293b9cc8c88c95394",
-        "type": "github"
-      },
-      "original": {
-        "owner": "hercules-ci",
-        "repo": "gitignore.nix",
-        "type": "github"
-      }
-    },
-    "nix": {
-      "inputs": {
-        "flake-compat": [
-          "cachix",
-          "devenv",
-          "flake-compat"
-        ],
-        "flake-parts": "flake-parts_2",
-        "git-hooks-nix": [
-          "cachix",
-          "devenv",
-          "git-hooks"
-        ],
-        "nixpkgs": [
-          "cachix",
-          "devenv",
-          "nixpkgs"
-        ],
-        "nixpkgs-23-11": [
-          "cachix",
-          "devenv"
-        ],
-        "nixpkgs-regression": [
-          "cachix",
-          "devenv"
-        ]
-      },
-      "locked": {
-        "lastModified": 1752773918,
-        "narHash": "sha256-dOi/M6yNeuJlj88exI+7k154z+hAhFcuB8tZktiW7rg=",
-        "owner": "cachix",
-        "repo": "nix",
-        "rev": "031c3cf42d2e9391eee373507d8c12e0f9606779",
-        "type": "github"
-      },
-      "original": {
-        "owner": "cachix",
-        "ref": "devenv-2.30",
-        "repo": "nix",
-        "type": "github"
-      }
-    },
-    "nix-filter": {
-      "locked": {
-        "lastModified": 1757882181,
-        "narHash": "sha256-+cCxYIh2UNalTz364p+QYmWHs0P+6wDhiWR4jDIKQIU=",
-        "owner": "numtide",
-        "repo": "nix-filter",
-        "rev": "59c44d1909c72441144b93cf0f054be7fe764de5",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "ref": "main",
-        "repo": "nix-filter",
-        "type": "github"
-      }
-    },
-    "nix-github-actions": {
-      "inputs": {
-        "nixpkgs": [
-          "attic",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1737420293,
-        "narHash": "sha256-F1G5ifvqTpJq7fdkT34e/Jy9VCyzd5XfJ9TO8fHhJWE=",
-        "owner": "nix-community",
-        "repo": "nix-github-actions",
-        "rev": "f4158fa080ef4503c8f4c820967d946c2af31ec9",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-community",
-        "repo": "nix-github-actions",
-        "type": "github"
-      }
-    },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1751949589,
-        "narHash": "sha256-mgFxAPLWw0Kq+C8P3dRrZrOYEQXOtKuYVlo9xvPntt8=",
+        "lastModified": 1760878510,
+        "narHash": "sha256-K5Osef2qexezUfs0alLvZ7nQFTGS9DL2oTVsIXsqLgs=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "9b008d60392981ad674e04016d25619281550a9d",
-        "type": "github"
-      },
-      "original": {
-        "owner": "NixOS",
-        "ref": "nixpkgs-unstable",
-        "repo": "nixpkgs",
-        "type": "github"
-      }
-    },
-    "nixpkgs-stable": {
-      "locked": {
-        "lastModified": 1751741127,
-        "narHash": "sha256-t75Shs76NgxjZSgvvZZ9qOmz5zuBE8buUaYD28BMTxg=",
-        "owner": "NixOS",
-        "repo": "nixpkgs",
-        "rev": "29e290002bfff26af1db6f64d070698019460302",
-        "type": "github"
-      },
-      "original": {
-        "owner": "NixOS",
-        "ref": "nixos-25.05",
-        "repo": "nixpkgs",
-        "type": "github"
-      }
-    },
-    "nixpkgs_2": {
-      "locked": {
-        "lastModified": 1754214453,
-        "narHash": "sha256-Q/I2xJn/j1wpkGhWkQnm20nShYnG7TI99foDBpXm1SY=",
-        "owner": "NixOS",
-        "repo": "nixpkgs",
-        "rev": "5b09dc45f24cf32316283e62aec81ffee3c3e376",
+        "rev": "5e2a59a5b1a82f89f2c7e598302a9cacebb72a67",
         "type": "github"
       },
       "original": {
@@ -453,42 +103,40 @@
         "type": "github"
       }
     },
-    "nixpkgs_3": {
+    "nixpkgs-lib": {
       "locked": {
-        "lastModified": 1760256791,
-        "narHash": "sha256-uTpzDHRASEDeFUuToWSQ46Re8beXyG9dx4W36FQa0/c=",
-        "owner": "NixOS",
-        "repo": "nixpkgs",
-        "rev": "832e3b6db48508ae436c2c7bfc0cf914eac6938e",
+        "lastModified": 1754788789,
+        "narHash": "sha256-x2rJ+Ovzq0sCMpgfgGaaqgBSwY+LST+WbZ6TytnT9Rk=",
+        "owner": "nix-community",
+        "repo": "nixpkgs.lib",
+        "rev": "a73b9c743612e4244d865a2fdee11865283c04e6",
         "type": "github"
       },
       "original": {
-        "owner": "NixOS",
-        "ref": "nixpkgs-unstable",
-        "repo": "nixpkgs",
+        "owner": "nix-community",
+        "repo": "nixpkgs.lib",
         "type": "github"
       }
     },
     "root": {
       "inputs": {
-        "attic": "attic",
-        "cachix": "cachix",
-        "crane": "crane_2",
+        "advisory-db": "advisory-db",
+        "crane": "crane",
         "fenix": "fenix",
-        "flake-compat": "flake-compat_3",
-        "flake-utils": "flake-utils",
-        "nix-filter": "nix-filter",
-        "nixpkgs": "nixpkgs_3"
+        "flake-compat": "flake-compat",
+        "flake-parts": "flake-parts",
+        "nixpkgs": "nixpkgs",
+        "treefmt-nix": "treefmt-nix"
       }
     },
     "rust-analyzer-src": {
       "flake": false,
       "locked": {
-        "lastModified": 1760260966,
-        "narHash": "sha256-pOVvZz/aa+laeaUKyE6PtBevdo4rywMwjhWdSZE/O1c=",
+        "lastModified": 1761077270,
+        "narHash": "sha256-O1uTuvI/rUlubJ8AXKyzh1WSWV3qCZX0huTFUvWLN4E=",
         "owner": "rust-lang",
         "repo": "rust-analyzer",
-        "rev": "c5181dbbe33af6f21b9d83e02fdb6fda298a3b65",
+        "rev": "39990a923c8bca38f5bd29dc4c96e20ee7808d5d",
         "type": "github"
       },
       "original": {
@@ -498,18 +146,23 @@
         "type": "github"
       }
     },
-    "systems": {
+    "treefmt-nix": {
+      "inputs": {
+        "nixpkgs": [
+          "nixpkgs"
+        ]
+      },
       "locked": {
-        "lastModified": 1681028828,
-        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
-        "owner": "nix-systems",
-        "repo": "default",
-        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+        "lastModified": 1760945191,
+        "narHash": "sha256-ZRVs8UqikBa4Ki3X4KCnMBtBW0ux1DaT35tgsnB1jM4=",
+        "owner": "numtide",
+        "repo": "treefmt-nix",
+        "rev": "f56b1934f5f8fcab8deb5d38d42fd692632b47c2",
         "type": "github"
       },
       "original": {
-        "owner": "nix-systems",
-        "repo": "default",
+        "owner": "numtide",
+        "repo": "treefmt-nix",
         "type": "github"
       }
     }

flake.nix

@@ -1,351 +1,46 @@
 {
+  description = "A nix flake for the continuwuity project";
+
   inputs = {
-    attic.url = "github:zhaofengli/attic?ref=main";
-    cachix.url = "github:cachix/cachix?ref=master";
-    crane = {
-      url = "github:ipetkov/crane?ref=master";
-    };
+    # basics
+    flake-parts.url = "github:hercules-ci/flake-parts";
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+
+    # for rust via nix
+    crane.url = "github:ipetkov/crane";
     fenix = {
-      url = "github:nix-community/fenix?ref=main";
+      url = "github:nix-community/fenix";
       inputs.nixpkgs.follows = "nixpkgs";
     };
+
+    # for vuln checks
+    advisory-db = {
+      url = "github:rustsec/advisory-db";
+      flake = false;
+    };
+
+    treefmt-nix = {
+      url = "github:numtide/treefmt-nix";
+      inputs.nixpkgs.follows = "nixpkgs";
+    };
+
+    # for default.nix
     flake-compat = {
       url = "github:edolstra/flake-compat?ref=master";
       flake = false;
     };
-    flake-utils.url = "github:numtide/flake-utils?ref=main";
-    nix-filter.url = "github:numtide/nix-filter?ref=main";
-    nixpkgs.url = "github:NixOS/nixpkgs?ref=nixpkgs-unstable";
+
   };
 
   outputs =
-    inputs:
-    inputs.flake-utils.lib.eachDefaultSystem (
-      system:
-      let
-        pkgsHost = import inputs.nixpkgs {
-          inherit system;
-        };
-
-        fnx = inputs.fenix.packages.${system};
-        # The Rust toolchain to use
-        toolchain = fnx.combine [
-          (fnx.fromToolchainFile {
-            file = ./rust-toolchain.toml;
-
-            # See also `rust-toolchain.toml`
-            sha256 = "sha256-+9FmLhAOezBZCOziO0Qct1NOrfpjNsXxc/8I0c7BdKE=";
-          })
-          fnx.complete.rustfmt
-        ];
-
-        mkScope =
-          pkgs:
-          pkgs.lib.makeScope pkgs.newScope (self: {
-            inherit pkgs inputs;
-            craneLib = (inputs.crane.mkLib pkgs).overrideToolchain (_: toolchain);
-            main = self.callPackage ./pkg/nix/pkgs/main { };
-            liburing = pkgs.liburing.overrideAttrs {
-              # Tests weren't building
-              outputs = [
-                "out"
-                "dev"
-                "man"
-              ];
-              buildFlags = [ "library" ];
-            };
-            rocksdb =
-              (pkgs.rocksdb_9_10.override {
-                # Override the liburing input for the build with our own so
-                # we have it built with the library flag
-                inherit (self) liburing;
-              }).overrideAttrs
-                (old: {
-                  src = pkgsHost.fetchFromGitea {
-                    domain = "forgejo.ellis.link";
-                    owner = "continuwuation";
-                    repo = "rocksdb";
-                    rev = "10.5.fb";
-                    sha256 = "sha256-X4ApGLkHF9ceBtBg77dimEpu720I79ffLoyPa8JMHaU=";
-                  };
-                  version = "v10.5.fb";
-                  cmakeFlags =
-                    pkgs.lib.subtractLists [
-                      # No real reason to have snappy or zlib, no one uses this
-                      "-DWITH_SNAPPY=1"
-                      "-DZLIB=1"
-                      "-DWITH_ZLIB=1"
-                      # We don't need to use ldb or sst_dump (core_tools)
-                      "-DWITH_CORE_TOOLS=1"
-                      # We don't need to build rocksdb tests
-                      "-DWITH_TESTS=1"
-                      # We use rust-rocksdb via C interface and don't need C++ RTTI
-                      "-DUSE_RTTI=1"
-                      # This doesn't exist in RocksDB, and USE_SSE is deprecated for
-                      # PORTABLE=$(march)
-                      "-DFORCE_SSE42=1"
-                      # PORTABLE will get set in main/default.nix
-                      "-DPORTABLE=1"
-                    ] old.cmakeFlags
-                    ++ [
-                      # No real reason to have snappy, no one uses this
-                      "-DWITH_SNAPPY=0"
-                      "-DZLIB=0"
-                      "-DWITH_ZLIB=0"
-                      # We don't need to use ldb or sst_dump (core_tools)
-                      "-DWITH_CORE_TOOLS=0"
-                      # We don't need trace tools
-                      "-DWITH_TRACE_TOOLS=0"
-                      # We don't need to build rocksdb tests
-                      "-DWITH_TESTS=0"
-                      # We use rust-rocksdb via C interface and don't need C++ RTTI
-                      "-DUSE_RTTI=0"
-                    ];
-
-                  # outputs has "tools" which we don't need or use
-                  outputs = [ "out" ];
-
-                  # preInstall hooks has stuff for messing with ldb/sst_dump which we don't need or use
-                  preInstall = "";
-
-                  # We have this already at https://forgejo.ellis.link/continuwuation/rocksdb/commit/a935c0273e1ba44eacf88ce3685a9b9831486155
-                  # Unsetting this so we don't have to revert it and make this nix exclusive
-                  patches = [ ];
-
-                  postPatch = ''
-                    # Fix gcc-13 build failures due to missing <cstdint> and
-                    # <system_error> includes, fixed upstream since 8.x
-                    sed -e '1i #include <cstdint>' -i db/compaction/compaction_iteration_stats.h
-                    sed -e '1i #include <cstdint>' -i table/block_based/data_block_hash_index.h
-                    sed -e '1i #include <cstdint>' -i util/string_util.h
-                    sed -e '1i #include <cstdint>' -i include/rocksdb/utilities/checkpoint.h
-                  '';
-                });
-          });
-
-        scopeHost = mkScope pkgsHost;
-        mkCrossScope =
-          crossSystem:
-          let
-            pkgsCrossStatic =
-              (import inputs.nixpkgs {
-                inherit system;
-                crossSystem = {
-                  config = crossSystem;
-                };
-              }).pkgsStatic;
-          in
-          mkScope pkgsCrossStatic;
-
-      in
-      {
-        packages =
-          {
-            default = scopeHost.main.override {
-              disable_features = [
-                # Don't include experimental features
-                "experimental"
-                # jemalloc profiling/stats features are expensive and shouldn't
-                # be expected on non-debug builds.
-                "jemalloc_prof"
-                "jemalloc_stats"
-                # This is non-functional on nix for some reason
-                "hardened_malloc"
-                # conduwuit_mods is a development-only hot reload feature
-                "conduwuit_mods"
-              ];
-            };
-            default-debug = scopeHost.main.override {
-              profile = "dev";
-              # Debug build users expect full logs
-              disable_release_max_log_level = true;
-              disable_features = [
-                # Don't include experimental features
-                "experimental"
-                # This is non-functional on nix for some reason
-                "hardened_malloc"
-                # conduwuit_mods is a development-only hot reload feature
-                "conduwuit_mods"
-              ];
-            };
-            # Just a test profile used for things like CI and complement
-            default-test = scopeHost.main.override {
-              profile = "test";
-              disable_release_max_log_level = true;
-              disable_features = [
-                # Don't include experimental features
-                "experimental"
-                # this is non-functional on nix for some reason
-                "hardened_malloc"
-                # conduwuit_mods is a development-only hot reload feature
-                "conduwuit_mods"
-              ];
-            };
-            all-features = scopeHost.main.override {
-              all_features = true;
-              disable_features = [
-                # Don't include experimental features
-                "experimental"
-                # jemalloc profiling/stats features are expensive and shouldn't
-                # be expected on non-debug builds.
-                "jemalloc_prof"
-                "jemalloc_stats"
-                # This is non-functional on nix for some reason
-                "hardened_malloc"
-                # conduwuit_mods is a development-only hot reload feature
-                "conduwuit_mods"
-              ];
-            };
-            all-features-debug = scopeHost.main.override {
-              profile = "dev";
-              all_features = true;
-              # Debug build users expect full logs
-              disable_release_max_log_level = true;
-              disable_features = [
-                # Don't include experimental features
-                "experimental"
-                # This is non-functional on nix for some reason
-                "hardened_malloc"
-                # conduwuit_mods is a development-only hot reload feature
-                "conduwuit_mods"
-              ];
-            };
-            hmalloc = scopeHost.main.override { features = [ "hardened_malloc" ]; };
-          }
-          // builtins.listToAttrs (
-            builtins.concatLists (
-              builtins.map
-                (
-                  crossSystem:
-                  let
-                    binaryName = "static-${crossSystem}";
-                    scopeCrossStatic = mkCrossScope crossSystem;
-                  in
-                  [
-                    # An output for a statically-linked binary
-                    {
-                      name = binaryName;
-                      value = scopeCrossStatic.main;
-                    }
-
-                    # An output for a statically-linked binary with x86_64 haswell
-                    # target optimisations
-                    {
-                      name = "${binaryName}-x86_64-haswell-optimised";
-                      value = scopeCrossStatic.main.override {
-                        x86_64_haswell_target_optimised =
-                          if (crossSystem == "x86_64-linux-gnu" || crossSystem == "x86_64-linux-musl") then true else false;
-                      };
-                    }
-
-                    # An output for a statically-linked unstripped debug ("dev") binary
-                    {
-                      name = "${binaryName}-debug";
-                      value = scopeCrossStatic.main.override {
-                        profile = "dev";
-                        # debug build users expect full logs
-                        disable_release_max_log_level = true;
-                      };
-                    }
-
-                    # An output for a statically-linked unstripped debug binary with the
-                    # "test" profile (for CI usage only)
-                    {
-                      name = "${binaryName}-test";
-                      value = scopeCrossStatic.main.override {
-                        profile = "test";
-                        disable_release_max_log_level = true;
-                        disable_features = [
-                          # dont include experimental features
-                          "experimental"
-                          # this is non-functional on nix for some reason
-                          "hardened_malloc"
-                          # conduwuit_mods is a development-only hot reload feature
-                          "conduwuit_mods"
-                        ];
-                      };
-                    }
-
-                    # An output for a statically-linked binary with `--all-features`
-                    {
-                      name = "${binaryName}-all-features";
-                      value = scopeCrossStatic.main.override {
-                        all_features = true;
-                        disable_features = [
-                          # dont include experimental features
-                          "experimental"
-                          # jemalloc profiling/stats features are expensive and shouldn't
-                          # be expected on non-debug builds.
-                          "jemalloc_prof"
-                          "jemalloc_stats"
-                          # this is non-functional on nix for some reason
-                          "hardened_malloc"
-                          # conduwuit_mods is a development-only hot reload feature
-                          "conduwuit_mods"
-                        ];
-                      };
-                    }
-
-                    # An output for a statically-linked binary with `--all-features` and with x86_64 haswell
-                    # target optimisations
-                    {
-                      name = "${binaryName}-all-features-x86_64-haswell-optimised";
-                      value = scopeCrossStatic.main.override {
-                        all_features = true;
-                        disable_features = [
-                          # dont include experimental features
-                          "experimental"
-                          # jemalloc profiling/stats features are expensive and shouldn't
-                          # be expected on non-debug builds.
-                          "jemalloc_prof"
-                          "jemalloc_stats"
-                          # this is non-functional on nix for some reason
-                          "hardened_malloc"
-                          # conduwuit_mods is a development-only hot reload feature
-                          "conduwuit_mods"
-                        ];
-                        x86_64_haswell_target_optimised =
-                          if (crossSystem == "x86_64-linux-gnu" || crossSystem == "x86_64-linux-musl") then true else false;
-                      };
-                    }
-
-                    # An output for a statically-linked unstripped debug ("dev") binary with `--all-features`
-                    {
-                      name = "${binaryName}-all-features-debug";
-                      value = scopeCrossStatic.main.override {
-                        profile = "dev";
-                        all_features = true;
-                        # debug build users expect full logs
-                        disable_release_max_log_level = true;
-                        disable_features = [
-                          # dont include experimental features
-                          "experimental"
-                          # this is non-functional on nix for some reason
-                          "hardened_malloc"
-                          # conduwuit_mods is a development-only hot reload feature
-                          "conduwuit_mods"
-                        ];
-                      };
-                    }
-
-                    # An output for a statically-linked binary with hardened_malloc
-                    {
-                      name = "${binaryName}-hmalloc";
-                      value = scopeCrossStatic.main.override {
-                        features = [ "hardened_malloc" ];
-                      };
-                    }
-                  ]
-                )
-                [
-                  #"x86_64-apple-darwin"
-                  #"aarch64-apple-darwin"
-                  "x86_64-linux-gnu"
-                  "x86_64-linux-musl"
-                  "aarch64-linux-musl"
-                ]
-            )
-          );
-      }
-    );
+    inputs@{ flake-parts, ... }:
+    flake-parts.lib.mkFlake { inherit inputs; } {
+      imports = [ ./nix ];
+      systems = [
+        # good support
+        "x86_64-linux"
+        # support untested but theoretically there
+        "aarch64-linux"
+      ];
+    };
 }

nix/checks/default.nix

@@ -0,0 +1,108 @@
+{ inputs, ... }:
+{
+  perSystem =
+    {
+      self',
+      lib,
+      pkgs,
+      ...
+    }:
+    let
+      uwulib = inputs.self.uwulib.init pkgs;
+
+      rocksdbAllFeatures = self'.packages.rocksdb.override {
+        enableJemalloc = true;
+        enableLiburing = true;
+      };
+
+      commonAttrs = (uwulib.build.commonAttrs { }) // {
+        buildInputs = [
+          pkgs.liburing
+          pkgs.rust-jemalloc-sys-unprefixed
+          rocksdbAllFeatures
+        ];
+        nativeBuildInputs = [
+          pkgs.pkg-config
+          # bindgen needs the build platform's libclang. Apparently due to "splicing
+          # weirdness", pkgs.rustPlatform.bindgenHook on its own doesn't quite do the
+          # right thing here.
+          pkgs.rustPlatform.bindgenHook
+        ];
+        env = {
+          LIBCLANG_PATH = lib.makeLibraryPath [ pkgs.llvmPackages.libclang.lib ];
+          LD_LIBRARY_PATH = lib.makeLibraryPath [
+            pkgs.liburing
+            pkgs.rust-jemalloc-sys-unprefixed
+            rocksdbAllFeatures
+          ];
+        }
+        // uwulib.environment.buildPackageEnv
+        // {
+          ROCKSDB_INCLUDE_DIR = "${rocksdbAllFeatures}/include";
+          ROCKSDB_LIB_DIR = "${rocksdbAllFeatures}/lib";
+        };
+      };
+      cargoArtifacts = self'.packages.continuwuity-all-features-deps;
+    in
+    {
+      # taken from
+      #
+      # https://crane.dev/examples/quick-start.html
+      checks = {
+        continuwuity-all-features-build = self'.packages.continuwuity-all-features-bin;
+
+        continuwuity-all-features-clippy = uwulib.build.craneLibForChecks.cargoClippy (
+          commonAttrs
+          // {
+            inherit cargoArtifacts;
+            cargoClippyExtraArgs = "-- --deny warnings";
+          }
+        );
+
+        continuwuity-all-features-docs = uwulib.build.craneLibForChecks.cargoDoc (
+          commonAttrs
+          // {
+            inherit cargoArtifacts;
+            # This can be commented out or tweaked as necessary, e.g. set to
+            # `--deny rustdoc::broken-intra-doc-links` to only enforce that lint
+            env.RUSTDOCFLAGS = "--deny warnings";
+          }
+        );
+
+        # Check formatting
+        continuwuity-all-features-fmt = uwulib.build.craneLibForChecks.cargoFmt {
+          src = uwulib.build.src;
+        };
+
+        continuwuity-all-features-toml-fmt = uwulib.build.craneLibForChecks.taploFmt {
+          src = pkgs.lib.sources.sourceFilesBySuffices uwulib.build.src [ ".toml" ];
+          # taplo arguments can be further customized below as needed
+          taploExtraArgs = "--config ${inputs.self}/taplo.toml";
+        };
+
+        # Audit dependencies
+        continuwuity-all-features-audit = uwulib.build.craneLibForChecks.cargoAudit {
+          inherit (inputs) advisory-db;
+          src = uwulib.build.src;
+        };
+
+        # Audit licenses
+        continuwuity-all-features-deny = uwulib.build.craneLibForChecks.cargoDeny {
+          src = uwulib.build.src;
+        };
+
+        # Run tests with cargo-nextest
+        # Consider setting `doCheck = false` on `continuwuity-all-features` if you do not want
+        # the tests to run twice
+        continuwuity-all-features-nextest = uwulib.build.craneLibForChecks.cargoNextest (
+          commonAttrs
+          // {
+            inherit cargoArtifacts;
+            partitions = 1;
+            partitionType = "count";
+            cargoNextestPartitionsExtraArgs = "--no-tests=pass";
+          }
+        );
+      };
+    };
+}

nix/default.nix

@@ -0,0 +1,11 @@
+{
+  imports = [
+    ./checks
+    ./packages
+    ./shells
+    ./tests
+
+    ./hydra.nix
+    ./fmt.nix
+  ];
+}

nix/fmt.nix

@@ -0,0 +1,37 @@
+{ inputs, ... }:
+{
+  # load the flake module from upstream
+  imports = [ inputs.treefmt-nix.flakeModule ];
+
+  perSystem =
+    { self', lib, ... }:
+    {
+      treefmt = {
+        # repo root as project root
+        projectRoot = inputs.self;
+
+        # the formatters
+        programs = {
+          nixfmt.enable = true;
+          typos = {
+            enable = true;
+            configFile = "${inputs.self}/.typos.toml";
+          };
+          taplo = {
+            enable = true;
+            settings = lib.importTOML "${inputs.self}/taplo.toml";
+          };
+        };
+
+        settings.formatter.rustfmt = {
+          command = "${lib.getExe' self'.packages.dev-toolchain "rustfmt"}";
+          includes = [ "**/*.rs" ];
+          options = [
+            "--unstable-features"
+            "--edition=2024"
+            "--config-path=${inputs.self}/rustfmt.toml"
+          ];
+        };
+      };
+    };
+}

nix/hydra.nix

@@ -0,0 +1,9 @@
+{ inputs, ... }:
+let
+  lib = inputs.nixpkgs.lib;
+in
+{
+  flake.hydraJobs.packages = builtins.mapAttrs (
+    _name: lib.hydraJob
+  ) inputs.self.packages.x86_64-linux;
+}

nix/packages/continuwuity/default.nix

@@ -0,0 +1,60 @@
+{ inputs, ... }:
+{
+  perSystem =
+    {
+      self',
+      lib,
+      pkgs,
+      ...
+    }:
+    let
+      uwulib = inputs.self.uwulib.init pkgs;
+    in
+    {
+      packages =
+        lib.pipe
+          [
+            # this is the default variant
+            {
+              variantName = "default";
+              commonAttrsArgs.profile = "release";
+              rocksdb = self'.packages.rocksdb;
+              features = { };
+            }
+            # this is the variant with all features enabled (liburing + jemalloc)
+            {
+              variantName = "all-features";
+              commonAttrsArgs.profile = "release";
+              rocksdb = self'.packages.rocksdb.override {
+                enableJemalloc = true;
+                enableLiburing = true;
+              };
+              features = {
+                enabledFeatures = "all";
+                disabledFeatures = uwulib.features.defaultDisabledFeatures ++ [ "bindgen-static" ];
+              };
+            }
+          ]
+          [
+            (builtins.map (cfg: rec {
+              deps = {
+                name = "continuwuity-${cfg.variantName}-deps";
+                value = uwulib.build.buildDeps {
+                  features = uwulib.features.calcFeatures cfg.features;
+                  inherit (cfg) commonAttrsArgs rocksdb;
+                };
+              };
+              bin = {
+                name = "continuwuity-${cfg.variantName}-bin";
+                value = uwulib.build.buildPackage {
+                  deps = self'.packages.${deps.name};
+                  features = uwulib.features.calcFeatures cfg.features;
+                  inherit (cfg) commonAttrsArgs rocksdb;
+                };
+              };
+            }))
+            (builtins.concatMap builtins.attrValues)
+            builtins.listToAttrs
+          ];
+    };
+}

nix/packages/default.nix

@@ -0,0 +1,14 @@
+{
+  imports = [
+    ./continuwuity
+    ./rocksdb
+    ./rust.nix
+    ./uwulib
+  ];
+
+  perSystem =
+    { self', ... }:
+    {
+      packages.default = self'.packages.continuwuity-default-bin;
+    };
+}

nix/packages/rocksdb/default.nix

@@ -0,0 +1,12 @@
+{
+  perSystem =
+    {
+      pkgs,
+      ...
+    }:
+    {
+      packages = {
+        rocksdb = pkgs.callPackage ./package.nix { };
+      };
+    };
+}

nix/packages/rocksdb/package.nix

@@ -0,0 +1,88 @@
+{
+  lib,
+  stdenv,
+
+  rocksdb,
+  liburing,
+  rust-jemalloc-sys-unprefixed,
+
+  enableJemalloc ? false,
+  enableLiburing ? false,
+
+  fetchFromGitea,
+
+  ...
+}:
+let
+  notDarwin = !stdenv.hostPlatform.isDarwin;
+in
+(rocksdb.override {
+  # Override the liburing input for the build with our own so
+  # we have it built with the library flag
+  inherit liburing;
+  jemalloc = rust-jemalloc-sys-unprefixed;
+
+  # rocksdb fails to build with prefixed jemalloc, which is required on
+  # darwin due to [1]. In this case, fall back to building rocksdb with
+  # libc malloc. This should not cause conflicts, because all of the
+  # jemalloc symbols are prefixed.
+  #
+  # [1]: https://github.com/tikv/jemallocator/blob/ab0676d77e81268cd09b059260c75b38dbef2d51/jemalloc-sys/src/env.rs#L17
+  enableJemalloc = enableJemalloc && notDarwin;
+
+  # for some reason enableLiburing in nixpkgs rocksdb is default true
+  # which breaks Darwin entirely
+  enableLiburing = enableLiburing && notDarwin;
+}).overrideAttrs
+  (old: {
+    src = fetchFromGitea {
+      domain = "forgejo.ellis.link";
+      owner = "continuwuation";
+      repo = "rocksdb";
+      rev = "10.5.fb";
+      sha256 = "sha256-X4ApGLkHF9ceBtBg77dimEpu720I79ffLoyPa8JMHaU=";
+    };
+    version = "10.5.fb";
+    cmakeFlags =
+      lib.subtractLists (builtins.map (flag: lib.cmakeBool flag true) [
+        # No real reason to have snappy or zlib, no one uses this
+        "WITH_SNAPPY"
+        "ZLIB"
+        "WITH_ZLIB"
+        # We don't need to use ldb or sst_dump (core_tools)
+        "WITH_CORE_TOOLS"
+        # We don't need to build rocksdb tests
+        "WITH_TESTS"
+        # We use rust-rocksdb via C interface and don't need C++ RTTI
+        "USE_RTTI"
+        # This doesn't exist in RocksDB, and USE_SSE is deprecated for
+        # PORTABLE=$(march)
+        "FORCE_SSE42"
+      ]) old.cmakeFlags
+      ++ (builtins.map (flag: lib.cmakeBool flag false) [
+        # No real reason to have snappy, no one uses this
+        "WITH_SNAPPY"
+        "ZLIB"
+        "WITH_ZLIB"
+        # We don't need to use ldb or sst_dump (core_tools)
+        "WITH_CORE_TOOLS"
+        # We don't need trace tools
+        "WITH_TRACE_TOOLS"
+        # We don't need to build rocksdb tests
+        "WITH_TESTS"
+        # We use rust-rocksdb via C interface and don't need C++ RTTI
+        "USE_RTTI"
+      ]);
+
+    enableLiburing = enableLiburing && notDarwin;
+
+    # outputs has "tools" which we don't need or use
+    outputs = [ "out" ];
+
+    # preInstall hooks has stuff for messing with ldb/sst_dump which we don't need or use
+    preInstall = "";
+
+    # We have this already at https://forgejo.ellis.link/continuwuation/rocksdb/commit/a935c0273e1ba44eacf88ce3685a9b9831486155
+    # Unsetting `patches` so we don't have to revert it and make this nix exclusive
+    patches = [ ];
+  })

nix/packages/rust.nix

@@ -0,0 +1,32 @@
+{ inputs, ... }:
+{
+  perSystem =
+    {
+      system,
+      lib,
+      ...
+    }:
+    {
+      packages =
+        let
+          fnx = inputs.fenix.packages.${system};
+
+          stable = fnx.fromToolchainFile {
+            file = inputs.self + "/rust-toolchain.toml";
+
+            # See also `rust-toolchain.toml`
+            sha256 = "sha256-SJwZ8g0zF2WrKDVmHrVG3pD2RGoQeo24MEXnNx5FyuI=";
+          };
+        in
+        {
+          # used for building nix stuff (doesn't include rustfmt overhead)
+          build-toolchain = stable;
+          # used for dev shells
+          dev-toolchain = fnx.combine [
+            stable
+            # use the nightly rustfmt because we use nightly features
+            fnx.complete.rustfmt
+          ];
+        };
+    };
+}

nix/packages/uwulib/build.nix

@@ -0,0 +1,108 @@
+args@{ pkgs, inputs, ... }:
+let
+  inherit (pkgs) lib;
+  uwuenv = import ./environment.nix args;
+  selfpkgs = inputs.self.packages.${pkgs.stdenv.system};
+in
+rec {
+  # basic, very minimal instance of the crane library with a minimal rust toolchain
+  craneLib = (inputs.crane.mkLib pkgs).overrideToolchain (_: selfpkgs.build-toolchain);
+  # the checks require more rust toolchain components, hence we have this separate instance of the crane library
+  craneLibForChecks = (inputs.crane.mkLib pkgs).overrideToolchain (_: selfpkgs.dev-toolchain);
+
+  # meta information (name, version, etc) of the rust crate based on the Cargo.toml
+  crateInfo = craneLib.crateNameFromCargoToml { cargoToml = "${inputs.self}/Cargo.toml"; };
+
+  src =
+    let
+      # see https://crane.dev/API.html#cranelibfiltercargosources
+      #
+      # we need to keep the `web` directory which would be filtered out by the regular source filtering function
+      #
+      # https://crane.dev/API.html#cranelibcleancargosource
+      isWebTemplate = path: _type: builtins.match ".*src/web.*" path != null;
+      isRust = craneLib.filterCargoSources;
+      isNix = path: _type: builtins.match ".+/nix.*" path != null;
+      webOrRustNotNix = p: t: !(isNix p t) && (isWebTemplate p t || isRust p t);
+    in
+    lib.cleanSourceWith {
+      src = inputs.self;
+      filter = webOrRustNotNix;
+      name = "source";
+    };
+
+  # common attrs that are shared between building continuwuity's deps and the package itself
+  commonAttrs =
+    {
+      profile ? "dev",
+      ...
+    }:
+    {
+      inherit (crateInfo)
+        pname
+        version
+        ;
+      inherit src;
+
+      # this prevents unnecessary rebuilds
+      strictDeps = true;
+
+      dontStrip = profile == "dev" || profile == "test";
+      dontPatchELF = profile == "dev" || profile == "test";
+
+      doCheck = true;
+
+      nativeBuildInputs = [
+        # bindgen needs the build platform's libclang. Apparently due to "splicing
+        # weirdness", pkgs.rustPlatform.bindgenHook on its own doesn't quite do the
+        # right thing here.
+        pkgs.rustPlatform.bindgenHook
+      ];
+    };
+
+  makeRocksDBEnv =
+    { rocksdb }:
+    {
+      ROCKSDB_INCLUDE_DIR = "${rocksdb}/include";
+      ROCKSDB_LIB_DIR = "${rocksdb}/lib";
+    };
+
+  # function that builds the continuwuity dependencies derivation
+  buildDeps =
+    {
+      rocksdb,
+      features,
+      commonAttrsArgs,
+    }:
+    craneLib.buildDepsOnly (
+      (commonAttrs commonAttrsArgs)
+      // {
+        env = uwuenv.buildDepsOnlyEnv // (makeRocksDBEnv { inherit rocksdb; });
+        inherit (features) cargoExtraArgs;
+      }
+
+    );
+
+  # function that builds the continuwuity package
+  buildPackage =
+    {
+      deps,
+      rocksdb,
+      features,
+      commonAttrsArgs,
+    }:
+    let
+      rocksdbEnv = makeRocksDBEnv { inherit rocksdb; };
+    in
+    craneLib.buildPackage (
+      (commonAttrs commonAttrsArgs)
+      // {
+        cargoArtifacts = deps;
+        doCheck = true;
+        env = uwuenv.buildPackageEnv // rocksdbEnv;
+        passthru.env = uwuenv.buildPackageEnv // rocksdbEnv;
+        meta.mainProgram = crateInfo.pname;
+        inherit (features) cargoExtraArgs;
+      }
+    );
+}

nix/packages/uwulib/default.nix

@@ -0,0 +1,10 @@
+{ inputs, ... }:
+{
+  flake.uwulib = {
+    init = pkgs: {
+      features = import ./features.nix { inherit pkgs inputs; };
+      environment = import ./environment.nix { inherit pkgs inputs; };
+      build = import ./build.nix { inherit pkgs inputs; };
+    };
+  };
+}

nix/packages/uwulib/environment.nix

@@ -0,0 +1,18 @@
+args@{ pkgs, inputs, ... }:
+let
+  uwubuild = import ./build.nix args;
+in
+rec {
+  buildDepsOnlyEnv = {
+    # https://crane.dev/faq/rebuilds-bindgen.html
+    NIX_OUTPATH_USED_AS_RANDOM_SEED = "aaaaaaaaaa";
+    CARGO_PROFILE = "release";
+  }
+  // uwubuild.craneLib.mkCrossToolchainEnv (p: pkgs.clangStdenv);
+
+  buildPackageEnv = {
+    GIT_COMMIT_HASH = inputs.self.rev or inputs.self.dirtyRev or "";
+    GIT_COMMIT_HASH_SHORT = inputs.self.shortRev or inputs.self.dirtyShortRev or "";
+  }
+  // buildDepsOnlyEnv;
+}

nix/packages/uwulib/features.nix

@@ -0,0 +1,77 @@
+{ pkgs, inputs, ... }:
+let
+  inherit (pkgs) lib;
+in
+rec {
+  defaultDisabledFeatures = [
+    # dont include experimental features
+    "experimental"
+    # jemalloc profiling/stats features are expensive and shouldn't
+    # be expected on non-debug builds.
+    "jemalloc_prof"
+    "jemalloc_stats"
+    # this is non-functional on nix for some reason
+    "hardened_malloc"
+    # conduwuit_mods is a development-only hot reload feature
+    "conduwuit_mods"
+    # we don't want to enable this feature set by default but be more specific about it
+    "full"
+  ];
+  # We perform default-feature unification in nix, because some of the dependencies
+  # on the nix side depend on feature values.
+  calcFeatures =
+    {
+      tomlPath ? "${inputs.self}/src/main",
+      # either a list of feature names or a string "all" which enables all non-default features
+      enabledFeatures ? [ ],
+      disabledFeatures ? defaultDisabledFeatures,
+      default_features ? true,
+      disable_release_max_log_level ? false,
+    }:
+    let
+      # simple helper to get the contents of a Cargo.toml file in a nix format
+      getToml = path: lib.importTOML "${path}/Cargo.toml";
+
+      # get all the features except for the default features
+      allFeatures = lib.pipe tomlPath [
+        getToml
+        (manifest: manifest.features)
+        lib.attrNames
+        (lib.remove "default")
+      ];
+
+      # get just the default enabled features
+      allDefaultFeatures = lib.pipe tomlPath [
+        getToml
+        (manifest: manifest.features.default)
+      ];
+
+      # depending on the value of enabledFeatures choose just a set or all non-default features
+      #
+      # - [ list of features ] -> choose exactly the features listed
+      # - "all" -> choose all non-default features
+      additionalFeatures = if enabledFeatures == "all" then allFeatures else enabledFeatures;
+
+      # unification with default features (if enabled)
+      features = lib.unique (additionalFeatures ++ lib.optionals default_features allDefaultFeatures);
+
+      # prepare the features that are subtracted from the set
+      disabledFeatures' =
+        disabledFeatures ++ lib.optionals disable_release_max_log_level [ "release_max_log_level" ];
+
+      # construct the final feature set
+      finalFeatures = lib.subtractLists disabledFeatures' features;
+    in
+    {
+      # final feature set, useful for querying it
+      features = finalFeatures;
+
+      # crane flag with the relevant features
+      cargoExtraArgs = builtins.concatStringsSep " " [
+        "--no-default-features"
+        "--locked"
+        (lib.optionalString (finalFeatures != [ ]) "--features")
+        (builtins.concatStringsSep "," finalFeatures)
+      ];
+    };
+}

nix/shells/default.nix

@@ -0,0 +1,29 @@
+{ inputs, ... }:
+{
+  perSystem =
+    {
+      self',
+      lib,
+      pkgs,
+      ...
+    }:
+    let
+      uwulib = inputs.self.uwulib.init pkgs;
+      rocksdbAllFeatures = self'.packages.rocksdb.override {
+        enableJemalloc = true;
+        enableLiburing = true;
+      };
+    in
+    {
+      # basic nix shell containing all things necessary to build continuwuity in all flavors manually (on x86_64-linux)
+      devShells.default = uwulib.build.craneLib.devShell {
+        packages = [
+          pkgs.pkg-config
+          pkgs.liburing
+          pkgs.rust-jemalloc-sys-unprefixed
+          rocksdbAllFeatures
+        ];
+        env.LIBCLANG_PATH = lib.makeLibraryPath [ pkgs.llvmPackages.libclang.lib ];
+      };
+    };
+}

nix/tests/default.nix

@@ -0,0 +1,124 @@
+{
+  perSystem =
+    {
+      self',
+      lib,
+      pkgs,
+      ...
+    }:
+    {
+      # run some nixos tests as checks
+      checks = lib.pipe self'.packages [
+        # we take all packages (names)
+        builtins.attrNames
+        # we filter out all packages that end with `-bin` (which we are interested in for testing)
+        (builtins.filter (lib.hasSuffix "-bin"))
+        # for each of these binaries we built the basic nixos test
+        #
+        # this test was initially yoinked from
+        #
+        # https://github.com/NixOS/nixpkgs/blob/960ce26339661b1b69c6f12b9063ca51b688615f/nixos/tests/matrix/continuwuity.nix
+        (builtins.map (name: {
+          name = "test-${name}";
+          value = pkgs.testers.runNixOSTest {
+            inherit name;
+
+            nodes = {
+              continuwuity = {
+                services.matrix-continuwuity = {
+                  enable = true;
+                  package = self'.packages.${name};
+                  settings.global = {
+                    server_name = name;
+                    address = [ "0.0.0.0" ];
+                    allow_registration = true;
+                    yes_i_am_very_very_sure_i_want_an_open_registration_server_prone_to_abuse = true;
+                  };
+                  extraEnvironment.RUST_BACKTRACE = "yes";
+                };
+                networking.firewall.allowedTCPPorts = [ 6167 ];
+              };
+              client =
+                { pkgs, ... }:
+                {
+                  environment.systemPackages = [
+                    (pkgs.writers.writePython3Bin "do_test" { libraries = [ pkgs.python3Packages.matrix-nio ]; } ''
+                      import asyncio
+                      import nio
+
+
+                      async def main() -> None:
+                          # Connect to continuwuity
+                          client = nio.AsyncClient("http://continuwuity:6167", "alice")
+
+                          # Register as user alice
+                          response = await client.register("alice", "my-secret-password")
+
+                          # Log in as user alice
+                          response = await client.login("my-secret-password")
+
+                          # Create a new room
+                          response = await client.room_create(federate=False)
+                          print("Matrix room create response:", response)
+                          assert isinstance(response, nio.RoomCreateResponse)
+                          room_id = response.room_id
+
+                          # Join the room
+                          response = await client.join(room_id)
+                          print("Matrix join response:", response)
+                          assert isinstance(response, nio.JoinResponse)
+
+                          # Send a message to the room
+                          response = await client.room_send(
+                              room_id=room_id,
+                              message_type="m.room.message",
+                              content={
+                                  "msgtype": "m.text",
+                                  "body": "Hello continuwuity!"
+                              }
+                          )
+                          print("Matrix room send response:", response)
+                          assert isinstance(response, nio.RoomSendResponse)
+
+                          # Sync responses
+                          response = await client.sync(timeout=30000)
+                          print("Matrix sync response:", response)
+                          assert isinstance(response, nio.SyncResponse)
+
+                          # Check the message was received by continuwuity
+                          last_message = response.rooms.join[room_id].timeline.events[-1].body
+                          assert last_message == "Hello continuwuity!"
+
+                          # Leave the room
+                          response = await client.room_leave(room_id)
+                          print("Matrix room leave response:", response)
+                          assert isinstance(response, nio.RoomLeaveResponse)
+
+                          # Close the client
+                          await client.close()
+
+
+                      if __name__ == "__main__":
+                          asyncio.run(main())
+                    '')
+                  ];
+                };
+            };
+
+            testScript = ''
+              start_all()
+
+              with subtest("start continuwuity"):
+                    continuwuity.wait_for_unit("continuwuity.service")
+                    continuwuity.wait_for_open_port(6167)
+
+              with subtest("ensure messages can be exchanged"):
+                    client.succeed("do_test >&2")
+            '';
+
+          };
+        }))
+        builtins.listToAttrs
+      ];
+    };
+}

pkg/conduwuit.service

@@ -12,13 +12,14 @@ Group=conduwuit
 Type=notify-reload
 ReloadSignal=SIGUSR1
 
-Environment="CONTINUWUITY_CONFIG=/etc/conduwuit/conduwuit.toml"
-
 Environment="CONTINUWUITY_LOG_TO_JOURNALD=true"
 Environment="CONTINUWUITY_JOURNALD_IDENTIFIER=%N"
-Environment="CONTINUWUITY_DATABASE_PATH=/var/lib/conduwuit"
+Environment="CONTINUWUITY_DATABASE_PATH=%S/conduwuit"
+Environment="CONTINUWUITY_CONFIG_RELOAD_SIGNAL=true"
 
-ExecStart=/usr/bin/conduwuit
+LoadCredential=conduwuit.toml:/etc/conduwuit/conduwuit.toml
+
+ExecStart=/usr/bin/conduwuit --config ${CREDENTIALS_DIRECTORY}/conduwuit.toml
 
 AmbientCapabilities=
 CapabilityBoundingSet=
@@ -52,8 +53,9 @@ SystemCallFilter=@system-service @resources
 SystemCallFilter=~@clock @debug @module @mount @reboot @swap @cpu-emulation @obsolete @timer @chown @setuid @privileged @keyring @ipc
 SystemCallErrorNumber=EPERM
 
+# ConfigurationDirectory isn't specified here because it's created by
+# the distro's package manager.
 StateDirectory=conduwuit
-ConfigurationDirectory=conduwuit
 RuntimeDirectory=conduwuit
 RuntimeDirectoryMode=0750
 
@@ -61,7 +63,7 @@ Restart=on-failure
 RestartSec=5
 
 TimeoutStopSec=4m
-TimeoutStartSec=4m
+TimeoutStartSec=10m
 
 StartLimitInterval=1m
 StartLimitBurst=5

pkg/fedora/continuwuity.spec.rpkg

@@ -51,7 +51,7 @@ find .cargo/registry/ -executable -name "*.rs" -exec chmod -x {} +
 %install
 install -Dpm0755 target/rpm/conduwuit -t %{buildroot}%{_bindir}
 install -Dpm0644 pkg/conduwuit.service -t %{buildroot}%{_unitdir}
-install -Dpm0644 conduwuit-example.toml %{buildroot}%{_sysconfdir}/conduwuit/conduwuit.toml
+install -Dpm0600 conduwuit-example.toml %{buildroot}%{_sysconfdir}/conduwuit/conduwuit.toml
 
 %files
 %license LICENSE
@@ -60,7 +60,7 @@ install -Dpm0644 conduwuit-example.toml %{buildroot}%{_sysconfdir}/conduwuit/con
 %doc CONTRIBUTING.md
 %doc README.md
 %doc SECURITY.md
-%config %{_sysconfdir}/conduwuit/conduwuit.toml
+%config(noreplace) %{_sysconfdir}/conduwuit/conduwuit.toml
 
 %{_bindir}/conduwuit
 %{_unitdir}/conduwuit.service

pkg/nix/pkgs/main/cross-compilation-env.nix

@@ -1,83 +0,0 @@
-{ lib
-, pkgsBuildHost
-, rust
-, stdenv
-}:
-
-lib.optionalAttrs stdenv.hostPlatform.isStatic
-  {
-    ROCKSDB_STATIC = "";
-  }
-//
-{
-  CARGO_BUILD_RUSTFLAGS =
-    lib.concatStringsSep
-      " "
-      (lib.optionals
-        stdenv.hostPlatform.isStatic
-        [ "-C" "relocation-model=static" ]
-      ++ lib.optionals
-        (stdenv.buildPlatform.config != stdenv.hostPlatform.config)
-        [
-          "-l"
-          "c"
-
-          "-l"
-          "stdc++"
-
-          "-L"
-          "${stdenv.cc.cc.lib}/${stdenv.hostPlatform.config}/lib"
-        ]
-      );
-}
-
-  # What follows is stolen from [here][0]. Its purpose is to properly
-  # configure compilers and linkers for various stages of the build, and
-  # even covers the case of build scripts that need native code compiled and
-  # run on the build platform (I think).
-  #
-  # [0]: https://github.com/NixOS/nixpkgs/blob/nixpkgs-unstable/pkgs/build-support/rust/lib/default.nix#L48-L68
-  //
-(
-  let
-    inherit (rust.lib) envVars;
-  in
-  lib.optionalAttrs
-    (stdenv.targetPlatform.rust.rustcTarget
-    != stdenv.hostPlatform.rust.rustcTarget)
-    (
-      let
-        inherit (stdenv.targetPlatform.rust) cargoEnvVarTarget;
-      in
-      {
-        "CC_${cargoEnvVarTarget}" = envVars.ccForTarget;
-        "CXX_${cargoEnvVarTarget}" = envVars.cxxForTarget;
-        "CARGO_TARGET_${cargoEnvVarTarget}_LINKER" = envVars.ccForTarget;
-      }
-    )
-  //
-  (
-    let
-      inherit (stdenv.hostPlatform.rust) cargoEnvVarTarget rustcTarget;
-    in
-    {
-      "CC_${cargoEnvVarTarget}" = envVars.ccForHost;
-      "CXX_${cargoEnvVarTarget}" = envVars.cxxForHost;
-      "CARGO_TARGET_${cargoEnvVarTarget}_LINKER" = envVars.ccForHost;
-      CARGO_BUILD_TARGET = rustcTarget;
-    }
-  )
-  //
-  (
-    let
-      inherit (stdenv.buildPlatform.rust) cargoEnvVarTarget;
-    in
-    {
-      "CC_${cargoEnvVarTarget}" = envVars.ccForBuild;
-      "CXX_${cargoEnvVarTarget}" = envVars.cxxForBuild;
-      "CARGO_TARGET_${cargoEnvVarTarget}_LINKER" = envVars.ccForBuild;
-      HOST_CC = "${pkgsBuildHost.stdenv.cc}/bin/cc";
-      HOST_CXX = "${pkgsBuildHost.stdenv.cc}/bin/c++";
-    }
-  )
-)

pkg/nix/pkgs/main/default.nix

@@ -1,224 +0,0 @@
-# Dependencies (keep sorted)
-{ craneLib
-, inputs
-, jq
-, lib
-, libiconv
-, liburing
-, pkgsBuildHost
-, rocksdb
-, removeReferencesTo
-, rust
-, rust-jemalloc-sys
-, stdenv
-
-  # Options (keep sorted)
-, all_features ? false
-, default_features ? true
-  # default list of disabled features
-, disable_features ? [
-    # dont include experimental features
-    "experimental"
-    # jemalloc profiling/stats features are expensive and shouldn't
-    # be expected on non-debug builds.
-    "jemalloc_prof"
-    "jemalloc_stats"
-    # this is non-functional on nix for some reason
-    "hardened_malloc"
-    # conduwuit_mods is a development-only hot reload feature
-    "conduwuit_mods"
-  ]
-, disable_release_max_log_level ? false
-, features ? [ ]
-, profile ? "release"
-  # rocksdb compiled with -march=haswell and target-cpu=haswell rustflag
-  # haswell is pretty much any x86 cpu made in the last 12 years, and
-  # supports modern CPU extensions that rocksdb can make use of.
-  # disable if trying to make a portable x86_64 build for very old hardware
-, x86_64_haswell_target_optimised ? false
-}:
-
-let
-  # We perform default-feature unification in nix, because some of the dependencies
-  # on the nix side depend on feature values.
-  crateFeatures = path:
-    let manifest = lib.importTOML "${path}/Cargo.toml"; in
-    lib.remove "default" (lib.attrNames manifest.features);
-  crateDefaultFeatures = path:
-    (lib.importTOML "${path}/Cargo.toml").features.default;
-  allDefaultFeatures = crateDefaultFeatures "${inputs.self}/src/main";
-  allFeatures = crateFeatures "${inputs.self}/src/main";
-  features' = lib.unique
-    (features ++
-      lib.optionals default_features allDefaultFeatures ++
-      lib.optionals all_features allFeatures);
-  disable_features' = disable_features ++ lib.optionals disable_release_max_log_level [ "release_max_log_level" ];
-  features'' = lib.subtractLists disable_features' features';
-
-  featureEnabled = feature: builtins.elem feature features'';
-
-  enableLiburing = featureEnabled "io_uring" && !stdenv.hostPlatform.isDarwin;
-
-  # This derivation will set the JEMALLOC_OVERRIDE variable, causing the
-  # tikv-jemalloc-sys crate to use the nixpkgs jemalloc instead of building it's
-  # own. In order for this to work, we need to set flags on the build that match
-  # whatever flags tikv-jemalloc-sys was going to use. These are dependent on
-  # which features we enable in tikv-jemalloc-sys.
-  rust-jemalloc-sys' = (rust-jemalloc-sys.override {
-    # tikv-jemalloc-sys/unprefixed_malloc_on_supported_platforms feature
-    unprefixed = true;
-  }).overrideAttrs (old: {
-    configureFlags = old.configureFlags ++
-      # we dont need docs
-      [ "--disable-doc" ] ++
-      # we dont need cxx/C++ integration
-      [ "--disable-cxx" ] ++
-      # tikv-jemalloc-sys/profiling feature
-      lib.optional (featureEnabled "jemalloc_prof") "--enable-prof" ++
-      # tikv-jemalloc-sys/stats feature
-      (if (featureEnabled "jemalloc_stats") then [ "--enable-stats" ] else [ "--disable-stats" ]);
-  });
-
-  buildDepsOnlyEnv =
-    let
-      rocksdb' = (rocksdb.override {
-        jemalloc = lib.optional (featureEnabled "jemalloc") rust-jemalloc-sys';
-        # rocksdb fails to build with prefixed jemalloc, which is required on
-        # darwin due to [1]. In this case, fall back to building rocksdb with
-        # libc malloc. This should not cause conflicts, because all of the
-        # jemalloc symbols are prefixed.
-        #
-        # [1]: https://github.com/tikv/jemallocator/blob/ab0676d77e81268cd09b059260c75b38dbef2d51/jemalloc-sys/src/env.rs#L17
-        enableJemalloc = featureEnabled "jemalloc" && !stdenv.hostPlatform.isDarwin;
-
-        # for some reason enableLiburing in nixpkgs rocksdb is default true
-        # which breaks Darwin entirely
-        inherit enableLiburing;
-      }).overrideAttrs (old: {
-        inherit enableLiburing;
-        cmakeFlags = (if x86_64_haswell_target_optimised then
-          (lib.subtractLists [
-            # dont make a portable build if x86_64_haswell_target_optimised is enabled
-            "-DPORTABLE=1"
-          ]
-            old.cmakeFlags
-          ++ [ "-DPORTABLE=haswell" ]) else [ "-DPORTABLE=1" ]
-        )
-        ++ old.cmakeFlags;
-
-        # outputs has "tools" which we dont need or use
-        outputs = [ "out" ];
-
-        # preInstall hooks has stuff for messing with ldb/sst_dump which we dont need or use
-        preInstall = "";
-      });
-    in
-    {
-      # https://crane.dev/faq/rebuilds-bindgen.html
-      NIX_OUTPATH_USED_AS_RANDOM_SEED = "aaaaaaaaaa";
-
-      CARGO_PROFILE = profile;
-      ROCKSDB_INCLUDE_DIR = "${rocksdb'}/include";
-      ROCKSDB_LIB_DIR = "${rocksdb'}/lib";
-    }
-    //
-    (import ./cross-compilation-env.nix {
-      # Keep sorted
-      inherit
-        lib
-        pkgsBuildHost
-        rust
-        stdenv;
-    });
-
-  buildPackageEnv = {
-    GIT_COMMIT_HASH = inputs.self.rev or inputs.self.dirtyRev or "";
-    GIT_COMMIT_HASH_SHORT = inputs.self.shortRev or inputs.self.dirtyShortRev or "";
-  } // buildDepsOnlyEnv // {
-    # Only needed in static stdenv because these are transitive dependencies of rocksdb
-    CARGO_BUILD_RUSTFLAGS = buildDepsOnlyEnv.CARGO_BUILD_RUSTFLAGS
-      + lib.optionalString (enableLiburing && stdenv.hostPlatform.isStatic)
-      " -L${lib.getLib liburing}/lib -luring"
-      + lib.optionalString x86_64_haswell_target_optimised
-      " -Ctarget-cpu=haswell";
-  };
-
-
-
-  commonAttrs = {
-    inherit
-      (craneLib.crateNameFromCargoToml {
-        cargoToml = "${inputs.self}/Cargo.toml";
-      })
-      pname
-      version;
-
-    src = let filter = inputs.nix-filter.lib; in filter {
-      root = inputs.self;
-
-      # Keep sorted
-      include = [
-        ".cargo"
-        "Cargo.lock"
-        "Cargo.toml"
-        "src"
-        "xtask"
-      ];
-    };
-
-    doCheck = true;
-
-    cargoExtraArgs = "--no-default-features --locked "
-      + lib.optionalString
-      (features'' != [ ])
-      "--features " + (builtins.concatStringsSep "," features'');
-
-    dontStrip = profile == "dev" || profile == "test";
-    dontPatchELF = profile == "dev" || profile == "test";
-
-    buildInputs = lib.optional (featureEnabled "jemalloc") rust-jemalloc-sys'
-      # needed to build Rust applications on macOS
-      ++ lib.optionals stdenv.hostPlatform.isDarwin [
-      # https://github.com/NixOS/nixpkgs/issues/206242
-      # ld: library not found for -liconv
-      libiconv
-      # https://stackoverflow.com/questions/69869574/properly-adding-darwin-apple-sdk-to-a-nix-shell
-      # https://discourse.nixos.org/t/compile-a-rust-binary-on-macos-dbcrossbar/8612
-      pkgsBuildHost.darwin.apple_sdk.frameworks.Security
-    ];
-
-    nativeBuildInputs = [
-      # bindgen needs the build platform's libclang. Apparently due to "splicing
-      # weirdness", pkgs.rustPlatform.bindgenHook on its own doesn't quite do the
-      # right thing here.
-      pkgsBuildHost.rustPlatform.bindgenHook
-
-      # We don't actually depend on `jq`, but crane's `buildPackage` does, but
-      # its `buildDepsOnly` doesn't. This causes those two derivations to have
-      # differing values for `NIX_CFLAGS_COMPILE`, which contributes to spurious
-      # rebuilds of bindgen and its depedents.
-      jq
-    ];
-  };
-in
-
-craneLib.buildPackage (commonAttrs // {
-  cargoArtifacts = craneLib.buildDepsOnly (commonAttrs // {
-    env = buildDepsOnlyEnv;
-  });
-
-  doCheck = true;
-
-  cargoExtraArgs = "--no-default-features --locked "
-    + lib.optionalString
-    (features'' != [ ])
-    "--features " + (builtins.concatStringsSep "," features'');
-
-  env = buildPackageEnv;
-
-  passthru = {
-    env = buildPackageEnv;
-  };
-
-  meta.mainProgram = commonAttrs.pname;
-})

renovate.json

@@ -10,15 +10,19 @@
     "nix": {
         "enabled": true
     },
+    "pre-commit": {
+        "enabled": true
+    },
     "labels": ["Dependencies", "Dependencies/Renovate"],
     "ignoreDeps": [
         "tikv-jemallocator",
         "tikv-jemalloc-sys",
         "tikv-jemalloc-ctl",
-        "opentelemetry",
-        "opentelemetry_sdk",
-        "opentelemetry-jaeger",
-        "tracing-opentelemetry"
+        "rustyline-async",
+        "event-listener",
+        "async-channel",
+        "core_affinity",
+        "hyper-util"
     ],
     "github-actions": {
         "enabled": true,

rust-toolchain.toml

@@ -10,7 +10,7 @@
 
 [toolchain]
 profile = "minimal"
-channel = "1.89.0"
+channel = "1.90.0"
 components = [
     # For rust-analyzer
     "rust-src",

src/admin/Cargo.toml

@@ -85,7 +85,7 @@ futures.workspace = true
 log.workspace = true
 ruma.workspace = true
 serde_json.workspace = true
-serde_yml.workspace = true
+serde-saphyr.workspace = true
 tokio.workspace = true
 tracing-subscriber.workspace = true
 tracing.workspace = true

src/admin/appservice/commands.rs

@@ -16,7 +16,7 @@ pub(super) async fn register(&self) -> Result {
 
 	let range = 1..checked!(body_len - 1)?;
 	let appservice_config_body = body[range].join("\n");
-	let parsed_config = serde_yml::from_str(&appservice_config_body);
+	let parsed_config = serde_saphyr::from_str(&appservice_config_body);
 	match parsed_config {
 		| Err(e) => return Err!("Could not parse appservice config as YAML: {e}"),
 		| Ok(registration) => match self
@@ -57,7 +57,7 @@ pub(super) async fn show_appservice_config(&self, appservice_identifier: String)
 	{
 		| None => return Err!("Appservice does not exist."),
 		| Some(config) => {
-			let config_str = serde_yml::to_string(&config)?;
+			let config_str = serde_saphyr::to_string(&config)?;
 			write!(self, "Config for {appservice_identifier}:\n\n```yaml\n{config_str}\n```")
 		},
 	}

src/admin/media/commands.rs

@@ -2,7 +2,8 @@
 
 use conduwuit::{
 	Err, Result, debug, debug_info, debug_warn, error, info, trace,
-	utils::time::parse_timepoint_ago, warn,
+	utils::time::{TimeDirection, parse_timepoint_ago},
+	warn,
 };
 use conduwuit_service::media::Dim;
 use ruma::{Mxc, OwnedEventId, OwnedMxcUri, OwnedServerName};
@@ -235,14 +236,19 @@ pub(super) async fn delete_past_remote_media(
 	}
 	assert!(!(before && after), "--before and --after should not be specified together");
 
-	let duration = parse_timepoint_ago(&duration)?;
+	let direction = if after {
+		TimeDirection::After
+	} else {
+		TimeDirection::Before
+	};
+
+	let time_boundary = parse_timepoint_ago(&duration)?;
 	let deleted_count = self
 		.services
 		.media
-		.delete_all_remote_media_at_after_time(
-			duration,
-			before,
-			after,
+		.delete_all_media_within_timeframe(
+			time_boundary,
+			direction,
 			yes_i_want_to_delete_local_media,
 		)
 		.await?;

src/admin/media/mod.rs

@@ -27,12 +27,24 @@ pub enum MediaCommand {
 	///   filesystem. This will always ignore errors.
 	DeleteList,
 
-	/// - Deletes all remote (and optionally local) media created before or
-	///   after [duration] time using filesystem metadata first created at date,
-	///   or fallback to last modified date. This will always ignore errors by
-	///   default.
+	/// Deletes all remote (and optionally local) media created before/after
+	/// [duration] ago, using filesystem metadata first created at date, or
+	/// fallback to last modified date. This will always ignore errors by
+	/// default.
+	///
+	/// * Examples:
+	///   * Delete all remote media older than a year:
+	///
+	///     `!admin media delete-past-remote-media -b 1y`
+	///
+	///   * Delete all remote and local media from 3 days ago, up until now:
+	///
+	///     `!admin media delete-past-remote-media -a 3d
+	/// --yes-i-want-to-delete-local-media`
+	#[command(verbatim_doc_comment)]
 	DeletePastRemoteMedia {
-		/// - The relative time (e.g. 30s, 5m, 7d) within which to search
+		/// - The relative time (e.g. 30s, 5m, 7d) from now within which to
+		///   search
 		duration: String,
 
 		/// - Only delete media created before [duration] ago

src/admin/query/account_data.rs

@@ -41,7 +41,7 @@ async fn changes_since(
 	let results: Vec<_> = self
 		.services
 		.account_data
-		.changes_since(room_id.as_deref(), &user_id, since, None)
+		.changes_since(room_id.as_deref(), &user_id, Some(since), None)
 		.collect()
 		.await;
 	let query_time = timer.elapsed();

src/api/client/keys.rs

@@ -389,7 +389,7 @@ pub(crate) async fn get_key_changes_route(
 	device_list_updates.extend(
 		services
 			.users
-			.keys_changed(sender_user, from, Some(to))
+			.keys_changed(sender_user, Some(from), Some(to))
 			.map(ToOwned::to_owned)
 			.collect::<Vec<_>>()
 			.await,
@@ -401,7 +401,7 @@ pub(crate) async fn get_key_changes_route(
 		device_list_updates.extend(
 			services
 				.users
-				.room_keys_changed(room_id, from, Some(to))
+				.room_keys_changed(room_id, Some(from), Some(to))
 				.map(|(user_id, _)| user_id)
 				.map(ToOwned::to_owned)
 				.collect::<Vec<_>>()

src/api/client/membership/knock.rs

@@ -5,7 +5,7 @@
 use conduwuit::{
 	Err, Result, debug, debug_info, debug_warn, err, info,
 	matrix::{
-		event::{Event, gen_event_id},
+		event::gen_event_id,
 		pdu::{PduBuilder, PduEvent},
 	},
 	result::FlatOk,
@@ -458,7 +458,7 @@ async fn knock_room_helper_local(
 			.await,
 	};
 
-	let send_knock_response = services
+	services
 		.sending
 		.send_federation_request(&remote_server, send_knock_request)
 		.await?;
@@ -477,20 +477,14 @@ async fn knock_room_helper_local(
 		.map_err(|e| err!(BadServerResponse("Invalid knock event PDU: {e:?}")))?;
 
 	info!("Updating membership locally to knock state with provided stripped state events");
+	// TODO: this call does not appear to do anything because `update_membership`
+	// doesn't call `mark_as_knock`. investigate further, ideally with the aim of
+	// removing this call entirely -- Ginger thinks `update_membership` should only
+	// be called from `force_state` and `append_pdu`.
 	services
 		.rooms
 		.state_cache
-		.update_membership(
-			room_id,
-			sender_user,
-			parsed_knock_pdu
-				.get_content::<RoomMemberEventContent>()
-				.expect("we just created this"),
-			sender_user,
-			Some(send_knock_response.knock_room_state),
-			None,
-			false,
-		)
+		.update_membership(room_id, sender_user, &parsed_knock_pdu, false)
 		.await?;
 
 	info!("Appending room knock event locally");
@@ -677,20 +671,11 @@ async fn knock_room_helper_remote(
 		.await?;
 
 	info!("Updating membership locally to knock state with provided stripped state events");
+	// TODO: see TODO on the other call to `update_membership`
 	services
 		.rooms
 		.state_cache
-		.update_membership(
-			room_id,
-			sender_user,
-			parsed_knock_pdu
-				.get_content::<RoomMemberEventContent>()
-				.expect("we just created this"),
-			sender_user,
-			Some(send_knock_response.knock_room_state),
-			None,
-			false,
-		)
+		.update_membership(room_id, sender_user, &parsed_knock_pdu, false)
 		.await?;
 
 	info!("Appending room knock event locally");

src/api/client/membership/leave.rs

@@ -2,12 +2,12 @@
 
 use axum::extract::State;
 use conduwuit::{
-	Err, Result, debug_info, debug_warn, err,
+	Err, Pdu, Result, debug_info, debug_warn, err,
 	matrix::{event::gen_event_id, pdu::PduBuilder},
 	utils::{self, FutureBoolExt, future::ReadyEqExt},
 	warn,
 };
-use futures::{FutureExt, StreamExt, TryFutureExt, pin_mut};
+use futures::{FutureExt, StreamExt, pin_mut};
 use ruma::{
 	CanonicalJsonObject, CanonicalJsonValue, OwnedServerName, RoomId, RoomVersionId, UserId,
 	api::{
@@ -81,42 +81,9 @@ pub async fn leave_room(
 	room_id: &RoomId,
 	reason: Option<String>,
 ) -> Result {
-	let default_member_content = RoomMemberEventContent {
-		membership: MembershipState::Leave,
-		reason: reason.clone(),
-		join_authorized_via_users_server: None,
-		is_direct: None,
-		avatar_url: None,
-		displayname: None,
-		third_party_invite: None,
-		blurhash: None,
-		redact_events: None,
-	};
-
 	let is_banned = services.rooms.metadata.is_banned(room_id);
 	let is_disabled = services.rooms.metadata.is_disabled(room_id);
 
-	pin_mut!(is_banned, is_disabled);
-	if is_banned.or(is_disabled).await {
-		// the room is banned/disabled, the room must be rejected locally since we
-		// cant/dont want to federate with this server
-		services
-			.rooms
-			.state_cache
-			.update_membership(
-				room_id,
-				user_id,
-				default_member_content,
-				user_id,
-				None,
-				None,
-				true,
-			)
-			.await?;
-
-		return Ok(());
-	}
-
 	let dont_have_room = services
 		.rooms
 		.state_cache
@@ -129,43 +96,41 @@ pub async fn leave_room(
 		.is_knocked(user_id, room_id)
 		.eq(&false);
 
-	// Ask a remote server if we don't have this room and are not knocking on it
-	if dont_have_room.and(not_knocked).await {
-		if let Err(e) = remote_leave_room(services, user_id, room_id, reason.clone())
-			.boxed()
-			.await
-		{
-			warn!(%user_id, "Failed to leave room {room_id} remotely: {e}");
-			// Don't tell the client about this error
-		}
+	pin_mut!(is_banned, is_disabled);
 
-		let last_state = services
-			.rooms
-			.state_cache
-			.invite_state(user_id, room_id)
-			.or_else(|_| services.rooms.state_cache.knock_state(user_id, room_id))
-			.or_else(|_| services.rooms.state_cache.left_state(user_id, room_id))
-			.await
-			.ok();
+	/*
+	there are three possible cases when leaving a room:
+	1. the room is banned or disabled, so we're not federating with it.
+	2. nobody on the homeserver is in the room, which can happen if the user is rejecting an invite
+	   to a room that we don't have any members in.
+	3. someone else on the homeserver is in the room. in this case we can leave like normal by sending a PDU over federation.
 
-		// We always drop the invite, we can't rely on other servers
-		services
-			.rooms
-			.state_cache
-			.update_membership(
-				room_id,
-				user_id,
-				default_member_content,
-				user_id,
-				last_state,
-				None,
-				true,
-			)
-			.await?;
+	in cases 1 and 2, we have to update the state cache using `mark_as_left` directly.
+	otherwise `build_and_append_pdu` will take care of updating the state cache for us.
+	*/
+
+	// `leave_pdu` is the outlier `m.room.member` event which will be synced to the
+	// user. if it's None the sync handler will create a dummy PDU.
+	let leave_pdu = if is_banned.or(is_disabled).await {
+		// case 1: the room is banned/disabled. we don't want to federate with another
+		// server to leave, so we can't create an outlier PDU.
+		None
+	} else if dont_have_room.and(not_knocked).await {
+		// case 2: ask a remote server to assist us with leaving
+		// we always mark the room as left locally, regardless of if the federated leave
+		// failed
+
+		remote_leave_room(services, user_id, room_id, reason.clone())
+			.await
+			.inspect_err(|err| {
+				warn!(%user_id, "Failed to leave room {room_id} remotely: {err}");
+			})
+			.ok()
 	} else {
+		// case 3: we can leave by sending a PDU.
 		let state_lock = services.rooms.state.mutex.lock(room_id).await;
 
-		let Ok(event) = services
+		let user_member_event_content = services
 			.rooms
 			.state_accessor
 			.room_state_get_content::<RoomMemberEventContent>(
@@ -173,44 +138,61 @@ pub async fn leave_room(
 				&StateEventType::RoomMember,
 				user_id.as_str(),
 			)
-			.await
-		else {
-			debug_warn!(
-				"Trying to leave a room you are not a member of, marking room as left locally."
-			);
+			.await;
 
-			return services
-				.rooms
-				.state_cache
-				.update_membership(
-					room_id,
-					user_id,
-					default_member_content,
-					user_id,
-					None,
-					None,
-					true,
-				)
-				.await;
-		};
+		match user_member_event_content {
+			| Ok(content) => {
+				services
+					.rooms
+					.timeline
+					.build_and_append_pdu(
+						PduBuilder::state(user_id.to_string(), &RoomMemberEventContent {
+							membership: MembershipState::Leave,
+							reason,
+							join_authorized_via_users_server: None,
+							is_direct: None,
+							..content
+						}),
+						user_id,
+						Some(room_id),
+						&state_lock,
+					)
+					.await?;
 
-		services
-			.rooms
-			.timeline
-			.build_and_append_pdu(
-				PduBuilder::state(user_id.to_string(), &RoomMemberEventContent {
-					membership: MembershipState::Leave,
-					reason,
-					join_authorized_via_users_server: None,
-					is_direct: None,
-					..event
-				}),
-				user_id,
-				Some(room_id),
-				&state_lock,
-			)
-			.await?;
-	}
+				// `build_and_append_pdu` calls `mark_as_left` internally, so we return early.
+				return Ok(());
+			},
+			| Err(_) => {
+				// an exception to case 3 is if the user isn't even in the room they're trying
+				// to leave. this can happen if the client's caching is wrong.
+				debug_warn!(
+					"Trying to leave a room you are not a member of, marking room as left \
+					 locally."
+				);
+
+				// return the existing leave state, if one exists. `mark_as_left` will then
+				// update the `roomuserid_leftcount` table, making the leave come down sync
+				// again.
+				services
+					.rooms
+					.state_cache
+					.left_state(user_id, room_id)
+					.await?
+			},
+		}
+	};
+
+	services
+		.rooms
+		.state_cache
+		.mark_as_left(user_id, room_id, leave_pdu)
+		.await;
+
+	services
+		.rooms
+		.state_cache
+		.update_joined_count(room_id)
+		.await;
 
 	Ok(())
 }
@@ -220,7 +202,7 @@ pub async fn remote_leave_room(
 	user_id: &UserId,
 	room_id: &RoomId,
 	reason: Option<String>,
-) -> Result<()> {
+) -> Result<Pdu> {
 	let mut make_leave_response_and_server =
 		Err!(BadServerResponse("No remote server available to assist in leaving {room_id}."));
 
@@ -373,7 +355,7 @@ pub async fn remote_leave_room(
 			&remote_server,
 			federation::membership::create_leave_event::v2::Request {
 				room_id: room_id.to_owned(),
-				event_id,
+				event_id: event_id.clone(),
 				pdu: services
 					.sending
 					.convert_to_outgoing_federation_event(leave_event.clone())
@@ -382,5 +364,14 @@ pub async fn remote_leave_room(
 		)
 		.await?;
 
-	Ok(())
+	services
+		.rooms
+		.outlier
+		.add_pdu_outlier(&event_id, &leave_event);
+
+	let leave_pdu = Pdu::from_id_val(&event_id, leave_event).map_err(|e| {
+		err!(BadServerResponse("Invalid leave PDU received during federated leave: {e:?}"))
+	})?;
+
+	Ok(leave_pdu)
 }

src/api/client/message.rs

@@ -16,7 +16,7 @@
 	Services,
 	rooms::{
 		lazy_loading,
-		lazy_loading::{Options, Witness},
+		lazy_loading::{MemberSet, Options},
 		timeline::PdusIterItem,
 	},
 };
@@ -162,7 +162,7 @@ pub(crate) async fn get_message_events_route(
 
 	let state = witness
 		.map(Option::into_iter)
-		.map(|option| option.flat_map(Witness::into_iter))
+		.map(|option| option.flat_map(MemberSet::into_iter))
 		.map(IterStream::stream)
 		.into_stream()
 		.flatten()
@@ -192,7 +192,7 @@ pub(crate) async fn lazy_loading_witness<'a, I>(
 	services: &Services,
 	lazy_loading_context: &lazy_loading::Context<'_>,
 	events: I,
-) -> Witness
+) -> MemberSet
 where
 	I: Iterator<Item = &'a PdusIterItem> + Clone + Send,
 {
@@ -213,10 +213,10 @@ pub(crate) async fn lazy_loading_witness<'a, I>(
 	let receipts = services
 		.rooms
 		.read_receipt
-		.readreceipts_since(lazy_loading_context.room_id, oldest.into_unsigned());
+		.readreceipts_since(lazy_loading_context.room_id, Some(oldest.into_unsigned()));
 
 	pin_mut!(receipts);
-	let witness: Witness = events
+	let witness: MemberSet = events
 		.stream()
 		.map(ref_at!(1))
 		.map(Event::sender)
@@ -224,7 +224,7 @@ pub(crate) async fn lazy_loading_witness<'a, I>(
 		.chain(
 			receipts
 				.ready_take_while(|(_, c, _)| *c <= newest.into_unsigned())
-				.map(|(user_id, ..)| user_id.to_owned()),
+				.map(|(user_id, ..)| user_id),
 		)
 		.collect()
 		.await;
@@ -232,7 +232,7 @@ pub(crate) async fn lazy_loading_witness<'a, I>(
 	services
 		.rooms
 		.lazy_loading
-		.witness_retain(witness, lazy_loading_context)
+		.retain_lazy_members(witness, lazy_loading_context)
 		.await
 }
 

src/api/client/space.rs

@@ -44,7 +44,7 @@ pub(crate) async fn get_hierarchy_route(
 		.as_ref()
 		.and_then(|s| PaginationToken::from_str(s).ok());
 
-	// Should prevent unexpeded behaviour in (bad) clients
+	// Should prevent unexpected behaviour in (bad) clients
 	if let Some(ref token) = key {
 		if token.suggested_only != body.suggested_only || token.max_depth != max_depth {
 			return Err!(Request(InvalidParam(

src/api/client/sync/mod.rs

@@ -1,65 +1,123 @@
 mod v3;
-mod v4;
 mod v5;
 
+use std::collections::VecDeque;
+
 use conduwuit::{
-	Error, PduCount, Result,
+	Event, PduCount, Result, err,
 	matrix::pdu::PduEvent,
+	ref_at, trace,
 	utils::stream::{BroadbandExt, ReadyExt, TryIgnore},
 };
 use conduwuit_service::Services;
-use futures::{StreamExt, pin_mut};
+use futures::StreamExt;
 use ruma::{
-	RoomId, UserId,
+	OwnedUserId, RoomId, UserId,
 	events::TimelineEventType::{
 		self, Beacon, CallInvite, PollStart, RoomEncrypted, RoomMessage, Sticker,
 	},
 };
 
-pub(crate) use self::{
-	v3::sync_events_route, v4::sync_events_v4_route, v5::sync_events_v5_route,
-};
+pub(crate) use self::{v3::sync_events_route, v5::sync_events_v5_route};
 
 pub(crate) const DEFAULT_BUMP_TYPES: &[TimelineEventType; 6] =
 	&[CallInvite, PollStart, Beacon, RoomEncrypted, RoomMessage, Sticker];
 
+#[derive(Default)]
+pub(crate) struct TimelinePdus {
+	pub pdus: VecDeque<(PduCount, PduEvent)>,
+	pub limited: bool,
+}
+
+impl TimelinePdus {
+	fn senders(&self) -> impl Iterator<Item = OwnedUserId> {
+		self.pdus
+			.iter()
+			.map(ref_at!(1))
+			.map(Event::sender)
+			.map(Into::into)
+	}
+}
+
+/// Load up to `limit` PDUs in the range (starting_count, ending_count].
 async fn load_timeline(
 	services: &Services,
 	sender_user: &UserId,
 	room_id: &RoomId,
-	roomsincecount: PduCount,
-	next_batch: Option<PduCount>,
+	starting_count: Option<PduCount>,
+	ending_count: Option<PduCount>,
 	limit: usize,
-) -> Result<(Vec<(PduCount, PduEvent)>, bool), Error> {
-	let last_timeline_count = services
-		.rooms
-		.timeline
-		.last_timeline_count(Some(sender_user), room_id)
-		.await?;
+) -> Result<TimelinePdus> {
+	let mut pdu_stream = match starting_count {
+		| Some(starting_count) => {
+			let last_timeline_count = services
+				.rooms
+				.timeline
+				.last_timeline_count(Some(sender_user), room_id)
+				.await
+				.map_err(|err| {
+					err!(Database(warn!("Failed to fetch end of room timeline: {}", err)))
+				})?;
 
-	if last_timeline_count <= roomsincecount {
-		return Ok((Vec::new(), false));
-	}
+			if last_timeline_count <= starting_count {
+				// no messages have been sent in this room since `starting_count`
+				return Ok(TimelinePdus::default());
+			}
 
-	let non_timeline_pdus = services
-		.rooms
-		.timeline
-		.pdus_rev(Some(sender_user), room_id, None)
-		.ignore_err()
-		.ready_skip_while(|&(pducount, _)| pducount > next_batch.unwrap_or_else(PduCount::max))
-		.ready_take_while(|&(pducount, _)| pducount > roomsincecount);
+			// for incremental sync, stream from the DB all PDUs which were sent after
+			// `starting_count` but before `ending_count`, including `ending_count` but
+			// not `starting_count`. this code is pretty similar to the initial sync
+			// branch, they're separate to allow for future optimization
+			services
+				.rooms
+				.timeline
+				.pdus_rev(
+					Some(sender_user),
+					room_id,
+					ending_count.map(|count| count.saturating_add(1)),
+				)
+				.ignore_err()
+				.ready_take_while(move |&(pducount, _)| pducount > starting_count)
+				.boxed()
+		},
+		| None => {
+			// For initial sync, stream from the DB all PDUs before and including
+			// `ending_count` in reverse order
+			services
+				.rooms
+				.timeline
+				.pdus_rev(
+					Some(sender_user),
+					room_id,
+					ending_count.map(|count| count.saturating_add(1)),
+				)
+				.ignore_err()
+				.boxed()
+		},
+	};
 
-	// Take the last events for the timeline
-	pin_mut!(non_timeline_pdus);
-	let timeline_pdus: Vec<_> = non_timeline_pdus.by_ref().take(limit).collect().await;
+	// Return at most `limit` PDUs from the stream
+	let pdus = pdu_stream
+		.by_ref()
+		.take(limit)
+		.ready_fold(VecDeque::with_capacity(limit), |mut pdus, item| {
+			pdus.push_front(item);
+			pdus
+		})
+		.await;
 
-	let timeline_pdus: Vec<_> = timeline_pdus.into_iter().rev().collect();
+	// The timeline is limited if there are still more PDUs in the stream
+	let limited = pdu_stream.next().await.is_some();
 
-	// They /sync response doesn't always return all messages, so we say the output
-	// is limited unless there are events in non_timeline_pdus
-	let limited = non_timeline_pdus.next().await.is_some();
+	trace!(
+		"syncing {:?} timeline pdus from {:?} to {:?} (limited = {:?})",
+		pdus.len(),
+		starting_count,
+		ending_count,
+		limited,
+	);
 
-	Ok((timeline_pdus, limited))
+	Ok(TimelinePdus { pdus, limited })
 }
 
 async fn share_encrypted_room(

src/api/client/sync/v3/joined.rs

@@ -0,0 +1,852 @@
+use std::collections::{BTreeMap, HashSet};
+
+use conduwuit::{
+	Result, at, debug_warn, err, extract_variant,
+	matrix::{
+		Event,
+		pdu::{PduCount, PduEvent},
+	},
+	trace,
+	utils::{
+		BoolExt, IterStream, ReadyExt, TryFutureExtExt,
+		math::ruma_from_u64,
+		stream::{TryIgnore, WidebandExt},
+	},
+	warn,
+};
+use conduwuit_service::Services;
+use futures::{
+	FutureExt, StreamExt, TryFutureExt,
+	future::{OptionFuture, join, join3, join4, try_join, try_join3},
+};
+use ruma::{
+	OwnedRoomId, OwnedUserId, RoomId, UserId,
+	api::client::sync::sync_events::{
+		UnreadNotificationsCount,
+		v3::{Ephemeral, JoinedRoom, RoomAccountData, RoomSummary, State as RoomState, Timeline},
+	},
+	events::{
+		AnyRawAccountDataEvent, StateEventType,
+		TimelineEventType::*,
+		room::member::{MembershipState, RoomMemberEventContent},
+	},
+	serde::Raw,
+	uint,
+};
+use service::rooms::short::ShortStateHash;
+
+use super::{load_timeline, share_encrypted_room};
+use crate::client::{
+	TimelinePdus, ignored_filter,
+	sync::v3::{
+		DEFAULT_TIMELINE_LIMIT, DeviceListUpdates, SyncContext, prepare_lazily_loaded_members,
+		state::{build_state_incremental, build_state_initial},
+	},
+};
+
+/// Generate the sync response for a room the user is joined to.
+#[tracing::instrument(
+	name = "joined",
+	level = "debug",
+	skip_all,
+	fields(
+		room_id = ?room_id,
+		syncing_user = ?sync_context.syncing_user,
+	),
+)]
+pub(super) async fn load_joined_room(
+	services: &Services,
+	sync_context: SyncContext<'_>,
+	ref room_id: OwnedRoomId,
+) -> Result<(JoinedRoom, DeviceListUpdates)> {
+	/*
+	Building a sync response involves many steps which all depend on each other.
+	To parallelize the process as much as possible, each step is divided into its own function,
+	and `join*` functions are used to perform steps in parallel which do not depend on each other.
+	*/
+
+	let (
+		account_data,
+		ephemeral,
+		StateAndTimeline {
+			state_events,
+			timeline,
+			summary,
+			notification_counts,
+			device_list_updates,
+		},
+	) = try_join3(
+		build_account_data(services, sync_context, room_id),
+		build_ephemeral(services, sync_context, room_id),
+		build_state_and_timeline(services, sync_context, room_id),
+	)
+	.boxed()
+	.await?;
+
+	if !timeline.is_empty() || !state_events.is_empty() {
+		trace!(
+			"syncing {} timeline events (limited = {}) and {} state events",
+			timeline.events.len(),
+			timeline.limited,
+			state_events.len()
+		);
+	}
+
+	let joined_room = JoinedRoom {
+		account_data,
+		summary: summary.unwrap_or_default(),
+		unread_notifications: notification_counts.unwrap_or_default(),
+		timeline,
+		state: RoomState {
+			events: state_events.into_iter().map(Event::into_format).collect(),
+		},
+		ephemeral,
+		unread_thread_notifications: BTreeMap::new(),
+	};
+
+	Ok((joined_room, device_list_updates))
+}
+
+/// Collect changes to the syncing user's account data events.
+#[tracing::instrument(level = "debug", skip_all)]
+async fn build_account_data(
+	services: &Services,
+	SyncContext {
+		syncing_user,
+		last_sync_end_count,
+		current_count,
+		..
+	}: SyncContext<'_>,
+	room_id: &RoomId,
+) -> Result<RoomAccountData> {
+	let account_data_changes = services
+		.account_data
+		.changes_since(Some(room_id), syncing_user, last_sync_end_count, Some(current_count))
+		.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
+		.collect()
+		.await;
+
+	Ok(RoomAccountData { events: account_data_changes })
+}
+
+/// Collect new ephemeral events.
+#[tracing::instrument(level = "debug", skip_all)]
+async fn build_ephemeral(
+	services: &Services,
+	SyncContext { syncing_user, last_sync_end_count, .. }: SyncContext<'_>,
+	room_id: &RoomId,
+) -> Result<Ephemeral> {
+	// note: some of the futures below are boxed. this is because, without the box,
+	// rustc produces over thirty inscrutable errors in `mod.rs` at the call-site
+	// of `load_joined_room`. I don't know why boxing them fixes this -- it seems
+	// to be related to the async closures and borrowing from the sync context.
+
+	// collect updates to read receipts
+	let receipt_events = services
+		.rooms
+		.read_receipt
+		.readreceipts_since(room_id, last_sync_end_count)
+		.filter_map(async |(read_user, _, edu)| {
+			let is_ignored = services
+				.users
+				.user_is_ignored(&read_user, syncing_user)
+				.await;
+
+			// filter out read receipts for ignored users
+			is_ignored.or_some(edu)
+		})
+		.collect::<Vec<_>>()
+		.boxed();
+
+	// collect the updated list of typing users, if it's changed
+	let typing_event = async {
+		let should_send_typing_event = match last_sync_end_count {
+			| Some(last_sync_end_count) => {
+				match services.rooms.typing.last_typing_update(room_id).await {
+					| Ok(last_typing_update) => {
+						// update the typing list if the users typing have changed since the last
+						// sync
+						last_typing_update > last_sync_end_count
+					},
+					| Err(err) => {
+						warn!("Error checking last typing update: {}", err);
+						return None;
+					},
+				}
+			},
+			// always update the typing list on an initial sync
+			| None => true,
+		};
+
+		if should_send_typing_event {
+			let event = services
+				.rooms
+				.typing
+				.typings_event_for_user(room_id, syncing_user)
+				.await;
+
+			if let Ok(event) = event {
+				return Some(
+					Raw::new(&event)
+						.expect("typing event should be valid")
+						.cast(),
+				);
+			}
+		}
+
+		None
+	};
+
+	// collect the syncing user's private-read marker, if it's changed
+	let private_read_event = async {
+		let should_send_private_read = match last_sync_end_count {
+			| Some(last_sync_end_count) => {
+				let last_privateread_update = services
+					.rooms
+					.read_receipt
+					.last_privateread_update(syncing_user, room_id)
+					.await;
+
+				// update the marker if it's changed since the last sync
+				last_privateread_update > last_sync_end_count
+			},
+			// always update the marker on an initial sync
+			| None => true,
+		};
+
+		if should_send_private_read {
+			services
+				.rooms
+				.read_receipt
+				.private_read_get(room_id, syncing_user)
+				.await
+				.ok()
+		} else {
+			None
+		}
+	};
+
+	let (receipt_events, typing_event, private_read_event) =
+		join3(receipt_events, typing_event, private_read_event).await;
+
+	let mut edus = receipt_events;
+	edus.extend(typing_event);
+	edus.extend(private_read_event);
+
+	Ok(Ephemeral { events: edus })
+}
+
+/// A struct to hold the state events, timeline, and other data which is
+/// computed from them.
+struct StateAndTimeline {
+	state_events: Vec<PduEvent>,
+	timeline: Timeline,
+	summary: Option<RoomSummary>,
+	notification_counts: Option<UnreadNotificationsCount>,
+	device_list_updates: DeviceListUpdates,
+}
+
+/// Compute changes to the room's state and timeline.
+#[tracing::instrument(level = "debug", skip_all)]
+async fn build_state_and_timeline(
+	services: &Services,
+	sync_context: SyncContext<'_>,
+	room_id: &RoomId,
+) -> Result<StateAndTimeline> {
+	let (shortstatehashes, timeline) = try_join(
+		fetch_shortstatehashes(services, sync_context, room_id),
+		build_timeline(services, sync_context, room_id),
+	)
+	.await?;
+
+	let (state_events, notification_counts, joined_since_last_sync) = try_join3(
+		build_state_events(services, sync_context, room_id, shortstatehashes, &timeline),
+		build_notification_counts(services, sync_context, room_id, &timeline),
+		check_joined_since_last_sync(services, shortstatehashes, sync_context),
+	)
+	.await?;
+
+	// the timeline should always include at least one PDU if the syncing user
+	// joined since the last sync, that being the syncing user's join event. if
+	// it's empty something is wrong.
+	if joined_since_last_sync && timeline.pdus.is_empty() {
+		warn!("timeline for newly joined room is empty");
+	}
+
+	let (summary, device_list_updates) = try_join(
+		build_room_summary(
+			services,
+			sync_context,
+			room_id,
+			shortstatehashes,
+			&timeline,
+			&state_events,
+			joined_since_last_sync,
+		),
+		build_device_list_updates(
+			services,
+			sync_context,
+			room_id,
+			shortstatehashes,
+			&state_events,
+			joined_since_last_sync,
+		),
+	)
+	.await?;
+
+	// the token which may be passed to the messages endpoint to backfill room
+	// history
+	let prev_batch = timeline.pdus.front().map(at!(0));
+
+	// note: we always indicate a limited timeline if the syncing user just joined
+	// the room, to indicate to the client that it should request backfill (and to
+	// copy Synapse's behavior). for federated room joins, the `timeline` will
+	// usually only include the syncing user's join event.
+	let limited = timeline.limited || joined_since_last_sync;
+
+	// filter out ignored events from the timeline and convert the PDUs into Ruma's
+	// AnySyncTimelineEvent type
+	let filtered_timeline = timeline
+		.pdus
+		.into_iter()
+		.stream()
+		.wide_filter_map(|item| ignored_filter(services, item, sync_context.syncing_user))
+		.map(at!(1))
+		.map(Event::into_format)
+		.collect::<Vec<_>>()
+		.await;
+
+	Ok(StateAndTimeline {
+		state_events,
+		timeline: Timeline {
+			limited,
+			prev_batch: prev_batch.as_ref().map(ToString::to_string),
+			events: filtered_timeline,
+		},
+		summary,
+		notification_counts,
+		device_list_updates,
+	})
+}
+
+/// Shortstatehashes necessary to compute what state events to sync.
+#[derive(Clone, Copy)]
+struct ShortStateHashes {
+	/// The current state of the syncing room.
+	current_shortstatehash: ShortStateHash,
+	/// The state of the syncing room at the end of the last sync.
+	last_sync_end_shortstatehash: Option<ShortStateHash>,
+}
+
+/// Fetch the current_shortstatehash and last_sync_end_shortstatehash.
+#[tracing::instrument(level = "debug", skip_all)]
+async fn fetch_shortstatehashes(
+	services: &Services,
+	SyncContext { last_sync_end_count, current_count, .. }: SyncContext<'_>,
+	room_id: &RoomId,
+) -> Result<ShortStateHashes> {
+	// the room state currently.
+	// TODO: this should be the room state as of `current_count`, but there's no way
+	// to get that right now.
+	let current_shortstatehash = services
+		.rooms
+		.state
+		.get_room_shortstatehash(room_id)
+		.map_err(|_| err!(Database(error!("Room {room_id} has no state"))));
+
+	// the room state as of the end of the last sync.
+	// this will be None if we are doing an initial sync or if we just joined this
+	// room.
+	let last_sync_end_shortstatehash =
+		OptionFuture::from(last_sync_end_count.map(|last_sync_end_count| {
+			// look up the shortstatehash saved by the last sync's call to
+			// `associate_token_shortstatehash`
+			services
+				.rooms
+				.user
+				.get_token_shortstatehash(room_id, last_sync_end_count)
+				.inspect_err(move |_| {
+					debug_warn!(
+						token = last_sync_end_count,
+						"Room has no shortstatehash for this token"
+					);
+				})
+				.ok()
+		}))
+		.map(Option::flatten)
+		.map(Ok);
+
+	let (current_shortstatehash, last_sync_end_shortstatehash) =
+		try_join(current_shortstatehash, last_sync_end_shortstatehash).await?;
+
+	/*
+	associate the `current_count` with the `current_shortstatehash`, so we can
+	use it on the next sync as the `last_sync_end_shortstatehash`.
+
+	TODO: the table written to by this call grows extremely fast, gaining one new entry for each
+	joined room on _every single sync request_. we need to find a better way to remember the shortstatehash
+	between syncs.
+	*/
+	services
+		.rooms
+		.user
+		.associate_token_shortstatehash(room_id, current_count, current_shortstatehash)
+		.await;
+
+	Ok(ShortStateHashes {
+		current_shortstatehash,
+		last_sync_end_shortstatehash,
+	})
+}
+
+/// Fetch recent timeline events.
+#[tracing::instrument(level = "debug", skip_all)]
+async fn build_timeline(
+	services: &Services,
+	sync_context: SyncContext<'_>,
+	room_id: &RoomId,
+) -> Result<TimelinePdus> {
+	let SyncContext {
+		syncing_user,
+		last_sync_end_count,
+		current_count,
+		filter,
+		..
+	} = sync_context;
+
+	/*
+	determine the maximum number of events to return in this sync.
+	if the sync filter specifies a limit, that will be used, otherwise
+	`DEFAULT_TIMELINE_LIMIT` will be used. `DEFAULT_TIMELINE_LIMIT` will also be
+	used if the limit is somehow greater than usize::MAX.
+	*/
+	let timeline_limit = filter
+		.room
+		.timeline
+		.limit
+		.and_then(|limit| limit.try_into().ok())
+		.unwrap_or(DEFAULT_TIMELINE_LIMIT);
+
+	load_timeline(
+		services,
+		syncing_user,
+		room_id,
+		last_sync_end_count.map(PduCount::Normal),
+		Some(PduCount::Normal(current_count)),
+		timeline_limit,
+	)
+	.await
+}
+
+/// Calculate the state events to sync.
+async fn build_state_events(
+	services: &Services,
+	sync_context: SyncContext<'_>,
+	room_id: &RoomId,
+	shortstatehashes: ShortStateHashes,
+	timeline: &TimelinePdus,
+) -> Result<Vec<PduEvent>> {
+	let SyncContext {
+		syncing_user,
+		last_sync_end_count,
+		full_state,
+		..
+	} = sync_context;
+
+	let ShortStateHashes {
+		current_shortstatehash,
+		last_sync_end_shortstatehash,
+	} = shortstatehashes;
+
+	// the spec states that the `state` property only includes state events up to
+	// the beginning of the timeline, so we determine the state of the syncing room
+	// as of the first timeline event. NOTE: this explanation is not entirely
+	// accurate; see the implementation of `build_state_incremental`.
+	let timeline_start_shortstatehash = async {
+		if let Some((_, pdu)) = timeline.pdus.front() {
+			if let Ok(shortstatehash) = services
+				.rooms
+				.state_accessor
+				.pdu_shortstatehash(&pdu.event_id)
+				.await
+			{
+				return shortstatehash;
+			}
+		}
+
+		current_shortstatehash
+	};
+
+	// the user IDs of members whose membership needs to be sent to the client, if
+	// lazy-loading is enabled.
+	let lazily_loaded_members =
+		prepare_lazily_loaded_members(services, sync_context, room_id, timeline.senders());
+
+	let (timeline_start_shortstatehash, lazily_loaded_members) =
+		join(timeline_start_shortstatehash, lazily_loaded_members).await;
+
+	// compute the state delta between the previous sync and this sync.
+	match (last_sync_end_count, last_sync_end_shortstatehash) {
+		/*
+		if `last_sync_end_count` is Some (meaning this is an incremental sync), and `last_sync_end_shortstatehash`
+		is Some (meaning the syncing user didn't just join this room for the first time ever), and `full_state` is false,
+		then use `build_state_incremental`.
+		*/
+		| (Some(last_sync_end_count), Some(last_sync_end_shortstatehash)) if !full_state =>
+			build_state_incremental(
+				services,
+				syncing_user,
+				room_id,
+				PduCount::Normal(last_sync_end_count),
+				last_sync_end_shortstatehash,
+				timeline_start_shortstatehash,
+				current_shortstatehash,
+				timeline,
+				lazily_loaded_members.as_ref(),
+			)
+			.boxed()
+			.await,
+		/*
+		otherwise use `build_state_initial`. note that this branch will be taken if the user joined this room since the last sync
+		for the first time ever, because in that case we have no `last_sync_end_shortstatehash` and can't correctly calculate
+		the state using the incremental sync algorithm.
+		*/
+		| _ =>
+			build_state_initial(
+				services,
+				syncing_user,
+				timeline_start_shortstatehash,
+				lazily_loaded_members.as_ref(),
+			)
+			.boxed()
+			.await,
+	}
+}
+
+/// Compute the number of unread notifications in this room.
+#[tracing::instrument(level = "debug", skip_all)]
+async fn build_notification_counts(
+	services: &Services,
+	SyncContext { syncing_user, last_sync_end_count, .. }: SyncContext<'_>,
+	room_id: &RoomId,
+	timeline: &TimelinePdus,
+) -> Result<Option<UnreadNotificationsCount>> {
+	// determine whether to actually update the notification counts
+	let should_send_notification_counts = async {
+		// if we're going to sync some timeline events, the notification count has
+		// definitely changed to include them
+		if !timeline.pdus.is_empty() {
+			return true;
+		}
+
+		// if this is an initial sync, we need to send notification counts because the
+		// client doesn't know what they are yet
+		let Some(last_sync_end_count) = last_sync_end_count else {
+			return true;
+		};
+
+		let last_notification_read = services
+			.rooms
+			.user
+			.last_notification_read(syncing_user, room_id)
+			.await;
+
+		// if the syncing user has read the events we sent during the last sync, we need
+		// to send a new notification count on this sync.
+		if last_notification_read > last_sync_end_count {
+			return true;
+		}
+
+		// otherwise, nothing's changed.
+		false
+	};
+
+	if should_send_notification_counts.await {
+		let (notification_count, highlight_count) = join(
+			services
+				.rooms
+				.user
+				.notification_count(syncing_user, room_id)
+				.map(TryInto::try_into)
+				.unwrap_or(uint!(0)),
+			services
+				.rooms
+				.user
+				.highlight_count(syncing_user, room_id)
+				.map(TryInto::try_into)
+				.unwrap_or(uint!(0)),
+		)
+		.await;
+
+		trace!(?notification_count, ?highlight_count, "syncing new notification counts");
+
+		Ok(Some(UnreadNotificationsCount {
+			notification_count: Some(notification_count),
+			highlight_count: Some(highlight_count),
+		}))
+	} else {
+		Ok(None)
+	}
+}
+
+/// Check if the syncing user joined the room since their last incremental sync.
+#[tracing::instrument(level = "debug", skip_all)]
+async fn check_joined_since_last_sync(
+	services: &Services,
+	ShortStateHashes { last_sync_end_shortstatehash, .. }: ShortStateHashes,
+	SyncContext { syncing_user, .. }: SyncContext<'_>,
+) -> Result<bool> {
+	// fetch the syncing user's membership event during the last sync.
+	// this will be None if `previous_sync_end_shortstatehash` is None.
+	let membership_during_previous_sync = match last_sync_end_shortstatehash {
+		| Some(last_sync_end_shortstatehash) => services
+			.rooms
+			.state_accessor
+			.state_get_content(
+				last_sync_end_shortstatehash,
+				&StateEventType::RoomMember,
+				syncing_user.as_str(),
+			)
+			.await
+			.inspect_err(|_| debug_warn!("User has no previous membership"))
+			.ok(),
+		| None => None,
+	};
+
+	// TODO: If the requesting user got state-reset out of the room, this
+	// will be `true` when it shouldn't be. this function should never be called
+	// in that situation, but it may be if the membership cache didn't get updated.
+	// the root cause of this needs to be addressed
+	let joined_since_last_sync =
+		membership_during_previous_sync.is_none_or(|content: RoomMemberEventContent| {
+			content.membership != MembershipState::Join
+		});
+
+	if joined_since_last_sync {
+		trace!("user joined since last sync");
+	}
+
+	Ok(joined_since_last_sync)
+}
+
+/// Build the `summary` field of the room object, which includes
+/// the number of joined and invited users and the room's heroes.
+#[tracing::instrument(level = "debug", skip_all)]
+async fn build_room_summary(
+	services: &Services,
+	SyncContext { syncing_user, .. }: SyncContext<'_>,
+	room_id: &RoomId,
+	ShortStateHashes { current_shortstatehash, .. }: ShortStateHashes,
+	timeline: &TimelinePdus,
+	state_events: &[PduEvent],
+	joined_since_last_sync: bool,
+) -> Result<Option<RoomSummary>> {
+	// determine whether any events in the state or timeline are membership events.
+	let are_syncing_membership_events = timeline
+		.pdus
+		.iter()
+		.map(|(_, pdu)| pdu)
+		.chain(state_events.iter())
+		.any(|event| event.kind == RoomMember);
+
+	/*
+	we only need to send an updated room summary if:
+	1. there are membership events in the state or timeline, because they might have changed the
+	   membership counts or heroes, or
+	2. the syncing user just joined this room, which usually implies #1 because their join event should be in the timeline.
+	*/
+	if !(are_syncing_membership_events || joined_since_last_sync) {
+		return Ok(None);
+	}
+
+	let joined_member_count = services
+		.rooms
+		.state_cache
+		.room_joined_count(room_id)
+		.unwrap_or(0);
+
+	let invited_member_count = services
+		.rooms
+		.state_cache
+		.room_invited_count(room_id)
+		.unwrap_or(0);
+
+	let has_name = services
+		.rooms
+		.state_accessor
+		.state_contains_type(current_shortstatehash, &StateEventType::RoomName);
+
+	let has_canonical_alias = services
+		.rooms
+		.state_accessor
+		.state_contains_type(current_shortstatehash, &StateEventType::RoomCanonicalAlias);
+
+	let (joined_member_count, invited_member_count, has_name, has_canonical_alias) =
+		join4(joined_member_count, invited_member_count, has_name, has_canonical_alias).await;
+
+	// only send heroes if the room has neither a name nor a canonical alias
+	let heroes = if !(has_name || has_canonical_alias) {
+		Some(build_heroes(services, room_id, syncing_user, current_shortstatehash).await)
+	} else {
+		None
+	};
+
+	trace!(
+		?joined_member_count,
+		?invited_member_count,
+		heroes_length = heroes.as_ref().map(HashSet::len),
+		"syncing updated summary"
+	);
+
+	Ok(Some(RoomSummary {
+		heroes: heroes
+			.map(|heroes| heroes.into_iter().collect())
+			.unwrap_or_default(),
+		joined_member_count: Some(ruma_from_u64(joined_member_count)),
+		invited_member_count: Some(ruma_from_u64(invited_member_count)),
+	}))
+}
+
+/// Fetch the user IDs to include in the `m.heroes` property of the room
+/// summary.
+async fn build_heroes(
+	services: &Services,
+	room_id: &RoomId,
+	syncing_user: &UserId,
+	current_shortstatehash: ShortStateHash,
+) -> HashSet<OwnedUserId> {
+	const MAX_HERO_COUNT: usize = 5;
+
+	// fetch joined members from the state cache first
+	let joined_members_stream = services
+		.rooms
+		.state_cache
+		.room_members(room_id)
+		.map(ToOwned::to_owned);
+
+	// then fetch invited members
+	let invited_members_stream = services
+		.rooms
+		.state_cache
+		.room_members_invited(room_id)
+		.map(ToOwned::to_owned);
+
+	// then as a last resort fetch every membership event
+	let all_members_stream = services
+		.rooms
+		.short
+		.multi_get_statekey_from_short(
+			services
+				.rooms
+				.state_accessor
+				.state_full_shortids(current_shortstatehash)
+				.ignore_err()
+				.ready_filter_map(|(key, _)| Some(key)),
+		)
+		.ignore_err()
+		.ready_filter_map(|(event_type, state_key)| {
+			if event_type == StateEventType::RoomMember {
+				state_key.to_string().try_into().ok()
+			} else {
+				None
+			}
+		});
+
+	joined_members_stream
+		.chain(invited_members_stream)
+		.chain(all_members_stream)
+		// the hero list should never include the syncing user
+		.ready_filter(|user_id| user_id != syncing_user)
+		.take(MAX_HERO_COUNT)
+		.collect()
+		.await
+}
+
+/// Collect updates to users' device lists for E2EE.
+#[tracing::instrument(level = "debug", skip_all)]
+async fn build_device_list_updates(
+	services: &Services,
+	SyncContext {
+		syncing_user,
+		last_sync_end_count,
+		current_count,
+		..
+	}: SyncContext<'_>,
+	room_id: &RoomId,
+	ShortStateHashes { current_shortstatehash, .. }: ShortStateHashes,
+	state_events: &Vec<PduEvent>,
+	joined_since_last_sync: bool,
+) -> Result<DeviceListUpdates> {
+	let is_encrypted_room = services
+		.rooms
+		.state_accessor
+		.state_get(current_shortstatehash, &StateEventType::RoomEncryption, "")
+		.is_ok();
+
+	// initial syncs don't include device updates, and rooms which aren't encrypted
+	// don't affect them, so return early in either of those cases
+	if last_sync_end_count.is_none() || !(is_encrypted_room.await) {
+		return Ok(DeviceListUpdates::new());
+	}
+
+	let mut device_list_updates = DeviceListUpdates::new();
+
+	// add users with changed keys to the `changed` list
+	services
+		.users
+		.room_keys_changed(room_id, last_sync_end_count, Some(current_count))
+		.map(at!(0))
+		.map(ToOwned::to_owned)
+		.ready_for_each(|user_id| {
+			device_list_updates.changed.insert(user_id);
+		})
+		.await;
+
+	// add users who now share encrypted rooms to `changed` and
+	// users who no longer share encrypted rooms to `left`
+	for state_event in state_events {
+		if state_event.kind == RoomMember {
+			let Some(content): Option<RoomMemberEventContent> = state_event.get_content().ok()
+			else {
+				continue;
+			};
+
+			let Some(user_id): Option<OwnedUserId> = state_event
+				.state_key
+				.as_ref()
+				.and_then(|key| key.parse().ok())
+			else {
+				continue;
+			};
+
+			{
+				use MembershipState::*;
+
+				if matches!(content.membership, Leave | Join) {
+					let shares_encrypted_room =
+						share_encrypted_room(services, syncing_user, &user_id, Some(room_id))
+							.await;
+					match content.membership {
+						| Leave if !shares_encrypted_room => {
+							device_list_updates.left.insert(user_id);
+						},
+						| Join if joined_since_last_sync || shares_encrypted_room => {
+							device_list_updates.changed.insert(user_id);
+						},
+						| _ => (),
+					}
+				}
+			}
+		}
+	}
+
+	if !device_list_updates.is_empty() {
+		trace!(
+			changed = device_list_updates.changed.len(),
+			left = device_list_updates.left.len(),
+			"syncing device list updates"
+		);
+	}
+
+	Ok(device_list_updates)
+}

src/api/client/sync/v3/left.rs

@@ -0,0 +1,349 @@
+use conduwuit::{
+	Event, PduCount, PduEvent, Result, at, debug_warn,
+	pdu::EventHash,
+	trace,
+	utils::{self, IterStream, future::ReadyEqExt, stream::WidebandExt as _},
+};
+use futures::{StreamExt, future::join};
+use ruma::{
+	EventId, OwnedRoomId, RoomId,
+	api::client::sync::sync_events::v3::{LeftRoom, RoomAccountData, State, Timeline},
+	events::{StateEventType, TimelineEventType},
+	uint,
+};
+use serde_json::value::RawValue;
+use service::{Services, rooms::short::ShortStateHash};
+
+use crate::client::{
+	TimelinePdus, ignored_filter,
+	sync::{
+		load_timeline,
+		v3::{
+			DEFAULT_TIMELINE_LIMIT, SyncContext, prepare_lazily_loaded_members,
+			state::build_state_initial,
+		},
+	},
+};
+
+#[tracing::instrument(
+	name = "left",
+	level = "debug",
+	skip_all,
+	fields(
+		room_id = %room_id,
+	),
+)]
+#[allow(clippy::too_many_arguments)]
+pub(super) async fn load_left_room(
+	services: &Services,
+	sync_context: SyncContext<'_>,
+	ref room_id: OwnedRoomId,
+	leave_membership_event: Option<PduEvent>,
+) -> Result<Option<LeftRoom>> {
+	let SyncContext {
+		syncing_user,
+		last_sync_end_count,
+		current_count,
+		filter,
+		..
+	} = sync_context;
+
+	// the global count as of the moment the user left the room
+	let Some(left_count) = services
+		.rooms
+		.state_cache
+		.get_left_count(room_id, syncing_user)
+		.await
+		.ok()
+	else {
+		// if we get here, the membership cache is incorrect, likely due to a state
+		// reset
+		debug_warn!("attempting to sync left room but no left count exists");
+		return Ok(None);
+	};
+
+	// return early if we haven't gotten to this leave yet.
+	// this can happen if the user leaves while a sync response is being generated
+	if current_count < left_count {
+		return Ok(None);
+	}
+
+	// return early if this is an incremental sync, and we've already synced this
+	// leave to the user, and `include_leave` isn't set on the filter.
+	if !filter.room.include_leave && last_sync_end_count >= Some(left_count) {
+		return Ok(None);
+	}
+
+	if let Some(ref leave_membership_event) = leave_membership_event {
+		debug_assert_eq!(
+			leave_membership_event.kind,
+			TimelineEventType::RoomMember,
+			"leave PDU should be m.room.member"
+		);
+	}
+
+	let does_not_exist = services.rooms.metadata.exists(room_id).eq(&false).await;
+
+	let (timeline, state_events) = match leave_membership_event {
+		| Some(leave_membership_event) if does_not_exist => {
+			/*
+			we have none PDUs with left beef for this room, likely because it was a rejected invite to a room
+			which nobody on this homeserver is in. `leave_pdu` is the remote-assisted outlier leave event for the room,
+			which is all we can send to the client.
+
+			if this is an initial sync, don't include this room at all to keep the client from asking for
+			state that we don't have.
+			*/
+
+			if last_sync_end_count.is_none() {
+				return Ok(None);
+			}
+
+			trace!("syncing remote-assisted leave PDU");
+			(TimelinePdus::default(), vec![leave_membership_event])
+		},
+		| Some(leave_membership_event) => {
+			// we have this room in our DB, and can fetch the state and timeline from when
+			// the user left.
+
+			let leave_state_key = syncing_user;
+			debug_assert_eq!(
+				Some(leave_state_key.as_str()),
+				leave_membership_event.state_key(),
+				"leave PDU should be for the user requesting the sync"
+			);
+
+			// the shortstatehash of the state _immediately before_ the syncing user left
+			// this room. the state represented here _does not_ include
+			// `leave_membership_event`.
+			let leave_shortstatehash = services
+				.rooms
+				.state_accessor
+				.pdu_shortstatehash(&leave_membership_event.event_id)
+				.await?;
+
+			let prev_membership_event = services
+				.rooms
+				.state_accessor
+				.state_get(
+					leave_shortstatehash,
+					&StateEventType::RoomMember,
+					leave_state_key.as_str(),
+				)
+				.await?;
+
+			build_left_state_and_timeline(
+				services,
+				sync_context,
+				room_id,
+				leave_membership_event,
+				leave_shortstatehash,
+				prev_membership_event,
+			)
+			.await?
+		},
+		| None => {
+			/*
+			no leave event was actually sent in this room, but we still need to pretend
+			like the user left it. this is usually because the room was banned by a server admin.
+
+			if this is an incremental sync, generate a fake leave event to make the room vanish from clients.
+			otherwise we don't tell the client about this room at all.
+			*/
+			if last_sync_end_count.is_none() {
+				return Ok(None);
+			}
+
+			trace!("syncing dummy leave event");
+			(TimelinePdus::default(), vec![create_dummy_leave_event(
+				services,
+				sync_context,
+				room_id,
+			)])
+		},
+	};
+
+	let raw_timeline_pdus = timeline
+		.pdus
+		.into_iter()
+		.stream()
+		// filter out ignored events from the timeline
+		.wide_filter_map(|item| ignored_filter(services, item, syncing_user))
+		.map(at!(1))
+		.map(Event::into_format)
+		.collect::<Vec<_>>()
+		.await;
+
+	Ok(Some(LeftRoom {
+		account_data: RoomAccountData { events: Vec::new() },
+		timeline: Timeline {
+			limited: timeline.limited,
+			prev_batch: Some(current_count.to_string()),
+			events: raw_timeline_pdus,
+		},
+		state: State {
+			events: state_events.into_iter().map(Event::into_format).collect(),
+		},
+	}))
+}
+
+async fn build_left_state_and_timeline(
+	services: &Services,
+	sync_context: SyncContext<'_>,
+	room_id: &RoomId,
+	leave_membership_event: PduEvent,
+	leave_shortstatehash: ShortStateHash,
+	prev_membership_event: PduEvent,
+) -> Result<(TimelinePdus, Vec<PduEvent>)> {
+	let SyncContext {
+		syncing_user,
+		last_sync_end_count,
+		filter,
+		..
+	} = sync_context;
+
+	let timeline_start_count = if let Some(last_sync_end_count) = last_sync_end_count {
+		// for incremental syncs, start the timeline after `since`
+		PduCount::Normal(last_sync_end_count)
+	} else {
+		// for initial syncs, start the timeline after the previous membership
+		// event. we don't want to include the membership event itself
+		// because clients get confused when they see a `join`
+		// membership event in a `leave` room.
+		services
+			.rooms
+			.timeline
+			.get_pdu_count(&prev_membership_event.event_id)
+			.await?
+	};
+
+	// end the timeline at the user's leave event
+	let timeline_end_count = services
+		.rooms
+		.timeline
+		.get_pdu_count(leave_membership_event.event_id())
+		.await?;
+
+	// limit the timeline using the same logic as for joined rooms
+	let timeline_limit = filter
+		.room
+		.timeline
+		.limit
+		.and_then(|limit| limit.try_into().ok())
+		.unwrap_or(DEFAULT_TIMELINE_LIMIT);
+
+	let timeline = load_timeline(
+		services,
+		syncing_user,
+		room_id,
+		Some(timeline_start_count),
+		Some(timeline_end_count),
+		timeline_limit,
+	)
+	.await?;
+
+	let timeline_start_shortstatehash = async {
+		if let Some((_, pdu)) = timeline.pdus.front() {
+			if let Ok(shortstatehash) = services
+				.rooms
+				.state_accessor
+				.pdu_shortstatehash(&pdu.event_id)
+				.await
+			{
+				return shortstatehash;
+			}
+		}
+
+		// the timeline generally should not be empty (see the TODO further down),
+		// but in case it is we use `leave_shortstatehash` as the state to
+		// send
+		leave_shortstatehash
+	};
+
+	let lazily_loaded_members =
+		prepare_lazily_loaded_members(services, sync_context, room_id, timeline.senders());
+
+	let (timeline_start_shortstatehash, lazily_loaded_members) =
+		join(timeline_start_shortstatehash, lazily_loaded_members).await;
+
+	// TODO: calculate incremental state for incremental syncs.
+	// always calculating initial state _works_ but returns more data and does
+	// more processing than strictly necessary.
+	let mut state = build_state_initial(
+		services,
+		syncing_user,
+		timeline_start_shortstatehash,
+		lazily_loaded_members.as_ref(),
+	)
+	.await?;
+
+	/*
+	remove membership events for the syncing user from state.
+	usually, `state` should include a `join` membership event and `timeline` should include a `leave` one.
+	however, the matrix-js-sdk gets confused when this happens (see [1]) and doesn't process the room leave,
+	so we have to filter out the membership from `state`.
+
+	NOTE: we are sending more information than synapse does in this scenario, because we always
+	calculate `state` for initial syncs, even when the sync being performed is incremental.
+	however, the specification does not forbid sending extraneous events in `state`.
+
+	TODO: there is an additional bug at play here. sometimes `load_joined_room` syncs the `leave` event
+	before `load_left_room` does, which means the `timeline` we sync immediately after a leave is empty.
+	this shouldn't happen -- `timeline` should always include the `leave` event. this is probably
+	a race condition with the membership state cache.
+
+	[1]: https://github.com/matrix-org/matrix-js-sdk/issues/5071
+	*/
+
+	// `state` should only ever include one membership event for the syncing user
+	let membership_event_index = state.iter().position(|pdu| {
+		*pdu.event_type() == TimelineEventType::RoomMember
+			&& pdu.state_key() == Some(syncing_user.as_str())
+	});
+
+	if let Some(index) = membership_event_index {
+		// the ordering of events in `state` does not matter
+		state.swap_remove(index);
+	}
+
+	trace!(
+		?timeline_start_count,
+		?timeline_end_count,
+		"syncing {} timeline events (limited = {}) and {} state events",
+		timeline.pdus.len(),
+		timeline.limited,
+		state.len()
+	);
+
+	Ok((timeline, state))
+}
+
+fn create_dummy_leave_event(
+	services: &Services,
+	SyncContext { syncing_user, .. }: SyncContext<'_>,
+	room_id: &RoomId,
+) -> PduEvent {
+	// TODO: because this event ID is random, it could cause caching issues with
+	// clients. perhaps a database table could be created to hold these dummy
+	// events, or they could be stored as outliers?
+	PduEvent {
+		event_id: EventId::new(services.globals.server_name()),
+		sender: syncing_user.to_owned(),
+		origin: None,
+		origin_server_ts: utils::millis_since_unix_epoch()
+			.try_into()
+			.expect("Timestamp is valid js_int value"),
+		kind: TimelineEventType::RoomMember,
+		content: RawValue::from_string(r#"{"membership": "leave"}"#.to_owned()).unwrap(),
+		state_key: Some(syncing_user.as_str().into()),
+		unsigned: None,
+		// The following keys are dropped on conversion
+		room_id: Some(room_id.to_owned()),
+		prev_events: vec![],
+		depth: uint!(1),
+		auth_events: vec![],
+		redacts: None,
+		hashes: EventHash { sha256: String::new() },
+		signatures: None,
+	}
+}

src/api/client/sync/v3/mod.rs

@@ -0,0 +1,494 @@
+mod joined;
+mod left;
+mod state;
+
+use std::{
+	cmp::{self},
+	collections::{BTreeMap, HashMap, HashSet},
+	time::Duration,
+};
+
+use axum::extract::State;
+use conduwuit::{
+	Result, extract_variant,
+	utils::{
+		ReadyExt, TryFutureExtExt,
+		stream::{BroadbandExt, Tools, WidebandExt},
+	},
+	warn,
+};
+use conduwuit_service::Services;
+use futures::{
+	FutureExt, StreamExt, TryFutureExt,
+	future::{OptionFuture, join3, join4, join5},
+};
+use ruma::{
+	DeviceId, OwnedUserId, RoomId, UserId,
+	api::client::{
+		filter::FilterDefinition,
+		sync::sync_events::{
+			self, DeviceLists,
+			v3::{
+				Filter, GlobalAccountData, InviteState, InvitedRoom, KnockState, KnockedRoom,
+				Presence, Rooms, ToDevice,
+			},
+		},
+		uiaa::UiaaResponse,
+	},
+	events::{
+		AnyRawAccountDataEvent,
+		presence::{PresenceEvent, PresenceEventContent},
+	},
+	serde::Raw,
+};
+use service::rooms::lazy_loading::{self, MemberSet, Options as _};
+
+use super::{load_timeline, share_encrypted_room};
+use crate::{
+	Ruma, RumaResponse,
+	client::{
+		is_ignored_invite,
+		sync::v3::{joined::load_joined_room, left::load_left_room},
+	},
+};
+
+/// The default maximum number of events to return in the `timeline` key of
+/// joined and left rooms. If the number of events sent since the last sync
+/// exceeds this number, the `timeline` will be `limited`.
+const DEFAULT_TIMELINE_LIMIT: usize = 30;
+
+/// A collection of updates to users' device lists, used for E2EE.
+struct DeviceListUpdates {
+	changed: HashSet<OwnedUserId>,
+	left: HashSet<OwnedUserId>,
+}
+
+impl DeviceListUpdates {
+	fn new() -> Self {
+		Self {
+			changed: HashSet::new(),
+			left: HashSet::new(),
+		}
+	}
+
+	fn merge(&mut self, other: Self) {
+		self.changed.extend(other.changed);
+		self.left.extend(other.left);
+	}
+
+	fn is_empty(&self) -> bool { self.changed.is_empty() && self.left.is_empty() }
+}
+
+impl From<DeviceListUpdates> for DeviceLists {
+	fn from(val: DeviceListUpdates) -> Self {
+		Self {
+			changed: val.changed.into_iter().collect(),
+			left: val.left.into_iter().collect(),
+		}
+	}
+}
+
+/// References to common data needed to calculate the sync response.
+#[derive(Clone, Copy)]
+struct SyncContext<'a> {
+	/// The ID of the user requesting this sync.
+	syncing_user: &'a UserId,
+	/// The ID of the device requesting this sync, which will belong to
+	/// `syncing_user`.
+	syncing_device: &'a DeviceId,
+	/// The global count at the end of the previous sync response.
+	/// The previous sync's `current_count` will become the next sync's
+	/// `last_sync_end_count`. This will be None if no `since` query parameter
+	/// was specified, indicating an initial sync.
+	last_sync_end_count: Option<u64>,
+	/// The global count as of when we started building the sync response.
+	/// This is used as an upper bound when querying the database to ensure the
+	/// response represents a snapshot in time and doesn't include data which
+	/// appeared while the response was being built.
+	current_count: u64,
+	/// The `full_state` query parameter, used when syncing state for joined and
+	/// left rooms.
+	full_state: bool,
+	/// The sync filter, which the client uses to specify what data should be
+	/// included in the sync response.
+	filter: &'a FilterDefinition,
+}
+
+impl<'a> SyncContext<'a> {
+	fn lazy_loading_context(&self, room_id: &'a RoomId) -> lazy_loading::Context<'a> {
+		lazy_loading::Context {
+			user_id: self.syncing_user,
+			device_id: Some(self.syncing_device),
+			room_id,
+			token: self.last_sync_end_count,
+			options: Some(&self.filter.room.state.lazy_load_options),
+		}
+	}
+
+	#[inline]
+	fn lazy_loading_enabled(&self) -> bool {
+		(self.filter.room.state.lazy_load_options.is_enabled()
+			|| self.filter.room.timeline.lazy_load_options.is_enabled())
+			&& !self.full_state
+	}
+}
+
+type PresenceUpdates = HashMap<OwnedUserId, PresenceEventContent>;
+
+/// # `GET /_matrix/client/r0/sync`
+///
+/// Synchronize the client's state with the latest state on the server.
+///
+/// - This endpoint takes a `since` parameter which should be the `next_batch`
+///   value from a previous request for incremental syncs.
+///
+/// Calling this endpoint without a `since` parameter returns:
+/// - Some of the most recent events of each timeline
+/// - Notification counts for each room
+/// - Joined and invited member counts, heroes
+/// - All state events
+///
+/// Calling this endpoint with a `since` parameter from a previous `next_batch`
+/// returns: For joined rooms:
+/// - Some of the most recent events of each timeline that happened after since
+/// - If user joined the room after since: All state events (unless lazy loading
+///   is activated) and all device list updates in that room
+/// - If the user was already in the room: A list of all events that are in the
+///   state now, but were not in the state at `since`
+/// - If the state we send contains a member event: Joined and invited member
+///   counts, heroes
+/// - Device list updates that happened after `since`
+/// - If there are events in the timeline we send or the user send updated his
+///   read mark: Notification counts
+/// - EDUs that are active now (read receipts, typing updates, presence)
+/// - TODO: Allow multiple sync streams to support Pantalaimon
+///
+/// For invited rooms:
+/// - If the user was invited after `since`: A subset of the state of the room
+///   at the point of the invite
+///
+/// For left rooms:
+/// - If the user left after `since`: `prev_batch` token, empty state (TODO:
+///   subset of the state at the point of the leave)
+#[tracing::instrument(
+	name = "sync",
+	level = "debug",
+	skip_all,
+	fields(
+		since = %body.body.since.as_deref().unwrap_or_default(),
+    )
+)]
+pub(crate) async fn sync_events_route(
+	State(services): State<crate::State>,
+	body: Ruma<sync_events::v3::Request>,
+) -> Result<sync_events::v3::Response, RumaResponse<UiaaResponse>> {
+	let (sender_user, sender_device) = body.sender();
+
+	// Presence update
+	if services.config.allow_local_presence {
+		services
+			.presence
+			.ping_presence(sender_user, &body.body.set_presence)
+			.await?;
+	}
+
+	// Setup watchers, so if there's no response, we can wait for them
+	let watcher = services.sync.watch(sender_user, sender_device);
+
+	let response = build_sync_events(&services, &body).await?;
+	if body.body.full_state
+		|| !(response.rooms.is_empty()
+			&& response.presence.is_empty()
+			&& response.account_data.is_empty()
+			&& response.device_lists.is_empty()
+			&& response.to_device.is_empty())
+	{
+		return Ok(response);
+	}
+
+	// Hang a few seconds so requests are not spammed
+	// Stop hanging if new info arrives
+	let default = Duration::from_secs(30);
+	let duration = cmp::min(body.body.timeout.unwrap_or(default), default);
+	_ = tokio::time::timeout(duration, watcher).await;
+
+	// Retry returning data
+	build_sync_events(&services, &body).await
+}
+
+pub(crate) async fn build_sync_events(
+	services: &Services,
+	body: &Ruma<sync_events::v3::Request>,
+) -> Result<sync_events::v3::Response, RumaResponse<UiaaResponse>> {
+	let (syncing_user, syncing_device) = body.sender();
+
+	let current_count = services.globals.current_count()?;
+
+	// the `since` token is the last sync end count stringified
+	let last_sync_end_count = body
+		.body
+		.since
+		.as_ref()
+		.and_then(|string| string.parse().ok());
+
+	let full_state = body.body.full_state;
+
+	// FilterDefinition is very large (0x1000 bytes), let's put it on the heap
+	let filter = Box::new(match body.body.filter.as_ref() {
+		// use the default filter if none was specified
+		| None => FilterDefinition::default(),
+		// use inline filters directly
+		| Some(Filter::FilterDefinition(filter)) => filter.clone(),
+		// look up filter IDs from the database
+		| Some(Filter::FilterId(filter_id)) => services
+			.users
+			.get_filter(syncing_user, filter_id)
+			.await
+			.unwrap_or_default(),
+	});
+
+	let context = SyncContext {
+		syncing_user,
+		syncing_device,
+		last_sync_end_count,
+		current_count,
+		full_state,
+		filter: &filter,
+	};
+
+	let joined_rooms = services
+		.rooms
+		.state_cache
+		.rooms_joined(syncing_user)
+		.map(ToOwned::to_owned)
+		.broad_filter_map(|room_id| async {
+			let joined_room = load_joined_room(services, context, room_id.clone()).await;
+
+			match joined_room {
+				| Ok((room, updates)) => Some((room_id, room, updates)),
+				| Err(err) => {
+					warn!(?err, ?room_id, "error loading joined room {}", room_id);
+					None
+				},
+			}
+		})
+		.ready_fold(
+			(BTreeMap::new(), DeviceListUpdates::new()),
+			|(mut joined_rooms, mut all_updates), (room_id, joined_room, updates)| {
+				all_updates.merge(updates);
+
+				if !joined_room.is_empty() {
+					joined_rooms.insert(room_id, joined_room);
+				}
+
+				(joined_rooms, all_updates)
+			},
+		);
+
+	let left_rooms = services
+		.rooms
+		.state_cache
+		.rooms_left(syncing_user)
+		.broad_filter_map(|(room_id, leave_pdu)| {
+			load_left_room(services, context, room_id.clone(), leave_pdu)
+				.map_ok(move |left_room| (room_id, left_room))
+				.ok()
+		})
+		.ready_filter_map(|(room_id, left_room)| left_room.map(|left_room| (room_id, left_room)))
+		.collect();
+
+	let invited_rooms = services
+		.rooms
+		.state_cache
+		.rooms_invited(syncing_user)
+		.wide_filter_map(async |(room_id, invite_state)| {
+			if is_ignored_invite(services, syncing_user, &room_id).await {
+				None
+			} else {
+				Some((room_id, invite_state))
+			}
+		})
+		.fold_default(|mut invited_rooms: BTreeMap<_, _>, (room_id, invite_state)| async move {
+			let invite_count = services
+				.rooms
+				.state_cache
+				.get_invite_count(&room_id, syncing_user)
+				.await
+				.ok();
+
+			// only sync this invite if it was sent after the last /sync call
+			if last_sync_end_count < invite_count {
+				let invited_room = InvitedRoom {
+					invite_state: InviteState { events: invite_state },
+				};
+
+				invited_rooms.insert(room_id, invited_room);
+			}
+			invited_rooms
+		});
+
+	let knocked_rooms = services
+		.rooms
+		.state_cache
+		.rooms_knocked(syncing_user)
+		.fold_default(|mut knocked_rooms: BTreeMap<_, _>, (room_id, knock_state)| async move {
+			let knock_count = services
+				.rooms
+				.state_cache
+				.get_knock_count(&room_id, syncing_user)
+				.await
+				.ok();
+
+			// only sync this knock if it was sent after the last /sync call
+			if last_sync_end_count < knock_count {
+				let knocked_room = KnockedRoom {
+					knock_state: KnockState { events: knock_state },
+				};
+
+				knocked_rooms.insert(room_id, knocked_room);
+			}
+			knocked_rooms
+		});
+
+	let presence_updates: OptionFuture<_> = services
+		.config
+		.allow_local_presence
+		.then(|| process_presence_updates(services, last_sync_end_count, syncing_user))
+		.into();
+
+	let account_data = services
+		.account_data
+		.changes_since(None, syncing_user, last_sync_end_count, Some(current_count))
+		.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
+		.collect();
+
+	// Look for device list updates of this account
+	let keys_changed = services
+		.users
+		.keys_changed(syncing_user, last_sync_end_count, Some(current_count))
+		.map(ToOwned::to_owned)
+		.collect::<HashSet<_>>();
+
+	let to_device_events = services
+		.users
+		.get_to_device_events(
+			syncing_user,
+			syncing_device,
+			last_sync_end_count,
+			Some(current_count),
+		)
+		.collect::<Vec<_>>();
+
+	let device_one_time_keys_count = services
+		.users
+		.count_one_time_keys(syncing_user, syncing_device);
+
+	// Remove all to-device events the device received *last time*
+	let remove_to_device_events =
+		services
+			.users
+			.remove_to_device_events(syncing_user, syncing_device, last_sync_end_count);
+
+	let rooms = join4(joined_rooms, left_rooms, invited_rooms, knocked_rooms);
+	let ephemeral = join3(remove_to_device_events, to_device_events, presence_updates);
+	let top = join5(account_data, ephemeral, device_one_time_keys_count, keys_changed, rooms)
+		.boxed()
+		.await;
+
+	let (account_data, ephemeral, device_one_time_keys_count, keys_changed, rooms) = top;
+	let ((), to_device_events, presence_updates) = ephemeral;
+	let (joined_rooms, left_rooms, invited_rooms, knocked_rooms) = rooms;
+	let (joined_rooms, mut device_list_updates) = joined_rooms;
+	device_list_updates.changed.extend(keys_changed);
+
+	let response = sync_events::v3::Response {
+		account_data: GlobalAccountData { events: account_data },
+		device_lists: device_list_updates.into(),
+		device_one_time_keys_count,
+		// Fallback keys are not yet supported
+		device_unused_fallback_key_types: None,
+		next_batch: current_count.to_string(),
+		presence: Presence {
+			events: presence_updates
+				.into_iter()
+				.flat_map(IntoIterator::into_iter)
+				.map(|(sender, content)| PresenceEvent { content, sender })
+				.map(|ref event| Raw::new(event))
+				.filter_map(Result::ok)
+				.collect(),
+		},
+		rooms: Rooms {
+			leave: left_rooms,
+			join: joined_rooms,
+			invite: invited_rooms,
+			knock: knocked_rooms,
+		},
+		to_device: ToDevice { events: to_device_events },
+	};
+
+	Ok(response)
+}
+
+#[tracing::instrument(name = "presence", level = "debug", skip_all)]
+async fn process_presence_updates(
+	services: &Services,
+	last_sync_end_count: Option<u64>,
+	syncing_user: &UserId,
+) -> PresenceUpdates {
+	services
+		.presence
+		.presence_since(last_sync_end_count.unwrap_or(0)) // send all presences on initial sync
+		.filter(|(user_id, ..)| {
+			services
+				.rooms
+				.state_cache
+				.user_sees_user(syncing_user, user_id)
+		})
+		.filter_map(|(user_id, _, presence_bytes)| {
+			services
+				.presence
+				.from_json_bytes_to_event(presence_bytes, user_id)
+				.map_ok(move |event| (user_id, event))
+				.ok()
+		})
+		.map(|(user_id, event)| (user_id.to_owned(), event.content))
+		.collect()
+		.await
+}
+
+/// Using the provided sync context and an iterator of user IDs in the
+/// `timeline`, return a HashSet of user IDs whose membership events should be
+/// sent to the client if lazy-loading is enabled.
+#[allow(clippy::let_and_return)]
+async fn prepare_lazily_loaded_members(
+	services: &Services,
+	sync_context: SyncContext<'_>,
+	room_id: &RoomId,
+	timeline_members: impl Iterator<Item = OwnedUserId>,
+) -> Option<MemberSet> {
+	let lazy_loading_context = &sync_context.lazy_loading_context(room_id);
+
+	// reset lazy loading state on initial sync.
+	// do this even if lazy loading is disabled so future lazy loads
+	// will have the correct members.
+	if sync_context.last_sync_end_count.is_none() {
+		services
+			.rooms
+			.lazy_loading
+			.reset(lazy_loading_context)
+			.await;
+	}
+
+	// filter the input members through `retain_lazy_members`, which
+	// contains the actual lazy loading logic.
+	let lazily_loaded_members =
+		OptionFuture::from(sync_context.lazy_loading_enabled().then(|| {
+			services
+				.rooms
+				.lazy_loading
+				.retain_lazy_members(timeline_members.collect(), lazy_loading_context)
+		}))
+		.await;
+
+	lazily_loaded_members
+}

src/api/client/sync/v3/state.rs

@@ -0,0 +1,280 @@
+use std::{collections::BTreeSet, ops::ControlFlow};
+
+use conduwuit::{
+	Result, at, is_equal_to,
+	matrix::{
+		Event,
+		pdu::{PduCount, PduEvent},
+	},
+	utils::{
+		BoolExt, IterStream, ReadyExt, TryFutureExtExt,
+		stream::{BroadbandExt, TryIgnore},
+	},
+};
+use conduwuit_service::{
+	Services,
+	rooms::{lazy_loading::MemberSet, short::ShortStateHash},
+};
+use futures::{FutureExt, StreamExt};
+use itertools::Itertools;
+use ruma::{OwnedEventId, RoomId, UserId, events::StateEventType};
+use service::rooms::short::ShortEventId;
+use tracing::trace;
+
+use crate::client::TimelinePdus;
+
+/// Calculate the state events to include in an initial sync response.
+///
+/// If lazy-loading is enabled (`lazily_loaded_members` is Some), the returned
+/// Vec will include the membership events of exclusively the members in
+/// `lazily_loaded_members`.
+#[tracing::instrument(
+	name = "initial",
+	level = "trace",
+	skip_all,
+	fields(current_shortstatehash)
+)]
+#[allow(clippy::too_many_arguments)]
+pub(super) async fn build_state_initial(
+	services: &Services,
+	sender_user: &UserId,
+	timeline_start_shortstatehash: ShortStateHash,
+	lazily_loaded_members: Option<&MemberSet>,
+) -> Result<Vec<PduEvent>> {
+	// load the keys and event IDs of the state events at the start of the timeline
+	let (shortstatekeys, event_ids): (Vec<_>, Vec<_>) = services
+		.rooms
+		.state_accessor
+		.state_full_ids(timeline_start_shortstatehash)
+		.unzip()
+		.await;
+
+	trace!("performing initial sync of {} state events", event_ids.len());
+
+	services
+		.rooms
+		.short
+		// look up the full state keys
+		.multi_get_statekey_from_short(shortstatekeys.into_iter().stream())
+		.zip(event_ids.into_iter().stream())
+		.ready_filter_map(|item| Some((item.0.ok()?, item.1)))
+		.ready_filter_map(|((event_type, state_key), event_id)| {
+			if let Some(lazily_loaded_members) = lazily_loaded_members {
+				/*
+				if lazy loading is enabled, filter out membership events which aren't for a user
+				included in `lazily_loaded_members` or for the user requesting the sync.
+				*/
+				let event_is_redundant = event_type == StateEventType::RoomMember
+					&& state_key.as_str().try_into().is_ok_and(|user_id: &UserId| {
+						sender_user != user_id && !lazily_loaded_members.contains(user_id)
+					});
+
+				event_is_redundant.or_some(event_id)
+			} else {
+				Some(event_id)
+			}
+		})
+		.broad_filter_map(|event_id: OwnedEventId| async move {
+			services.rooms.timeline.get_pdu(&event_id).await.ok()
+		})
+		.collect()
+		.map(Ok)
+		.await
+}
+
+/// Calculate the state events to include in an incremental sync response.
+///
+/// If lazy-loading is enabled (`lazily_loaded_members` is Some), the returned
+/// Vec will include the membership events of all the members in
+/// `lazily_loaded_members`.
+#[tracing::instrument(name = "incremental", level = "trace", skip_all)]
+#[allow(clippy::too_many_arguments)]
+pub(super) async fn build_state_incremental<'a>(
+	services: &Services,
+	sender_user: &'a UserId,
+	room_id: &RoomId,
+	last_sync_end_count: PduCount,
+	last_sync_end_shortstatehash: ShortStateHash,
+	timeline_start_shortstatehash: ShortStateHash,
+	timeline_end_shortstatehash: ShortStateHash,
+	timeline: &TimelinePdus,
+	lazily_loaded_members: Option<&'a MemberSet>,
+) -> Result<Vec<PduEvent>> {
+	/*
+	NB: a limited sync is one where `timeline.limited == true`. Synapse calls this a "gappy" sync internally.
+
+	The algorithm implemented in this function is, currently, quite different from the algorithm vaguely described
+	by the Matrix specification. This is because the specification's description of the `state` property does not accurately
+	reflect how Synapse behaves, and therefore how client SDKs behave. Notable differences include:
+	1. We do not compute the delta using the naive approach of "every state event from the end of the last sync
+	   up to the start of this sync's timeline". see below for details.
+	2. If lazy-loading is enabled, we include lazily-loaded membership events. The specific users to include are determined
+	   elsewhere and supplied to this function in the `lazily_loaded_members` parameter.
+	*/
+
+	/*
+	the `state` property of an incremental sync which isn't limited are _usually_ empty.
+	(note: the specification says that the `state` property is _always_ empty for limited syncs, which is incorrect.)
+	however, if an event in the timeline (`timeline.pdus`) merges a split in the room's DAG (i.e. has multiple `prev_events`),
+	the state at the _end_ of the timeline may include state events which were merged in and don't exist in the state
+	at the _start_ of the timeline. because this is uncommon, we check here to see if any events in the timeline
+	merged a split in the DAG.
+
+	see: https://github.com/element-hq/synapse/issues/16941
+	*/
+
+	let timeline_is_linear = timeline.pdus.is_empty() || {
+		let last_pdu_of_last_sync = services
+			.rooms
+			.timeline
+			.pdus_rev(Some(sender_user), room_id, Some(last_sync_end_count.saturating_add(1)))
+			.boxed()
+			.next()
+			.await
+			.transpose()
+			.expect("last sync should have had some PDUs")
+			.map(at!(1));
+
+		// make sure the prev_events of each pdu in the timeline refer only to the
+		// previous pdu
+		timeline
+			.pdus
+			.iter()
+			.try_fold(last_pdu_of_last_sync.map(|pdu| pdu.event_id), |prev_event_id, (_, pdu)| {
+				if let Ok(pdu_prev_event_id) = pdu.prev_events.iter().exactly_one() {
+					if prev_event_id
+						.as_ref()
+						.is_none_or(is_equal_to!(pdu_prev_event_id))
+					{
+						return ControlFlow::Continue(Some(pdu_prev_event_id.to_owned()));
+					}
+				}
+
+				trace!(
+					"pdu {:?} has split prev_events (expected {:?}): {:?}",
+					pdu.event_id, prev_event_id, pdu.prev_events
+				);
+				ControlFlow::Break(())
+			})
+			.is_continue()
+	};
+
+	if timeline_is_linear && !timeline.limited {
+		// if there are no splits in the DAG and the timeline isn't limited, then
+		// `state` will always be empty unless lazy loading is enabled.
+
+		if let Some(lazily_loaded_members) = lazily_loaded_members {
+			if !timeline.pdus.is_empty() {
+				// lazy loading is enabled, so we return the membership events which were
+				// requested by the caller.
+				let lazy_membership_events: Vec<_> = lazily_loaded_members
+					.iter()
+					.stream()
+					.broad_filter_map(|user_id| async move {
+						if user_id == sender_user {
+							return None;
+						}
+
+						services
+							.rooms
+							.state_accessor
+							.state_get(
+								timeline_start_shortstatehash,
+								&StateEventType::RoomMember,
+								user_id.as_str(),
+							)
+							.ok()
+							.await
+					})
+					.collect()
+					.await;
+
+				if !lazy_membership_events.is_empty() {
+					trace!(
+						"syncing lazy membership events for members: {:?}",
+						lazy_membership_events
+							.iter()
+							.map(|pdu| pdu.state_key().unwrap())
+							.collect::<Vec<_>>()
+					);
+				}
+				return Ok(lazy_membership_events);
+			}
+		}
+
+		// lazy loading is disabled, `state` is empty.
+		return Ok(vec![]);
+	}
+
+	/*
+	at this point, either the timeline is `limited` or the DAG has a split in it. this necessitates
+	computing the incremental state (which may be empty).
+
+	NOTE: this code path does not use the `lazy_membership_events` parameter. any changes to membership will be included
+	in the incremental state. therefore, the incremental state may include "redundant" membership events,
+	which we do not filter out because A. the spec forbids lazy-load filtering if the timeline is `limited`,
+	and B. DAG splits which require sending extra membership state events are (probably) uncommon enough that
+	the performance penalty is acceptable.
+	*/
+
+	trace!(?timeline_is_linear, ?timeline.limited, "computing state for incremental sync");
+
+	// fetch the shorteventids of state events in the timeline
+	let state_events_in_timeline: BTreeSet<ShortEventId> = services
+		.rooms
+		.short
+		.multi_get_or_create_shorteventid(timeline.pdus.iter().filter_map(|(_, pdu)| {
+			if pdu.state_key().is_some() {
+				Some(pdu.event_id.as_ref())
+			} else {
+				None
+			}
+		}))
+		.collect()
+		.await;
+
+	trace!("{} state events in timeline", state_events_in_timeline.len());
+
+	/*
+	fetch the state events which were added since the last sync.
+
+	specifically we fetch the difference between the state at the last sync and the state at the _end_
+	of the timeline, and then we filter out state events in the timeline itself using the shorteventids we fetched.
+	this is necessary to account for splits in the DAG, as explained above.
+	*/
+	let state_diff = services
+		.rooms
+		.short
+		.multi_get_eventid_from_short::<'_, OwnedEventId, _>(
+			services
+				.rooms
+				.state_accessor
+				.state_added((last_sync_end_shortstatehash, timeline_end_shortstatehash))
+				.await?
+				.stream()
+				.ready_filter_map(|(_, shorteventid)| {
+					if state_events_in_timeline.contains(&shorteventid) {
+						None
+					} else {
+						Some(shorteventid)
+					}
+				}),
+		)
+		.ignore_err();
+
+	// finally, fetch the PDU contents and collect them into a vec
+	let state_diff_pdus = state_diff
+		.broad_filter_map(|event_id| async move {
+			services
+				.rooms
+				.timeline
+				.get_non_outlier_pdu(&event_id)
+				.await
+				.ok()
+		})
+		.collect::<Vec<_>>()
+		.await;
+
+	trace!(?state_diff_pdus, "collected state PDUs for incremental sync");
+	Ok(state_diff_pdus)
+}

src/api/client/sync/v4.rs

@@ -1,848 +0,0 @@
-use std::{
-	cmp::{self, Ordering},
-	collections::{BTreeMap, BTreeSet, HashMap, HashSet},
-	time::Duration,
-};
-
-use axum::extract::State;
-use conduwuit::{
-	Err, Error, Event, PduCount, Result, at, debug, error, extract_variant,
-	matrix::TypeStateKey,
-	utils::{
-		BoolExt, IterStream, ReadyExt, TryFutureExtExt,
-		math::{ruma_from_usize, usize_from_ruma, usize_from_u64_truncated},
-		stream::WidebandExt,
-	},
-	warn,
-};
-use conduwuit_service::{
-	Services,
-	rooms::read_receipt::pack_receipts,
-	sync::{into_db_key, into_snake_key},
-};
-use futures::{FutureExt, StreamExt, TryFutureExt};
-use ruma::{
-	MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, RoomId, UInt, UserId,
-	api::client::sync::sync_events::{
-		self, DeviceLists, UnreadNotificationsCount,
-		v4::{SlidingOp, SlidingSyncRoomHero},
-	},
-	directory::RoomTypeFilter,
-	events::{
-		AnyRawAccountDataEvent, AnySyncEphemeralRoomEvent, StateEventType,
-		TimelineEventType::*,
-		room::member::{MembershipState, RoomMemberEventContent},
-	},
-	serde::Raw,
-	uint,
-};
-
-use super::{load_timeline, share_encrypted_room};
-use crate::{
-	Ruma,
-	client::{DEFAULT_BUMP_TYPES, ignored_filter, is_ignored_invite},
-};
-
-type TodoRooms = BTreeMap<OwnedRoomId, (BTreeSet<TypeStateKey>, usize, u64)>;
-const SINGLE_CONNECTION_SYNC: &str = "single_connection_sync";
-
-#[allow(clippy::cognitive_complexity)]
-/// POST `/_matrix/client/unstable/org.matrix.msc3575/sync`
-///
-/// Sliding Sync endpoint (future endpoint: `/_matrix/client/v4/sync`)
-pub(crate) async fn sync_events_v4_route(
-	State(services): State<crate::State>,
-	body: Ruma<sync_events::v4::Request>,
-) -> Result<sync_events::v4::Response> {
-	debug_assert!(DEFAULT_BUMP_TYPES.is_sorted(), "DEFAULT_BUMP_TYPES is not sorted");
-	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
-	let sender_device = body.sender_device.as_ref().expect("user is authenticated");
-	let mut body = body.body;
-
-	// Setup watchers, so if there's no response, we can wait for them
-	let watcher = services.sync.watch(sender_user, sender_device);
-
-	let next_batch = services.globals.next_count()?;
-
-	let conn_id = body
-		.conn_id
-		.clone()
-		.unwrap_or_else(|| SINGLE_CONNECTION_SYNC.to_owned());
-
-	let globalsince = body
-		.pos
-		.as_ref()
-		.and_then(|string| string.parse().ok())
-		.unwrap_or(0);
-
-	let db_key = into_db_key(sender_user, sender_device, conn_id.clone());
-	if globalsince != 0 && !services.sync.remembered(&db_key) {
-		debug!("Restarting sync stream because it was gone from the database");
-		return Err!(Request(UnknownPos("Connection data lost since last time")));
-	}
-
-	if globalsince == 0 {
-		services.sync.forget_sync_request_connection(&db_key);
-	}
-
-	// Get sticky parameters from cache
-	let snake_key = into_snake_key(sender_user, sender_device, conn_id.clone());
-	let known_rooms = services
-		.sync
-		.update_sync_request_with_cache(&snake_key, &mut body);
-
-	let all_joined_rooms: Vec<_> = services
-		.rooms
-		.state_cache
-		.rooms_joined(sender_user)
-		.map(ToOwned::to_owned)
-		.collect()
-		.await;
-
-	let all_invited_rooms: Vec<_> = services
-		.rooms
-		.state_cache
-		.rooms_invited(sender_user)
-		.wide_filter_map(async |(room_id, invite_state)| {
-			if is_ignored_invite(&services, sender_user, &room_id).await {
-				None
-			} else {
-				Some((room_id, invite_state))
-			}
-		})
-		.map(|r| r.0)
-		.collect()
-		.await;
-
-	let all_knocked_rooms: Vec<_> = services
-		.rooms
-		.state_cache
-		.rooms_knocked(sender_user)
-		.map(|r| r.0)
-		.collect()
-		.await;
-
-	let all_invited_rooms: Vec<&RoomId> = all_invited_rooms.iter().map(AsRef::as_ref).collect();
-	let all_knocked_rooms: Vec<&RoomId> = all_knocked_rooms.iter().map(AsRef::as_ref).collect();
-
-	let all_rooms: Vec<&RoomId> = all_joined_rooms
-		.iter()
-		.map(AsRef::as_ref)
-		.chain(all_invited_rooms.iter().map(AsRef::as_ref))
-		.chain(all_knocked_rooms.iter().map(AsRef::as_ref))
-		.collect();
-
-	let all_joined_rooms = all_joined_rooms.iter().map(AsRef::as_ref).collect();
-	let all_invited_rooms = all_invited_rooms.iter().map(AsRef::as_ref).collect();
-
-	if body.extensions.to_device.enabled.unwrap_or(false) {
-		services
-			.users
-			.remove_to_device_events(sender_user, sender_device, globalsince)
-			.await;
-	}
-
-	let mut left_encrypted_users = HashSet::new(); // Users that have left any encrypted rooms the sender was in
-	let mut device_list_changes = HashSet::new();
-	let mut device_list_left = HashSet::new();
-
-	let mut receipts = sync_events::v4::Receipts { rooms: BTreeMap::new() };
-
-	let mut account_data = sync_events::v4::AccountData {
-		global: Vec::new(),
-		rooms: BTreeMap::new(),
-	};
-	if body.extensions.account_data.enabled.unwrap_or(false) {
-		account_data.global = services
-			.account_data
-			.changes_since(None, sender_user, globalsince, Some(next_batch))
-			.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
-			.collect()
-			.await;
-
-		if let Some(rooms) = body.extensions.account_data.rooms {
-			for room in rooms {
-				account_data.rooms.insert(
-					room.clone(),
-					services
-						.account_data
-						.changes_since(Some(&room), sender_user, globalsince, Some(next_batch))
-						.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
-						.collect()
-						.await,
-				);
-			}
-		}
-	}
-
-	if body.extensions.e2ee.enabled.unwrap_or(false) {
-		// Look for device list updates of this account
-		device_list_changes.extend(
-			services
-				.users
-				.keys_changed(sender_user, globalsince, None)
-				.map(ToOwned::to_owned)
-				.collect::<Vec<_>>()
-				.await,
-		);
-
-		for room_id in &all_joined_rooms {
-			let room_id: &&RoomId = room_id;
-			let Ok(current_shortstatehash) =
-				services.rooms.state.get_room_shortstatehash(room_id).await
-			else {
-				error!("Room {room_id} has no state");
-				continue;
-			};
-
-			let since_shortstatehash = services
-				.rooms
-				.user
-				.get_token_shortstatehash(room_id, globalsince)
-				.await
-				.ok();
-
-			let encrypted_room = services
-				.rooms
-				.state_accessor
-				.state_get(current_shortstatehash, &StateEventType::RoomEncryption, "")
-				.await
-				.is_ok();
-
-			if let Some(since_shortstatehash) = since_shortstatehash {
-				// Skip if there are only timeline changes
-				if since_shortstatehash == current_shortstatehash {
-					continue;
-				}
-
-				let since_encryption = services
-					.rooms
-					.state_accessor
-					.state_get(since_shortstatehash, &StateEventType::RoomEncryption, "")
-					.await;
-
-				let since_sender_member: Option<RoomMemberEventContent> = services
-					.rooms
-					.state_accessor
-					.state_get_content(
-						since_shortstatehash,
-						&StateEventType::RoomMember,
-						sender_user.as_str(),
-					)
-					.ok()
-					.await;
-
-				let joined_since_last_sync = since_sender_member
-					.as_ref()
-					.is_none_or(|member| member.membership != MembershipState::Join);
-
-				let new_encrypted_room = encrypted_room && since_encryption.is_err();
-
-				if encrypted_room {
-					let current_state_ids: HashMap<_, OwnedEventId> = services
-						.rooms
-						.state_accessor
-						.state_full_ids(current_shortstatehash)
-						.collect()
-						.await;
-
-					let since_state_ids: HashMap<_, _> = services
-						.rooms
-						.state_accessor
-						.state_full_ids(since_shortstatehash)
-						.collect()
-						.await;
-
-					for (key, id) in current_state_ids {
-						if since_state_ids.get(&key) != Some(&id) {
-							let Ok(pdu) = services.rooms.timeline.get_pdu(&id).await else {
-								error!("Pdu in state not found: {id}");
-								continue;
-							};
-							if pdu.kind == RoomMember {
-								if let Some(Ok(user_id)) =
-									pdu.state_key.as_deref().map(UserId::parse)
-								{
-									if user_id == sender_user {
-										continue;
-									}
-
-									let content: RoomMemberEventContent = pdu.get_content()?;
-									match content.membership {
-										| MembershipState::Join => {
-											// A new user joined an encrypted room
-											if !share_encrypted_room(
-												&services,
-												sender_user,
-												user_id,
-												Some(room_id),
-											)
-											.await
-											{
-												device_list_changes.insert(user_id.to_owned());
-											}
-										},
-										| MembershipState::Leave => {
-											// Write down users that have left encrypted rooms we
-											// are in
-											left_encrypted_users.insert(user_id.to_owned());
-										},
-										| _ => {},
-									}
-								}
-							}
-						}
-					}
-					if joined_since_last_sync || new_encrypted_room {
-						// If the user is in a new encrypted room, give them all joined users
-						device_list_changes.extend(
-							services
-								.rooms
-								.state_cache
-								.room_members(room_id)
-								// Don't send key updates from the sender to the sender
-								.ready_filter(|&user_id| sender_user != user_id)
-								// Only send keys if the sender doesn't share an encrypted room with the target
-								// already
-								.filter_map(|user_id| {
-									share_encrypted_room(&services, sender_user, user_id, Some(room_id))
-										.map(|res| res.or_some(user_id.to_owned()))
-								})
-								.collect::<Vec<_>>()
-								.await,
-						);
-					}
-				}
-			}
-			// Look for device list updates in this room
-			device_list_changes.extend(
-				services
-					.users
-					.room_keys_changed(room_id, globalsince, None)
-					.map(|(user_id, _)| user_id)
-					.map(ToOwned::to_owned)
-					.collect::<Vec<_>>()
-					.await,
-			);
-		}
-
-		for user_id in left_encrypted_users {
-			let dont_share_encrypted_room =
-				!share_encrypted_room(&services, sender_user, &user_id, None).await;
-
-			// If the user doesn't share an encrypted room with the target anymore, we need
-			// to tell them
-			if dont_share_encrypted_room {
-				device_list_left.insert(user_id);
-			}
-		}
-	}
-
-	let mut lists = BTreeMap::new();
-	let mut todo_rooms: TodoRooms = BTreeMap::new(); // and required state
-
-	for (list_id, list) in &body.lists {
-		let active_rooms = match list.filters.clone().and_then(|f| f.is_invite) {
-			| Some(true) => &all_invited_rooms,
-			| Some(false) => &all_joined_rooms,
-			| None => &all_rooms,
-		};
-
-		let active_rooms = match list.filters.clone().map(|f| f.not_room_types) {
-			| Some(filter) if filter.is_empty() => active_rooms.clone(),
-			| Some(value) => filter_rooms(&services, active_rooms, &value, true).await,
-			| None => active_rooms.clone(),
-		};
-
-		let active_rooms = match list.filters.clone().map(|f| f.room_types) {
-			| Some(filter) if filter.is_empty() => active_rooms.clone(),
-			| Some(value) => filter_rooms(&services, &active_rooms, &value, false).await,
-			| None => active_rooms,
-		};
-
-		let mut new_known_rooms: BTreeSet<OwnedRoomId> = BTreeSet::new();
-
-		let ranges = list.ranges.clone();
-		lists.insert(list_id.clone(), sync_events::v4::SyncList {
-			ops: ranges
-				.into_iter()
-				.map(|mut r| {
-					r.0 = r.0.clamp(
-						uint!(0),
-						UInt::try_from(active_rooms.len().saturating_sub(1)).unwrap_or(UInt::MAX),
-					);
-					r.1 = r.1.clamp(
-						r.0,
-						UInt::try_from(active_rooms.len().saturating_sub(1)).unwrap_or(UInt::MAX),
-					);
-
-					let room_ids = if !active_rooms.is_empty() {
-						active_rooms[usize_from_ruma(r.0)..=usize_from_ruma(r.1)].to_vec()
-					} else {
-						Vec::new()
-					};
-
-					new_known_rooms.extend(room_ids.clone().into_iter().map(ToOwned::to_owned));
-					for room_id in &room_ids {
-						let todo_room = todo_rooms.entry((*room_id).to_owned()).or_insert((
-							BTreeSet::new(),
-							0_usize,
-							u64::MAX,
-						));
-
-						let limit: usize = list
-							.room_details
-							.timeline_limit
-							.map(u64::from)
-							.map_or(10, usize_from_u64_truncated)
-							.min(100);
-
-						todo_room.0.extend(
-							list.room_details
-								.required_state
-								.iter()
-								.map(|(ty, sk)| (ty.clone(), sk.as_str().into())),
-						);
-
-						todo_room.1 = todo_room.1.max(limit);
-						// 0 means unknown because it got out of date
-						todo_room.2 = todo_room.2.min(
-							known_rooms
-								.get(list_id.as_str())
-								.and_then(|k| k.get(*room_id))
-								.copied()
-								.unwrap_or(0),
-						);
-					}
-					sync_events::v4::SyncOp {
-						op: SlidingOp::Sync,
-						range: Some(r),
-						index: None,
-						room_ids: room_ids.into_iter().map(ToOwned::to_owned).collect(),
-						room_id: None,
-					}
-				})
-				.collect(),
-			count: ruma_from_usize(active_rooms.len()),
-		});
-
-		if let Some(conn_id) = &body.conn_id {
-			let db_key = into_db_key(sender_user, sender_device, conn_id);
-			services.sync.update_sync_known_rooms(
-				&db_key,
-				list_id.clone(),
-				new_known_rooms,
-				globalsince,
-			);
-		}
-	}
-
-	let mut known_subscription_rooms = BTreeSet::new();
-	for (room_id, room) in &body.room_subscriptions {
-		if !services.rooms.metadata.exists(room_id).await
-			|| services.rooms.metadata.is_disabled(room_id).await
-			|| services.rooms.metadata.is_banned(room_id).await
-		{
-			continue;
-		}
-		let todo_room =
-			todo_rooms
-				.entry(room_id.clone())
-				.or_insert((BTreeSet::new(), 0_usize, u64::MAX));
-
-		let limit: usize = room
-			.timeline_limit
-			.map(u64::from)
-			.map_or(10, usize_from_u64_truncated)
-			.min(100);
-
-		todo_room.0.extend(
-			room.required_state
-				.iter()
-				.map(|(ty, sk)| (ty.clone(), sk.as_str().into())),
-		);
-		todo_room.1 = todo_room.1.max(limit);
-		// 0 means unknown because it got out of date
-		todo_room.2 = todo_room.2.min(
-			known_rooms
-				.get("subscriptions")
-				.and_then(|k| k.get(room_id))
-				.copied()
-				.unwrap_or(0),
-		);
-		known_subscription_rooms.insert(room_id.clone());
-	}
-
-	for r in body.unsubscribe_rooms {
-		known_subscription_rooms.remove(&r);
-		body.room_subscriptions.remove(&r);
-	}
-
-	if let Some(conn_id) = &body.conn_id {
-		let db_key = into_db_key(sender_user, sender_device, conn_id);
-		services.sync.update_sync_known_rooms(
-			&db_key,
-			"subscriptions".to_owned(),
-			known_subscription_rooms,
-			globalsince,
-		);
-	}
-
-	if let Some(conn_id) = body.conn_id.clone() {
-		let db_key = into_db_key(sender_user, sender_device, conn_id);
-		services
-			.sync
-			.update_sync_subscriptions(&db_key, body.room_subscriptions);
-	}
-
-	let mut rooms = BTreeMap::new();
-	for (room_id, (required_state_request, timeline_limit, roomsince)) in &todo_rooms {
-		let roomsincecount = PduCount::Normal(*roomsince);
-
-		let mut timestamp: Option<_> = None;
-		let mut invite_state = None;
-		let (timeline_pdus, limited);
-		let new_room_id: &RoomId = (*room_id).as_ref();
-		if all_invited_rooms.contains(&new_room_id) {
-			// TODO: figure out a timestamp we can use for remote invites
-			invite_state = services
-				.rooms
-				.state_cache
-				.invite_state(sender_user, room_id)
-				.await
-				.ok();
-
-			(timeline_pdus, limited) = (Vec::new(), true);
-		} else {
-			(timeline_pdus, limited) = match load_timeline(
-				&services,
-				sender_user,
-				room_id,
-				roomsincecount,
-				None,
-				*timeline_limit,
-			)
-			.await
-			{
-				| Ok(value) => value,
-				| Err(err) => {
-					warn!("Encountered missing timeline in {}, error {}", room_id, err);
-					continue;
-				},
-			};
-		}
-
-		account_data.rooms.insert(
-			room_id.to_owned(),
-			services
-				.account_data
-				.changes_since(Some(room_id), sender_user, *roomsince, Some(next_batch))
-				.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
-				.collect()
-				.await,
-		);
-
-		let last_privateread_update = services
-			.rooms
-			.read_receipt
-			.last_privateread_update(sender_user, room_id)
-			.await > *roomsince;
-
-		let private_read_event = if last_privateread_update {
-			services
-				.rooms
-				.read_receipt
-				.private_read_get(room_id, sender_user)
-				.await
-				.ok()
-		} else {
-			None
-		};
-
-		let mut vector: Vec<Raw<AnySyncEphemeralRoomEvent>> = services
-			.rooms
-			.read_receipt
-			.readreceipts_since(room_id, *roomsince)
-			.filter_map(|(read_user, _ts, v)| async move {
-				services
-					.users
-					.user_is_ignored(read_user, sender_user)
-					.await
-					.or_some(v)
-			})
-			.collect()
-			.await;
-
-		if let Some(private_read_event) = private_read_event {
-			vector.push(private_read_event);
-		}
-
-		let receipt_size = vector.len();
-		receipts
-			.rooms
-			.insert(room_id.clone(), pack_receipts(Box::new(vector.into_iter())));
-
-		if roomsince != &0
-			&& timeline_pdus.is_empty()
-			&& account_data.rooms.get(room_id).is_some_and(Vec::is_empty)
-			&& receipt_size == 0
-		{
-			continue;
-		}
-
-		let prev_batch = timeline_pdus
-			.first()
-			.map_or(Ok::<_, Error>(None), |(pdu_count, _)| {
-				Ok(Some(match pdu_count {
-					| PduCount::Backfilled(_) => {
-						error!("timeline in backfill state?!");
-						"0".to_owned()
-					},
-					| PduCount::Normal(c) => c.to_string(),
-				}))
-			})?
-			.or_else(|| {
-				if roomsince != &0 {
-					Some(roomsince.to_string())
-				} else {
-					None
-				}
-			});
-
-		let room_events: Vec<_> = timeline_pdus
-			.iter()
-			.stream()
-			.filter_map(|item| ignored_filter(&services, item.clone(), sender_user))
-			.map(at!(1))
-			.map(Event::into_format)
-			.collect()
-			.await;
-
-		for (_, pdu) in timeline_pdus {
-			let ts = MilliSecondsSinceUnixEpoch(pdu.origin_server_ts);
-			if DEFAULT_BUMP_TYPES.binary_search(&pdu.kind).is_ok()
-				&& timestamp.is_none_or(|time| time <= ts)
-			{
-				timestamp = Some(ts);
-			}
-		}
-
-		let required_state = required_state_request
-			.iter()
-			.stream()
-			.filter_map(|state| async move {
-				services
-					.rooms
-					.state_accessor
-					.room_state_get(room_id, &state.0, &state.1)
-					.await
-					.map(Event::into_format)
-					.ok()
-			})
-			.collect()
-			.await;
-
-		// Heroes
-		let heroes: Vec<_> = services
-			.rooms
-			.state_cache
-			.room_members(room_id)
-			.ready_filter(|&member| member != sender_user)
-			.filter_map(|user_id| {
-				services
-					.rooms
-					.state_accessor
-					.get_member(room_id, user_id)
-					.map_ok(|memberevent| SlidingSyncRoomHero {
-						user_id: user_id.into(),
-						name: memberevent.displayname,
-						avatar: memberevent.avatar_url,
-					})
-					.ok()
-			})
-			.take(5)
-			.collect()
-			.await;
-
-		let name = match heroes.len().cmp(&(1_usize)) {
-			| Ordering::Greater => {
-				let firsts = heroes[1..]
-					.iter()
-					.map(|h| h.name.clone().unwrap_or_else(|| h.user_id.to_string()))
-					.collect::<Vec<_>>()
-					.join(", ");
-
-				let last = heroes[0]
-					.name
-					.clone()
-					.unwrap_or_else(|| heroes[0].user_id.to_string());
-
-				Some(format!("{firsts} and {last}"))
-			},
-			| Ordering::Equal => Some(
-				heroes[0]
-					.name
-					.clone()
-					.unwrap_or_else(|| heroes[0].user_id.to_string()),
-			),
-			| Ordering::Less => None,
-		};
-
-		let heroes_avatar = if heroes.len() == 1 {
-			heroes[0].avatar.clone()
-		} else {
-			None
-		};
-
-		rooms.insert(room_id.clone(), sync_events::v4::SlidingSyncRoom {
-			name: services
-				.rooms
-				.state_accessor
-				.get_name(room_id)
-				.await
-				.ok()
-				.or(name),
-			avatar: match heroes_avatar {
-				| Some(heroes_avatar) => ruma::JsOption::Some(heroes_avatar),
-				| _ => match services.rooms.state_accessor.get_avatar(room_id).await {
-					| ruma::JsOption::Some(avatar) => ruma::JsOption::from_option(avatar.url),
-					| ruma::JsOption::Null => ruma::JsOption::Null,
-					| ruma::JsOption::Undefined => ruma::JsOption::Undefined,
-				},
-			},
-			initial: Some(roomsince == &0),
-			is_dm: None,
-			invite_state,
-			unread_notifications: UnreadNotificationsCount {
-				highlight_count: Some(
-					services
-						.rooms
-						.user
-						.highlight_count(sender_user, room_id)
-						.await
-						.try_into()
-						.expect("notification count can't go that high"),
-				),
-				notification_count: Some(
-					services
-						.rooms
-						.user
-						.notification_count(sender_user, room_id)
-						.await
-						.try_into()
-						.expect("notification count can't go that high"),
-				),
-			},
-			timeline: room_events,
-			required_state,
-			prev_batch,
-			limited,
-			joined_count: Some(
-				services
-					.rooms
-					.state_cache
-					.room_joined_count(room_id)
-					.await
-					.unwrap_or(0)
-					.try_into()
-					.unwrap_or_else(|_| uint!(0)),
-			),
-			invited_count: Some(
-				services
-					.rooms
-					.state_cache
-					.room_invited_count(room_id)
-					.await
-					.unwrap_or(0)
-					.try_into()
-					.unwrap_or_else(|_| uint!(0)),
-			),
-			num_live: None, // Count events in timeline greater than global sync counter
-			timestamp,
-			heroes: Some(heroes),
-		});
-	}
-
-	if rooms.iter().all(|(id, r)| {
-		r.timeline.is_empty() && r.required_state.is_empty() && !receipts.rooms.contains_key(id)
-	}) {
-		// Hang a few seconds so requests are not spammed
-		// Stop hanging if new info arrives
-		let default = Duration::from_secs(30);
-		let duration = cmp::min(body.timeout.unwrap_or(default), default);
-		_ = tokio::time::timeout(duration, watcher).await;
-	}
-
-	Ok(sync_events::v4::Response {
-		initial: globalsince == 0,
-		txn_id: body.txn_id.clone(),
-		pos: next_batch.to_string(),
-		lists,
-		rooms,
-		extensions: sync_events::v4::Extensions {
-			to_device: if body.extensions.to_device.enabled.unwrap_or(false) {
-				Some(sync_events::v4::ToDevice {
-					events: services
-						.users
-						.get_to_device_events(
-							sender_user,
-							sender_device,
-							Some(globalsince),
-							Some(next_batch),
-						)
-						.collect()
-						.await,
-					next_batch: next_batch.to_string(),
-				})
-			} else {
-				None
-			},
-			e2ee: sync_events::v4::E2EE {
-				device_lists: DeviceLists {
-					changed: device_list_changes.into_iter().collect(),
-					left: device_list_left.into_iter().collect(),
-				},
-				device_one_time_keys_count: services
-					.users
-					.count_one_time_keys(sender_user, sender_device)
-					.await,
-				// Fallback keys are not yet supported
-				device_unused_fallback_key_types: None,
-			},
-			account_data,
-			receipts,
-			typing: sync_events::v4::Typing { rooms: BTreeMap::new() },
-		},
-		delta_token: None,
-	})
-}
-
-async fn filter_rooms<'a>(
-	services: &Services,
-	rooms: &[&'a RoomId],
-	filter: &[RoomTypeFilter],
-	negate: bool,
-) -> Vec<&'a RoomId> {
-	rooms
-		.iter()
-		.stream()
-		.filter_map(|r| async move {
-			let room_type = services.rooms.state_accessor.get_room_type(r).await;
-
-			if room_type.as_ref().is_err_and(|e| !e.is_not_found()) {
-				return None;
-			}
-
-			let room_type_filter = RoomTypeFilter::from(room_type.ok());
-
-			let include = if negate {
-				!filter.contains(&room_type_filter)
-			} else {
-				filter.is_empty() || filter.contains(&room_type_filter)
-			};
-
-			include.then_some(r)
-		})
-		.collect()
-		.await
-}

src/api/client/sync/v5.rs

@@ -1,6 +1,6 @@
 use std::{
 	cmp::{self, Ordering},
-	collections::{BTreeMap, BTreeSet, HashMap, HashSet},
+	collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque},
 	ops::Deref,
 	time::Duration,
 };
@@ -31,6 +31,7 @@
 	events::{
 		AnyRawAccountDataEvent, AnySyncEphemeralRoomEvent, StateEventType, TimelineEventType,
 		room::member::{MembershipState, RoomMemberEventContent},
+		typing::TypingEventContent,
 	},
 	serde::Raw,
 	uint,
@@ -39,7 +40,9 @@
 use super::share_encrypted_room;
 use crate::{
 	Ruma,
-	client::{DEFAULT_BUMP_TYPES, ignored_filter, is_ignored_invite, sync::load_timeline},
+	client::{
+		DEFAULT_BUMP_TYPES, TimelinePdus, ignored_filter, is_ignored_invite, sync::load_timeline,
+	},
 };
 
 type SyncInfo<'a> = (&'a UserId, &'a DeviceId, u64, &'a sync_events::v5::Request);
@@ -210,6 +213,9 @@ pub(crate) async fn sync_events_v5_route(
 		_ = tokio::time::timeout(duration, watcher).await;
 	}
 
+	let typing = collect_typing_events(services, sender_user, &body, &todo_rooms).await?;
+	response.extensions.typing = typing;
+
 	trace!(
 		rooms = ?response.rooms.len(),
 		account_data = ?response.extensions.account_data.rooms.len(),
@@ -293,6 +299,8 @@ async fn handle_lists<'a, Rooms, AllRooms>(
 	Rooms: Iterator<Item = &'a RoomId> + Clone + Send + 'a,
 	AllRooms: Iterator<Item = &'a RoomId> + Clone + Send + 'a,
 {
+	// TODO MSC4186: Implement remaining list filters: is_dm, is_encrypted,
+	// room_types.
 	for (list_id, list) in &body.lists {
 		let active_rooms: Vec<_> = match list.filters.as_ref().and_then(|f| f.is_invite) {
 			| None => all_rooms.clone().collect(),
@@ -409,13 +417,13 @@ async fn process_rooms<'a, Rooms>(
 				.await
 				.ok();
 
-			(timeline_pdus, limited) = (Vec::new(), true);
+			(timeline_pdus, limited) = (VecDeque::new(), true);
 		} else {
-			(timeline_pdus, limited) = match load_timeline(
+			TimelinePdus { pdus: timeline_pdus, limited } = match load_timeline(
 				services,
 				sender_user,
 				room_id,
-				roomsincecount,
+				Some(roomsincecount),
 				Some(PduCount::from(next_batch)),
 				*timeline_limit,
 			)
@@ -434,7 +442,7 @@ async fn process_rooms<'a, Rooms>(
 				room_id.to_owned(),
 				services
 					.account_data
-					.changes_since(Some(room_id), sender_user, *roomsince, Some(next_batch))
+					.changes_since(Some(room_id), sender_user, Some(*roomsince), Some(next_batch))
 					.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
 					.collect()
 					.await,
@@ -460,11 +468,11 @@ async fn process_rooms<'a, Rooms>(
 		let mut receipts: Vec<Raw<AnySyncEphemeralRoomEvent>> = services
 			.rooms
 			.read_receipt
-			.readreceipts_since(room_id, *roomsince)
+			.readreceipts_since(room_id, Some(*roomsince))
 			.filter_map(|(read_user, _ts, v)| async move {
 				services
 					.users
-					.user_is_ignored(read_user, sender_user)
+					.user_is_ignored(&read_user, sender_user)
 					.await
 					.or_some(v)
 			})
@@ -499,7 +507,7 @@ async fn process_rooms<'a, Rooms>(
 		}
 
 		let prev_batch = timeline_pdus
-			.first()
+			.front()
 			.map_or(Ok::<_, Error>(None), |(pdu_count, _)| {
 				Ok(Some(match pdu_count {
 					| PduCount::Backfilled(_) => {
@@ -672,6 +680,62 @@ async fn process_rooms<'a, Rooms>(
 	}
 	Ok(rooms)
 }
+
+async fn collect_typing_events(
+	services: &Services,
+	sender_user: &UserId,
+	body: &sync_events::v5::Request,
+	todo_rooms: &TodoRooms,
+) -> Result<sync_events::v5::response::Typing> {
+	if !body.extensions.typing.enabled.unwrap_or(false) {
+		return Ok(sync_events::v5::response::Typing::default());
+	}
+	let rooms: Vec<_> = body.extensions.typing.rooms.clone().unwrap_or_else(|| {
+		body.room_subscriptions
+			.keys()
+			.map(ToOwned::to_owned)
+			.collect()
+	});
+	let lists: Vec<_> = body
+		.extensions
+		.typing
+		.lists
+		.clone()
+		.unwrap_or_else(|| body.lists.keys().map(ToOwned::to_owned).collect::<Vec<_>>());
+
+	if rooms.is_empty() && lists.is_empty() {
+		return Ok(sync_events::v5::response::Typing::default());
+	}
+
+	let mut typing_response = sync_events::v5::response::Typing::default();
+	for (room_id, (_, _, roomsince)) in todo_rooms {
+		if services.rooms.typing.last_typing_update(room_id).await? <= *roomsince {
+			continue;
+		}
+
+		match services
+			.rooms
+			.typing
+			.typing_users_for_user(room_id, sender_user)
+			.await
+		{
+			| Ok(typing_users) => {
+				typing_response.rooms.insert(
+					room_id.to_owned(), // Already OwnedRoomId
+					Raw::new(&sync_events::v5::response::SyncTypingEvent {
+						content: TypingEventContent::new(typing_users),
+					})?,
+				);
+			},
+			| Err(e) => {
+				warn!(%room_id, "Failed to get typing events for room: {}", e);
+			},
+		}
+	}
+
+	Ok(typing_response)
+}
+
 async fn collect_account_data(
 	services: &Services,
 	(sender_user, _, globalsince, body): (&UserId, &DeviceId, u64, &sync_events::v5::Request),
@@ -687,7 +751,7 @@ async fn collect_account_data(
 
 	account_data.global = services
 		.account_data
-		.changes_since(None, sender_user, globalsince, None)
+		.changes_since(None, sender_user, Some(globalsince), None)
 		.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
 		.collect()
 		.await;
@@ -698,7 +762,7 @@ async fn collect_account_data(
 				room.clone(),
 				services
 					.account_data
-					.changes_since(Some(room), sender_user, globalsince, None)
+					.changes_since(Some(room), sender_user, Some(globalsince), None)
 					.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
 					.collect()
 					.await,
@@ -732,7 +796,7 @@ async fn collect_e2ee<'a, Rooms>(
 	device_list_changes.extend(
 		services
 			.users
-			.keys_changed(sender_user, globalsince, None)
+			.keys_changed(sender_user, Some(globalsince), None)
 			.map(ToOwned::to_owned)
 			.collect::<Vec<_>>()
 			.await,
@@ -868,7 +932,7 @@ async fn collect_e2ee<'a, Rooms>(
 		device_list_changes.extend(
 			services
 				.users
-				.room_keys_changed(room_id, globalsince, None)
+				.room_keys_changed(room_id, Some(globalsince), None)
 				.map(|(user_id, _)| user_id)
 				.map(ToOwned::to_owned)
 				.collect::<Vec<_>>()

src/api/client/unversioned.rs

@@ -52,7 +52,6 @@ pub(crate) async fn get_supported_versions_route(
 			("org.matrix.msc3026.busy_presence".to_owned(), true), /* busy presence status (https://github.com/matrix-org/matrix-spec-proposals/pull/3026) */
 			("org.matrix.msc3827".to_owned(), true), /* filtering of /publicRooms by room type (https://github.com/matrix-org/matrix-spec-proposals/pull/3827) */
 			("org.matrix.msc3952_intentional_mentions".to_owned(), true), /* intentional mentions (https://github.com/matrix-org/matrix-spec-proposals/pull/3952) */
-			("org.matrix.msc3575".to_owned(), true), /* sliding sync (https://github.com/matrix-org/matrix-spec-proposals/pull/3575/files#r1588877046) */
 			("org.matrix.msc3916.stable".to_owned(), true), /* authenticated media (https://github.com/matrix-org/matrix-spec-proposals/pull/3916) */
 			("org.matrix.msc4180".to_owned(), true), /* stable flag for 3916 (https://github.com/matrix-org/matrix-spec-proposals/pull/4180) */
 			("uk.tcpip.msc4133".to_owned(), true), /* Extending User Profile API with Key:Value Pairs (https://github.com/matrix-org/matrix-spec-proposals/pull/4133) */

src/api/client/well_known.rs

@@ -78,7 +78,7 @@ pub(crate) async fn well_known_support(
 			while let Some(user_id) = stream.next().await {
 				// Skip server user
 				if *user_id == services.globals.server_user {
-					break;
+					continue;
 				}
 				contacts.push(Contact {
 					role: role_value.clone(),

src/api/router.rs

@@ -143,7 +143,6 @@ pub fn build(router: Router<State>, server: &Server) -> Router<State> {
 				.put(client::send_state_event_for_empty_key_route),
 		)
 		.ruma_route(&client::sync_events_route)
-		.ruma_route(&client::sync_events_v4_route)
 		.ruma_route(&client::sync_events_v5_route)
 		.ruma_route(&client::get_context_route)
 		.ruma_route(&client::get_message_events_route)
@@ -226,6 +225,7 @@ pub fn build(router: Router<State>, server: &Server) -> Router<State> {
 			.ruma_route(&server::well_known_server)
 			.ruma_route(&server::get_content_route)
 			.ruma_route(&server::get_content_thumbnail_route)
+			.ruma_route(&server::get_edutypes_route)
 			.route("/_conduwuit/local_user_count", get(client::conduwuit_local_user_count))
 			.route("/_continuwuity/local_user_count", get(client::conduwuit_local_user_count));
 	} else {

src/api/server/edutypes.rs

@@ -0,0 +1,19 @@
+use axum::extract::State;
+use conduwuit::Result;
+use ruma::api::federation::edutypes::get_edutypes;
+
+use crate::Ruma;
+
+/// # `GET /_matrix/federation/v1/edutypes`
+///
+/// Lists EDU types we wish to receive
+pub(crate) async fn get_edutypes_route(
+	State(services): State<crate::State>,
+	_body: Ruma<get_edutypes::unstable::Request>,
+) -> Result<get_edutypes::unstable::Response> {
+	Ok(get_edutypes::unstable::Response {
+		typing: services.config.allow_incoming_typing,
+		presence: services.config.allow_incoming_presence,
+		receipt: services.config.allow_incoming_read_receipts,
+	})
+}

src/api/server/invite.rs

@@ -10,7 +10,6 @@
 use ruma::{
 	CanonicalJsonValue, OwnedUserId, UserId,
 	api::{client::error::ErrorKind, federation::membership::create_invite},
-	events::room::member::{MembershipState, RoomMemberEventContent},
 	serde::JsonObject,
 };
 
@@ -133,17 +132,21 @@ pub(crate) async fn create_invite_route(
 		services
 			.rooms
 			.state_cache
-			.update_membership(
-				&body.room_id,
+			.mark_as_invited(
 				&recipient_user,
-				RoomMemberEventContent::new(MembershipState::Invite),
+				&body.room_id,
 				sender_user,
 				Some(invite_state),
 				body.via.clone(),
-				true,
 			)
 			.await?;
 
+		services
+			.rooms
+			.state_cache
+			.update_joined_count(&body.room_id)
+			.await;
+
 		for appservice in services.appservice.read().await.values() {
 			if appservice.is_user_match(&recipient_user) {
 				services

src/api/server/make_join.rs

@@ -1,6 +1,6 @@
 use axum::extract::State;
 use conduwuit::{
-	Err, Error, Result, debug_info, matrix::pdu::PduBuilder, utils::IterStream, warn,
+	Err, Error, Result, debug_info, info, matrix::pdu::PduBuilder, utils::IterStream, warn,
 };
 use conduwuit_service::Services;
 use futures::StreamExt;
@@ -22,6 +22,7 @@
 /// # `GET /_matrix/federation/v1/make_join/{roomId}/{userId}`
 ///
 /// Creates a join template.
+#[tracing::instrument(skip_all, fields(room_id = %body.room_id, user_id = %body.user_id, origin = %body.origin()))]
 pub(crate) async fn create_join_event_template_route(
 	State(services): State<crate::State>,
 	body: Ruma<prepare_join_event::v1::Request>,
@@ -72,11 +73,16 @@ pub(crate) async fn create_join_event_template_route(
 	}
 
 	let state_lock = services.rooms.state.mutex.lock(&body.room_id).await;
-
+	let is_invited = services
+		.rooms
+		.state_cache
+		.is_invited(&body.user_id, &body.room_id)
+		.await;
 	let join_authorized_via_users_server: Option<OwnedUserId> = {
 		use RoomVersionId::*;
-		if matches!(room_version_id, V1 | V2 | V3 | V4 | V5 | V6 | V7) {
-			// room version does not support restricted join rules
+		if matches!(room_version_id, V1 | V2 | V3 | V4 | V5 | V6 | V7) || is_invited {
+			// room version does not support restricted join rules, or the user is currently
+			// already invited
 			None
 		} else if user_can_perform_restricted_join(
 			&services,
@@ -103,6 +109,10 @@ pub(crate) async fn create_join_event_template_route(
 				.await
 				.map(ToOwned::to_owned)
 			else {
+				info!(
+					"No local user is able to authorize the join of {} into {}",
+					&body.user_id, &body.room_id
+				);
 				return Err!(Request(UnableToGrantJoin(
 					"No user on this server is able to assist in joining."
 				)));
@@ -167,6 +177,7 @@ pub(crate) async fn user_can_perform_restricted_join(
 		)
 		.await
 	else {
+		// No join rules means there's nothing to authorise (defaults to invite)
 		return Ok(false);
 	};
 

src/api/server/mod.rs

@@ -1,4 +1,5 @@
 pub(super) mod backfill;
+pub(super) mod edutypes;
 pub(super) mod event;
 pub(super) mod event_auth;
 pub(super) mod get_missing_events;
@@ -23,6 +24,7 @@
 pub(super) mod well_known;
 
 pub(super) use backfill::*;
+pub(super) use edutypes::*;
 pub(super) use event::*;
 pub(super) use event_auth::*;
 pub(super) use get_missing_events::*;

src/core/Cargo.toml

@@ -92,7 +92,7 @@ ruma.workspace = true
 sanitize-filename.workspace = true
 serde_json.workspace = true
 serde_regex.workspace = true
-serde_yml.workspace = true
+serde-saphyr.workspace = true
 serde.workspace = true
 smallvec.workspace = true
 smallstr.workspace = true

src/core/config/mod.rs

@@ -1128,6 +1128,23 @@ pub struct Config {
 	#[serde(default = "true_fn")]
 	pub rocksdb_bottommost_compression: bool,
 
+	/// Compression algorithm for RocksDB's Write-Ahead-Log (WAL).
+	///
+	/// At present, only ZSTD compression is supported by RocksDB for WAL
+	/// compression. Enabling this can reduce WAL size at the expense of some
+	/// CPU usage during writes.
+	///
+	/// The options are:
+	/// - "none" = No compression
+	/// - "zstd" = ZSTD compression
+	///
+	/// For more information on WAL compression, see:
+	/// https://github.com/facebook/rocksdb/wiki/WAL-Compression
+	///
+	/// default: "zstd"
+	#[serde(default = "default_rocksdb_wal_compression")]
+	pub rocksdb_wal_compression: String,
+
 	/// Database recovery mode (for RocksDB WAL corruption).
 	///
 	/// Use this option when the server reports corruption and refuses to start.
@@ -1710,6 +1727,19 @@ pub struct Config {
 	#[serde(default)]
 	pub block_non_admin_invites: bool,
 
+	/// Enable or disable making requests to MSC4284 Policy Servers.
+	/// It is recommended you keep this enabled unless you experience frequent
+	/// connectivity issues, such as in a restricted networking environment.
+	#[serde(default = "true_fn")]
+	pub enable_msc4284_policy_servers: bool,
+
+	/// Enable running locally generated events through configured MSC4284
+	/// policy servers. You may wish to disable this if your server is
+	/// single-user for a slight speed benefit in some rooms, but otherwise
+	/// should leave it enabled.
+	#[serde(default = "true_fn")]
+	pub policy_server_check_own_events: bool,
+
 	/// Allow admins to enter commands in rooms other than "#admins" (admin
 	/// room) by prefixing your message with "\!admin" or "\\!admin" followed up
 	/// a normal continuwuity admin command. The reply will be publicly visible
@@ -2441,6 +2471,8 @@ fn default_rocksdb_compression_algo() -> String {
 		.to_owned()
 }
 
+fn default_rocksdb_wal_compression() -> String { "zstd".to_owned() }
+
 /// Default RocksDB compression level is 32767, which is internally read by
 /// RocksDB as the default magic number and translated to the library's default
 /// compression level as they all differ. See their `kDefaultCompressionLevel`.

src/core/error/mod.rs

@@ -83,7 +83,9 @@ pub enum Error {
 	#[error(transparent)]
 	TypedHeader(#[from] axum_extra::typed_header::TypedHeaderRejection),
 	#[error(transparent)]
-	Yaml(#[from] serde_yml::Error),
+	YamlDe(#[from] serde_saphyr::Error),
+	#[error(transparent)]
+	YamlSer(#[from] serde_saphyr::ser_error::Error),
 
 	// ruma/conduwuit
 	#[error("Arithmetic operation failed: {0}")]

src/core/info/rustc.rs

@@ -5,13 +5,17 @@
 
 use std::{collections::BTreeMap, sync::OnceLock};
 
-use crate::{SyncMutex, utils::exchange};
+use crate::utils::exchange;
 
 /// Raw capture of rustc flags used to build each crate in the project. Informed
 /// by rustc_flags_capture macro (one in each crate's mod.rs). This is
 /// done during static initialization which is why it's mutex-protected and pub.
 /// Should not be written to by anything other than our macro.
-pub static FLAGS: SyncMutex<BTreeMap<&str, &[&str]>> = SyncMutex::new(BTreeMap::new());
+///
+/// We specifically use a std mutex here because parking_lot cannot be used
+/// after thread local storage is destroyed on MacOS.
+pub static FLAGS: std::sync::Mutex<BTreeMap<&str, &[&str]>> =
+	std::sync::Mutex::new(BTreeMap::new());
 
 /// Processed list of enabled features across all project crates. This is
 /// generated from the data in FLAGS.
@@ -24,6 +28,7 @@ fn init_features() -> Vec<&'static str> {
 	let mut features = Vec::new();
 	FLAGS
 		.lock()
+		.expect("locked")
 		.iter()
 		.for_each(|(_, flags)| append_features(&mut features, flags));
 

src/core/matrix/state_res/event_auth.rs

@@ -615,15 +615,21 @@ pub async fn auth_check<E, F, Fut>(
 	Ok(true)
 }
 
-fn is_creator<EV>(v: &RoomVersion, c: &BTreeSet<OwnedUserId>, ce: &EV, user_id: &UserId) -> bool
+fn is_creator<EV>(
+	v: &RoomVersion,
+	c: &BTreeSet<OwnedUserId>,
+	ce: &EV,
+	user_id: &UserId,
+	have_pls: bool,
+) -> bool
 where
 	EV: Event + Send + Sync,
 {
 	if v.explicitly_privilege_room_creators {
 		c.contains(user_id)
-	} else if v.use_room_create_sender {
+	} else if v.use_room_create_sender && !have_pls {
 		ce.sender() == user_id
-	} else {
+	} else if !have_pls {
 		#[allow(deprecated)]
 		let creator = from_json_str::<RoomCreateEventContent>(ce.content().get())
 			.unwrap()
@@ -632,6 +638,8 @@ fn is_creator<EV>(v: &RoomVersion, c: &BTreeSet<OwnedUserId>, ce: &EV, user_id:
 			.unwrap();
 
 		creator == user_id
+	} else {
+		false
 	}
 }
 
@@ -724,10 +732,11 @@ struct GetThirdPartyInvite {
 	}
 	trace!(?creators, "creators for room");
 
-	let mut join_rules = JoinRule::Invite;
-	if let Some(jr) = &join_rules_event {
-		join_rules = from_json_str::<RoomJoinRulesEventContent>(jr.content().get())?.join_rule;
-	}
+	let join_rules = if let Some(jr) = &join_rules_event {
+		from_json_str::<RoomJoinRulesEventContent>(jr.content().get())?.join_rule
+	} else {
+		JoinRule::Invite
+	};
 
 	let power_levels_event_id = power_levels_event.as_ref().map(Event::event_id);
 	let sender_membership_event_id = sender_membership_event.as_ref().map(Event::event_id);
@@ -753,8 +762,13 @@ struct GetThirdPartyInvite {
 			(int!(0), int!(0))
 		};
 		let user_joined = user_for_join_auth_membership == &MembershipState::Join;
-		let okay_power = is_creator(room_version, &creators, create_room, user_for_join_auth)
-			|| auth_user_pl >= invite_level;
+		let okay_power = is_creator(
+			room_version,
+			&creators,
+			create_room,
+			user_for_join_auth,
+			power_levels_event.as_ref().is_some(),
+		) || auth_user_pl >= invite_level;
 		trace!(
 			auth_user_pl=?auth_user_pl,
 			invite_level=?invite_level,
@@ -769,8 +783,20 @@ struct GetThirdPartyInvite {
 		trace!("No auth user given for join auth");
 		false
 	};
-	let sender_creator = is_creator(room_version, &creators, create_room, sender);
-	let target_creator = is_creator(room_version, &creators, create_room, target_user);
+	let sender_creator = is_creator(
+		room_version,
+		&creators,
+		create_room,
+		sender,
+		power_levels_event.as_ref().is_some(),
+	);
+	let target_creator = is_creator(
+		room_version,
+		&creators,
+		create_room,
+		target_user,
+		power_levels_event.as_ref().is_some(),
+	);
 
 	Ok(match target_membership {
 		| MembershipState::Join => {
@@ -882,7 +908,7 @@ struct GetThirdPartyInvite {
 							false
 						}
 					},
-					| JoinRule::Restricted(_) =>
+					| JoinRule::Restricted(_) => {
 						if membership_allows_join || user_for_join_auth_is_valid {
 							trace!(
 								%sender,
@@ -902,7 +928,8 @@ struct GetThirdPartyInvite {
 								 valid authorising user given to permit the join"
 							);
 							false
-						},
+						}
+					},
 					| JoinRule::Public => {
 						trace!(%sender, "join rule is public, allowing join");
 						true
@@ -985,7 +1012,7 @@ struct GetThirdPartyInvite {
 		},
 		| MembershipState::Leave => {
 			let can_unban = if target_user_current_membership == MembershipState::Ban {
-				sender_creator || sender_power.filter(|&p| p < &power_levels.ban).is_some()
+				sender_creator || sender_power.filter(|&p| p >= &power_levels.ban).is_some()
 			} else {
 				true
 			};
@@ -993,7 +1020,24 @@ struct GetThirdPartyInvite {
 				target_user_current_membership,
 				MembershipState::Ban | MembershipState::Leave
 			) {
-				sender_creator || sender_power.filter(|&p| p < &power_levels.kick).is_some()
+				if sender_creator {
+					// sender is a creator
+					true
+				} else if sender_power.filter(|&p| p >= &power_levels.kick).is_none() {
+					// sender lacks kick power level
+					false
+				} else if let Some(sp) = sender_power {
+					if let Some(tp) = target_power {
+						// sender must have more power than target
+						sp > tp
+					} else {
+						// target has default power level
+						true
+					}
+				} else {
+					// sender has default power level
+					false
+				}
 			} else {
 				true
 			};
@@ -1023,7 +1067,7 @@ struct GetThirdPartyInvite {
 					"sender cannot kick another user as they are not joined to the room",
 				);
 				false
-			} else if !can_unban {
+			} else if !(can_unban && can_kick) {
 				// If the target is banned, only a room creator or someone with ban power
 				// level can unban them
 				warn!(

src/core/matrix/state_res/mod.rs

@@ -101,40 +101,40 @@ pub async fn resolve<'a, Pdu, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Ex
 	debug!(version = ?stateres_version, "State resolution starting");
 
 	// Split non-conflicting and conflicting state
-	let (clean, conflicting) = separate(state_sets.into_iter());
+	let (unconflicted, conflicting) = separate(state_sets.into_iter());
 
-	debug!(count = clean.len(), "non-conflicting events");
-	trace!(map = ?clean, "non-conflicting events");
+	debug!(count = unconflicted.len(), "non-conflicting events");
+	trace!(map = ?unconflicted, "non-conflicting events");
 
 	if conflicting.is_empty() {
 		debug!("no conflicting state found");
-		return Ok(clean);
+		return Ok(unconflicted);
 	}
 
 	debug!(count = conflicting.len(), "conflicting events");
 	trace!(map = ?conflicting, "conflicting events");
-	let conflicted_state_subgraph: HashSet<_> = match stateres_version {
-		| StateResolutionVersion::V2_1 =>
-			calculate_conflicted_subgraph(&conflicting, event_fetch)
+	let (conflicted_state_subgraph, initial_state) =
+		if stateres_version == StateResolutionVersion::V2_1 {
+			let csg = calculate_conflicted_subgraph(&conflicting, event_fetch)
 				.await
 				.ok_or_else(|| {
 					Error::InvalidPdu("Failed to calculate conflicted subgraph".to_owned())
-				})?,
-		| _ => HashSet::new(),
-	};
-	debug!(count = conflicted_state_subgraph.len(), "conflicted subgraph");
-	trace!(set = ?conflicted_state_subgraph, "conflicted subgraph");
-
-	let conflicting_values = conflicting.into_values().flatten().stream();
+				})?;
+			debug!(count = csg.len(), "conflicted subgraph");
+			trace!(set = ?csg, "conflicted subgraph");
+			(csg, HashMap::new())
+		} else {
+			(HashSet::new(), unconflicted.clone())
+		};
 
 	// `all_conflicted` contains unique items
 	// synapse says `full_set = {eid for eid in full_conflicted_set if eid in
 	// event_map}`
 	// Hydra: Also consider the conflicted state subgraph
 	let all_conflicted: HashSet<_> = get_auth_chain_diff(auth_chain_sets)
-		.chain(conflicting_values)
-		.chain(conflicted_state_subgraph.into_iter().stream())
+		.chain(conflicting.into_values().flatten().stream())
 		.broad_filter_map(async |id| event_exists(id.clone()).await.then_some(id))
+		.chain(conflicted_state_subgraph.into_iter().stream())
 		.collect()
 		.await;
 
@@ -171,7 +171,7 @@ pub async fn resolve<'a, Pdu, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Ex
 		&room_version,
 		&stateres_version,
 		sorted_control_levels.iter().stream().map(AsRef::as_ref),
-		clean.clone(),
+		initial_state,
 		&event_fetch,
 	)
 	.await?;
@@ -201,7 +201,7 @@ pub async fn resolve<'a, Pdu, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Ex
 	let power_levels_ty_sk = (StateEventType::RoomPowerLevels, StateKey::new());
 	let power_event = resolved_control.get(&power_levels_ty_sk);
 
-	debug!(event_id = ?power_event, "power event");
+	trace!(event_id = ?power_event, "power event");
 
 	let sorted_left_events =
 		mainline_sort(&events_to_resolve, power_event.cloned(), &event_fetch).await?;
@@ -212,19 +212,13 @@ pub async fn resolve<'a, Pdu, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Ex
 		&room_version,
 		&stateres_version,
 		sorted_left_events.iter().stream().map(AsRef::as_ref),
-		resolved_control.clone(), // The control events are added to the final resolved state
+		resolved_control, // The control events are added to the final resolved state
 		&event_fetch,
 	)
 	.await?;
 
-	// Add unconflicted state to the resolved state
-	// We priorities the unconflicting state
-	resolved_state.extend(clean);
-	if stateres_version == StateResolutionVersion::V2_1 {
-		resolved_state.extend(resolved_control);
-		// TODO(hydra): this feels disgusting and wrong but it allows
-		// the state to resolve properly?
-	}
+	// Ensure unconflicting state is in the final state
+	resolved_state.extend(unconflicted);
 
 	debug!("state resolution finished");
 	trace!( map = ?resolved_state, "final resolved state" );
@@ -427,8 +421,8 @@ async fn reverse_topological_power_sort<E, F, Fut>(
 /// `key_fn` is used as to obtain the power level and age of an event for
 /// breaking ties (together with the event ID).
 #[tracing::instrument(level = "debug", skip_all)]
-pub async fn lexicographical_topological_sort<Id, F, Fut, Hasher>(
-	graph: &HashMap<Id, HashSet<Id, Hasher>>,
+pub async fn lexicographical_topological_sort<Id, F, Fut, Hasher, S>(
+	graph: &HashMap<Id, HashSet<Id, Hasher>, S>,
 	key_fn: &F,
 ) -> Result<Vec<Id>>
 where
@@ -436,6 +430,7 @@ pub async fn lexicographical_topological_sort<Id, F, Fut, Hasher>(
 	Fut: Future<Output = Result<(Int, MilliSecondsSinceUnixEpoch)>> + Send,
 	Id: Borrow<EventId> + Clone + Eq + Hash + Ord + Send + Sync,
 	Hasher: BuildHasher + Default + Clone + Send + Sync,
+	S: BuildHasher + Clone + Send + Sync,
 {
 	#[derive(PartialEq, Eq)]
 	struct TieBreaker<'a, Id> {

src/core/utils/future/try_ext_ext.rs

@@ -36,7 +36,7 @@ fn map_ok_or<U, F>(
 	) -> MapOkOrElse<Self, impl FnOnce(Self::Ok) -> U, impl FnOnce(Self::Error) -> U>
 	where
 		F: FnOnce(Self::Ok) -> U,
-		Self: Send + Sized;
+		Self: Sized;
 
 	fn ok(
 		self,
@@ -100,7 +100,7 @@ fn map_ok_or<U, F>(
 	) -> MapOkOrElse<Self, impl FnOnce(Self::Ok) -> U, impl FnOnce(Self::Error) -> U>
 	where
 		F: FnOnce(Self::Ok) -> U,
-		Self: Send + Sized,
+		Self: Sized,
 	{
 		self.map_ok_or_else(|_| default, f)
 	}

src/core/utils/tests.rs

@@ -276,3 +276,22 @@ async fn set_intersection_sorted_stream2() {
 		.await;
 	assert!(r.eq(&["ccc", "ggg", "iii"]));
 }
+
+#[test]
+fn is_within_bounds() {
+	use std::time::{Duration, SystemTime};
+
+	use utils::time::{TimeDirection, is_within_bounds};
+
+	let now = SystemTime::now();
+	let yesterday = now - Duration::from_secs(86400);
+	assert!(is_within_bounds(yesterday, now, TimeDirection::Before));
+	assert!(!is_within_bounds(yesterday, now, TimeDirection::After));
+
+	let tomorrow = now + Duration::from_secs(86400);
+	assert!(is_within_bounds(tomorrow, now, TimeDirection::After));
+	assert!(!is_within_bounds(tomorrow, now, TimeDirection::Before));
+
+	assert!(is_within_bounds(now, now, TimeDirection::Before));
+	assert!(is_within_bounds(now, now, TimeDirection::After));
+}

src/core/utils/time.rs

@@ -126,3 +126,24 @@ pub enum Unit {
 	Micros(u128),
 	Nanos(u128),
 }
+
+#[derive(Eq, PartialEq, Clone, Copy, Debug)]
+pub enum TimeDirection {
+	Before,
+	After,
+}
+
+/// Checks if `item_time` is before or after `time_boundary`.
+/// If both times are the same, it will return true for both directions, as the
+/// matching is inclusive.
+#[must_use]
+pub fn is_within_bounds(
+	item_time: SystemTime,
+	time_boundary: SystemTime,
+	direction: TimeDirection,
+) -> bool {
+	match direction {
+		| TimeDirection::Before => item_time <= time_boundary,
+		| TimeDirection::After => item_time >= time_boundary,
+	}
+}

src/database/engine/db_opts.rs

@@ -1,7 +1,9 @@
 use std::{cmp, convert::TryFrom};
 
-use conduwuit::{Config, Result, utils};
-use rocksdb::{Cache, DBRecoveryMode, Env, LogLevel, Options, statistics::StatsLevel};
+use conduwuit::{Config, Result, utils, warn};
+use rocksdb::{
+	Cache, DBCompressionType, DBRecoveryMode, Env, LogLevel, Options, statistics::StatsLevel,
+};
 
 use super::{cf_opts::cache_size_f64, logger::handle as handle_log};
 
@@ -58,6 +60,20 @@ pub(crate) fn db_options(config: &Config, env: &Env, row_cache: &Cache) -> Resul
 	opts.set_max_total_wal_size(1024 * 1024 * 512);
 	opts.set_writable_file_max_buffer_size(1024 * 1024 * 2);
 
+	// WAL compression
+	let wal_compression = match config.rocksdb_wal_compression.as_ref() {
+		| "zstd" => DBCompressionType::Zstd,
+		| "none" => DBCompressionType::None,
+		| value => {
+			warn!(
+				"Invalid rocksdb_wal_compression value '{value}'. Supported values are 'none' \
+				 or 'zstd'. Defaulting to 'none'."
+			);
+			DBCompressionType::None
+		},
+	};
+	opts.set_wal_compression_type(wal_compression);
+
 	// Misc
 	opts.set_disable_auto_compactions(!config.rocksdb_compaction);
 	opts.create_missing_column_families(true);

src/database/map/get_batch.rs

@@ -10,7 +10,7 @@
 use futures::{Stream, StreamExt, TryStreamExt};
 use rocksdb::{DBPinnableSlice, ReadOptions};
 
-use super::get::{cached_handle_from, handle_from};
+use super::get::handle_from;
 use crate::Handle;
 
 pub trait Get<'a, K, S>
@@ -58,20 +58,6 @@ pub(crate) fn get_batch<'a, S, K>(
 		.try_flatten()
 }
 
-#[implement(super::Map)]
-#[tracing::instrument(name = "batch_cached", level = "trace", skip_all)]
-pub(crate) fn get_batch_cached<'a, I, K>(
-	&self,
-	keys: I,
-) -> impl Iterator<Item = Result<Option<Handle<'_>>>> + Send + use<'_, I, K>
-where
-	I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
-	K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
-{
-	self.get_batch_blocking_opts(keys, &self.cache_read_options)
-		.map(cached_handle_from)
-}
-
 #[implement(super::Map)]
 #[tracing::instrument(name = "batch_blocking", level = "trace", skip_all)]
 pub(crate) fn get_batch_blocking<'a, I, K>(

src/database/pool.rs

@@ -184,7 +184,7 @@ fn spawn_one(
 	let handle = thread::Builder::new()
 		.name(WORKER_NAME.into())
 		.stack_size(WORKER_STACK_SIZE)
-		.spawn(move || self.worker(id, recv))?;
+		.spawn(move || self.worker(id, &recv))?;
 
 	workers.push(handle);
 
@@ -260,9 +260,9 @@ async fn execute(&self, queue: &Sender<Cmd>, cmd: Cmd) -> Result {
 		tid = ?thread::current().id(),
 	),
 )]
-fn worker(self: Arc<Self>, id: usize, recv: Receiver<Cmd>) {
+fn worker(self: Arc<Self>, id: usize, recv: &Receiver<Cmd>) {
 	self.worker_init(id);
-	self.worker_loop(&recv);
+	self.worker_loop(recv);
 }
 
 #[implement(Pool)]
@@ -309,7 +309,7 @@ fn worker_loop(self: &Arc<Self>, recv: &Receiver<Cmd>) {
 	self.busy.fetch_add(1, Ordering::Relaxed);
 
 	while let Ok(cmd) = self.worker_wait(recv) {
-		self.worker_handle(cmd);
+		Pool::worker_handle(cmd);
 	}
 }
 
@@ -331,11 +331,11 @@ fn worker_wait(self: &Arc<Self>, recv: &Receiver<Cmd>) -> Result<Cmd, RecvError>
 }
 
 #[implement(Pool)]
-fn worker_handle(self: &Arc<Self>, cmd: Cmd) {
+fn worker_handle(cmd: Cmd) {
 	match cmd {
-		| Cmd::Get(cmd) if cmd.key.len() == 1 => self.handle_get(cmd),
-		| Cmd::Get(cmd) => self.handle_batch(cmd),
-		| Cmd::Iter(cmd) => self.handle_iter(cmd),
+		| Cmd::Get(cmd) if cmd.key.len() == 1 => Pool::handle_get(cmd),
+		| Cmd::Get(cmd) => Pool::handle_batch(cmd),
+		| Cmd::Iter(cmd) => Pool::handle_iter(cmd),
 	}
 }
 
@@ -346,7 +346,7 @@ fn worker_handle(self: &Arc<Self>, cmd: Cmd) {
 	skip_all,
 	fields(%cmd.map),
 )]
-fn handle_iter(&self, mut cmd: Seek) {
+fn handle_iter(mut cmd: Seek) {
 	let chan = cmd.res.take().expect("missing result channel");
 
 	if chan.is_canceled() {
@@ -375,7 +375,7 @@ fn handle_iter(&self, mut cmd: Seek) {
 		keys = %cmd.key.len(),
 	),
 )]
-fn handle_batch(self: &Arc<Self>, mut cmd: Get) {
+fn handle_batch(mut cmd: Get) {
 	debug_assert!(cmd.key.len() > 1, "should have more than one key");
 	debug_assert!(!cmd.key.iter().any(SmallVec::is_empty), "querying for empty key");
 
@@ -401,7 +401,7 @@ fn handle_batch(self: &Arc<Self>, mut cmd: Get) {
 	skip_all,
 	fields(%cmd.map),
 )]
-fn handle_get(&self, mut cmd: Get) {
+fn handle_get(mut cmd: Get) {
 	debug_assert!(!cmd.key[0].is_empty(), "querying for empty key");
 
 	// Obtain the result channel.

src/macros/refutable.rs

@@ -25,13 +25,13 @@ pub(super) fn refutable(mut item: ItemFn, _args: &[Meta]) -> Result<TokenStream>
 		};
 
 		let name = format!("_args_{i}");
-		*pat = Box::new(Pat::Ident(PatIdent {
+		**pat = Pat::Ident(PatIdent {
 			ident: Ident::new(&name, Span::call_site().into()),
 			attrs: Vec::new(),
 			by_ref: None,
 			mutability: None,
 			subpat: None,
-		}));
+		});
 
 		let field = fields.iter();
 		let refute = quote! {

src/macros/rustc.rs

@@ -15,13 +15,13 @@ pub(super) fn flags_capture(args: TokenStream) -> TokenStream {
 
 		#[ctor]
 		fn _set_rustc_flags() {
-			conduwuit_core::info::rustc::FLAGS.lock().insert(#crate_name, &RUSTC_FLAGS);
+			conduwuit_core::info::rustc::FLAGS.lock().expect("locked").insert(#crate_name, &RUSTC_FLAGS);
 		}
 
 		// static strings have to be yanked on module unload
 		#[dtor]
 		fn _unset_rustc_flags() {
-			conduwuit_core::info::rustc::FLAGS.lock().remove(#crate_name);
+			conduwuit_core::info::rustc::FLAGS.lock().expect("locked").remove(#crate_name);
 		}
 	};
 

src/main/Cargo.toml

@@ -62,7 +62,8 @@ standard = [
 	"media_thumbnail",
 	"systemd",
 	"url_preview",
-	"zstd_compression"
+	"zstd_compression",
+    "sentry_telemetry"
 ]
 full = [
     "standard",
@@ -129,7 +130,6 @@ perf_measurements = [
 	"dep:tracing-opentelemetry",
 	"dep:opentelemetry_sdk",
 	"dep:opentelemetry-otlp",
-	"dep:opentelemetry-jaeger-propagator",
 	"conduwuit-core/perf_measurements",
 	"conduwuit-core/sentry_telemetry",
 ]
@@ -201,6 +201,7 @@ conduwuit-core.workspace = true
 conduwuit-database.workspace = true
 conduwuit-router.workspace = true
 conduwuit-service.workspace = true
+conduwuit-build-metadata.workspace = true
 
 clap.workspace = true
 console-subscriber.optional = true
@@ -212,8 +213,6 @@ opentelemetry.optional = true
 opentelemetry.workspace = true
 opentelemetry-otlp.optional = true
 opentelemetry-otlp.workspace = true
-opentelemetry-jaeger-propagator.optional = true
-opentelemetry-jaeger-propagator.workspace = true
 opentelemetry_sdk.optional = true
 opentelemetry_sdk.workspace = true
 sentry-tower.optional = true

src/main/logging.rs

@@ -94,7 +94,7 @@ pub(crate) fn init(
 
 		let otlp_layer = config.allow_otlp.then(|| {
 			opentelemetry::global::set_text_map_propagator(
-				opentelemetry_jaeger_propagator::Propagator::new(),
+				opentelemetry_sdk::propagation::TraceContextPropagator::new(),
 			);
 
 			let exporter = opentelemetry_otlp::SpanExporter::builder()

src/main/sentry.rs

@@ -1,10 +1,12 @@
 #![cfg(feature = "sentry_telemetry")]
 
 use std::{
+	borrow::Cow,
 	str::FromStr,
 	sync::{Arc, OnceLock},
 };
 
+use conduwuit_build_metadata as build;
 use conduwuit_core::{config::Config, debug, trace};
 use sentry::{
 	Breadcrumb, ClientOptions, Level,
@@ -44,7 +46,7 @@ fn options(config: &Config) -> ClientOptions {
 		server_name,
 		traces_sample_rate: config.sentry_traces_sample_rate,
 		debug: cfg!(debug_assertions),
-		release: sentry::release_name!(),
+		release: release_name(),
 		user_agent: conduwuit_core::version::user_agent().into(),
 		attach_stacktrace: config.sentry_attach_stacktrace,
 		before_send: Some(Arc::new(before_send)),
@@ -91,3 +93,21 @@ fn before_breadcrumb(crumb: Breadcrumb) -> Option<Breadcrumb> {
 	trace!("Sentry breadcrumb: {crumb:?}");
 	Some(crumb)
 }
+
+fn release_name() -> Option<Cow<'static, str>> {
+	static RELEASE: OnceLock<Option<String>> = OnceLock::new();
+
+	RELEASE
+		.get_or_init(|| {
+			let pkg_name = env!("CARGO_PKG_NAME");
+			let pkg_version = env!("CARGO_PKG_VERSION");
+
+			if let Some(commit_short) = build::GIT_COMMIT_HASH_SHORT {
+				Some(format!("{pkg_name}@{pkg_version}+{commit_short}"))
+			} else {
+				Some(format!("{pkg_name}@{pkg_version}"))
+			}
+		})
+		.as_ref()
+		.map(|s| Cow::Borrowed(s.as_str()))
+}

src/service/Cargo.toml

@@ -108,7 +108,7 @@ rustyline-async.workspace = true
 rustyline-async.optional = true
 serde_json.workspace = true
 serde.workspace = true
-serde_yml.workspace = true
+serde-saphyr.workspace = true
 sha2.workspace = true
 termimad.workspace = true
 termimad.optional = true

src/service/account_data/mod.rs

@@ -129,13 +129,14 @@ pub fn changes_since<'a>(
 	&'a self,
 	room_id: Option<&'a RoomId>,
 	user_id: &'a UserId,
-	since: u64,
+	since: Option<u64>,
 	to: Option<u64>,
 ) -> impl Stream<Item = AnyRawAccountDataEvent> + Send + 'a {
 	type Key<'a> = (Option<&'a RoomId>, &'a UserId, u64, Ignore);
 
 	// Skip the data that's exactly at since, because we sent that last time
-	let first_possible = (room_id, user_id, since.saturating_add(1));
+	// ...unless this is an initial sync, in which case send everything
+	let first_possible = (room_id, user_id, since.map_or(0, |since| since.saturating_add(1)));
 
 	self.db
 		.roomuserdataid_accountdata

src/service/appservice/mod.rs

@@ -271,7 +271,7 @@ pub async fn get_db_registration(&self, id: &str) -> Result<Registration> {
 			.id_appserviceregistrations
 			.get(id)
 			.await
-			.and_then(|ref bytes| serde_yml::from_slice(bytes).map_err(Into::into))
+			.and_then(|ref bytes| serde_saphyr::from_slice(bytes).map_err(Into::into))
 			.map_err(|e| err!(Database("Invalid appservice {id:?} registration: {e:?}")))
 	}
 

src/service/media/mod.rs

@@ -11,7 +11,10 @@
 use base64::{Engine as _, engine::general_purpose};
 use conduwuit::{
 	Err, Result, Server, debug, debug_error, debug_info, debug_warn, err, error, trace,
-	utils::{self, MutexMap},
+	utils::{
+		self, MutexMap,
+		time::{self, TimeDirection},
+	},
 	warn,
 };
 use ruma::{Mxc, OwnedMxcUri, UserId, http_headers::ContentDisposition};
@@ -226,13 +229,12 @@ pub async fn get_all_mxcs(&self) -> Result<Vec<OwnedMxcUri>> {
 		Ok(mxcs)
 	}
 
-	/// Deletes all remote only media files in the given at or after
-	/// time/duration. Returns a usize with the amount of media files deleted.
-	pub async fn delete_all_remote_media_at_after_time(
+	/// Deletes all media files in the given time frame.
+	/// Returns a usize with the amount of media files deleted.
+	pub async fn delete_all_media_within_timeframe(
 		&self,
-		time: SystemTime,
-		before: bool,
-		after: bool,
+		time_boundary: SystemTime,
+		direction: TimeDirection,
 		yes_i_want_to_delete_local_media: bool,
 	) -> Result<usize> {
 		let all_keys = self.db.get_all_media_keys().await;
@@ -299,18 +301,14 @@ pub async fn delete_all_remote_media_at_after_time(
 
 			debug!("File created at: {file_created_at:?}");
 
-			if file_created_at >= time && before {
+			if time::is_within_bounds(file_created_at, time_boundary, direction) {
 				debug!(
-					"File is within (before) user duration, pushing to list of file paths and \
-					 keys to delete."
-				);
-				remote_mxcs.push(mxc.to_string());
-			} else if file_created_at <= time && after {
-				debug!(
-					"File is not within (after) user duration, pushing to list of file paths \
-					 and keys to delete."
+					"File is within bounds ({direction:?} {time_boundary:?}), pushing to list \
+					 of file paths and keys to delete.",
 				);
 				remote_mxcs.push(mxc.to_string());
+			} else {
+				debug!("File is outside bounds ({direction:?} {time_boundary:?}), ignoring.");
 			}
 		}
 
@@ -318,7 +316,7 @@ pub async fn delete_all_remote_media_at_after_time(
 			return Err!(Database("Did not found any eligible MXCs to delete."));
 		}
 
-		debug_info!("Deleting media now in the past {time:?}");
+		debug_info!("Deleting media now {direction:?} {time_boundary:?}");
 
 		let mut deletion_count: usize = 0;
 

src/service/migrations.rs

@@ -1,7 +1,7 @@
-use std::cmp;
+use std::{cmp, collections::HashMap};
 
 use conduwuit::{
-	Err, Result, debug, debug_info, debug_warn, error, info,
+	Err, Pdu, Result, debug, debug_info, debug_warn, error, info,
 	result::NotFound,
 	utils::{
 		IterStream, ReadyExt,
@@ -13,14 +13,16 @@
 use futures::{FutureExt, StreamExt, TryStreamExt};
 use itertools::Itertools;
 use ruma::{
-	OwnedUserId, RoomId, UserId,
+	OwnedRoomId, OwnedUserId, RoomId, UserId,
 	events::{
-		GlobalAccountDataEventType, push_rules::PushRulesEvent, room::member::MembershipState,
+		GlobalAccountDataEventType, StateEventType, push_rules::PushRulesEvent,
+		room::member::MembershipState,
 	},
 	push::Ruleset,
+	serde::Raw,
 };
 
-use crate::{Services, media};
+use crate::{Services, media, rooms::short::ShortStateHash};
 
 /// The current schema version.
 /// - If database is opened at greater version we reject with error. The
@@ -152,6 +154,14 @@ async fn migrate(services: &Services) -> Result<()> {
 		info!("Migration: Bumped database version to 18");
 	}
 
+	if db["global"]
+		.get(POPULATED_USERROOMID_LEFTSTATE_TABLE_MARKER)
+		.await
+		.is_not_found()
+	{
+		populate_userroomid_leftstate_table(services).await?;
+	}
+
 	assert_eq!(
 		services.globals.db.database_version().await,
 		DATABASE_VERSION,
@@ -456,7 +466,11 @@ async fn retroactively_fix_bad_data_from_roomuserid_joined(services: &Services)
 
 		for user_id in &non_joined_members {
 			debug_info!("User is left or banned, marking as left");
-			services.rooms.state_cache.mark_as_left(user_id, room_id);
+			services
+				.rooms
+				.state_cache
+				.mark_as_left(user_id, room_id, None)
+				.await;
 		}
 	}
 
@@ -624,3 +638,78 @@ async fn fix_corrupt_msc4133_fields(services: &Services) -> Result {
 	db.db.sort()?;
 	Ok(())
 }
+
+const POPULATED_USERROOMID_LEFTSTATE_TABLE_MARKER: &str = "populate_userroomid_leftstate_table";
+async fn populate_userroomid_leftstate_table(services: &Services) -> Result {
+	type KeyVal<'a> = (Key<'a>, Raw<Option<Pdu>>);
+	type Key<'a> = (&'a UserId, &'a RoomId);
+
+	let db = &services.db;
+	let cork = db.cork_and_sync();
+	let userroomid_leftstate = db["userroomid_leftstate"].clone();
+
+	let (total, fixed, _) = userroomid_leftstate
+		.stream()
+		.try_fold(
+			(0_usize, 0_usize, HashMap::<OwnedRoomId, ShortStateHash>::new()),
+			async |(mut total, mut fixed, mut shortstatehash_cache): (
+				usize,
+				usize,
+				HashMap<_, _>,
+			),
+			       ((user_id, room_id), state): KeyVal<'_>|
+			       -> Result<(usize, usize, HashMap<_, _>)> {
+				if state.deserialize().is_err() {
+					let latest_shortstatehash =
+						if let Some(shortstatehash) = shortstatehash_cache.get(room_id) {
+							*shortstatehash
+						} else if let Ok(shortstatehash) =
+							services.rooms.state.get_room_shortstatehash(room_id).await
+						{
+							shortstatehash_cache.insert(room_id.to_owned(), shortstatehash);
+							shortstatehash
+						} else {
+							warn!(?room_id, ?user_id, "room has no shortstatehash");
+							return Ok((total, fixed, shortstatehash_cache));
+						};
+
+					let leave_state_event = services
+						.rooms
+						.state_accessor
+						.state_get(
+							latest_shortstatehash,
+							&StateEventType::RoomMember,
+							user_id.as_str(),
+						)
+						.await;
+
+					match leave_state_event {
+						| Ok(leave_state_event) => {
+							userroomid_leftstate.put((user_id, room_id), Json(leave_state_event));
+							fixed = fixed.saturating_add(1);
+						},
+						| Err(_) => {
+							warn!(
+								?room_id,
+								?user_id,
+								"room cached as left has no leave event for user, removing \
+								 cache entry"
+							);
+							userroomid_leftstate.del((user_id, room_id));
+						},
+					}
+				}
+
+				total = total.saturating_add(1);
+				Ok((total, fixed, shortstatehash_cache))
+			},
+		)
+		.await?;
+
+	drop(cork);
+	info!(?total, ?fixed, "Fixed entries in `userroomid_leftstate`.");
+
+	db["global"].insert(POPULATED_USERROOMID_LEFTSTATE_TABLE_MARKER, []);
+	db.db.sort()?;
+	Ok(())
+}

src/service/rooms/alias/mod.rs

@@ -188,9 +188,7 @@ pub fn local_aliases_for_room<'a>(
 	}
 
 	#[tracing::instrument(skip(self), level = "debug")]
-	pub fn all_local_aliases<'a>(
-		&'a self,
-	) -> impl Stream<Item = (&'a RoomId, &'a str)> + Send + 'a {
+	pub fn all_local_aliases(&self) -> impl Stream<Item = (&RoomId, &str)> + Send + '_ {
 		self.db
 			.alias_roomid
 			.stream()

src/service/rooms/event_handler/policy_server.rs

@@ -5,9 +5,9 @@
 
 use std::time::Duration;
 
-use conduwuit::{Err, Event, PduEvent, Result, debug, implement, warn};
+use conduwuit::{Err, Event, PduEvent, Result, debug, debug_info, implement, trace, warn};
 use ruma::{
-	RoomId, ServerName,
+	CanonicalJsonObject, RoomId, ServerName,
 	api::federation::room::policy::v1::Request as PolicyRequest,
 	events::{StateEventType, room::policy::RoomPolicyEventContent},
 };
@@ -25,7 +25,25 @@
 /// fail-open operation.
 #[implement(super::Service)]
 #[tracing::instrument(skip_all, level = "debug")]
-pub async fn ask_policy_server(&self, pdu: &PduEvent, room_id: &RoomId) -> Result<bool> {
+pub async fn ask_policy_server(
+	&self,
+	pdu: &PduEvent,
+	pdu_json: &CanonicalJsonObject,
+	room_id: &RoomId,
+) -> Result<bool> {
+	if !self.services.server.config.enable_msc4284_policy_servers {
+		return Ok(true); // don't ever contact policy servers
+	}
+	if self.services.server.config.policy_server_check_own_events
+		&& pdu.origin.is_some()
+		&& self
+			.services
+			.server
+			.is_ours(pdu.origin.as_ref().unwrap().as_str())
+	{
+		return Ok(true); // don't contact policy servers for locally generated events
+	}
+
 	if *pdu.event_type() == StateEventType::RoomPolicy.into() {
 		debug!(
 			room_id = %room_id,
@@ -47,12 +65,12 @@ pub async fn ask_policy_server(&self, pdu: &PduEvent, room_id: &RoomId) -> Resul
 	let via = match policyserver.via {
 		| Some(ref via) => ServerName::parse(via)?,
 		| None => {
-			debug!("No policy server configured for room {room_id}");
+			trace!("No policy server configured for room {room_id}");
 			return Ok(true);
 		},
 	};
 	if via.is_empty() {
-		debug!("Policy server is empty for room {room_id}, skipping spam check");
+		trace!("Policy server is empty for room {room_id}, skipping spam check");
 		return Ok(true);
 	}
 	if !self.services.state_cache.server_in_room(via, room_id).await {
@@ -66,12 +84,12 @@ pub async fn ask_policy_server(&self, pdu: &PduEvent, room_id: &RoomId) -> Resul
 	let outgoing = self
 		.services
 		.sending
-		.convert_to_outgoing_federation_event(pdu.to_canonical_object())
+		.convert_to_outgoing_federation_event(pdu_json.clone())
 		.await;
-	debug!(
+	debug_info!(
 		room_id = %room_id,
 		via = %via,
-		outgoing = ?outgoing,
+		outgoing = ?pdu_json,
 		"Checking event for spam with policy server"
 	);
 	let response = tokio::time::timeout(
@@ -85,7 +103,10 @@ pub async fn ask_policy_server(&self, pdu: &PduEvent, room_id: &RoomId) -> Resul
 	)
 	.await;
 	let response = match response {
-		| Ok(Ok(response)) => response,
+		| Ok(Ok(response)) => {
+			debug!("Response from policy server: {:?}", response);
+			response
+		},
 		| Ok(Err(e)) => {
 			warn!(
 				via = %via,
@@ -97,16 +118,18 @@ pub async fn ask_policy_server(&self, pdu: &PduEvent, room_id: &RoomId) -> Resul
 			// default.
 			return Err(e);
 		},
-		| Err(_) => {
+		| Err(elapsed) => {
 			warn!(
-				via = %via,
+				%via,
 				event_id = %pdu.event_id(),
-				room_id = %room_id,
+				%room_id,
+				%elapsed,
 				"Policy server request timed out after 10 seconds"
 			);
 			return Err!("Request to policy server timed out");
 		},
 	};
+	trace!("Recommendation from policy server was {}", response.recommendation);
 	if response.recommendation == "spam" {
 		warn!(
 			via = %via,

src/service/rooms/event_handler/upgrade_outlier_pdu.rs

@@ -255,7 +255,10 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
 		// 14-pre. If the event is not a state event, ask the policy server about it
 		if incoming_pdu.state_key.is_none() {
 			debug!(event_id = %incoming_pdu.event_id, "Checking policy server for event");
-			match self.ask_policy_server(&incoming_pdu, room_id).await {
+			match self
+				.ask_policy_server(&incoming_pdu, &incoming_pdu.to_canonical_object(), room_id)
+				.await
+			{
 				| Ok(false) => {
 					warn!(
 						event_id = %incoming_pdu.event_id,

src/service/rooms/lazy_loading/mod.rs

@@ -39,7 +39,7 @@ pub enum Status {
 	Seen(u64),
 }
 
-pub type Witness = HashSet<OwnedUserId>;
+pub type MemberSet = HashSet<OwnedUserId>;
 type Key<'a> = (&'a UserId, Option<&'a DeviceId>, &'a RoomId, &'a UserId);
 
 impl crate::Service for Service {
@@ -67,9 +67,11 @@ pub async fn reset(&self, ctx: &Context<'_>) {
 		.await;
 }
 
+/// Returns only the subset of `senders` which should be sent to the client
+/// according to the provided lazy loading context.
 #[implement(Service)]
 #[tracing::instrument(name = "retain", level = "debug", skip_all)]
-pub async fn witness_retain(&self, senders: Witness, ctx: &Context<'_>) -> Witness {
+pub async fn retain_lazy_members(&self, senders: MemberSet, ctx: &Context<'_>) -> MemberSet {
 	debug_assert!(
 		ctx.options.is_none_or(Options::is_enabled),
 		"lazy loading should be enabled by your options"
@@ -84,7 +86,7 @@ pub async fn witness_retain(&self, senders: Witness, ctx: &Context<'_>) -> Witne
 
 	pin_mut!(witness);
 	let _cork = self.db.db.cork();
-	let mut senders = Witness::with_capacity(senders.len());
+	let mut senders = MemberSet::with_capacity(senders.len());
 	while let Some((status, sender)) = witness.next().await {
 		if include_redundant || status == Status::Unseen {
 			senders.insert(sender.into());

src/service/rooms/read_receipt/data.rs

@@ -7,7 +7,7 @@
 use database::{Deserialized, Json, Map};
 use futures::{Stream, StreamExt};
 use ruma::{
-	CanonicalJsonObject, RoomId, UserId,
+	CanonicalJsonObject, OwnedUserId, RoomId, UserId,
 	events::{AnySyncEphemeralRoomEvent, receipt::ReceiptEvent},
 	serde::Raw,
 };
@@ -25,7 +25,7 @@ struct Services {
 	globals: Dep<globals::Service>,
 }
 
-pub(super) type ReceiptItem<'a> = (&'a UserId, u64, Raw<AnySyncEphemeralRoomEvent>);
+pub(super) type ReceiptItem = (OwnedUserId, u64, Raw<AnySyncEphemeralRoomEvent>);
 
 impl Data {
 	pub(super) fn new(args: &crate::Args<'_>) -> Self {
@@ -65,7 +65,7 @@ pub(super) fn readreceipts_since<'a>(
 		&'a self,
 		room_id: &'a RoomId,
 		since: u64,
-	) -> impl Stream<Item = ReceiptItem<'a>> + Send + 'a {
+	) -> impl Stream<Item = ReceiptItem> + Send + 'a {
 		type Key<'a> = (&'a RoomId, u64, &'a UserId);
 		type KeyVal<'a> = (Key<'a>, CanonicalJsonObject);
 
@@ -81,7 +81,7 @@ pub(super) fn readreceipts_since<'a>(
 
 				let event = serde_json::value::to_raw_value(&json)?;
 
-				Ok((user_id, count, Raw::from_json(event)))
+				Ok((user_id.to_owned(), count, Raw::from_json(event)))
 			})
 			.ignore_err()
 	}

src/service/rooms/read_receipt/mod.rs

@@ -104,16 +104,16 @@ pub async fn private_read_get(
 		Ok(Raw::from_json(event))
 	}
 
-	/// Returns an iterator over the most recent read_receipts in a room that
-	/// happened after the event with id `since`.
+	/// Returns an iterator over the most recent read_receipts in a room,
+	/// optionally after the event with id `since`.
 	#[inline]
 	#[tracing::instrument(skip(self), level = "debug")]
 	pub fn readreceipts_since<'a>(
 		&'a self,
 		room_id: &'a RoomId,
-		since: u64,
-	) -> impl Stream<Item = ReceiptItem<'a>> + Send + 'a {
-		self.db.readreceipts_since(room_id, since)
+		since: Option<u64>,
+	) -> impl Stream<Item = ReceiptItem> + Send + 'a {
+		self.db.readreceipts_since(room_id, since.unwrap_or(0))
 	}
 
 	/// Sets a private read marker at PDU `count`.

src/service/rooms/short/mod.rs

@@ -1,10 +1,18 @@
 use std::{borrow::Borrow, fmt::Debug, mem::size_of_val, sync::Arc};
 
 pub use conduwuit::matrix::pdu::{ShortEventId, ShortId, ShortRoomId, ShortStateKey};
-use conduwuit::{Result, err, implement, matrix::StateKey, utils, utils::IterStream};
+use conduwuit::{
+	Result, err, implement,
+	matrix::StateKey,
+	pair_of,
+	utils::{self, IterStream, ReadyExt},
+};
 use database::{Deserialized, Get, Map, Qry};
-use futures::{Stream, StreamExt};
-use ruma::{EventId, RoomId, events::StateEventType};
+use futures::{
+	Stream, StreamExt,
+	stream::{self},
+};
+use ruma::{EventId, OwnedEventId, RoomId, events::StateEventType};
 use serde::Deserialize;
 
 use crate::{Dep, globals};
@@ -258,3 +266,23 @@ pub async fn get_or_create_shortroomid(&self, room_id: &RoomId) -> ShortRoomId {
 			short
 		})
 }
+
+#[implement(Service)]
+pub async fn multi_get_state_from_short<'a, S>(
+	&'a self,
+	short_state: S,
+) -> impl Stream<Item = Result<((StateEventType, StateKey), OwnedEventId)>> + Send + 'a
+where
+	S: Stream<Item = (ShortStateKey, ShortEventId)> + Send + 'a,
+{
+	let (short_state_keys, short_event_ids): pair_of!(Vec<_>) = short_state.unzip().await;
+
+	StreamExt::zip(
+		self.multi_get_statekey_from_short(stream::iter(short_state_keys.into_iter())),
+		self.multi_get_eventid_from_short(stream::iter(short_event_ids.into_iter())),
+	)
+	.ready_filter_map(|state_event| match state_event {
+		| (Ok(state_key), Ok(event_id)) => Some(Ok((state_key, event_id))),
+		| (Err(e), _) | (_, Err(e)) => Some(Err(e)),
+	})
+}

src/service/rooms/state/mod.rs

@@ -20,7 +20,7 @@
 	EventId, OwnedEventId, OwnedRoomId, RoomId, RoomVersionId, UserId,
 	events::{
 		AnyStrippedStateEvent, StateEventType, TimelineEventType,
-		room::{create::RoomCreateEventContent, member::RoomMemberEventContent},
+		room::create::RoomCreateEventContent,
 	},
 	serde::Raw,
 };
@@ -126,21 +126,9 @@ pub async fn force_state(
 						continue;
 					};
 
-					let Ok(membership_event) = pdu.get_content::<RoomMemberEventContent>() else {
-						continue;
-					};
-
 					self.services
 						.state_cache
-						.update_membership(
-							room_id,
-							user_id,
-							membership_event,
-							&pdu.sender,
-							None,
-							None,
-							false,
-						)
+						.update_membership(room_id, user_id, &pdu, false)
 						.await?;
 				},
 				| TimelineEventType::SpaceChild => {

src/service/rooms/state_accessor/state.rs

@@ -1,7 +1,7 @@
 use std::{borrow::Borrow, ops::Deref, sync::Arc};
 
 use conduwuit::{
-	Result, at, err, implement,
+	Pdu, Result, at, err, implement,
 	matrix::{Event, StateKey},
 	pair_of,
 	utils::{
@@ -10,7 +10,7 @@
 	},
 };
 use database::Deserialized;
-use futures::{FutureExt, Stream, StreamExt, TryFutureExt, future::try_join, pin_mut};
+use futures::{FutureExt, Stream, StreamExt, TryFutureExt, pin_mut};
 use ruma::{
 	EventId, OwnedEventId, UserId,
 	events::{
@@ -125,7 +125,7 @@ pub async fn state_get(
 	shortstatehash: ShortStateHash,
 	event_type: &StateEventType,
 	state_key: &str,
-) -> Result<impl Event> {
+) -> Result<Pdu> {
 	self.state_get_id(shortstatehash, event_type, state_key)
 		.and_then(async |event_id: OwnedEventId| self.services.timeline.get_pdu(&event_id).await)
 		.await
@@ -286,28 +286,28 @@ pub fn state_keys<'a>(
 /// not in .1)
 #[implement(super::Service)]
 #[inline]
-pub fn state_removed(
+pub async fn state_removed(
 	&self,
 	shortstatehash: pair_of!(ShortStateHash),
-) -> impl Stream<Item = (ShortStateKey, ShortEventId)> + Send + '_ {
-	self.state_added((shortstatehash.1, shortstatehash.0))
+) -> Result<Vec<(ShortStateKey, ShortEventId)>> {
+	self.state_added((shortstatehash.1, shortstatehash.0)).await
 }
 
 /// Returns the state events added between the interval (present in .1 but
 /// not in .0)
 #[implement(super::Service)]
-pub fn state_added(
+pub async fn state_added(
 	&self,
 	shortstatehash: pair_of!(ShortStateHash),
-) -> impl Stream<Item = (ShortStateKey, ShortEventId)> + Send + '_ {
-	let a = self.load_full_state(shortstatehash.0);
-	let b = self.load_full_state(shortstatehash.1);
-	try_join(a, b)
-		.map_ok(|(a, b)| b.difference(&a).copied().collect::<Vec<_>>())
-		.map_ok(IterStream::try_stream)
-		.try_flatten_stream()
-		.ignore_err()
+) -> Result<Vec<(ShortStateKey, ShortEventId)>> {
+	let full_state_a = self.load_full_state(shortstatehash.0).await?;
+	let full_state_b = self.load_full_state(shortstatehash.1).await?;
+
+	Ok(full_state_b
+		.difference(&full_state_a)
+		.copied()
 		.map(parse_compressed_state_event)
+		.collect())
 }
 
 #[implement(super::Service)]

src/service/rooms/state_accessor/user_can.rs

@@ -1,9 +1,10 @@
-use conduwuit::{Err, Result, implement, matrix::Event, pdu::PduBuilder};
+use conduwuit::{Err, Result, RoomVersion, implement, matrix::Event, pdu::PduBuilder};
 use ruma::{
 	EventId, RoomId, UserId,
 	events::{
 		StateEventType, TimelineEventType,
 		room::{
+			create::RoomCreateEventContent,
 			history_visibility::{HistoryVisibility, RoomHistoryVisibilityEventContent},
 			member::{MembershipState, RoomMemberEventContent},
 			power_levels::{RoomPowerLevels, RoomPowerLevelsEventContent},
@@ -44,6 +45,23 @@ pub async fn user_can_redact(
 		)));
 	}
 
+	let room_create = self
+		.room_state_get(room_id, &StateEventType::RoomCreate, "")
+		.await?;
+	let create_content: RoomCreateEventContent =
+		serde_json::from_str(room_create.content().get())?;
+	let room_features = RoomVersion::new(&create_content.room_version)?;
+	if room_features.explicitly_privilege_room_creators {
+		let sender_owned = sender.to_owned();
+		if sender == room_create.sender()
+			|| create_content
+				.additional_creators
+				.is_some_and(|cs| cs.contains(&sender_owned))
+		{
+			return Ok(true);
+		}
+	}
+
 	match self
 		.room_state_get_content::<RoomPowerLevelsEventContent>(
 			room_id,
@@ -68,18 +86,10 @@ pub async fn user_can_redact(
 		},
 		| _ => {
 			// Falling back on m.room.create to judge power level
-			match self
-				.room_state_get(room_id, &StateEventType::RoomCreate, "")
-				.await
-			{
-				| Ok(room_create) => Ok(room_create.sender() == sender
-					|| redacting_event
-						.as_ref()
-						.is_ok_and(|redacting_event| redacting_event.sender() == sender)),
-				| _ => Err!(Database(
-					"No m.room.power_levels or m.room.create events in database for room"
-				)),
-			}
+			Ok(room_create.sender() == sender
+				|| redacting_event
+					.as_ref()
+					.is_ok_and(|redacting_event| redacting_event.sender() == sender))
 		},
 	}
 }

src/service/rooms/state_cache/mod.rs

@@ -4,7 +4,7 @@
 use std::{collections::HashMap, sync::Arc};
 
 use conduwuit::{
-	Result, SyncRwLock, implement,
+	Pdu, Result, SyncRwLock, implement,
 	result::LogErr,
 	utils::{ReadyExt, stream::TryIgnore},
 	warn,
@@ -13,7 +13,7 @@
 use futures::{Stream, StreamExt, future::join5, pin_mut};
 use ruma::{
 	OwnedRoomId, OwnedUserId, RoomId, ServerName, UserId,
-	events::{AnyStrippedStateEvent, AnySyncStateEvent, room::member::MembershipState},
+	events::{AnyStrippedStateEvent, room::member::MembershipState},
 	serde::Raw,
 };
 
@@ -54,7 +54,6 @@ struct Data {
 
 type AppServiceInRoomCache = SyncRwLock<HashMap<OwnedRoomId, HashMap<String, bool>>>;
 type StrippedStateEventItem = (OwnedRoomId, Vec<Raw<AnyStrippedStateEvent>>);
-type SyncStateEventItem = (OwnedRoomId, Vec<Raw<AnySyncStateEvent>>);
 
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
@@ -431,18 +430,9 @@ pub async fn knock_state(
 
 #[implement(Service)]
 #[tracing::instrument(skip(self), level = "trace")]
-pub async fn left_state(
-	&self,
-	user_id: &UserId,
-	room_id: &RoomId,
-) -> Result<Vec<Raw<AnyStrippedStateEvent>>> {
+pub async fn left_state(&self, user_id: &UserId, room_id: &RoomId) -> Result<Option<Pdu>> {
 	let key = (user_id, room_id);
-	self.db
-		.userroomid_leftstate
-		.qry(&key)
-		.await
-		.deserialized()
-		.and_then(|val: Raw<Vec<AnyStrippedStateEvent>>| val.deserialize_as().map_err(Into::into))
+	self.db.userroomid_leftstate.qry(&key).await.deserialized()
 }
 
 /// Returns an iterator over all rooms a user left.
@@ -451,8 +441,8 @@ pub async fn left_state(
 pub fn rooms_left<'a>(
 	&'a self,
 	user_id: &'a UserId,
-) -> impl Stream<Item = SyncStateEventItem> + Send + 'a {
-	type KeyVal<'a> = (Key<'a>, Raw<Vec<Raw<AnySyncStateEvent>>>);
+) -> impl Stream<Item = (OwnedRoomId, Option<Pdu>)> + Send + 'a {
+	type KeyVal<'a> = (Key<'a>, Raw<Option<Pdu>>);
 	type Key<'a> = (&'a UserId, &'a RoomId);
 
 	let prefix = (user_id, Interfix);
@@ -461,7 +451,7 @@ pub fn rooms_left<'a>(
 		.stream_prefix(&prefix)
 		.ignore_err()
 		.map(|((_, room_id), state): KeyVal<'_>| (room_id.to_owned(), state))
-		.map(|(room_id, state)| Ok((room_id, state.deserialize_as()?)))
+		.map(|(room_id, state)| Ok((room_id, state.deserialize()?)))
 		.ignore_err()
 }
 

src/service/rooms/state_cache/update.rs

@@ -1,13 +1,13 @@
 use std::collections::HashSet;
 
-use conduwuit::{Err, Result, implement, is_not_empty, utils::ReadyExt, warn};
+use conduwuit::{Err, Event, Pdu, Result, implement, is_not_empty, utils::ReadyExt, warn};
 use database::{Json, serialize_key};
 use futures::StreamExt;
 use ruma::{
 	OwnedServerName, RoomId, UserId,
 	events::{
-		AnyStrippedStateEvent, AnySyncStateEvent, GlobalAccountDataEventType,
-		RoomAccountDataEventType, StateEventType,
+		AnyStrippedStateEvent, GlobalAccountDataEventType, RoomAccountDataEventType,
+		StateEventType,
 		direct::DirectEvent,
 		invite_permission_config::FilterLevel,
 		room::{
@@ -26,8 +26,7 @@
 		fields(
 			%room_id,
 			%user_id,
-			%sender,
-			?membership_event,
+			?pdu,
 		),
 	)]
 #[allow(clippy::too_many_arguments)]
@@ -35,13 +34,10 @@ pub async fn update_membership(
 	&self,
 	room_id: &RoomId,
 	user_id: &UserId,
-	membership_event: RoomMemberEventContent,
-	sender: &UserId,
-	last_state: Option<Vec<Raw<AnyStrippedStateEvent>>>,
-	invite_via: Option<Vec<OwnedServerName>>,
+	pdu: &Pdu,
 	update_joined_count: bool,
 ) -> Result {
-	let membership = membership_event.membership;
+	let membership = pdu.get_content::<RoomMemberEventContent>()?;
 
 	// Keep track what remote users exist by adding them as "deactivated" users
 	//
@@ -54,7 +50,7 @@ pub async fn update_membership(
 		}
 	}
 
-	match &membership {
+	match &membership.membership {
 		| MembershipState::Join => {
 			// Check if the user never joined this room
 			if !self.once_joined(user_id, room_id).await {
@@ -122,33 +118,14 @@ pub async fn update_membership(
 			self.mark_as_joined(user_id, room_id);
 		},
 		| MembershipState::Invite => {
-			// return an error for blocked invites. ignored invites aren't handled here
-			// since the recipient's membership should still be changed to `invite`.
-			// they're filtered out in the individual /sync handlers
-			if matches!(
-				self.services
-					.users
-					.invite_filter_level(sender, user_id)
-					.await,
-				FilterLevel::Block
-			) {
-				return Err!(Request(InviteBlocked(
-					"{user_id} has blocked invites from {sender}."
-				)));
-			}
-			self.mark_as_invited(user_id, room_id, sender, last_state, invite_via)
-				.await;
+			// TODO: make sure that passing None for `last_state` is correct behavior.
+			// the call from `append_pdu` used to use `services.state.summary_stripped`
+			// to fill that parameter.
+			self.mark_as_invited(user_id, room_id, pdu.sender(), None, None)
+				.await?;
 		},
 		| MembershipState::Leave | MembershipState::Ban => {
-			self.mark_as_left(user_id, room_id);
-
-			if self.services.globals.user_is_local(user_id)
-				&& (self.services.config.forget_forced_upon_leave
-					|| self.services.metadata.is_banned(room_id).await
-					|| self.services.metadata.is_disabled(room_id).await)
-			{
-				self.forget(room_id, user_id);
-			}
+			self.mark_as_left(user_id, room_id, Some(pdu.clone())).await;
 		},
 		| _ => {},
 	}
@@ -252,24 +229,24 @@ pub fn mark_as_joined(&self, user_id: &UserId, room_id: &RoomId) {
 	self.db.roomid_inviteviaservers.remove(room_id);
 }
 
-/// Direct DB function to directly mark a user as left. It is not
-/// recommended to use this directly. You most likely should use
-/// `update_membership` instead
+/// Mark a user as having left a room.
+///
+/// `leave_pdu` represents the m.room.member event which the user sent to leave
+/// the room. If this is None, no event was actually sent, but we must still
+/// behave as if the user is no longer in the room. This may occur, for example,
+/// if the room being left has been server-banned by an administrator.
 #[implement(super::Service)]
 #[tracing::instrument(skip(self), level = "debug")]
-pub fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId) {
+pub async fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId, leave_pdu: Option<Pdu>) {
 	let userroom_id = (user_id, room_id);
 	let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id");
 
 	let roomuser_id = (room_id, user_id);
 	let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
 
-	// (timo) TODO
-	let leftstate = Vec::<Raw<AnySyncStateEvent>>::new();
-
 	self.db
 		.userroomid_leftstate
-		.raw_put(&userroom_id, Json(leftstate));
+		.raw_put(&userroom_id, Json(leave_pdu));
 	self.db
 		.roomuserid_leftcount
 		.raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap());
@@ -285,6 +262,14 @@ pub fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId) {
 	self.db.roomuserid_knockedcount.remove(&roomuser_id);
 
 	self.db.roomid_inviteviaservers.remove(room_id);
+
+	if self.services.globals.user_is_local(user_id)
+		&& (self.services.config.forget_forced_upon_leave
+			|| self.services.metadata.is_banned(room_id).await
+			|| self.services.metadata.is_disabled(room_id).await)
+	{
+		self.forget(room_id, user_id);
+	}
 }
 
 /// Direct DB function to directly mark a user as knocked. It is not
@@ -351,7 +336,20 @@ pub async fn mark_as_invited(
 	sender_user: &UserId,
 	last_state: Option<Vec<Raw<AnyStrippedStateEvent>>>,
 	invite_via: Option<Vec<OwnedServerName>>,
-) {
+) -> Result<()> {
+	// return an error for blocked invites. ignored invites aren't handled here
+	// since the recipient's membership should still be changed to `invite`.
+	// they're filtered out in the individual /sync handlers
+	if matches!(
+		self.services
+			.users
+			.invite_filter_level(sender_user, user_id)
+			.await,
+		FilterLevel::Block
+	) {
+		return Err!(Request(InviteBlocked("{user_id} has blocked invites from {sender_user}.")));
+	}
+
 	let roomuser_id = (room_id, user_id);
 	let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
 
@@ -366,7 +364,7 @@ pub async fn mark_as_invited(
 		.raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap());
 	self.db
 		.userroomid_invitesender
-		.raw_put(&userroom_id, sender_user);
+		.insert(&userroom_id, sender_user);
 
 	self.db.userroomid_joined.remove(&userroom_id);
 	self.db.roomuserid_joined.remove(&roomuser_id);
@@ -380,4 +378,6 @@ pub async fn mark_as_invited(
 	if let Some(servers) = invite_via.filter(is_not_empty!()) {
 		self.add_servers_invite_via(room_id, servers).await;
 	}
+
+	Ok(())
 }

src/service/rooms/state_compressor/mod.rs

@@ -526,7 +526,7 @@ pub(crate) fn compress_state_event(
 
 #[inline]
 #[must_use]
-pub(crate) fn parse_compressed_state_event(
+pub fn parse_compressed_state_event(
 	compressed_event: CompressedStateEvent,
 ) -> (ShortStateKey, ShortEventId) {
 	use utils::u64_from_u8;

src/service/rooms/timeline/append.rs

@@ -19,9 +19,7 @@
 		GlobalAccountDataEventType, StateEventType, TimelineEventType,
 		push_rules::PushRulesEvent,
 		room::{
-			encrypted::Relation,
-			member::{MembershipState, RoomMemberEventContent},
-			power_levels::RoomPowerLevelsEventContent,
+			encrypted::Relation, power_levels::RoomPowerLevelsEventContent,
 			redaction::RoomRedactionEventContent,
 		},
 	},
@@ -323,31 +321,12 @@ pub async fn append_pdu<'a, Leaves>(
 				let target_user_id =
 					UserId::parse(state_key).expect("This state_key was previously validated");
 
-				let content: RoomMemberEventContent = pdu.get_content()?;
-				let stripped_state = match content.membership {
-					| MembershipState::Invite | MembershipState::Knock => self
-						.services
-						.state
-						.summary_stripped(pdu, room_id)
-						.await
-						.into(),
-					| _ => None,
-				};
-
 				// Update our membership info, we do this here incase a user is invited or
 				// knocked and immediately leaves we need the DB to record the invite or
 				// knock event for auth
 				self.services
 					.state_cache
-					.update_membership(
-						room_id,
-						target_user_id,
-						content,
-						pdu.sender(),
-						stripped_state,
-						None,
-						true,
-					)
+					.update_membership(room_id, target_user_id, pdu, true)
 					.await?;
 			}
 		},

src/service/rooms/timeline/create.rs

@@ -9,6 +9,7 @@
 		state_res::{self, RoomVersion},
 	},
 	utils::{self, IterStream, ReadyExt, stream::TryIgnore},
+	warn,
 };
 use futures::{StreamExt, TryStreamExt, future, future::ready};
 use ruma::{
@@ -19,7 +20,6 @@
 	uint,
 };
 use serde_json::value::{RawValue, to_raw_value};
-use tracing::warn;
 
 use super::RoomMutexGuard;
 
@@ -267,23 +267,19 @@ fn from_evt(
 			| _ => Err!(Request(Unknown(warn!("Signing event failed: {e}")))),
 		};
 	}
-
 	// Generate event id
 	pdu.event_id = gen_event_id(&pdu_json, &room_version_id)?;
-
-	pdu_json.insert("event_id".into(), CanonicalJsonValue::String(pdu.event_id.clone().into()));
-
 	// Check with the policy server
+	pdu_json.insert("event_id".into(), CanonicalJsonValue::String(pdu.event_id.clone().into()));
 	if room_id.is_some() {
 		trace!(
-			"Checking event {} in room {} with policy server",
-			pdu.event_id,
+			"Checking event in room {} with policy server",
 			pdu.room_id.as_ref().map_or("None", |id| id.as_str())
 		);
 		match self
 			.services
 			.event_handler
-			.ask_policy_server(&pdu, &pdu.room_id_or_hash())
+			.ask_policy_server(&pdu, &pdu_json, pdu.room_id().expect("has room ID"))
 			.await
 		{
 			| Ok(true) => {},

src/service/rooms/timeline/mod.rs

@@ -186,10 +186,8 @@ pub async fn get_pdu_id(&self, event_id: &EventId) -> Result<RawPduId> {
 	}
 
 	/// Returns the pdu.
-	///
-	/// Checks the `eventid_outlierpdu` Tree if not found in the timeline.
 	#[inline]
-	pub async fn get_non_outlier_pdu(&self, event_id: &EventId) -> Result<impl Event> {
+	pub async fn get_non_outlier_pdu(&self, event_id: &EventId) -> Result<PduEvent> {
 		self.db.get_non_outlier_pdu(event_id).await
 	}
 
@@ -243,7 +241,7 @@ pub fn all_pdus<'a>(
 		self.pdus(Some(user_id), room_id, None).ignore_err()
 	}
 
-	/// Reverse iteration starting at from.
+	/// Reverse iteration starting after `until`.
 	#[tracing::instrument(skip(self), level = "debug")]
 	pub fn pdus_rev<'a>(
 		&'a self,
@@ -255,7 +253,7 @@ pub fn pdus_rev<'a>(
 			.pdus_rev(user_id, room_id, until.unwrap_or_else(PduCount::max))
 	}
 
-	/// Forward iteration starting at from.
+	/// Forward iteration starting after `from`.
 	#[tracing::instrument(skip(self), level = "debug")]
 	pub fn pdus<'a>(
 		&'a self,