x

queue/logic/order/activateOrderLogic.go

@@ -28,6 +28,7 @@ import (
 	"github.com/perfect-panel/server/pkg/uuidx"
 	queueTypes "github.com/perfect-panel/server/queue/types"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 // Order type constants define the different types of orders that can be processed
@@ -71,8 +72,10 @@ func NewActivateOrderLogic(svc *svc.ServiceContext) *ActivateOrderLogic {
 // It handles the complete workflow of activating a paid order including validation,
 // processing based on order type, and finalization.
 func (l *ActivateOrderLogic) ProcessTask(ctx context.Context, task *asynq.Task) error {
-	logger.WithContext(ctx).Info("[ActivateOrderLogic] 开始处理订单激活任务",
-		logger.Field("payload", string(task.Payload())))
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "activation_task_received",
+		"[SubscriptionFlow] activation task received",
+		logger.Field("payload", string(task.Payload())),
+	)
 
 	payload, err := l.parsePayload(ctx, task.Payload())
 	if err != nil {
@@ -81,8 +84,11 @@ func (l *ActivateOrderLogic) ProcessTask(ctx context.Context, task *asynq.Task)
 		return nil // payload 解析失败不重试，因为重试也会失败
 	}
 
-	logger.WithContext(ctx).Info("[ActivateOrderLogic] 正在验证订单",
-		logger.Field("order_no", payload.OrderNo))
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "activation_order_lookup",
+		"[SubscriptionFlow] activation task is loading order",
+		logger.Field("order_no", payload.OrderNo),
+		logger.Field("iap_expire_at", payload.IAPExpireAt),
+	)
 
 	orderInfo, err := l.claimAndGetOrder(ctx, payload.OrderNo)
 	if err != nil {
@@ -104,10 +110,10 @@ func (l *ActivateOrderLogic) ProcessTask(ctx context.Context, task *asynq.Task)
 		return nil
 	}
 
-	logger.WithContext(ctx).Info("[ActivateOrderLogic] 订单验证通过，开始处理",
-		logger.Field("order_no", orderInfo.OrderNo),
-		logger.Field("order_type", orderInfo.Type),
-		logger.Field("user_id", orderInfo.UserId))
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "activation_order_claimed",
+		"[SubscriptionFlow] activation worker claimed paid order",
+		commonLogic.OrderTraceFields(orderInfo)...,
+	)
 
 	if err = l.processOrderByType(ctx, orderInfo, payload.IAPExpireAt); err != nil {
 		l.releaseClaim(ctx, orderInfo.OrderNo)
@@ -118,12 +124,21 @@ func (l *ActivateOrderLogic) ProcessTask(ctx context.Context, task *asynq.Task)
 		return err // 返回 err 允许 asynq 重试
 	}
 
+	if err = l.reconcilePostOrderSubscriptions(ctx, orderInfo); err != nil {
+		l.releaseClaim(ctx, orderInfo.OrderNo)
+		logger.WithContext(ctx).Error("[ActivateOrderLogic] 订单订阅兜底合并失败，将重试",
+			logger.Field("order_no", orderInfo.OrderNo),
+			logger.Field("order_type", orderInfo.Type),
+			logger.Field("error", err.Error()))
+		return err
+	}
+
 	l.finalizeCouponAndOrder(ctx, orderInfo)
 
-	logger.WithContext(ctx).Info("[ActivateOrderLogic] 订单激活成功",
-		logger.Field("order_no", orderInfo.OrderNo),
-		logger.Field("order_type", orderInfo.Type),
-		logger.Field("user_id", orderInfo.UserId))
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "activation_finished",
+		"[SubscriptionFlow] order activation completed",
+		commonLogic.OrderTraceFields(orderInfo)...,
+	)
 	return nil
 }
 
@@ -217,6 +232,311 @@ func (l *ActivateOrderLogic) processOrderByType(ctx context.Context, orderInfo *
 	}
 }
 
