refactor(chat): refresh storagelayer auth, mappers, mock and models

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

@@ -90,7 +90,7 @@

9090 .filter(att_entity::Entity::COLUMN.owner_type.eq(owner_type))

9191 .filter(att_entity::Entity::COLUMN.owner_id.eq(owner_id))

9292 .order_by_asc(att_entity::Entity::COLUMN.created_at)

93 .limit(page.limit as u64)

93 .limit(page.clamped_limit())

9494 .all(&db)

9595 .await?;

9696 Ok(models.into_iter().map(Into::into).collect())

Modified chat/storagelayer/mapper/conversation.rs +2 −3 View chat/storagelayer/mapper/conversation.rs

@@ -81,11 +81,10 @@

8181 async fn list_by_channel(&self, ctx: &ChatStorageCtx, channel_id: Uuid, page: Page) -> ChatStorageResult<Vec<Conversation>> {

8282 auth::require_authenticated(ctx).await?;

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

84 let limit = page.limit as u64;

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

8685 .filter(chat_entity::COLUMN.channel_id.eq(channel_id))

8786 .order_by_desc(chat_entity::COLUMN.created_at)

88 .limit(limit)

87 .limit(page.clamped_limit())

8988 .all(&db)

9089 .await?;

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

@@ -142,7 +141,7 @@

142141 let rows = member_entity::Entity::find()

143142 .filter(member_entity::COLUMN.chat_id.eq(chat_id))

144143 .order_by_asc(member_entity::COLUMN.joined_at)

145 .limit(page.limit as u64)

144 .limit(page.clamped_limit())

146145 .all(&db)

147146 .await?;

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

Modified chat/storagelayer/mapper/member.rs +65 −44 View chat/storagelayer/mapper/member.rs

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

22

33use async_trait::async_trait;

44use sea_orm::sea_query::OnConflict;

5use sea_orm::{EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect, Set};

5use sea_orm::{EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect, Set, TransactionTrait};

66use time::OffsetDateTime;

77use uuid::Uuid;

88

@@ -32,56 +32,77 @@

3232 chat_id: Uuid,

3333 user_id: Uuid,

3434 ) -> ChatStorageResult<ConversationMember> {

35 crate::auth::require_admin(&self.db, ctx, chat_id).await?;

36 let db = self.db.writer();

35 let caller = ctx.require_user()?;

3736 let now = OffsetDateTime::now_utc();

38 let active = member_entity::ActiveModel {

39 chat_id: Set(chat_id),

40 user_id: Set(user_id),

41 joined_at: Set(now),

42 last_read_at: Set(None),

43 muted: Set(false),

44 };

45 member_entity::Entity::insert(active)

46 .on_conflict(

47 OnConflict::columns([

48 member_entity::Entity::COLUMN.chat_id,

49 member_entity::Entity::COLUMN.user_id,

50 ])

51 .do_nothing()

52 .to_owned(),

53 )

54 .exec(&db)

55 .await?;

56 Ok(ConversationMember {

57 chat_id,

58 user_id,

59 joined_at: now,

60 last_read_at: None,

61 muted: false,

62 })

37 self.db

38 .writer()

39 .transaction(|txn| {

40 Box::pin(async move {

41 crate::auth::require_admin_in_txn(txn, caller, chat_id).await?;

42 let active = member_entity::ActiveModel {

43 chat_id: Set(chat_id),

44 user_id: Set(user_id),

45 joined_at: Set(now),

46 last_read_at: Set(None),

47 muted: Set(false),

48 };

49 member_entity::Entity::insert(active)

50 .on_conflict(

51 OnConflict::columns([

52 member_entity::Entity::COLUMN.chat_id,

53 member_entity::Entity::COLUMN.user_id,

54 ])

55 .do_nothing()

56 .to_owned(),

57 )

58 .exec(txn)

59 .await?;

60 Ok(ConversationMember {

61 chat_id,

62 user_id,

63 joined_at: now,

64 last_read_at: None,

65 muted: false,

66 })

67 })

68 })

69 .await

70 .map_err(|e| match e {

71 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),

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

73 })

6374 }

6475

