From 64b0c7612054333bb948bbd025764ee3619b8c40 Mon Sep 17 00:00:00 2001 From: Nell Date: Sun, 18 Oct 2026 18:46:52 +0200 Subject: [PATCH] hotfix --- .junie/plans/fix-new-member-history.md | 155 ++++++++++++++++++++ frontend/src/pages/server/channel/index.vue | 14 +- frontend/src/stores/message.ts | 3 + frontend/src/stores/user.ts | 14 +- src/core/mod.rs | 6 + src/repositories/computed_permission.rs | 78 +++++++++- src/routes/gateway/mod.rs | 46 +++++- src/routes/message/mod.rs | 3 + src/routes/message/tests.rs | 114 ++++++++++++++ src/services/realtime_registry.rs | 2 +- src/services/server.rs | 23 +++ 11 files changed, 453 insertions(+), 5 deletions(-) create mode 100644 .junie/plans/fix-new-member-history.md create mode 100644 src/routes/message/tests.rs diff --git a/.junie/plans/fix-new-member-history.md b/.junie/plans/fix-new-member-history.md new file mode 100644 index 0000000..dee5e10 --- /dev/null +++ b/.junie/plans/fix-new-member-history.md @@ -0,0 +1,155 @@ +--- +sessionId: session-261018-151548-e6aw +--- + +# Requirements + +### Overview & Goals +Quand un nouvel utilisateur rejoint un serveur, il doit voir l'historique des messages des canaux texte du serveur, et recevoir en temps réel les nouveaux messages. Aujourd'hui, certains membres n'ont pas de cache de permissions calculées, ce qui casse la diffusion temps réel et masque les erreurs côté client. + +### Scope + +**In Scope** +- Diagnostiquer précisément pourquoi un membre récent ne voit pas l'historique. +- Corriger le flux d'arrivée sur un serveur (`join_server`) pour que les permissions soient immédiatement disponibles. +- Réparer les membres existants qui n'ont aucune permission calculée. +- Rendre visibles les échecs de chargement de l'historique côté frontend. + +**Out of Scope** +- Changer le modèle de rôles ou de permissions par canal. +- Ajouter de nouvelles permissions par défaut configurables par l'utilisateur. +- Refonte du système d'événements de l'EventBus. + +### User Stories +- En tant que nouveau membre, je veux voir les anciens messages d'un canal dès que je le rejoins, afin de suivre la conversation. +- En tant que membre, je veux recevoir les nouveaux messages en direct sans recharger la page. +- En tant qu'administrateur, je veux que les membres existants soient réparés sans manipulation manuelle de la base. + +### Functional Requirements +- Après `join_server`, un membre standard peut lire (`READ_CHANNEL`) tous les canaux texte du serveur via `GET /messages`. +- Après `join_server`, le membre reçoit les événements temps réel de messages des canaux du serveur. +- Les membres existants sans cache (5 sur le serveur principal dans la base locale) sont réparés au démarrage ou par une opération de réparation. +- Si le chargement de l'historique échoue, l'interface affiche un message d'erreur au lieu d'un canal vide silencieux. + +### Non-Functional Requirements +- L'opération de join doit rester transactionnelle et ne pas bloquer la réponse HTTP plus que nécessaire. +- La réparation doit être idempotente (relançable sans effet de bord). + +# Technical Design + +### Current Implementation + +**Lecture de l'historique (chemin REST)** +- `src/routes/message/handlers.rs` `get_all` exige `READ_CHANNEL` via `require_channel_permission`. +- `src/http/permissions.rs` `check_channel_permission` (l.192-218) calcule les droits **en direct** à partir de `server_user`, `role_user`, `channel_role_permission` et `channel_user_permission`. Pour un canal de serveur, la base de bits part de `0` ; `PermissionSet::DEFAULT` (`src/permissions.rs` l.96-105) n'est appliqué qu'aux canaux sans serveur. +- Le rôle par défaut « Member » reçoit `READ_CHANNEL` sur chaque canal créé via `src/services/channel.rs` `create_channel` (l.69-86). Dans la base locale, chaque rôle par défaut a une permission par canal (11/11 et 4/4), et tous les membres ont ce rôle. Le contrôle REST passe donc pour les membres actuels. + +**Arrivée d'un membre** +- `src/services/server.rs` `join_server` (l.77-137) insère `server_user` et `role_user` puis émet `ServerUserCreatedEvent` après commit. Il n'appelle pas directement la synchronisation du cache. +- `src/services/permission_sync.rs` `start_listen_event` (l.85+) écoute `ServerUserCreatedEvent` (l.100) et appelle `computed_permission.full_sync_user` de façon asynchrone. +- `src/repositories/computed_permission.rs` `full_sync_user` (l.108+) recalcule les lignes `computed_permission` du membre. + +**Diffusion temps réel** +- `src/services/realtime_registry.rs` construit l'audience des canaux (`refresh_channel` l.25-57, `initialize` l.76+) **à partir de la table `computed_permission`**, uniquement pour les lignes portant `READ_CHANNEL`. +- `src/routes/gateway/mod.rs` (l.141-293) envoie tous les événements de messages et de canaux via `users_for_channel`. Un membre absent de ce registre ne reçoit rien en direct. + +**Constat dans la base locale (lecture seule, `oxspeak.db`)** +- Serveur `B3D2…` (9 membres) : 4 membres ont 12 lignes de cache (11 canaux + niveau serveur), 5 membres ont **0 ligne**. +- Serveur `F142…` (2 membres) : un membre a 5 lignes, l'autre **0 ligne**. +- Le serveur `B3D2…` contient des messages (ex. canal `Test` : 11 messages), donc le problème n'est pas une absence de données. + +**Affichage côté client** +- `frontend/src/stores/message.ts` `requestPage` (l.94-112) lève une erreur sur une réponse non OK, mais `fetchMessages` (l.148-175) ne fait que `console.error`. Un 403 apparaît donc comme un canal vide. + +### Key Decisions + +**Décision 1 : Source de vérité de l'audience temps réel** +- Option A (recommandée) : garder `computed_permission` comme cache, mais le rafraîchir **de façon synchrone** dans `join_server` (appel à `full_sync_user` puis `realtime_registry.refresh_user`) avant de répondre. +- Option B : rendre l'audience temps réel dépendante de `check_channel_permission` (plus coûteux, à évaluer sur des serveurs volumineux). + +**Décision 2 : Réparation des membres existants** +- Option A (recommandée) : au démarrage, lancer `full_sync_server` pour chaque serveur (idempotent, coût acceptable à faible volume). +- Option B : une route d'administration ou une commande ponctuelle de réparation. + +**Décision 3 : Erreur côté frontend** +- Afficher un état d'erreur dans `fetchMessages` au lieu d'un canal vide. Ce n'est pas une correction de la cause, mais elle rend le problème visible. + +### Proposed Changes +- `services/server.rs` `join_server` : après commit, synchroniser le cache du membre et rafraîchir le registre temps réel, en conservant l'émission de `ServerUserCreatedEvent`. +- `services/permission_sync.rs` : journaliser les échecs du listener `sync_user` (aujourd'hui silencieux) pour expliquer les membres restés sans cache. +- Démarrage (`core/mod.rs`, après `realtime_registry.initialize`) : appeler une réparation idempotente `full_sync_server` pour chaque serveur. +- `frontend/src/stores/message.ts` `fetchMessages` : exposer une erreur de chargement et l'afficher. + +### Architecture Diagram +```mermaid +graph LR + J[join_server] --> DB[(server_user, role_user)] + J --> S[full_sync_user] + S --> C[(computed_permission)] + S --> R[realtime_registry] + R --> G[gateway: message events] + F[GET /messages] --> P[check_channel_permission] + P --> DB +``` + +### Risks +- Le listener asynchrone peut échouer silencieusement : la cause exacte des 5 membres sans cache doit être confirmée (journaux, rejeu). +- La réparation au démarrage peut être longue sur de gros serveurs : la limiter aux serveurs dont un membre n'a pas de cache. +- Le symptôme exact « ne voit pas les anciens messages » n'a pas encore été reproduit à l'interface ; l'historique REST semble autorisé pour les membres actuels. Il faut reproduire avec un compte réellement concerné avant de conclure. +- Verrou `lock_scope` de `full_sync_user` : éviter les appels imbriqués pendant la transaction de join. + +# Testing + +### Validation Approach +- Reproduire le problème par un test d'intégration qui crée un serveur, ajoute un canal avec des messages, fait rejoindre un nouvel utilisateur via `join_server`, puis vérifie `GET /messages` et la présence dans le registre temps réel. +- Utiliser les helpers existants de `src/http/test_support.rs` et le style des tests de `src/routes/channel/tests.rs`. + +### Key Scenarios +- Un nouvel membre rejoint un serveur ayant des canaux et des messages : `GET /messages` renvoie les messages, `computed_permission` contient une ligne `READ_CHANNEL` par canal. +- Après join, `users_for_channel(channel_id)` contient le nouveau membre pour chaque canal texte. +- La réparation au démarrage ajoute les lignes manquantes pour les 5 membres sans cache et ne modifie pas les autres. +- Côté frontend, un 403 sur `/messages` affiche un message d'erreur au lieu d'un canal vide. + +### Edge Cases +- Membre qui rejoint un serveur déjà joint (chemin « déjà membre » de `join_server`) : pas de doublon, cache inchangé. +- Canal créé avant l'existence du rôle par défaut : vérifier la présence de `channel_role_permission`. +- Relance de la réparation : idempotente. +- Mauvais mot de passe de serveur : aucune ligne créée, aucun événement émis. +- Régression : les DM (canaux sans serveur) continuent d'utiliser `channel_user` pour leur audience. + +### Test Changes +- Ajouter un test de join dans le module de tests des routes message (à créer à côté de `src/routes/message/handlers.rs`). +- Ajouter un test unitaire sur `full_sync_server` idempotent dans `src/repositories/computed_permission.rs`. +- Ne pas modifier les tests existants de `realtime_registry.rs`. + +# Delivery Steps + +### ✓ Step 1: Reproduire et confirmer la cause avec un test de join +Un test d'intégration reproduit le cas « nouveau membre sans historique » et confirme la cause côté cache et audience temps réel. + +- Écrire un test qui crée un serveur et un canal texte avec des messages, puis fait rejoindre un nouvel utilisateur via `join_server`. +- Vérifier `GET /messages` (handler `get_all` dans `src/routes/message/handlers.rs`) et l'absence éventuelle de lignes `computed_permission`. +- Vérifier la présence du membre dans `RealtimeRegistry::users_for_channel`. +- Consigner dans les journaux pourquoi le listener `sync_user` de `src/services/permission_sync.rs` n'a pas rempli le cache. + +### ✓ Step 2: Synchroniser les permissions dès l'arrivée d'un membre +Un membre qui rejoint un serveur dispose immédiatement de son cache de permissions et de sa place dans le registre temps réel. + +- Dans `join_server` (`src/services/server.rs`), après commit, appeler `computed_permission.full_sync_user` pour le nouveau membre. +- Appeler `realtime_registry.refresh_user` pour mettre à jour les canaux de l'utilisateur. +- Garder l'émission de `ServerUserCreatedEvent` pour les autres abonnés. +- Journaliser l'échec de synchronisation au lieu de l'ignorer dans le listener `sync_user`. + +### ✓ Step 3: Réparer les membres existants au démarrage +Les membres déjà présents sans cache de permissions récupèrent leurs lignes au prochain démarrage du serveur. + +- Ajouter une réparation idempotente qui appelle `full_sync_server` pour chaque serveur dont un membre n'a pas de ligne `computed_permission`. +- Brancher la réparation dans `src/core/mod.rs` après `realtime_registry.initialize`. +- Ajouter un test unitaire vérifiant que deux exécutions successives ne modifient pas les lignes. + +### ✓ Step 4: Afficher les erreurs de chargement de l'historique dans le frontend +Un échec de chargement de l'historique est visible par l'utilisateur au lieu d'un canal vide silencieux. + +- Dans `frontend/src/stores/message.ts`, `fetchMessages` expose l'état d'erreur (par exemple `loadError`) au lieu de seulement `console.error`. +- Afficher un message d'erreur avec un bouton de nouvel essai dans le composant de liste de messages. +- Ne pas modifier le comportement de `fetchBefore` et `fetchAfter`. \ No newline at end of file diff --git a/frontend/src/pages/server/channel/index.vue b/frontend/src/pages/server/channel/index.vue index c2c8221..a416f99 100644 --- a/frontend/src/pages/server/channel/index.vue +++ b/frontend/src/pages/server/channel/index.vue @@ -30,7 +30,7 @@ const router = useRouter() const {renderMarkdown} = useMarkdown() const messageContainer = ref(null); -const {messages, loading, loadingBefore, loadingAfter, hasMoreAfter, isAtBottom, newestId} = storeToRefs(messageStore); +const {messages, loading, loadError, loadingBefore, loadingAfter, hasMoreAfter, isAtBottom, newestId} = storeToRefs(messageStore); const newMessage = ref(''); const fileInput = ref(null); type PendingAttachment = { @@ -435,6 +435,18 @@ watch(() => props.serverId, (newServerId) => { indeterminate /> + +
+ Impossible de charger l'historique des messages. + Réessayer +
+ + ({ - users: [] as User[] + users: [] as User[], + serverId: null as string | null }), getters: { usersById: (state): Record => { @@ -21,11 +23,21 @@ export const useUserStore = defineStore('user', { if (serverId) { url += `?server_id=${serverId}`; } + this.serverId = serverId ?? null; const response = await api.get(url); this.users = await response.json(); }, reset() { this.users = []; + this.serverId = null; } } }); + +onGatewayEvent("ServerUser", (payload) => { + const store = useUserStore(); + const content = payload.content as { server_id?: string | null }; + if (content?.server_id && content.server_id === store.serverId) { + void store.fetchUsers(content.server_id); + } +}); diff --git a/src/core/mod.rs b/src/core/mod.rs index 9deafee..4b08710 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -72,6 +72,12 @@ impl App { let services = Arc::new(Services::new(repositories.clone(), event_bus.clone())); services.permission_sync.start_listen_event().await; services.realtime_registry.initialize(&repositories).await?; + for (user_id, _server_id) in repositories.computed_permission.repair_missing_members().await? { + services + .realtime_registry + .refresh_user(&repositories, user_id) + .await?; + } services .realtime_registry .start_listening(repositories.clone(), event_bus.clone()); diff --git a/src/repositories/computed_permission.rs b/src/repositories/computed_permission.rs index bca99a3..13a1a3d 100644 --- a/src/repositories/computed_permission.rs +++ b/src/repositories/computed_permission.rs @@ -8,7 +8,7 @@ use crate::repositories::{AnyResult, RepositoryContext}; use sea_orm::{ ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QuerySelect, Set, TransactionTrait, }; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::{Arc, OnceLock}; use uuid::Uuid; @@ -81,6 +81,39 @@ impl ComputedPermissionRepository { Ok(()) } + /// Répare les membres sans aucune ligne de cache. Les membres déjà synchronisés ne sont pas recalculés. + pub async fn repair_missing_members(&self) -> AnyResult> { + let members = server_user::Entity::find() + .select_only() + .column(server_user::Column::UserId) + .column(server_user::Column::ServerId) + .into_tuple::<(Uuid, Uuid)>() + .all(&self.context.db) + .await?; + + let cached: HashSet<(Uuid, Uuid)> = computed_permission::Entity::find() + .select_only() + .column(computed_permission::Column::UserId) + .column(computed_permission::Column::ServerId) + .distinct() + .into_tuple::<(Uuid, Uuid)>() + .all(&self.context.db) + .await? + .into_iter() + .collect(); + + let mut repaired = Vec::new(); + for (user_id, server_id) in members { + if cached.contains(&(user_id, server_id)) { + continue; + } + self.full_sync_user(user_id, server_id).await?; + repaired.push((user_id, server_id)); + } + + Ok(repaired) + } + /// Scope RÔLE : Recalcule le cache uniquement pour les membres d'un rôle spécifique. pub async fn sync_role_members(&self, role_id: Uuid, server_id: Uuid) -> AnyResult<()> { let user_ids = role_user::Entity::find() @@ -263,3 +296,46 @@ impl ComputedPermissionRepository { Ok(()) } } + + +#[cfg(test)] +mod tests { + use crate::http::test_support::{state, user}; + use crate::models::computed_permission; + use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; + + #[tokio::test] + async fn repair_missing_members_is_idempotent() { + let state = state().await; + let owner = user(&state, false).await; + let server = state + .services + .server + .create_server("repair".into(), None, false, owner.id) + .await + .unwrap(); + + computed_permission::Entity::delete_many() + .filter(computed_permission::Column::UserId.eq(owner.id)) + .filter(computed_permission::Column::ServerId.eq(server.id)) + .exec(&state.db) + .await + .unwrap(); + + let first = state + .repositories + .computed_permission + .repair_missing_members() + .await + .unwrap(); + let second = state + .repositories + .computed_permission + .repair_missing_members() + .await + .unwrap(); + + assert!(first.contains(&(owner.id, server.id))); + assert!(second.is_empty()); + } +} diff --git a/src/routes/gateway/mod.rs b/src/routes/gateway/mod.rs index 31bcfce..7439fc9 100644 --- a/src/routes/gateway/mod.rs +++ b/src/routes/gateway/mod.rs @@ -10,7 +10,10 @@ use crate::domain::events::message::{ MessageCreatedEvent, MessageDeletedEvent, MessageReactionAddedEvent, MessageReactionRemovedEvent, MessageUpdatedEvent, }; -use crate::domain::events::server::{ServerCreatedEvent, ServerDeletedEvent, ServerUpdatedEvent}; +use crate::domain::events::server::{ + ServerCreatedEvent, ServerDeletedEvent, ServerUpdatedEvent, ServerUserCreatedEvent, + ServerUserDeletedEvent, +}; use crate::domain::events::server_tree::ServerTreeInvalidatedEvent; use crate::domain::events::voice_presence::VoicePresenceEvent; use crate::models::{server_user, user}; @@ -95,6 +98,12 @@ impl GatewayManager { } } +#[derive(Debug, Serialize)] +pub struct ServerUserPayload { + pub server_id: Uuid, + pub user_id: Uuid, +} + /// Résout les audiences puis délègue l'envoi au transport WebSocket. pub struct RealtimeRouter { gateway: Arc, @@ -381,6 +390,41 @@ impl RealtimeRouter { } }); + let router = Arc::clone(self); + event_bus.on_async::(move |event| { + let router = Arc::clone(&router); + async move { + let server_id = event.server_user.server_id; + let users = router.server_users(server_id).await; + router.gateway.send_to_users( + users, + "ServerUser", + "add", + ServerUserPayload { + server_id, + user_id: event.server_user.user_id, + }, + ); + } + }); + + let router = Arc::clone(self); + event_bus.on_async::(move |event| { + let router = Arc::clone(&router); + async move { + let server_id = event.server_user.server_id; + let user_id = event.server_user.user_id; + let mut users = router.server_users(server_id).await; + users.push(user_id); + router.gateway.send_to_users( + users, + "ServerUser", + "remove", + ServerUserPayload { server_id, user_id }, + ); + } + }); + let router = Arc::clone(self); event_bus.on_async::(move |event| { let router = Arc::clone(&router); diff --git a/src/routes/message/mod.rs b/src/routes/message/mod.rs index a376414..5406b88 100644 --- a/src/routes/message/mod.rs +++ b/src/routes/message/mod.rs @@ -3,3 +3,6 @@ pub mod handlers; pub mod mapper; pub mod routes; pub mod service; + +#[cfg(test)] +mod tests; diff --git a/src/routes/message/tests.rs b/src/routes/message/tests.rs new file mode 100644 index 0000000..76cd5e8 --- /dev/null +++ b/src/routes/message/tests.rs @@ -0,0 +1,114 @@ +use crate::core::AppState; +use crate::domain::dto::channel::CreateChannelRequest; +use crate::http::test_support::{request, state, user}; +use crate::models::{channel, computed_permission, computed_permission::PermissionScopeType, message}; +use crate::permissions::ChannelPermission; +use axum::{body::{to_bytes, Body}, http::{Method, StatusCode}, Router}; +use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, Set}; +use serde_json::Value; +use std::time::Duration; +use tower::ServiceExt; +use uuid::Uuid; + +async fn call(router: &Router, uri: &str, actor: crate::models::user::Model) -> axum::response::Response { + let req = request(Method::GET, uri, Body::empty(), Some(actor)); + router.clone().oneshot(req).await.unwrap() +} + +async fn wait_for_read_cache(state: &AppState, user_id: Uuid, channel_id: Uuid) -> bool { + for _ in 0..50 { + let cached = computed_permission::Entity::find() + .filter(computed_permission::Column::UserId.eq(user_id)) + .filter(computed_permission::Column::ScopeType.eq(PermissionScopeType::Channel)) + .filter(computed_permission::Column::ResourceId.eq(channel_id)) + .one(&state.db) + .await + .unwrap(); + if cached.is_some_and(|row| { + ChannelPermission::from_bits_retain(row.permissions as u64).contains(ChannelPermission::READ_CHANNEL) + }) { + return true; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + false +} + +#[tokio::test] +async fn new_member_reads_history_and_receives_realtime_after_join() { + let state = state().await; + let owner = user(&state, false).await; + let member = user(&state, false).await; + + let server = state + .services + .server + .create_server("join-history".into(), None, false, owner.id) + .await + .unwrap(); + let channel = state + .services + .channel + .create_channel(CreateChannelRequest { + server_id: Some(server.id), + category_id: None, + channel_type: channel::ChannelType::Text, + name: Some("history".into()), + }) + .await + .unwrap(); + + for index in 0..3 { + state + .repositories + .message + .create(message::ActiveModel { + id: Set(Uuid::now_v7()), + channel_id: Set(channel.id), + user_id: Set(owner.id), + content: Set(format!("message {index}")), + created_at: Set(chrono::Utc::now()), + updated_at: Set(None), + reply_to_id: Set(None), + }) + .await + .unwrap(); + } + + assert!(wait_for_read_cache(&state, owner.id, channel.id).await); + state + .services + .realtime_registry + .refresh_channel(&state.repositories, channel.id) + .await + .unwrap(); + + state + .services + .server + .join_server(server.id, member.id, None) + .await + .unwrap(); + + assert!( + wait_for_read_cache(&state, member.id, channel.id).await, + "join must fill the computed permission cache" + ); + + let router = super::routes::router().with_state(state.clone()); + let uri = format!("/messages?channel_id={}&limit=50", channel.id); + let response = call(&router, &uri, member.clone()).await; + assert_eq!(response.status(), StatusCode::OK); + let data: Value = + serde_json::from_slice(&to_bytes(response.into_body(), 1024 * 1024).await.unwrap()).unwrap(); + assert_eq!(data["messages"].as_array().unwrap().len(), 3); + + assert!( + state + .services + .realtime_registry + .users_for_channel(channel.id) + .contains(&member.id), + "a joined member must be in the realtime audience of the channel" + ); +} diff --git a/src/services/realtime_registry.rs b/src/services/realtime_registry.rs index 71662d9..c1691dc 100644 --- a/src/services/realtime_registry.rs +++ b/src/services/realtime_registry.rs @@ -56,7 +56,7 @@ impl RealtimeRegistry { Ok(()) } - async fn refresh_user(&self, repositories: &Repositories, user_id: Uuid) -> anyhow::Result<()> { + pub async fn refresh_user(&self, repositories: &Repositories, user_id: Uuid) -> anyhow::Result<()> { let channels = repositories .computed_permission .get_all() diff --git a/src/services/server.rs b/src/services/server.rs index 115d28b..62e1cf8 100644 --- a/src/services/server.rs +++ b/src/services/server.rs @@ -125,6 +125,29 @@ impl ServerService { .await?; txn.commit().await?; + + let services = self + .service_context + .services + .get() + .expect("services initialized"); + if let Err(error) = self + .service_context + .repositories + .computed_permission + .full_sync_user(user_id, server_id) + .await + { + tracing::error!(%user_id, %server_id, ?error, "Unable to sync permissions on join"); + } + if let Err(error) = services + .realtime_registry + .refresh_user(&self.service_context.repositories, user_id) + .await + { + tracing::error!(%user_id, ?error, "Unable to refresh realtime registry on join"); + } + event_bus.emit(ServerUserCreatedEvent { server_user: membership, });