starpivot/apps/services/modules/memory/report.go

package memory

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"yunyan/comm"
	"yunyan/lego/base"
	"yunyan/lego/core"
	"yunyan/lego/core/cbase"
	"yunyan/lego/sys/cron"
	redissys "yunyan/lego/sys/redis"
	"yunyan/pb"

	"github.com/redis/go-redis/v9"
)

/*
周期报告：cron 建单入队 → N 个 worker BRPop → SQL 算 stat_json → LLM 组织播报文案。

三条要点，全是踩过或推演出来的：

 1. **金额与完成数走 SQL，LLM 只组织文案**。数字算错比话说得干巴严重得多。
 2. **锁与队列 key 必须过 redissys.RKey 加应用前缀**。测试机上多个应用共用一个
    Redis 实例，锁 key 不带前缀会让 A 应用抢到的锁挡住 B 应用整周不生成报告，
    而且安安静静什么都不报。
 3. **漏跑比重复更可能**。当前 compose 是单副本，"多副本重复生成"是前瞻性风险；
    真会发生的是 00:30 跑到一半容器重启、cron 不补跑，报告永远停在 pending。
    所以 memory_getreport 里有一条超时重新入队的补偿（见 requeueIfStale）。
*/

type reportComp struct {
	cbase.ModuleCompBase
	module  *Memory
	service base.IRPCXService
	options *Options
	ctx     context.Context
	cancel  context.CancelFunc
}

func (this *reportComp) Init(service core.IService, module core.IModule, comp core.IModuleComp, opt core.IModuleOptions) error {
	this.ModuleCompBase.Init(service, module, comp, opt)
	this.module = module.(*Memory)
	this.service = service.(base.IRPCXService)
	this.options = opt.(*Options)
	this.ctx, this.cancel = context.WithCancel(context.Background())
	return nil
}

func (this *reportComp) Start() error {
	if err := this.ModuleCompBase.Start(); err != nil {
		return err
	}
	for i := 0; i < this.options.MaxReportProcess; i++ {
		go this.consume(i)
	}
	if _, err := cron.AddFunc(this.options.ReportCronWeek, func() { this.tick(comm.MemoryPeriodWeek) }); err != nil {
		this.module.Errorf("注册周报 cron 失败 spec:%s err:%v", this.options.ReportCronWeek, err)
	}
	if _, err := cron.AddFunc(this.options.ReportCronMonth, func() { this.tick(comm.MemoryPeriodMonth) }); err != nil {
		this.module.Errorf("注册月报 cron 失败 spec:%s err:%v", this.options.ReportCronMonth, err)
	}
	this.module.Infof("周期报告已启动 workers:%d week:%q month:%q",
		this.options.MaxReportProcess, this.options.ReportCronWeek, this.options.ReportCronMonth)
	return nil
}

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

// ===== key（一律经 RKey 加应用前缀）=====

func (this *reportComp) awaitKey() string {
	return redissys.RKey(fmt.Sprintf("%s:%s", this.service.GetTag(), QueueReportAwait))
}

func (this *reportComp) lockKey(ptype, pkey string) string {
	return redissys.RKey(fmt.Sprintf("%s:%s:%s:%s", this.service.GetTag(), LockReportPrefix, ptype, pkey))
}

// ===== cron =====

