use async_trait::async_trait;
use chrono::{DateTime, TimeDelta, Utc};
use minco_plugin_audit::{
AuditAppendReport, AuditCursor, AuditJournalEntry, AuditJournalStatus, AuditJournalStore,
AuditLedgerError, AuditLedgerWriter, AuditLifecyclePolicy, AuditPage, AuditQuery, AuditReader,
AuditRecordV2, AuditSegmentState, AuditSegmentStatus, AuditStorageHealth,
AuditStorageInspector, AuditStorageSnapshot, evaluate_storage_health,
};
use sqlx::{QueryBuilder, Row, Sqlite, SqlitePool, Transaction};
use std::collections::BTreeMap;
use uuid::Uuid;
#[derive(Debug, Clone)]
pub struct SqliteAuditJournal {
pool: SqlitePool,
}
impl SqliteAuditJournal {
pub const fn new(pool: SqlitePool) -> Self {
Self { pool }
}
pub async fn enqueue_in(
&self,
transaction: &mut Transaction<'_, Sqlite>,
entry: AuditJournalEntry,
) -> Result<(), AuditLedgerError> {
validate_pending_entry(&entry)?;
let record =
serde_json::to_string(&entry.record).map_err(|_| AuditLedgerError::Encoding)?;
let encoded_bytes = i64::try_from(entry.encoded_bytes)
.map_err(|_| AuditLedgerError::InvalidJournalEntry)?;
let result = sqlx::query(
"INSERT INTO minco_audit_journal
(event_id, occurred_at, record, encoded_bytes, status, attempt_count,
available_at, claimed_by, claim_expires_at, failure_code)
VALUES (?, ?, ?, ?, 'pending', ?, ?, ?, ?, ?)
ON CONFLICT(event_id) DO NOTHING",
)
.bind(entry.record.event_id)
.bind(entry.record.occurred_at)
.bind(&record)
.bind(encoded_bytes)
.bind(i64::from(entry.attempt_count))
.bind(entry.available_at)
.bind(entry.claimed_by)
.bind(entry.claim_expires_at)
.bind(entry.failure_code)
.execute(&mut **transaction)
.await
.map_err(infrastructure)?;
if result.rows_affected() == 1 {
return Ok(());
}
let existing: String =
sqlx::query_scalar("SELECT record FROM minco_audit_journal WHERE event_id = ?")
.bind(entry.record.event_id)
.fetch_one(&mut **transaction)
.await
.map_err(infrastructure)?;
if existing == record {
Ok(())
} else {
Err(AuditLedgerError::EventConflict(entry.record.event_id))
}
}
async fn transition(
&self,
event_ids: &[Uuid],
worker_id: &str,
transition: JournalTransition<'_>,
) -> Result<(), AuditLedgerError> {
validate_transition(event_ids, worker_id)?;
let mut transaction = self
.pool
.begin_with("BEGIN IMMEDIATE")
.await
.map_err(infrastructure)?;
let mut count = QueryBuilder::<Sqlite>::new(
"SELECT COUNT(*) FROM minco_audit_journal WHERE status = 'claimed' AND claimed_by = ",
);
count.push_bind(worker_id).push(" AND event_id IN (");
push_uuid_list(&mut count, event_ids);
count.push(")");
let claimed: i64 = count
.build_query_scalar()
.fetch_one(&mut *transaction)
.await
.map_err(infrastructure)?;
if usize::try_from(claimed).ok() != Some(event_ids.len()) {
return Err(AuditLedgerError::JournalClaimLost);
}
let mut statement = match transition {
JournalTransition::Delivered => {
QueryBuilder::<Sqlite>::new("DELETE FROM minco_audit_journal WHERE event_id IN (")
}
JournalTransition::Retry { .. } | JournalTransition::Quarantine { .. } => {
let mut builder = QueryBuilder::<Sqlite>::new("UPDATE minco_audit_journal SET ");
match transition {
JournalTransition::Retry {
failure_code,
retry_at,
} => {
validate_failure_code(failure_code)?;
builder
.push("status = 'failed', available_at = ")
.push_bind(retry_at)
.push(", failure_code = ")
.push_bind(failure_code);
}
JournalTransition::Quarantine { failure_code } => {
validate_failure_code(failure_code)?;
builder
.push("status = 'quarantined', failure_code = ")
.push_bind(failure_code);
}
JournalTransition::Delivered => unreachable!(),
}
builder.push(", claimed_by = NULL, claim_expires_at = NULL WHERE event_id IN (");
builder
}
};
push_uuid_list(&mut statement, event_ids);
statement.push(")");
let affected = statement
.build()
.execute(&mut *transaction)
.await
.map_err(infrastructure)?
.rows_affected();
if usize::try_from(affected).ok() != Some(event_ids.len()) {
return Err(AuditLedgerError::JournalClaimLost);
}
transaction.commit().await.map_err(infrastructure)
}
}
enum JournalTransition<'a> {
Delivered,
Retry {
failure_code: &'a str,
retry_at: DateTime<Utc>,
},
Quarantine {
failure_code: &'a str,
},
}
#[async_trait]
impl AuditJournalStore for SqliteAuditJournal {
async fn enqueue(&self, entry: AuditJournalEntry) -> Result<(), AuditLedgerError> {
let mut transaction = self
.pool
.begin_with("BEGIN IMMEDIATE")
.await
.map_err(infrastructure)?;
self.enqueue_in(&mut transaction, entry).await?;
transaction.commit().await.map_err(infrastructure)
}
async fn claim_pending(
&self,
worker_id: &str,
limit: usize,
claim_expires_at: DateTime<Utc>,
) -> Result<Vec<AuditJournalEntry>, AuditLedgerError> {
validate_claim(worker_id, limit, claim_expires_at)?;
let limit = i64::try_from(limit).map_err(|_| AuditLedgerError::InvalidJournalClaim)?;
let now = Utc::now();
let mut transaction = self
.pool
.begin_with("BEGIN IMMEDIATE")
.await
.map_err(infrastructure)?;
let ids = sqlx::query_scalar::<_, Uuid>(
"SELECT event_id FROM minco_audit_journal
WHERE status IN ('pending', 'failed') AND available_at <= ?
ORDER BY available_at, occurred_at, event_id LIMIT ?",
)
.bind(now)
.bind(limit)
.fetch_all(&mut *transaction)
.await
.map_err(infrastructure)?;
if ids.is_empty() {
transaction.commit().await.map_err(infrastructure)?;
return Ok(Vec::new());
}
let mut update = QueryBuilder::<Sqlite>::new(
"UPDATE minco_audit_journal SET status = 'claimed', claimed_by = ",
);
update
.push_bind(worker_id)
.push(", claim_expires_at = ")
.push_bind(claim_expires_at)
.push(", attempt_count = attempt_count + 1 WHERE event_id IN (");
push_uuid_list(&mut update, &ids);
update.push(")");
update
.build()
.execute(&mut *transaction)
.await
.map_err(infrastructure)?;
let entries = fetch_entries(&mut transaction, &ids).await?;
transaction.commit().await.map_err(infrastructure)?;
Ok(entries)
}
async fn mark_delivered(
&self,
event_ids: &[Uuid],
worker_id: &str,
) -> Result<(), AuditLedgerError> {
self.transition(event_ids, worker_id, JournalTransition::Delivered)
.await
}
async fn mark_retry(
&self,
event_ids: &[Uuid],
worker_id: &str,
failure_code: &str,
retry_at: DateTime<Utc>,
) -> Result<(), AuditLedgerError> {
self.transition(
event_ids,
worker_id,
JournalTransition::Retry {
failure_code,
retry_at,
},
)
.await
}
async fn quarantine(
&self,
event_ids: &[Uuid],
worker_id: &str,
failure_code: &str,
) -> Result<(), AuditLedgerError> {
self.transition(
event_ids,
worker_id,
JournalTransition::Quarantine { failure_code },
)
.await
}
async fn recover_expired_claims(&self, now: DateTime<Utc>) -> Result<usize, AuditLedgerError> {
let result = sqlx::query(
"UPDATE minco_audit_journal
SET status = 'failed', available_at = ?, claimed_by = NULL,
claim_expires_at = NULL, failure_code = 'AUDIT-CLAIM-EXPIRED'
WHERE status = 'claimed' AND claim_expires_at <= ?",
)
.bind(now)
.bind(now)
.execute(&self.pool)
.await
.map_err(infrastructure)?;
usize::try_from(result.rows_affected()).map_err(|_| AuditLedgerError::Infrastructure)
}
}
async fn fetch_entries(
transaction: &mut Transaction<'_, Sqlite>,
ids: &[Uuid],
) -> Result<Vec<AuditJournalEntry>, AuditLedgerError> {
let mut query =
QueryBuilder::<Sqlite>::new("SELECT * FROM minco_audit_journal WHERE event_id IN (");
push_uuid_list(&mut query, ids);
query.push(") ORDER BY available_at, occurred_at, event_id");
query
.build()
.fetch_all(&mut **transaction)
.await
.map_err(infrastructure)?
.iter()
.map(decode_journal_entry)
.collect()
}
fn decode_journal_entry(
row: &sqlx::sqlite::SqliteRow,
) -> Result<AuditJournalEntry, AuditLedgerError> {
let record: String = row.try_get("record").map_err(infrastructure)?;
let status: String = row.try_get("status").map_err(infrastructure)?;
let record: AuditRecordV2 =
serde_json::from_str(&record).map_err(|_| AuditLedgerError::Encoding)?;
let encoded_bytes: i64 = row.try_get("encoded_bytes").map_err(infrastructure)?;
let attempt_count: i64 = row.try_get("attempt_count").map_err(infrastructure)?;
Ok(AuditJournalEntry {
record,
status: decode_status(&status)?,
attempt_count: u32::try_from(attempt_count)
.map_err(|_| AuditLedgerError::Infrastructure)?,
encoded_bytes: usize::try_from(encoded_bytes)
.map_err(|_| AuditLedgerError::Infrastructure)?,
available_at: row.try_get("available_at").map_err(infrastructure)?,
claimed_by: row.try_get("claimed_by").map_err(infrastructure)?,
claim_expires_at: row.try_get("claim_expires_at").map_err(infrastructure)?,
failure_code: row.try_get("failure_code").map_err(infrastructure)?,
})
}
fn decode_status(value: &str) -> Result<AuditJournalStatus, AuditLedgerError> {
match value {
"pending" => Ok(AuditJournalStatus::Pending),
"claimed" => Ok(AuditJournalStatus::Claimed),
"failed" => Ok(AuditJournalStatus::Failed),
"quarantined" => Ok(AuditJournalStatus::Quarantined),
_ => Err(AuditLedgerError::Infrastructure),
}
}
#[derive(Debug, Clone)]
pub struct SqliteAuditLedger {
pool: SqlitePool,
}
impl SqliteAuditLedger {
pub const fn new(pool: SqlitePool) -> Self {
Self { pool }
}
}
#[async_trait]
impl AuditLedgerWriter for SqliteAuditLedger {
async fn append_batch(
&self,
records: &[AuditRecordV2],
) -> Result<AuditAppendReport, AuditLedgerError> {
let prepared = prepare_batch(records)?;
let mut transaction = self
.pool
.begin_with("BEGIN IMMEDIATE")
.await
.map_err(infrastructure)?;
let existing = fetch_existing_records(&mut transaction, prepared.keys().copied()).await?;
let mut new = Vec::new();
let mut duplicates = records.len().saturating_sub(prepared.len());
for (event_id, item) in &prepared {
match existing.get(event_id) {
Some(value) if value == &item.1 => duplicates += 1,
Some(_) => return Err(AuditLedgerError::EventConflict(*event_id)),
None => new.push(item),
}
}
if !new.is_empty() {
let mut insert = QueryBuilder::<Sqlite>::new(
"INSERT INTO minco_audit_records
(event_id, tenant_scope, resource_type, resource_id, occurred_at,
recorded_at, encoded_bytes, record) ",
);
insert.push_values(&new, |mut row, item| {
row.push_bind(item.0.event_id)
.push_bind(&item.0.tenant_scope)
.push_bind(&item.0.resource.resource_type)
.push_bind(&item.0.resource.resource_id)
.push_bind(item.0.occurred_at)
.push_bind(item.0.recorded_at)
.push_bind(i64::try_from(item.2).expect("validated audit record size"))
.push_bind(&item.1);
});
insert
.build()
.execute(&mut *transaction)
.await
.map_err(infrastructure)?;
let related = new
.iter()
.flat_map(|item| {
let record = &item.0;
record
.related_resources
.iter()
.map(move |related| (record, related))
})
.collect::<Vec<_>>();
if !related.is_empty() {
let mut insert_related = QueryBuilder::<Sqlite>::new(
"INSERT INTO minco_audit_related_resources
(event_id, tenant_scope, relation, resource_type, resource_id, occurred_at) ",
);
insert_related.push_values(related, |mut row, (record, related)| {
row.push_bind(record.event_id)
.push_bind(&record.tenant_scope)
.push_bind(&related.relation)
.push_bind(&related.resource.resource_type)
.push_bind(&related.resource.resource_id)
.push_bind(record.occurred_at);
});
insert_related
.build()
.execute(&mut *transaction)
.await
.map_err(infrastructure)?;
}
}
transaction.commit().await.map_err(infrastructure)?;
Ok(AuditAppendReport {
requested: records.len(),
inserted: new.len(),
duplicates,
})
}
}
type PreparedBatch = BTreeMap<Uuid, (AuditRecordV2, String, usize)>;
fn prepare_batch(records: &[AuditRecordV2]) -> Result<PreparedBatch, AuditLedgerError> {
if records.is_empty() || records.len() > minco_plugin_audit::MAX_AUDIT_BATCH_RECORDS {
return Err(AuditLedgerError::InvalidBatch(
"invalid record count".into(),
));
}
let mut prepared = BTreeMap::new();
let mut bytes = 0usize;
for record in records {
let encoded_bytes = record.validate()?;
bytes = bytes
.checked_add(encoded_bytes)
.ok_or_else(|| AuditLedgerError::InvalidBatch("batch bytes overflow".into()))?;
if bytes > minco_plugin_audit::MAX_AUDIT_BATCH_BYTES {
return Err(AuditLedgerError::BatchTooLarge {
bytes,
maximum: minco_plugin_audit::MAX_AUDIT_BATCH_BYTES,
});
}
let json = serde_json::to_string(record).map_err(|_| AuditLedgerError::Encoding)?;
if let Some(existing) =
prepared.insert(record.event_id, (record.clone(), json, encoded_bytes))
&& existing.0 != *record
{
return Err(AuditLedgerError::EventConflict(record.event_id));
}
}
Ok(prepared)
}
async fn fetch_existing_records(
transaction: &mut Transaction<'_, Sqlite>,
ids: impl Iterator<Item = Uuid>,
) -> Result<BTreeMap<Uuid, String>, AuditLedgerError> {
let ids = ids.collect::<Vec<_>>();
if ids.is_empty() {
return Ok(BTreeMap::new());
}
let mut query = QueryBuilder::<Sqlite>::new(
"SELECT event_id, record FROM minco_audit_records WHERE event_id IN (",
);
push_uuid_list(&mut query, &ids);
query.push(")");
query
.build()
.fetch_all(&mut **transaction)
.await
.map_err(infrastructure)?
.iter()
.map(|row| {
Ok((
row.try_get("event_id").map_err(infrastructure)?,
row.try_get("record").map_err(infrastructure)?,
))
})
.collect()
}
#[async_trait]
impl AuditReader for SqliteAuditLedger {
async fn list_resource_history(
&self,
query: &AuditQuery,
) -> Result<AuditPage, AuditLedgerError> {
query.validate()?;
let mut statement = QueryBuilder::<Sqlite>::new(
"SELECT record, occurred_at, event_id FROM minco_audit_records AS audit WHERE tenant_scope = ",
);
statement
.push_bind(&query.tenant_scope)
.push(" AND ((resource_type = ")
.push_bind(&query.resource.resource_type)
.push(" AND resource_id = ")
.push_bind(&query.resource.resource_id)
.push(")");
if query.include_related {
statement.push(
" OR EXISTS (SELECT 1 FROM minco_audit_related_resources AS related
WHERE related.event_id = audit.event_id AND related.tenant_scope = ",
);
statement
.push_bind(&query.tenant_scope)
.push(" AND related.resource_type = ")
.push_bind(&query.resource.resource_type)
.push(" AND related.resource_id = ")
.push_bind(&query.resource.resource_id);
if let Some(relation) = &query.relation {
statement
.push(" AND related.relation = ")
.push_bind(relation);
}
statement.push(")");
}
statement.push(")");
if let Some(after) = query.after {
let comparator = match query.direction {
minco_plugin_audit::AuditSortDirection::OldestFirst => ">",
minco_plugin_audit::AuditSortDirection::NewestFirst => "<",
};
statement
.push(" AND (occurred_at, event_id) ")
.push(comparator)
.push(" (")
.push_bind(after.occurred_at)
.push(", ")
.push_bind(after.event_id)
.push(")");
}
let direction = match query.direction {
minco_plugin_audit::AuditSortDirection::OldestFirst => "ASC",
minco_plugin_audit::AuditSortDirection::NewestFirst => "DESC",
};
statement
.push(" ORDER BY occurred_at ")
.push(direction)
.push(", event_id ")
.push(direction)
.push(" LIMIT ")
.push_bind(
i64::try_from(query.limit + 1)
.map_err(|_| AuditLedgerError::InvalidQuery("limit".into()))?,
);
let rows = statement
.build()
.fetch_all(&self.pool)
.await
.map_err(infrastructure)?;
decode_page(rows, query.limit)
}
}
fn decode_page(
rows: Vec<sqlx::sqlite::SqliteRow>,
limit: usize,
) -> Result<AuditPage, AuditLedgerError> {
let has_more = rows.len() > limit;
let mut records = rows
.into_iter()
.take(limit)
.map(|row| {
let value: String = row.try_get("record").map_err(infrastructure)?;
let record: AuditRecordV2 =
serde_json::from_str(&value).map_err(|_| AuditLedgerError::Encoding)?;
record.validate()?;
Ok(record)
})
.collect::<Result<Vec<_>, AuditLedgerError>>()?;
let next_cursor = has_more.then(|| {
records
.last()
.map(AuditCursor::from)
.expect("positive validated query limit")
});
Ok(AuditPage {
records: std::mem::take(&mut records),
next_cursor,
})
}
#[derive(Debug, Clone)]
pub struct SqliteAuditStorageInspector {
source: SqlitePool,
ledger: SqlitePool,
policy: AuditLifecyclePolicy,
}
impl SqliteAuditStorageInspector {
pub fn new(
source: SqlitePool,
ledger: SqlitePool,
policy: AuditLifecyclePolicy,
) -> Result<Self, AuditLedgerError> {
policy.validate()?;
Ok(Self {
source,
ledger,
policy,
})
}
}
#[async_trait]
impl AuditStorageInspector for SqliteAuditStorageInspector {
async fn storage_health(&self) -> Result<AuditStorageHealth, AuditLedgerError> {
let page_count: i64 = sqlx::query_scalar("PRAGMA page_count")
.fetch_one(&self.ledger)
.await
.map_err(infrastructure)?;
let page_size: i64 = sqlx::query_scalar("PRAGMA page_size")
.fetch_one(&self.ledger)
.await
.map_err(infrastructure)?;
let hot_bytes = u64::try_from(page_count.saturating_mul(page_size))
.map_err(|_| AuditLedgerError::Infrastructure)?;
let ledger_file = main_database_file(&self.ledger).await?;
let free_bytes = fs2::available_space(ledger_file).map_err(infrastructure)?;
let row = sqlx::query(
"SELECT COUNT(*) AS pending_records,
COALESCE(SUM(encoded_bytes), 0) AS pending_bytes,
MIN(occurred_at) AS oldest_pending
FROM minco_audit_journal WHERE status IN ('pending', 'failed', 'claimed')",
)
.fetch_one(&self.source)
.await
.map_err(infrastructure)?;
let pending_records: i64 = row.try_get("pending_records").map_err(infrastructure)?;
let pending_bytes: i64 = row.try_get("pending_bytes").map_err(infrastructure)?;
let oldest_pending: Option<DateTime<Utc>> =
row.try_get("oldest_pending").map_err(infrastructure)?;
let quarantined: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM minco_audit_journal WHERE status = 'quarantined'",
)
.fetch_one(&self.source)
.await
.map_err(infrastructure)?;
let record_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM minco_audit_records")
.fetch_one(&self.ledger)
.await
.map_err(infrastructure)?;
let bounds = ledger_bounds(&self.ledger).await?;
let snapshot = AuditStorageSnapshot {
provider: "sqlite".into(),
hot_bytes,
free_bytes: Some(free_bytes),
pending_records: u64::try_from(pending_records)
.map_err(|_| AuditLedgerError::Infrastructure)?,
pending_bytes: u64::try_from(pending_bytes)
.map_err(|_| AuditLedgerError::Infrastructure)?,
oldest_pending_seconds: oldest_pending.map(|time| {
u64::try_from((Utc::now() - time).num_seconds().max(0)).unwrap_or(u64::MAX)
}),
quarantined_records: u64::try_from(quarantined)
.map_err(|_| AuditLedgerError::Infrastructure)?,
archive_watermark: None,
segments: vec![AuditSegmentStatus {
segment_id: 1,
state: AuditSegmentState::Active,
record_count: u64::try_from(record_count)
.map_err(|_| AuditLedgerError::Infrastructure)?,
encoded_bytes: hot_bytes,
first: bounds.0,
last: bounds.1,
archive_receipt: None,
}],
};
evaluate_storage_health(self.policy, snapshot)
}
}
async fn ledger_bounds(
pool: &SqlitePool,
) -> Result<(Option<AuditCursor>, Option<AuditCursor>), AuditLedgerError> {
let first = sqlx::query(
"SELECT occurred_at, event_id FROM minco_audit_records ORDER BY occurred_at, event_id LIMIT 1",
)
.fetch_optional(pool)
.await
.map_err(infrastructure)?
.map(|row| decode_cursor(&row))
.transpose()?;
let last = sqlx::query(
"SELECT occurred_at, event_id FROM minco_audit_records ORDER BY occurred_at DESC, event_id DESC LIMIT 1",
)
.fetch_optional(pool)
.await
.map_err(infrastructure)?
.map(|row| decode_cursor(&row))
.transpose()?;
Ok((first, last))
}
fn decode_cursor(row: &sqlx::sqlite::SqliteRow) -> Result<AuditCursor, AuditLedgerError> {
Ok(AuditCursor {
occurred_at: row.try_get("occurred_at").map_err(infrastructure)?,
event_id: row.try_get("event_id").map_err(infrastructure)?,
})
}
pub async fn validate_separate_audit_pools(
source: &SqlitePool,
ledger: &SqlitePool,
) -> Result<(), AuditLedgerError> {
let source_file = main_database_file(source).await?;
let ledger_file = main_database_file(ledger).await?;
if source_file.is_empty() || ledger_file.is_empty() || source_file == ledger_file {
return Err(AuditLedgerError::InvalidLifecycle(
"SQLite audit ledger requires a distinct file-backed database".into(),
));
}
Ok(())
}
async fn main_database_file(pool: &SqlitePool) -> Result<String, AuditLedgerError> {
sqlx::query("PRAGMA database_list")
.fetch_all(pool)
.await
.map_err(infrastructure)?
.into_iter()
.find(|row| row.try_get::<String, _>("name").ok().as_deref() == Some("main"))
.ok_or(AuditLedgerError::Infrastructure)?
.try_get("file")
.map_err(infrastructure)
}
pub async fn migrate_audit_ledger(pool: &SqlitePool) -> Result<(), sqlx::migrate::MigrateError> {
let mut migrator = sqlx::migrate!("migrations/audit-ledger");
migrator.dangerous_set_table_name("_minco_audit_ledger_migrations");
migrator.run(pool).await
}
fn validate_pending_entry(entry: &AuditJournalEntry) -> Result<(), AuditLedgerError> {
if entry.status != AuditJournalStatus::Pending
|| entry.encoded_bytes != entry.record.validate()?
{
Err(AuditLedgerError::InvalidJournalEntry)
} else {
Ok(())
}
}
fn validate_claim(
worker_id: &str,
limit: usize,
claim_expires_at: DateTime<Utc>,
) -> Result<(), AuditLedgerError> {
let now = Utc::now();
if worker_id.trim().is_empty()
|| worker_id.len() > 128
|| worker_id.chars().any(char::is_control)
|| limit == 0
|| limit > minco_plugin_audit::MAX_AUDIT_BATCH_RECORDS
|| claim_expires_at <= now
|| claim_expires_at > now + TimeDelta::hours(1)
{
Err(AuditLedgerError::InvalidJournalClaim)
} else {
Ok(())
}
}
fn validate_transition(event_ids: &[Uuid], worker_id: &str) -> Result<(), AuditLedgerError> {
if event_ids.is_empty()
|| event_ids.len() > minco_plugin_audit::MAX_AUDIT_BATCH_RECORDS
|| worker_id.trim().is_empty()
|| worker_id.len() > 128
|| worker_id.chars().any(char::is_control)
{
Err(AuditLedgerError::InvalidJournalClaim)
} else {
Ok(())
}
}
fn validate_failure_code(value: &str) -> Result<(), AuditLedgerError> {
if value.is_empty()
|| value.len() > 128
|| !value
.bytes()
.all(|byte| byte.is_ascii_uppercase() || byte.is_ascii_digit() || byte == b'-')
{
Err(AuditLedgerError::InvalidJournalEntry)
} else {
Ok(())
}
}
fn push_uuid_list(builder: &mut QueryBuilder<Sqlite>, ids: &[Uuid]) {
let mut separated = builder.separated(", ");
for id in ids {
separated.push_bind(*id);
}
}
fn infrastructure(_: impl std::fmt::Display) -> AuditLedgerError {
AuditLedgerError::Infrastructure
}
#[cfg(test)]
mod tests {
use super::*;
use minco_plugin_audit::{
AuditActor, AuditRelatedResource, AuditRelay, AuditResourceRef, AuditSortDirection,
};
use std::{path::PathBuf, sync::Arc};
struct Databases {
_directory: tempfile::TempDir,
source: SqlitePool,
ledger: SqlitePool,
source_path: PathBuf,
ledger_path: PathBuf,
}
impl Databases {
async fn new() -> Self {
let directory = tempfile::tempdir().unwrap();
let source_path = directory.path().join("source.sqlite");
let ledger_path = directory.path().join("audit.sqlite");
let source = crate::connect(&crate::SqlitePoolConfig::file(&source_path))
.await
.unwrap();
let ledger = crate::connect(&crate::SqlitePoolConfig::file(&ledger_path))
.await
.unwrap();
crate::plugin_adapters::migrate_plugin_storage(&source)
.await
.unwrap();
migrate_audit_ledger(&ledger).await.unwrap();
validate_separate_audit_pools(&source, &ledger)
.await
.unwrap();
Self {
_directory: directory,
source,
ledger,
source_path,
ledger_path,
}
}
}
fn record(index: u32) -> AuditRecordV2 {
let mut record = AuditRecordV2::new(
"tenant",
"order.status_changed",
AuditResourceRef::new("order", "one"),
AuditActor::human("subject"),
"updateOrder",
Uuid::now_v7(),
);
record.event_id = Uuid::from_u128(10_000 + u128::from(index));
record.occurred_at = DateTime::from_timestamp(1_800_000_000 + i64::from(index), 0).unwrap();
record.recorded_at = record.occurred_at + TimeDelta::seconds(1);
record.resource_revision = Some(u64::from(index));
record
}
#[tokio::test]
async fn source_intent_rolls_back_with_domain_mutation() {
let databases = Databases::new().await;
sqlx::query("CREATE TABLE orders (id TEXT PRIMARY KEY)")
.execute(&databases.source)
.await
.unwrap();
let journal = SqliteAuditJournal::new(databases.source.clone());
let mut transaction = databases
.source
.begin_with("BEGIN IMMEDIATE")
.await
.unwrap();
sqlx::query("INSERT INTO orders (id) VALUES ('one')")
.execute(&mut *transaction)
.await
.unwrap();
journal
.enqueue_in(
&mut transaction,
AuditJournalEntry::pending(record(1)).unwrap(),
)
.await
.unwrap();
transaction.rollback().await.unwrap();
let orders: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM orders")
.fetch_one(&databases.source)
.await
.unwrap();
let intents: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM minco_audit_journal")
.fetch_one(&databases.source)
.await
.unwrap();
assert_eq!((orders, intents), (0, 0));
}
#[tokio::test]
async fn retry_after_ledger_commit_is_deduplicated_and_acknowledged() {
let databases = Databases::new().await;
let journal = Arc::new(SqliteAuditJournal::new(databases.source.clone()));
let ledger = Arc::new(SqliteAuditLedger::new(databases.ledger.clone()));
journal
.enqueue(AuditJournalEntry::pending(record(1)).unwrap())
.await
.unwrap();
let expires = Utc::now() + TimeDelta::milliseconds(1);
let claimed = journal
.claim_pending("crashed-worker", 10, expires)
.await
.unwrap();
ledger
.append_batch(&[claimed[0].record.clone()])
.await
.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
let report = AuditRelay::new(journal, ledger)
.dispatch_once("recovery-worker", 10, TimeDelta::minutes(1))
.await
.unwrap();
assert_eq!(report.duplicates, 1);
assert_eq!(report.inserted, 0);
let pending: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM minco_audit_journal")
.fetch_one(&databases.source)
.await
.unwrap();
assert_eq!(pending, 0);
}
#[tokio::test]
async fn pages_and_related_history_are_stable_in_separate_ledger() {
let databases = Databases::new().await;
assert_ne!(databases.source_path, databases.ledger_path);
let ledger = SqliteAuditLedger::new(databases.ledger.clone());
let mut records = (1..=5).map(record).collect::<Vec<_>>();
records[2].resource = AuditResourceRef::new("shift", "shift-one");
records[2].related_resources.push(AuditRelatedResource {
relation: "order".into(),
resource: AuditResourceRef::new("order", "one"),
});
let report = ledger.append_batch(&records).await.unwrap();
assert_eq!(report.inserted, 5);
let mut query = AuditQuery::for_resource("tenant", AuditResourceRef::new("order", "one"));
query.direction = AuditSortDirection::OldestFirst;
query.include_related = true;
query.relation = Some("order".into());
query.limit = 2;
let first = ledger.list_resource_history(&query).await.unwrap();
assert_eq!(first.records.len(), 2);
query.after = first.next_cursor;
let second = ledger.list_resource_history(&query).await.unwrap();
assert_eq!(second.records.len(), 2);
query.after = second.next_cursor;
let third = ledger.list_resource_history(&query).await.unwrap();
assert_eq!(third.records.len(), 1);
let revisions = first
.records
.into_iter()
.chain(second.records)
.chain(third.records)
.map(|record| record.resource_revision.unwrap())
.collect::<Vec<_>>();
assert_eq!(revisions, vec![1, 2, 3, 4, 5]);
}
#[tokio::test]
async fn same_pool_is_rejected_and_health_reports_real_ledger_bytes() {
let databases = Databases::new().await;
assert!(
validate_separate_audit_pools(&databases.source, &databases.source)
.await
.is_err()
);
let inspector = SqliteAuditStorageInspector::new(
databases.source,
databases.ledger,
AuditLifecyclePolicy::sqlite_100_mib(64 * 1024 * 1024),
)
.unwrap();
let health = inspector.storage_health().await.unwrap();
assert_eq!(health.snapshot.provider, "sqlite");
assert!(health.snapshot.hot_bytes > 0);
assert!(health.snapshot.free_bytes.is_some_and(|bytes| bytes > 0));
}
}