Compare commits

...
11 Commits
18 changed files with 226 additions and 104 deletions
Generated
+8 -8
View File
@@ -3408,7 +3408,7 @@ dependencies = [
[[package]]
name = "revolt-bonfire"
version = "0.7.4"
version = "0.7.5"
dependencies = [
"async-std",
"async-tungstenite",
@@ -3436,7 +3436,7 @@ dependencies = [
[[package]]
name = "revolt-config"
version = "0.7.4"
version = "0.7.5"
dependencies = [
"async-std",
"cached",
@@ -3452,7 +3452,7 @@ dependencies = [
[[package]]
name = "revolt-database"
version = "0.7.4"
version = "0.7.5"
dependencies = [
"async-lock",
"async-recursion",
@@ -3498,7 +3498,7 @@ dependencies = [
[[package]]
name = "revolt-delta"
version = "0.7.4"
version = "0.7.5"
dependencies = [
"async-channel",
"async-std",
@@ -3544,7 +3544,7 @@ dependencies = [
[[package]]
name = "revolt-models"
version = "0.7.4"
version = "0.7.5"
dependencies = [
"indexmap",
"iso8601-timestamp 0.2.11",
@@ -3561,7 +3561,7 @@ dependencies = [
[[package]]
name = "revolt-permissions"
version = "0.7.4"
version = "0.7.5"
dependencies = [
"async-std",
"async-trait",
@@ -3576,7 +3576,7 @@ dependencies = [
[[package]]
name = "revolt-presence"
version = "0.7.4"
version = "0.7.5"
dependencies = [
"async-std",
"log",
@@ -3587,7 +3587,7 @@ dependencies = [
[[package]]
name = "revolt-result"
version = "0.7.4"
version = "0.7.5"
dependencies = [
"revolt_okapi",
"revolt_rocket_okapi",
+24
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,8 @@ 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:
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-bonfire"
version = "0.7.4"
version = "0.7.5"
license = "AGPL-3.0-or-later"
edition = "2021"
@@ -39,7 +39,7 @@ 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.4", path = "../core/permissions" }
revolt-permissions = { version = "0.7.5", 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
+87 -32
View File
@@ -24,6 +24,7 @@ use async_std::{
sync::{Mutex, RwLock},
};
use revolt_result::create_error;
use sentry::Level;
use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
use crate::events::state::{State, SubscriptionStateChange};
@@ -52,6 +53,7 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
let Ok(mut config) = receiver.await else {
return;
};
info!(
"User {addr:?} provided protocol configuration (version = {}, format = {:?})",
config.get_protocol_version(),
@@ -63,10 +65,8 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
// 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())
{
while let Ok(Some(message)) = read.try_next().await {
if let Ok(ClientMessage::Authenticate { token }) = config.decode(&message) {
config.set_session_token(token);
break;
}
@@ -75,7 +75,10 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
// Try to authenticate the user.
let Some(token) = config.get_session_token().as_ref() else {
write.send(config.encode(&create_error!(InvalidSession))).await.ok();
write
.send(config.encode(&create_error!(InvalidSession)))
.await
.ok();
return;
};
@@ -94,20 +97,24 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
let user_id = state.cache.user_id.clone();
// Notify socket we have authenticated.
if write
.send(config.encode(&EventV1::Authenticated))
.await
.is_err()
{
if let Err(err) = write.send(config.encode(&EventV1::Authenticated)).await {
error!("Failed to write: {err:?}");
sentry::capture_error(&err);
return;
}
// Download required data to local cache and send Ready payload.
let Ok(ready_payload) = state.generate_ready_payload(db).await else {
return;
let ready_payload = match state.generate_ready_payload(db).await {
Ok(ready_payload) => ready_payload,
Err(err) => {
sentry::capture_error(&err);
return;
}
};
if write.send(config.encode(&ready_payload)).await.is_err() {
if let Err(err) = write.send(config.encode(&ready_payload)).await {
error!("Failed to write: {err:?}");
sentry::capture_error(&err);
return;
}
@@ -154,23 +161,41 @@ async fn listener(
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;
let subscriber = match fred::types::Builder::from_config(redis_config).build_subscriber_client()
{
Ok(subscriber) => subscriber,
Err(err) => {
error!("Failed to build a subscriber: {err:?}");
sentry::capture_error(&err);
return;
}
};
if subscriber.init().await.is_err() {
if let Err(err) = subscriber.init().await {
error!("Failed to init subscriber: {err:?}");
sentry::capture_error(&err);
return;
};
let mut message_rx = subscriber.message_rx();
loop {
// Check for state changes for subscriptions.
match state.apply_state().await {
SubscriptionStateChange::Reset => {
subscriber.unsubscribe_all().await.unwrap();
if let Err(err) = subscriber.unsubscribe_all().await {
error!("Unsubscribe all failed: {err:?}");
sentry::capture_error(&err);
return;
}
let subscribed = state.subscribed.read().await;
for id in subscribed.iter() {
subscriber.subscribe(id).await.unwrap();
if let Err(err) = subscriber.subscribe(id).await {
error!("Subscribe failed: {err:?}");
sentry::capture_error(&err);
return;
}
}
#[cfg(debug_assertions)]
@@ -181,25 +206,35 @@ async fn listener(
#[cfg(debug_assertions)]
info!("{addr:?} unsubscribing from {id}");
subscriber.unsubscribe(id).await.unwrap();
if let Err(err) = subscriber.unsubscribe(id).await {
error!("Unsubscribe failed: {err:?}");
sentry::capture_error(&err);
return;
}
}
for id in add {
#[cfg(debug_assertions)]
info!("{addr:?} subscribing to {id}");
subscriber.subscribe(id).await.unwrap();
if let Err(err) = subscriber.subscribe(id).await {
error!("Subscribe failed: {err:?}");
sentry::capture_error(&err);
return;
}
}
}
SubscriptionStateChange::None => {}
}
// Handle incoming events.
let Ok(message) = message_rx.recv().await.map_err(|e| {
warn!("Error while consuming pub/sub messages: {e:?}");
sentry::capture_error(&e);
}) else {
return;
let message = match message_rx.recv().await {
Ok(message) => message,
Err(e) => {
error!("Error while consuming pub/sub messages: {e:?}");
sentry::capture_error(&e);
return;
}
};
let event = match *REDIS_PAYLOAD_TYPE {
@@ -218,7 +253,17 @@ async fn listener(
};
let Some(mut event) = event else {
warn!("Failed to deserialise an event for {}!", message.channel);
let err = format!(
"Failed to deserialise an event for {}! Introspection: `{:?}`",
message.channel,
message
.value
.as_string()
.map(|x| x.chars().take(32).collect::<String>())
);
error!("{}", err);
sentry::capture_message(&err, Level::Error);
return;
};
@@ -250,13 +295,16 @@ async fn listener(
if let Err(e) = result {
use async_tungstenite::tungstenite::Error;
if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
warn!("Error while sending an event to {addr:?}: {e:?}");
let err = format!("Error while sending an event to {addr:?}: {e:?}");
warn!("{}", err);
sentry::capture_message(&err, Level::Warning);
}
return;
}
if let EventV1::Logout = event {
info!("User {addr:?} received log out event!");
return;
}
}
@@ -274,12 +322,19 @@ async fn worker(
let result = read.try_next().await;
let msg = match result {
Ok(Some(msg)) => msg,
Ok(None) => return,
Ok(None) => {
warn!("Received a None message!");
sentry::capture_message("Received a None message!", Level::Warning);
return;
}
Err(e) => {
use async_tungstenite::tungstenite::Error;
if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
warn!("Error while reading an event from {addr:?}: {e:?}");
let err = format!("Error while reading an event from {addr:?}: {e:?}");
warn!("{}", err);
sentry::capture_message(&err, Level::Warning);
}
return;
}
};
@@ -291,7 +346,7 @@ async fn worker(
match payload {
ClientMessage::BeginTyping { channel } => {
if !subscribed.read().await.contains(&channel) {
break;
continue;
}
EventV1::ChannelStartTyping {
@@ -303,7 +358,7 @@ async fn worker(
}
ClientMessage::EndTyping { channel } => {
if !subscribed.read().await.contains(&channel) {
break;
continue;
}
EventV1::ChannelStopTyping {
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-config"
version = "0.7.4"
version = "0.7.5"
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.4"
version = "0.7.5"
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.4", path = "../config" }
revolt-result = { version = "0.7.4", path = "../result" }
revolt-models = { version = "0.7.4", path = "../models", features = [
revolt-config = { version = "0.7.5", path = "../config" }
revolt-result = { version = "0.7.5", path = "../result" }
revolt-models = { version = "0.7.5", path = "../models", features = [
"validator",
] }
revolt-presence = { version = "0.7.4", path = "../presence" }
revolt-permissions = { version = "0.7.4", path = "../permissions", features = [
revolt-presence = { version = "0.7.5", path = "../presence" }
revolt-permissions = { version = "0.7.5", path = "../permissions", features = [
"serde",
"bson",
] }
@@ -504,7 +504,35 @@ impl Message {
if let Some(true) = include_users {
let user_ids = messages
.iter()
.map(|m| m.author.clone())
.flat_map(|m| {
let mut users = vec![m.author.clone()];
if let Some(system) = &m.system {
match system {
v0::SystemMessage::ChannelDescriptionChanged { by } => {
users.push(by.clone())
}
v0::SystemMessage::ChannelIconChanged { by } => users.push(by.clone()),
v0::SystemMessage::ChannelOwnershipChanged { from, to, .. } => {
users.push(from.clone());
users.push(to.clone())
}
v0::SystemMessage::ChannelRenamed { by, .. } => users.push(by.clone()),
v0::SystemMessage::UserAdded { by, id, .. }
| v0::SystemMessage::UserRemove { by, id, .. } => {
users.push(by.clone());
users.push(id.clone());
}
v0::SystemMessage::UserBanned { id, .. }
| v0::SystemMessage::UserKicked { id, .. }
| v0::SystemMessage::UserJoined { id, .. }
| v0::SystemMessage::UserLeft { id, .. } => {
users.push(id.clone());
}
v0::SystemMessage::Text { .. } => {}
}
}
users
})
.collect::<HashSet<String>>()
.into_iter()
.collect::<Vec<String>>();
+1 -13
View File
@@ -1009,11 +1009,6 @@ impl crate::User {
} else {
None
},
profile: if can_see_profile {
self.profile.map(|profile| profile.into())
} else {
None
},
flags: self.flags.unwrap_or_default() as u32,
privileged: self.privileged,
bot: self.bot.map(|bot| bot.into()),
@@ -1086,11 +1081,6 @@ impl crate::User {
} else {
None
},
profile: if can_see_profile {
self.profile.map(|profile| profile.into())
} else {
None
},
flags: self.flags.unwrap_or_default() as u32,
privileged: self.privileged,
bot: self.bot.map(|bot| bot.into()),
@@ -1124,7 +1114,6 @@ impl crate::User {
})
),
status: self.status.map(|status| status.into()),
profile: self.profile.map(|profile| profile.into()),
flags: self.flags.unwrap_or_default() as u32,
privileged: self.privileged,
bot: self.bot.map(|bot| bot.into()),
@@ -1152,7 +1141,7 @@ impl From<User> for crate::User {
relations: None,
badges: Some(value.badges as i32),
status: value.status.map(Into::into),
profile: value.profile.map(Into::into),
profile: None,
flags: Some(value.flags as i32),
privileged: value.privileged,
bot: value.bot.map(Into::into),
@@ -1175,7 +1164,6 @@ impl From<crate::PartialUser> for PartialUser {
}),
badges: value.badges.map(|badges| badges as u32),
status: value.status.map(|status| status.into()),
profile: value.profile.map(|profile| profile.into()),
flags: value.flags.map(|flags| flags as u32),
privileged: value.privileged,
bot: value.bot.map(|bot| bot.into()),
+22 -2
View File
@@ -1,3 +1,5 @@
use std::str::FromStr;
use revolt_result::Result;
#[cfg(feature = "rocket-impl")]
use rocket::request::FromParam;
@@ -44,9 +46,27 @@ impl Reference {
db.fetch_channel(&self.id).await
}
/// Fetch invite from Ref
/// Fetch invite from Ref or create invite to server if discoverable
pub async fn as_invite(&self, db: &Database) -> Result<Invite> {
db.fetch_invite(&self.id).await
if ulid::Ulid::from_str(&self.id).is_ok() {
let server = self.as_server(db).await?;
if !server.discoverable {
return Err(create_error!(NotFound));
}
Ok(Invite::Server {
code: self.id.to_string(),
server: server.id,
creator: server.owner,
channel: server
.channels
.into_iter()
.next()
.ok_or(create_error!(NotFound))?,
})
} else {
db.fetch_invite(&self.id).await
}
}
/// Fetch message from Ref
+3 -3
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-models"
version = "0.7.4"
version = "0.7.5"
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.4", path = "../config" }
revolt-permissions = { version = "0.7.4", path = "../permissions" }
revolt-config = { version = "0.7.5", path = "../config" }
revolt-permissions = { version = "0.7.5", path = "../permissions" }
# Utility
regex = "1"
-3
View File
@@ -50,9 +50,6 @@ auto_derived_partial!(
/// User's current status
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
pub status: Option<UserStatus>,
/// User's profile page
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
pub profile: Option<UserProfile>,
/// Enum of user flags
#[cfg_attr(
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-permissions"
version = "0.7.4"
version = "0.7.5"
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.4", path = "../result" }
revolt-result = { version = "0.7.5", path = "../result" }
# Utility
auto_ops = "0.3.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-presence"
version = "0.7.4"
version = "0.7.5"
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.4"
version = "0.7.5"
edition = "2021"
license = "MIT"
authors = ["Paul Makles <me@insrt.uk>"]
+10
View File
@@ -1,3 +1,5 @@
use std::fmt::Display;
#[cfg(feature = "serde")]
#[macro_use]
extern crate serde;
@@ -28,6 +30,14 @@ pub struct Error {
pub location: String,
}
impl Display for Error {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{:?} occurred in {}", self.error_type, self.location)
}
}
impl std::error::Error for Error {}
/// Possible error types
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
#[cfg_attr(feature = "serde", serde(tag = "type"))]
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "revolt-delta"
version = "0.7.4"
version = "0.7.5"
license = "AGPL-3.0-or-later"
authors = ["Paul Makles <paulmakles@gmail.com>"]
edition = "2018"
@@ -30,12 +30,12 @@ pub async fn edit(
})
})?;
// Fetch server, target member and current permissions
// Fetch server and target member
let mut server = server.as_server(db).await?;
let mut member = target.as_member(db, &server.id).await?;
let mut query = DatabasePermissionQuery::new(db, &user)
.server(&server)
.member(&member);
// Fetch our currrent permissions
let mut query = DatabasePermissionQuery::new(db, &user).server(&server);
let permissions = calculate_server_permissions(&mut query).await;
// Check permissions in server