// CreateListingTask 创建商品上架任务（幂等性保证）
func (s *ListingService) CreateListingTask(req *ListingRequest) (*ListingTask, error) {
    // 1. 生成幂等性标识符
    taskCode := generateTaskCode(req.CategoryID, req.CreatedBy, time.Now())
    
    // 2. 尝试创建任务（唯一索引保证幂等性）
    task := &ListingTask{
        TaskCode:   taskCode,
        ItemInfo:   req.ItemInfo,
        Status:     StatusDraft,
        CreatedBy:  req.CreatedBy,
        CreatedAt:  time.Now(),
    }
    
    // 3. 插入数据库（如果 task_code 已存在，返回已有记录）
    result := s.db.Where("task_code = ?", taskCode).FirstOrCreate(task)
    if result.Error != nil {
        return nil, fmt.Errorf("create listing task failed: %w", result.Error)
    }
    
    // 4. 如果是首次创建，触发审核流程
    if result.RowsAffected > 0 {
        s.eventPublisher.Publish(&ListingTaskCreatedEvent{
            TaskCode: taskCode,
            ItemInfo: req.ItemInfo,
        })
    }
    
    return task, nil
}

// generateTaskCode 生成上架任务唯一标识符
func generateTaskCode(categoryID, createdBy int64, timestamp time.Time) string {
    data := fmt.Sprintf("%d-%d-%d", categoryID, createdBy, timestamp.Unix())
    hash := sha256.Sum256([]byte(data))
    return hex.EncodeToString(hash[:8]) // 取前16个字符
}

// ProcessSupplierData 处理供应商商品数据同步
func (s *SupplierSyncService) ProcessSupplierData(supplierID int64, externalItems []*ExternalItem) error {
    for _, extItem := range externalItems {
        // 1. 根据 (supplier_id, external_id) 查询商品是否存在
        item, err := s.repo.GetItemByExternalID(supplierID, extItem.ExternalID)
        
        if err == ErrItemNotFound {
            // 2. 商品不存在，创建新商品（类似上架流程）
            if err := s.createItemFromSupplier(supplierID, extItem); err != nil {
                return fmt.Errorf("create item failed: %w", err)
            }
        } else if err == nil {
            // 3. 商品已存在，计算差异并决定审核策略
            diff := s.calculateDiff(item, extItem)
            if err := s.applyChanges(item, diff); err != nil {
                return fmt.Errorf("apply changes failed: %w", err)
            }
        } else {
            return fmt.Errorf("query item failed: %w", err)
        }
    }
    return nil
}

// calculateDiff 计算商品数据差异
func (s *SupplierSyncService) calculateDiff(current *Item, external *ExternalItem) *ItemDiff {
    diff := &ItemDiff{
        ItemID:     current.ItemID,
        SupplierID: current.SupplierID,
        ExternalID: current.ExternalID,
        Changes:    make(map[string]*FieldChange),
    }
    
    // 价格变动
    if current.Price != external.Price {
        diff.Changes["price"] = &FieldChange{
            Field:      "price",
            OldValue:   current.Price,
            NewValue:   external.Price,
            ChangeRate: (external.Price - current.Price) / current.Price,
        }
    }
    
    // 库存变动
    if current.Stock != external.Stock {
        diff.Changes["stock"] = &FieldChange{
            Field:      "stock",
            OldValue:   current.Stock,
            NewValue:   external.Stock,
            ChangeRate: float64(external.Stock-current.Stock) / float64(current.Stock),
        }
    }
    
    // 商品标题变动
    if current.Title != external.Title {
        diff.Changes["title"] = &FieldChange{
            Field:    "title",
            OldValue: current.Title,
            NewValue: external.Title,
        }
    }
    
    return diff
}

// applyChanges 应用变更并根据风险等级决定审核策略
func (s *SupplierSyncService) applyChanges(item *Item, diff *ItemDiff) error {
    // 1. 评估审核策略
    strategy := s.evaluateApprovalStrategy(diff)
    
    switch strategy {
    case ApprovalStrategyNone:
        // 直接更新，无需审核（例如库存变动、小幅价格调整）
        return s.repo.UpdateItem(item, diff)
        
    case ApprovalStrategyAuto:
        // 自动审核（规则引擎快速验证）
        if s.autoApprove(diff) {
            return s.repo.UpdateItem(item, diff)
        } else {
            return s.createChangeRequest(item, diff, ApprovalStrategyManual)
        }
        
    case ApprovalStrategyManual:
        // 人工审核（推送审核队列）
        return s.createChangeRequest(item, diff, ApprovalStrategyManual)
        
    case ApprovalStrategyStrict:
        // 严格审核（多级审核）
        return s.createChangeRequest(item, diff, ApprovalStrategyStrict)
    }
    
    return nil
}