6576 async fn remove_member(

6677 &self,

6778 ctx: &ChatStorageCtx,

6879 chat_id: Uuid,

69 user_id: Uuid,

80 target_user_id: Uuid,

7081 ) -> ChatStorageResult<()> {

71 let caller = crate::auth::require_authenticated(ctx).await?;

72 if caller != user_id {

73 crate::auth::require_admin(&self.db, ctx, chat_id).await?;

74 }

75 let db = self.db.writer();

76 let res = member_entity::Entity::delete_many()

77 .filter(member_entity::Entity::COLUMN.chat_id.eq(chat_id))

78 .filter(member_entity::Entity::COLUMN.user_id.eq(user_id))

79 .exec(&db)

80 .await?;

81 if res.rows_affected == 0 {

82 return Err(ChatStorageError::MemberNotFound(chat_id, user_id));

83 }

84 Ok(())

82 let caller = ctx.require_user()?;

83 self.db

84 .writer()

85 .transaction(|txn| {

86 Box::pin(async move {

87 if caller != target_user_id {

88 crate::auth::require_admin_in_txn(txn, caller, chat_id).await?;

89 }

90 let res = member_entity::Entity::delete_many()

91 .filter(member_entity::Entity::COLUMN.chat_id.eq(chat_id))

92 .filter(member_entity::Entity::COLUMN.user_id.eq(target_user_id))

93 .exec(txn)

94 .await?;

95 if res.rows_affected == 0 {

96 return Err(ChatStorageError::MemberNotFound(chat_id, target_user_id));

97 }

98 Ok(())

99 })

100 })

101 .await

102 .map_err(|e| match e {

103 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),

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

105 })

85106 }

86107

87108 async fn list_members(

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

95116 let models = member_entity::Entity::find()

96117 .filter(member_entity::Entity::COLUMN.chat_id.eq(chat_id))

97118 .order_by_asc(member_entity::Entity::COLUMN.joined_at)

98 .limit(page.limit as u64)

119 .limit(page.clamped_limit())

99120 .all(&db)

100121 .await?;

101122 Ok(models.into_iter().map(Into::into).collect())

Modified chat/storagelayer/mapper/mention.rs +1 −1 View chat/storagelayer/mapper/mention.rs

@@ -81,7 +81,7 @@

8181 .add(mention_entity::Entity::COLUMN.is_here.eq(true)),

8282 )

8383 .order_by_desc(mention_entity::Entity::COLUMN.created_at)

84 .limit(limit as u64)

84 .limit(limit.clamp(1, crate::MAX_PAGE_LIMIT) as u64)

8585 .all(&db)

8686 .await?;

8787 Ok(models.into_iter().map(Into::into).collect())

Modified chat/storagelayer/mapper/message.rs +91 −43 View chat/storagelayer/mapper/message.rs

@@ -205,7 +205,7 @@

205205 }

206206 }

207207

208 let limit = query.limit.max(1) as u64;

208 let limit = query.clamped_limit();

209209 let models = select.limit(limit + 1).all(&db).await?;

210210 let has_more = (models.len() as u64) > limit;

211211 let items: Vec<Message> = models

@@ -236,7 +236,7 @@

236236 .filter(message_entity::COLUMN.thread_id.eq(root_message_id))

237237 .order_by_asc(message_entity::COLUMN.created_at)

238238 .order_by_asc(message_entity::COLUMN.id)

239 .limit(page.limit as u64)

239 .limit(page.clamped_limit())

240240 .all(&db)

241241 .await?;

242242 let items = models.into_iter().map(Into::into).collect();

@@ -249,53 +249,101 @@

249249 id: Uuid,

250250 new_body: String,

251251 ) -> ChatStorageResult<Message> {

252 crate::auth::require_sender_or_admin(&self.db, ctx, id).await?;

253 let db = self.db.writer();

254 let model = message_entity::Entity::find_by_id(id)

255 .one(&db)

256 .await?

257 .ok_or_else(|| ChatStorageError::MessageNotFound(id))?;

258 let flags = MessageFlags::from_bits(model.flags);

259 let mut active = model.into_active_model();

260 active.body = Set(new_body);

261 active.edited_at = Set(Some(OffsetDateTime::now_utc()));

262 active.flags = Set(flags.with(MessageFlags::EDITED).bits());

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

264 Ok(updated.into())

252 let user_id = ctx.require_user()?;

253 self.db

254 .writer()

255 .transaction(|txn| {

256 Box::pin(async move {

257 let model = message_entity::Entity::find_by_id(id)

258 .one(txn)

259 .await?

260 .ok_or(ChatStorageError::MessageNotFound(id))?;

261 if model.deleted_at.is_some() {

262 return Err(ChatStorageError::MessageDeleted(id));

263 }

264 if model.sender_id != user_id {

265 crate::auth::require_admin_in_txn(txn, user_id, model.chat_id)

266 .await?;

267 }

268 let flags = MessageFlags::from_bits(model.flags);

269 let mut active = model.into_active_model();

270 active.body = Set(new_body);

271 active.edited_at = Set(Some(OffsetDateTime::now_utc()));

272 active.flags = Set(flags.with(MessageFlags::EDITED).bits());

273 let updated = active.update(txn).await?;

274 Ok(updated.into())

275 })

276 })

277 .await

278 .map_err(|e| match e {

279 sea_orm::TransactionError::Connection(err) => {

280 ChatStorageError::SeaOrm(err)

281 }

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

283 })

265284 }