// tick 到点建单入队。多副本时用 SET NX EX 抢锁，只有一个实例真的干活。
func (this *reportComp) tick(ptype string) {
	now := time.Now()
	var start, end time.Time
	var pkey string
	switch ptype {
	case comm.MemoryPeriodWeek:
		// 周一 00:30 跑的是「上周」，往前退一天落到上周日再取那一周
		prev := now.AddDate(0, 0, -1)
		start, end = comm.MemoryWeekRange(prev)
		pkey = comm.MemoryWeekKey(prev)
	case comm.MemoryPeriodMonth:
		prev := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location()).AddDate(0, 0, -1)
		start, end = comm.MemoryMonthRange(prev)
		pkey = comm.MemoryMonthKey(prev)
	default:
		return
	}

	// 锁存 2 小时：足够覆盖一次生成，又不会长到把下一个周期也挡掉
	ok, err := redissys.Conn().SetNX(this.ctx, this.lockKey(ptype, pkey), "1", 2*time.Hour).Result()
	if err != nil {
		this.module.Errorf("报告 cron 抢锁失败 %s/%s err:%v", ptype, pkey, err)
		return
	}
	if !ok {
		this.module.Infof("报告 cron %s/%s 已被其它实例领走，跳过", ptype, pkey)
		return
	}

	sd, ed := comm.FormatMemoryDate(start), comm.FormatMemoryDate(end)
	uids, err := this.module.model.uidsWithItemsBetween(sd, ed, this.options.ReportMinItems)
	if err != nil {
		this.module.Errorf("报告 cron 扫描用户失败 %s/%s err:%v", ptype, pkey, err)
		return
	}
	this.module.Infof("报告 cron %s/%s 范围 %s~%s 命中 %d 个用户", ptype, pkey, sd, ed, len(uids))

	for _, uid := range uids {
		rec := &pb.DBMemoryReport{
			Uid: uid, PeriodType: ptype, PeriodKey: pkey,
			PeriodStart: sd, PeriodEnd: ed,
			State: pb.MemoryReportState_MemoryReportState_Pending,
		}
		if err := this.module.model.upsertReport(rec); err != nil {
			this.module.Errorf("报告建单失败 uid:%s %s/%s err:%v", uid, ptype, pkey, err)
			continue
		}
		if rec.State == pb.MemoryReportState_MemoryReportState_Done && rec.Confirmed {
			continue // upsert 里对已确认的直接返回原记录，不重算
		}
		this.enqueueOnce(rec.Id)
	}
}

// procLockKey 单份报告的生成锁，防止 cron 与 ensureRecent 同时把它交给两个 worker。
func (this *reportComp) procLockKey(id uint64) string {
	return redissys.RKey(fmt.Sprintf("%s:%s:%d", this.service.GetTag(), LockReportProcPrefix, id))
}

// enqueueOnce 入队，但先看它在不在队列里。
//
// cron、requeueIfStale、ensureRecent 三条路都会入队，不判重的话同一份报告能在队列里
// 排好几次，每次出队都是一次 LLM 调用（且后跑的覆盖先跑的）。做法同 echomeet 的
// sweepStuckSummarize：LPos 找不到会返回 redis.Nil，所以 err==nil 才算命中。
func (this *reportComp) enqueueOnce(id uint64) {
	member := strconv.FormatUint(id, 10)
	if _, err := redissys.Conn().LPos(this.ctx, this.awaitKey(), member, redis.LPosArgs{}).Result(); err == nil {
		this.module.Debugf("报告 id:%d 已在队列里，跳过入队", id)
		return
	}
	if err := redissys.Conn().LPush(this.ctx, this.awaitKey(), member).Err(); err != nil {
		this.module.Errorf("报告入队失败 id:%d err:%v", id, err)
	}
}

// requeueIfStale 补偿：报告卡在 pending/processing 超过阈值就重新入队。
//
// 由 memory_getreport 驱动（同 echomeet 用 PollTranscribe 兜「转写回调丢了」的思路）。
// 没有这条，容器在 00:30 重启一次就够让那一周的报告永远停在 pending —— 而且不报任何错。
func (this *reportComp) requeueIfStale(rec *pb.DBMemoryReport) bool {
	if rec == nil {
		return false
	}
	if rec.State != pb.MemoryReportState_MemoryReportState_Pending &&
		rec.State != pb.MemoryReportState_MemoryReportState_Processing {
		return false
	}
	if time.Now().Unix()-rec.UpdateTime < this.options.ReportStaleSec {
		return false
	}
	this.module.Warnf("报告 id:%d 卡在 state:%d 超过 %ds，重新入队", rec.Id, rec.State, this.options.ReportStaleSec)
	rec.State = pb.MemoryReportState_MemoryReportState_Pending
	if err := this.module.model.saveReport(rec); err != nil {
		this.module.Errorf("报告 id:%d 重置状态失败 err:%v", rec.Id, err)
		return false
	}
	this.enqueueOnce(rec.Id)
	return true
}

