fix: reconcile pre-watermark Stripe invoices

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;

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
+    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());
+            .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;
+    }
 }