kra-new/internal/data/business/effectiveness.go

package business

import (
	"context"
	"time"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"
	"kra/internal/biz/business"
)

type effectivenessRepo struct{ data Provider }

func NewEffectivenessRepo(data Provider) business.EffectivenessRepo {
	return &effectivenessRepo{data: data}
}
func (r *effectivenessRepo) State(ctx context.Context) (*business.EffectivenessState, error) {
	var po effectivenessStatePO
	if err := r.data.DB().WithContext(ctx).First(&po, "id=1").Error; err != nil {
		return nil, business.ErrStorage
	}
	s := &business.EffectivenessState{RuleKey: po.RuleKey, Revision: po.Revision}
	if po.BuiltAt != nil {
		s.BuiltAt = *po.BuiltAt
	}
	return s, nil
}
func (r *effectivenessRepo) Candidates(ctx context.Context, key string, duration, now int64, full bool) ([]*business.EffectivenessCandidate, error) {
	db := r.data.DB().WithContext(ctx)
	available, err := businessTablesAvailable(db, "tab_promote", "tab_spend")
	if err != nil {
		return nil, err
	}
	if !available {
		return nil, business.ErrEffectivenessUnavailable
	}
	q := db.Table("tab_promote p").Joins("LEFT JOIN sys_business_channel_effectiveness e ON e.rule_key=? AND e.source_db=p.source_db AND e.promote_id=p.id", key).Where("p.source_db IN ?", []string{"lx", "hw", "yq"})
	if !full {
		q = q.Where("e.promote_id IS NULL OR e.complete=?", false)
	}
	var pos []struct {
		ID                      uint64
		SourceDB                string `gorm:"column:source_db"`
		CreateTime, QualifiedAt *int64
		WindowPay               string
	}
	if err := q.Select("p.id,p.source_db,p.create_time,e.qualified_at,e.window_pay").Order("p.id,p.source_db").Scan(&pos).Error; err != nil {
		return nil, business.ErrStorage
	}
	rows := make([]*business.EffectivenessCandidate, 0, len(pos))
	for _, po := range pos {
		rows = append(rows, &business.EffectivenessCandidate{ChannelKey: business.ChannelKey{ID: po.ID, SourceDB: po.SourceDB}, CreatedAt: po.CreateTime, PreviousAt: po.QualifiedAt, PreviousPay: po.WindowPay})
	}
	return rows, nil
}
func (r *effectivenessRepo) Charges(ctx context.Context, candidates []*business.EffectivenessCandidate, duration, now int64, full bool, consume func(business.EffectivenessCharge) error) error {
	db := r.data.DB().WithContext(ctx)
	// Bound by now as future charges are excluded. Avoid unsigned subtraction
	// on legacy pay_time columns and overflow for very large configured windows.
	windowSeconds := min(duration, now+1)
	// The full-history scan is reserved for explicit rebuilds. Minute jobs
	// must keep their candidate restriction in SQL, even with many open windows.
	if full && len(candidates) > 3000 {
		return r.scanAllCharges(db, candidates, windowSeconds, now, consume)
	}
	for start := 0; start < len(candidates); start += 300 {
		keys := [][]any{}
		ids := []uint64{}
		for _, c := range candidates[start:min(start+300, len(candidates))] {
			if c.CreatedAt != nil && *c.CreatedAt > 0 && *c.CreatedAt <= now {
				keys = append(keys, []any{c.SourceDB, c.ID})
				ids = append(ids, c.ID)
			}
		}
		if len(keys) == 0 {
			continue
		}
		q := db.Table("tab_spend s").Joins("JOIN tab_promote p ON p.source_db=s.source_db AND p.id=s.promote_id").Where("s.promote_id IN ? AND (s.source_db,s.promote_id) IN ? AND s.pay_status=1", ids, keys).Where("s.pay_time>=p.create_time AND s.pay_time<=? AND s.pay_time<p.create_time+?", now, windowSeconds)
		rows, err := q.Select("s.source_db,s.promote_id,s.pay_time,CAST(COALESCE(s.pay_amount,0) AS CHAR)").Order("s.promote_id,s.source_db,s.pay_time,s.id").Rows()
		if err != nil {
			return business.ErrStorage
		}
		for rows.Next() {
			var c business.EffectivenessCharge
			if err := rows.Scan(&c.SourceDB, &c.ID, &c.PaidAt, &c.Amount); err != nil {
				rows.Close()
				return business.ErrStorage
			}
			if err := consume(c); err != nil {
				rows.Close()
				return err
			}
		}
		err = rows.Err()
		rows.Close()
		if err != nil {
			return business.ErrStorage
		}
	}
	return nil
}