// ===== 用户回来时的补偿 =====

/*
ensureRecent 是 requeueIfStale 之外的第二道补偿，管的是**它管不到的那些情况**。

原先只有 requeueIfStale 一条路，而它由 memory_getreport 驱动；客户端调的是无参版本，
服务端于是走 latestUnconfirmedReport —— 那条 SQL **只查 state=Done**。
也就是说 pending / processing / failed 的报告永远不会被查出来，也就永远不会被重新入队，
注释里写的「容器 00:30 重启一次就靠它救」实际一次都没救到过。

三种卡死，这里一并兜住：

  - cron 那一刻容器不在线 → 那一周连 memory_report 行都没建，下周 tick 算的是新的
    period_key，不回头补；
  - 建了单但卡在 pending / processing（跑到一半重启）；
  - state=Failed 之后再没人碰过它，下一次 tick 已是新周期，旧的永远 Failed。

只看「最近一个已结束的周」和「最近一个已结束的月」：与「弹窗只弹最近一份」口径一致，
也避免半年没开 App 的用户一回来就给他建二十份报告、连着烧二十次 LLM。
*/

// reportPeriod 一个待确认的周期
type reportPeriod struct {
	ptype string
	key   string
	start string // YYYY-MM-DD
	end   string
}

// lastFinishedWeek / lastFinishedMonth 与 tick 里的算法保持同一口径：
// 往前退到上一个周期里的任意一天，再取那个周期的范围与键。
func lastFinishedWeek(now time.Time) reportPeriod {
	prev := now.AddDate(0, 0, -7)
	start, end := comm.MemoryWeekRange(prev)
	return reportPeriod{comm.MemoryPeriodWeek, comm.MemoryWeekKey(prev),
		comm.FormatMemoryDate(start), comm.FormatMemoryDate(end)}
}

func lastFinishedMonth(now time.Time) reportPeriod {
	prev := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location()).AddDate(0, 0, -1)
	start, end := comm.MemoryMonthRange(prev)
	return reportPeriod{comm.MemoryPeriodMonth, comm.MemoryMonthKey(prev),
		comm.FormatMemoryDate(start), comm.FormatMemoryDate(end)}
}

// maxReportRetry 失败报告最多自动重试几次。
// 不设上限的话，一个必然失败的原因（比如统计 SQL 撞上坏数据）会让用户每次切前台
// 都触发一次重试，日志刷屏、LLM 白烧。用完这几次就等人来查。
const maxReportRetry = 3

// ensureRecent 用户回来时（memory_today）补建 / 重跑最近两个周期的报告。
//
// ⚠️ 必须在 goroutine 里调：它有两次主键查询、至多一次 count，不该挂在 today 的响应路径上。
// 自带 recover —— 它是附赠功能，炸了也不能连累弹窗。
func (this *reportComp) ensureRecent(uid string) {
	defer func() {
		if r := recover(); r != nil {
			this.module.Errorf("报告补偿 panic uid:%s err:%v", uid, r)
		}
	}()
	if uid == "" {
		return
	}
	now := time.Now()
	for _, p := range []reportPeriod{lastFinishedWeek(now), lastFinishedMonth(now)} {
		rec, err := this.module.model.getReportByPeriod(uid, p.ptype, p.key)
		if err != nil {
			this.module.Warnf("报告补偿 uid:%s %s/%s 查询失败已跳过: %v", uid, p.ptype, p.key, err)
			continue
		}
		switch {
		case rec == nil:
			this.ensureBuilt(uid, p)
		case rec.State == pb.MemoryReportState_MemoryReportState_Pending,
			rec.State == pb.MemoryReportState_MemoryReportState_Processing:
			// 复用同一套超时判定，别在这里另写一份阈值
			this.requeueIfStale(rec)
		case rec.State == pb.MemoryReportState_MemoryReportState_Failed && !rec.Confirmed:
			this.retryFailed(rec)
		}
	}
}