+func (l *ActivateOrderLogic) reconcilePostOrderSubscriptions(ctx context.Context, orderInfo *order.Order) error {
+	if !shouldReconcilePostOrderSubscriptions(orderInfo) {
+		return nil
+	}
+
+	effectiveUserID := orderInfo.UserId
+	if orderInfo.SubscriptionUserId > 0 {
+		effectiveUserID = orderInfo.SubscriptionUserId
+	}
+	if effectiveUserID == 0 || orderInfo.Id == 0 {
+		return nil
+	}
+
+	var (
+		survivor             user.Subscribe
+		survivorBefore       user.Subscribe
+		losers               []user.Subscribe
+		mergedIDs            []int64
+		subscribeIDsToClear  = make(map[int64]struct{})
+		missingSurvivor      bool
+		ownerMismatchSkipped bool
+		identitySourceID     int64
+	)
+
+	err := l.svc.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
+			Model(&user.Subscribe{}).
+			Where("order_id = ?", orderInfo.Id).
+			First(&survivor).Error; err != nil {
+			if errors.Is(err, gorm.ErrRecordNotFound) {
+				missingSurvivor = true
+				return nil
+			}
+			return err
+		}
+		survivorBefore = survivor
+
+		var ownerSubs []user.Subscribe
+		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
+			Model(&user.Subscribe{}).
+			Where("user_id = ?", effectiveUserID).
+			Order("id ASC").
+			Find(&ownerSubs).Error; err != nil {
+			return err
+		}
+
+		if survivor.UserId != effectiveUserID {
+			if len(ownerSubs) == 0 {
+				ownerMismatchSkipped = true
+				return nil
+			}
+
+			if err := tx.Model(&user.Subscribe{}).
+				Where("id = ?", survivor.Id).
+				Update("user_id", effectiveUserID).Error; err != nil {
+				return err
+			}
+			survivor.UserId = effectiveUserID
+
+			if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
+				Model(&user.Subscribe{}).
+				Where("user_id = ?", effectiveUserID).
+				Order("id ASC").
+				Find(&ownerSubs).Error; err != nil {
+				return err
+			}
+		}
+
+		if len(ownerSubs) <= 1 {
+			return nil
+		}
+
+		maxExpire := survivor.ExpireTime
+		for i := range ownerSubs {
+			item := ownerSubs[i]
+			if item.Id == survivor.Id {
+				if item.ExpireTime.After(maxExpire) {
+					maxExpire = item.ExpireTime
+				}
+				continue
+			}
+
+			losers = append(losers, item)
+			mergedIDs = append(mergedIDs, item.Id)
+			if item.ExpireTime.After(maxExpire) {
+				maxExpire = item.ExpireTime
+			}
+			if item.SubscribeId > 0 {
+				subscribeIDsToClear[item.SubscribeId] = struct{}{}
+			}
+		}
+
+		if len(losers) == 0 {
+			return nil
+		}
+
+		if survivor.SubscribeId > 0 {
+			subscribeIDsToClear[survivor.SubscribeId] = struct{}{}
+		}
+
+		identitySource := pickSubscriptionIdentitySource(losers)
+		if identitySource != nil {
+			identitySourceID = identitySource.Id
+		}
+
+		updateFields := map[string]interface{}{
+			"status":      1,
+			"finished_at": nil,
+		}
+		if maxExpire.After(survivor.ExpireTime) {
+			survivor.ExpireTime = maxExpire
+			updateFields["expire_time"] = maxExpire
+		}
+		if identitySource != nil {
+			if identitySource.Token != "" {
+				survivor.Token = identitySource.Token
+				updateFields["token"] = identitySource.Token
+			}
+			if identitySource.UUID != "" {
+				survivor.UUID = identitySource.UUID
+				updateFields["uuid"] = identitySource.UUID
+			}
+		}
+
+		loserIDs := make([]int64, 0, len(losers))
+		for i := range losers {
+			loserIDs = append(loserIDs, losers[i].Id)
+		}
+		if len(loserIDs) == 0 {
+			return nil
+		}
+
+		// user_subscribe 当前没有 deleted_at 字段，这里沿用项目现有删除语义清理 loser 记录。
+		if err := tx.Where("id IN ?", loserIDs).Delete(&user.Subscribe{}).Error; err != nil {
+			return err
+		}
+
+		if err := tx.Model(&user.Subscribe{}).
+			Where("id = ?", survivor.Id).
+			Updates(updateFields).Error; err != nil {
+			return err
+		}
+		survivor.Status = 1
+		survivor.FinishedAt = nil
+
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	if missingSurvivor {
+		commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "post_order_reconcile_skipped",
+			"[SubscriptionFlow] post-order reconcile skipped because survivor subscription was not found",
+			append(commonLogic.OrderTraceFields(orderInfo),
+				logger.Field("reason", "post_order_reconcile"),
+				logger.Field("effective_user_id", effectiveUserID),
+			)...,
+		)
+		return nil
+	}
+
+	if ownerMismatchSkipped {
+		commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "post_order_reconcile_skipped",
+			"[SubscriptionFlow] post-order reconcile skipped because survivor owner mismatch had no duplicates",
+			append(commonLogic.OrderTraceFields(orderInfo),
+				logger.Field("reason", "post_order_reconcile"),
+				logger.Field("effective_user_id", effectiveUserID),
+				logger.Field("survivor_subscribe_id", survivor.Id),
+				logger.Field("survivor_user_id", survivorBefore.UserId),
+			)...,
+		)
+		return nil
+	}
+
+	if len(losers) == 0 {
+		return nil
+	}
+
+	l.clearPostOrderReconcileCache(ctx, &survivorBefore, &survivor, losers, subscribeIDsToClear)
+
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "post_order_reconciled",
+		"[SubscriptionFlow] post-order reconcile merged duplicate subscriptions",
+		append(commonLogic.OrderTraceFields(orderInfo),
+			logger.Field("reason", "post_order_reconcile"),
+			logger.Field("effective_user_id", effectiveUserID),
+			logger.Field("survivor_subscribe_id", survivor.Id),
+			logger.Field("identity_source_subscribe_id", identitySourceID),
+			logger.Field("merged_subscribe_ids", mergedIDs),
+			logger.Field("merged_count", len(mergedIDs)),
+		)...,
+	)
+
+	return nil
+}
+
+func shouldReconcilePostOrderSubscriptions(orderInfo *order.Order) bool {
+	if orderInfo == nil {
+		return false
+	}
+
+	switch orderInfo.Type {
+	case OrderTypeSubscribe, OrderTypeRenewal, OrderTypeRedemption:
+		return true
+	default:
+		return false
+	}
+}
+
+func pickSubscriptionIdentitySource(candidates []user.Subscribe) *user.Subscribe {
+	if len(candidates) == 0 {
+		return nil
+	}
+
+	best := &candidates[0]
+	for i := 1; i < len(candidates); i++ {
+		candidate := &candidates[i]
+		if subscriptionIdentityPriority(candidate, best) {
+			best = candidate
+		}
+	}
+	return best
+}
+
+func subscriptionIdentityPriority(candidate *user.Subscribe, current *user.Subscribe) bool {
+	if candidate == nil {
+		return false
+	}
+	if current == nil {
+		return true
+	}
+
+	candidateUsable := candidate.Token != "" || candidate.UUID != ""
+	currentUsable := current.Token != "" || current.UUID != ""
+	if candidateUsable != currentUsable {
+		return candidateUsable
+	}
+
+	if candidate.ExpireTime.After(current.ExpireTime) {
+		return true
+	}
+	if current.ExpireTime.After(candidate.ExpireTime) {
+		return false
+	}
+
+	if candidate.UpdatedAt.After(current.UpdatedAt) {
+		return true
+	}
+	if current.UpdatedAt.After(candidate.UpdatedAt) {
+		return false
+	}
+
+	return candidate.Id > current.Id
+}
+
+func (l *ActivateOrderLogic) clearPostOrderReconcileCache(
+	ctx context.Context,
+	survivorBefore *user.Subscribe,
+	survivorAfter *user.Subscribe,
+	losers []user.Subscribe,
+	subscribeIDs map[int64]struct{},
+) {
+	cacheModels := make([]*user.Subscribe, 0, len(losers)+2)
+	if survivorBefore != nil {
+		cacheModels = append(cacheModels, survivorBefore)
+	}
+	if survivorAfter != nil {
+		cacheModels = append(cacheModels, survivorAfter)
+	}
+	for i := range losers {
+		loser := losers[i]
+		cacheModels = append(cacheModels, &loser)
+	}
+
+	if len(cacheModels) > 0 {
+		if err := l.svc.UserModel.ClearSubscribeCache(ctx, cacheModels...); err != nil {
+			logger.WithContext(ctx).Error("Post-order reconcile clear subscribe cache failed",
+				logger.Field("reason", "post_order_reconcile"),
+				logger.Field("error", err.Error()),
+			)
+		}
+	}
+
+	if l.svc.SubscribeModel != nil {
+		for subscribeID := range subscribeIDs {
+			if err := l.svc.SubscribeModel.ClearCache(ctx, subscribeID); err != nil {
+				logger.WithContext(ctx).Error("Post-order reconcile clear plan cache failed",
+					logger.Field("reason", "post_order_reconcile"),
+					logger.Field("subscribe_id", subscribeID),
+					logger.Field("error", err.Error()),
+				)
+			}
+		}
+	}
+
+	if l.svc.NodeModel != nil {
+		if err := l.svc.NodeModel.ClearServerAllCache(ctx); err != nil {
+			logger.WithContext(ctx).Error("Post-order reconcile clear node cache failed",
+				logger.Field("reason", "post_order_reconcile"),
+				logger.Field("error", err.Error()),
+			)
+		}
+	}
+}
+
 // finalizeCouponAndOrder handles post-processing tasks including coupon updates
 // and order status finalization
 func (l *ActivateOrderLogic) finalizeCouponAndOrder(ctx context.Context, orderInfo *order.Order) {
@@ -238,6 +558,10 @@ func (l *ActivateOrderLogic) finalizeCouponAndOrder(ctx context.Context, orderIn
 		)
 	}
 	orderInfo.Status = OrderStatusFinished
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "order_status_finished",
+		"[SubscriptionFlow] order status updated to finished",
+		commonLogic.OrderTraceFields(orderInfo)...,
+	)
 }
 
 // NewPurchase handles new subscription purchase including user creation,