266285

267286 async fn soft_delete(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {

268 crate::auth::require_sender_or_admin(&self.db, ctx, id).await?;

269 let db = self.db.writer();

270 let model = message_entity::Entity::find_by_id(id)

271 .one(&db)

272 .await?

273 .ok_or_else(|| ChatStorageError::MessageNotFound(id))?;

274 let flags = MessageFlags::from_bits(model.flags);

275 let mut active = model.into_active_model();

276 active.deleted_at = Set(Some(OffsetDateTime::now_utc()));

277 let actor = ctx.require_user()?;

278 active.deleted_by = Set(Some(actor));

279 active.flags = Set(flags.with(MessageFlags::DELETED).bits());

280 active.update(&db).await?;

281 Ok(())

287 let user_id = ctx.require_user()?;

288 self.db

289 .writer()

290 .transaction(|txn| {

291 Box::pin(async move {

292 let model = message_entity::Entity::find_by_id(id)

293 .one(txn)

294 .await?

295 .ok_or(ChatStorageError::MessageNotFound(id))?;

296 if model.deleted_at.is_some() {

297 return Ok(());

298 }

299 if model.sender_id != user_id {

300 crate::auth::require_admin_in_txn(txn, user_id, model.chat_id)

301 .await?;

302 }

303 let flags = MessageFlags::from_bits(model.flags);

304 let mut active = model.into_active_model();

305 active.deleted_at = Set(Some(OffsetDateTime::now_utc()));

306 active.deleted_by = Set(Some(user_id));

307 active.flags = Set(flags.with(MessageFlags::DELETED).bits());

308 active.update(txn).await?;

309 Ok(())

310 })

311 })

312 .await

313 .map_err(|e| match e {

314 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),

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

316 })

282317 }

283318

284319 async fn restore(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {

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

286 crate::auth::require_admin(&self.db, ctx, chat_id).await?;

287 let db = self.db.writer();

288 let model = message_entity::Entity::find_by_id(id)

289 .one(&db)

290 .await?

291 .ok_or_else(|| ChatStorageError::MessageNotFound(id))?;

292 let flags = MessageFlags::from_bits(model.flags);

293 let mut active = model.into_active_model();

294 active.deleted_at = Set(None);

295 active.deleted_by = Set(None);

296 active.flags = Set(flags.without(MessageFlags::DELETED).bits());

297 active.update(&db).await?;

298 Ok(())

320 let user_id = ctx.require_user()?;

321 self.db

322 .writer()

323 .transaction(|txn| {

324 Box::pin(async move {

325 let model = message_entity::Entity::find_by_id(id)

326 .one(txn)

327 .await?

328 .ok_or(ChatStorageError::MessageNotFound(id))?;

329 if model.deleted_at.is_none() {

330 return Ok(());

331 }

332 crate::auth::require_admin_in_txn(txn, user_id, model.chat_id).await?;

333 let flags = MessageFlags::from_bits(model.flags);

334 let mut active = model.into_active_model();

335 active.deleted_at = Set(None);

336 active.deleted_by = Set(None);

337 active.flags = Set(flags.without(MessageFlags::DELETED).bits());

338 active.update(txn).await?;

339 Ok(())

340 })

341 })

342 .await

343 .map_err(|e| match e {

344 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),

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

346 })

299347 }

300348

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

Modified chat/storagelayer/mapper/pinned.rs +59 −37 View chat/storagelayer/mapper/pinned.rs

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

11use std::sync::Arc;

22

33use async_trait::async_trait;

4use sea_orm::{EntityTrait, QueryFilter, QueryOrder, Set};

4use sea_orm::{EntityTrait, QueryFilter, QueryOrder, Set, TransactionTrait};

55use time::OffsetDateTime;

66use uuid::Uuid;

77

@@ -30,28 +30,39 @@

3030 ctx: &ChatStorageCtx,

