Merge remote-tracking branch 'origin/pr/604'

crates/aether-data/src/repository/wallet/memory.rs

@@ -769,6 +769,41 @@ impl WalletReadRepository for InMemoryWalletRepository {
             .cloned())
     }
 
+    async fn find_pending_plan_purchase_order_by_user_id(
+        &self,
+        user_id: &str,
+        product_id: &str,
+    ) -> Result<Option<StoredAdminPaymentOrder>, DataLayerError> {
+        let now = current_unix_secs();
+        Ok(self
+            .payment_orders_by_id
+            .read()
+            .expect("wallet repo lock")
+            .values()
+            .filter(|order| {
+                order.user_id.as_deref() == Some(user_id)
+                    && order.status == "pending"
+                    && order
+                        .expires_at_unix_secs
+                        .is_some_and(|expires_at| expires_at > now)
+                    && order
+                        .gateway_response
+                        .as_ref()
+                        .is_some_and(|gateway_response| {
+                            gateway_response
+                                .get("order_kind")
+                                .and_then(serde_json::Value::as_str)
+                                == Some("plan_purchase")
+                                && gateway_response
+                                    .get("product_id")
+                                    .and_then(serde_json::Value::as_str)
+                                    == Some(product_id)
+                        })
+            })
+            .max_by_key(|order| order.created_at_unix_ms)
+            .cloned())
+    }
+
     async fn find_wallet_refund(
         &self,
         wallet_id: &str,

crates/aether-data/src/repository/wallet/mysql.rs

@@ -536,6 +536,32 @@ WHERE wallet_id = ?
             .await
     }
 
