refactor: refresh chat stack (chatlayer, glayer, storagelayer, live) and lib/space

Modified chat/chatlayer/actor/runtime.rs +24 −8 View chat/chatlayer/actor/runtime.rs

@@ -25,9 +25,14 @@

2525 + Send

2626 + Sync;

2727

28struct ActorEntry<M> {

29 tx: mpsc::UnboundedSender<M>,

30 generation: u64,

31}

32

2833pub struct ActorRegistry<M> {

2934 name: &'static str,

30 actors: Mutex<HashMap<Uuid, mpsc::UnboundedSender<M>>>,

35 actors: Mutex<HashMap<Uuid, ActorEntry<M>>>,

3136 spawn: Arc<SpawnFn<M>>,

3237 shutdown: CancellationToken,

3338}

@@ -71,8 +76,8 @@

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

7277 {

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

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

75 match tx.send(msg) {

79 if let Some(entry) = guard.get(&key) {

80 match entry.tx.send(msg) {

7681 Ok(()) => return,

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

7883 }

@@ -80,22 +85,33 @@

8085 }

8186 let (tx, rx) = mpsc::unbounded_channel();

8287 let mailbox = Mailbox { tx: tx.clone() };

83 {

88 let generation = {

8489 let mut guard = self.actors.lock().unwrap();

8590 if let Some(existing) = guard.get(&key) {

86 match existing.send(msg) {

91 match existing.tx.send(msg) {

8792 Ok(()) => return,

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

8994 }

9095 }

91 guard.insert(key, tx);

92 }

96 let generation = guard

97 .get(&key)

98 .map(|entry| entry.generation.wrapping_add(1))

99 .unwrap_or(0);

100 guard.insert(key, ActorEntry { tx, generation });

101 generation

102 };

93103 let registry = self.clone();

94104 let spawn = self.spawn.clone();

95105 tokio::spawn(async move {

96106 let fut = (spawn)(registry.clone(), key, rx);

97107 fut.await;

98 registry.actors.lock().unwrap().remove(&key);

108 let mut guard = registry.actors.lock().unwrap();

109 if guard

110 .get(&key)

111 .is_some_and(|entry| entry.generation == generation)

112 {

113 guard.remove(&key);

114 }

99115 });

100116 let _ = mailbox.send(msg);

101117 crate::metrics::actor_spawned(self.name);

Modified chat/chatlayer/events/room.rs +12 −0 View chat/chatlayer/events/room.rs

@@ -1,4 +1,6 @@

11use serde::{Deserialize, Serialize};

2use storagelayer::models::conversation::ConversationKind;

3use time::OffsetDateTime;

24use uuid::Uuid;

35

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

@@ -48,6 +50,13 @@

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

4951#[serde(tag = "type", rename_all = "snake_case")]

5052pub enum ChatRoomEvent {

53 ConversationCreated {

54 conversation_id: Uuid,

55 kind: ConversationKind,

56 title: Option<String>,

57 creator_user_id: Uuid,

58 ts: OffsetDateTime,

59 },

5160 MessageCreated(MessageCommand),

5261 MessageEdited {

5362 conversation_id: Uuid,

@@ -88,6 +97,9 @@

8897impl ChatRoomEvent {

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

9099 match self {

100 Self::ConversationCreated {

101 conversation_id, ..

102 } => *conversation_id,

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

92104 Self::MessageEdited {

93105 conversation_id, ..

Modified chat/chatlayer/events/user.rs +11 −1 View chat/chatlayer/events/user.rs

@@ -1,10 +1,19 @@

11use serde::{Deserialize, Serialize};

2use storagelayer::models::conversation::ConversationKind;

23use time::OffsetDateTime;

34use uuid::Uuid;

45

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

67#[serde(tag = "type", rename_all = "snake_case")]

78pub enum ChatUserEvent {

9 ConversationCreated {

10 user_id: Uuid,

11 conversation_id: Uuid,

12 kind: ConversationKind,

13 title: Option<String>,

14 creator_user_id: Uuid,

15 ts: OffsetDateTime,

16 },

817 MessageDelivered {

918 user_id: Uuid,

1019 conversation_id: Uuid,

@@ -34,7 +43,8 @@

3443impl ChatUserEvent {

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

3645 match self {

37 Self::MessageDelivered { user_id, .. }

46 Self::ConversationCreated { user_id, .. }

47 | Self::MessageDelivered { user_id, .. }

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

3949 | Self::ReadReceiptAck { user_id, .. }

4050 | Self::PresenceHeartbeat { user_id, .. } => *user_id,

Modified chat/chatlayer/events/user_view.rs +11 −1 View chat/chatlayer/events/user_view.rs

@@ -1,10 +1,19 @@

11use serde::{Deserialize, Serialize};

2use storagelayer::models::conversation::ConversationKind;

23use time::OffsetDateTime;

34use uuid::Uuid;

45

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

67#[serde(tag = "type", rename_all = "snake_case")]

78pub enum ChatUserViewEvent {

9 ConversationCreated {

10 user_id: Uuid,

11 conversation_id: Uuid,

12 kind: ConversationKind,

13 title: Option<String>,

14 creator_user_id: Uuid,

15 ts: OffsetDateTime,

16 },

817 NewMessage {

918 user_id: Uuid,

1019 conversation_id: Uuid,

@@ -42,7 +51,8 @@

4251impl ChatUserViewEvent {

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

4453 match self {

45 Self::NewMessage { user_id, .. }

54 Self::ConversationCreated { user_id, .. }

55 | Self::NewMessage { user_id, .. }

4656 | Self::MessageEditedView { user_id, .. }

4757 | Self::MessageDeletedView { user_id, .. }

4858 | Self::UnreadChanged { user_id, .. }

Modified chat/chatlayer/projection/live.rs +34 −11 View chat/chatlayer/projection/live.rs

@@ -62,6 +62,10 @@

6262 }

6363}

6464

65fn effective_actor(actor_id: Uuid, subject: Uuid) -> Uuid {

66 if actor_id.is_nil() { subject } else { actor_id }

67}

68

6569async fn apply_meet_started(

6670 store: &DynChatStorage,

6771 actor_id: Uuid,

@@ -71,27 +75,44 @@

7175 ts: time::OffsetDateTime,

7276) -> ChatLayerResult<Vec<ChatLiveViewEvent>> {

7377 let ctx = ChatStorageCtx::new(actor_id);

74 let meet = match store.meets().find_by_room_name(&ctx, room_name).await? {

75 Some(meet) => store.meets().mark_active(&ctx, meet.id).await?,

78 let (meet, operator) = match store.meets().find_by_room_name(&ctx, room_name).await? {

79 Some(meet) => {

80 let creator = effective_actor(actor_id, meet.created_by);

81 let creator_ctx = ChatStorageCtx::new(creator);

82 let meet = store.meets().mark_active(&creator_ctx, meet.id).await?;

83 (meet, creator)

84 }

7685 None => {

86 if by.is_nil() {

87 tracing::warn!(

88 channel_id = %channel_id,

89 room_name,

90 "webhook meet start without a known creator; no meet row to activate"

91 );

92 return Ok(Vec::new());

93 }

94 let creator = effective_actor(actor_id, by);

95 let creator_ctx = ChatStorageCtx::new(creator);

7796 let created = store

7897 .meets()

7998 .create(

80 &ctx,

99 &creator_ctx,

81100 NewMeet {

82101 channel_id,

83102 room_name: room_name.to_string(),

84 created_by: by,

103 created_by: creator,

85104 },

86105 )

87106 .await?;

88 store.meets().mark_active(&ctx, created.id).await?

107 let meet = store.meets().mark_active(&creator_ctx, created.id).await?;

108 (meet, creator)

89109 }

90110 };

91111

112 let op_ctx = ChatStorageCtx::new(operator);

92113 let members = store

93114 .meets()

94 .list_channel_member_user_ids(&ctx, channel_id)

115 .list_channel_member_user_ids(&op_ctx, channel_id)

95116 .await?;

96117 Ok(members

97118 .into_iter()

@@ -117,15 +138,17 @@

117138 let Some(meet) = store.meets().find_by_room_name(&ctx, room_name).await? else {

118139 return Ok(Vec::new());

119140 };

120 store.meets().mark_ended(&ctx, meet.id).await?;

141 let operator = effective_actor(actor_id, meet.created_by);

142 let op_ctx = ChatStorageCtx::new(operator);

143 store.meets().mark_ended(&op_ctx, meet.id).await?;

121144 store

122145 .meet_participants()

123 .mark_all_ended(&ctx, meet.id)

146 .mark_all_ended(&op_ctx, meet.id)

124147 .await?;

125148

126149 let members = store

127150 .meets()

128 .list_channel_member_user_ids(&ctx, channel_id)

151 .list_channel_member_user_ids(&op_ctx, channel_id)

129152 .await?;

130153 Ok(members

131154 .into_iter()

@@ -146,7 +169,7 @@

146169 user_id: Uuid,

147170 ts: time::OffsetDateTime,

148171) -> ChatLayerResult<Vec<ChatLiveViewEvent>> {

149 let ctx = ChatStorageCtx::new(actor_id);

172 let ctx = ChatStorageCtx::new(effective_actor(actor_id, user_id));

150173 let Some(meet) = store.meets().find_by_room_name(&ctx, room_name).await? else {

151174 return Ok(Vec::new());

152175 };

@@ -185,7 +208,7 @@

185208 reason: storagelayer::LeaveReason,

186209 ts: time::OffsetDateTime,

187210) -> ChatLayerResult<Vec<ChatLiveViewEvent>> {

188 let ctx = ChatStorageCtx::new(actor_id);

211 let ctx = ChatStorageCtx::new(effective_actor(actor_id, user_id));

189212 let Some(meet) = store.meets().find_by_room_name(&ctx, room_name).await? else {

190213 return Ok(Vec::new());

191214 };

Modified chat/chatlayer/projection/room.rs +35 −2 View chat/chatlayer/projection/room.rs

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

1use std::collections::HashSet;

12use std::sync::Arc;

23

34use storagelayer::DynChatStorage;

@@ -23,6 +24,30 @@

2324) -> ChatLayerResult<Vec<ChatUserEvent>> {

2425 let ctx = ChatStorageCtx::new(envelope.actor_id);

2526 match &envelope.payload {

27 ChatRoomEvent::ConversationCreated {

28 conversation_id,

29 kind,

30 title,

31 creator_user_id,

32 ts,

33 } => {

34 let members = store

35 .conversations()

36 .list_member_user_ids(&ctx, *conversation_id)

37 .await?;

38 let out = members

39 .into_iter()

40 .map(|user_id| ChatUserEvent::ConversationCreated {

41 user_id,

42 conversation_id: *conversation_id,

43 kind: kind.clone(),

44 title: title.clone(),

45 creator_user_id: *creator_user_id,

46 ts: *ts,

47 })

48 .collect();

49 Ok(out)

50 }

2651 ChatRoomEvent::MessageCreated(cmd) => {

2752 apply_message_created(store, idempotency, media, &ctx, envelope.actor_id, cmd).await

2853 }

@@ -158,9 +183,17 @@

158183 }

159184

160185 let mut out = Vec::new();

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

186 let mentioned: HashSet<Uuid> = cmd.mentions.iter().filter_map(mention_user).collect();

187 let members = store

188 .conversations()

189 .list_member_user_ids(ctx, cmd.conversation_id)

190 .await?;

191 for member in members {

192 if member == cmd.sender_id || mentioned.contains(&member) {

193 continue;

194 }

162195 out.push(ChatUserEvent::MessageDelivered {

163 user_id: cmd.sender_id,

196 user_id: member,

164197 conversation_id: cmd.conversation_id,

165198 message_id: message.id,

166199 ts: message.created_at,

Modified chat/chatlayer/projection/user.rs +15 −0 View chat/chatlayer/projection/user.rs

@@ -37,6 +37,21 @@

3737 envelope: &EventEnvelope<ChatUserEvent>,

3838) -> ChatLayerResult<Vec<ChatUserViewEvent>> {

3939 match &envelope.payload {

40 ChatUserEvent::ConversationCreated {

41 user_id,

42 conversation_id,

43 kind,

44 title,

45 creator_user_id,

46 ts,

47 } => Ok(vec![ChatUserViewEvent::ConversationCreated {

48 user_id: *user_id,

49 conversation_id: *conversation_id,

50 kind: kind.clone(),

51 title: title.clone(),

52 creator_user_id: *creator_user_id,

53 ts: *ts,

54 }]),

4055 ChatUserEvent::MessageDelivered {

4156 user_id,

4257 conversation_id,

Modified chat/chatlayer/auth_bridge.rs +5 −10 View chat/chatlayer/auth_bridge.rs

@@ -64,10 +64,10 @@

6464 self.jwt.settings.clock_skew_seconds,

6565 )?;

6666

67 if let Some(blacklist) = &self.blacklist {

68 if blacklist.is_revoked(claims.jti, claims.sub).await? {

69 return Err(ChatLayerError::Jwt(JwtError::SessionRevoked));

70 }

67 if let Some(blacklist) = &self.blacklist

68 && blacklist.is_revoked(claims.jti, claims.sub).await?

69 {

70 return Err(ChatLayerError::Jwt(JwtError::SessionRevoked));

7171 }

7272 if let Some(db) = &self.db {

7373 let conn = db.reader();

@@ -96,11 +96,6 @@

9696 }

9797

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

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

100 ChatLayerError::Jwt(_) => {

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

102 }

103 other => other,

104 })

99 self.verify(token).await

105100 }

106101}

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

@@ -175,13 +175,13 @@

175175 let attachment_id = Uuid::now_v7();

176176 let path = AttachmentPath::new(owner, Uuid::nil(), filename);

177177 let key = path.as_key();

178 let object_path = Self::object_path(&key);

178 let object_path = Self::object_path(key);

179179 let upload_url = self.storage.presigned_upload(&object_path, ttl).await?;

180180

181181 let draft = UploadDraft {

182182 sender_id: identity.user_id,

183183 client_msg_id: owner_hint.map(|h| h.to_string()),

184 key: key.clone(),

184 key: key.to_string(),

185185 filename: filename.to_string(),

186186 content_type: content_type.to_string(),

187187 size: size_hint.unwrap_or(0),

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

215215 Ok(PresignedUpload {

216216 attachment_id,

217217 upload_url,

218 key,

218 key: key.to_string(),

219219 filename: filename.to_string(),

220220 content_type: content_type.to_string(),

221221 size_hint,

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

252252 )));

253253 }

254254

255 draft.size = observed_size;

255 draft.size = meta.size as i64;

256256 self.kv

257257 .set_bytes(

258258 &Self::draft_key(attachment_id),

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

267267 key: draft.key.clone(),

268268 filename: draft.filename.clone(),

269269 content_type: draft.content_type.clone(),

270 size: observed_size,

270 size: meta.size as i64,

271271 })

272272 }

273273

Modified chat/chatlayer/service.rs +64 −3 View chat/chatlayer/service.rs

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

33use cache::redis::RedisCache;

44use serde::{Deserialize, Serialize};

55use storagelayer::DynChatStorage;

6use storagelayer::models::conversation::Conversation;

7use storagelayer::models::message::{MessagePage, MessageQuery};

6use storagelayer::models::conversation::{Conversation, NewConversation};

7use storagelayer::models::message::{MessagePage, MessageQuery, Page};

88use storagelayer::repository::ChatStorageCtx;

99use tokio_util::sync::CancellationToken;

1010use uuid::Uuid;

@@ -201,7 +201,16 @@

201201 server_msg_id: Some(id),

202202 });

203203 }

204 ClientMessageClaim::Fresh | ClientMessageClaim::InFlight => {}

204 // A request is already being processed for this idempotency key;

205 // publishing again would only add a duplicate envelope (the

206 // projection dedupe would discard it). Ack without publishing.

207 ClientMessageClaim::InFlight => {

208 return Ok(AcceptedMsg {

209 client_msg_id: cmd.client_msg_id,

210 server_msg_id: None,

211 });

212 }

213 ClientMessageClaim::Fresh => {}

205214 }

206215

207216 let sanitized_body = sanitize_message_body(&cmd.body);

@@ -440,6 +449,58 @@

440449 Ok(self.store.conversations().find_by_id(&ctx, id).await?)

441450 }

442451

452 pub async fn create_conversation(

453 &self,

454 auth: &AuthBridge,

455 token: &str,

456 cmd: NewConversation,

457 ) -> ChatLayerResult<Conversation> {

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

459 if cmd.counterpart_user_id == Some(identity.user_id) {

460 return Err(ChatLayerError::invalid(

461 "counterpart_user_id cannot match the creator",

462 ));

463 }

464 if let Some(t) = cmd.title.as_deref() {

465 let trimmed = t.trim();

466 if trimmed.is_empty() {

467 return Err(ChatLayerError::invalid(

468 "conversation title cannot be empty",

469 ));

470 }

471 if trimmed.len() > 128 {

472 return Err(ChatLayerError::invalid(

473 "conversation title is too long (max 128 bytes)",

474 ));

475 }

476 }

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

478 let conversation = self.store.conversations().create(&ctx, cmd).await?;

479 self.producer

480 .publish_room(&EventEnvelope::new(

481 identity.user_id,

482 ChatRoomEvent::ConversationCreated {

483 conversation_id: conversation.id,

484 kind: conversation.kind,

485 title: conversation.title.clone(),

486 creator_user_id: conversation.creator_user_id,

487 ts: conversation.created_at,

488 },

489 ))

490 .await?;

491 Ok(conversation)

492 }

493

494 pub async fn list_conversations(

495 &self,

496 auth: &AuthBridge,

497 token: &str,

498 page: Page,

499 ) -> ChatLayerResult<Vec<Conversation>> {

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

501 Ok(self.store.conversations().list_for_user(&ctx, page).await?)

502 }

503

443504 pub async fn lookup_by_client_msg_id(

444505 &self,

445506 auth: &AuthBridge,

Modified chat/glayer/handler/conversations.rs +40 −3 View chat/glayer/handler/conversations.rs

@@ -1,12 +1,49 @@

1use axum::extract::{Path, State};

2use export::data::AppResponseData;

3use storagelayer::models::conversation::Conversation;

1use axum::extract::{Path, Query, State};

2use export::data::{AppRequest, AppResponseData};

3use serde::Deserialize;

4use storagelayer::models::conversation::{Conversation, NewConversation};

5use storagelayer::Page;

46use uuid::Uuid;

57

68use crate::auth::BearToken;

79use crate::error::{GatewayError, GatewayResult};

810use crate::state::GatewayState;

911

12#[derive(Debug, Default, Deserialize)]

13#[serde(default)]

14pub(crate) struct ListConversationsParams {

15 pub limit: Option<u32>,

16 pub cursor: Option<String>,

17}

18

19pub(crate) async fn create_conversation(

20 State(state): State<GatewayState>,

21 BearToken(token): BearToken,

22 AppRequest(cmd): AppRequest<NewConversation>,

23) -> GatewayResult<AppResponseData<Conversation>> {

24 let conversation = state

25 .chat

26 .create_conversation(state.chat.auth(), &token, cmd)

27 .await?;

28 Ok(AppResponseData::ok(conversation).with_code(201))

29}

30

31pub(crate) async fn list_conversations(

32 State(state): State<GatewayState>,

33 BearToken(token): BearToken,

34 Query(params): Query<ListConversationsParams>,

35) -> GatewayResult<AppResponseData<Vec<Conversation>>> {

36 let page = Page {

37 limit: params.limit.unwrap_or(50),

38 cursor: params.cursor,

39 };

40 let list = state

41 .chat

42 .list_conversations(state.chat.auth(), &token, page)

43 .await?;

44 Ok(AppResponseData::ok(list))

45}

46

1047pub(crate) async fn get_conversation(

1148 State(state): State<GatewayState>,

1249 BearToken(token): BearToken,

Modified chat/glayer/auth.rs +5 −0 View chat/glayer/auth.rs

@@ -27,6 +27,11 @@

2727 let token = it

2828 .next()

2929 .ok_or_else(|| GatewayError::Unauthorized("missing bearer token".into()))?;

30 if it.next().is_some() {

31 return Err(GatewayError::Unauthorized(

32 "malformed authorization header".into(),

33 ));

34 }

3035 Ok(BearToken(token.to_string()))

3136 }

3237}

Modified chat/glayer/router.rs +5 −0 View chat/glayer/router.rs

@@ -54,6 +54,11 @@

5454 post(handler::read::mark_read),

5555 )

5656 .route(

57 "/conversations",

58 get(handler::conversations::list_conversations)

59 .post(handler::conversations::create_conversation),

60 )

61 .route(

5762 "/conversations/{conversation_id}",

5863 get(handler::conversations::get_conversation),

5964 )

Modified chat/glayer/state.rs +6 −3 View chat/glayer/state.rs

@@ -94,15 +94,18 @@

9494 .map_err(GatewayError::from)

9595 }

9696

97 pub async fn publish_typing(&self, signal: &TypingSignal) -> GatewayResult<()> {

97 /// Publishes the signal to the shared typing topic. Returns `true` when

98 /// the signal was published (delivery via the TypingConsumer), or

99 /// `false` when the queue is disabled and a local broadcast is required.

100 pub async fn publish_typing(&self, signal: &TypingSignal) -> GatewayResult<bool> {

98101 let Some(queue) = self.chat.queue() else {

99 return Ok(());

102 return Ok(false);

100103 };

101104 let topic = typing_topic(&self.settings.topic_prefix);

102105 let payload = serde_json::to_vec(signal)?;

103106 queue

104107 .publish_raw(&topic, Some(signal.conversation_id.as_bytes()), &payload)

105108 .await?;

106 Ok(())

109 Ok(true)

107110 }

108111}

Modified chat/glayer/ws.rs +11 −3 View chat/glayer/ws.rs

@@ -159,9 +159,17 @@

159159 user_id,

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

161161 };

162 broadcast_typing(&store, &state.hub, &signal).await;

163 if let Err(e) = state.publish_typing(&signal).await {

164 tracing::warn!(error = %e, "failed to publish typing signal");

162 // Route through the shared typing topic when a queue is available; the

163 // TypingConsumer broadcasts to members of every node (including this

164 // one). Fall back to a local broadcast only when the queue is disabled,

165 // avoiding a duplicate delivery on the source node.

166 match state.publish_typing(&signal).await {

167 Ok(true) => {}

168 Ok(false) => broadcast_typing(&store, &state.hub, &signal).await,

169 Err(e) => {

170 tracing::warn!(error = %e, "failed to publish typing signal");

171 broadcast_typing(&store, &state.hub, &signal).await;

172 }

165173 }

166174}

167175

Modified chat/live/service.rs +33 −8 View chat/live/service.rs

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

22

33use axum::http::HeaderMap;

44use cache::redis::RedisCache;

5use config::rate_limit::RateLimitConfig;

56use jwt::{HasJwtState, JwtState};

67use storagelayer::{ChatStorageCtx, ChatStorageError, DynChatStorage};

78use uuid::Uuid;

@@ -19,6 +20,7 @@

1920 pub(crate) producer: Arc<chatlayer::ChatProducer>,

2021 pub(crate) client: LiveKitClient,

2122 pub(crate) settings: LiveSettings,

23 pub(crate) rate_limit: Arc<RateLimitConfig>,

2224 pub(crate) kv: Arc<RedisCache>,

2325 pub(crate) jwt: JwtState,

2426}

@@ -39,6 +41,7 @@

3941 kv: Arc<RedisCache>,

4042 jwt: JwtState,

4143 settings: LiveSettings,

44 rate_limit: Arc<RateLimitConfig>,

4245 ) -> Self {

4346 let client = LiveKitClient::new(

4447 settings.base_url.clone(),

@@ -50,6 +53,7 @@

5053 producer,

5154 client,

5255 settings,

56 rate_limit,

5357 kv,

5458 jwt,

5559 }

@@ -102,12 +106,33 @@

102106 created_by: user_id,

103107 },

104108 )

105 .await?;

106 self.client

107 .create_room(&room_name, self.settings.empty_timeout_seconds)

108 .await?;

109 let activated = self.store.meets().mark_active(&ctx, created.id).await?;

110 (activated, true)

109 .await;

110 match created {

111 Ok(created) => {

112 self.client

113 .create_room(&room_name, self.settings.empty_timeout_seconds)

114 .await?;

115 let activated = self.store.meets().mark_active(&ctx, created.id).await?;

116 (activated, true)

117 }

118 Err(_race) => {

119 // A concurrent first join may have won the race; retry

120 // the lookup before giving up.

121 match self

122 .store

123 .meets()

124 .find_active_by_channel(&ctx, channel_id)

125 .await?

126 {

127 Some(meet) => (meet, false),

128 None => {

129 return Err(LiveError::Conflict(format!(

130 "failed to create meet for channel {channel_id}"

131 )));

132 }

133 }

134 }

135 }

111136 }

112137 };

113138

@@ -308,7 +333,7 @@

308333 if !self

309334 .kv

310335 .set_nx(

311 &format!("chat:live:webhook:{sub}", sub = claims.sub),

336 &format!("chat:live:webhook:{}", claims.dedup_id()),

312337 b"1".to_vec(),

313338 300,

314339 )

@@ -348,6 +373,6 @@

348373 }

349374

350375 fn rate_limit_settings(&self) -> Arc<config::rate_limit::RateLimitConfig> {

351 Arc::new(config::rate_limit::RateLimitConfig::default())

376 self.rate_limit.clone()

352377 }

353378}

Modified chat/live/webhook.rs +12 −0 View chat/live/webhook.rs

@@ -46,6 +46,9 @@

4646 #[serde(rename = "iss", default)]

4747 iss: Option<String>,

4848 #[allow(dead_code)]

49 #[serde(rename = "jti", default)]

50 jti: Option<String>,

51 #[allow(dead_code)]

4952 #[serde(rename = "nbf")]

5053 nbf: i64,

5154 #[allow(dead_code)]

@@ -55,6 +58,15 @@

5558 nh: String,

5659}

