max / balanced_breakfast

2.9 KB · 82 lines History Blame Raw

1	use super::{BusserStateId, DbBusserState, SqlitePool, TIMESTAMP_FMT, Utc};
2
3	#[derive(Clone)]
4	/// Repository for busser key-value state (cursors, tokens, pagination markers)
5	pub struct StateRepository {
6	pool: SqlitePool,
7	}
8
9	impl StateRepository {
10	/// Create a new state repository backed by the given pool.
11	#[tracing::instrument(skip_all)]
12	pub fn new(pool: SqlitePool) -> Self {
13	Self { pool }
14	}
15
16	/// Get a single state value for a busser by key.
17	#[tracing::instrument(skip_all)]
18	pub async fn get(&self, busser_id: &str, key: &str) -> Result<Option<String>, sqlx::Error> {
19	let row: Option<(String,)> =
20	sqlx::query_as("SELECT value FROM busser_state WHERE busser_id = ?1 AND key = ?2")
21	.bind(busser_id)
22	.bind(key)
23	.fetch_optional(&self.pool)
24	.await?;
25	Ok(row.map(\|(v,)\| v))
26	}
27
28	/// Set a state value, inserting or updating on conflict.
29	/// Uses upsert on the `(busser_id, key)` composite unique constraint so
30	/// callers don't need to check existence first.
31	#[tracing::instrument(skip_all)]
32	pub async fn set(&self, busser_id: &str, key: &str, value: &str) -> Result<(), sqlx::Error> {
33	let id = BusserStateId::new();
34	let now = Utc::now().format(TIMESTAMP_FMT).to_string();
35	sqlx::query(
36	r"
37	INSERT INTO busser_state (id, busser_id, key, value, created_at, updated_at)
38	VALUES (?1, ?2, ?3, ?4, ?5, ?5)
39	ON CONFLICT (busser_id, key) DO UPDATE SET value = EXCLUDED.value, updated_at = EXCLUDED.updated_at
40	",
41	)
42	.bind(id)
43	.bind(busser_id)
44	.bind(key)
45	.bind(value)
46	.bind(&now)
47	.execute(&self.pool)
48	.await?;
49	Ok(())
50	}
51
52	/// Delete a single state entry by busser ID and key.
53	#[tracing::instrument(skip_all)]
54	pub async fn delete(&self, busser_id: &str, key: &str) -> Result<(), sqlx::Error> {
55	sqlx::query("DELETE FROM busser_state WHERE busser_id = ?1 AND key = ?2")
56	.bind(busser_id)
57	.bind(key)
58	.execute(&self.pool)
59	.await?;
60	Ok(())
61	}
62
63	/// Delete all state entries for a busser. Returns the number of rows removed.
64	#[tracing::instrument(skip_all)]
65	pub async fn delete_all(&self, busser_id: &str) -> Result<u64, sqlx::Error> {
66	let result = sqlx::query("DELETE FROM busser_state WHERE busser_id = ?1")
67	.bind(busser_id)
68	.execute(&self.pool)
69	.await?;
70	Ok(result.rows_affected())
71	}
72
73	/// List all state entries for a busser, ordered by key.
74	#[tracing::instrument(skip_all)]
75	pub async fn list(&self, busser_id: &str) -> Result<Vec<DbBusserState>, sqlx::Error> {
76	sqlx::query_as("SELECT * FROM busser_state WHERE busser_id = ?1 ORDER BY key")
77	.bind(busser_id)
78	.fetch_all(&self.pool)
79	.await
80	}
81	}
82