Merge branch 'main' into devon/element-docker

This commit is contained in:
Quentin Gliech
2026-02-17 11:50:54 +01:00
committed by GitHub
41 changed files with 2670 additions and 1532 deletions
+1 -1
View File
@@ -13,7 +13,7 @@ runs:
uses: open-policy-agent/setup-opa@v2.2.0
with:
# Keep in sync with the Dockerfile and policies/Makefile
version: 1.8.0
version: 1.13.1
- name: Build the policies
run: make
+1 -1
View File
@@ -271,7 +271,7 @@ jobs:
mirrors = ["mirror.gcr.io"]
- name: Login to GitHub Container Registry
uses: docker/login-action@v3.6.0
uses: docker/login-action@v3.7.0
with:
registry: ghcr.io
username: ${{ github.repository_owner }}
+5 -5
View File
@@ -42,7 +42,7 @@ jobs:
uses: StyraInc/setup-regal@v1
with:
# Keep in sync with policies/Makefile
version: 0.36.1
version: 0.38.1
- name: Lint policies
working-directory: ./policies
@@ -64,7 +64,7 @@ jobs:
uses: actions/checkout@v6
- name: Install Node
uses: actions/setup-node@v6.1.0
uses: actions/setup-node@v6.2.0
with:
node-version: 24
@@ -88,7 +88,7 @@ jobs:
uses: actions/checkout@v6
- name: Install Node
uses: actions/setup-node@v6.1.0
uses: actions/setup-node@v6.2.0
with:
node-version: 24
@@ -112,7 +112,7 @@ jobs:
uses: actions/checkout@v6
- name: Install Node
uses: actions/setup-node@v6.1.0
uses: actions/setup-node@v6.2.0
with:
node-version: 24
@@ -216,7 +216,7 @@ jobs:
uses: actions/checkout@v6
- name: Install Rust toolchain
uses: dtolnay/rust-toolchain@1.89.0
uses: dtolnay/rust-toolchain@1.93.0
with:
components: clippy
+1 -1
View File
@@ -39,7 +39,7 @@ jobs:
tool: mdbook
- name: Install Node
uses: actions/setup-node@v6.1.0
uses: actions/setup-node@v6.2.0
with:
node-version: 24
+1 -1
View File
@@ -64,7 +64,7 @@ jobs:
uses: actions/checkout@v6
- name: Install Node
uses: actions/setup-node@v6.1.0
uses: actions/setup-node@v6.2.0
with:
node-version: 24
+2 -2
View File
@@ -22,7 +22,7 @@ jobs:
uses: actions/checkout@v6
- name: Install Node
uses: actions/setup-node@v6.1.0
uses: actions/setup-node@v6.2.0
with:
node-version: 24
@@ -42,7 +42,7 @@ jobs:
- name: Create Pull Request
id: cpr
uses: peter-evans/create-pull-request@v8.0.0
uses: peter-evans/create-pull-request@v8.1.0
with:
sign-commits: true
token: ${{ secrets.BOT_GITHUB_TOKEN }}
+1 -1
View File
@@ -21,7 +21,7 @@ jobs:
uses: actions/checkout@v6
- name: Install Node
uses: actions/setup-node@v6.1.0
uses: actions/setup-node@v6.2.0
with:
node-version: 24
Generated
+91 -79
View File
@@ -197,9 +197,12 @@ checksum = "c3d036a3c4ab069c7b410a2ce876bd74808d2d0888a82667669f8e783a898bf1"
[[package]]
name = "arc-swap"
version = "1.7.1"
version = "1.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457"
checksum = "9ded5f9a03ac8f24d1b8a25101ee812cd32cdc8c50a4c50237de2c4915850e73"
dependencies = [
"rustversion",
]
[[package]]
name = "argon2"
@@ -677,9 +680,9 @@ checksum = "55248b47b0caf0546f7988906588779981c43bb1bc9d0c44087278f80cdb44ba"
[[package]]
name = "bcrypt"
version = "0.17.1"
version = "0.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "abaf6da45c74385272ddf00e1ac074c7d8a6c1a1dda376902bd6a427522a8b2c"
checksum = "9a0f5948f30df5f43ac29d310b7476793be97c50787e6ef4a63d960a0d0be827"
dependencies = [
"base64",
"blowfish",
@@ -805,9 +808,9 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
[[package]]
name = "bytes"
version = "1.10.1"
version = "1.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a"
checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"
dependencies = [
"serde",
]
@@ -882,6 +885,12 @@ version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2fd1289c04a9ea8cb22300a459a72a385d7c73d3259e2ed7dcb2af674838cfa9"
[[package]]
name = "cfg_aliases"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
[[package]]
name = "chacha20"
version = "0.9.1"
@@ -1736,7 +1745,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
dependencies = [
"libc",
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
@@ -2779,14 +2788,15 @@ dependencies = [
[[package]]
name = "insta"
version = "1.43.2"
version = "1.46.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "46fdb647ebde000f43b5b53f773c30cf9b0cb4300453208713fa38b2c70935a0"
checksum = "e82db8c87c7f1ccecb34ce0c24399b8a73081427f3c7c50a5d597925356115e4"
dependencies = [
"console",
"once_cell",
"serde",
"similar",
"tempfile",
]
[[package]]
@@ -3097,7 +3107,7 @@ dependencies = [
[[package]]
name = "mas-axum-utils"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"axum",
@@ -3131,7 +3141,7 @@ dependencies = [
[[package]]
name = "mas-cli"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"axum",
@@ -3206,7 +3216,7 @@ dependencies = [
[[package]]
name = "mas-config"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"camino",
@@ -3237,7 +3247,7 @@ dependencies = [
[[package]]
name = "mas-context"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"console",
"opentelemetry",
@@ -3253,7 +3263,7 @@ dependencies = [
[[package]]
name = "mas-data-model"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"base64ct",
"chrono",
@@ -3276,7 +3286,7 @@ dependencies = [
[[package]]
name = "mas-email"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"async-trait",
"lettre",
@@ -3287,7 +3297,7 @@ dependencies = [
[[package]]
name = "mas-handlers"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"aide",
"anyhow",
@@ -3367,7 +3377,7 @@ dependencies = [
[[package]]
name = "mas-http"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"futures-util",
"headers",
@@ -3388,7 +3398,7 @@ dependencies = [
[[package]]
name = "mas-i18n"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"camino",
"icu_calendar",
@@ -3410,7 +3420,7 @@ dependencies = [
[[package]]
name = "mas-i18n-scan"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"camino",
"clap",
@@ -3424,7 +3434,7 @@ dependencies = [
[[package]]
name = "mas-iana"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"schemars 0.9.0",
"serde",
@@ -3432,7 +3442,7 @@ dependencies = [
[[package]]
name = "mas-iana-codegen"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"async-trait",
@@ -3449,7 +3459,7 @@ dependencies = [
[[package]]
name = "mas-jose"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"base64ct",
"chrono",
@@ -3479,7 +3489,7 @@ dependencies = [
[[package]]
name = "mas-keystore"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"aead",
"base64ct",
@@ -3507,7 +3517,7 @@ dependencies = [
[[package]]
name = "mas-listener"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"bytes",
@@ -3531,7 +3541,7 @@ dependencies = [
[[package]]
name = "mas-matrix"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"async-trait",
@@ -3541,7 +3551,7 @@ dependencies = [
[[package]]
name = "mas-matrix-synapse"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"async-trait",
@@ -3558,7 +3568,7 @@ dependencies = [
[[package]]
name = "mas-oidc-client"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"assert_matches",
"async-trait",
@@ -3594,7 +3604,7 @@ dependencies = [
[[package]]
name = "mas-policy"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"arc-swap",
@@ -3611,7 +3621,7 @@ dependencies = [
[[package]]
name = "mas-router"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"axum",
"serde",
@@ -3622,7 +3632,7 @@ dependencies = [
[[package]]
name = "mas-spa"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"camino",
"serde",
@@ -3631,7 +3641,7 @@ dependencies = [
[[package]]
name = "mas-storage"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"async-trait",
"chrono",
@@ -3653,7 +3663,7 @@ dependencies = [
[[package]]
name = "mas-storage-pg"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"async-trait",
"chrono",
@@ -3683,7 +3693,7 @@ dependencies = [
[[package]]
name = "mas-tasks"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"async-trait",
@@ -3715,7 +3725,7 @@ dependencies = [
[[package]]
name = "mas-templates"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"arc-swap",
@@ -3747,7 +3757,7 @@ dependencies = [
[[package]]
name = "mas-tower"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"http",
"opentelemetry",
@@ -3950,9 +3960,9 @@ dependencies = [
[[package]]
name = "num-conv"
version = "0.1.0"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9"
checksum = "cf97ec579c3c42f953ef76dbf8d55ac91fb219dde70e49aa4a6b7d74e9919050"
[[package]]
name = "num-integer"
@@ -4016,7 +4026,7 @@ dependencies = [
[[package]]
name = "oauth2-types"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"assert_matches",
"base64ct",
@@ -4086,7 +4096,7 @@ dependencies = [
"sha1",
"sha2",
"sprintf",
"thiserror 1.0.69",
"thiserror 2.0.17",
"tokio",
"tracing",
"urlencoding",
@@ -4708,7 +4718,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9120690fafc389a67ba3803df527d0ec9cbbc9cc45e4cc20b332996dfb672425"
dependencies = [
"anyhow",
"itertools 0.13.0",
"itertools 0.14.0",
"proc-macro2",
"quote",
"syn",
@@ -4968,9 +4978,9 @@ checksum = "caf4aa5b0f434c91fe5c7f1ecb6a5ece2130b02ad2a590589dda5146df959001"
[[package]]
name = "reqwest"
version = "0.12.24"
version = "0.12.28"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9d0946410b9f7b082a427e4ef5c8ff541a88b357bc6c637c40db3a68ac70a36f"
checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147"
dependencies = [
"base64",
"bytes",
@@ -5146,7 +5156,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys",
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
@@ -5418,10 +5428,11 @@ checksum = "d767eb0aabc880b29956c35734170f26ed551a859dbd361d140cdbeca61ab1e2"
[[package]]
name = "sentry"
version = "0.46.0"
version = "0.46.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d9794f69ad475e76c057e326175d3088509649e3aed98473106b9fe94ba59424"
checksum = "d92d893ba7469d361a6958522fa440e4e2bc8bf4c5803cd1bf40b9af63f8f9a8"
dependencies = [
"cfg_aliases",
"httpdate",
"reqwest",
"sentry-backtrace",
@@ -5435,9 +5446,9 @@ dependencies = [
[[package]]
name = "sentry-backtrace"
version = "0.46.0"
version = "0.46.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e81137ad53b8592bd0935459ad74c0376053c40084aa170451e74eeea8dbc6c3"
checksum = "5f8784d0a27b5cd4b5f75769ffc84f0b7580e3c35e1af9cd83cb90b612d769cc"
dependencies = [
"backtrace",
"regex",
@@ -5446,9 +5457,9 @@ dependencies = [
[[package]]
name = "sentry-contexts"
version = "0.46.0"
version = "0.46.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cfb403c66cc2651a01b9bacda2e7c22cd51f7e8f56f206aa4310147eb3259282"
checksum = "0e5eb42f4cd4f9fdfec9e3b07b25a4c9769df83d218a7e846658984d5948ad3e"
dependencies = [
"hostname",
"libc",
@@ -5460,9 +5471,9 @@ dependencies = [
[[package]]
name = "sentry-core"
version = "0.46.0"
version = "0.46.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cfc409727ae90765ca8ea76fe6c949d6f159a11d02e130b357fa652ee9efcada"
checksum = "b0b1e7ca40f965db239da279bf278d87b7407469b98835f27f0c8e59ed189b06"
dependencies = [
"rand 0.9.2",
"sentry-types",
@@ -5473,9 +5484,9 @@ dependencies = [
[[package]]
name = "sentry-panic"
version = "0.46.0"
version = "0.46.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3df79f4e1e72b2a8b75a0ebf49e78709ceb9b3f0b451f13adc92a0361b0aaabe"
checksum = "8906f8be87aea5ac7ef937323fb655d66607427f61007b99b7cb3504dc5a156c"
dependencies = [
"sentry-backtrace",
"sentry-core",
@@ -5483,9 +5494,9 @@ dependencies = [
[[package]]
name = "sentry-tower"
version = "0.46.0"
version = "0.46.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7eec9885bceb8ba374858d015bb6fa39dbb341d94ca088bc8f13bee2e64e2c68"
checksum = "56aebe376310840b49dad4cca55c7b32d9abdc14946cd071d4158ecb149b63a4"
dependencies = [
"axum",
"http",
@@ -5498,9 +5509,9 @@ dependencies = [
[[package]]
name = "sentry-tracing"
version = "0.46.0"
version = "0.46.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ff2046f527fd4b75e0b6ab3bd656c67dce42072f828dc4d03c206d15dca74a93"
checksum = "5b07eefe04486316c57aba08ab53dd44753c25102d1d3fe05775cc93a13262d9"
dependencies = [
"bitflags",
"sentry-backtrace",
@@ -5511,9 +5522,9 @@ dependencies = [
[[package]]
name = "sentry-types"
version = "0.46.0"
version = "0.46.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c7b9b4e4c03a4d3643c18c78b8aa91d2cbee5da047d2fa0ca4bb29bc67e6c55c"
checksum = "567711f01f86a842057e1fc17779eba33a336004227e1a1e7e6cc2599e22e259"
dependencies = [
"debugid",
"hex",
@@ -6120,7 +6131,7 @@ dependencies = [
[[package]]
name = "syn2mas"
version = "1.10.0"
version = "1.11.0"
dependencies = [
"anyhow",
"arc-swap",
@@ -6191,7 +6202,7 @@ dependencies = [
"getrandom 0.3.3",
"once_cell",
"rustix",
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
@@ -6276,31 +6287,32 @@ dependencies = [
[[package]]
name = "time"
version = "0.3.43"
version = "0.3.47"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "83bde6f1ec10e72d583d91623c939f623002284ef622b87de38cfd546cbf2031"
checksum = "743bd48c283afc0388f9b8827b976905fb217ad9e647fae3a379a9283c4def2c"
dependencies = [
"deranged",
"itoa",
"libc",
"num-conv",
"num_threads",
"powerfmt",
"serde",
"serde_core",
"time-core",
"time-macros",
]
[[package]]
name = "time-core"
version = "0.1.6"
version = "0.1.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "40868e7c1d2f0b8d73e4a8c7f0ff63af4f6d19be117e90bd73eb1d62cf831c6b"
checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca"
[[package]]
name = "time-macros"
version = "0.2.24"
version = "0.2.27"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "30cfb0125f12d9c277f35663a0a33f8c30190f4e4574868a330595412d34ebf3"
checksum = "2e70e4c5a0e0a8a4823ad65dfe1a6930e4f4d756dcd9dd7939022b5e8c501215"
dependencies = [
"num-conv",
"time-core",
@@ -6491,9 +6503,9 @@ dependencies = [
[[package]]
name = "tower-http"
version = "0.6.6"
version = "0.6.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "adc82fd73de2a9722ac5da747f12383d2bfdb93591ee6c58486e0097890f05f2"
checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8"
dependencies = [
"bitflags",
"bytes",
@@ -6531,9 +6543,9 @@ checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3"
[[package]]
name = "tracing"
version = "0.1.41"
version = "0.1.44"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0"
checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
dependencies = [
"log",
"pin-project-lite",
@@ -6555,9 +6567,9 @@ dependencies = [
[[package]]
name = "tracing-attributes"
version = "0.1.30"
version = "0.1.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "81383ab64e72a7a8b8e13130c49e3dab29def6d0c7d76a03087b3cf71c5c6903"
checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da"
dependencies = [
"proc-macro2",
"quote",
@@ -6566,9 +6578,9 @@ dependencies = [
[[package]]
name = "tracing-core"
version = "0.1.34"
version = "0.1.36"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b9d12581f227e93f094d3af2ae690a574abb8a2b9b7a96e7cfe9647b2b617678"
checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a"
dependencies = [
"once_cell",
"valuable",
@@ -6616,9 +6628,9 @@ dependencies = [
[[package]]
name = "tracing-subscriber"
version = "0.3.20"
version = "0.3.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2054a14f5307d601f88daf0553e1cbf472acc4f2c51afab632431cdcd72124d5"
checksum = "2f30143827ddab0d256fd843b7a66d164e9f271cfa0dde49142c5ca0ca291f1e"
dependencies = [
"matchers",
"nu-ansi-term",
@@ -7318,7 +7330,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
+35 -35
View File
@@ -9,7 +9,7 @@ members = ["crates/*"]
resolver = "2"
# Updated in the CI with a `sed` command
package.version = "1.10.0"
package.version = "1.11.0"
package.license = "AGPL-3.0-only OR LicenseRef-Element-Commercial"
package.authors = ["Element Backend Team"]
package.edition = "2024"
@@ -34,35 +34,35 @@ broken_intra_doc_links = "deny"
[workspace.dependencies]
# Workspace crates
mas-axum-utils = { path = "./crates/axum-utils/", version = "=1.10.0" }
mas-cli = { path = "./crates/cli/", version = "=1.10.0" }
mas-config = { path = "./crates/config/", version = "=1.10.0" }
mas-context = { path = "./crates/context/", version = "=1.10.0" }
mas-data-model = { path = "./crates/data-model/", version = "=1.10.0" }
mas-email = { path = "./crates/email/", version = "=1.10.0" }
mas-graphql = { path = "./crates/graphql/", version = "=1.10.0" }
mas-handlers = { path = "./crates/handlers/", version = "=1.10.0" }
mas-http = { path = "./crates/http/", version = "=1.10.0" }
mas-i18n = { path = "./crates/i18n/", version = "=1.10.0" }
mas-i18n-scan = { path = "./crates/i18n-scan/", version = "=1.10.0" }
mas-iana = { path = "./crates/iana/", version = "=1.10.0" }
mas-iana-codegen = { path = "./crates/iana-codegen/", version = "=1.10.0" }
mas-jose = { path = "./crates/jose/", version = "=1.10.0" }
mas-keystore = { path = "./crates/keystore/", version = "=1.10.0" }
mas-listener = { path = "./crates/listener/", version = "=1.10.0" }
mas-matrix = { path = "./crates/matrix/", version = "=1.10.0" }
mas-matrix-synapse = { path = "./crates/matrix-synapse/", version = "=1.10.0" }
mas-oidc-client = { path = "./crates/oidc-client/", version = "=1.10.0" }
mas-policy = { path = "./crates/policy/", version = "=1.10.0" }
mas-router = { path = "./crates/router/", version = "=1.10.0" }
mas-spa = { path = "./crates/spa/", version = "=1.10.0" }
mas-storage = { path = "./crates/storage/", version = "=1.10.0" }
mas-storage-pg = { path = "./crates/storage-pg/", version = "=1.10.0" }
mas-tasks = { path = "./crates/tasks/", version = "=1.10.0" }
mas-templates = { path = "./crates/templates/", version = "=1.10.0" }
mas-tower = { path = "./crates/tower/", version = "=1.10.0" }
oauth2-types = { path = "./crates/oauth2-types/", version = "=1.10.0" }
syn2mas = { path = "./crates/syn2mas", version = "=1.10.0" }
mas-axum-utils = { path = "./crates/axum-utils/", version = "=1.11.0" }
mas-cli = { path = "./crates/cli/", version = "=1.11.0" }
mas-config = { path = "./crates/config/", version = "=1.11.0" }
mas-context = { path = "./crates/context/", version = "=1.11.0" }
mas-data-model = { path = "./crates/data-model/", version = "=1.11.0" }
mas-email = { path = "./crates/email/", version = "=1.11.0" }
mas-graphql = { path = "./crates/graphql/", version = "=1.11.0" }
mas-handlers = { path = "./crates/handlers/", version = "=1.11.0" }
mas-http = { path = "./crates/http/", version = "=1.11.0" }
mas-i18n = { path = "./crates/i18n/", version = "=1.11.0" }
mas-i18n-scan = { path = "./crates/i18n-scan/", version = "=1.11.0" }
mas-iana = { path = "./crates/iana/", version = "=1.11.0" }
mas-iana-codegen = { path = "./crates/iana-codegen/", version = "=1.11.0" }
mas-jose = { path = "./crates/jose/", version = "=1.11.0" }
mas-keystore = { path = "./crates/keystore/", version = "=1.11.0" }
mas-listener = { path = "./crates/listener/", version = "=1.11.0" }
mas-matrix = { path = "./crates/matrix/", version = "=1.11.0" }
mas-matrix-synapse = { path = "./crates/matrix-synapse/", version = "=1.11.0" }
mas-oidc-client = { path = "./crates/oidc-client/", version = "=1.11.0" }
mas-policy = { path = "./crates/policy/", version = "=1.11.0" }
mas-router = { path = "./crates/router/", version = "=1.11.0" }
mas-spa = { path = "./crates/spa/", version = "=1.11.0" }
mas-storage = { path = "./crates/storage/", version = "=1.11.0" }
mas-storage-pg = { path = "./crates/storage-pg/", version = "=1.11.0" }
mas-tasks = { path = "./crates/tasks/", version = "=1.11.0" }
mas-templates = { path = "./crates/templates/", version = "=1.11.0" }
mas-tower = { path = "./crates/tower/", version = "=1.11.0" }
oauth2-types = { path = "./crates/oauth2-types/", version = "=1.11.0" }
syn2mas = { path = "./crates/syn2mas", version = "=1.11.0" }
# OpenAPI schema generation and validation
[workspace.dependencies.aide]
@@ -71,7 +71,7 @@ features = ["axum", "axum-extra", "axum-extra-query", "axum-json", "macros"]
# An `Arc` that can be atomically updated
[workspace.dependencies.arc-swap]
version = "1.7.1"
version = "1.8.1"
# GraphQL server
[workspace.dependencies.async-graphql]
@@ -124,7 +124,7 @@ features = ["std"]
# Bcrypt password hashing
[workspace.dependencies.bcrypt]
version = "0.17.1"
version = "0.18.0"
default-features = true
# Packed bitfields
@@ -330,7 +330,7 @@ version = "2.0.6"
# Snapshot testing
[workspace.dependencies.insta]
version = "1.43.2"
version = "1.46.3"
features = ["yaml", "json"]
# IP network address types
@@ -567,7 +567,7 @@ features = [
# Sentry error tracking
[workspace.dependencies.sentry]
version = "0.46.0"
version = "0.46.2"
default-features = false
features = ["backtrace", "contexts", "panic", "tower", "reqwest"]
@@ -685,7 +685,7 @@ features = ["cors", "fs", "add-extension", "set-header"]
[workspace.dependencies.tracing]
version = "0.1.41"
[workspace.dependencies.tracing-subscriber]
version = "0.3.20"
version = "0.3.22"
features = ["env-filter"]
[workspace.dependencies.tracing-appender]
version = "0.2.4"
+9 -7
View File
@@ -1,4 +1,5 @@
# syntax = docker/dockerfile:1.7.1
# syntax = docker/dockerfile:1.21.0
# Copyright 2025, 2026 Element Creations Ltd.
# Copyright 2025 New Vector Ltd.
#
# SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
@@ -10,13 +11,14 @@
# being the platform being built.
# The Debian version and version name must be in sync
ARG DEBIAN_VERSION=12
ARG DEBIAN_VERSION_NAME=bookworm
ARG RUSTC_VERSION=1.89.0
ARG NODEJS_VERSION=24.11.0
ARG DEBIAN_VERSION=13
ARG DEBIAN_VERSION_NAME=trixie
# Keep in sync with .github/workflows/ci.yaml
ARG RUSTC_VERSION=1.93.0
ARG NODEJS_VERSION=24.13.0
# Keep in sync with .github/actions/build-policies/action.yml and policies/Makefile
ARG OPA_VERSION=1.8.0
ARG CARGO_AUDITABLE_VERSION=0.7.0
ARG OPA_VERSION=1.13.1
ARG CARGO_AUDITABLE_VERSION=0.7.2
##########################################
## Build stage that builds the frontend ##
-13
View File
@@ -16,19 +16,6 @@ use url::Url;
use super::{ClientSecret, ClientSecretRaw, ConfigurationSection};
#[derive(JsonSchema, Serialize, Deserialize, Clone, Debug)]
#[serde(rename_all = "snake_case")]
pub enum JwksOrJwksUri {
Jwks(PublicJsonWebKeySet),
JwksUri(Url),
}
impl From<PublicJsonWebKeySet> for JwksOrJwksUri {
fn from(jwks: PublicJsonWebKeySet) -> Self {
Self::Jwks(jwks)
}
}
/// Authentication method used by clients
#[derive(JsonSchema, Serialize, Deserialize, Copy, Clone, Debug)]
#[serde(rename_all = "snake_case")]
-9
View File
@@ -14,15 +14,6 @@ use serde::{Deserialize, Serialize, de::Error};
use super::ConfigurationSection;
#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
pub struct Credentials {
/// Username for use to authenticate when connecting to the SMTP server
pub username: String,
/// Password for use to authenticate when connecting to the SMTP server
pub password: String,
}
/// Encryption mode to use
#[derive(Clone, Copy, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "lowercase")]
+774
View File
@@ -0,0 +1,774 @@
// Copyright 2026 Element Creations Ltd.
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.
//! Integration tests for session cleanup jobs.
//!
//! These tests verify that cleanup jobs correctly respect the session hierarchy
//! required for OIDC Backchannel Logout to function properly.
//!
//! Session hierarchy:
//! ```text
//! upstream_oauth_authorization_sessions (matched by sub/sid claims)
//! │ user_session_id
//! ▼
//! user_sessions (browser sessions)
//! │ user_session_id FK
//! ┌────┴──────────────┐
//! │ │
//! ▼ ▼
//! compat_sessions oauth2_sessions
//! ```
use chrono::Duration;
use hyper::{Request, StatusCode};
use mas_data_model::{
BrowserSession, Clock as _, CompatSession, Device, UpstreamOAuthAuthorizationSession,
UpstreamOAuthLink, UpstreamOAuthProvider, UpstreamOAuthProviderClaimsImports,
UpstreamOAuthProviderDiscoveryMode, UpstreamOAuthProviderOnBackchannelLogout,
UpstreamOAuthProviderPkceMode, UpstreamOAuthProviderTokenAuthMethod, User,
};
use mas_iana::jose::JsonWebSignatureAlg;
use mas_jose::jwt::{JsonWebSignatureHeader, Jwt};
use mas_storage::{
RepositoryAccess,
queue::{DeactivateUserJob, QueueJobRepositoryExt},
upstream_oauth2::UpstreamOAuthProviderParams,
};
use oauth2_types::scope::{OPENID, Scope};
use sqlx::PgPool;
use wiremock::{
Mock, MockServer, ResponseTemplate,
matchers::{method, path},
};
use crate::test_utils::{RequestBuilderExt, ResponseExt, TestState, setup};
/// Helper struct to hold all the entities created for testing the session
/// hierarchy.
struct TestSessionHierarchy {
user: User,
browser_session: BrowserSession,
compat_session: Option<CompatSession>,
oauth2_session: Option<mas_data_model::Session>,
upstream_session: Option<UpstreamOAuthAuthorizationSession>,
#[expect(dead_code)]
upstream_link: Option<UpstreamOAuthLink>,
provider: Option<UpstreamOAuthProvider>,
#[expect(dead_code)]
mock_server: MockServer,
}
const UPSTREAM_OAUTH_ISSUER: &str = "https://idp.example.com";
const UPSTREAM_OAUTH_CLIENT_ID: &str = "test-client";
const UPSTREAM_OAUTH_SESSION_ID: &str = "upstream-oauth-session-id";
const UPSTREAM_OAUTH_SUBJECT: &str = "upstream-oauth-sub";
/// Create the complete session hierarchy for testing.
///
/// This creates:
/// - A user
/// - A browser session
/// - Optionally a compat session linked to the browser session
/// - Optionally an OAuth 2.0 session linked to the browser session
/// - Optionally an upstream OAuth session linked to the browser session
async fn create_session_hierarchy(
state: &TestState,
with_compat: bool,
with_oauth2: bool,
with_upstream: bool,
) -> TestSessionHierarchy {
let mut rng = state.rng();
let mut repo = state.repository().await.unwrap();
// Start a mock server to answer to JWKS requests for the backchannel logout
// tests
let mock_server = MockServer::start().await;
let jwks = state.key_store.public_jwks();
let mock_jwks = Mock::given(method("GET"))
.and(path("jwks.json"))
.respond_with(ResponseTemplate::new(200).set_body_json(jwks));
mock_server.register(mock_jwks).await;
// Create user
let user = repo
.user()
.add(&mut rng, &state.clock, "testuser".to_owned())
.await
.unwrap();
// Create browser session
let browser_session = repo
.browser_session()
.add(&mut rng, &state.clock, &user, None)
.await
.unwrap();
// Create compat session if requested
let compat_session = if with_compat {
let device = Device::generate(&mut rng);
let session = repo
.compat_session()
.add(
&mut rng,
&state.clock,
&user,
device,
Some(&browser_session),
false,
None,
)
.await
.unwrap();
Some(session)
} else {
None
};
// Create OAuth2 session if requested
let oauth2_session = if with_oauth2 {
// First create an OAuth2 client
let client = repo
.oauth2_client()
.add(
&mut rng,
&state.clock,
vec!["https://example.com/callback".parse().unwrap()],
None,
None,
None,
vec![],
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
)
.await
.unwrap();
let session = repo
.oauth2_session()
.add_from_browser_session(
&mut rng,
&state.clock,
&client,
&browser_session,
Scope::from_iter([OPENID]),
)
.await
.unwrap();
Some(session)
} else {
None
};
// Create upstream OAuth session if requested
let (provider, upstream_link, upstream_session) = if with_upstream {
let params = UpstreamOAuthProviderParams {
issuer: Some(UPSTREAM_OAUTH_ISSUER.to_owned()),
human_name: Some("Test IdP".to_owned()),
brand_name: None,
scope: Scope::from_iter([OPENID]),
token_endpoint_auth_method: UpstreamOAuthProviderTokenAuthMethod::ClientSecretBasic,
token_endpoint_signing_alg: None,
id_token_signed_response_alg: JsonWebSignatureAlg::Rs256,
fetch_userinfo: false,
userinfo_signed_response_alg: None,
client_id: UPSTREAM_OAUTH_CLIENT_ID.to_owned(),
encrypted_client_secret: None,
claims_imports: UpstreamOAuthProviderClaimsImports::default(),
authorization_endpoint_override: None,
token_endpoint_override: None,
userinfo_endpoint_override: None,
// Point to the mock server to have it use a JWKS we can use for signing
jwks_uri_override: Some(format!("{}/jwks.json", mock_server.uri()).parse().unwrap()),
discovery_mode: UpstreamOAuthProviderDiscoveryMode::Disabled,
pkce_mode: UpstreamOAuthProviderPkceMode::Auto,
response_mode: None,
additional_authorization_parameters: vec![],
forward_login_hint: false,
ui_order: 0,
on_backchannel_logout: UpstreamOAuthProviderOnBackchannelLogout::LogoutAll,
};
let provider = repo
.upstream_oauth_provider()
.add(&mut rng, &state.clock, params)
.await
.unwrap();
// Create a link
let link = repo
.upstream_oauth_link()
.add(
&mut rng,
&state.clock,
&provider,
UPSTREAM_OAUTH_SUBJECT.to_owned(),
Some("test@idp.example.com".to_owned()),
)
.await
.unwrap();
// Associate link to user
repo.upstream_oauth_link()
.associate_to_user(&link, &user)
.await
.unwrap();
// Create an upstream session
let session = repo
.upstream_oauth_session()
.add(
&mut rng,
&state.clock,
&provider,
"state123".to_owned(),
Some("verifier123".to_owned()),
Some("nonce123".to_owned()),
)
.await
.unwrap();
// Complete the session with the link and ID token claims (including sub and
// sid)
let id_token_claims = serde_json::json!({
"sub": UPSTREAM_OAUTH_SUBJECT,
"sid": UPSTREAM_OAUTH_SESSION_ID,
});
let session = repo
.upstream_oauth_session()
.complete_with_link(
&state.clock,
session,
&link,
Some("fake-id-token".to_owned()),
Some(id_token_claims),
None,
None,
)
.await
.unwrap();
// Consume the session and link it to the browser session
let session = repo
.upstream_oauth_session()
.consume(&state.clock, session, &browser_session)
.await
.unwrap();
(Some(provider), Some(link), Some(session))
} else {
(None, None, None)
};
repo.save().await.unwrap();
TestSessionHierarchy {
user,
browser_session,
compat_session,
oauth2_session,
upstream_session,
upstream_link,
provider,
mock_server,
}
}
/// Test that sessions finished less than 30 days ago are NOT deleted.
#[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
async fn test_cleanup_sessions_within_retention_preserved(pool: PgPool) {
setup();
let state = TestState::from_pool(pool).await.unwrap();
// Create a full hierarchy with all session types
let hierarchy = create_session_hierarchy(&state, true, true, false).await;
// Finish all sessions
let mut repo = state.repository().await.unwrap();
let browser_session = repo
.browser_session()
.finish(&state.clock, hierarchy.browser_session)
.await
.unwrap();
let compat_session = repo
.compat_session()
.finish(&state.clock, hierarchy.compat_session.unwrap())
.await
.unwrap();
let oauth2_session = repo
.oauth2_session()
.finish(&state.clock, hierarchy.oauth2_session.unwrap())
.await
.unwrap();
repo.save().await.unwrap();
// Wait one day and run the cleanup jobs a few times
state.clock.advance(Duration::try_days(1).unwrap());
state.run_jobs_in_queue().await;
for _ in 0..5 {
state.clock.advance(Duration::try_hours(1).unwrap());
state.run_jobs_in_queue().await;
}
// Verify all sessions still exist
let mut repo = state.repository().await.unwrap();
assert!(
repo.browser_session()
.lookup(browser_session.id)
.await
.unwrap()
.is_some(),
"Browser session should still exist"
);
assert!(
repo.compat_session()
.lookup(compat_session.id)
.await
.unwrap()
.is_some(),
"Compat session should still exist"
);
assert!(
repo.oauth2_session()
.lookup(oauth2_session.id)
.await
.unwrap()
.is_some(),
"OAuth2 session should still exist"
);
}
/// Test that deactivated users eventually get all their sessions cleaned up
#[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
async fn test_cleanup_deactivated_users(pool: PgPool) {
setup();
let state = TestState::from_pool(pool).await.unwrap();
let mut rng = state.rng();
// Create a hierarchy with all session types
let hierarchy = create_session_hierarchy(&state, true, true, true).await;
// Deactivate the user
let mut repo = state.repository().await.unwrap();
let user = repo
.user()
.lookup(hierarchy.user.id)
.await
.unwrap()
.unwrap();
let user = repo.user().deactivate(&state.clock, user).await.unwrap();
repo.queue_job()
.schedule_job(&mut rng, &state.clock, DeactivateUserJob::new(&user, false))
.await
.unwrap();
repo.save().await.unwrap();
state.run_jobs_in_queue().await;
// Verify all sessions are finished
let mut repo = state.repository().await.unwrap();
assert!(
repo.compat_session()
.lookup(hierarchy.compat_session.as_ref().unwrap().id)
.await
.unwrap()
.unwrap()
.is_finished(),
"Compat session should be finished"
);
assert!(
repo.oauth2_session()
.lookup(hierarchy.oauth2_session.as_ref().unwrap().id)
.await
.unwrap()
.unwrap()
.is_finished(),
"OAuth2 session should be finished"
);
assert!(
repo.browser_session()
.lookup(hierarchy.browser_session.id)
.await
.unwrap()
.unwrap()
.finished_at
.is_some(),
"Browser session should be there"
);
assert!(
repo.upstream_oauth_session()
.lookup(hierarchy.upstream_session.as_ref().unwrap().id)
.await
.unwrap()
.is_some(),
"Upstream OAuth session should be there"
);
// Wait 31 days and run the cleanup jobs a few times
state.clock.advance(Duration::try_days(31).unwrap());
state.run_jobs_in_queue().await;
for _ in 0..5 {
state.clock.advance(Duration::try_hours(1).unwrap());
state.run_jobs_in_queue().await;
}
// Verify all sessions are deleted
let mut repo = state.repository().await.unwrap();
assert!(
repo.compat_session()
.lookup(hierarchy.compat_session.as_ref().unwrap().id)
.await
.unwrap()
.is_none(),
"Compat session should be deleted"
);
assert!(
repo.oauth2_session()
.lookup(hierarchy.oauth2_session.as_ref().unwrap().id)
.await
.unwrap()
.is_none(),
"OAuth2 session should be deleted"
);
assert!(
repo.browser_session()
.lookup(hierarchy.browser_session.id)
.await
.unwrap()
.is_none(),
"Browser session should be deleted"
);
assert!(
repo.upstream_oauth_session()
.lookup(hierarchy.upstream_session.as_ref().unwrap().id)
.await
.unwrap()
.is_none(),
"Upstream OAuth session should be deleted"
);
}
/// Test that sessions finished more than 30 days ago ARE deleted.
#[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
async fn test_cleanup_sessions_after_retention_deleted(pool: PgPool) {
setup();
let state = TestState::from_pool(pool).await.unwrap();
// Create hierarchy with compat and oauth2 sessions
let hierarchy = create_session_hierarchy(&state, true, true, false).await;
// Finish all sessions
let mut repo = state.repository().await.unwrap();
let browser_session = repo
.browser_session()
.finish(&state.clock, hierarchy.browser_session)
.await
.unwrap();
let compat_session = repo
.compat_session()
.finish(&state.clock, hierarchy.compat_session.unwrap())
.await
.unwrap();
let oauth2_session = repo
.oauth2_session()
.finish(&state.clock, hierarchy.oauth2_session.unwrap())
.await
.unwrap();
repo.save().await.unwrap();
// Wait 31 days and run the cleanup jobs a few times
state.clock.advance(Duration::try_days(31).unwrap());
state.run_jobs_in_queue().await;
for _ in 0..5 {
state.clock.advance(Duration::try_hours(1).unwrap());
state.run_jobs_in_queue().await;
}
// Verify all sessions are deleted
let mut repo = state.repository().await.unwrap();
assert!(
repo.compat_session()
.lookup(compat_session.id)
.await
.unwrap()
.is_none(),
"Compat session should be deleted"
);
assert!(
repo.oauth2_session()
.lookup(oauth2_session.id)
.await
.unwrap()
.is_none(),
"OAuth2 session should be deleted"
);
// Browser session should also be deleted since children are gone
assert!(
repo.browser_session()
.lookup(browser_session.id)
.await
.unwrap()
.is_none(),
"Browser session should be deleted after children are gone"
);
}
/// Test that user sessions with remaining child sessions are NOT deleted.
#[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
async fn test_cleanup_user_session_blocked_by_child_sessions(pool: PgPool) {
setup();
let state = TestState::from_pool(pool).await.unwrap();
// Create hierarchy with compat session only
let hierarchy = create_session_hierarchy(&state, true, false, false).await;
// Finish only the browser session (not the compat session)
let mut repo = state.repository().await.unwrap();
let browser_session = repo
.browser_session()
.finish(&state.clock, hierarchy.browser_session)
.await
.unwrap();
repo.save().await.unwrap();
let compat_session_id = hierarchy.compat_session.as_ref().unwrap().id;
// Wait 31 days and run the cleanup jobs a few times
state.clock.advance(Duration::try_days(31).unwrap());
state.run_jobs_in_queue().await;
for _ in 0..5 {
state.clock.advance(Duration::try_hours(1).unwrap());
state.run_jobs_in_queue().await;
}
// Verify browser session still exists because compat session is still active
let mut repo = state.repository().await.unwrap();
assert!(
repo.browser_session()
.lookup(browser_session.id)
.await
.unwrap()
.is_some(),
"Browser session should NOT be deleted because it has an active child session"
);
assert!(
repo.compat_session()
.lookup(compat_session_id)
.await
.unwrap()
.is_some(),
"Compat session should still exist (not finished)"
);
}
/// Test that backchannel logout can find sessions before cleanup.
#[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
async fn test_backchannel_logout_works_before_cleanup(pool: PgPool) {
setup();
let state = TestState::from_pool(pool).await.unwrap();
// Create hierarchy with upstream session
let hierarchy = create_session_hierarchy(&state, true, true, true).await;
let provider = hierarchy.provider.as_ref().unwrap();
// The edge case we're trying to make works, is that if the browser session
// is finished for 30 days but *not* the child sessions, that browser
// session and the upstream sessions stay there so that backchannel logout
// still works
let mut repo = state.repository().await.unwrap();
let browser_session = repo
.browser_session()
.lookup(hierarchy.browser_session.id)
.await
.unwrap()
.unwrap();
repo.browser_session()
.finish(&state.clock, browser_session)
.await
.unwrap();
repo.save().await.unwrap();
// Now wait 31 days and run the cleanup jobs a few times
state.clock.advance(Duration::try_days(31).unwrap());
state.run_jobs_in_queue().await;
for _ in 0..5 {
state.clock.advance(Duration::try_hours(1).unwrap());
state.run_jobs_in_queue().await;
}
// Now let's craft a backchannel logout request
let ts = state.clock.now().timestamp();
let payload = serde_json::json!({
"iss": UPSTREAM_OAUTH_ISSUER,
"aud": UPSTREAM_OAUTH_CLIENT_ID,
"sub": UPSTREAM_OAUTH_SUBJECT,
"sid": UPSTREAM_OAUTH_SESSION_ID,
"jti": "iswearthisisrandom",
"iat": ts,
"exp": ts + 300,
"events": {
"http://schemas.openid.net/event/backchannel-logout": {}
}
});
let key = state
.key_store
.signing_key_for_algorithm(&JsonWebSignatureAlg::Rs256)
.unwrap();
let signer = key
.params()
.signing_key_for_alg(&JsonWebSignatureAlg::Rs256)
.unwrap();
let signed = Jwt::sign(
JsonWebSignatureHeader::new(JsonWebSignatureAlg::Rs256),
payload,
&signer,
)
.unwrap();
let request = Request::post(format!("/upstream/backchannel-logout/{}", provider.id)).form(
serde_json::json!({
"logout_token": signed.as_str(),
}),
);
let response = state.request(request).await;
response.assert_status(StatusCode::OK);
// The session should still exist, but are finished
let mut repo = state.repository().await.unwrap();
assert!(
!repo
.browser_session()
.lookup(hierarchy.browser_session.id)
.await
.unwrap()
.unwrap()
.active(),
"Inactive browser session should not be cleaned up"
);
assert!(
repo.compat_session()
.lookup(hierarchy.compat_session.as_ref().unwrap().id)
.await
.unwrap()
.unwrap()
.is_finished(),
"Active compat session should not be cleaned up"
);
assert!(
repo.oauth2_session()
.lookup(hierarchy.oauth2_session.as_ref().unwrap().id)
.await
.unwrap()
.unwrap()
.is_finished(),
"Active OAuth2 session should not be cleaned up"
);
// Wait again, then the sessions should be completely deleted
state.clock.advance(Duration::try_days(31).unwrap());
state.run_jobs_in_queue().await;
for _ in 0..5 {
state.clock.advance(Duration::try_hours(1).unwrap());
state.run_jobs_in_queue().await;
}
let mut repo = state.repository().await.unwrap();
assert!(
repo.browser_session()
.lookup(hierarchy.browser_session.id)
.await
.unwrap()
.is_none(),
"Browser session should be deleted"
);
assert!(
repo.compat_session()
.lookup(hierarchy.compat_session.as_ref().unwrap().id)
.await
.unwrap()
.is_none(),
"Compat session should be deleted"
);
assert!(
repo.oauth2_session()
.lookup(hierarchy.oauth2_session.as_ref().unwrap().id)
.await
.unwrap()
.is_none(),
"OAuth2 session should be deleted"
);
assert!(
repo.upstream_oauth_session()
.lookup(hierarchy.upstream_session.as_ref().unwrap().id)
.await
.unwrap()
.is_none(),
"Upstream OAuth session should be deleted"
);
}
/// Test that active sessions are not cleaned up even after retention period.
#[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
async fn test_active_sessions_not_cleaned_up(pool: PgPool) {
setup();
let state = TestState::from_pool(pool).await.unwrap();
// Create hierarchy without finishing any sessions
let hierarchy = create_session_hierarchy(&state, true, true, false).await;
let browser_session_id = hierarchy.browser_session.id;
let compat_session_id = hierarchy.compat_session.as_ref().unwrap().id;
let oauth2_session_id = hierarchy.oauth2_session.as_ref().unwrap().id;
// Wait 31 days and run the cleanup jobs a few times
state.clock.advance(Duration::try_days(31).unwrap());
state.run_jobs_in_queue().await;
for _ in 0..5 {
state.clock.advance(Duration::try_hours(1).unwrap());
state.run_jobs_in_queue().await;
}
// All sessions should still exist because they're active
let mut repo = state.repository().await.unwrap();
assert!(
repo.browser_session()
.lookup(browser_session_id)
.await
.unwrap()
.is_some(),
"Active browser session should not be cleaned up"
);
assert!(
repo.compat_session()
.lookup(compat_session_id)
.await
.unwrap()
.is_some(),
"Active compat session should not be cleaned up"
);
assert!(
repo.oauth2_session()
.lookup(oauth2_session_id)
.await
.unwrap()
.is_some(),
"Active OAuth2 session should not be cleaned up"
);
}
@@ -83,7 +83,7 @@ pub async fn get(
Some(CompatLoginSsoAction::Register) => {
url_builder.redirect(&mas_router::Register::and_continue_compat_sso_login(id))
}
Some(CompatLoginSsoAction::Login) | None => {
Some(CompatLoginSsoAction::Login | CompatLoginSsoAction::Unknown) | None => {
url_builder.redirect(&mas_router::Login::and_continue_compat_sso_login(id))
}
};
@@ -224,7 +224,7 @@ pub async fn post(
Some(CompatLoginSsoAction::Register) => {
url_builder.redirect(&mas_router::Register::and_continue_compat_sso_login(id))
}
Some(CompatLoginSsoAction::Login) | None => {
Some(CompatLoginSsoAction::Login | CompatLoginSsoAction::Unknown) | None => {
url_builder.redirect(&mas_router::Login::and_continue_compat_sso_login(id))
}
};
@@ -13,7 +13,6 @@ use mas_router::{CompatLoginSsoAction, CompatLoginSsoComplete, UrlBuilder};
use mas_storage::{BoxRepository, compat::CompatSsoLoginRepository};
use rand::distributions::{Alphanumeric, DistString};
use serde::Deserialize;
use serde_with::serde;
use thiserror::Error;
use url::Url;
@@ -23,12 +22,21 @@ use crate::impl_from_error_for_route;
pub struct Params {
#[serde(rename = "redirectUrl")]
redirect_url: Option<String>,
action: Option<CompatLoginSsoAction>,
#[serde(rename = "org.matrix.msc3824.action")]
unstable_action: Option<CompatLoginSsoAction>,
}
impl Params {
fn action(&self) -> Option<CompatLoginSsoAction> {
self.action
.filter(CompatLoginSsoAction::is_known)
.or(self.unstable_action.filter(CompatLoginSsoAction::is_known))
}
}
#[derive(Debug, Error)]
pub enum RouteError {
#[error(transparent)]
@@ -62,6 +70,8 @@ pub async fn get(
State(url_builder): State<UrlBuilder>,
Query(params): Query<Params>,
) -> Result<impl IntoResponse, RouteError> {
let action = params.action();
// Check the redirectUrl parameter
let redirect_url = params.redirect_url.ok_or(RouteError::MissingRedirectUrl)?;
let redirect_url = Url::parse(&redirect_url).map_err(|_| RouteError::InvalidRedirectUrl)?;
@@ -84,10 +94,7 @@ pub async fn get(
repo.save().await?;
Ok(url_builder.absolute_redirect(&CompatLoginSsoComplete::new(
login.id,
params.action.or(params.unstable_action),
)))
Ok(url_builder.absolute_redirect(&CompatLoginSsoComplete::new(login.id, action)))
}
#[cfg(test)]
@@ -121,4 +128,29 @@ mod tests {
assert!(location.contains("org.matrix.msc3824.action=register"));
assert!(location.contains("action=register"));
}
#[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
async fn test_unknown_action(pool: PgPool) {
let state: TestState = TestState::from_pool(pool).await.unwrap();
let request = Request::get(
"/_matrix/client/v3/login/sso/redirect?\
redirectUrl=http://example.com/\
&org.matrix.msc3824.action=undefinedaction",
)
.empty();
let response = state.request(request).await;
response.assert_status(StatusCode::SEE_OTHER);
let location = response
.headers()
.get("Location")
.unwrap()
.to_str()
.unwrap();
assert!(!location.contains("org.matrix.msc3824.action"));
assert!(!location.contains("action"));
}
}
+3
View File
@@ -25,6 +25,9 @@ pub(crate) mod logout;
pub(crate) mod logout_all;
pub(crate) mod refresh;
#[cfg(test)]
mod tests;
#[derive(Debug, Serialize)]
struct MatrixError {
errcode: &'static str,
+1 -1
View File
@@ -156,7 +156,7 @@ pub(crate) async fn post(
.await?;
repo.compat_refresh_token()
.consume(&clock, refresh_token)
.consume_and_replace(&clock, refresh_token, &new_refresh_token)
.await?;
if let Some(access_token) = access_token {
+233
View File
@@ -0,0 +1,233 @@
// Copyright 2026 Element Creations Ltd.
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.
use axum::http::{Request, StatusCode};
use mas_matrix::{HomeserverConnection, ProvisionRequest};
use serde::{Deserialize, Serialize};
use zeroize::Zeroizing;
use crate::test_utils::{RequestBuilderExt, ResponseExt, TestState, setup};
#[derive(Debug, Serialize)]
#[serde(tag = "type")]
enum LoginCredentials {
#[serde(rename = "m.login.password")]
Password {
identifier: LoginIdentifier,
password: String,
},
}
#[derive(Debug, Serialize)]
#[serde(tag = "type")]
enum LoginIdentifier {
#[serde(rename = "m.id.user")]
User { user: String },
}
#[derive(Debug, Serialize)]
struct LoginRequest {
#[serde(flatten)]
credentials: LoginCredentials,
#[serde(default)]
refresh_token: bool,
}
#[derive(Debug, Deserialize)]
struct LoginResponse {
#[allow(dead_code)]
access_token: String,
#[allow(dead_code)]
user_id: String,
#[allow(dead_code)]
device_id: Option<String>,
refresh_token: Option<String>,
}
#[derive(Debug, Serialize)]
struct RefreshRequest {
refresh_token: String,
}
#[derive(Debug, Deserialize, PartialEq, Eq)]
struct RefreshResponse {
access_token: String,
refresh_token: String,
expires_in_ms: i64,
}
/// Test using a compatibility refresh token.
#[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
async fn test_compat_refresh(pool: sqlx::PgPool) {
setup();
let state = TestState::from_pool(pool).await.unwrap();
// Create a user
create_test_user(&state, "testuser").await;
// Login to get initial tokens
let login_request = Request::post("/_matrix/client/v3/login").json(&LoginRequest {
credentials: LoginCredentials::Password {
identifier: LoginIdentifier::User {
user: "testuser".to_owned(),
},
password: "password".to_owned(),
},
refresh_token: true,
});
let login_response = state.request(login_request).await;
login_response.assert_status(StatusCode::OK);
let login_response: LoginResponse = login_response.json();
let initial_refresh_token = login_response
.refresh_token
.expect("Login should return a refresh token");
// First refresh
let refresh_request = Request::post("/_matrix/client/v3/refresh").json(&RefreshRequest {
refresh_token: initial_refresh_token.clone(),
});
let first_refresh_response = state.request(refresh_request).await;
first_refresh_response.assert_status(StatusCode::OK);
let first_refresh_response: RefreshResponse = first_refresh_response.json();
let first_new_refresh_token = first_refresh_response.refresh_token.clone();
assert_eq!(
first_refresh_response,
RefreshResponse {
access_token: "mct_fNbm5KAQovodfVQz7IvDc44woP66fR_fsaiD1".to_owned(),
refresh_token: "mcr_42oTpLoieH5IecxG6gZXyvelQWW9Xq_a8g5N3".to_owned(),
expires_in_ms: 300_000
}
);
// Use the token from the /refresh response to /refresh again,
// proving that it works.
// This is a regression test: we were previously consuming the refresh token
// before it was returned from /refresh.
let second_refresh_request =
Request::post("/_matrix/client/v3/refresh").json(&RefreshRequest {
refresh_token: first_new_refresh_token.clone(),
});
let second_refresh_response = state.request(second_refresh_request).await;
second_refresh_response.assert_status(StatusCode::OK);
let second_refresh_response: RefreshResponse = second_refresh_response.json();
assert_eq!(
second_refresh_response,
RefreshResponse {
access_token: "mct_Wc6Hx4l9DGzqGtgLoYqtrtBUBcWlE4_ZFyTp2".to_owned(),
refresh_token: "mcr_Yp7FM44zJN5qePGMLvvMXC4Ds1A3lC_0YcYCM".to_owned(),
expires_in_ms: 300_000
}
);
}
#[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
async fn test_refresh_with_invalid_token(pool: sqlx::PgPool) {
setup();
let state = TestState::from_pool(pool).await.unwrap();
let refresh_request = RefreshRequest {
refresh_token: "invalid_token".to_owned(),
};
let refresh_request = Request::post("/_matrix/client/v3/refresh").json(&refresh_request);
let response = state.request(refresh_request).await;
response.assert_status(StatusCode::UNAUTHORIZED);
}
#[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
async fn test_refresh_with_consumed_token(pool: sqlx::PgPool) {
setup();
let state = TestState::from_pool(pool).await.unwrap();
// Create a user and login
create_test_user(&state, "testuser").await;
let login_request = LoginRequest {
credentials: LoginCredentials::Password {
identifier: LoginIdentifier::User {
user: "testuser".to_owned(),
},
password: "password".to_owned(),
},
refresh_token: true,
};
let login_request = Request::post("/_matrix/client/v3/login").json(&login_request);
let login_response = state.request(login_request).await;
login_response.assert_status(StatusCode::OK);
let login_response: LoginResponse = login_response.json();
let refresh_token = login_response
.refresh_token
.expect("Login should return a refresh token");
let refresh_request = RefreshRequest {
refresh_token: refresh_token.clone(),
};
// Use the refresh token once
let first_refresh_request = Request::post("/_matrix/client/v3/refresh").json(&refresh_request);
let first_refresh_response = state.request(first_refresh_request).await;
first_refresh_response.assert_status(StatusCode::OK);
let _first_refresh_response: RefreshResponse = first_refresh_response.json();
// Try to use the same refresh token again - should fail because it's consumed
let second_refresh_request = Request::post("/_matrix/client/v3/refresh").json(&refresh_request);
let second_refresh_response = state.request(second_refresh_request).await;
second_refresh_response.assert_status(StatusCode::UNAUTHORIZED);
}
async fn create_test_user(state: &TestState, username: &str) -> mas_data_model::User {
let mut repo = state.repository().await.unwrap();
let mut rng = state.rng();
let user = repo
.user()
.add(&mut rng, &state.clock, username.to_owned())
.await
.unwrap();
let password = Zeroizing::new("password".to_owned());
let (version, hashed_password) = state
.password_manager
.hash(&mut rng, password)
.await
.unwrap();
repo.user_password()
.add(
&mut rng,
&state.clock,
&user,
version,
hashed_password,
None,
)
.await
.unwrap();
// Provision the user on the homeserver
state
.homeserver_connection
.provision_user(&ProvisionRequest::new(&user.username, &user.sub))
.await
.unwrap();
repo.save().await.unwrap();
user
}
+2
View File
@@ -62,6 +62,8 @@ mod views;
mod activity_tracker;
mod captcha;
#[cfg(test)]
mod cleanup_tests;
mod preferred_language;
mod rate_limit;
mod session;
+10
View File
@@ -628,6 +628,16 @@ impl SimpleRoute for CompatLoginSsoRedirectIdp {
pub enum CompatLoginSsoAction {
Login,
Register,
#[serde(other)]
Unknown,
}
impl CompatLoginSsoAction {
/// Returns true if the action is a known action.
#[must_use]
pub fn is_known(&self) -> bool {
!matches!(self, Self::Unknown)
}
}
#[derive(Debug, Serialize, Deserialize, Clone, Copy)]
@@ -1,15 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE compat_refresh_tokens\n SET consumed_at = $2\n WHERE compat_session_id = $1\n AND consumed_at IS NULL\n ",
"query": "\n UPDATE compat_refresh_tokens\n SET consumed_at = $2\n WHERE compat_session_id = $1\n AND consumed_at IS NULL\n AND compat_refresh_token_id <> $3\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Timestamptz"
"Timestamptz",
"Uuid"
]
},
"nullable": []
},
"hash": "f75e44b528234dac708640ad9a111f3f6b468a91bf0d5b574795bf8c80605f19"
"hash": "4e64540bbffe5f4b9c4a6589012cf69eb67adaa4d40fc1910dfcd2640e32ab37"
}
+17 -4
View File
@@ -437,6 +437,7 @@ mod tests {
async fn test_refresh_token_repository(pool: PgPool) {
const ACCESS_TOKEN: &str = "access_token";
const REFRESH_TOKEN: &str = "refresh_token";
const REFRESH_TOKEN2: &str = "refresh_token2";
let mut rng = ChaChaRng::seed_from_u64(42);
let clock = MockClock::default();
let mut repo = PgRepository::from_pool(&pool).await.unwrap().boxed();
@@ -508,16 +509,28 @@ mod tests {
assert!(refresh_token_lookup.is_valid());
assert!(!refresh_token_lookup.is_consumed());
// Consume it
// Consume the first token, but to do so we need a 2nd to replace it with
let refresh_token2 = repo
.compat_refresh_token()
.add(
&mut rng,
&clock,
&session,
&access_token,
REFRESH_TOKEN2.to_owned(),
)
.await
.unwrap();
let refresh_token = repo
.compat_refresh_token()
.consume(&clock, refresh_token)
.consume_and_replace(&clock, refresh_token, &refresh_token2)
.await
.unwrap();
assert!(!refresh_token.is_valid());
assert!(refresh_token.is_consumed());
// Reload it and check again
// Reload the first token and check again
let refresh_token_lookup = repo
.compat_refresh_token()
.find_by_token(REFRESH_TOKEN)
@@ -530,7 +543,7 @@ mod tests {
// Consuming it again should not work
assert!(
repo.compat_refresh_token()
.consume(&clock, refresh_token)
.consume_and_replace(&clock, refresh_token, &refresh_token2)
.await
.is_err()
);
+10 -2
View File
@@ -185,20 +185,26 @@ impl CompatRefreshTokenRepository for PgCompatRefreshTokenRepository<'_> {
}
#[tracing::instrument(
name = "db.compat_refresh_token.consume",
name = "db.compat_refresh_token.consume_and_replace",
skip_all,
fields(
db.query.text,
%compat_refresh_token.id,
%successor_compat_refresh_token.id,
compat_session.id = %compat_refresh_token.session_id,
),
err,
)]
async fn consume(
async fn consume_and_replace(
&mut self,
clock: &dyn Clock,
compat_refresh_token: CompatRefreshToken,
successor_compat_refresh_token: &CompatRefreshToken,
) -> Result<CompatRefreshToken, Self::Error> {
if compat_refresh_token.session_id != successor_compat_refresh_token.session_id {
return Err(DatabaseError::invalid_operation());
}
let consumed_at = clock.now();
let res = sqlx::query!(
r#"
@@ -206,9 +212,11 @@ impl CompatRefreshTokenRepository for PgCompatRefreshTokenRepository<'_> {
SET consumed_at = $2
WHERE compat_session_id = $1
AND consumed_at IS NULL
AND compat_refresh_token_id <> $3
"#,
Uuid::from(compat_refresh_token.session_id),
consumed_at,
Uuid::from(successor_compat_refresh_token.id),
)
.traced()
.execute(&mut *self.conn)
+18 -7
View File
@@ -69,16 +69,22 @@ pub trait CompatRefreshTokenRepository: Send + Sync {
token: String,
) -> Result<CompatRefreshToken, Self::Error>;
/// Consume a compat refresh token.
/// Consume the given compat refresh token, as well as all other refresh
/// tokens from the same session, except for the given successor compat
/// refresh token.
///
/// This also marks other refresh tokens in the same session as consumed.
/// The given successor refresh token will thereafter be the only valid
/// refresh token for the session.
///
/// # Historical context
///
/// When using a refresh token, we must be able to mark multiple other
/// refresh tokens in the same session as consumed.
/// This is desirable because the syn2mas migration process can import
/// multiple refresh tokens for one device (compat session).
/// But once the user uses one of those, the others should no longer
/// be valid.
///
/// Returns the consumed compat refresh token
///
/// # Parameters
///
/// * `clock`: The clock used to generate timestamps
@@ -86,11 +92,15 @@ pub trait CompatRefreshTokenRepository: Send + Sync {
///
/// # Errors
///
/// Returns [`Self::Error`] if the underlying repository fails
async fn consume(
/// - Returns [`Self::Error`] if the underlying repository fails
/// - Returns an error if `compat_refresh_token` is not valid to be
/// consumed.
/// - Returns an error if no refresh tokens would be consumed.
async fn consume_and_replace(
&mut self,
clock: &dyn Clock,
compat_refresh_token: CompatRefreshToken,
successor_compat_refresh_token: &CompatRefreshToken,
) -> Result<CompatRefreshToken, Self::Error>;
}
@@ -111,9 +121,10 @@ repository_impl!(CompatRefreshTokenRepository:
token: String,
) -> Result<CompatRefreshToken, Self::Error>;
async fn consume(
async fn consume_and_replace(
&mut self,
clock: &dyn Clock,
compat_refresh_token: CompatRefreshToken,
successor_compat_refresh_token: &CompatRefreshToken,
) -> Result<CompatRefreshToken, Self::Error>;
);
+88
View File
@@ -0,0 +1,88 @@
// Copyright 2026 Element Creations Ltd.
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.
//! Miscellaneous cleanup tasks
use std::time::Duration;
use async_trait::async_trait;
use mas_storage::queue::{CleanupQueueJobsJob, PruneStalePolicyDataJob};
use tracing::{debug, info};
use ulid::Ulid;
use super::BATCH_SIZE;
use crate::{
State,
new_queue::{JobContext, JobError, RunnableJob},
};
#[async_trait]
impl RunnableJob for CleanupQueueJobsJob {
#[tracing::instrument(name = "job.cleanup_queue_jobs", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove completed and failed queue jobs after 30 days.
// Keep them for debugging purposes.
let until = state.clock.now() - chrono::Duration::days(30);
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, cursor) = repo
.queue_job()
.cleanup(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no queue jobs to clean up");
} else {
info!(count = total, "cleaned up queue jobs");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for PruneStalePolicyDataJob {
#[tracing::instrument(name = "job.prune_stale_policy_data", skip_all)]
async fn run(&self, state: &State, _context: JobContext) -> Result<(), JobError> {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// Keep the last 10 policy data
let count = repo
.policy_data()
.prune(10)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
if count == 0 {
debug!("no stale policy data to prune");
} else {
info!(count, "pruned stale policy data");
}
Ok(())
}
}
+24
View File
@@ -0,0 +1,24 @@
// Copyright 2026 Element Creations Ltd.
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.
//! Database cleanup tasks
//!
//! This module contains tasks for cleaning up old data from the database.
//! Tasks are grouped by domain:
//!
//! - [`tokens`]: OAuth token cleanup (access and refresh tokens)
//! - [`sessions`]: Session cleanup (compat, `OAuth2`, user sessions and their
//! IPs)
//! - [`oauth`]: OAuth grants and upstream OAuth cleanup
//! - [`user`]: User-related cleanup (registrations, recovery, email auth)
//! - [`misc`]: Miscellaneous cleanup (queue jobs, policy data)
mod misc;
mod oauth;
mod sessions;
mod tokens;
mod user;
pub(crate) const BATCH_SIZE: usize = 1000;
+216
View File
@@ -0,0 +1,216 @@
// Copyright 2026 Element Creations Ltd.
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.
//! OAuth grants and upstream OAuth cleanup tasks
use std::time::Duration;
use async_trait::async_trait;
use mas_storage::queue::{
CleanupOAuthAuthorizationGrantsJob, CleanupOAuthDeviceCodeGrantsJob,
CleanupUpstreamOAuthLinksJob, CleanupUpstreamOAuthSessionsJob,
};
use tracing::{debug, info};
use ulid::Ulid;
use super::BATCH_SIZE;
use crate::{
State,
new_queue::{JobContext, JobError, RunnableJob},
};
#[async_trait]
impl RunnableJob for CleanupOAuthAuthorizationGrantsJob {
#[tracing::instrument(name = "job.cleanup_oauth_authorization_grants", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove authorization grants after 7 days. They are in practice only
// valid for a short time, but keeping them around helps investigate abuse
// patterns.
let until = state.clock.now() - chrono::Duration::days(7);
// We use the fact that ULIDs include the creation time in their first 48 bits
// as a cursor
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted grants, and the greatest ULID processed
let (count, cursor) = repo
.oauth2_authorization_grant()
.cleanup(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no authorization grants to clean up");
} else {
info!(count = total, "cleaned up authorization grants");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupOAuthDeviceCodeGrantsJob {
#[tracing::instrument(name = "job.cleanup_oauth_device_code_grants", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove device code grants after 7 days. They are in practice only
// valid for a short time, but keeping them around helps investigate abuse
// patterns.
let until = state.clock.now() - chrono::Duration::days(7);
// We use the fact that ULIDs include the creation time in their first 48 bits
// as a cursor
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted grants, and the greatest ULID processed
let (count, cursor) = repo
.oauth2_device_code_grant()
.cleanup(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no device code grants to clean up");
} else {
info!(count = total, "cleaned up device code grants");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupUpstreamOAuthSessionsJob {
#[tracing::instrument(name = "job.cleanup_upstream_oauth_sessions", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove pending upstream OAuth authorization sessions after 7 days.
let until = state.clock.now() - chrono::Duration::days(7);
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, cursor) = repo
.upstream_oauth_session()
.cleanup_orphaned(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no pending upstream OAuth sessions to clean up");
} else {
info!(count = total, "cleaned up pending upstream OAuth sessions");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupUpstreamOAuthLinksJob {
#[tracing::instrument(name = "job.cleanup_upstream_oauth_links", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove orphaned upstream OAuth links after 7 days.
let until = state.clock.now() - chrono::Duration::days(7);
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, cursor) = repo
.upstream_oauth_link()
.cleanup_orphaned(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no orphaned upstream OAuth links to clean up");
} else {
info!(count = total, "cleaned up orphaned upstream OAuth links");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
+290
View File
@@ -0,0 +1,290 @@
// Copyright 2026 Element Creations Ltd.
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.
//! Session cleanup tasks
use std::time::Duration;
use async_trait::async_trait;
use mas_storage::queue::{
CleanupFinishedCompatSessionsJob, CleanupFinishedOAuth2SessionsJob,
CleanupFinishedUserSessionsJob, CleanupInactiveCompatSessionIpsJob,
CleanupInactiveOAuth2SessionIpsJob, CleanupInactiveUserSessionIpsJob,
};
use tracing::{debug, info};
use super::BATCH_SIZE;
use crate::{
State,
new_queue::{JobContext, JobError, RunnableJob},
};
#[async_trait]
impl RunnableJob for CleanupFinishedCompatSessionsJob {
#[tracing::instrument(name = "job.cleanup_finished_compat_sessions", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup compat sessions that were finished more than 30 days ago
let until = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted sessions, and the last finished_at
// timestamp
let (count, last_finished_at) = repo
.compat_session()
.cleanup_finished(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_finished_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no finished compat sessions to clean up");
} else {
info!(count = total, "cleaned up finished compat sessions");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupFinishedOAuth2SessionsJob {
#[tracing::instrument(name = "job.cleanup_finished_oauth2_sessions", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup OAuth2 sessions that were finished more than 30 days ago
let until = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted sessions, and the last finished_at
// timestamp
let (count, last_finished_at) = repo
.oauth2_session()
.cleanup_finished(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_finished_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no finished OAuth2 sessions to clean up");
} else {
info!(count = total, "cleaned up finished OAuth2 sessions");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupFinishedUserSessionsJob {
#[tracing::instrument(name = "job.cleanup_finished_user_sessions", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup user/browser sessions that were finished more than 30 days ago
let until = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted sessions, and the last finished_at
// timestamp. Only deletes sessions that have no child sessions
// (compat_sessions or oauth2_sessions).
let (count, last_finished_at) = repo
.browser_session()
.cleanup_finished(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_finished_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no finished user sessions to clean up");
} else {
info!(count = total, "cleaned up finished user sessions");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupInactiveOAuth2SessionIpsJob {
#[tracing::instrument(name = "job.cleanup_inactive_oauth2_session_ips", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Clear IPs from sessions inactive for 30+ days
let threshold = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, last_active_at) = repo
.oauth2_session()
.cleanup_inactive_ips(since, threshold, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_active_at;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no OAuth2 session IPs to clean up");
} else {
info!(count = total, "cleaned up inactive OAuth2 session IPs");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupInactiveCompatSessionIpsJob {
#[tracing::instrument(name = "job.cleanup_inactive_compat_session_ips", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Clear IPs from sessions inactive for 30+ days
let threshold = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, last_active_at) = repo
.compat_session()
.cleanup_inactive_ips(since, threshold, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_active_at;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no compat session IPs to clean up");
} else {
info!(count = total, "cleaned up inactive compat session IPs");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupInactiveUserSessionIpsJob {
#[tracing::instrument(name = "job.cleanup_inactive_user_session_ips", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Clear IPs from sessions inactive for 30+ days
let threshold = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, last_active_at) = repo
.browser_session()
.cleanup_inactive_ips(since, threshold, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_active_at;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no user session IPs to clean up");
} else {
info!(count = total, "cleaned up inactive user session IPs");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
Some(Duration::from_secs(10 * 60))
}
}
+214
View File
@@ -0,0 +1,214 @@
// Copyright 2026 Element Creations Ltd.
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.
//! OAuth token cleanup tasks
use std::time::Duration;
use async_trait::async_trait;
use mas_storage::queue::{
CleanupConsumedOAuthRefreshTokensJob, CleanupExpiredOAuthAccessTokensJob,
CleanupRevokedOAuthAccessTokensJob, CleanupRevokedOAuthRefreshTokensJob,
};
use tracing::{debug, info};
use super::BATCH_SIZE;
use crate::{
State,
new_queue::{JobContext, JobError, RunnableJob},
};
#[async_trait]
impl RunnableJob for CleanupRevokedOAuthAccessTokensJob {
#[tracing::instrument(name = "job.cleanup_revoked_oauth_access_tokens", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup tokens that were revoked more than an hour ago
let until = state.clock.now() - chrono::Duration::hours(1);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted tokens, and the last revoked_at timestamp
let (count, last_revoked_at) = repo
.oauth2_access_token()
.cleanup_revoked(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_revoked_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no token to clean up");
} else {
info!(count = total, "cleaned up revoked tokens");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupExpiredOAuthAccessTokensJob {
#[tracing::instrument(name = "job.cleanup_expired_oauth_access_tokens", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup tokens that expired more than a month ago
// It is important to keep them around for a bit because of refresh
// token idempotency. When we see a refresh token twice, we allow
// reusing it *only* if both the next refresh token and the next access
// tokens were not used. By keeping expired access tokens around for a
// month, we cannot make the *correct* decision, we will assume that the
// token wasn't used. Refer to the token refresh logic for details.
let until = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted tokens, and the last expires_at timestamp
let (count, last_expires_at) = repo
.oauth2_access_token()
.cleanup_expired(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_expires_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no token to clean up");
} else {
info!(count = total, "cleaned up expired tokens");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
Some(Duration::from_secs(60))
}
}
#[async_trait]
impl RunnableJob for CleanupRevokedOAuthRefreshTokensJob {
#[tracing::instrument(name = "job.cleanup_revoked_oauth_refresh_tokens", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup tokens that were revoked more than an hour ago
let until = state.clock.now() - chrono::Duration::hours(1);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted tokens, and the last revoked_at timestamp
let (count, last_revoked_at) = repo
.oauth2_refresh_token()
.cleanup_revoked(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_revoked_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no token to clean up");
} else {
info!(count = total, "cleaned up revoked tokens");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupConsumedOAuthRefreshTokensJob {
#[tracing::instrument(name = "job.cleanup_consumed_oauth_refresh_tokens", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup tokens that were consumed more than an hour ago
let until = state.clock.now() - chrono::Duration::hours(1);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted tokens, and the last consumed_at timestamp
let (count, last_consumed_at) = repo
.oauth2_refresh_token()
.cleanup_consumed(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_consumed_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no token to clean up");
} else {
info!(count = total, "cleaned up consumed tokens");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
+181
View File
@@ -0,0 +1,181 @@
// Copyright 2026 Element Creations Ltd.
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.
//! User-related cleanup tasks
use std::time::Duration;
use async_trait::async_trait;
use mas_storage::queue::{
CleanupUserEmailAuthenticationsJob, CleanupUserRecoverySessionsJob, CleanupUserRegistrationsJob,
};
use tracing::{debug, info};
use ulid::Ulid;
use super::BATCH_SIZE;
use crate::{
State,
new_queue::{JobContext, JobError, RunnableJob},
};
#[async_trait]
impl RunnableJob for CleanupUserRegistrationsJob {
#[tracing::instrument(name = "job.cleanup_user_registrations", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove user registrations after 30 days. They are in practice only
// valid for 1h, but keeping them around helps investigate abuse patterns.
let until = state.clock.now() - chrono::Duration::days(30);
// We use the fact that ULIDs include the creation time in their first 48 bits
// as a cursor
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted registrations, and the greatest ULID
// processed
let (count, cursor) = repo
.user_registration()
.cleanup(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no user registrations to clean up");
} else {
info!(count = total, "cleaned up user registrations");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupUserRecoverySessionsJob {
#[tracing::instrument(name = "job.cleanup_user_recovery_sessions", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove recovery sessions after 7 days. They are in practice only
// valid for a short time (tickets expire after 10 minutes), but keeping
// them around helps investigate abuse patterns.
let until = state.clock.now() - chrono::Duration::days(7);
// We use the fact that ULIDs include the creation time in their first 48 bits
// as a cursor
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted sessions, and the greatest ULID processed
let (count, cursor) = repo
.user_recovery()
.cleanup(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no user recovery sessions to clean up");
} else {
info!(count = total, "cleaned up user recovery sessions");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupUserEmailAuthenticationsJob {
#[tracing::instrument(name = "job.cleanup_user_email_authentications", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove email authentications after 7 days. They are in practice only
// valid for a short time (codes expire after 10 minutes), but keeping
// them around helps investigate abuse patterns.
let until = state.clock.now() - chrono::Duration::days(7);
// We use the fact that ULIDs include the creation time in their first 48 bits
// as a cursor
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted authentications, and the greatest ULID
// processed
let (count, cursor) = repo
.user_email()
.cleanup_authentications(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no user email authentications to clean up");
} else {
info!(count = total, "cleaned up user email authentications");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
-916
View File
@@ -1,916 +0,0 @@
// Copyright 2025, 2026 Element Creations Ltd.
// Copyright 2024, 2025 New Vector Ltd.
// Copyright 2023, 2024 The Matrix.org Foundation C.I.C.
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.
//! Database-related tasks
use std::time::Duration;
use async_trait::async_trait;
use mas_storage::queue::{
CleanupConsumedOAuthRefreshTokensJob, CleanupExpiredOAuthAccessTokensJob,
CleanupFinishedCompatSessionsJob, CleanupFinishedOAuth2SessionsJob,
CleanupFinishedUserSessionsJob, CleanupInactiveCompatSessionIpsJob,
CleanupInactiveOAuth2SessionIpsJob, CleanupInactiveUserSessionIpsJob,
CleanupOAuthAuthorizationGrantsJob, CleanupOAuthDeviceCodeGrantsJob, CleanupQueueJobsJob,
CleanupRevokedOAuthAccessTokensJob, CleanupRevokedOAuthRefreshTokensJob,
CleanupUpstreamOAuthLinksJob, CleanupUpstreamOAuthSessionsJob,
CleanupUserEmailAuthenticationsJob, CleanupUserRecoverySessionsJob,
CleanupUserRegistrationsJob, PruneStalePolicyDataJob,
};
use tracing::{debug, info};
use ulid::Ulid;
use crate::{
State,
new_queue::{JobContext, JobError, RunnableJob},
};
const BATCH_SIZE: usize = 1000;
#[async_trait]
impl RunnableJob for CleanupRevokedOAuthAccessTokensJob {
#[tracing::instrument(name = "job.cleanup_revoked_oauth_access_tokens", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup tokens that were revoked more than an hour ago
let until = state.clock.now() - chrono::Duration::hours(1);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted tokens, and the last revoked_at timestamp
let (count, last_revoked_at) = repo
.oauth2_access_token()
.cleanup_revoked(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_revoked_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no token to clean up");
} else {
info!(count = total, "cleaned up revoked tokens");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupExpiredOAuthAccessTokensJob {
#[tracing::instrument(name = "job.cleanup_expired_oauth_access_tokens", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup tokens that expired more than a month ago
// It is important to keep them around for a bit because of refresh
// token idempotency. When we see a refresh token twice, we allow
// reusing it *only* if both the next refresh token and the next access
// tokens were not used. By keeping expired access tokens around for a
// month, we cannot make the *correct* decision, we will assume that the
// token wasn't used. Refer to the token refresh logic for details.
let until = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted tokens, and the last expires_at timestamp
let (count, last_expires_at) = repo
.oauth2_access_token()
.cleanup_expired(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_expires_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no token to clean up");
} else {
info!(count = total, "cleaned up expired tokens");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
Some(Duration::from_secs(60))
}
}
#[async_trait]
impl RunnableJob for CleanupRevokedOAuthRefreshTokensJob {
#[tracing::instrument(name = "job.cleanup_revoked_oauth_refresh_tokens", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup tokens that were revoked more than an hour ago
let until = state.clock.now() - chrono::Duration::hours(1);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted tokens, and the last revoked_at timestamp
let (count, last_revoked_at) = repo
.oauth2_refresh_token()
.cleanup_revoked(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_revoked_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no token to clean up");
} else {
info!(count = total, "cleaned up revoked tokens");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupConsumedOAuthRefreshTokensJob {
#[tracing::instrument(name = "job.cleanup_consumed_oauth_refresh_tokens", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup tokens that were consumed more than an hour ago
let until = state.clock.now() - chrono::Duration::hours(1);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted tokens, and the last consumed_at timestamp
let (count, last_consumed_at) = repo
.oauth2_refresh_token()
.cleanup_consumed(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_consumed_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no token to clean up");
} else {
info!(count = total, "cleaned up consumed tokens");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupUserRecoverySessionsJob {
#[tracing::instrument(name = "job.cleanup_user_recovery_sessions", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove recovery sessions after 7 days. They are in practice only
// valid for a short time (tickets expire after 10 minutes), but keeping
// them around helps investigate abuse patterns.
let until = state.clock.now() - chrono::Duration::days(7);
// We use the fact that ULIDs include the creation time in their first 48 bits
// as a cursor
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted sessions, and the greatest ULID processed
let (count, cursor) = repo
.user_recovery()
.cleanup(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no user recovery sessions to clean up");
} else {
info!(count = total, "cleaned up user recovery sessions");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupUserEmailAuthenticationsJob {
#[tracing::instrument(name = "job.cleanup_user_email_authentications", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove email authentications after 7 days. They are in practice only
// valid for a short time (codes expire after 10 minutes), but keeping
// them around helps investigate abuse patterns.
let until = state.clock.now() - chrono::Duration::days(7);
// We use the fact that ULIDs include the creation time in their first 48 bits
// as a cursor
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted authentications, and the greatest ULID
// processed
let (count, cursor) = repo
.user_email()
.cleanup_authentications(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no user email authentications to clean up");
} else {
info!(count = total, "cleaned up user email authentications");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupUpstreamOAuthSessionsJob {
#[tracing::instrument(name = "job.cleanup_upstream_oauth_sessions", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove pending upstream OAuth authorization sessions after 7 days.
let until = state.clock.now() - chrono::Duration::days(7);
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, cursor) = repo
.upstream_oauth_session()
.cleanup_orphaned(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no pending upstream OAuth sessions to clean up");
} else {
info!(count = total, "cleaned up pending upstream OAuth sessions");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupUpstreamOAuthLinksJob {
#[tracing::instrument(name = "job.cleanup_upstream_oauth_links", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove orphaned upstream OAuth links after 7 days.
let until = state.clock.now() - chrono::Duration::days(7);
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, cursor) = repo
.upstream_oauth_link()
.cleanup_orphaned(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no orphaned upstream OAuth links to clean up");
} else {
info!(count = total, "cleaned up orphaned upstream OAuth links");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupQueueJobsJob {
#[tracing::instrument(name = "job.cleanup_queue_jobs", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove completed and failed queue jobs after 30 days.
// Keep them for debugging purposes.
let until = state.clock.now() - chrono::Duration::days(30);
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, cursor) = repo
.queue_job()
.cleanup(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no queue jobs to clean up");
} else {
info!(count = total, "cleaned up queue jobs");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupUserRegistrationsJob {
#[tracing::instrument(name = "job.cleanup_user_registrations", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove user registrations after 30 days. They are in practice only
// valid for 1h, but keeping them around helps investigate abuse patterns.
let until = state.clock.now() - chrono::Duration::days(30);
// We use the fact that ULIDs include the creation time in their first 48 bits
// as a cursor
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted registrations, and the greatest ULID
// processed
let (count, cursor) = repo
.user_registration()
.cleanup(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no user registrations to clean up");
} else {
info!(count = total, "cleaned up user registrations");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupFinishedCompatSessionsJob {
#[tracing::instrument(name = "job.cleanup_finished_compat_sessions", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup compat sessions that were finished more than 30 days ago
let until = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted sessions, and the last finished_at
// timestamp
let (count, last_finished_at) = repo
.compat_session()
.cleanup_finished(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_finished_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no finished compat sessions to clean up");
} else {
info!(count = total, "cleaned up finished compat sessions");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupFinishedOAuth2SessionsJob {
#[tracing::instrument(name = "job.cleanup_finished_oauth2_sessions", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup OAuth2 sessions that were finished more than 30 days ago
let until = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted sessions, and the last finished_at
// timestamp
let (count, last_finished_at) = repo
.oauth2_session()
.cleanup_finished(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_finished_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no finished OAuth2 sessions to clean up");
} else {
info!(count = total, "cleaned up finished OAuth2 sessions");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupFinishedUserSessionsJob {
#[tracing::instrument(name = "job.cleanup_finished_user_sessions", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Cleanup user/browser sessions that were finished more than 30 days ago
let until = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted sessions, and the last finished_at
// timestamp. Only deletes sessions that have no child sessions
// (compat_sessions or oauth2_sessions).
let (count, last_finished_at) = repo
.browser_session()
.cleanup_finished(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_finished_at;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no finished user sessions to clean up");
} else {
info!(count = total, "cleaned up finished user sessions");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupOAuthAuthorizationGrantsJob {
#[tracing::instrument(name = "job.cleanup_oauth_authorization_grants", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove authorization grants after 7 days. They are in practice only
// valid for a short time, but keeping them around helps investigate abuse
// patterns.
let until = state.clock.now() - chrono::Duration::days(7);
// We use the fact that ULIDs include the creation time in their first 48 bits
// as a cursor
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted grants, and the greatest ULID processed
let (count, cursor) = repo
.oauth2_authorization_grant()
.cleanup(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no authorization grants to clean up");
} else {
info!(count = total, "cleaned up authorization grants");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupOAuthDeviceCodeGrantsJob {
#[tracing::instrument(name = "job.cleanup_oauth_device_code_grants", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Remove device code grants after 7 days. They are in practice only
// valid for a short time, but keeping them around helps investigate abuse
// patterns.
let until = state.clock.now() - chrono::Duration::days(7);
// We use the fact that ULIDs include the creation time in their first 48 bits
// as a cursor
let until = Ulid::from_parts(
u64::try_from(until.timestamp_millis()).unwrap_or(u64::MIN),
u128::MAX,
);
let mut total = 0;
// Run until we get cancelled. We don't schedule a retry if we get cancelled, as
// this is a scheduled job and it will end up being rescheduled later anyway.
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// This returns the number of deleted grants, and the greatest ULID processed
let (count, cursor) = repo
.oauth2_device_code_grant()
.cleanup(since, until, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = cursor;
total += count;
// Check how many we deleted. If we deleted exactly BATCH_SIZE,
// there might be more to delete
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no device code grants to clean up");
} else {
info!(count = total, "cleaned up device code grants");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
// This job runs every hour, so having it running it for 10 minutes is fine
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for PruneStalePolicyDataJob {
#[tracing::instrument(name = "job.prune_stale_policy_data", skip_all)]
async fn run(&self, state: &State, _context: JobContext) -> Result<(), JobError> {
let mut repo = state.repository().await.map_err(JobError::retry)?;
// Keep the last 10 policy data
let count = repo
.policy_data()
.prune(10)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
if count == 0 {
debug!("no stale policy data to prune");
} else {
info!(count, "pruned stale policy data");
}
Ok(())
}
}
#[async_trait]
impl RunnableJob for CleanupInactiveOAuth2SessionIpsJob {
#[tracing::instrument(name = "job.cleanup_inactive_oauth2_session_ips", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Clear IPs from sessions inactive for 30+ days
let threshold = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, last_active_at) = repo
.oauth2_session()
.cleanup_inactive_ips(since, threshold, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_active_at;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no OAuth2 session IPs to clean up");
} else {
info!(count = total, "cleaned up inactive OAuth2 session IPs");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupInactiveCompatSessionIpsJob {
#[tracing::instrument(name = "job.cleanup_inactive_compat_session_ips", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Clear IPs from sessions inactive for 30+ days
let threshold = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, last_active_at) = repo
.compat_session()
.cleanup_inactive_ips(since, threshold, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_active_at;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no compat session IPs to clean up");
} else {
info!(count = total, "cleaned up inactive compat session IPs");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
Some(Duration::from_secs(10 * 60))
}
}
#[async_trait]
impl RunnableJob for CleanupInactiveUserSessionIpsJob {
#[tracing::instrument(name = "job.cleanup_inactive_user_session_ips", skip_all)]
async fn run(&self, state: &State, context: JobContext) -> Result<(), JobError> {
// Clear IPs from sessions inactive for 30+ days
let threshold = state.clock.now() - chrono::Duration::days(30);
let mut total = 0;
let mut since = None;
while !context.cancellation_token.is_cancelled() {
let mut repo = state.repository().await.map_err(JobError::retry)?;
let (count, last_active_at) = repo
.browser_session()
.cleanup_inactive_ips(since, threshold, BATCH_SIZE)
.await
.map_err(JobError::retry)?;
repo.save().await.map_err(JobError::retry)?;
since = last_active_at;
total += count;
if count != BATCH_SIZE {
break;
}
}
if total == 0 {
debug!("no user session IPs to clean up");
} else {
info!(count = total, "cleaned up inactive user session IPs");
}
Ok(())
}
fn timeout(&self) -> Option<Duration> {
Some(Duration::from_secs(10 * 60))
}
}
+64 -61
View File
@@ -21,7 +21,7 @@ use tokio_util::{sync::CancellationToken, task::TaskTracker};
pub use crate::new_queue::QueueWorker;
mod database;
mod cleanup;
mod email;
mod matrix;
mod new_queue;
@@ -162,125 +162,128 @@ pub async fn init(
.register_handler::<mas_storage::queue::CleanupInactiveCompatSessionIpsJob>()
.register_handler::<mas_storage::queue::CleanupInactiveUserSessionIpsJob>()
.register_deprecated_queue("cleanup-expired-tokens")
// Recurring jobs are spread across the hour at ~5 minute intervals
// to avoid clustering and distribute database load evenly.
.add_schedule(
"cleanup-revoked-oauth-access-tokens",
// Run this job every hour
// Run this job every hour at minute 0
"0 0 * * * *".parse()?,
mas_storage::queue::CleanupRevokedOAuthAccessTokensJob,
)
.add_schedule(
"cleanup-revoked-oauth-refresh-tokens",
// Run this job every hour
"0 10 * * * *".parse()?,
// Run this job every hour at minute 5
"0 5 * * * *".parse()?,
mas_storage::queue::CleanupRevokedOAuthRefreshTokensJob,
)
.add_schedule(
"cleanup-consumed-oauth-refresh-tokens",
// Run this job every hour
"0 20 * * * *".parse()?,
// Run this job every hour at minute 5 (safe to parallelize with revoked)
"0 5 * * * *".parse()?,
mas_storage::queue::CleanupConsumedOAuthRefreshTokensJob,
)
.add_schedule(
"cleanup-user-registrations",
// Run this job every hour
"0 30 * * * *".parse()?,
mas_storage::queue::CleanupUserRegistrationsJob,
)
.add_schedule(
"cleanup-finished-compat-sessions",
// Run this job every hour
"0 40 * * * *".parse()?,
// Run this job every hour at minute 10
"0 10 * * * *".parse()?,
mas_storage::queue::CleanupFinishedCompatSessionsJob,
)
.add_schedule(
"cleanup-finished-oauth2-sessions",
// Run this job every hour
"0 42 * * * *".parse()?,
// Run this job every hour at minute 15
"0 15 * * * *".parse()?,
mas_storage::queue::CleanupFinishedOAuth2SessionsJob,
)
.add_schedule(
"cleanup-finished-user-sessions",
// Run this job every hour
"0 44 * * * *".parse()?,
// Run this job every hour at minute 20
"0 20 * * * *".parse()?,
mas_storage::queue::CleanupFinishedUserSessionsJob,
)
.add_schedule(
"cleanup-inactive-oauth2-session-ips",
// Run this job every hour at minute 25
"0 25 * * * *".parse()?,
mas_storage::queue::CleanupInactiveOAuth2SessionIpsJob,
)
.add_schedule(
"cleanup-inactive-compat-session-ips",
// Run this job every hour at minute 25
"0 25 * * * *".parse()?,
mas_storage::queue::CleanupInactiveCompatSessionIpsJob,
)
.add_schedule(
"cleanup-inactive-user-session-ips",
// Run this job every hour at minute 25
"0 25 * * * *".parse()?,
mas_storage::queue::CleanupInactiveUserSessionIpsJob,
)
.add_schedule(
"cleanup-oauth-authorization-grants",
// Run this job every hour
"0 50 * * * *".parse()?,
// Run this job every hour at minute 30
"0 30 * * * *".parse()?,
mas_storage::queue::CleanupOAuthAuthorizationGrantsJob,
)
.add_schedule(
"cleanup-oauth-device-code-grants",
// Run this job every hour
"0 55 * * * *".parse()?,
// Run this job every hour at minute 35
"0 35 * * * *".parse()?,
mas_storage::queue::CleanupOAuthDeviceCodeGrantsJob,
)
.add_schedule(
"cleanup-user-recovery-sessions",
// Run this job every hour
"0 56 * * * *".parse()?,
mas_storage::queue::CleanupUserRecoverySessionsJob,
)
.add_schedule(
"cleanup-user-email-authentications",
// Run this job every hour
"0 57 * * * *".parse()?,
mas_storage::queue::CleanupUserEmailAuthenticationsJob,
)
.add_schedule(
"cleanup-upstream-oauth-sessions",
// Run this job every hour
"0 58 * * * *".parse()?,
// Run this job every hour at minute 40 (independent, safe to parallelize)
"0 40 * * * *".parse()?,
mas_storage::queue::CleanupUpstreamOAuthSessionsJob,
)
.add_schedule(
"cleanup-upstream-oauth-links",
// Run this job every hour
"0 59 * * * *".parse()?,
// Run this job every hour at minute 40
"0 40 * * * *".parse()?,
mas_storage::queue::CleanupUpstreamOAuthLinksJob,
)
// User cleanup jobs (minutes 45, 50)
.add_schedule(
"cleanup-user-registrations",
// Run this job every hour at minute 45
"0 45 * * * *".parse()?,
mas_storage::queue::CleanupUserRegistrationsJob,
)
.add_schedule(
"cleanup-user-recovery-sessions",
// Run this job every hour at minute 50
"0 50 * * * *".parse()?,
mas_storage::queue::CleanupUserRecoverySessionsJob,
)
.add_schedule(
"cleanup-user-email-authentications",
// Run this job every hour at minute 50
"0 50 * * * *".parse()?,
mas_storage::queue::CleanupUserEmailAuthenticationsJob,
)
.add_schedule(
"cleanup-queue-jobs",
// Run this job every hour
"0 45 * * * *".parse()?,
// Run this job every hour at minute 55
"0 55 * * * *".parse()?,
mas_storage::queue::CleanupQueueJobsJob,
)
.add_schedule(
"cleanup-expired-oauth-access-tokens",
// Run this job every 4 hours
// Run this job every 4 hours at minute 5
"0 5 */4 * * *".parse()?,
mas_storage::queue::CleanupExpiredOAuthAccessTokensJob,
)
.add_schedule(
"expire-inactive-sessions",
// Run this job every 15 minutes
// Run this job every 15 minutes at second 30
"30 */15 * * * *".parse()?,
mas_storage::queue::ExpireInactiveSessionsJob,
)
.add_schedule(
"prune-stale-policy-data",
// Run once a day
// Run once a day at 2:00 AM
"0 0 2 * * *".parse()?,
mas_storage::queue::PruneStalePolicyDataJob,
)
.add_schedule(
"cleanup-inactive-oauth2-session-ips",
// Run this job every hour
"0 46 * * * *".parse()?,
mas_storage::queue::CleanupInactiveOAuth2SessionIpsJob,
)
.add_schedule(
"cleanup-inactive-compat-session-ips",
// Run this job every hour
"0 47 * * * *".parse()?,
mas_storage::queue::CleanupInactiveCompatSessionIpsJob,
)
.add_schedule(
"cleanup-inactive-user-session-ips",
// Run this job every hour
"0 48 * * * *".parse()?,
mas_storage::queue::CleanupInactiveUserSessionIpsJob,
);
Ok(worker)
+16 -2
View File
@@ -11,9 +11,23 @@ Cleanup jobs are scheduled tasks that hard-delete old data from the database. Th
1. **Job struct** in `crates/storage/src/queue/tasks.rs` - Defines the job and queue name
2. **Storage trait** in `crates/storage/src/{domain}/` - Declares the cleanup method interface
3. **PostgreSQL implementation** in `crates/storage-pg/src/{domain}/` - Implements the actual cleanup logic
4. **Job runner** in `crates/tasks/src/database.rs` - Implements the `RunnableJob` trait with batching logic
4. **Job runner** in `crates/tasks/src/cleanup/` - Implements the `RunnableJob` trait with batching logic
5. **Registration** in `crates/tasks/src/lib.rs` - Registers the handler and schedules execution
### Module Structure
The cleanup job implementations are organized into submodules by domain:
```
crates/tasks/src/cleanup/
├── mod.rs # Re-exports, shared BATCH_SIZE constant
├── tokens.rs # OAuth token cleanup (access and refresh tokens)
├── sessions.rs # Session cleanup (compat, OAuth2, user sessions and their IPs)
├── oauth.rs # OAuth grants and upstream OAuth cleanup
├── user.rs # User-related cleanup (registrations, recovery, email auth)
└── misc.rs # Queue jobs, policy data cleanup
```
## All Cleanup Jobs
| Job | Entity | Retention | Notes |
@@ -183,7 +197,7 @@ The partial index (`WHERE timestamp_col IS NOT NULL`) makes queries more efficie
### 5. Implement RunnableJob
In `crates/tasks/src/database.rs`:
In the appropriate submodule under `crates/tasks/src/cleanup/` (e.g., `tokens.rs`, `sessions.rs`, `oauth.rs`, `user.rs`, or `misc.rs`):
```rust
#[async_trait]
+33 -33
View File
@@ -27,7 +27,7 @@ export type LocalazyMetadata = {
};
const localazyMetadata: LocalazyMetadata = {
projectUrl: "https://localazy.com/p/matrix-authentication-service!v1.10",
projectUrl: "https://localazy.com/p/matrix-authentication-service!v1.11",
baseLocale: "en",
languages: [
{
@@ -181,22 +181,22 @@ const localazyMetadata: LocalazyMetadata = {
file: "frontend.json",
path: "",
cdnFiles: {
"cs": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/cs/frontend.json",
"da": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/da/frontend.json",
"de": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/de/frontend.json",
"en": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/en/frontend.json",
"et": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/et/frontend.json",
"fi": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/fi/frontend.json",
"fr": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/fr/frontend.json",
"hu": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/hu/frontend.json",
"nb_NO": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/nb-NO/frontend.json",
"nl": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/nl/frontend.json",
"pl": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/pl/frontend.json",
"pt": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/pt/frontend.json",
"ru": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/ru/frontend.json",
"sv": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/sv/frontend.json",
"uk": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/uk/frontend.json",
"zh#Hans": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/zh-Hans/frontend.json"
"cs": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/cs/frontend.json",
"da": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/da/frontend.json",
"de": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/de/frontend.json",
"en": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/en/frontend.json",
"et": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/et/frontend.json",
"fi": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/fi/frontend.json",
"fr": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/fr/frontend.json",
"hu": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/hu/frontend.json",
"nb_NO": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/nb-NO/frontend.json",
"nl": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/nl/frontend.json",
"pl": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/pl/frontend.json",
"pt": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/pt/frontend.json",
"ru": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/ru/frontend.json",
"sv": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/sv/frontend.json",
"uk": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/uk/frontend.json",
"zh#Hans": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/7c203a8ac8bd48c3c4609a8effcd0fbac430f9b2/zh-Hans/frontend.json"
}
},
{
@@ -204,22 +204,22 @@ const localazyMetadata: LocalazyMetadata = {
file: "file.json",
path: "",
cdnFiles: {
"cs": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/cs/file.json",
"da": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/da/file.json",
"de": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/de/file.json",
"en": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/en/file.json",
"et": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/et/file.json",
"fi": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/fi/file.json",
"fr": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/fr/file.json",
"hu": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/hu/file.json",
"nb_NO": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/nb-NO/file.json",
"nl": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/nl/file.json",
"pl": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/pl/file.json",
"pt": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/pt/file.json",
"ru": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/ru/file.json",
"sv": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/sv/file.json",
"uk": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/uk/file.json",
"zh#Hans": "https://delivery.localazy.com/_a6657523900524156441ad8bd4f1/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/zh-Hans/file.json"
"cs": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/cs/file.json",
"da": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/da/file.json",
"de": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/de/file.json",
"en": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/en/file.json",
"et": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/et/file.json",
"fi": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/fi/file.json",
"fr": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/fr/file.json",
"hu": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/hu/file.json",
"nb_NO": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/nb-NO/file.json",
"nl": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/nl/file.json",
"pl": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/pl/file.json",
"pt": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/pt/file.json",
"ru": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/ru/file.json",
"sv": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/sv/file.json",
"uk": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/uk/file.json",
"zh#Hans": "https://delivery.localazy.com/_a6642094778790161378d4638916/_e0/5b69b0350dccfd47c245a5d41c1b9fdf6912cc6e/zh-Hans/file.json"
}
}
]
@@ -7,7 +7,7 @@
* - Please do NOT modify this file.
*/
const PACKAGE_VERSION = '2.12.4'
const PACKAGE_VERSION = '2.12.8'
const INTEGRITY_CHECKSUM = '4db4a41e972cec1b64cc569c66952d82'
const IS_MOCKED_RESPONSE = Symbol('isMockedResponse')
const activeClientIds = new Set()
+261 -316
View File
File diff suppressed because it is too large Load Diff
+14 -14
View File
@@ -22,7 +22,7 @@
"@fontsource/inter": "^5.2.8",
"@radix-ui/react-collapsible": "^1.1.12",
"@radix-ui/react-dialog": "^1.1.15",
"@tanstack/react-query": "^5.90.16",
"@tanstack/react-query": "^5.90.20",
"@tanstack/react-router": "^1.150.0",
"@vector-im/compound-design-tokens": "6.4.3",
"@vector-im/compound-web": "^8.3.5",
@@ -31,8 +31,8 @@
"classnames": "^2.5.1",
"date-fns": "^4.1.0",
"i18next": "^25.7.4",
"react": "^19.2.3",
"react-dom": "^19.2.3",
"react": "^19.2.4",
"react-dom": "^19.2.4",
"react-i18next": "^16.5.3",
"swagger-ui-dist": "^5.31.0",
"valibot": "^1.2.0",
@@ -43,34 +43,34 @@
"@browser-logos/chrome": "^2.0.0",
"@browser-logos/firefox": "^3.0.10",
"@browser-logos/safari": "^2.1.0",
"@graphql-codegen/cli": "^6.1.0",
"@graphql-codegen/cli": "^6.1.1",
"@graphql-codegen/client-preset": "^5.2.2",
"@graphql-codegen/typescript-msw": "^3.0.1",
"@storybook/addon-docs": "^10.1.11",
"@storybook/react-vite": "^10.1.11",
"@tanstack/react-query-devtools": "^5.91.2",
"@storybook/addon-docs": "^10.2.8",
"@storybook/react-vite": "^10.2.8",
"@tanstack/react-query-devtools": "^5.91.3",
"@tanstack/react-router-devtools": "^1.150.0",
"@tanstack/router-plugin": "^1.150.0",
"@testing-library/jest-dom": "^6.9.1",
"@testing-library/react": "^16.3.1",
"@testing-library/user-event": "^14.6.1",
"@types/node": "^25.0.3",
"@types/react": "19.2.7",
"@types/node": "^25.2.3",
"@types/react": "19.2.14",
"@types/react-dom": "19.2.3",
"@types/swagger-ui-dist": "^3.30.6",
"@vitejs/plugin-react": "^5.1.2",
"@vitest/coverage-v8": "^4.0.16",
"@vitejs/plugin-react": "^5.1.3",
"@vitest/coverage-v8": "^4.0.18",
"autoprefixer": "^10.4.23",
"browserslist-to-esbuild": "^2.1.1",
"graphql": "^16.12.0",
"happy-dom": "^20.0.11",
"happy-dom": "^20.5.0",
"i18next-cli": "^1.36.1",
"knip": "^5.81.0",
"msw": "^2.12.4",
"msw": "^2.12.8",
"msw-storybook-addon": "^2.0.6",
"postcss": "^8.5.6",
"postcss-import": "^16.1.1",
"postcss-nesting": "^13.0.2",
"postcss-nesting": "^14.0.0",
"rimraf": "^6.1.2",
"storybook": "^10.1.11",
"tailwindcss": "^3.4.19",
@@ -109,8 +109,8 @@ const AccountDeleteButton: React.FC<Props> = (props) => {
setAllowSubmitting(false);
}, [isMaybeValid]);
const onPasswordChange = useCallback(
(e: React.ChangeEvent<HTMLInputElement>) => {
const onPasswordInput = useCallback(
(e: React.InputEvent<HTMLInputElement>) => {
// We don't know if the password is correct, so we consider the form as
// valid if the field is not empty
setIsMaybeValid(e.currentTarget.value !== "");
@@ -118,8 +118,8 @@ const AccountDeleteButton: React.FC<Props> = (props) => {
[],
);
const onMxidChange = useCallback(
(e: React.ChangeEvent<HTMLInputElement>) => {
const onMxidInput = useCallback(
(e: React.InputEvent<HTMLInputElement>) => {
setIsMaybeValid(e.currentTarget.value === user.matrix.mxid);
},
[user.matrix.mxid],
@@ -208,7 +208,7 @@ const AccountDeleteButton: React.FC<Props> = (props) => {
<Form.PasswordControl
autoComplete="current-password"
required
onInput={onPasswordChange}
onInput={onPasswordInput}
/>
<Form.ErrorMessage match="valueMissing">
@@ -232,7 +232,7 @@ const AccountDeleteButton: React.FC<Props> = (props) => {
<Form.TextControl
required
placeholder={user.matrix.mxid}
onInput={onMxidChange}
onInput={onMxidInput}
/>
<Form.ErrorMessage match="valueMissing">
+1 -1
View File
@@ -10,7 +10,7 @@
// Ideally later on we could find a way to hydrate full React components instead
// of doing this, as this can very quickly get out of hands.
const VALID_USERNAME_RE = /^\s*([a-z0-9.=_/-]+|@[a-z0-9.=_/-]+(:.*)?)\s*$/g;
const VALID_USERNAME_RE = /^\s*([a-z0-9.=_/+-]+|@[a-z0-9.=_/+-]+(:.*)?)\s*$/g;
/** Grab the nearest error message inserted by the templates by error kind and code */
function grabErrorMessage(
+2 -2
View File
@@ -7,9 +7,9 @@
DOCKER := 0
PODMAN := 0
# Keep in sync with Dockerfile and .github/actions/build-policies/action.yml
OPA_DOCKER_IMAGE := docker.io/openpolicyagent/opa:1.8.0-debug
OPA_DOCKER_IMAGE := docker.io/openpolicyagent/opa:1.13.1
# Keep in sync with .github/workflows/ci.yaml
REGAL_DOCKER_IMAGE := ghcr.io/open-policy-agent/regal:0.36.1
REGAL_DOCKER_IMAGE := ghcr.io/open-policy-agent/regal:0.38.1
INPUTS := \
common/common.rego \
+1 -1
View File
@@ -49,7 +49,7 @@ violation contains {
"field": "username", "code": "username-invalid-chars",
"msg": "username contains invalid characters",
} if {
not regex.match(`^[a-z0-9.=_/-]+$`, input.username)
not regex.match(`^[a-z0-9.=_/+-]+$`, input.username)
}
violation contains {