// evaluateApprovalStrategy 评估审核策略
func (s *SupplierSyncService) evaluateApprovalStrategy(diff *ItemDiff) ApprovalStrategy {
    for _, change := range diff.Changes {
        switch change.Field {
        case "price":
            if math.Abs(change.ChangeRate) >= 0.3 { // 价格变动 >= 30%
                return ApprovalStrategyManual
            }
            return ApprovalStrategyNone
            
        case "stock":
            return ApprovalStrategyNone // 库存变动无需审核
            
        case "title", "description":
            return ApprovalStrategyAuto // 自动审核（敏感词过滤）
            
        case "category_id":
            return ApprovalStrategyStrict // 类目变更需严格审核
        }
    }
    return ApprovalStrategyNone
}

// BatchUpdateItems 批量更新商品
func (s *OperationService) BatchUpdateItems(req *BatchUpdateRequest) (*BatchOperationResult, error) {
    // 1. 生成操作批次ID（幂等性保证）
    batchID := s.idGenerator.Generate()
    
    // 2. 创建批量操作记录
    batch := &BatchOperation{
        BatchID:    batchID,
        Type:       req.OperationType,
        Status:     BatchStatusPending,
        TotalCount: len(req.Items),
        CreatedBy:  req.OperatorID,
        CreatedAt:  time.Now(),
    }
    if err := s.repo.CreateBatchOperation(batch); err != nil {
        return nil, fmt.Errorf("create batch operation failed: %w", err)
    }
    
    // 3. 使用 Worker Pool 并发处理
    results := make(chan *ItemUpdateResult, len(req.Items))
    wp := NewWorkerPool(10) // 10个并发 Worker
    
    for _, itemReq := range req.Items {
        wp.Submit(func() {
            result := s.processItemUpdate(batchID, itemReq)
            results <- result
        })
    }
    
    // 4. 收集结果
    wp.Wait()
    close(results)
    
    return s.summarizeBatchResult(batchID, results), nil
}

// processItemUpdate 处理单个商品更新
func (s *OperationService) processItemUpdate(batchID string, req *ItemUpdateRequest) *ItemUpdateResult {
    // 1. 获取当前商品信息
    item, err := s.repo.GetItemByID(req.ItemID)
    if err != nil {
        return &ItemUpdateResult{ItemID: req.ItemID, Success: false, Error: err}
    }
    
    // 2. 计算差异
    diff := s.calculateDiff(item, req.Changes)
    
    // 3. 应用变更（差异化审核）
    if err := s.applyChangesByType(item, diff, batchID); err != nil {
        return &ItemUpdateResult{ItemID: req.ItemID, Success: false, Error: err}
    }
    
    return &ItemUpdateResult{ItemID: req.ItemID, Success: true}
}

// applyChangesByType 根据变更类型应用不同的审核策略
func (s *OperationService) applyChangesByType(item *Item, diff *ItemDiff, batchID string) error {
    // 1. 评估审核策略
    strategy := s.evaluateApprovalStrategy(diff)
    
    // 2. 记录操作日志
    log := &OperationLog{
        BatchID:    batchID,
        ItemID:     item.ItemID,
        ChangeType: diff.ChangeType,
        OldValue:   diff.OldValue,
        NewValue:   diff.NewValue,
        Strategy:   strategy,
        CreatedAt:  time.Now(),
    }
    s.repo.CreateOperationLog(log)
    
    // 3. 根据策略处理
    switch strategy {
    case ApprovalStrategyNone:
        // 直接更新
        return s.repo.UpdateItemWithVersion(item, diff)
        
    case ApprovalStrategyManual:
        // 创建审批单
        return s.createChangeRequest(item, diff, batchID)
    }
    
    return nil
}

// ApprovalRouter 审核策略路由器
type ApprovalRouter struct {
    riskEvaluator *RiskEvaluator
    ruleEngine    *RuleEngine
}

// Route 根据变更内容路由到合适的审核策略
func (r *ApprovalRouter) Route(diff *ItemDiff) ApprovalStrategy {
    // 1. 计算风险分数
    riskScore := r.riskEvaluator.Evaluate(diff)
    
    // 2. 根据风险分数决定审核策略
    if riskScore <= 3 {
        return ApprovalStrategyNone // 免审核
    } else if riskScore <= 5 {
        return ApprovalStrategyAuto // 自动审核
    } else if riskScore <= 8 {
        return ApprovalStrategyManual // 人工审核
    } else {
        return ApprovalStrategyStrict // 严格审核
    }
}

// ApprovalStrategy 审核策略
type ApprovalStrategy int

