minco_sqlx_sqlite/

audit_v2.rs

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));
    }
}