88 lines
3.0 KiB
Go
88 lines
3.0 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"hyapp/pkg/appcode"
|
|
"hyapp/pkg/logx"
|
|
"hyapp/services/room-service/internal/room/outbox"
|
|
)
|
|
|
|
const (
|
|
// robotDisplayPublishTimeout 是机器人展示直发 IM 的单条上限；超时或失败只记日志，也绝不能补写数据库或 MQ。
|
|
robotDisplayPublishTimeout = 3 * time.Second
|
|
// robotDisplayMaxConcurrentPublishes 给高频机器人展示设置进程级硬上限，避免腾讯 IM 变慢时无界堆积 goroutine。
|
|
robotDisplayMaxConcurrentPublishes = 32
|
|
)
|
|
|
|
func (s *Service) publishRobotDisplayRecordsBestEffort(ctx context.Context, records []outbox.Record) {
|
|
if s == nil || len(records) == 0 {
|
|
return
|
|
}
|
|
if s.robotDisplayPublisher == nil {
|
|
// 缺少腾讯 IM publisher 时只把整批纯展示事件视为 best-effort 投递失败，不创建持久化补偿事实。
|
|
logx.Warn(ctx, "robot_display_im_publish_skipped",
|
|
slog.String("node_id", s.nodeID),
|
|
slog.Int("record_count", len(records)),
|
|
slog.String("reason", "publisher_not_configured"),
|
|
)
|
|
return
|
|
}
|
|
scopedApp := appcode.FromContext(ctx)
|
|
copied := append([]outbox.Record(nil), records...)
|
|
for index := range copied {
|
|
copied[index].AppCode = appcode.Normalize(firstNonEmpty(copied[index].AppCode, scopedApp))
|
|
if copied[index].Envelope != nil {
|
|
copied[index].Envelope.AppCode = copied[index].AppCode
|
|
}
|
|
}
|
|
sortOutboxRecords(copied)
|
|
s.robotDisplayPublishOnce.Do(func() {
|
|
s.robotDisplayPublishSlots = make(chan struct{}, robotDisplayMaxConcurrentPublishes)
|
|
})
|
|
select {
|
|
case s.robotDisplayPublishSlots <- struct{}{}:
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
// 机器人消息没有业务事实语义；发布槽位耗尽时只丢当前展示，不能排队到 DB、MQ 或无界内存。
|
|
logx.Warn(ctx, "robot_display_im_publish_skipped",
|
|
slog.String("node_id", s.nodeID),
|
|
slog.Int("record_count", len(copied)),
|
|
slog.String("reason", "publisher_busy"),
|
|
)
|
|
return
|
|
}
|
|
|
|
go func(records []outbox.Record) {
|
|
defer func() { <-s.robotDisplayPublishSlots }()
|
|
for _, record := range records {
|
|
if record.Envelope == nil {
|
|
logx.Warn(ctx, "robot_display_im_publish_skipped",
|
|
slog.String("node_id", s.nodeID),
|
|
slog.String("event_id", record.EventID),
|
|
slog.String("event_type", record.EventType),
|
|
slog.String("room_id", record.RoomID),
|
|
slog.String("reason", "nil_envelope"),
|
|
)
|
|
continue
|
|
}
|
|
publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), robotDisplayPublishTimeout)
|
|
err := s.robotDisplayPublisher.PublishOutboxEvent(publishCtx, record.Envelope)
|
|
cancel()
|
|
if err != nil {
|
|
// 机器人展示事件允许极小概率丢失；这里只有日志且不补偿，房间快照只保证真实房态不被这次失败污染。
|
|
logx.Warn(ctx, "robot_display_im_publish_failed",
|
|
slog.String("node_id", s.nodeID),
|
|
slog.String("event_id", record.EventID),
|
|
slog.String("event_type", record.EventType),
|
|
slog.String("room_id", record.RoomID),
|
|
slog.String("error", trimOutboxError(err.Error())),
|
|
)
|
|
}
|
|
}
|
|
}(copied)
|
|
}
|