use std::time::Duration;
use chrono::{DateTime, Utc};
use sqlx::{PgPool, Row};
use syrup_rail::{
ActorId, BillingScopeId, GatewayAccountId, GatewayLifecycleAccount,
GatewayLifecycleQuarantineReason, GatewayLifecycleQuarantineResolutionReason,
};
use uuid::Uuid;
use crate::lifecycle_reconciliation::{
GatewayLifecycleReconciliationError, ensure_account, set_timeouts,
};
const MAX_REVIEW_PAGE_SIZE: i64 = 101;
const INVALID_QUARANTINE_REASON: &str = "canonical gateway lifecycle quarantine reason is invalid";
const INVALID_QUARANTINE_HISTORY: &str =
"canonical gateway lifecycle quarantine resolution history is invalid";
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct GatewayLifecycleQuarantineAlert {
unresolved_count: i64,
oldest_first_seen_at: DateTime<Utc>,
latest_last_seen_at: DateTime<Utc>,
}
impl GatewayLifecycleQuarantineAlert {
pub const fn unresolved_count(self) -> i64 {
self.unresolved_count
}
pub const fn oldest_first_seen_at(self) -> DateTime<Utc> {
self.oldest_first_seen_at
}
pub const fn latest_last_seen_at(self) -> DateTime<Utc> {
self.latest_last_seen_at
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct GatewayLifecycleQuarantineResolutionRecord {
quarantine_id: Uuid,
actor_id: ActorId,
reason: GatewayLifecycleQuarantineResolutionReason,
observed_occurrence_count: i64,
observed_last_seen_at: DateTime<Utc>,
resolved_at: DateTime<Utc>,
}
impl GatewayLifecycleQuarantineResolutionRecord {
pub const fn quarantine_id(&self) -> Uuid {
self.quarantine_id
}
pub const fn actor_id(&self) -> ActorId {
self.actor_id
}
pub const fn reason(&self) -> &GatewayLifecycleQuarantineResolutionReason {
&self.reason
}
pub const fn observed_occurrence_count(&self) -> i64 {
self.observed_occurrence_count
}
pub const fn observed_last_seen_at(&self) -> DateTime<Utc> {
self.observed_last_seen_at
}
pub const fn resolved_at(&self) -> DateTime<Utc> {
self.resolved_at
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct GatewayLifecycleQuarantineReviewRecord {
id: Uuid,
billing_scope_id: BillingScopeId,
gateway_account_id: GatewayAccountId,
transaction_id: Option<String>,
order_id: Option<String>,
reason: GatewayLifecycleQuarantineReason,
first_seen_at: DateTime<Utc>,
last_seen_at: DateTime<Utc>,
occurrence_count: i64,
resolution_count: i64,
latest_resolution: Option<GatewayLifecycleQuarantineResolutionRecord>,
}
impl GatewayLifecycleQuarantineReviewRecord {
pub const fn id(&self) -> Uuid {
self.id
}
pub const fn billing_scope_id(&self) -> BillingScopeId {
self.billing_scope_id
}
pub const fn gateway_account_id(&self) -> GatewayAccountId {
self.gateway_account_id
}
pub fn transaction_id(&self) -> Option<&str> {
self.transaction_id.as_deref()
}
pub fn order_id(&self) -> Option<&str> {
self.order_id.as_deref()
}
pub const fn reason(&self) -> GatewayLifecycleQuarantineReason {
self.reason
}
pub const fn first_seen_at(&self) -> DateTime<Utc> {
self.first_seen_at
}
pub const fn last_seen_at(&self) -> DateTime<Utc> {
self.last_seen_at
}
pub const fn occurrence_count(&self) -> i64 {
self.occurrence_count
}
pub const fn resolution_count(&self) -> i64 {
self.resolution_count
}
pub const fn latest_resolution(&self) -> Option<&GatewayLifecycleQuarantineResolutionRecord> {
self.latest_resolution.as_ref()
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum GatewayLifecycleQuarantineResolutionOutcome {
Resolved(GatewayLifecycleQuarantineResolutionRecord),
Replayed(GatewayLifecycleQuarantineResolutionRecord),
AlreadyResolved,
Stale,
NotFound,
}
pub async fn claim_gateway_lifecycle_quarantine_alert(
pool: &PgPool,
account: &GatewayLifecycleAccount,
alert_after: Duration,
) -> Result<Option<GatewayLifecycleQuarantineAlert>, GatewayLifecycleReconciliationError> {
let alert_after_seconds = i64::try_from(alert_after.as_secs()).map_err(|_| {
GatewayLifecycleReconciliationError::InvalidState(
"gateway lifecycle quarantine alert cadence is too large",
)
})?;
let mut transaction = pool.begin().await?;
set_timeouts(&mut transaction).await?;
ensure_account(&mut transaction, account).await?;
let row = sqlx::query(
r#"
WITH due AS MATERIALIZED (
SELECT EXISTS (
SELECT 1
FROM billing_gateway_lifecycle_quarantines
WHERE billing_scope_id = $1
AND gateway_account_id = $2
AND resolved_at IS NULL
AND (
last_operator_alerted_at IS NULL
OR last_operator_alerted_at
<= now() - ($3::bigint * interval '1 second')
)
) AS should_claim
),
claimed AS (
UPDATE billing_gateway_lifecycle_quarantines
SET last_operator_alerted_at = now()
WHERE billing_scope_id = $1
AND gateway_account_id = $2
AND resolved_at IS NULL
AND (SELECT should_claim FROM due)
RETURNING id
)
SELECT
COUNT(*)::bigint AS unresolved_count,
MIN(first_seen_at) AS oldest_first_seen_at,
MAX(last_seen_at) AS latest_last_seen_at
FROM billing_gateway_lifecycle_quarantines
WHERE billing_scope_id = $1
AND gateway_account_id = $2
AND resolved_at IS NULL
HAVING EXISTS (SELECT 1 FROM claimed)
"#,
)
.bind(account.billing_scope_id().as_uuid())
.bind(account.gateway_account_id().as_uuid())
.bind(alert_after_seconds)
.fetch_optional(&mut *transaction)
.await?;
transaction.commit().await?;
row.map(|row| {
Ok(GatewayLifecycleQuarantineAlert {
unresolved_count: row.try_get("unresolved_count")?,
oldest_first_seen_at: row.try_get("oldest_first_seen_at")?,
latest_last_seen_at: row.try_get("latest_last_seen_at")?,
})
})
.transpose()
}
pub async fn gateway_lifecycle_quarantine_review_page(
pool: &PgPool,
limit: i64,
cursor: Option<(DateTime<Utc>, Uuid)>,
) -> Result<Vec<GatewayLifecycleQuarantineReviewRecord>, GatewayLifecycleReconciliationError> {
if !(1..=MAX_REVIEW_PAGE_SIZE).contains(&limit) {
return Err(GatewayLifecycleReconciliationError::InvalidState(
"gateway lifecycle quarantine review page size is invalid",
));
}
let rows = sqlx::query(
r#"
WITH page AS (
SELECT quarantine.id,
quarantine.billing_scope_id,
quarantine.gateway_account_id,
quarantine.gateway_transaction_id,
quarantine.gateway_order_id,
quarantine.reason_code,
quarantine.first_seen_at,
quarantine.last_seen_at,
quarantine.occurrence_count
FROM billing_gateway_lifecycle_quarantines quarantine
WHERE quarantine.resolved_at IS NULL
AND (
$1::timestamptz IS NULL
OR (quarantine.first_seen_at, quarantine.id)
> ($1::timestamptz, $2::uuid)
)
ORDER BY quarantine.first_seen_at, quarantine.id
LIMIT $3
)
SELECT page.*,
COALESCE(history.resolution_count, 0)::bigint AS resolution_count,
history.actor_id AS latest_resolution_actor_id,
history.reason AS latest_resolution_reason,
history.observed_occurrence_count AS latest_observed_occurrence_count,
history.observed_last_seen_at AS latest_observed_last_seen_at,
history.resolved_at AS latest_resolved_at
FROM page
LEFT JOIN LATERAL (
SELECT resolution.actor_id,
resolution.reason,
resolution.observed_occurrence_count,
resolution.observed_last_seen_at,
resolution.resolved_at,
COUNT(*) OVER () AS resolution_count
FROM billing_gateway_lifecycle_quarantine_resolutions resolution
WHERE resolution.quarantine_id = page.id
ORDER BY resolution.resolved_at DESC, resolution.id DESC
LIMIT 1
) history ON TRUE
ORDER BY page.first_seen_at, page.id
"#,
)
.bind(cursor.as_ref().map(|(first_seen_at, _)| first_seen_at))
.bind(cursor.as_ref().map(|(_, id)| id))
.bind(limit)
.fetch_all(pool)
.await?;
rows.into_iter().map(review_record).collect()
}
pub async fn resolve_gateway_lifecycle_quarantine(
pool: &PgPool,
quarantine_id: Uuid,
actor_id: ActorId,
expected_occurrence_count: i64,
expected_last_seen_at: DateTime<Utc>,
reason: &GatewayLifecycleQuarantineResolutionReason,
) -> Result<GatewayLifecycleQuarantineResolutionOutcome, GatewayLifecycleReconciliationError> {
if expected_occurrence_count <= 0 {
return Err(GatewayLifecycleReconciliationError::InvalidState(
"gateway lifecycle quarantine observation is invalid",
));
}
let mut transaction = pool.begin().await?;
set_timeouts(&mut transaction).await?;
let quarantine = sqlx::query(
r#"
SELECT occurrence_count, last_seen_at, resolved_at
FROM billing_gateway_lifecycle_quarantines
WHERE id = $1
FOR UPDATE
"#,
)
.bind(quarantine_id)
.fetch_optional(&mut *transaction)
.await?;
let Some(quarantine) = quarantine else {
transaction.commit().await?;
return Ok(GatewayLifecycleQuarantineResolutionOutcome::NotFound);
};
let occurrence_count: i64 = quarantine.try_get("occurrence_count")?;
let last_seen_at: DateTime<Utc> = quarantine.try_get("last_seen_at")?;
if occurrence_count != expected_occurrence_count || last_seen_at != expected_last_seen_at {
transaction.commit().await?;
return Ok(GatewayLifecycleQuarantineResolutionOutcome::Stale);
}
if quarantine
.try_get::<Option<DateTime<Utc>>, _>("resolved_at")?
.is_some()
{
let existing = sqlx::query(
r#"
SELECT quarantine_id, actor_id, reason,
observed_occurrence_count, observed_last_seen_at, resolved_at
FROM billing_gateway_lifecycle_quarantine_resolutions
WHERE quarantine_id = $1
AND observed_occurrence_count = $2
AND observed_last_seen_at = $3
"#,
)
.bind(quarantine_id)
.bind(expected_occurrence_count)
.bind(expected_last_seen_at)
.fetch_optional(&mut *transaction)
.await?;
transaction.commit().await?;
return Ok(match existing {
Some(existing)
if existing.try_get::<Uuid, _>("actor_id")? == *actor_id.as_uuid()
&& existing.try_get::<String, _>("reason")? == reason.expose() =>
{
GatewayLifecycleQuarantineResolutionOutcome::Replayed(resolution_record(existing)?)
}
_ => GatewayLifecycleQuarantineResolutionOutcome::AlreadyResolved,
});
}
let resolution = sqlx::query(
r#"
INSERT INTO billing_gateway_lifecycle_quarantine_resolutions (
quarantine_id, actor_id, reason,
observed_occurrence_count, observed_last_seen_at
) VALUES ($1, $2, $3, $4, $5)
RETURNING quarantine_id, actor_id, reason,
observed_occurrence_count, observed_last_seen_at, resolved_at
"#,
)
.bind(quarantine_id)
.bind(actor_id.as_uuid())
.bind(reason.expose())
.bind(expected_occurrence_count)
.bind(expected_last_seen_at)
.fetch_one(&mut *transaction)
.await?;
let resolved_at: DateTime<Utc> = resolution.try_get("resolved_at")?;
let updated = sqlx::query(
r#"
UPDATE billing_gateway_lifecycle_quarantines
SET resolved_at = $2
WHERE id = $1
AND resolved_at IS NULL
AND occurrence_count = $3
AND last_seen_at = $4
"#,
)
.bind(quarantine_id)
.bind(resolved_at)
.bind(expected_occurrence_count)
.bind(expected_last_seen_at)
.execute(&mut *transaction)
.await?;
if updated.rows_affected() != 1 {
return Err(GatewayLifecycleReconciliationError::InvalidState(
"locked gateway lifecycle quarantine did not accept its operator resolution",
));
}
transaction.commit().await?;
Ok(GatewayLifecycleQuarantineResolutionOutcome::Resolved(
resolution_record(resolution)?,
))
}
fn review_record(
row: sqlx::postgres::PgRow,
) -> Result<GatewayLifecycleQuarantineReviewRecord, GatewayLifecycleReconciliationError> {
let latest_resolution = match (
row.try_get::<Option<Uuid>, _>("latest_resolution_actor_id")?,
row.try_get::<Option<String>, _>("latest_resolution_reason")?,
row.try_get::<Option<i64>, _>("latest_observed_occurrence_count")?,
row.try_get::<Option<DateTime<Utc>>, _>("latest_observed_last_seen_at")?,
row.try_get::<Option<DateTime<Utc>>, _>("latest_resolved_at")?,
) {
(Some(actor_id), Some(reason), Some(count), Some(last_seen_at), Some(resolved_at)) => {
Some(GatewayLifecycleQuarantineResolutionRecord {
quarantine_id: row.try_get("id")?,
actor_id: ActorId::new(actor_id),
reason: GatewayLifecycleQuarantineResolutionReason::new(reason).map_err(|_| {
GatewayLifecycleReconciliationError::InvalidState(INVALID_QUARANTINE_HISTORY)
})?,
observed_occurrence_count: count,
observed_last_seen_at: last_seen_at,
resolved_at,
})
}
(None, None, None, None, None) => None,
_ => {
return Err(GatewayLifecycleReconciliationError::InvalidState(
INVALID_QUARANTINE_HISTORY,
));
}
};
let resolution_count: i64 = row.try_get("resolution_count")?;
if (resolution_count == 0) != latest_resolution.is_none() {
return Err(GatewayLifecycleReconciliationError::InvalidState(
INVALID_QUARANTINE_HISTORY,
));
}
Ok(GatewayLifecycleQuarantineReviewRecord {
id: row.try_get("id")?,
billing_scope_id: BillingScopeId::new(row.try_get("billing_scope_id")?),
gateway_account_id: GatewayAccountId::new(row.try_get("gateway_account_id")?),
transaction_id: row.try_get("gateway_transaction_id")?,
order_id: row.try_get("gateway_order_id")?,
reason: quarantine_reason(&row.try_get::<String, _>("reason_code")?)?,
first_seen_at: row.try_get("first_seen_at")?,
last_seen_at: row.try_get("last_seen_at")?,
occurrence_count: row.try_get("occurrence_count")?,
resolution_count,
latest_resolution,
})
}
fn resolution_record(
row: sqlx::postgres::PgRow,
) -> Result<GatewayLifecycleQuarantineResolutionRecord, GatewayLifecycleReconciliationError> {
Ok(GatewayLifecycleQuarantineResolutionRecord {
quarantine_id: row.try_get("quarantine_id")?,
actor_id: ActorId::new(row.try_get("actor_id")?),
reason: GatewayLifecycleQuarantineResolutionReason::new(
row.try_get::<String, _>("reason")?,
)
.map_err(|_| {
GatewayLifecycleReconciliationError::InvalidState(INVALID_QUARANTINE_HISTORY)
})?,
observed_occurrence_count: row.try_get("observed_occurrence_count")?,
observed_last_seen_at: row.try_get("observed_last_seen_at")?,
resolved_at: row.try_get("resolved_at")?,
})
}
fn quarantine_reason(
value: &str,
) -> Result<GatewayLifecycleQuarantineReason, GatewayLifecycleReconciliationError> {
match value {
"ambiguous_reversal_success" => {
Ok(GatewayLifecycleQuarantineReason::AmbiguousReversalSuccess)
}
"invalid_refund_economics" => Ok(GatewayLifecycleQuarantineReason::InvalidRefundEconomics),
"malformed_report_structure" => {
Ok(GatewayLifecycleQuarantineReason::MalformedReportStructure)
}
_ => Err(GatewayLifecycleReconciliationError::InvalidState(
INVALID_QUARANTINE_REASON,
)),
}
}
#[cfg(test)]
mod tests {
use std::error::Error;
use super::*;
use crate::{
record_gateway_lifecycle_quarantines,
test_support::{TestDatabase, create_gateway_account},
};
use syrup_rail::{GatewayLifecycleQuarantine, GatewayProviderKey};
#[tokio::test]
async fn alert_review_resolution_replay_and_reopen_are_exact_and_auditable()
-> Result<(), Box<dyn Error>> {
let database = TestDatabase::start("rail_quarantine").await?;
let fixture = create_gateway_account(&database.pool, "nmi").await?;
let account = GatewayLifecycleAccount::new(
BillingScopeId::new(fixture.billing_scope_id),
GatewayAccountId::new(fixture.gateway_account_id),
GatewayProviderKey::new("nmi")?,
);
let quarantine = GatewayLifecycleQuarantine::new(
Some(syrup_rail::GatewayTransactionId::new("txn-review")?),
None,
GatewayLifecycleQuarantineReason::MalformedReportStructure,
)?;
record_gateway_lifecycle_quarantines(
&database.pool,
&account,
std::slice::from_ref(&quarantine),
)
.await?;
let alert = claim_gateway_lifecycle_quarantine_alert(
&database.pool,
&account,
Duration::from_secs(3_600),
)
.await?
.expect("first unresolved alert is due");
assert_eq!(alert.unresolved_count(), 1);
assert!(
claim_gateway_lifecycle_quarantine_alert(
&database.pool,
&account,
Duration::from_secs(3_600),
)
.await?
.is_none()
);
let page = gateway_lifecycle_quarantine_review_page(&database.pool, 10, None).await?;
assert_eq!(page.len(), 1);
let incident = &page[0];
assert_eq!(incident.billing_scope_id(), account.billing_scope_id());
assert_eq!(incident.gateway_account_id(), account.gateway_account_id());
assert_eq!(incident.transaction_id(), Some("txn-review"));
assert_eq!(
incident.reason(),
GatewayLifecycleQuarantineReason::MalformedReportStructure
);
assert_eq!(incident.resolution_count(), 0);
let actor = ActorId::new(Uuid::now_v7());
let reason = GatewayLifecycleQuarantineResolutionReason::new("provider report reviewed")?;
let resolved = resolve_gateway_lifecycle_quarantine(
&database.pool,
incident.id(),
actor,
incident.occurrence_count(),
incident.last_seen_at(),
&reason,
)
.await?;
let GatewayLifecycleQuarantineResolutionOutcome::Resolved(resolution) = resolved else {
panic!("expected a new resolution");
};
assert_eq!(resolution.actor_id(), actor);
assert_eq!(resolution.reason(), &reason);
assert!(
gateway_lifecycle_quarantine_review_page(&database.pool, 10, None)
.await?
.is_empty()
);
assert!(matches!(
resolve_gateway_lifecycle_quarantine(
&database.pool,
incident.id(),
actor,
incident.occurrence_count(),
incident.last_seen_at(),
&reason,
)
.await?,
GatewayLifecycleQuarantineResolutionOutcome::Replayed(_)
));
record_gateway_lifecycle_quarantines(&database.pool, &account, &[quarantine]).await?;
let reopened = gateway_lifecycle_quarantine_review_page(&database.pool, 10, None).await?;
assert_eq!(reopened.len(), 1);
assert_eq!(reopened[0].resolution_count(), 1);
assert_eq!(reopened[0].latest_resolution(), Some(&resolution));
assert!(matches!(
resolve_gateway_lifecycle_quarantine(
&database.pool,
reopened[0].id(),
actor,
incident.occurrence_count(),
incident.last_seen_at(),
&reason,
)
.await?,
GatewayLifecycleQuarantineResolutionOutcome::Stale
));
let wrong_provider = GatewayLifecycleAccount::new(
account.billing_scope_id(),
account.gateway_account_id(),
GatewayProviderKey::new("other")?,
);
assert!(matches!(
claim_gateway_lifecycle_quarantine_alert(
&database.pool,
&wrong_provider,
Duration::ZERO,
)
.await,
Err(GatewayLifecycleReconciliationError::AccountNotFound)
));
database.cleanup().await?;
Ok(())
}
}