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::{PgPool, Postgres, QueryBuilder, Row, Transaction, postgres::PgRow};
use std::collections::{BTreeMap, BTreeSet};
use uuid::Uuid;
#[derive(Debug, Clone)]
pub struct PostgresAuditJournal {
pool: PgPool,
}
impl PostgresAuditJournal {
pub const fn new(pool: PgPool) -> Self {
Self { pool }
}
pub async fn enqueue_in(
&self,
transaction: &mut Transaction<'_, Postgres>,
entry: AuditJournalEntry,
) -> Result<(), AuditLedgerError> {
validate_pending_entry(&entry)?;
let record = serde_json::to_value(&entry.record).map_err(|_| AuditLedgerError::Encoding)?;
let encoded_bytes = i32::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 ($1, $2, $3, $4, 'pending', $5, $6, $7, $8, $9)
ON CONFLICT(event_id) DO NOTHING",
)
.bind(entry.record.event_id)
.bind(entry.record.occurred_at)
.bind(&record)
.bind(encoded_bytes)
.bind(
i32::try_from(entry.attempt_count)
.map_err(|_| AuditLedgerError::InvalidJournalEntry)?,
)
.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: serde_json::Value =
sqlx::query_scalar("SELECT record FROM minco_audit_journal WHERE event_id = $1")
.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 result = match transition {
JournalTransition::Delivered => {
sqlx::query(
"DELETE FROM minco_audit_journal
WHERE event_id = ANY($1) AND status = 'claimed' AND claimed_by = $2",
)
.bind(event_ids)
.bind(worker_id)
.execute(&self.pool)
.await
}
JournalTransition::Retry {
failure_code,
retry_at,
} => {
validate_failure_code(failure_code)?;
sqlx::query(
"UPDATE minco_audit_journal
SET status = 'failed', available_at = $3, failure_code = $4,
claimed_by = NULL, claim_expires_at = NULL
WHERE event_id = ANY($1) AND status = 'claimed' AND claimed_by = $2",
)
.bind(event_ids)
.bind(worker_id)
.bind(retry_at)
.bind(failure_code)
.execute(&self.pool)
.await
}
JournalTransition::Quarantine { failure_code } => {
validate_failure_code(failure_code)?;
sqlx::query(
"UPDATE minco_audit_journal
SET status = 'quarantined', failure_code = $3,
claimed_by = NULL, claim_expires_at = NULL
WHERE event_id = ANY($1) AND status = 'claimed' AND claimed_by = $2",
)
.bind(event_ids)
.bind(worker_id)
.bind(failure_code)
.execute(&self.pool)
.await
}
}
.map_err(infrastructure)?;
if usize::try_from(result.rows_affected()).ok() != Some(event_ids.len()) {
return Err(AuditLedgerError::JournalClaimLost);
}
Ok(())
}
}
enum JournalTransition<'a> {
Delivered,
Retry {
failure_code: &'a str,
retry_at: DateTime<Utc>,
},
Quarantine {
failure_code: &'a str,
},
}
#[async_trait]
impl AuditJournalStore for PostgresAuditJournal {
async fn enqueue(&self, entry: AuditJournalEntry) -> Result<(), AuditLedgerError> {
let mut transaction = self.pool.begin().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)?;
sqlx::query(
"WITH claimable AS (
SELECT event_id FROM minco_audit_journal
WHERE status IN ('pending', 'failed') AND available_at <= $1
ORDER BY available_at, occurred_at, event_id
FOR UPDATE SKIP LOCKED LIMIT $2
)
UPDATE minco_audit_journal AS journal
SET status = 'claimed', claimed_by = $3, claim_expires_at = $4,
attempt_count = journal.attempt_count + 1
FROM claimable
WHERE journal.event_id = claimable.event_id
RETURNING journal.*",
)
.bind(Utc::now())
.bind(limit)
.bind(worker_id)
.bind(claim_expires_at)
.fetch_all(&self.pool)
.await
.map_err(infrastructure)?
.iter()
.map(decode_journal_entry)
.collect()
}
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 = $1, claimed_by = NULL,
claim_expires_at = NULL, failure_code = 'AUDIT-CLAIM-EXPIRED'
WHERE status = 'claimed' AND claim_expires_at <= $1",
)
.bind(now)
.execute(&self.pool)
.await
.map_err(infrastructure)?;
usize::try_from(result.rows_affected()).map_err(|_| AuditLedgerError::Infrastructure)
}
}
fn decode_journal_entry(row: &PgRow) -> Result<AuditJournalEntry, AuditLedgerError> {
let value: serde_json::Value = row.try_get("record").map_err(infrastructure)?;
let status: String = row.try_get("status").map_err(infrastructure)?;
let record: AuditRecordV2 =
serde_json::from_value(value).map_err(|_| AuditLedgerError::Encoding)?;
let encoded_bytes: i32 = row.try_get("encoded_bytes").map_err(infrastructure)?;
let attempt_count: i32 = 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 PostgresAuditLedger {
pool: PgPool,
}
impl PostgresAuditLedger {
pub const fn new(pool: PgPool) -> Self {
Self { pool }
}
}
#[async_trait]
impl AuditLedgerWriter for PostgresAuditLedger {
async fn append_batch(
&self,
records: &[AuditRecordV2],
) -> Result<AuditAppendReport, AuditLedgerError> {
let prepared = prepare_batch(records)?;
let mut transaction = self.pool.begin().await.map_err(infrastructure)?;
let ids = prepared.keys().copied().collect::<Vec<_>>();
let existing = fetch_existing_records(&mut transaction, &ids).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),
}
}
let inserted = if new.is_empty() {
BTreeSet::new()
} else {
let inserted = insert_records(&mut transaction, &new).await?;
if inserted.len() != new.len() {
let raced_ids = new
.iter()
.filter(|item| !inserted.contains(&item.0.event_id))
.map(|item| item.0.event_id)
.collect::<Vec<_>>();
let raced = fetch_existing_records(&mut transaction, &raced_ids).await?;
for item in &new {
if !inserted.contains(&item.0.event_id)
&& raced.get(&item.0.event_id) != Some(&item.1)
{
return Err(AuditLedgerError::EventConflict(item.0.event_id));
}
}
duplicates += raced_ids.len();
}
insert_related_records(&mut transaction, &new, &inserted).await?;
inserted
};
transaction.commit().await.map_err(infrastructure)?;
Ok(AuditAppendReport {
requested: records.len(),
inserted: inserted.len(),
duplicates,
})
}
}
type PreparedBatch = BTreeMap<Uuid, (AuditRecordV2, serde_json::Value, 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 value = serde_json::to_value(record).map_err(|_| AuditLedgerError::Encoding)?;
if let Some(existing) =
prepared.insert(record.event_id, (record.clone(), value, encoded_bytes))
&& existing.0 != *record
{
return Err(AuditLedgerError::EventConflict(record.event_id));
}
}
Ok(prepared)
}
async fn fetch_existing_records(
transaction: &mut Transaction<'_, Postgres>,
ids: &[Uuid],
) -> Result<BTreeMap<Uuid, serde_json::Value>, AuditLedgerError> {
sqlx::query("SELECT event_id, record FROM minco_audit_records WHERE event_id = ANY($1)")
.bind(ids)
.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 fn insert_records(
transaction: &mut Transaction<'_, Postgres>,
records: &[&(AuditRecordV2, serde_json::Value, usize)],
) -> Result<BTreeSet<Uuid>, AuditLedgerError> {
let mut insert = QueryBuilder::<Postgres>::new(
"INSERT INTO minco_audit_records
(event_id, tenant_scope, resource_type, resource_id, occurred_at,
recorded_at, encoded_bytes, record) ",
);
insert.push_values(records, |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(i32::try_from(item.2).expect("validated audit record size"))
.push_bind(&item.1);
});
insert.push(" ON CONFLICT(event_id) DO NOTHING RETURNING event_id");
insert
.build()
.fetch_all(&mut **transaction)
.await
.map_err(infrastructure)?
.iter()
.map(|row| row.try_get("event_id").map_err(infrastructure))
.collect()
}
async fn insert_related_records(
transaction: &mut Transaction<'_, Postgres>,
records: &[&(AuditRecordV2, serde_json::Value, usize)],
inserted: &BTreeSet<Uuid>,
) -> Result<(), AuditLedgerError> {
let related = records
.iter()
.filter(|item| inserted.contains(&item.0.event_id))
.flat_map(|item| {
item.0
.related_resources
.iter()
.map(move |related| (&item.0, related))
})
.collect::<Vec<_>>();
if related.is_empty() {
return Ok(());
}
let mut insert = QueryBuilder::<Postgres>::new(
"INSERT INTO minco_audit_related_resources
(event_id, tenant_scope, relation, resource_type, resource_id, occurred_at) ",
);
insert.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
.build()
.execute(&mut **transaction)
.await
.map_err(infrastructure)?;
Ok(())
}
#[async_trait]
impl AuditReader for PostgresAuditLedger {
async fn list_resource_history(
&self,
query: &AuditQuery,
) -> Result<AuditPage, AuditLedgerError> {
query.validate()?;
let mut statement = QueryBuilder::<Postgres>::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<PgRow>, limit: usize) -> Result<AuditPage, AuditLedgerError> {
let has_more = rows.len() > limit;
let records = rows
.into_iter()
.take(limit)
.map(|row| {
let value: serde_json::Value = row.try_get("record").map_err(infrastructure)?;
let record: AuditRecordV2 =
serde_json::from_value(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,
next_cursor,
})
}
#[derive(Debug, Clone)]
pub struct PostgresAuditStorageInspector {
source: PgPool,
ledger: PgPool,
policy: AuditLifecyclePolicy,
}
impl PostgresAuditStorageInspector {
pub fn new(
source: PgPool,
ledger: PgPool,
policy: AuditLifecyclePolicy,
) -> Result<Self, AuditLedgerError> {
policy.validate()?;
Ok(Self {
source,
ledger,
policy,
})
}
}
#[async_trait]
impl AuditStorageInspector for PostgresAuditStorageInspector {
async fn storage_health(&self) -> Result<AuditStorageHealth, AuditLedgerError> {
let hot_bytes: i64 = sqlx::query_scalar(
"SELECT pg_total_relation_size('minco_audit_records')
+ pg_total_relation_size('minco_audit_related_resources')",
)
.fetch_one(&self.ledger)
.await
.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 hot_bytes = u64::try_from(hot_bytes).map_err(|_| AuditLedgerError::Infrastructure)?;
let snapshot = AuditStorageSnapshot {
provider: "postgresql".into(),
hot_bytes,
free_bytes: None,
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: &PgPool,
) -> 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: &PgRow) -> 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: &PgPool,
ledger: &PgPool,
) -> Result<(), AuditLedgerError> {
let source_identity = database_identity(source).await?;
let ledger_identity = database_identity(ledger).await?;
if source_identity == ledger_identity {
return Err(AuditLedgerError::InvalidLifecycle(
"PostgreSQL audit ledger requires a distinct database".into(),
));
}
Ok(())
}
async fn database_identity(
pool: &PgPool,
) -> Result<(String, Option<String>, Option<i32>), AuditLedgerError> {
let row = sqlx::query(
"SELECT current_database() AS database,
inet_server_addr()::text AS server_address,
inet_server_port() AS server_port",
)
.fetch_one(pool)
.await
.map_err(infrastructure)?;
Ok((
row.try_get("database").map_err(infrastructure)?,
row.try_get("server_address").map_err(infrastructure)?,
row.try_get("server_port").map_err(infrastructure)?,
))
}
pub async fn migrate_audit_ledger(pool: &PgPool) -> 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 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::sync::{Arc, OnceLock};
fn test_lock() -> &'static tokio::sync::Mutex<()> {
static LOCK: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| tokio::sync::Mutex::new(()))
}
async fn databases() -> Option<(PgPool, PgPool)> {
let source_url = std::env::var("MINCO_TEST_POSTGRES_URL").ok()?;
let ledger_url = std::env::var("MINCO_TEST_POSTGRES_AUDIT_URL").ok()?;
let source = PgPool::connect(&source_url).await.ok()?;
let ledger = PgPool::connect(&ledger_url).await.ok()?;
crate::plugin_adapters::migrate_plugin_storage(&source)
.await
.ok()?;
migrate_audit_ledger(&ledger).await.ok()?;
validate_separate_audit_pools(&source, &ledger).await.ok()?;
Some((source, ledger))
}
fn record() -> 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::now_v7();
record
}
#[tokio::test]
async fn transaction_and_crash_retry_are_behavioral_when_two_databases_are_configured() {
let _guard = test_lock().lock().await;
let Some((source, ledger)) = databases().await else {
eprintln!(
"MINCO_TEST_POSTGRES_URL and MINCO_TEST_POSTGRES_AUDIT_URL must name distinct databases; PostgreSQL audit proof skipped"
);
return;
};
sqlx::query("DELETE FROM minco_audit_journal")
.execute(&source)
.await
.unwrap();
sqlx::query("DELETE FROM minco_audit_related_resources")
.execute(&ledger)
.await
.unwrap();
sqlx::query("DELETE FROM minco_audit_records")
.execute(&ledger)
.await
.unwrap();
let journal = Arc::new(PostgresAuditJournal::new(source));
let ledger = Arc::new(PostgresAuditLedger::new(ledger));
let rolled_back = record();
let mut transaction = journal.pool.begin().await.unwrap();
journal
.enqueue_in(
&mut transaction,
AuditJournalEntry::pending(rolled_back.clone()).unwrap(),
)
.await
.unwrap();
transaction.rollback().await.unwrap();
let rollback_count: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM minco_audit_journal WHERE event_id = $1")
.bind(rolled_back.event_id)
.fetch_one(&journal.pool)
.await
.unwrap();
assert_eq!(rollback_count, 0);
let record = record();
journal
.enqueue(AuditJournalEntry::pending(record).unwrap())
.await
.unwrap();
let claimed = journal
.claim_pending(
"crashed-worker",
10,
Utc::now() + TimeDelta::milliseconds(1),
)
.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.inserted, report.duplicates), (0, 1));
}
#[tokio::test]
async fn concurrent_claims_and_relationship_query_are_behavioral_when_configured() {
let _guard = test_lock().lock().await;
let Some((source, ledger_pool)) = databases().await else {
eprintln!(
"MINCO_TEST_POSTGRES_URL and MINCO_TEST_POSTGRES_AUDIT_URL must name distinct databases; PostgreSQL audit proof skipped"
);
return;
};
sqlx::query("DELETE FROM minco_audit_journal")
.execute(&source)
.await
.unwrap();
sqlx::query("DELETE FROM minco_audit_related_resources")
.execute(&ledger_pool)
.await
.unwrap();
sqlx::query("DELETE FROM minco_audit_records")
.execute(&ledger_pool)
.await
.unwrap();
let journal = Arc::new(PostgresAuditJournal::new(source));
let ledger = PostgresAuditLedger::new(ledger_pool);
let first = record();
let second = record();
journal
.enqueue(AuditJournalEntry::pending(first).unwrap())
.await
.unwrap();
journal
.enqueue(AuditJournalEntry::pending(second).unwrap())
.await
.unwrap();
let lease = Utc::now() + TimeDelta::minutes(1);
let (claim_a, claim_b) = tokio::join!(
journal.claim_pending("worker-a", 1, lease),
journal.claim_pending("worker-b", 1, lease)
);
let claim_a = claim_a.unwrap();
let claim_b = claim_b.unwrap();
assert_eq!((claim_a.len(), claim_b.len()), (1, 1));
assert_ne!(claim_a[0].record.event_id, claim_b[0].record.event_id);
let records = claim_a
.iter()
.chain(&claim_b)
.map(|entry| entry.record.clone())
.collect::<Vec<_>>();
ledger.append_batch(&records).await.unwrap();
journal
.mark_delivered(&[claim_a[0].record.event_id], "worker-a")
.await
.unwrap();
journal
.mark_delivered(&[claim_b[0].record.event_id], "worker-b")
.await
.unwrap();
let mut related = record();
related.resource = AuditResourceRef::new("shift", "one");
related.related_resources.push(AuditRelatedResource {
relation: "order".into(),
resource: AuditResourceRef::new("order", "related-one"),
});
ledger.append_batch(&[related]).await.unwrap();
let mut query =
AuditQuery::for_resource("tenant", AuditResourceRef::new("order", "related-one"));
query.include_related = true;
query.relation = Some("order".into());
query.direction = AuditSortDirection::OldestFirst;
let page = ledger.list_resource_history(&query).await.unwrap();
assert_eq!(page.records.len(), 1);
assert_eq!(page.records[0].resource.resource_type, "shift");
}
}