refactor(chat): migrate chatlayer and storagelayer to central lib/config

Modified chat/chatlayer/actor/mod.rs +1 −0 View chat/chatlayer/actor/mod.rs

@@ -1,3 +1,4 @@

1pub mod live;

12pub mod room;

23pub mod runtime;

34pub mod user;

Modified chat/chatlayer/actor/room.rs +3 −3 View chat/chatlayer/actor/room.rs

@@ -34,15 +34,15 @@

3434}

3535

3636impl RoomActorCtx {

37 /// Errors are logged and swallowed; Kafka at-least-once redelivery is

38 pub async fn handle(&self, envelope: EventEnvelope<ChatRoomEvent>) {

37 /// Errors are logged and swallowed; Kafka at-least-once redelivery is

38 pub async fn handle(&self, envelope: EventEnvelope<ChatRoomEvent>) {

3939 match self.handle_inner(&envelope).await {

4040 Ok(()) => crate::metrics::room_event_applied("success"),

4141 Err(e) => {

4242 crate::metrics::room_event_applied("failure");

4343 tracing::error!(

4444 event_id = %envelope.event_id,

45 channel = %envelope.payload.channel_id(),

45 channel = %envelope.payload.conversation_id(),

4646 error = %e,

4747 "room event apply failed"

4848 );

Modified chat/chatlayer/actor/runtime.rs +3 −3 View chat/chatlayer/actor/runtime.rs

@@ -64,17 +64,17 @@

6464 self.shutdown.clone()

6565 }

6666

67 pub fn live_count(&self) -> usize {

67 pub fn live_count(&self) -> usize {

6868 self.actors.lock().unwrap().len()

6969 }

7070

71 pub fn dispatch(self: &Arc<Self>, key: Uuid, mut msg: M) {

71 pub fn dispatch(self: &Arc<Self>, key: Uuid, mut msg: M) {

7272 {

7373 let guard = self.actors.lock().unwrap();

7474 if let Some(tx) = guard.get(&key) {

7575 match tx.send(msg) {

7676 Ok(()) => return,

77 Err(e) => msg = e.0,

77 Err(e) => msg = e.0,

7878 }

7979 }

8080 }

Modified chat/chatlayer/actor/user.rs +3 −3 View chat/chatlayer/actor/user.rs

@@ -4,13 +4,13 @@

44use tokio_util::sync::CancellationToken;

55

66use crate::actor::runtime::{ActorRegistry, run_actor};

7use crate::config::AppChatSettings;

87use crate::dedup;

98use crate::error::ChatLayerResult;

109use crate::events::envelope::EventEnvelope;

1110use crate::events::user::ChatUserEvent;

1211use crate::producer::ChatProducer;

1312use crate::projection;

13use config::chat::AppChatSettings;

1414

1515#[derive(Clone)]

1616pub struct UserActorCtx {

@@ -28,8 +28,8 @@

2828}

2929

3030impl UserActorCtx {

31 /// view events for the (future) gateway. Errors are logged and swallowed;

32 pub async fn handle(&self, envelope: EventEnvelope<ChatUserEvent>) {

31 /// view events for the (future) gateway. Errors are logged and swallowed;

32 pub async fn handle(&self, envelope: EventEnvelope<ChatUserEvent>) {

3333 match self.handle_inner(&envelope).await {

3434 Ok(()) => crate::metrics::user_event_applied("success"),

3535 Err(e) => {

Modified chat/chatlayer/consumer/mod.rs +9 −1 View chat/chatlayer/consumer/mod.rs

@@ -1,9 +1,10 @@

1pub mod live_worker;

12pub mod room_worker;

23pub mod user_worker;

34

45use queue::{AutoOffsetReset, SubscribeHandle, SubscriptionMode};

56

6use crate::config::AppChatSettings;

7use config::chat::AppChatSettings;

78

89pub fn group_mode(group_id: &str) -> SubscriptionMode {

910 SubscriptionMode::Group {

@@ -59,4 +60,11 @@

5960 group_id: cfg.user_group_id.clone(),

6061 }

6162 }

63

64 pub fn live(cfg: &AppChatSettings) -> Self {

65 Self {

66 topic: crate::topics::live_topic(&cfg.topic_prefix),

67 group_id: cfg.live_group_id.clone(),

68 }

69 }

6270}

Modified chat/chatlayer/consumer/room_worker.rs +1 −1 View chat/chatlayer/consumer/room_worker.rs

@@ -45,7 +45,7 @@

4545 Box::pin(async move {

4646 match serde_json::from_slice::<EventEnvelope<ChatRoomEvent>>(&msg.payload) {

4747 Ok(envelope) => {

48 registry.dispatch(envelope.payload.channel_id(), envelope);

48 registry.dispatch(envelope.payload.conversation_id(), envelope);

4949 Ok(())

5050 }

5151 Err(e) => {

Modified chat/chatlayer/events/envelope.rs +1 −1 View chat/chatlayer/events/envelope.rs

@@ -31,7 +31,7 @@

3131}

3232

3333impl<T: Clone> EventEnvelope<&T> {

34 pub fn into_owned(self) -> EventEnvelope<T> {

34 pub fn into_owned(self) -> EventEnvelope<T> {

3535 EventEnvelope {

3636 event_id: self.event_id,

3737 occurred_at: self.occurred_at,

Modified chat/chatlayer/events/mod.rs +9 −7 View chat/chatlayer/events/mod.rs

@@ -1,9 +1,11 @@

11pub mod envelope;

2pub mod live;

23pub mod room;

34pub mod user;

45pub mod user_view;

56

67pub use envelope::EventEnvelope;

8pub use live::{ChatLiveEvent, ChatLiveViewEvent};

79pub use room::{ChatRoomEvent, MentionSpec, MessageCommand};

810pub use user::ChatUserEvent;

911pub use user_view::ChatUserViewEvent;

@@ -16,7 +18,7 @@

1618 #[test]

1719 fn room_event_serde_roundtrip() {

1820 let cmd = MessageCommand {

19 channel_id: Uuid::now_v7(),

21 conversation_id: Uuid::now_v7(),

2022 sender_id: Uuid::now_v7(),

2123 body: "hi".into(),

2224 reply_to: None,

@@ -34,21 +36,21 @@

3436 }

3537

3638 #[test]

37 fn room_event_channel_id() {

39 fn room_event_conversation_id() {

3840 let channel = Uuid::now_v7();

3941 let evt = ChatRoomEvent::MessageDeleted {

40 channel_id: channel,

42 conversation_id: channel,

4143 message_id: Uuid::now_v7(),

4244 by: Uuid::now_v7(),

4345 };

44 assert_eq!(evt.channel_id(), channel);

46 assert_eq!(evt.conversation_id(), channel);

4547 }

4648

4749 #[test]

4850 fn user_event_serde_roundtrip() {

4951 let evt = ChatUserEvent::ReadReceiptAck {

5052 user_id: Uuid::now_v7(),

51 channel_id: Uuid::now_v7(),

53 conversation_id: Uuid::now_v7(),

5254 last_read_message_id: Uuid::now_v7(),

5355 ts: time::OffsetDateTime::now_utc(),

5456 };

@@ -61,12 +63,12 @@

6163 fn envelope_key_derivation() {

6264 let channel = Uuid::now_v7();

6365 let evt = ChatRoomEvent::PinToggle {

64 channel_id: channel,

66 conversation_id: channel,

6567 message_id: Uuid::now_v7(),

6668 by: Uuid::now_v7(),

6769 pinned: true,

6870 };

69 assert_eq!(evt.channel_id(), channel);

71 assert_eq!(evt.conversation_id(), channel);

7072 assert_eq!(crate::topics::room_key(channel), channel.to_string());

7173 }

7274}

Modified chat/chatlayer/events/room.rs +27 −15 View chat/chatlayer/events/room.rs

@@ -34,7 +34,7 @@

3434

3535#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]

3636pub struct MessageCommand {

37 pub channel_id: Uuid,

37 pub conversation_id: Uuid,

3838 pub sender_id: Uuid,

3939 pub body: String,

4040 pub reply_to: Option<Uuid>,

@@ -50,35 +50,35 @@

5050pub enum ChatRoomEvent {

5151 MessageCreated(MessageCommand),

5252 MessageEdited {

53 channel_id: Uuid,

53 conversation_id: Uuid,

5454 message_id: Uuid,

5555 editor_id: Uuid,

5656 new_body: String,

5757 },

5858 MessageDeleted {

59 channel_id: Uuid,

59 conversation_id: Uuid,

6060 message_id: Uuid,

6161 by: Uuid,

6262 },

6363 ReactionAdded {

64 channel_id: Uuid,

64 conversation_id: Uuid,

6565 message_id: Uuid,

6666 user_id: Uuid,

6767 emoji: String,

6868 },

6969 ReactionRemoved {

70 channel_id: Uuid,

70 conversation_id: Uuid,

7171 message_id: Uuid,

7272 user_id: Uuid,

7373 emoji: String,

7474 },

7575 MentionAdded {

76 channel_id: Uuid,

76 conversation_id: Uuid,

7777 message_id: Uuid,

7878 mentions: Vec<MentionSpec>,

7979 },

8080 PinToggle {

81 channel_id: Uuid,

81 conversation_id: Uuid,

8282 message_id: Uuid,

8383 by: Uuid,

8484 pinned: bool,

@@ -86,15 +86,27 @@

8686}

8787

8888impl ChatRoomEvent {

89 pub fn channel_id(&self) -> Uuid {

89 pub fn conversation_id(&self) -> Uuid {

9090 match self {

91 Self::MessageCreated(cmd) => cmd.channel_id,

92 Self::MessageEdited { channel_id, .. }

93 | Self::MessageDeleted { channel_id, .. }

94 | Self::ReactionAdded { channel_id, .. }

95 | Self::ReactionRemoved { channel_id, .. }

96 | Self::MentionAdded { channel_id, .. }

97 | Self::PinToggle { channel_id, .. } => *channel_id,

91 Self::MessageCreated(cmd) => cmd.conversation_id,

92 Self::MessageEdited {

93 conversation_id, ..

94 }

95 | Self::MessageDeleted {

96 conversation_id, ..

97 }

98 | Self::ReactionAdded {

99 conversation_id, ..

100 }

101 | Self::ReactionRemoved {

102 conversation_id, ..

103 }

104 | Self::MentionAdded {

105 conversation_id, ..

106 }

107 | Self::PinToggle {

108 conversation_id, ..

109 } => *conversation_id,

98110 }

99111 }

100112}

Modified chat/chatlayer/events/user.rs +4 −4 View chat/chatlayer/events/user.rs

@@ -7,20 +7,20 @@

77pub enum ChatUserEvent {

88 MessageDelivered {

99 user_id: Uuid,

10 channel_id: Uuid,

10 conversation_id: Uuid,

1111 message_id: Uuid,

1212 ts: OffsetDateTime,

1313 },

1414 MentionReceived {

1515 user_id: Uuid,

16 channel_id: Uuid,

16 conversation_id: Uuid,

1717 message_id: Uuid,

1818 by: Uuid,

1919 ts: OffsetDateTime,

2020 },

2121 ReadReceiptAck {

2222 user_id: Uuid,

23 channel_id: Uuid,

23 conversation_id: Uuid,

2424 last_read_message_id: Uuid,

2525 ts: OffsetDateTime,

2626 },

@@ -32,7 +32,7 @@

3232}

3333

3434impl ChatUserEvent {

35 pub fn user_id(&self) -> Uuid {

35 pub fn user_id(&self) -> Uuid {

3636 match self {

3737 Self::MessageDelivered { user_id, .. }

3838 | Self::MentionReceived { user_id, .. }

Modified chat/chatlayer/events/user_view.rs +4 −4 View chat/chatlayer/events/user_view.rs

@@ -7,7 +7,7 @@

77pub enum ChatUserViewEvent {

88 NewMessage {

99 user_id: Uuid,

10 channel_id: Uuid,

10 conversation_id: Uuid,

1111 message_id: Uuid,

1212 sender_id: Uuid,

1313 body: String,

@@ -15,20 +15,20 @@

1515 },

1616 MessageEditedView {

1717 user_id: Uuid,

18 channel_id: Uuid,

18 conversation_id: Uuid,

1919 message_id: Uuid,

2020 new_body: String,

2121 ts: OffsetDateTime,

2222 },

2323 MessageDeletedView {

2424 user_id: Uuid,

25 channel_id: Uuid,

25 conversation_id: Uuid,

2626 message_id: Uuid,

2727 ts: OffsetDateTime,

2828 },

2929 UnreadChanged {

3030 user_id: Uuid,

31 channel_id: Uuid,

31 conversation_id: Uuid,

3232 unread: u64,

3333 ts: OffsetDateTime,

3434 },

Modified chat/chatlayer/projection/mod.rs +2 −0 View chat/chatlayer/projection/mod.rs

@@ -1,5 +1,7 @@

1pub mod live;

12pub mod room;

23pub mod user;

34

5pub use live::apply_live_event;

46pub use room::apply_room_event;

57pub use user::apply_user_event;

Modified chat/chatlayer/projection/room.rs +11 −8 View chat/chatlayer/projection/room.rs

@@ -78,7 +78,7 @@

7878 Ok(Vec::new())

7979 }

8080 ChatRoomEvent::PinToggle {

81 channel_id,

81 conversation_id,

8282 message_id,

8383 pinned,

8484 ..

@@ -89,14 +89,17 @@

8989 .pin(

9090 &ctx,

9191 NewPinnedMessage {

92 chat_id: *channel_id,

92 chat_id: *conversation_id,

9393 message_id: *message_id,

9494 pinned_by: envelope.actor_id,

9595 },

9696 )

9797 .await?;

9898 } else {

99 store.pinned().unpin(&ctx, *channel_id, *message_id).await?;

99 store

100 .pinned()

101 .unpin(&ctx, *conversation_id, *message_id)

102 .await?;

100103 }

101104 Ok(Vec::new())

102105 }

@@ -121,7 +124,7 @@

121124 .create(

122125 ctx,

123126 NewMessage {

124 chat_id: cmd.channel_id,

127 chat_id: cmd.conversation_id,

125128 sender_id: cmd.sender_id,

126129 body: cmd.body.clone(),

127130 reply_to_message_id: cmd.reply_to,

@@ -153,7 +156,7 @@

153156 if cmd.sender_id != ctx.user_id.unwrap_or(Uuid::nil()) {

154157 out.push(ChatUserEvent::MessageDelivered {

155158 user_id: cmd.sender_id,

156 channel_id: cmd.channel_id,

159 conversation_id: cmd.conversation_id,

157160 message_id: message.id,

158161 ts: message.created_at,

159162 });

@@ -162,7 +165,7 @@

162165 if let Some(target) = mention_user(mention) {

163166 out.push(ChatUserEvent::MentionReceived {

164167 user_id: target,

165 channel_id: cmd.channel_id,

168 conversation_id: cmd.conversation_id,

166169 message_id: message.id,

167170 by: cmd.sender_id,

168171 ts: message.created_at,

@@ -215,7 +218,7 @@

215218 use storagelayer::models::message::MessageQuery;

216219 use storagelayer::repository::ChatStorageCtx;

217220

218 struct MemIdempotency(Mutex<HashMap<String, Uuid>>);

221 struct MemIdempotency(Mutex<HashMap<String, Uuid>>);

219222

220223 impl MessageIdempotencyStore for MemIdempotency {

221224 fn is_resolved<'a>(

@@ -243,7 +246,7 @@

243246

244247 fn cmd(channel: Uuid, sender: Uuid, body: &str, client_msg_id: &str) -> MessageCommand {

245248 MessageCommand {

246 channel_id: channel,

249 conversation_id: channel,

247250 sender_id: sender,

248251 body: body.to_string(),

249252 reply_to: None,

Modified chat/chatlayer/projection/user.rs +24 −18 View chat/chatlayer/projection/user.rs

@@ -4,19 +4,19 @@

44use time::OffsetDateTime;

55use uuid::Uuid;

66

7use crate::config::AppChatSettings;

87use crate::error::ChatLayerResult;

98use crate::events::envelope::EventEnvelope;

109use crate::events::user::ChatUserEvent;

1110use crate::events::user_view::ChatUserViewEvent;

11use config::chat::AppChatSettings;

1212

1313pub const UNREAD_PREFIX: &str = "chat:unread:";

1414pub const PRESENCE_PREFIX: &str = "chat:presence:";

1515pub const MENTION_QUEUE_PREFIX: &str = "chat:mentions:";

1616pub const READ_RECEIPT_PREFIX: &str = "chat:read:";

1717

18pub fn unread_key(user_id: Uuid, channel_id: Uuid) -> String {

19 format!("{UNREAD_PREFIX}{user_id}:{channel_id}")

18pub fn unread_key(user_id: Uuid, conversation_id: Uuid) -> String {

19 format!("{UNREAD_PREFIX}{user_id}:{conversation_id}")

2020}

2121

2222pub fn presence_key(user_id: Uuid) -> String {

@@ -27,8 +27,8 @@

2727 format!("{MENTION_QUEUE_PREFIX}{user_id}")

2828}

2929

30fn read_receipt_key(user_id: Uuid, channel_id: Uuid) -> String {

31 format!("{READ_RECEIPT_PREFIX}{user_id}:{channel_id}")

30fn read_receipt_key(user_id: Uuid, conversation_id: Uuid) -> String {

31 format!("{READ_RECEIPT_PREFIX}{user_id}:{conversation_id}")

3232}

3333

3434pub async fn apply_user_event(

@@ -39,25 +39,28 @@

3939 match &envelope.payload {

4040 ChatUserEvent::MessageDelivered {

4141 user_id,

42 channel_id,

42 conversation_id,

4343 message_id,

4444 ts,

4545 } => {

4646 let count = kv

47 .incr_with_ttl(&unread_key(*user_id, *channel_id), cfg.unread_ttl_seconds)

47 .incr_with_ttl(

48 &unread_key(*user_id, *conversation_id),

49 cfg.unread_ttl_seconds,

50 )

4851 .await?;

4952 Ok(vec![

5053 ChatUserViewEvent::NewMessage {

5154 user_id: *user_id,

52 channel_id: *channel_id,

55 conversation_id: *conversation_id,

5356 message_id: *message_id,

5457 sender_id: envelope.actor_id,

55 body: String::new(),

58 body: String::new(),

5659 ts: *ts,

5760 },

5861 ChatUserViewEvent::UnreadChanged {

5962 user_id: *user_id,

60 channel_id: *channel_id,

63 conversation_id: *conversation_id,

6164 unread: count,

6265 ts: *ts,

6366 },

@@ -65,13 +68,16 @@

6568 }

6669 ChatUserEvent::MentionReceived {

6770 user_id,

68 channel_id,

71 conversation_id,

6972 message_id,

7073 by,

7174 ts,

7275 } => {

7376 let count = kv

74 .incr_with_ttl(&unread_key(*user_id, *channel_id), cfg.unread_ttl_seconds)

77 .incr_with_ttl(

78 &unread_key(*user_id, *conversation_id),

79 cfg.unread_ttl_seconds,

80 )

7581 .await?;

7682 kv.rpush(

7783 &mention_queue_key(*user_id),

@@ -81,7 +87,7 @@

8187 Ok(vec![

8288 ChatUserViewEvent::NewMessage {

8389 user_id: *user_id,

84 channel_id: *channel_id,

90 conversation_id: *conversation_id,

8591 message_id: *message_id,

8692 sender_id: *by,

8793 body: String::new(),

@@ -89,7 +95,7 @@

8995 },

9096 ChatUserViewEvent::UnreadChanged {

9197 user_id: *user_id,

92 channel_id: *channel_id,

98 conversation_id: *conversation_id,

9399 unread: count,

94100 ts: *ts,

95101 },

@@ -97,20 +103,20 @@

97103 }

98104 ChatUserEvent::ReadReceiptAck {

99105 user_id,

100 channel_id,

106 conversation_id,

101107 last_read_message_id,

102108 ts,

103109 } => {

104110 kv.set_bytes(

105 &read_receipt_key(*user_id, *channel_id),

111 &read_receipt_key(*user_id, *conversation_id),

106112 last_read_message_id.as_bytes().to_vec(),

107113 cfg.unread_ttl_seconds,

108114 )

109115 .await?;

110 kv.delete(&unread_key(*user_id, *channel_id)).await?;

116 kv.delete(&unread_key(*user_id, *conversation_id)).await?;

111117 Ok(vec![ChatUserViewEvent::UnreadChanged {

112118 user_id: *user_id,

113 channel_id: *channel_id,

119 conversation_id: *conversation_id,

114120 unread: 0,

115121 ts: *ts,

116122 }])

Modified chat/chatlayer/auth_bridge.rs +3 −3 View chat/chatlayer/auth_bridge.rs

@@ -28,7 +28,7 @@

2828 Self { jwt }

2929 }

3030

31 pub async fn verify(&self, token: &str) -> ChatLayerResult<AuthIdentity> {

31 pub async fn verify(&self, token: &str) -> ChatLayerResult<AuthIdentity> {

3232 let kid = extract_kid(token).ok_or(JwtError::InvalidHeader)?;

3333 let cache = self.jwt.cache.read().await;

3434 let cached = cache

@@ -48,12 +48,12 @@

4848 })

4949 }

5050

51 pub async fn ctx_for(&self, token: &str) -> ChatLayerResult<ChatStorageCtx> {

51 pub async fn ctx_for(&self, token: &str) -> ChatLayerResult<ChatStorageCtx> {

5252 let identity = self.verify(token).await?;

5353 Ok(ChatStorageCtx::new(identity.user_id))

5454 }

5555

56 pub async fn require_identity(&self, token: &str) -> ChatLayerResult<AuthIdentity> {

56 pub async fn require_identity(&self, token: &str) -> ChatLayerResult<AuthIdentity> {

5757 self.verify(token).await.map_err(|e| match e {

5858 ChatLayerError::Jwt(_) => {

5959 ChatLayerError::PermissionDenied("valid bearer token required".to_string())

Modified chat/chatlayer/Cargo.toml +1 −0 View chat/chatlayer/Cargo.toml

@@ -14,6 +14,7 @@

1414

1515[dependencies]

1616storagelayer.workspace = true

17config.workspace = true

1718queue.workspace = true

1819cache.workspace = true

1920storage.workspace = true

Deleted chat/chatlayer/config.rs +0 −42

@@ -1,42 +0,0 @@

1use serde::{Deserialize, Serialize};

2

3#[derive(Debug, Clone, Serialize, Deserialize)]

4#[serde(default, deny_unknown_fields)]

5pub struct AppChatSettings {

6 pub enabled: bool,

7

8 pub topic_prefix: String,

9

10 pub room_group_id: String,

11

12 pub user_group_id: String,

13

14 pub user_view_group_id: String,

15

16 pub dedup_ttl_seconds: u64,

17

18 pub idempotency_ttl_seconds: u64,

19

20 pub upload_ttl_seconds: u64,

21

22 pub presence_ttl_seconds: u64,

23

24 pub unread_ttl_seconds: u64,

25}

26

27impl Default for AppChatSettings {

28 fn default() -> Self {

29 Self {

30 enabled: true,

31 topic_prefix: "chat".to_string(),

32 room_group_id: "chat.room.v1".to_string(),

33 user_group_id: "chat.user.v1".to_string(),

34 user_view_group_id: "chat.user.view.v1".to_string(),

35 dedup_ttl_seconds: 24 * 3600,

36 idempotency_ttl_seconds: 24 * 3600,

37 upload_ttl_seconds: 3600,

38 presence_ttl_seconds: 120,

39 unread_ttl_seconds: 30 * 24 * 3600,

40 }

41 }

42}

Modified chat/chatlayer/lib.rs +3 −4 View chat/chatlayer/lib.rs

@@ -3,7 +3,6 @@

33

44pub mod actor;

55pub mod auth_bridge;

6pub mod config;

76pub mod consumer;

87pub mod dedup;

98pub mod error;

@@ -17,14 +16,14 @@

1716pub mod topics;

1817

1918pub use auth_bridge::{AuthBridge, AuthIdentity};

20pub use config::AppChatSettings;

2119pub use error::{ChatLayerError, ChatLayerResult};

2220pub use events::{

2321 envelope::EventEnvelope,

22 live::{ChatLiveEvent, ChatLiveViewEvent},

2423 room::{ChatRoomEvent, MentionSpec, MessageCommand},

2524 user::ChatUserEvent,

2625 user_view::ChatUserViewEvent,

2726};

28pub use media::{MediaService, PresignedUpload, UploadDraft};

27pub use media::{AttachmentMeta, MediaService, PresignedUpload, UploadDraft};

2928pub use producer::ChatProducer;

30pub use service::{AcceptedMsg, ChatService, WorkerHandle};

29pub use service::{AcceptedMsg, ChatService, NewMessageCmd, WorkerHandle};

Modified chat/chatlayer/media.rs +5 −5 View chat/chatlayer/media.rs

@@ -96,7 +96,7 @@

9696 ObjectPath::from(format!("{BLOB_PREFIX}/{key}"))

9797 }

9898

99 #[allow(clippy::too_many_arguments)]

99 #[allow(clippy::too_many_arguments)]

100100 pub async fn presign_upload(

101101 &self,

102102 auth: &AuthBridge,

@@ -183,7 +183,7 @@

183183 })

184184 }

185185

186 pub async fn finalize_upload(

186 pub async fn finalize_upload(

187187 &self,

188188 auth: &AuthBridge,

189189 token: &str,

@@ -232,7 +232,7 @@

232232 })

233233 }

234234

235 pub async fn abort_upload(

235 pub async fn abort_upload(

236236 &self,

237237 auth: &AuthBridge,

238238 token: &str,

@@ -253,7 +253,7 @@

253253 Ok(())

254254 }

255255

256 pub async fn attach_to_message(

256 pub async fn attach_to_message(

257257 &self,

258258 auth: &AuthBridge,

259259 token: &str,

@@ -265,7 +265,7 @@

265265 .await

266266 }

267267

268 pub async fn attach_draft(

268 pub async fn attach_draft(

269269 &self,

270270 actor_id: Uuid,

271271 attachment_id: Uuid,

Modified chat/chatlayer/metrics.rs +4 −0 View chat/chatlayer/metrics.rs

@@ -27,3 +27,7 @@

2727pub fn media_op(op: &'static str, result: &'static str) {

2828 counter!("chat.media_ops_total", "op" => op, "result" => result).increment(1);

2929}

30

31pub fn live_event_applied(result: &'static str) {

32 counter!("chat.live_events_applied_total", "result" => result).increment(1);

33}

Modified chat/chatlayer/producer.rs +30 −1 View chat/chatlayer/producer.rs

@@ -2,6 +2,7 @@

22

33use crate::error::{ChatLayerError, ChatLayerResult};

44use crate::events::envelope::EventEnvelope;

5use crate::events::live::{ChatLiveEvent, ChatLiveViewEvent};

56use crate::events::room::ChatRoomEvent;

67use crate::events::user::ChatUserEvent;

78use crate::events::user_view::ChatUserViewEvent;

@@ -50,7 +51,7 @@

5051 ) -> ChatLayerResult<()> {

5152 let queue = self.require_queue()?;

5253 let topic = topics::room_topic(&self.topic_prefix);

53 let key = topics::room_key(envelope.payload.channel_id());

54 let key = topics::room_key(envelope.payload.conversation_id());

5455 let payload = serde_json::to_vec(envelope)?;

5556 queue

5657 .publish_raw(&topic, Some(key.as_bytes()), &payload)

@@ -85,4 +86,32 @@

8586 .await

8687 .map_err(ChatLayerError::from)

8788 }

89

90 pub async fn publish_live(

91 &self,

92 envelope: &EventEnvelope<ChatLiveEvent>,

93 ) -> ChatLayerResult<()> {

94 let queue = self.require_queue()?;

95 let topic = topics::live_topic(&self.topic_prefix);

96 let key = topics::live_key(envelope.payload.channel_id());

97 let payload = serde_json::to_vec(envelope)?;

98 queue

99 .publish_raw(&topic, Some(key.as_bytes()), &payload)

100 .await

101 .map_err(ChatLayerError::from)

102 }

103

104 pub async fn publish_live_view(

105 &self,

106 envelope: &EventEnvelope<ChatLiveViewEvent>,

107 ) -> ChatLayerResult<()> {

108 let queue = self.require_queue()?;

109 let topic = topics::live_view_topic(&self.topic_prefix);

110 let key = topics::live_user_key(envelope.payload.user_id());

111 let payload = serde_json::to_vec(envelope)?;

112 queue

113 .publish_raw(&topic, Some(key.as_bytes()), &payload)

114 .await

115 .map_err(ChatLayerError::from)

116 }

88117}

Modified chat/chatlayer/service.rs +46 −24 View chat/chatlayer/service.rs

@@ -9,10 +9,10 @@

99use tokio_util::sync::CancellationToken;

1010use uuid::Uuid;

1111

12use crate::actor::live::{MeetActorCtx, spawn_meet_registry};

1213use crate::actor::room::{RoomActorCtx, spawn_room_registry};

1314use crate::actor::user::{UserActorCtx, spawn_user_registry};

1415use crate::auth_bridge::AuthBridge;

15use crate::config::AppChatSettings;

1616use crate::consumer::WorkerSettings;

1717use crate::error::{ChatLayerError, ChatLayerResult};

1818use crate::events::envelope::EventEnvelope;

@@ -21,10 +21,11 @@

2121use crate::media::MediaService;

2222use crate::producer::ChatProducer;

2323use crate::storage_bridge::{ClientMessageClaim, MessageIdempotency};

24use config::chat::AppChatSettings;

2425

2526#[derive(Debug, Clone, Serialize, Deserialize)]

2627pub struct NewMessageCmd {

27 pub channel_id: Uuid,

28 pub conversation_id: Uuid,

2829 pub body: String,

2930 pub reply_to: Option<Uuid>,

3031 pub thread_root: Option<Uuid>,

@@ -76,6 +77,7 @@

7677 shutdown: CancellationToken,

7778 room_ctx: Arc<RoomActorCtx>,

7879 user_ctx: Arc<UserActorCtx>,

80 meet_ctx: Arc<MeetActorCtx>,

7981}

8082

8183impl std::fmt::Debug for ChatService {

@@ -111,6 +113,12 @@

111113 producer: producer.clone(),

112114 settings: settings.clone(),

113115 });

116 let meet_ctx = Arc::new(MeetActorCtx {

117 store: store.clone(),

118 producer: producer.clone(),

119 kv: kv.clone(),

120 dedup_ttl_seconds: settings.dedup_ttl_seconds,

121 });

114122 Self {

115123 store,

116124 producer,

@@ -121,6 +129,7 @@

121129 shutdown,

122130 room_ctx,

123131 user_ctx,

132 meet_ctx,

124133 }

125134 }

126135

@@ -135,7 +144,6 @@

135144 pub fn auth(&self) -> &AuthBridge {

136145 &self.auth

137146 }

138

139147

140148 pub async fn send_message(

141149 &self,

@@ -163,7 +171,7 @@

163171 }

164172

165173 let command = crate::events::room::MessageCommand {

166 channel_id: cmd.channel_id,

174 conversation_id: cmd.conversation_id,

167175 sender_id: identity.user_id,

168176 body: cmd.body,

169177 reply_to: cmd.reply_to,

@@ -181,7 +189,7 @@

181189 })

182190 }

183191

184 async fn channel_for_message(

192 async fn conversation_for_message(

185193 &self,

186194 ctx: &ChatStorageCtx,

187195 message_id: Uuid,

@@ -204,11 +212,11 @@

204212 ) -> ChatLayerResult<()> {

205213 let identity = auth.require_identity(token).await?;

206214 let ctx = ChatStorageCtx::new(identity.user_id);

207 let channel_id = self.channel_for_message(&ctx, message_id).await?;

215 let conversation_id = self.conversation_for_message(&ctx, message_id).await?;

208216 let envelope = EventEnvelope::new(

209217 identity.user_id,

210218 ChatRoomEvent::MessageEdited {

211 channel_id,

219 conversation_id,

212220 message_id,

213221 editor_id: identity.user_id,

214222 new_body,

@@ -225,11 +233,11 @@

225233 ) -> ChatLayerResult<()> {

226234 let identity = auth.require_identity(token).await?;

227235 let ctx = ChatStorageCtx::new(identity.user_id);

228 let channel_id = self.channel_for_message(&ctx, message_id).await?;

236 let conversation_id = self.conversation_for_message(&ctx, message_id).await?;

229237 let envelope = EventEnvelope::new(

230238 identity.user_id,

231239 ChatRoomEvent::MessageDeleted {

232 channel_id,

240 conversation_id,

233241 message_id,

234242 by: identity.user_id,

235243 },

@@ -249,11 +257,11 @@

249257 return Err(ChatLayerError::invalid("emoji is required"));

250258 }

251259 let ctx = ChatStorageCtx::new(identity.user_id);

252 let channel_id = self.channel_for_message(&ctx, message_id).await?;

260 let conversation_id = self.conversation_for_message(&ctx, message_id).await?;

253261 let envelope = EventEnvelope::new(

254262 identity.user_id,

255263 ChatRoomEvent::ReactionAdded {

256 channel_id,

264 conversation_id,

257265 message_id,

258266 user_id: identity.user_id,

259267 emoji,

@@ -271,11 +279,11 @@

271279 ) -> ChatLayerResult<()> {

272280 let identity = auth.require_identity(token).await?;

273281 let ctx = ChatStorageCtx::new(identity.user_id);

274 let channel_id = self.channel_for_message(&ctx, message_id).await?;

282 let conversation_id = self.conversation_for_message(&ctx, message_id).await?;

275283 let envelope = EventEnvelope::new(

276284 identity.user_id,

277285 ChatRoomEvent::ReactionRemoved {

278 channel_id,

286 conversation_id,

279287 message_id,

280288 user_id: identity.user_id,

281289 emoji,

@@ -293,11 +301,11 @@

293301 ) -> ChatLayerResult<()> {

294302 let identity = auth.require_identity(token).await?;

295303 let ctx = ChatStorageCtx::new(identity.user_id);

296 let channel_id = self.channel_for_message(&ctx, message_id).await?;

304 let conversation_id = self.conversation_for_message(&ctx, message_id).await?;

297305 let envelope = EventEnvelope::new(

298306 identity.user_id,

299307 ChatRoomEvent::MentionAdded {

300 channel_id,

308 conversation_id,

301309 message_id,

302310 mentions,

303311 },

@@ -314,11 +322,11 @@

314322 ) -> ChatLayerResult<()> {

315323 let identity = auth.require_identity(token).await?;

316324 let ctx = ChatStorageCtx::new(identity.user_id);

317 let channel_id = self.channel_for_message(&ctx, message_id).await?;

325 let conversation_id = self.conversation_for_message(&ctx, message_id).await?;

318326 let envelope = EventEnvelope::new(

319327 identity.user_id,

320328 ChatRoomEvent::PinToggle {

321 channel_id,

329 conversation_id,

322330 message_id,

323331 by: identity.user_id,

324332 pinned,

@@ -331,7 +339,7 @@

331339 &self,

332340 auth: &AuthBridge,

333341 token: &str,

334 channel_id: Uuid,

342 conversation_id: Uuid,

335343 last_read_message_id: Uuid,

336344 ) -> ChatLayerResult<()> {

337345 let identity = auth.require_identity(token).await?;

@@ -339,7 +347,7 @@

339347 identity.user_id,

340348 ChatUserEvent::ReadReceiptAck {

341349 user_id: identity.user_id,

342 channel_id,

350 conversation_id,

343351 last_read_message_id,

344352 ts: time::OffsetDateTime::now_utc(),

345353 },

@@ -364,20 +372,19 @@

364372 );

365373 self.producer.publish_user(&envelope).await

366374 }

367

368375

369376 pub async fn list_messages(

370377 &self,

371378 auth: &AuthBridge,

372379 token: &str,

373 channel_id: Uuid,

380 conversation_id: Uuid,

374381 query: MessageQuery,

375382 ) -> ChatLayerResult<MessagePage> {

376383 let ctx = auth.ctx_for(token).await?;

377384 Ok(self

378385 .store

379386 .messages()

380 .list_by_conversation(&ctx, channel_id, query)

387 .list_by_conversation(&ctx, conversation_id, query)

381388 .await?)

382389 }

383390

@@ -401,7 +408,6 @@

401408 self.idempotency.lookup(client_msg_id).await

402409 }

403410

404

405411 pub fn queue(&self) -> Option<Arc<queue::Queue>> {

406412 self.producer.queue()

407413 }

@@ -438,7 +444,23 @@

438444 })

439445 }

440446

441 pub async fn shutdown(&self) -> ChatLayerResult<()> {

447 pub fn spawn_live_worker(self: &Arc<Self>) -> ChatLayerResult<WorkerHandle> {

448 let queue = self

449 .queue()

450 .ok_or_else(|| ChatLayerError::invalid("chat queue disabled"))?;

451 let settings = WorkerSettings::live(&self.settings);

452 let registry = spawn_meet_registry(self.meet_ctx.clone(), self.shutdown.clone());

453 let worker =

454 crate::consumer::live_worker::LiveWorker::new(queue, settings, registry.clone());

455 let handle = worker.run()?;

456 Ok(WorkerHandle {

457 role: "live",

458 handle,

459 live_actors: Box::new(move || registry.live_count()),

460 })

461 }

462

463 pub async fn shutdown(&self) -> ChatLayerResult<()> {

442464 self.shutdown.cancel();

443465 Ok(())

444466 }

Modified chat/chatlayer/storage_bridge.rs +7 −7 View chat/chatlayer/storage_bridge.rs

@@ -12,18 +12,18 @@

1212

1313#[derive(Debug, Clone, Copy, PartialEq, Eq)]

1414pub enum ClientMessageClaim {

15 Fresh,

16 AlreadyResolved(Uuid),

17 InFlight,

15 Fresh,

16 AlreadyResolved(Uuid),

17 InFlight,

1818}

1919

2020pub trait MessageIdempotencyStore: Send + Sync {

21 fn is_resolved<'a>(

21 fn is_resolved<'a>(

2222 &'a self,

2323 client_msg_id: &'a str,

2424 ) -> std::pin::Pin<Box<dyn Future<Output = ChatLayerResult<bool>> + Send + 'a>>;

2525

26 fn resolve<'a>(

26 fn resolve<'a>(

2727 &'a self,

2828 client_msg_id: &'a str,

2929 message_id: Uuid,

@@ -53,7 +53,7 @@

5353 format!("{MSG_IDEMPOTENCY_PREFIX}{client_msg_id}")

5454 }

5555

56 pub async fn claim(&self, client_msg_id: &str) -> ChatLayerResult<ClientMessageClaim> {

56 pub async fn claim(&self, client_msg_id: &str) -> ChatLayerResult<ClientMessageClaim> {

5757 let key = Self::key(client_msg_id);

5858 if self

5959 .kv

@@ -72,7 +72,7 @@

7272 }

7373 }

7474

75 pub async fn lookup(&self, client_msg_id: &str) -> ChatLayerResult<Option<Uuid>> {

75 pub async fn lookup(&self, client_msg_id: &str) -> ChatLayerResult<Option<Uuid>> {

7676 let key = Self::key(client_msg_id);

7777 match self.kv.get_bytes(&key).await? {

7878 Some(v) if v.len() == 16 => {

Modified chat/chatlayer/topics.rs +20 −2 View chat/chatlayer/topics.rs

@@ -3,6 +3,8 @@

33pub const ROOM_SUFFIX: &str = "room.v1";

44pub const USER_SUFFIX: &str = "user.v1";

55pub const USER_VIEW_SUFFIX: &str = "user.view.v1";

6pub const LIVE_SUFFIX: &str = "live.v1";

7pub const LIVE_VIEW_SUFFIX: &str = "live.view.v1";

68

79pub fn room_topic(prefix: &str) -> String {

810 format!("{prefix}.{ROOM_SUFFIX}")

@@ -16,10 +18,26 @@

1618 format!("{prefix}.{USER_VIEW_SUFFIX}")

1719}

1820

19pub fn room_key(channel_id: Uuid) -> String {

21pub fn live_topic(prefix: &str) -> String {

22 format!("{prefix}.{LIVE_SUFFIX}")

23}

24

25pub fn live_view_topic(prefix: &str) -> String {

26 format!("{prefix}.{LIVE_VIEW_SUFFIX}")

27}

28

29pub fn room_key(conversation_id: Uuid) -> String {

30 conversation_id.to_string()

31}

32

33pub fn user_key(user_id: Uuid) -> String {

34 user_id.to_string()

35}

36

37pub fn live_key(channel_id: Uuid) -> String {

2038 channel_id.to_string()

2139}

2240

23pub fn user_key(user_id: Uuid) -> String {

41pub fn live_user_key(user_id: Uuid) -> String {

2442 user_id.to_string()

2543}

Modified chat/storagelayer/mapper/conversation.rs +40 −2 View chat/storagelayer/mapper/conversation.rs

@@ -2,8 +2,8 @@

22

33use async_trait::async_trait;

44use sea_orm::{

5 ActiveModelTrait, EntityTrait, IntoActiveModel, PaginatorTrait, QueryFilter, QueryOrder,

6 QuerySelect, Set, TransactionTrait,

5 ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, PaginatorTrait, QueryFilter,

6 QueryOrder, QuerySelect, Set, TransactionTrait,

77};

88use space::error::SpaceError;

99use space::features::channels::usecase::ensure_channel_for_resource;

@@ -17,7 +17,9 @@

1717use crate::repository::ChatStorageCtx;

1818use crate::repository::conversation::ConversationRepository;

1919

20use db::channel::channels as channel_entity;

2021use db::chat::chats as chat_entity;

22use db::space::space_members;

2123

2224impl From<SpaceError> for ChatStorageError {

2325 fn from(e: SpaceError) -> Self {

@@ -190,6 +192,42 @@

190192 let db = self.db.reader();

191193 Ok(chat_entity::Entity::find_by_id(id).count(&db).await? > 0)

192194 }

195

196 async fn list_member_user_ids(

197 &self,

198 ctx: &ChatStorageCtx,

199 id: Uuid,

200 ) -> ChatStorageResult<Vec<Uuid>> {

201 auth::require_member(&self.db, ctx, id).await?;

202 let db = self.db.reader();

203 let chat = chat_entity::Entity::find_by_id(id)

204 .one(&db)

205 .await?

206 .ok_or(ChatStorageError::ConversationNotFound(id))?;

207 match chat.kind {

208 chat_entity::ChatKind::Private => {

209 let mut out = vec![chat.creator_user_id];

210 if let Some(c) = chat.counterpart_user_id

211 && c != chat.creator_user_id

212 {

213 out.push(c);

214 }

215 Ok(out)

216 }

217 chat_entity::ChatKind::Group => {

218 let channel = channel_entity::Entity::find_by_id(chat.channel_id)

219 .one(&db)

220 .await?

221 .ok_or(ChatStorageError::ConversationNotFound(id))?;

222 let rows = space_members::Entity::find()

223 .filter(space_members::Column::SpaceId.eq(channel.space_id))

224 .filter(space_members::Column::Status.eq(space_members::MemberStatus::Active))

225 .all(&db)

226 .await?;

227 Ok(rows.into_iter().map(|m| m.user_id).collect())

228 }

229 }

230 }

193231}

194232impl std::fmt::Debug for SeaConversationRepository {

195233 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

Modified chat/storagelayer/mock/in_memory.rs +341 −0 View chat/storagelayer/mock/in_memory.rs

@@ -14,6 +14,8 @@

1414use crate::error::{ChatStorageError, ChatStorageResult};

1515use crate::models::attachment::{Attachment, AttachmentOwner, NewAttachment};

1616use crate::models::conversation::{Conversation, ConversationKind, NewConversation};

17use crate::models::meet::{Meet, MeetStatus, NewMeet};

18use crate::models::meet_participant::{LeaveReason, MeetParticipant, NewMeetParticipant};

1719use crate::models::mention::{Mention, NewMention};

1820use crate::models::message::{

1921 ListDirection, Message, MessageFlags, MessagePage, MessageQuery, NewMessage, Page,

@@ -24,6 +26,10 @@

2426use crate::models::thread::Thread;

2527use crate::repository::attachment::{AttachmentRepository, DynAttachmentRepository};

2628use crate::repository::conversation::{ConversationRepository, DynConversationRepository};

29use crate::repository::meet::{DynMeetRepository, MeetRepository};

30use crate::repository::meet_participant::{

31 DynMeetParticipantRepository, MeetParticipantRepository,

32};

2733use crate::repository::mention::{DynMentionRepository, MentionRepository};

2834use crate::repository::message::{DynMessageRepository, MessageRepository};

2935use crate::repository::pinned::{DynPinnedMessageRepository, PinnedMessageRepository};

@@ -43,6 +49,9 @@

4349 pub read_receipts: RwLock<HashMap<Uuid, ReadReceipt>>,

4450 pub pinned: RwLock<HashMap<(Uuid, Uuid), PinnedMessage>>,

4551 pub threads: RwLock<HashMap<(Uuid, Uuid), Thread>>,

52 pub meets: RwLock<HashMap<Uuid, Meet>>,

53 pub meet_participants: RwLock<HashMap<Uuid, MeetParticipant>>,

54 pub members: RwLock<HashMap<Uuid, Vec<Uuid>>>,

4655}

4756

4857pub type SharedState = Arc<InMemoryState>;

@@ -51,6 +60,16 @@

5160 ctx.require_user()

5261}

5362

63fn require_self(ctx: &ChatStorageCtx, target_user_id: Uuid) -> ChatStorageResult<Uuid> {

64 let user_id = require_user(ctx)?;

65 if user_id != target_user_id {

66 return Err(ChatStorageError::PermissionDenied(format!(

67 "user {user_id} cannot act on behalf of {target_user_id}"

68 )));

69 }

70 Ok(user_id)

71}

72

5473fn is_admin(state: &SharedState, chat_id: Uuid, user_id: Uuid) -> bool {

5574 state

5675 .creators

@@ -277,6 +296,32 @@

277296 async fn exists(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<bool> {

278297 Ok(self.state.conversations.read().unwrap().contains_key(&id))

279298 }

299

300 async fn list_member_user_ids(

301 &self,

302 ctx: &ChatStorageCtx,

303 id: Uuid,

304 ) -> ChatStorageResult<Vec<Uuid>> {

305 require_member(&self.state, ctx, id)?;

306 if let Some(ms) = self.state.members.read().unwrap().get(&id).cloned() {

307 return Ok(ms);

308 }

309 let conv = self

310 .state

311 .conversations

312 .read()

313 .unwrap()

314 .get(&id)

315 .cloned()

316 .ok_or(ChatStorageError::ConversationNotFound(id))?;

317 let mut out = vec![conv.creator_user_id];

318 if let Some(c) = conv.counterpart_user_id

319 && c != conv.creator_user_id

320 {

321 out.push(c);

322 }

323 Ok(out)

324 }

280325}

281326

282327#[derive(Debug)]

@@ -1009,6 +1054,296 @@

10091054 }

10101055}

10111056

1057pub struct InMemoryMeetRepository {

1058 state: SharedState,

1059}

1060

1061impl InMemoryMeetRepository {

1062 pub fn new(state: SharedState) -> Self {

1063 Self { state }

1064 }

1065}

1066

1067#[async_trait]

1068impl MeetRepository for InMemoryMeetRepository {

1069 async fn create(&self, ctx: &ChatStorageCtx, input: NewMeet) -> ChatStorageResult<Meet> {

1070 let caller = require_user(ctx)?;

1071 if caller != input.created_by {

1072 return Err(ChatStorageError::PermissionDenied(format!(

1073 "user {caller} cannot create meet as user {}",

1074 input.created_by

1075 )));

1076 }

1077 let mut map = self.state.meets.write().unwrap();

1078 if map

1079 .values()

1080 .any(|m| m.channel_id == input.channel_id && m.status == MeetStatus::Active)

1081 {

1082 return Err(ChatStorageError::Conflict(

1083 "channel already has an active meet".into(),

1084 ));

1085 }

1086 let now = OffsetDateTime::now_utc();

1087 let meet = Meet {

1088 id: Uuid::now_v7(),

1089 channel_id: input.channel_id,

1090 room_name: input.room_name,

1091 status: MeetStatus::Pending,

1092 created_by: input.created_by,

1093 started_at: None,

1094 ended_at: None,

1095 created_at: now,

1096 updated_at: None,

1097 };

1098 map.insert(meet.id, meet.clone());

1099 Ok(meet)

1100 }

1101

1102 async fn find_active_by_channel(

1103 &self,

1104 ctx: &ChatStorageCtx,

1105 channel_id: Uuid,

1106 ) -> ChatStorageResult<Option<Meet>> {

1107 require_user(ctx)?;

1108 let map = self.state.meets.read().unwrap();

1109 Ok(map

1110 .values()

1111 .find(|m| m.channel_id == channel_id && m.status == MeetStatus::Active)

1112 .cloned())

1113 }

1114

1115 async fn find_by_room_name(

1116 &self,

1117 _ctx: &ChatStorageCtx,

1118 room_name: &str,

1119 ) -> ChatStorageResult<Option<Meet>> {

1120 let map = self.state.meets.read().unwrap();

1121 Ok(map.values().find(|m| m.room_name == room_name).cloned())

1122 }

1123

1124 async fn find_by_id(

1125 &self,

1126 _ctx: &ChatStorageCtx,

1127 meet_id: Uuid,

1128 ) -> ChatStorageResult<Option<Meet>> {

1129 let map = self.state.meets.read().unwrap();

1130 Ok(map.get(&meet_id).cloned())

1131 }

1132

1133 async fn mark_active(&self, ctx: &ChatStorageCtx, meet_id: Uuid) -> ChatStorageResult<Meet> {

1134 let caller = require_user(ctx)?;

1135 let mut map = self.state.meets.write().unwrap();

1136 let meet = map

1137 .get_mut(&meet_id)

1138 .ok_or(ChatStorageError::MeetNotFound(meet_id))?;

1139 if meet.created_by != caller {

1140 return Err(ChatStorageError::PermissionDenied(format!(

1141 "user {caller} cannot activate meet {meet_id}"

1142 )));

1143 }

1144 meet.status = MeetStatus::Active;

1145 meet.started_at = Some(OffsetDateTime::now_utc());

1146 meet.updated_at = Some(OffsetDateTime::now_utc());

1147 Ok(meet.clone())

1148 }

1149

1150 async fn mark_ended(&self, ctx: &ChatStorageCtx, meet_id: Uuid) -> ChatStorageResult<Meet> {

1151 require_user(ctx)?;

1152 let mut map = self.state.meets.write().unwrap();

1153 let meet = map

1154 .get_mut(&meet_id)

1155 .ok_or(ChatStorageError::MeetNotFound(meet_id))?;

1156 meet.status = MeetStatus::Ended;

1157 meet.ended_at = Some(OffsetDateTime::now_utc());

1158 meet.updated_at = Some(OffsetDateTime::now_utc());

1159 Ok(meet.clone())

1160 }

1161

1162 async fn list_for_channel(

1163 &self,

1164 ctx: &ChatStorageCtx,

1165 channel_id: Uuid,

1166 limit: u32,

1167 ) -> ChatStorageResult<Vec<Meet>> {

1168 require_user(ctx)?;

1169 let mut meets: Vec<_> = self

1170 .state

1171 .meets

1172 .read()

1173 .unwrap()

1174 .values()

1175 .filter(|m| m.channel_id == channel_id)

1176 .cloned()

1177 .collect();

1178 meets.sort_by_key(|m| std::cmp::Reverse(m.created_at));

1179 meets.truncate(limit as usize);

1180 Ok(meets)

1181 }

1182

1183 async fn list_channel_member_user_ids(

1184 &self,

1185 ctx: &ChatStorageCtx,

1186 channel_id: Uuid,

1187 ) -> ChatStorageResult<Vec<Uuid>> {

1188 require_user(ctx)?;

1189 let members = self.state.members.read().unwrap();

1190 Ok(members.get(&channel_id).cloned().unwrap_or_default())

1191 }

1192}

1193

1194impl std::fmt::Debug for InMemoryMeetRepository {

1195 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

1196 f.debug_struct("InMemoryMeetRepository")

1197 .finish_non_exhaustive()

1198 }

1199}

1200

1201pub struct InMemoryMeetParticipantRepository {

1202 state: SharedState,

1203}

1204

1205impl InMemoryMeetParticipantRepository {

1206 pub fn new(state: SharedState) -> Self {

1207 Self { state }

1208 }

1209}

1210

1211#[async_trait]

1212impl MeetParticipantRepository for InMemoryMeetParticipantRepository {

1213 async fn join(

1214 &self,

1215 ctx: &ChatStorageCtx,

1216 input: NewMeetParticipant,

1217 ) -> ChatStorageResult<MeetParticipant> {

1218 let caller = require_user(ctx)?;

1219 if caller != input.user_id {

1220 return Err(ChatStorageError::PermissionDenied(format!(

1221 "user {caller} cannot join as user {}",

1222 input.user_id

1223 )));

1224 }

1225 {

1226 let meets = self.state.meets.read().unwrap();

1227 let meet = meets

1228 .get(&input.meet_id)

1229 .ok_or(ChatStorageError::MeetNotFound(input.meet_id))?;

1230 if meet.status != MeetStatus::Active {

1231 return Err(ChatStorageError::InvalidInput(

1232 "meet is not active; cannot join".into(),

1233 ));

1234 }

1235 }

1236 let mut map = self.state.meet_participants.write().unwrap();

1237 if map.values().any(|p| {

1238 p.meet_id == input.meet_id && p.user_id == input.user_id && p.left_at.is_none()

1239 }) {

1240 return Err(ChatStorageError::Conflict(format!(

1241 "user {} already in meet {}",

1242 input.user_id, input.meet_id

1243 )));

1244 }

1245 let participant = MeetParticipant {

1246 id: Uuid::now_v7(),

1247 meet_id: input.meet_id,

1248 user_id: input.user_id,

1249 joined_at: input.joined_at,

1250 left_at: None,

1251 leave_reason: None,

1252 };

1253 map.insert(participant.id, participant.clone());

1254 Ok(participant)

1255 }

1256

1257 async fn leave(

1258 &self,

1259 ctx: &ChatStorageCtx,

1260 participant_id: Uuid,

1261 reason: LeaveReason,

1262 ) -> ChatStorageResult<MeetParticipant> {

1263 require_user(ctx)?;

1264 let mut map = self.state.meet_participants.write().unwrap();

1265 let p = map

1266 .get_mut(&participant_id)

1267 .ok_or(ChatStorageError::MeetParticipantNotFound(participant_id))?;

1268 if p.left_at.is_some() {

1269 return Err(ChatStorageError::Conflict(format!(

1270 "participant {participant_id} already left"

1271 )));

1272 }

1273 p.left_at = Some(OffsetDateTime::now_utc());

1274 p.leave_reason = Some(reason);

1275 Ok(p.clone())

1276 }

1277

1278 async fn list_active(

1279 &self,

1280 ctx: &ChatStorageCtx,

1281 meet_id: Uuid,

1282 ) -> ChatStorageResult<Vec<MeetParticipant>> {

1283 require_user(ctx)?;

1284 let map = self.state.meet_participants.read().unwrap();

1285 let mut list: Vec<_> = map

1286 .values()

1287 .filter(|p| p.meet_id == meet_id && p.left_at.is_none())

1288 .cloned()

1289 .collect();

1290 list.sort_by_key(|p| p.joined_at);

1291 Ok(list)

1292 }

1293

1294 async fn list_history(

1295 &self,

1296 ctx: &ChatStorageCtx,

1297 user_id: Uuid,

1298 limit: u32,

1299 ) -> ChatStorageResult<Vec<MeetParticipant>> {

1300 require_self(ctx, user_id)?;

1301 let map = self.state.meet_participants.read().unwrap();

1302 let mut list: Vec<_> = map

1303 .values()

1304 .filter(|p| p.user_id == user_id)

1305 .cloned()

1306 .collect();

1307 list.sort_by_key(|p| std::cmp::Reverse(p.joined_at));

1308 list.truncate(limit as usize);

1309 Ok(list)

1310 }

1311

1312 async fn is_active(

1313 &self,

1314 ctx: &ChatStorageCtx,

1315 meet_id: Uuid,

1316 user_id: Uuid,

1317 ) -> ChatStorageResult<bool> {

1318 require_user(ctx)?;

1319 let map = self.state.meet_participants.read().unwrap();

1320 Ok(map

1321 .values()

1322 .any(|p| p.meet_id == meet_id && p.user_id == user_id && p.left_at.is_none()))

1323 }

1324

1325 async fn mark_all_ended(&self, ctx: &ChatStorageCtx, meet_id: Uuid) -> ChatStorageResult<u64> {

1326 require_user(ctx)?;

1327 let mut map = self.state.meet_participants.write().unwrap();

1328 let mut count = 0u64;

1329 for p in map.values_mut() {

1330 if p.meet_id == meet_id && p.left_at.is_none() {

1331 p.left_at = Some(OffsetDateTime::now_utc());

1332 p.leave_reason = Some(LeaveReason::RoomEnded);

1333 count += 1;

1334 }

1335 }

1336 Ok(count)

1337 }

1338}

1339

1340impl std::fmt::Debug for InMemoryMeetParticipantRepository {

1341 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

1342 f.debug_struct("InMemoryMeetParticipantRepository")

1343 .finish_non_exhaustive()

1344 }

1345}

1346

10121347pub struct InMemoryChatStorage {

10131348 state: SharedState,

10141349 blobs: Arc<dyn ChatMediaStore>,

@@ -1107,6 +1442,12 @@

11071442 fn threads(&self) -> DynThreadRepository {

11081443 Arc::new(InMemoryThreadRepository::new(self.state.clone()))

11091444 }

1445 fn meets(&self) -> DynMeetRepository {

1446 Arc::new(InMemoryMeetRepository::new(self.state.clone()))

1447 }

1448 fn meet_participants(&self) -> DynMeetParticipantRepository {

1449 Arc::new(InMemoryMeetParticipantRepository::new(self.state.clone()))

1450 }

11101451 fn blobs(&self) -> Arc<dyn ChatMediaStore> {

11111452 self.blobs.clone()

11121453 }

Modified chat/storagelayer/models/conversation.rs +2 −2 View chat/storagelayer/models/conversation.rs

@@ -24,8 +24,8 @@

2424

2525#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]

2626pub struct NewConversation {

27 pub space_id: Uuid,

28 pub parent_dir: Option<Uuid>,

27 pub space_id: Uuid,

28 pub parent_dir: Option<Uuid>,

2929 pub kind: ConversationKind,

3030 pub counterpart_user_id: Option<Uuid>,

3131 pub title: Option<String>,

Modified chat/storagelayer/repository/attachment.rs +1 −1 View chat/storagelayer/repository/attachment.rs

@@ -32,7 +32,7 @@

3232

3333 async fn delete(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()>;

3434

35 async fn attach_to_message(

35 async fn attach_to_message(

3636 &self,

3737 ctx: &ChatStorageCtx,

3838 id: Uuid,

Modified chat/storagelayer/repository/conversation.rs +6 −0 View chat/storagelayer/repository/conversation.rs

@@ -45,6 +45,12 @@

4545 async fn delete(&self, ctx: &ChatStorageCtx, id: uuid::Uuid) -> ChatStorageResult<()>;

4646

4747 async fn exists(&self, ctx: &ChatStorageCtx, id: uuid::Uuid) -> ChatStorageResult<bool>;

48

49 async fn list_member_user_ids(

50 &self,

51 ctx: &ChatStorageCtx,

52 id: uuid::Uuid,

53 ) -> ChatStorageResult<Vec<uuid::Uuid>>;

4854}

4955

5056pub type DynConversationRepository = Arc<dyn ConversationRepository>;

Modified chat/storagelayer/repository/read_receipt.rs +1 −1 View chat/storagelayer/repository/read_receipt.rs

@@ -29,7 +29,7 @@

2929 thread_id: Option<Uuid>,

3030 ) -> ChatStorageResult<Option<ReadReceipt>>;

3131

32 async fn set_muted(

32 async fn set_muted(

3333 &self,

3434 ctx: &ChatStorageCtx,

3535 chat_id: Uuid,

Modified chat/storagelayer/auth.rs +45 −0 View chat/storagelayer/auth.rs

@@ -50,6 +50,51 @@

5050 ))

5151}

5252

53async fn space_id_for_channel<C: ConnectionTrait>(

54 db: &C,

55 channel_id: Uuid,

56) -> ChatStorageResult<Option<Uuid>> {

57 let Some(channel) = channel_entity::Entity::find_by_id(channel_id)

58 .one(db)

59 .await?

60 else {

61 return Ok(None);

62 };

63 Ok(channel.space_id)

64}

65

66fn channel_not_in_space(channel_id: Uuid) -> ChatStorageError {

67 ChatStorageError::PermissionDenied(format!(

68 "channel {channel_id} is not bound to a space (channel or space missing)"

69 ))

70}

71

72pub async fn require_channel_member(

73 db: &db::conn::Db,

74 ctx: &ChatStorageCtx,

75 channel_id: Uuid,

76) -> ChatStorageResult<Uuid> {

77 let user_id = require_authenticated(ctx).await?;

78 let space_id = space_id_for_channel(&db.reader(), channel_id)

79 .await?

80 .ok_or_else(|| channel_not_in_space(channel_id))?;

81 require_active_member(&db.reader(), space_id, user_id).await?;

82 Ok(user_id)

83}

84

85pub async fn require_channel_admin(

86 db: &db::conn::Db,

87 ctx: &ChatStorageCtx,

88 channel_id: Uuid,

89) -> ChatStorageResult<Uuid> {

90 let user_id = require_authenticated(ctx).await?;

91 let space_id = space_id_for_channel(&db.reader(), channel_id)

92 .await?

93 .ok_or_else(|| channel_not_in_space(channel_id))?;

94 require_admin_in_space(&db.reader(), space_id, user_id).await?;

95 Ok(user_id)

96}

97

5398async fn require_active_member<C: ConnectionTrait>(

5499 db: &C,

55100 space_id: Uuid,

Modified chat/storagelayer/error.rs +9 −0 View chat/storagelayer/error.rs

@@ -33,6 +33,15 @@

3333 #[error("thread not found: chat={0} root={1}")]

3434 ThreadNotFound(Uuid, Uuid),

3535

36 #[error("meet not found: {0}")]

37 MeetNotFound(Uuid),

38

39 #[error("no active meet for channel: {0}")]

40 NoActiveMeet(Uuid),

41

42 #[error("meet participant not found: {0}")]

43 MeetParticipantNotFound(Uuid),

44

3645 #[error("invalid input: {0}")]

3746 InvalidInput(String),

3847

Modified chat/storagelayer/lib.rs +6 −3 View chat/storagelayer/lib.rs

@@ -21,6 +21,8 @@

2121pub use models::{

2222 attachment::{Attachment, AttachmentOwner, NewAttachment},

2323 conversation::{Conversation, ConversationKind, NewConversation},

24 meet::{Meet, MeetStatus, NewMeet},

25 meet_participant::{LeaveReason, MeetParticipant, NewMeetParticipant},

2426 mention::{Mention, NewMention},

2527 message::{

2628 ListDirection, MAX_PAGE_LIMIT, Message, MessageFlags, MessagePage, MessageQuery,

@@ -33,9 +35,10 @@

3335};

3436pub use repository::{

3537 ChatStorage, ChatStorageCtx, ChatStorageFactory, DynChatStorage,

36 attachment::AttachmentRepository, conversation::ConversationRepository,

37 mention::MentionRepository, message::MessageRepository, pinned::PinnedMessageRepository,

38 reaction::ReactionRepository, read_receipt::ReadReceiptRepository, thread::ThreadRepository,

38 attachment::AttachmentRepository, conversation::ConversationRepository, meet::MeetRepository,

39 meet_participant::MeetParticipantRepository, mention::MentionRepository,

40 message::MessageRepository, pinned::PinnedMessageRepository, reaction::ReactionRepository,

41 read_receipt::ReadReceiptRepository, thread::ThreadRepository,

3942};

4043

4144#[cfg(feature = "mock")]

Modified chat/storagelayer/mapper.rs +14 −0 View chat/storagelayer/mapper.rs

@@ -1,5 +1,7 @@

11pub mod attachment;

22pub mod conversation;

3pub mod meet;

4pub mod meet_participant;

35pub mod mention;

46pub mod message;

57pub mod pinned;

@@ -14,6 +16,8 @@

1416

1517use self::attachment::SeaAttachmentRepository;

1618use self::conversation::SeaConversationRepository;

19use self::meet::SeaMeetRepository;

20use self::meet_participant::SeaMeetParticipantRepository;

1721use self::mention::SeaMentionRepository;

1822use self::message::SeaMessageRepository;

1923use self::pinned::SeaPinnedMessageRepository;

@@ -65,6 +69,16 @@

6569 Arc::new(SeaThreadRepository::new(self.db.clone()))

6670 }

6771

72 fn meets(&self) -> crate::repository::meet::DynMeetRepository {

73 Arc::new(SeaMeetRepository::new(self.db.clone()))

74 }

75

76 fn meet_participants(

77 &self,

78 ) -> crate::repository::meet_participant::DynMeetParticipantRepository {

79 Arc::new(SeaMeetParticipantRepository::new(self.db.clone()))

80 }

81

6882 fn blobs(&self) -> Arc<dyn ChatMediaStore> {

6983 self.blobs.clone()

7084 }

Modified chat/storagelayer/mapping.rs +111 −0 View chat/storagelayer/mapping.rs

@@ -10,11 +10,15 @@

1010use db::chat::chat_reads as read_entity;

1111use db::chat::chats as chat_entity;

1212use db::chat::threads as thread_entity;

13use db::live::meet_participants as meet_participant_entity;

14use db::live::meets as meet_entity;

1315use sea_orm::Set;

1416

1517use crate::error::{ChatStorageError, ChatStorageResult};

1618use crate::models::attachment::{Attachment, AttachmentOwner};

1719use crate::models::conversation::{Conversation, ConversationKind};

20use crate::models::meet::{Meet, MeetStatus, NewMeet};

21use crate::models::meet_participant::{LeaveReason, MeetParticipant, NewMeetParticipant};

1822use crate::models::mention::{Mention, NewMention};

1923use crate::models::message::{Message, NewMessage};

2024use crate::models::pinned::{NewPinnedMessage, PinnedMessage};

@@ -286,3 +290,110 @@

286290 m.edited_at,

287291 )

288292}

293

294impl From<meet_entity::MeetStatus> for MeetStatus {

295 fn from(status: meet_entity::MeetStatus) -> Self {

296 match status {

297 meet_entity::MeetStatus::Pending => Self::Pending,

298 meet_entity::MeetStatus::Active => Self::Active,

299 meet_entity::MeetStatus::Ended => Self::Ended,

300 }

301 }

302}

303

304impl From<MeetStatus> for meet_entity::MeetStatus {

305 fn from(status: MeetStatus) -> Self {

306 match status {

307 MeetStatus::Pending => Self::Pending,

308 MeetStatus::Active => Self::Active,

309 MeetStatus::Ended => Self::Ended,

310 }

311 }

312}

313

314impl From<meet_entity::Model> for Meet {

315 fn from(m: meet_entity::Model) -> Self {

316 Self {

317 id: m.id,

318 channel_id: m.channel_id,

319 room_name: m.room_name,

320 status: m.status.into(),

321 created_by: m.created_by,

322 started_at: m.started_at,

323 ended_at: m.ended_at,

324 created_at: m.created_at,

325 updated_at: m.updated_at,

326 }

327 }

328}

329

330impl TryFrom<NewMeet> for meet_entity::ActiveModel {

331 type Error = ChatStorageError;

332 fn try_from(m: NewMeet) -> ChatStorageResult<Self> {

333 Ok(Self {

334 id: Set(Uuid::now_v7()),

335 channel_id: Set(m.channel_id),

336 room_name: Set(m.room_name),

337 status: Set(meet_entity::MeetStatus::Pending),

338 created_by: Set(m.created_by),

339 started_at: Set(None),

340 ended_at: Set(None),

341 created_at: Set(OffsetDateTime::now_utc()),

342 updated_at: Set(None),

343 })

344 }

345}

346

347impl From<meet_participant_entity::LeaveReason> for LeaveReason {

348 fn from(reason: meet_participant_entity::LeaveReason) -> Self {

349 match reason {

350 meet_participant_entity::LeaveReason::NotLeft => Self::NotLeft,

351 meet_participant_entity::LeaveReason::Left => Self::Left,

352 meet_participant_entity::LeaveReason::Removed => Self::Removed,

353 meet_participant_entity::LeaveReason::RoomEnded => Self::RoomEnded,

354 meet_participant_entity::LeaveReason::NetworkDrop => Self::NetworkDrop,

355 meet_participant_entity::LeaveReason::Kicked => Self::Kicked,

356 }

357 }

358}

359

360impl From<LeaveReason> for meet_participant_entity::LeaveReason {

361 fn from(reason: LeaveReason) -> Self {

362 match reason {

363 LeaveReason::NotLeft => Self::NotLeft,

364 LeaveReason::Left => Self::Left,

365 LeaveReason::Removed => Self::Removed,

366 LeaveReason::RoomEnded => Self::RoomEnded,

367 LeaveReason::NetworkDrop => Self::NetworkDrop,

368 LeaveReason::Kicked => Self::Kicked,

369 }

370 }

371}

372

373impl From<meet_participant_entity::Model> for MeetParticipant {

374 fn from(m: meet_participant_entity::Model) -> Self {

375 Self {

376 id: m.id,

377 meet_id: m.meet_id,

378 user_id: m.user_id,

379 joined_at: m.joined_at,

380 left_at: m.left_at,

381 leave_reason: m.leave_reason.map(Into::into),

382 }

383 }

384}

385

386impl TryFrom<NewMeetParticipant> for meet_participant_entity::ActiveModel {

387 type Error = ChatStorageError;

388 fn try_from(p: NewMeetParticipant) -> ChatStorageResult<Self> {

389 Ok(Self {

390 id: Set(Uuid::now_v7()),

391 meet_id: Set(p.meet_id),

392 user_id: Set(p.user_id),

393 joined_at: Set(p.joined_at),

394 left_at: Set(None),

395 leave_reason: Set(None),

396 connection_id: Set(p.connection_id),

397 })

398 }

399}

Modified chat/storagelayer/models.rs +2 −0 View chat/storagelayer/models.rs

@@ -1,5 +1,7 @@

11pub mod attachment;

22pub mod conversation;

3pub mod meet;

4pub mod meet_participant;

35pub mod mention;

46pub mod message;

57pub mod pinned;

Modified chat/storagelayer/repository.rs +4 −0 View chat/storagelayer/repository.rs

@@ -4,6 +4,8 @@

44

55pub mod attachment;

66pub mod conversation;

7pub mod meet;

8pub mod meet_participant;

79pub mod mention;

810pub mod message;

911pub mod pinned;

@@ -59,6 +61,8 @@

5961 fn reads(&self) -> read_receipt::DynReadReceiptRepository;

6062 fn pinned(&self) -> pinned::DynPinnedMessageRepository;

6163 fn threads(&self) -> thread::DynThreadRepository;

64 fn meets(&self) -> meet::DynMeetRepository;

65 fn meet_participants(&self) -> meet_participant::DynMeetParticipantRepository;

6266 fn blobs(&self) -> Arc<dyn crate::blob::ChatMediaStore>;

6367}

6468