// Large rebuilds stream the covering index once. Repeating hundreds of IN
// batches can make the optimizer repeatedly sort large payment ranges.
func (r *effectivenessRepo) scanAllCharges(db *gorm.DB, candidates []*business.EffectivenessCandidate, windowSeconds, now int64, consume func(business.EffectivenessCharge) error) error {
	selected := make(map[business.ChannelKey]bool, len(candidates))
	for _, c := range candidates {
		selected[c.ChannelKey] = true
	}
	table, join := "tab_spend s", "JOIN tab_promote p ON p.source_db=s.source_db AND p.id=s.promote_id"
	if db.Dialector.Name() == "mysql" {
		join = "STRAIGHT_JOIN tab_promote p ON p.source_db=s.source_db AND p.id=s.promote_id"
		if db.Migrator().HasIndex("tab_spend", "idx_sw_spend_effective_scan") {
			table += " FORCE INDEX (idx_sw_spend_effective_scan)"
		}
	}
	q := db.Table(table).Joins(join).Where("s.pay_status=1 AND s.source_db IN ? AND p.create_time>0 AND s.pay_time>=p.create_time AND s.pay_time<=? AND s.pay_time<p.create_time+?", []string{"lx", "hw", "yq"}, now, windowSeconds)
	rows, err := q.Select("s.source_db,s.promote_id,s.pay_time,CAST(COALESCE(s.pay_amount,0) AS CHAR)").Order("s.promote_id,s.source_db,s.pay_time,s.id").Rows()
	if err != nil {
		return business.ErrStorage
	}
	defer rows.Close()
	for rows.Next() {
		var c business.EffectivenessCharge
		if err := rows.Scan(&c.SourceDB, &c.ID, &c.PaidAt, &c.Amount); err != nil {
			return business.ErrStorage
		}
		if selected[c.ChannelKey] {
			if err := consume(c); err != nil {
				return err
			}
		}
	}
	return storageError(rows.Err())
}
func (r *effectivenessRepo) Publish(ctx context.Context, expected *business.EffectivenessState, rule business.OtherSettings, key string, rows []business.EffectivenessResult, save bool) (*business.OtherSettings, error) {
	var saved = rule
	err := r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		var state effectivenessStatePO
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&state, "id=1").Error; err != nil {
			return err
		}
		if state.Revision != expected.Revision {
			return business.ErrEffectivenessConflict
		}
		if !save {
			var current otherSettingsPO
			result := tx.Where(map[string]any{"key": otherSettingsKey}).Limit(1).Find(&current)
			if result.Error != nil {
				return result.Error
			}
			active := business.DefaultOtherSettings()
			if result.RowsAffected > 0 {
				v, err := toBizOtherSettings(current)
				if err != nil {
					return err
				}
				active = *v
			}
			if business.EffectivenessRuleKey(active) != key {
				return business.ErrEffectivenessConflict
			}
		}
		now := time.Now()
		for start := 0; start < len(rows); start += 500 {
			pos := make([]effectivenessPO, 0, min(500, len(rows)-start))
			for _, row := range rows[start:min(start+500, len(rows))] {
				pos = append(pos, effectivenessPO{RuleKey: key, SourceDB: row.SourceDB, PromoteID: row.ID, QualifiedAt: row.QualifiedAt, WindowPay: row.WindowPay, Complete: row.Complete, UpdatedAt: now})
			}
			if err := tx.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "rule_key"}, {Name: "source_db"}, {Name: "promote_id"}}, DoUpdates: clause.AssignmentColumns([]string{"qualified_at", "window_pay", "complete", "updated_at"})}).Create(&pos).Error; err != nil {
				return err
			}
		}
		if save {
			po := newOtherSettings(rule)
			var old otherSettingsPO
			result := tx.Where(map[string]any{"key": otherSettingsKey}).Limit(1).Find(&old)
			if result.Error != nil {
				return result.Error
			}
			if result.RowsAffected == 0 {
				if err := tx.Create(&po).Error; err != nil {
					return err
				}
			} else {
				if err := tx.Model(&old).Updates(map[string]any{"value": po.Value, "updated_at": now}).Error; err != nil {
					return err
				}
			}
			saved.UpdatedAt = now
		}
		return tx.Model(&effectivenessStatePO{}).Where("id=1").Updates(map[string]any{"rule_key": key, "revision": state.Revision + 1, "built_at": now}).Error
	})
	if err != nil {
		return nil, storageError(err)
	}
	return &saved, nil
}