5760

61impl WebhookClaims {

62 /// Unique per-delivery identifier: prefers the JWT `jti` when the LiveKit

63 /// server is configured to emit one, otherwise falls back to the body

64 /// hash (`nh`), which is unique per event payload.

65 pub fn dedup_id(&self) -> &str {

66 self.jti.as_deref().unwrap_or(&self.nh)

67 }

68}

69

5870pub fn verify_webhook(

5971 headers: &HeaderMap,

6072 body: &[u8],

Modified chat/storagelayer/blob/media.rs +13 −10 View chat/storagelayer/blob/media.rs

@@ -13,26 +13,29 @@

1313 pub owner: AttachmentOwner,

1414 pub owner_id: Uuid,

1515 pub filename: String,

16 key: String,

1617}

1718

1819impl AttachmentPath {

1920 pub fn new(owner: AttachmentOwner, owner_id: Uuid, filename: impl Into<String>) -> Self {

21 let filename = filename.into();

22 let key = format!(

23 "{}/{}/{}-{}",

24 owner.as_str(),

25 owner_id,

26 Uuid::now_v7(),

27 sanitize_filename(&filename)

28 );

2029 Self {

2130 owner,

2231 owner_id,

23 filename: filename.into(),

32 filename,

33 key,

2434 }

2535 }

2636

27 pub fn as_key(&self) -> String {

28 let sanitized = sanitize_filename(&self.filename);

29 format!(

30 "{}/{}/{}-{}",

31 self.owner.as_str(),

32 self.owner_id,

33 Uuid::now_v7(),

34 sanitized

35 )

37 pub fn as_key(&self) -> &str {

38 &self.key

3639 }

3740}

3841

Modified chat/storagelayer/blob/object_store.rs +5 −9 View chat/storagelayer/blob/object_store.rs

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

6262 ) -> ChatStorageResult<AttachmentPath> {

6363 let attachment = AttachmentPath::new(owner, owner_id, filename);

6464 let key = attachment.as_key();

65 let path = self.object_path(&key);

65 let path = self.object_path(key);

6666 self.inner

6767 .put(&path, bytes)

6868 .await

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

7272

7373 async fn get_attachment_bytes(&self, path: &AttachmentPath) -> ChatStorageResult<Bytes> {

7474 use futures::TryStreamExt;

75 let key = path.as_key();

76 let op = self.object_path(&key);

75 let op = self.object_path(path.as_key());

7776 let result = self.inner.get(&op).await.map_err(ChatStorageError::from)?;

7877 let collected: Vec<Bytes> = result

7978 .into_stream()

@@ -93,8 +92,7 @@

9392 }

9493

9594 async fn delete_attachment(&self, path: &AttachmentPath) -> ChatStorageResult<()> {

96 let key = path.as_key();

97 let op = self.object_path(&key);

95 let op = self.object_path(path.as_key());

9896 self.inner.delete(&op).await.map_err(ChatStorageError::from)

9997 }

10098

@@ -106,8 +104,7 @@

106104 ttl: Duration,

107105 ) -> ChatStorageResult<Url> {

108106 let attachment = AttachmentPath::new(owner, owner_id, filename);

109 let key = attachment.as_key();

110 let op = self.object_path(&key);

107 let op = self.object_path(attachment.as_key());

111108 self.inner

112109 .presigned_upload(&op, ttl)

113110 .await

@@ -119,8 +116,7 @@

119116 path: &AttachmentPath,

120117 ttl: Duration,

121118 ) -> ChatStorageResult<Url> {

122 let key = path.as_key();

123 let op = self.object_path(&key);

119 let op = self.object_path(path.as_key());

124120 self.inner

125121 .presigned_download(&op, ttl)

126122 .await

Modified chat/storagelayer/mapper/attachment.rs +7 −7 View chat/storagelayer/mapper/attachment.rs

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

22

33use async_trait::async_trait;

44use sea_orm::{

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

6 QueryOrder, QuerySelect, Set,

5 ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter, QueryOrder,

6 QuerySelect, Set,

77};

88use time::OffsetDateTime;

99use uuid::Uuid;

@@ -130,13 +130,13 @@

130130 "attachment {id} is already attached to a message"

131131 )));

