feat(chat): add live and meet modules to chatlayer/storagelayer
zhenyi committed
Added chat/chatlayer/actor/live.rs chat/chatlayer/actor/live.rs +86 −0 View chat/chatlayer/actor/live.rs
@@ -0,0 +1,86 @@
1use std::sync::Arc;
2
3use cache::redis::RedisCache;
4use storagelayer::DynChatStorage;
5use tokio_util::sync::CancellationToken;
6
7use crate::actor::runtime::{ActorRegistry, run_actor};
8use crate::dedup;
9use crate::error::ChatLayerResult;
10use crate::events::envelope::EventEnvelope;
11use crate::events::live::ChatLiveEvent;
12use crate::producer::ChatProducer;
13use crate::projection;
14
15#[derive(Clone)]
16pub struct MeetActorCtx {
17 pub store: DynChatStorage,
18 pub producer: Arc<ChatProducer>,
19 pub kv: Arc<RedisCache>,
20 pub dedup_ttl_seconds: u64,
21}
22
23impl std::fmt::Debug for MeetActorCtx {
24 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
25 f.debug_struct("MeetActorCtx")
26 .field("dedup_ttl_seconds", &self.dedup_ttl_seconds)
27 .finish_non_exhaustive()
28 }
29}
30
31impl MeetActorCtx {
32 /// Errors are logged and swallowed; Kafka at-least-once redelivery is
33 pub async fn handle(&self, envelope: EventEnvelope<ChatLiveEvent>) {
34 match self.handle_inner(&envelope).await {
35 Ok(()) => crate::metrics::live_event_applied("success"),
36 Err(e) => {
37 crate::metrics::live_event_applied("failure");
38 tracing::error!(
39 event_id = %envelope.event_id,
40 channel = %envelope.payload.channel_id(),
41 error = %e,
42 "live event apply failed"
43 );
44 }
45 }
46 }
47
48 async fn handle_inner(&self, envelope: &EventEnvelope<ChatLiveEvent>) -> ChatLayerResult<()> {
49 if !dedup::is_new_event(&self.kv, envelope.event_id, self.dedup_ttl_seconds).await? {
50 crate::metrics::dedup_hit();
51 return Ok(());
52 }
53 let view_events = projection::live::apply_live_event(&self.store, envelope).await?;
54 for evt in view_events {
55 let sub = EventEnvelope::new(envelope.actor_id, evt);
56 if let Err(e) = self.producer.publish_live_view(&sub).await {
57 tracing::warn!(
58 event_id = %envelope.event_id,
59 error = %e,
60 "failed to fan out live view event"
61 );
62 }
63 }
64 Ok(())
65 }
66}
67
68pub fn spawn_meet_registry(
69 ctx: Arc<MeetActorCtx>,
70 shutdown: CancellationToken,
71) -> Arc<ActorRegistry<EventEnvelope<ChatLiveEvent>>> {
72 ActorRegistry::new(
73 "live",
74 move |reg, _key, rx| {
75 let ctx = ctx.clone();
76 async move {
77 run_actor(rx, reg.shutdown_token(), move |msg| {
78 let ctx = ctx.clone();
79 Box::pin(async move { ctx.handle(msg).await })
80 })
81 .await;
82 }
83 },
84 shutdown,
85 )
86}
Added chat/chatlayer/consumer/live_worker.rs chat/chatlayer/consumer/live_worker.rs +78 −0 View chat/chatlayer/consumer/live_worker.rs
@@ -0,0 +1,78 @@
1use std::sync::Arc;
2
3use queue::{Queue, QueueError, SubscribeHandle};
4use tokio_util::sync::CancellationToken;
5
6use crate::actor::live::MeetActorCtx;
7use crate::actor::runtime::ActorRegistry;
8use crate::consumer::WorkerSettings;
9use crate::events::envelope::EventEnvelope;
10use crate::events::live::ChatLiveEvent;
11
12pub struct LiveWorker {
13 queue: Arc<Queue>,
14 settings: WorkerSettings,
15 registry: Arc<ActorRegistry<EventEnvelope<ChatLiveEvent>>>,
16}
17
18impl std::fmt::Debug for LiveWorker {
19 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
20 f.debug_struct("LiveWorker")
21 .field("settings", &self.settings)
22 .finish_non_exhaustive()
23 }
24}
25
26impl LiveWorker {
27 pub fn new(
28 queue: Arc<Queue>,
29 settings: WorkerSettings,
30 registry: Arc<ActorRegistry<EventEnvelope<ChatLiveEvent>>>,
31 ) -> Self {
32 Self {
33 queue,
34 settings,
35 registry,
36 }
37 }
38
39 pub fn run(&self) -> Result<SubscribeHandle, QueueError> {
40 let mode = crate::consumer::group_mode(&self.settings.group_id);
41 let topics = [self.settings.topic.as_str()];
42 let registry = self.registry.clone();
43 self.queue.subscribe_raw(mode, &topics, move |msg| {
44 let registry = registry.clone();
45 Box::pin(async move {
46 match serde_json::from_slice::<EventEnvelope<ChatLiveEvent>>(&msg.payload) {
47 Ok(envelope) => {
48 registry.dispatch(envelope.payload.channel_id(), envelope);
49 Ok(())
50 }
51 Err(e) => {
52 tracing::warn!(
53 topic = %msg.topic,
54 offset = msg.offset,
55 error = %e,
56 "malformed live event dropped"
57 );
58 Ok(())
59 }
60 }
61 })
62 })
63 }
64
65 pub fn registry(&self) -> Arc<ActorRegistry<EventEnvelope<ChatLiveEvent>>> {
66 self.registry.clone()
67 }
68}
69
70pub fn build_live_worker(
71 queue: Arc<Queue>,
72 settings: WorkerSettings,
73 ctx: Arc<MeetActorCtx>,
74 shutdown: CancellationToken,
75) -> LiveWorker {
76 let registry = crate::actor::live::spawn_meet_registry(ctx, shutdown);
77 LiveWorker::new(queue, settings, registry)
78}
Added chat/chatlayer/events/live.rs chat/chatlayer/events/live.rs +150 −0 View chat/chatlayer/events/live.rs
@@ -0,0 +1,150 @@
1use serde::{Deserialize, Serialize};
2use storagelayer::LeaveReason;
3use time::OffsetDateTime;
4use uuid::Uuid;
5
6#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
7#[serde(tag = "type", rename_all = "snake_case")]
8pub enum ChatLiveEvent {
9 MeetStarted {
10 channel_id: Uuid,
11 room_name: String,
12 by: Uuid,
13 ts: OffsetDateTime,
14 },
15 MeetEnded {
16 channel_id: Uuid,
17 room_name: String,
18 by: Uuid,
19 ts: OffsetDateTime,
20 },
21 ParticipantJoined {
22 channel_id: Uuid,
23 room_name: String,
24 user_id: Uuid,
25 ts: OffsetDateTime,
26 },
27 ParticipantLeft {
28 channel_id: Uuid,
29 room_name: String,
30 user_id: Uuid,
31 reason: LeaveReason,
32 ts: OffsetDateTime,
33 },
34}
35
36impl ChatLiveEvent {
37 pub fn channel_id(&self) -> Uuid {
38 match self {
39 Self::MeetStarted { channel_id, .. }
40 | Self::MeetEnded { channel_id, .. }
41 | Self::ParticipantJoined { channel_id, .. }
42 | Self::ParticipantLeft { channel_id, .. } => *channel_id,
43 }
44 }
45}
46
47#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
48#[serde(tag = "type", rename_all = "snake_case")]
49pub enum ChatLiveViewEvent {
50 MeetStartedView {
51 user_id: Uuid,
52 channel_id: Uuid,
53 room_name: String,
54 by: Uuid,
55 ts: OffsetDateTime,
56 },
57 MeetEndedView {
58 user_id: Uuid,
59 channel_id: Uuid,
60 room_name: String,
61 ts: OffsetDateTime,
62 },
63 ParticipantJoinedView {
64 user_id: Uuid,
65 channel_id: Uuid,
66 room_name: String,
67 participant_id: Uuid,
68 ts: OffsetDateTime,
69 },
70 ParticipantLeftView {
71 user_id: Uuid,
72 channel_id: Uuid,
73 room_name: String,
74 participant_id: Uuid,
75 reason: LeaveReason,
76 ts: OffsetDateTime,
77 },
78}
79
80impl ChatLiveViewEvent {
81 pub fn user_id(&self) -> Uuid {
82 match self {
83 Self::MeetStartedView { user_id, .. }
84 | Self::MeetEndedView { user_id, .. }
85 | Self::ParticipantJoinedView { user_id, .. }
86 | Self::ParticipantLeftView { user_id, .. } => *user_id,
87 }
88 }
89}
90
91#[cfg(test)]
92mod tests {
93 use super::*;
94 use uuid::Uuid;
95
96 #[test]
97 fn live_event_serde_roundtrip() {
98 let evt = ChatLiveEvent::ParticipantLeft {
99 channel_id: Uuid::now_v7(),
100 room_name: "room-1".into(),
101 user_id: Uuid::now_v7(),
102 reason: storagelayer::LeaveReason::NetworkDrop,
103 ts: time::OffsetDateTime::now_utc(),
104 };
105 let json = serde_json::to_string(&evt).unwrap();
106 let back: ChatLiveEvent = serde_json::from_str(&json).unwrap();
107 assert_eq!(back, evt);
108 assert!(json.contains("\"type\":\"participant_left\""));
109 }
110
111 #[test]
112 fn live_event_channel_id_key() {
113 let channel = Uuid::now_v7();
114 let evt = ChatLiveEvent::MeetStarted {
115 channel_id: channel,
116 room_name: "room-2".into(),
117 by: Uuid::now_v7(),
118 ts: time::OffsetDateTime::now_utc(),
119 };
120 assert_eq!(evt.channel_id(), channel);
121 }
122
123 #[test]
124 fn live_view_event_serde_roundtrip() {
125 let evt = ChatLiveViewEvent::ParticipantJoinedView {
126 user_id: Uuid::now_v7(),
127 channel_id: Uuid::now_v7(),
128 room_name: "room-3".into(),
129 participant_id: Uuid::now_v7(),
130 ts: time::OffsetDateTime::now_utc(),
131 };
132 let json = serde_json::to_string(&evt).unwrap();
133 let back: ChatLiveViewEvent = serde_json::from_str(&json).unwrap();
134 assert_eq!(back, evt);
135 }
136
137 #[test]
138 fn live_view_event_user_id_key() {
139 let user = Uuid::now_v7();
140 let evt = ChatLiveViewEvent::MeetStartedView {
141 user_id: user,
142 channel_id: Uuid::now_v7(),
143 room_name: "room-4".into(),
144 by: Uuid::now_v7(),
145 ts: time::OffsetDateTime::now_utc(),
146 };
147 assert_eq!(evt.user_id(), user);
148 assert_eq!(crate::topics::live_user_key(user), user.to_string());
149 }
150}
Added chat/chatlayer/projection/live.rs chat/chatlayer/projection/live.rs +400 −0 View chat/chatlayer/projection/live.rs
@@ -0,0 +1,400 @@
1use storagelayer::DynChatStorage;
2use storagelayer::models::meet::NewMeet;
3use storagelayer::models::meet_participant::NewMeetParticipant;
4use storagelayer::repository::ChatStorageCtx;
5use uuid::Uuid;
6
7use crate::error::ChatLayerResult;
8use crate::events::envelope::EventEnvelope;
9use crate::events::live::{ChatLiveEvent, ChatLiveViewEvent};
10
11pub async fn apply_live_event(
12 store: &DynChatStorage,
13 envelope: &EventEnvelope<ChatLiveEvent>,
14) -> ChatLayerResult<Vec<ChatLiveViewEvent>> {
15 match &envelope.payload {
16 ChatLiveEvent::MeetStarted {
17 channel_id,
18 room_name,
19 by,
20 ts,
21 } => apply_meet_started(store, envelope.actor_id, *channel_id, room_name, *by, *ts).await,
22 ChatLiveEvent::MeetEnded {
23 channel_id,
24 room_name,
25 by,
26 ts,
27 } => apply_meet_ended(store, envelope.actor_id, *channel_id, room_name, *by, *ts).await,
28 ChatLiveEvent::ParticipantJoined {
29 channel_id,
30 room_name,
31 user_id,
32 ts,
33 } => {
34 apply_participant_joined(
35 store,
36 envelope.actor_id,
37 *channel_id,
38 room_name,
39 *user_id,
40 *ts,
41 )
42 .await
43 }
44 ChatLiveEvent::ParticipantLeft {
45 channel_id,
46 room_name,
47 user_id,
48 reason,
49 ts,
50 } => {
51 apply_participant_left(
52 store,
53 envelope.actor_id,
54 *channel_id,
55 room_name,
56 *user_id,
57 *reason,
58 *ts,
59 )
60 .await
61 }
62 }
63}
64
65async fn apply_meet_started(
66 store: &DynChatStorage,
67 actor_id: Uuid,
68 channel_id: Uuid,
69 room_name: &str,
70 by: Uuid,
71 ts: time::OffsetDateTime,
72) -> ChatLayerResult<Vec<ChatLiveViewEvent>> {
73 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?,
76 None => {
77 let created = store
78 .meets()
79 .create(
80 &ctx,
81 NewMeet {
82 channel_id,
83 room_name: room_name.to_string(),
84 created_by: by,
85 },
86 )
87 .await?;
88 store.meets().mark_active(&ctx, created.id).await?
89 }
90 };
91
92 let members = store
93 .meets()
94 .list_channel_member_user_ids(&ctx, channel_id)
95 .await?;
96 Ok(members
97 .into_iter()
98 .map(|user_id| ChatLiveViewEvent::MeetStartedView {
99 user_id,
100 channel_id,
101 room_name: meet.room_name.clone(),
102 by,
103 ts,
104 })
105 .collect())
106}
107
108async fn apply_meet_ended(
109 store: &DynChatStorage,
110 actor_id: Uuid,
111 channel_id: Uuid,
112 room_name: &str,
113 _by: Uuid,
114 ts: time::OffsetDateTime,
115) -> ChatLayerResult<Vec<ChatLiveViewEvent>> {
116 let ctx = ChatStorageCtx::new(actor_id);
117 let Some(meet) = store.meets().find_by_room_name(&ctx, room_name).await? else {
118 return Ok(Vec::new());
119 };
120 store.meets().mark_ended(&ctx, meet.id).await?;
121 store
122 .meet_participants()
123 .mark_all_ended(&ctx, meet.id)
124 .await?;
125
126 let members = store
127 .meets()
128 .list_channel_member_user_ids(&ctx, channel_id)
129 .await?;
130 Ok(members
131 .into_iter()
132 .map(|user_id| ChatLiveViewEvent::MeetEndedView {
133 user_id,
134 channel_id,
135 room_name: meet.room_name.clone(),
136 ts,
137 })
138 .collect())
139}
140
141async fn apply_participant_joined(
142 store: &DynChatStorage,
143 actor_id: Uuid,
144 channel_id: Uuid,
145 room_name: &str,
146 user_id: Uuid,
147 ts: time::OffsetDateTime,
148) -> ChatLayerResult<Vec<ChatLiveViewEvent>> {
149 let ctx = ChatStorageCtx::new(actor_id);
150 let Some(meet) = store.meets().find_by_room_name(&ctx, room_name).await? else {
151 return Ok(Vec::new());
152 };
153 let participant = store
154 .meet_participants()
155 .join(
156 &ctx,
157 NewMeetParticipant {
158 meet_id: meet.id,
159 user_id,
160 joined_at: ts,
161 connection_id: None,
162 },
163 )
164 .await?;
165
166 let members = store.meet_participants().list_active(&ctx, meet.id).await?;
167 Ok(members
168 .into_iter()
169 .map(|m| ChatLiveViewEvent::ParticipantJoinedView {
170 user_id: m.user_id,
171 channel_id,
172 room_name: meet.room_name.clone(),
173 participant_id: participant.id,
174 ts,
175 })
176 .collect())
177}
178
179async fn apply_participant_left(
180 store: &DynChatStorage,
181 actor_id: Uuid,
182 channel_id: Uuid,
183 room_name: &str,
184 user_id: Uuid,
185 reason: storagelayer::LeaveReason,
186 ts: time::OffsetDateTime,
187) -> ChatLayerResult<Vec<ChatLiveViewEvent>> {
188 let ctx = ChatStorageCtx::new(actor_id);
189 let Some(meet) = store.meets().find_by_room_name(&ctx, room_name).await? else {
190 return Ok(Vec::new());
191 };
192 let active_before = store.meet_participants().list_active(&ctx, meet.id).await?;
193 let Some(row) = active_before
194 .iter()
195 .find(|p| p.user_id == user_id)
196 .map(|p| p.id)
197 else {
198 return Ok(Vec::new());
199 };
200 store.meet_participants().leave(&ctx, row, reason).await?;
201
202 Ok(active_before
203 .into_iter()
204 .map(|m| ChatLiveViewEvent::ParticipantLeftView {
205 user_id: m.user_id,
206 channel_id,
207 room_name: meet.room_name.clone(),
208 participant_id: row,
209 reason,
210 ts,
211 })
212 .collect())
213}
214#[cfg(test)]
215mod tests {
216 use super::*;
217 use storagelayer::ChatStorageFactory;
218 use storagelayer::LeaveReason;
219
220 fn ts() -> time::OffsetDateTime {
221 time::OffsetDateTime::now_utc()
222 }
223
224 #[tokio::test]
225 async fn meet_started_creates_active_row() {
226 let store = ChatStorageFactory::in_memory();
227 let creator = Uuid::now_v7();
228 let channel = Uuid::now_v7();
229 let room_name = "room-start".to_string();
230
231 let envelope = EventEnvelope::new(
232 creator,
233 ChatLiveEvent::MeetStarted {
234 channel_id: channel,
235 room_name: room_name.clone(),
236 by: creator,
237 ts: ts(),
238 },
239 );
240 let _fanout = apply_live_event(&store, &envelope).await.unwrap();
241
242 let ctx = ChatStorageCtx::new(creator);
243 let meet = store
244 .meets()
245 .find_by_room_name(&ctx, &room_name)
246 .await
247 .unwrap()
248 .expect("meet row exists");
249 assert_eq!(meet.status, storagelayer::MeetStatus::Active);
250 assert_eq!(meet.channel_id, channel);
251 }
252
253 #[tokio::test]
254 async fn participant_join_and_left_persist_records() {
255 let store = ChatStorageFactory::in_memory();
256 let creator = Uuid::now_v7();
257 let channel = Uuid::now_v7();
258 let alice = Uuid::now_v7();
259 let room_name = "room-join".to_string();
260
261 let ctx = ChatStorageCtx::new(creator);
262 let meet = store
263 .meets()
264 .create(
265 &ctx,
266 NewMeet {
267 channel_id: channel,
268 room_name: room_name.clone(),
269 created_by: creator,
270 },
271 )
272 .await
273 .unwrap();
274 store.meets().mark_active(&ctx, meet.id).await.unwrap();
275
276 let join_env = EventEnvelope::new(
277 alice,
278 ChatLiveEvent::ParticipantJoined {
279 channel_id: channel,
280 room_name: room_name.clone(),
281 user_id: alice,
282 ts: ts(),
283 },
284 );
285 let fanout = apply_live_event(&store, &join_env).await.unwrap();
286 assert!(fanout.iter().any(|e| matches!(
287 e,
288 ChatLiveViewEvent::ParticipantJoinedView { user_id, .. } if *user_id == alice
289 )));
290
291 let active = store
292 .meet_participants()
293 .list_active(&ctx, meet.id)
294 .await
295 .unwrap();
296 assert_eq!(active.len(), 1);
297 assert_eq!(active[0].user_id, alice);
298 assert!(active[0].left_at.is_none());
299
300 let leave_env = EventEnvelope::new(
301 alice,
302 ChatLiveEvent::ParticipantLeft {
303 channel_id: channel,
304 room_name: room_name.clone(),
305 user_id: alice,
306 reason: LeaveReason::NetworkDrop,
307 ts: ts(),
308 },
309 );
310 let fanout = apply_live_event(&store, &leave_env).await.unwrap();
311 assert!(fanout.iter().any(|e| matches!(
312 e,
313 ChatLiveViewEvent::ParticipantLeftView { reason, .. }
314 if *reason == LeaveReason::NetworkDrop
315 )));
316
317 let active = store
318 .meet_participants()
319 .list_active(&ctx, meet.id)
320 .await
321 .unwrap();
322 assert!(active.is_empty(), "no active participants after leave");
323 let alice_ctx = ChatStorageCtx::new(alice);
324 let history = store
325 .meet_participants()
326 .list_history(&alice_ctx, alice, 10)
327 .await
328 .unwrap();
329 assert_eq!(history.len(), 1);
330 assert_eq!(history[0].leave_reason, Some(LeaveReason::NetworkDrop));
331 assert!(history[0].left_at.is_some());
332 }
333
334 #[tokio::test]
335 async fn meet_ended_marks_remaining_participants_room_ended() {
336 let store = ChatStorageFactory::in_memory();
337 let creator = Uuid::now_v7();
338 let channel = Uuid::now_v7();
339 let room_name = "room-end".to_string();
340
341 let ctx = ChatStorageCtx::new(creator);
342 let meet = store
343 .meets()
344 .create(
345 &ctx,
346 NewMeet {
347 channel_id: channel,
348 room_name: room_name.clone(),
349 created_by: creator,
350 },
351 )
352 .await
353 .unwrap();
354 store.meets().mark_active(&ctx, meet.id).await.unwrap();
355 store
356 .meet_participants()
357 .join(
358 &ctx,
359 NewMeetParticipant {
360 meet_id: meet.id,
361 user_id: creator,
362 joined_at: ts(),
363 connection_id: None,
364 },
365 )
366 .await
367 .unwrap();
368
369 let end_env = EventEnvelope::new(
370 creator,
371 ChatLiveEvent::MeetEnded {
372 channel_id: channel,
373 room_name: room_name.clone(),
374 by: creator,
375 ts: ts(),
376 },
377 );
378 let _fanout = apply_live_event(&store, &end_env).await.unwrap();
379
380 let meet = store
381 .meets()
382 .find_by_room_name(&ctx, &room_name)
383 .await
384 .unwrap()
385 .expect("meet row exists");
386 assert_eq!(meet.status, storagelayer::MeetStatus::Ended);
387 let active = store
388 .meet_participants()
389 .list_active(&ctx, meet.id)
390 .await
391 .unwrap();
392 assert!(active.is_empty());
393 let history = store
394 .meet_participants()
395 .list_history(&ctx, creator, 10)
396 .await
397 .unwrap();
398 assert_eq!(history[0].leave_reason, Some(LeaveReason::RoomEnded));
399 }
400}
Added chat/storagelayer/mapper/meet.rs chat/storagelayer/mapper/meet.rs +211 −0 View chat/storagelayer/mapper/meet.rs
@@ -0,0 +1,211 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use sea_orm::{
5 ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set,
6 TransactionTrait,
7};
8use time::OffsetDateTime;
9use uuid::Uuid;
10
11use crate::auth;
12use crate::error::{ChatStorageError, ChatStorageResult};
13use crate::models::meet::{Meet, NewMeet};
14use crate::repository::ChatStorageCtx;
15use crate::repository::meet::MeetRepository;
16
17use db::channel::channels as channel_entity;
18use db::live::meets as meet_entity;
19use db::space::space_members as member_entity;
20
21pub struct SeaMeetRepository {
22 db: Arc<db::conn::Db>,
23}
24
25impl SeaMeetRepository {
26 pub fn new(db: Arc<db::conn::Db>) -> Self {
27 Self { db }
28 }
29}
30
31#[async_trait]
32impl MeetRepository for SeaMeetRepository {
33 async fn create(&self, ctx: &ChatStorageCtx, input: NewMeet) -> ChatStorageResult<Meet> {
34 let caller = auth::require_channel_member(&self.db, ctx, input.channel_id).await?;
35 if caller != input.created_by {
36 return Err(ChatStorageError::PermissionDenied(format!(
37 "user {caller} cannot create meet as user {}",
38 input.created_by
39 )));
40 }
41 self.db
42 .writer()
43 .transaction(|txn| {
44 Box::pin(async move {
45 let existing = meet_entity::Entity::find()
46 .filter(meet_entity::Column::ChannelId.eq(input.channel_id))
47 .filter(meet_entity::Column::Status.eq(meet_entity::MeetStatus::Active))
48 .one(txn)
49 .await?;
50 if existing.is_some() {
51 return Err(ChatStorageError::Conflict(
52 "channel already has an active meet".into(),
53 ));
54 }
55 let active = meet_entity::ActiveModel {
56 id: Set(Uuid::now_v7()),
57 channel_id: Set(input.channel_id),
58 room_name: Set(input.room_name.clone()),
59 status: Set(meet_entity::MeetStatus::Pending),
60 created_by: Set(input.created_by),
61 started_at: Set(None),
62 ended_at: Set(None),
63 created_at: Set(OffsetDateTime::now_utc()),
64 updated_at: Set(None),
65 };
66 let model = meet_entity::Entity::insert(active)
67 .exec_with_returning(txn)
68 .await?;
69 Ok(model.into())
70 })
71 })
72 .await
73 .map_err(|e| match e {
74 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),
75 sea_orm::TransactionError::Transaction(err) => err,
76 })
77 }
78
79 async fn find_active_by_channel(
80 &self,
81 ctx: &ChatStorageCtx,
82 channel_id: Uuid,
83 ) -> ChatStorageResult<Option<Meet>> {
84 auth::require_channel_member(&self.db, ctx, channel_id).await?;
85 let db = self.db.reader();
86 let model = meet_entity::Entity::find()
87 .filter(meet_entity::Column::ChannelId.eq(channel_id))
88 .filter(meet_entity::Column::Status.eq(meet_entity::MeetStatus::Active))
89 .one(&db)
90 .await?;
91 Ok(model.map(Into::into))
92 }
93
94 async fn find_by_room_name(
95 &self,
96 ctx: &ChatStorageCtx,
97 room_name: &str,
98 ) -> ChatStorageResult<Option<Meet>> {
99 auth::require_authenticated(ctx).await?;
100 let db = self.db.reader();
101 let model = meet_entity::Entity::find()
102 .filter(meet_entity::Column::RoomName.eq(room_name))
103 .one(&db)
104 .await?;
105 Ok(model.map(Into::into))
106 }
107
108 async fn find_by_id(
109 &self,
110 ctx: &ChatStorageCtx,
111 meet_id: Uuid,
112 ) -> ChatStorageResult<Option<Meet>> {
113 auth::require_authenticated(ctx).await?;
114 let db = self.db.reader();
115 let model = meet_entity::Entity::find_by_id(meet_id).one(&db).await?;
116 Ok(model.map(Into::into))
117 }
118
119 async fn mark_active(&self, ctx: &ChatStorageCtx, meet_id: Uuid) -> ChatStorageResult<Meet> {
120 let caller = auth::require_authenticated(ctx).await?;
121 self.db
122 .writer()
123 .transaction(|txn| {
124 Box::pin(async move {
125 let model = meet_entity::Entity::find_by_id(meet_id)
126 .one(txn)
127 .await?
128 .ok_or(ChatStorageError::MeetNotFound(meet_id))?;
129 if model.created_by != caller {
130 return Err(ChatStorageError::PermissionDenied(format!(
131 "user {caller} cannot activate meet {meet_id}"
132 )));
133 }
134 let mut active: meet_entity::ActiveModel = model.into();
135 active.status = Set(meet_entity::MeetStatus::Active);
136 active.started_at = Set(Some(OffsetDateTime::now_utc()));
137 active.updated_at = Set(Some(OffsetDateTime::now_utc()));
138 let model = active.update(txn).await?;
139 Ok(model.into())
140 })
141 })
142 .await
143 .map_err(|e| match e {
144 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),
145 sea_orm::TransactionError::Transaction(err) => err,
146 })
147 }
148
149 async fn mark_ended(&self, ctx: &ChatStorageCtx, meet_id: Uuid) -> ChatStorageResult<Meet> {
150 auth::require_authenticated(ctx).await?;
151 let db = self.db.writer();
152 let model = meet_entity::Entity::find_by_id(meet_id)
153 .one(&db)
154 .await?
155 .ok_or(ChatStorageError::MeetNotFound(meet_id))?;
156 let mut ended: meet_entity::ActiveModel = model.into();
157 ended.status = Set(meet_entity::MeetStatus::Ended);
158 ended.ended_at = Set(Some(OffsetDateTime::now_utc()));
159 ended.updated_at = Set(Some(OffsetDateTime::now_utc()));
160 let model = ended.update(&db).await?;
161 Ok(model.into())
162 }
163
164 async fn list_for_channel(
165 &self,
166 ctx: &ChatStorageCtx,
167 channel_id: Uuid,
168 limit: u32,
169 ) -> ChatStorageResult<Vec<Meet>> {
170 auth::require_channel_member(&self.db, ctx, channel_id).await?;
171 let limit = limit.clamp(1, 100);
172 let db = self.db.reader();
173 let models = meet_entity::Entity::find()
174 .filter(meet_entity::Column::ChannelId.eq(channel_id))
175 .order_by_desc(meet_entity::Column::CreatedAt)
176 .limit(Some(u64::from(limit)))
177 .all(&db)
178 .await?;
179 Ok(models.into_iter().map(Into::into).collect())
180 }
181
182 async fn list_channel_member_user_ids(
183 &self,
184 ctx: &ChatStorageCtx,
185 channel_id: Uuid,
186 ) -> ChatStorageResult<Vec<Uuid>> {
187 auth::require_channel_member(&self.db, ctx, channel_id).await?;
188 let db = self.db.reader();
189 let channel = channel_entity::Entity::find_by_id(channel_id)
190 .one(&db)
191 .await?
192 .ok_or(ChatStorageError::PermissionDenied(format!(
193 "channel {channel_id} not found"
194 )))?;
195 let Some(space_id) = channel.space_id else {
196 return Ok(Vec::new());
197 };
198 let rows = member_entity::Entity::find()
199 .filter(member_entity::Column::SpaceId.eq(space_id))
200 .filter(member_entity::Column::Status.eq(member_entity::MemberStatus::Active))
201 .all(&db)
202 .await?;
203 Ok(rows.into_iter().map(|m| m.user_id).collect())
204 }
205}
206
207impl std::fmt::Debug for SeaMeetRepository {
208 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
209 f.debug_struct("SeaMeetRepository").finish_non_exhaustive()
210 }
211}
Added chat/storagelayer/mapper/meet_participant.rs chat/storagelayer/mapper/meet_participant.rs +198 −0 View chat/storagelayer/mapper/meet_participant.rs
@@ -0,0 +1,198 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use sea_orm::{
5 ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set,
6 TransactionTrait,
7};
8use time::OffsetDateTime;
9use uuid::Uuid;
10
11use crate::auth;
12use crate::error::{ChatStorageError, ChatStorageResult};
13use crate::models::meet_participant::{LeaveReason, MeetParticipant, NewMeetParticipant};
14use crate::repository::ChatStorageCtx;
15use crate::repository::meet_participant::MeetParticipantRepository;
16
17use db::live::meet_participants as participant_entity;
18use db::live::meets as meet_entity;
19
20pub struct SeaMeetParticipantRepository {
21 db: Arc<db::conn::Db>,
22}
23
24impl SeaMeetParticipantRepository {
25 pub fn new(db: Arc<db::conn::Db>) -> Self {
26 Self { db }
27 }
28}
29
30#[async_trait]
31impl MeetParticipantRepository for SeaMeetParticipantRepository {
32 async fn join(
33 &self,
34 ctx: &ChatStorageCtx,
35 input: NewMeetParticipant,
36 ) -> ChatStorageResult<MeetParticipant> {
37 let caller = auth::require_authenticated(ctx).await?;
38 if caller != input.user_id {
39 return Err(ChatStorageError::PermissionDenied(format!(
40 "user {caller} cannot join as user {}",
41 input.user_id
42 )));
43 }
44 self.db
45 .writer()
46 .transaction(|txn| {
47 Box::pin(async move {
48 let meet = meet_entity::Entity::find_by_id(input.meet_id)
49 .one(txn)
50 .await?
51 .ok_or(ChatStorageError::MeetNotFound(input.meet_id))?;
52 if meet.status != meet_entity::MeetStatus::Active {
53 return Err(ChatStorageError::InvalidInput(
54 "meet is not active; cannot join".into(),
55 ));
56 }
57 let existing = participant_entity::Entity::find()
58 .filter(participant_entity::Column::MeetId.eq(input.meet_id))
59 .filter(participant_entity::Column::UserId.eq(input.user_id))
60 .filter(participant_entity::Column::LeftAt.is_null())
61 .one(txn)
62 .await?;
63 if existing.is_some() {
64 return Err(ChatStorageError::Conflict(format!(
65 "user {} already in meet {}",
66 input.user_id, input.meet_id
67 )));
68 }
69 let active = participant_entity::ActiveModel {
70 id: Set(Uuid::now_v7()),
71 meet_id: Set(input.meet_id),
72 user_id: Set(input.user_id),
73 joined_at: Set(input.joined_at),
74 left_at: Set(None),
75 leave_reason: Set(None),
76 connection_id: Set(input.connection_id),
77 };
78 let model = participant_entity::Entity::insert(active)
79 .exec_with_returning(txn)
80 .await?;
81 Ok(model.into())
82 })
83 })
84 .await
85 .map_err(|e| match e {
86 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),
87 sea_orm::TransactionError::Transaction(err) => err,
88 })
89 }
90
91 async fn leave(
92 &self,
93 ctx: &ChatStorageCtx,
94 participant_id: Uuid,
95 reason: LeaveReason,
96 ) -> ChatStorageResult<MeetParticipant> {
97 auth::require_authenticated(ctx).await?;
98 self.db
99 .writer()
100 .transaction(|txn| {
101 Box::pin(async move {
102 let model = participant_entity::Entity::find_by_id(participant_id)
103 .one(txn)
104 .await?
105 .ok_or(ChatStorageError::MeetParticipantNotFound(participant_id))?;
106 if model.left_at.is_some() {
107 return Err(ChatStorageError::Conflict(format!(
108 "participant {participant_id} already left"
109 )));
110 }
111 let mut active: participant_entity::ActiveModel = model.into();
112 active.left_at = Set(Some(OffsetDateTime::now_utc()));
113 active.leave_reason = Set(Some(reason.into()));
114 let model = active.update(txn).await?;
115 Ok(model.into())
116 })
117 })
118 .await
119 .map_err(|e| match e {
120 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),
121 sea_orm::TransactionError::Transaction(err) => err,
122 })
123 }
124
125 async fn list_active(
126 &self,
127 ctx: &ChatStorageCtx,
128 meet_id: Uuid,
129 ) -> ChatStorageResult<Vec<MeetParticipant>> {
130 auth::require_authenticated(ctx).await?;
131 let db = self.db.reader();
132 let models = participant_entity::Entity::find()
133 .filter(participant_entity::Column::MeetId.eq(meet_id))
134 .filter(participant_entity::Column::LeftAt.is_null())
135 .order_by_asc(participant_entity::Column::JoinedAt)
136 .all(&db)
137 .await?;
138 Ok(models.into_iter().map(Into::into).collect())
139 }
140
141 async fn list_history(
142 &self,
143 ctx: &ChatStorageCtx,
144 user_id: Uuid,
145 limit: u32,
146 ) -> ChatStorageResult<Vec<MeetParticipant>> {
147 auth::require_self(ctx, user_id)?;
148 let limit = limit.clamp(1, 100);
149 let db = self.db.reader();
150 let models = participant_entity::Entity::find()
151 .filter(participant_entity::Column::UserId.eq(user_id))
152 .order_by_desc(participant_entity::Column::JoinedAt)
153 .limit(Some(u64::from(limit)))
154 .all(&db)
155 .await?;
156 Ok(models.into_iter().map(Into::into).collect())
157 }
158
159 async fn is_active(
160 &self,
161 ctx: &ChatStorageCtx,
162 meet_id: Uuid,
163 user_id: Uuid,
164 ) -> ChatStorageResult<bool> {
165 auth::require_authenticated(ctx).await?;
166 let db = self.db.reader();
167 let found = participant_entity::Entity::find()
168 .filter(participant_entity::Column::MeetId.eq(meet_id))
169 .filter(participant_entity::Column::UserId.eq(user_id))
170 .filter(participant_entity::Column::LeftAt.is_null())
171 .one(&db)
172 .await?;
173 Ok(found.is_some())
174 }
175
176 async fn mark_all_ended(&self, ctx: &ChatStorageCtx, meet_id: Uuid) -> ChatStorageResult<u64> {
177 auth::require_authenticated(ctx).await?;
178 let db = self.db.writer();
179 let res = participant_entity::Entity::update_many()
180 .set(participant_entity::ActiveModel {
181 left_at: Set(Some(OffsetDateTime::now_utc())),
182 leave_reason: Set(Some(participant_entity::LeaveReason::RoomEnded)),
183 ..Default::default()
184 })
185 .filter(participant_entity::Column::MeetId.eq(meet_id))
186 .filter(participant_entity::Column::LeftAt.is_null())
187 .exec(&db)
188 .await?;
189 Ok(res.rows_affected)
190 }
191}
192
193impl std::fmt::Debug for SeaMeetParticipantRepository {
194 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
195 f.debug_struct("SeaMeetParticipantRepository")
196 .finish_non_exhaustive()
197 }
198}
Added chat/storagelayer/models/meet.rs chat/storagelayer/models/meet.rs +32 −0 View chat/storagelayer/models/meet.rs
@@ -0,0 +1,32 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
6#[serde(rename_all = "snake_case")]
7pub enum MeetStatus {
8 #[default]
9 Pending,
10 Active,
11 Ended,
12}
13
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15pub struct Meet {
16 pub id: Uuid,
17 pub channel_id: Uuid,
18 pub room_name: String,
19 pub status: MeetStatus,
20 pub created_by: Uuid,
21 pub started_at: Option<OffsetDateTime>,
22 pub ended_at: Option<OffsetDateTime>,
23 pub created_at: OffsetDateTime,
24 pub updated_at: Option<OffsetDateTime>,
25}
26
27#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
28pub struct NewMeet {
29 pub channel_id: Uuid,
30 pub room_name: String,
31 pub created_by: Uuid,
32}
Added chat/storagelayer/models/meet_participant.rs chat/storagelayer/models/meet_participant.rs +33 −0 View chat/storagelayer/models/meet_participant.rs
@@ -0,0 +1,33 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
6#[serde(rename_all = "snake_case")]
7pub enum LeaveReason {
8 #[default]
9 NotLeft,
10 Left,
11 Removed,
12 RoomEnded,
13 NetworkDrop,
14 Kicked,
15}
16
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18pub struct MeetParticipant {
19 pub id: Uuid,
20 pub meet_id: Uuid,
21 pub user_id: Uuid,
22 pub joined_at: OffsetDateTime,
23 pub left_at: Option<OffsetDateTime>,
24 pub leave_reason: Option<LeaveReason>,
25}
26
27#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
28pub struct NewMeetParticipant {
29 pub meet_id: Uuid,
30 pub user_id: Uuid,
31 pub joined_at: OffsetDateTime,
32 pub connection_id: Option<String>,
33}
Added chat/storagelayer/repository/meet.rs chat/storagelayer/repository/meet.rs +50 −0 View chat/storagelayer/repository/meet.rs
@@ -0,0 +1,50 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use uuid::Uuid;
4
5use crate::error::ChatStorageResult;
6use crate::models::meet::{Meet, NewMeet};
7
8use super::ChatStorageCtx;
9
10#[async_trait]
11pub trait MeetRepository: Send + Sync {
12 async fn create(&self, ctx: &ChatStorageCtx, input: NewMeet) -> ChatStorageResult<Meet>;
13
14 async fn find_active_by_channel(
15 &self,
16 ctx: &ChatStorageCtx,
17 channel_id: Uuid,
18 ) -> ChatStorageResult<Option<Meet>>;
19
20 async fn find_by_room_name(
21 &self,
22 ctx: &ChatStorageCtx,
23 room_name: &str,
24 ) -> ChatStorageResult<Option<Meet>>;
25
26 async fn find_by_id(
27 &self,
28 ctx: &ChatStorageCtx,
29 meet_id: Uuid,
30 ) -> ChatStorageResult<Option<Meet>>;
31
32 async fn mark_active(&self, ctx: &ChatStorageCtx, meet_id: Uuid) -> ChatStorageResult<Meet>;
33
34 async fn mark_ended(&self, ctx: &ChatStorageCtx, meet_id: Uuid) -> ChatStorageResult<Meet>;
35
36 async fn list_for_channel(
37 &self,
38 ctx: &ChatStorageCtx,
39 channel_id: Uuid,
40 limit: u32,
41 ) -> ChatStorageResult<Vec<Meet>>;
42
43 async fn list_channel_member_user_ids(
44 &self,
45 ctx: &ChatStorageCtx,
46 channel_id: Uuid,
47 ) -> ChatStorageResult<Vec<Uuid>>;
48}
49
50pub type DynMeetRepository = Arc<dyn MeetRepository>;
Added chat/storagelayer/repository/meet_participant.rs chat/storagelayer/repository/meet_participant.rs +48 −0 View chat/storagelayer/repository/meet_participant.rs
@@ -0,0 +1,48 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use uuid::Uuid;
4
5use crate::error::ChatStorageResult;
6use crate::models::meet_participant::{LeaveReason, MeetParticipant, NewMeetParticipant};
7
8use super::ChatStorageCtx;
9
10#[async_trait]
11pub trait MeetParticipantRepository: Send + Sync {
12 async fn join(
13 &self,
14 ctx: &ChatStorageCtx,
15 input: NewMeetParticipant,
16 ) -> ChatStorageResult<MeetParticipant>;
17
18 async fn leave(
19 &self,
20 ctx: &ChatStorageCtx,
21 participant_id: Uuid,
22 reason: LeaveReason,
23 ) -> ChatStorageResult<MeetParticipant>;
24
25 async fn list_active(
26 &self,
27 ctx: &ChatStorageCtx,
28 meet_id: Uuid,
29 ) -> ChatStorageResult<Vec<MeetParticipant>>;
30
31 async fn list_history(
32 &self,
33 ctx: &ChatStorageCtx,
34 user_id: Uuid,
35 limit: u32,
36 ) -> ChatStorageResult<Vec<MeetParticipant>>;
37
38 async fn is_active(
39 &self,
40 ctx: &ChatStorageCtx,
41 meet_id: Uuid,
42 user_id: Uuid,
43 ) -> ChatStorageResult<bool>;
44
45 async fn mark_all_ended(&self, ctx: &ChatStorageCtx, meet_id: Uuid) -> ChatStorageResult<u64>;
46}
47
48pub type DynMeetParticipantRepository = Arc<dyn MeetParticipantRepository>;