diff --git a/migrations/020_existing_invoice_watermarks.sql b/migrations/020_existing_invoice_watermarks.sql new file mode 100644 index 0000000..6f3bb06 --- /dev/null +++ b/migrations/020_existing_invoice_watermarks.sql @@ -0,0 +1,31 @@ +-- Existing Stripe invoices predate object watermarks. Force the first later +-- event to reconcile against Stripe instead of treating it as authoritative. +INSERT INTO provider_object_event_watermarks ( + provider, object_type, provider_object_id, + last_event_created, last_event_rank, last_event_id, + is_deleted, requires_reconciliation, reconciliation_reason, + updated_at +) +SELECT + 'stripe', 'invoice', provider_invoice_id, + 0, 0, 'reconcile:migration:020', + false, true, 'migration_020_existing_invoice', + NOW() +FROM invoices +WHERE provider = 'stripe' + AND provider_invoice_id IS NOT NULL +ON CONFLICT (provider, object_type, provider_object_id) DO NOTHING; + +-- A non-paid invoice must never retain the timestamp from an older paid +-- payload. Clean historical contradictions before enforcing the invariant. +UPDATE invoices +SET paid_at = NULL +WHERE status <> 'paid' + AND paid_at IS NOT NULL; + +ALTER TABLE invoices + ADD CONSTRAINT invoices_paid_at_status_check + CHECK (paid_at IS NULL OR status = 'paid') NOT VALID; + +ALTER TABLE invoices + VALIDATE CONSTRAINT invoices_paid_at_status_check; diff --git a/src/api/webhooks.rs b/src/api/webhooks.rs index 974c14c..3173221 100644 --- a/src/api/webhooks.rs +++ b/src/api/webhooks.rs @@ -1150,10 +1150,13 @@ async fn resolve_invoice( .get("period_end") .and_then(|value| value.as_i64()) .and_then(|timestamp| Utc.timestamp_opt(timestamp, 0).single()); - let paid_at = object - .pointer("/status_transitions/paid_at") - .and_then(|value| value.as_i64()) - .and_then(|timestamp| Utc.timestamp_opt(timestamp, 0).single()); + let paid_at = (status == "paid").then(|| { + object + .pointer("/status_transitions/paid_at") + .and_then(|value| value.as_i64()) + .and_then(|timestamp| Utc.timestamp_opt(timestamp, 0).single()) + }); + let paid_at = paid_at.flatten(); Ok(ResolvedInvoice { provider_invoice_id: provider_invoice_id.to_string(), @@ -1200,7 +1203,11 @@ async fn write_invoice( pdf_url = COALESCE(EXCLUDED.pdf_url, invoices.pdf_url), period_start = COALESCE(EXCLUDED.period_start, invoices.period_start), period_end = COALESCE(EXCLUDED.period_end, invoices.period_end), - paid_at = COALESCE(EXCLUDED.paid_at, invoices.paid_at) + paid_at = CASE + WHEN EXCLUDED.status = 'paid' + THEN COALESCE(EXCLUDED.paid_at, invoices.paid_at) + ELSE NULL + END "#, ) .bind(invoice.user_id) @@ -2216,4 +2223,177 @@ mod tests { ); assert!(index_exists.is_none(), "unique index was partially applied"); } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[ignore = "requires isolated IMAGEFORGE_TEST_DATABASE_URL and IMAGEFORGE_TEST_REDIS_URL with CREATE DATABASE"] + async fn existing_paid_invoice_reconciles_before_accepting_a_later_event() { + let database_url = std::env::var("IMAGEFORGE_TEST_DATABASE_URL") + .expect("IMAGEFORGE_TEST_DATABASE_URL must be set"); + assert!( + database_url.to_ascii_lowercase().contains("test"), + "refusing to run destructive integration test outside a test database" + ); + let redis_url = std::env::var("IMAGEFORGE_TEST_REDIS_URL") + .expect("IMAGEFORGE_TEST_REDIS_URL must be set"); + let mut admin_url = url::Url::parse(&database_url).expect("parse test database URL"); + admin_url.set_path("/postgres"); + let admin_pool = PgPoolOptions::new() + .max_connections(1) + .connect(admin_url.as_str()) + .await + .expect("connect PostgreSQL admin database"); + let marker = Uuid::new_v4().simple().to_string(); + let child_database = format!("imageforge_test_invoice_seed_{marker}"); + sqlx::query(&format!("CREATE DATABASE {child_database}")) + .execute(&admin_pool) + .await + .expect("create invoice seed test database"); + + let mut child_url = url::Url::parse(&database_url).expect("parse child database URL"); + child_url.set_path(&format!("/{child_database}")); + let child_pool = PgPoolOptions::new() + .max_connections(8) + .connect(child_url.as_str()) + .await + .expect("connect invoice seed test database"); + let all_migrations = sqlx::migrate!(); + let through_019 = sqlx::migrate::Migrator { + migrations: std::borrow::Cow::Owned( + all_migrations + .iter() + .filter(|migration| migration.version <= 19) + .cloned() + .collect(), + ), + ignore_missing: false, + locking: true, + no_tx: false, + }; + through_019 + .run(&child_pool) + .await + .expect("run migrations through 019"); + + let user_id = Uuid::new_v4(); + let customer_id = format!("cus_{marker}_existing_invoice"); + let invoice_id = format!("inv_{marker}_existing_paid"); + let invoice_number = format!("INV-{marker}-EXISTING"); + sqlx::query( + r#" + INSERT INTO users ( + id, email, username, password_hash, billing_customer_id + ) VALUES ($1, $2, $3, 'test-only', $4) + "#, + ) + .bind(user_id) + .bind(format!("invoice-seed-{marker}@example.test")) + .bind(format!("invoice_seed_{marker}")) + .bind(&customer_id) + .execute(&child_pool) + .await + .expect("insert existing invoice user"); + sqlx::query( + r#" + INSERT INTO invoices ( + user_id, invoice_number, status, provider, provider_invoice_id, paid_at + ) VALUES ($1, $2, 'paid', 'stripe', $3, to_timestamp($4)) + "#, + ) + .bind(user_id) + .bind(&invoice_number) + .bind(&invoice_id) + .bind(1_700_020_000_i64) + .execute(&child_pool) + .await + .expect("insert existing paid invoice"); + + let pre_migration_watermarks: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM provider_object_event_watermarks WHERE object_type = 'invoice' AND provider_object_id = $1", + ) + .bind(&invoice_id) + .fetch_one(&child_pool) + .await + .expect("count pre-migration invoice watermarks"); + assert_eq!(pre_migration_watermarks, 0); + + all_migrations + .run(&child_pool) + .await + .expect("run invoice seed correction migration"); + let seeded: (bool, String) = sqlx::query_as( + r#" + SELECT requires_reconciliation, last_event_id + FROM provider_object_event_watermarks + WHERE provider = 'stripe' + AND object_type = 'invoice' + AND provider_object_id = $1 + "#, + ) + .bind(&invoice_id) + .fetch_one(&child_pool) + .await + .expect("query seeded invoice watermark"); + assert!(seeded.0); + assert_eq!(seeded.1, "reconcile:migration:020"); + + let (stripe_base_url, stripe_mock, stripe_mock_task) = spawn_stripe_snapshot_mock().await; + let authoritative = invoice_event( + &format!("evt_{marker}_snapshot"), + "invoice.paid", + 1_700_020_000, + &invoice_id, + &customer_id, + Some(&invoice_number), + ); + stripe_mock + .objects + .write() + .await + .insert(invoice_id.clone(), authoritative.data.object); + let state = build_test_state( + child_pool.clone(), + child_url.to_string(), + redis_url, + stripe_base_url, + ) + .await; + let stale_failure = invoice_event( + &format!("evt_{marker}_stale_failure"), + "invoice.payment_failed", + 1_700_010_000, + &invoice_id, + &customer_id, + Some(&invoice_number), + ); + apply_test_event(&state, &stale_failure) + .await + .expect("reconcile existing invoice before applying stale event"); + assert_invoice_once(&child_pool, &invoice_id, "paid", &invoice_number).await; + let watermark: (i16, bool, String) = sqlx::query_as( + r#" + SELECT last_event_rank, requires_reconciliation, last_event_id + FROM provider_object_event_watermarks + WHERE provider = 'stripe' + AND object_type = 'invoice' + AND provider_object_id = $1 + "#, + ) + .bind(&invoice_id) + .fetch_one(&child_pool) + .await + .expect("query reconciled invoice watermark"); + assert_eq!(watermark.0, 100); + assert!(!watermark.1); + assert!(watermark.2.starts_with("snapshot:")); + assert_eq!(stripe_mock.calls.load(Ordering::SeqCst), 1); + + drop(state); + stripe_mock_task.abort(); + child_pool.close().await; + sqlx::query(&format!("DROP DATABASE {child_database} WITH (FORCE)")) + .execute(&admin_pool) + .await + .expect("drop invoice seed test database"); + admin_pool.close().await; + } }