132132 }

133 let message_exists = message_entity::Entity::find_by_id(message_id)

134 .count(&db)

133 let Some(message) = message_entity::Entity::find_by_id(message_id)

134 .one(&db)

135135 .await?

136 > 0;

137 if !message_exists {

136 else {

138137 return Err(ChatStorageError::MessageNotFound(message_id));

139 }

138 };

139 auth::require_member(&self.db, ctx, message.chat_id).await?;

140140 let mut active = att.into_active_model();

141141 active.owner_id = Set(message_id);

142142 let updated = active.update(&db).await?;

Modified chat/storagelayer/mapper/conversation.rs +94 −8 View chat/storagelayer/mapper/conversation.rs

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

22

33use async_trait::async_trait;

44use sea_orm::{

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

6 QueryOrder, QuerySelect, Set, TransactionTrait,

5 ActiveModelTrait, ColumnTrait, Condition, EntityTrait, IntoActiveModel, PaginatorTrait,

6 QueryFilter, QueryOrder, QuerySelect, Set, TransactionTrait,

77};

88use space::error::SpaceError;

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

@@ -64,14 +64,55 @@

6464 let kind = input.kind;

6565 let counterpart_user_id = input.counterpart_user_id;

6666 let channel_name = input.title.as_deref().unwrap_or("chat").to_string();