// ensureBuilt cron 那一刻容器不在线 → 补建。必须过与 cron 同一道条数闸门，
// 否则会给「上周只记了一条」的用户建出 cron 本来就不打算建的报告。
func (this *reportComp) ensureBuilt(uid string, p reportPeriod) {
	n, err := this.module.model.countItemsBetween(uid, p.start, p.end)
	if err != nil {
		this.module.Warnf("报告补偿 uid:%s %s/%s 统计条数失败已跳过: %v", uid, p.ptype, p.key, err)
		return
	}
	if n < int64(this.options.ReportMinItems) {
		return // 条数不够，本来就不该有报告，不是卡死
	}
	rec := &pb.DBMemoryReport{
		Uid: uid, PeriodType: p.ptype, PeriodKey: p.key,
		PeriodStart: p.start, PeriodEnd: p.end,
		State: pb.MemoryReportState_MemoryReportState_Pending,
	}
	if err := this.module.model.upsertReport(rec); err != nil {
		this.module.Errorf("报告补偿建单失败 uid:%s %s/%s err:%v", uid, p.ptype, p.key, err)
		return
	}
	this.module.Infof("报告补偿 uid:%s %s/%s cron 当时未建单（%d 条记录），现在补上", uid, p.ptype, p.key, n)
	this.enqueueOnce(rec.Id)
}

// retryFailed 失败的报告重新排一次，至多 maxReportRetry 次。
// 同样受 ReportStaleSec 节流：一次失败之后至少隔那么久才会再试。
func (this *reportComp) retryFailed(rec *pb.DBMemoryReport) {
	if time.Now().Unix()-rec.UpdateTime < this.options.ReportStaleSec {
		return
	}
	n := retryCount(rec.ErrorMsg)
	if n >= maxReportRetry {
		this.module.Warnf("报告 id:%d 已重试 %d 次仍失败，不再自动重试: %s", rec.Id, n, rec.ErrorMsg)
		return
	}
	rec.ErrorMsg = withRetryCount(rec.ErrorMsg, n+1)
	rec.State = pb.MemoryReportState_MemoryReportState_Pending
	if err := this.module.model.saveReport(rec); err != nil {
		this.module.Errorf("报告 id:%d 重置为待生成失败 err:%v", rec.Id, err)
		return
	}
	this.module.Warnf("报告 id:%d 上次失败，第 %d 次重试", rec.Id, n+1)
	this.enqueueOnce(rec.Id)
}

// 重试次数记在 error_msg 的 "retry=N;" 前缀里。
//
// 为什么不加一列：那要改 proto 重生成 pb（生成器版本那个坑）再改表结构，
// 而这个计数只在「失败之后」有意义、失败本来就要写 error_msg，寄在它前面代价最小。
// 展示端（后台/日志）看到的就是 "retry=2;统计失败: ..."，一眼能看出重试过几次。
const retryPrefix = "retry="

func retryCount(errMsg string) int {
	if !strings.HasPrefix(errMsg, retryPrefix) {
		return 0
	}
	i := strings.Index(errMsg, ";")
	if i < 0 {
		return 0
	}
	n, err := strconv.Atoi(errMsg[len(retryPrefix):i])
	if err != nil || n < 0 {
		return 0
	}
	return n
}

// withRetryCount 换掉（或加上）前缀，保留原来的错误正文。
func withRetryCount(errMsg string, n int) string {
	body := errMsg
	if strings.HasPrefix(errMsg, retryPrefix) {
		if i := strings.Index(errMsg, ";"); i >= 0 {
			body = errMsg[i+1:]
		}
	}
	return fmt.Sprintf("%s%d;%s", retryPrefix, n, body)
}

// ===== worker =====

func (this *reportComp) consume(idx int) {
	key := this.awaitKey()
	for {
		select {
		case <-this.ctx.Done():
			return
		default:
		}
		res, err := redissys.Conn().BRPop(this.ctx, 5*time.Second, key).Result()
		if err != nil {
			continue // 超时是常态，不刷日志
		}
		if len(res) < 2 {
			continue
		}
		id, e := strconv.ParseUint(res[1], 10, 64)
		if e != nil {
			continue
		}
		this.process(id)
	}
}

