Compare commits

..
Author SHA1 Message Date
Paul Makles d677716f93 docs: update development guide 2024-06-11 14:16:44 +01:00
Paul Makles 8099310f89 fix: wait on unwaited futures 2024-06-11 14:16:35 +01:00
Paul Makles 5c40f66010 fix: wrong match keyword causing disconnects on typing event 2024-06-11 14:01:59 +01:00
Paul Makles 8e7dd21bce docs: add note about creating GitHub release [skip ci] 2024-06-11 13:08:24 +01:00
Paul Makles 4868205df4 chore: bump version to 0.7.4 2024-06-11 12:52:45 +01:00
Paul Makles 96fb0eecca fix: don't allow sending typing notification to unknown channel
closes #177
2024-06-11 12:51:52 +01:00
Paul Makles 2cb20618da feat(bonfire): add disconnection mechanism
feat(bonfire): handle session deletion and logout events
feat(core): trigger logout on bot token reset
2024-06-11 12:38:32 +01:00
Paul Makles 92e948aabc chore: update README [skip ci] 2024-06-09 14:11:30 +01:00
19 changed files with 202 additions and 99 deletions
Generated
+9 -8
View File
@@ -3408,10 +3408,11 @@ dependencies = [
[[package]]
name = "revolt-bonfire"
version = "0.7.3"
version = "0.7.4"
dependencies = [
"async-std",
"async-tungstenite",
"authifier",
"bincode",
"fred",
"futures",
@@ -3435,7 +3436,7 @@ dependencies = [
[[package]]
name = "revolt-config"
version = "0.7.3"
version = "0.7.4"
dependencies = [
"async-std",
"cached",
@@ -3451,7 +3452,7 @@ dependencies = [
[[package]]
name = "revolt-database"
version = "0.7.3"
version = "0.7.4"
dependencies = [
"async-lock",
"async-recursion",
@@ -3497,7 +3498,7 @@ dependencies = [
[[package]]
name = "revolt-delta"
version = "0.7.3"
version = "0.7.4"
dependencies = [
"async-channel",
"async-std",
@@ -3543,7 +3544,7 @@ dependencies = [
[[package]]
name = "revolt-models"
version = "0.7.3"
version = "0.7.4"
dependencies = [
"indexmap",
"iso8601-timestamp 0.2.11",
@@ -3560,7 +3561,7 @@ dependencies = [
[[package]]
name = "revolt-permissions"
version = "0.7.3"
version = "0.7.4"
dependencies = [
"async-std",
"async-trait",
@@ -3575,7 +3576,7 @@ dependencies = [
[[package]]
name = "revolt-presence"
version = "0.7.3"
version = "0.7.4"
dependencies = [
"async-std",
"log",
@@ -3586,7 +3587,7 @@ dependencies = [
[[package]]
name = "revolt-result"
version = "0.7.3"
version = "0.7.4"
dependencies = [
"revolt_okapi",
"revolt_rocket_okapi",
+39
View File
@@ -48,7 +48,29 @@ cp .env.example .env
# (optionally) copy the default configuration file
cp crates/core/config/Revolt.toml Revolt.toml
# configure as necessary...
```
You may want to copy the following configuration:
```toml
# Revolt.toml
[database]
mongodb = "mongodb://localhost"
redis = "redis://localhost"
[hosts]
app = "http://local.revolt.chat"
api = "http://local.revolt.chat:8000"
events = "ws://local.revolt.chat:9000"
autumn = "http://local.revolt.chat:3000"
january = "http://local.revolt.chat:7000"
voso_legacy = ""
voso_legacy_ws = ""
```
Then continue:
```bash
# start other necessary services
docker compose up -d
@@ -105,6 +127,23 @@ Tag and push a new release by running:
just release
```
If you have bumped the crate versions, proceed to [GitHub releases](https://github.com/revoltchat/backend/releases/new) to create a changelog.
## Testing
First, start the required services:
```sh
docker compose -f docker-compose.db.yml up -d
```
Now run tests for whichever database:
```sh
TEST_DB=REFERENCE cargo nextest run
TEST_DB=MONGOBD cargo nextest run
```
## License
The Revolt backend is generally licensed under the [GNU Affero General Public License v3.0](https://github.com/revoltchat/backend/blob/master/LICENSE).
+3 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-bonfire"
version = "0.7.3"
version = "0.7.4"
license = "AGPL-3.0-or-later"
edition = "2021"
@@ -34,11 +34,12 @@ async-std = { version = "1.8.0", features = [
] }
# core
authifier = { version = "1.0.8" }
revolt-result = { path = "../core/result" }
revolt-models = { path = "../core/models" }
revolt-config = { path = "../core/config" }
revolt-database = { path = "../core/database" }
revolt-permissions = { version = "0.7.3", path = "../core/permissions" }
revolt-permissions = { version = "0.7.4", path = "../core/permissions" }
revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
# redis
+24 -24
View File
@@ -21,11 +21,11 @@ impl Cache {
let server = self.servers.get(server);
let mut query =
DatabasePermissionQuery::new(db, self.users.get(&self.user_id).unwrap())
.channel(&channel);
.channel(channel);
// let mut perms = perms(self.users.get(&self.user_id).unwrap()).channel(channel);
if let Some(member) = member {
query = query.member(&member);
query = query.member(member);
}
if let Some(server) = server {
@@ -182,19 +182,19 @@ impl State {
users.push(user.into_self().await);
// Set subscription state internally.
self.reset_state();
self.insert_subscription(self.private_topic.clone());
self.reset_state().await;
self.insert_subscription(self.private_topic.clone()).await;
for user in &users {
self.insert_subscription(user.id.clone());
self.insert_subscription(user.id.clone()).await;
}
for server in &servers {
self.insert_subscription(server.id.clone());
self.insert_subscription(server.id.clone()).await;
}
for channel in &channels {
self.insert_subscription(channel.id().to_string());
self.insert_subscription(channel.id().to_string()).await;
}
Ok(EventV1::Ready {
@@ -236,11 +236,11 @@ impl State {
let mut bulk_events = vec![];
for id in added_channels {
self.insert_subscription(id);
self.insert_subscription(id).await;
}
for id in removed_channels {
self.remove_subscription(&id);
self.remove_subscription(&id).await;
self.cache.channels.remove(&id);
bulk_events.push(EventV1::ChannelDelete { id });
@@ -263,7 +263,7 @@ impl State {
.channels
.insert(channel.id().to_string(), channel.clone());
self.insert_subscription(channel.id().to_string());
self.insert_subscription(channel.id().to_string()).await;
bulk_events.push(EventV1::ChannelCreate(channel.into()));
}
}
@@ -336,7 +336,7 @@ impl State {
match event {
EventV1::ChannelCreate(channel) => {
let id = channel.id().to_string();
self.insert_subscription(id.clone());
self.insert_subscription(id.clone()).await;
self.cache.channels.insert(id, channel.clone().into());
}
EventV1::ChannelUpdate {
@@ -376,17 +376,17 @@ impl State {
}
}
EventV1::ChannelDelete { id } => {
self.remove_subscription(id);
self.remove_subscription(id).await;
self.cache.channels.remove(id);
}
EventV1::ChannelGroupJoin { user, .. } => {
self.insert_subscription(user.clone());
self.insert_subscription(user.clone()).await;
}
EventV1::ChannelGroupLeave { id, user, .. } => {
if user == &self.cache.user_id {
self.remove_subscription(id);
self.remove_subscription(id).await;
} else if !self.cache.can_subscribe_to_user(user) {
self.remove_subscription(user);
self.remove_subscription(user).await;
}
}
@@ -396,7 +396,7 @@ impl State {
channels,
emojis: _,
} => {
self.insert_subscription(id.clone());
self.insert_subscription(id.clone()).await;
self.cache.servers.insert(id.clone(), server.clone().into());
let member = Member {
id: MemberCompositeKey {
@@ -435,11 +435,11 @@ impl State {
}
EventV1::ServerMemberLeave { id, user } => {
if user == &self.cache.user_id {
self.remove_subscription(id);
self.remove_subscription(id).await;
if let Some(server) = self.cache.servers.remove(id) {
for channel in &server.channels {
self.remove_subscription(channel);
self.remove_subscription(channel).await;
self.cache.channels.remove(channel);
}
}
@@ -447,11 +447,11 @@ impl State {
}
}
EventV1::ServerDelete { id } => {
self.remove_subscription(id);
self.remove_subscription(id).await;
if let Some(server) = self.cache.servers.remove(id) {
for channel in &server.channels {
self.remove_subscription(channel);
self.remove_subscription(channel).await;
self.cache.channels.remove(channel);
}
}
@@ -524,9 +524,9 @@ impl State {
self.cache.users.insert(id.clone(), user.clone().into());
if self.cache.can_subscribe_to_user(id) {
self.insert_subscription(id.clone());
self.insert_subscription(id.clone()).await;
} else {
self.remove_subscription(id);
self.remove_subscription(id).await;
}
}
@@ -540,11 +540,11 @@ impl State {
// Sub / unsub accordingly.
if let Some(id) = queue_add {
self.insert_subscription(id);
self.insert_subscription(id).await;
}
if let Some(id) = queue_remove {
self.remove_subscription(&id);
self.remove_subscription(&id).await;
}
true
+25 -21
View File
@@ -1,5 +1,9 @@
use std::collections::{HashMap, HashSet};
use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use async_std::sync::RwLock;
use lru::LruCache;
use revolt_database::{Channel, Member, Server, User};
@@ -58,14 +62,15 @@ impl Default for Cache {
pub struct State {
pub cache: Cache,
pub session_id: String,
pub private_topic: String,
subscribed: HashSet<String>,
state: SubscriptionStateChange,
pub state: SubscriptionStateChange,
pub subscribed: Arc<RwLock<HashSet<String>>>,
}
impl State {
/// Create state from User
pub fn from(user: User) -> State {
pub fn from(user: User, session_id: String) -> State {
let mut subscribed = HashSet::new();
let private_topic = format!("{}!", user.id);
subscribed.insert(private_topic.clone());
@@ -80,22 +85,24 @@ impl State {
State {
cache,
subscribed,
subscribed: Arc::new(RwLock::new(subscribed)),
session_id,
private_topic,
state: SubscriptionStateChange::Reset,
}
}
/// Apply currently queued state
pub fn apply_state(&mut self) -> SubscriptionStateChange {
pub async fn apply_state(&mut self) -> SubscriptionStateChange {
let state = std::mem::replace(&mut self.state, SubscriptionStateChange::None);
let mut subscribed = self.subscribed.write().await;
if let SubscriptionStateChange::Change { add, remove } = &state {
for id in add {
self.subscribed.insert(id.clone());
subscribed.insert(id.clone());
}
for id in remove {
self.subscribed.remove(id);
subscribed.remove(id);
}
}
@@ -107,20 +114,16 @@ impl State {
self.cache.users.get(&self.cache.user_id).unwrap().clone()
}
/// Iterate through all subscriptions
pub fn iter_subscriptions(&self) -> std::collections::hash_set::Iter<'_, std::string::String> {
self.subscribed.iter()
}
/// Reset the current state
pub fn reset_state(&mut self) {
pub async fn reset_state(&mut self) {
self.state = SubscriptionStateChange::Reset;
self.subscribed.clear();
self.subscribed.write().await.clear();
}
/// Add a new subscription
pub fn insert_subscription(&mut self, subscription: String) {
if self.subscribed.contains(&subscription) {
pub async fn insert_subscription(&mut self, subscription: String) {
let mut subscribed = self.subscribed.write().await;
if subscribed.contains(&subscription) {
return;
}
@@ -137,12 +140,13 @@ impl State {
SubscriptionStateChange::Reset => {}
}
self.subscribed.insert(subscription);
subscribed.insert(subscription);
}
/// Remove existing subscription
pub fn remove_subscription(&mut self, subscription: &str) {
if !self.subscribed.contains(&subscription.to_string()) {
pub async fn remove_subscription(&mut self, subscription: &str) {
let mut subscribed = self.subscribed.write().await;
if !subscribed.contains(&subscription.to_string()) {
return;
}
@@ -159,6 +163,6 @@ impl State {
SubscriptionStateChange::Reset => panic!("Should not remove during a reset!"),
}
self.subscribed.remove(subscription);
subscribed.remove(subscription);
}
}
+59 -10
View File
@@ -1,6 +1,7 @@
use std::net::SocketAddr;
use std::{collections::HashSet, net::SocketAddr, sync::Arc};
use async_tungstenite::WebSocketStream;
use authifier::AuthifierEvent;
use fred::{
interfaces::{ClientLike, EventInterface, PubsubInterface},
types::RedisConfig,
@@ -18,7 +19,10 @@ use revolt_database::{
};
use revolt_presence::{create_session, delete_session};
use async_std::{net::TcpStream, sync::Mutex};
use async_std::{
net::TcpStream,
sync::{Mutex, RwLock},
};
use revolt_result::create_error;
use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
@@ -43,6 +47,7 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
else {
return;
};
// Verify we've received a valid config, otherwise we should just drop the connection.
let Ok(mut config) = receiver.await else {
return;
@@ -73,17 +78,19 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
write.send(config.encode(&create_error!(InvalidSession))).await.ok();
return;
};
let user = match User::from_token(db, token, UserHint::Any).await {
let (user, session_id) = 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 mut state = State::from(user, session_id);
let user_id = state.cache.user_id.clone();
// Notify socket we have authenticated.
@@ -99,6 +106,7 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
let Ok(ready_payload) = state.generate_ready_payload(db).await else {
return;
};
if write.send(config.encode(&ready_payload)).await.is_err() {
return;
}
@@ -113,10 +121,12 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
{
let write = Mutex::new(write);
let subscribed = state.subscribed.clone();
// Create a PubSub connection to poll on.
let listener = listener(db, &mut state, addr, &config, &write).fuse();
// Read from WebSocket stream.
let worker = worker(addr, user_id.clone(), &config, read, &write).fuse();
let worker = worker(addr, subscribed, user_id.clone(), &config, read, &write).fuse();
// Pin both tasks.
pin_mut!(listener, worker);
@@ -154,10 +164,12 @@ async fn listener(
let mut message_rx = subscriber.message_rx();
loop {
// Check for state changes for subscriptions.
match state.apply_state() {
match state.apply_state().await {
SubscriptionStateChange::Reset => {
subscriber.unsubscribe_all().await.unwrap();
for id in state.iter_subscriptions() {
let subscribed = state.subscribed.read().await;
for id in subscribed.iter() {
subscriber.subscribe(id).await.unwrap();
}
@@ -189,6 +201,7 @@ async fn listener(
}) else {
return;
};
let event = match *REDIS_PAYLOAD_TYPE {
PayloadType::Json => message
.value
@@ -203,13 +216,34 @@ async fn listener(
.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 {
continue;
if let EventV1::Auth(auth) = &event {
if let AuthifierEvent::DeleteSession { session_id, .. } = auth {
if &state.session_id == session_id {
event = EventV1::Logout;
}
} else if let AuthifierEvent::DeleteAllSessions {
exclude_session_id, ..
} = auth
{
if let Some(excluded) = exclude_session_id {
if &state.session_id != excluded {
event = EventV1::Logout;
}
} else {
event = EventV1::Logout;
}
}
} else {
let should_send = state.handle_incoming_event_v1(db, &mut event).await;
if !should_send {
continue;
}
}
let result = write.lock().await.send(config.encode(&event)).await;
@@ -218,6 +252,11 @@ async fn listener(
if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
warn!("Error while sending an event to {addr:?}: {e:?}");
}
return;
}
if let EventV1::Logout = event {
return;
}
}
@@ -225,6 +264,7 @@ async fn listener(
async fn worker(
addr: SocketAddr,
subscribed: Arc<RwLock<HashSet<String>>>,
user_id: String,
config: &ProtocolConfiguration,
mut read: WsReader,
@@ -247,8 +287,13 @@ async fn worker(
let Ok(payload) = config.decode(&msg) else {
continue;
};
match payload {
ClientMessage::BeginTyping { channel } => {
if !subscribed.read().await.contains(&channel) {
continue;
}
EventV1::ChannelStartTyping {
id: channel.clone(),
user: user_id.clone(),
@@ -257,6 +302,10 @@ async fn worker(
.await;
}
ClientMessage::EndTyping { channel } => {
if !subscribed.read().await.contains(&channel) {
continue;
}
EventV1::ChannelStopTyping {
id: channel.clone(),
user: user_id.clone(),
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-config"
version = "0.7.3"
version = "0.7.4"
edition = "2021"
license = "MIT"
authors = ["Paul Makles <me@insrt.uk>"]
+6 -6
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-database"
version = "0.7.3"
version = "0.7.4"
edition = "2021"
license = "AGPL-3.0-or-later"
authors = ["Paul Makles <me@insrt.uk>"]
@@ -23,13 +23,13 @@ default = ["mongodb", "async-std-runtime", "tasks"]
[dependencies]
# Core
revolt-config = { version = "0.7.3", path = "../config" }
revolt-result = { version = "0.7.3", path = "../result" }
revolt-models = { version = "0.7.3", path = "../models", features = [
revolt-config = { version = "0.7.4", path = "../config" }
revolt-result = { version = "0.7.4", path = "../result" }
revolt-models = { version = "0.7.4", path = "../models", features = [
"validator",
] }
revolt-presence = { version = "0.7.3", path = "../presence" }
revolt-permissions = { version = "0.7.3", path = "../permissions", features = [
revolt-presence = { version = "0.7.4", path = "../presence" }
revolt-permissions = { version = "0.7.4", path = "../permissions", features = [
"serde",
"bson",
] }
@@ -48,6 +48,8 @@ pub enum EventV1 {
/// Successfully authenticated
Authenticated,
/// Logged out
Logout,
/// Basic data to cache
Ready {
users: Vec<User>,
@@ -2,7 +2,7 @@ use revolt_config::config;
use revolt_result::Result;
use ulid::Ulid;
use crate::{BotInformation, Database, PartialUser, User};
use crate::{events::client::EventV1, BotInformation, Database, PartialUser, User};
auto_derived_partial!(
/// Bot
@@ -142,6 +142,10 @@ impl Bot {
db.update_bot(&self.id, &partial, remove).await?;
if partial.token.is_some() {
EventV1::Logout.private(self.id.clone()).await;
}
self.apply_options(partial);
Ok(())
}
+11 -7
View File
@@ -276,23 +276,27 @@ impl User {
Ok(username)
}
/// Find a user from a given token and hint
/// Find a user and session ID from a given token and hint
#[async_recursion]
pub async fn from_token(db: &Database, token: &str, hint: UserHint) -> Result<User> {
pub async fn from_token(db: &Database, token: &str, hint: UserHint) -> Result<(User, String)> {
match hint {
UserHint::Bot => {
UserHint::Bot => Ok((
db.fetch_user(
&db.fetch_bot_by_token(token)
.await
.map_err(|_| create_error!(InvalidSession))?
.id,
)
.await
.await?,
String::new(),
)),
UserHint::User => {
let session = db.fetch_session_by_token(token).await?;
Ok((db.fetch_user(&session.user_id).await?, session.id))
}
UserHint::User => db.fetch_user_by_token(token).await,
UserHint::Any => {
if let Ok(user) = User::from_token(db, token, UserHint::User).await {
Ok(user)
if let Ok(result) = User::from_token(db, token, UserHint::User).await {
Ok(result)
} else {
User::from_token(db, token, UserHint::Bot).await
}
+3 -2
View File
@@ -1,3 +1,4 @@
use authifier::models::Session;
use revolt_result::Result;
use crate::{FieldsUser, PartialUser, RelationshipStatus, User};
@@ -16,8 +17,8 @@ pub trait AbstractUsers: Sync + Send {
/// Fetch a user from the database by their username
async fn fetch_user_by_username(&self, username: &str, discriminator: &str) -> Result<User>;
/// Fetch a user from the database by their session token
async fn fetch_user_by_token(&self, token: &str) -> Result<User>;
/// Fetch a session from the database by token
async fn fetch_session_by_token(&self, token: &str) -> Result<Session>;
/// Fetch multiple users by their ids
async fn fetch_users<'a>(&self, ids: &'a [String]) -> Result<Vec<User>>;
@@ -46,10 +46,9 @@ impl AbstractUsers for MongoDb {
.ok_or_else(|| create_error!(NotFound))
}
/// Fetch a user from the database by their session token
async fn fetch_user_by_token(&self, token: &str) -> Result<User> {
let session = self
.col::<Session>("sessions")
/// Fetch a session from the database by token
async fn fetch_session_by_token(&self, token: &str) -> Result<Session> {
self.col::<Session>("sessions")
.find_one(
doc! {
"token": token
@@ -58,9 +57,7 @@ impl AbstractUsers for MongoDb {
)
.await
.map_err(|_| create_database_error!("find_one", "sessions"))?
.ok_or_else(|| create_error!(InvalidSession))?;
self.fetch_user(&session.user_id).await
.ok_or_else(|| create_error!(InvalidSession))
}
/// Fetch multiple users by their ids
@@ -1,3 +1,4 @@
use authifier::models::Session;
use revolt_result::Result;
use crate::{FieldsUser, PartialUser, RelationshipStatus, User};
@@ -40,8 +41,8 @@ impl AbstractUsers for ReferenceDb {
.ok_or_else(|| create_error!(NotFound))
}
/// Fetch a user from the database by their session token
async fn fetch_user_by_token(&self, _token: &str) -> Result<User> {
/// Fetch a session from the database by token
async fn fetch_session_by_token(&self, _token: &str) -> Result<Session> {
todo!()
}
+3 -3
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-models"
version = "0.7.3"
version = "0.7.4"
edition = "2021"
license = "MIT"
authors = ["Paul Makles <me@insrt.uk>"]
@@ -19,8 +19,8 @@ default = ["serde", "partials", "rocket"]
[dependencies]
# Core
revolt-config = { version = "0.7.3", path = "../config" }
revolt-permissions = { version = "0.7.3", path = "../permissions" }
revolt-config = { version = "0.7.4", path = "../config" }
revolt-permissions = { version = "0.7.4", path = "../permissions" }
# Utility
regex = "1"
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-permissions"
version = "0.7.3"
version = "0.7.4"
edition = "2021"
license = "MIT"
authors = ["Paul Makles <me@insrt.uk>"]
@@ -21,7 +21,7 @@ async-std = { version = "1.8.0", features = ["attributes"] }
[dependencies]
# Core
revolt-result = { version = "0.7.3", path = "../result" }
revolt-result = { version = "0.7.4", path = "../result" }
# Utility
auto_ops = "0.3.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-presence"
version = "0.7.3"
version = "0.7.4"
edition = "2021"
license = "AGPL-3.0-or-later"
authors = ["Paul Makles <me@insrt.uk>"]
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-result"
version = "0.7.3"
version = "0.7.4"
edition = "2021"
license = "MIT"
authors = ["Paul Makles <me@insrt.uk>"]
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-delta"
version = "0.7.3"
version = "0.7.4"
license = "AGPL-3.0-or-later"
authors = ["Paul Makles <paulmakles@gmail.com>"]
edition = "2018"