const (
    ApprovalStrategyNone   ApprovalStrategy = 0 // 免审核
    ApprovalStrategyAuto   ApprovalStrategy = 1 // 自动审核
    ApprovalStrategyManual ApprovalStrategy = 2 // 人工审核
    ApprovalStrategyStrict ApprovalStrategy = 3 // 严格审核
)

// RiskEvaluator 风险评估器
type RiskEvaluator struct {
    fieldWeights  map[string]float64
    itemRepo      ItemRepository
}

// Evaluate 评估变更风险分数
func (e *RiskEvaluator) Evaluate(diff *ItemDiff) float64 {
    var totalRisk float64
    
    // 1. 获取商品当前状态
    item, _ := e.itemRepo.GetItemByID(diff.ItemID)
    itemFactor := e.calculateItemFactor(item)
    
    // 2. 遍历所有变更字段，计算风险分数
    for _, change := range diff.Changes {
        fieldWeight := e.fieldWeights[change.Field]
        changeMagnitude := e.calculateChangeMagnitude(change)
        
        // 风险分数 = 字段权重 × 变更幅度 × 商品因子
        risk := fieldWeight * changeMagnitude * itemFactor
        totalRisk += risk
    }
    
    return totalRisk
}

// calculateItemFactor 计算商品状态风险系数
func (e *RiskEvaluator) calculateItemFactor(item *Item) float64 {
    if item.MonthlySales > 1000 {
        return 1.5 // 热销商品
    } else if item.MonthlySales > 100 {
        return 1.0 // 普通商品
    } else if item.Status == StatusOffline {
        return 0.5 // 已下架商品
    } else {
        return 0.8 // 新品
    }
}

// calculateChangeMagnitude 计算变更幅度风险系数
func (e *RiskEvaluator) calculateChangeMagnitude(change *FieldChange) float64 {
    switch change.Field {
    case "price":
        absRate := math.Abs(change.ChangeRate)
        if absRate < 0.1 {
            return 0.5
        } else if absRate < 0.3 {
            return 1.0
        } else if absRate < 0.5 {
            return 2.0
        } else {
            return 3.0
        }
        
    case "category_id":
        return 2.0 // 类目变更固定高风险
        
    case "title":
        // 根据标题变更的相似度计算
        similarity := e.calculateSimilarity(change.OldValue, change.NewValue)
        return 1.0 - similarity // 相似度越低，风险越高
        
    default:
        return 1.0
    }
}

// createChangeRequest 创建变更审批单
func (s *ApprovalService) createChangeRequest(item *Item, diff *ItemDiff, strategy ApprovalStrategy) error {
    // 1. 生成审批单唯一标识
    requestCode := s.generateRequestCode(item.ItemID)
    
    // 2. 计算风险分数
    riskScore := s.riskEvaluator.Evaluate(diff)
    
    // 3. 创建审批单
    request := &ChangeRequest{
        RequestCode:      requestCode,
        ItemID:           item.ItemID,
        ChangeType:       diff.ChangeType,
        ChangeFields:     diff.ToJSON(),
        BeforeSnapshot:   item.ToJSON(),
        AfterSnapshot:    diff.ApplyTo(item).ToJSON(),
        Status:           StatusPendingApproval,
        ApprovalStrategy: strategy,
        RiskScore:        riskScore,
        ImpactAnalysis:   s.analyzeImpact(item, diff),
        CreatedBy:        diff.OperatorID,
        CreatedAt:        time.Now(),
    }
    
    // 4. 保存到数据库
    if err := s.repo.CreateChangeRequest(request); err != nil {
        return fmt.Errorf("create change request failed: %w", err)
    }
    
    // 5. 推送到审核队列
    event := &ChangeRequestCreatedEvent{
        RequestCode:      requestCode,
        ItemID:           item.ItemID,
        ApprovalStrategy: strategy,
        RiskScore:        riskScore,
    }
    return s.eventPublisher.Publish("approval.change_request.created", event)
}

// ProcessApprovalResult 处理审核结果
func (s *ApprovalService) ProcessApprovalResult(requestCode string, result *ApprovalResult) error {
    // 1. 获取审批单
    request, err := s.repo.GetChangeRequest(requestCode)
    if err != nil {
        return fmt.Errorf("get change request failed: %w", err)
    }
    
    // 2. 更新审批单状态
    request.ApproverID = result.ApproverID
    request.ApprovedAt = time.Now()
    
    if result.Approved {
        // 审核通过
        request.Status = StatusApproved
        
        // 应用变更
        if err := s.applyChange(request); err != nil {
            return fmt.Errorf("apply change failed: %w", err)
        }
    } else {
        // 审核驳回
        request.Status = StatusRejected
        request.RejectReason = result.RejectReason
    }
    
    // 3. 保存审批单
    if err := s.repo.UpdateChangeRequest(request); err != nil {
        return fmt.Errorf("update change request failed: %w", err)
    }
    
    // 4. 发送通知
    s.sendNotification(request)
    
    return nil
}

