minco_plugin_feedback/persistence/
sqlite.rs

1use super::{
2    MIGRATION_HISTORY_TABLE, decode_thread, encode_thread, revision_from_i64, revision_to_i64,
3};
4use crate::{
5    FeedbackId, FeedbackListFilter, FeedbackStore, FeedbackStoreError, FeedbackSummary,
6    FeedbackThread,
7};
8use async_trait::async_trait;
9use sqlx::{Row, SqlitePool};
10
11#[derive(Debug, Clone)]
12pub struct SqliteFeedbackStore {
13    pool: SqlitePool,
14}
15
16impl SqliteFeedbackStore {
17    pub const fn new(pool: SqlitePool) -> Self {
18        Self { pool }
19    }
20
21    pub const fn pool(&self) -> &SqlitePool {
22        &self.pool
23    }
24
25    pub async fn migrate(&self) -> Result<(), FeedbackStoreError> {
26        let mut migrator = sqlx::migrate!("migrations/sqlite");
27        migrator.dangerous_set_table_name(MIGRATION_HISTORY_TABLE);
28        migrator
29            .run(&self.pool)
30            .await
31            .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))
32    }
33}
34
35#[async_trait]
36impl FeedbackStore for SqliteFeedbackStore {
37    async fn create(
38        &self,
39        thread: FeedbackThread,
40        client_token_hash: String,
41    ) -> Result<(), FeedbackStoreError> {
42        let document = serde_json::to_string(&encode_thread(&thread)?)
43            .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))?;
44        let result = sqlx::query(
45            r"
46            INSERT INTO minco_feedback_threads (
47                id, client_token_hash, project_id, status, document,
48                revision, created_at, updated_at
49            )
50            VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
51            ",
52        )
53        .bind(thread.id.to_string())
54        .bind(client_token_hash)
55        .bind(&thread.project_id)
56        .bind(thread.status.to_string())
57        .bind(document)
58        .bind(revision_to_i64(thread.revision)?)
59        .bind(thread.created_at.to_rfc3339())
60        .bind(thread.updated_at.to_rfc3339())
61        .execute(&self.pool)
62        .await;
63
64        match result {
65            Ok(_) => Ok(()),
66            Err(error)
67                if error
68                    .as_database_error()
69                    .is_some_and(sqlx::error::DatabaseError::is_unique_violation) =>
70            {
71                Err(FeedbackStoreError::AlreadyExists(thread.id))
72            }
73            Err(error) => Err(FeedbackStoreError::Infrastructure(error.to_string())),
74        }
75    }
76
77    async fn get(&self, id: FeedbackId) -> Result<Option<FeedbackThread>, FeedbackStoreError> {
78        let row = sqlx::query("SELECT document FROM minco_feedback_threads WHERE id = ?1")
79            .bind(id.to_string())
80            .fetch_optional(&self.pool)
81            .await
82            .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))?;
83        row.map(|row| decode_row(&row)).transpose()
84    }
85
86    async fn get_for_client(
87        &self,
88        id: FeedbackId,
89        client_token_hash: &str,
90    ) -> Result<Option<FeedbackThread>, FeedbackStoreError> {
91        let row = sqlx::query(
92            "SELECT document FROM minco_feedback_threads WHERE id = ?1 AND client_token_hash = ?2",
93        )
94        .bind(id.to_string())
95        .bind(client_token_hash)
96        .fetch_optional(&self.pool)
97        .await
98        .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))?;
99        row.map(|row| decode_row(&row)).transpose()
100    }
101
102    async fn list(
103        &self,
104        filter: FeedbackListFilter,
105    ) -> Result<Vec<FeedbackSummary>, FeedbackStoreError> {
106        let status = filter.status.map(|value| value.to_string());
107        let limit = i64::try_from(filter.limit.clamp(1, 200))
108            .map_err(|_| FeedbackStoreError::Infrastructure("invalid list limit".into()))?;
109        let rows = sqlx::query(
110            r"
111            SELECT document
112            FROM minco_feedback_threads
113            WHERE (?1 IS NULL OR status = ?1)
114              AND (?2 IS NULL OR project_id = ?2)
115            ORDER BY updated_at DESC, id DESC
116            LIMIT ?3
117            ",
118        )
119        .bind(status)
120        .bind(filter.project_id)
121        .bind(limit)
122        .fetch_all(&self.pool)
123        .await
124        .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))?;
125
126        rows.into_iter()
127            .map(|row| decode_row(&row).map(|thread| FeedbackSummary::from(&thread)))
128            .collect()
129    }
130
131    async fn ready(&self) -> Result<(), FeedbackStoreError> {
132        sqlx::query_scalar::<_, i64>("SELECT 1")
133            .fetch_one(&self.pool)
134            .await
135            .map(|_| ())
136            .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))
137    }
138
139    async fn save(
140        &self,
141        thread: FeedbackThread,
142        expected_revision: u64,
143    ) -> Result<(), FeedbackStoreError> {
144        let document = serde_json::to_string(&encode_thread(&thread)?)
145            .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))?;
146        let result = sqlx::query(
147            r"
148            UPDATE minco_feedback_threads
149            SET project_id = ?2,
150                status = ?3,
151                document = ?4,
152                revision = ?5,
153                updated_at = ?6
154            WHERE id = ?1 AND revision = ?7
155            ",
156        )
157        .bind(thread.id.to_string())
158        .bind(&thread.project_id)
159        .bind(thread.status.to_string())
160        .bind(document)
161        .bind(revision_to_i64(thread.revision)?)
162        .bind(thread.updated_at.to_rfc3339())
163        .bind(revision_to_i64(expected_revision)?)
164        .execute(&self.pool)
165        .await
166        .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))?;
167
168        if result.rows_affected() == 1 {
169            return Ok(());
170        }
171        let actual = sqlx::query("SELECT revision FROM minco_feedback_threads WHERE id = ?1")
172            .bind(thread.id.to_string())
173            .fetch_optional(&self.pool)
174            .await
175            .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))?;
176        match actual {
177            None => Err(FeedbackStoreError::NotFound(thread.id)),
178            Some(row) => Err(FeedbackStoreError::ConcurrentModification {
179                id: thread.id,
180                expected_revision,
181                actual_revision: revision_from_i64(
182                    row.try_get::<i64, _>("revision")
183                        .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))?,
184                )?,
185            }),
186        }
187    }
188}
189
190fn decode_row(row: &sqlx::sqlite::SqliteRow) -> Result<FeedbackThread, FeedbackStoreError> {
191    let document = row
192        .try_get::<String, _>("document")
193        .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))?;
194    let value = serde_json::from_str(&document)
195        .map_err(|error| FeedbackStoreError::Infrastructure(error.to_string()))?;
196    decode_thread(value)
197}
minco_plugin_feedback/persistence/sqlite.rs

minco_plugin_feedback/persistence/
sqlite.rs
