feat(chat): add storagelayer mapper module (10 entity mappers)
zhenyi committed
Added chat/storagelayer/mapper/attachment.rs chat/storagelayer/mapper/attachment.rs +96 −0 View chat/storagelayer/mapper/attachment.rs
@@ -0,0 +1,96 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set};
5use time::OffsetDateTime;
6use uuid::Uuid;
7
8use crate::error::{ChatStorageError, ChatStorageResult};
9use crate::models::attachment::{Attachment, AttachmentOwner, NewAttachment};
10use crate::Page;
11use crate::repository::attachment::AttachmentRepository;
12use crate::repository::ChatStorageCtx;
13
14use db::attachment::attachments as att_entity;
15
16
17pub struct SeaAttachmentRepository {
18 db: Arc<db::conn::Db>,
19}
20
21impl SeaAttachmentRepository {
22 pub fn new(db: Arc<db::conn::Db>) -> Self {
23 Self { db }
24 }
25}
26
27#[async_trait]
28impl AttachmentRepository for SeaAttachmentRepository {
29 async fn create_metadata(
30 &self,
31 _ctx: &ChatStorageCtx,
32 input: NewAttachment,
33 ) -> ChatStorageResult<Attachment> {
34 let db = self.db.writer();
35 let active = att_entity::ActiveModel {
36 id: Set(Uuid::new_v4()),
37 owner_type: Set(input.owner.into()),
38 owner_id: Set(input.owner_id),
39 filename: Set(input.filename),
40 content_type: Set(input.content_type),
41 size: Set(input.size),
42 url: Set(input.url),
43 uploaded_by: Set(input.uploaded_by),
44 created_at: Set(OffsetDateTime::now_utc()),
45 };
46 let model = att_entity::Entity::insert(active)
47 .exec_with_returning(&db)
48 .await?;
49 Ok(model.into())
50 }
51
52 async fn find_by_id(
53 &self,
54 _ctx: &ChatStorageCtx,
55 id: Uuid,
56 ) -> ChatStorageResult<Option<Attachment>> {
57 let db = self.db.reader();
58 Ok(att_entity::Entity::find_by_id(id)
59 .one(&db)
60 .await?
61 .map(Into::into))
62 }
63
64 async fn list_for_owner(
65 &self,
66 _ctx: &ChatStorageCtx,
67 owner: AttachmentOwner,
68 owner_id: Uuid,
69 page: Page,
70 ) -> ChatStorageResult<Vec<Attachment>> {
71 let db = self.db.reader();
72 let owner_type: att_entity::OwnerType = owner.into();
73 let models = att_entity::Entity::find()
74 .filter(att_entity::Entity::COLUMN.owner_type.eq(owner_type))
75 .filter(att_entity::Entity::COLUMN.owner_id.eq(owner_id))
76 .order_by_asc(att_entity::Entity::COLUMN.created_at)
77 .limit(page.limit as u64)
78 .all(&db)
79 .await?;
80 Ok(models.into_iter().map(Into::into).collect())
81 }
82
83 async fn delete(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {
84 let db = self.db.writer();
85 let res = att_entity::Entity::delete_by_id(id).exec(&db).await?;
86 if res.rows_affected == 0 {
87 return Err(ChatStorageError::AttachmentNotFound(id));
88 }
89 Ok(())
90 }
91}
92impl std::fmt::Debug for SeaAttachmentRepository {
93 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
94 f.debug_struct("SeaAttachmentRepository").finish_non_exhaustive()
95 }
96}
Added chat/storagelayer/mapper/conversation.rs chat/storagelayer/mapper/conversation.rs +141 −0 View chat/storagelayer/mapper/conversation.rs
@@ -0,0 +1,141 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use sea_orm::{ActiveModelTrait, EntityTrait, IntoActiveModel, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect, Set};
5use time::OffsetDateTime;
6use uuid::Uuid;
7
8use crate::error::{ChatStorageError, ChatStorageResult};
9use crate::models::conversation::{Conversation, ConversationKind, ConversationMember, NewConversation};
10use crate::Page;
11use crate::repository::conversation::ConversationRepository;
12use crate::repository::ChatStorageCtx;
13
14use db::chat::chat_members as member_entity;
15use db::chat::chats as chat_entity;
16
17pub struct SeaConversationRepository {
18 db: Arc<db::conn::Db>,
19}
20
21impl SeaConversationRepository {
22 pub fn new(db: Arc<db::conn::Db>) -> Self {
23 Self { db }
24 }
25}
26
27#[async_trait]
28impl ConversationRepository for SeaConversationRepository {
29 async fn create(&self, _ctx: &ChatStorageCtx, input: NewConversation) -> ChatStorageResult<Conversation> {
30 let db = self.db.writer();
31 let id = Uuid::new_v4();
32 let now = OffsetDateTime::now_utc();
33
34 let chat = chat_entity::ActiveModel {
35 id: Set(id),
36 channel_id: Set(input.channel_id),
37 kind: Set(input.kind.into()),
38 counterpart_user_id: Set(input.counterpart_user_id),
39 title: Set(input.title.clone()),
40 created_at: Set(now),
41 updated_at: Set(None),
42 };
43 chat_entity::Entity::insert(chat).exec(&db).await?;
44
45 for user_id in input.initial_members {
46 let member = member_entity::ActiveModel {
47 chat_id: Set(id),
48 user_id: Set(user_id),
49 joined_at: Set(now),
50 last_read_at: Set(None),
51 muted: Set(false),
52 };
53 member_entity::Entity::insert(member).exec(&db).await?;
54 }
55
56 Ok(Conversation {
57 id,
58 channel_id: input.channel_id,
59 kind: input.kind,
60 counterpart_user_id: input.counterpart_user_id,
61 title: input.title,
62 created_at: now,
63 updated_at: None,
64 })
65 }
66
67 async fn find_by_id(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<Option<Conversation>> {
68 let db = self.db.reader();
69 Ok(chat_entity::Entity::find_by_id(id).one(&db).await?.map(Into::into))
70 }
71
72 async fn list_by_channel(&self, _ctx: &ChatStorageCtx, channel_id: Uuid, page: Page) -> ChatStorageResult<Vec<Conversation>> {
73 let db = self.db.reader();
74 let limit = page.limit as u64;
75 let rows = chat_entity::Entity::find()
76 .filter(chat_entity::COLUMN.channel_id.eq(channel_id))
77 .order_by_desc(chat_entity::COLUMN.created_at)
78 .limit(limit)
79 .all(&db)
80 .await?;
81 Ok(rows.into_iter().map(Into::into).collect())
82 }
83
84 async fn update_title(&self, _ctx: &ChatStorageCtx, id: Uuid, title: Option<String>) -> ChatStorageResult<()> {
85 let db = self.db.writer();
86 let Some(chat) = chat_entity::Entity::find_by_id(id).one(&db).await? else {
87 return Err(ChatStorageError::ConversationNotFound(id));
88 };
89 let mut active = chat.into_active_model();
90 active.title = Set(title);
91 active.updated_at = Set(Some(OffsetDateTime::now_utc()));
92 active.update(&db).await?;
93 Ok(())
94 }
95
96 async fn update_kind(&self, _ctx: &ChatStorageCtx, id: Uuid, kind: ConversationKind) -> ChatStorageResult<()> {
97 let db = self.db.writer();
98 let Some(chat) = chat_entity::Entity::find_by_id(id).one(&db).await? else {
99 return Err(ChatStorageError::ConversationNotFound(id));
100 };
101 let mut active = chat.into_active_model();
102 active.kind = Set(kind.into());
103 active.updated_at = Set(Some(OffsetDateTime::now_utc()));
104 active.update(&db).await?;
105 Ok(())
106 }
107
108 async fn delete(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {
109 let db = self.db.writer();
110 let result = chat_entity::Entity::delete_by_id(id).exec(&db).await?;
111 if result.rows_affected == 0 {
112 return Err(ChatStorageError::ConversationNotFound(id));
113 }
114 member_entity::Entity::delete_many()
115 .filter(member_entity::COLUMN.chat_id.eq(id))
116 .exec(&db)
117 .await?;
118 Ok(())
119 }
120
121 async fn exists(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<bool> {
122 let db = self.db.reader();
123 Ok(chat_entity::Entity::find_by_id(id).count(&db).await? > 0)
124 }
125
126 async fn list_members(&self, _ctx: &ChatStorageCtx, chat_id: Uuid, page: Page) -> ChatStorageResult<Vec<ConversationMember>> {
127 let db = self.db.reader();
128 let rows = member_entity::Entity::find()
129 .filter(member_entity::COLUMN.chat_id.eq(chat_id))
130 .order_by_asc(member_entity::COLUMN.joined_at)
131 .limit(page.limit as u64)
132 .all(&db)
133 .await?;
134 Ok(rows.into_iter().map(Into::into).collect())
135 }
136}
137impl std::fmt::Debug for SeaConversationRepository {
138 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
139 f.debug_struct("SeaConversationRepository").finish_non_exhaustive()
140 }
141}
Added chat/storagelayer/mapper/member.rs chat/storagelayer/mapper/member.rs +163 −0 View chat/storagelayer/mapper/member.rs
@@ -0,0 +1,163 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use sea_orm::sea_query::OnConflict;
5use sea_orm::{EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect, Set};
6use time::OffsetDateTime;
7use uuid::Uuid;
8
9use crate::error::{ChatStorageError, ChatStorageResult};
10use crate::models::conversation::ConversationMember;
11use crate::Page;
12use crate::repository::member::MemberRepository;
13use crate::repository::ChatStorageCtx;
14
15use db::chat::chat_members as member_entity;
16
17pub struct SeaMemberRepository {
18 db: Arc<db::conn::Db>,
19}
20
21impl SeaMemberRepository {
22 pub fn new(db: Arc<db::conn::Db>) -> Self {
23 Self { db }
24 }
25}
26
27#[async_trait]
28impl MemberRepository for SeaMemberRepository {
29 async fn add_member(
30 &self,
31 _ctx: &ChatStorageCtx,
32 chat_id: Uuid,
33 user_id: Uuid,
34 ) -> ChatStorageResult<ConversationMember> {
35 let db = self.db.writer();
36 let now = OffsetDateTime::now_utc();
37 let active = member_entity::ActiveModel {
38 chat_id: Set(chat_id),
39 user_id: Set(user_id),
40 joined_at: Set(now),
41 last_read_at: Set(None),
42 muted: Set(false),
43 };
44 member_entity::Entity::insert(active)
45 .on_conflict(
46 OnConflict::columns([
47 member_entity::Entity::COLUMN.chat_id,
48 member_entity::Entity::COLUMN.user_id,
49 ])
50 .do_nothing()
51 .to_owned(),
52 )
53 .exec(&db)
54 .await?;
55 Ok(ConversationMember {
56 chat_id,
57 user_id,
58 joined_at: now,
59 last_read_at: None,
60 muted: false,
61 })
62 }
63
64 async fn remove_member(
65 &self,
66 _ctx: &ChatStorageCtx,
67 chat_id: Uuid,
68 user_id: Uuid,
69 ) -> ChatStorageResult<()> {
70 let db = self.db.writer();
71 let res = member_entity::Entity::delete_many()
72 .filter(member_entity::Entity::COLUMN.chat_id.eq(chat_id))
73 .filter(member_entity::Entity::COLUMN.user_id.eq(user_id))
74 .exec(&db)
75 .await?;
76 if res.rows_affected == 0 {
77 return Err(ChatStorageError::MemberNotFound(chat_id, user_id));
78 }
79 Ok(())
80 }
81
82 async fn list_members(
83 &self,
84 _ctx: &ChatStorageCtx,
85 chat_id: Uuid,
86 page: Page,
87 ) -> ChatStorageResult<Vec<ConversationMember>> {
88 let db = self.db.reader();
89 let models = member_entity::Entity::find()
90 .filter(member_entity::Entity::COLUMN.chat_id.eq(chat_id))
91 .order_by_asc(member_entity::Entity::COLUMN.joined_at)
92 .limit(page.limit as u64)
93 .all(&db)
94 .await?;
95 Ok(models.into_iter().map(Into::into).collect())
96 }
97
98 async fn mark_read(
99 &self,
100 _ctx: &ChatStorageCtx,
101 chat_id: Uuid,
102 user_id: Uuid,
103 at: OffsetDateTime,
104 ) -> ChatStorageResult<()> {
105 let db = self.db.writer();
106 let res = member_entity::Entity::update_many()
107 .filter(member_entity::Entity::COLUMN.chat_id.eq(chat_id))
108 .filter(member_entity::Entity::COLUMN.user_id.eq(user_id))
109 .set(member_entity::ActiveModel {
110 last_read_at: Set(Some(at)),
111 ..Default::default()
112 })
113 .exec(&db)
114 .await?;
115 if res.rows_affected == 0 {
116 return Err(ChatStorageError::MemberNotFound(chat_id, user_id));
117 }
118 Ok(())
119 }
120
121 async fn set_muted(
122 &self,
123 _ctx: &ChatStorageCtx,
124 chat_id: Uuid,
125 user_id: Uuid,
126 muted: bool,
127 ) -> ChatStorageResult<()> {
128 let db = self.db.writer();
129 let res = member_entity::Entity::update_many()
130 .filter(member_entity::Entity::COLUMN.chat_id.eq(chat_id))
131 .filter(member_entity::Entity::COLUMN.user_id.eq(user_id))
132 .set(member_entity::ActiveModel {
133 muted: Set(muted),
134 ..Default::default()
135 })
136 .exec(&db)
137 .await?;
138 if res.rows_affected == 0 {
139 return Err(ChatStorageError::MemberNotFound(chat_id, user_id));
140 }
141 Ok(())
142 }
143
144 async fn is_member(
145 &self,
146 _ctx: &ChatStorageCtx,
147 chat_id: Uuid,
148 user_id: Uuid,
149 ) -> ChatStorageResult<bool> {
150 let db = self.db.reader();
151 let count = member_entity::Entity::find()
152 .filter(member_entity::Entity::COLUMN.chat_id.eq(chat_id))
153 .filter(member_entity::Entity::COLUMN.user_id.eq(user_id))
154 .count(&db)
155 .await?;
156 Ok(count > 0)
157 }
158}
159impl std::fmt::Debug for SeaMemberRepository {
160 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
161 f.debug_struct("SeaMemberRepository").finish_non_exhaustive()
162 }
163}
Added chat/storagelayer/mapper/mention.rs chat/storagelayer/mapper/mention.rs +88 −0 View chat/storagelayer/mapper/mention.rs
@@ -0,0 +1,88 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use sea_orm::{Condition, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set};
5use time::OffsetDateTime;
6use uuid::Uuid;
7
8use crate::error::ChatStorageResult;
9use crate::models::mention::{Mention, NewMention};
10use crate::repository::mention::MentionRepository;
11use crate::repository::ChatStorageCtx;
12
13use db::chat::chat_message_mentions as mention_entity;
14
15pub struct SeaMentionRepository {
16 db: Arc<db::conn::Db>,
17}
18
19impl SeaMentionRepository {
20 pub fn new(db: Arc<db::conn::Db>) -> Self {
21 Self { db }
22 }
23}
24
25#[async_trait]
26impl MentionRepository for SeaMentionRepository {
27 async fn add_mention(
28 &self,
29 _ctx: &ChatStorageCtx,
30 input: NewMention,
31 ) -> ChatStorageResult<Mention> {
32 let db = self.db.writer();
33 let active = mention_entity::ActiveModel {
34 id: Set(Uuid::new_v4()),
35 message_id: Set(input.message_id),
36 user_id: Set(input.user_id),
37 role_id: Set(input.role_id),
38 channel_id: Set(input.channel_id),
39 is_everyone: Set(input.is_everyone),
40 is_here: Set(input.is_here),
41 created_at: Set(OffsetDateTime::now_utc()),
42 };
43 let model = mention_entity::Entity::insert(active)
44 .exec_with_returning(&db)
45 .await?;
46 Ok(model.into())
47 }
48
49 async fn list_for_message(
50 &self,
51 _ctx: &ChatStorageCtx,
52 message_id: Uuid,
53 ) -> ChatStorageResult<Vec<Mention>> {
54 let db = self.db.reader();
55 let models = mention_entity::Entity::find()
56 .filter(mention_entity::Entity::COLUMN.message_id.eq(message_id))
57 .order_by_asc(mention_entity::Entity::COLUMN.created_at)
58 .all(&db)
59 .await?;
60 Ok(models.into_iter().map(Into::into).collect())
61 }
62
63 async fn list_for_user(
64 &self,
65 _ctx: &ChatStorageCtx,
66 user_id: Uuid,
67 limit: u32,
68 ) -> ChatStorageResult<Vec<Mention>> {
69 let db = self.db.reader();
70 let models = mention_entity::Entity::find()
71 .filter(
72 Condition::any()
73 .add(mention_entity::Entity::COLUMN.user_id.eq(user_id))
74 .add(mention_entity::Entity::COLUMN.is_everyone.eq(true))
75 .add(mention_entity::Entity::COLUMN.is_here.eq(true)),
76 )
77 .order_by_desc(mention_entity::Entity::COLUMN.created_at)
78 .limit(limit as u64)
79 .all(&db)
80 .await?;
81 Ok(models.into_iter().map(Into::into).collect())
82 }
83}
84impl std::fmt::Debug for SeaMentionRepository {
85 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
86 f.debug_struct("SeaMentionRepository").finish_non_exhaustive()
87 }
88}
Added chat/storagelayer/mapper/message.rs chat/storagelayer/mapper/message.rs +301 −0 View chat/storagelayer/mapper/message.rs
@@ -0,0 +1,301 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use futures::FutureExt;
5use sea_orm::sea_query::OnConflict;
6use sea_orm::{
7 ActiveModelTrait, Condition, EntityTrait, IntoActiveModel, PaginatorTrait, QueryFilter,
8 QueryOrder, QuerySelect, Set, TransactionTrait,
9};
10use time::OffsetDateTime;
11use uuid::Uuid;
12
13use crate::error::{ChatStorageError, ChatStorageResult};
14use crate::models::message::{
15 ListDirection, Message, MessageFlags, MessagePage, MessageQuery, NewMessage,
16};
17use crate::Page;
18use crate::repository::message::MessageRepository;
19use crate::repository::ChatStorageCtx;
20
21use db::attachment::attachments as att_entity;
22use db::chat::chat_messages as message_entity;
23use db::chat::threads as thread_entity;
24
25pub struct SeaMessageRepository {
26 db: Arc<db::conn::Db>,
27}
28
29impl SeaMessageRepository {
30 pub fn new(db: Arc<db::conn::Db>) -> Self {
31 Self { db }
32 }
33}
34
35#[async_trait]
36impl MessageRepository for SeaMessageRepository {
37 async fn create(
38 &self,
39 _ctx: &ChatStorageCtx,
40 input: NewMessage,
41 ) -> ChatStorageResult<Message> {
42 let db = self.db.writer();
43 let id = Uuid::new_v4();
44 let now = OffsetDateTime::now_utc();
45
46 let created = db
47 .transaction(|txn| {
48 async move {
49 let flags = input.flags.unwrap_or(0);
50 let model = message_entity::ActiveModel {
51 id: Set(id),
52 chat_id: Set(input.chat_id),
53 sender_id: Set(input.sender_id),
54 body: Set(input.body.clone()),
55 reply_to_message_id: Set(input.reply_to_message_id),
56 thread_id: Set(input.thread_id),
57 is_thread_start: Set(input.is_thread_start),
58 flags: Set(flags),
59 deleted_at: Set(None),
60 deleted_by: Set(None),
61 edited_at: Set(None),
62 created_at: Set(now),
63 };
64 message_entity::Entity::insert(model).exec(txn).await?;
65
66 if input.is_thread_start {
67 let thread = thread_entity::ActiveModel {
68 chat_id: Set(input.chat_id),
69 root_message_id: Set(id),
70 reply_count: Set(0),
71 last_reply_at: Set(None),
72 last_reply_by: Set(None),
73 };
74 thread_entity::Entity::insert(thread)
75 .on_conflict(
76 OnConflict::columns([
77 thread_entity::COLUMN.chat_id,
78 thread_entity::COLUMN.root_message_id,
79 ])
80 .do_nothing()
81 .to_owned(),
82 )
83 .exec(txn)
84 .await?;
85 } else if let Some(root) = input.thread_id {
86 let res = thread_entity::Entity::update_many()
87 .filter(thread_entity::COLUMN.chat_id.eq(input.chat_id))
88 .filter(thread_entity::COLUMN.root_message_id.eq(root))
89 .exec(txn)
90 .await?;
91 if res.rows_affected == 0 {
92 let thread = thread_entity::ActiveModel {
93 chat_id: Set(input.chat_id),
94 root_message_id: Set(root),
95 reply_count: Set(1),
96 last_reply_at: Set(Some(now)),
97 last_reply_by: Set(Some(input.sender_id)),
98 };
99 thread_entity::Entity::insert(thread).exec(txn).await?;
100 }
101 }
102
103 for att_id in &input.attachment_ids {
104 let att = att_entity::ActiveModel {
105 id: Set(*att_id),
106 owner_type: Set(att_entity::OwnerType::ChatMessage),
107 owner_id: Set(id),
108 filename: Set(String::new()),
109 content_type: Set(String::new()),
110 size: Set(0),
111 url: Set(String::new()),
112 uploaded_by: Set(None),
113 created_at: Set(now),
114 };
115 att_entity::Entity::insert(att).exec(txn).await?;
116 }
117
118 Ok(Message {
119 id,
120 chat_id: input.chat_id,
121 sender_id: input.sender_id,
122 body: input.body,
123 reply_to_message_id: input.reply_to_message_id,
124 thread_id: input.thread_id,
125 is_thread_start: input.is_thread_start,
126 flags,
127 deleted_at: None,
128 deleted_by: None,
129 edited_at: None,
130 created_at: now,
131 })
132 }
133 .boxed()
134 })
135 .await
136 .map_err(|e| match e {
137 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),
138 sea_orm::TransactionError::Transaction(err) => err,
139 })?;
140
141 Ok(created)
142 }
143
144 async fn find_by_id(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<Option<Message>> {
145 let db = self.db.reader();
146 let model = message_entity::Entity::find_by_id(id).one(&db).await?;
147 Ok(model.map(Into::into))
148 }
149
150 async fn list_by_conversation(
151 &self,
152 _ctx: &ChatStorageCtx,
153 chat_id: Uuid,
154 query: MessageQuery,
155 ) -> ChatStorageResult<MessagePage> {
156 let db = self.db.reader();
157 let mut select =
158 message_entity::Entity::find().filter(message_entity::COLUMN.chat_id.eq(chat_id));
159 if !query.include_deleted {
160 select = select.filter(message_entity::COLUMN.deleted_at.is_null());
161 }
162
163 if let Some(cursor) = &query.cursor {
164 let cursor_id = Uuid::parse_str(cursor)
165 .map_err(|_| ChatStorageError::InvalidInput("invalid cursor".to_string()))?;
166 let cursor_msg = message_entity::Entity::find_by_id(cursor_id)
167 .one(&db)
168 .await?
169 .ok_or_else(|| ChatStorageError::MessageNotFound(cursor_id))?;
170 let cursor_ts = cursor_msg.created_at;
171 let cursor_cond = match query.direction {
172 ListDirection::Forward => Condition::any()
173 .add(message_entity::COLUMN.created_at.gt(cursor_ts))
174 .add(
175 Condition::all()
176 .add(message_entity::COLUMN.created_at.eq(cursor_ts))
177 .add(message_entity::COLUMN.id.gt(cursor_id)),
178 ),
179 ListDirection::Backward => Condition::any()
180 .add(message_entity::COLUMN.created_at.lt(cursor_ts))
181 .add(
182 Condition::all()
183 .add(message_entity::COLUMN.created_at.eq(cursor_ts))
184 .add(message_entity::COLUMN.id.lt(cursor_id)),
185 ),
186 };
187 select = select.filter(cursor_cond);
188 }
189
190 match query.direction {
191 ListDirection::Forward => {
192 select = select
193 .order_by_asc(message_entity::COLUMN.created_at)
194 .order_by_asc(message_entity::COLUMN.id);
195 }
196 ListDirection::Backward => {
197 select = select
198 .order_by_desc(message_entity::COLUMN.created_at)
199 .order_by_desc(message_entity::COLUMN.id);
200 }
201 }
202
203 let limit = query.limit.max(1) as u64;
204 let models = select.limit(limit + 1).all(&db).await?;
205 let has_more = (models.len() as u64) > limit;
206 let items: Vec<Message> = models
207 .into_iter()
208 .take(limit as usize)
209 .map(Into::into)
210 .collect();
211 let next_cursor = if has_more {
212 items.last().map(|m| m.id.to_string())
213 } else {
214 None
215 };
216
217 Ok(MessagePage { items, next_cursor })
218 }
219
220 async fn list_thread(
221 &self,
222 _ctx: &ChatStorageCtx,
223 chat_id: Uuid,
224 root_message_id: Uuid,
225 page: Page,
226 ) -> ChatStorageResult<MessagePage> {
227 let db = self.db.reader();
228 let models = message_entity::Entity::find()
229 .filter(message_entity::COLUMN.chat_id.eq(chat_id))
230 .filter(message_entity::COLUMN.thread_id.eq(root_message_id))
231 .order_by_asc(message_entity::COLUMN.created_at)
232 .order_by_asc(message_entity::COLUMN.id)
233 .limit(page.limit as u64)
234 .all(&db)
235 .await?;
236 let items = models.into_iter().map(Into::into).collect();
237 Ok(MessagePage { items, next_cursor: None })
238 }
239
240 async fn edit(
241 &self,
242 _ctx: &ChatStorageCtx,
243 id: Uuid,
244 _editor: Uuid,
245 new_body: String,
246 ) -> ChatStorageResult<Message> {
247 let db = self.db.writer();
248 let model = message_entity::Entity::find_by_id(id)
249 .one(&db)
250 .await?
251 .ok_or_else(|| ChatStorageError::MessageNotFound(id))?;
252 let flags = MessageFlags::from_bits(model.flags);
253 let mut active = model.into_active_model();
254 active.body = Set(new_body);
255 active.edited_at = Set(Some(OffsetDateTime::now_utc()));
256 active.flags = Set(flags.with(MessageFlags::EDITED).bits());
257 let updated = active.update(&db).await?;
258 Ok(updated.into())
259 }
260
261 async fn soft_delete(&self, _ctx: &ChatStorageCtx, id: Uuid, by: Uuid) -> ChatStorageResult<()> {
262 let db = self.db.writer();
263 let model = message_entity::Entity::find_by_id(id)
264 .one(&db)
265 .await?
266 .ok_or_else(|| ChatStorageError::MessageNotFound(id))?;
267 let flags = MessageFlags::from_bits(model.flags);
268 let mut active = model.into_active_model();
269 active.deleted_at = Set(Some(OffsetDateTime::now_utc()));
270 active.deleted_by = Set(Some(by));
271 active.flags = Set(flags.with(MessageFlags::DELETED).bits());
272 active.update(&db).await?;
273 Ok(())
274 }
275
276 async fn restore(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {
277 let db = self.db.writer();
278 let model = message_entity::Entity::find_by_id(id)
279 .one(&db)
280 .await?
281 .ok_or_else(|| ChatStorageError::MessageNotFound(id))?;
282 let flags = MessageFlags::from_bits(model.flags);
283 let mut active = model.into_active_model();
284 active.deleted_at = Set(None);
285 active.deleted_by = Set(None);
286 active.flags = Set(flags.without(MessageFlags::DELETED).bits());
287 active.update(&db).await?;
288 Ok(())
289 }
290
291 async fn exists(&self, _ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<bool> {
292 let db = self.db.reader();
293 let count = message_entity::Entity::find_by_id(id).count(&db).await?;
294 Ok(count > 0)
295 }
296}
297impl std::fmt::Debug for SeaMessageRepository {
298 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
299 f.debug_struct("SeaMessageRepository").finish_non_exhaustive()
300 }
301}
Added chat/storagelayer/mapper/pinned.rs chat/storagelayer/mapper/pinned.rs +94 −0 View chat/storagelayer/mapper/pinned.rs
@@ -0,0 +1,94 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use sea_orm::{EntityTrait, QueryFilter, QueryOrder, Set};
5use time::OffsetDateTime;
6use uuid::Uuid;
7
8use crate::error::{ChatStorageError, ChatStorageResult};
9use crate::models::pinned::{NewPinnedMessage, PinnedMessage};
10use crate::repository::pinned::PinnedMessageRepository;
11use crate::repository::ChatStorageCtx;
12
13use db::chat::chat_pinned_messages as pinned_entity;
14
15pub struct SeaPinnedMessageRepository {
16 db: Arc<db::conn::Db>,
17}
18
19impl SeaPinnedMessageRepository {
20 pub fn new(db: Arc<db::conn::Db>) -> Self {
21 Self { db }
22 }
23}
24
25#[async_trait]
26impl PinnedMessageRepository for SeaPinnedMessageRepository {
27 async fn pin(
28 &self,
29 _ctx: &ChatStorageCtx,
30 input: NewPinnedMessage,
31 ) -> ChatStorageResult<PinnedMessage> {
32 let db = self.db.writer();
33 let existing = pinned_entity::Entity::find()
34 .filter(pinned_entity::Entity::COLUMN.chat_id.eq(input.chat_id))
35 .filter(pinned_entity::Entity::COLUMN.message_id.eq(input.message_id))
36 .one(&db)
37 .await?;
38 if existing.is_some() {
39 return Err(ChatStorageError::Conflict("message already pinned".into()));
40 }
41 let active = pinned_entity::ActiveModel {
42 id: Set(Uuid::new_v4()),
43 chat_id: Set(input.chat_id),
44 message_id: Set(input.message_id),
45 pinned_by: Set(input.pinned_by),
46 created_at: Set(OffsetDateTime::now_utc()),
47 };
48 let model = pinned_entity::Entity::insert(active)
49 .exec_with_returning(&db)
50 .await?;
51 Ok(model.into())
52 }
53
54 async fn unpin(
55 &self,
56 _ctx: &ChatStorageCtx,
57 chat_id: Uuid,
58 message_id: Uuid,
59 ) -> ChatStorageResult<()> {
60 let db = self.db.writer();
61 let found = pinned_entity::Entity::find()
62 .filter(pinned_entity::Entity::COLUMN.chat_id.eq(chat_id))
63 .filter(pinned_entity::Entity::COLUMN.message_id.eq(message_id))
64 .one(&db)
65 .await?;
66 let Some(model) = found else {
67 return Err(ChatStorageError::PinnedMessageNotFound(message_id));
68 };
69 let res = pinned_entity::Entity::delete_by_id(model.id).exec(&db).await?;
70 if res.rows_affected == 0 {
71 return Err(ChatStorageError::PinnedMessageNotFound(model.id));
72 }
73 Ok(())
74 }
75
76 async fn list_for_chat(
77 &self,
78 _ctx: &ChatStorageCtx,
79 chat_id: Uuid,
80 ) -> ChatStorageResult<Vec<PinnedMessage>> {
81 let db = self.db.reader();
82 let models = pinned_entity::Entity::find()
83 .filter(pinned_entity::Entity::COLUMN.chat_id.eq(chat_id))
84 .order_by_desc(pinned_entity::Entity::COLUMN.created_at)
85 .all(&db)
86 .await?;
87 Ok(models.into_iter().map(Into::into).collect())
88 }
89}
90impl std::fmt::Debug for SeaPinnedMessageRepository {
91 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
92 f.debug_struct("SeaPinnedMessageRepository").finish_non_exhaustive()
93 }
94}
Added chat/storagelayer/mapper/reaction.rs chat/storagelayer/mapper/reaction.rs +93 −0 View chat/storagelayer/mapper/reaction.rs
@@ -0,0 +1,93 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use sea_orm::{EntityTrait, QueryFilter, QueryOrder, Set};
5use time::OffsetDateTime;
6use uuid::Uuid;
7
8use crate::error::{ChatStorageError, ChatStorageResult};
9use crate::models::reaction::{NewReaction, Reaction};
10use crate::repository::reaction::ReactionRepository;
11use crate::repository::ChatStorageCtx;
12
13use db::chat::chat_message_reactions as reaction_entity;
14
15pub struct SeaReactionRepository {
16 db: Arc<db::conn::Db>,
17}
18
19impl SeaReactionRepository {
20 pub fn new(db: Arc<db::conn::Db>) -> Self {
21 Self { db }
22 }
23}
24
25#[async_trait]
26impl ReactionRepository for SeaReactionRepository {
27 async fn add_reaction(
28 &self,
29 _ctx: &ChatStorageCtx,
30 input: NewReaction,
31 ) -> ChatStorageResult<Reaction> {
32 let db = self.db.writer();
33 let existing = reaction_entity::Entity::find()
34 .filter(reaction_entity::Entity::COLUMN.message_id.eq(input.message_id))
35 .filter(reaction_entity::Entity::COLUMN.user_id.eq(input.user_id))
36 .filter(reaction_entity::Entity::COLUMN.emoji.eq(&input.emoji))
37 .one(&db)
38 .await?;
39 if existing.is_some() {
40 return Err(ChatStorageError::Conflict("reaction already exists".into()));
41 }
42 let active = reaction_entity::ActiveModel {
43 id: Set(Uuid::new_v4()),
44 message_id: Set(input.message_id),
45 user_id: Set(input.user_id),
46 emoji: Set(input.emoji),
47 created_at: Set(OffsetDateTime::now_utc()),
48 };
49 let model = reaction_entity::Entity::insert(active)
50 .exec_with_returning(&db)
51 .await?;
52 Ok(model.into())
53 }
54
55 async fn remove_reaction(
56 &self,
57 _ctx: &ChatStorageCtx,
58 message_id: Uuid,
59 user_id: Uuid,
60 emoji: &str,
61 ) -> ChatStorageResult<()> {
62 let db = self.db.writer();
63 let res = reaction_entity::Entity::delete_many()
64 .filter(reaction_entity::Entity::COLUMN.message_id.eq(message_id))
65 .filter(reaction_entity::Entity::COLUMN.user_id.eq(user_id))
66 .filter(reaction_entity::Entity::COLUMN.emoji.eq(emoji))
67 .exec(&db)
68 .await?;
69 if res.rows_affected == 0 {
70 return Err(ChatStorageError::ReactionNotFound(message_id));
71 }
72 Ok(())
73 }
74
75 async fn list_reactions(
76 &self,
77 _ctx: &ChatStorageCtx,
78 message_id: Uuid,
79 ) -> ChatStorageResult<Vec<Reaction>> {
80 let db = self.db.reader();
81 let models = reaction_entity::Entity::find()
82 .filter(reaction_entity::Entity::COLUMN.message_id.eq(message_id))
83 .order_by_asc(reaction_entity::Entity::COLUMN.created_at)
84 .all(&db)
85 .await?;
86 Ok(models.into_iter().map(Into::into).collect())
87 }
88}
89impl std::fmt::Debug for SeaReactionRepository {
90 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91 f.debug_struct("SeaReactionRepository").finish_non_exhaustive()
92 }
93}
Added chat/storagelayer/mapper/read_receipt.rs chat/storagelayer/mapper/read_receipt.rs +102 −0 View chat/storagelayer/mapper/read_receipt.rs
@@ -0,0 +1,102 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use sea_orm::{ActiveModelTrait, EntityTrait, IntoActiveModel, QueryFilter, QueryOrder, Set};
5use time::OffsetDateTime;
6use uuid::Uuid;
7
8use crate::error::ChatStorageResult;
9use crate::models::read_receipt::{NewReadReceipt, ReadReceipt};
10use crate::repository::read_receipt::ReadReceiptRepository;
11use crate::repository::ChatStorageCtx;
12
13use db::chat::chat_reads as read_entity;
14
15pub struct SeaReadReceiptRepository {
16 db: Arc<db::conn::Db>,
17}
18
19impl SeaReadReceiptRepository {
20 pub fn new(db: Arc<db::conn::Db>) -> Self {
21 Self { db }
22 }
23}
24
25#[async_trait]
26impl ReadReceiptRepository for SeaReadReceiptRepository {
27 async fn upsert(
28 &self,
29 _ctx: &ChatStorageCtx,
30 input: NewReadReceipt,
31 ) -> ChatStorageResult<ReadReceipt> {
32 let db = self.db.writer();
33 let now = OffsetDateTime::now_utc();
34 let existing = read_entity::Entity::find()
35 .filter(read_entity::Entity::COLUMN.chat_id.eq(input.chat_id))
36 .filter(read_entity::Entity::COLUMN.user_id.eq(input.user_id))
37 .filter(match input.thread_id {
38 Some(thread_id) => read_entity::Entity::COLUMN.thread_id.eq(thread_id),
39 None => read_entity::Entity::COLUMN.thread_id.is_null(),
40 })
41 .one(&db)
42 .await?;
43 if let Some(model) = existing {
44 let mut active = model.into_active_model();
45 active.last_read_message_id = Set(input.last_read_message_id);
46 active.updated_at = Set(now);
47 let updated = active.update(&db).await?;
48 return Ok(updated.into());
49 }
50 let active = read_entity::ActiveModel {
51 id: Set(Uuid::new_v4()),
52 chat_id: Set(input.chat_id),
53 user_id: Set(input.user_id),
54 thread_id: Set(input.thread_id),
55 last_read_message_id: Set(input.last_read_message_id),
56 updated_at: Set(now),
57 };
58 let model = read_entity::Entity::insert(active)
59 .exec_with_returning(&db)
60 .await?;
61 Ok(model.into())
62 }
63
64 async fn list_for_chat(
65 &self,
66 _ctx: &ChatStorageCtx,
67 chat_id: Uuid,
68 ) -> ChatStorageResult<Vec<ReadReceipt>> {
69 let db = self.db.reader();
70 let models = read_entity::Entity::find()
71 .filter(read_entity::Entity::COLUMN.chat_id.eq(chat_id))
72 .order_by_desc(read_entity::Entity::COLUMN.updated_at)
73 .all(&db)
74 .await?;
75 Ok(models.into_iter().map(Into::into).collect())
76 }
77
78 async fn find(
79 &self,
80 _ctx: &ChatStorageCtx,
81 chat_id: Uuid,
82 user_id: Uuid,
83 thread_id: Option<Uuid>,
84 ) -> ChatStorageResult<Option<ReadReceipt>> {
85 let db = self.db.reader();
86 let model = read_entity::Entity::find()
87 .filter(read_entity::Entity::COLUMN.chat_id.eq(chat_id))
88 .filter(read_entity::Entity::COLUMN.user_id.eq(user_id))
89 .filter(match thread_id {
90 Some(thread_id) => read_entity::Entity::COLUMN.thread_id.eq(thread_id),
91 None => read_entity::Entity::COLUMN.thread_id.is_null(),
92 })
93 .one(&db)
94 .await?;
95 Ok(model.map(Into::into))
96 }
97}
98impl std::fmt::Debug for SeaReadReceiptRepository {
99 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
100 f.debug_struct("SeaReadReceiptRepository").finish_non_exhaustive()
101 }
102}
Added chat/storagelayer/mapper/thread.rs chat/storagelayer/mapper/thread.rs +101 −0 View chat/storagelayer/mapper/thread.rs
@@ -0,0 +1,101 @@
1use std::sync::Arc;
2
3use async_trait::async_trait;
4use sea_orm::{EntityTrait, QueryFilter, QueryOrder, Set};
5use time::OffsetDateTime;
6use uuid::Uuid;
7
8use crate::error::{ChatStorageError, ChatStorageResult};
9use crate::models::thread::Thread;
10use crate::repository::thread::ThreadRepository;
11use crate::repository::ChatStorageCtx;
12
13use db::chat::threads as thread_entity;
14
15pub struct SeaThreadRepository {
16 db: Arc<db::conn::Db>,
17}
18
19impl SeaThreadRepository {
20 pub fn new(db: Arc<db::conn::Db>) -> Self {
21 Self { db }
22 }
23}
24
25#[async_trait]
26impl ThreadRepository for SeaThreadRepository {
27 async fn increment_reply(
28 &self,
29 _ctx: &ChatStorageCtx,
30 chat_id: Uuid,
31 root_message_id: Uuid,
32 reply_by: Uuid,
33 at: OffsetDateTime,
34 ) -> ChatStorageResult<Thread> {
35 let db = self.db.writer();
36 let existing = thread_entity::Entity::find_by_id((chat_id, root_message_id))
37 .one(&db)
38 .await?;
39 if let Some(model) = existing {
40 thread_entity::Entity::update_many()
41 .filter(thread_entity::Entity::COLUMN.chat_id.eq(chat_id))
42 .filter(thread_entity::Entity::COLUMN.root_message_id.eq(root_message_id))
43 .set(thread_entity::ActiveModel {
44 reply_count: Set(model.reply_count + 1),
45 last_reply_at: Set(Some(at)),
46 last_reply_by: Set(Some(reply_by)),
47 ..Default::default()
48 })
49 .exec(&db)
50 .await?;
51 let updated = thread_entity::Entity::find_by_id((chat_id, root_message_id))
52 .one(&db)
53 .await?
54 .ok_or(ChatStorageError::ThreadNotFound(chat_id, root_message_id))?;
55 return Ok(updated.into());
56 }
57 let active = thread_entity::ActiveModel {
58 chat_id: Set(chat_id),
59 root_message_id: Set(root_message_id),
60 reply_count: Set(1),
61 last_reply_at: Set(Some(at)),
62 last_reply_by: Set(Some(reply_by)),
63 };
64 let model = thread_entity::Entity::insert(active)
65 .exec_with_returning(&db)
66 .await?;
67 Ok(model.into())
68 }
69
70 async fn find(
71 &self,
72 _ctx: &ChatStorageCtx,
73 chat_id: Uuid,
74 root_message_id: Uuid,
75 ) -> ChatStorageResult<Option<Thread>> {
76 let db = self.db.reader();
77 let model = thread_entity::Entity::find_by_id((chat_id, root_message_id))
78 .one(&db)
79 .await?;
80 Ok(model.map(Into::into))
81 }
82
83 async fn list_for_chat(
84 &self,
85 _ctx: &ChatStorageCtx,
86 chat_id: Uuid,
87 ) -> ChatStorageResult<Vec<Thread>> {
88 let db = self.db.reader();
89 let models = thread_entity::Entity::find()
90 .filter(thread_entity::Entity::COLUMN.chat_id.eq(chat_id))
91 .order_by_desc(thread_entity::Entity::COLUMN.last_reply_at)
92 .all(&db)
93 .await?;
94 Ok(models.into_iter().map(Into::into).collect())
95 }
96}
97impl std::fmt::Debug for SeaThreadRepository {
98 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
99 f.debug_struct("SeaThreadRepository").finish_non_exhaustive()
100 }
101}
Added chat/storagelayer/mapper.rs chat/storagelayer/mapper.rs +85 −0 View chat/storagelayer/mapper.rs
@@ -0,0 +1,85 @@
1pub mod attachment;
2pub mod conversation;
3pub mod member;
4pub mod mention;
5pub mod message;
6pub mod pinned;
7pub mod reaction;
8pub mod read_receipt;
9pub mod thread;
10
11use std::sync::Arc;
12
13use crate::blob::ChatMediaStore;
14use crate::repository::ChatStorage;
15
16use self::attachment::SeaAttachmentRepository;
17use self::conversation::SeaConversationRepository;
18use self::member::SeaMemberRepository;
19use self::mention::SeaMentionRepository;
20use self::message::SeaMessageRepository;
21use self::pinned::SeaPinnedMessageRepository;
22use self::reaction::SeaReactionRepository;
23use self::read_receipt::SeaReadReceiptRepository;
24use self::thread::SeaThreadRepository;
25
26pub struct SeaChatStorage {
27 db: Arc<db::conn::Db>,
28 blobs: Arc<dyn ChatMediaStore>,
29}
30
31impl SeaChatStorage {
32 pub fn new(db: Arc<db::conn::Db>, blobs: Arc<dyn ChatMediaStore>) -> Self {
33 Self { db, blobs }
34 }
35}
36
37impl ChatStorage for SeaChatStorage {
38 fn conversations(&self) -> crate::repository::conversation::DynConversationRepository {
39 Arc::new(SeaConversationRepository::new(self.db.clone()))
40 }
41
42 fn messages(&self) -> crate::repository::message::DynMessageRepository {
43 Arc::new(SeaMessageRepository::new(self.db.clone()))
44 }
45
46 fn attachments(&self) -> crate::repository::attachment::DynAttachmentRepository {
47 Arc::new(SeaAttachmentRepository::new(self.db.clone()))
48 }
49
50 fn members(&self) -> crate::repository::member::DynMemberRepository {
51 Arc::new(SeaMemberRepository::new(self.db.clone()))
52 }
53
54 fn reactions(&self) -> crate::repository::reaction::DynReactionRepository {
55 Arc::new(SeaReactionRepository::new(self.db.clone()))
56 }
57
58 fn mentions(&self) -> crate::repository::mention::DynMentionRepository {
59 Arc::new(SeaMentionRepository::new(self.db.clone()))
60 }
61
62 fn reads(&self) -> crate::repository::read_receipt::DynReadReceiptRepository {
63 Arc::new(SeaReadReceiptRepository::new(self.db.clone()))
64 }
65
66 fn pinned(&self) -> crate::repository::pinned::DynPinnedMessageRepository {
67 Arc::new(SeaPinnedMessageRepository::new(self.db.clone()))
68 }
69
70 fn threads(&self) -> crate::repository::thread::DynThreadRepository {
71 Arc::new(SeaThreadRepository::new(self.db.clone()))
72 }
73
74 fn blobs(&self) -> Arc<dyn ChatMediaStore> {
75 self.blobs.clone()
76 }
77}
78impl std::fmt::Debug for SeaChatStorage {
79 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80 f.debug_struct("SeaChatStorage")
81 .field("db", &"<Db>")
82 .field("blobs", &"<ChatMediaStore>")
83 .finish()
84 }
85}