// CheckSLA 检查 SLA 超时
func (s *ApprovalService) CheckSLA() error {
    // 1. 查询超时的审批单
    requests, err := s.repo.GetTimeoutRequests()
    if err != nil {
        return err
    }
    
    // 2. 处理超时审批单
    for _, req := range requests {
        switch req.ApprovalStrategy {
        case ApprovalStrategyAuto:
            // 自动审核超时，自动通过
            s.autoApprove(req)
            
        case ApprovalStrategyManual:
            // 人工审核超时，升级到严格审核
            s.escalateToStrict(req)
            
        case ApprovalStrategyStrict:
            // 严格审核超时，告警通知
            s.sendAlert(req)
        }
    }
    
    return nil
}

// StateMachine 商品生命周期状态机
type StateMachine struct {
    transitions map[ItemStatus][]ItemStatus
    repo        ItemRepository
    logRepo     LogRepository
}

// NewStateMachine 创建状态机
func NewStateMachine() *StateMachine {
    return &StateMachine{
        transitions: map[ItemStatus][]ItemStatus{
            StatusDraft:     {StatusPending, StatusArchived},
            StatusPending:   {StatusApproved, StatusRejected},
            StatusRejected:  {StatusDraft, StatusArchived},
            StatusApproved:  {StatusPublished},
            StatusPublished: {StatusOnline},
            StatusOnline:    {StatusOffline, StatusArchived},
            StatusOffline:   {StatusOnline, StatusArchived},
        },
    }
}

// CanTransition 检查是否可以进行状态转换
func (sm *StateMachine) CanTransition(from, to ItemStatus) bool {
    allowedStates, ok := sm.transitions[from]
    if !ok {
        return false
    }
    
    for _, allowed := range allowedStates {
        if allowed == to {
            return true
        }
    }
    return false
}

// Transition 执行状态转换
func (sm *StateMachine) Transition(item *Item, to ItemStatus, operator int64) error {
    // 1. 检查状态转换是否合法
    if !sm.CanTransition(item.Status, to) {
        return fmt.Errorf("invalid transition from %s to %s", item.Status, to)
    }
    
    // 2. 检查前置条件
    if err := sm.checkPreconditions(item, to); err != nil {
        return fmt.Errorf("precondition check failed: %w", err)
    }
    
    // 3. 记录状态变更前的快照
    oldStatus := item.Status
    
    // 4. 更新状态
    item.Status = to
    item.UpdatedAt = time.Now()
    item.UpdatedBy = operator
    item.Version++
    
    // 5. 保存到数据库（带乐观锁）
    if err := sm.repo.UpdateItemWithVersion(item); err != nil {
        return fmt.Errorf("update item failed: %w", err)
    }
    
    // 6. 记录状态变更日志
    sm.logStatusChange(item.ItemID, oldStatus, to, operator)
    
    // 7. 发布生命周期事件
    sm.publishLifecycleEvent(item, oldStatus, to)
    
    return nil
}

// checkPreconditions 检查状态转换的前置条件
func (sm *StateMachine) checkPreconditions(item *Item, to ItemStatus) error {
    switch to {
    case StatusPending:
        if item.Title == "" || item.CategoryID == 0 {
            return errors.New("item info incomplete")
        }
        
    case StatusPublished:
        if item.Price <= 0 {
            return errors.New("price not set")
        }
        
    case StatusOnline:
        if item.Stock <= 0 {
            return errors.New("stock is zero")
        }
        
    case StatusArchived:
        if item.Status == StatusOnline {
            orderCount, _ := sm.orderRepo.CountPendingOrders(item.ItemID)
            if orderCount > 0 {
                return errors.New("has pending orders")
            }
        }
    }
    
    return nil
}

// checkPermission 检查状态变更权限
func (sm *StateMachine) checkPermission(item *Item, to ItemStatus, operator *Operator) error {
    switch operator.Role {
    case RoleOperator:
        return nil
        
    case RoleMerchant:
        allowedTransitions := map[ItemStatus][]ItemStatus{
            StatusDraft:   {StatusPending},
            StatusOffline: {StatusOnline},
            StatusOnline:  {StatusOffline},
        }
        allowed, ok := allowedTransitions[item.Status]
        if !ok {
            return errors.New("permission denied")
        }
        for _, s := range allowed {
            if s == to {
                return nil
            }
        }
        return errors.New("permission denied")
        
    case RoleSystem:
        if to == StatusOffline && item.Stock == 0 {
            return nil
        }
        if to == StatusApproved && item.ApprovalStrategy == ApprovalStrategyAuto {
            return nil
        }
        return errors.New("permission denied")
        
    case RoleApprover:
        if item.Status == StatusPending && (to == StatusApproved || to == StatusRejected) {
            return nil
        }
        return errors.New("permission denied")
    }
    
    return errors.New("unknown role")
}

