stateset_db/sqlite/
print_stations.rs

1//! SQLite implementation of the print station / print job repository
2
3use super::{
4    map_db_error, parse_datetime_opt_row, parse_datetime_row, parse_enum_row, parse_json_row,
5    parse_uuid_row, with_immediate_transaction,
6};
7use chrono::Utc;
8use r2d2::Pool;
9use r2d2_sqlite::SqliteConnectionManager;
10use rusqlite::OptionalExtension;
11use sha2::{Digest, Sha256};
12use stateset_core::{
13    CommerceError, CreatePrintStation, EnqueuePrintJob, PairStationResult, PrintJob,
14    PrintJobFilter, PrintJobId, PrintJobStatus, PrintPayloadKind, PrintStation, PrintStationId,
15    PrintStationRepository, Result,
16};
17use uuid::Uuid;
18
19#[derive(Debug)]
20pub struct SqlitePrintStationRepository {
21    pool: Pool<SqliteConnectionManager>,
22}
23
24impl SqlitePrintStationRepository {
25    #[must_use]
26    pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
27        Self { pool }
28    }
29
30    fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
31        self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
32    }
33
34    /// SHA-256 hex digest of a token.
35    fn hash_token(token: &str) -> String {
36        let mut hasher = Sha256::new();
37        hasher.update(token.as_bytes());
38        format!("{:x}", hasher.finalize())
39    }
40
41    fn row_to_station(row: &rusqlite::Row<'_>) -> rusqlite::Result<PrintStation> {
42        let printers_json: String = row.get("printers")?;
43        Ok(PrintStation {
44            id: parse_uuid_row(&row.get::<_, String>("id")?, "print_station", "id")?.into(),
45            name: row.get("name")?,
46            printers: parse_json_row(&printers_json, "print_station", "printers")?,
47            revoked: row.get::<_, i32>("revoked")? != 0,
48            last_seen_at: parse_datetime_opt_row(
49                row.get::<_, Option<String>>("last_seen_at")?,
50                "print_station",
51                "last_seen_at",
52            )?,
53            created_at: parse_datetime_row(
54                &row.get::<_, String>("created_at")?,
55                "print_station",
56                "created_at",
57            )?,
58            updated_at: parse_datetime_row(
59                &row.get::<_, String>("updated_at")?,
60                "print_station",
61                "updated_at",
62            )?,
63        })
64    }
65
66    fn row_to_job(row: &rusqlite::Row<'_>) -> rusqlite::Result<PrintJob> {
67        Ok(PrintJob {
68            id: parse_uuid_row(&row.get::<_, String>("id")?, "print_job", "id")?.into(),
69            station_id: parse_uuid_row(
70                &row.get::<_, String>("station_id")?,
71                "print_job",
72                "station_id",
73            )?
74            .into(),
75            printer_name: row.get("printer_name")?,
76            payload_kind: parse_enum_row::<PrintPayloadKind>(
77                &row.get::<_, String>("payload_kind")?,
78                "print_job",
79                "payload_kind",
80            )?,
81            payload: row.get("payload")?,
82            status: parse_enum_row::<PrintJobStatus>(
83                &row.get::<_, String>("status")?,
84                "print_job",
85                "status",
86            )?,
87            created_at: parse_datetime_row(
88                &row.get::<_, String>("created_at")?,
89                "print_job",
90                "created_at",
91            )?,
92            picked_up_at: parse_datetime_opt_row(
93                row.get::<_, Option<String>>("picked_up_at")?,
94                "print_job",
95                "picked_up_at",
96            )?,
97        })
98    }
99
100    fn conflict(msg: &str) -> rusqlite::Error {
101        rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::Conflict(msg.to_string())))
102    }
103}
104
105impl PrintStationRepository for SqlitePrintStationRepository {
106    fn pair(&self, input: CreatePrintStation) -> Result<PairStationResult> {
107        let id = PrintStationId::new();
108        let id_str = id.to_string();
109        let now = Utc::now().to_rfc3339();
110        // Token = two random UUIDs (~244 bits). Only its hash is persisted.
111        let token = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
112        let token_hash = Self::hash_token(&token);
113        let printers_json = serde_json::to_string(&input.printers)
114            .map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
115        let station = with_immediate_transaction(&self.pool, |tx| {
116            tx.execute(
117                "INSERT INTO print_stations (id, name, token_hash, printers, revoked, created_at, updated_at)
118                 VALUES (?, ?, ?, ?, 0, ?, ?)",
119                rusqlite::params![&id_str, &input.name, &token_hash, &printers_json, &now, &now],
120            )?;
121            tx.query_row(
122                "SELECT * FROM print_stations WHERE id = ?",
123                [&id_str],
124                Self::row_to_station,
125            )
126        })?;
127        Ok(PairStationResult { station, token })
128    }
129
130    fn list_stations(&self) -> Result<Vec<PrintStation>> {
131        let conn = self.conn()?;
132        let mut stmt = conn
133            .prepare("SELECT * FROM print_stations ORDER BY created_at DESC")
134            .map_err(map_db_error)?;
135        let rows = stmt
136            .query_map([], Self::row_to_station)
137            .map_err(map_db_error)?
138            .collect::<std::result::Result<Vec<_>, _>>()
139            .map_err(map_db_error)?;
140        Ok(rows)
141    }
142
143    fn get_station(&self, id: PrintStationId) -> Result<Option<PrintStation>> {
144        let conn = self.conn()?;
145        match conn.query_row(
146            "SELECT * FROM print_stations WHERE id = ?",
147            [id.to_string()],
148            Self::row_to_station,
149        ) {
150            Ok(s) => Ok(Some(s)),
151            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
152            Err(e) => Err(map_db_error(e)),
153        }
154    }
155
156    fn revoke_station(&self, id: PrintStationId) -> Result<PrintStation> {
157        let id_str = id.to_string();
158        let now = Utc::now().to_rfc3339();
159        with_immediate_transaction(&self.pool, |tx| {
160            tx.execute(
161                "UPDATE print_stations SET revoked = 1, updated_at = ? WHERE id = ?",
162                rusqlite::params![&now, &id_str],
163            )?;
164            tx.query_row(
165                "SELECT * FROM print_stations WHERE id = ?",
166                [&id_str],
167                Self::row_to_station,
168            )
169        })
170    }
171
172    fn enqueue_job(&self, station_id: PrintStationId, input: EnqueuePrintJob) -> Result<PrintJob> {
173        let station_str = station_id.to_string();
174        let job_id = PrintJobId::new();
175        let job_str = job_id.to_string();
176        let now = Utc::now().to_rfc3339();
177        with_immediate_transaction(&self.pool, |tx| {
178            let revoked: Option<i32> = tx
179                .query_row("SELECT revoked FROM print_stations WHERE id = ?", [&station_str], |r| {
180                    r.get(0)
181                })
182                .optional()?;
183            match revoked {
184                None => return Err(rusqlite::Error::QueryReturnedNoRows),
185                Some(r) if r != 0 => return Err(Self::conflict("station is revoked")),
186                _ => {}
187            }
188            tx.execute(
189                "INSERT INTO print_jobs (id, station_id, printer_name, payload_kind, payload, status, created_at)
190                 VALUES (?, ?, ?, ?, ?, 'queued', ?)",
191                rusqlite::params![
192                    &job_str,
193                    &station_str,
194                    &input.printer_name,
195                    input.payload_kind.to_string(),
196                    &input.payload,
197                    &now,
198                ],
199            )?;
200            tx.query_row("SELECT * FROM print_jobs WHERE id = ?", [&job_str], Self::row_to_job)
201        })
202    }
203
204    fn next_job(&self, station_id: PrintStationId) -> Result<Option<PrintJob>> {
205        let station_str = station_id.to_string();
206        let now = Utc::now().to_rfc3339();
207        with_immediate_transaction(&self.pool, |tx| {
208            // Touch last-seen on every poll.
209            tx.execute(
210                "UPDATE print_stations SET last_seen_at = ?, updated_at = ? WHERE id = ?",
211                rusqlite::params![&now, &now, &station_str],
212            )?;
213            let job_id: Option<String> = tx
214                .query_row(
215                    "SELECT id FROM print_jobs WHERE station_id = ? AND status = 'queued' ORDER BY created_at ASC LIMIT 1",
216                    [&station_str],
217                    |r| r.get(0),
218                )
219                .optional()?;
220            let Some(job_id) = job_id else {
221                return Ok(None);
222            };
223            tx.execute(
224                "UPDATE print_jobs SET status = 'picked_up', picked_up_at = ? WHERE id = ?",
225                rusqlite::params![&now, &job_id],
226            )?;
227            let job =
228                tx.query_row("SELECT * FROM print_jobs WHERE id = ?", [&job_id], Self::row_to_job)?;
229            Ok(Some(job))
230        })
231    }
232
233    fn complete_job(&self, job_id: PrintJobId, success: bool) -> Result<PrintJob> {
234        let job_str = job_id.to_string();
235        let status = if success { PrintJobStatus::Printed } else { PrintJobStatus::Failed };
236        with_immediate_transaction(&self.pool, |tx| {
237            tx.execute(
238                "UPDATE print_jobs SET status = ? WHERE id = ?",
239                rusqlite::params![status.to_string(), &job_str],
240            )?;
241            tx.query_row("SELECT * FROM print_jobs WHERE id = ?", [&job_str], Self::row_to_job)
242        })
243    }
244
245    fn list_jobs(
246        &self,
247        station_id: PrintStationId,
248        filter: PrintJobFilter,
249    ) -> Result<Vec<PrintJob>> {
250        let conn = self.conn()?;
251        let mut sql = "SELECT * FROM print_jobs WHERE station_id = ?".to_string();
252        let mut params: Vec<Box<dyn rusqlite::types::ToSql>> =
253            vec![Box::new(station_id.to_string())];
254        if let Some(status) = filter.status {
255            sql.push_str(" AND status = ?");
256            params.push(Box::new(status.to_string()));
257        }
258        sql.push_str(" ORDER BY created_at DESC");
259        crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
260        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
261            params.iter().map(|p| p.as_ref()).collect();
262        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
263        let rows = stmt
264            .query_map(param_refs.as_slice(), Self::row_to_job)
265            .map_err(map_db_error)?
266            .collect::<std::result::Result<Vec<_>, _>>()
267            .map_err(map_db_error)?;
268        Ok(rows)
269    }
270}
271
272#[cfg(test)]
273mod tests {
274    use super::*;
275    use crate::DatabaseConfig;
276    use crate::sqlite::SqliteDatabase;
277
278    fn test_repo() -> SqlitePrintStationRepository {
279        let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
280        SqlitePrintStationRepository::new(db.pool().clone())
281    }
282
283    fn pair(repo: &SqlitePrintStationRepository) -> PairStationResult {
284        repo.pair(CreatePrintStation {
285            name: "Packing Bench 1".into(),
286            printers: vec!["Zebra-1".into()],
287        })
288        .expect("pair")
289    }
290
291    fn job(printer: Option<&str>) -> EnqueuePrintJob {
292        EnqueuePrintJob {
293            printer_name: printer.map(String::from),
294            payload_kind: PrintPayloadKind::Zpl,
295            payload: "^XA^FO50,50^FDhi^FS^XZ".into(),
296        }
297    }
298
299    #[test]
300    fn pair_returns_token_and_persists_hash() {
301        let repo = test_repo();
302        let result = pair(&repo);
303        assert!(!result.token.is_empty());
304        assert!(!result.station.revoked);
305        // token hash is not surfaced on the station struct
306        let fetched = repo.get_station(result.station.id).expect("get").expect("found");
307        assert_eq!(fetched.printers, vec!["Zebra-1".to_string()]);
308    }
309
310    #[test]
311    fn enqueue_and_pick_up_in_fifo_order() {
312        let repo = test_repo();
313        let station = pair(&repo).station;
314        repo.enqueue_job(station.id, job(Some("Zebra-1"))).expect("enqueue 1");
315        repo.enqueue_job(station.id, job(Some("Zebra-1"))).expect("enqueue 2");
316
317        let first = repo.next_job(station.id).expect("next").expect("job");
318        assert_eq!(first.status, PrintJobStatus::PickedUp);
319        assert!(first.picked_up_at.is_some());
320        // station last-seen updated
321        let s = repo.get_station(station.id).expect("get").expect("found");
322        assert!(s.last_seen_at.is_some());
323
324        // completing the first leaves one queued
325        let done = repo.complete_job(first.id, true).expect("complete");
326        assert_eq!(done.status, PrintJobStatus::Printed);
327        let second = repo.next_job(station.id).expect("next").expect("job");
328        assert_ne!(second.id, first.id);
329        // queue now empty
330        assert!(repo.next_job(station.id).expect("next").is_none());
331    }
332
333    #[test]
334    fn revoked_station_rejects_jobs() {
335        let repo = test_repo();
336        let station = pair(&repo).station;
337        repo.revoke_station(station.id).expect("revoke");
338        assert!(repo.enqueue_job(station.id, job(None)).is_err());
339    }
340
341    #[test]
342    fn list_jobs_filters_by_status() {
343        let repo = test_repo();
344        let station = pair(&repo).station;
345        let j = repo.enqueue_job(station.id, job(None)).expect("enqueue");
346        repo.enqueue_job(station.id, job(None)).expect("enqueue2");
347        repo.complete_job(j.id, false).expect("fail");
348        let failed = repo
349            .list_jobs(
350                station.id,
351                PrintJobFilter { status: Some(PrintJobStatus::Failed), ..Default::default() },
352            )
353            .expect("list");
354        assert_eq!(failed.len(), 1);
355    }
356}
stateset_db/sqlite/print_stations.rs

stateset_db/sqlite/
print_stations.rs