3131 input: NewPinnedMessage,

3232 ) -> ChatStorageResult<PinnedMessage> {

33 auth::require_admin(&self.db, ctx, input.chat_id).await?;

33 let caller = ctx.require_user()?;

3434 auth::require_self(ctx, input.pinned_by)?;

35 let db = self.db.writer();

36 let existing = pinned_entity::Entity::find()

37 .filter(pinned_entity::Entity::COLUMN.chat_id.eq(input.chat_id))

38 .filter(pinned_entity::Entity::COLUMN.message_id.eq(input.message_id))

39 .one(&db)

40 .await?;

41 if existing.is_some() {

42 return Err(ChatStorageError::Conflict("message already pinned".into()));

43 }

44 let active = pinned_entity::ActiveModel {

45 id: Set(Uuid::new_v4()),

46 chat_id: Set(input.chat_id),

47 message_id: Set(input.message_id),

48 pinned_by: Set(input.pinned_by),

49 created_at: Set(OffsetDateTime::now_utc()),

50 };

51 let model = pinned_entity::Entity::insert(active)

52 .exec_with_returning(&db)

53 .await?;

54 Ok(model.into())

35 self.db

36 .writer()

37 .transaction(|txn| {

38 Box::pin(async move {

39 auth::require_admin_in_txn(txn, caller, input.chat_id).await?;

40 let existing = pinned_entity::Entity::find()

41 .filter(pinned_entity::Entity::COLUMN.chat_id.eq(input.chat_id))

42 .filter(pinned_entity::Entity::COLUMN.message_id.eq(input.message_id))

43 .one(txn)

44 .await?;

45 if existing.is_some() {

46 return Err(ChatStorageError::Conflict("message already pinned".into()));

47 }

48 let active = pinned_entity::ActiveModel {

49 id: Set(Uuid::new_v4()),

50 chat_id: Set(input.chat_id),

51 message_id: Set(input.message_id),

52 pinned_by: Set(input.pinned_by),

53 created_at: Set(OffsetDateTime::now_utc()),

54 };

55 let model = pinned_entity::Entity::insert(active)

56 .exec_with_returning(txn)

57 .await?;

58 Ok(model.into())

59 })

60 })

61 .await

62 .map_err(|e| match e {

63 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),

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

65 })

5566 }

5667

5768 async fn unpin(

@@ -60,21 +71,32 @@

6071 chat_id: Uuid,

6172 message_id: Uuid,

6273 ) -> ChatStorageResult<()> {

63 auth::require_admin(&self.db, ctx, chat_id).await?;

64 let db = self.db.writer();

65 let found = pinned_entity::Entity::find()

66 .filter(pinned_entity::Entity::COLUMN.chat_id.eq(chat_id))

67 .filter(pinned_entity::Entity::COLUMN.message_id.eq(message_id))

68 .one(&db)

69 .await?;

70 let Some(model) = found else {

71 return Err(ChatStorageError::PinnedMessageNotFound(message_id));

72 };

73 let res = pinned_entity::Entity::delete_by_id(model.id).exec(&db).await?;

74 if res.rows_affected == 0 {

75 return Err(ChatStorageError::PinnedMessageNotFound(model.id));

76 }

77 Ok(())

74 let caller = ctx.require_user()?;

75 self.db

76 .writer()

77 .transaction(|txn| {

78 Box::pin(async move {

79 auth::require_admin_in_txn(txn, caller, chat_id).await?;

80 let found = pinned_entity::Entity::find()

81 .filter(pinned_entity::Entity::COLUMN.chat_id.eq(chat_id))

82 .filter(pinned_entity::Entity::COLUMN.message_id.eq(message_id))

83 .one(txn)

84 .await?;

85 let Some(model) = found else {

86 return Err(ChatStorageError::PinnedMessageNotFound(message_id));

87 };

88 let res = pinned_entity::Entity::delete_by_id(model.id).exec(txn).await?;

89 if res.rows_affected == 0 {

90 return Err(ChatStorageError::PinnedMessageNotFound(model.id));

91 }

92 Ok(())

93 })

94 })

95 .await

96 .map_err(|e| match e {

97 sea_orm::TransactionError::Connection(err) => ChatStorageError::SeaOrm(err),

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

99 })

78100 }

79101

80102 async fn list_for_chat(

Modified chat/storagelayer/mock/in_memory.rs +28 −9 View chat/storagelayer/mock/in_memory.rs

@@ -58,14 +58,15 @@

5858}

