gorb/backend
5
0
Fork 0
Code Issues 16 Pull requests 2 Projects Releases Packages 1 Wiki Activity Actions

Compare commits

base: gorb:wip/messaging-features
Branches Tags
gorb:main
gorb:wip/auditlog
gorb:wip/messaging-features
gorb:staging
gorb:wip/member-improvements
gorb:wip/ban
gorb:wip/kick
gorb:wip/message-editing
gorb:wip/categories
gorb:wip/federation-impl
gorb:wip/local-data-folder
..
compare: gorb:main
Branches Tags
gorb:wip/auditlog
gorb:wip/messaging-features
gorb:main
gorb:staging
gorb:wip/member-improvements
gorb:wip/ban
gorb:wip/kick
gorb:wip/message-editing
gorb:wip/categories
gorb:wip/federation-impl
gorb:wip/local-data-folder

No commits in common. "wip/messaging-features" and "main" have entirely different histories.
wip/messag ... main

2 changed files with 24 additions and 173 deletions
Download patch file Download diff file Expand all files Collapse all files

167
src/api/v1/channels/uuid/socket.rs
View file

@ -5,58 +5,22 @@ use axum::{
http::HeaderMap, http::HeaderMap,
response::IntoResponse, response::IntoResponse,
}; };
use diesel::{ExpressionMethods, QueryDsl, SelectableHelper, delete, update};
use diesel_async::RunQueryDsl;
use futures_util::{SinkExt, StreamExt}; use futures_util::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize}; use serde::Deserialize;
use uuid::Uuid; use uuid::Uuid;
use crate::{ use crate::{
AppState, AppState,
api::v1::auth::check_access_token, api::v1::auth::check_access_token,
error::Error, error::Error,
objects::{self, Channel, Member, message::MessageBuilder}, objects::{Channel, Member},
schema::messages,
utils::global_checks, utils::global_checks,
}; };
#[derive(Deserialize)] #[derive(Deserialize)]
#[serde(tag = "event")] struct MessageBody {
enum ReceiveEvent {
MessageSend { entity: MessageSend },
MessageEdit { entity: MessageEdit },
MessageDelete { entity: MessageDelete },
}
#[derive(Deserialize)]
struct MessageSend {
text: String,
reply_to: Option<Uuid>,
}
#[derive(Deserialize)]
struct MessageEdit {
uuid: Uuid,
text: String,
}
#[derive(Deserialize, Serialize)]
struct MessageDelete {
uuid: Uuid,
}
#[derive(Serialize)]
#[serde(tag = "event")]
enum SendEvent {
MessageSend { entity: objects::Message },
MessageEdit { entity: objects::Message },
MessageDelete { entity: MessageDelete },
Error { entity: SendError },
}
#[derive(Serialize)]
struct SendError {
message: String, message: String,
reply_to: Option<Uuid>,
} }
pub async fn ws( pub async fn ws(
@ -135,27 +99,20 @@ pub async fn ws(
tokio::spawn(async move { tokio::spawn(async move {
while let Some(msg) = receiver.next().await { while let Some(msg) = receiver.next().await {
if let Ok(Message::Text(text)) = msg { if let Ok(Message::Text(text)) = msg {
let message_body: ReceiveEvent = serde_json::from_str(&text)?; let message_body: MessageBody = serde_json::from_str(&text)?;
match message_body {
ReceiveEvent::MessageSend { entity } => {
let message = channel let message = channel
.new_message( .new_message(
&mut app_state.pool.get().await?, &mut conn,
&app_state.cache_pool, &app_state.cache_pool,
uuid, uuid,
entity.text, message_body.message,
entity.reply_to, message_body.reply_to,
) )
.await?; .await?;
redis::cmd("PUBLISH") redis::cmd("PUBLISH")
.arg(&[ .arg(&[channel_uuid.to_string(), serde_json::to_string(&message)?])
channel_uuid.to_string(),
serde_json::to_string(&SendEvent::MessageSend {
entity: message,
})?,
])
.exec_async( .exec_async(
&mut app_state &mut app_state
.cache_pool .cache_pool
@ -164,112 +121,6 @@ pub async fn ws(
) )
.await?; .await?;
} }
ReceiveEvent::MessageEdit { entity } => {
use messages::dsl;
let mut message: MessageBuilder = dsl::messages
.filter(dsl::uuid.eq(entity.uuid))
.select(MessageBuilder::as_select())
.get_result(&mut app_state.pool.get().await?)
.await?;
if uuid != message.user_uuid {
redis::cmd("PUBLISH")
.arg(&[
channel_uuid.to_string(),
serde_json::to_string(&SendEvent::Error {
entity: SendError {
message: "Not allowed".to_string(),
},
})?,
])
.exec_async(
&mut app_state
.cache_pool
.get_multiplexed_tokio_connection()
.await?,
)
.await?;
continue;
}
update(messages::table)
.filter(dsl::uuid.eq(entity.uuid))
.set(dsl::message.eq(&entity.text))
.execute(&mut app_state.pool.get().await?)
.await?;
message.message = entity.text;
redis::cmd("PUBLISH")
.arg(&[
channel_uuid.to_string(),
serde_json::to_string(&SendEvent::MessageEdit {
entity: message
.build(
&mut app_state.pool.get().await?,
&app_state.cache_pool,
)
.await?,
})?,
])
.exec_async(
&mut app_state
.cache_pool
.get_multiplexed_tokio_connection()
.await?,
)
.await?;
}
ReceiveEvent::MessageDelete { entity } => {
use messages::dsl;
let message: MessageBuilder = dsl::messages
.filter(dsl::uuid.eq(entity.uuid))
.select(MessageBuilder::as_select())
.get_result(&mut app_state.pool.get().await?)
.await?;
if uuid != message.user_uuid {
redis::cmd("PUBLISH")
.arg(&[
channel_uuid.to_string(),
serde_json::to_string(&SendEvent::Error {
entity: SendError {
message: "Not allowed".to_string(),
},
})?,
])
.exec_async(
&mut app_state
.cache_pool
.get_multiplexed_tokio_connection()
.await?,
)
.await?;
continue;
}
delete(messages::table)
.filter(dsl::uuid.eq(entity.uuid))
.execute(&mut app_state.pool.get().await?)
.await?;
redis::cmd("PUBLISH")
.arg(&[
channel_uuid.to_string(),
serde_json::to_string(&SendEvent::MessageDelete { entity })?,
])
.exec_async(
&mut app_state
.cache_pool
.get_multiplexed_tokio_connection()
.await?,
)
.await?;
}
}
}
} }
Ok::<(), crate::error::Error>(()) Ok::<(), crate::error::Error>(())

2
src/objects/mod.rs
View file

@ -15,7 +15,7 @@ mod guild;
mod invite; mod invite;
mod me; mod me;
mod member; mod member;
pub mod message; mod message;
mod password_reset_token; mod password_reset_token;
mod role; mod role;
mod user; mod user;