pavel/discord
1
0
Fork 0
Code Issues Pull requests Projects Releases 60 Packages Wiki Activity Actions

ui stuff

Browse source
This commit is contained in:
pavel 2026-02-13 19:52:42 +01:00
commit d1a68c635c
5 changed files with 384 additions and 94 deletions
Download patch file Download diff file Expand all files Collapse all files

88
src/chat.rs Normal file
View file

@ -0,0 +1,88 @@
use crate::AppState;
use axum::extract::ws::{Message, WebSocket};
use futures_util::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use tokio::sync::{RwLock, mpsc};
use uuid::Uuid;
#[derive(Default)]
pub struct ChatHub {
// user_id -> sender
clients: RwLock<HashMap<Uuid, mpsc::UnboundedSender<ServerEvent>>>,
}
#[derive(Serialize, Clone)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ServerEvent {
MessageCreated {
channel_id: Uuid,
message: serde_json::Value,
},
DmCreated {
other_user_id: Uuid,
message: serde_json::Value,
},
}
#[derive(Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum ClientEvent {
// Currently no interactive client events for the general chat WS
Ping,
}
impl ChatHub {
pub async fn add_client(&self, user_id: Uuid, tx: mpsc::UnboundedSender<ServerEvent>) {
let mut clients = self.clients.write().await;
clients.insert(user_id, tx);
}
pub async fn remove_client(&self, user_id: Uuid) {
let mut clients = self.clients.write().await;
clients.remove(&user_id);
}
pub async fn broadcast_to_user(&self, user_id: Uuid, event: ServerEvent) {
let clients = self.clients.read().await;
if let Some(tx) = clients.get(&user_id) {
let _ = tx.send(event);
}
}
pub async fn broadcast_to_many(&self, user_ids: Vec<Uuid>, event: ServerEvent) {
let clients = self.clients.read().await;
for user_id in user_ids {
if let Some(tx) = clients.get(&user_id) {
let _ = tx.send(event.clone());
}
}
}
}
pub async fn handle_socket(state: AppState, socket: WebSocket, user_id: Uuid) {
let (mut ws_sender, mut ws_receiver) = socket.split();
let (tx, mut rx) = mpsc::unbounded_channel::<ServerEvent>();
state.chat.add_client(user_id, tx).await;
let send_task = tokio::spawn(async move {
while let Some(event) = rx.recv().await {
let Ok(payload) = serde_json::to_string(&event) else {
continue;
};
if ws_sender.send(Message::Text(payload.into())).await.is_err() {
break;
}
}
});
while let Some(Ok(msg)) = ws_receiver.next().await {
if let Message::Close(_) = msg {
break;
}
}
send_task.abort();
state.chat.remove_client(user_id).await;
}

126
src/db.rs
View file