// ItemStatusLog 商品状态变更日志
type ItemStatusLog struct {
    LogID        int64      `json:"log_id"`
    ItemID       int64      `json:"item_id"`
    OldStatus    ItemStatus `json:"old_status"`
    NewStatus    ItemStatus `json:"new_status"`
    Operator     int64      `json:"operator"`
    OperatorRole string     `json:"operator_role"`
    Reason       string     `json:"reason"`
    CreatedAt    time.Time  `json:"created_at"`
}

// logStatusChange 记录状态变更日志
func (sm *StateMachine) logStatusChange(itemID int64, oldStatus, newStatus ItemStatus, operator int64) {
    log := &ItemStatusLog{
        ItemID:    itemID,
        OldStatus: oldStatus,
        NewStatus: newStatus,
        Operator:  operator,
        CreatedAt: time.Now(),
    }
    sm.logRepo.CreateStatusLog(log)
}

// LifecycleEvent 生命周期事件
type LifecycleEvent struct {
    EventID   string     `json:"event_id"`
    EventType string     `json:"event_type"`
    ItemID    int64      `json:"item_id"`
    OldStatus ItemStatus `json:"old_status"`
    NewStatus ItemStatus `json:"new_status"`
    Operator  int64      `json:"operator"`
    Timestamp time.Time  `json:"timestamp"`
    Payload   map[string]interface{} `json:"payload"`
}

// publishLifecycleEvent 发布生命周期事件
func (sm *StateMachine) publishLifecycleEvent(item *Item, oldStatus, newStatus ItemStatus) {
    eventType := sm.mapEventType(oldStatus, newStatus)
    
    event := &LifecycleEvent{
        EventID:   sm.generateEventID(),
        EventType: eventType,
        ItemID:    item.ItemID,
        OldStatus: oldStatus,
        NewStatus: newStatus,
        Timestamp: time.Now(),
        Payload: map[string]interface{}{
            "title":       item.Title,
            "category_id": item.CategoryID,
            "price":       item.Price,
            "stock":       item.Stock,
        },
    }
    
    sm.eventPublisher.Publish("product.lifecycle", event)
}

// mapEventType 根据状态转换映射事件类型
func (sm *StateMachine) mapEventType(oldStatus, newStatus ItemStatus) string {
    if oldStatus == StatusDraft && newStatus == StatusPending {
        return "ProductListed"
    }
    if oldStatus == StatusPending && newStatus == StatusApproved {
        return "ProductApproved"
    }
    if oldStatus == StatusApproved && newStatus == StatusPublished {
        return "ProductPublished"
    }
    if oldStatus == StatusPublished && newStatus == StatusOnline {
        return "ProductOnline"
    }
    if newStatus == StatusOffline {
        return "ProductOffline"
    }
    if newStatus == StatusArchived {
        return "ProductArchived"
    }
    return "ProductStatusChanged"
}

// SearchEngineConsumer 搜索引擎消费者
type SearchEngineConsumer struct {
    esClient *elasticsearch.Client
}

// Consume 消费生命周期事件
func (c *SearchEngineConsumer) Consume(event *LifecycleEvent) error {
    switch event.EventType {
    case "ProductOnline":
        return c.addToIndex(event.ItemID)
        
    case "ProductOffline", "ProductArchived":
        return c.removeFromIndex(event.ItemID)
        
    case "ProductPriceChanged", "ProductStockChanged":
        return c.updateIndex(event.ItemID, event.Payload)
    }
    
    return nil
}

// TransitionWithOutbox 使用 Outbox 模式进行状态转换
func (sm *StateMachine) TransitionWithOutbox(item *Item, to ItemStatus, operator int64) error {
    tx := sm.db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx.Rollback()
        }
    }()
    
    // 更新商品状态
    oldStatus := item.Status
    item.Status = to
    if err := tx.Save(item).Error; err != nil {
        tx.Rollback()
        return err
    }
    
    // 写入 Outbox 表
    event := &LifecycleEvent{
        EventID:   sm.generateEventID(),
        EventType: sm.mapEventType(oldStatus, to),
        ItemID:    item.ItemID,
        OldStatus: oldStatus,
        NewStatus: to,
        Timestamp: time.Now(),
    }
    outbox := &EventOutbox{
        EventID:   event.EventID,
        EventType: event.EventType,
        Payload:   event.ToJSON(),
        Status:    OutboxStatusPending,
        CreatedAt: time.Now(),
    }
    if err := tx.Create(outbox).Error; err != nil {
        tx.Rollback()
        return err
    }
    
    return tx.Commit().Error
}

