starpivot/apps/services/modules/api/stat_consumer.go

package api

import (
	"context"
	"yunyan/comm"
	"yunyan/lego/core"
	"yunyan/lego/core/cbase"
	"yunyan/lego/sys/log"
	"yunyan/lego/sys/mysql"
	redissys "yunyan/lego/sys/redis"
	"yunyan/pb"
	"yunyan/sys/appstat"
	"encoding/json"
	"sync"
	"time"
)

// statConsumerComp 消费 Redis 统计队列并写入日/月/年三张 DB 表。
// 内存中缓存当前日/月/年三行数据，启动时从 DB 加载一次；
// 每次 flush 在内存累加后立即写入 DB，日期切换时加载新行。
type statConsumerComp struct {
	cbase.ModuleCompBase
	module *API
	cancel context.CancelFunc

	// 内存缓存（三个时间粒度 + 全局累计）
	// mu 保护以下四个缓存指针的读写，避免 consumer flush 与外部全量重建并发改写。
	mu      sync.Mutex
	daily   *pb.DBAppStat
	monthly *pb.DBAppStat
	yearly  *pb.DBAppStat
	global  *pb.DBAppStat
}

// 全局累计行固定 stat_date 标识
const GlobalStatDate = "all"

func (this *statConsumerComp) Init(service core.IService, module core.IModule, comp core.IModuleComp, opt core.IModuleOptions) (err error) {
	this.ModuleCompBase.Init(service, module, comp, opt)
	this.module = module.(*API)
	return
}

func (this *statConsumerComp) Start() (err error) {
	if err = this.ModuleCompBase.Start(); err != nil {
		return
	}
	// 启动时从 DB 加载当前日/月/年缓存行
	this.loadCache(time.Now())

	ctx, cancel := context.WithCancel(context.Background())
	this.cancel = cancel
	go this.consume(ctx)
	log.Infoln("appstat consumer started")
	return
}

func (this *statConsumerComp) Destroy() (err error) {
	if this.cancel != nil {
		this.cancel()
	}
	return this.ModuleCompBase.Destroy()
}

// loadCache 从 DB 加载指定时间点对应的日/月/年/全局行（不存在则初始化空行）
func (this *statConsumerComp) loadCache(t time.Time) {
	this.daily = this.loadOrNew(comm.TableAppStatDaily, t.Format("2006-01-02"))
	this.monthly = this.loadOrNew(comm.TableAppStatMonthly, t.Format("2006-01"))
	this.yearly = this.loadOrNew(comm.TableAppStatYearly, t.Format("2006"))
	this.global = this.loadOrNew(comm.TableAppStatGlobal, GlobalStatDate)
}

func (this *statConsumerComp) loadOrNew(tbl, date string) *pb.DBAppStat {
	var row pb.DBAppStat
	if err := mysql.FindOne(tbl, &row, "stat_date=?", date); err != nil {
		row = pb.DBAppStat{StatDate: date}
	}
	return &row
}

// consume 从 Redis 队列消费统计增量
func (this *statConsumerComp) consume(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}
		vals, err := redissys.Conn().BRPop(ctx, 5*time.Second, redissys.RKey(appstat.RedisQueueKey)).Result()
		if err != nil {
			continue
		}
		if len(vals) < 2 {
			continue
		}
		var delta pb.DBAppStat
		if err = json.Unmarshal([]byte(vals[1]), &delta); err != nil {
			this.module.Warn("stat consumer unmarshal", log.Field{Key: "err", Value: err.Error()})
			continue
		}
		this.flush(&delta)
	}
}

// checkExpired 各粒度独立检查是否过期，过期则创建新的空统计对象；全局行不过期
func (this *statConsumerComp) checkExpired(now time.Time) {
	if today := now.Format("2006-01-02"); this.daily == nil || this.daily.StatDate != today {
		this.daily = &pb.DBAppStat{StatDate: today}
	}
	if month := now.Format("2006-01"); this.monthly == nil || this.monthly.StatDate != month {
		this.monthly = &pb.DBAppStat{StatDate: month}
	}
	if year := now.Format("2006"); this.yearly == nil || this.yearly.StatDate != year {
		this.yearly = &pb.DBAppStat{StatDate: year}
	}
	if this.global == nil {
		this.global = this.loadOrNew(comm.TableAppStatGlobal, GlobalStatDate)
	}
}

// flush 将 delta 累加到内存缓存后立即写入 DB
func (this *statConsumerComp) flush(delta *pb.DBAppStat) {
	if len(delta.StatDate) < 10 {
		this.module.Warn("stat consumer flush: invalid StatDate", log.Field{Key: "date", Value: delta.StatDate})
		return
	}
	this.mu.Lock()
	defer this.mu.Unlock()
	this.checkExpired(time.Now())

	accum(this.daily, delta)
	accum(this.monthly, delta)
	accum(this.yearly, delta)
	accum(this.global, delta)

	this.saveRow(comm.TableAppStatDaily, this.daily)
	this.saveRow(comm.TableAppStatMonthly, this.monthly)
	this.saveRow(comm.TableAppStatYearly, this.yearly)
	this.saveRow(comm.TableAppStatGlobal, this.global)
}

// ReloadCache 由外部（如全量重建）在改写 DB 后调用，丢弃过期内存缓存重新从 DB 加载。
func (this *statConsumerComp) ReloadCache() {
	this.mu.Lock()
	defer this.mu.Unlock()
	this.loadCache(time.Now())
}

// WithLock 提供给外部对四张统计表执行原子改写（清空+重建）使用，
// 期间 consumer flush 被阻塞，避免被半成品状态覆盖。
func (this *statConsumerComp) WithLock(fn func() error) error {
	this.mu.Lock()
	defer this.mu.Unlock()
	return fn()
}

func (this *statConsumerComp) saveRow(tbl string, row *pb.DBAppStat) {
	if row == nil {
		return
	}
	row.UpdateTime = time.Now().Unix()
	if err := mysql.Save(tbl, row); err != nil {
		this.module.Warn("stat consumer save", log.Field{Key: "table", Value: tbl}, log.Field{Key: "err", Value: err.Error()})
	}
}

// accum 将 delta 各字段累加到 row 上
func accum(row, delta *pb.DBAppStat) {
	row.PayAmount += delta.PayAmount
	row.OrderCount += delta.OrderCount
	row.PayCount += delta.PayCount
	row.FailedOrderCount += delta.FailedOrderCount
	row.PayUserCount += delta.PayUserCount
	row.AddVipDays += delta.AddVipDays
	row.AddAiIntegral += delta.AddAiIntegral
	row.AddTradeSecond += delta.AddTradeSecond
	row.AddMeetSecond += delta.AddMeetSecond
	row.AiChatCount += delta.AiChatCount
	row.AiUpToken += delta.AiUpToken
	row.AiDownToken += delta.AiDownToken
	row.TradeCount += delta.TradeCount
	row.TradeTime += delta.TradeTime
	row.TradeWords += delta.TradeWords
	row.MeetCount += delta.MeetCount
	row.MeetTime += delta.MeetTime
	row.NewUserCount += delta.NewUserCount
	row.LoginCount += delta.LoginCount
	row.ActiveUserCount += delta.ActiveUserCount
	row.TotalUserCount = delta.TotalUserCount // 快照值，直接覆盖
	row.VipUserCount += delta.VipUserCount
	row.BindDeviceCount += delta.BindDeviceCount
	row.ActiveDeviceCount += delta.ActiveDeviceCount
}