zt-closet-backend/pkg/tokenhub/scheduler.go

package tokenhub

import (
	"context"
	"log/slog"
	"strconv"
	"sync"
	"time"

	"git.lzcwy.cn/zhengchiliu/zt-closet-backend/global"
	"git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/mysqldao"
	"git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/redisdao"
	"github.com/google/uuid"
)

// fortuneEnsureMu 用于同一实例内串行化当日运势兜底生成，避免并发重复触发 LLM。
var fortuneEnsureMu sync.Mutex

// StartFortuneScheduler 启动后台定时任务：每天 0:00 生成当日运势并入库。
func StartFortuneScheduler() {
	go func() {
		defer func() {
			if r := recover(); r != nil {
				slog.Error("fortune scheduler panic", "recover", r)
			}
		}()
		slog.Info("fortune scheduler started")

		client := NewClient()
		horoscopeDao := mysqldao.NewHoroscopeDao()

		for {
			// 每次循环重新计算到下一个午夜的等待时间，避免漂移累积。
			now := time.Now()
			next := time.Date(now.Year(), now.Month(), now.Day()+1, 0, 0, 0, 0, now.Location())
			wait := next.Sub(now)

			slog.Info("fortune scheduler: next run",
				"at", next.Format("2006-01-02 15:04:05"),
				"in", wait.Round(time.Second).String(),
			)

			time.Sleep(wait)

			// 生成 & 入库（带分布式锁，多实例仅一个执行）。
			func() {
				defer func() {
					if r := recover(); r != nil {
						slog.Error("fortune generation panic", "recover", r)
					}
				}()
				runFortuneGenerationWithLock(client, horoscopeDao)
			}()
		}
	}()
}

// runFortuneGeneration 调用混元生成运势，写入数据库。
func runFortuneGeneration(client *Client, horoscopeDao *mysqldao.HoroscopeDao) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()

	ctx, span := global.Tracer.Start(ctx, "scheduler.FortuneGeneration")
	defer span.End()

	slog.InfoContext(ctx, "generating daily fortunes...")

	result, err := client.GenerateDailyFortunes(ctx)
	if err != nil {
		span.RecordError(err)
		slog.ErrorContext(ctx, "generate fortunes failed", "error", err)
		return
	}

	// result.Date 形如 "2006-01-02"，转换为 Horoscope.StatDate 的 YYYYMMDD 整数。
	date, err := time.Parse("2006-01-02", result.Date)
	if err != nil {
		slog.ErrorContext(ctx, "parse fortune date failed", "date", result.Date, "error", err)
		return
	}
	statDate, err := strconv.ParseInt(date.Format("20060102"), 10, 64)
	if err != nil {
		slog.ErrorContext(ctx, "parse stat date failed", "date", result.Date, "error", err)
		return
	}

	successCount := 0
	for _, f := range result.Fortunes {
		horoscope := f.ToHoroscope(statDate)
		if err := horoscopeDao.HoroscopeCreate(ctx, horoscope); err != nil {
			slog.ErrorContext(ctx, "save horoscope failed", "zodiac", f.Zodiac, "error", err)
		} else {
			successCount++
		}
	}

	slog.InfoContext(ctx, "daily fortune generation completed",
		"total", len(result.Fortunes),
		"saved", successCount,
	)
}

// runFortuneGenerationWithLock 带 Redis 分布式锁执行当日运势生成：
// 多实例部署时仅一个实例执行；锁释放校验持有者 value，防止超时误删其它实例的锁。
func runFortuneGenerationWithLock(client *Client, horoscopeDao *mysqldao.HoroscopeDao) {
	today := time.Now().Format("2006-01-02")
	lockKey := "ztcloset:lock:fortune:daily:" + today
	lockVal := uuid.NewString()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()

	ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, 5*time.Minute)
	if lerr != nil || !ok {
		slog.Info("fortune generation already in progress, skip", "date", today)
		return
	}
	defer redisdao.Unlock(context.Background(), lockKey, lockVal)

	runFortuneGeneration(client, horoscopeDao)
}

// EnsureTodayFortunes 检查今日运势数据是否完整，不完整则自动生成。
// 该函数异步执行，适用于服务器启动时兜底。
func EnsureTodayFortunes() {
	go func() {
		defer func() {
			if r := recover(); r != nil {
				slog.Error("EnsureTodayFortunes panic", "recover", r)
			}
		}()
		// 进程内防重：已有兜底生成在进行中则直接跳过。
		if !fortuneEnsureMu.TryLock() {
			return
		}
		defer fortuneEnsureMu.Unlock()

		client := NewClient()
		horoscopeDao := mysqldao.NewHoroscopeDao()

		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
		defer cancel()

		now := time.Now()
		today := now.Format("2006-01-02")
		statDate, err := strconv.ParseInt(now.Format("20060102"), 10, 64)
		if err != nil {
			slog.Error("EnsureTodayFortunes: parse stat date failed", "error", err)
			return
		}

		horoscopes, err := horoscopeDao.HoroscopeGetByDate(ctx, statDate)
		if err != nil {
			slog.Warn("EnsureTodayFortunes: query today fortunes failed, will try to generate", "error", err, "date", today)
		}

		if len(horoscopes) >= 12 {
			slog.Info("EnsureTodayFortunes: today fortunes already complete, skip", "date", today, "count", len(horoscopes))
			return
		}

		slog.Info("EnsureTodayFortunes: today fortunes incomplete, generating...",
			"date", today,
			"existing", len(horoscopes),
			"needed", 12,
		)

		runFortuneGenerationWithLock(client, horoscopeDao)
	}()
}