// SnowflakeIDGenerator 雪花算法ID生成器
type SnowflakeIDGenerator struct {
    machineID int64
    sequence  int64
    lastTime  int64
    mu        sync.Mutex
}

// Generate 生成雪花ID
func (g *SnowflakeIDGenerator) Generate() int64 {
    g.mu.Lock()
    defer g.mu.Unlock()
    
    now := time.Now().UnixMilli()
    
    if now == g.lastTime {
        g.sequence = (g.sequence + 1) & 0xFFF
        if g.sequence == 0 {
            for now <= g.lastTime {
                now = time.Now().UnixMilli()
            }
        }
    } else {
        g.sequence = 0
    }
    
    g.lastTime = now
    
    id := ((now - 1640995200000) << 22) | (g.machineID << 12) | g.sequence
    return id
}

// generateTaskCode 生成上架任务唯一标识符
func generateTaskCode(categoryID, createdBy int64, timestamp time.Time) string {
    data := fmt.Sprintf("%d-%d-%d", categoryID, createdBy, timestamp.Unix())
    hash := sha256.Sum256([]byte(data))
    return hex.EncodeToString(hash[:8])
}

// CreateOrGetListingTask 创建或获取上架任务（幂等性保证）
func (s *ListingService) CreateOrGetListingTask(req *ListingRequest) (*ListingTask, bool, error) {
    taskCode := generateTaskCode(req.CategoryID, req.CreatedBy, time.Now())
    
    existingTask, err := s.repo.GetTaskByCode(taskCode)
    if err == nil {
        return existingTask, false, nil
    }
    
    task := &ListingTask{
        TaskCode:  taskCode,
        ItemInfo:  req.ItemInfo,
        Status:    StatusDraft,
        CreatedBy: req.CreatedBy,
        CreatedAt: time.Now(),
    }
    
    if err := s.repo.CreateTask(task); err != nil {
        if isDuplicateKeyError(err) {
            existingTask, _ = s.repo.GetTaskByCode(taskCode)
            return existingTask, false, nil
        }
        return nil, false, fmt.Errorf("create task failed: %w", err)
    }
    
    return task, true, nil
}

// UpdateItemWithVersion 使用乐观锁更新商品
func (r *ItemRepository) UpdateItemWithVersion(item *Item) error {
    currentVersion := item.Version
    item.Version++
    item.UpdatedAt = time.Now()
    
    result := r.db.Model(&Item{}).
        Where("item_id = ? AND version = ?", item.ItemID, currentVersion).
        Updates(map[string]interface{}{
            "title":      item.Title,
            "price":      item.Price,
            "stock":      item.Stock,
            "status":     item.Status,
            "version":    item.Version,
            "updated_at": item.UpdatedAt,
        })
    
    if result.Error != nil {
        return fmt.Errorf("update item failed: %w", result.Error)
    }
    
    if result.RowsAffected == 0 {
        return ErrVersionConflict
    }
    
    return nil
}

// UpdateItemWithRetry 乐观锁更新失败时重试
func (s *OperationService) UpdateItemWithRetry(itemID int64, updateFn func(*Item) error, maxRetries int) error {
    for i := 0; i < maxRetries; i++ {
        item, err := s.repo.GetItemByID(itemID)
        if err != nil {
            return err
        }
        
        if err := updateFn(item); err != nil {
            return err
        }
        
        if err := s.repo.UpdateItemWithVersion(item); err == nil {
            return nil
        } else if err == ErrVersionConflict {
            time.Sleep(time.Duration(i*10) * time.Millisecond)
            continue
        } else {
            return err
        }
    }
    
    return errors.New("update failed after max retries")
}

// resolveConflict 解决并发冲突
func (s *ConflictResolver) resolveConflict(item *Item, updateA, updateB *ItemUpdate) (*ItemUpdate, error) {
    priorityA := s.getOperatorPriority(updateA.Operator)
    priorityB := s.getOperatorPriority(updateB.Operator)
    
    if priorityA > priorityB {
        return updateA, nil
    } else if priorityB > priorityA {
        return updateB, nil
    }
    
    if updateA.Timestamp.After(updateB.Timestamp) {
        return updateA, nil
    } else {
        return updateB, nil
    }
}

// getOperatorPriority 获取操作者优先级
func (s *ConflictResolver) getOperatorPriority(operator *Operator) int {
    switch operator.Type {
    case OperatorTypeManual:
        return 100
    case OperatorTypeSupplier:
        return 50
    case OperatorTypeSystem:
        return 10
    default:
        return 0
    }
}

