234 lines
7.0 KiB
Go
234 lines
7.0 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"log/slog"
|
||
"sync"
|
||
"time"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/logx"
|
||
"hyapp/services/room-service/internal/room/outbox"
|
||
)
|
||
|
||
const (
|
||
// roomRocketProgressCoalesceWindow 固定按 UTC 毫秒切 1 秒桶，把高频燃料变化压缩成客户端可接受的展示粒度。
|
||
roomRocketProgressCoalesceWindow = time.Second
|
||
// roomRocketProgressFlushInterval 控制到期 bucket 的扫描频率；100ms 让实际延迟稳定落在 1 秒窗口附近。
|
||
roomRocketProgressFlushInterval = 100 * time.Millisecond
|
||
)
|
||
|
||
type roomRocketProgressPending struct {
|
||
record outbox.Record
|
||
robotLane bool
|
||
rocketID string
|
||
level int32
|
||
currentFuel int64
|
||
roomVersion int64
|
||
}
|
||
|
||
type roomRocketProgressKey struct {
|
||
appCode string
|
||
roomID string
|
||
rocketID string
|
||
level int32
|
||
robotLane bool
|
||
windowStartMS int64
|
||
}
|
||
|
||
type roomRocketProgressEntry struct {
|
||
key roomRocketProgressKey
|
||
record outbox.Record
|
||
roomVersion int64
|
||
currentFuel int64
|
||
flushAtMS int64
|
||
}
|
||
|
||
type roomRocketProgressCoalescer struct {
|
||
mu sync.Mutex
|
||
repository Repository
|
||
pending map[roomRocketProgressKey]roomRocketProgressEntry
|
||
}
|
||
|
||
func newRoomRocketProgressCoalescer(repository Repository) *roomRocketProgressCoalescer {
|
||
return &roomRocketProgressCoalescer{
|
||
repository: repository,
|
||
pending: make(map[roomRocketProgressKey]roomRocketProgressEntry),
|
||
}
|
||
}
|
||
|
||
func (c *roomRocketProgressCoalescer) enqueue(progress roomRocketProgressPending) {
|
||
if c == nil || progress.record.Envelope == nil || progress.rocketID == "" || progress.level <= 0 {
|
||
// 入队只接受已经构造成 outbox envelope 的火箭进度；异常输入说明调用方没有走标准 SendGift 构造链路。
|
||
return
|
||
}
|
||
|
||
record := progress.record
|
||
record.AppCode = appcode.Normalize(record.AppCode)
|
||
record.Envelope.AppCode = record.AppCode
|
||
if record.CreatedAtMS <= 0 {
|
||
record.CreatedAtMS = record.Envelope.GetOccurredAtMs()
|
||
}
|
||
if progress.roomVersion <= 0 {
|
||
progress.roomVersion = record.Envelope.GetRoomVersion()
|
||
}
|
||
windowStartMS := roomRocketProgressWindowStartMS(record.CreatedAtMS)
|
||
key := roomRocketProgressKey{
|
||
appCode: record.AppCode,
|
||
roomID: record.RoomID,
|
||
rocketID: progress.rocketID,
|
||
level: progress.level,
|
||
robotLane: progress.robotLane,
|
||
windowStartMS: windowStartMS,
|
||
}
|
||
entry := roomRocketProgressEntry{
|
||
key: key,
|
||
record: record,
|
||
roomVersion: progress.roomVersion,
|
||
currentFuel: progress.currentFuel,
|
||
flushAtMS: windowStartMS + roomRocketProgressCoalesceWindow.Milliseconds(),
|
||
}
|
||
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
c.mergeLocked(entry)
|
||
}
|
||
|
||
func (c *roomRocketProgressCoalescer) flushDue(ctx context.Context, nowMS int64) ([]outbox.Record, error) {
|
||
if c == nil || c.repository == nil {
|
||
return nil, nil
|
||
}
|
||
due := c.takeDue(nowMS)
|
||
if len(due) == 0 {
|
||
return nil, nil
|
||
}
|
||
normalEntries, robotRecords := splitRoomRocketProgressDue(due)
|
||
if err := c.saveNormalDue(ctx, normalEntries); err != nil {
|
||
// 主 outbox 保存失败时只把主链路进度放回队列；机器人展示允许丢失，仍交给直发 IM best-effort 处理。
|
||
c.requeue(normalEntries, nowMS+roomRocketProgressFlushInterval.Milliseconds())
|
||
return robotRecords, err
|
||
}
|
||
return robotRecords, nil
|
||
}
|
||
|
||
func (c *roomRocketProgressCoalescer) takeDue(nowMS int64) []roomRocketProgressEntry {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
due := make([]roomRocketProgressEntry, 0)
|
||
for key, entry := range c.pending {
|
||
if entry.flushAtMS > nowMS {
|
||
continue
|
||
}
|
||
due = append(due, entry)
|
||
delete(c.pending, key)
|
||
}
|
||
return due
|
||
}
|
||
|
||
func (c *roomRocketProgressCoalescer) requeue(entries []roomRocketProgressEntry, nextFlushAtMS int64) {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
for _, entry := range entries {
|
||
entry.flushAtMS = nextFlushAtMS
|
||
c.mergeLocked(entry)
|
||
}
|
||
}
|
||
|
||
func (c *roomRocketProgressCoalescer) mergeLocked(entry roomRocketProgressEntry) {
|
||
current, exists := c.pending[entry.key]
|
||
if !exists || roomRocketProgressEntryNewer(entry, current) {
|
||
c.pending[entry.key] = entry
|
||
}
|
||
}
|
||
|
||
func splitRoomRocketProgressDue(entries []roomRocketProgressEntry) ([]roomRocketProgressEntry, []outbox.Record) {
|
||
normal := make([]roomRocketProgressEntry, 0, len(entries))
|
||
robotRecords := make([]outbox.Record, 0)
|
||
for _, entry := range entries {
|
||
record := entry.record
|
||
record.AppCode = appcode.Normalize(record.AppCode)
|
||
if record.Envelope != nil {
|
||
record.Envelope.AppCode = record.AppCode
|
||
}
|
||
if entry.key.robotLane {
|
||
robotRecords = append(robotRecords, record)
|
||
continue
|
||
}
|
||
normal = append(normal, entry)
|
||
}
|
||
return normal, robotRecords
|
||
}
|
||
|
||
func (c *roomRocketProgressCoalescer) saveNormalDue(ctx context.Context, entries []roomRocketProgressEntry) error {
|
||
normal := make(map[string][]outbox.Record)
|
||
for _, entry := range entries {
|
||
record := entry.record
|
||
record.AppCode = appcode.Normalize(record.AppCode)
|
||
if record.Envelope != nil {
|
||
record.Envelope.AppCode = record.AppCode
|
||
}
|
||
normal[record.AppCode] = append(normal[record.AppCode], record)
|
||
}
|
||
|
||
for scopedApp, records := range normal {
|
||
// repository 会按 context app_code 二次归一，分 app 批量写避免跨租户记录被同一个上下文覆盖。
|
||
if err := c.repository.SaveOutbox(appcode.WithContext(ctx, scopedApp), records); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func roomRocketProgressEntryNewer(candidate roomRocketProgressEntry, current roomRocketProgressEntry) bool {
|
||
if candidate.roomVersion != current.roomVersion {
|
||
return candidate.roomVersion > current.roomVersion
|
||
}
|
||
if candidate.currentFuel != current.currentFuel {
|
||
return candidate.currentFuel > current.currentFuel
|
||
}
|
||
return candidate.record.CreatedAtMS >= current.record.CreatedAtMS
|
||
}
|
||
|
||
func roomRocketProgressWindowStartMS(createdAtMS int64) int64 {
|
||
windowMS := roomRocketProgressCoalesceWindow.Milliseconds()
|
||
if createdAtMS <= 0 {
|
||
return 0
|
||
}
|
||
return createdAtMS / windowMS * windowMS
|
||
}
|
||
|
||
// FlushRoomRocketProgress 把已经超过 1 秒窗口的火箭进度写回主 outbox；机器人进度只 best-effort 直发 IM。
|
||
func (s *Service) FlushRoomRocketProgress(ctx context.Context) error {
|
||
if s == nil || s.roomRocketProgressCoalescer == nil {
|
||
return nil
|
||
}
|
||
robotRecords, err := s.roomRocketProgressCoalescer.flushDue(ctx, s.clock.Now().UTC().UnixMilli())
|
||
s.publishRobotDisplayRecordsBestEffort(ctx, robotRecords)
|
||
return err
|
||
}
|
||
|
||
// RunRoomRocketProgressFlushWorker 周期性把内存合并后的火箭进度落入主 outbox；机器人展示进度不经过 MySQL。
|
||
func (s *Service) RunRoomRocketProgressFlushWorker(ctx context.Context, interval time.Duration) {
|
||
if interval <= 0 {
|
||
interval = roomRocketProgressFlushInterval
|
||
}
|
||
if err := s.FlushRoomRocketProgress(ctx); err != nil && ctx.Err() == nil {
|
||
logx.Warn(ctx, "room_rocket_progress_flush_failed", slog.String("error", err.Error()))
|
||
}
|
||
|
||
ticker := time.NewTicker(interval)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
if err := s.FlushRoomRocketProgress(ctx); err != nil && ctx.Err() == nil {
|
||
logx.Warn(ctx, "room_rocket_progress_flush_failed", slog.String("error", err.Error()))
|
||
}
|
||
}
|
||
}
|
||
}
|