fix(chat): scope idempotency keys by sender + ownership-check lookup (H8)
zhenyi committed
Modified chat/chatlayer/projection/room.rs chat/chatlayer/projection/room.rs +7 −2 View chat/chatlayer/projection/room.rs
@@ -114,7 +114,10 @@
114114 actor_id: Uuid,
115115 cmd: &MessageCommand,
116116) -> ChatLayerResult<Vec<ChatUserEvent>> {
117 if idempotency.is_resolved(&cmd.client_msg_id).await? {
117 if idempotency
118 .is_resolved(cmd.sender_id, &cmd.client_msg_id)
119 .await?
120 {
118121 crate::metrics::idempotency_hit();
119122 return Ok(Vec::new());
120123 }
@@ -136,7 +139,9 @@
136139 )
137140 .await?;
138141
139 idempotency.resolve(&cmd.client_msg_id, message.id).await?;
142 idempotency
143 .resolve(cmd.sender_id, &cmd.client_msg_id, cmd.sender_id, message.id)
144 .await?;
140145 crate::metrics::message_persisted("success");
141146
142147 if let Some(media) = media {
Modified chat/chatlayer/service.rs chat/chatlayer/service.rs +11 −3 View chat/chatlayer/service.rs
@@ -190,7 +190,11 @@
190190 "client_msg_id exceeds {MAX_CLIENT_MSG_ID_BYTES} bytes"
191191 )));
192192 }
193 match self.idempotency.claim(&cmd.client_msg_id).await? {
193 match self
194 .idempotency
195 .claim(identity.user_id, &cmd.client_msg_id)
196 .await?
197 {
194198 ClientMessageClaim::AlreadyResolved(id) => {
195199 return Ok(AcceptedMsg {
196200 client_msg_id: cmd.client_msg_id,
@@ -442,8 +446,12 @@
442446 token: &str,
443447 client_msg_id: &str,
444448 ) -> ChatLayerResult<Option<Uuid>> {
445 let _ = auth.require_identity(token).await?;
446 self.idempotency.lookup(client_msg_id).await
449 let identity = auth.require_identity(token).await?;
450 let rec = self
451 .idempotency
452 .lookup_for_caller(identity.user_id, client_msg_id)
453 .await?;
454 Ok(rec.map(|r| r.message_id))
447455 }
448456
449457 pub fn queue(&self) -> Option<Arc<queue::Queue>> {
Modified chat/chatlayer/storage_bridge.rs chat/chatlayer/storage_bridge.rs +95 −25 View chat/chatlayer/storage_bridge.rs
@@ -4,7 +4,7 @@
44use cache::redis::RedisCache;
55use uuid::Uuid;
66
7use crate::error::ChatLayerResult;
7use crate::error::{ChatLayerError, ChatLayerResult};
88
99pub const MSG_IDEMPOTENCY_PREFIX: &str = "chat:msg:";
1010
@@ -17,15 +17,53 @@
1717 InFlight,
1818}
1919
20/// Resolved idempotency record stored once a message has been persisted.
21///
22/// Layout: 32 bytes — first 16 are `message_id`, next 16 are `sender_id`.
23/// Both Uuids are stored in big-endian network order via `Uuid::as_bytes`.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub struct ResolvedRecord {
26 pub message_id: Uuid,
27 pub sender_id: Uuid,
28}
29
30impl ResolvedRecord {
31 const LEN: usize = 32;
32
33 fn encode(&self) -> Vec<u8> {
34 let mut buf = Vec::with_capacity(Self::LEN);
35 buf.extend_from_slice(self.message_id.as_bytes());
36 buf.extend_from_slice(self.sender_id.as_bytes());
37 buf
38 }
39
40 fn decode(bytes: &[u8]) -> Option<Self> {
41 if bytes.len() != Self::LEN {
42 return None;
43 }
44 let mut mid = [0u8; 16];
45 mid.copy_from_slice(&bytes[..16]);
46 let mut sid = [0u8; 16];
47 sid.copy_from_slice(&bytes[16..]);
48 Some(Self {
49 message_id: Uuid::from_bytes(mid),
50 sender_id: Uuid::from_bytes(sid),
51 })
52 }
53}
54
2055pub trait MessageIdempotencyStore: Send + Sync {
2156 fn is_resolved<'a>(
2257 &'a self,
58 user_id: Uuid,
2359 client_msg_id: &'a str,
2460 ) -> std::pin::Pin<Box<dyn Future<Output = ChatLayerResult<bool>> + Send + 'a>>;
2561
2662 fn resolve<'a>(
2763 &'a self,
64 user_id: Uuid,
2865 client_msg_id: &'a str,
66 sender_id: Uuid,
2967 message_id: Uuid,
3068 ) -> std::pin::Pin<Box<dyn Future<Output = ChatLayerResult<()>> + Send + 'a>>;
3169}
@@ -49,12 +87,16 @@
4987 Self { kv, ttl_seconds }
5088 }
5189
52 fn key(client_msg_id: &str) -> String {
53 format!("{MSG_IDEMPOTENCY_PREFIX}{client_msg_id}")
90 fn key(user_id: Uuid, client_msg_id: &str) -> String {
91 format!("{MSG_IDEMPOTENCY_PREFIX}{user_id}:{client_msg_id}")
5492 }
5593
56 pub async fn claim(&self, client_msg_id: &str) -> ChatLayerResult<ClientMessageClaim> {
57 let key = Self::key(client_msg_id);
94 pub async fn claim(
95 &self,
96 user_id: Uuid,
97 client_msg_id: &str,
98 ) -> ChatLayerResult<ClientMessageClaim> {
99 let key = Self::key(user_id, client_msg_id);
58100 if self
59101 .kv
60102 .set_nx(&key, PENDING.to_vec(), self.ttl_seconds)
@@ -63,24 +105,46 @@
63105 return Ok(ClientMessageClaim::Fresh);
64106 }
65107 match self.kv.get_bytes(&key).await? {
66 Some(v) if v.len() == 16 => {
67 let mut b = [0u8; 16];
68 b.copy_from_slice(&v);
69 Ok(ClientMessageClaim::AlreadyResolved(Uuid::from_bytes(b)))
70 }
71 _ => Ok(ClientMessageClaim::InFlight),
108 Some(v) => match ResolvedRecord::decode(&v) {
109 Some(rec) => Ok(ClientMessageClaim::AlreadyResolved(rec.message_id)),
110 None => Ok(ClientMessageClaim::InFlight),
111 },
112 None => Ok(ClientMessageClaim::InFlight),
72113 }
73114 }
74115
75 pub async fn lookup(&self, client_msg_id: &str) -> ChatLayerResult<Option<Uuid>> {
76 let key = Self::key(client_msg_id);
77 match self.kv.get_bytes(&key).await? {
78 Some(v) if v.len() == 16 => {
79 let mut b = [0u8; 16];
80 b.copy_from_slice(&v);
81 Ok(Some(Uuid::from_bytes(b)))
82 }
83 _ => Ok(None),
116 pub async fn lookup(
117 &self,
118 user_id: Uuid,
119 client_msg_id: &str,
120 ) -> ChatLayerResult<Option<ResolvedRecord>> {
121 let key = Self::key(user_id, client_msg_id);
122 let bytes = self.kv.get_bytes(&key).await?;
123 Ok(bytes.and_then(|v| ResolvedRecord::decode(&v)))
124 }
125
126 /// Resolve a lookup, additionally enforcing that the resolved record
127 /// belongs to the requesting `caller`. Returns:
128 /// - `Ok(Some(record))` — caller owns the record
129 /// - `Ok(None)` — no record exists for this caller/key
130 /// - `Err(ChatLayerError::PermissionDenied)` — record exists but belongs
131 /// to a different sender
132 pub async fn lookup_for_caller(
133 &self,
134 caller: Uuid,
135 client_msg_id: &str,
136 ) -> ChatLayerResult<Option<ResolvedRecord>> {
137 if caller.is_nil() {
138 return Err(ChatLayerError::PermissionDenied(
139 "authentication required".to_string(),
140 ));
141 }
142 match self.lookup(caller, client_msg_id).await? {
143 Some(rec) if rec.sender_id == caller => Ok(Some(rec)),
144 Some(_) => Err(ChatLayerError::PermissionDenied(
145 "client_msg_id belongs to another user".to_string(),
146 )),
147 None => Ok(None),
84148 }
85149 }
86150}
@@ -88,29 +152,35 @@
88152impl MessageIdempotencyStore for MessageIdempotency {
89153 fn is_resolved<'a>(
90154 &'a self,
155 user_id: Uuid,
91156 client_msg_id: &'a str,
92157 ) -> std::pin::Pin<Box<dyn Future<Output = ChatLayerResult<bool>> + Send + 'a>> {
93 let key = Self::key(client_msg_id);
158 let key = Self::key(user_id, client_msg_id);
94159 let kv = self.kv.clone();
95160 Box::pin(async move {
96161 Ok(matches!(
97162 kv.get_bytes(&key).await?,
98 Some(v) if v.len() == 16
163 Some(v) if v.len() == ResolvedRecord::LEN
99164 ))
100165 })
101166 }
102167
103168 fn resolve<'a>(
104169 &'a self,
170 user_id: Uuid,
105171 client_msg_id: &'a str,
172 sender_id: Uuid,
106173 message_id: Uuid,
107174 ) -> std::pin::Pin<Box<dyn Future<Output = ChatLayerResult<()>> + Send + 'a>> {
108 let key = Self::key(client_msg_id);
175 let key = Self::key(user_id, client_msg_id);
109176 let kv = self.kv.clone();
110177 let ttl = self.ttl_seconds;
111178 Box::pin(async move {
112 kv.set_bytes(&key, message_id.as_bytes().to_vec(), ttl)
113 .await?;
179 let rec = ResolvedRecord {
180 message_id,
181 sender_id,
182 };
183 kv.set_bytes(&key, rec.encode(), ttl).await?;
114184 Ok(())
115185 })
116186 }