5959

6060fn is_admin(state: &SharedState, chat_id: Uuid, user_id: Uuid) -> bool {

61 state

61 let is_creator = state

6262 .creators

6363 .read()

6464 .unwrap()

6565 .get(&chat_id)

6666 .copied()

6767 .map(|c| c == user_id)

68 .unwrap_or(false)

68 .unwrap_or(false);

69 is_creator && is_member(state, chat_id, user_id)

6970}

7071

7172fn chat_id_for_message(state: &SharedState, message_id: Uuid) -> Option<Uuid> {

@@ -241,7 +242,7 @@

241242 .cloned()

242243 .collect();

243244 out.sort_by_key(|c| std::cmp::Reverse(c.created_at));

244 let limit = page.limit as usize;

245 let limit = page.clamped_limit() as usize;

245246 Ok(out.into_iter().take(limit).collect())

246247 }

247248

@@ -314,7 +315,7 @@

314315 .cloned()

315316 .collect();

316317 out.sort_by_key(|m| m.joined_at);

317 let limit = page.limit as usize;

318 let limit = page.clamped_limit() as usize;

318319 Ok(out.into_iter().take(limit).collect())

319320 }

320321}

@@ -395,7 +396,7 @@

395396 ListDirection::Forward => items.sort_by_key(|m| m.created_at),

396397 ListDirection::Backward => items.sort_by_key(|m| std::cmp::Reverse(m.created_at)),

397398 }

398 let limit = query.limit as usize;

399 let limit = query.clamped_limit() as usize;

399400 let items: Vec<Message> = items.into_iter().take(limit).collect();

400401 let next_cursor = items.last().map(|m| m.id.to_string());

401402 Ok(MessagePage { items, next_cursor })

@@ -416,7 +417,7 @@

416417 .cloned()

417418 .collect();

418419 items.sort_by_key(|m| m.created_at);

419 let limit = page.limit as usize;

420 let limit = page.clamped_limit() as usize;

420421 let items: Vec<Message> = items.into_iter().take(limit).collect();

421422 let next_cursor = items.last().map(|m| m.id.to_string());

422423 Ok(MessagePage { items, next_cursor })

@@ -434,6 +435,9 @@

434435 let msg = map

435436 .get_mut(&id)

436437 .ok_or(ChatStorageError::MessageNotFound(id))?;

438 if msg.deleted_at.is_some() {

439 return Err(ChatStorageError::MessageDeleted(id));

440 }

437441 msg.body = new_body;

438442 msg.edited_at = Some(now);

439443 msg.flags = MessageFlags::from_bits_truncate(msg.flags)

@@ -445,6 +449,13 @@

445449

446450 async fn soft_delete(&self, ctx: &ChatStorageCtx, id: Uuid) -> ChatStorageResult<()> {

447451 let actor = require_sender_or_admin(&self.state, ctx, id)?;

452 {

453 let map = self.state.messages.read().unwrap();

454 let msg = map.get(&id).ok_or(ChatStorageError::MessageNotFound(id))?;

455 if msg.deleted_at.is_some() {

456 return Ok(());

457 }

458 }

448459 let now = OffsetDateTime::now_utc();

449460 let mut map = self.state.messages.write().unwrap();

450461 let msg = map

@@ -462,6 +473,13 @@

462473 let chat_id = chat_id_for_message(&self.state, id)

463474 .ok_or(ChatStorageError::MessageNotFound(id))?;

464475 require_admin(&self.state, ctx, chat_id)?;

476 {

477 let map = self.state.messages.read().unwrap();

478 let msg = map.get(&id).ok_or(ChatStorageError::MessageNotFound(id))?;

479 if msg.deleted_at.is_none() {

480 return Ok(());

481 }

482 }

465483 let mut map = self.state.messages.write().unwrap();

466484 let msg = map

467485 .get_mut(&id)

@@ -561,7 +579,7 @@

561579 .cloned()

562580 .collect();

563581 items.sort_by_key(|a| a.created_at);

564 let limit = page.limit as usize;

582 let limit = page.clamped_limit() as usize;

565583 Ok(items.into_iter().take(limit).collect())

566584 }

567585

@@ -658,7 +676,7 @@

658676 .cloned()

659677 .collect();

660678 out.sort_by_key(|m| m.joined_at);

661 let limit = page.limit as usize;

679 let limit = page.clamped_limit() as usize;

662680 Ok(out.into_iter().take(limit).collect())

663681 }

664682

@@ -864,7 +882,8 @@