67 let title = input.title;

68 let title_for_chat = title.clone();

67 let title_for_chat = input.title;

68 let title_in_txn = title_for_chat.clone();

6969

70 let channel_id = self

70 let (channel_id, existing) = self

7171 .db

7272 .writer()

7373 .transaction(|txn| {

7474 Box::pin(async move {

75 auth::require_active_member(txn, space_id, creator).await?;

76 if kind == ConversationKind::Private {

77 let counterpart = counterpart_user_id.ok_or_else(|| {

78 ChatStorageError::InvalidInput(

79 "private conversation requires counterpart_user_id".to_string(),

80 )

81 })?;

82 if counterpart == creator {

83 return Err(ChatStorageError::InvalidInput(

84 "counterpart_user_id cannot match the creator".to_string(),

85 ));

86 }

87 let existing = chat_entity::Entity::find()

88 .inner_join(channel_entity::Entity)

89 .filter(channel_entity::Column::SpaceId.eq(space_id))

90 .filter(chat_entity::Column::Kind.eq(chat_entity::ChatKind::Private))

91 .filter(

92 Condition::any()

93 .add(

94 Condition::all()

95 .add(chat_entity::Column::CreatorUserId.eq(creator))

96 .add(

97 chat_entity::Column::CounterpartUserId

98 .eq(counterpart),

99 ),

100 )

101 .add(

102 Condition::all()

103 .add(chat_entity::Column::CreatorUserId.eq(counterpart))

104 .add(

105 chat_entity::Column::CounterpartUserId.eq(creator),

106 ),

107 ),

108 )

109 .lock_exclusive()

110 .one(txn)

111 .await?;

112 if let Some(chat) = existing {

113 return Ok((chat.channel_id, Some(chat.into())));

114 }

115 }

75116 let object_channel = ensure_channel_for_resource(

76117 txn,

77118 space_id,

@@ -85,13 +126,13 @@

85126 channel_id: Set(object_channel.id),

86127 kind: Set(kind.into()),

87128 counterpart_user_id: Set(counterpart_user_id),

88 title: Set(title_for_chat.clone()),

129 title: Set(title_in_txn.clone()),

89130 creator_user_id: Set(creator),

90131 created_at: Set(now),

91132 updated_at: Set(None),

92133 };

93134 chat_entity::Entity::insert(chat).exec(txn).await?;

94 Ok(object_channel.id)

135 Ok((object_channel.id, None))

95136 })

96137 })

