Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

chore(test): remove standalone mock and integration test files (storagelayer mock, dvcs tests, jwt tests)

zhenyi committed
parent 5979b6b commit 1a38d21 Browse files

6 files changed

+0 −2755

File tree 6
Deleted chat/storagelayer/mock/in_memory.rs
@@ -1,1454 +0,0 @@
1use std::collections::HashMap;
2use std::sync::Arc;
3use std::sync::RwLock;
4use std::time::Duration;
5
6use async_trait::async_trait;
7use bytes::Bytes;
8use time::OffsetDateTime;
9use url::Url;
10use uuid::Uuid;
11
12use crate::blob::ChatMediaStore;
13use crate::blob::media::AttachmentPath;
14use crate::error::{ChatStorageError, ChatStorageResult};
15use crate::models::attachment::{Attachment, AttachmentOwner, NewAttachment};
16use crate::models::conversation::{Conversation, ConversationKind, NewConversation};
17use crate::models::meet::{Meet, MeetStatus, NewMeet};
18use crate::models::meet_participant::{LeaveReason, MeetParticipant, NewMeetParticipant};
19use crate::models::mention::{Mention, NewMention};
20use crate::models::message::{
21 ListDirection, Message, MessageFlags, MessagePage, MessageQuery, NewMessage, Page,
22};
23use crate::models::pinned::{NewPinnedMessage, PinnedMessage};
24use crate::models::reaction::{NewReaction, Reaction};
25use crate::models::read_receipt::{NewReadReceipt, ReadReceipt};
26use crate::models::thread::Thread;
27use crate::repository::attachment::{AttachmentRepository, DynAttachmentRepository};
28use crate::repository::conversation::{ConversationRepository, DynConversationRepository};
29use crate::repository::meet::{DynMeetRepository, MeetRepository};
30use crate::repository::meet_participant::{
31 DynMeetParticipantRepository, MeetParticipantRepository,
32};
33use crate::repository::mention::{DynMentionRepository, MentionRepository};
34use crate::repository::message::{DynMessageRepository, MessageRepository};
35use crate::repository::pinned::{DynPinnedMessageRepository, PinnedMessageRepository};
36use crate::repository::reaction::{DynReactionRepository, ReactionRepository};
37use crate::repository::read_receipt::{DynReadReceiptRepository, ReadReceiptRepository};
38use crate::repository::thread::{DynThreadRepository, ThreadRepository};
39use crate::repository::{ChatStorage, ChatStorageCtx, DynChatStorage};
40
41#[derive(Debug, Default)]
42pub struct InMemoryState {
43 pub conversations: RwLock<HashMap<Uuid, Conversation>>,
44 pub creators: RwLock<HashMap<Uuid, Uuid>>,
45 pub messages: RwLock<HashMap<Uuid, Message>>,
46 pub attachments: RwLock<HashMap<Uuid, Attachment>>,
47 pub reactions: RwLock<HashMap<(Uuid, Uuid, String), Reaction>>,
48 pub mentions: RwLock<HashMap<Uuid, Mention>>,
49 pub read_receipts: RwLock<HashMap<Uuid, ReadReceipt>>,
50 pub pinned: RwLock<HashMap<(Uuid, Uuid), PinnedMessage>>,
51 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>>>,
55}
56
57pub type SharedState = Arc<InMemoryState>;
58
59fn require_user(ctx: &ChatStorageCtx) -> ChatStorageResult<Uuid> {
60 ctx.require_user()
61}
62
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
73fn is_admin(state: &SharedState, chat_id: Uuid, user_id: Uuid) -> bool {
74 state
75 .creators
76 .read()
77 .unwrap()
78 .get(&chat_id)
79 .copied()
80 .map(|c| c == user_id)
81 .unwrap_or(false)
82}
83
84fn chat_id_for_message(state: &SharedState, message_id: Uuid) -> Option<Uuid> {
85 state
86 .messages
87 .read()
88 .unwrap()
89 .get(&message_id)
90 .map(|m| m.chat_id)
91}
92
93fn chat_id_for_attachment(state: &SharedState, att: &Attachment) -> Option<Uuid> {
94 if att.owner == AttachmentOwner::ChatMessage {
95 chat_id_for_message(state, att.owner_id)
96 } else {
97 None
98 }
99}
100
101fn require_member(
102 state: &SharedState,
103 ctx: &ChatStorageCtx,
104 _chat_id: Uuid,
105) -> ChatStorageResult<Uuid> {
106 let _ = state;
107 require_user(ctx)
108}
109
110fn require_admin(
111 state: &SharedState,
112 ctx: &ChatStorageCtx,
113 chat_id: Uuid,
114) -> ChatStorageResult<Uuid> {
115 let user_id = require_user(ctx)?;
116 if !is_admin(state, chat_id, user_id) {
117 return Err(ChatStorageError::PermissionDenied(format!(
118 "user {user_id} is not admin of chat {chat_id}"
119 )));
120 }
121 Ok(user_id)
122}
123
124fn require_sender_or_admin(
125 state: &SharedState,
126 ctx: &ChatStorageCtx,
127 message_id: Uuid,
128) -> ChatStorageResult<Uuid> {
129 let user_id = require_user(ctx)?;
130 let msg = state
131 .messages
132 .read()
133 .unwrap()
134 .get(&message_id)
135 .cloned()
136 .ok_or(ChatStorageError::MessageNotFound(message_id))?;
137 if msg.sender_id == user_id {
138 return Ok(user_id);
139 }
140 if is_admin(state, msg.chat_id, user_id) {
141 return Ok(user_id);
142 }
143 Err(ChatStorageError::PermissionDenied(format!(
144 "user {user_id} is neither sender nor admin of message {message_id}"
145 )))
146}
147
148fn require_uploader_or_admin(
149 state: &SharedState,
150 ctx: &ChatStorageCtx,
151 attachment_id: Uuid,
152) -> ChatStorageResult<Uuid> {
153 let user_id = require_user(ctx)?;
154 let att = state
155 .attachments
156 .read()
157 .unwrap()
158 .get(&attachment_id)
159 .cloned()
160 .ok_or(ChatStorageError::AttachmentNotFound(attachment_id))?;
161 if att.uploaded_by == Some(user_id) {
162 return Ok(user_id);
163 }
164 let chat_id = chat_id_for_attachment(state, &att).ok_or_else(|| {
165 ChatStorageError::PermissionDenied(format!(
166 "user {user_id} cannot delete attachment {attachment_id}: owner not found"
167 ))
168 })?;
169 if is_admin(state, chat_id, user_id) {
170 return Ok(user_id);
171 }
172 Err(ChatStorageError::PermissionDenied(format!(
173 "user {user_id} cannot delete attachment {attachment_id}"
174 )))
175}
176
177#[derive(Debug)]
178pub struct InMemoryConversationRepository {
179 state: SharedState,
180}
181
182impl InMemoryConversationRepository {
183 pub fn new(state: SharedState) -> Self {
184 Self { state }
185 }
186}
187
188#[async_trait]
189impl ConversationRepository for InMemoryConversationRepository {
190 async fn create(
191 &self,
192 ctx: &ChatStorageCtx,
193 input: NewConversation,
194 ) -> ChatStorageResult<Conversation> {
195 let creator = require_user(ctx)?;
196 let now = OffsetDateTime::now_utc();
197 let conv = Conversation {
198 id: Uuid::now_v7(),
199 channel_id: Uuid::now_v7(),
200 kind: input.kind,
201 counterpart_user_id: input.counterpart_user_id,
202 title: input.title,
203 creator_user_id: creator,
204 created_at: now,
205 updated_at: Some(now),
206 };
207 self.state
208 .conversations
209 .write()
210 .unwrap()
211 .insert(conv.id, conv.clone());
212 self.state
213 .creators
214 .write()
215 .unwrap()
216 .insert(conv.id, creator);
217
218 Ok(conv)
219 }
220
221 async fn find_by_id(
222 &self,
223 ctx: &ChatStorageCtx,
224 id: Uuid,
225 ) -> ChatStorageResult<Option<Conversation>> {
226 require_member(&self.state, ctx, id)?;
227 Ok(self.state.conversations.read().unwrap().get(&id).cloned())
228 }
229
230 async fn list_by_channel(
231 &self,
232 ctx: &ChatStorageCtx,
233 channel_id: Uuid,
234 page: Page,
235 ) -> ChatStorageResult<Vec<Conversation>> {
236 require_user(ctx)?;
237 let map = self.state.conversations.read().unwrap();
238 let mut out: Vec<Conversation> = map
239 .values()
240 .filter(|c| c.channel_id == channel_id)
241 .cloned()
242 .collect();
243 out.sort_by_key(|c| std::cmp::Reverse(c.created_at));
244 let limit = page.clamped_limit() as usize;
245 Ok(out.into_iter().take(limit).collect())
246 }
247
248 async fn update_title(
249 &self,
250 ctx: &ChatStorageCtx,
251 id: Uuid,
252 title: Option<String>,
253 ) -> ChatStorageResult<()> {
254 require_admin(&self.state, ctx, id)?;
255 let mut map = self.state.conversations.write().unwrap();
256 let conv = map
257 .get_mut(&id)
258 .ok_or(ChatStorageError::ConversationNotFound(id))?;
259 conv.title = title;
260 conv.updated_at = Some(OffsetDateTime::now_utc());
261 Ok(())
262 }
263
264 async fn update_kind(
265 &self,
266 ctx: &ChatStorageCtx,
267 id: Uuid,
268 kind: ConversationKind,
269 ) -> ChatStorageResult<()> {
270 require_admin(&self.state, ctx, id)?;
271 let mut map = self.state.conversations.write().unwrap();
272 let conv = map
273 .get_mut(&id)
274 .ok_or(ChatStorageError::ConversationNotFound(id))?;
275 conv.kind = kind;
276 conv.updated_at = Some(OffsetDateTime::now_utc());
277 Ok(())
278 }
279
280 async fn delete(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {
281 require_admin(&self.state, ctx, id)?;
282 let removed = self
283 .state
284 .conversations
285 .write()
286 .unwrap()
287 .remove(&id)
288 .is_some();
289 if !removed {
290 return Err(ChatStorageError::ConversationNotFound(id));
291 }
292 self.state.creators.write().unwrap().remove(&id);
293 Ok(())
294 }
295
296 async fn exists(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<bool> {
297 Ok(self.state.conversations.read().unwrap().contains_key(&id))
298 }
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 }
325}
326
327#[derive(Debug)]
328pub struct InMemoryMessageRepository {
329 state: SharedState,
330}
331
332impl InMemoryMessageRepository {
333 pub fn new(state: SharedState) -> Self {
334 Self { state }
335 }
336}
337
338#[async_trait]
339impl MessageRepository for InMemoryMessageRepository {
340 async fn create(&self, ctx: &ChatStorageCtx, input: NewMessage) -> ChatStorageResult<Message> {
341 let caller = require_user(ctx)?;
342 require_member(&self.state, ctx, input.chat_id)?;
343 if caller != input.sender_id {
344 return Err(ChatStorageError::PermissionDenied(format!(
345 "user {caller} cannot send as user {}",
346 input.sender_id
347 )));
348 }
349 let now = OffsetDateTime::now_utc();
350 let msg = Message {
351 id: Uuid::now_v7(),
352 chat_id: input.chat_id,
353 sender_id: input.sender_id,
354 body: input.body,
355 reply_to_message_id: input.reply_to_message_id,
356 thread_id: input.thread_id,
357 is_thread_start: input.is_thread_start,
358 flags: input.flags.unwrap_or(0),
359 deleted_at: None,
360 deleted_by: None,
361 edited_at: None,
362 created_at: now,
363 };
364 self.state
365 .messages
366 .write()
367 .unwrap()
368 .insert(msg.id, msg.clone());
369 Ok(msg)
370 }
371
372 async fn find_by_id(
373 &self,
374 ctx: &ChatStorageCtx,
375 id: Uuid,
376 ) -> ChatStorageResult<Option<Message>> {
377 let chat_id = match chat_id_for_message(&self.state, id) {
378 Some(c) => c,
379 None => return Ok(None),
380 };
381 require_member(&self.state, ctx, chat_id)?;
382 Ok(self.state.messages.read().unwrap().get(&id).cloned())
383 }
384
385 async fn list_by_conversation(
386 &self,
387 ctx: &ChatStorageCtx,
388 chat_id: Uuid,
389 query: MessageQuery,
390 ) -> ChatStorageResult<MessagePage> {
391 require_member(&self.state, ctx, chat_id)?;
392 let map = self.state.messages.read().unwrap();
393 let mut items: Vec<Message> = map
394 .values()
395 .filter(|m| m.chat_id == chat_id)
396 .filter(|m| query.include_deleted || m.deleted_at.is_none())
397 .cloned()
398 .collect();
399 match query.direction {
400 ListDirection::Forward => items.sort_by_key(|m| m.created_at),
401 ListDirection::Backward => items.sort_by_key(|m| std::cmp::Reverse(m.created_at)),
402 }
403 let limit = query.clamped_limit() as usize;
404 let items: Vec<Message> = items.into_iter().take(limit).collect();
405 let next_cursor = items.last().map(|m| m.id.to_string());
406 Ok(MessagePage { items, next_cursor })
407 }
408
409 async fn list_thread(
410 &self,
411 ctx: &ChatStorageCtx,
412 chat_id: Uuid,
413 root_message_id: Uuid,
414 page: Page,
415 ) -> ChatStorageResult<MessagePage> {
416 require_member(&self.state, ctx, chat_id)?;
417 let map = self.state.messages.read().unwrap();
418 let mut items: Vec<Message> = map
419 .values()
420 .filter(|m| m.chat_id == chat_id && m.thread_id == Some(root_message_id))
421 .cloned()
422 .collect();
423 items.sort_by_key(|m| m.created_at);
424 let limit = page.clamped_limit() as usize;
425 let items: Vec<Message> = items.into_iter().take(limit).collect();
426 let next_cursor = items.last().map(|m| m.id.to_string());
427 Ok(MessagePage { items, next_cursor })
428 }
429
430 async fn edit(
431 &self,
432 ctx: &ChatStorageCtx,
433 id: Uuid,
434 new_body: String,
435 ) -> ChatStorageResult<Message> {
436 require_sender_or_admin(&self.state, ctx, id)?;
437 let now = OffsetDateTime::now_utc();
438 let mut map = self.state.messages.write().unwrap();
439 let msg = map
440 .get_mut(&id)
441 .ok_or(ChatStorageError::MessageNotFound(id))?;
442 if msg.deleted_at.is_some() {
443 return Err(ChatStorageError::MessageDeleted(id));
444 }
445 msg.body = new_body;
446 msg.edited_at = Some(now);
447 msg.flags = MessageFlags::from_bits_truncate(msg.flags)
448 .with(MessageFlags::EDITED)
449 .bits();
450 msg.deleted_by = None;
451 Ok(msg.clone())
452 }
453
454 async fn soft_delete(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {
455 let actor = require_sender_or_admin(&self.state, ctx, id)?;
456 {
457 let map = self.state.messages.read().unwrap();
458 let msg = map.get(&id).ok_or(ChatStorageError::MessageNotFound(id))?;
459 if msg.deleted_at.is_some() {
460 return Ok(());
461 }
462 }
463 let now = OffsetDateTime::now_utc();
464 let mut map = self.state.messages.write().unwrap();
465 let msg = map
466 .get_mut(&id)
467 .ok_or(ChatStorageError::MessageNotFound(id))?;
468 msg.deleted_at = Some(now);
469 msg.deleted_by = Some(actor);
470 msg.flags = MessageFlags::from_bits_truncate(msg.flags)
471 .with(MessageFlags::DELETED)
472 .bits();
473 Ok(())
474 }
475
476 async fn restore(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {
477 let chat_id =
478 chat_id_for_message(&self.state, id).ok_or(ChatStorageError::MessageNotFound(id))?;
479 require_admin(&self.state, ctx, chat_id)?;
480 {
481 let map = self.state.messages.read().unwrap();
482 let msg = map.get(&id).ok_or(ChatStorageError::MessageNotFound(id))?;
483 if msg.deleted_at.is_none() {
484 return Ok(());
485 }
486 }
487 let mut map = self.state.messages.write().unwrap();
488 let msg = map
489 .get_mut(&id)
490 .ok_or(ChatStorageError::MessageNotFound(id))?;
491 msg.deleted_at = None;
492 msg.deleted_by = None;
493 msg.flags = MessageFlags::from_bits_truncate(msg.flags)
494 .without(MessageFlags::DELETED)
495 .bits();
496 Ok(())
497 }
498
499 async fn exists(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<bool> {
500 Ok(self.state.messages.read().unwrap().contains_key(&id))
501 }
502}
503
504#[derive(Debug)]
505pub struct InMemoryAttachmentRepository {
506 state: SharedState,
507}
508
509impl InMemoryAttachmentRepository {
510 pub fn new(state: SharedState) -> Self {
511 Self { state }
512 }
513}
514
515#[async_trait]
516impl AttachmentRepository for InMemoryAttachmentRepository {
517 async fn create_metadata(
518 &self,
519 ctx: &ChatStorageCtx,
520 input: NewAttachment,
521 ) -> ChatStorageResult<Attachment> {
522 require_user(ctx)?;
523 let att = Attachment {
524 id: Uuid::now_v7(),
525 owner: input.owner,
526 owner_id: input.owner_id,
527 filename: input.filename,
528 content_type: input.content_type,
529 size: input.size,
530 url: input.url,
531 uploaded_by: input.uploaded_by,
532 created_at: OffsetDateTime::now_utc(),
533 };
534 self.state
535 .attachments
536 .write()
537 .unwrap()
538 .insert(att.id, att.clone());
539 Ok(att)
540 }
541
542 async fn find_by_id(
543 &self,
544 ctx: &ChatStorageCtx,
545 id: Uuid,
546 ) -> ChatStorageResult<Option<Attachment>> {
547 let att = self.state.attachments.read().unwrap().get(&id).cloned();
548 let Some(att) = att else {
549 return Ok(None);
550 };
551 if let Some(chat_id) = chat_id_for_attachment(&self.state, &att) {
552 require_member(&self.state, ctx, chat_id)?;
553 } else {
554 require_user(ctx)?;
555 }
556 Ok(Some(att))
557 }
558
559 async fn list_for_owner(
560 &self,
561 ctx: &ChatStorageCtx,
562 owner: AttachmentOwner,
563 owner_id: Uuid,
564 page: Page,
565 ) -> ChatStorageResult<Vec<Attachment>> {
566 if let Some(chat_id) = chat_id_for_attachment_owner_in_state(&self.state, owner, owner_id) {
567 require_member(&self.state, ctx, chat_id)?;
568 } else {
569 require_user(ctx)?;
570 }
571 let map = self.state.attachments.read().unwrap();
572 let mut items: Vec<Attachment> = map
573 .values()
574 .filter(|a| a.owner == owner && a.owner_id == owner_id)
575 .cloned()
576 .collect();
577 items.sort_by_key(|a| a.created_at);
578 let limit = page.clamped_limit() as usize;
579 Ok(items.into_iter().take(limit).collect())
580 }
581
582 async fn delete(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {
583 require_uploader_or_admin(&self.state, ctx, id)?;
584 let removed = self
585 .state
586 .attachments
587 .write()
588 .unwrap()
589 .remove(&id)
590 .is_some();
591 if !removed {
592 return Err(ChatStorageError::AttachmentNotFound(id));
593 }
594 Ok(())
595 }
596
597 async fn attach_to_message(
598 &self,
599 ctx: &ChatStorageCtx,
600 id: Uuid,
601 message_id: Uuid,
602 ) -> ChatStorageResult<Attachment> {
603 require_uploader_or_admin(&self.state, ctx, id)?;
604 let mut map = self.state.attachments.write().unwrap();
605 let Some(att) = map.get(&id).cloned() else {
606 return Err(ChatStorageError::AttachmentNotFound(id));
607 };
608 if !att.owner_id.is_nil() {
609 return Err(ChatStorageError::Conflict(format!(
610 "attachment {id} is already attached to a message"
611 )));
612 }
613 if !self
614 .state
615 .messages
616 .read()
617 .unwrap()
618 .contains_key(&message_id)
619 {
620 return Err(ChatStorageError::MessageNotFound(message_id));
621 }
622 let attached = Attachment {
623 owner_id: message_id,
624 ..att
625 };
626 map.insert(id, attached.clone());
627 Ok(attached)
628 }
629}
630
631fn chat_id_for_attachment_owner_in_state(
632 state: &SharedState,
633 owner: AttachmentOwner,
634 owner_id: Uuid,
635) -> Option<Uuid> {
636 match owner {
637 AttachmentOwner::ChatMessage => chat_id_for_message(state, owner_id),
638 _ => None,
639 }
640}
641
642#[derive(Debug)]
643pub struct InMemoryReactionRepository {
644 state: SharedState,
645}
646
647impl InMemoryReactionRepository {
648 pub fn new(state: SharedState) -> Self {
649 Self { state }
650 }
651}
652
653#[async_trait]
654impl ReactionRepository for InMemoryReactionRepository {
655 async fn add_reaction(
656 &self,
657 ctx: &ChatStorageCtx,
658 input: NewReaction,
659 ) -> ChatStorageResult<Reaction> {
660 let caller = require_user(ctx)?;
661 if caller != input.user_id {
662 return Err(ChatStorageError::PermissionDenied(format!(
663 "user {caller} cannot react as user {}",
664 input.user_id
665 )));
666 }
667 let chat_id = chat_id_for_message(&self.state, input.message_id)
668 .ok_or(ChatStorageError::MessageNotFound(input.message_id))?;
669 require_member(&self.state, ctx, chat_id)?;
670 let r = Reaction {
671 id: Uuid::now_v7(),
672 message_id: input.message_id,
673 user_id: input.user_id,
674 emoji: input.emoji,
675 created_at: OffsetDateTime::now_utc(),
676 };
677 self.state
678 .reactions
679 .write()
680 .unwrap()
681 .insert((r.message_id, r.user_id, r.emoji.clone()), r.clone());
682 Ok(r)
683 }
684
685 async fn remove_reaction(
686 &self,
687 ctx: &ChatStorageCtx,
688 message_id: Uuid,
689 emoji: &str,
690 ) -> ChatStorageResult<()> {
691 let user_id = require_user(ctx)?;
692 let chat_id = chat_id_for_message(&self.state, message_id)
693 .ok_or(ChatStorageError::MessageNotFound(message_id))?;
694 require_member(&self.state, ctx, chat_id)?;
695 let removed = self
696 .state
697 .reactions
698 .write()
699 .unwrap()
700 .remove(&(message_id, user_id, emoji.to_string()))
701 .is_some();
702 if !removed {
703 return Err(ChatStorageError::ReactionNotFound(message_id));
704 }
705 Ok(())
706 }
707
708 async fn list_reactions(
709 &self,
710 ctx: &ChatStorageCtx,
711 message_id: Uuid,
712 ) -> ChatStorageResult<Vec<Reaction>> {
713 let chat_id = chat_id_for_message(&self.state, message_id)
714 .ok_or(ChatStorageError::MessageNotFound(message_id))?;
715 require_member(&self.state, ctx, chat_id)?;
716 let map = self.state.reactions.read().unwrap();
717 Ok(map
718 .values()
719 .filter(|r| r.message_id == message_id)
720 .cloned()
721 .collect())
722 }
723}
724
725#[derive(Debug)]
726pub struct InMemoryMentionRepository {
727 state: SharedState,
728}
729
730impl InMemoryMentionRepository {
731 pub fn new(state: SharedState) -> Self {
732 Self { state }
733 }
734}
735
736#[async_trait]
737impl MentionRepository for InMemoryMentionRepository {
738 async fn add_mention(
739 &self,
740 ctx: &ChatStorageCtx,
741 input: NewMention,
742 ) -> ChatStorageResult<Mention> {
743 let chat_id = chat_id_for_message(&self.state, input.message_id)
744 .ok_or(ChatStorageError::MessageNotFound(input.message_id))?;
745 require_member(&self.state, ctx, chat_id)?;
746 let m = Mention {
747 id: Uuid::now_v7(),
748 message_id: input.message_id,
749 user_id: input.user_id,
750 role_id: input.role_id,
751 channel_id: input.channel_id,
752 is_everyone: input.is_everyone,
753 is_here: input.is_here,
754 created_at: OffsetDateTime::now_utc(),
755 };
756 self.state.mentions.write().unwrap().insert(m.id, m.clone());
757 Ok(m)
758 }
759
760 async fn list_for_message(
761 &self,
762 ctx: &ChatStorageCtx,
763 message_id: Uuid,
764 ) -> ChatStorageResult<Vec<Mention>> {
765 let chat_id = chat_id_for_message(&self.state, message_id)
766 .ok_or(ChatStorageError::MessageNotFound(message_id))?;
767 require_member(&self.state, ctx, chat_id)?;
768 let map = self.state.mentions.read().unwrap();
769 Ok(map
770 .values()
771 .filter(|m| m.message_id == message_id)
772 .cloned()
773 .collect())
774 }
775
776 async fn list_for_user(
777 &self,
778 ctx: &ChatStorageCtx,
779 user_id: Uuid,
780 limit: u32,
781 ) -> ChatStorageResult<Vec<Mention>> {
782 let caller = require_user(ctx)?;
783 if caller != user_id {
784 return Err(ChatStorageError::PermissionDenied(format!(
785 "user {caller} cannot list mentions for user {user_id}"
786 )));
787 }
788 let map = self.state.mentions.read().unwrap();
789 let mut items: Vec<Mention> = map
790 .values()
791 .filter(|m| m.user_id == Some(user_id) || m.is_everyone || m.is_here)
792 .cloned()
793 .collect();
794 items.sort_by_key(|m| std::cmp::Reverse(m.created_at));
795 let limit = limit.clamp(1, crate::MAX_PAGE_LIMIT) as usize;
796 Ok(items.into_iter().take(limit).collect())
797 }
798}
799
800#[derive(Debug)]
801pub struct InMemoryReadReceiptRepository {
802 state: SharedState,
803}
804
805impl InMemoryReadReceiptRepository {
806 pub fn new(state: SharedState) -> Self {
807 Self { state }
808 }
809}
810
811#[async_trait]
812impl ReadReceiptRepository for InMemoryReadReceiptRepository {
813 async fn upsert(
814 &self,
815 ctx: &ChatStorageCtx,
816 input: NewReadReceipt,
817 ) -> ChatStorageResult<ReadReceipt> {
818 let caller = require_user(ctx)?;
819 if caller != input.user_id {
820 return Err(ChatStorageError::PermissionDenied(format!(
821 "user {caller} cannot mark read for user {}",
822 input.user_id
823 )));
824 }
825 require_member(&self.state, ctx, input.chat_id)?;
826 let mut map = self.state.read_receipts.write().unwrap();
827 let now = OffsetDateTime::now_utc();
828 if let Some(existing) = map.values_mut().find(|r| {
829 r.chat_id == input.chat_id
830 && r.user_id == input.user_id
831 && r.thread_id == input.thread_id
832 }) {
833 existing.last_read_message_id = input.last_read_message_id;
834 existing.updated_at = now;
835 return Ok(existing.clone());
836 }
837 let r = ReadReceipt {
838 id: Uuid::now_v7(),
839 chat_id: input.chat_id,
840 user_id: input.user_id,
841 thread_id: input.thread_id,
842 last_read_message_id: input.last_read_message_id,
843 muted: false,
844 updated_at: now,
845 };
846 map.insert(r.id, r.clone());
847 Ok(r)
848 }
849
850 async fn list_for_chat(
851 &self,
852 ctx: &ChatStorageCtx,
853 chat_id: Uuid,
854 ) -> ChatStorageResult<Vec<ReadReceipt>> {
855 require_member(&self.state, ctx, chat_id)?;
856 let map = self.state.read_receipts.read().unwrap();
857 Ok(map
858 .values()
859 .filter(|r| r.chat_id == chat_id)
860 .cloned()
861 .collect())
862 }
863
864 async fn find(
865 &self,
866 ctx: &ChatStorageCtx,
867 chat_id: Uuid,
868 _user_id: Uuid,
869 thread_id: Option<Uuid>,
870 ) -> ChatStorageResult<Option<ReadReceipt>> {
871 require_member(&self.state, ctx, chat_id)?;
872 let map = self.state.read_receipts.read().unwrap();
873 let caller = require_user(ctx)?;
874 let r = map
875 .values()
876 .find(|r| r.chat_id == chat_id && r.user_id == caller && r.thread_id == thread_id)
877 .cloned();
878 Ok(r)
879 }
880
881 async fn set_muted(
882 &self,
883 ctx: &ChatStorageCtx,
884 chat_id: Uuid,
885 muted: bool,
886 ) -> ChatStorageResult<()> {
887 require_member(&self.state, ctx, chat_id)?;
888 let caller = require_user(ctx)?;
889 let mut map = self.state.read_receipts.write().unwrap();
890 let target = map
891 .values_mut()
892 .find(|r| r.chat_id == chat_id && r.user_id == caller && r.thread_id.is_none());
893 match target {
894 Some(r) => {
895 r.muted = muted;
896 r.updated_at = OffsetDateTime::now_utc();
897 }
898 None => {
899 let r = ReadReceipt {
900 id: Uuid::now_v7(),
901 chat_id,
902 user_id: caller,
903 thread_id: None,
904 last_read_message_id: None,
905 muted,
906 updated_at: OffsetDateTime::now_utc(),
907 };
908 map.insert(r.id, r);
909 }
910 }
911 Ok(())
912 }
913}
914
915#[derive(Debug)]
916pub struct InMemoryPinnedMessageRepository {
917 state: SharedState,
918}
919
920impl InMemoryPinnedMessageRepository {
921 pub fn new(state: SharedState) -> Self {
922 Self { state }
923 }
924}
925
926#[async_trait]
927impl PinnedMessageRepository for InMemoryPinnedMessageRepository {
928 async fn pin(
929 &self,
930 ctx: &ChatStorageCtx,
931 input: NewPinnedMessage,
932 ) -> ChatStorageResult<PinnedMessage> {
933 let caller = require_user(ctx)?;
934 if caller != input.pinned_by {
935 return Err(ChatStorageError::PermissionDenied(format!(
936 "user {caller} cannot pin as user {}",
937 input.pinned_by
938 )));
939 }
940 require_admin(&self.state, ctx, input.chat_id)?;
941 let p = PinnedMessage {
942 id: Uuid::now_v7(),
943 chat_id: input.chat_id,
944 message_id: input.message_id,
945 pinned_by: input.pinned_by,
946 created_at: OffsetDateTime::now_utc(),
947 };
948 self.state
949 .pinned
950 .write()
951 .unwrap()
952 .insert((p.chat_id, p.message_id), p.clone());
953 Ok(p)
954 }
955
956 async fn unpin(
957 &self,
958 ctx: &ChatStorageCtx,
959 chat_id: Uuid,
960 message_id: Uuid,
961 ) -> ChatStorageResult<()> {
962 require_admin(&self.state, ctx, chat_id)?;
963 let removed = self
964 .state
965 .pinned
966 .write()
967 .unwrap()
968 .remove(&(chat_id, message_id))
969 .is_some();
970 if !removed {
971 return Err(ChatStorageError::PinnedMessageNotFound(message_id));
972 }
973 Ok(())
974 }
975
976 async fn list_for_chat(
977 &self,
978 ctx: &ChatStorageCtx,
979 chat_id: Uuid,
980 ) -> ChatStorageResult<Vec<PinnedMessage>> {
981 require_member(&self.state, ctx, chat_id)?;
982 let map = self.state.pinned.read().unwrap();
983 Ok(map
984 .values()
985 .filter(|p| p.chat_id == chat_id)
986 .cloned()
987 .collect())
988 }
989}
990
991#[derive(Debug)]
992pub struct InMemoryThreadRepository {
993 state: SharedState,
994}
995
996impl InMemoryThreadRepository {
997 pub fn new(state: SharedState) -> Self {
998 Self { state }
999 }
1000}
1001
1002#[async_trait]
1003impl ThreadRepository for InMemoryThreadRepository {
1004 async fn increment_reply(
1005 &self,
1006 ctx: &ChatStorageCtx,
1007 chat_id: Uuid,
1008 root_message_id: Uuid,
1009 reply_by: Uuid,
1010 at: OffsetDateTime,
1011 ) -> ChatStorageResult<Thread> {
1012 require_member(&self.state, ctx, chat_id)?;
1013 let mut map = self.state.threads.write().unwrap();
1014 let entry = map.entry((chat_id, root_message_id)).or_insert(Thread {
1015 chat_id,
1016 root_message_id,
1017 reply_count: 0,
1018 last_reply_at: None,
1019 last_reply_by: None,
1020 });
1021 entry.reply_count = entry.reply_count.saturating_add(1);
1022 entry.last_reply_at = Some(at);
1023 entry.last_reply_by = Some(reply_by);
1024 Ok(entry.clone())
1025 }
1026
1027 async fn find(
1028 &self,
1029 _ctx: &ChatStorageCtx,
1030 chat_id: Uuid,
1031 root_message_id: Uuid,
1032 ) -> ChatStorageResult<Option<Thread>> {
1033 Ok(self
1034 .state
1035 .threads
1036 .read()
1037 .unwrap()
1038 .get(&(chat_id, root_message_id))
1039 .cloned())
1040 }
1041
1042 async fn list_for_chat(
1043 &self,
1044 ctx: &ChatStorageCtx,
1045 chat_id: Uuid,
1046 ) -> ChatStorageResult<Vec<Thread>> {
1047 require_member(&self.state, ctx, chat_id)?;
1048 let map = self.state.threads.read().unwrap();
1049 Ok(map
1050 .values()
1051 .filter(|t| t.chat_id == chat_id)
1052 .cloned()
1053 .collect())
1054 }
1055}
1056
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
1347pub struct InMemoryChatStorage {
1348 state: SharedState,
1349 blobs: Arc<dyn ChatMediaStore>,
1350}
1351
1352impl std::fmt::Debug for InMemoryChatStorage {
1353 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1354 f.debug_struct("InMemoryChatStorage")
1355 .field("state", &self.state)
1356 .field("blobs", &"<dyn ChatMediaStore>")
1357 .finish()
1358 }
1359}
1360
1361impl InMemoryChatStorage {
1362 pub fn new(blobs: Arc<dyn ChatMediaStore>) -> Self {
1363 Self {
1364 state: Arc::new(InMemoryState::default()),
1365 blobs,
1366 }
1367 }
1368
1369 pub fn new_shared() -> DynChatStorage {
1370 let blobs: Arc<dyn ChatMediaStore> = Arc::new(NullChatMedia);
1371 Arc::new(Self::new(blobs))
1372 }
1373}
1374
1375#[derive(Debug)]
1376pub struct NullChatMedia;
1377
1378fn unsupported<T>() -> ChatStorageResult<T> {
1379 Err(ChatStorageError::Blob(storage::StorageError::Unsupported(
1380 "null backend".to_string(),
1381 )))
1382}
1383
1384#[async_trait]
1385impl ChatMediaStore for NullChatMedia {
1386 async fn put_attachment_bytes(
1387 &self,
1388 _owner: AttachmentOwner,
1389 _owner_id: Uuid,
1390 _filename: &str,
1391 _bytes: Bytes,
1392 ) -> ChatStorageResult<AttachmentPath> {
1393 unsupported()
1394 }
1395 async fn get_attachment_bytes(&self, _path: &AttachmentPath) -> ChatStorageResult<Bytes> {
1396 unsupported()
1397 }
1398 async fn delete_attachment(&self, _path: &AttachmentPath) -> ChatStorageResult<()> {
1399 unsupported()
1400 }
1401 async fn presign_upload(
1402 &self,
1403 _owner: AttachmentOwner,
1404 _owner_id: Uuid,
1405 _filename: &str,
1406 _ttl: Duration,
1407 ) -> ChatStorageResult<Url> {
1408 unsupported()
1409 }
1410 async fn presign_download(
1411 &self,
1412 _path: &AttachmentPath,
1413 _ttl: Duration,
1414 ) -> ChatStorageResult<Url> {
1415 unsupported()
1416 }
1417}
1418
1419#[async_trait]
1420impl ChatStorage for InMemoryChatStorage {
1421 fn conversations(&self) -> DynConversationRepository {
1422 Arc::new(InMemoryConversationRepository::new(self.state.clone()))
1423 }
1424 fn messages(&self) -> DynMessageRepository {
1425 Arc::new(InMemoryMessageRepository::new(self.state.clone()))
1426 }
1427 fn attachments(&self) -> DynAttachmentRepository {
1428 Arc::new(InMemoryAttachmentRepository::new(self.state.clone()))
1429 }
1430 fn reactions(&self) -> DynReactionRepository {
1431 Arc::new(InMemoryReactionRepository::new(self.state.clone()))
1432 }
1433 fn mentions(&self) -> DynMentionRepository {
1434 Arc::new(InMemoryMentionRepository::new(self.state.clone()))
1435 }
1436 fn reads(&self) -> DynReadReceiptRepository {
1437 Arc::new(InMemoryReadReceiptRepository::new(self.state.clone()))
1438 }
1439 fn pinned(&self) -> DynPinnedMessageRepository {
1440 Arc::new(InMemoryPinnedMessageRepository::new(self.state.clone()))
1441 }
1442 fn threads(&self) -> DynThreadRepository {
1443 Arc::new(InMemoryThreadRepository::new(self.state.clone()))
1444 }
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 }
1451 fn blobs(&self) -> Arc<dyn ChatMediaStore> {
1452 self.blobs.clone()
1453 }
1454}
Deleted chat/storagelayer/mock.rs
@@ -1,2 +0,0 @@
1#[cfg(feature = "mock")]
2pub mod in_memory;
Deleted dvcs/gitauth/testing.rs
@@ -1,24 +0,0 @@
1use db::conn::sea_orm::{
2 ConnectionTrait, Database, DatabaseBackend, DatabaseConnection, EntityTrait, Schema,
3};
4
5async fn create_table<E: EntityTrait>(db: &DatabaseConnection) {
6 let stmt = Schema::new(DatabaseBackend::Sqlite).create_table_from_entity(E::default());
7 db.execute(&stmt).await.expect("create table");
8}
9
10pub async fn setup_db() -> DatabaseConnection {
11 let db = Database::connect("sqlite::memory:")
12 .await
13 .expect("connect sqlite memory");
14 create_table::<db::user::users::Entity>(&db).await;
15 create_table::<db::space::spaces::Entity>(&db).await;
16 create_table::<db::channel::channels::Entity>(&db).await;
17 create_table::<db::space::space_members::Entity>(&db).await;
18 create_table::<db::space::space_roles::Entity>(&db).await;
19 create_table::<db::space::space_member_roles::Entity>(&db).await;
20 create_table::<db::space::space_role_permissions::Entity>(&db).await;
21 create_table::<db::repo::repo_collaborations::Entity>(&db).await;
22 create_table::<db::repo::repos::Entity>(&db).await;
23 db
24}
Deleted dvcs/gitsmart/tests/smart.rs
@@ -1,661 +0,0 @@
1use std::net::SocketAddr;
2use std::path::Path;
3use std::sync::Arc;
4use std::time::Duration;
5
6use config::git::SmartConfig;
7use config::jwt::JwtSettings;
8use config::storage::{LocalStorageSettings, StorageSettings};
9use db::channel::channels;
10use db::conn::Db;
11use db::conn::sea_orm::{
12 ActiveModelTrait, ConnectionTrait, Database, DatabaseBackend, DatabaseConnection, EntityTrait,
13 Schema, Set,
14};
15use db::repo::repos;
16use db::space::spaces;
17use db::user::users;
18use gitauth::GitAuthService;
19use gitsmart::SmartService;
20use gitsmart::handler::router;
21use jwt::{JwtState, SigningKeyCache};
22use password_hash::rand_core::OsRng;
23use password_hash::{PasswordHasher, SaltString};
24use tempfile::TempDir;
25use time::OffsetDateTime;
26use tokio::sync::RwLock;
27use uuid::Uuid;
28
29fn now() -> OffsetDateTime {
30 OffsetDateTime::now_utc()
31}
32
33fn hash_for(password: &str) -> String {
34 let salt = SaltString::generate(&mut OsRng);
35 argon2::Argon2::default()
36 .hash_password(password.as_bytes(), &salt)
37 .expect("hash")
38 .to_string()
39}
40
41async fn create_table<E: EntityTrait>(db: &DatabaseConnection) {
42 let stmt = Schema::new(DatabaseBackend::Sqlite).create_table_from_entity(E::default());
43 db.execute(&stmt).await.expect("create table");
44}
45
46async fn setup_db() -> DatabaseConnection {
47 let db = Database::connect("sqlite::memory:")
48 .await
49 .expect("connect sqlite memory");
50 create_table::<users::Entity>(&db).await;
51 create_table::<spaces::Entity>(&db).await;
52 create_table::<channels::Entity>(&db).await;
53 create_table::<db::space::space_members::Entity>(&db).await;
54 create_table::<db::space::space_roles::Entity>(&db).await;
55 create_table::<db::space::space_member_roles::Entity>(&db).await;
56 create_table::<db::space::space_role_permissions::Entity>(&db).await;
57 create_table::<db::repo::repo_collaborations::Entity>(&db).await;
58 create_table::<repos::Entity>(&db).await;
59 db
60}
61
62async fn insert_user(
63 db: &impl ConnectionTrait,
64 uid: Uuid,
65 username: &str,
66 passwd_hash: Option<&str>,
67) {
68 users::ActiveModel {
69 uid: Set(uid),
70 username: Set(username.to_string()),
71 lower_name: Set(username.to_lowercase()),
72 display_name: Set(username.to_string()),
73 avatar: Set(String::new()),
74 is_active: Set(true),
75 is_admin: Set(false),
76 is_restricted: Set(false),
77 prohibit_login: Set(false),
78 visibility: Set(users::ProfileVisibility::Public),
79 bot: Set(false),
80 system: Set(false),
81 totp_enabled: Set(false),
82 version: Set(1),
83 passwd_hash: Set(passwd_hash.map(|s| s.to_string())),
84 created_at: Set(now()),
85 updated_at: Set(now()),
86 ..Default::default()
87 }
88 .insert(db)
89 .await
90 .expect("insert user");
91}
92
93async fn insert_space(
94 db: &impl ConnectionTrait,
95 id: Uuid,
96 lower_name: &str,
97 owner_id: Uuid,
98 is_public: bool,
99) {
100 spaces::ActiveModel {
101 id: Set(id),
102 lower_name: Set(lower_name.to_string()),
103 name: Set(lower_name.to_string()),
104 owner_id: Set(owner_id),
105 public: Set(is_public),
106 is_active: Set(true),
107 allow_join: Set(true),
108 allow_invite: Set(true),
109 member_count: Set(0),
110 allow_join_level: Set(spaces::AllowJoinLevel::None),
111 member_must_enable_2fa: Set(false),
112 created_at: Set(now()),
113 updated_at: Set(now()),
114 ..Default::default()
115 }
116 .insert(db)
117 .await
118 .expect("insert space");
119}
120
121async fn insert_channel(
122 db: &impl ConnectionTrait,
123 id: Uuid,
124 space_id: Uuid,
125 name: &str,
126 created_by: Uuid,
127) {
128 channels::ActiveModel {
129 id: Set(id),
130 space_id: Set(Some(space_id)),
131 name: Set(name.to_string()),
132 position: Set(0),
133 is_archived: Set(false),
134 created_at: Set(now()),
135 updated_at: Set(now()),
136 created_by: Set(created_by),
137 ..Default::default()
138 }
139 .insert(db)
140 .await
141 .expect("insert channel");
142}
143
144async fn insert_repo(
145 db: &impl ConnectionTrait,
146 id: Uuid,
147 channel_id: Uuid,
148 name: &str,
149 is_private: bool,
150) {
151 repos::ActiveModel {
152 id: Set(id),
153 channel_id: Set(channel_id),
154 lower_name: Set(name.to_lowercase()),
155 name: Set(name.to_string()),
156 default_branch: Set("main".to_string()),
157 nums_watch: Set(0),
158 nums_star: Set(0),
159 nums_pr: Set(0),
160 nums_commits: Set(0),
161 nums_branches: Set(0),
162 nums_tags: Set(0),
163 enable_lfs: Set(true),
164 is_private: Set(is_private),
165 is_archived: Set(false),
166 size: Set(0),
167 git_size: Set(0),
168 lfs_size: Set(0),
169 allow_rebase: Set(true),
170 allow_rebase_explicit: Set(true),
171 allow_squash_merge: Set(true),
172 allow_merge_commits: Set(true),
173 created_at: Set(now()),
174 object_format: Set("sha1".to_string()),
175 ..Default::default()
176 }
177 .insert(db)
178 .await
179 .expect("insert repo");
180}
181
182async fn git(dir: &Path, args: &[&str]) -> String {
183 let out = tokio::process::Command::new("git")
184 .args(args)
185 .current_dir(dir)
186 .env("GIT_TERMINAL_PROMPT", "0")
187 .output()
188 .await
189 .expect("run git");
190 assert!(
191 out.status.success(),
192 "git {:?} failed: {}",
193 args,
194 String::from_utf8_lossy(&out.stderr)
195 );
196 String::from_utf8_lossy(&out.stdout).into_owned()
197}
198
199fn basic_auth(username: &str, password: &str) -> String {
200 use base64::Engine;
201 format!(
202 "Basic {}",
203 base64::engine::general_purpose::STANDARD.encode(format!("{username}:{password}"))
204 )
205}
206
207struct TestServer {
208 _tmp: TempDir,
209 addr: SocketAddr,
210 base: String,
211 bare: std::path::PathBuf,
212 db: DatabaseConnection,
213 git_root: std::path::PathBuf,
214 space_id: Uuid,
215 channel_id: Uuid,
216}
217
218async fn start_server() -> TestServer {
219 let tmp = TempDir::new().expect("tempdir");
220 let root = tmp.path().to_path_buf();
221
222 let git_root = root.join("git");
223 let lfs_root = root.join("lfs");
224 let storage_root = root.join("storage");
225 std::fs::create_dir_all(&git_root).expect("git root");
226 std::fs::create_dir_all(&lfs_root).expect("lfs root");
227 std::fs::create_dir_all(&storage_root).expect("storage root");
228
229 let conn = setup_db().await;
230 let db = Db::from_connection(conn.clone());
231
232 let alice = Uuid::now_v7();
233 let space_id = Uuid::now_v7();
234 let channel_id = Uuid::now_v7();
235 let repo_id = Uuid::now_v7();
236
237 insert_user(&conn, alice, "alice", Some(hash_for("secret").as_str())).await;
238 insert_space(&conn, space_id, "acme", alice, true).await;
239 insert_channel(&conn, channel_id, space_id, "dev", alice).await;
240 insert_repo(&conn, repo_id, channel_id, "widget", false).await;
241
242 let bare = git_root.join(format!("{}/{}.git", channel_id, repo_id));
243 std::fs::create_dir_all(bare.parent().expect("bare parent")).expect("bare dirs");
244 git(&root, &["init", "--bare", bare.to_str().expect("bare str")]).await;
245
246 let mut settings = StorageSettings::default();
247 settings.set_local(
248 LocalStorageSettings::builder()
249 .root_dir(storage_root.to_path_buf())
250 .build(),
251 );
252 let storage = storage::repo::from_config(&settings).expect("storage");
253
254 let jwt = JwtState::new(
255 conn.clone(),
256 JwtSettings::default(),
257 Arc::new(RwLock::new(SigningKeyCache::new())),
258 );
259 let auth = GitAuthService::new(db, None, jwt);
260 let config = SmartConfig {
261 git_storage_root: git_root.to_str().expect("git root str").to_string(),
262 lfs_storage_root: lfs_root.to_str().expect("lfs root str").to_string(),
263 object_cache_prefix: "gitsmart".to_string(),
264 object_cache_ttl: 300,
265 };
266 let svc = SmartService::new(auth, config, storage)
267 .await
268 .expect("smart service");
269 let app = router(Arc::new(svc));
270
271 let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
272 .await
273 .expect("bind");
274 let addr = listener.local_addr().expect("addr");
275 tokio::spawn(async move {
276 axum::serve(listener, app).await.expect("serve");
277 });
278
279 let base = format!("http://{addr}/acme/widget.git");
280 TestServer {
281 _tmp: tmp,
282 addr,
283 base,
284 bare,
285 db: conn,
286 git_root,
287 space_id,
288 channel_id,
289 }
290}
291
292async fn seed_bare(root: &Path, bare: &Path) {
293 let src = root.join("seed-src");
294 std::fs::create_dir_all(&src).expect("src dir");
295 git(&src, &["init", "-q"]).await;
296 git(&src, &["config", "user.email", "alice@example.com"]).await;
297 git(&src, &["config", "user.name", "Alice"]).await;
298 std::fs::write(src.join("hello.txt"), b"hello world\n").expect("write file");
299 git(&src, &["add", "."]).await;
300 git(&src, &["commit", "-q", "-m", "init"]).await;
301 git(
302 &src,
303 &["remote", "add", "origin", bare.to_str().expect("bare str")],
304 )
305 .await;
306 git(&src, &["push", "-q", "origin", "HEAD:main"]).await;
307 git(
308 root,
309 &[
310 "--git-dir",
311 bare.to_str().expect("bare str"),
312 "update-server-info",
313 ],
314 )
315 .await;
316}
317
318#[tokio::test]
319async fn smart_clone_and_push() {
320 let srv = start_server().await;
321
322 let src = srv._tmp.path().join("src");
323 std::fs::create_dir_all(&src).expect("src dir");
324 git(&src, &["init", "-q"]).await;
325 git(&src, &["config", "user.email", "alice@example.com"]).await;
326 git(&src, &["config", "user.name", "Alice"]).await;
327 std::fs::write(src.join("hello.txt"), b"hello world\n").expect("write file");
328 git(&src, &["add", "."]).await;
329 git(&src, &["commit", "-q", "-m", "init"]).await;
330 git(
331 &src,
332 &[
333 "remote",
334 "add",
335 "origin",
336 &format!("http://alice:secret@{}/acme/widget.git", srv.addr),
337 ],
338 )
339 .await;
340 git(&src, &["push", "-q", "-u", "origin", "main"]).await;
341
342 let dst = srv._tmp.path().join("dst");
343 git(
344 &srv._tmp.path(),
345 &[
346 "clone",
347 "-q",
348 &format!("http://alice:secret@{}/acme/widget.git", srv.addr),
349 dst.to_str().expect("dst str"),
350 ],
351 )
352 .await;
353 let content = std::fs::read_to_string(dst.join("hello.txt")).expect("read cloned file");
354 assert_eq!(content, "hello world\n");
355}
356
357#[tokio::test]
358async fn dumb_info_refs_text() {
359 let srv = start_server().await;
360 seed_bare(&srv._tmp.path().to_path_buf(), &srv.bare).await;
361 let client = reqwest::Client::new();
362 let resp = client
363 .get(format!("{}/info/refs", srv.base))
364 .send()
365 .await
366 .expect("get info/refs");
367 assert_eq!(resp.status(), reqwest::StatusCode::OK, "dumb info/refs");
368 let body = resp.text().await.expect("body");
369 assert!(body.contains("refs/heads/main"), "body: {body}");
370}
371
372#[tokio::test]
373async fn lfs_put_get_roundtrip() {
374 let srv = start_server().await;
375 let client = reqwest::Client::new();
376 let data = b"large binary payload".to_vec();
377 let oid = gitlfs::LfsOid::compute(&data);
378 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());
379
380 let put = client
381 .put(&url)
382 .header("Authorization", basic_auth("alice", "secret"))
383 .body(data.clone())
384 .send()
385 .await
386 .expect("lfs put");
387 assert_eq!(put.status(), reqwest::StatusCode::OK, "lfs put failed");
388
389 let get = client
390 .get(&url)
391 .header("Authorization", basic_auth("alice", "secret"))
392 .send()
393 .await
394 .expect("lfs get");
395 assert_eq!(get.status(), reqwest::StatusCode::OK);
396 let got = get.bytes().await.expect("lfs bytes");
397 assert_eq!(got.as_ref(), data.as_slice());
398}
399
400#[tokio::test]
401async fn lfs_batch_upload_action() {
402 let srv = start_server().await;
403 let client = reqwest::Client::new();
404 let data = b"batch test payload".to_vec();
405 let oid = gitlfs::LfsOid::compute(&data);
406 let body = serde_json::json!({
407 "operation": "upload",
408 "transfers": ["basic"],
409 "objects": [{"oid": oid.as_str(), "size": data.len() as i64}]
410 });
411 let resp = client
412 .post(format!("{}/info/lfs/objects/batch", srv.base))
413 .header("Authorization", basic_auth("alice", "secret"))
414 .header("Content-Type", "application/vnd.git-lfs+json")
415 .body(body.to_string())
416 .send()
417 .await
418 .expect("lfs batch");
419 assert_eq!(resp.status(), reqwest::StatusCode::OK, "lfs batch failed");
420 let json: serde_json::Value = resp.json().await.expect("batch json");
421 let href = json["objects"][0]["actions"]["upload"]["href"]
422 .as_str()
423 .expect("upload href");
424 assert!(
425 href.contains(&format!("/info/lfs/objects/{}", oid.as_str())),
426 "href: {href}"
427 );
428}
429
430#[tokio::test]
431async fn anonymous_cannot_read_private() {
432 let srv = start_server().await;
433 let client = reqwest::Client::new();
434
435 let repo_id = Uuid::now_v7();
436 insert_repo(&srv.db, repo_id, srv.channel_id, "secret", true).await;
437
438 let bare = srv
439 .git_root
440 .join(format!("{}/{}.git", srv.channel_id, repo_id));
441 std::fs::create_dir_all(bare.parent().expect("bare parent")).expect("bare dirs");
442 git(
443 &srv._tmp.path(),
444 &["init", "--bare", bare.to_str().expect("bare str")],
445 )
446 .await;
447
448 let resp = client
449 .get(format!(
450 "http://{}/acme/secret.git/info/refs?service=git-upload-pack",
451 srv.addr
452 ))
453 .send()
454 .await
455 .expect("private info refs");
456 assert_eq!(resp.status(), reqwest::StatusCode::UNAUTHORIZED);
457}
458
459#[tokio::test]
460async fn unknown_repo_404() {
461 let srv = start_server().await;
462 let client = reqwest::Client::new();
463 let resp = client
464 .get(format!(
465 "http://{}/acme/ghost.git/info/refs?service=git-upload-pack",
466 srv.addr
467 ))
468 .send()
469 .await
470 .expect("ghost info refs");
471 assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND);
472}
473
474#[tokio::test]
475async fn smart_advertisement_responds() {
476 let srv = start_server().await;
477 let client = reqwest::Client::new();
478 let resp = client
479 .get(format!("{}/info/refs?service=git-upload-pack", srv.base))
480 .timeout(Duration::from_secs(10))
481 .send()
482 .await
483 .expect("info refs");
484 assert_eq!(resp.status(), reqwest::StatusCode::OK);
485 let body = resp.bytes().await.expect("bytes");
486 assert!(!body.is_empty());
487 assert!(
488 String::from_utf8_lossy(&body).contains("service=git-upload-pack"),
489 "advertisement header missing: {:?}",
490 String::from_utf8_lossy(&body)
491 );
492}
493
494fn pseudo_random_bytes(len: usize, seed: u64) -> Vec<u8> {
495 let mut out = Vec::with_capacity(len);
496 let mut state: u64 = seed;
497 for _ in 0..len {
498 state ^= state << 13;
499 state ^= state >> 7;
500 state ^= state << 17;
501 out.push(state as u8);
502 }
503 out
504}
505
506fn compressible_payload(len: usize) -> Vec<u8> {
507 let pattern = b"the quick brown fox jumps over the lazy dog\n";
508 let mut out = Vec::with_capacity(len);
509 while out.len() < len {
510 let take = (len - out.len()).min(pattern.len());
511 out.extend_from_slice(&pattern[..take]);
512 }
513 out
514}
515
516fn gzip_compress(data: &[u8]) -> Vec<u8> {
517 use flate2::Compression;
518 use flate2::write::GzEncoder;
519 use std::io::Write;
520 let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
521 encoder.write_all(data).expect("gzip write");
522 encoder.finish().expect("gzip finish")
523}
524
525#[tokio::test]
526async fn lfs_upload_streams_large() {
527 let srv = start_server().await;
528 let client = reqwest::Client::new();
529
530 let data = pseudo_random_bytes(5 * 1024 * 1024, 0xA1B2_C3D4_E5F6_0701);
531 let oid = gitlfs::LfsOid::compute(&data);
532 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());
533
534 let put = client
535 .put(&url)
536 .header("Authorization", basic_auth("alice", "secret"))
537 .body(data.clone())
538 .timeout(Duration::from_secs(30))
539 .send()
540 .await
541 .expect("lfs put");
542 assert_eq!(put.status(), reqwest::StatusCode::OK, "lfs put failed");
543
544 let got = client
545 .get(&url)
546 .header("Authorization", basic_auth("alice", "secret"))
547 .send()
548 .await
549 .expect("lfs get");
550 assert_eq!(got.status(), reqwest::StatusCode::OK);
551 let body = got.bytes().await.expect("lfs bytes");
552 assert_eq!(body.len(), data.len(), "size mismatch");
553 assert_eq!(gitlfs::LfsOid::compute(&body), oid, "sha256 mismatch");
554 assert_eq!(body.as_ref(), data.as_slice(), "content mismatch");
555}
556
557#[tokio::test]
558async fn lfs_upload_gzip() {
559 let srv = start_server().await;
560 let client = reqwest::Client::new();
561
562 let data = compressible_payload(2 * 1024 * 1024);
563 let oid = gitlfs::LfsOid::compute(&data);
564 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());
565
566 let gz = gzip_compress(&data);
567 assert!(gz.len() < data.len(), "gzip did not compress random data");
568
569 let put = client
570 .put(&url)
571 .header("Authorization", basic_auth("alice", "secret"))
572 .header("Content-Encoding", "gzip")
573 .body(gz)
574 .timeout(Duration::from_secs(30))
575 .send()
576 .await
577 .expect("lfs put gzip");
578 assert_eq!(put.status(), reqwest::StatusCode::OK, "lfs gzip put failed");
579
580 let got = client
581 .get(&url)
582 .header("Authorization", basic_auth("alice", "secret"))
583 .send()
584 .await
585 .expect("lfs get");
586 assert_eq!(got.status(), reqwest::StatusCode::OK);
587 let body = got.bytes().await.expect("lfs bytes");
588 assert_eq!(body.as_ref(), data.as_slice(), "decoded mismatch");
589 assert_eq!(gitlfs::LfsOid::compute(&body), oid);
590}
591
592#[tokio::test]
593async fn lfs_upload_hash_mismatch() {
594 let srv = start_server().await;
595 let client = reqwest::Client::new();
596
597 let real = pseudo_random_bytes(64 * 1024, 0x1111_1111_1111_1111);
598 let fake = pseudo_random_bytes(64 * 1024, 0x2222_2222_2222_2222);
599 let oid = gitlfs::LfsOid::compute(&real);
600 assert_ne!(real, fake);
601 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());
602
603 let put = client
604 .put(&url)
605 .header("Authorization", basic_auth("alice", "secret"))
606 .body(fake)
607 .timeout(Duration::from_secs(30))
608 .send()
609 .await
610 .expect("lfs put");
611 assert!(
612 put.status().is_client_error() || put.status().is_server_error(),
613 "expected error status, got {}",
614 put.status()
615 );
616
617 let head = client
618 .get(&url)
619 .header("Authorization", basic_auth("alice", "secret"))
620 .send()
621 .await
622 .expect("lfs get");
623 assert_eq!(head.status(), reqwest::StatusCode::NOT_FOUND);
624}
625
626#[tokio::test]
627async fn lfs_download_streams_large() {
628 let srv = start_server().await;
629 let client = reqwest::Client::new();
630
631 let data = pseudo_random_bytes(3 * 1024 * 1024, 0xCAFE_BABE_DEAD_BEEF);
632 let oid = gitlfs::LfsOid::compute(&data);
633 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());
634
635 let put = client
636 .put(&url)
637 .header("Authorization", basic_auth("alice", "secret"))
638 .body(data.clone())
639 .send()
640 .await
641 .expect("seed put");
642 assert_eq!(put.status(), reqwest::StatusCode::OK);
643
644 let resp = client
645 .get(&url)
646 .header("Authorization", basic_auth("alice", "secret"))
647 .send()
648 .await
649 .expect("lfs get");
650 assert_eq!(resp.status(), reqwest::StatusCode::OK);
651
652 use futures::StreamExt;
653 let mut stream = resp.bytes_stream();
654 let mut got = Vec::with_capacity(data.len());
655 while let Some(chunk) = stream.next().await {
656 got.extend_from_slice(&chunk.expect("chunk"));
657 }
658 assert_eq!(got.len(), data.len(), "size mismatch");
659 assert_eq!(gitlfs::LfsOid::compute(&got), oid, "sha256 mismatch");
660 assert_eq!(got, data, "content mismatch");
661}
Deleted dvcs/gitssh/tests/ssh_roundtrip.rs
@@ -1,412 +0,0 @@
1use std::net::SocketAddr;
2use std::os::unix::fs::PermissionsExt;
3use std::path::Path;
4use std::sync::Arc;
5use std::time::Duration;
6
7use config::git::GitSshConfig;
8use config::jwt::JwtSettings;
9use db::channel::channels;
10use db::conn::Db;
11use db::conn::sea_orm::{
12 ActiveModelTrait, ConnectionTrait, Database, DatabaseBackend, DatabaseConnection, EntityTrait,
13 Schema, Set,
14};
15use db::repo::repos;
16use db::space::spaces;
17use db::user::user_ssh_keys;
18use db::user::users;
19use gitauth::GitAuthService;
20use gitssh::GitSshService;
21use gitssh::server::{generate_host_key, serve_on_listener};
22use jwt::{JwtState, SigningKeyCache};
23use sha2::{Digest, Sha256};
24use tempfile::TempDir;
25use time::OffsetDateTime;
26use tokio::sync::RwLock;
27use uuid::Uuid;
28
29fn now() -> OffsetDateTime {
30 OffsetDateTime::now_utc()
31}
32
33fn sha256_hex(data: &[u8]) -> String {
34 let mut hasher = Sha256::new();
35 hasher.update(data);
36 let digest = hasher.finalize();
37 let mut out = String::with_capacity(64);
38 for b in digest {
39 out.push_str(&format!("{b:02x}"));
40 }
41 out
42}
43
44async fn create_table<E: EntityTrait>(db: &DatabaseConnection) {
45 let stmt = Schema::new(DatabaseBackend::Sqlite).create_table_from_entity(E::default());
46 db.execute(&stmt).await.expect("create table");
47}
48
49async fn setup_db() -> DatabaseConnection {
50 let db = Database::connect("sqlite::memory:")
51 .await
52 .expect("connect sqlite memory");
53 create_table::<users::Entity>(&db).await;
54 create_table::<spaces::Entity>(&db).await;
55 create_table::<channels::Entity>(&db).await;
56 create_table::<db::space::space_members::Entity>(&db).await;
57 create_table::<db::space::space_roles::Entity>(&db).await;
58 create_table::<db::space::space_member_roles::Entity>(&db).await;
59 create_table::<db::space::space_role_permissions::Entity>(&db).await;
60 create_table::<db::repo::repo_collaborations::Entity>(&db).await;
61 create_table::<repos::Entity>(&db).await;
62 create_table::<user_ssh_keys::Entity>(&db).await;
63 db
64}
65
66async fn insert_user(db: &impl ConnectionTrait, uid: Uuid, username: &str) {
67 users::ActiveModel {
68 uid: Set(uid),
69 username: Set(username.to_string()),
70 lower_name: Set(username.to_lowercase()),
71 display_name: Set(username.to_string()),
72 avatar: Set(String::new()),
73 is_active: Set(true),
74 is_admin: Set(false),
75 is_restricted: Set(false),
76 prohibit_login: Set(false),
77 visibility: Set(users::ProfileVisibility::Public),
78 bot: Set(false),
79 system: Set(false),
80 totp_enabled: Set(false),
81 version: Set(1),
82 passwd_hash: Set(None),
83 created_at: Set(now()),
84 updated_at: Set(now()),
85 ..Default::default()
86 }
87 .insert(db)
88 .await
89 .expect("insert user");
90}
91
92async fn insert_space(db: &impl ConnectionTrait, id: Uuid, lower_name: &str, owner_id: Uuid) {
93 spaces::ActiveModel {
94 id: Set(id),
95 lower_name: Set(lower_name.to_string()),
96 name: Set(lower_name.to_string()),
97 owner_id: Set(owner_id),
98 public: Set(true),
99 is_active: Set(true),
100 allow_join: Set(true),
101 allow_invite: Set(true),
102 member_count: Set(0),
103 allow_join_level: Set(spaces::AllowJoinLevel::None),
104 member_must_enable_2fa: Set(false),
105 created_at: Set(now()),
106 updated_at: Set(now()),
107 ..Default::default()
108 }
109 .insert(db)
110 .await
111 .expect("insert space");
112}
113
114async fn insert_channel(
115 db: &impl ConnectionTrait,
116 id: Uuid,
117 space_id: Uuid,
118 name: &str,
119 created_by: Uuid,
120) {
121 channels::ActiveModel {
122 id: Set(id),
123 space_id: Set(Some(space_id)),
124 name: Set(name.to_string()),
125 position: Set(0),
126 is_archived: Set(false),
127 created_at: Set(now()),
128 updated_at: Set(now()),
129 created_by: Set(created_by),
130 ..Default::default()
131 }
132 .insert(db)
133 .await
134 .expect("insert channel");
135}
136
137async fn insert_repo(
138 db: &impl ConnectionTrait,
139 id: Uuid,
140 channel_id: Uuid,
141 name: &str,
142 is_private: bool,
143) {
144 repos::ActiveModel {
145 id: Set(id),
146 channel_id: Set(channel_id),
147 lower_name: Set(name.to_lowercase()),
148 name: Set(name.to_string()),
149 default_branch: Set("main".to_string()),
150 nums_watch: Set(0),
151 nums_star: Set(0),
152 nums_pr: Set(0),
153 nums_commits: Set(0),
154 nums_branches: Set(0),
155 nums_tags: Set(0),
156 enable_lfs: Set(true),
157 is_private: Set(is_private),
158 is_archived: Set(false),
159 size: Set(0),
160 git_size: Set(0),
161 lfs_size: Set(0),
162 allow_rebase: Set(true),
163 allow_rebase_explicit: Set(true),
164 allow_squash_merge: Set(true),
165 allow_merge_commits: Set(true),
166 created_at: Set(now()),
167 object_format: Set("sha1".to_string()),
168 ..Default::default()
169 }
170 .insert(db)
171 .await
172 .expect("insert repo");
173}
174
175async fn insert_ssh_key(db: &impl ConnectionTrait, user_id: Uuid, content: &str) {
176 let fingerprint = sha256_hex(content.as_bytes());
177 let am: user_ssh_keys::ActiveModel = user_ssh_keys::ActiveModel::builder()
178 .set_id(Uuid::now_v7())
179 .set_user_id(user_id)
180 .set_key_type(user_ssh_keys::SshKeyType::SshEd25519)
181 .set_fingerprint(fingerprint)
182 .set_name("test-key".to_string())
183 .set_content(content.to_string())
184 .set_created_at(now())
185 .set_updated_at(now())
186 .set_last_used_at(None::<time::OffsetDateTime>)
187 .into();
188 am.insert(db).await.expect("insert ssh key");
189}
190
191async fn git(dir: &Path, args: &[&str]) -> String {
192 let out = tokio::process::Command::new("git")
193 .args(args)
194 .current_dir(dir)
195 .env("GIT_TERMINAL_PROMPT", "0")
196 .env("GIT_SSH_COMMAND", "")
197 .output()
198 .await
199 .expect("run git");
200 assert!(
201 out.status.success(),
202 "git {:?} failed: {}",
203 args,
204 String::from_utf8_lossy(&out.stderr)
205 );
206 String::from_utf8_lossy(&out.stdout).into_owned()
207}
208
209struct TestServer {
210 _tmp: TempDir,
211 addr: SocketAddr,
212 port: u16,
213 git_root: std::path::PathBuf,
214 channel_id: Uuid,
215 repo_id: Uuid,
216 ssh_env: String,
217}
218
219async fn start_server() -> TestServer {
220 let tmp = TempDir::new().expect("tempdir");
221 let root = tmp.path().to_path_buf();
222
223 let git_root = root.join("git");
224 std::fs::create_dir_all(&git_root).expect("git root");
225
226 let conn = setup_db().await;
227 let db = Db::from_connection(conn.clone());
228
229 let alice = Uuid::now_v7();
230 let space_id = Uuid::now_v7();
231 let channel_id = Uuid::now_v7();
232 let repo_id = Uuid::now_v7();
233
234 insert_user(&conn, alice, "alice").await;
235 insert_space(&conn, space_id, "acme", alice).await;
236 insert_channel(&conn, channel_id, space_id, "dev", alice).await;
237 insert_repo(&conn, repo_id, channel_id, "widget", false).await;
238
239 let mut rng = rand::rng();
240 let client_key = russh::keys::PrivateKey::random(&mut rng, russh::keys::Algorithm::Ed25519)
241 .expect("client key");
242 let client_pub = client_key
243 .public_key()
244 .to_openssh()
245 .expect("client pub openssh");
246 insert_ssh_key(&conn, alice, &client_pub).await;
247
248 let client_key_path = root.join("id_ed25519");
249 let client_key_pem = client_key
250 .to_openssh(ssh_key::LineEnding::LF)
251 .expect("client key pem");
252 std::fs::write(&client_key_path, client_key_pem.as_bytes()).expect("write client key");
253 std::fs::set_permissions(&client_key_path, std::fs::Permissions::from_mode(0o600))
254 .expect("chmod client key");
255
256 let bare = git_root.join(format!("{}/{}.git", channel_id, repo_id));
257 std::fs::create_dir_all(bare.parent().expect("bare parent")).expect("bare dirs");
258 git(&root, &["init", "--bare", bare.to_str().expect("bare str")]).await;
259
260 let seed_dir = root.join("seed");
261 std::fs::create_dir_all(&seed_dir).expect("seed dir");
262 git(
263 &root,
264 &[
265 "init",
266 "--initial-branch=main",
267 seed_dir.to_str().expect("seed str"),
268 ],
269 )
270 .await;
271 std::fs::write(seed_dir.join("README.md"), "seed\n").expect("write seed");
272 git(&seed_dir, &["add", "README.md"]).await;
273 git(
274 &seed_dir,
275 &[
276 "-c",
277 "user.name=Seed",
278 "-c",
279 "user.email=seed@example.com",
280 "commit",
281 "-m",
282 "initial",
283 ],
284 )
285 .await;
286 git(
287 &seed_dir,
288 &["remote", "add", "origin", bare.to_str().expect("bare str")],
289 )
290 .await;
291 git(&seed_dir, &["push", "-u", "origin", "main"]).await;
292
293 let host_key = generate_host_key().expect("host key");
294 let host_key_path = root.join("host_key");
295 let host_key_pem = host_key
296 .to_openssh(ssh_key::LineEnding::LF)
297 .expect("host key pem");
298 std::fs::write(&host_key_path, host_key_pem.as_bytes()).expect("write host key");
299
300 let jwt = JwtState::new(
301 conn.clone(),
302 JwtSettings::default(),
303 Arc::new(RwLock::new(SigningKeyCache::new())),
304 );
305 let auth = GitAuthService::new(db, None, jwt);
306 let config = GitSshConfig {
307 git_storage_root: git_root.to_str().expect("git root str").to_string(),
308 host_key_path: host_key_path.to_str().expect("host key str").to_string(),
309 ..Default::default()
310 };
311 let svc = GitSshService::new(auth, config);
312
313 let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
314 .await
315 .expect("bind");
316 let addr = listener.local_addr().expect("addr");
317 let port = addr.port();
318 tokio::spawn(async move {
319 let _ = serve_on_listener(svc, listener).await;
320 });
321 tokio::time::sleep(Duration::from_millis(100)).await;
322
323 let ssh_env = format!(
324 "ssh -i {} -p {} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o IdentitiesOnly=yes",
325 client_key_path.display(),
326 port
327 );
328
329 TestServer {
330 _tmp: tmp,
331 addr,
332 port,
333 git_root,
334 channel_id,
335 repo_id,
336 ssh_env,
337 }
338}
339
340async fn git_ssh(dir: &Path, ssh_env: &str, args: &[&str]) -> String {
341 let out = tokio::process::Command::new("git")
342 .args(args)
343 .current_dir(dir)
344 .env("GIT_TERMINAL_PROMPT", "0")
345 .env("GIT_SSH_COMMAND", ssh_env)
346 .output()
347 .await
348 .expect("run git");
349 assert!(
350 out.status.success(),
351 "git {:?} failed: {}",
352 args,
353 String::from_utf8_lossy(&out.stderr)
354 );
355 String::from_utf8_lossy(&out.stdout).into_owned()
356}
357
358#[tokio::test]
359async fn ssh_clone_and_push_roundtrip() {
360 let server = start_server().await;
361 let url = format!("ssh://alice@127.0.0.1:{}/acme/widget.git", server.port);
362
363 let clone_dir = server._tmp.path().join("clone");
364 git_ssh(
365 server._tmp.path(),
366 &server.ssh_env,
367 &["clone", &url, clone_dir.to_str().expect("clone str")],
368 )
369 .await;
370
371 std::fs::write(clone_dir.join("hello.txt"), "hello ssh\n").expect("write file");
372 git(&clone_dir, &["add", "hello.txt"]).await;
373 git(
374 &clone_dir,
375 &[
376 "-c",
377 "user.name=Alice",
378 "-c",
379 "user.email=alice@example.com",
380 "commit",
381 "-m",
382 "add hello",
383 ],
384 )
385 .await;
386 git_ssh(
387 &clone_dir,
388 &server.ssh_env,
389 &["push", "origin", "HEAD:main"],
390 )
391 .await;
392
393 let bare = server
394 .git_root
395 .join(format!("{}/{}.git", server.channel_id, server.repo_id));
396 let out = tokio::process::Command::new("git")
397 .args(["log", "--oneline", "-1"])
398 .current_dir(&bare)
399 .output()
400 .await
401 .expect("git log");
402 assert!(
403 out.status.success(),
404 "git log failed: {}",
405 String::from_utf8_lossy(&out.stderr)
406 );
407 let log = String::from_utf8_lossy(&out.stdout);
408 assert!(
409 log.contains("add hello"),
410 "expected commit in bare repo, got: {log}"
411 );
412}
Deleted lib/jwt/tests.rs
@@ -1,202 +0,0 @@
1use time::OffsetDateTime;
2use uuid::Uuid;
3
4use crate::claims::{AccessClaims, RefreshClaims};
5use crate::error::JwtError;
6use crate::grace;
7use crate::keys::CachedKey;
8use crate::signer::{encode_access_token, encode_refresh_token};
9
10fn ed25519_key() -> (jsonwebtoken::EncodingKey, jsonwebtoken::DecodingKey) {
11 use ed25519_dalek::SigningKey;
12 use ed25519_dalek::pkcs8::EncodePrivateKey;
13 use rand_core::OsRng;
14 let sk = SigningKey::generate(&mut OsRng);
15 let vk = sk.verifying_key();
16 let encoding = jsonwebtoken::EncodingKey::from_ed_der(sk.to_pkcs8_der().unwrap().as_bytes());
17 let decoding = jsonwebtoken::DecodingKey::from_ed_der(&vk.to_bytes());
18 (encoding, decoding)
19}
20
21fn access_claims(kid: &str) -> AccessClaims {
22 AccessClaims::new(
23 Uuid::now_v7(),
24 "alice".into(),
25 true,
26 false,
27 1,
28 900,
29 "test-issuer",
30 "test-aud",
31 kid,
32 )
33}
34
35#[test]
36fn access_roundtrip() {
37 let (enc, dec) = ed25519_key();
38 let kid = "kid-1";
39 let claims = access_claims(kid);
40 let token = encode_access_token(&claims, &enc, jsonwebtoken::Algorithm::EdDSA).unwrap();
41 let mut validation = jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::EdDSA);
42 validation.set_issuer(&["test-issuer".to_string()]);
43 validation.set_audience(&["test-aud".to_string()]);
44 let data = jsonwebtoken::decode::<AccessClaims>(&token, &dec, &validation).unwrap();
45 assert_eq!(data.claims.sub, claims.sub);
46 assert_eq!(data.claims.kid, kid);
47 assert!(data.claims.is_admin);
48}
49
50#[test]
51fn refresh_roundtrip() {
52 let (enc, dec) = ed25519_key();
53 let kid = "kid-1";
54 let family = Uuid::now_v7();
55 let claims = RefreshClaims::new(
56 Uuid::now_v7(),
57 family,
58 None,
59 1,
60 2592000,
61 "test-issuer",
62 "test-aud",
63 kid,
64 );
65 let token = encode_refresh_token(&claims, &enc, jsonwebtoken::Algorithm::EdDSA).unwrap();
66 let mut validation = jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::EdDSA);
67 validation.set_issuer(&["test-issuer".to_string()]);
68 validation.set_audience(&["test-aud".to_string()]);
69 let data = jsonwebtoken::decode::<RefreshClaims>(&token, &dec, &validation).unwrap();
70 assert_eq!(data.claims.family_id, family);
71 assert_eq!(data.claims.kid, kid);
72}
73
74#[test]
75fn expired_token_rejected() {
76 let (enc, dec) = ed25519_key();
77 let mut claims = access_claims("kid-1");
78 claims.exp = 0;
79 let token = encode_access_token(&claims, &enc, jsonwebtoken::Algorithm::EdDSA).unwrap();
80 let mut validation = jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::EdDSA);
81 validation.set_issuer(&["test-issuer".to_string()]);
82 validation.set_audience(&["test-aud".to_string()]);
83 let res = jsonwebtoken::decode::<AccessClaims>(&token, &dec, &validation);
84 assert!(res.is_err());
85 let err = JwtError::from(res.unwrap_err());
86 assert!(matches!(err, JwtError::Expired));
87}
88
89#[test]
90fn wrong_audience_rejected() {
91 let (enc, dec) = ed25519_key();
92 let claims = access_claims("kid-1");
93 let token = encode_access_token(&claims, &enc, jsonwebtoken::Algorithm::EdDSA).unwrap();
94 let mut validation = jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::EdDSA);
95 validation.set_issuer(&["test-issuer".to_string()]);
96 validation.set_audience(&["wrong-aud".to_string()]);
97 let res = jsonwebtoken::decode::<AccessClaims>(&token, &dec, &validation);
98 assert!(res.is_err());
99 let err = JwtError::from(res.unwrap_err());
100 assert!(matches!(err, JwtError::InvalidAudience));
101}
102
103#[test]
104fn wrong_key_rejected() {
105 let (enc, _dec) = ed25519_key();
106 let (_, dec2) = ed25519_key();
107 let claims = access_claims("kid-1");
108 let token = encode_access_token(&claims, &enc, jsonwebtoken::Algorithm::EdDSA).unwrap();
109 let mut validation = jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::EdDSA);
110 validation.set_issuer(&["test-issuer".to_string()]);
111 validation.set_audience(&["test-aud".to_string()]);
112 let res = jsonwebtoken::decode::<AccessClaims>(&token, &dec2, &validation);
113 assert!(res.is_err());
114 let err = JwtError::from(res.unwrap_err());
115 assert!(matches!(err, JwtError::InvalidSignature));
116}
117
118fn cached_key(rotating: bool, grace_ends_at: Option<OffsetDateTime>) -> CachedKey {
119 let (_, dec) = ed25519_key();
120 CachedKey {
121 decoding: dec,
122 algorithm: jsonwebtoken::Algorithm::EdDSA,
123 is_rotating: rotating,
124 grace_ends_at,
125 expires_at: None,
126 }
127}
128
129#[test]
130fn active_key_usable_for_verify() {
131 let key = cached_key(false, None);
132 assert!(grace::is_key_usable_for_verify(
133 &key,
134 OffsetDateTime::now_utc()
135 ));
136}
137
138#[test]
139fn rotating_key_usable_inside_grace() {
140 let key = cached_key(
141 true,
142 Some(OffsetDateTime::now_utc() + time::Duration::seconds(60)),
143 );
144 assert!(grace::is_key_usable_for_verify(
145 &key,
146 OffsetDateTime::now_utc()
147 ));
148}
149
150#[test]
151fn rotating_key_unusable_after_grace() {
152 let key = cached_key(
153 true,
154 Some(OffsetDateTime::now_utc() - time::Duration::seconds(1)),
155 );
156 assert!(!grace::is_key_usable_for_verify(
157 &key,
158 OffsetDateTime::now_utc()
159 ));
160 assert!(grace::ensure_key_usable(&key).is_err());
161}
162
163#[test]
164fn expired_key_unusable() {
165 let mut key = cached_key(false, None);
166 key.expires_at = Some(OffsetDateTime::now_utc() - time::Duration::seconds(1));
167 assert!(!grace::is_key_usable_for_verify(
168 &key,
169 OffsetDateTime::now_utc()
170 ));
171}
172
173#[test]
174fn replay_rejected_outside_grace() {
175 let now = OffsetDateTime::now_utc();
176 assert!(grace::reject_replay(None, now));
177 assert!(grace::reject_replay(
178 Some(now - time::Duration::seconds(1)),
179 now
180 ));
181 assert!(!grace::reject_replay(
182 Some(now + time::Duration::seconds(60)),
183 now
184 ));
185}
186
187#[test]
188fn unknown_kid_returns_error() {
189 let kid = Uuid::now_v7();
190 let cache = crate::keys::SigningKeyCache::new();
191 assert!(cache.get(&kid).is_none());
192 assert!(!cache.in_grace_window(&kid));
193}
194
195#[test]
196fn kid_from_token_roundtrip() {
197 let (enc, _dec) = ed25519_key();
198 let claims = access_claims("kid-xyz");
199 let token = encode_access_token(&claims, &enc, jsonwebtoken::Algorithm::EdDSA).unwrap();
200 let kid = crate::signer::kid_from_token(&token);
201 assert_eq!(kid.as_deref(), Some("kid-xyz"));
202}