@ -1,15 +1,18 @@
use anyhow::{Result, anyhow};
use chrono::{Duration, Utc};
use sea_orm::{
ActiveModelTrait, ActiveValue::Set, ColumnTrait, DatabaseConnection,
Condition, DatabaseTransaction, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect,
ActiveModelTrait, ActiveValue::Set, ColumnTrait, Condition, DatabaseConnection,
DatabaseTransaction, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect,
TransactionTrait, sea_query::OnConflict,
};
use uuid::Uuid;
use crate::{
entity::{channels, direct_messages, guild_members, guilds, invites, messages, users},
models::{BasicUser, Channel, DmConversation, DmMessageWithAuthor, Guild, Invite, MessageWithAuthor, User},
models::{
BasicUser, Channel, DmConversation, DmMessageWithAuthor, Guild, Invite, MessageWithAuthor,
User,
},
};
pub const CHANNEL_KIND_TEXT: &str = "text";
@ -42,7 +45,10 @@ pub async fn upsert_user_from_oidc(
users::Column::DisplayName,
users::Column::AvatarUrl,
])
.value(users::Column::UpdatedAt, sea_orm::sea_query::Expr::current_timestamp())
.value(
users::Column::UpdatedAt,
sea_orm::sea_query::Expr::current_timestamp(),
)
.to_owned(),
)
.exec_with_returning(db)
@ -75,6 +81,15 @@ pub async fn list_guild_members(db: &DatabaseConnection, guild_id: Uuid) -> Resu
.collect())
}
pub async fn list_guild_member_ids(db: &DatabaseConnection, guild_id: Uuid) -> Result<Vec<Uuid>> {
let rows = guild_members::Entity::find()
.filter(guild_members::Column::GuildId.eq(guild_id))
.all(db)
.await?;
Ok(rows.into_iter().map(|m| m.user_id).collect())
}
pub async fn list_guilds_for_user(db: &DatabaseConnection, user_id: Uuid) -> Result<Vec<Guild>> {
let rows = guild_members::Entity::find()
.filter(guild_members::Column::UserId.eq(user_id))
@ -88,7 +103,11 @@ pub async fn list_guilds_for_user(db: &DatabaseConnection, user_id: Uuid) -> Res
.collect())
}
pub async fn create_guild(db: &DatabaseConnection, owner_user_id: Uuid, name: &str) -> Result<Guild> {
pub async fn create_guild(
db: &DatabaseConnection,
owner_user_id: Uuid,
name: &str,
) -> Result<Guild> {
let guild = guilds::Entity::insert(guilds::ActiveModel {
id: Set(Uuid::new_v4()),
name: Set(name.to_string()),
@ -104,9 +123,12 @@ pub async fn create_guild(db: &DatabaseConnection, owner_user_id: Uuid, name: &s
..Default::default()
})
.on_conflict(
OnConflict::columns([guild_members::Column::GuildId, guild_members::Column::UserId])
.do_nothing()
.to_owned(),
OnConflict::columns([
guild_members::Column::GuildId,
guild_members::Column::UserId,
])
.do_nothing()
.to_owned(),
)
.exec(db)
.await?;
@ -176,9 +198,12 @@ pub async fn join_invite(db: &DatabaseConnection, code: &str, user_id: Uuid) ->
..Default::default()
})
.on_conflict(
OnConflict::columns([guild_members::Column::GuildId, guild_members::Column::UserId])
.do_nothing()
.to_owned(),
OnConflict::columns([
guild_members::Column::GuildId,
guild_members::Column::UserId,
])
.do_nothing()
.to_owned(),
)
.exec(&txn)
.await?;
@ -215,12 +240,18 @@ pub async fn create_channel(
Ok(map_channel(channel))
}
pub async fn get_channel_by_id(db: &DatabaseConnection, channel_id: Uuid) -> Result<Option<Channel>> {
pub async fn get_channel_by_id(
db: &DatabaseConnection,
channel_id: Uuid,
) -> Result<Option<Channel>> {
let row = channels::Entity::find_by_id(channel_id).one(db).await?;
Ok(row.map(map_channel))
}
pub async fn list_channels_for_guild(db: &DatabaseConnection, guild_id: Uuid) -> Result<Vec<Channel>> {
pub async fn list_channels_for_guild(
db: &DatabaseConnection,
guild_id: Uuid,
) -> Result<Vec<Channel>> {
let channels = channels::Entity::find()
.filter(channels::Column::GuildId.eq(guild_id))
.order_by_asc(channels::Column::CreatedAt)
@ -230,7 +261,10 @@ pub async fn list_channels_for_guild(db: &DatabaseConnection, guild_id: Uuid) ->
Ok(channels.into_iter().map(map_channel).collect())
}
pub async fn list_voice_channels_for_guild(db: &DatabaseConnection, guild_id: Uuid) -> Result<Vec<Channel>> {
pub async fn list_voice_channels_for_guild(
db: &DatabaseConnection,
guild_id: Uuid,
) -> Result<Vec<Channel>> {
let channels = channels::Entity::find()
.filter(channels::Column::GuildId.eq(guild_id))
.filter(channels::Column::Kind.eq(CHANNEL_KIND_VOICE))
@ -241,7 +275,11 @@ pub async fn list_voice_channels_for_guild(db: &DatabaseConnection, guild_id: Uu
Ok(channels.into_iter().map(map_channel).collect())
}
pub async fn is_member_of_guild(db: &DatabaseConnection, guild_id: Uuid, user_id: Uuid) -> Result<bool> {
pub async fn is_member_of_guild(
db: &DatabaseConnection,
guild_id: Uuid,
user_id: Uuid,
) -> Result<bool> {
let count = guild_members::Entity::find()
.filter(guild_members::Column::GuildId.eq(guild_id))
.filter(guild_members::Column::UserId.eq(user_id))
@ -251,7 +289,10 @@ pub async fn is_member_of_guild(db: &DatabaseConnection, guild_id: Uuid, user_id
Ok(count > 0)
}
pub async fn guild_id_for_channel(db: &DatabaseConnection, channel_id: Uuid) -> Result<Option<Uuid>> {
pub async fn guild_id_for_channel(
db: &DatabaseConnection,
channel_id: Uuid,
) -> Result<Option<Uuid>> {
let guild_id = channels::Entity::find_by_id(channel_id)
.select_only()
.column(channels::Column::GuildId)
@ -267,18 +308,30 @@ pub async fn create_message(
channel_id: Uuid,
author_user_id: Uuid,
body: &str,
) -> Result<()> {
messages::Entity::insert(messages::ActiveModel {
) -> Result<MessageWithAuthor> {
let model = messages::Entity::insert(messages::ActiveModel {
id: Set(Uuid::new_v4()),
channel_id: Set(channel_id),
author_user_id: Set(author_user_id),
body: Set(body.to_string()),
..Default::default()
})
.exec(db)
.exec_with_returning(db)
.await?;
Ok(())
let user = users::Entity::find_by_id(author_user_id)
.one(db)
.await?
.ok_or_else(|| anyhow!("author not found"))?;
Ok(MessageWithAuthor {
id: model.id,
channel_id: model.channel_id,
author_user_id: model.author_user_id,
author_display_name: user.display_name,
body: model.body,
created_at: model.created_at,
})
}
pub async fn list_messages(
@ -317,18 +370,30 @@ pub async fn create_direct_message(
sender_user_id: Uuid,
recipient_user_id: Uuid,
body: &str,
) -> Result<()> {
direct_messages::Entity::insert(direct_messages::ActiveModel {
) -> Result<DmMessageWithAuthor> {
let model = direct_messages::Entity::insert(direct_messages::ActiveModel {
id: Set(Uuid::new_v4()),
sender_user_id: Set(sender_user_id),
recipient_user_id: Set(recipient_user_id),
body: Set(body.to_string()),
..Default::default()
})
.exec(db)
.exec_with_returning(db)
.await?;
Ok(())
let user = users::Entity::find_by_id(sender_user_id)
.one(db)
.await?
.ok_or_else(|| anyhow!("sender not found"))?;
Ok(DmMessageWithAuthor {
id: model.id,
author_user_id: model.sender_user_id,
recipient_user_id: model.recipient_user_id,
author_display_name: user.display_name,
body: model.body,
created_at: model.created_at,
})
}
pub async fn list_direct_messages(
@ -385,7 +450,10 @@ pub async fn list_direct_messages(
.collect())
}
pub async fn list_dm_conversations(db: &DatabaseConnection, current_user_id: Uuid) -> Result<Vec<DmConversation>> {
pub async fn list_dm_conversations(
db: &DatabaseConnection,
current_user_id: Uuid,
) -> Result<Vec<DmConversation>> {
let rows = direct_messages::Entity::find()
.filter(
Condition::any()
@ -396,7 +464,8 @@ pub async fn list_dm_conversations(db: &DatabaseConnection, current_user_id: Uui
.all(db)
.await?;
let mut latest_by_peer = std::collections::HashMap::<Uuid, chrono::DateTime<chrono::FixedOffset>>::new();
let mut latest_by_peer =
std::collections::HashMap::<Uuid, chrono::DateTime<chrono::FixedOffset>>::new();
for row in rows {
let peer_id = if row.sender_user_id == current_user_id {
row.recipient_user_id
@ -496,7 +565,10 @@ fn validate_invite(invite: &invites::Model) -> Result<()> {
Ok(())
}
async fn increment_invite_use_count(txn: &DatabaseTransaction, invite: invites::Model) -> Result<()> {
async fn increment_invite_use_count(
txn: &DatabaseTransaction,
invite: invites::Model,
) -> Result<()> {
let next_count = invite.use_count + 1;
let mut active: invites::ActiveModel = invite.into();
active.use_count = Set(next_count);

125
src/handlers.rs
View file

@ -11,7 +11,7 @@ use uuid::Uuid;
use crate::{
AppState,
auth::{self, ApiError, AuthUser},
db, voice,
chat, db, voice,
};
pub fn routes() -> Router<AppState> {
@ -22,17 +22,27 @@ pub fn routes() -> Router<AppState> {
.route("/auth/logout", post(auth_logout))
.route("/me", get(me))
.route("/dms", get(list_dm_conversations))
.route("/dms/{other_user_id}/messages", get(list_dm_messages).post(send_dm_message))
.route(
"/dms/{other_user_id}/messages",
get(list_dm_messages).post(send_dm_message),
)
.route("/rtc-config", get(rtc_config))
.route("/guilds", get(list_guilds).post(create_guild))
.route("/guilds/{guild_id}/members", get(list_guild_members))
.route("/guilds/{guild_id}/channels", get(list_channels))
.route("/guilds/{guild_id}/voice-presence", get(guild_voice_presence))
.route(
"/guilds/{guild_id}/voice-presence",
get(guild_voice_presence),
)
.route("/guilds/{guild_id}/invites", post(create_invite))
.route("/invites/{code}/join", post(join_invite))
.route("/channels", post(create_channel))
.route("/channels/{channel_id}/messages", get(list_messages).post(send_message))
.route(
"/channels/{channel_id}/messages",
get(list_messages).post(send_message),
)
.route("/channels/{channel_id}/voice/ws", get(voice_ws))
.route("/ws", get(chat_ws))
}
pub async fn health() -> &'static str {
@ -142,7 +152,10 @@ async fn auth_callback(
.map_err(|e| ApiError::bad_request(&format!("token exchange failed: {e}")))?;
if !token_res.status().is_success() {
let text = token_res.text().await.unwrap_or_else(|_| "<no body>".to_string());
let text = token_res
.text()
.await
.unwrap_or_else(|_| "<no body>".to_string());
return Err(ApiError::bad_request(&format!(
"token exchange returned non-success: {text}"
)));
@ -162,7 +175,10 @@ async fn auth_callback(
.map_err(|e| ApiError::bad_request(&format!("userinfo request failed: {e}")))?;
if !userinfo_res.status().is_success() {
let text = userinfo_res.text().await.unwrap_or_else(|_| "<no body>".to_string());
let text = userinfo_res
.text()
.await
.unwrap_or_else(|_| "<no body>".to_string());
return Err(ApiError::bad_request(&format!(
"userinfo returned non-success: {text}"
)));
@ -185,9 +201,12 @@ async fn auth_callback(
.await
.map_err(|e| ApiError::internal(&format!("failed to persist user: {e}")))?;
let session_cookie =
auth::new_session_cookie(user.id, &state.settings.session_secret, state.settings.cookie_secure)
.map_err(|e| ApiError::internal(&e.to_string()))?;
let session_cookie = auth::new_session_cookie(
user.id,
&state.settings.session_secret,
state.settings.cookie_secure,
)
.map_err(|e| ApiError::internal(&e.to_string()))?;
let mut headers = HeaderMap::new();
headers.append(
@ -232,7 +251,10 @@ struct RtcIceServer {
credential: Option<String>,
}
async fn rtc_config(State(state): State<AppState>, _user: AuthUser) -> Result<impl IntoResponse, ApiError> {
async fn rtc_config(
State(state): State<AppState>,
_user: AuthUser,
) -> Result<impl IntoResponse, ApiError> {
let mut ice_servers = Vec::new();
if !state.settings.stun_urls.is_empty() {
@ -441,9 +463,13 @@ async fn create_channel(
return Err(ApiError::bad_request("channel name must be 1..64 chars"));
}
let kind = body.kind.unwrap_or_else(|| db::CHANNEL_KIND_TEXT.to_string());
let kind = body
.kind
.unwrap_or_else(|| db::CHANNEL_KIND_TEXT.to_string());
if kind != db::CHANNEL_KIND_TEXT && kind != db::CHANNEL_KIND_VOICE {
return Err(ApiError::bad_request("channel kind must be 'text' or 'voice'"));
return Err(ApiError::bad_request(
"channel kind must be 'text' or 'voice'",
));
}
let channel = db::create_channel(&state.db, body.guild_id, trimmed, &kind)
@ -497,10 +523,31 @@ async fn send_message(
return Err(ApiError::bad_request("message body must be 1..4000 chars"));
}
db::create_message(&state.db, channel_id, user.id, content)
let message = db::create_message(&state.db, channel_id, user.id, content)
.await
.map_err(|e| ApiError::internal(&format!("failed to create message: {e}")))?;
// Broadcast to all guild members
let guild_id = db::guild_id_for_channel(&state.db, channel_id)
.await
.map_err(|e| ApiError::internal(&format!("channel lookup failed: {e}")))?
.ok_or_else(|| ApiError::internal("channel not found after creation"))?;
let members = db::list_guild_member_ids(&state.db, guild_id)
.await
.map_err(|e| ApiError::internal(&format!("failed to list guild members: {e}")))?;
state
.chat
.broadcast_to_many(
members,
chat::ServerEvent::MessageCreated {
channel_id,
message: serde_json::to_value(&message).unwrap_or_default(),
},
)
.await;
Ok((StatusCode::CREATED, Json(SendMessageResponse { ok: true })))
}
@ -557,10 +604,32 @@ async fn send_dm_message(
return Err(ApiError::bad_request("message body must be 1..4000 chars"));
}
db::create_direct_message(&state.db, user.id, other_user_id, content)
let message = db::create_direct_message(&state.db, user.id, other_user_id, content)
.await
.map_err(|e| ApiError::internal(&format!("failed to create dm message: {e}")))?;
// Broadcast to both users
state
.chat
.broadcast_to_user(
user.id,
chat::ServerEvent::DmCreated {
other_user_id,
message: serde_json::to_value(&message).unwrap_or_default(),
},
)
.await;
state
.chat
.broadcast_to_user(
other_user_id,
chat::ServerEvent::DmCreated {
other_user_id: user.id,
message: serde_json::to_value(&message).unwrap_or_default(),
},
)
.await;
Ok((StatusCode::CREATED, Json(SendMessageResponse { ok: true })))
}
@ -581,15 +650,27 @@ async fn voice_ws(
})?;
if channel.kind != db::CHANNEL_KIND_VOICE {
return Err(ApiError::bad_request("voice websocket requires a voice channel"));
return Err(ApiError::bad_request(
"voice websocket requires a voice channel",
));
}
Ok(ws.on_upgrade(move |socket| {
voice::handle_socket(state, socket, channel_id, user.id)
}))
Ok(ws.on_upgrade(move |socket| voice::handle_socket(state, socket, channel_id, user.id)))
}
async fn ensure_guild_member(state: &AppState, guild_id: Uuid, user_id: Uuid) -> Result<(), ApiError> {
async fn chat_ws(
ws: WebSocketUpgrade,
State(state): State<AppState>,
user: AuthUser,
) -> Result<impl IntoResponse, ApiError> {
Ok(ws.on_upgrade(move |socket| chat::handle_socket(state, socket, user.id)))
}
async fn ensure_guild_member(
state: &AppState,
guild_id: Uuid,
user_id: Uuid,
) -> Result<(), ApiError> {
let is_member = db::is_member_of_guild(&state.db, guild_id, user_id)
.await
.map_err(|e| ApiError::internal(&format!("membership check failed: {e}")))?;
@ -604,7 +685,11 @@ async fn ensure_guild_member(state: &AppState, guild_id: Uuid, user_id: Uuid) ->
Ok(())
}
async fn ensure_channel_member(state: &AppState, channel_id: Uuid, user_id: Uuid) -> Result<(), ApiError> {
async fn ensure_channel_member(
state: &AppState,
channel_id: Uuid,
user_id: Uuid,
) -> Result<(), ApiError> {
let guild_id = db::guild_id_for_channel(&state.db, channel_id)
.await
.map_err(|e| ApiError::internal(&format!("channel lookup failed: {e}")))?

3
src/main.rs
View file

@ -1,4 +1,5 @@
mod auth;
mod chat;
mod config;
mod db;
mod entity;
@ -24,6 +25,7 @@ pub struct AppState {
pub settings: Arc<Settings>,
pub http: reqwest::Client,
pub voice: Arc<VoiceHub>,
pub chat: Arc<chat::ChatHub>,
}
#[tokio::main]
@ -46,6 +48,7 @@ async fn main() -> anyhow::Result<()> {
settings,
http: reqwest::Client::new(),
voice: Arc::new(VoiceHub::default()),
chat: Arc::new(chat::ChatHub::default()),
};
let port = state.settings.port;