package distribution import ( "context" "errors" "fmt" "slices" "strings" "time" "hfb_sys/backend/internal/model" "hfb_sys/backend/internal/modules/wallet" "gorm.io/gorm" "gorm.io/gorm/clause" ) // autoBatchPrefix 区分自动释放批次与主站财务手工创建的结算批次(DS 前缀)。 const autoBatchPrefix = "DA" // autoReleaseBlockedRefundStatuses 进行中的退款不得自动释放,避免先入账再回冲。 var autoReleaseBlockedRefundStatuses = []string{"pending", "pending_review", "refunding"} // autoReleaseBlockedDisputeStatuses 开放中的维权订单暂缓释放。 var autoReleaseBlockedDisputeStatuses = []string{"open", "processing"} // AutoReleaseDueSettlements 扫描到达释放时间的分销订单,按站点自动创建批次并释放到可提现余额。 // 返回实际触达释放的订单数。 func (r *Repository) AutoReleaseDueSettlements(ctx context.Context) (int, error) { if r == nil || r.db == nil { return 0, ErrDependencyUnavailable } now := time.Now() var orderIDs []uint64 if err := r.db.WithContext(ctx). Table("distribution_order_financials f"). Select("DISTINCT f.order_id"). Joins("JOIN rental_orders ro ON ro.id = f.order_id"). Where("f.settlement_status = ? AND f.release_at IS NOT NULL AND f.release_at <= ?", "pending", now). Where("ro.status IN ?", settlementReadyOrderStatuses). Where("ro.refund_status NOT IN ?", autoReleaseBlockedRefundStatuses). Where("NOT EXISTS (SELECT 1 FROM disputes d WHERE d.order_id = ro.id AND d.status IN ?)", autoReleaseBlockedDisputeStatuses). Order("f.order_id ASC"). Limit(1000). Scan(&orderIDs).Error; err != nil { return 0, err } if len(orderIDs) == 0 { return 0, nil } var siteIDs []uint64 if err := r.db.WithContext(ctx). Table("distribution_order_parties p"). Select("DISTINCT p.party_id"). Where("p.order_id IN ? AND p.party_type = ?", orderIDs, PartyTypeSite). Where("NOT EXISTS (SELECT 1 FROM distribution_settlement_items i WHERE i.party_id = p.id)"). Order("p.party_id ASC"). Scan(&siteIDs).Error; err != nil { return 0, err } released := 0 for _, siteID := range siteIDs { if siteID == 0 { continue } count, err := r.releaseSiteSettlements(ctx, siteID, orderIDs, now) if err != nil { return released, fmt.Errorf("释放站点 %d 分销结算失败: %w", siteID, err) } released += count } return released, nil } // releaseSiteSettlements 在事务内释放指定站点上所有到期订单的份额: // 复用当天的自动批次,生成结算明细并把总额入账到站点钱包。 func (r *Repository) releaseSiteSettlements(ctx context.Context, siteID uint64, orderIDs []uint64, now time.Time) (int, error) { releasedOrders := 0 err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { // 先不加锁筛出本站尚未入账的候选参与方对应的订单,收敛后续加锁范围: // 调用方传入的是全站到期订单,其中可能大多数已被手工批次纳入,无需加锁。 var candidateOrderIDs []uint64 if err := tx.Table("distribution_order_parties p"). Select("DISTINCT p.order_id"). Where("p.order_id IN ? AND p.party_type = ? AND p.party_id = ?", orderIDs, PartyTypeSite, siteID). Where("NOT EXISTS (SELECT 1 FROM distribution_settlement_items i WHERE i.party_id = p.id)"). Order("p.order_id ASC"). Scan(&candidateOrderIDs).Error; err != nil { return err } if len(candidateOrderIDs) == 0 { return nil } // 先锁财务快照再锁参与方,与 ApplyRefundTx 保持同一顺序,避免并发退款回冲时死锁。 var lockedFinancials []model.DistributionOrderFinancial if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). Where("order_id IN ?", candidateOrderIDs). Order("id ASC").Find(&lockedFinancials).Error; err != nil { return err } var parties []model.DistributionOrderParty if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). Where("order_id IN ? AND party_type = ? AND party_id = ?", candidateOrderIDs, PartyTypeSite, siteID). Where("NOT EXISTS (SELECT 1 FROM distribution_settlement_items i WHERE i.party_id = distribution_order_parties.id)"). Order("order_id ASC, id ASC").Find(&parties).Error; err != nil { return err } releasable := make(map[uint64]bool, len(parties)) checked := make(map[uint64]struct{}, len(parties)) eligible := make([]model.DistributionOrderParty, 0, len(parties)) for _, party := range parties { if _, done := checked[party.OrderID]; !done { ok, err := releasableOrder(tx, party.OrderID, now) if err != nil { return err } checked[party.OrderID] = struct{}{} releasable[party.OrderID] = ok } if !releasable[party.OrderID] { continue } eligible = append(eligible, party) } if len(eligible) == 0 { return nil } batch, err := ensureAutoBatch(tx, siteID, now) if err != nil { return err } var total int64 orderSet := make(map[uint64]struct{}, len(eligible)) for _, party := range eligible { item := model.DistributionSettlementItem{ BatchID: batch.ID, OrderID: party.OrderID, PartyID: party.ID, AmountCent: party.IncomeAmountCent, Status: "settled", } if err := tx.Create(&item).Error; err != nil { if isDuplicateSettlementItem(err) { // 唯一键命中说明另一条路径(手工批次或并发自动释放)已为该参与方建过明细, // 跳过它,避免重复入账。 continue } return err } total += party.IncomeAmountCent orderSet[party.OrderID] = struct{}{} } if total > 0 { if err := wallet.AppendEntries(tx, wallet.Entry{ UserID: 0, DistributionSiteID: siteID, Direction: "in", AmountCent: total, BalanceType: "available", BizType: "distribution_settlement", BizNo: batch.BatchNo, Remark: fmt.Sprintf("分销自动结算释放 %s", batch.BatchNo), }); err != nil { return err } } if err := tx.Model(&model.DistributionSettlementBatch{}).Where("id = ?", batch.ID). Updates(map[string]any{ "total_amount_cent": gorm.Expr("total_amount_cent + ?", total), "status": "settled", "settled_at": now, }).Error; err != nil { return err } orderList := make([]uint64, 0, len(orderSet)) for orderID := range orderSet { orderList = append(orderList, orderID) } if err := markFinancialsSettledIfSitePartiesDone(tx, orderList, now); err != nil { return err } releasedOrders = len(orderList) return nil }) if err != nil { return 0, err } return releasedOrders, nil } // isDuplicateSettlementItem 判断错误是否为结算明细唯一键冲突。 // GORM 未开启 TranslateError 时 MySQL 1062 不会映射为 ErrDuplicatedKey,因此同时检查错误链与错误文本。 func isDuplicateSettlementItem(err error) bool { if err == nil { return false } if errors.Is(err, gorm.ErrDuplicatedKey) { return true } return strings.Contains(err.Error(), "1062") } // releasableOrder 复核订单是否仍满足释放条件,并返回财务快照。 // 订单在扫描后可能进入退款或维权流程,因此事务内需要重新校验。 func releasableOrder(tx *gorm.DB, orderID uint64, now time.Time) (bool, error) { var financial model.DistributionOrderFinancial if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). Where("order_id = ?", orderID).First(&financial).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return false, nil } return false, err } if financial.SettlementStatus != "pending" || financial.ReleaseAt == nil || financial.ReleaseAt.After(now) { return false, nil } var order model.RentalOrder if err := tx.Select("id, status, refund_status").First(&order, orderID).Error; err != nil { return false, err } if !orderStatusAllowsRelease(order.Status) || slices.Contains(autoReleaseBlockedRefundStatuses, order.RefundStatus) { return false, nil } var openDisputes int64 if err := tx.Table("disputes"). Where("order_id = ? AND status IN ?", orderID, autoReleaseBlockedDisputeStatuses). Count(&openDisputes).Error; err != nil { return false, err } return openDisputes == 0, nil } // ensureAutoBatch 复用当天的自动批次(不论 pending/settled),没有则创建,保证对账按站点按天聚合。 func ensureAutoBatch(tx *gorm.DB, siteID uint64, now time.Time) (*model.DistributionSettlementBatch, error) { startOfDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()) var batch model.DistributionSettlementBatch err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). Where("site_id = ? AND batch_no LIKE ? AND period_start = ?", siteID, autoBatchPrefix+"%", startOfDay). Order("id ASC").First(&batch).Error if err == nil { return &batch, nil } if !errors.Is(err, gorm.ErrRecordNotFound) { return nil, err } batch = model.DistributionSettlementBatch{ SiteID: siteID, BatchNo: fmt.Sprintf("%s%s%d", autoBatchPrefix, now.Format("20060102150405.000000000"), siteID), PeriodStart: startOfDay, PeriodEnd: startOfDay.AddDate(0, 0, 1), Status: "pending", } if err := tx.Create(&batch).Error; err != nil { return nil, err } return &batch, nil }