+    async fn find_pending_plan_purchase_order_by_user_id(
+        &self,
+        user_id: &str,
+        product_id: &str,
+    ) -> Result<Option<StoredAdminPaymentOrder>, DataLayerError> {
+        let sql = payment_order_select_sql(
+            r#"
+WHERE user_id = ?
+  AND product_id = ?
+  AND order_kind = 'plan_purchase'
+  AND status = 'pending'
+  AND expires_at > ?
+ORDER BY created_at DESC
+LIMIT 1
+"#,
+        );
+        let row = sqlx::query(&sql)
+            .bind(user_id)
+            .bind(product_id)
+            .bind(current_unix_secs_i64())
+            .fetch_optional(&self.pool)
+            .await
+            .map_sql_err()?;
+        row.as_ref().map(map_payment_order_row).transpose()
+    }
+
     async fn find_wallet_refund(
         &self,
         wallet_id: &str,

crates/aether-data/src/repository/wallet/postgres.rs

@@ -577,6 +577,41 @@ WHERE user_id = $1
 LIMIT 1
 "#;
 
+const FIND_PENDING_PLAN_PURCHASE_ORDER_BY_USER_SQL: &str = r#"
+SELECT
+  id,
+  order_no,
+  wallet_id,
+  user_id,
+  CAST(amount_usd AS DOUBLE PRECISION) AS amount_usd,
+  CAST(pay_amount AS DOUBLE PRECISION) AS pay_amount,
+  pay_currency,
+  CAST(exchange_rate AS DOUBLE PRECISION) AS exchange_rate,
+  CAST(refunded_amount_usd AS DOUBLE PRECISION) AS refunded_amount_usd,
+  CAST(refundable_amount_usd AS DOUBLE PRECISION) AS refundable_amount_usd,
+  payment_method,
+  payment_provider,
+  payment_channel,
+  order_kind,
+  product_id,
+  product_snapshot,
+  gateway_order_id,
+  gateway_response,
+  status,
+  CAST(EXTRACT(EPOCH FROM created_at) AS BIGINT) AS created_at_unix_ms,
+  CAST(EXTRACT(EPOCH FROM paid_at) AS BIGINT) AS paid_at_unix_secs,
+  CAST(EXTRACT(EPOCH FROM credited_at) AS BIGINT) AS credited_at_unix_secs,
+  CAST(EXTRACT(EPOCH FROM expires_at) AS BIGINT) AS expires_at_unix_secs
+FROM payment_orders
+WHERE user_id = $1
+  AND product_id = $2
+  AND order_kind = 'plan_purchase'
+  AND status = 'pending'
+  AND expires_at > NOW()
+ORDER BY created_at DESC
+LIMIT 1
+"#;
+
 const FIND_WALLET_REFUND_SQL: &str = r#"
 SELECT
   id,
@@ -1112,6 +1147,20 @@ impl WalletReadRepository for SqlxWalletRepository {
         row.as_ref().map(map_admin_payment_order_row).transpose()
     }
 
+    async fn find_pending_plan_purchase_order_by_user_id(
+        &self,
+        user_id: &str,
+        product_id: &str,
+    ) -> Result<Option<StoredAdminPaymentOrder>, DataLayerError> {
+        let row = sqlx::query(FIND_PENDING_PLAN_PURCHASE_ORDER_BY_USER_SQL)
+            .bind(user_id)
+            .bind(product_id)
+            .fetch_optional(&self.pool)
+            .await
+            .map_postgres_err()?;
+        row.as_ref().map(map_admin_payment_order_row).transpose()
+    }
+
     async fn find_wallet_refund(
         &self,
         wallet_id: &str,

crates/aether-data/src/repository/wallet/sqlite.rs

@@ -783,6 +783,32 @@ LIMIT 1
         row.as_ref().map(map_payment_order_row).transpose()
     }
 
+    async fn find_pending_plan_purchase_order_by_user_id(
+        &self,
+        user_id: &str,
+        product_id: &str,
+    ) -> Result<Option<StoredAdminPaymentOrder>, DataLayerError> {
+        let sql = payment_order_select_sql(
+            r#"
+WHERE user_id = ?
+  AND product_id = ?
+  AND order_kind = 'plan_purchase'
+  AND status = 'pending'
+  AND expires_at > ?
+ORDER BY created_at DESC
+LIMIT 1
+"#,
+        );
+        let row = sqlx::query(&sql)
+            .bind(user_id)
+            .bind(product_id)
+            .bind(current_unix_secs_i64())
+            .fetch_optional(&self.pool)
+            .await
+            .map_sql_err()?;
+        row.as_ref().map(map_payment_order_row).transpose()
+    }
+
     async fn find_wallet_refund(
         &self,
         wallet_id: &str,
@@ -5460,6 +5486,175 @@ INSERT INTO billing_plans (
         assert_eq!(wallet_balance, 0.0);
     }
 
+    #[tokio::test]
+    async fn sqlite_finds_reusable_pending_plan_purchase_order() {
+        let pool = sqlx::sqlite::SqlitePoolOptions::new()
+            .max_connections(1)
+            .connect("sqlite::memory:")
+            .await
+            .expect("sqlite pool should connect");
+        run_sqlite_migrations(&pool)
+            .await
+            .expect("sqlite migrations should run");
+
+        let repository = SqliteWalletReadRepository::new(pool);
+        sqlx::query(
+            "INSERT INTO users (id, username, email, auth_source, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)",
+        )
+        .bind("user-pending-plan-1")
+        .bind("Pending Buyer")
+        .bind("pending-plan@example.com")
+        .bind("local")
+        .bind(1_i64)
+        .bind(1_i64)
+        .execute(repository.pool())
+        .await
+        .expect("user should seed");
+
+        let _wallet_order = match repository
+            .create_wallet_recharge_order(CreateWalletRechargeOrderInput {
+                preferred_wallet_id: Some("wallet-pending-plan-1".to_string()),
+                user_id: "user-pending-plan-1".to_string(),
+                amount_usd: 1.0,
+                pay_amount: Some(1.0),
+                pay_currency: Some("USD".to_string()),
+                exchange_rate: Some(1.0),
+                payment_method: "bootstrap".to_string(),
+                payment_provider: None,
+                payment_channel: None,
+                gateway_order_id: "gateway-bootstrap-pending-plan-1".to_string(),
+                gateway_response: json!({ "bootstrap": true }),
+                order_no: "order-bootstrap-pending-plan-1".to_string(),
+                expires_at_unix_secs: 4_102_444_800,
+            })
+            .await
+            .expect("wallet should be created")
+        {
+            CreateWalletRechargeOrderOutcome::Created(order) => order,
+            CreateWalletRechargeOrderOutcome::WalletInactive => {
+                panic!("new wallet should be active")
+            }
+        };
+
+        let plan_snapshot = json!({
+            "id": "pending-plan",
+            "title": "每日额度月卡",
+            "duration_unit": "month",
+            "duration_value": 1,
+            "max_active_per_user": 1,
+            "purchase_limit_scope": "active_period",
+            "entitlements": [
+                {
+                    "type": "daily_quota",
+                    "daily_quota_usd": 50.0,
+                    "reset_timezone": "Asia/Shanghai",
+                    "allow_wallet_overage": false
+                }
+            ]
+        });
+        let pending_order = match repository
+            .create_plan_purchase_order(CreatePlanPurchaseOrderInput {
+                preferred_wallet_id: None,
+                user_id: "user-pending-plan-1".to_string(),
+                amount_usd: 13.8,
+                pay_amount: 100.0,
+                pay_currency: "CNY".to_string(),
+                exchange_rate: 7.24637681,
+                payment_method: "alipay".to_string(),
+                payment_provider: Some("epay".to_string()),
+                payment_channel: Some("alipay".to_string()),
+                gateway_order_id: "gateway-pending-plan-1".to_string(),
+                gateway_response: json!({ "checkout": true }),
+                order_no: "order-pending-plan-1".to_string(),
+                product_id: "pending-plan".to_string(),
+                product_snapshot: plan_snapshot.clone(),
+                expires_at_unix_secs: 4_102_444_800,
+            })
+            .await
+            .expect("pending plan order should create")
+        {
+            CreatePlanPurchaseOrderOutcome::Created(order) => order,
+            other => panic!("pending plan order should be created, got {other:?}"),
+        };
+        let now = chrono::Utc::now().timestamp().max(0);
+        for (id, order_no, status, product_id, user_id, expires_at, created_at) in [
+            (
+                "expired-pending-plan-order",
+                "order-expired-pending-plan",
+                "pending",
+                "pending-plan",
+                "user-pending-plan-1",
+                now - 10,
+                now + 10,
+            ),
+            (
+                "credited-pending-plan-order",
+                "order-credited-pending-plan",
+                "credited",
+                "pending-plan",
+                "user-pending-plan-1",
+                now + 3_600,
+                now + 20,
+            ),
+            (
+                "other-user-pending-plan-order",
+                "order-other-user-pending-plan",
+                "pending",
+                "pending-plan",
+                "other-user",
+                now + 3_600,
+                now + 30,
+            ),
+        ] {
+            sqlx::query(
+                r#"
+INSERT INTO payment_orders (
+  id, order_no, wallet_id, user_id, amount_usd, pay_amount, pay_currency,
+  exchange_rate, refunded_amount_usd, refundable_amount_usd, payment_method,
+  payment_provider, payment_channel, order_kind, product_id, product_snapshot,
+  fulfillment_status, gateway_order_id, gateway_response, status, created_at, expires_at
+) VALUES (?, ?, ?, ?, 13.8, 100.0, 'CNY', 7.24637681, 0, 0, 'alipay',
+  'epay', 'alipay', 'plan_purchase', ?, ?, 'pending', ?, ?, ?, ?, ?)
+                "#,
+            )
+            .bind(id)
+            .bind(order_no)
+            .bind("wallet-pending-plan-1")
+            .bind(user_id)
+            .bind(product_id)
+            .bind(plan_snapshot.to_string())
+            .bind(format!("gateway-{id}"))
+            .bind(json!({ "checkout": id }).to_string())
+            .bind(status)
+            .bind(created_at)
+            .bind(expires_at)
+            .execute(repository.pool())
+            .await
+            .expect("extra payment order should seed");
+        }
+
+        let found = repository
+            .find_pending_plan_purchase_order_by_user_id("user-pending-plan-1", "pending-plan")
+            .await
+            .expect("pending plan lookup should run")
+            .expect("pending plan order should be found");
+        assert_eq!(found.id, pending_order.id);
+        assert_eq!(
+            repository
+                .find_pending_plan_purchase_order_by_user_id("user-pending-plan-1", "missing-plan")
+                .await
+                .expect("missing plan lookup should run"),
+            None
+        );
+        assert_eq!(
+            repository
+                .find_pending_plan_purchase_order_by_user_id("missing-user", "pending-plan")
+                .await
+                .expect("missing user lookup should run"),
+            None
+        );
+    }
+
     #[tokio::test]
     async fn sqlite_plan_purchase_replaces_same_class_entitlements_on_manual_credit() {
         let pool = sqlx::sqlite::SqlitePoolOptions::new()

crates/aether-data/src/repository/wallet/types.rs

@@ -927,6 +927,12 @@ pub trait WalletReadRepository: Send + Sync {
         order_id: &str,
     ) -> Result<Option<StoredAdminPaymentOrder>, crate::DataLayerError>;
 
+    async fn find_pending_plan_purchase_order_by_user_id(
+        &self,
+        user_id: &str,
+        product_id: &str,
+    ) -> Result<Option<StoredAdminPaymentOrder>, crate::DataLayerError>;
+
     async fn find_wallet_refund(
         &self,
         wallet_id: &str,