func (this *reportComp) process(id uint64) {
	defer func() {
		if r := recover(); r != nil {
			this.module.Errorf("报告生成 panic id:%d err:%v", id, r)
		}
	}()

	// 同一份报告同一时刻只许一个 worker 真的去跑。
	// cron 建单入队、ensureRecent 补偿入队两条路撞上时，没有这把锁就是两次 LLM 调用，
	// 后完成的覆盖先完成的。600s 覆盖一次生成（LLM 超时 60s），拿不到就直接返回——
	// 说明别人正在跑，本次不用做任何事。
	lock := this.procLockKey(id)
	ok, err := redissys.Conn().SetNX(this.ctx, lock, "1", 600*time.Second).Result()
	if err != nil {
		this.module.Warnf("报告 id:%d 抢生成锁失败，仍继续生成: %v", id, err)
	} else if !ok {
		this.module.Infof("报告 id:%d 正在别处生成，跳过", id)
		return
	} else {
		defer redissys.Conn().Del(this.ctx, lock)
	}

	rec, err := this.module.model.getReport(id)
	if err != nil || rec == nil {
		this.module.Errorf("报告生成取记录失败 id:%d err:%v", id, err)
		return
	}
	if rec.State == pb.MemoryReportState_MemoryReportState_Done {
		return
	}
	rec.State = pb.MemoryReportState_MemoryReportState_Processing
	_ = this.module.model.saveReport(rec)

	// 1) 统计走 SQL，确定性的
	st, err := this.module.computeStats(rec.Uid, rec.PeriodStart, rec.PeriodEnd)
	if err != nil {
		this.fail(rec, "统计失败: "+err.Error())
		return
	}
	if !st.hasContent() {
		// 到这儿说明入队后数据被删光了。空报告没有播报价值，直接标完成并确认掉，
		// 免得它挂在「待确认」里天天弹。
		rec.StatJson = toJSON(st)
		rec.Summary = ""
		rec.State = pb.MemoryReportState_MemoryReportState_Done
		rec.Confirmed = true
		rec.ConfirmTime = time.Now().Unix()
		_ = this.module.model.saveReport(rec)
		return
	}
	rec.StatJson = toJSON(st)

	// 2) LLM 只把 stat_json 组织成一段适合朗读的话
	if this.module.echo != nil {
		ctx, cancel := context.WithTimeout(this.ctx, 60*time.Second)
		text, svcId, e := this.module.echo.ChatLLM(ctx, "", this.options.ReportPrompt, rec.StatJson)
		cancel()
		rec.LlmSvcId = svcId
		if e != nil {
			// 文案生成失败**不算整体失败**：统计数据是完整的，客户端拿 stat_json
			// 一样能显示，只是少一段朗读文案。宁可少一句话，不可让报告整个没有。
			this.module.Warnf("报告 id:%d 文案生成失败，仅保留统计 err:%v", rec.Id, e)
		} else {
			rec.Summary = text
		}
	}

	rec.State = pb.MemoryReportState_MemoryReportState_Done
	rec.ErrorMsg = ""
	if err := this.module.model.saveReport(rec); err != nil {
		this.module.Errorf("报告 id:%d 回写失败 err:%v", rec.Id, err)
		return
	}
	this.module.Infof("报告生成完成 id:%d uid:%s %s/%s svc:%s",
		rec.Id, rec.Uid, rec.PeriodType, rec.PeriodKey, rec.LlmSvcId)
}

func (this *reportComp) fail(rec *pb.DBMemoryReport, msg string) {
	rec.State = pb.MemoryReportState_MemoryReportState_Failed
	if len([]rune(msg)) > 400 {
		msg = string([]rune(msg)[:400])
	}
	// ⚠️ 保住已有的 "retry=N;" 前缀：直接覆盖会把计数清零，
	// 于是一份永远失败的报告可以无限重试下去。
	if n := retryCount(rec.ErrorMsg); n > 0 {
		msg = withRetryCount(msg, n)
	}
	rec.ErrorMsg = msg
	_ = this.module.model.saveReport(rec)
	this.module.Errorf("报告生成失败 id:%d uid:%s %s", rec.Id, rec.Uid, msg)
}

func toJSON(v interface{}) string {
	b, err := json.Marshal(v)
	if err != nil {
		return "{}"
	}
	return string(b)
}