@@ -248,6 +572,13 @@ func (l *ActivateOrderLogic) NewPurchase(ctx context.Context, orderInfo *order.O
 		return err
 	}
 
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "activation_user_resolved",
+		"[SubscriptionFlow] activation resolved subscription recipient user",
+		append(commonLogic.OrderTraceFields(orderInfo),
+			logger.Field("resolved_user_id", userInfo.Id),
+		)...,
+	)
+
 	sub, err := l.getSubscribeInfo(ctx, orderInfo.SubscribeId)
 	if err != nil {
 		return err
@@ -288,12 +619,14 @@ func (l *ActivateOrderLogic) NewPurchase(ctx context.Context, orderInfo *order.O
 				)
 			} else {
 				userSub = anchorSub
-				logger.WithContext(ctx).Infow("Single mode purchase routed to renewal in activation",
-					logger.Field("mode", "single"),
-					logger.Field("route", "purchase_to_renewal"),
-					logger.Field("plan_changed", anchorSub.SubscribeId != orderInfo.SubscribeId),
-					logger.Field("anchor_user_subscribe_id", anchorSub.Id),
-					logger.Field("order_no", orderInfo.OrderNo),
+				commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "subscription_reused",
+					"[SubscriptionFlow] activation reused single-mode anchor subscription",
+					append(commonLogic.OrderTraceFields(orderInfo),
+						append(commonLogic.UserSubscribeTraceFields(anchorSub),
+							logger.Field("reuse_reason", "single_mode_purchase_to_renewal"),
+							logger.Field("plan_changed", anchorSub.SubscribeId != orderInfo.SubscribeId),
+						)...,
+					)...,
 				)
 			}
 		case errors.Is(anchorErr, gorm.ErrRecordNotFound):