97138 .await

@@ -100,12 +141,16 @@

100141 sea_orm::TransactionError::Transaction(err) => err,

101142 })?;

102143

144 if let Some(existing) = existing {

145 return Ok(existing);

146 }

147

103148 Ok(Conversation {

104149 id,

105150 channel_id,

106151 kind,

107152 counterpart_user_id,

108 title,

153 title: title_for_chat,

109154 creator_user_id: creator,

110155 created_at: now,

111156 updated_at: None,

@@ -142,6 +187,47 @@

142187 Ok(rows.into_iter().map(Into::into).collect())

143188 }

144189

190 async fn list_for_user(

191 &self,

192 ctx: &ChatStorageCtx,

193 page: Page,

194 ) -> ChatStorageResult<Vec<Conversation>> {

195 let user_id = auth::require_authenticated(ctx).await?;

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

197 let member_spaces: Vec<Uuid> = space_members::Entity::find()

198 .filter(space_members::Column::UserId.eq(user_id))

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

200 .select_only()

201 .column(space_members::Column::SpaceId)

202 .into_tuple()

203 .all(&db)

204 .await?;

205 let rows = chat_entity::Entity::find()

206 .inner_join(channel_entity::Entity)

207 .filter(

208 Condition::any()

209 .add(

210 Condition::all()

211 .add(chat_entity::Column::Kind.eq(chat_entity::ChatKind::Private))

212 .add(

213 Condition::any()

214 .add(chat_entity::Column::CreatorUserId.eq(user_id))

215 .add(chat_entity::Column::CounterpartUserId.eq(user_id)),

216 ),

217 )

218 .add(

219 Condition::all()

220 .add(chat_entity::Column::Kind.eq(chat_entity::ChatKind::Group))

221 .add(channel_entity::Column::SpaceId.is_in(member_spaces)),

222 ),

223 )

224 .order_by_desc(chat_entity::COLUMN.created_at)

225 .limit(page.clamped_limit())

226 .all(&db)

227 .await?;

228 Ok(rows.into_iter().map(Into::into).collect())

229 }

230

145231 async fn update_title(

146232 &self,

147233 ctx: &ChatStorageCtx,

Modified chat/storagelayer/mapper/message.rs +24 −10 View chat/storagelayer/mapper/message.rs

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

44use futures::FutureExt;

55use sea_orm::sea_query::OnConflict;

66use sea_orm::{

7 ActiveModelTrait, Condition, EntityTrait, IntoActiveModel, PaginatorTrait, QueryFilter,

8 QueryOrder, QuerySelect, Set, TransactionTrait,

7 ActiveModelTrait, Condition, EntityTrait, IntoActiveModel, QueryFilter, QueryOrder,

8 QuerySelect, Set, TransactionTrait,

99};

1010use time::OffsetDateTime;

1111use uuid::Uuid;

@@ -144,11 +144,12 @@

144144 ctx: &ChatStorageCtx,

145145 id: Uuid,

146146 ) -> ChatStorageResult<Option<Message>> {

147 let chat_id = crate::auth::chat_id_for_message(&self.db, id).await?;

148 crate::auth::require_member(&self.db, ctx, chat_id).await?;

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

150 let model = message_entity::Entity::find_by_id(id).one(&db).await?;

151 Ok(model.map(Into::into))

148 let Some(model) = message_entity::Entity::find_by_id(id).one(&db).await? else {

149 return Ok(None);

150 };

151 crate::auth::require_member(&self.db, ctx, model.chat_id).await?;

152 Ok(Some(model.into()))

152153 }

153154

154155 async fn list_by_conversation(

@@ -169,9 +170,14 @@

169170 let cursor_id = Uuid::parse_str(cursor)

170171 .map_err(|_| ChatStorageError::InvalidInput("invalid cursor".to_string()))?;

171172 let cursor_msg = message_entity::Entity::find_by_id(cursor_id)

173 .filter(message_entity::COLUMN.chat_id.eq(chat_id))

172174 .one(&db)

173175 .await?

174 .ok_or_else(|| ChatStorageError::MessageNotFound(cursor_id))?;

176 .ok_or_else(|| {

177 ChatStorageError::InvalidInput(

178 "cursor does not belong to this conversation".to_string(),

179 )

180 })?;

175181 let cursor_ts = cursor_msg.created_at;

176182 let cursor_cond = match query.direction {

177183 ListDirection::Forward => Condition::any()

@@ -345,10 +351,18 @@

345351 })

346352 }

