234 lines
8.5 KiB
Go
234 lines
8.5 KiB
Go
package host
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"errors"
|
||
"fmt"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp/pkg/tencentim"
|
||
"hyapp/pkg/xerr"
|
||
hostdomain "hyapp/services/user-service/internal/domain/host"
|
||
)
|
||
|
||
const (
|
||
privateMessageHistoryHours = 7 * 24
|
||
privateMessageHistoryReadyDelay = 2 * time.Hour
|
||
privateMessageHistoryRetryDelay = 10 * time.Minute
|
||
privateMessageHistoryImportMax = 500
|
||
privateMessageHistoryEmptyMaxWait = 4 * time.Hour
|
||
)
|
||
|
||
// PrivateMessageHistoryClient 是腾讯 IM 离线归档边界；它只在 cron 批处理中调用，
|
||
// 不参与客户端发送、接收或 UserSig 签发链路。
|
||
type PrivateMessageHistoryClient interface {
|
||
ListC2CHistoryFiles(ctx context.Context, hourStartUTC time.Time) ([]tencentim.C2CHistoryFile, error)
|
||
StreamC2CHistory(ctx context.Context, file tencentim.C2CHistoryFile, consume func(tencentim.C2CHistoryMessage) error) (int, error)
|
||
}
|
||
|
||
// PrivateMessageHistoryRepository 保存小时同步租约和最小私聊统计事实。
|
||
type PrivateMessageHistoryRepository interface {
|
||
PreparePrivateMessageHistoryHours(ctx context.Context, sdkAppID int64, fromHourMS int64, throughHourMS int64, nowMS int64) error
|
||
ClaimPrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, runID string, fromHourMS int64, throughHourMS int64, nowMS int64, lockTTL time.Duration) (hourStartMS int64, attempts int, claimed bool, err error)
|
||
ImportPrivateMessageEvents(ctx context.Context, events []hostdomain.PrivateMessageEvent) (int, error)
|
||
CompletePrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, hourStartMS int64, runID string, sourceMessages int, importedEvents int, nowMS int64) error
|
||
RetryPrivateMessageHistoryHour(ctx context.Context, sdkAppID int64, hourStartMS int64, runID string, retryAtMS int64, errorMessage string, nowMS int64) error
|
||
}
|
||
|
||
// PrivateMessageHistoryBatchResult 汇总一次小时归档处理，不把下载地址或消息正文带到 cron 日志。
|
||
type PrivateMessageHistoryBatchResult struct {
|
||
Claimed bool
|
||
SourceCount int
|
||
ImportedCount int
|
||
RetryPending bool
|
||
}
|
||
|
||
// WithPrivateMessageHistorySource 挂载腾讯归档和 owner repository。SDKAppID 是共享归档主键，
|
||
// 即使 cron 按多个 app_code 并发触发，同一小时也只会被认领一次。
|
||
func WithPrivateMessageHistorySource(sdkAppID int64, client PrivateMessageHistoryClient) Option {
|
||
return func(s *Service) {
|
||
if sdkAppID <= 0 || client == nil {
|
||
return
|
||
}
|
||
repository, ok := s.repository.(PrivateMessageHistoryRepository)
|
||
if !ok {
|
||
return
|
||
}
|
||
s.privateMessageHistorySDKAppID = sdkAppID
|
||
s.privateMessageHistoryClient = client
|
||
s.privateMessageHistoryRepository = repository
|
||
}
|
||
}
|
||
|
||
// ProcessPrivateMessageHistoryBatch 处理最近七天中一个已完成的 UTC 小时。
|
||
// get_history 文件由腾讯异步生成，因此统计最多延迟约两小时，但不会给 IM 发消息增加等待。
|
||
func (s *Service) ProcessPrivateMessageHistoryBatch(ctx context.Context, runID string, lockTTL time.Duration, importBatchSize int) (PrivateMessageHistoryBatchResult, error) {
|
||
if s == nil || s.privateMessageHistoryClient == nil || s.privateMessageHistoryRepository == nil || s.privateMessageHistorySDKAppID <= 0 {
|
||
return PrivateMessageHistoryBatchResult{}, xerr.New(xerr.Unavailable, "private message history sync is not configured")
|
||
}
|
||
runID = strings.TrimSpace(runID)
|
||
if runID == "" {
|
||
return PrivateMessageHistoryBatchResult{}, xerr.New(xerr.InvalidArgument, "run_id is required")
|
||
}
|
||
if importBatchSize <= 0 || importBatchSize > privateMessageHistoryImportMax {
|
||
importBatchSize = privateMessageHistoryImportMax
|
||
}
|
||
if lockTTL <= 0 {
|
||
lockTTL = 5 * time.Minute
|
||
}
|
||
|
||
now := s.now().UTC()
|
||
throughHour := now.Truncate(time.Hour).Add(-privateMessageHistoryReadyDelay)
|
||
fromHour := throughHour.Add(-time.Duration(privateMessageHistoryHours-1) * time.Hour)
|
||
if err := s.privateMessageHistoryRepository.PreparePrivateMessageHistoryHours(
|
||
ctx,
|
||
s.privateMessageHistorySDKAppID,
|
||
fromHour.UnixMilli(),
|
||
throughHour.UnixMilli(),
|
||
now.UnixMilli(),
|
||
); err != nil {
|
||
return PrivateMessageHistoryBatchResult{}, err
|
||
}
|
||
|
||
hourStartMS, attempts, claimed, err := s.privateMessageHistoryRepository.ClaimPrivateMessageHistoryHour(
|
||
ctx,
|
||
s.privateMessageHistorySDKAppID,
|
||
runID,
|
||
fromHour.UnixMilli(),
|
||
throughHour.UnixMilli(),
|
||
now.UnixMilli(),
|
||
lockTTL,
|
||
)
|
||
if err != nil || !claimed {
|
||
return PrivateMessageHistoryBatchResult{}, err
|
||
}
|
||
result := PrivateMessageHistoryBatchResult{Claimed: true}
|
||
hourStart := time.UnixMilli(hourStartMS).UTC()
|
||
|
||
files, err := s.privateMessageHistoryClient.ListC2CHistoryFiles(ctx, hourStart)
|
||
if err != nil {
|
||
if tencentim.IsRESTErrorCode(err, 1004) {
|
||
// 1004 同时表示“无消息”和“文件尚未生成”。最近小时先重试；超过生成窗口后
|
||
// 视为空小时完成，避免一个永久空小时阻塞七天回补。
|
||
if attempts < 3 && now.Sub(hourStart) < privateMessageHistoryEmptyMaxWait {
|
||
if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, err, now); retryErr != nil {
|
||
return result, retryErr
|
||
}
|
||
result.RetryPending = true
|
||
return result, nil
|
||
}
|
||
if completeErr := s.privateMessageHistoryRepository.CompletePrivateMessageHistoryHour(
|
||
ctx, s.privateMessageHistorySDKAppID, hourStartMS, runID, 0, 0, now.UnixMilli(),
|
||
); completeErr != nil {
|
||
return result, completeErr
|
||
}
|
||
return result, nil
|
||
}
|
||
if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, err, now); retryErr != nil {
|
||
return result, retryErr
|
||
}
|
||
return result, err
|
||
}
|
||
|
||
pending := make([]hostdomain.PrivateMessageEvent, 0, importBatchSize)
|
||
flush := func() error {
|
||
if len(pending) == 0 {
|
||
return nil
|
||
}
|
||
imported, importErr := s.privateMessageHistoryRepository.ImportPrivateMessageEvents(ctx, pending)
|
||
if importErr != nil {
|
||
return importErr
|
||
}
|
||
result.ImportedCount += imported
|
||
pending = pending[:0]
|
||
return nil
|
||
}
|
||
consume := func(message tencentim.C2CHistoryMessage) error {
|
||
event, ok := privateMessageEventFromHistory(message)
|
||
if !ok {
|
||
return nil
|
||
}
|
||
pending = append(pending, event)
|
||
if len(pending) >= importBatchSize {
|
||
return flush()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
for _, file := range files {
|
||
processed, streamErr := s.privateMessageHistoryClient.StreamC2CHistory(ctx, file, consume)
|
||
result.SourceCount += processed
|
||
if streamErr != nil {
|
||
if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, streamErr, now); retryErr != nil {
|
||
return result, retryErr
|
||
}
|
||
return result, streamErr
|
||
}
|
||
}
|
||
if err := flush(); err != nil {
|
||
if retryErr := s.markPrivateMessageHistoryRetry(ctx, hourStartMS, runID, err, now); retryErr != nil {
|
||
return result, retryErr
|
||
}
|
||
return result, err
|
||
}
|
||
if err := s.privateMessageHistoryRepository.CompletePrivateMessageHistoryHour(
|
||
ctx,
|
||
s.privateMessageHistorySDKAppID,
|
||
hourStartMS,
|
||
runID,
|
||
result.SourceCount,
|
||
result.ImportedCount,
|
||
s.now().UTC().UnixMilli(),
|
||
); err != nil {
|
||
return result, err
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
func (s *Service) markPrivateMessageHistoryRetry(ctx context.Context, hourStartMS int64, runID string, cause error, now time.Time) error {
|
||
retryErr := s.privateMessageHistoryRepository.RetryPrivateMessageHistoryHour(
|
||
ctx,
|
||
s.privateMessageHistorySDKAppID,
|
||
hourStartMS,
|
||
runID,
|
||
now.Add(privateMessageHistoryRetryDelay).UnixMilli(),
|
||
cause.Error(),
|
||
now.UnixMilli(),
|
||
)
|
||
if retryErr == nil {
|
||
return nil
|
||
}
|
||
return errors.Join(cause, retryErr)
|
||
}
|
||
|
||
func privateMessageEventFromHistory(message tencentim.C2CHistoryMessage) (hostdomain.PrivateMessageEvent, bool) {
|
||
senderText := strings.TrimSpace(message.FromAccount)
|
||
targetText := strings.TrimSpace(message.ToAccount)
|
||
senderUserID, senderErr := strconv.ParseInt(senderText, 10, 64)
|
||
targetUserID, targetErr := strconv.ParseInt(targetText, 10, 64)
|
||
if senderErr != nil || targetErr != nil || senderUserID <= 0 || targetUserID <= 0 || message.Timestamp <= 0 {
|
||
// 共享 SDKAppID 可能包含管理员/第三方非数字 identifier；只有项目内部不可变 user_id
|
||
// 才能安全映射 users 主键，其余消息不属于 Host Center 统计事实。
|
||
return hostdomain.PrivateMessageEvent{}, false
|
||
}
|
||
identity := fmt.Sprintf(
|
||
"%s\x00%s\x00%d\x00%d\x00%d",
|
||
senderText,
|
||
targetText,
|
||
message.Timestamp,
|
||
message.Sequence,
|
||
message.Random,
|
||
)
|
||
hash := sha256.Sum256([]byte(identity))
|
||
return hostdomain.PrivateMessageEvent{
|
||
EventID: "im-history:" + hex.EncodeToString(hash[:]),
|
||
SenderUserID: senderUserID,
|
||
TargetUserID: targetUserID,
|
||
OccurredAtMS: message.Timestamp * int64(time.Second/time.Millisecond),
|
||
}, true
|
||
}
|