@@ -359,11 +692,15 @@ func (l *ActivateOrderLogic) NewPurchase(ctx context.Context, orderInfo *order.O
 				)
 			} else {
 				userSub = &existingSub
-				logger.WithContext(ctx).Infow("Fallback: renewed existing subscription instead of creating duplicate",
-					logger.Field("existing_subscribe_id", existingSub.Id),
-					logger.Field("order_no", orderInfo.OrderNo),
-					logger.Field("candidate_user_ids", candidateUserIds),
-					logger.Field("owner_corrected_to", effectiveOwner),
+				commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "subscription_reused",
+					"[SubscriptionFlow] activation renewed an existing subscription instead of creating a duplicate",
+					append(commonLogic.OrderTraceFields(orderInfo),
+						append(commonLogic.UserSubscribeTraceFields(&existingSub),
+							logger.Field("reuse_reason", "fallback_existing_subscription"),
+							logger.Field("candidate_user_ids", candidateUserIds),
+							logger.Field("owner_corrected_to", effectiveOwner),
+						)...,
+					)...,
 				)
 			}
 		}
@@ -386,7 +723,12 @@ func (l *ActivateOrderLogic) NewPurchase(ctx context.Context, orderInfo *order.O
 	// Clear cache
 	l.clearServerCache(ctx, sub)
 
-	logger.WithContext(ctx).Info("Insert user subscribe success")
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "subscription_issued",
+		"[SubscriptionFlow] activation finished issuing subscription entitlement",
+		append(commonLogic.OrderTraceFields(orderInfo),
+			commonLogic.UserSubscribeTraceFields(userSub)...,
+		)...,
+	)
 	return nil
 }
 