347353

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

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

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

350 let count = message_entity::Entity::find_by_id(id).count(&db).await?;

351 Ok(count > 0)

356 let Some(model) = message_entity::Entity::find_by_id(id).one(&db).await? else {

357 return Ok(false);

358 };

359 if crate::auth::require_member(&self.db, ctx, model.chat_id)

360 .await

361 .is_err()

362 {

363 return Ok(false);

364 }

365 Ok(true)

352366 }

353367}

354368impl std::fmt::Debug for SeaMessageRepository {

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

@@ -28,6 +28,12 @@

2828 page: Page,

2929 ) -> ChatStorageResult<Vec<Conversation>>;

3030

31 async fn list_for_user(

32 &self,

33 ctx: &ChatStorageCtx,

34 page: Page,

35 ) -> ChatStorageResult<Vec<Conversation>>;

36

3137 async fn update_title(

3238 &self,

3339 ctx: &ChatStorageCtx,

Modified chat/storagelayer/auth.rs +1 −1 View chat/storagelayer/auth.rs

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

9595 Ok(user_id)

9696}

9797

98async fn require_active_member<C: ConnectionTrait>(

98pub(crate) async fn require_active_member<C: ConnectionTrait>(

9999 db: &C,

100100 space_id: Uuid,

101101 user_id: Uuid,

Modified lib/space/features/channels/impls.rs +14 −2 View lib/space/features/channels/impls.rs

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

44use crate::features::SpaceService;

55

66use super::usecase;

7use super::{ChannelPage, ChannelView};

7use super::{ChannelPage, ChannelView, UpdateChannelCmd};

88

99impl SpaceService {

1010 pub async fn get_channel(

@@ -49,7 +49,19 @@

4949 space_id: Uuid,

5050 parent_id: Option<Uuid>,

5151 name: &str,

52 icon: Option<String>,

53 color: Option<String>,

54 ) -> SpaceResult<ChannelView> {

55 usecase::create_directory(self, caller, space_id, parent_id, name, icon, color).await

56 }

57

58 pub async fn update_channel(

59 &self,

60 caller: Uuid,

61 space_id: Uuid,

62 channel_id: Uuid,

63 cmd: UpdateChannelCmd,

5264 ) -> SpaceResult<ChannelView> {

53 usecase::create_directory(self, caller, space_id, parent_id, name).await

65 usecase::update_channel(self, caller, space_id, channel_id, cmd).await

5466 }

5567}

Modified lib/space/features/channels/usecase.rs +84 −3 View lib/space/features/channels/usecase.rs

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

1212use crate::features::SpaceService;

1313use crate::metrics;

1414

15use super::{ChannelPage, ChannelView};

15use super::{ChannelPage, ChannelView, UpdateChannelCmd};

1616

1717async fn member_space(

1818 svc: &SpaceService,

@@ -156,6 +156,8 @@

156156 space_id: Uuid,

157157 parent_id: Option<Uuid>,

158158 name: &str,

159 icon: Option<String>,

160 color: Option<String>,

159161) -> SpaceResult<ChannelView> {

160162 let _timer = metrics::Timer::start(metrics::FEATURE_CHANNELS_ENSURE);

161163 let result: SpaceResult<ChannelView> = async {

@@ -169,9 +171,9 @@

169171 .set_name(name.to_string())

170172 .set_kind(channel_entity::ChannelKind::Directory)

171173 .set_parent_id(parent_id)

172 .set_color(None)

174 .set_icon(icon)

173175 .set_position(0)

174 .set_icon(None)

176 .set_color(color)

175177 .set_topic(None)

176178 .set_purpose(None)

177179 .set_is_archived(false)

@@ -190,6 +192,85 @@

190192 result

191193}

192194

195pub async fn update_channel(

196 svc: &SpaceService,

197 caller: Uuid,

198 space_id: Uuid,

199 channel_id: Uuid,

200 cmd: UpdateChannelCmd,

201) -> SpaceResult<ChannelView> {

202 let _timer = metrics::Timer::start(metrics::FEATURE_CHANNELS_UPDATE);

203 let result: SpaceResult<ChannelView> = async {

204 member_space(svc, caller, space_id).await?;

205 let writer = svc.db.writer();

206 let txn = writer.begin().await?;

207

208 let channel = channel_entity::Entity::find_by_id(channel_id)

209 .one(&txn)

210 .await?

211 .ok_or_else(|| SpaceError::NotFound(format!("channel {channel_id}")))?;

212 if channel.space_id != Some(space_id) {

213 return Err(SpaceError::NotFound(format!("channel {channel_id}")));

214 }

215

216 let mut active: channel_entity::ActiveModel = channel.clone().into();

217 if let Some(name) = cmd.name {

218 let name = name.trim();

219 if name.is_empty() {

220 return Err(SpaceError::BadRequest("channel name must not be empty".into()));

221 }

222 active.name = Set(name.to_string());

223 }

224 if let Some(icon) = cmd.icon {

225 active.icon = Set(icon);

226 }

227 if let Some(position) = cmd.position {

228 active.position = Set(position);

229 }

230 if let Some(color) = cmd.color {

231 active.color = Set(color);

232 }

233 if let Some(parent_id) = cmd.parent_id {

234 if parent_id == Some(channel_id) {

235 return Err(SpaceError::BadRequest(

236 "a channel cannot be its own parent".into(),

237 ));

238 }

239 if let Some(pid) = parent_id {

240 if channel.kind == channel_entity::ChannelKind::Directory {

241 let mut current = Some(pid);

242 while let Some(cid) = current {

243 if cid == channel_id {

244 return Err(SpaceError::BadRequest(

245 "cannot move a directory into its own subtree".into(),

246 ));

247 }

248 current = channel_entity::Entity::find_by_id(cid)

249 .one(&txn)

250 .await?

251 .ok_or_else(|| {

252 SpaceError::NotFound(format!("parent channel {cid}"))

253 })?

254 .parent_id;

255 }

256 }

257 }

258 require_directory_parent(&txn, space_id, parent_id).await?;

259 active.parent_id = Set(parent_id);

260 }

261 active.updated_at = Set(OffsetDateTime::now_utc());

262 let saved = active.update(&txn).await?;

263 txn.commit().await?;

264 let created_by_user = load_user(&svc.db, saved.created_by)

265 .await?

266 .map(UserBasicInfo::from);

267 Ok(ChannelView::from_model(saved, None, created_by_user))

268 }

269 .await;

270 metrics::record_outcome(metrics::FEATURE_CHANNELS_UPDATE, &result);

271 result

272}

273

193274pub async fn ensure_channel_for_resource<C: ConnectionTrait>(

194275 db: &C,

195276 space_id: Uuid,

Modified lib/space/features/invites/usecase.rs +15 −0 View lib/space/features/invites/usecase.rs

@@ -291,6 +291,21 @@

291291 "user is already a member of this space".to_string(),

292292 ));

293293 }

294 let space_model = space_entity::Entity::find_by_id(invite.space_id)

295 .one(&txn)

296 .await?

297 .ok_or_else(|| SpaceError::NotFound(format!("space {}", invite.space_id)))?;

298 if space_model.member_must_enable_2fa {

299 let user_model = user_entity::Entity::find_by_id(caller)

300 .one(&txn)

301 .await?

302 .ok_or_else(|| SpaceError::NotFound(format!("user {caller}")))?;

303 if !user_model.totp_enabled {

304 return Err(SpaceError::Forbidden(

305 "two-factor authentication must be enabled to join this space".to_string(),

306 ));

307 }

308 }

294309 let now = OffsetDateTime::now_utc();

295310 let member: member_entity::ActiveModel = member_entity::ActiveModel::builder()

296311 .set_space_id(invite.space_id)

Modified lib/space/features/members/usecase.rs +14 −1 View lib/space/features/members/usecase.rs

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

22use db::conn::sea_orm::*;

33use db::space::space_members as member_entity;

44use db::space::spaces as space_entity;

5use db::user::users as user_entity;

56use db::user::users_loader::{load_user, load_users};

67use export::user::basic::UserBasicInfo;

78use time::OffsetDateTime;

@@ -92,7 +93,7 @@

9293) -> SpaceResult<SpaceMemberView> {

9394 let _timer = metrics::Timer::start(metrics::FEATURE_MEMBERS_ADD);

9495 let result: SpaceResult<SpaceMemberView> = async {

95 space(caller, space_id, svc).await?;

96 let space_model = space(caller, space_id, svc).await?;

9697 let db = svc.db.writer();

9798 let now = OffsetDateTime::now_utc();

9899 let existing = member_entity::Entity::find_by_id((space_id, cmd.user_id))

@@ -104,6 +105,18 @@

104105 cmd.user_id

105106 )));

