minco_plugin_feedback/
store.rs

1use crate::{FeedbackAccessToken, FeedbackId, FeedbackListFilter, FeedbackSummary, FeedbackThread};
2use async_trait::async_trait;
3use sha2::{Digest, Sha256};
4use std::{collections::BTreeMap, sync::Arc};
5use subtle::ConstantTimeEq;
6use tokio::sync::RwLock;
7
8#[async_trait]
9pub trait FeedbackStore: Send + Sync + std::fmt::Debug {
10    async fn create(
11        &self,
12        thread: FeedbackThread,
13        client_token_hash: String,
14    ) -> Result<(), FeedbackStoreError>;
15
16    async fn get(&self, id: FeedbackId) -> Result<Option<FeedbackThread>, FeedbackStoreError>;
17
18    async fn get_for_client(
19        &self,
20        id: FeedbackId,
21        client_token_hash: &str,
22    ) -> Result<Option<FeedbackThread>, FeedbackStoreError>;
23
24    async fn list(
25        &self,
26        filter: FeedbackListFilter,
27    ) -> Result<Vec<FeedbackSummary>, FeedbackStoreError>;
28
29    async fn save(
30        &self,
31        thread: FeedbackThread,
32        expected_revision: u64,
33    ) -> Result<(), FeedbackStoreError>;
34
35    /// Bounded readiness check for deployment health reporting.
36    async fn ready(&self) -> Result<(), FeedbackStoreError> {
37        Ok(())
38    }
39}
40
41#[derive(Clone)]
42pub struct FeedbackStoreService(pub Arc<dyn FeedbackStore>);
43
44impl std::fmt::Debug for FeedbackStoreService {
45    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        formatter.debug_tuple("FeedbackStoreService").finish()
47    }
48}
49
50impl FeedbackStoreService {
51    pub fn new(store: Arc<dyn FeedbackStore>) -> Self {
52        Self(store)
53    }
54
55    pub async fn create(
56        &self,
57        thread: FeedbackThread,
58        client_token_hash: String,
59    ) -> Result<(), FeedbackStoreError> {
60        self.0.create(thread, client_token_hash).await
61    }
62
63    pub async fn get(&self, id: FeedbackId) -> Result<Option<FeedbackThread>, FeedbackStoreError> {
64        self.0.get(id).await
65    }
66
67    pub async fn get_for_client(
68        &self,
69        id: FeedbackId,
70        client_token_hash: &str,
71    ) -> Result<Option<FeedbackThread>, FeedbackStoreError> {
72        self.0.get_for_client(id, client_token_hash).await
73    }
74
75    pub async fn list(
76        &self,
77        filter: FeedbackListFilter,
78    ) -> Result<Vec<FeedbackSummary>, FeedbackStoreError> {
79        self.0.list(filter).await
80    }
81
82    pub async fn save(
83        &self,
84        thread: FeedbackThread,
85        expected_revision: u64,
86    ) -> Result<(), FeedbackStoreError> {
87        self.0.save(thread, expected_revision).await
88    }
89
90    pub async fn ready(&self) -> Result<(), FeedbackStoreError> {
91        self.0.ready().await
92    }
93}
94
95#[derive(Debug, Clone)]
96struct MemoryFeedbackEntry {
97    thread: FeedbackThread,
98    client_token_hash: String,
99}
100
101#[derive(Debug, Default)]
102pub struct MemoryFeedbackStore {
103    entries: RwLock<BTreeMap<FeedbackId, MemoryFeedbackEntry>>,
104}
105
106#[async_trait]
107impl FeedbackStore for MemoryFeedbackStore {
108    async fn create(
109        &self,
110        thread: FeedbackThread,
111        client_token_hash: String,
112    ) -> Result<(), FeedbackStoreError> {
113        let mut entries = self.entries.write().await;
114        if entries.contains_key(&thread.id) {
115            return Err(FeedbackStoreError::AlreadyExists(thread.id));
116        }
117        entries.insert(
118            thread.id,
119            MemoryFeedbackEntry {
120                thread,
121                client_token_hash,
122            },
123        );
124        drop(entries);
125        Ok(())
126    }
127
128    async fn get(&self, id: FeedbackId) -> Result<Option<FeedbackThread>, FeedbackStoreError> {
129        Ok(self
130            .entries
131            .read()
132            .await
133            .get(&id)
134            .map(|entry| entry.thread.clone()))
135    }
136
137    async fn get_for_client(
138        &self,
139        id: FeedbackId,
140        client_token_hash: &str,
141    ) -> Result<Option<FeedbackThread>, FeedbackStoreError> {
142        Ok(self.entries.read().await.get(&id).and_then(|entry| {
143            if constant_time_equals(&entry.client_token_hash, client_token_hash) {
144                Some(entry.thread.clone())
145            } else {
146                None
147            }
148        }))
149    }
150
151    async fn list(
152        &self,
153        filter: FeedbackListFilter,
154    ) -> Result<Vec<FeedbackSummary>, FeedbackStoreError> {
155        let limit = filter.limit.clamp(1, 200);
156        let mut threads = self
157            .entries
158            .read()
159            .await
160            .values()
161            .map(|entry| entry.thread.clone())
162            .filter(|thread| {
163                filter.status.is_none_or(|status| thread.status == status)
164                    && filter
165                        .project_id
166                        .as_deref()
167                        .is_none_or(|project_id| thread.project_id == project_id)
168            })
169            .collect::<Vec<_>>();
170        threads.sort_by(|left, right| {
171            right
172                .updated_at
173                .cmp(&left.updated_at)
174                .then_with(|| right.id.cmp(&left.id))
175        });
176        Ok(threads
177            .iter()
178            .take(limit)
179            .map(FeedbackSummary::from)
180            .collect())
181    }
182
183    async fn save(
184        &self,
185        thread: FeedbackThread,
186        expected_revision: u64,
187    ) -> Result<(), FeedbackStoreError> {
188        let mut entries = self.entries.write().await;
189        {
190            let entry = entries
191                .get_mut(&thread.id)
192                .ok_or(FeedbackStoreError::NotFound(thread.id))?;
193            if entry.thread.revision != expected_revision {
194                return Err(FeedbackStoreError::ConcurrentModification {
195                    id: thread.id,
196                    expected_revision,
197                    actual_revision: entry.thread.revision,
198                });
199            }
200            entry.thread = thread;
201        }
202        drop(entries);
203        Ok(())
204    }
205}
206
207pub fn hash_access_token(token: &FeedbackAccessToken) -> String {
208    format!("{:x}", Sha256::digest(token.expose().as_bytes()))
209}
210
211fn constant_time_equals(left: &str, right: &str) -> bool {
212    left.as_bytes().ct_eq(right.as_bytes()).into()
213}
214
215#[derive(Debug, thiserror::Error)]
216pub enum FeedbackStoreError {
217    #[error("feedback already exists: {0}")]
218    AlreadyExists(FeedbackId),
219    #[error("feedback was not found: {0}")]
220    NotFound(FeedbackId),
221    #[error(
222        "feedback {id} changed concurrently: expected revision {expected_revision}, actual {actual_revision}"
223    )]
224    ConcurrentModification {
225        id: FeedbackId,
226        expected_revision: u64,
227        actual_revision: u64,
228    },
229    #[error("feedback store failed: {0}")]
230    Infrastructure(String),
231}
232
233#[cfg(test)]
234mod tests {
235    use super::*;
236    use crate::{CreateFeedbackInput, FeedbackContext, FeedbackKind, FeedbackPriority};
237    use std::collections::BTreeSet;
238
239    fn thread() -> FeedbackThread {
240        FeedbackThread::create(CreateFeedbackInput {
241            project_id: "example".into(),
242            kind: FeedbackKind::Bug,
243            priority: FeedbackPriority::Normal,
244            title: "Problem".into(),
245            description: "Something did not work.".into(),
246            context: FeedbackContext {
247                page_url: "https://example.test".into(),
248                route_name: None,
249                release_id: None,
250                environment: None,
251                request_id: None,
252                user_agent: None,
253                viewport: None,
254                client_subject: None,
255            },
256            tags: BTreeSet::new(),
257        })
258        .unwrap()
259    }
260
261    #[tokio::test]
262    async fn access_token_is_required_for_client_reads() {
263        let store = MemoryFeedbackStore::default();
264        let feedback = thread();
265        let id = feedback.id;
266        let token = FeedbackAccessToken::generate();
267        store
268            .create(feedback, hash_access_token(&token))
269            .await
270            .unwrap();
271        assert!(
272            store
273                .get_for_client(id, &hash_access_token(&token))
274                .await
275                .unwrap()
276                .is_some()
277        );
278        assert!(
279            store
280                .get_for_client(id, &hash_access_token(&FeedbackAccessToken::generate()))
281                .await
282                .unwrap()
283                .is_none()
284        );
285    }
286
287    #[tokio::test]
288    async fn optimistic_revision_prevents_lost_updates() {
289        let store = MemoryFeedbackStore::default();
290        let feedback = thread();
291        let id = feedback.id;
292        store
293            .create(feedback.clone(), "token".into())
294            .await
295            .unwrap();
296        let mut updated = feedback;
297        updated.append_message(crate::FeedbackMessage::client("More detail").unwrap());
298        store.save(updated.clone(), 1).await.unwrap();
299        assert!(matches!(
300            store.save(updated, 1).await,
301            Err(FeedbackStoreError::ConcurrentModification { id: conflict_id, .. })
302                if conflict_id == id
303        ));
304    }
305}
minco_plugin_feedback/store.rs

minco_plugin_feedback/
store.rs
