# SubscriptionApplicationService.java

    package com.yumu.noveltranslator.application.service;

    import com.stripe.model.Event;
    import com.yumu.noveltranslator.domain.model.StripeCustomer;
    import com.yumu.noveltranslator.domain.model.StripeSubscription;
    import com.yumu.noveltranslator.domain.model.User;
    import com.yumu.noveltranslator.domain.model.UserPlanHistory;
    import com.yumu.noveltranslator.enums.BillingCycle;
    import com.yumu.noveltranslator.enums.SubscriptionPlan;
    import com.yumu.noveltranslator.port.dto.subscription.CheckoutSessionRequest;
    import com.yumu.noveltranslator.port.dto.subscription.CheckoutSessionResponse;
    import com.yumu.noveltranslator.port.dto.subscription.PaymentVerificationResponse;
    import com.yumu.noveltranslator.port.dto.subscription.PortalSessionResponse;
    import com.yumu.noveltranslator.port.dto.subscription.SubscriptionStatusResponse;
    import com.yumu.noveltranslator.port.out.BillingRepositoryPort;
    import com.yumu.noveltranslator.port.out.PaymentPort;
    import com.yumu.noveltranslator.port.out.TokenRevocationPort;
    import com.yumu.noveltranslator.port.out.UserRepositoryPort;
    import com.yumu.noveltranslator.port.out.payment.CustomerInfo;
    import com.yumu.noveltranslator.port.out.payment.PaymentSessionInfo;
    import com.yumu.noveltranslator.port.out.payment.SubscriptionInfo;
    import com.yumu.noveltranslator.port.out.payment.SubscriptionUpdateRequest;
    import com.yumu.noveltranslator.properties.StripeProperties;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.dao.DuplicateKeyException;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.PlatformTransactionManager;
    import org.springframework.transaction.TransactionDefinition;
    import org.springframework.transaction.support.TransactionSynchronization;
    import org.springframework.transaction.support.TransactionSynchronizationManager;
    import org.springframework.transaction.support.TransactionTemplate;

    import java.time.Instant;
    import java.time.LocalDateTime;
    import java.time.ZoneId;
    import java.util.Map;
    import java.util.Optional;

    /**
     * 订阅应用服务
     *
     * <p>事务边界原则：Stripe HTTP 调用在事务外执行（通过 PaymentPort），DB 操作通过 TransactionTemplate 编程式事务执行。
     */
    @Slf4j
    @Service
    public class SubscriptionApplicationService implements com.yumu.noveltranslator.port.in.SubscriptionPort {

        private final StripeProperties stripeProperties;
        private final BillingRepositoryPort billingPort;
        private final UserRepositoryPort userRepositoryPort;
        private final StringRedisTemplate stringRedisTemplate;
        private final TokenRevocationPort tokenRevocationPort;
        private final PaymentPort paymentPort;
        private final PlatformTransactionManager transactionManager;

        private final TransactionTemplate tx;

        public SubscriptionApplicationService(StripeProperties stripeProperties,
                                   BillingRepositoryPort billingPort,
                                   UserRepositoryPort userRepositoryPort,
                                   StringRedisTemplate stringRedisTemplate,
                                   TokenRevocationPort tokenRevocationPort,
                                   PaymentPort paymentPort,
                                   PlatformTransactionManager transactionManager) {
            this.stripeProperties = stripeProperties;
            this.billingPort = billingPort;
            this.userRepositoryPort = userRepositoryPort;
            this.stringRedisTemplate = stringRedisTemplate;
            this.tokenRevocationPort = tokenRevocationPort;
            this.paymentPort = paymentPort;
            this.transactionManager = transactionManager;

            TransactionDefinition def = new TransactionDefinition() {
                @Override public int getPropagationBehavior() { return TransactionDefinition.PROPAGATION_REQUIRED; }
                @Override public int getIsolationLevel() { return TransactionDefinition.ISOLATION_DEFAULT; }
                @Override public int getTimeout() { return 30; }
                @Override public boolean isReadOnly() { return false; }
                @Override public String getName() { return "subscription-tx"; }
            };
            this.tx = new TransactionTemplate(transactionManager, def);
        }

        // ==================== 用户端 API ====================

        /**
         * 验证 Checkout Session 支付结果（前端回调时主动查询）
         */
        public PaymentVerificationResponse verifyCheckoutSession(String sessionId, Long userId) {
            if (sessionId == null || sessionId.isBlank()) {
                return new PaymentVerificationResponse(false, sessionId, null, null, "缺少 session_id 参数");
            }

            try {
                PaymentSessionInfo session = paymentPort.retrieveCheckoutSession(sessionId);

                String metadataUserId = Optional.ofNullable(session.metadata())
                    .map(m -> m.get("userId"))
                    .orElse(null);
                if (metadataUserId == null || !metadataUserId.equals(String.valueOf(userId))) {
                    return new PaymentVerificationResponse(false, sessionId, null, null, "支付会话不属于当前用户");
                }

                String paymentStatus = session.paymentStatus();
                String plan = Optional.ofNullable(session.metadata())
                    .map(m -> m.get("plan"))
                    .orElse(null);

                if ("paid".equals(paymentStatus)) {
                    StripeSubscription sub = Optional.ofNullable(session.subscriptionId())
                        .map(billingPort::findSubscriptionByStripeId)
                        .orElse(null);

                    if (sub != null) {
                        return new PaymentVerificationResponse(true, sessionId, plan, sub.getStatus(), "支付成功，订阅已激活");
                    }
                    return new PaymentVerificationResponse(true, sessionId, plan, "pending", "支付已确认，订阅正在激活中");
                } else if ("unpaid".equals(paymentStatus)) {
                    return new PaymentVerificationResponse(false, sessionId, plan, "unpaid", "支付尚未完成");
                } else if ("no_payment_required".equals(paymentStatus)) {
                    return new PaymentVerificationResponse(false, sessionId, plan, "no_payment_required", "无需支付");
                }

                return new PaymentVerificationResponse(false, sessionId, plan, paymentStatus, "支付状态: " + paymentStatus);
            } catch (Exception e) {
                log.error("Failed to verify checkout session {}: {}", sessionId, e.getMessage(), e);
                return new PaymentVerificationResponse(false, sessionId, null, null, "无法验证支付状态");
            }
        }

        /**
         * 创建订阅支付会话
         */
        public CheckoutSessionResponse createCheckoutSession(Long userId, CheckoutSessionRequest request) {
            SubscriptionPlan plan = validatePlan(request.getPlan());
            BillingCycle billingCycle = validateBillingCycle(request.getBillingCycle());
            String priceId = getPriceId(plan, billingCycle);

            // 1. Stripe HTTP 调用在事务外
            StripeCustomer customer = getOrCreateCustomer(userId);
            StripeSubscription existingSub = billingPort.findActiveSubscriptionByUserId(userId);

            // 2. 已有活跃订阅 → 升级现有订阅
            if (existingSub != null) {
                return upgradeSubscription(userId, existingSub, priceId, plan, billingCycle);
            }

            // 3. 无活跃订阅 → 创建新 Checkout Session
            try {
                String url = paymentPort.createCheckoutSession(
                    customer.getStripeCustomerId(), priceId,
                    stripeProperties.getSuccessUrl(), stripeProperties.getCancelUrl());
                return new CheckoutSessionResponse(url);
            } catch (Exception e) {
                log.error("Failed to create Stripe Checkout Session for user {}: {}", userId, e.getMessage(), e);
                throw new RuntimeException("创建支付会话失败", e);
            }
        }

        /**
         * 升级现有订阅
         */
        private CheckoutSessionResponse upgradeSubscription(Long userId,
                                                             StripeSubscription existingSub,
                                                             String newPriceId,
                                                             SubscriptionPlan newPlan,
                                                             BillingCycle billingCycle) {
            // Stripe HTTP 调用在事务外
            SubscriptionUpdateRequest request = new SubscriptionUpdateRequest(
                existingSub.getStripeSubscriptionId(), newPriceId, null);

            SubscriptionInfo updated;
            try {
                // Retrieve to get the item ID
                SubscriptionInfo subInfo = paymentPort.retrieveSubscription(existingSub.getStripeSubscriptionId());
                String oldItemId = subInfo.firstItemId();
                if (oldItemId == null) {
                    throw new IllegalStateException("Subscription " + existingSub.getStripeSubscriptionId() + " has no items");
                }

                updated = paymentPort.updateSubscription(existingSub.getStripeSubscriptionId(),
                    new SubscriptionUpdateRequest(oldItemId, newPriceId, null));
            } catch (Exception e) {
                log.error("Failed to upgrade subscription for user {}: {}", userId, e.getMessage(), e);
                throw new RuntimeException("升级订阅失败", e);
            }

            // DB 操作在编程式事务内
            tx.execute(status -> {
                doUpgradeSubscription(existingSub, updated, newPriceId, newPlan, billingCycle);
                return null;
            });

            return new CheckoutSessionResponse(null, true);
        }

        /**
         * 获取当前订阅状态
         */
        public SubscriptionStatusResponse getSubscriptionStatus(Long userId) {
            var subscriptions = billingPort.findSubscriptionsByUserId(userId);
            StripeSubscription sub = subscriptions.isEmpty() ? null : subscriptions.get(0);

            if (sub == null) {
                return new SubscriptionStatusResponse("FREE", "none", null, false);
            }

            return new SubscriptionStatusResponse(
                sub.getPlan(),
                sub.getStatus(),
                sub.getCurrentPeriodEnd(),
                Boolean.TRUE.equals(sub.getCancelAtPeriodEnd())
            );
        }

        /**
         * 取消订阅（在周期结束时取消）
         */
        public SubscriptionStatusResponse cancelSubscription(Long userId) {
            StripeSubscription sub = billingPort.findActiveSubscriptionByUserId(userId);

            if (sub == null) {
                throw new RuntimeException("没有可取消的活跃订阅");
            }

            // Stripe HTTP 调用在事务外
            try {
                paymentPort.updateSubscription(sub.getStripeSubscriptionId(),
                    new SubscriptionUpdateRequest(null, null, true));
            } catch (Exception e) {
                log.error("Failed to cancel Stripe subscription {}: {}", sub.getStripeSubscriptionId(), e.getMessage(), e);
                throw new RuntimeException("取消订阅失败", e);
            }

            // DB 操作在编程式事务内
            return tx.execute(status -> doCancelSubscription(sub));
        }

        /**
         * 创建 Portal Session（账单管理跳转）
         */
        public PortalSessionResponse createPortalSession(Long userId) {
            StripeCustomer customer = billingPort.findCustomerByUserIdAndNotDeleted(userId);

            if (customer == null) {
                throw new RuntimeException("未找到 Stripe 客户");
            }

            String url = paymentPort.createBillingPortalSession(customer.getStripeCustomerId(), stripeProperties.getCancelUrl());
            return new PortalSessionResponse(url);
        }

        // ==================== Webhook 事件处理 ====================

        public void handleCheckoutSessionCompleted(Event event) {
            SessionInfo session = deserializeSessionInfo(event, "checkout.session.completed");
            if (session == null) {
                throw new IllegalStateException("Failed to deserialize checkout.session.completed event: " + event.getId());
            }

            Map<String, String> metadata = session.metadata();
            String userIdStr = metadata != null ? metadata.get("userId") : null;
            String planStr = metadata != null ? metadata.get("plan") : null;
            String billingCycleStr = metadata != null ? metadata.get("billingCycle") : null;

            if (userIdStr == null || planStr == null) {
                throw new IllegalStateException("Missing required metadata in session " + session.id() + ": userId=" + userIdStr + ", plan=" + planStr);
            }

            Long userId = Long.parseLong(userIdStr);
            SubscriptionPlan plan = validatePlan(planStr);
            BillingCycle billingCycle = BillingCycle.fromValue(billingCycleStr != null ? billingCycleStr : "monthly");

            String subscriptionId = session.subscriptionId();
            if (subscriptionId == null) {
                throw new IllegalStateException("No subscription in session " + session.id());
            }

            // Stripe HTTP 调用在事务外
            SubscriptionInfo stripeSub;
            StripeCustomer customer;
            try {
                stripeSub = paymentPort.retrieveSubscription(subscriptionId);
                customer = getOrCreateCustomer(userId);
            } catch (Exception e) {
                log.error("checkout.session.completed: Stripe API error for session {}: {}", session.id(), e.getMessage(), e);
                throw new RuntimeException("Stripe API call failed for session " + session.id(), e);
            }

            // DB 操作在编程式事务内
            tx.execute(status -> {
                doHandleCheckoutSessionCompleted(event, userId, plan, billingCycle, subscriptionId, stripeSub, customer);
                return null;
            });
        }

        public void handleSubscriptionUpdated(Event event) {
            SubscriptionInfo stripeSub = deserializeSubscriptionInfo(event, "subscription.updated");
            if (stripeSub == null) {
                throw new IllegalStateException("Failed to deserialize subscription.updated event: " + event.getId());
            }

            String subscriptionId = stripeSub.id();
            StripeSubscription subRecord = billingPort.findSubscriptionByStripeId(subscriptionId);

            if (subRecord == null) {
                log.warn("subscription.updated: no local record for subscription {}", subscriptionId);
                return;
            }

            tx.execute(status -> {
                long eventCreated = event.getCreated();
                int rows = billingPort.atomicUpdateSubscription(subRecord.getId(), event.getId(), stripeSub.status(), eventCreated);

                if (rows == 0) {
                    log.info("subscription.updated: already processed event {}, skipping", event.getId());
                    return null;
                }

                // 更新时间字段
                LocalDateTime newPeriodStart = toLocalDateTime(stripeSub.currentPeriodStart());
                LocalDateTime newPeriodEnd = toLocalDateTime(stripeSub.currentPeriodEnd());
                LocalDateTime newCanceledAt = toLocalDateTime(stripeSub.canceledAt());
                if (newPeriodStart != null || newPeriodEnd != null || newCanceledAt != null) {
                    billingPort.updateSubscriptionFields(subRecord.getId(), newPeriodStart, newPeriodEnd, newCanceledAt);
                }

                // 同步 userLevel（通过 afterCommit 保证 Redis 在 DB 提交后写入）
                scheduleAfterCommit(() -> syncUserLevelAfterStatusChange(subRecord, stripeSub.status(), event));
                return null;
            });
        }

        public void handleSubscriptionDeleted(Event event) {
            SubscriptionInfo stripeSub = deserializeSubscriptionInfo(event, "subscription.deleted");
            if (stripeSub == null) {
                throw new IllegalStateException("Failed to deserialize subscription.deleted event: " + event.getId());
            }

            String subscriptionId = stripeSub.id();
            StripeSubscription subRecord = billingPort.findSubscriptionByStripeId(subscriptionId);

            if (subRecord == null) {
                throw new IllegalStateException("subscription.deleted: no local record for subscription " + subscriptionId);
            }

            tx.execute(status -> {
                long eventCreated = event.getCreated();
                int rows = billingPort.atomicUpdateSubscription(subRecord.getId(), event.getId(), "canceled", eventCreated);

                if (rows == 0) {
                    log.info("subscription.deleted: already processed event {}, skipping", event.getId());
                    return null;
                }

                scheduleAfterCommit(() -> updateUserLevel(subRecord.getUserId(), "FREE", "subscription.deleted", event.getId()));
                return null;
            });
        }

        public void handleSubscriptionResumed(Event event) {
            SubscriptionInfo stripeSub = deserializeSubscriptionInfo(event, "subscription.resumed");
            if (stripeSub == null) {
                throw new IllegalStateException("Failed to deserialize subscription.resumed event: " + event.getId());
            }

            String subscriptionId = stripeSub.id();
            StripeSubscription subRecord = billingPort.findSubscriptionByStripeId(subscriptionId);

            if (subRecord == null) {
                throw new IllegalStateException("subscription.resumed: no local record for subscription " + subscriptionId);
            }

            tx.execute(status -> {
                long eventCreated = event.getCreated();
                int rows = billingPort.atomicUpdateSubscription(subRecord.getId(), event.getId(), stripeSub.status(), eventCreated);

                if (rows == 0) {
                    log.info("subscription.resumed: already processed event {}, skipping", event.getId());
                }
                return null;
            });
        }

        public void handleInvoicePaymentFailed(Event event) {
            com.stripe.model.Invoice invoice = deserializeStripeObject(event, com.stripe.model.Invoice.class, "invoice.payment_failed");
            if (invoice == null) {
                throw new IllegalStateException("Failed to deserialize invoice.payment_failed event: " + event.getId());
            }

            String subscriptionId = invoice.getSubscription();
            if (subscriptionId == null) {
                log.warn("invoice.payment_failed: no subscription in invoice {}", event.getId());
                return;
            }

            StripeSubscription subRecord = billingPort.findSubscriptionByStripeId(subscriptionId);

            if (subRecord != null) {
                tx.execute(status -> {
                    long eventCreated = event.getCreated();
                    int rows = billingPort.atomicUpdateSubscription(subRecord.getId(), event.getId(), "past_due", eventCreated);

                    if (rows == 0) {
                        log.info("invoice.payment_failed: already processed event {}, skipping", event.getId());
                    } else {
                        log.warn("invoice.payment_failed: subscription {} marked past_due (grace period applies)", subscriptionId);
                    }
                    return null;
                });
            } else {
                log.warn("invoice.payment_failed: no local record for subscription {}", subscriptionId);
            }
        }

        public void handleInvoicePaymentSucceeded(Event event) {
            com.stripe.model.Invoice invoice = deserializeStripeObject(event, com.stripe.model.Invoice.class, "invoice.payment_succeeded");
            if (invoice == null) {
                throw new IllegalStateException("Failed to deserialize invoice.payment_succeeded event: " + event.getId());
            }

            String subscriptionId = invoice.getSubscription();
            if (subscriptionId == null) {
                log.info("invoice.payment_succeeded: no subscription in invoice {}, skipping", event.getId());
                return;
            }

            StripeSubscription subRecord = billingPort.findSubscriptionByStripeId(subscriptionId);

            // 已有记录且状态已是 active/trialing → 已由 checkout.session.completed 处理
            if (subRecord != null
                && ("active".equals(subRecord.getStatus()) || "trialing".equals(subRecord.getStatus()))) {
                log.info("invoice.payment_succeeded: subscription {} already active, skipping (handled by checkout.session.completed)",
                    subscriptionId);
                return;
            }

            // 已有记录但状态不是 active → 用原子更新激活
            if (subRecord != null) {
                tx.execute(status -> {
                    doActivateSubscriptionFromInvoice(event, subRecord);
                    return null;
                });
                return;
            }

            // 无本地记录 → 从孤立发票创建订阅（Stripe HTTP 在事务外，DB 在事务内）
            createSubscriptionFromOrphanedInvoice(event, invoice, subscriptionId);
        }

        /**
         * 从孤立发票创建订阅记录（Stripe HTTP 调用在事务外）
         */
        private void createSubscriptionFromOrphanedInvoice(Event event,
                                                            com.stripe.model.Invoice invoice,
                                                            String subscriptionId) {
            String userIdStr = invoice.getMetadata() != null ? invoice.getMetadata().get("userId") : null;
            if (userIdStr == null) {
                throw new IllegalStateException("invoice.payment_succeeded: orphaned invoice " + event.getId()
                    + " has no userId in metadata, cannot create subscription");
            }

            Long userId = Long.parseLong(userIdStr);

            StripeCustomer customer;
            SubscriptionInfo stripeSub;
            try {
                customer = getOrCreateCustomer(userId);
                stripeSub = paymentPort.retrieveSubscription(subscriptionId);
            } catch (Exception e) {
                log.error("invoice.payment_succeeded: Stripe API error for orphaned invoice {}: {}",
                    event.getId(), e.getMessage(), e);
                throw new RuntimeException("Stripe API call failed for orphaned invoice " + event.getId(), e);
            }

            tx.execute(status -> {
                doCreateSubscriptionFromOrphanedInvoice(event, userId, subscriptionId, stripeSub, customer, invoice);
                return null;
            });
        }

        // ==================== 窄事务 DB 操作方法（由 TransactionTemplate 包裹） ====================

        void doHandleCheckoutSessionCompleted(Event event, Long userId, SubscriptionPlan plan,
                                               BillingCycle billingCycle, String subscriptionId,
                                               SubscriptionInfo stripeSub, StripeCustomer customer) {
            StripeSubscription subRecord = billingPort.findSubscriptionByStripeId(subscriptionId);

            if (subRecord != null && event.getId().equals(subRecord.getLastWebhookEventId())) {
                log.info("checkout.session.completed: already processed event {}, skipping", event.getId());
                return;
            }

            if (subRecord == null) {
                subRecord = new StripeSubscription();
                subRecord.setUserId(userId);
                subRecord.setStripeCustomerId(customer.getStripeCustomerId());
                subRecord.setStripeSubscriptionId(subscriptionId);
                subRecord.setPlan(plan.getValue());
                subRecord.setBillingCycle(billingCycle.getValue());

                subRecord.setStripePriceId(stripeSub.priceId());
                subRecord.setStatus(stripeSub.status());
                subRecord.setCancelAtPeriodEnd(stripeSub.cancelAtPeriodEnd());
                subRecord.setCurrentPeriodStart(toLocalDateTime(stripeSub.currentPeriodStart()));
                subRecord.setCurrentPeriodEnd(toLocalDateTime(stripeSub.currentPeriodEnd()));
                subRecord.setLastWebhookEventId(event.getId());
                subRecord.setLastEventCreated(event.getCreated());

                boolean insertSucceeded = false;
                try {
                    billingPort.saveSubscription(subRecord);
                    insertSucceeded = true;
                } catch (DuplicateKeyException e) {
                    subRecord = billingPort.findSubscriptionByStripeId(subscriptionId);
                    if (subRecord == null) {
                        throw new IllegalStateException("DuplicateKeyException but re-query returned null for subscriptionId " + subscriptionId, e);
                    }
                    log.info("checkout.session.completed: duplicate insert caught, re-queried existing record for subscriptionId {}", subscriptionId);
                }

                if (!insertSucceeded) {
                    int claimed = billingPort.claimWebhookEvent(subRecord.getId(), event.getId());
                    if (claimed == 0) {
                        log.info("checkout.session.completed: event {} already claimed by concurrent thread, skipping", event.getId());
                        return;
                    }
                }

                log.info("Created subscription record for user {}, plan {}, subscriptionId {}", userId, plan, subscriptionId);
            } else {
                int claimed = billingPort.claimWebhookEvent(subRecord.getId(), event.getId());
                if (claimed == 0) {
                    log.info("checkout.session.completed: event {} already claimed by concurrent thread for existing record, skipping", event.getId());
                    return;
                }
            }

            scheduleAfterCommit(() -> updateUserLevel(userId, plan.getValue(), "checkout.session.completed", event.getId()));
        }

        void doUpgradeSubscription(StripeSubscription existingSub,
                                    SubscriptionInfo updated,
                                    String newPriceId,
                                    SubscriptionPlan newPlan,
                                    BillingCycle billingCycle) {
            String oldPlan = existingSub.getPlan();
            existingSub.setPlan(newPlan.getValue());
            existingSub.setBillingCycle(billingCycle.getValue());
            existingSub.setStripePriceId(updated.priceId());
            existingSub.setStatus(updated.status());
            existingSub.setCancelAtPeriodEnd(updated.cancelAtPeriodEnd());
            existingSub.setCurrentPeriodStart(toLocalDateTime(updated.currentPeriodStart()));
            existingSub.setCurrentPeriodEnd(toLocalDateTime(updated.currentPeriodEnd()));
            existingSub.setLastOperationSource("upgrade_" + System.currentTimeMillis());
            billingPort.updateSubscription(existingSub);

            scheduleAfterCommit(() -> updateUserLevel(
                existingSub.getUserId(), newPlan.getValue(), "subscription_upgrade"));

            log.info("Upgraded subscription {} for user {}: {} -> {}, priceId {}",
                existingSub.getStripeSubscriptionId(), existingSub.getUserId(),
                oldPlan, newPlan, newPriceId);
        }

        SubscriptionStatusResponse doCancelSubscription(StripeSubscription sub) {
            sub.setCancelAtPeriodEnd(true);
            sub.setCanceledAt(LocalDateTime.now(ZoneId.of("UTC")));
            sub.setLastOperationSource("manual_cancel_" + System.currentTimeMillis());
            billingPort.updateSubscription(sub);

            return new SubscriptionStatusResponse(
                sub.getPlan(),
                sub.getStatus(),
                sub.getCurrentPeriodEnd(),
                true
            );
        }

        void doActivateSubscriptionFromInvoice(Event event, StripeSubscription subRecord) {
            long eventCreated = event.getCreated();
            int rows = billingPort.atomicUpdateSubscription(subRecord.getId(), event.getId(), "active", eventCreated);

            if (rows == 0) {
                log.info("invoice.payment_succeeded: already processed event {}, skipping", event.getId());
                return;
            }

            scheduleAfterCommit(() -> updateUserLevel(
                subRecord.getUserId(), subRecord.getPlan(),
                "invoice.payment_succeeded -> fallback activate", event.getId()));

            log.info("invoice.payment_succeeded: fallback activated subscription {} for user {}",
                subRecord.getStripeSubscriptionId(), subRecord.getUserId());
        }

        void doCreateSubscriptionFromOrphanedInvoice(Event event, Long userId,
                                                      String subscriptionId,
                                                      SubscriptionInfo stripeSub,
                                                      StripeCustomer customer,
                                                      com.stripe.model.Invoice invoice) {
            // 双重检查：并发情况下可能已被 checkout.session.completed 创建
            StripeSubscription existing = billingPort.findSubscriptionByStripeId(subscriptionId);
            if (existing != null
                && ("active".equals(existing.getStatus()) || "trialing".equals(existing.getStatus()))) {
                log.info("invoice.payment_succeeded: orphaned path race, subscription {} already active, skipping",
                    subscriptionId);
                return;
            }

            StripeSubscription subRecord = new StripeSubscription();
            subRecord.setUserId(userId);
            subRecord.setStripeCustomerId(customer.getStripeCustomerId());
            subRecord.setStripeSubscriptionId(subscriptionId);
            subRecord.setLastWebhookEventId(event.getId());
            subRecord.setLastEventCreated(event.getCreated());

            subRecord.setStripePriceId(stripeSub.priceId());
            subRecord.setStatus(stripeSub.status());
            subRecord.setCancelAtPeriodEnd(stripeSub.cancelAtPeriodEnd());
            subRecord.setCurrentPeriodStart(toLocalDateTime(stripeSub.currentPeriodStart()));
            subRecord.setCurrentPeriodEnd(toLocalDateTime(stripeSub.currentPeriodEnd()));

            // 从 Stripe Subscription metadata 或 invoice metadata 推断 plan
            String planStr = null;
            // Try to get metadata from the original event object
            try {
                var deserializer = event.getDataObjectDeserializer();
                var obj = deserializer.getObject().orElse(null);
                if (obj instanceof com.stripe.model.Subscription s && s.getMetadata() != null) {
                    planStr = s.getMetadata().get("plan");
                }
            } catch (Exception ignored) {}

            if (planStr == null && invoice.getMetadata() != null) {
                planStr = invoice.getMetadata().get("plan");
            }
            if (planStr == null) {
                planStr = "PRO";
                log.warn("invoice.payment_succeeded: orphaned invoice {} has no plan metadata, defaulting to PRO", event.getId());
            }
            subRecord.setPlan(planStr);

            String billingCycleStr = "monthly";
            try {
                var deserializer = event.getDataObjectDeserializer();
                var obj = deserializer.getObject().orElse(null);
                if (obj instanceof com.stripe.model.Subscription s && s.getMetadata() != null) {
                    billingCycleStr = s.getMetadata().getOrDefault("billingCycle", "monthly");
                }
            } catch (Exception ignored) {}
            if (invoice.getMetadata() != null && invoice.getMetadata().containsKey("billingCycle")) {
                billingCycleStr = invoice.getMetadata().get("billingCycle");
            }
            subRecord.setBillingCycle(billingCycleStr);

            boolean insertSucceeded = false;
            try {
                billingPort.saveSubscription(subRecord);
                insertSucceeded = true;
            } catch (DuplicateKeyException e) {
                subRecord = billingPort.findSubscriptionByStripeId(subscriptionId);
                if (subRecord == null) {
                    throw new IllegalStateException("DuplicateKeyException but re-query returned null for subscriptionId " + subscriptionId, e);
                }
            }

            // Capture final reference for lambda
            final StripeSubscription savedRecord = subRecord;

            if (!insertSucceeded) {
                int claimed = billingPort.claimWebhookEvent(savedRecord.getId(), event.getId());
                if (claimed == 0) {
                    log.info("invoice.payment_succeeded: event {} already claimed by concurrent thread, skipping", event.getId());
                    return;
                }
            }

            scheduleAfterCommit(() -> updateUserLevel(
                userId, savedRecord.getPlan(), "invoice.payment_succeeded -> orphaned create", event.getId()));
        }

        // ==================== 内部工具方法 ====================

        /**
         * 泛型反序列化：提取 Stripe 对象，减少重复代码
         */
        private <T> T deserializeStripeObject(Event event, Class<T> clazz, String eventType) {
            try {
                var deserializer = event.getDataObjectDeserializer();
                @SuppressWarnings("unchecked")
                T obj = (T) deserializer.getObject().orElse(null);
                if (obj == null) {
                    obj = clazz.cast(deserializer.deserializeUnsafe());
                    if (obj == null) {
                        log.warn("{}: failed to deserialize event object", eventType);
                        return null;
                    }
                }
                return obj;
            } catch (Exception e) {
                log.error("{}: deserialization error", eventType, e);
                return null;
            }
        }

        /**
         * 内部 record：从事件反序列化的 session 信息
         */
        record SessionInfo(String id, String subscriptionId, Map<String, String> metadata) {}

        private SessionInfo deserializeSessionInfo(Event event, String eventType) {
            try {
                var deserializer = event.getDataObjectDeserializer();
                var obj = deserializer.getObject().orElse(null);
                if (obj == null) {
                    obj = deserializer.deserializeUnsafe();
                }
                if (obj instanceof com.stripe.model.checkout.Session session) {
                    return new SessionInfo(session.getId(), session.getSubscription(), session.getMetadata());
                }
                log.warn("{}: failed to deserialize event object", eventType);
                return null;
            } catch (Exception e) {
                log.error("{}: deserialization error", eventType, e);
                return null;
            }
        }

        private SubscriptionInfo deserializeSubscriptionInfo(Event event, String eventType) {
            try {
                var deserializer = event.getDataObjectDeserializer();
                var obj = deserializer.getObject().orElse(null);
                if (obj == null) {
                    obj = deserializer.deserializeUnsafe();
                }
                if (obj instanceof com.stripe.model.Subscription sub) {
                    String firstItemId = Optional.ofNullable(sub.getItems())
                        .map(items -> items.getData())
                        .filter(list -> !list.isEmpty())
                        .map(list -> list.get(0).getId())
                        .orElse(null);
                    String priceId = Optional.ofNullable(sub.getItems())
                        .map(items -> items.getData())
                        .filter(list -> !list.isEmpty())
                        .map(list -> list.get(0))
                        .map(item -> item.getPrice())
                        .filter(price -> price != null)
                        .map(price -> price.getId())
                        .orElse(null);
                    return new SubscriptionInfo(
                        sub.getId(), sub.getStatus(),
                        sub.getCurrentPeriodStart(), sub.getCurrentPeriodEnd(),
                        sub.getCancelAtPeriodEnd(), firstItemId, priceId, sub.getCanceledAt());
                }
                log.warn("{}: failed to deserialize event object", eventType);
                return null;
            } catch (Exception e) {
                log.error("{}: deserialization error", eventType, e);
                return null;
            }
        }

        private StripeCustomer getOrCreateCustomer(Long userId) {
            StripeCustomer existing = billingPort.findCustomerByUserIdAndNotDeleted(userId);

            if (existing != null) {
                return existing;
            }

            User user = userRepositoryPort.findById(userId).orElse(null);
            if (user == null) {
                throw new RuntimeException("用户不存在，userId=" + userId);
            }

            try {
                CustomerInfo customer = paymentPort.createCustomer(user.getEmail());
                StripeCustomer newCustomer = new StripeCustomer();
                newCustomer.setUserId(userId);
                newCustomer.setStripeCustomerId(customer.id());
                billingPort.saveCustomer(newCustomer);

                log.info("Created Stripe Customer for user {}: {}", userId, customer.id());
                return newCustomer;
            } catch (Exception e) {
                log.error("Failed to create Stripe Customer for user {}: {}", userId, e.getMessage(), e);
                throw new RuntimeException("创建 Stripe 客户失败", e);
            }
        }

        private String getPriceId(SubscriptionPlan plan, BillingCycle billingCycle) {
            Map<String, StripeProperties.PlanPrices> prices = stripeProperties.getPrices();
            if (prices == null) {
                throw new RuntimeException("Stripe prices not configured");
            }

            StripeProperties.PlanPrices planPrices = prices.get(plan.getValue().toLowerCase());
            if (planPrices == null) {
                throw new RuntimeException("No Stripe prices configured for plan: " + plan);
            }

            String priceId = switch (billingCycle) {
                case MONTHLY -> planPrices.getMonthlyPriceId();
                case YEARLY -> planPrices.getYearlyPriceId();
            };

            if (priceId == null || priceId.isBlank()) {
                throw new RuntimeException("No price configured for " + plan + " " + billingCycle);
            }

            return priceId;
        }

        private SubscriptionPlan validatePlan(String plan) {
            try {
                return SubscriptionPlan.fromValue(plan);
            } catch (IllegalArgumentException e) {
                String validValues = String.join(", ", SubscriptionPlan.VALUES);
                throw new IllegalArgumentException("无效的套餐类型: " + plan + "，可选值: " + validValues);
            }
        }

        private BillingCycle validateBillingCycle(String billingCycle) {
            try {
                return BillingCycle.fromValue(billingCycle);
            } catch (IllegalArgumentException e) {
                String validValues = String.join(", ", BillingCycle.VALUES);
                throw new IllegalArgumentException("无效的计费周期: " + billingCycle + "，可选值: " + validValues);
            }
        }

        /**
         * 将 Stripe epoch 秒数转换为 UTC LocalDateTime
         */
        private LocalDateTime toLocalDateTime(Long epochSeconds) {
            if (epochSeconds == null) {
                return null;
            }
            return LocalDateTime.ofInstant(Instant.ofEpochSecond(epochSeconds), ZoneId.of("UTC"));
        }

        /**
         * 注册在事务提交后执行的回调，用于 Redis 幂等标记等非事务操作。
         */
        private void scheduleAfterCommit(Runnable action) {
            if (TransactionSynchronizationManager.isSynchronizationActive()) {
                TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                    @Override
                    public void afterCommit() {
                        action.run();
                    }
                });
            } else {
                action.run();
            }
        }

        /**
         * 基于 Redis SETNX 的事件级幂等检查
         */
        private boolean markEventProcessed(String eventId) {
            String key = "webhook:event_processed:" + eventId;
            Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", java.time.Duration.ofHours(24));
            return Boolean.TRUE.equals(success);
        }

        /**
         * 同步用户等级（状态变更后调用）
         */
        private void syncUserLevelAfterStatusChange(StripeSubscription subRecord, String newStatus, Event event) {
            String targetLevel;
            if ("active".equals(newStatus) || "trialing".equals(newStatus)) {
                targetLevel = subRecord.getPlan();
            } else if ("past_due".equals(newStatus)) {
                log.warn("subscription {}: past_due, not downgrading yet (grace period)", subRecord.getStripeSubscriptionId());
                return;
            } else if ("canceled".equals(newStatus) || "unpaid".equals(newStatus)) {
                targetLevel = "FREE";
            } else if ("paused".equals(newStatus)) {
                log.info("subscription {}: paused, keeping userLevel unchanged", subRecord.getStripeSubscriptionId());
                return;
            } else {
                log.info("subscription {}: unknown status {}, keeping userLevel unchanged", subRecord.getStripeSubscriptionId(), newStatus);
                return;
            }

            scheduleAfterCommit(() -> updateUserLevel(subRecord.getUserId(), targetLevel,
                "subscription.updated -> " + newStatus, event.getId()));
        }

        private void updateUserLevel(Long userId, String newLevel, String reason, String eventId) {
            if (eventId != null && !markEventProcessed(eventId)) {
                log.info("updateUserLevel: event {} already processed, skipping", eventId);
                return;
            }

            updateUserLevel(userId, newLevel, reason);
        }

        private void updateUserLevel(Long userId, String newLevel, String reason) {
            User user = userRepositoryPort.findById(userId).orElse(null);
            if (user == null) {
                log.error("Cannot update userLevel: user {} not found", userId);
                return;
            }

            String oldLevel = user.getUserLevel();
            if (newLevel.equals(oldLevel)) {
                return;
            }

            user.setUserLevel(newLevel);
            userRepositoryPort.update(user);

            UserPlanHistory history = new UserPlanHistory();
            history.setUserId(userId);
            history.setOldPlan(oldLevel != null ? oldLevel : "UNKNOWN");
            history.setNewPlan(newLevel);
            history.setNote(reason);
            userRepositoryPort.savePlanHistory(history);

            log.info("User {} level changed: {} -> {} (reason: {})", userId, oldLevel, newLevel, reason);

            // 降级到 FREE 时：通过端口吊销令牌
            if ("FREE".equals(newLevel) && !"FREE".equals(oldLevel)) {
                tokenRevocationPort.revokeAllTokens(userId, user.getEmail(), "subscription_downgrade: " + reason);
            }
        }
    }