106107 }

108 if space_model.member_must_enable_2fa {

109 let target = user_entity::Entity::find_by_id(cmd.user_id)

110 .one(&db)

111 .await?

112 .ok_or_else(|| SpaceError::NotFound(format!("user {}", cmd.user_id)))?;

113 if !target.totp_enabled {

114 return Err(SpaceError::Forbidden(format!(

115 "user {} must enable two-factor authentication to join this space",

116 cmd.user_id

117 )));

118 }

119 }

107120 let mute_until = parse_mute_until(cmd.mute_until.as_deref())?;

108121 let active: member_entity::ActiveModel = member_entity::ActiveModel::builder()

109122 .set_space_id(space_id)

Modified lib/space/features/sidebar/usecase.rs +1 −14 View lib/space/features/sidebar/usecase.rs

@@ -128,21 +128,8 @@

128128 })

129129 .collect();

130130

131 let group = |id: u128, name: &str, position: i32| SidebarNode {

132 id: Uuid::from_u128(id),

133 kind: SidebarNodeKind::Group,

134 name: name.to_string(),

135 position,

136 color: None,

137 is_archived: false,

138 parent_id: None,

139 };

140131

141 let mut tree = vec![

142 group(GROUP_CHANNELS, "频道", 0),

143 group(GROUP_REPOS, "仓库", 1),

144 group(GROUP_MEMBERS, "私信", 2),

145 ];

