1use crate::audit_chain::{
2 self, AuditEventKind, AuditKey, ChainContent, ChainIdentity, ChainRow, EventContent, EventRow,
3 VerifyOutcome, CURRENT_KEY_ID,
4};
5use crate::audit_watermark::emit_chain_tip_watermark;
6use crate::auth::CallerPrincipal;
7use rusqlite::{params, Connection, TransactionBehavior};
8use serde::{de::DeserializeOwned, Deserialize, Serialize};
9use std::path::{Path, PathBuf};
10use std::sync::Arc;
11use subtle::ConstantTimeEq;
12use sysknife_types::{CallerRole, JobState, PreviewEnvelope, RiskLevel, TransactionRecord};
13use uuid::Uuid;
14
15pub(crate) const APPROVAL_RECEIPT_TTL_MINUTES: i64 = 15;
26
27#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
35pub struct JobHistoryEntry {
36 pub transaction_id: String,
37 pub action_name: String,
38 pub risk_level: RiskLevel,
39 pub status: JobState,
40 pub summary: String,
41 pub created_at: String,
42}
43
44#[derive(Clone, Debug, PartialEq, Eq)]
50pub struct NewTransaction {
51 pub request_id: String,
52 pub request_hash: String,
53 pub action_name: String,
54 pub risk_level: RiskLevel,
55 pub summary: String,
75 pub warnings: Vec<String>,
76 pub caller_role: CallerRole,
85 pub caller_principal: CallerPrincipal,
93}
94
95struct SqliteMigration {
104 version: i64,
105 name: &'static str,
106 sql: &'static str,
107}
108
109const SQLITE_MIGRATIONS: &[SqliteMigration] = &[
127 SqliteMigration {
128 version: 1,
129 name: "initial_audit_schema",
130 sql: r#"
131 CREATE TABLE IF NOT EXISTS transactions (
132 transaction_id TEXT PRIMARY KEY,
133 request_id TEXT NOT NULL,
134 request_hash TEXT NOT NULL,
135 action_name TEXT NOT NULL,
136 risk_level TEXT NOT NULL,
137 status TEXT NOT NULL,
138 approval_id TEXT,
139 summary TEXT NOT NULL,
140 warnings_json TEXT NOT NULL,
141 created_at TEXT NOT NULL DEFAULT (datetime('now')),
142 seq INTEGER NOT NULL UNIQUE,
143 key_id TEXT NOT NULL,
144 chain_hash TEXT NOT NULL,
145 prev_chain_hash TEXT NOT NULL DEFAULT ''
146 );
147
148 CREATE TABLE IF NOT EXISTS transaction_previews (
149 transaction_id TEXT PRIMARY KEY,
150 preview_json TEXT NOT NULL
151 );
152
153 CREATE TABLE IF NOT EXISTS transaction_approvals (
154 transaction_id TEXT PRIMARY KEY,
155 receipt_digest TEXT NOT NULL,
156 approved_at TEXT NOT NULL DEFAULT (datetime('now')),
157 consumed_at TEXT
158 );
159
160 CREATE INDEX IF NOT EXISTS transactions_seq_idx ON transactions(seq);
161 "#,
162 },
163 SqliteMigration {
172 version: 2,
173 name: "caller_identity_and_approval_events",
174 sql: r#"
175 ALTER TABLE transactions ADD COLUMN chain_version INTEGER NOT NULL DEFAULT 1;
176 ALTER TABLE transactions ADD COLUMN caller_role TEXT;
177 ALTER TABLE transactions ADD COLUMN event_tip TEXT;
178
179 CREATE TABLE IF NOT EXISTS audit_events (
180 seq INTEGER PRIMARY KEY,
181 key_id TEXT NOT NULL,
182 kind TEXT NOT NULL,
183 transaction_id TEXT NOT NULL,
184 receipt_digest TEXT NOT NULL,
185 created_at TEXT NOT NULL,
186 chain_hash TEXT NOT NULL,
187 prev_chain_hash TEXT NOT NULL DEFAULT ''
188 );
189
190 CREATE INDEX IF NOT EXISTS audit_events_transaction_idx
191 ON audit_events(transaction_id);
192 "#,
193 },
194 SqliteMigration {
199 version: 3,
200 name: "caller_principal",
201 sql: r#"
202 ALTER TABLE transactions ADD COLUMN caller_principal TEXT;
203 "#,
204 },
205];
206
207const CHAIN_ROW_COLUMNS: &str = "seq, key_id, transaction_id, request_id, request_hash, \
214 action_name, risk_level, summary, approval_id, warnings_json, \
215 created_at, prev_chain_hash, chain_hash, chain_version, caller_role, event_tip, \
216 caller_principal";
217
218fn chain_row_from_sqlite(row: &rusqlite::Row<'_>) -> rusqlite::Result<ChainRow> {
219 Ok(ChainRow {
220 seq: row.get::<_, i64>(0)? as u64,
221 key_id: row.get(1)?,
222 transaction_id: row.get(2)?,
223 request_id: row.get(3)?,
224 request_hash: row.get(4)?,
225 action_name: row.get(5)?,
226 risk_level: deserialize_field(&row.get::<_, String>(6)?).map_err(|e| {
227 rusqlite::Error::FromSqlConversionFailure(6, rusqlite::types::Type::Text, Box::new(e))
228 })?,
229 summary: row.get(7)?,
230 approval_id: row.get(8)?,
231 warnings_json: row.get(9)?,
232 created_at: row.get(10)?,
233 prev_chain_hash: row.get(11)?,
234 chain_hash: row.get(12)?,
235 chain_version: row.get::<_, i64>(13)? as u32,
236 caller_role: row.get(14)?,
237 event_tip: row.get(15)?,
238 caller_principal: row.get(16)?,
239 })
240}
241
242fn event_row_from_sqlite(row: &rusqlite::Row<'_>) -> rusqlite::Result<EventRow> {
243 Ok(EventRow {
244 seq: row.get::<_, i64>(0)? as u64,
245 key_id: row.get(1)?,
246 kind: row.get(2)?,
247 transaction_id: row.get(3)?,
248 receipt_digest: row.get(4)?,
249 created_at: row.get(5)?,
250 prev_chain_hash: row.get(6)?,
251 chain_hash: row.get(7)?,
252 })
253}
254
255fn query_optional<T>(
261 conn: &Connection,
262 sql: &str,
263 params: impl rusqlite::Params,
264 f: impl FnOnce(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
265) -> rusqlite::Result<Option<T>> {
266 match conn.query_row(sql, params, f) {
267 Ok(v) => Ok(Some(v)),
268 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
269 Err(e) => Err(e),
270 }
271}
272
273#[derive(Clone, Debug)]
274pub struct TransactionStore {
275 path: PathBuf,
276 audit_key: Option<Arc<AuditKey>>,
279}
280
281#[derive(Clone, Debug, PartialEq)]
282pub struct RecordedPreviewedTransaction {
283 pub transaction: TransactionRecord,
284 pub preview: PreviewEnvelope,
285}
286
287struct InsertedTransaction {
288 record: TransactionRecord,
289 seq: u64,
290 chain_hash: String,
291}
292
293#[derive(Debug, thiserror::Error)]
294pub enum TransactionStoreError {
295 #[error("io error: {0}")]
296 Io(#[from] std::io::Error),
297
298 #[error("sqlite error: {0}")]
299 Sqlite(#[from] rusqlite::Error),
300
301 #[error("json error: {0}")]
302 Json(#[from] serde_json::Error),
303
304 #[error("database invariant violation: {0}")]
305 DatabaseInvariant(String),
306
307 #[error("transaction not found: {0}")]
308 NotFound(String),
309
310 #[error("invalid transition from {from:?} to {to:?}")]
311 InvalidTransition { from: JobState, to: JobState },
312
313 #[error("audit chain misconfiguration: {0}")]
314 AuditChainMissing(&'static str),
315
316 #[error("transaction {0} status changed concurrently; retry")]
324 ConcurrentStatusChange(String),
325}
326
327impl TransactionStore {
328 pub fn open_read_only(path: impl AsRef<Path>) -> Result<Self, TransactionStoreError> {
332 let path = path.as_ref().to_path_buf();
333 if let Some(parent) = path.parent() {
334 ensure_private_dir(parent)?;
335 }
336
337 let store = Self {
338 path,
339 audit_key: None,
340 };
341 store.initialize()?;
342 Ok(store)
343 }
344
345 pub fn open(path: impl AsRef<Path>) -> Result<Self, TransactionStoreError> {
352 let db_path = path.as_ref();
353 let key_path = std::env::var("SYSKNIFE_AUDIT_KEY_PATH")
354 .map(PathBuf::from)
355 .unwrap_or_else(|_| {
356 db_path
357 .parent()
358 .unwrap_or_else(|| Path::new("."))
359 .join("audit-key")
360 });
361 let key = AuditKey::load_or_generate(&key_path).map_err(|e| {
362 TransactionStoreError::Io(std::io::Error::new(
363 std::io::ErrorKind::InvalidData,
364 format!("audit key load failed: {e}"),
365 ))
366 })?;
367 Self::open_with_key(path, Arc::new(key))
368 }
369
370 pub fn open_with_key(
373 path: impl AsRef<Path>,
374 audit_key: Arc<AuditKey>,
375 ) -> Result<Self, TransactionStoreError> {
376 let path = path.as_ref().to_path_buf();
377 if let Some(parent) = path.parent() {
378 ensure_private_dir(parent)?;
379 }
380
381 let store = Self {
382 path,
383 audit_key: Some(audit_key),
384 };
385 store.initialize()?;
386 restrict_db_file_mode(&store.path);
387 Ok(store)
388 }
389
390 fn record_with(
399 &self,
400 transaction: NewTransaction,
401 in_tx: impl FnOnce(&Connection, &InsertedTransaction) -> Result<(), TransactionStoreError>,
402 ) -> Result<InsertedTransaction, TransactionStoreError> {
403 let key = self
404 .audit_key
405 .as_ref()
406 .ok_or(TransactionStoreError::AuditChainMissing(
407 "this TransactionStore was opened read-only; cannot record",
408 ))?;
409 let mut conn = self.connection()?;
410 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
415 let transaction_id = Uuid::new_v4().to_string();
416 let inserted = Self::insert_transaction(&tx, key, &transaction_id, transaction)?;
417 in_tx(&tx, &inserted)?;
418 tx.commit()?;
419 emit_chain_tip_watermark(inserted.seq, &inserted.chain_hash);
420 Ok(inserted)
421 }
422
423 pub fn record(
424 &self,
425 transaction: NewTransaction,
426 ) -> Result<TransactionRecord, TransactionStoreError> {
427 let inserted = self.record_with(transaction, |_, _| Ok(()))?;
428 Ok(inserted.record)
429 }
430
431 pub fn record_previewed(
432 &self,
433 transaction: NewTransaction,
434 preview: PreviewEnvelope,
435 ) -> Result<RecordedPreviewedTransaction, TransactionStoreError> {
436 let inserted = self.record_with(transaction, |conn, inserted| {
437 Self::insert_preview(conn, &inserted.record.transaction_id, &preview)
438 })?;
439 Ok(RecordedPreviewedTransaction {
440 transaction: inserted.record,
441 preview,
442 })
443 }
444
445 pub fn get(
446 &self,
447 transaction_id: &str,
448 ) -> Result<Option<TransactionRecord>, TransactionStoreError> {
449 let conn = self.connection()?;
450 let mut stmt = conn.prepare(
451 "SELECT
452 transaction_id,
453 request_id,
454 request_hash,
455 action_name,
456 risk_level,
457 status,
458 approval_id,
459 summary,
460 warnings_json
461 FROM transactions
462 WHERE transaction_id = ?1",
463 )?;
464 let mut rows = stmt.query(params![transaction_id])?;
465 if let Some(row) = rows.next()? {
466 Ok(Some(row_to_record(row)?))
467 } else {
468 Ok(None)
469 }
470 }
471
472 pub fn get_preview(
473 &self,
474 transaction_id: &str,
475 ) -> Result<Option<PreviewEnvelope>, TransactionStoreError> {
476 let conn = self.connection()?;
477 let mut stmt = conn.prepare(
478 "SELECT preview_json
479 FROM transaction_previews
480 WHERE transaction_id = ?1",
481 )?;
482 let mut rows = stmt.query(params![transaction_id])?;
483 if let Some(row) = rows.next()? {
484 let preview_json: String = row.get(0)?;
485 Ok(Some(serde_json::from_str(&preview_json)?))
486 } else {
487 Ok(None)
488 }
489 }
490
491 pub fn update_status(
492 &self,
493 transaction_id: &str,
494 new_status: JobState,
495 ) -> Result<(), TransactionStoreError> {
496 let conn = self.connection()?;
497
498 let current_status: String = conn
500 .query_row(
501 "SELECT status FROM transactions WHERE transaction_id = ?1",
502 params![transaction_id],
503 |row| row.get(0),
504 )
505 .map_err(|e| match e {
506 rusqlite::Error::QueryReturnedNoRows => {
507 TransactionStoreError::NotFound(transaction_id.to_string())
508 }
509 other => TransactionStoreError::Sqlite(other),
510 })?;
511
512 let current: JobState = deserialize_field(¤t_status)?;
513 if !crate::jobs::allowed_transition(¤t, &new_status) {
514 return Err(TransactionStoreError::InvalidTransition {
515 from: current,
516 to: new_status,
517 });
518 }
519
520 let rows_affected = conn.execute(
532 "UPDATE transactions SET status = ?1 WHERE transaction_id = ?2 AND status = ?3",
533 params![
534 serialize_field(&new_status)?,
535 transaction_id,
536 current_status
537 ],
538 )?;
539 if rows_affected == 0 {
540 return Err(TransactionStoreError::ConcurrentStatusChange(
541 transaction_id.to_string(),
542 ));
543 }
544 Ok(())
545 }
546
547 pub fn approve_transaction(
549 &self,
550 transaction_id: &str,
551 ) -> Result<Option<String>, TransactionStoreError> {
552 let key = self
553 .audit_key
554 .as_ref()
555 .ok_or(TransactionStoreError::AuditChainMissing(
556 "this TransactionStore was opened read-only; cannot approve",
557 ))?;
558 let Some(record) = self.get(transaction_id)? else {
559 return Ok(None);
560 };
561 let receipt = key.approval_receipt(transaction_id, &record.request_hash);
562 let receipt_digest = audit_chain::approval_receipt_digest(&receipt);
563 let Some(committed_digest) = record.approval_id.as_deref() else {
564 return Err(TransactionStoreError::DatabaseInvariant(format!(
565 "transaction {transaction_id} has no signed approval commitment"
566 )));
567 };
568 if !bool::from(receipt_digest.as_bytes().ct_eq(committed_digest.as_bytes())) {
569 return Err(TransactionStoreError::DatabaseInvariant(format!(
570 "transaction {transaction_id} approval commitment does not match its signed preview"
571 )));
572 }
573
574 let mut conn = self.connection()?;
577 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
578 let queued_json = serialize_field(&JobState::Queued)?;
579 let rows_affected = tx.execute(
580 &format!(
581 "INSERT INTO transaction_approvals (transaction_id, receipt_digest) \
582 SELECT transaction_id, ?1 FROM transactions \
583 WHERE transaction_id = ?2 \
584 AND status = ?3 \
585 AND julianday(created_at) > julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes') \
586 AND NOT EXISTS ( \
587 SELECT 1 FROM transaction_approvals WHERE transaction_id = ?2 \
588 )"
589 ),
590 params![receipt_digest, transaction_id, queued_json],
591 )?;
592 if rows_affected > 0 {
593 Self::append_event(
594 &tx,
595 key,
596 AuditEventKind::ApprovalGranted,
597 transaction_id,
598 &receipt_digest,
599 )?;
600 }
601 tx.commit()?;
602 Ok((rows_affected > 0).then_some(receipt))
603 }
604
605 pub fn revoke_unconsumed_approval(
608 &self,
609 transaction_id: &str,
610 ) -> Result<bool, TransactionStoreError> {
611 let key = self
612 .audit_key
613 .as_ref()
614 .ok_or(TransactionStoreError::AuditChainMissing(
615 "this TransactionStore was opened read-only; cannot revoke",
616 ))?;
617 let mut conn = self.connection()?;
618 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
619 let revoked = Self::revoke_unconsumed_approval_in_tx(&tx, key, transaction_id)?;
620 tx.commit()?;
621 Ok(revoked)
622 }
623
624 fn revoke_unconsumed_approval_in_tx(
625 conn: &Connection,
626 key: &AuditKey,
627 transaction_id: &str,
628 ) -> Result<bool, TransactionStoreError> {
629 let digest: Option<String> = query_optional(
632 conn,
633 "SELECT receipt_digest FROM transaction_approvals \
634 WHERE transaction_id = ?1 AND consumed_at IS NULL",
635 params![transaction_id],
636 |row| row.get(0),
637 )?;
638 let rows_affected = conn.execute(
639 "DELETE FROM transaction_approvals \
640 WHERE transaction_id = ?1 AND consumed_at IS NULL",
641 params![transaction_id],
642 )?;
643 if rows_affected > 0 {
644 let digest = digest.ok_or_else(|| {
645 TransactionStoreError::DatabaseInvariant(format!(
646 "revoked an approval for {transaction_id} that had no receipt digest"
647 ))
648 })?;
649 Self::append_event(
650 conn,
651 key,
652 AuditEventKind::ApprovalRevoked,
653 transaction_id,
654 &digest,
655 )?;
656 }
657 Ok(rows_affected > 0)
658 }
659
660 pub fn claim_approved_for_execution(
662 &self,
663 transaction_id: &str,
664 receipt_digest: &str,
665 ) -> Result<bool, TransactionStoreError> {
666 let key = self
667 .audit_key
668 .as_ref()
669 .ok_or(TransactionStoreError::AuditChainMissing(
670 "this TransactionStore was opened read-only; cannot claim",
671 ))?;
672 let mut conn = self.connection()?;
673 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
674 let queued_json = serialize_field(&JobState::Queued)?;
675 let running_json = serialize_field(&JobState::Running)?;
676 let rows_affected = tx.execute(
677 &format!(
678 "UPDATE transactions SET status = ?1 \
679 WHERE transaction_id = ?2 \
680 AND status = ?3 \
681 AND julianday(created_at) > julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes') \
682 AND EXISTS ( \
683 SELECT 1 FROM transaction_approvals \
684 WHERE transaction_id = ?2 \
685 AND receipt_digest = ?4 \
686 AND consumed_at IS NULL \
687 )"
688 ),
689 params![running_json, transaction_id, queued_json, receipt_digest],
690 )?;
691 if rows_affected > 0 {
692 tx.execute(
693 "UPDATE transaction_approvals \
694 SET consumed_at = datetime('now') \
695 WHERE transaction_id = ?1 AND consumed_at IS NULL",
696 params![transaction_id],
697 )?;
698 Self::append_event(
699 &tx,
700 key,
701 AuditEventKind::ApprovalConsumed,
702 transaction_id,
703 receipt_digest,
704 )?;
705 }
706 tx.commit()?;
707 Ok(rows_affected > 0)
708 }
709
710 pub fn cleanup_stale_queued(&self) -> Result<u64, TransactionStoreError> {
725 let key = self
726 .audit_key
727 .as_ref()
728 .ok_or(TransactionStoreError::AuditChainMissing(
729 "this TransactionStore was opened read-only; cannot clean up",
730 ))?;
731 let mut conn = self.connection()?;
732 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
733 let canceled_json = serialize_field(&JobState::Canceled)?;
734 let queued_json = serialize_field(&JobState::Queued)?;
735 let stale_ids: Vec<String> = {
736 let mut statement = tx.prepare(&format!(
737 "SELECT transaction_id FROM transactions \
738 WHERE status = ?1 \
739 AND julianday(created_at) <= julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes')"
740 ))?;
741 let stale_ids = statement
742 .query_map(params![queued_json], |row| row.get(0))?
743 .collect::<Result<_, _>>()?;
744 stale_ids
745 };
746 let mut canceled = 0;
747 for transaction_id in stale_ids {
748 let rows_affected = tx.execute(
749 "UPDATE transactions SET status = ?1 \
750 WHERE transaction_id = ?2 AND status = ?3",
751 params![canceled_json, transaction_id, queued_json],
752 )?;
753 if rows_affected > 0 {
754 Self::revoke_unconsumed_approval_in_tx(&tx, key, &transaction_id)?;
755 canceled += rows_affected;
756 }
757 }
758 tx.commit()?;
759 Ok(canceled as u64)
760 }
761
762 pub fn cancel_queued(&self, transaction_id: &str) -> Result<bool, TransactionStoreError> {
771 let key = self
772 .audit_key
773 .as_ref()
774 .ok_or(TransactionStoreError::AuditChainMissing(
775 "this TransactionStore was opened read-only; cannot cancel",
776 ))?;
777 let mut conn = self.connection()?;
778 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
779 let canceled_json = serialize_field(&JobState::Canceled)?;
780 let queued_json = serialize_field(&JobState::Queued)?;
781 let rows_affected = tx.execute(
782 "UPDATE transactions SET status = ?1 \
783 WHERE transaction_id = ?2 AND status = ?3",
784 params![canceled_json, transaction_id, queued_json],
785 )?;
786 if rows_affected > 0 {
787 Self::revoke_unconsumed_approval_in_tx(&tx, key, transaction_id)?;
788 }
789 tx.commit()?;
790 Ok(rows_affected > 0)
791 }
792
793 pub fn list_transactions(
802 &self,
803 limit: u32,
804 status_filter: Option<&str>,
805 action_filter: Option<&str>,
806 since_hours: Option<u32>,
807 ) -> Result<Vec<TransactionRecord>, TransactionStoreError> {
808 let conn = self.connection()?;
809 let (filter_sql, param_values) =
810 Self::build_history_filter(limit, status_filter, action_filter, since_hours)?;
811 let sql = format!(
812 "SELECT transaction_id, request_id, request_hash, action_name, \
813 risk_level, status, approval_id, summary, warnings_json \
814 FROM transactions WHERE 1=1{filter_sql}"
815 );
816 let params_ref: Vec<&dyn rusqlite::types::ToSql> =
817 param_values.iter().map(|b| b.as_ref()).collect();
818
819 let mut stmt = conn.prepare(&sql)?;
820 let rows = stmt.query_map(params_ref.as_slice(), |row| Ok(row_to_record(row)))?;
821
822 let mut results = Vec::new();
823 for row in rows {
824 results.push(row??);
825 }
826 Ok(results)
827 }
828
829 pub fn list_history(
834 &self,
835 limit: u32,
836 status_filter: Option<&str>,
837 action_filter: Option<&str>,
838 since_hours: Option<u32>,
839 ) -> Result<Vec<JobHistoryEntry>, TransactionStoreError> {
840 let conn = self.connection()?;
841 let (filter_sql, param_values) =
842 Self::build_history_filter(limit, status_filter, action_filter, since_hours)?;
843 let sql = format!(
844 "SELECT transaction_id, action_name, risk_level, status, summary, created_at \
845 FROM transactions WHERE 1=1{filter_sql}"
846 );
847 let params_ref: Vec<&dyn rusqlite::types::ToSql> =
848 param_values.iter().map(|b| b.as_ref()).collect();
849
850 let mut stmt = conn.prepare(&sql)?;
851 let rows = stmt.query_map(params_ref.as_slice(), |row| {
852 Ok((|| {
853 Ok::<JobHistoryEntry, TransactionStoreError>(JobHistoryEntry {
854 transaction_id: row.get(0)?,
855 action_name: row.get(1)?,
856 risk_level: deserialize_field(&row.get::<_, String>(2)?)?,
857 status: deserialize_field(&row.get::<_, String>(3)?)?,
858 summary: row.get(4)?,
859 created_at: row.get(5)?,
860 })
861 })())
862 })?;
863
864 let mut results = Vec::new();
865 for row in rows {
866 results.push(row??);
867 }
868 Ok(results)
869 }
870
871 fn build_history_filter(
876 limit: u32,
877 status_filter: Option<&str>,
878 action_filter: Option<&str>,
879 since_hours: Option<u32>,
880 ) -> Result<(String, Vec<Box<dyn rusqlite::types::ToSql>>), TransactionStoreError> {
881 let mut sql = String::new();
882 let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
883
884 if let Some(status) = status_filter {
885 let validated: JobState = deserialize_field(&format!("\"{status}\""))?;
889 let status_json = serialize_field(&validated)?;
890 sql.push_str(" AND status = ?");
891 param_values.push(Box::new(status_json));
892 }
893
894 if let Some(action) = action_filter {
895 sql.push_str(" AND action_name = ?");
896 param_values.push(Box::new(action.to_string()));
897 }
898
899 if let Some(hours) = since_hours {
900 sql.push_str(" AND julianday(created_at) > julianday('now', '-' || ? || ' hours')");
901 param_values.push(Box::new(hours));
902 }
903
904 sql.push_str(" ORDER BY seq DESC LIMIT ?");
905 param_values.push(Box::new(limit.min(100)));
906
907 Ok((sql, param_values))
908 }
909
910 fn connection(&self) -> Result<Connection, TransactionStoreError> {
911 let conn = Connection::open(&self.path)?;
912 conn.pragma_update(None, "journal_mode", "WAL")?;
923 conn.pragma_update(None, "synchronous", "NORMAL")?;
924 conn.pragma_update(None, "foreign_keys", "ON")?;
925 conn.busy_timeout(std::time::Duration::from_secs(5))?;
926 Ok(conn)
927 }
928
929 fn initialize(&self) -> Result<(), TransactionStoreError> {
930 let mut conn = self.connection()?;
931 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
932 tx.execute_batch(
933 "CREATE TABLE IF NOT EXISTS schema_migrations (\
934 version INTEGER PRIMARY KEY,\
935 name TEXT NOT NULL,\
936 applied_at TEXT NOT NULL DEFAULT (datetime('now'))\
937 );",
938 )?;
939 let current: i64 = tx.query_row(
940 "SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
941 [],
942 |row| row.get(0),
943 )?;
944 let latest = SQLITE_MIGRATIONS.last().map_or(0, |m| m.version);
945 if current > latest {
946 return Err(TransactionStoreError::DatabaseInvariant(format!(
947 "sqlite schema version {current} is newer than this binary supports ({latest})"
948 )));
949 }
950 for migration in SQLITE_MIGRATIONS.iter().filter(|m| m.version > current) {
951 tx.execute_batch(migration.sql)?;
952 tx.execute(
953 "INSERT INTO schema_migrations (version, name) VALUES (?1, ?2)",
954 params![migration.version, migration.name],
955 )?;
956 }
957 tx.commit()?;
958 Ok(())
959 }
960
961 pub fn fetch_chain_rows(&self) -> Result<Vec<ChainRow>, TransactionStoreError> {
963 let conn = self.connection()?;
964 let mut stmt = conn.prepare(&format!(
965 "SELECT {CHAIN_ROW_COLUMNS} FROM transactions ORDER BY seq ASC"
966 ))?;
967 let rows = stmt.query_map([], chain_row_from_sqlite)?;
968 let mut out = Vec::new();
969 for row in rows {
970 out.push(row?);
971 }
972 Ok(out)
973 }
974
975 pub fn fetch_event_rows(&self) -> Result<Vec<EventRow>, TransactionStoreError> {
977 let conn = self.connection()?;
978 let mut stmt = conn.prepare(
979 "SELECT seq, key_id, kind, transaction_id, receipt_digest, \
980 created_at, prev_chain_hash, chain_hash \
981 FROM audit_events ORDER BY seq ASC",
982 )?;
983 let rows = stmt.query_map([], event_row_from_sqlite)?;
984 let mut out = Vec::new();
985 for row in rows {
986 out.push(row?);
987 }
988 Ok(out)
989 }
990
991 pub fn verify_audit_chain(
993 &self,
994 key: &AuditKey,
995 ) -> Result<VerifyOutcome, TransactionStoreError> {
996 let rows = self.fetch_chain_rows()?;
997 Ok(audit_chain::verify_chain(key, &rows))
998 }
999
1000 pub fn verify_event_chain(
1002 &self,
1003 key: &AuditKey,
1004 ) -> Result<VerifyOutcome, TransactionStoreError> {
1005 let rows = self.fetch_event_rows()?;
1006 Ok(audit_chain::verify_event_chain(key, &rows))
1007 }
1008
1009 pub fn verify_event_binding(
1012 &self,
1013 ) -> Result<audit_chain::BindingOutcome, TransactionStoreError> {
1014 let tx_rows = self.fetch_chain_rows()?;
1015 let event_rows = self.fetch_event_rows()?;
1016 Ok(audit_chain::verify_event_binding(&tx_rows, &event_rows))
1017 }
1018
1019 pub fn verify_audit_chain_with_pubkey(
1022 &self,
1023 verifying_key_hex: &str,
1024 ) -> Result<VerifyOutcome, TransactionStoreError> {
1025 let rows = self.fetch_chain_rows()?;
1026 Ok(audit_chain::verify_chain_with_pubkey(
1027 verifying_key_hex,
1028 &rows,
1029 ))
1030 }
1031
1032 pub fn fetch_chain_row(
1035 &self,
1036 transaction_id: &str,
1037 ) -> Result<Option<ChainRow>, TransactionStoreError> {
1038 let conn = self.connection()?;
1039 let mut stmt = conn.prepare(&format!(
1040 "SELECT {CHAIN_ROW_COLUMNS} FROM transactions WHERE transaction_id = ?1"
1041 ))?;
1042 let mut rows = stmt.query(params![transaction_id])?;
1043 if let Some(row) = rows.next()? {
1044 Ok(Some(chain_row_from_sqlite(row)?))
1045 } else {
1046 Ok(None)
1047 }
1048 }
1049
1050 fn next_seq_and_prev_hash(conn: &Connection) -> Result<(u64, String), TransactionStoreError> {
1054 let prev: Option<(i64, String)> = query_optional(
1055 conn,
1056 "SELECT seq, chain_hash FROM transactions ORDER BY seq DESC LIMIT 1",
1057 [],
1058 |row| Ok((row.get(0)?, row.get(1)?)),
1059 )?;
1060 Ok(match prev {
1061 Some((seq, hash)) => ((seq as u64) + 1, hash),
1062 None => (1, String::new()),
1063 })
1064 }
1065
1066 fn event_chain_tip(conn: &Connection) -> Result<Option<String>, TransactionStoreError> {
1069 Ok(query_optional(
1070 conn,
1071 "SELECT chain_hash FROM audit_events ORDER BY seq DESC LIMIT 1",
1072 [],
1073 |row| row.get(0),
1074 )?)
1075 }
1076
1077 fn append_event(
1082 conn: &Connection,
1083 key: &AuditKey,
1084 kind: AuditEventKind,
1085 transaction_id: &str,
1086 receipt_digest: &str,
1087 ) -> Result<(), TransactionStoreError> {
1088 let prev_chain_hash = Self::event_chain_tip(conn)?.unwrap_or_default();
1089 let seq: i64 = conn.query_row(
1090 "SELECT COALESCE(MAX(seq), 0) + 1 FROM audit_events",
1091 [],
1092 |row| row.get(0),
1093 )?;
1094 let created_at: String =
1095 conn.query_row("SELECT strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", [], |row| {
1096 row.get(0)
1097 })?;
1098 let key_id = CURRENT_KEY_ID.to_string();
1099 let chain_hash = key.event_hash(
1100 &EventContent {
1101 seq: seq as u64,
1102 key_id: &key_id,
1103 kind,
1104 transaction_id,
1105 receipt_digest,
1106 created_at: &created_at,
1107 },
1108 &prev_chain_hash,
1109 );
1110 conn.execute(
1111 "INSERT INTO audit_events (
1112 seq, key_id, kind, transaction_id, receipt_digest,
1113 created_at, chain_hash, prev_chain_hash
1114 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
1115 params![
1116 seq,
1117 key_id,
1118 kind.as_str(),
1119 transaction_id,
1120 receipt_digest,
1121 created_at,
1122 chain_hash,
1123 prev_chain_hash,
1124 ],
1125 )?;
1126 Ok(())
1127 }
1128
1129 fn insert_transaction(
1130 conn: &Connection,
1131 key: &AuditKey,
1132 transaction_id: &str,
1133 transaction: NewTransaction,
1134 ) -> Result<InsertedTransaction, TransactionStoreError> {
1135 let request_id = transaction.request_id;
1136 let request_hash = transaction.request_hash;
1137 let action_name = transaction.action_name;
1138 let risk_level = transaction.risk_level;
1139 let status = JobState::Queued;
1141 let approval_id = Some(key.approval_commitment(transaction_id, &request_hash));
1148 let summary = transaction.summary;
1149 let warnings = transaction.warnings;
1150 let warnings_json = serde_json::to_string(&warnings)?;
1151
1152 let (seq, prev_chain_hash) = Self::next_seq_and_prev_hash(conn)?;
1155
1156 let created_at: String =
1160 conn.query_row("SELECT strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", [], |row| {
1161 row.get(0)
1162 })?;
1163
1164 let key_id = CURRENT_KEY_ID.to_string();
1165 let event_tip = Self::event_chain_tip(conn)?.unwrap_or_default();
1169 let caller_role = transaction.caller_role.as_str();
1170 let caller_principal = transaction.caller_principal.as_signed_str();
1171 let identity = ChainIdentity::V3 {
1175 caller_role,
1176 event_tip: &event_tip,
1177 caller_principal: &caller_principal,
1178 };
1179 let chain_hash = key.chain_hash(
1180 &ChainContent {
1181 seq,
1182 key_id: &key_id,
1183 transaction_id,
1184 request_id: &request_id,
1185 request_hash: &request_hash,
1186 action_name: &action_name,
1187 risk_level,
1188 summary: &summary,
1189 approval_id: approval_id.as_deref(),
1190 warnings_json: &warnings_json,
1191 created_at: &created_at,
1192 identity,
1193 },
1194 &prev_chain_hash,
1195 );
1196
1197 let record = TransactionRecord {
1198 transaction_id: transaction_id.to_string(),
1199 request_id: request_id.clone(),
1200 request_hash: request_hash.clone(),
1201 action_name: action_name.clone(),
1202 risk_level,
1203 status,
1204 approval_id: approval_id.clone(),
1205 summary: summary.clone(),
1206 warnings: warnings.clone(),
1207 };
1208
1209 conn.execute(
1210 "INSERT INTO transactions (
1211 transaction_id,
1212 request_id,
1213 request_hash,
1214 action_name,
1215 risk_level,
1216 status,
1217 approval_id,
1218 summary,
1219 warnings_json,
1220 created_at,
1221 seq,
1222 key_id,
1223 chain_hash,
1224 prev_chain_hash,
1225 chain_version,
1226 caller_role,
1227 event_tip,
1228 caller_principal
1229 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)",
1230 params![
1231 transaction_id,
1232 request_id,
1233 request_hash,
1234 action_name,
1235 serialize_field(&risk_level)?,
1236 serialize_field(&status)?,
1237 approval_id,
1238 summary,
1239 warnings_json,
1240 created_at,
1241 seq as i64,
1242 key_id,
1243 chain_hash,
1244 prev_chain_hash,
1245 identity.version() as i64,
1246 caller_role,
1247 event_tip,
1248 caller_principal,
1249 ],
1250 )?;
1251
1252 Ok(InsertedTransaction {
1253 record,
1254 seq,
1255 chain_hash,
1256 })
1257 }
1258
1259 fn insert_preview(
1260 conn: &Connection,
1261 transaction_id: &str,
1262 preview: &PreviewEnvelope,
1263 ) -> Result<(), TransactionStoreError> {
1264 conn.execute(
1265 "INSERT INTO transaction_previews (transaction_id, preview_json)
1266 VALUES (?1, ?2)",
1267 params![transaction_id, serde_json::to_string(preview)?],
1268 )?;
1269 Ok(())
1270 }
1271}
1272
1273fn ensure_private_dir(dir: &Path) -> std::io::Result<()> {
1280 use std::os::unix::fs::DirBuilderExt;
1281 if dir.exists() {
1282 return Ok(());
1283 }
1284 std::fs::DirBuilder::new()
1285 .recursive(true)
1286 .mode(0o700)
1287 .create(dir)
1288}
1289
1290fn restrict_db_file_mode(db_path: &Path) {
1304 use std::os::unix::fs::PermissionsExt;
1305
1306 let sidecars = [
1310 db_path.to_path_buf(),
1311 with_suffix(db_path, "-wal"),
1312 with_suffix(db_path, "-shm"),
1313 ];
1314 for path in sidecars {
1315 let Ok(metadata) = std::fs::metadata(&path) else {
1316 continue; };
1318 let mode = metadata.permissions().mode() & 0o777;
1319 if mode & 0o077 == 0 {
1320 continue; }
1322 if let Err(err) = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)) {
1323 eprintln!(
1324 "[sysknife-daemon] WARNING: {} is mode {mode:o} and could not be \
1325 narrowed to 0600 ({err}); it is readable by anyone who can reach \
1326 it. The parent directory should be 0700 — check it.",
1327 path.display()
1328 );
1329 }
1330 }
1331}
1332
1333fn with_suffix(path: &Path, suffix: &str) -> PathBuf {
1336 let mut name = path.as_os_str().to_os_string();
1337 name.push(suffix);
1338 PathBuf::from(name)
1339}
1340
1341fn row_to_record(row: &rusqlite::Row) -> Result<TransactionRecord, TransactionStoreError> {
1342 Ok(TransactionRecord {
1343 transaction_id: row.get(0)?,
1344 request_id: row.get(1)?,
1345 request_hash: row.get(2)?,
1346 action_name: row.get(3)?,
1347 risk_level: deserialize_field(&row.get::<_, String>(4)?)?,
1348 status: deserialize_field(&row.get::<_, String>(5)?)?,
1349 approval_id: row.get(6)?,
1350 summary: row.get(7)?,
1351 warnings: serde_json::from_str(&row.get::<_, String>(8)?)?,
1352 })
1353}
1354
1355fn serialize_field<T: Serialize>(value: &T) -> Result<String, serde_json::Error> {
1356 serde_json::to_string(value)
1357}
1358
1359fn deserialize_field<T: DeserializeOwned>(value: &str) -> Result<T, serde_json::Error> {
1360 serde_json::from_str(value)
1361}
1362
1363#[cfg(test)]
1364mod tests {
1365 use super::*;
1366 use crate::audit_chain::CHAIN_VERSION_CURRENT;
1367 use crate::audit_chain::CHAIN_VERSION_LEGACY;
1368 use std::os::unix::fs::PermissionsExt;
1369 use tempfile::tempdir;
1370
1371 fn test_store(path: impl AsRef<Path>) -> TransactionStore {
1375 let key = Arc::new(AuditKey::from_bytes(vec![0x42; 32]));
1376 TransactionStore::open_with_key(path, key).unwrap()
1377 }
1378
1379 fn legacy_v1_database(path: &Path, key: &AuditKey) -> String {
1386 let conn = Connection::open(path).unwrap();
1387 conn.execute_batch(
1388 "CREATE TABLE schema_migrations (
1389 version INTEGER PRIMARY KEY,
1390 name TEXT NOT NULL,
1391 applied_at TEXT NOT NULL DEFAULT (datetime('now'))
1392 );",
1393 )
1394 .unwrap();
1395 conn.execute_batch(SQLITE_MIGRATIONS[0].sql).unwrap();
1396 conn.execute(
1397 "INSERT INTO schema_migrations (version, name) VALUES (1, ?1)",
1398 params![SQLITE_MIGRATIONS[0].name],
1399 )
1400 .unwrap();
1401
1402 let transaction_id = "tx-legacy";
1403 let created_at = "2026-04-24T12:00:00.000Z";
1404 let chain_hash = key.chain_hash(
1405 &ChainContent {
1406 seq: 1,
1407 key_id: CURRENT_KEY_ID,
1408 transaction_id,
1409 request_id: "req-legacy",
1410 request_hash: "hash-legacy",
1411 action_name: "UpdateSystem",
1412 risk_level: RiskLevel::High,
1413 summary: "Upgrade the system",
1414 approval_id: None,
1415 warnings_json: "[]",
1416 created_at,
1417 identity: ChainIdentity::LegacyV1,
1418 },
1419 "",
1420 );
1421 conn.execute(
1422 "INSERT INTO transactions (
1423 transaction_id, request_id, request_hash, action_name, risk_level,
1424 status, approval_id, summary, warnings_json, created_at,
1425 seq, key_id, chain_hash, prev_chain_hash
1426 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8, ?9, 1, ?10, ?11, '')",
1427 params![
1428 transaction_id,
1429 "req-legacy",
1430 "hash-legacy",
1431 "UpdateSystem",
1432 serialize_field(&RiskLevel::High).unwrap(),
1433 serialize_field(&JobState::Queued).unwrap(),
1434 "Upgrade the system",
1435 "[]",
1436 created_at,
1437 CURRENT_KEY_ID,
1438 chain_hash,
1439 ],
1440 )
1441 .unwrap();
1442 transaction_id.to_string()
1443 }
1444
1445 fn v2_database(path: &Path, key: &AuditKey) -> String {
1452 let conn = Connection::open(path).unwrap();
1453 conn.execute_batch(
1454 "CREATE TABLE schema_migrations (
1455 version INTEGER PRIMARY KEY,
1456 name TEXT NOT NULL,
1457 applied_at TEXT NOT NULL DEFAULT (datetime('now'))
1458 );",
1459 )
1460 .unwrap();
1461 for migration in &SQLITE_MIGRATIONS[0..=1] {
1462 conn.execute_batch(migration.sql).unwrap();
1463 conn.execute(
1464 "INSERT INTO schema_migrations (version, name) VALUES (?1, ?2)",
1465 params![migration.version, migration.name],
1466 )
1467 .unwrap();
1468 }
1469
1470 let transaction_id = "tx-v2";
1471 let created_at = "2026-07-01T12:00:00.000Z";
1472 let chain_hash = key.chain_hash(
1473 &ChainContent {
1474 seq: 1,
1475 key_id: CURRENT_KEY_ID,
1476 transaction_id,
1477 request_id: "req-v2",
1478 request_hash: "hash-v2",
1479 action_name: "UpdateSystem",
1480 risk_level: RiskLevel::High,
1481 summary: "Upgrade the system",
1482 approval_id: None,
1483 warnings_json: "[]",
1484 created_at,
1485 identity: ChainIdentity::V2 {
1486 caller_role: "admin",
1487 event_tip: "",
1488 },
1489 },
1490 "",
1491 );
1492 conn.execute(
1493 "INSERT INTO transactions (
1494 transaction_id, request_id, request_hash, action_name, risk_level,
1495 status, approval_id, summary, warnings_json, created_at,
1496 seq, key_id, chain_hash, prev_chain_hash,
1497 chain_version, caller_role, event_tip
1498 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8, ?9, 1, ?10, ?11, '', 2, 'admin', '')",
1499 params![
1500 transaction_id,
1501 "req-v2",
1502 "hash-v2",
1503 "UpdateSystem",
1504 serialize_field(&RiskLevel::High).unwrap(),
1505 serialize_field(&JobState::Queued).unwrap(),
1506 "Upgrade the system",
1507 "[]",
1508 created_at,
1509 CURRENT_KEY_ID,
1510 chain_hash,
1511 ],
1512 )
1513 .unwrap();
1514 transaction_id.to_string()
1515 }
1516
1517 #[test]
1525 fn a_v2_database_keeps_verifying_and_accepts_v3_rows() {
1526 let dir = tempdir().unwrap();
1527 let db_path = dir.path().join("tx.db");
1528 let key = AuditKey::from_bytes(vec![0x42; 32]);
1529 let old_transaction = v2_database(&db_path, &key);
1530
1531 let store = test_store(&db_path);
1532
1533 let version: i64 = store
1534 .connection()
1535 .unwrap()
1536 .query_row("SELECT MAX(version) FROM schema_migrations", [], |row| {
1537 row.get(0)
1538 })
1539 .unwrap();
1540 assert_eq!(version, 3, "opening a v2 database must apply migration 3");
1541
1542 let rows = store.fetch_chain_rows().unwrap();
1543 assert_eq!(rows.len(), 1);
1544 assert_eq!(rows[0].transaction_id, old_transaction);
1545 assert_eq!(rows[0].chain_version, audit_chain::CHAIN_VERSION_V2);
1546 assert_eq!(
1547 rows[0].caller_principal, None,
1548 "migration 3 must not invent a principal for a row signed without one"
1549 );
1550 assert!(matches!(
1551 store.verify_audit_chain(&key).unwrap(),
1552 audit_chain::VerifyOutcome::Intact { rows_checked: 1 }
1553 ));
1554
1555 let mut fresh = queued_transaction();
1557 fresh.caller_principal = CallerPrincipal::Uid(4242);
1558 store.record(fresh).unwrap();
1559
1560 let rows = store.fetch_chain_rows().unwrap();
1561 assert_eq!(rows.len(), 2);
1562 assert_eq!(rows[1].chain_version, audit_chain::CHAIN_VERSION_CURRENT);
1563 assert_eq!(rows[1].caller_principal.as_deref(), Some("uid:4242"));
1564 assert_eq!(
1565 rows[1].prev_chain_hash, rows[0].chain_hash,
1566 "the v3 row must chain onto the v2 row, not start a new chain"
1567 );
1568 assert!(matches!(
1569 store.verify_audit_chain(&key).unwrap(),
1570 audit_chain::VerifyOutcome::Intact { rows_checked: 2 }
1571 ));
1572
1573 let census = audit_chain::AttributionCensus::of(&rows);
1581 assert_eq!(census.rows(), 2);
1582 assert_eq!(census.named(), 1, "only the v3 row names uid:4242");
1583 assert_eq!(
1584 census.not_recorded(),
1585 1,
1586 "the v2 row's encoding never signed a principal"
1587 );
1588 assert_eq!(census.attribution_failed(), 0);
1589 assert_eq!(census.unattested(), 0);
1590 assert_eq!(census.unnamed(), 1);
1591 }
1592
1593 #[test]
1594 fn a_database_written_before_the_migration_still_verifies_after_upgrading() {
1595 let dir = tempdir().unwrap();
1600 let db_path = dir.path().join("tx.db");
1601 let key = AuditKey::from_bytes(vec![0x42; 32]);
1602 legacy_v1_database(&db_path, &key);
1603
1604 let store = test_store(&db_path);
1605 assert_eq!(
1606 store.verify_audit_chain(&key).unwrap(),
1607 VerifyOutcome::Intact { rows_checked: 1 }
1608 );
1609 let rows = store.fetch_chain_rows().unwrap();
1610 assert_eq!(rows[0].chain_version, CHAIN_VERSION_LEGACY);
1611 assert_eq!(rows[0].caller_role, None);
1612 }
1613
1614 #[test]
1615 fn rows_appended_after_the_migration_chain_onto_legacy_rows() {
1616 let dir = tempdir().unwrap();
1620 let db_path = dir.path().join("tx.db");
1621 let key = AuditKey::from_bytes(vec![0x42; 32]);
1622 legacy_v1_database(&db_path, &key);
1623
1624 let store = test_store(&db_path);
1625 store.record(queued_transaction()).unwrap();
1626
1627 let rows = store.fetch_chain_rows().unwrap();
1628 assert_eq!(rows.len(), 2);
1629 assert_eq!(rows[1].prev_chain_hash, rows[0].chain_hash);
1630 assert_eq!(rows[1].chain_version, CHAIN_VERSION_CURRENT);
1631 assert_eq!(
1632 store.verify_audit_chain(&key).unwrap(),
1633 VerifyOutcome::Intact { rows_checked: 2 }
1634 );
1635 }
1636
1637 #[test]
1638 fn a_schema_newer_than_this_binary_is_refused() {
1639 let dir = tempdir().unwrap();
1640 let db_path = dir.path().join("tx.db");
1641 let key = AuditKey::from_bytes(vec![0x42; 32]);
1642 legacy_v1_database(&db_path, &key);
1643 let conn = Connection::open(&db_path).unwrap();
1644 conn.execute(
1645 "INSERT INTO schema_migrations (version, name) VALUES (99, 'from the future')",
1646 [],
1647 )
1648 .unwrap();
1649 drop(conn);
1650
1651 let err = TransactionStore::open_with_key(&db_path, Arc::new(key)).unwrap_err();
1652 assert!(
1653 matches!(err, TransactionStoreError::DatabaseInvariant(ref m) if m.contains("99")),
1654 "expected a refusal naming the unsupported version, got {err:?}"
1655 );
1656 }
1657
1658 #[test]
1661 fn the_signed_row_records_which_role_asked() {
1662 let dir = tempdir().unwrap();
1663 let store = test_store(dir.path().join("tx.db"));
1664 let mut tx = queued_transaction();
1665 tx.caller_role = CallerRole::Admin;
1666 store.record(tx).unwrap();
1667
1668 let rows = store.fetch_chain_rows().unwrap();
1669 assert_eq!(rows[0].caller_role.as_deref(), Some("admin"));
1670 }
1671
1672 #[test]
1673 fn approving_consuming_and_revoking_each_append_a_chained_event() {
1674 let dir = tempdir().unwrap();
1675 let key = AuditKey::from_bytes(vec![0x42; 32]);
1676 let store = test_store(dir.path().join("tx.db"));
1677
1678 let a = store.record(queued_transaction()).unwrap();
1680 let receipt = store
1681 .approve_transaction(&a.transaction_id)
1682 .unwrap()
1683 .expect("queued transaction approves");
1684 let digest = audit_chain::approval_receipt_digest(&receipt);
1685 assert!(store
1686 .claim_approved_for_execution(&a.transaction_id, &digest)
1687 .unwrap());
1688
1689 let b = store.record(queued_transaction()).unwrap();
1691 store.approve_transaction(&b.transaction_id).unwrap();
1692 assert!(store.revoke_unconsumed_approval(&b.transaction_id).unwrap());
1693
1694 let events = store.fetch_event_rows().unwrap();
1695 let kinds: Vec<&str> = events.iter().map(|e| e.kind.as_str()).collect();
1696 assert_eq!(
1697 kinds,
1698 vec![
1699 "approval_granted",
1700 "approval_consumed",
1701 "approval_granted",
1702 "approval_revoked",
1703 ]
1704 );
1705 assert_eq!(
1706 store.verify_event_chain(&key).unwrap(),
1707 VerifyOutcome::Intact { rows_checked: 4 }
1708 );
1709 }
1710
1711 #[test]
1712 fn a_failed_approval_appends_no_event() {
1713 let dir = tempdir().unwrap();
1717 let store = test_store(dir.path().join("tx.db"));
1718 let tx = store.record(queued_transaction()).unwrap();
1719 store.approve_transaction(&tx.transaction_id).unwrap();
1720 store.approve_transaction(&tx.transaction_id).unwrap();
1721 assert_eq!(store.fetch_event_rows().unwrap().len(), 1);
1722
1723 assert!(!store.revoke_unconsumed_approval("no-such-tx").unwrap());
1725 assert_eq!(store.fetch_event_rows().unwrap().len(), 1);
1726 }
1727
1728 #[test]
1729 fn deleting_the_record_that_an_approval_happened_no_longer_goes_unnoticed() {
1730 let dir = tempdir().unwrap();
1736 let db_path = dir.path().join("tx.db");
1737 let key = AuditKey::from_bytes(vec![0x42; 32]);
1738 let store = test_store(&db_path);
1739
1740 let a = store.record(queued_transaction()).unwrap();
1741 store.approve_transaction(&a.transaction_id).unwrap();
1742 store.record(queued_transaction()).unwrap();
1745
1746 assert_eq!(
1747 store.verify_event_chain(&key).unwrap(),
1748 VerifyOutcome::Intact { rows_checked: 1 }
1749 );
1750 assert_eq!(
1751 store.verify_event_binding().unwrap(),
1752 audit_chain::BindingOutcome::Consistent {
1753 bindings_checked: 1
1754 }
1755 );
1756
1757 let conn = Connection::open(&db_path).unwrap();
1759 conn.execute("DELETE FROM transaction_approvals", [])
1760 .unwrap();
1761 conn.execute("DELETE FROM audit_events", []).unwrap();
1762 drop(conn);
1763
1764 assert_eq!(
1768 store.verify_event_chain(&key).unwrap(),
1769 VerifyOutcome::Intact { rows_checked: 0 }
1770 );
1771 match store.verify_event_binding().unwrap() {
1772 audit_chain::BindingOutcome::MissingEvent {
1773 transaction_seq, ..
1774 } => assert_eq!(transaction_seq, 2),
1775 other => panic!("expected MissingEvent, got {other:?}"),
1776 }
1777 }
1778
1779 #[test]
1780 fn a_read_only_store_cannot_revoke_or_claim() {
1781 let dir = tempdir().unwrap();
1784 let db_path = dir.path().join("tx.db");
1785 let store = test_store(&db_path);
1786 store.record(queued_transaction()).unwrap();
1787
1788 let read_only = TransactionStore::open_read_only(&db_path).unwrap();
1789 assert!(matches!(
1790 read_only.revoke_unconsumed_approval("tx"),
1791 Err(TransactionStoreError::AuditChainMissing(_))
1792 ));
1793 assert!(matches!(
1794 read_only.claim_approved_for_execution("tx", "digest"),
1795 Err(TransactionStoreError::AuditChainMissing(_))
1796 ));
1797 }
1798
1799 #[test]
1806 fn record_persists_the_caller_principal_under_the_current_encoding() {
1807 let dir = tempdir().unwrap();
1808 let store = test_store(dir.path().join("tx.db"));
1809 let mut tx = queued_transaction();
1810 tx.caller_principal = CallerPrincipal::Uid(4242);
1811 store.record(tx).unwrap();
1812
1813 let conn = store.connection().unwrap();
1814 let (version, role, principal): (i64, Option<String>, Option<String>) = conn
1815 .query_row(
1816 "SELECT chain_version, caller_role, caller_principal FROM transactions",
1817 [],
1818 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
1819 )
1820 .unwrap();
1821
1822 assert_eq!(version, i64::from(audit_chain::CHAIN_VERSION_CURRENT));
1823 assert_eq!(role.as_deref(), Some("dev"));
1824 assert_eq!(
1825 principal.as_deref(),
1826 Some("uid:4242"),
1827 "the row must name the account the dispatcher resolved"
1828 );
1829
1830 let key = AuditKey::from_bytes(vec![0x42; 32]);
1832 assert!(matches!(
1833 store.verify_audit_chain(&key).unwrap(),
1834 audit_chain::VerifyOutcome::Intact { .. }
1835 ));
1836 }
1837
1838 #[test]
1839 fn record_writes_audit_chain_columns() {
1840 let dir = tempdir().unwrap();
1841 let store = test_store(dir.path().join("tx.db"));
1842 let _record = store.record(queued_transaction()).unwrap();
1843
1844 let conn = store.connection().unwrap();
1845 let (seq, key_id, chain_hash, prev): (i64, String, String, String) = conn
1846 .query_row(
1847 "SELECT seq, key_id, chain_hash, prev_chain_hash FROM transactions",
1848 [],
1849 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
1850 )
1851 .unwrap();
1852 assert_eq!(seq, 1, "first row gets seq=1");
1853 assert_eq!(key_id, audit_chain::CURRENT_KEY_ID);
1854 assert_eq!(prev, "", "first row has empty prev_chain_hash");
1855 assert_eq!(
1856 chain_hash.len(),
1857 audit_chain::HASH_HEX_LEN,
1858 "chain_hash is a hex-encoded Ed25519 signature"
1859 );
1860 }
1861
1862 #[test]
1863 fn sequential_records_produce_chained_hashes() {
1864 let dir = tempdir().unwrap();
1865 let store = test_store(dir.path().join("tx.db"));
1866 store.record(queued_transaction()).unwrap();
1867 store.record(queued_transaction()).unwrap();
1868 store.record(queued_transaction()).unwrap();
1869
1870 let rows = store.fetch_chain_rows().unwrap();
1871 assert_eq!(rows.len(), 3);
1872 assert_eq!(rows[0].seq, 1);
1873 assert_eq!(rows[1].seq, 2);
1874 assert_eq!(rows[2].seq, 3);
1875 assert_eq!(rows[1].prev_chain_hash, rows[0].chain_hash);
1877 assert_eq!(rows[2].prev_chain_hash, rows[1].chain_hash);
1878 }
1879
1880 #[test]
1892 fn concurrent_record_keeps_chain_intact_and_seqs_contiguous() {
1893 const WORKERS: usize = 8;
1894 const RECORDS_PER_WORKER: usize = 10;
1895 const TOTAL: usize = WORKERS * RECORDS_PER_WORKER;
1896
1897 let dir = tempdir().unwrap();
1898 let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
1899
1900 let mut handles = Vec::with_capacity(WORKERS);
1901 for w in 0..WORKERS {
1902 let store = std::sync::Arc::clone(&store);
1903 handles.push(std::thread::spawn(move || {
1904 for r in 0..RECORDS_PER_WORKER {
1905 let tx = NewTransaction {
1906 request_id: format!("worker-{w}-record-{r}"),
1907 request_hash: format!("hash-{w}-{r}"),
1908 action_name: "GetSystemState".to_string(),
1909 risk_level: RiskLevel::Low,
1910 summary: format!("worker {w} record {r}"),
1911 warnings: vec![],
1912 caller_role: CallerRole::Dev,
1913 caller_principal: CallerPrincipal::Uid(3000 + w as u32),
1917 };
1918 store
1919 .record(tx)
1920 .expect("record must succeed under contention");
1921 }
1922 }));
1923 }
1924 for h in handles {
1925 h.join().expect("worker thread did not panic");
1926 }
1927
1928 let key = AuditKey::from_bytes(vec![0x42; 32]);
1930 let outcome = store.verify_audit_chain(&key).unwrap();
1931 match outcome {
1932 VerifyOutcome::Intact { rows_checked } => {
1933 assert_eq!(
1934 rows_checked, TOTAL as u64,
1935 "expected {TOTAL} rows checked, got {rows_checked}"
1936 );
1937 }
1938 other => panic!("chain must be Intact under concurrent writes; got {other:?}"),
1939 }
1940
1941 {
1945 let conn = store.connection().unwrap();
1946 let mut stmt = conn
1947 .prepare("SELECT summary, caller_principal FROM transactions")
1948 .unwrap();
1949 let pairs: Vec<(String, Option<String>)> = stmt
1950 .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
1951 .unwrap()
1952 .map(|r| r.unwrap())
1953 .collect();
1954 assert_eq!(pairs.len(), TOTAL);
1955 for (summary, principal) in pairs {
1956 let worker: u32 = summary
1957 .split_whitespace()
1958 .nth(1)
1959 .and_then(|w| w.parse().ok())
1960 .unwrap_or_else(|| panic!("unexpected summary {summary:?}"));
1961 assert_eq!(
1962 principal.as_deref(),
1963 Some(format!("uid:{}", 3000 + worker).as_str()),
1964 "row {summary:?} was attributed to the wrong account"
1965 );
1966 }
1967 }
1968
1969 let conn = store.connection().unwrap();
1971 let mut stmt = conn
1972 .prepare("SELECT seq FROM transactions ORDER BY seq ASC")
1973 .unwrap();
1974 let seqs: Vec<i64> = stmt
1975 .query_map([], |row| row.get::<_, i64>(0))
1976 .unwrap()
1977 .map(|r| r.unwrap())
1978 .collect();
1979 assert_eq!(seqs.len(), TOTAL, "row count mismatch");
1980 for (i, s) in seqs.iter().enumerate() {
1981 assert_eq!(
1982 *s,
1983 (i as i64) + 1,
1984 "seq column must be contiguous 1..={TOTAL}; saw {s} at position {i}"
1985 );
1986 }
1987 }
1988
1989 #[test]
1990 fn verify_audit_chain_intact_after_inserts() {
1991 let dir = tempdir().unwrap();
1992 let store = test_store(dir.path().join("tx.db"));
1993 for _ in 0..3 {
1994 store.record(queued_transaction()).unwrap();
1995 }
1996 let key = AuditKey::from_bytes(vec![0x42; 32]);
1997 let outcome = store.verify_audit_chain(&key).unwrap();
1998 assert!(matches!(outcome, VerifyOutcome::Intact { rows_checked: 3 }));
1999 }
2000
2001 #[test]
2002 fn verify_audit_chain_with_pubkey_intact_after_inserts() {
2003 let dir = tempdir().unwrap();
2004 let store = test_store(dir.path().join("tx.db"));
2005 for _ in 0..3 {
2006 store.record(queued_transaction()).unwrap();
2007 }
2008 let key = AuditKey::from_bytes(vec![0x42; 32]);
2010 let outcome = store
2011 .verify_audit_chain_with_pubkey(&key.verifying_key_hex())
2012 .unwrap();
2013 assert!(matches!(outcome, VerifyOutcome::Intact { rows_checked: 3 }));
2014 }
2015
2016 #[test]
2017 fn verify_audit_chain_with_wrong_pubkey_is_broken() {
2018 let dir = tempdir().unwrap();
2019 let store = test_store(dir.path().join("tx.db"));
2020 store.record(queued_transaction()).unwrap();
2021 let other = AuditKey::from_bytes(vec![0x99; 32]);
2023 let outcome = store
2024 .verify_audit_chain_with_pubkey(&other.verifying_key_hex())
2025 .unwrap();
2026 assert!(matches!(outcome, VerifyOutcome::Broken { .. }));
2027 }
2028
2029 #[test]
2030 fn verify_detects_tampered_summary() {
2031 let dir = tempdir().unwrap();
2032 let store = test_store(dir.path().join("tx.db"));
2033 let tx = store.record(queued_transaction()).unwrap();
2034
2035 let conn = store.connection().unwrap();
2038 conn.execute(
2039 "UPDATE transactions SET summary = ?1 WHERE transaction_id = ?2",
2040 params!["EVIL CHANGE", tx.transaction_id],
2041 )
2042 .unwrap();
2043
2044 let key = AuditKey::from_bytes(vec![0x42; 32]);
2045 let outcome = store.verify_audit_chain(&key).unwrap();
2046 assert!(matches!(outcome, VerifyOutcome::Broken { .. }));
2047 }
2048
2049 #[test]
2050 fn status_update_does_not_break_chain() {
2051 let dir = tempdir().unwrap();
2053 let store = test_store(dir.path().join("tx.db"));
2054 let tx = store.record(queued_transaction()).unwrap();
2055 store
2056 .update_status(&tx.transaction_id, JobState::Running)
2057 .unwrap();
2058 store
2059 .update_status(&tx.transaction_id, JobState::Succeeded)
2060 .unwrap();
2061
2062 let key = AuditKey::from_bytes(vec![0x42; 32]);
2063 let outcome = store.verify_audit_chain(&key).unwrap();
2064 assert!(
2065 matches!(outcome, VerifyOutcome::Intact { rows_checked: 1 }),
2066 "status mutation must not break the chain (status not in chain content): {outcome:?}"
2067 );
2068 }
2069
2070 #[test]
2071 fn open_read_only_rejects_record() {
2072 let dir = tempdir().unwrap();
2073 let key_path = dir.path().join("audit-key");
2074 std::fs::write(&key_path, vec![0x42; 32]).unwrap();
2075 std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)).unwrap();
2076
2077 let store = TransactionStore::open_read_only(dir.path().join("tx.db")).unwrap();
2078 let result = store.record(queued_transaction());
2079 assert!(matches!(
2080 result,
2081 Err(TransactionStoreError::AuditChainMissing(_))
2082 ));
2083 }
2084
2085 #[test]
2086 fn ensure_private_dir_creates_with_0700_mode() {
2087 let root = tempdir().unwrap();
2088 let target = root.path().join("a/b/c");
2089 ensure_private_dir(&target).unwrap();
2090 assert!(target.is_dir());
2091 let mode = std::fs::metadata(&target).unwrap().permissions().mode() & 0o777;
2092 assert_eq!(mode, 0o700, "leaf dir must be 0o700, got {mode:o}");
2093 }
2094
2095 #[test]
2103 fn the_audit_database_is_created_owner_only() {
2104 let dir = tempdir().unwrap();
2105 let db = dir.path().join("daemon.sqlite");
2106 let store = TransactionStore::open(&db).unwrap();
2107 store
2109 .record(NewTransaction {
2110 request_id: "mode-probe".to_string(),
2111 request_hash: "hash-mode-probe".to_string(),
2112 action_name: "GetSystemState".to_string(),
2113 risk_level: RiskLevel::Low,
2114 summary: "probe".to_string(),
2115 warnings: vec![],
2116 caller_role: CallerRole::Dev,
2117 caller_principal: CallerPrincipal::Uid(3000),
2118 })
2119 .unwrap();
2120
2121 for path in [db.clone(), db.with_extension("sqlite-wal")] {
2122 if !path.exists() {
2123 continue;
2124 }
2125 let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
2126 assert_eq!(
2127 mode,
2128 0o600,
2129 "{} must be owner-only, got {mode:o}",
2130 path.display()
2131 );
2132 }
2133 }
2134
2135 #[test]
2136 fn ensure_private_dir_is_idempotent_and_does_not_widen_existing_mode() {
2137 let root = tempdir().unwrap();
2138 let target = root.path().join("preexisting");
2139 std::fs::create_dir(&target).unwrap();
2140 std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o755)).unwrap();
2141 ensure_private_dir(&target).unwrap();
2142 let mode = std::fs::metadata(&target).unwrap().permissions().mode() & 0o777;
2144 assert_eq!(mode, 0o755);
2145 }
2146
2147 #[test]
2148 fn open_creates_parent_with_private_mode() {
2149 let root = tempdir().unwrap();
2150 let db_path = root.path().join("nested/dirs/daemon.sqlite");
2151 let _store = test_store(&db_path);
2152 let parent = db_path.parent().unwrap();
2153 let mode = std::fs::metadata(parent).unwrap().permissions().mode() & 0o777;
2154 assert_eq!(mode, 0o700);
2155 }
2156
2157 fn queued_transaction() -> NewTransaction {
2158 NewTransaction {
2159 request_id: "req-1".to_string(),
2160 request_hash: "hash-abc".to_string(),
2161 action_name: "UpdateSystem".to_string(),
2162 risk_level: RiskLevel::High,
2163 summary: "Upgrade the system".to_string(),
2164 warnings: vec![],
2165 caller_role: CallerRole::Dev,
2166 caller_principal: CallerPrincipal::Uid(1000),
2167 }
2168 }
2169
2170 #[test]
2171 fn update_status_transitions_queued_to_running() {
2172 let dir = tempdir().unwrap();
2173 let store = test_store(dir.path().join("tx.db"));
2174 let tx = store.record(queued_transaction()).unwrap();
2175
2176 store
2177 .update_status(&tx.transaction_id, JobState::Running)
2178 .unwrap();
2179
2180 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2181 assert_eq!(updated.status, JobState::Running);
2182 }
2183
2184 #[test]
2185 fn update_status_transitions_running_to_succeeded() {
2186 let dir = tempdir().unwrap();
2187 let store = test_store(dir.path().join("tx.db"));
2188 let tx = store.record(queued_transaction()).unwrap();
2189
2190 store
2191 .update_status(&tx.transaction_id, JobState::Running)
2192 .unwrap();
2193 store
2194 .update_status(&tx.transaction_id, JobState::Succeeded)
2195 .unwrap();
2196
2197 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2198 assert_eq!(updated.status, JobState::Succeeded);
2199 }
2200
2201 #[test]
2202 fn update_status_for_unknown_id_returns_not_found() {
2203 let dir = tempdir().unwrap();
2204 let store = test_store(dir.path().join("tx.db"));
2205
2206 let result = store.update_status("does-not-exist", JobState::Running);
2207 assert!(
2208 matches!(result, Err(TransactionStoreError::NotFound(ref id)) if id == "does-not-exist"),
2209 "expected NotFound, got: {result:?}"
2210 );
2211 }
2212
2213 #[test]
2214 fn update_status_leaves_other_fields_intact() {
2215 let dir = tempdir().unwrap();
2216 let store = test_store(dir.path().join("tx.db"));
2217 let tx = store.record(queued_transaction()).unwrap();
2218
2219 store
2220 .update_status(&tx.transaction_id, JobState::Running)
2221 .unwrap();
2222 store
2223 .update_status(&tx.transaction_id, JobState::Failed)
2224 .unwrap();
2225
2226 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2227 assert_eq!(updated.action_name, "UpdateSystem");
2228 assert_eq!(updated.risk_level, RiskLevel::High);
2229 assert_eq!(updated.status, JobState::Failed);
2230 }
2231
2232 #[test]
2241 fn update_status_is_atomic_under_concurrent_conflicting_transitions() {
2242 let dir = tempdir().unwrap();
2243 let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
2244 let tx = store.record(queued_transaction()).unwrap();
2245 store
2246 .update_status(&tx.transaction_id, JobState::Running)
2247 .unwrap();
2248
2249 let store_a = std::sync::Arc::clone(&store);
2250 let id_a = tx.transaction_id.clone();
2251 let succeed = std::thread::spawn(move || store_a.update_status(&id_a, JobState::Succeeded));
2252
2253 let store_b = std::sync::Arc::clone(&store);
2254 let id_b = tx.transaction_id.clone();
2255 let fail = std::thread::spawn(move || store_b.update_status(&id_b, JobState::Failed));
2256
2257 let succeed_result = succeed.join().unwrap();
2258 let fail_result = fail.join().unwrap();
2259
2260 let oks = [succeed_result.is_ok(), fail_result.is_ok()]
2262 .into_iter()
2263 .filter(|ok| *ok)
2264 .count();
2265 assert_eq!(
2266 oks, 1,
2267 "exactly one concurrent transition should succeed: succeed={succeed_result:?} \
2268 fail={fail_result:?}"
2269 );
2270
2271 let loser = if succeed_result.is_ok() {
2287 &fail_result
2288 } else {
2289 &succeed_result
2290 };
2291 match loser {
2292 Err(TransactionStoreError::ConcurrentStatusChange(id)) => {
2293 assert_eq!(id, &tx.transaction_id);
2294 }
2295 Err(TransactionStoreError::InvalidTransition { from, to }) => {
2296 let winner = if succeed_result.is_ok() {
2298 JobState::Succeeded
2299 } else {
2300 JobState::Failed
2301 };
2302 assert_eq!(
2303 *from, winner,
2304 "the loser must have observed the winner's committed status"
2305 );
2306 assert_ne!(*to, winner, "the loser must be the other transition");
2307 }
2308 other => panic!(
2309 "the losing transition must be refused with either \
2310 ConcurrentStatusChange or InvalidTransition, got: {other:?}"
2311 ),
2312 }
2313
2314 let final_status = store.get(&tx.transaction_id).unwrap().unwrap().status;
2317 if succeed_result.is_ok() {
2318 assert_eq!(final_status, JobState::Succeeded);
2319 } else {
2320 assert_eq!(final_status, JobState::Failed);
2321 }
2322 }
2323
2324 #[test]
2325 fn revoke_unconsumed_approval_removes_an_unused_receipt() {
2326 let dir = tempdir().unwrap();
2329 let store = test_store(dir.path().join("tx.db"));
2330 let tx = store.record(queued_transaction()).unwrap();
2331 let receipt = store
2332 .approve_transaction(&tx.transaction_id)
2333 .unwrap()
2334 .expect("approved");
2335 let digest = audit_chain::approval_receipt_digest(&receipt);
2336
2337 assert!(
2338 store
2339 .revoke_unconsumed_approval(&tx.transaction_id)
2340 .unwrap(),
2341 "an unconsumed approval must be revocable"
2342 );
2343 assert!(
2344 !store
2345 .claim_approved_for_execution(&tx.transaction_id, &digest)
2346 .unwrap(),
2347 "a revoked receipt must no longer be claimable"
2348 );
2349 }
2350
2351 #[test]
2352 fn revoke_cannot_retract_an_approval_that_was_already_used() {
2353 let dir = tempdir().unwrap();
2357 let store = test_store(dir.path().join("tx.db"));
2358 let tx = store.record(queued_transaction()).unwrap();
2359 let receipt = store
2360 .approve_transaction(&tx.transaction_id)
2361 .unwrap()
2362 .expect("approved");
2363 let digest = audit_chain::approval_receipt_digest(&receipt);
2364 assert!(store
2365 .claim_approved_for_execution(&tx.transaction_id, &digest)
2366 .unwrap());
2367
2368 assert!(
2369 !store
2370 .revoke_unconsumed_approval(&tx.transaction_id)
2371 .unwrap(),
2372 "a consumed approval must not be revocable"
2373 );
2374 assert_eq!(
2375 store.get(&tx.transaction_id).unwrap().unwrap().status,
2376 JobState::Running,
2377 "a failed revocation must not disturb the running transaction"
2378 );
2379 }
2380
2381 #[test]
2382 fn approve_refuses_a_transaction_that_is_no_longer_queued() {
2383 let dir = tempdir().unwrap();
2387 let store = test_store(dir.path().join("tx.db"));
2388 let tx = store.record(queued_transaction()).unwrap();
2389 let receipt = store
2390 .approve_transaction(&tx.transaction_id)
2391 .unwrap()
2392 .expect("approved");
2393 let digest = audit_chain::approval_receipt_digest(&receipt);
2394 assert!(store
2395 .claim_approved_for_execution(&tx.transaction_id, &digest)
2396 .unwrap());
2397
2398 assert!(
2400 store
2401 .approve_transaction(&tx.transaction_id)
2402 .unwrap()
2403 .is_none(),
2404 "a Running transaction must not be approvable"
2405 );
2406
2407 store
2409 .update_status(&tx.transaction_id, JobState::Succeeded)
2410 .unwrap();
2411 assert!(
2412 store
2413 .approve_transaction(&tx.transaction_id)
2414 .unwrap()
2415 .is_none(),
2416 "a completed transaction must not be approvable"
2417 );
2418 }
2419
2420 #[test]
2421 fn only_one_of_two_concurrent_claims_can_execute() {
2422 let dir = tempdir().unwrap();
2427 let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
2428 let tx = store.record(queued_transaction()).unwrap();
2429 let receipt = store
2430 .approve_transaction(&tx.transaction_id)
2431 .unwrap()
2432 .expect("approved");
2433 let digest = audit_chain::approval_receipt_digest(&receipt);
2434
2435 let mut handles = Vec::new();
2436 for _ in 0..2 {
2437 let store = std::sync::Arc::clone(&store);
2438 let id = tx.transaction_id.clone();
2439 let digest = digest.clone();
2440 handles.push(std::thread::spawn(move || {
2441 store.claim_approved_for_execution(&id, &digest)
2442 }));
2443 }
2444 let claims: Vec<bool> = handles
2445 .into_iter()
2446 .map(|h| h.join().unwrap().unwrap_or(false))
2447 .collect();
2448
2449 assert_eq!(
2450 claims.iter().filter(|c| **c).count(),
2451 1,
2452 "exactly one claim may win: {claims:?}"
2453 );
2454 assert_eq!(
2455 store.get(&tx.transaction_id).unwrap().unwrap().status,
2456 JobState::Running
2457 );
2458 }
2459
2460 #[test]
2461 fn cancel_queued_cancels_a_queued_transaction_once() {
2462 let dir = tempdir().unwrap();
2463 let store = test_store(dir.path().join("tx.db"));
2464 let tx = store.record(queued_transaction()).unwrap();
2465
2466 assert!(
2467 store.cancel_queued(&tx.transaction_id).unwrap(),
2468 "a queued transaction is cancelable"
2469 );
2470 assert_eq!(
2471 store.get(&tx.transaction_id).unwrap().unwrap().status,
2472 JobState::Canceled
2473 );
2474 assert!(
2475 !store.cancel_queued(&tx.transaction_id).unwrap(),
2476 "an already-canceled transaction is not cancelable again"
2477 );
2478 assert!(
2479 !store.cancel_queued("no-such-transaction").unwrap(),
2480 "a missing transaction is not cancelable"
2481 );
2482 assert!(
2483 store.fetch_event_rows().unwrap().is_empty(),
2484 "canceling an unapproved transaction must not append an approval event"
2485 );
2486 }
2487
2488 #[test]
2489 fn cancel_queued_revokes_an_unconsumed_approval_and_appends_event() {
2490 let dir = tempdir().unwrap();
2491 let store = test_store(dir.path().join("tx.db"));
2492 let tx = store.record(queued_transaction()).unwrap();
2493 let receipt = store
2494 .approve_transaction(&tx.transaction_id)
2495 .unwrap()
2496 .expect("queued transaction is approvable");
2497 let digest = audit_chain::approval_receipt_digest(&receipt);
2498
2499 assert!(store.cancel_queued(&tx.transaction_id).unwrap());
2500 assert_eq!(
2501 store.get(&tx.transaction_id).unwrap().unwrap().status,
2502 JobState::Canceled
2503 );
2504 assert!(
2505 !store
2506 .claim_approved_for_execution(&tx.transaction_id, &digest)
2507 .unwrap(),
2508 "canceling must revoke the unconsumed receipt"
2509 );
2510 assert_eq!(
2511 store
2512 .fetch_event_rows()
2513 .unwrap()
2514 .iter()
2515 .map(|event| event.kind.as_str())
2516 .collect::<Vec<_>>(),
2517 vec!["approval_granted", "approval_revoked"]
2518 );
2519 }
2520
2521 #[test]
2522 fn cancel_queued_refuses_a_running_transaction() {
2523 let dir = tempdir().unwrap();
2526 let store = test_store(dir.path().join("tx.db"));
2527 let tx = store.record(queued_transaction()).unwrap();
2528 let receipt = store
2529 .approve_transaction(&tx.transaction_id)
2530 .unwrap()
2531 .expect("approved");
2532 let digest = audit_chain::approval_receipt_digest(&receipt);
2533 assert!(store
2534 .claim_approved_for_execution(&tx.transaction_id, &digest)
2535 .unwrap());
2536
2537 assert!(
2538 !store.cancel_queued(&tx.transaction_id).unwrap(),
2539 "a running transaction must not be cancelable"
2540 );
2541 assert_eq!(
2542 store.get(&tx.transaction_id).unwrap().unwrap().status,
2543 JobState::Running,
2544 "cancel must not disturb a running transaction"
2545 );
2546 }
2547
2548 #[test]
2549 fn approved_receipt_is_required_and_consumed_once() {
2550 let dir = tempdir().unwrap();
2551 let store = test_store(dir.path().join("tx.db"));
2552 let tx = store.record(queued_transaction()).unwrap();
2553
2554 assert!(
2555 !store
2556 .claim_approved_for_execution(&tx.transaction_id, "digest-a")
2557 .unwrap(),
2558 "an unapproved preview must not execute"
2559 );
2560 let receipt = store
2561 .approve_transaction(&tx.transaction_id)
2562 .unwrap()
2563 .expect("first approval must succeed");
2564 let digest = audit_chain::approval_receipt_digest(&receipt);
2565 assert_eq!(tx.approval_id.as_deref(), Some(digest.as_str()));
2566 assert!(
2567 store
2568 .approve_transaction(&tx.transaction_id)
2569 .unwrap()
2570 .is_none(),
2571 "approval is immutable once issued"
2572 );
2573 assert!(
2574 !store
2575 .claim_approved_for_execution(&tx.transaction_id, "wrong-digest")
2576 .unwrap(),
2577 "a forged receipt must not execute"
2578 );
2579 assert!(
2580 store
2581 .claim_approved_for_execution(&tx.transaction_id, &digest)
2582 .unwrap(),
2583 "the exact approved receipt must execute"
2584 );
2585 assert!(
2586 !store
2587 .claim_approved_for_execution(&tx.transaction_id, &digest)
2588 .unwrap(),
2589 "the receipt must be one-time"
2590 );
2591 }
2592
2593 #[test]
2594 fn approval_commitment_is_covered_by_the_signed_chain() {
2595 let dir = tempdir().unwrap();
2596 let store = test_store(dir.path().join("tx.db"));
2597 let tx = store.record(queued_transaction()).unwrap();
2598 assert!(tx.approval_id.is_some());
2599
2600 store
2601 .connection()
2602 .unwrap()
2603 .execute(
2604 "UPDATE transactions SET approval_id = 'forged' WHERE transaction_id = ?1",
2605 params![tx.transaction_id],
2606 )
2607 .unwrap();
2608
2609 let key = AuditKey::from_bytes(vec![0x42; 32]);
2610 assert!(matches!(
2611 store.verify_audit_chain(&key).unwrap(),
2612 VerifyOutcome::Broken { .. }
2613 ));
2614 }
2615
2616 #[test]
2617 fn approve_rejects_a_forged_commitment_at_runtime() {
2618 let dir = tempdir().unwrap();
2622 let store = test_store(dir.path().join("tx.db"));
2623 let tx = store.record(queued_transaction()).unwrap();
2624
2625 store
2626 .connection()
2627 .unwrap()
2628 .execute(
2629 "UPDATE transactions SET approval_id = 'forged' WHERE transaction_id = ?1",
2630 params![tx.transaction_id],
2631 )
2632 .unwrap();
2633
2634 let err = store
2635 .approve_transaction(&tx.transaction_id)
2636 .expect_err("a forged commitment must be rejected, not approved");
2637 assert!(
2638 matches!(err, TransactionStoreError::DatabaseInvariant(_)),
2639 "forged commitment must surface as a DatabaseInvariant, got {err:?}"
2640 );
2641 }
2642
2643 #[test]
2644 fn stale_iso_timestamp_cannot_be_approved() {
2645 let dir = tempdir().unwrap();
2646 let store = test_store(dir.path().join("tx.db"));
2647 let tx = store.record(queued_transaction()).unwrap();
2648 let conn = store.connection().unwrap();
2649 conn.execute(
2650 "UPDATE transactions \
2651 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
2652 WHERE transaction_id = ?1",
2653 params![tx.transaction_id],
2654 )
2655 .unwrap();
2656
2657 assert!(
2658 store
2659 .approve_transaction(&tx.transaction_id)
2660 .unwrap()
2661 .is_none(),
2662 "a production-format timestamp outside the TTL must not be approved"
2663 );
2664 }
2665
2666 #[test]
2667 fn stale_iso_timestamp_cannot_be_claimed_for_execution() {
2668 let dir = tempdir().unwrap();
2673 let store = test_store(dir.path().join("tx.db"));
2674 let tx = store.record(queued_transaction()).unwrap();
2675 let receipt = store
2676 .approve_transaction(&tx.transaction_id)
2677 .unwrap()
2678 .expect("a fresh approval succeeds");
2679 let digest = audit_chain::approval_receipt_digest(&receipt);
2680
2681 let conn = store.connection().unwrap();
2683 conn.execute(
2684 "UPDATE transactions \
2685 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
2686 WHERE transaction_id = ?1",
2687 params![tx.transaction_id],
2688 )
2689 .unwrap();
2690
2691 assert!(
2692 !store
2693 .claim_approved_for_execution(&tx.transaction_id, &digest)
2694 .unwrap(),
2695 "an approval aged past the TTL must not be claimable at execute time"
2696 );
2697 assert_eq!(
2698 store.get(&tx.transaction_id).unwrap().unwrap().status,
2699 JobState::Queued,
2700 "a TTL-expired claim must leave the transaction Queued, never Running"
2701 );
2702 }
2703
2704 #[test]
2705 fn cleanup_stale_queued_cancels_old_records() {
2706 let dir = tempdir().unwrap();
2707 let store = test_store(dir.path().join("tx.db"));
2708
2709 let fresh = store.record(queued_transaction()).unwrap();
2711 let stale = store.record(queued_transaction()).unwrap();
2712
2713 let conn = store.connection().unwrap();
2715 conn.execute(
2716 "UPDATE transactions \
2717 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
2718 WHERE transaction_id = ?1",
2719 params![stale.transaction_id],
2720 )
2721 .unwrap();
2722
2723 let canceled = store.cleanup_stale_queued().unwrap();
2724 assert_eq!(canceled, 1, "only the stale record should be canceled");
2725
2726 let stale_record = store.get(&stale.transaction_id).unwrap().unwrap();
2728 assert_eq!(stale_record.status, JobState::Canceled);
2729
2730 let fresh_record = store.get(&fresh.transaction_id).unwrap().unwrap();
2732 assert_eq!(fresh_record.status, JobState::Queued);
2733 }
2734
2735 #[test]
2736 fn cleanup_stale_queued_revokes_every_unconsumed_approval() {
2737 let dir = tempdir().unwrap();
2738 let store = test_store(dir.path().join("tx.db"));
2739 let transactions: Vec<_> = (0..3)
2740 .map(|_| {
2741 let transaction = store.record(queued_transaction()).unwrap();
2742 store
2743 .approve_transaction(&transaction.transaction_id)
2744 .unwrap()
2745 .expect("stale transaction is approvable");
2746 transaction
2747 })
2748 .collect();
2749
2750 let conn = store.connection().unwrap();
2751 for transaction in &transactions {
2752 conn.execute(
2753 "UPDATE transactions \
2754 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
2755 WHERE transaction_id = ?1",
2756 params![transaction.transaction_id],
2757 )
2758 .unwrap();
2759 }
2760
2761 assert_eq!(store.cleanup_stale_queued().unwrap(), 3);
2762 for transaction in &transactions {
2763 assert_eq!(
2764 store
2765 .get(&transaction.transaction_id)
2766 .unwrap()
2767 .unwrap()
2768 .status,
2769 JobState::Canceled
2770 );
2771 assert!(!store
2772 .revoke_unconsumed_approval(&transaction.transaction_id)
2773 .unwrap());
2774 }
2775 assert_eq!(
2776 store
2777 .fetch_event_rows()
2778 .unwrap()
2779 .iter()
2780 .map(|event| event.kind.as_str())
2781 .collect::<Vec<_>>(),
2782 vec![
2783 "approval_granted",
2784 "approval_granted",
2785 "approval_granted",
2786 "approval_revoked",
2787 "approval_revoked",
2788 "approval_revoked"
2789 ]
2790 );
2791 }
2792
2793 #[test]
2796 fn update_status_rejects_queued_to_succeeded() {
2797 let dir = tempdir().unwrap();
2798 let store = test_store(dir.path().join("tx.db"));
2799 let tx = store.record(queued_transaction()).unwrap();
2800
2801 let result = store.update_status(&tx.transaction_id, JobState::Succeeded);
2802 assert!(
2803 matches!(
2804 result,
2805 Err(TransactionStoreError::InvalidTransition {
2806 from: JobState::Queued,
2807 to: JobState::Succeeded,
2808 })
2809 ),
2810 "Queued -> Succeeded must be rejected (must go through Running first): {result:?}"
2811 );
2812 }
2813
2814 #[test]
2815 fn update_status_rejects_succeeded_to_running() {
2816 let dir = tempdir().unwrap();
2817 let store = test_store(dir.path().join("tx.db"));
2818 let tx = store.record(queued_transaction()).unwrap();
2819
2820 store
2821 .update_status(&tx.transaction_id, JobState::Running)
2822 .unwrap();
2823 store
2824 .update_status(&tx.transaction_id, JobState::Succeeded)
2825 .unwrap();
2826
2827 let result = store.update_status(&tx.transaction_id, JobState::Running);
2828 assert!(
2829 matches!(
2830 result,
2831 Err(TransactionStoreError::InvalidTransition {
2832 from: JobState::Succeeded,
2833 to: JobState::Running,
2834 })
2835 ),
2836 "Succeeded -> Running must be rejected (terminal state): {result:?}"
2837 );
2838 }
2839
2840 #[test]
2841 fn update_status_accepts_running_to_failed() {
2842 let dir = tempdir().unwrap();
2843 let store = test_store(dir.path().join("tx.db"));
2844 let tx = store.record(queued_transaction()).unwrap();
2845
2846 store
2847 .update_status(&tx.transaction_id, JobState::Running)
2848 .unwrap();
2849 store
2850 .update_status(&tx.transaction_id, JobState::Failed)
2851 .unwrap();
2852
2853 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2854 assert_eq!(updated.status, JobState::Failed);
2855 }
2856
2857 #[test]
2858 fn update_status_accepts_running_to_rolled_back() {
2859 let dir = tempdir().unwrap();
2860 let store = test_store(dir.path().join("tx.db"));
2861 let tx = store.record(queued_transaction()).unwrap();
2862
2863 store
2864 .update_status(&tx.transaction_id, JobState::Running)
2865 .unwrap();
2866 store
2867 .update_status(&tx.transaction_id, JobState::RolledBack)
2868 .unwrap();
2869
2870 let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2871 assert_eq!(updated.status, JobState::RolledBack);
2872 }
2873
2874 #[test]
2877 fn list_transactions_returns_empty_for_fresh_store() {
2878 let dir = tempdir().unwrap();
2879 let store = test_store(dir.path().join("tx.db"));
2880 let results = store.list_transactions(10, None, None, None).unwrap();
2881 assert!(results.is_empty());
2882 }
2883
2884 #[test]
2885 fn list_transactions_returns_all_records_ordered_by_newest_first() {
2886 let dir = tempdir().unwrap();
2887 let store = test_store(dir.path().join("tx.db"));
2888 store.record(queued_transaction()).unwrap();
2889
2890 let mut second = queued_transaction();
2891 second.action_name = "GetDiskUsage".to_string();
2892 second.risk_level = RiskLevel::Low;
2893 store.record(second).unwrap();
2894
2895 let results = store.list_transactions(10, None, None, None).unwrap();
2896 assert_eq!(results.len(), 2);
2897 assert_eq!(results[0].action_name, "GetDiskUsage");
2899 assert_eq!(results[1].action_name, "UpdateSystem");
2900 }
2901
2902 #[test]
2903 fn list_history_populates_created_at_and_risk_level() {
2904 let dir = tempdir().unwrap();
2905 let store = test_store(dir.path().join("tx.db"));
2906 store.record(queued_transaction()).unwrap();
2907
2908 let entries = store.list_history(10, None, None, None).unwrap();
2909 assert_eq!(entries.len(), 1);
2910 let entry = &entries[0];
2911 assert_eq!(entry.action_name, "UpdateSystem");
2912 assert_eq!(entry.risk_level, RiskLevel::High);
2913 assert_eq!(entry.status, JobState::Queued);
2914 assert!(
2915 !entry.created_at.is_empty(),
2916 "created_at must be populated from the stored row, not left blank"
2917 );
2918 }
2919
2920 #[test]
2921 fn list_history_applies_the_same_filters_as_list_transactions() {
2922 let dir = tempdir().unwrap();
2923 let store = test_store(dir.path().join("tx.db"));
2924 store.record(queued_transaction()).unwrap();
2925 let mut low = queued_transaction();
2926 low.action_name = "GetDiskUsage".to_string();
2927 low.risk_level = RiskLevel::Low;
2928 store.record(low).unwrap();
2929
2930 let only = store
2931 .list_history(10, None, Some("GetDiskUsage"), None)
2932 .unwrap();
2933 assert_eq!(only.len(), 1);
2934 assert_eq!(only[0].action_name, "GetDiskUsage");
2935 }
2936
2937 #[test]
2938 fn list_transactions_respects_limit() {
2939 let dir = tempdir().unwrap();
2940 let store = test_store(dir.path().join("tx.db"));
2941 for _ in 0..5 {
2942 store.record(queued_transaction()).unwrap();
2943 }
2944 let results = store.list_transactions(3, None, None, None).unwrap();
2945 assert_eq!(results.len(), 3);
2946 }
2947
2948 #[test]
2949 fn list_transactions_filters_by_status() {
2950 let dir = tempdir().unwrap();
2951 let store = test_store(dir.path().join("tx.db"));
2952 let tx = store.record(queued_transaction()).unwrap();
2953 store
2954 .update_status(&tx.transaction_id, JobState::Running)
2955 .unwrap();
2956 store
2957 .update_status(&tx.transaction_id, JobState::Succeeded)
2958 .unwrap();
2959
2960 store.record(queued_transaction()).unwrap();
2962
2963 let succeeded = store
2964 .list_transactions(10, Some("succeeded"), None, None)
2965 .unwrap();
2966 assert_eq!(succeeded.len(), 1);
2967 assert_eq!(succeeded[0].status, JobState::Succeeded);
2968
2969 let queued = store
2970 .list_transactions(10, Some("queued"), None, None)
2971 .unwrap();
2972 assert_eq!(queued.len(), 1);
2973 assert_eq!(queued[0].status, JobState::Queued);
2974 }
2975
2976 #[test]
2977 fn list_transactions_filters_by_action_name() {
2978 let dir = tempdir().unwrap();
2979 let store = test_store(dir.path().join("tx.db"));
2980 store.record(queued_transaction()).unwrap(); let mut disk = queued_transaction();
2983 disk.action_name = "GetDiskUsage".to_string();
2984 store.record(disk).unwrap();
2985
2986 let results = store
2987 .list_transactions(10, None, Some("GetDiskUsage"), None)
2988 .unwrap();
2989 assert_eq!(results.len(), 1);
2990 assert_eq!(results[0].action_name, "GetDiskUsage");
2991 }
2992
2993 #[test]
2994 fn list_transactions_filters_by_since_hours() {
2995 let dir = tempdir().unwrap();
2996 let store = test_store(dir.path().join("tx.db"));
2997
2998 let old = store.record(queued_transaction()).unwrap();
3000 let conn = store.connection().unwrap();
3001 conn.execute(
3002 "UPDATE transactions \
3003 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-48 hours') \
3004 WHERE transaction_id = ?1",
3005 params![old.transaction_id],
3006 )
3007 .unwrap();
3008
3009 store.record(queued_transaction()).unwrap();
3011
3012 let results = store.list_transactions(10, None, None, Some(24)).unwrap();
3014 assert_eq!(results.len(), 1);
3015
3016 let results = store.list_transactions(10, None, None, Some(72)).unwrap();
3018 assert_eq!(results.len(), 2);
3019 }
3020
3021 #[test]
3022 fn list_transactions_rejects_invalid_status_filter() {
3023 let dir = tempdir().unwrap();
3024 let store = test_store(dir.path().join("tx.db"));
3025 store.record(queued_transaction()).unwrap();
3026 let result = store.list_transactions(10, Some("bogus"), None, None);
3027 assert!(result.is_err(), "invalid status filter should return error");
3028 }
3029
3030 #[test]
3038 fn record_emits_one_watermark_per_entry() {
3039 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3040 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3041
3042 let dir = tempdir().unwrap();
3043 let store = test_store(dir.path().join("tx.db"));
3044 store.record(queued_transaction()).unwrap();
3045
3046 let calls = crate::audit_watermark::take_watermarks(&sink);
3047 assert_eq!(calls.len(), 1, "expected exactly 1 watermark per record()");
3048 }
3049
3050 #[test]
3052 fn record_previewed_emits_one_watermark() {
3053 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3054 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3055
3056 let dir = tempdir().unwrap();
3057 let store = test_store(dir.path().join("tx.db"));
3058 let preview = PreviewEnvelope {
3059 summary: "Upgrade the system".to_string(),
3060 risk_level: RiskLevel::High,
3061 current_state: serde_json::Value::Null,
3062 proposed_change: serde_json::Value::Null,
3063 expected_side_effects: vec![],
3064 reboot_required: false,
3065 rollback_available: false,
3066 warnings: vec![],
3067 request_hash: sysknife_types::RequestHash::from("hash-abc".to_string()),
3068 };
3069 store
3070 .record_previewed(queued_transaction(), preview)
3071 .unwrap();
3072
3073 let calls = crate::audit_watermark::take_watermarks(&sink);
3074 assert_eq!(
3075 calls.len(),
3076 1,
3077 "expected exactly 1 watermark per record_previewed()"
3078 );
3079 }
3080
3081 #[test]
3083 fn watermark_seq_and_hash_match_chain_row() {
3084 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3085 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3086
3087 let dir = tempdir().unwrap();
3088 let store = test_store(dir.path().join("tx.db"));
3089 store.record(queued_transaction()).unwrap();
3090
3091 let rows = store.fetch_chain_rows().unwrap();
3092 assert_eq!(rows.len(), 1);
3093 let row = &rows[0];
3094
3095 let calls = crate::audit_watermark::take_watermarks(&sink);
3096 assert_eq!(calls.len(), 1);
3097 assert_eq!(calls[0].seq, row.seq, "watermark seq must match chain row");
3098 assert_eq!(
3099 calls[0].chain_hash_hex, row.chain_hash,
3100 "watermark chain_hash_hex must match stored chain_hash"
3101 );
3102 }
3103
3104 #[test]
3106 fn multiple_records_produce_one_watermark_each() {
3107 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3108 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3109
3110 let dir = tempdir().unwrap();
3111 let store = test_store(dir.path().join("tx.db"));
3112 for _ in 0..3 {
3113 store.record(queued_transaction()).unwrap();
3114 }
3115
3116 let calls = crate::audit_watermark::take_watermarks(&sink);
3117 assert_eq!(calls.len(), 3, "one watermark per record call");
3118 assert_eq!(calls[0].seq, 1);
3119 assert_eq!(calls[1].seq, 2);
3120 assert_eq!(calls[2].seq, 3);
3121 }
3122
3123 #[test]
3137 fn no_watermark_emitted_before_any_record() {
3138 let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3139 crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3140
3141 let dir = tempdir().unwrap();
3142 let _store = test_store(dir.path().join("tx.db"));
3143
3144 let calls = crate::audit_watermark::take_watermarks(&sink);
3146 assert!(
3147 calls.is_empty(),
3148 "no watermark must be emitted without a record() call"
3149 );
3150 }
3151}