diff --git a/Cargo.lock b/Cargo.lock
index c4f23907..3a9fe11a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -104,6 +104,12 @@ version = "1.0.68"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2cb2f989d18dd141ab8ae82f64d1a8cdd37e0840f73a406896cf5e99502fab61"
 
+[[package]]
+name = "arc-swap"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bddcadddf5e9015d310179a59bb28c4d4b9920ad0f11e8e14dbadf654890c9a6"
+
 [[package]]
 name = "arrayref"
 version = "0.3.6"
@@ -165,7 +171,7 @@ dependencies = [
  "futures-lite",
  "once_cell",
  "tokio 0.2.25",
- "tokio 1.18.2",
+ "tokio 1.35.1",
 ]
 
 [[package]]
@@ -246,7 +252,7 @@ dependencies = [
  "log",
  "memchr",
  "once_cell",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "pin-utils",
  "slab",
  "wasm-bindgen-futures",
@@ -315,7 +321,7 @@ dependencies = [
  "futures-io",
  "futures-util",
  "log",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "tungstenite",
 ]
 
@@ -588,9 +594,19 @@ checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38"
 
 [[package]]
 name = "bytes"
-version = "1.1.0"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c4872d67bab6358e59559027aa3b9157c53d9358c51423c17554809a8858e0f8"
+checksum = "a2bd12c1caf447e69cd4528f47f94d203fd2582878ecb9e9465484c4148a8223"
+
+[[package]]
+name = "bytes-utils"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7dafe3a8757b027e2be6e4e5601ed563c55989fcf1546e933c66c8eb3a058d35"
+dependencies = [
+ "bytes 1.5.0",
+ "either",
+]
 
 [[package]]
 name = "cache-padded"
@@ -612,7 +628,7 @@ dependencies = [
  "instant",
  "once_cell",
  "thiserror",
- "tokio 1.18.2",
+ "tokio 1.35.1",
 ]
 
 [[package]]
@@ -712,11 +728,11 @@ version = "4.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2a604e93b79d1808327a6fca85a6f2d69de66461e7620f5a4cbf5fb4d1d7c948"
 dependencies = [
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "futures-core",
  "memchr",
- "pin-project-lite 0.2.9",
- "tokio 1.18.2",
+ "pin-project-lite 0.2.13",
+ "tokio 1.35.1",
  "tokio-util 0.7.2",
 ]
 
@@ -784,6 +800,12 @@ dependencies = [
  "version_check",
 ]
 
+[[package]]
+name = "cookie-factory"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "396de984970346b0d9e93d1415082923c679e5ae5c3ee3dcbd104f5610af126b"
+
 [[package]]
 name = "core-foundation"
 version = "0.9.3"
@@ -809,6 +831,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "crc16"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "338089f42c427b86394a5ee60ff321da23a5c89c9d89514c829687b26359fcff"
+
 [[package]]
 name = "crossbeam-queue"
 version = "0.3.5"
@@ -1002,7 +1030,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "16a2561fd313df162315935989dceb8c99db4ee1933358270a57a3cfb8c957f3"
 dependencies = [
  "crossbeam-queue",
- "tokio 1.18.2",
+ "tokio 1.35.1",
 ]
 
 [[package]]
@@ -1348,6 +1376,15 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "float-cmp"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "98de4bbd547a563b716d8dfa9aad1cb19bfab00f4fa09a6a4ed21dbcf44ce9c4"
+dependencies = [
+ "num-traits",
+]
+
 [[package]]
 name = "fnv"
 version = "1.0.7"
@@ -1371,14 +1408,40 @@ checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b"
 
 [[package]]
 name = "form_urlencoded"
-version = "1.0.1"
+version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5fc25a87fa4fd2094bffb06925852034d90a17f0d1e05197d4956d3555752191"
+checksum = "e13624c2627564efccf4934284bdd98cbaa14e79b0b5a141218e507b3a823456"
 dependencies = [
- "matches",
  "percent-encoding",
 ]
 
+[[package]]
+name = "fred"
+version = "8.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3b2a2ac060e3266004c552235c241b481e438e2b1ea75715ea1176914ef2868"
+dependencies = [
+ "arc-swap",
+ "async-trait",
+ "bytes 1.5.0",
+ "bytes-utils",
+ "crossbeam-queue",
+ "float-cmp",
+ "futures",
+ "lazy_static",
+ "log",
+ "parking_lot",
+ "rand 0.8.5",
+ "redis-protocol",
+ "semver 1.0.9",
+ "socket2 0.5.5",
+ "tokio 1.35.1",
+ "tokio-stream",
+ "tokio-util 0.7.2",
+ "url",
+ "urlencoding",
+]
+
 [[package]]
 name = "fuchsia-cprng"
 version = "0.1.1"
@@ -1460,7 +1523,7 @@ dependencies = [
  "futures-io",
  "memchr",
  "parking",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "waker-fn",
 ]
 
@@ -1472,7 +1535,7 @@ checksum = "45ec6fe3675af967e67c5536c0b9d44e34e6c52f86bedc4ea49c5317b8e94d06"
 dependencies = [
  "futures-channel",
  "futures-task",
- "tokio 1.18.2",
+ "tokio 1.35.1",
 ]
 
 [[package]]
@@ -1517,7 +1580,7 @@ dependencies = [
  "futures-sink",
  "futures-task",
  "memchr",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "pin-utils",
  "slab",
 ]
@@ -1661,7 +1724,7 @@ version = "0.3.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "37a82c6d637fc9515a4694bbf1cb2457b79d81ce52b3108bdeea58b07dd34a57"
 dependencies = [
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "fnv",
  "futures-core",
  "futures-sink",
@@ -1669,7 +1732,7 @@ dependencies = [
  "http",
  "indexmap",
  "slab",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "tokio-util 0.7.2",
  "tracing",
 ]
@@ -1802,7 +1865,7 @@ version = "0.2.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ff8670570af52249509a86f5e3e18a08c60b177071826898fde8997cf5f6bfbb"
 dependencies = [
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "fnv",
  "itoa 1.0.2",
 ]
@@ -1823,9 +1886,9 @@ version = "0.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d5f38f16d184e36f2408a55281cd658ecbd3ca05cce6d6510a176eca393e26d1"
 dependencies = [
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "http",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
 ]
 
 [[package]]
@@ -1885,7 +1948,7 @@ version = "0.14.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42dc3c131584288d375f2d07f822b0cb012d8c6fb899a5b9fdb3cb7eb9b6004f"
 dependencies = [
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "futures-channel",
  "futures-core",
  "futures-util",
@@ -1895,9 +1958,9 @@ dependencies = [
  "httparse",
  "httpdate 1.0.2",
  "itoa 1.0.2",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "socket2 0.4.4",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "tower-service",
  "tracing",
  "want",
@@ -1922,10 +1985,10 @@ version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905"
 dependencies = [
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "hyper 0.14.19",
  "native-tls",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "tokio-native-tls",
 ]
 
@@ -1946,6 +2009,16 @@ dependencies = [
  "unicode-normalization",
 ]
 
+[[package]]
+name = "idna"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "634d9b1461af396cad843f47fdba5597a4f9e6ddd4bfb6ff5d85028c25cb12f6"
+dependencies = [
+ "unicode-bidi",
+ "unicode-normalization",
+]
+
 [[package]]
 name = "if_chain"
 version = "1.0.2"
@@ -2183,7 +2256,7 @@ dependencies = [
  "futures-util",
  "hostname",
  "httpdate 1.0.2",
- "idna",
+ "idna 0.2.3",
  "mime",
  "native-tls",
  "nom",
@@ -2194,9 +2267,9 @@ dependencies = [
 
 [[package]]
 name = "libc"
-version = "0.2.126"
+version = "0.2.153"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "349d5a591cd28b49e1d1037471617a32ddcda5731b99419008085f72d5a53836"
+checksum = "9c198f91728a82281a64e1f4f9eeb25d82cb32a5de251c6bd1b5154d63a8e7bd"
 
 [[package]]
 name = "libgit2-sys"
@@ -2429,14 +2502,13 @@ dependencies = [
 
 [[package]]
 name = "mio"
-version = "0.8.3"
+version = "0.8.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "713d550d9b44d89174e066b7a6217ae06234c10cb47819a88290d2b353c31799"
+checksum = "8f3d0b296e374a4e6f3c7b0a1f5a51d748a0d34c85e7dc48fc3fa9a87657fe09"
 dependencies = [
  "libc",
- "log",
  "wasi 0.11.0+wasi-snapshot-preview1",
- "windows-sys",
+ "windows-sys 0.48.0",
 ]
 
 [[package]]
@@ -2466,7 +2538,7 @@ dependencies = [
  "log",
  "metrics",
  "thiserror",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "tracing",
  "tracing-subscriber",
 ]
@@ -2518,7 +2590,7 @@ dependencies = [
  "strsim",
  "take_mut",
  "thiserror",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "tokio-rustls",
  "tokio-util 0.7.2",
  "trust-dns-proto",
@@ -2534,7 +2606,7 @@ version = "2.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5f8f35e687561d5c1667590911e6698a8cb714a134a7505718a182e7bc9d3836"
 dependencies = [
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "encoding_rs",
  "futures-util",
  "http",
@@ -2543,7 +2615,7 @@ dependencies = [
  "memchr",
  "mime",
  "spin 0.9.3",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "tokio-util 0.6.10",
  "version_check",
 ]
@@ -2846,7 +2918,7 @@ dependencies = [
  "libc",
  "redox_syscall",
  "smallvec",
- "windows-sys",
+ "windows-sys 0.36.1",
 ]
 
 [[package]]
@@ -2924,9 +2996,9 @@ dependencies = [
 
 [[package]]
 name = "percent-encoding"
-version = "2.1.0"
+version = "2.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e"
+checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
 
 [[package]]
 name = "pest"
@@ -2999,9 +3071,9 @@ checksum = "257b64915a082f7811703966789728173279bdebb956b143dbcd23f6f970a777"
 
 [[package]]
 name = "pin-project-lite"
-version = "0.2.9"
+version = "0.2.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e0a7ae3ac2f1173085d398531c705756c94a4c56843785df85a60c1a0afac116"
+checksum = "8afb450f006bf6385ca15ef45d71d2288452bc3683ce2e2cacc0d18e4be60b58"
 
 [[package]]
 name = "pin-utils"
@@ -3114,9 +3186,9 @@ dependencies = [
 
 [[package]]
 name = "proc-macro2"
-version = "1.0.56"
+version = "1.0.78"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2b63bdb0cd06f1f4dedf69b254734f9b45af66e4a031e42a7480257d9898b435"
+checksum = "e2422ad645d89c99f8f3e6b88a9fdeca7fabeac836b1002371c4367c8f984aae"
 dependencies = [
  "unicode-ident",
 ]
@@ -3334,15 +3406,15 @@ source = "git+https://github.com/revoltchat/redis-rs?rev=1a41faf356fd21aebba71ce
 dependencies = [
  "async-std",
  "async-trait",
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "combine",
  "futures-util",
  "itoa 1.0.2",
  "percent-encoding",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "ryu",
  "sha1_smol",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "tokio-util 0.7.2",
  "url",
 ]
@@ -3354,14 +3426,14 @@ source = "git+https://github.com/revoltchat/redis-rs?rev=f8ca28ab85da59d2ccde526
 dependencies = [
  "async-std",
  "async-trait",
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "combine",
  "futures-util",
  "itoa 1.0.2",
  "percent-encoding",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "ryu",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "tokio-util 0.7.2",
  "url",
 ]
@@ -3382,6 +3454,20 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "redis-protocol"
+version = "4.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9c31deddf734dc0a39d3112e73490e88b61a05e83e074d211f348404cee4d2c6"
+dependencies = [
+ "bytes 1.5.0",
+ "bytes-utils",
+ "cookie-factory",
+ "crc16",
+ "log",
+ "nom",
+]
+
 [[package]]
 name = "redox_syscall"
 version = "0.2.13"
@@ -3453,7 +3539,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "46a1f7aa4f35e5e8b4160449f51afc758f0ce6454315a9fa7d0d113e958c41eb"
 dependencies = [
  "base64 0.13.0",
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "encoding_rs",
  "futures-core",
  "futures-util",
@@ -3469,11 +3555,11 @@ dependencies = [
  "mime",
  "native-tls",
  "percent-encoding",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "serde",
  "serde_json",
  "serde_urlencoded",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "tokio-native-tls",
  "url",
  "wasm-bindgen",
@@ -3498,6 +3584,8 @@ version = "0.6.9"
 dependencies = [
  "async-std",
  "async-tungstenite",
+ "bincode",
+ "fred",
  "futures",
  "log",
  "once_cell",
@@ -3830,7 +3918,7 @@ dependencies = [
  "atomic",
  "atty",
  "binascii",
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "either",
  "figment",
  "futures",
@@ -3840,7 +3928,7 @@ dependencies = [
  "multer",
  "num_cpus",
  "parking_lot",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "rand 0.8.5",
  "ref-cast",
  "rocket_codegen",
@@ -3850,7 +3938,7 @@ dependencies = [
  "state",
  "tempfile",
  "time 0.3.17",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "tokio-stream",
  "tokio-util 0.7.2",
  "ubyte",
@@ -3933,14 +4021,14 @@ dependencies = [
  "memchr",
  "pear",
  "percent-encoding",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "ref-cast",
  "serde",
  "smallvec",
  "stable-pattern",
  "state",
  "time 0.3.17",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "uncased",
 ]
 
@@ -4088,7 +4176,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "88d6731146462ea25d9244b2ed5fd1d716d25c52e4d54aa4fb0f3c4e9854dbe2"
 dependencies = [
  "lazy_static",
- "windows-sys",
+ "windows-sys 0.36.1",
 ]
 
 [[package]]
@@ -4222,7 +4310,7 @@ dependencies = [
  "sentry-debug-images",
  "sentry-panic",
  "sentry-tracing",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "ureq",
 ]
 
@@ -4523,6 +4611,16 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "socket2"
+version = "0.5.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b5fac59a5cb5dd637972e5fca70daf0523c9067fcdc4842f053dae04a18f8e9"
+dependencies = [
+ "libc",
+ "windows-sys 0.48.0",
+]
+
 [[package]]
 name = "spin"
 version = "0.5.2"
@@ -4783,33 +4881,32 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.18.2"
+version = "1.35.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4903bf0427cf68dddd5aa6a93220756f8be0c34fcfa9f5e6191e103e15a31395"
+checksum = "c89b4efa943be685f629b149f53829423f8f5531ea21249408e8e2f8671ec104"
 dependencies = [
- "bytes 1.1.0",
+ "backtrace",
+ "bytes 1.5.0",
  "libc",
- "memchr",
- "mio 0.8.3",
+ "mio 0.8.10",
  "num_cpus",
- "once_cell",
  "parking_lot",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "signal-hook-registry",
- "socket2 0.4.4",
+ "socket2 0.5.5",
  "tokio-macros",
- "winapi 0.3.9",
+ "windows-sys 0.48.0",
 ]
 
 [[package]]
 name = "tokio-macros"
-version = "1.7.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b557f72f448c511a979e2564e55d74e6c4432fc96ff4f6241bc6bded342643b7"
+checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
 dependencies = [
  "proc-macro2",
  "quote 1.0.26",
- "syn 1.0.107",
+ "syn 2.0.15",
 ]
 
 [[package]]
@@ -4819,7 +4916,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f7d995660bd2b7f8c1568414c1126076c13fbb725c40112dc0120b78eb9b717b"
 dependencies = [
  "native-tls",
- "tokio 1.18.2",
+ "tokio 1.35.1",
 ]
 
 [[package]]
@@ -4829,7 +4926,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c43ee83903113e03984cb9e5cebe6c04a5116269e900e3ddba8f068a62adda59"
 dependencies = [
  "rustls",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "webpki",
 ]
 
@@ -4840,8 +4937,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "50145484efff8818b5ccd256697f36863f587da82cf8b409c53adf1e840798e3"
 dependencies = [
  "futures-core",
- "pin-project-lite 0.2.9",
- "tokio 1.18.2",
+ "pin-project-lite 0.2.13",
+ "tokio 1.35.1",
 ]
 
 [[package]]
@@ -4874,12 +4971,12 @@ version = "0.6.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "36943ee01a6d67977dd3f84a5a1d2efeb4ada3a1ae771cadfaa535d9d9fc6507"
 dependencies = [
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "futures-core",
  "futures-sink",
  "log",
- "pin-project-lite 0.2.9",
- "tokio 1.18.2",
+ "pin-project-lite 0.2.13",
+ "tokio 1.35.1",
 ]
 
 [[package]]
@@ -4888,12 +4985,12 @@ version = "0.7.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f988a1a1adc2fb21f9c12aa96441da33a1728193ae0b95d2be22dbd17fcb4e5c"
 dependencies = [
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "futures-core",
  "futures-io",
  "futures-sink",
- "pin-project-lite 0.2.9",
- "tokio 1.18.2",
+ "pin-project-lite 0.2.13",
+ "tokio 1.35.1",
  "tracing",
 ]
 
@@ -4932,7 +5029,7 @@ checksum = "5d0ecdcb44a79f0fe9844f0c4f33a342cbcbb5117de8001e6ba0dc2351327d09"
 dependencies = [
  "cfg-if 1.0.0",
  "log",
- "pin-project-lite 0.2.9",
+ "pin-project-lite 0.2.13",
  "tracing-attributes",
  "tracing-core",
 ]
@@ -5010,7 +5107,7 @@ dependencies = [
  "futures-channel",
  "futures-io",
  "futures-util",
- "idna",
+ "idna 0.2.3",
  "ipnet",
  "lazy_static",
  "log",
@@ -5018,7 +5115,7 @@ dependencies = [
  "smallvec",
  "thiserror",
  "tinyvec",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "url",
 ]
 
@@ -5038,7 +5135,7 @@ dependencies = [
  "resolv-conf",
  "smallvec",
  "thiserror",
- "tokio 1.18.2",
+ "tokio 1.35.1",
  "trust-dns-proto",
 ]
 
@@ -5056,7 +5153,7 @@ checksum = "d96a2dea40e7570482f28eb57afbe42d97551905da6a9400acc5c328d24004f5"
 dependencies = [
  "base64 0.13.0",
  "byteorder",
- "bytes 1.1.0",
+ "bytes 1.5.0",
  "http",
  "httparse",
  "log",
@@ -5169,9 +5266,9 @@ dependencies = [
 
 [[package]]
 name = "unicode-bidi"
-version = "0.3.8"
+version = "0.3.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "099b7128301d285f79ddd55b9a83d5e6b9e97c92e0ea0daebee7263e932de992"
+checksum = "08f95100a766bf4f8f28f90d77e0a5461bbdb219042e7679bebe79004fed8d75"
 
 [[package]]
 name = "unicode-ident"
@@ -5181,9 +5278,9 @@ checksum = "d22af068fba1eb5edcb4aea19d382b2a3deb4c8f9d475c589b6ada9e0fd493ee"
 
 [[package]]
 name = "unicode-normalization"
-version = "0.1.19"
+version = "0.1.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d54590932941a9e9266f0832deed84ebe1bf2e4c9e4a3554d393d18f5e854bf9"
+checksum = "5c5713f0fc4b5db668a2ac63cdb7bb4469d8c9fed047b1d0292cc7b0ce2ba921"
 dependencies = [
  "tinyvec",
 ]
@@ -5243,13 +5340,12 @@ dependencies = [
 
 [[package]]
 name = "url"
-version = "2.2.2"
+version = "2.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a507c383b2d33b5fc35d1861e77e6b383d158b2da5e14fe51b83dfedf6fd578c"
+checksum = "31e6302e3bb753d46e83516cae55ae196fc0c309407cf11ab35cc51a4c2a4633"
 dependencies = [
  "form_urlencoded",
- "idna",
- "matches",
+ "idna 0.5.0",
  "percent-encoding",
  "serde",
 ]
@@ -5263,6 +5359,12 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "urlencoding"
+version = "2.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
+
 [[package]]
 name = "utf-8"
 version = "0.7.6"
@@ -5295,7 +5397,7 @@ version = "0.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f07b0a1390e01c0fc35ebb26b28ced33c9a3808f7f9fbe94d3cc01e233bfeed5"
 dependencies = [
- "idna",
+ "idna 0.2.3",
  "lazy_static",
  "regex",
  "serde",
@@ -5310,7 +5412,7 @@ version = "0.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32ad5bf234c7d3ad1042e5252b7eddb2c4669ee23f32c7dd0e9b7705f07ef591"
 dependencies = [
- "idna",
+ "idna 0.2.3",
  "lazy_static",
  "regex",
  "serde",
@@ -5626,43 +5728,109 @@ version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ea04155a16a59f9eab786fe12a4a450e75cdb175f9e0d80da1e17db09f55b8d2"
 dependencies = [
- "windows_aarch64_msvc",
- "windows_i686_gnu",
- "windows_i686_msvc",
- "windows_x86_64_gnu",
- "windows_x86_64_msvc",
+ "windows_aarch64_msvc 0.36.1",
+ "windows_i686_gnu 0.36.1",
+ "windows_i686_msvc 0.36.1",
+ "windows_x86_64_gnu 0.36.1",
+ "windows_x86_64_msvc 0.36.1",
 ]
 
+[[package]]
+name = "windows-sys"
+version = "0.48.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
+dependencies = [
+ "windows-targets",
+]
+
+[[package]]
+name = "windows-targets"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c"
+dependencies = [
+ "windows_aarch64_gnullvm",
+ "windows_aarch64_msvc 0.48.5",
+ "windows_i686_gnu 0.48.5",
+ "windows_i686_msvc 0.48.5",
+ "windows_x86_64_gnu 0.48.5",
+ "windows_x86_64_gnullvm",
+ "windows_x86_64_msvc 0.48.5",
+]
+
+[[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8"
+
 [[package]]
 name = "windows_aarch64_msvc"
 version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9bb8c3fd39ade2d67e9874ac4f3db21f0d710bee00fe7cab16949ec184eeaa47"
 
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc"
+
 [[package]]
 name = "windows_i686_gnu"
 version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "180e6ccf01daf4c426b846dfc66db1fc518f074baa793aa7d9b9aaeffad6a3b6"
 
+[[package]]
+name = "windows_i686_gnu"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e"
+
 [[package]]
 name = "windows_i686_msvc"
 version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e2e7917148b2812d1eeafaeb22a97e4813dfa60a3f8f78ebe204bcc88f12f024"
 
+[[package]]
+name = "windows_i686_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406"
+
 [[package]]
 name = "windows_x86_64_gnu"
 version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4dcd171b8776c41b97521e5da127a2d86ad280114807d0b2ab1e462bc764d9e1"
 
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e"
+
+[[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc"
+
 [[package]]
 name = "windows_x86_64_msvc"
 version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c811ca4a8c853ef420abd8592ba53ddbbac90410fab6903b3e79972a631f7680"
 
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538"
+
 [[package]]
 name = "winreg"
 version = "0.7.0"
diff --git a/crates/bonfire/Cargo.toml b/crates/bonfire/Cargo.toml
index 55260818..3d52dce2 100644
--- a/crates/bonfire/Cargo.toml
+++ b/crates/bonfire/Cargo.toml
@@ -18,6 +18,7 @@ querystring = "1.1.0"
 revolt-quark = { path = "../quark" }
 
 # serde
+bincode = "1.3.3"
 serde_json = "1.0.79"
 rmp-serde = "1.0.0"
 serde = "1.0.136"
@@ -34,3 +35,6 @@ async-std = { version = "1.8.0", features = [
 # core
 revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
 sentry = "0.31.5"
+
+# redis
+fred = { version = "8.0.1", features = ["subscriber-client"] }
\ No newline at end of file
diff --git a/crates/bonfire/src/main.rs b/crates/bonfire/src/main.rs
index e9bcef88..765ce4e2 100644
--- a/crates/bonfire/src/main.rs
+++ b/crates/bonfire/src/main.rs
@@ -29,6 +29,10 @@ async fn main() {
 
     // Start accepting new connections and spawn a client for each connection.
     while let Ok((stream, addr)) = listener.accept().await {
-        websocket::spawn_client(database::get_db(), stream, addr);
+        async_std::task::spawn(async move {
+            info!("User connected from {addr:?}");
+            websocket::client(database::get_db(), stream, addr).await;
+            info!("User disconnected from {addr:?}");
+        });
     }
 }
diff --git a/crates/bonfire/src/websocket.rs b/crates/bonfire/src/websocket.rs
index b0cb8f45..6fd5aa08 100644
--- a/crates/bonfire/src/websocket.rs
+++ b/crates/bonfire/src/websocket.rs
@@ -1,6 +1,16 @@
 use std::net::SocketAddr;
 
-use futures::{channel::oneshot, pin_mut, select, FutureExt, SinkExt, StreamExt, TryStreamExt};
+use async_tungstenite::WebSocketStream;
+use fred::{
+    interfaces::{ClientLike, EventInterface, PubsubInterface},
+    types::RedisConfig,
+};
+use futures::{
+    channel::oneshot,
+    pin_mut, select,
+    stream::{SplitSink, SplitStream},
+    FutureExt, SinkExt, StreamExt, TryStreamExt,
+};
 use revolt_presence::{create_session, delete_session};
 use revolt_quark::{
     events::{
@@ -9,249 +19,249 @@ use revolt_quark::{
         state::{State, SubscriptionStateChange},
     },
     models::{user::UserHint, User},
-    redis_kiss, Database,
+    redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI},
+    Database,
 };
 
-use async_std::{net::TcpStream, sync::Mutex, task};
+use async_std::{net::TcpStream, sync::Mutex};
 
-use crate::config::WebsocketHandshakeCallback;
+use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
 
-/// Spawn a new WebSocket client worker given access to the database,
+type WsReader = SplitStream<WebSocketStream<TcpStream>>;
+type WsWriter = SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>;
+
+/// Start a new WebSocket client worker given access to the database,
 /// the relevant TCP stream and the remote address of the client.
-pub fn spawn_client(db: &'static Database, stream: TcpStream, addr: SocketAddr) {
-    // Spawn a new Async task to work on.
-    task::spawn(async move {
-        info!("User connected from {addr:?}");
+pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr) {
+    // Upgrade the TCP connection to a WebSocket connection.
+    // In this process, we also parse any additional parameters given.
+    // e.g. wss://example.com?format=json&version=1
+    let (sender, receiver) = oneshot::channel();
+    let Ok(ws) = async_tungstenite::accept_hdr_async_with_config(
+        stream,
+        WebsocketHandshakeCallback::from(sender),
+        None,
+    )
+    .await
+    else {
+        return;
+    };
+    // Verify we've received a valid config, otherwise we should just drop the connection.
+    let Ok(mut config) = receiver.await else {
+        return;
+    };
+    info!(
+        "User {addr:?} provided protocol configuration (version = {}, format = {:?})",
+        config.get_protocol_version(),
+        config.get_protocol_format()
+    );
 
-        // Upgrade the TCP connection to a WebSocket connection.
-        // In this process, we also parse any additional parameters given.
-        // e.g. wss://example.com?format=json&version=1
-        let (sender, receiver) = oneshot::channel();
-        if let Ok(ws) = async_tungstenite::accept_hdr_async_with_config(
-            stream,
-            WebsocketHandshakeCallback::from(sender),
-            None,
-        )
+    // Split the socket for simultaneously read and write.
+    let (mut write, mut read) = ws.split();
+
+    // If the user has not provided authentication, request information.
+    if config.get_session_token().is_none() {
+        while let Ok(message) = read.try_next().await {
+            if let Ok(ClientMessage::Authenticate { token }) =
+                config.decode(message.as_ref().unwrap())
+            {
+                config.set_session_token(token);
+                break;
+            }
+        }
+    }
+
+    // Try to authenticate the user.
+    let Some(token) = config.get_session_token().as_ref() else {
+        return;
+    };
+    let user = match User::from_token(db, token, UserHint::Any).await {
+        Ok(user) => user,
+        Err(err) => {
+            write.send(config.encode(&err)).await.ok();
+            return;
+        }
+    };
+    info!("User {addr:?} authenticated as @{}", user.username);
+
+    // Create local state.
+    let mut state = State::from(user);
+    let user_id = state.cache.user_id.clone();
+
+    // Notify socket we have authenticated.
+    if write
+        .send(config.encode(&EventV1::Authenticated))
         .await
-        {
-            // Verify we've received a valid config, otherwise we should just drop the connection.
-            if let Ok(mut config) = receiver.await {
-                info!(
-                    "User {addr:?} provided protocol configuration (version = {}, format = {:?})",
-                    config.get_protocol_version(),
-                    config.get_protocol_format()
-                );
+        .is_err()
+    {
+        return;
+    }
 
-                // Split the socket for simultaneously read and write.
-                let (write, mut read) = ws.split();
-                let write = Mutex::new(write);
+    // Download required data to local cache and send Ready payload.
+    let Ok(ready_payload) = state.generate_ready_payload(db).await else {
+        return;
+    };
+    if write.send(config.encode(&ready_payload)).await.is_err() {
+        return;
+    }
 
-                // If the user has not provided authentication, request information.
-                if config.get_session_token().is_none() {
-                    'outer: while let Ok(message) = read.try_next().await {
-                        if let Ok(ClientMessage::Authenticate { token }) =
-                            config.decode(message.as_ref().unwrap())
-                        {
-                            config.set_session_token(token);
-                            break 'outer;
-                        }
-                    }
+    // Create presence session.
+    let (first_session, session_id) = create_session(&user_id, 0).await;
+
+    // If this was the first session, notify other users that we just went online.
+    if first_session {
+        state.broadcast_presence_change(true).await;
+    }
+
+    {
+        let write = Mutex::new(write);
+        // Create a PubSub connection to poll on.
+        let listener = listener(db, &mut state, addr, &config, &write).fuse();
+        // Read from WebSocket stream.
+        let worker = worker(user_id.clone(), &config, read, &write).fuse();
+
+        // Pin both tasks.
+        pin_mut!(listener, worker);
+
+        // Wait for either disconnect or for listener to die.
+        select!(
+            () = listener => {},
+            () = worker => {}
+        );
+    }
+    // Clean up presence session.
+    let last_session = delete_session(&user_id, session_id).await;
+
+    // If this was the last session, notify other users that we just went offline.
+    if last_session {
+        state.broadcast_presence_change(false).await;
+    }
+}
+
+async fn listener(
+    db: &'static Database,
+    state: &mut State,
+    addr: SocketAddr,
+    config: &ProtocolConfiguration,
+    write: &Mutex<WsWriter>,
+) {
+    let redis_config = RedisConfig::from_url(&REDIS_URI).unwrap();
+    let Ok(subscriber) = fred::types::Builder::from_config(redis_config).build_subscriber_client()
+    else {
+        return;
+    };
+    if subscriber.init().await.is_err() {
+        return;
+    };
+    let mut message_rx = subscriber.message_rx();
+    loop {
+        // Check for state changes for subscriptions.
+        match state.apply_state() {
+            SubscriptionStateChange::Reset => {
+                subscriber.unsubscribe_all().await.unwrap();
+                for id in state.iter_subscriptions() {
+                    subscriber.subscribe(id).await.unwrap();
                 }
 
-                // Try to authenticate the user.
-                if let Some(token) = config.get_session_token().as_ref() {
-                    match User::from_token(db, token, UserHint::Any).await {
-                        Ok(user) => {
-                            info!("User {addr:?} authenticated as @{}", user.username);
+                #[cfg(debug_assertions)]
+                info!("{addr:?} has reset their subscriptions");
+            }
+            SubscriptionStateChange::Change { add, remove } => {
+                for id in remove {
+                    #[cfg(debug_assertions)]
+                    info!("{addr:?} unsubscribing from {id}");
 
-                            // Create local state.
-                            let mut state = State::from(user);
-                            let user_id = state.cache.user_id.clone();
+                    subscriber.unsubscribe(id).await.unwrap();
+                }
 
-                            // Create presence session.
-                            let (first_session, session_id) = create_session(&user_id, 0).await;
+                for id in add {
+                    #[cfg(debug_assertions)]
+                    info!("{addr:?} subscribing to {id}");
 
-                            // Notify socket we have authenticated.
-                            write
-                                .lock()
-                                .await
-                                .send(config.encode(&EventV1::Authenticated))
-                                .await
-                                .ok();
-
-                            // Download required data to local cache and send Ready payload.
-                            if let Ok(ready_payload) = state.generate_ready_payload(db).await {
-                                write
-                                    .lock()
-                                    .await
-                                    .send(config.encode(&ready_payload))
-                                    .await
-                                    .ok();
-
-                                // If this was the first session, notify other users that we just went online.
-                                if first_session {
-                                    state.broadcast_presence_change(true).await;
-                                }
-
-                                // Create a PubSub connection to poll on.
-                                let listener = async {
-                                    if let Ok(mut conn) = redis_kiss::open_pubsub_connection().await
-                                    {
-                                        loop {
-                                            // Check for state changes for subscriptions.
-                                            match state.apply_state() {
-                                                SubscriptionStateChange::Reset => {
-                                                    for id in state.iter_subscriptions() {
-                                                        conn.subscribe(id).await.unwrap();
-                                                    }
-
-                                                    #[cfg(debug_assertions)]
-                                                    info!("{addr:?} has reset their subscriptions");
-                                                }
-                                                SubscriptionStateChange::Change { add, remove } => {
-                                                    for id in remove {
-                                                        #[cfg(debug_assertions)]
-                                                        info!("{addr:?} unsubscribing from {id}");
-
-                                                        conn.unsubscribe(id).await.unwrap();
-                                                    }
-
-                                                    for id in add {
-                                                        #[cfg(debug_assertions)]
-                                                        info!("{addr:?} subscribing to {id}");
-
-                                                        conn.subscribe(id).await.unwrap();
-                                                    }
-                                                }
-                                                SubscriptionStateChange::None => {}
-                                            }
-
-                                            // * Debug logging of current subscriptions.
-                                            /*#[cfg(debug_assertions)]
-                                            info!(
-                                                "User {addr:?} is subscribed to {:?}",
-                                                state
-                                                    .iter_subscriptions()
-                                                    .collect::<Vec<&String>>()
-                                            );*/
-
-                                            // Handle incoming events.
-                                            match conn.on_message().next().await.map(|res| {
-                                                res.map(|item|(
-                                                    item.get_channel_name().to_string(),
-                                                    redis_kiss::decode_payload::<EventV1>(&item),
-                                                ))
-                                            }) {
-                                                Some(Ok((channel, item))) => {
-                                                    if let Ok(mut event) = item {
-                                                        if state
-                                                            .handle_incoming_event_v1(
-                                                                db, &mut event,
-                                                            )
-                                                            .await
-                                                            && write.lock().await
-                                                                .send(config.encode(&event))
-                                                                .await
-                                                                .is_err()
-                                                        {
-                                                            break;
-                                                        }
-                                                    } else {
-                                                        warn!("Failed to deserialise an event for {channel}!");
-                                                    }
-                                                }
-                                                Some(Err(e)) => {
-                                                    info!("Error while consuming pub/sub messages: {e:?}");
-                                                    sentry::capture_error(&e);
-                                                    break
-                                                }
-                                                // No more data, assume we disconnected or otherwise
-                                                // something bad occurred, so disconnect user.
-                                                None => break,
-                                            }
-                                        }
-                                    }
-                                }
-                                .fuse();
-
-                                // Read from WebSocket stream.
-                                let worker =
-                                    async {
-                                        while let Ok(Some(msg)) = read.try_next().await {
-                                            if let Ok(payload) = config.decode(&msg) {
-                                                match payload {
-                                                    ClientMessage::BeginTyping { channel } => {
-                                                        EventV1::ChannelStartTyping {
-                                                            id: channel.clone(),
-                                                            user: user_id.clone(),
-                                                        }
-                                                        .p(channel.clone())
-                                                        .await;
-                                                    }
-                                                    ClientMessage::EndTyping { channel } => {
-                                                        EventV1::ChannelStopTyping {
-                                                            id: channel.clone(),
-                                                            user: user_id.clone(),
-                                                        }
-                                                        .p(channel.clone())
-                                                        .await;
-                                                    }
-                                                    ClientMessage::Ping { data, responded } => {
-                                                        if responded.is_none() {
-                                                            write
-                                                                .lock()
-                                                                .await
-                                                                .send(config.encode(
-                                                                    &EventV1::Pong { data },
-                                                                ))
-                                                                .await
-                                                                .ok();
-                                                        }
-                                                    }
-                                                    _ => {}
-                                                }
-                                            }
-                                        }
-                                    }
-                                    .fuse();
-
-                                // Pin both tasks.
-                                pin_mut!(listener, worker);
-
-                                // Wait for either disconnect or for listener to die.
-                                select!(
-                                    () = listener => {},
-                                    () = worker => {}
-                                );
-
-                                // * Combine the streams back once we are ready to disconnect.
-                                /* ws = read.reunite(write).unwrap(); */
-                            }
-
-                            // Clean up presence session.
-                            let last_session = delete_session(&user_id, session_id).await;
-
-                            // If this was the last session, notify other users that we just went offline.
-                            if last_session {
-                                state.broadcast_presence_change(false).await;
-                            }
-                        }
-                        Err(err) => {
-                            write.lock().await.send(config.encode(&err)).await.ok();
-                        }
-                    }
+                    subscriber.subscribe(id).await.unwrap();
                 }
             }
-
-            // * Disconnect the WebSocket if it isn't already.
-            /*ws.close(Some(CloseFrame {
-                code: CloseCode::Normal,
-                reason: std::borrow::Cow::from(""),
-            }))
-            .await
-            .unwrap();*/
+            SubscriptionStateChange::None => {}
         }
 
-        info!("User disconnected from {addr:?}");
-    });
+        // Handle incoming events.
+        let Ok(message) = message_rx.recv().await.map_err(|e| {
+            info!("Error while consuming pub/sub messages: {e:?}");
+            sentry::capture_error(&e);
+        }) else {
+            return;
+        };
+        let event = match *REDIS_PAYLOAD_TYPE {
+            PayloadType::Json => message
+                .value
+                .as_str()
+                .and_then(|s| serde_json::from_str::<EventV1>(s.as_ref()).ok()),
+            PayloadType::Msgpack => message
+                .value
+                .as_bytes()
+                .and_then(|b| rmp_serde::from_slice::<EventV1>(b).ok()),
+            PayloadType::Bincode => message
+                .value
+                .as_bytes()
+                .and_then(|b| bincode::deserialize::<EventV1>(b).ok()),
+        };
+        let Some(mut event) = event else {
+            warn!("Failed to deserialise an event for {}!", message.channel);
+            return;
+        };
+        let should_send = state.handle_incoming_event_v1(db, &mut event).await;
+
+        if should_send
+            && write
+                .lock()
+                .await
+                .send(config.encode(&event))
+                .await
+                .is_err()
+        {
+            return;
+        }
+    }
+}
+
+async fn worker(
+    user_id: String,
+    config: &ProtocolConfiguration,
+    mut read: WsReader,
+    write: &Mutex<WsWriter>,
+) {
+    while let Ok(Some(msg)) = read.try_next().await {
+        let Ok(payload) = config.decode(&msg) else {
+            continue;
+        };
+        match payload {
+            ClientMessage::BeginTyping { channel } => {
+                EventV1::ChannelStartTyping {
+                    id: channel.clone(),
+                    user: user_id.clone(),
+                }
+                .p(channel.clone())
+                .await;
+            }
+            ClientMessage::EndTyping { channel } => {
+                EventV1::ChannelStopTyping {
+                    id: channel.clone(),
+                    user: user_id.clone(),
+                }
+                .p(channel.clone())
+                .await;
+            }
+            ClientMessage::Ping { data, responded } => {
+                if responded.is_none() {
+                    write
+                        .lock()
+                        .await
+                        .send(config.encode(&EventV1::Pong { data }))
+                        .await
+                        .ok();
+                }
+            }
+            _ => {}
+        }
+    }
 }