chore: replace async-std with tokio (#813)

Signed-off-by: Zomatree <me@zomatree.live>

Cargo.toml

@@ -23,11 +23,11 @@ async-trait = "0.1.89"
 tokio = "1.49.0"
 async-channel = "2.3.1"
 futures = "0.3.32"
-async-std = "1.8.0"
 async-tungstenite = "0.17.0"
 futures-locks = "0.7.1"
 async-lock = "2.8.0"
 async-recursion = "1.0.4"
+tokio-util = { version = "0.7.18" }
 
 # Error Handling
 anyhow = "1.0.100"
@@ -89,7 +89,7 @@ log = "0.4.29"
 pretty_env_logger = "0.4.0"
 
 # Redis
-redis-kiss = "0.1.4"
+redis-kiss = { version = "0.1.4", default-features = false }
 fred = "8.0.1"
 
 # Serialisation
@@ -160,8 +160,8 @@ opentelemetry-appender-tracing = "0.31.1"
 lapin = "4.7.1"
 
 # Voice
-livekit-api = "0.4.4"
+livekit-api = "=0.4.23"
-livekit-protocol = "0.7.4"
+livekit-protocol = "=0.7.7"
 livekit-runtime = "0.4.0"
 
 # Other Utilities

crates/bonfire/Cargo.toml

@@ -14,7 +14,7 @@ sentry = { workspace = true }
 lru = { workspace = true }
 ulid = { workspace = true }
 once_cell = { workspace = true }
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 lru_time_cache = { workspace = true }
 async-channel = { workspace = true }
 
@@ -30,8 +30,9 @@ serde = { workspace = true }
 
 # async
 futures = { workspace = true }
-async-tungstenite = { workspace = true, features = ["async-std-runtime"] }
+async-tungstenite = { workspace = true, features = ["tokio-runtime"] }
-async-std = { workspace = true }
+tokio = { workspace = true }
+tokio-util = { workspace = true, features = ["compat"] }
 
 # core
 revolt-result = { workspace = true }

crates/bonfire/src/events/state.rs

@@ -2,7 +2,7 @@ use std::{
     collections::{HashMap, HashSet}, num::NonZeroUsize, sync::Arc, time::Duration
 };
 
-use async_std::sync::{Mutex, RwLock};
+use tokio::sync::{Mutex, RwLock};
 use lru::LruCache;
 use lru_time_cache::{LruCache as LruTimeCache, TimedEntry};
 use revolt_database::{Channel, Member, Server, User};

crates/bonfire/src/main.rs

@@ -1,6 +1,6 @@
 use std::env;
 
-use async_std::net::TcpListener;
+use tokio::net::TcpListener;
 use revolt_presence::clear_region;
 
 #[macro_use]
@@ -12,7 +12,7 @@ pub mod events;
 mod database;
 mod websocket;
 
-#[async_std::main]
+#[tokio::main]
 async fn main() {
     // Configure requirements for Bonfire.
     revolt_config::configure!(events);
@@ -33,7 +33,7 @@ async fn main() {
 
     // Start accepting new connections and spawn a client for each connection.
     while let Ok((stream, addr)) = listener.accept().await {
-        async_std::task::spawn(async move {
+        tokio::task::spawn(async move {
             info!("User connected from {addr:?}");
             websocket::client(database::get_db(), stream, addr).await;
             info!("User disconnected from {addr:?}");

crates/bonfire/src/websocket.rs

@@ -21,11 +21,12 @@ use revolt_database::{
 };
 use revolt_presence::{create_session, delete_session};
 
-use async_std::{
+use tokio::{
     net::TcpStream,
     sync::{Mutex, RwLock},
     task::spawn,
 };
+use tokio_util::compat::{TokioAsyncReadCompatExt, Compat};
 use revolt_result::create_error;
 use sentry::Level;
 
@@ -33,8 +34,8 @@ use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
 use crate::events::state::{State, SubscriptionStateChange};
 use revolt_models::v0;
 
-type WsReader = SplitStream<WebSocketStream<TcpStream>>;
+type WsReader = SplitStream<WebSocketStream<Compat<TcpStream>>>;
-type WsWriter = SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>;
+type WsWriter = SplitSink<WebSocketStream<Compat<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.
@@ -44,7 +45,7 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
     // 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,
+        stream.compat(),
         WebsocketHandshakeCallback::from(sender),
         None,
     )

crates/core/config/Cargo.toml

@@ -13,7 +13,7 @@ repository = "https://github.com/stoatchat/stoatchat"
 anyhow = ["dep:sentry-anyhow"]
 report-macros = ["revolt-result"]
 sentry = ["dep:sentry"]
-test = ["async-std"]
+test = ["tokio"]
 default = ["sentry"]
 
 [dependencies]
@@ -26,7 +26,7 @@ serde = { workspace = true }
 
 # Async
 futures-locks = { workspace = true }
-async-std = { workspace = true, features = ["attributes"], optional = true }
+tokio = { workspace = true, optional = true }
 
 # Logging
 log = { workspace = true }

crates/core/config/src/lib.rs

@@ -586,7 +586,7 @@ macro_rules! configure {
 mod tests {
     use crate::init;
 
-    #[async_std::test]
+    #[tokio::test]
     async fn it_works() {
         init().await;
     }

crates/core/database/Cargo.toml

@@ -15,7 +15,7 @@ mongodb = ["dep:mongodb", "bson"]
 
 # ... Other
 tasks = ["isahc", "linkify", "url-escape"]
-async-std-runtime = ["async-std"]
+tokio-runtime = ["tokio"]
 rocket-impl = [
     "rocket",
     "schemars",
@@ -27,7 +27,7 @@ redis-is-patched = ["revolt-presence/redis-is-patched"]
 voice = ["livekit-api", "livekit-protocol", "livekit-runtime"]
 
 # Default Features
-default = ["mongodb", "async-std-runtime", "tasks"]
+default = ["mongodb", "tokio-runtime", "tasks"]
 
 [dependencies]
 # Core
@@ -64,7 +64,7 @@ serde = { workspace = true }
 iso8601-timestamp = { workspace = true, features = ["serde", "bson"] }
 
 # Events
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 
 # Database
 bson = { workspace = true, optional = true }
@@ -81,7 +81,7 @@ async-trait = { workspace = true }
 async-recursion = { workspace = true }
 
 # Async
-async-std = { workspace = true, features = ["attributes"], optional = true }
+tokio = { workspace = true, optional = true }
 
 # Axum Impl
 axum = { workspace = true, optional = true }

crates/core/database/src/lib.rs

@@ -25,8 +25,8 @@ pub use mongodb;
 #[macro_use]
 extern crate bson;
 
-#[cfg(not(feature = "async-std-runtime"))]
+#[cfg(not(feature = "tokio-runtime"))]
-compile_error!("async-std-runtime feature must be enabled.");
+compile_error!("tokio-runtime feature must be enabled.");
 
 #[macro_export]
 #[cfg(debug_assertions)]

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

@@ -11,7 +11,7 @@ auto_derived!(
 
 #[cfg(test)]
 mod tests {
-    #[async_std::test]
+    #[tokio::test]
     async fn migrate() {
         database_test!(|db| async move {
             // Initialise the database

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

@@ -452,7 +452,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
         warn!("This is a destructive operation and will wipe existing permission data (excl. defaults for SendMessage).");
         warn!("Taking a backup is advised.");
         warn!("Continuing in 10 seconds...");
-        async_std::task::sleep(Duration::from_secs(10)).await;
+        tokio::time::sleep(Duration::from_secs(10)).await;
 
         let servers = db.col::<Document>("servers");
         let mut cursor = servers.find(doc! {}).await.unwrap();

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

@@ -164,7 +164,7 @@ impl Bot {
 mod tests {
     use crate::{Bot, FieldsBot, PartialBot, User};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn crud() {
         database_test!(|db| async move {
             let owner = User::create(&db, "Owner".to_string(), None, None)

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

@@ -128,7 +128,7 @@ impl Webhook {
 mod tests {
     use crate::{FieldsWebhook, PartialWebhook, Webhook};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn crud() {
         database_test!(|db| async move {
             let webhook_id = "webhook";

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

@@ -787,7 +787,7 @@ mod tests {
 
     use crate::{fixture, util::permissions::DatabasePermissionQuery};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn permissions_group_channel() {
         database_test!(|db| async move {
             fixture!(db, "group_with_members",
@@ -813,7 +813,7 @@ mod tests {
         });
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn permissions_text_channel() {
         database_test!(|db| async move {
             fixture!(db, "server_with_roles",

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

@@ -320,7 +320,7 @@ mod tests {
 
     use crate::{Member, PartialMember, RemovalIntention, Server, User};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn muted_member_rejoin() {
         database_test!(|db| async move {
             match db {

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

@@ -420,7 +420,7 @@ mod tests {
 
     use crate::{fixture, util::permissions::DatabasePermissionQuery};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn permissions() {
         database_test!(|db| async move {
             fixture!(db, "server_with_roles",

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

@@ -926,7 +926,7 @@ mod tests {
         assert!(User::validate_username(username_https).is_err());
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn username_sanitisation_clean() {
         let username_clean = "Test";
 
@@ -936,7 +936,7 @@ mod tests {
         assert_eq!(username_clean, username_clean_sanitised.unwrap());
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn username_sanitisation_homoglyphs() {
         let username_homoglyphs = "𝔽𝕌Ňℕｙ";
 
@@ -947,7 +947,7 @@ mod tests {
         assert_eq!("funny", username_homoglyphs_sanitised);
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn username_sanitisation_padding() {
         let username_padding = "a";
 
@@ -956,7 +956,7 @@ mod tests {
         assert_eq!("a_", username);
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn create_user() {
         use revolt_result::Result;
 

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

@@ -305,6 +305,6 @@ pub async fn worker(db: Database, amqp: AMQP) {
         }
 
         // Sleep for an arbitrary amount of time.
-        async_std::task::sleep(Duration::from_secs(1)).await;
+        tokio::time::sleep(Duration::from_secs(1)).await;
     }
 }

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

@@ -82,6 +82,6 @@ pub async fn worker(db: Database) {
         }
 
         // Sleep for an arbitrary amount of time.
-        async_std::task::sleep(Duration::from_secs(1)).await;
+        tokio::time::sleep(Duration::from_secs(1)).await;
     }
 }

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

@@ -2,7 +2,7 @@
 
 use crate::{Database, AMQP};
 
-use async_std::task;
+use tokio::task;
 use std::time::Instant;
 
 const WORKER_COUNT: usize = 5;

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

@@ -7,11 +7,11 @@ use revolt_config::config;
 use revolt_result::Result;
 
 use async_lock::Semaphore;
-use async_std::task::spawn;
 use deadqueue::limited::Queue;
 use once_cell::sync::Lazy;
 use revolt_models::v0::Embed;
 use std::{collections::HashSet, sync::Arc};
+use tokio::task::spawn;
 
 use isahc::prelude::*;
 
@@ -159,6 +159,7 @@ pub async fn generate(
         .await
         .into_iter()
         .flatten()
+        .flatten()
         .collect::<Vec<Embed>>();
 
     // Prevent database update when no embeds are found.

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

@@ -5,7 +5,7 @@ use revolt_result::{create_error, Result};
 #[cfg(feature = "rocket-impl")]
 use revolt_result::Error;
 
-use async_std::sync::Mutex;
+use tokio::sync::Mutex;
 use once_cell::sync::Lazy;
 use serde::{Deserialize, Serialize};
 

crates/core/permissions/Cargo.toml

@@ -18,7 +18,7 @@ try-from-primitive = ["dep:num_enum"]
 
 [dev-dependencies]
 # Async
-async-std = { workspace = true, features = ["attributes"] }
+tokio = { workspace = true }
 
 [dependencies]
 # Core

crates/core/permissions/src/test.rs

@@ -4,7 +4,7 @@ use crate::{
     DEFAULT_PERMISSION_SERVER, DEFAULT_PERMISSION_VIEW_ONLY,
 };
 
-#[async_std::test]
+#[tokio::test]
 async fn validate_user_permissions() {
     /// Scenario in which we are friends with a user
     /// and we have a DM channel open with them
@@ -102,7 +102,7 @@ async fn validate_user_permissions() {
     }
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn validate_group_permissions() {
     /// Scenario in which we are in a group channel with only talking permission
     struct Scenario {}
@@ -202,7 +202,7 @@ async fn validate_group_permissions() {
     }
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn validate_server_permissions() {
     /// Scenario in which we are in a server channel where:
     /// - the server grants reading history and sending messages by default
@@ -314,7 +314,7 @@ async fn validate_server_permissions() {
     }
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn validate_timed_out_member() {
     /// Scenario in which we are in a server that we have been timed out from
     struct Scenario {}

crates/core/presence/Cargo.toml

@@ -14,7 +14,7 @@ redis-is-patched = []
 
 [dev-dependencies]
 # Async
-async-std = { workspace = true, features = ["attributes"] }
+tokio = { workspace = true }
 
 # Config for loading Redis URI
 revolt-config = { workspace = true }
@@ -26,4 +26,4 @@ rand = { workspace = true }
 once_cell = { workspace = true }
 
 # Redis
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }

crates/core/presence/src/lib.rs

@@ -195,7 +195,7 @@ mod tests {
     use crate::{clear_region, create_session, delete_session, filter_online, is_online};
     use rand::Rng;
 
-    #[async_std::test]
+    #[tokio::test]
     async fn it_works() {
         revolt_config::config().await;
 

crates/daemons/crond/Cargo.toml

@@ -18,7 +18,7 @@ tokio = { workspace = true }
 futures = { workspace = true }
 
 # Redis
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 
 # RabbitMQ
 lapin = { workspace = true }

crates/daemons/pushd/Cargo.toml

@@ -20,7 +20,7 @@ fcm_v1 = { workspace = true }
 web-push = { workspace = true }
 isahc = { workspace = true, features = ["json"], optional = true }
 revolt_a2 = { workspace = true, features = ["ring"] }
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 tokio = { workspace = true }
 async-trait = { workspace = true }
 ulid = { workspace = true }

crates/daemons/voice-ingress/Cargo.toml

@@ -13,7 +13,7 @@ log = { workspace = true }
 sentry = { workspace = true }
 lru = { workspace = true }
 ulid = { workspace = true }
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 chrono = { workspace = true }
 
 # Serde
@@ -27,11 +27,6 @@ rocket_empty = { workspace = true }
 
 # Async
 futures = { workspace = true }
-async-std = { workspace = true, features = [
-    "tokio1",
-    "tokio02",
-    "attributes",
-] }
 
 # Core
 revolt-result = { workspace = true, features = ["rocket"] }

crates/delta/Cargo.toml

@@ -11,7 +11,7 @@ publish = false
 [dependencies]
 # Test
 rand = { workspace = true }
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 
 # Utility
 lru = { workspace = true }
@@ -42,11 +42,7 @@ futures = { workspace = true }
 chrono = { workspace = true }
 async-channel = { workspace = true }
 reqwest = { workspace = true, features = ["json"] }
-async-std = { workspace = true, features = [
+tokio = { workspace = true }
-    "tokio1",
-    "tokio02",
-    "attributes",
-] }
 
 # internal util
 lettre = { workspace = true }

crates/delta/src/routes/account/create_account.rs

@@ -2,7 +2,7 @@
 //! POST /account/create
 use std::time::Duration;
 
-use async_std::task::sleep;
+use tokio::time::sleep;
 use revolt_config::config;
 use revolt_database::{
     util::{

crates/delta/src/routes/account/resend_verification.rs

@@ -2,7 +2,7 @@
 //! POST /account/reverify
 use std::time::Duration;
 
-use async_std::task::sleep;
+use tokio::time::sleep;
 use rocket::{serde::json::Json, State};
 use rocket_empty::EmptyResponse;
 use revolt_result::Result;

crates/delta/src/routes/account/send_password_reset.rs

@@ -2,7 +2,7 @@
 //! POST /account/reset_password
 use std::time::Duration;
 
-use async_std::task::sleep;
+use tokio::time::sleep;
 use rocket::serde::json::Json;
 use rocket::State;
 use rocket_empty::EmptyResponse;

crates/delta/src/routes/session/login.rs

@@ -3,7 +3,7 @@
 use std::ops::Add;
 use std::time::Duration;
 
-use async_std::task::sleep;
+use tokio::time::sleep;
 use iso8601_timestamp::Timestamp;
 use revolt_database::{
     util::{email::normalise_email, password::assert_safe},