132 let mut tree = vec![];

146133 tree.extend(channel_nodes);

147134 tree.extend(repo_nodes);

148135 tree.extend(member_nodes);

Modified lib/space/features/spaces/usecase.rs +13 −0 View lib/space/features/spaces/usecase.rs

@@ -5,6 +5,7 @@

55use db::conn::sea_orm::*;

66use db::space::space_members as member_entity;

77use db::space::spaces as space_entity;

8use db::user::users as user_entity;

89use db::user::users_loader::{load_user, load_users};

910use export::user::basic::UserBasicInfo;

1011use time::OffsetDateTime;

@@ -36,6 +37,18 @@

3637 let txn = writer.begin().await?;

3738 let now = OffsetDateTime::now_utc();

3839 let id = Uuid::now_v7();

40 if cmd.member_must_enable_2fa {

41 let db = svc.db.reader();

42 let creator = user_entity::Entity::find_by_id(caller)

43 .one(&db)

44 .await?

45 .ok_or_else(|| SpaceError::NotFound(format!("user {caller}")))?;

46 if !creator.totp_enabled {

47 return Err(SpaceError::Forbidden(

48 "two-factor authentication must be enabled to create this space".into(),

49 ));

50 }

51 }

3952 let active: space_entity::ActiveModel = space_entity::ActiveModel::builder()

4053 .set_id(id)

4154 .set_lower_name(name.to_lowercase())

Modified lib/space/handler/channels/mod.rs +2 −0 View lib/space/handler/channels/mod.rs

@@ -1,2 +1,4 @@

1pub(crate) mod create;

12pub(crate) mod get;

23pub(crate) mod list;

4pub(crate) mod update;

Modified lib/space/handler/openapi.rs +4 −2 View lib/space/handler/openapi.rs

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

11use utoipa::OpenApi;

22

33use export::space::bans::{CreateBanCmd, SpaceBanPage, SpaceBanView, UpdateBanCmd};

4use export::space::channels::{ChannelPage, ChannelView};

4use export::space::channels::{ChannelPage, ChannelView, CreateDirectoryCmd, UpdateChannelCmd};

55use export::space::invites::{AcceptInviteCmd, CreateInviteCmd, SpaceInvitePage, SpaceInviteView};

66use export::space::members::{

77 AddMemberCmd, MessageNotifications, SpaceMemberPage, SpaceMemberStatus, SpaceMemberView,

@@ -55,8 +55,10 @@

5555 crate::handler::bans::list::handler,

5656 crate::handler::bans::update::handler,

5757 crate::handler::bans::unban::handler,

58 crate::handler::channels::create::handler,

5859 crate::handler::channels::get::handler,

5960 crate::handler::channels::list::handler,

61 crate::handler::channels::update::handler,

6062 crate::handler::sidebar::get::handler,

6163 crate::handler::overwrites::set_role::handler,

6264 crate::handler::overwrites::remove_role::handler,

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

7274 RoleView, CreateRoleCmd, UpdateRoleCmd, RolePage, AssignRoleCmd, UnassignRoleCmd,

7375 SpaceInviteView, CreateInviteCmd, AcceptInviteCmd, SpaceInvitePage,

7476 SpaceBanView, CreateBanCmd, UpdateBanCmd, SpaceBanPage,

75 ChannelView, ChannelPage,

77 ChannelView, CreateDirectoryCmd, UpdateChannelCmd, ChannelPage,

7678 SidebarTreeView, SidebarNode, SidebarNodeKind,

7779 RoleOverwriteView, SetRoleOverwriteCmd, UserOverwriteView, SetUserOverwriteCmd,

7880 ))

Modified lib/space/handler/routes.rs +2 −2 View lib/space/handler/routes.rs

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

7575 )

7676 .route(

7777 "/spaces/{space_name}/channels",

78 get(super::channels::list::handler),

78 post(super::channels::create::handler).get(super::channels::list::handler),

7979 )

8080 .route(

8181 "/spaces/{space_name}/sidebar",

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

8383 )

8484 .route(

8585 "/spaces/{space_name}/channels/{cid}",

86 get(super::channels::get::handler),

86 get(super::channels::get::handler).patch(super::channels::update::handler),

8787 )

8888 .route(

8989 "/spaces/{space_name}/channels/{cid}/overwrites/roles",

Modified lib/space/metrics.rs +1 −0 View lib/space/metrics.rs

@@ -37,6 +37,7 @@

3737

3838pub const FEATURE_CHANNELS_GET: &str = "channels_get";

3939pub const FEATURE_CHANNELS_LIST: &str = "channels_list";

40pub const FEATURE_CHANNELS_UPDATE: &str = "channels_update";

4041pub const FEATURE_CHANNELS_ENSURE: &str = "channels_ensure";

4142

4243pub const FEATURE_OVERWRITES_SET_ROLE: &str = "overwrites_set_role";