Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

feat(chat): add chat storagelayer crate (models, repository, blob, mock)

zhenyi committed
parent afe9a6b commit b3475d6 Browse files

28 files changed

+2300 −0

File tree 28
Added chat/storagelayer/blob/media.rs View chat/storagelayer/blob/media.rs
@@ -0,0 +1,81 @@
1use std::time::Duration;
2
3use async_trait::async_trait;
4use bytes::Bytes;
5use url::Url;
6use uuid::Uuid;
7
8use crate::error::ChatStorageResult;
9use crate::models::attachment::AttachmentOwner;
10
11#[derive(Debug, Clone, PartialEq, Eq, Hash)]
12pub struct AttachmentPath {
13 pub owner: AttachmentOwner,
14 pub owner_id: Uuid,
15 pub filename: String,
16}
17
18impl AttachmentPath {
19 pub fn new(owner: AttachmentOwner, owner_id: Uuid, filename: impl Into<String>) -> Self {
20 Self {
21 owner,
22 owner_id,
23 filename: filename.into(),
24 }
25 }
26
27 pub fn as_key(&self) -> String {
28 let sanitized = sanitize_filename(&self.filename);
29 format!(
30 "{}/{}/{}-{}",
31 self.owner.as_str(),
32 self.owner_id,
33 Uuid::new_v4(),
34 sanitized
35 )
36 }
37}
38
39fn sanitize_filename(input: &str) -> String {
40 input
41 .chars()
42 .map(|c| match c {
43 '/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|' | ' ' => '_',
44 _ => c,
45 })
46 .collect()
47}
48
49#[async_trait]
50pub trait ChatMediaStore: Send + Sync {
51 async fn put_attachment_bytes(
52 &self,
53 owner: AttachmentOwner,
54 owner_id: Uuid,
55 filename: &str,
56 bytes: Bytes,
57 ) -> ChatStorageResult<AttachmentPath>;
58
59 async fn get_attachment_bytes(
60 &self,
61 path: &AttachmentPath,
62 ) -> ChatStorageResult<Bytes>;
63
64 async fn delete_attachment(&self, path: &AttachmentPath) -> ChatStorageResult<()>;
65
66 async fn presign_upload(
67 &self,
68 owner: AttachmentOwner,
69 owner_id: Uuid,
70 filename: &str,
71 ttl: Duration,
72 ) -> ChatStorageResult<Url>;
73
74 async fn presign_download(
75 &self,
76 path: &AttachmentPath,
77 ttl: Duration,
78 ) -> ChatStorageResult<Url>;
79}
80
81pub type DynChatMediaStore = std::sync::Arc<dyn ChatMediaStore>;
No newline at end of file
Added chat/storagelayer/blob/object_store.rs View chat/storagelayer/blob/object_store.rs
@@ -0,0 +1,139 @@
1use std::path::PathBuf;
2use std::sync::Arc;
3use std::time::Duration;
4
5use async_trait::async_trait;
6use bytes::Bytes;
7use object_store::path::Path as ObjectPath;
8use url::Url;
9use uuid::Uuid;
10
11use crate::blob::media::{AttachmentPath, ChatMediaStore};
12use crate::error::{ChatStorageError, ChatStorageResult};
13use crate::models::attachment::AttachmentOwner;
14
15pub struct ObjectStoreChatMedia {
16 inner: Arc<dyn storage::StorageRepository>,
17 prefix: PathBuf,
18}
19
20impl std::fmt::Debug for ObjectStoreChatMedia {
21 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
22 f.debug_struct("ObjectStoreChatMedia")
23 .field("prefix", &self.prefix)
24 .field("inner", &"<dyn storage::StorageRepository>")
25 .finish()
26 }
27}
28
29impl ObjectStoreChatMedia {
30 pub fn new(inner: Arc<dyn storage::StorageRepository>) -> Self {
31 Self {
32 inner,
33 prefix: PathBuf::from("chat"),
34 }
35 }
36
37 pub fn with_prefix(mut self, prefix: impl Into<PathBuf>) -> Self {
38 self.prefix = prefix.into();
39 self
40 }
41
42 fn object_path(&self, key: &str) -> ObjectPath {
43 let mut p = self.prefix.clone();
44 for part in key.split('/') {
45 if !part.is_empty() {
46 p.push(part);
47 }
48 }
49 let s = p.to_string_lossy().trim_start_matches('/').to_string();
50 ObjectPath::from(s)
51 }
52}
53
54#[async_trait]
55impl ChatMediaStore for ObjectStoreChatMedia {
56 async fn put_attachment_bytes(
57 &self,
58 owner: AttachmentOwner,
59 owner_id: Uuid,
60 filename: &str,
61 bytes: Bytes,
62 ) -> ChatStorageResult<AttachmentPath> {
63 let attachment = AttachmentPath::new(owner, owner_id, filename);
64 let key = attachment.as_key();
65 let path = self.object_path(&key);
66 self.inner
67 .put(&path, bytes)
68 .await
69 .map_err(ChatStorageError::from)?;
70 Ok(attachment)
71 }
72
73 async fn get_attachment_bytes(
74 &self,
75 path: &AttachmentPath,
76 ) -> ChatStorageResult<Bytes> {
77 use futures::TryStreamExt;
78 let key = path.as_key();
79 let op = self.object_path(&key);
80 let result = self
81 .inner
82 .get(&op)
83 .await
84 .map_err(ChatStorageError::from)?;
85 let collected: Vec<Bytes> = result
86 .into_stream()
87 .try_collect()
88 .await
89 .map_err(|e| ChatStorageError::Blob(storage::StorageError::from(e)))?;
90 Ok(collected.into_iter().fold(Bytes::new(), |mut acc, b| {
91 if acc.is_empty() {
92 acc = b;
93 } else {
94 let mut v = acc.to_vec();
95 v.extend_from_slice(&b);
96 acc = Bytes::from(v);
97 }
98 acc
99 }))
100 }
101
102 async fn delete_attachment(&self, path: &AttachmentPath) -> ChatStorageResult<()> {
103 let key = path.as_key();
104 let op = self.object_path(&key);
105 self.inner
106 .delete(&op)
107 .await
108 .map_err(ChatStorageError::from)
109 }
110
111 async fn presign_upload(
112 &self,
113 owner: AttachmentOwner,
114 owner_id: Uuid,
115 filename: &str,
116 ttl: Duration,
117 ) -> ChatStorageResult<Url> {
118 let attachment = AttachmentPath::new(owner, owner_id, filename);
119 let key = attachment.as_key();
120 let op = self.object_path(&key);
121 self.inner
122 .presigned_upload(&op, ttl)
123 .await
124 .map_err(ChatStorageError::from)
125 }
126
127 async fn presign_download(
128 &self,
129 path: &AttachmentPath,
130 ttl: Duration,
131 ) -> ChatStorageResult<Url> {
132 let key = path.as_key();
133 let op = self.object_path(&key);
134 self.inner
135 .presigned_download(&op, ttl)
136 .await
137 .map_err(ChatStorageError::from)
138 }
139}
No newline at end of file
Added chat/storagelayer/mock/in_memory.rs View chat/storagelayer/mock/in_memory.rs
@@ -0,0 +1,743 @@
1use std::collections::HashMap;
2use std::sync::Arc;
3use std::sync::RwLock;
4
5use async_trait::async_trait;
6use time::OffsetDateTime;
7use uuid::Uuid;
8
9use crate::blob::ChatMediaStore;
10use crate::error::{ChatStorageError, ChatStorageResult};
11use crate::blob::media::AttachmentPath;
12use bytes::Bytes;
13use std::time::Duration;
14use url::Url;
15use crate::models::attachment::{Attachment, AttachmentOwner, NewAttachment};
16use crate::models::conversation::{Conversation, ConversationKind, ConversationMember, NewConversation};
17use crate::models::mention::{Mention, NewMention};
18use crate::models::message::{Message, MessageFlags, MessagePage, MessageQuery, NewMessage, Page};
19use crate::models::pinned::{NewPinnedMessage, PinnedMessage};
20use crate::models::reaction::{NewReaction, Reaction};
21use crate::models::read_receipt::{NewReadReceipt, ReadReceipt};
22use crate::models::thread::Thread;
23use crate::repository::attachment::{AttachmentRepository, DynAttachmentRepository};
24use crate::repository::conversation::{ConversationRepository, DynConversationRepository};
25use crate::repository::member::{DynMemberRepository, MemberRepository};
26use crate::repository::mention::{DynMentionRepository, MentionRepository};
27use crate::repository::message::{DynMessageRepository, MessageRepository};
28use crate::repository::pinned::{DynPinnedMessageRepository, PinnedMessageRepository};
29use crate::repository::reaction::{DynReactionRepository, ReactionRepository};
30use crate::repository::read_receipt::{DynReadReceiptRepository, ReadReceiptRepository};
31use crate::repository::thread::{DynThreadRepository, ThreadRepository};
32use crate::repository::{ChatStorage, ChatStorageCtx, DynChatStorage};
33
34
35#[derive(Debug, Default)]
36pub struct InMemoryState {
37 pub conversations: RwLock<HashMap<Uuid, Conversation>>,
38 pub messages: RwLock<HashMap<Uuid, Message>>,
39 pub attachments: RwLock<HashMap<Uuid, Attachment>>,
40 pub members: RwLock<HashMap<(Uuid, Uuid), ConversationMember>>,
41 pub reactions: RwLock<HashMap<(Uuid, Uuid, String), Reaction>>,
42 pub mentions: RwLock<HashMap<Uuid, Mention>>,
43 pub read_receipts: RwLock<HashMap<Uuid, ReadReceipt>>,
44 pub pinned: RwLock<HashMap<(Uuid, Uuid), PinnedMessage>>,
45 pub threads: RwLock<HashMap<(Uuid, Uuid), Thread>>,
46}
47
48pub type SharedState = Arc<InMemoryState>;
49
50fn require_user(ctx: &ChatStorageCtx) -> ChatStorageResult<Uuid> {
51 ctx.require_user()
52}
53
54
55#[derive(Debug)]
56pub struct InMemoryConversationRepository {
57 state: SharedState,
58}
59
60impl InMemoryConversationRepository {
61 pub fn new(state: SharedState) -> Self {
62 Self { state }
63 }
64}
65
66#[async_trait]
67impl ConversationRepository for InMemoryConversationRepository {
68 async fn create(&self, ctx: &ChatStorageCtx, input: NewConversation) -> ChatStorageResult<Conversation> {
69 require_user(ctx)?;
70 let now = OffsetDateTime::now_utc();
71 let conv = Conversation {
72 id: Uuid::new_v4(),
73 channel_id: input.channel_id,
74 kind: input.kind,
75 counterpart_user_id: input.counterpart_user_id,
76 title: input.title,
77 created_at: now,
78 updated_at: Some(now),
79 };
80 self.state.conversations.write().unwrap().insert(conv.id, conv.clone());
81
82 let joined_at = now;
83 let mut members = self.state.members.write().unwrap();
84 for uid in &input.initial_members {
85 let m = ConversationMember {
86 chat_id: conv.id,
87 user_id: *uid,
88 joined_at,
89 last_read_at: None,
90 muted: false,
91 };
92 members.insert((conv.id, *uid), m);
93 }
94 Ok(conv)
95 }
96
97 async fn find_by_id(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<Option<Conversation>> {
98 Ok(self.state.conversations.read().unwrap().get(&id).cloned())
99 }
100
101 async fn list_by_channel(&self, _ctx: &ChatStorageCtx, channel_id: Uuid, page: Page) -> ChatStorageResult<Vec<Conversation>> {
102 let map = self.state.conversations.read().unwrap();
103 let mut out: Vec<Conversation> = map
104 .values()
105 .filter(|c| c.channel_id == channel_id)
106 .cloned()
107 .collect();
108 out.sort_by_key(|c| c.created_at);
109 let limit = page.limit as usize;
110 Ok(out.into_iter().take(limit).collect())
111 }
112
113 async fn update_title(&self, ctx: &ChatStorageCtx, id: Uuid, title: Option<String>) -> ChatStorageResult<()> {
114 require_user(ctx)?;
115 let mut map = self.state.conversations.write().unwrap();
116 let conv = map.get_mut(&id).ok_or(ChatStorageError::ConversationNotFound(id))?;
117 conv.title = title;
118 conv.updated_at = Some(OffsetDateTime::now_utc());
119 Ok(())
120 }
121
122 async fn update_kind(&self, ctx: &ChatStorageCtx, id: Uuid, kind: ConversationKind) -> ChatStorageResult<()> {
123 require_user(ctx)?;
124 let mut map = self.state.conversations.write().unwrap();
125 let conv = map.get_mut(&id).ok_or(ChatStorageError::ConversationNotFound(id))?;
126 conv.kind = kind;
127 conv.updated_at = Some(OffsetDateTime::now_utc());
128 Ok(())
129 }
130
131 async fn delete(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {
132 require_user(ctx)?;
133 let removed = self.state.conversations.write().unwrap().remove(&id).is_some();
134 if !removed {
135 return Err(ChatStorageError::ConversationNotFound(id));
136 }
137 Ok(())
138 }
139
140 async fn exists(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<bool> {
141 Ok(self.state.conversations.read().unwrap().contains_key(&id))
142 }
143
144 async fn list_members(&self, _ctx: &ChatStorageCtx, chat_id: Uuid, page: Page) -> ChatStorageResult<Vec<ConversationMember>> {
145 let map = self.state.members.read().unwrap();
146 let mut out: Vec<ConversationMember> = map
147 .values()
148 .filter(|m| m.chat_id == chat_id)
149 .cloned()
150 .collect();
151 out.sort_by_key(|m| m.joined_at);
152 let limit = page.limit as usize;
153 Ok(out.into_iter().take(limit).collect())
154 }
155}
156
157
158#[derive(Debug)]
159pub struct InMemoryMessageRepository {
160 state: SharedState,
161}
162
163impl InMemoryMessageRepository {
164 pub fn new(state: SharedState) -> Self {
165 Self { state }
166 }
167}
168
169#[async_trait]
170impl MessageRepository for InMemoryMessageRepository {
171 async fn create(&self, ctx: &ChatStorageCtx, input: NewMessage) -> ChatStorageResult<Message> {
172 let sender_id = ctx.user_id.unwrap_or(input.sender_id);
173 let now = OffsetDateTime::now_utc();
174 let msg = Message {
175 id: Uuid::new_v4(),
176 chat_id: input.chat_id,
177 sender_id,
178 body: input.body,
179 reply_to_message_id: input.reply_to_message_id,
180 thread_id: input.thread_id,
181 is_thread_start: input.is_thread_start,
182 flags: input.flags.unwrap_or(0),
183 deleted_at: None,
184 deleted_by: None,
185 edited_at: None,
186 created_at: now,
187 };
188 self.state.messages.write().unwrap().insert(msg.id, msg.clone());
189 Ok(msg)
190 }
191
192 async fn find_by_id(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<Option<Message>> {
193 let map = self.state.messages.read().unwrap();
194 let msg = map.get(&id).cloned();
195 let include_deleted = true;
196 let _ = include_deleted;
197 Ok(msg)
198 }
199
200 async fn list_by_conversation(&self, _ctx: &ChatStorageCtx, chat_id: Uuid, query: MessageQuery) -> ChatStorageResult<MessagePage> {
201 let map = self.state.messages.read().unwrap();
202 let mut items: Vec<Message> = map
203 .values()
204 .filter(|m| m.chat_id == chat_id)
205 .filter(|m| query.include_deleted || m.deleted_at.is_none())
206 .cloned()
207 .collect();
208 use crate::models::message::ListDirection;
209 match query.direction {
210 ListDirection::Forward => items.sort_by_key(|m| m.created_at),
211 ListDirection::Backward => items.sort_by_key(|m| std::cmp::Reverse(m.created_at)),
212 }
213 let limit = query.limit as usize;
214 let items: Vec<Message> = items.into_iter().take(limit).collect();
215 let next_cursor = items.last().map(|m| m.id.to_string());
216 Ok(MessagePage { items, next_cursor })
217 }
218
219 async fn list_thread(&self, _ctx: &ChatStorageCtx, chat_id: Uuid, root_message_id: Uuid, page: Page) -> ChatStorageResult<MessagePage> {
220 let map = self.state.messages.read().unwrap();
221 let mut items: Vec<Message> = map
222 .values()
223 .filter(|m| m.chat_id == chat_id && m.thread_id == Some(root_message_id))
224 .cloned()
225 .collect();
226 items.sort_by_key(|m| m.created_at);
227 let limit = page.limit as usize;
228 let items: Vec<Message> = items.into_iter().take(limit).collect();
229 let next_cursor = items.last().map(|m| m.id.to_string());
230 Ok(MessagePage { items, next_cursor })
231 }
232
233 async fn edit(&self, ctx: &ChatStorageCtx, id: Uuid, editor: Uuid, new_body: String) -> ChatStorageResult<Message> {
234 require_user(ctx)?;
235 let now = OffsetDateTime::now_utc();
236 let mut map = self.state.messages.write().unwrap();
237 let msg = map.get_mut(&id).ok_or(ChatStorageError::MessageNotFound(id))?;
238 msg.body = new_body;
239 msg.edited_at = Some(now);
240 msg.flags = MessageFlags::from_bits_truncate(msg.flags).with(MessageFlags::EDITED).bits();
241 msg.deleted_by = None;
242 let _ = editor;
243 Ok(msg.clone())
244 }
245
246 async fn soft_delete(&self, ctx: &ChatStorageCtx, id: Uuid, by: Uuid) -> ChatStorageResult<()> {
247 require_user(ctx)?;
248 let now = OffsetDateTime::now_utc();
249 let mut map = self.state.messages.write().unwrap();
250 let msg = map.get_mut(&id).ok_or(ChatStorageError::MessageNotFound(id))?;
251 msg.deleted_at = Some(now);
252 msg.deleted_by = Some(by);
253 msg.flags = MessageFlags::from_bits_truncate(msg.flags).with(MessageFlags::DELETED).bits();
254 Ok(())
255 }
256
257 async fn restore(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {
258 require_user(ctx)?;
259 let mut map = self.state.messages.write().unwrap();
260 let msg = map.get_mut(&id).ok_or(ChatStorageError::MessageNotFound(id))?;
261 msg.deleted_at = None;
262 msg.deleted_by = None;
263 msg.flags = MessageFlags::from_bits_truncate(msg.flags).without(MessageFlags::DELETED).bits();
264 Ok(())
265 }
266
267 async fn exists(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<bool> {
268 Ok(self.state.messages.read().unwrap().contains_key(&id))
269 }
270}
271
272
273#[derive(Debug)]
274pub struct InMemoryAttachmentRepository {
275 state: SharedState,
276}
277
278impl InMemoryAttachmentRepository {
279 pub fn new(state: SharedState) -> Self {
280 Self { state }
281 }
282}
283
284#[async_trait]
285impl AttachmentRepository for InMemoryAttachmentRepository {
286 async fn create_metadata(&self, ctx: &ChatStorageCtx, input: NewAttachment) -> ChatStorageResult<Attachment> {
287 require_user(ctx)?;
288 let att = Attachment {
289 id: Uuid::new_v4(),
290 owner: input.owner,
291 owner_id: input.owner_id,
292 filename: input.filename,
293 content_type: input.content_type,
294 size: input.size,
295 url: input.url,
296 uploaded_by: input.uploaded_by,
297 created_at: OffsetDateTime::now_utc(),
298 };
299 self.state.attachments.write().unwrap().insert(att.id, att.clone());
300 Ok(att)
301 }
302
303 async fn find_by_id(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<Option<Attachment>> {
304 Ok(self.state.attachments.read().unwrap().get(&id).cloned())
305 }
306
307 async fn list_for_owner(&self, _ctx: &ChatStorageCtx, owner: AttachmentOwner, owner_id: Uuid, page: Page) -> ChatStorageResult<Vec<Attachment>> {
308 let map = self.state.attachments.read().unwrap();
309 let mut items: Vec<Attachment> = map
310 .values()
311 .filter(|a| a.owner == owner && a.owner_id == owner_id)
312 .cloned()
313 .collect();
314 items.sort_by_key(|a| a.created_at);
315 let limit = page.limit as usize;
316 Ok(items.into_iter().take(limit).collect())
317 }
318
319 async fn delete(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {
320 require_user(ctx)?;
321 let removed = self.state.attachments.write().unwrap().remove(&id).is_some();
322 if !removed {
323 return Err(ChatStorageError::AttachmentNotFound(id));
324 }
325 Ok(())
326 }
327}
328
329
330#[derive(Debug)]
331pub struct InMemoryMemberRepository {
332 state: SharedState,
333}
334
335impl InMemoryMemberRepository {
336 pub fn new(state: SharedState) -> Self {
337 Self { state }
338 }
339}
340
341#[async_trait]
342impl MemberRepository for InMemoryMemberRepository {
343 async fn add_member(&self, ctx: &ChatStorageCtx, chat_id: Uuid, user_id: Uuid) -> ChatStorageResult<ConversationMember> {
344 require_user(ctx)?;
345 let m = ConversationMember {
346 chat_id,
347 user_id,
348 joined_at: OffsetDateTime::now_utc(),
349 last_read_at: None,
350 muted: false,
351 };
352 self.state.members.write().unwrap().insert((chat_id, user_id), m.clone());
353 Ok(m)
354 }
355
356 async fn remove_member(&self, ctx: &ChatStorageCtx, chat_id: Uuid, user_id: Uuid) -> ChatStorageResult<()> {
357 require_user(ctx)?;
358 let removed = self.state.members.write().unwrap().remove(&(chat_id, user_id)).is_some();
359 if !removed {
360 return Err(ChatStorageError::MemberNotFound(chat_id, user_id));
361 }
362 Ok(())
363 }
364
365 async fn list_members(&self, _ctx: &ChatStorageCtx, chat_id: Uuid, page: Page) -> ChatStorageResult<Vec<ConversationMember>> {
366 self.conversation_repo_list_members(chat_id, page)
367 }
368
369 async fn mark_read(&self, ctx: &ChatStorageCtx, chat_id: Uuid, user_id: Uuid, at: OffsetDateTime) -> ChatStorageResult<()> {
370 require_user(ctx)?;
371 let mut map = self.state.members.write().unwrap();
372 let m = map.get_mut(&(chat_id, user_id)).ok_or(ChatStorageError::MemberNotFound(chat_id, user_id))?;
373 m.last_read_at = Some(at);
374 Ok(())
375 }
376
377 async fn set_muted(&self, ctx: &ChatStorageCtx, chat_id: Uuid, user_id: Uuid, muted: bool) -> ChatStorageResult<()> {
378 require_user(ctx)?;
379 let mut map = self.state.members.write().unwrap();
380 let m = map.get_mut(&(chat_id, user_id)).ok_or(ChatStorageError::MemberNotFound(chat_id, user_id))?;
381 m.muted = muted;
382 Ok(())
383 }
384
385 async fn is_member(&self, _ctx: &ChatStorageCtx, chat_id: Uuid, user_id: Uuid) -> ChatStorageResult<bool> {
386 Ok(self.state.members.read().unwrap().contains_key(&(chat_id, user_id)))
387 }
388}
389
390impl InMemoryMemberRepository {
391 fn conversation_repo_list_members(&self, chat_id: Uuid, page: Page) -> ChatStorageResult<Vec<ConversationMember>> {
392 let map = self.state.members.read().unwrap();
393 let mut out: Vec<ConversationMember> = map
394 .values()
395 .filter(|m| m.chat_id == chat_id)
396 .cloned()
397 .collect();
398 out.sort_by_key(|m| m.joined_at);
399 let limit = page.limit as usize;
400 Ok(out.into_iter().take(limit).collect())
401 }
402}
403
404
405#[derive(Debug)]
406pub struct InMemoryReactionRepository {
407 state: SharedState,
408}
409
410impl InMemoryReactionRepository {
411 pub fn new(state: SharedState) -> Self {
412 Self { state }
413 }
414}
415
416#[async_trait]
417impl ReactionRepository for InMemoryReactionRepository {
418 async fn add_reaction(&self, ctx: &ChatStorageCtx, input: NewReaction) -> ChatStorageResult<Reaction> {
419 require_user(ctx)?;
420 let r = Reaction {
421 id: Uuid::new_v4(),
422 message_id: input.message_id,
423 user_id: input.user_id,
424 emoji: input.emoji,
425 created_at: OffsetDateTime::now_utc(),
426 };
427 self.state
428 .reactions
429 .write()
430 .unwrap()
431 .insert((r.message_id, r.user_id, r.emoji.clone()), r.clone());
432 Ok(r)
433 }
434
435 async fn remove_reaction(&self, ctx: &ChatStorageCtx, message_id: Uuid, user_id: Uuid, emoji: &str) -> ChatStorageResult<()> {
436 require_user(ctx)?;
437 let removed = self
438 .state
439 .reactions
440 .write()
441 .unwrap()
442 .remove(&(message_id, user_id, emoji.to_string()))
443 .is_some();
444 if !removed {
445 return Err(ChatStorageError::ReactionNotFound(message_id));
446 }
447 Ok(())
448 }
449
450 async fn list_reactions(&self, _ctx: &ChatStorageCtx, message_id: Uuid) -> ChatStorageResult<Vec<Reaction>> {
451 let map = self.state.reactions.read().unwrap();
452 Ok(map
453 .values()
454 .filter(|r| r.message_id == message_id)
455 .cloned()
456 .collect())
457 }
458}
459
460
461#[derive(Debug)]
462pub struct InMemoryMentionRepository {
463 state: SharedState,
464}
465
466impl InMemoryMentionRepository {
467 pub fn new(state: SharedState) -> Self {
468 Self { state }
469 }
470}
471
472#[async_trait]
473impl MentionRepository for InMemoryMentionRepository {
474 async fn add_mention(&self, ctx: &ChatStorageCtx, input: NewMention) -> ChatStorageResult<Mention> {
475 require_user(ctx)?;
476 let m = Mention {
477 id: Uuid::new_v4(),
478 message_id: input.message_id,
479 user_id: input.user_id,
480 role_id: input.role_id,
481 channel_id: input.channel_id,
482 is_everyone: input.is_everyone,
483 is_here: input.is_here,
484 created_at: OffsetDateTime::now_utc(),
485 };
486 self.state.mentions.write().unwrap().insert(m.id, m.clone());
487 Ok(m)
488 }
489
490 async fn list_for_message(&self, _ctx: &ChatStorageCtx, message_id: Uuid) -> ChatStorageResult<Vec<Mention>> {
491 let map = self.state.mentions.read().unwrap();
492 Ok(map.values().filter(|m| m.message_id == message_id).cloned().collect())
493 }
494
495 async fn list_for_user(&self, _ctx: &ChatStorageCtx, user_id: Uuid, limit: u32) -> ChatStorageResult<Vec<Mention>> {
496 let map = self.state.mentions.read().unwrap();
497 let mut items: Vec<Mention> = map.values().filter(|m| m.user_id == Some(user_id)).cloned().collect();
498 items.sort_by_key(|m| std::cmp::Reverse(m.created_at));
499 Ok(items.into_iter().take(limit as usize).collect())
500 }
501}
502
503
504#[derive(Debug)]
505pub struct InMemoryReadReceiptRepository {
506 state: SharedState,
507}
508
509impl InMemoryReadReceiptRepository {
510 pub fn new(state: SharedState) -> Self {
511 Self { state }
512 }
513}
514
515#[async_trait]
516impl ReadReceiptRepository for InMemoryReadReceiptRepository {
517 async fn upsert(&self, ctx: &ChatStorageCtx, input: NewReadReceipt) -> ChatStorageResult<ReadReceipt> {
518 require_user(ctx)?;
519 let mut map = self.state.read_receipts.write().unwrap();
520 let now = OffsetDateTime::now_utc();
521 if let Some(existing) = map
522 .values_mut()
523 .find(|r| r.chat_id == input.chat_id && r.user_id == input.user_id && r.thread_id == input.thread_id)
524 {
525 existing.last_read_message_id = input.last_read_message_id;
526 existing.updated_at = now;
527 return Ok(existing.clone());
528 }
529 let r = ReadReceipt {
530 id: Uuid::new_v4(),
531 chat_id: input.chat_id,
532 user_id: input.user_id,
533 thread_id: input.thread_id,
534 last_read_message_id: input.last_read_message_id,
535 updated_at: now,
536 };
537 map.insert(r.id, r.clone());
538 Ok(r)
539 }
540
541 async fn list_for_chat(&self, _ctx: &ChatStorageCtx, chat_id: Uuid) -> ChatStorageResult<Vec<ReadReceipt>> {
542 let map = self.state.read_receipts.read().unwrap();
543 Ok(map.values().filter(|r| r.chat_id == chat_id).cloned().collect())
544 }
545
546 async fn find(&self, _ctx: &ChatStorageCtx, chat_id: Uuid, user_id: Uuid, thread_id: Option<Uuid>) -> ChatStorageResult<Option<ReadReceipt>> {
547 let map = self.state.read_receipts.read().unwrap();
548 Ok(map
549 .values()
550 .find(|r| r.chat_id == chat_id && r.user_id == user_id && r.thread_id == thread_id)
551 .cloned())
552 }
553}
554
555
556#[derive(Debug)]
557pub struct InMemoryPinnedMessageRepository {
558 state: SharedState,
559}
560
561impl InMemoryPinnedMessageRepository {
562 pub fn new(state: SharedState) -> Self {
563 Self { state }
564 }
565}
566
567#[async_trait]
568impl PinnedMessageRepository for InMemoryPinnedMessageRepository {
569 async fn pin(&self, ctx: &ChatStorageCtx, input: NewPinnedMessage) -> ChatStorageResult<PinnedMessage> {
570 require_user(ctx)?;
571 let p = PinnedMessage {
572 id: Uuid::new_v4(),
573 chat_id: input.chat_id,
574 message_id: input.message_id,
575 pinned_by: input.pinned_by,
576 created_at: OffsetDateTime::now_utc(),
577 };
578 self.state.pinned.write().unwrap().insert((p.chat_id, p.message_id), p.clone());
579 Ok(p)
580 }
581
582 async fn unpin(&self, ctx: &ChatStorageCtx, chat_id: Uuid, message_id: Uuid) -> ChatStorageResult<()> {
583 require_user(ctx)?;
584 let removed = self.state.pinned.write().unwrap().remove(&(chat_id, message_id)).is_some();
585 if !removed {
586 return Err(ChatStorageError::PinnedMessageNotFound(message_id));
587 }
588 Ok(())
589 }
590
591 async fn list_for_chat(&self, _ctx: &ChatStorageCtx, chat_id: Uuid) -> ChatStorageResult<Vec<PinnedMessage>> {
592 let map = self.state.pinned.read().unwrap();
593 Ok(map.values().filter(|p| p.chat_id == chat_id).cloned().collect())
594 }
595}
596
597
598#[derive(Debug)]
599pub struct InMemoryThreadRepository {
600 state: SharedState,
601}
602
603impl InMemoryThreadRepository {
604 pub fn new(state: SharedState) -> Self {
605 Self { state }
606 }
607}
608
609#[async_trait]
610impl ThreadRepository for InMemoryThreadRepository {
611 async fn increment_reply(&self, ctx: &ChatStorageCtx, chat_id: Uuid, root_message_id: Uuid, reply_by: Uuid, at: OffsetDateTime) -> ChatStorageResult<Thread> {
612 require_user(ctx)?;
613 let mut map = self.state.threads.write().unwrap();
614 let entry = map.entry((chat_id, root_message_id)).or_insert(Thread {
615 chat_id,
616 root_message_id,
617 reply_count: 0,
618 last_reply_at: None,
619 last_reply_by: None,
620 });
621 entry.reply_count = entry.reply_count.saturating_add(1);
622 entry.last_reply_at = Some(at);
623 entry.last_reply_by = Some(reply_by);
624 Ok(entry.clone())
625 }
626
627 async fn find(&self, _ctx: &ChatStorageCtx, chat_id: Uuid, root_message_id: Uuid) -> ChatStorageResult<Option<Thread>> {
628 let map = self.state.threads.read().unwrap();
629 Ok(map.get(&(chat_id, root_message_id)).cloned())
630 }
631
632 async fn list_for_chat(&self, _ctx: &ChatStorageCtx, chat_id: Uuid) -> ChatStorageResult<Vec<Thread>> {
633 let map = self.state.threads.read().unwrap();
634 Ok(map.values().filter(|t| t.chat_id == chat_id).cloned().collect())
635 }
636}
637
638
639pub struct InMemoryChatStorage {
640 state: SharedState,
641 blobs: Arc<dyn ChatMediaStore>,
642}
643
644impl std::fmt::Debug for InMemoryChatStorage {
645 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
646 f.debug_struct("InMemoryChatStorage")
647 .field("state", &self.state)
648 .field("blobs", &"<dyn ChatMediaStore>")
649 .finish()
650 }
651}
652
653impl InMemoryChatStorage {
654 pub fn new(blobs: Arc<dyn ChatMediaStore>) -> Self {
655 Self {
656 state: Arc::new(InMemoryState::default()),
657 blobs,
658 }
659 }
660
661 pub fn new_shared() -> DynChatStorage {
662 let blobs: Arc<dyn ChatMediaStore> = Arc::new(NullChatMedia);
663 Arc::new(Self::new(blobs))
664 }
665}
666
667#[derive(Debug)]
668pub struct NullChatMedia;
669
670#[async_trait]
671impl ChatMediaStore for NullChatMedia {
672 async fn put_attachment_bytes(
673 &self,
674 _owner: AttachmentOwner,
675 _owner_id: Uuid,
676 _filename: &str,
677 _bytes: Bytes,
678 ) -> ChatStorageResult<AttachmentPath> {
679 Err(ChatStorageError::Blob(storage::StorageError::Unsupported(
680 "null backend".to_string(),
681 )))
682 }
683 async fn get_attachment_bytes(&self, _path: &AttachmentPath) -> ChatStorageResult<Bytes> {
684 Err(ChatStorageError::Blob(storage::StorageError::Unsupported(
685 "null backend".to_string(),
686 )))
687 }
688 async fn delete_attachment(&self, _path: &AttachmentPath) -> ChatStorageResult<()> {
689 Err(ChatStorageError::Blob(storage::StorageError::Unsupported(
690 "null backend".to_string(),
691 )))
692 }
693 async fn presign_upload(
694 &self,
695 _owner: AttachmentOwner,
696 _owner_id: Uuid,
697 _filename: &str,
698 _ttl: Duration,
699 ) -> ChatStorageResult<Url> {
700 Err(ChatStorageError::Blob(storage::StorageError::Unsupported(
701 "null backend".to_string(),
702 )))
703 }
704 async fn presign_download(&self, _path: &AttachmentPath, _ttl: Duration) -> ChatStorageResult<Url> {
705 Err(ChatStorageError::Blob(storage::StorageError::Unsupported(
706 "null backend".to_string(),
707 )))
708 }
709}
710
711#[async_trait]
712impl ChatStorage for InMemoryChatStorage {
713 fn conversations(&self) -> DynConversationRepository {
714 Arc::new(InMemoryConversationRepository::new(self.state.clone()))
715 }
716 fn messages(&self) -> DynMessageRepository {
717 Arc::new(InMemoryMessageRepository::new(self.state.clone()))
718 }
719 fn attachments(&self) -> DynAttachmentRepository {
720 Arc::new(InMemoryAttachmentRepository::new(self.state.clone()))
721 }
722 fn members(&self) -> DynMemberRepository {
723 Arc::new(InMemoryMemberRepository::new(self.state.clone()))
724 }
725 fn reactions(&self) -> DynReactionRepository {
726 Arc::new(InMemoryReactionRepository::new(self.state.clone()))
727 }
728 fn mentions(&self) -> DynMentionRepository {
729 Arc::new(InMemoryMentionRepository::new(self.state.clone()))
730 }
731 fn reads(&self) -> DynReadReceiptRepository {
732 Arc::new(InMemoryReadReceiptRepository::new(self.state.clone()))
733 }
734 fn pinned(&self) -> DynPinnedMessageRepository {
735 Arc::new(InMemoryPinnedMessageRepository::new(self.state.clone()))
736 }
737 fn threads(&self) -> DynThreadRepository {
738 Arc::new(InMemoryThreadRepository::new(self.state.clone()))
739 }
740 fn blobs(&self) -> Arc<dyn ChatMediaStore> {
741 self.blobs.clone()
742 }
743}
No newline at end of file
Added chat/storagelayer/models/attachment.rs View chat/storagelayer/models/attachment.rs
@@ -0,0 +1,66 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
6#[serde(rename_all = "snake_case")]
7pub enum AttachmentOwner {
8 Issue,
9 Comment,
10 Release,
11 ChatMessage,
12 Post,
13 PostReply,
14}
15
16impl AttachmentOwner {
17 pub fn as_str(&self) -> &'static str {
18 match self {
19 Self::Issue => "issue",
20 Self::Comment => "comment",
21 Self::Release => "release",
22 Self::ChatMessage => "chat_message",
23 Self::Post => "post",
24 Self::PostReply => "post_reply",
25 }
26 }
27}
28
29impl std::str::FromStr for AttachmentOwner {
30 type Err = String;
31 fn from_str(s: &str) -> Result<Self, Self::Err> {
32 match s {
33 "issue" => Ok(Self::Issue),
34 "comment" => Ok(Self::Comment),
35 "release" => Ok(Self::Release),
36 "chat_message" => Ok(Self::ChatMessage),
37 "post" => Ok(Self::Post),
38 "post_reply" => Ok(Self::PostReply),
39 other => Err(format!("unknown attachment owner: {other}")),
40 }
41 }
42}
43
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45pub struct Attachment {
46 pub id: Uuid,
47 pub owner: AttachmentOwner,
48 pub owner_id: Uuid,
49 pub filename: String,
50 pub content_type: String,
51 pub size: i64,
52 pub url: String,
53 pub uploaded_by: Option<Uuid>,
54 pub created_at: OffsetDateTime,
55}
56
57#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
58pub struct NewAttachment {
59 pub owner: AttachmentOwner,
60 pub owner_id: Uuid,
61 pub filename: String,
62 pub content_type: String,
63 pub size: i64,
64 pub url: String,
65 pub uploaded_by: Option<Uuid>,
66}
No newline at end of file
Added chat/storagelayer/models/conversation.rs View chat/storagelayer/models/conversation.rs
@@ -0,0 +1,40 @@
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 ConversationKind {
8 #[default]
9 Group,
10 Private,
11}
12
13#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
14pub struct Conversation {
15 pub id: Uuid,
16 pub channel_id: Uuid,
17 pub kind: ConversationKind,
18 pub counterpart_user_id: Option<Uuid>,
19 pub title: Option<String>,
20 pub created_at: OffsetDateTime,
21 pub updated_at: Option<OffsetDateTime>,
22}
23
24#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
25pub struct ConversationMember {
26 pub chat_id: Uuid,
27 pub user_id: Uuid,
28 pub joined_at: OffsetDateTime,
29 pub last_read_at: Option<OffsetDateTime>,
30 pub muted: bool,
31}
32
33#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
34pub struct NewConversation {
35 pub channel_id: Uuid,
36 pub kind: ConversationKind,
37 pub counterpart_user_id: Option<Uuid>,
38 pub title: Option<String>,
39 pub initial_members: Vec<Uuid>,
40}
No newline at end of file
Added chat/storagelayer/models/mention.rs View chat/storagelayer/models/mention.rs
@@ -0,0 +1,25 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
6pub struct Mention {
7 pub id: Uuid,
8 pub message_id: Uuid,
9 pub user_id: Option<Uuid>,
10 pub role_id: Option<Uuid>,
11 pub channel_id: Option<Uuid>,
12 pub is_everyone: bool,
13 pub is_here: bool,
14 pub created_at: OffsetDateTime,
15}
16
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18pub struct NewMention {
19 pub message_id: Uuid,
20 pub user_id: Option<Uuid>,
21 pub role_id: Option<Uuid>,
22 pub channel_id: Option<Uuid>,
23 pub is_everyone: bool,
24 pub is_here: bool,
25}
No newline at end of file
Added chat/storagelayer/models/message.rs View chat/storagelayer/models/message.rs
@@ -0,0 +1,211 @@
1use serde::{Deserialize, Serialize};
2use std::fmt;
3use time::OffsetDateTime;
4use uuid::Uuid;
5
6#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)]
7#[serde(transparent)]
8pub struct MessageFlags(i64);
9
10impl MessageFlags {
11 pub const EDITED_FLAG: i64 = 1 << 0;
12 pub const DELETED_FLAG: i64 = 1 << 1;
13 pub const THREAD_START_FLAG: i64 = 1 << 2;
14 pub const PINNED_FLAG: i64 = 1 << 3;
15 pub const SYSTEM_FLAG: i64 = 1 << 4;
16 pub const SILENT_FLAG: i64 = 1 << 5;
17
18 pub const fn empty() -> Self {
19 Self(0)
20 }
21
22 pub const fn from_bits(bits: i64) -> Self {
23 Self(bits)
24 }
25
26 pub const fn from_bits_truncate(bits: i64) -> Self {
27 Self(bits)
28 }
29
30 pub const fn bits(self) -> i64 {
31 self.0
32 }
33
34 pub const fn contains(self, other: Self) -> bool {
35 (self.0 & other.0) == other.0
36 }
37
38 pub const fn intersects(self, other: Self) -> bool {
39 (self.0 & other.0) != 0
40 }
41
42 pub const fn is_empty(self) -> bool {
43 self.0 == 0
44 }
45
46 pub fn set(&mut self, other: Self) {
47 self.0 |= other.0;
48 }
49
50 pub fn unset(&mut self, other: Self) {
51 self.0 &= !other.0;
52 }
53
54 pub fn with(self, other: Self) -> Self {
55 Self(self.0 | other.0)
56 }
57
58 pub fn without(self, other: Self) -> Self {
59 Self(self.0 & !other.0)
60 }
61
62 pub const EDITED: Self = Self(Self::EDITED_FLAG);
63 pub const DELETED: Self = Self(Self::DELETED_FLAG);
64 pub const THREAD_START: Self = Self(Self::THREAD_START_FLAG);
65 pub const PINNED: Self = Self(Self::PINNED_FLAG);
66 pub const SYSTEM: Self = Self(Self::SYSTEM_FLAG);
67 pub const SILENT: Self = Self(Self::SILENT_FLAG);
68}
69
70impl fmt::Display for MessageFlags {
71 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
72 let mut first = true;
73 let mut emit = |label: &str| -> fmt::Result {
74 if !first {
75 f.write_str(" | ")?;
76 }
77 first = false;
78 f.write_str(label)
79 };
80 if self.contains(Self::EDITED) {
81 emit("EDITED")?;
82 }
83 if self.contains(Self::DELETED) {
84 emit("DELETED")?;
85 }
86 if self.contains(Self::THREAD_START) {
87 emit("THREAD_START")?;
88 }
89 if self.contains(Self::PINNED) {
90 emit("PINNED")?;
91 }
92 if self.contains(Self::SYSTEM) {
93 emit("SYSTEM")?;
94 }
95 if self.contains(Self::SILENT) {
96 emit("SILENT")?;
97 }
98 if first {
99 f.write_str("0")?;
100 }
101 Ok(())
102 }
103}
104
105impl std::ops::BitOr for MessageFlags {
106 type Output = Self;
107 fn bitor(self, rhs: Self) -> Self {
108 self.with(rhs)
109 }
110}
111
112impl std::ops::BitOrAssign for MessageFlags {
113 fn bitor_assign(&mut self, rhs: Self) {
114 self.set(rhs);
115 }
116}
117
118impl std::ops::BitAnd for MessageFlags {
119 type Output = Self;
120 fn bitand(self, rhs: Self) -> Self {
121 Self(self.0 & rhs.0)
122 }
123}
124
125impl std::ops::BitAndAssign for MessageFlags {
126 fn bitand_assign(&mut self, rhs: Self) {
127 self.0 &= rhs.0;
128 }
129}
130
131#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
132pub struct Message {
133 pub id: Uuid,
134 pub chat_id: Uuid,
135 pub sender_id: Uuid,
136 pub body: String,
137 pub reply_to_message_id: Option<Uuid>,
138 pub thread_id: Option<Uuid>,
139 pub is_thread_start: bool,
140 pub flags: i64,
141 pub deleted_at: Option<OffsetDateTime>,
142 pub deleted_by: Option<Uuid>,
143 pub edited_at: Option<OffsetDateTime>,
144 pub created_at: OffsetDateTime,
145}
146
147impl Message {
148 pub fn flags_typed(&self) -> MessageFlags {
149 MessageFlags::from_bits_truncate(self.flags)
150 }
151}
152
153#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
154pub struct NewMessage {
155 pub chat_id: Uuid,
156 pub sender_id: Uuid,
157 pub body: String,
158 pub reply_to_message_id: Option<Uuid>,
159 pub thread_id: Option<Uuid>,
160 pub is_thread_start: bool,
161 pub flags: Option<i64>,
162 pub attachment_ids: Vec<Uuid>,
163}
164
165#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
166pub struct MessagePage {
167 pub items: Vec<Message>,
168 pub next_cursor: Option<String>,
169}
170
171#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
172#[serde(rename_all = "snake_case")]
173pub enum ListDirection {
174 #[default]
175 Forward,
176 Backward,
177}
178
179#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
180pub struct MessageQuery {
181 pub direction: ListDirection,
182 pub limit: u32,
183 pub cursor: Option<String>,
184 pub include_deleted: bool,
185}
186
187impl Default for MessageQuery {
188 fn default() -> Self {
189 Self {
190 direction: ListDirection::Forward,
191 limit: 50,
192 cursor: None,
193 include_deleted: false,
194 }
195 }
196}
197
198#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
199pub struct Page {
200 pub limit: u32,
201 pub cursor: Option<String>,
202}
203
204impl Default for Page {
205 fn default() -> Self {
206 Self {
207 limit: 50,
208 cursor: None,
209 }
210 }
211}
No newline at end of file
Added chat/storagelayer/models/pinned.rs View chat/storagelayer/models/pinned.rs
@@ -0,0 +1,19 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
6pub struct PinnedMessage {
7 pub id: Uuid,
8 pub chat_id: Uuid,
9 pub message_id: Uuid,
10 pub pinned_by: Uuid,
11 pub created_at: OffsetDateTime,
12}
13
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15pub struct NewPinnedMessage {
16 pub chat_id: Uuid,
17 pub message_id: Uuid,
18 pub pinned_by: Uuid,
19}
No newline at end of file
Added chat/storagelayer/models/reaction.rs View chat/storagelayer/models/reaction.rs
@@ -0,0 +1,19 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
6pub struct Reaction {
7 pub id: Uuid,
8 pub message_id: Uuid,
9 pub user_id: Uuid,
10 pub emoji: String,
11 pub created_at: OffsetDateTime,
12}
13
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15pub struct NewReaction {
16 pub message_id: Uuid,
17 pub user_id: Uuid,
18 pub emoji: String,
19}
No newline at end of file
Added chat/storagelayer/models/read_receipt.rs View chat/storagelayer/models/read_receipt.rs
@@ -0,0 +1,21 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
6pub struct ReadReceipt {
7 pub id: Uuid,
8 pub chat_id: Uuid,
9 pub user_id: Uuid,
10 pub thread_id: Option<Uuid>,
11 pub last_read_message_id: Option<Uuid>,
12 pub updated_at: OffsetDateTime,
13}
14
15#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
16pub struct NewReadReceipt {
17 pub chat_id: Uuid,
18 pub user_id: Uuid,
19 pub thread_id: Option<Uuid>,
20 pub last_read_message_id: Option<Uuid>,
21}
No newline at end of file
Added chat/storagelayer/models/thread.rs View chat/storagelayer/models/thread.rs
@@ -0,0 +1,12 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
6pub struct Thread {
7 pub chat_id: Uuid,
8 pub root_message_id: Uuid,
9 pub reply_count: i32,
10 pub last_reply_at: Option<OffsetDateTime>,
11 pub last_reply_by: Option<Uuid>,
12}
No newline at end of file
Added chat/storagelayer/repository/attachment.rs View chat/storagelayer/repository/attachment.rs
@@ -0,0 +1,36 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use uuid::Uuid;
4
5use crate::error::ChatStorageResult;
6use crate::models::attachment::{Attachment, AttachmentOwner, NewAttachment};
7use crate::models::message::Page;
8
9use super::ChatStorageCtx;
10
11#[async_trait]
12pub trait AttachmentRepository: Send + Sync {
13 async fn create_metadata(
14 &self,
15 ctx: &ChatStorageCtx,
16 input: NewAttachment,
17 ) -> ChatStorageResult<Attachment>;
18
19 async fn find_by_id(
20 &self,
21 ctx: &ChatStorageCtx,
22 id: Uuid,
23 ) -> ChatStorageResult<Option<Attachment>>;
24
25 async fn list_for_owner(
26 &self,
27 ctx: &ChatStorageCtx,
28 owner: AttachmentOwner,
29 owner_id: Uuid,
30 page: Page,
31 ) -> ChatStorageResult<Vec<Attachment>>;
32
33 async fn delete(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()>;
34}
35
36pub type DynAttachmentRepository = Arc<dyn AttachmentRepository>;
No newline at end of file
Added chat/storagelayer/repository/conversation.rs View chat/storagelayer/repository/conversation.rs
@@ -0,0 +1,59 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3
4use crate::error::ChatStorageResult;
5use crate::models::conversation::{
6 Conversation, ConversationKind, ConversationMember, NewConversation,
7};
8use crate::models::message::Page;
9
10use super::ChatStorageCtx;
11
12#[async_trait]
13pub trait ConversationRepository: Send + Sync {
14 async fn create(
15 &self,
16 ctx: &ChatStorageCtx,
17 input: NewConversation,
18 ) -> ChatStorageResult<Conversation>;
19
20 async fn find_by_id(
21 &self,
22 ctx: &ChatStorageCtx,
23 id: uuid::Uuid,
24 ) -> ChatStorageResult<Option<Conversation>>;
25
26 async fn list_by_channel(
27 &self,
28 ctx: &ChatStorageCtx,
29 channel_id: uuid::Uuid,
30 page: Page,
31 ) -> ChatStorageResult<Vec<Conversation>>;
32
33 async fn update_title(
34 &self,
35 ctx: &ChatStorageCtx,
36 id: uuid::Uuid,
37 title: Option<String>,
38 ) -> ChatStorageResult<()>;
39
40 async fn update_kind(
41 &self,
42 ctx: &ChatStorageCtx,
43 id: uuid::Uuid,
44 kind: ConversationKind,
45 ) -> ChatStorageResult<()>;
46
47 async fn delete(&self, ctx: &ChatStorageCtx, id: uuid::Uuid) -> ChatStorageResult<()>;
48
49 async fn exists(&self, ctx: &ChatStorageCtx, id: uuid::Uuid) -> ChatStorageResult<bool>;
50
51 async fn list_members(
52 &self,
53 ctx: &ChatStorageCtx,
54 chat_id: uuid::Uuid,
55 page: Page,
56 ) -> ChatStorageResult<Vec<ConversationMember>>;
57}
58
59pub type DynConversationRepository = Arc<dyn ConversationRepository>;
No newline at end of file
Added chat/storagelayer/repository/member.rs View chat/storagelayer/repository/member.rs
@@ -0,0 +1,59 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use time::OffsetDateTime;
4use uuid::Uuid;
5
6use crate::error::ChatStorageResult;
7use crate::models::conversation::ConversationMember;
8use crate::models::message::Page;
9
10use super::ChatStorageCtx;
11
12#[async_trait]
13pub trait MemberRepository: Send + Sync {
14 async fn add_member(
15 &self,
16 ctx: &ChatStorageCtx,
17 chat_id: Uuid,
18 user_id: Uuid,
19 ) -> ChatStorageResult<ConversationMember>;
20
21 async fn remove_member(
22 &self,
23 ctx: &ChatStorageCtx,
24 chat_id: Uuid,
25 user_id: Uuid,
26 ) -> ChatStorageResult<()>;
27
28 async fn list_members(
29 &self,
30 ctx: &ChatStorageCtx,
31 chat_id: Uuid,
32 page: Page,
33 ) -> ChatStorageResult<Vec<ConversationMember>>;
34
35 async fn mark_read(
36 &self,
37 ctx: &ChatStorageCtx,
38 chat_id: Uuid,
39 user_id: Uuid,
40 at: OffsetDateTime,
41 ) -> ChatStorageResult<()>;
42
43 async fn set_muted(
44 &self,
45 ctx: &ChatStorageCtx,
46 chat_id: Uuid,
47 user_id: Uuid,
48 muted: bool,
49 ) -> ChatStorageResult<()>;
50
51 async fn is_member(
52 &self,
53 ctx: &ChatStorageCtx,
54 chat_id: Uuid,
55 user_id: Uuid,
56 ) -> ChatStorageResult<bool>;
57}
58
59pub type DynMemberRepository = Arc<dyn MemberRepository>;
No newline at end of file
Added chat/storagelayer/repository/mention.rs View chat/storagelayer/repository/mention.rs
@@ -0,0 +1,32 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use uuid::Uuid;
4
5use crate::error::ChatStorageResult;
6use crate::models::mention::{Mention, NewMention};
7
8use super::ChatStorageCtx;
9
10#[async_trait]
11pub trait MentionRepository: Send + Sync {
12 async fn add_mention(
13 &self,
14 ctx: &ChatStorageCtx,
15 input: NewMention,
16 ) -> ChatStorageResult<Mention>;
17
18 async fn list_for_message(
19 &self,
20 ctx: &ChatStorageCtx,
21 message_id: Uuid,
22 ) -> ChatStorageResult<Vec<Mention>>;
23
24 async fn list_for_user(
25 &self,
26 ctx: &ChatStorageCtx,
27 user_id: Uuid,
28 limit: u32,
29 ) -> ChatStorageResult<Vec<Mention>>;
30}
31
32pub type DynMentionRepository = Arc<dyn MentionRepository>;
No newline at end of file
Added chat/storagelayer/repository/message.rs View chat/storagelayer/repository/message.rs
@@ -0,0 +1,59 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use uuid::Uuid;
4
5use crate::error::ChatStorageResult;
6use crate::models::message::{Message, MessagePage, MessageQuery, NewMessage, Page};
7
8use super::ChatStorageCtx;
9
10#[async_trait]
11pub trait MessageRepository: Send + Sync {
12 async fn create(
13 &self,
14 ctx: &ChatStorageCtx,
15 input: NewMessage,
16 ) -> ChatStorageResult<Message>;
17
18 async fn find_by_id(
19 &self,
20 ctx: &ChatStorageCtx,
21 id: Uuid,
22 ) -> ChatStorageResult<Option<Message>>;
23
24 async fn list_by_conversation(
25 &self,
26 ctx: &ChatStorageCtx,
27 chat_id: Uuid,
28 query: MessageQuery,
29 ) -> ChatStorageResult<MessagePage>;
30
31 async fn list_thread(
32 &self,
33 ctx: &ChatStorageCtx,
34 chat_id: Uuid,
35 root_message_id: Uuid,
36 page: Page,
37 ) -> ChatStorageResult<MessagePage>;
38
39 async fn edit(
40 &self,
41 ctx: &ChatStorageCtx,
42 id: Uuid,
43 editor: Uuid,
44 new_body: String,
45 ) -> ChatStorageResult<Message>;
46
47 async fn soft_delete(
48 &self,
49 ctx: &ChatStorageCtx,
50 id: Uuid,
51 by: Uuid,
52 ) -> ChatStorageResult<()>;
53
54 async fn restore(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()>;
55
56 async fn exists(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<bool>;
57}
58
59pub type DynMessageRepository = Arc<dyn MessageRepository>;
No newline at end of file
Added chat/storagelayer/repository/pinned.rs View chat/storagelayer/repository/pinned.rs
@@ -0,0 +1,32 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use uuid::Uuid;
4
5use crate::error::ChatStorageResult;
6use crate::models::pinned::{NewPinnedMessage, PinnedMessage};
7
8use super::ChatStorageCtx;
9
10#[async_trait]
11pub trait PinnedMessageRepository: Send + Sync {
12 async fn pin(
13 &self,
14 ctx: &ChatStorageCtx,
15 input: NewPinnedMessage,
16 ) -> ChatStorageResult<PinnedMessage>;
17
18 async fn unpin(
19 &self,
20 ctx: &ChatStorageCtx,
21 chat_id: Uuid,
22 message_id: Uuid,
23 ) -> ChatStorageResult<()>;
24
25 async fn list_for_chat(
26 &self,
27 ctx: &ChatStorageCtx,
28 chat_id: Uuid,
29 ) -> ChatStorageResult<Vec<PinnedMessage>>;
30}
31
32pub type DynPinnedMessageRepository = Arc<dyn PinnedMessageRepository>;
No newline at end of file
Added chat/storagelayer/repository/reaction.rs View chat/storagelayer/repository/reaction.rs
@@ -0,0 +1,33 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use uuid::Uuid;
4
5use crate::error::ChatStorageResult;
6use crate::models::reaction::{NewReaction, Reaction};
7
8use super::ChatStorageCtx;
9
10#[async_trait]
11pub trait ReactionRepository: Send + Sync {
12 async fn add_reaction(
13 &self,
14 ctx: &ChatStorageCtx,
15 input: NewReaction,
16 ) -> ChatStorageResult<Reaction>;
17
18 async fn remove_reaction(
19 &self,
20 ctx: &ChatStorageCtx,
21 message_id: Uuid,
22 user_id: Uuid,
23 emoji: &str,
24 ) -> ChatStorageResult<()>;
25
26 async fn list_reactions(
27 &self,
28 ctx: &ChatStorageCtx,
29 message_id: Uuid,
30 ) -> ChatStorageResult<Vec<Reaction>>;
31}
32
33pub type DynReactionRepository = Arc<dyn ReactionRepository>;
No newline at end of file
Added chat/storagelayer/repository/read_receipt.rs View chat/storagelayer/repository/read_receipt.rs
@@ -0,0 +1,33 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use uuid::Uuid;
4
5use crate::error::ChatStorageResult;
6use crate::models::read_receipt::{NewReadReceipt, ReadReceipt};
7
8use super::ChatStorageCtx;
9
10#[async_trait]
11pub trait ReadReceiptRepository: Send + Sync {
12 async fn upsert(
13 &self,
14 ctx: &ChatStorageCtx,
15 input: NewReadReceipt,
16 ) -> ChatStorageResult<ReadReceipt>;
17
18 async fn list_for_chat(
19 &self,
20 ctx: &ChatStorageCtx,
21 chat_id: Uuid,
22 ) -> ChatStorageResult<Vec<ReadReceipt>>;
23
24 async fn find(
25 &self,
26 ctx: &ChatStorageCtx,
27 chat_id: Uuid,
28 user_id: Uuid,
29 thread_id: Option<Uuid>,
30 ) -> ChatStorageResult<Option<ReadReceipt>>;
31}
32
33pub type DynReadReceiptRepository = Arc<dyn ReadReceiptRepository>;
No newline at end of file
Added chat/storagelayer/repository/thread.rs View chat/storagelayer/repository/thread.rs
@@ -0,0 +1,36 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use time::OffsetDateTime;
4use uuid::Uuid;
5
6use crate::error::ChatStorageResult;
7use crate::models::thread::Thread;
8
9use super::ChatStorageCtx;
10
11#[async_trait]
12pub trait ThreadRepository: Send + Sync {
13 async fn increment_reply(
14 &self,
15 ctx: &ChatStorageCtx,
16 chat_id: Uuid,
17 root_message_id: Uuid,
18 reply_by: Uuid,
19 at: OffsetDateTime,
20 ) -> ChatStorageResult<Thread>;
21
22 async fn find(
23 &self,
24 ctx: &ChatStorageCtx,
25 chat_id: Uuid,
26 root_message_id: Uuid,
27 ) -> ChatStorageResult<Option<Thread>>;
28
29 async fn list_for_chat(
30 &self,
31 ctx: &ChatStorageCtx,
32 chat_id: Uuid,
33 ) -> ChatStorageResult<Vec<Thread>>;
34}
35
36pub type DynThreadRepository = Arc<dyn ThreadRepository>;
No newline at end of file
Added chat/storagelayer/blob.rs View chat/storagelayer/blob.rs
@@ -0,0 +1,5 @@
1pub mod media;
2pub mod object_store;
3
4pub use media::{AttachmentPath, ChatMediaStore, DynChatMediaStore};
5pub use object_store::ObjectStoreChatMedia;
No newline at end of file
Added chat/storagelayer/Cargo.toml View chat/storagelayer/Cargo.toml
@@ -0,0 +1,38 @@
1[package]
2name = "storagelayer"
3version.workspace = true
4edition.workspace = true
5authors.workspace = true
6categories.workspace = true
7keywords.workspace = true
8repository.workspace = true
9homepage.workspace = true
10
11[lib]
12path = "lib.rs"
13name = "storagelayer"
14
15[dependencies]
16storage.workspace = true
17db.workspace = true
18queue.workspace = true
19jwt.workspace = true
20cache.workspace = true
21
22tokio.workspace = true
23async-trait.workspace = true
24thiserror.workspace = true
25serde.workspace = true
26time.workspace = true
27uuid.workspace = true
28bytes.workspace = true
29object_store.workspace = true
30tracing.workspace = true
31url.workspace = true
32sea-orm.workspace = true
33
34futures = { workspace = true }
35
36[features]
37default = []
38mock = []
No newline at end of file
Added chat/storagelayer/error.rs View chat/storagelayer/error.rs
@@ -0,0 +1,55 @@
1use thiserror::Error;
2use uuid::Uuid;
3
4#[derive(Debug, Error)]
5pub enum ChatStorageError {
6 #[error("message not found: {0}")]
7 MessageNotFound(Uuid),
8
9 #[error("conversation not found: {0}")]
10 ConversationNotFound(Uuid),
11
12 #[error("attachment not found: {0}")]
13 AttachmentNotFound(Uuid),
14
15 #[error("member not found: chat={0} user={1}")]
16 MemberNotFound(Uuid, Uuid),
17
18 #[error("reaction not found: {0}")]
19 ReactionNotFound(Uuid),
20
21 #[error("mention not found: {0}")]
22 MentionNotFound(Uuid),
23
24 #[error("read receipt not found: {0}")]
25 ReadReceiptNotFound(Uuid),
26
27 #[error("pinned message not found: {0}")]
28 PinnedMessageNotFound(Uuid),
29
30 #[error("thread not found: chat={0} root={1}")]
31 ThreadNotFound(Uuid, Uuid),
32
33 #[error("invalid input: {0}")]
34 InvalidInput(String),
35
36 #[error("permission denied: {0}")]
37 PermissionDenied(String),
38
39 #[error("conflict: {0}")]
40 Conflict(String),
41
42 #[error("database error: {0}")]
43 Database(#[from] db::conn::DbError),
44
45 #[error("sea_orm error: {0}")]
46 SeaOrm(#[from] sea_orm::DbErr),
47
48 #[error("blob storage error: {0}")]
49 Blob(#[from] storage::StorageError),
50
51 #[error("internal error: {0}")]
52 Internal(String),
53}
54
55pub type ChatStorageResult<T> = Result<T, ChatStorageError>;
No newline at end of file
Added chat/storagelayer/lib.rs View chat/storagelayer/lib.rs
@@ -0,0 +1,45 @@
1//! # #[cfg(feature = "mock")]
2#![warn(missing_debug_implementations)]
3#![warn(unreachable_pub)]
4
5pub mod blob;
6pub mod error;
7pub mod mapping;
8pub mod models;
9pub mod repository;
10
11#[cfg(feature = "mock")]
12pub mod mock;
13
14pub use blob::{
15 media::{AttachmentPath, ChatMediaStore, DynChatMediaStore},
16 object_store::ObjectStoreChatMedia,
17};
18pub use error::{ChatStorageError, ChatStorageResult};
19pub use models::{
20 attachment::{Attachment, AttachmentOwner, NewAttachment},
21 conversation::{Conversation, ConversationKind, ConversationMember, NewConversation},
22 mention::{Mention, NewMention},
23 message::{
24 ListDirection, Message, MessageFlags, MessagePage, MessageQuery, NewMessage, Page,
25 },
26 pinned::{NewPinnedMessage, PinnedMessage},
27 reaction::{NewReaction, Reaction},
28 read_receipt::{NewReadReceipt, ReadReceipt},
29 thread::Thread,
30};
31pub use repository::{
32 attachment::AttachmentRepository,
33 conversation::ConversationRepository,
34 member::MemberRepository,
35 mention::MentionRepository,
36 message::MessageRepository,
37 pinned::PinnedMessageRepository,
38 reaction::ReactionRepository,
39 read_receipt::ReadReceiptRepository,
40 thread::ThreadRepository,
41 ChatStorage, ChatStorageCtx, ChatStorageFactory, DynChatStorage,
42};
43
44#[cfg(feature = "mock")]
45pub use mock::in_memory::InMemoryChatStorage;
No newline at end of file
Added chat/storagelayer/mapping.rs View chat/storagelayer/mapping.rs
@@ -0,0 +1,315 @@
1use time::OffsetDateTime;
2use uuid::Uuid;
3
4use db::attachment::attachments as att_entity;
5use db::chat::chat_message_edits as edit_entity;
6use db::chat::chat_message_mentions as mention_entity;
7use db::chat::chat_message_reactions as reaction_entity;
8use db::chat::chat_members as member_entity;
9use db::chat::chat_messages as message_entity;
10use db::chat::chat_pinned_messages as pinned_entity;
11use db::chat::chat_reads as read_entity;
12use db::chat::chats as chat_entity;
13use db::chat::threads as thread_entity;
14use sea_orm::Set;
15
16use crate::error::{ChatStorageError, ChatStorageResult};
17use crate::models::attachment::{Attachment, AttachmentOwner};
18use crate::models::conversation::{Conversation, ConversationKind, ConversationMember};
19use crate::models::mention::{Mention, NewMention};
20use crate::models::message::{Message, NewMessage};
21use crate::models::pinned::{NewPinnedMessage, PinnedMessage};
22use crate::models::reaction::{NewReaction, Reaction};
23use crate::models::read_receipt::{NewReadReceipt, ReadReceipt};
24use crate::models::thread::Thread;
25
26
27impl From<chat_entity::ChatKind> for ConversationKind {
28 fn from(kind: chat_entity::ChatKind) -> Self {
29 match kind {
30 chat_entity::ChatKind::Group => Self::Group,
31 chat_entity::ChatKind::Private => Self::Private,
32 }
33 }
34}
35
36impl From<ConversationKind> for chat_entity::ChatKind {
37 fn from(kind: ConversationKind) -> Self {
38 match kind {
39 ConversationKind::Group => Self::Group,
40 ConversationKind::Private => Self::Private,
41 }
42 }
43}
44
45impl From<att_entity::OwnerType> for AttachmentOwner {
46 fn from(owner: att_entity::OwnerType) -> Self {
47 match owner {
48 att_entity::OwnerType::Issue => Self::Issue,
49 att_entity::OwnerType::Comment => Self::Comment,
50 att_entity::OwnerType::Release => Self::Release,
51 att_entity::OwnerType::ChatMessage => Self::ChatMessage,
52 att_entity::OwnerType::Post => Self::Post,
53 att_entity::OwnerType::PostReply => Self::PostReply,
54 }
55 }
56}
57
58impl From<AttachmentOwner> for att_entity::OwnerType {
59 fn from(owner: AttachmentOwner) -> Self {
60 match owner {
61 AttachmentOwner::Issue => Self::Issue,
62 AttachmentOwner::Comment => Self::Comment,
63 AttachmentOwner::Release => Self::Release,
64 AttachmentOwner::ChatMessage => Self::ChatMessage,
65 AttachmentOwner::Post => Self::Post,
66 AttachmentOwner::PostReply => Self::PostReply,
67 }
68 }
69}
70
71
72impl From<chat_entity::Model> for Conversation {
73 fn from(m: chat_entity::Model) -> Self {
74 Self {
75 id: m.id,
76 channel_id: m.channel_id,
77 kind: m.kind.into(),
78 counterpart_user_id: m.counterpart_user_id,
79 title: m.title,
80 created_at: m.created_at,
81 updated_at: m.updated_at,
82 }
83 }
84}
85
86impl TryFrom<Conversation> for chat_entity::ActiveModel {
87 type Error = ChatStorageError;
88 fn try_from(c: Conversation) -> ChatStorageResult<Self> {
89 Ok(Self {
90 id: Set(c.id),
91 channel_id: Set(c.channel_id),
92 kind: Set(c.kind.into()),
93 counterpart_user_id: Set(c.counterpart_user_id),
94 title: Set(c.title),
95 created_at: Set(c.created_at),
96 updated_at: Set(c.updated_at),
97 })
98 }
99}
100
101
102impl From<member_entity::Model> for ConversationMember {
103 fn from(m: member_entity::Model) -> Self {
104 Self {
105 chat_id: m.chat_id,
106 user_id: m.user_id,
107 joined_at: m.joined_at,
108 last_read_at: m.last_read_at,
109 muted: m.muted,
110 }
111 }
112}
113
114impl TryFrom<ConversationMember> for member_entity::ActiveModel {
115 type Error = ChatStorageError;
116 fn try_from(m: ConversationMember) -> ChatStorageResult<Self> {
117 Ok(Self {
118 chat_id: Set(m.chat_id),
119 user_id: Set(m.user_id),
120 joined_at: Set(m.joined_at),
121 last_read_at: Set(m.last_read_at),
122 muted: Set(m.muted),
123 })
124 }
125}
126
127
128impl From<message_entity::Model> for Message {
129 fn from(m: message_entity::Model) -> Self {
130 Self {
131 id: m.id,
132 chat_id: m.chat_id,
133 sender_id: m.sender_id,
134 body: m.body,
135 reply_to_message_id: m.reply_to_message_id,
136 thread_id: m.thread_id,
137 is_thread_start: m.is_thread_start,
138 flags: m.flags,
139 deleted_at: m.deleted_at,
140 deleted_by: m.deleted_by,
141 edited_at: m.edited_at,
142 created_at: m.created_at,
143 }
144 }
145}
146
147impl TryFrom<NewMessage> for message_entity::ActiveModel {
148 type Error = ChatStorageError;
149 fn try_from(m: NewMessage) -> ChatStorageResult<Self> {
150 Ok(Self {
151 id: Set(Uuid::new_v4()),
152 chat_id: Set(m.chat_id),
153 sender_id: Set(m.sender_id),
154 body: Set(m.body),
155 reply_to_message_id: Set(m.reply_to_message_id),
156 thread_id: Set(m.thread_id),
157 is_thread_start: Set(m.is_thread_start),
158 flags: Set(m.flags.unwrap_or(0)),
159 deleted_at: Set(None),
160 deleted_by: Set(None),
161 edited_at: Set(None),
162 created_at: Set(OffsetDateTime::now_utc()),
163 })
164 }
165}
166
167
168impl From<att_entity::Model> for Attachment {
169 fn from(m: att_entity::Model) -> Self {
170 Self {
171 id: m.id,
172 owner: m.owner_type.into(),
173 owner_id: m.owner_id,
174 filename: m.filename,
175 content_type: m.content_type,
176 size: m.size,
177 url: m.url,
178 uploaded_by: m.uploaded_by,
179 created_at: m.created_at,
180 }
181 }
182}
183
184
185impl From<reaction_entity::Model> for Reaction {
186 fn from(m: reaction_entity::Model) -> Self {
187 Self {
188 id: m.id,
189 message_id: m.message_id,
190 user_id: m.user_id,
191 emoji: m.emoji,
192 created_at: m.created_at,
193 }
194 }
195}
196
197impl TryFrom<NewReaction> for reaction_entity::ActiveModel {
198 type Error = ChatStorageError;
199 fn try_from(r: NewReaction) -> ChatStorageResult<Self> {
200 Ok(Self {
201 id: Set(Uuid::new_v4()),
202 message_id: Set(r.message_id),
203 user_id: Set(r.user_id),
204 emoji: Set(r.emoji),
205 created_at: Set(OffsetDateTime::now_utc()),
206 })
207 }
208}
209
210
211impl From<mention_entity::Model> for Mention {
212 fn from(m: mention_entity::Model) -> Self {
213 Self {
214 id: m.id,
215 message_id: m.message_id,
216 user_id: m.user_id,
217 role_id: m.role_id,
218 channel_id: m.channel_id,
219 is_everyone: m.is_everyone,
220 is_here: m.is_here,
221 created_at: m.created_at,
222 }
223 }
224}
225
226impl TryFrom<NewMention> for mention_entity::ActiveModel {
227 type Error = ChatStorageError;
228 fn try_from(m: NewMention) -> ChatStorageResult<Self> {
229 Ok(Self {
230 id: Set(Uuid::new_v4()),
231 message_id: Set(m.message_id),
232 user_id: Set(m.user_id),
233 role_id: Set(m.role_id),
234 channel_id: Set(m.channel_id),
235 is_everyone: Set(m.is_everyone),
236 is_here: Set(m.is_here),
237 created_at: Set(OffsetDateTime::now_utc()),
238 })
239 }
240}
241
242
243impl From<pinned_entity::Model> for PinnedMessage {
244 fn from(m: pinned_entity::Model) -> Self {
245 Self {
246 id: m.id,
247 chat_id: m.chat_id,
248 message_id: m.message_id,
249 pinned_by: m.pinned_by,
250 created_at: m.created_at,
251 }
252 }
253}
254
255impl TryFrom<NewPinnedMessage> for pinned_entity::ActiveModel {
256 type Error = ChatStorageError;
257 fn try_from(p: NewPinnedMessage) -> ChatStorageResult<Self> {
258 Ok(Self {
259 id: Set(Uuid::new_v4()),
260 chat_id: Set(p.chat_id),
261 message_id: Set(p.message_id),
262 pinned_by: Set(p.pinned_by),
263 created_at: Set(OffsetDateTime::now_utc()),
264 })
265 }
266}
267
268
269impl From<read_entity::Model> for ReadReceipt {
270 fn from(m: read_entity::Model) -> Self {
271 Self {
272 id: m.id,
273 chat_id: m.chat_id,
274 user_id: m.user_id,
275 thread_id: m.thread_id,
276 last_read_message_id: m.last_read_message_id,
277 updated_at: m.updated_at,
278 }
279 }
280}
281
282impl TryFrom<NewReadReceipt> for read_entity::ActiveModel {
283 type Error = ChatStorageError;
284 fn try_from(r: NewReadReceipt) -> ChatStorageResult<Self> {
285 Ok(Self {
286 id: Set(Uuid::new_v4()),
287 chat_id: Set(r.chat_id),
288 user_id: Set(r.user_id),
289 thread_id: Set(r.thread_id),
290 last_read_message_id: Set(r.last_read_message_id),
291 updated_at: Set(OffsetDateTime::now_utc()),
292 })
293 }
294}
295
296
297impl From<thread_entity::Model> for Thread {
298 fn from(m: thread_entity::Model) -> Self {
299 Self {
300 chat_id: m.chat_id,
301 root_message_id: m.root_message_id,
302 reply_count: m.reply_count,
303 last_reply_at: m.last_reply_at,
304 last_reply_by: m.last_reply_by,
305 }
306 }
307}
308
309
310#[allow(dead_code)]
311pub(crate) fn message_edit_model_to_fields(
312 m: edit_entity::Model,
313) -> (Uuid, Uuid, Uuid, String, OffsetDateTime) {
314 (m.id, m.message_id, m.editor_id, m.previous_body, m.edited_at)
315}
No newline at end of file
Added chat/storagelayer/mock.rs View chat/storagelayer/mock.rs
@@ -0,0 +1,2 @@
1#[cfg(feature = "mock")]
2pub mod in_memory;
No newline at end of file
Added chat/storagelayer/models.rs View chat/storagelayer/models.rs
@@ -0,0 +1,8 @@
1pub mod attachment;
2pub mod conversation;
3pub mod mention;
4pub mod message;
5pub mod pinned;
6pub mod reaction;
7pub mod read_receipt;
8pub mod thread;
No newline at end of file
Added chat/storagelayer/repository.rs View chat/storagelayer/repository.rs
@@ -0,0 +1,77 @@
1use async_trait::async_trait;
2use std::sync::Arc;
3use uuid::Uuid;
4
5pub mod attachment;
6pub mod conversation;
7pub mod member;
8pub mod mention;
9pub mod message;
10pub mod pinned;
11pub mod reaction;
12pub mod read_receipt;
13pub mod thread;
14
15#[derive(Debug, Clone, Default)]
16pub struct ChatStorageCtx {
17 pub user_id: Option<Uuid>,
18 pub request_id: Option<String>,
19 pub trace_id: Option<String>,
20}
21
22impl ChatStorageCtx {
23 pub fn new(user_id: Uuid) -> Self {
24 Self {
25 user_id: Some(user_id),
26 ..Self::default()
27 }
28 }
29
30 pub fn anonymous() -> Self {
31 Self::default()
32 }
33
34 pub fn with_request_id(mut self, request_id: impl Into<String>) -> Self {
35 self.request_id = Some(request_id.into());
36 self
37 }
38
39 pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
40 self.trace_id = Some(trace_id.into());
41 self
42 }
43
44 pub fn require_user(&self) -> crate::error::ChatStorageResult<Uuid> {
45 self.user_id.ok_or_else(|| {
46 crate::error::ChatStorageError::PermissionDenied(
47 "anonymous context requires authenticated actor".to_string(),
48 )
49 })
50 }
51}
52
53#[async_trait]
54pub trait ChatStorage: Send + Sync {
55 fn conversations(&self) -> conversation::DynConversationRepository;
56 fn messages(&self) -> message::DynMessageRepository;
57 fn attachments(&self) -> attachment::DynAttachmentRepository;
58 fn members(&self) -> member::DynMemberRepository;
59 fn reactions(&self) -> reaction::DynReactionRepository;
60 fn mentions(&self) -> mention::DynMentionRepository;
61 fn reads(&self) -> read_receipt::DynReadReceiptRepository;
62 fn pinned(&self) -> pinned::DynPinnedMessageRepository;
63 fn threads(&self) -> thread::DynThreadRepository;
64 fn blobs(&self) -> Arc<dyn crate::blob::ChatMediaStore>;
65}
66
67pub type DynChatStorage = Arc<dyn ChatStorage>;
68
69#[derive(Debug)]
70pub struct ChatStorageFactory;
71
72impl ChatStorageFactory {
73 #[cfg(feature = "mock")]
74 pub fn in_memory() -> DynChatStorage {
75 crate::InMemoryChatStorage::new_shared()
76 }
77}
No newline at end of file