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

minco_plugin_feedback/persistence/
postgres.rs
