fix(server): migrate legacy user info (#15621)

#### PR Dependency Tree


* **PR #15621** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- **Bug Fixes**
- Improved payment and customer portal reliability when Stripe customer
records already exist.
- Existing legacy customer records are now recognized and associated
with the correct account context.
- Prevented customer records from being used across incompatible account
contexts.
- Preserved the existing error behavior when no customer record is
available.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/backend/native/src/runtime/backend_runtime/payment/command/catalog.rs

@@ -197,15 +197,7 @@ impl PaymentRuntime {
   ) -> RuntimeResult<String> {
     validate_identity(user_id, "payment user")?;
     let namespace_key = canonical_namespace(namespace)?;
-    if let Some(customer_id) = sqlx::query_scalar::<_, String>(
-      "SELECT stripe_customer_id FROM user_stripe_customers WHERE user_id=$1 AND provider_namespace=$2",
-    )
-    .bind(user_id)
-    .bind(&namespace_key)
-    .fetch_optional(&self.pool)
-    .await
-    .map_err(|error| RuntimeError::database("load Stripe customer", error))?
-    {
+    if let Some(customer_id) = load_or_adopt_stripe_customer(&self.pool, user_id, &namespace_key).await? {
       return Ok(customer_id);
     }
     let email = email

packages/backend/native/src/runtime/backend_runtime/payment/command/mod.rs

@@ -27,7 +27,8 @@ use sqlx::Row;
 use super::{
   CustomerSnapshot, OperationCompletion, OperationIntent, PaymentConnection, PaymentFormField, PaymentFormValue,
   PaymentRuntime, PaymentScope, PaymentSendDecision, PaymentSnapshot, PaymentStep, PaymentStepState, SnapshotCoverage,
-  TrialSnapshot, freeze_operation, mark_operation_step_sent, record_operation_error, record_operation_step_result,
+  TrialSnapshot, freeze_operation, load_or_adopt_stripe_customer, mark_operation_step_sent, record_operation_error,
+  record_operation_step_result,
   snapshot::parse_lookup_key,
   stripe_client::{
     StripeCheckoutSession, StripeCustomer, StripeForm, StripeFormValue, StripePortalSession, StripePrice,

packages/backend/native/src/runtime/backend_runtime/payment/command/portal.rs

@@ -8,15 +8,9 @@ impl PaymentRuntime {
     validate_identity(actor_user_id, "payment actor")?;
     let stripe = self.stripe()?;
     let namespace_key = canonical_namespace(stripe.namespace())?;
-    let customer_id: String = sqlx::query_scalar(
-      "SELECT stripe_customer_id FROM user_stripe_customers WHERE user_id=$1 AND provider_namespace=$2",
-    )
-    .bind(actor_user_id)
-    .bind(&namespace_key)
-    .fetch_optional(&self.pool)
-    .await
-    .map_err(|error| RuntimeError::database("load Stripe portal customer", error))?
-    .ok_or_else(|| RuntimeError::invalid_state("payment_customer_not_found"))?;
+    let customer_id = load_or_adopt_stripe_customer(&self.pool, actor_user_id, &namespace_key)
+      .await?
+      .ok_or_else(|| RuntimeError::invalid_state("payment_customer_not_found"))?;
     let mut form = StripeForm::default();
     form.push("customer", StripeFormValue::Text(customer_id));
     let portal: StripePortalSession = stripe

packages/backend/native/src/runtime/backend_runtime/payment/mod.rs

@@ -42,7 +42,8 @@ use types::*;
 use worker::PaymentWorker;
 use write::{
   adopt_legacy_entitlement, complete_receipts_and_operation, entitlement_subject, expire_missing_sources,
-  upsert_customers, upsert_financial_facts, upsert_invoices, upsert_licenses, upsert_subscription, upsert_trials,
+  load_or_adopt_stripe_customer, upsert_customers, upsert_financial_facts, upsert_invoices, upsert_licenses,
+  upsert_subscription, upsert_trials,
 };
 
 #[cfg(test)]

packages/backend/native/src/runtime/backend_runtime/payment/tests.rs

@@ -1124,6 +1124,29 @@ async fn operation_intent_is_frozen_before_send_and_blocks_overlapping_work() {
       .all(|price| price.get("created") == Some(&json!(false)))
   );
   assert!(catalog_requests.recv().await.unwrap().starts_with("GET /v1/prices?"));
+
+  let legacy_user = insert_user(&pool, &format!("{account_marker}-legacy")).await;
+  let legacy_customer = format!("cus-{account_marker}-legacy");
+  sqlx::query("INSERT INTO user_stripe_customers(user_id,stripe_customer_id,provider_namespace) VALUES($1,$2,NULL)")
+    .bind(&legacy_user)
+    .bind(&legacy_customer)
+    .execute(&pool)
+    .await
+    .unwrap();
+  assert_eq!(
+    load_or_adopt_stripe_customer(&pool, &legacy_user, &account_namespace_key)
+      .await
+      .unwrap(),
+    Some(legacy_customer)
+  );
+  assert_eq!(
+    sqlx::query_scalar::<_, String>("SELECT provider_namespace FROM user_stripe_customers WHERE user_id=$1")
+      .bind(&legacy_user)
+      .fetch_one(&pool)
+      .await
+      .unwrap(),
+    account_namespace_key
+  );
   cleanup(&pool, &account_namespace_key, &account_marker).await;
 }
 

packages/backend/native/src/runtime/backend_runtime/payment/write.rs

@@ -1,13 +1,57 @@
 use std::collections::BTreeSet;
 
 use affine_core::payment::Provider;
-use sqlx::{Postgres, Row, Transaction};
+use sqlx::{PgPool, Postgres, Row, Transaction};
 
 use super::{
   super::{RuntimeError, RuntimeResult},
   PaymentSnapshot, SnapshotCoverage, StoredSubscription, SubscriptionSnapshot,
 };
 
+pub(super) async fn load_or_adopt_stripe_customer(
+  pool: &PgPool,
+  user_id: &str,
+  namespace: &str,
+) -> RuntimeResult<Option<String>> {
+  let mut tx = pool
+    .begin()
+    .await
+    .map_err(|error| RuntimeError::database("begin Stripe customer lookup", error))?;
+  let customer =
+    sqlx::query("SELECT stripe_customer_id,provider_namespace FROM user_stripe_customers WHERE user_id=$1 FOR UPDATE")
+      .bind(user_id)
+      .fetch_optional(&mut *tx)
+      .await
+      .map_err(|error| RuntimeError::database("load Stripe customer", error))?;
+  let Some(customer) = customer else {
+    tx.commit()
+      .await
+      .map_err(|error| RuntimeError::database("commit empty Stripe customer lookup", error))?;
+    return Ok(None);
+  };
+  let customer_id: String = customer.get("stripe_customer_id");
+  match customer.get::<Option<String>, _>("provider_namespace") {
+    Some(stored) if stored != namespace => {
+      return Err(RuntimeError::invalid_state(
+        "Stripe user customer belongs to another provider namespace",
+      ));
+    }
+    None => {
+      sqlx::query("UPDATE user_stripe_customers SET provider_namespace=$2 WHERE user_id=$1")
+        .bind(user_id)
+        .bind(namespace)
+        .execute(&mut *tx)
+        .await
+        .map_err(|error| RuntimeError::database("adopt legacy Stripe customer", error))?;
+    }
+    Some(_) => {}
+  }
+  tx.commit()
+    .await
+    .map_err(|error| RuntimeError::database("commit Stripe customer lookup", error))?;
+  Ok(Some(customer_id))
+}
+
 pub(super) async fn upsert_customers(
   tx: &mut Transaction<'_, Postgres>,
   snapshot: &PaymentSnapshot,