@@ -457,6 +799,14 @@ func (l *ActivateOrderLogic) createGuestUser(ctx context.Context, orderInfo *ord
 		logger.Field("identifier", tempOrder.Identifier),
 		logger.Field("auth_type", tempOrder.AuthType),
 	)
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "guest_user_created",
+		"[SubscriptionFlow] guest user created during order activation",
+		append(commonLogic.OrderTraceFields(orderInfo),
+			logger.Field("created_user_id", userInfo.Id),
+			logger.Field("identifier", tempOrder.Identifier),
+			logger.Field("auth_type", tempOrder.AuthType),
+		)...,
+	)
 
 	return userInfo, nil
 }
@@ -570,6 +920,13 @@ func (l *ActivateOrderLogic) createUserSubscription(ctx context.Context, orderIn
 		return nil, err
 	}
 
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "subscription_created",
+		"[SubscriptionFlow] new user subscription record created",
+		append(commonLogic.OrderTraceFields(orderInfo),
+			commonLogic.UserSubscribeTraceFields(userSub)...,
+		)...,
+	)
+
 	return userSub, nil
 }
 
@@ -622,11 +979,15 @@ func (l *ActivateOrderLogic) extendGiftSubscription(ctx context.Context, giftSub
 		return nil, err
 	}
 
-	logger.WithContext(ctx).Info("Extended gift subscription successfully",
-		logger.Field("subscribe_id", giftSub.Id),
-		logger.Field("old_expire_time", baseTime),
-		logger.Field("new_expire_time", newExpireTime),
-		logger.Field("order_id", orderInfo.Id),
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "subscription_reused",
+		"[SubscriptionFlow] paid order extended an existing gift subscription",
+		append(commonLogic.OrderTraceFields(orderInfo),
+			append(commonLogic.UserSubscribeTraceFields(giftSub),
+				logger.Field("reuse_reason", "gift_subscription_promoted"),
+				logger.Field("old_expire_time", baseTime),
+				logger.Field("new_expire_time", newExpireTime),
+			)...,
+		)...,
 	)
 
 	return giftSub, nil
@@ -997,6 +1358,15 @@ func (l *ActivateOrderLogic) Renewal(ctx context.Context, orderInfo *order.Order
 	// Handle commission
 	go l.handleCommission(context.Background(), userInfo, orderInfo)
 
+	commonLogic.SubscriptionTraceInfo(logger.WithContext(ctx), commonLogic.SubscriptionTraceFlowOrder, "subscription_renewed",
+		"[SubscriptionFlow] renewal order updated existing subscription",
+		append(commonLogic.OrderTraceFields(orderInfo),
+			append(commonLogic.UserSubscribeTraceFields(userSub),
+				logger.Field("iap_expire_at", iapExpireAt),
+			)...,
+		)...,
+	)
+
 	return nil
 }