// PriceService 价格服务
type PriceService struct {
    itemRepo      ItemRepository
    priceClient   PriceEngineClient
    eventPublisher EventPublisher
    cache         *redis.Client
}

// UpdateItemPrice 更新商品价格
func (s *PriceService) UpdateItemPrice(itemID int64, newPrice float64, operator int64) error {
    // 1. 获取商品
    item, err := s.itemRepo.GetItemByID(itemID)
    if err != nil {
        return err
    }
    
    // 2. 更新商品基础价格
    oldPrice := item.Price
    item.Price = newPrice
    item.UpdatedAt = time.Now()
    
    if err := s.itemRepo.UpdateItem(item); err != nil {
        return fmt.Errorf("update item price failed: %w", err)
    }
    
    // 3. 发布价格变更事件（异步）
    event := &PriceChangedEvent{
        ItemID:    itemID,
        OldPrice:  oldPrice,
        NewPrice:  newPrice,
        Operator:  operator,
        Timestamp: time.Now(),
    }
    s.eventPublisher.Publish("product.price.changed", event)
    
    // 4. 清除缓存
    s.cache.Del(fmt.Sprintf("item:price:%d", itemID))
    
    return nil
}

// GetFinalPrice 获取商品最终价格（含促销）
func (s *PriceService) GetFinalPrice(itemID int64, userID int64) (float64, error) {
    // 1. 尝试从缓存获取
    cacheKey := fmt.Sprintf("item:price:%d", itemID)
    if cachedPrice, err := s.cache.Get(cacheKey).Float64(); err == nil {
        return cachedPrice, nil
    }
    
    // 2. 调用定价引擎计算最终价格
    finalPrice, err := s.priceClient.CalculateFinalPrice(itemID, userID)
    if err != nil {
        return 0, fmt.Errorf("calculate final price failed: %w", err)
    }
    
    // 3. 缓存最终价格（TTL 5分钟）
    s.cache.Set(cacheKey, finalPrice, 5*time.Minute)
    
    return finalPrice, nil
}

// StockService 库存服务
type StockService struct {
    itemRepo       ItemRepository
    stockClient    StockSystemClient
    eventPublisher EventPublisher
    cache          *redis.Client
}

// InitializeStock 初始化商品库存
func (s *StockService) InitializeStock(itemID int64, initialStock int) error {
    // 同步调用库存系统
    if err := s.stockClient.CreateStock(itemID, initialStock); err != nil {
        return fmt.Errorf("initialize stock failed: %w", err)
    }
    
    // 更新商品表的库存快照
    if err := s.itemRepo.UpdateStockSnapshot(itemID, initialStock); err != nil {
        return fmt.Errorf("update stock snapshot failed: %w", err)
    }
    
    return nil
}

// HandleStockChangedEvent 处理库存变更事件
func (s *StockService) HandleStockChangedEvent(event *StockChangedEvent) error {
    // 1. 更新商品表的库存快照
    if err := s.itemRepo.UpdateStockSnapshot(event.ItemID, event.NewStock); err != nil {
        return fmt.Errorf("update stock snapshot failed: %w", err)
    }
    
    // 2. 更新缓存
    cacheKey := fmt.Sprintf("item:stock:%d", event.ItemID)
    s.cache.Set(cacheKey, event.NewStock, 10*time.Minute)
    
    // 3. 如果库存为0，自动下架商品
    if event.NewStock == 0 {
        item, _ := s.itemRepo.GetItemByID(event.ItemID)
        if item.Status == StatusOnline {
            if err := s.offlineItem(item, "库存为0自动下架"); err != nil {
                return fmt.Errorf("offline item failed: %w", err)
            }
        }
    }
    
    return nil
}

// ReconcileStock 库存对账
func (s *StockService) ReconcileStock() error {
    // 1. 获取所有在售商品
    items, err := s.itemRepo.GetOnlineItems()
    if err != nil {
        return err
    }
    
    // 2. 批量查询库存系统
    itemIDs := make([]int64, len(items))
    for i, item := range items {
        itemIDs[i] = item.ItemID
    }
    stocks, err := s.stockClient.BatchGetStock(itemIDs)
    if err != nil {
        return err
    }
    
    // 3. 对比库存快照
    for _, item := range items {
        actualStock := stocks[item.ItemID]
        if item.StockSnapshot != actualStock {
            // 库存不一致，更新快照
            s.itemRepo.UpdateStockSnapshot(item.ItemID, actualStock)
            
            // 如果库存为0，自动下架
            if actualStock == 0 && item.Status == StatusOnline {
                s.offlineItem(item, "对账发现库存为0，自动下架")
            }
        }
    }
    
    return nil
}