864882 .cloned()

865883 .collect();

866884 items.sort_by_key(|m| std::cmp::Reverse(m.created_at));

867 Ok(items.into_iter().take(limit as usize).collect())

885 let limit = limit.clamp(1, crate::MAX_PAGE_LIMIT) as usize;

886 Ok(items.into_iter().take(limit).collect())

868887 }

869888}

870889

Modified chat/storagelayer/models/message.rs +14 −0 View chat/storagelayer/models/message.rs

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

33use time::OffsetDateTime;

44use uuid::Uuid;

55

6pub const MAX_PAGE_LIMIT: u32 = 200;

7

68#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)]

79#[serde(transparent)]

810pub struct MessageFlags(i64);

@@ -195,6 +197,12 @@

195197 }

196198}

197199

200impl MessageQuery {

201 pub fn clamped_limit(&self) -> u64 {

202 self.limit.clamp(1, MAX_PAGE_LIMIT) as u64

203 }

204}

205

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

199207pub struct Page {

200208 pub limit: u32,

@@ -208,4 +216,10 @@

208216 cursor: None,

209217 }

210218 }

219}

220

221impl Page {

222 pub fn clamped_limit(&self) -> u64 {

223 self.limit.clamp(1, MAX_PAGE_LIMIT) as u64

224 }

211225}

No newline at end of file

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

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

1use sea_orm::EntityTrait;

1use sea_orm::{ConnectionTrait, DatabaseConnection, EntityTrait};

22use uuid::Uuid;

33

44use crate::error::{ChatStorageError, ChatStorageResult};

@@ -56,9 +56,51 @@

5656 "user {user_id} is not admin of chat {chat_id}"

5757 )));

5858 }

59 verify_creator_member(&db.reader(), chat_id, user_id).await?;

5960 Ok(user_id)

6061}

6162

63pub async fn require_admin_in_txn<C: ConnectionTrait>(

64 txn: &C,

65 user_id: Uuid,

66 chat_id: Uuid,

67) -> ChatStorageResult<()> {

68 let chat = chat_entity::Entity::find_by_id(chat_id)

69 .one(txn)

70 .await?

71 .ok_or(ChatStorageError::ConversationNotFound(chat_id))?;

72 if chat.creator_user_id != user_id {

73 return Err(ChatStorageError::PermissionDenied(format!(

74 "user {user_id} is not admin of chat {chat_id}"

75 )));

76 }

77 verify_creator_member_in_txn(txn, chat_id, user_id).await

78}

79

80async fn verify_creator_member(

81 db: &DatabaseConnection,

82 chat_id: Uuid,

83 user_id: Uuid,

84) -> ChatStorageResult<()> {

85 verify_creator_member_in_txn(db, chat_id, user_id).await

86}

87

88async fn verify_creator_member_in_txn<C: ConnectionTrait>(

89 txn: &C,

90 chat_id: Uuid,

91 user_id: Uuid,

92) -> ChatStorageResult<()> {

93 let member = member_entity::Entity::find_by_id((chat_id, user_id))

94 .one(txn)

95 .await?;

96 if member.is_none() {

97 return Err(ChatStorageError::PermissionDenied(format!(

98 "creator/admin {user_id} is no longer a member of chat {chat_id}"

99 )));

100 }

101 Ok(())

102}

103

62104pub async fn require_sender_or_admin(

63105 db: &db::conn::Db,

64106 ctx: &ChatStorageCtx,

Modified chat/storagelayer/error.rs +3 −0 View chat/storagelayer/error.rs

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

66 #[error("message not found: {0}")]

77 MessageNotFound(Uuid),

88

9 #[error("message already deleted: {0}")]

10 MessageDeleted(Uuid),

11

912 #[error("conversation not found: {0}")]

1013 ConversationNotFound(Uuid),

1114

Modified chat/storagelayer/lib.rs +2 −1 View chat/storagelayer/lib.rs

@@ -23,7 +23,8 @@

2323 conversation::{Conversation, ConversationKind, ConversationMember, NewConversation},

2424 mention::{Mention, NewMention},

2525 message::{

26 ListDirection, Message, MessageFlags, MessagePage, MessageQuery, NewMessage, Page,

26 ListDirection, MAX_PAGE_LIMIT, Message, MessageFlags, MessagePage, MessageQuery,

27 NewMessage, Page,

2728 },

2829 pinned::{NewPinnedMessage, PinnedMessage},

2930 reaction::{NewReaction, Reaction},