1use async_trait::async_trait;
2use chrono::{DateTime, TimeDelta, Utc};
3use minco_plugin_audit::{AuditError, AuditEvent, AuditSink};
4use minco_plugin_idempotency::{
5 BeginOutcome, IdempotencyError, IdempotencyKey, IdempotencyLease, IdempotencyRecord,
6 IdempotencyStore, RequestFingerprint, validate_claim_timeout,
7};
8use minco_plugin_sessions::{
9 SessionError, SessionId, SessionRecord, SessionStore, SessionTokenHash,
10};
11use sqlx::{Row, SqlitePool};
12use uuid::Uuid;
13
14#[derive(Debug, Clone)]
15pub struct SqliteSessionStore {
16 pool: SqlitePool,
17}
18
19impl SqliteSessionStore {
20 pub const fn new(pool: SqlitePool) -> Self {
21 Self { pool }
22 }
23}
24
25#[async_trait]
26impl SessionStore for SqliteSessionStore {
27 async fn create(
28 &self,
29 token_hash: SessionTokenHash,
30 session: SessionRecord,
31 ) -> Result<(), SessionError> {
32 let attributes = serde_json::to_string(&session.attributes).map_err(session_store_error)?;
33 let result = sqlx::query(
34 "INSERT INTO minco_sessions
35 (id, token_hash, subject, created_at, expires_at, revoked_at, attributes)
36 VALUES (?, ?, ?, ?, ?, ?, ?)",
37 )
38 .bind(session.id.0)
39 .bind(token_hash.as_bytes().as_slice())
40 .bind(session.subject)
41 .bind(session.created_at)
42 .bind(session.expires_at)
43 .bind(session.revoked_at)
44 .bind(attributes)
45 .execute(&self.pool)
46 .await;
47 match result {
48 Ok(_) => Ok(()),
49 Err(error) if is_unique_violation(&error) => Err(SessionError::Duplicate),
50 Err(error) => Err(session_store_error(error)),
51 }
52 }
53
54 async fn find_by_token_hash(
55 &self,
56 token_hash: SessionTokenHash,
57 ) -> Result<Option<SessionRecord>, SessionError> {
58 let row = sqlx::query(
59 "SELECT id, subject, created_at, expires_at, revoked_at, attributes
60 FROM minco_sessions WHERE token_hash = ?",
61 )
62 .bind(token_hash.as_bytes().as_slice())
63 .fetch_optional(&self.pool)
64 .await
65 .map_err(session_store_error)?;
66 row.map(|row| decode_session(&row)).transpose()
67 }
68
69 async fn revoke(&self, id: SessionId, at: DateTime<Utc>) -> Result<bool, SessionError> {
70 let result = sqlx::query(
71 "UPDATE minco_sessions SET revoked_at = ?
72 WHERE id = ? AND revoked_at IS NULL",
73 )
74 .bind(at)
75 .bind(id.0)
76 .execute(&self.pool)
77 .await
78 .map_err(session_store_error)?;
79 Ok(result.rows_affected() == 1)
80 }
81
82 async fn revoke_subject(
83 &self,
84 subject: &str,
85 at: DateTime<Utc>,
86 ) -> Result<usize, SessionError> {
87 let result = sqlx::query(
88 "UPDATE minco_sessions SET revoked_at = ?
89 WHERE subject = ? AND revoked_at IS NULL",
90 )
91 .bind(at)
92 .bind(subject)
93 .execute(&self.pool)
94 .await
95 .map_err(session_store_error)?;
96 usize::try_from(result.rows_affected())
97 .map_err(|error| session_store_error(error.to_string()))
98 }
99}
100
101fn decode_session(row: &sqlx::sqlite::SqliteRow) -> Result<SessionRecord, SessionError> {
102 let attributes: String = row.try_get("attributes").map_err(session_store_error)?;
103 Ok(SessionRecord {
104 id: SessionId(row.try_get("id").map_err(session_store_error)?),
105 subject: row.try_get("subject").map_err(session_store_error)?,
106 created_at: row.try_get("created_at").map_err(session_store_error)?,
107 expires_at: row.try_get("expires_at").map_err(session_store_error)?,
108 revoked_at: row.try_get("revoked_at").map_err(session_store_error)?,
109 attributes: serde_json::from_str(&attributes).map_err(session_store_error)?,
110 })
111}
112
113#[derive(Debug, Clone)]
114pub struct SqliteIdempotencyStore {
115 pool: SqlitePool,
116}
117
118impl SqliteIdempotencyStore {
119 pub const fn new(pool: SqlitePool) -> Self {
120 Self { pool }
121 }
122}
123
124#[async_trait]
125impl IdempotencyStore for SqliteIdempotencyStore {
126 async fn get(
127 &self,
128 key: &IdempotencyKey,
129 ) -> Result<Option<IdempotencyRecord>, IdempotencyError> {
130 let row = sqlx::query(
131 "SELECT fingerprint, response, completed_at
132 FROM minco_idempotency WHERE key = ? AND state = 'completed'",
133 )
134 .bind(key.as_str())
135 .fetch_optional(&self.pool)
136 .await
137 .map_err(idempotency_store_error)?;
138 row.map(|row| decode_completed(&row)).transpose()
139 }
140
141 async fn begin(
142 &self,
143 key: IdempotencyKey,
144 fingerprint: RequestFingerprint,
145 now: DateTime<Utc>,
146 stale_after: TimeDelta,
147 ) -> Result<BeginOutcome, IdempotencyError> {
148 validate_claim_timeout(stale_after)?;
149 let lease_id = Uuid::now_v7();
150 let mut transaction = self
151 .pool
152 .begin_with("BEGIN IMMEDIATE")
153 .await
154 .map_err(idempotency_store_error)?;
155 let inserted = sqlx::query(
156 "INSERT INTO minco_idempotency
157 (key, fingerprint, state, lease_id, started_at)
158 VALUES (?, ?, 'in_progress', ?, ?)
159 ON CONFLICT(key) DO NOTHING",
160 )
161 .bind(key.as_str())
162 .bind(fingerprint.as_str())
163 .bind(lease_id)
164 .bind(now)
165 .execute(&mut *transaction)
166 .await
167 .map_err(idempotency_store_error)?
168 .rows_affected()
169 == 1;
170 if inserted {
171 transaction
172 .commit()
173 .await
174 .map_err(idempotency_store_error)?;
175 return Ok(BeginOutcome::Started(IdempotencyLease {
176 key,
177 fingerprint,
178 lease_id,
179 started_at: now,
180 }));
181 }
182
183 let row = sqlx::query(
184 "SELECT fingerprint, state, lease_id, started_at, response, completed_at
185 FROM minco_idempotency WHERE key = ?",
186 )
187 .bind(key.as_str())
188 .fetch_one(&mut *transaction)
189 .await
190 .map_err(idempotency_store_error)?;
191 let stored_fingerprint = RequestFingerprint::parse(
192 row.try_get::<String, _>("fingerprint")
193 .map_err(idempotency_store_error)?,
194 )?;
195 if stored_fingerprint != fingerprint {
196 transaction
197 .commit()
198 .await
199 .map_err(idempotency_store_error)?;
200 return Ok(BeginOutcome::Conflict);
201 }
202 let state: String = row.try_get("state").map_err(idempotency_store_error)?;
203 if state == "completed" {
204 let record = decode_completed(&row)?;
205 transaction
206 .commit()
207 .await
208 .map_err(idempotency_store_error)?;
209 return Ok(BeginOutcome::Replay(record));
210 }
211 let started_at: DateTime<Utc> =
212 row.try_get("started_at").map_err(idempotency_store_error)?;
213 if started_at > now - stale_after {
214 transaction
215 .commit()
216 .await
217 .map_err(idempotency_store_error)?;
218 return Ok(BeginOutcome::InProgress { started_at });
219 }
220 sqlx::query(
221 "UPDATE minco_idempotency
222 SET lease_id = ?, started_at = ?, response = NULL, completed_at = NULL
223 WHERE key = ? AND state = 'in_progress'",
224 )
225 .bind(lease_id)
226 .bind(now)
227 .bind(key.as_str())
228 .execute(&mut *transaction)
229 .await
230 .map_err(idempotency_store_error)?;
231 transaction
232 .commit()
233 .await
234 .map_err(idempotency_store_error)?;
235 Ok(BeginOutcome::Started(IdempotencyLease {
236 key,
237 fingerprint,
238 lease_id,
239 started_at: now,
240 }))
241 }
242
243 async fn complete(
244 &self,
245 lease: IdempotencyLease,
246 response: serde_json::Value,
247 completed_at: DateTime<Utc>,
248 ) -> Result<IdempotencyRecord, IdempotencyError> {
249 let encoded = serde_json::to_string(&response).map_err(idempotency_store_error)?;
250 let result = sqlx::query(
251 "UPDATE minco_idempotency
252 SET state = 'completed', response = ?, completed_at = ?, lease_id = NULL
253 WHERE key = ? AND fingerprint = ? AND state = 'in_progress' AND lease_id = ?",
254 )
255 .bind(encoded)
256 .bind(completed_at)
257 .bind(lease.key.as_str())
258 .bind(lease.fingerprint.as_str())
259 .bind(lease.lease_id)
260 .execute(&self.pool)
261 .await
262 .map_err(idempotency_store_error)?;
263 if result.rows_affected() != 1 {
264 return Err(IdempotencyError::InvalidLease);
265 }
266 Ok(IdempotencyRecord {
267 fingerprint: lease.fingerprint,
268 response,
269 created_at: completed_at,
270 })
271 }
272
273 async fn abort(&self, lease: &IdempotencyLease) -> Result<bool, IdempotencyError> {
274 let result = sqlx::query(
275 "DELETE FROM minco_idempotency
276 WHERE key = ? AND fingerprint = ? AND state = 'in_progress' AND lease_id = ?",
277 )
278 .bind(lease.key.as_str())
279 .bind(lease.fingerprint.as_str())
280 .bind(lease.lease_id)
281 .execute(&self.pool)
282 .await
283 .map_err(idempotency_store_error)?;
284 Ok(result.rows_affected() == 1)
285 }
286}
287
288fn decode_completed(row: &sqlx::sqlite::SqliteRow) -> Result<IdempotencyRecord, IdempotencyError> {
289 let response: String = row.try_get("response").map_err(idempotency_store_error)?;
290 Ok(IdempotencyRecord {
291 fingerprint: RequestFingerprint::parse(
292 row.try_get::<String, _>("fingerprint")
293 .map_err(idempotency_store_error)?,
294 )?,
295 response: serde_json::from_str(&response).map_err(idempotency_store_error)?,
296 created_at: row
297 .try_get("completed_at")
298 .map_err(idempotency_store_error)?,
299 })
300}
301
302#[derive(Debug, Clone)]
303pub struct SqliteAuditSink {
304 pool: SqlitePool,
305}
306
307impl SqliteAuditSink {
308 pub const fn new(pool: SqlitePool) -> Self {
309 Self { pool }
310 }
311}
312
313#[async_trait]
314impl AuditSink for SqliteAuditSink {
315 async fn append(&self, event: AuditEvent) -> Result<(), AuditError> {
316 if event.action.trim().is_empty() || event.resource_id.trim().is_empty() {
317 return Err(AuditError::InvalidEvent);
318 }
319 let metadata = serde_json::to_string(&event.metadata)
320 .map_err(|error| AuditError::Append(error.to_string()))?;
321 sqlx::query(
322 "INSERT INTO minco_audit
323 (id, action, resource_type, resource_id, actor_subject, correlation_id,
324 occurred_at, metadata)
325 VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
326 )
327 .bind(event.id)
328 .bind(event.action)
329 .bind(event.resource_type)
330 .bind(event.resource_id)
331 .bind(event.actor_subject)
332 .bind(event.correlation_id)
333 .bind(event.occurred_at)
334 .bind(metadata)
335 .execute(&self.pool)
336 .await
337 .map_err(|error| AuditError::Append(error.to_string()))?;
338 Ok(())
339 }
340}
341
342pub async fn migrate_plugin_storage(pool: &SqlitePool) -> Result<(), sqlx::migrate::MigrateError> {
343 let mut migrator = sqlx::migrate!("migrations/plugins");
344 migrator.dangerous_set_table_name("_minco_plugin_storage_migrations");
345 migrator.run(pool).await
346}
347
348fn session_store_error(error: impl std::fmt::Display) -> SessionError {
349 SessionError::Store(error.to_string())
350}
351
352fn idempotency_store_error(error: impl std::fmt::Display) -> IdempotencyError {
353 IdempotencyError::Store(error.to_string())
354}
355
356fn is_unique_violation(error: &sqlx::Error) -> bool {
357 error
358 .as_database_error()
359 .is_some_and(sqlx::error::DatabaseError::is_unique_violation)
360}
361
362#[cfg(test)]
363mod tests {
364 use super::*;
365 use minco_plugin_audit::AuditEvent;
366 use minco_plugin_idempotency::RequestFingerprint;
367 use minco_plugin_sessions::{CreateSession, SessionService};
368 use std::{collections::BTreeMap, path::PathBuf, sync::Arc};
369
370 struct TestDatabasePath(PathBuf);
371
372 impl Drop for TestDatabasePath {
373 fn drop(&mut self) {
374 for suffix in ["", "-shm", "-wal"] {
375 let path = PathBuf::from(format!("{}{suffix}", self.0.display()));
376 let _ = std::fs::remove_file(path);
377 }
378 }
379 }
380
381 async fn pool() -> SqlitePool {
382 let pool = crate::connect(&crate::SqlitePoolConfig::memory())
383 .await
384 .unwrap();
385 migrate_plugin_storage(&pool).await.unwrap();
386 pool
387 }
388
389 #[tokio::test]
390 async fn persistent_sessions_resolve_and_revoke() {
391 let pool = pool().await;
392 let migration_count: i64 =
393 sqlx::query_scalar("SELECT COUNT(*) FROM _minco_plugin_storage_migrations")
394 .fetch_one(&pool)
395 .await
396 .unwrap();
397 assert_eq!(migration_count, 1);
398 let service = SessionService::new(Arc::new(SqliteSessionStore::new(pool)));
399 let issued = service
400 .issue(CreateSession {
401 subject: "subject-1".into(),
402 ttl: TimeDelta::minutes(5),
403 attributes: BTreeMap::new(),
404 })
405 .await
406 .unwrap();
407 assert_eq!(
408 service.resolve(&issued.token).await.unwrap().subject,
409 "subject-1"
410 );
411 assert!(service.revoke(issued.session.id).await.unwrap());
412 assert!(matches!(
413 service.resolve(&issued.token).await,
414 Err(SessionError::Unauthenticated)
415 ));
416 }
417
418 #[tokio::test]
419 async fn idempotency_leases_replay_and_reject_stale_completion() {
420 let store = SqliteIdempotencyStore::new(pool().await);
421 let key = IdempotencyKey::parse("sqlite-request").unwrap();
422 let fingerprint =
423 RequestFingerprint::from_serializable(&serde_json::json!({"request": 1})).unwrap();
424 let now = Utc::now();
425 let BeginOutcome::Started(stale_lease) = store
426 .begin(
427 key.clone(),
428 fingerprint.clone(),
429 now - TimeDelta::minutes(10),
430 TimeDelta::minutes(5),
431 )
432 .await
433 .unwrap()
434 else {
435 panic!("expected a lease");
436 };
437 let BeginOutcome::Started(current_lease) = store
438 .begin(key.clone(), fingerprint.clone(), now, TimeDelta::minutes(5))
439 .await
440 .unwrap()
441 else {
442 panic!("expected replacement lease");
443 };
444 assert!(matches!(
445 store
446 .complete(stale_lease, serde_json::json!({"status": 409}), now)
447 .await,
448 Err(IdempotencyError::InvalidLease)
449 ));
450 store
451 .complete(current_lease, serde_json::json!({"status": 201}), now)
452 .await
453 .unwrap();
454 let BeginOutcome::Replay(record) = store
455 .begin(key, fingerprint, now, TimeDelta::minutes(5))
456 .await
457 .unwrap()
458 else {
459 panic!("expected completed response replay");
460 };
461 assert_eq!(record.response, serde_json::json!({"status": 201}));
462 }
463
464 #[tokio::test]
465 async fn file_backed_idempotency_serializes_concurrent_begin() {
466 let database_path = TestDatabasePath(
467 std::env::temp_dir().join(format!("minco-idempotency-{}.sqlite", Uuid::now_v7())),
468 );
469 let mut config = crate::SqlitePoolConfig::file(&database_path.0);
470 config.max_connections = 2;
471 config.acquire_timeout_seconds = 2;
472 let pool = crate::connect(&config).await.unwrap();
473 migrate_plugin_storage(&pool).await.unwrap();
474
475 let store = SqliteIdempotencyStore::new(pool);
476 let key = IdempotencyKey::parse("sqlite-concurrent-request").unwrap();
477 let fingerprint =
478 RequestFingerprint::from_serializable(&serde_json::json!({"request": 2})).unwrap();
479 let now = Utc::now();
480 let (first, second) = tokio::join!(
481 store.begin(key.clone(), fingerprint.clone(), now, TimeDelta::minutes(5),),
482 store.begin(key, fingerprint, now, TimeDelta::minutes(5)),
483 );
484 let outcomes = [first.unwrap(), second.unwrap()];
485 assert_eq!(
486 outcomes
487 .iter()
488 .filter(|outcome| matches!(outcome, BeginOutcome::Started(_)))
489 .count(),
490 1
491 );
492 assert_eq!(
493 outcomes
494 .iter()
495 .filter(|outcome| matches!(outcome, BeginOutcome::InProgress { .. }))
496 .count(),
497 1
498 );
499 }
500
501 #[tokio::test]
502 async fn audit_is_append_only_with_database_ordering() {
503 let pool = pool().await;
504 let sink = SqliteAuditSink::new(pool.clone());
505 sink.append(AuditEvent::new(
506 "feedback.created",
507 "feedback",
508 "one",
509 Uuid::now_v7(),
510 ))
511 .await
512 .unwrap();
513 let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM minco_audit")
514 .fetch_one(&pool)
515 .await
516 .unwrap();
517 assert_eq!(count, 1);
518 }
519}