// ListingSaga 商品上架的 Saga 编排器
type ListingSaga struct {
    itemRepo       ItemRepository
    priceClient    PriceEngineClient
    stockClient    StockSystemClient
    eventPublisher EventPublisher
}

// Execute 执行 Saga
func (s *ListingSaga) Execute(req *ListingRequest) error {
    // 1. 创建 Saga 状态记录
    saga := &SagaState{
        SagaID:    generateSagaID(),
        Type:      "ProductListing",
        Status:    SagaStatusPending,
        CreatedAt: time.Now(),
    }
    
    // 2. 步骤1：创建商品
    item, err := s.createItem(saga, req)
    if err != nil {
        saga.Status = SagaStatusFailed
        s.saveSagaState(saga)
        return err
    }
    saga.Steps = append(saga.Steps, &SagaStep{
        StepName: "CreateItem",
        Status:   SagaStepStatusCompleted,
        Data:     map[string]interface{}{"item_id": item.ItemID},
    })
    
    // 3. 步骤2：初始化价格
    if err := s.initPrice(saga, item.ItemID, req.Price); err != nil {
        // 失败，执行补偿
        s.compensate(saga)
        saga.Status = SagaStatusFailed
        s.saveSagaState(saga)
        return err
    }
    saga.Steps = append(saga.Steps, &SagaStep{
        StepName: "InitPrice",
        Status:   SagaStepStatusCompleted,
    })
    
    // 4. 步骤3：初始化库存
    if err := s.initStock(saga, item.ItemID, req.Stock); err != nil {
        // 失败，执行补偿
        s.compensate(saga)
        saga.Status = SagaStatusFailed
        s.saveSagaState(saga)
        return err
    }
    saga.Steps = append(saga.Steps, &SagaStep{
        StepName: "InitStock",
        Status:   SagaStepStatusCompleted,
    })
    
    // 5. 步骤4：发布事件
    s.eventPublisher.Publish("product.listed", &ProductListedEvent{
        ItemID: item.ItemID,
    })
    
    // 6. Saga 完成
    saga.Status = SagaStatusCompleted
    s.saveSagaState(saga)
    
    return nil
}

// compensate 执行补偿操作
func (s *ListingSaga) compensate(saga *SagaState) {
    // 从后往前补偿
    for i := len(saga.Steps) - 1; i >= 0; i-- {
        step := saga.Steps[i]
        
        switch step.StepName {
        case "CreateItem":
            // 补偿：删除商品
            itemID := step.Data["item_id"].(int64)
            s.itemRepo.DeleteItem(itemID)
            step.Status = SagaStepStatusCompensated
            
        case "InitPrice":
            // 补偿：删除价格
            itemID := step.Data["item_id"].(int64)
            s.priceClient.DeletePrice(itemID)
            step.Status = SagaStepStatusCompensated
            
        case "InitStock":
            // 补偿：删除库存
            itemID := step.Data["item_id"].(int64)
            s.stockClient.DeleteStock(itemID)
            step.Status = SagaStepStatusCompensated
        }
    }
}

// createItem 创建商品
func (s *ListingSaga) createItem(saga *SagaState, req *ListingRequest) (*Item, error) {
    item := &Item{
        ItemID:     s.idGenerator.Generate(),
        Title:      req.Title,
        CategoryID: req.CategoryID,
        Status:     StatusDraft,
        CreatedAt:  time.Now(),
    }
    
    if err := s.itemRepo.CreateItem(item); err != nil {
        return nil, fmt.Errorf("create item failed: %w", err)
    }
    
    return item, nil
}

// initPrice 初始化价格
func (s *ListingSaga) initPrice(saga *SagaState, itemID int64, price float64) error {
    if err := s.priceClient.CreatePrice(itemID, price); err != nil {
        return fmt.Errorf("init price failed: %w", err)
    }
    return nil
}

// initStock 初始化库存
func (s *ListingSaga) initStock(saga *SagaState, itemID int64, stock int) error {
    if err := s.stockClient.CreateStock(itemID, stock); err != nil {
        return fmt.Errorf("init stock failed: %w", err)
    }
    return nil
}

// ExecuteWithTimeout 执行 Saga（带超时）
func (s *ListingSaga) ExecuteWithTimeout(req *ListingRequest, timeout time.Duration) error {
    ctx, cancel := context.WithTimeout(context.Background(), timeout)
    defer cancel()
    
    errChan := make(chan error, 1)
    
    go func() {
        errChan <- s.Execute(req)
    }()
    
    select {
    case err := <-errChan:
        return err
    case <-ctx.Done():
        // 超时，执行补偿
        return errors.New("saga execution timeout")
    }
}