244 lines
8.3 KiB
Go
244 lines
8.3 KiB
Go
package message
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"hyapp/pkg/appcode"
|
|
"hyapp/pkg/xerr"
|
|
messagedomain "hyapp/services/activity-service/internal/domain/message"
|
|
)
|
|
|
|
const (
|
|
defaultFanoutPollInterval = time.Second
|
|
defaultFanoutLockTTL = 30 * time.Second
|
|
defaultFanoutMaxRetry = 8
|
|
fanoutProducer = "message-fanout"
|
|
fanoutProducerEventType = "message_fanout"
|
|
)
|
|
|
|
// FanoutWorkerOptions carries runtime knobs for the durable fanout worker.
|
|
type FanoutWorkerOptions struct {
|
|
WorkerID string
|
|
PollInterval time.Duration
|
|
LockTTL time.Duration
|
|
BatchSize int
|
|
MaxRetry int
|
|
}
|
|
|
|
type fanoutTargetFilter struct {
|
|
TargetUserID int64 `json:"target_user_id"`
|
|
UserID int64 `json:"user_id"`
|
|
UserIDs []int64 `json:"user_ids"`
|
|
RegionID int64 `json:"region_id"`
|
|
Country string `json:"country"`
|
|
}
|
|
|
|
type fanoutTemplateSnapshot struct {
|
|
ProducerEventType string `json:"producer_event_type"`
|
|
AggregateType string `json:"aggregate_type"`
|
|
AggregateID string `json:"aggregate_id"`
|
|
TemplateID string `json:"template_id"`
|
|
TemplateVersion string `json:"template_version"`
|
|
Title string `json:"title"`
|
|
Summary string `json:"summary"`
|
|
Body string `json:"body"`
|
|
IconURL string `json:"icon_url"`
|
|
ImageURL string `json:"image_url"`
|
|
ActionType string `json:"action_type"`
|
|
ActionParam string `json:"action_param"`
|
|
Priority int32 `json:"priority"`
|
|
SentAtMS int64 `json:"sent_at_ms"`
|
|
ExpireAtMS int64 `json:"expire_at_ms"`
|
|
MetadataJSON json.RawMessage `json:"metadata_json"`
|
|
}
|
|
|
|
// ProcessNextFanoutJob claims one job, processes one target page, then commits progress.
|
|
func (s *Service) ProcessNextFanoutJob(ctx context.Context, options FanoutWorkerOptions) (bool, error) {
|
|
if err := s.requireRepository(); err != nil {
|
|
return false, err
|
|
}
|
|
options = normalizeFanoutWorkerOptions(options, s.cfg.NodeID)
|
|
nowMS := s.now().UnixMilli()
|
|
job, claimed, err := s.repository.ClaimFanoutJob(ctx, options.WorkerID, nowMS, nowMS+options.LockTTL.Milliseconds(), options.MaxRetry)
|
|
if err != nil || !claimed {
|
|
return claimed, err
|
|
}
|
|
|
|
jobCtx := appcode.WithContext(ctx, job.AppCode)
|
|
progress, processErr := s.processFanoutJobPage(jobCtx, job, options)
|
|
if processErr != nil {
|
|
failErr := s.repository.FailFanoutJob(jobCtx, job.JobID, options.WorkerID, nowMS, nowMS+options.PollInterval.Milliseconds(), options.MaxRetry, processErr.Error())
|
|
if failErr != nil {
|
|
return true, failErr
|
|
}
|
|
return true, processErr
|
|
}
|
|
if err := s.repository.FinishFanoutJobBatch(jobCtx, job.JobID, options.WorkerID, progress.nextCursorUserID, progress.totalDelta, progress.successDelta, progress.failureDelta, progress.done, nowMS, progress.errorMessage); err != nil {
|
|
return true, err
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
type fanoutProgress struct {
|
|
nextCursorUserID int64
|
|
totalDelta int64
|
|
successDelta int64
|
|
failureDelta int64
|
|
done bool
|
|
errorMessage string
|
|
}
|
|
|
|
func (s *Service) processFanoutJobPage(ctx context.Context, job messagedomain.FanoutJob, options FanoutWorkerOptions) (fanoutProgress, error) {
|
|
var target fanoutTargetFilter
|
|
if err := json.Unmarshal([]byte(job.TargetFilterJSON), &target); err != nil {
|
|
return fanoutProgress{}, xerr.New(xerr.InvalidArgument, "target_filter_json is invalid")
|
|
}
|
|
var snapshot fanoutTemplateSnapshot
|
|
if err := json.Unmarshal([]byte(job.TemplateSnapshotJSON), &snapshot); err != nil {
|
|
return fanoutProgress{}, xerr.New(xerr.InvalidArgument, "template_snapshot_json is invalid")
|
|
}
|
|
pageSize := int(normalizeBatchSize(firstPositiveInt32(job.BatchSize, int32(options.BatchSize))))
|
|
userIDs, nextCursor, done, err := s.fanoutTargetPage(ctx, job, target, pageSize)
|
|
if err != nil {
|
|
return fanoutProgress{}, err
|
|
}
|
|
progress := fanoutProgress{nextCursorUserID: nextCursor, totalDelta: int64(len(userIDs)), done: done}
|
|
for _, userID := range userIDs {
|
|
_, _, err := s.CreateInboxMessage(ctx, messagedomain.CreateInput{
|
|
TargetUserID: userID,
|
|
Producer: fanoutProducer,
|
|
ProducerEventID: fmt.Sprintf("fanout:%s:%d", job.JobID, userID),
|
|
ProducerEventType: firstNonEmpty(snapshot.ProducerEventType, fanoutProducerEventType),
|
|
MessageType: job.MessageType,
|
|
AggregateType: snapshot.AggregateType,
|
|
AggregateID: snapshot.AggregateID,
|
|
TemplateID: snapshot.TemplateID,
|
|
TemplateVersion: snapshot.TemplateVersion,
|
|
Title: snapshot.Title,
|
|
Summary: snapshot.Summary,
|
|
Body: snapshot.Body,
|
|
IconURL: snapshot.IconURL,
|
|
ImageURL: snapshot.ImageURL,
|
|
ActionType: snapshot.ActionType,
|
|
ActionParam: snapshot.ActionParam,
|
|
Priority: snapshot.Priority,
|
|
SentAtMS: snapshot.SentAtMS,
|
|
ExpireAtMS: snapshot.ExpireAtMS,
|
|
MetadataJSON: rawJSONToString(snapshot.MetadataJSON),
|
|
})
|
|
if err != nil {
|
|
return fanoutProgress{}, err
|
|
}
|
|
progress.successDelta++
|
|
}
|
|
return progress, nil
|
|
}
|
|
|
|
func (s *Service) fanoutTargetPage(ctx context.Context, job messagedomain.FanoutJob, target fanoutTargetFilter, pageSize int) ([]int64, int64, bool, error) {
|
|
switch job.TargetScope {
|
|
case messagedomain.TargetScopeSingleUser:
|
|
userID := firstPositive(target.TargetUserID, target.UserID)
|
|
if userID <= 0 {
|
|
return nil, job.CursorUserID, false, xerr.New(xerr.InvalidArgument, "target_user_id is required")
|
|
}
|
|
if job.CursorUserID >= userID {
|
|
return nil, job.CursorUserID, true, nil
|
|
}
|
|
return []int64{userID}, userID, true, nil
|
|
case messagedomain.TargetScopeUserIDs:
|
|
return userIDsTargetPage(target.UserIDs, job.CursorUserID, pageSize)
|
|
case messagedomain.TargetScopeRegion, messagedomain.TargetScopeCountry, messagedomain.TargetScopeAllActiveUsers:
|
|
if s.targetSource == nil {
|
|
return nil, job.CursorUserID, false, xerr.New(xerr.Unavailable, "fanout target source is not configured")
|
|
}
|
|
return s.targetSource.ListFanoutUserIDs(ctx, TargetQuery{
|
|
TargetScope: job.TargetScope,
|
|
CursorUserID: job.CursorUserID,
|
|
PageSize: pageSize,
|
|
RegionID: target.RegionID,
|
|
Country: target.Country,
|
|
})
|
|
default:
|
|
return nil, job.CursorUserID, false, xerr.New(xerr.InvalidArgument, "target_scope is invalid")
|
|
}
|
|
}
|
|
|
|
func userIDsTargetPage(userIDs []int64, cursorUserID int64, pageSize int) ([]int64, int64, bool, error) {
|
|
unique := make([]int64, 0, len(userIDs))
|
|
seen := make(map[int64]bool, len(userIDs))
|
|
for _, userID := range userIDs {
|
|
if userID <= 0 {
|
|
return nil, cursorUserID, false, xerr.New(xerr.InvalidArgument, "user_ids contains invalid value")
|
|
}
|
|
if !seen[userID] {
|
|
seen[userID] = true
|
|
unique = append(unique, userID)
|
|
}
|
|
}
|
|
slices.Sort(unique)
|
|
start := sort.Search(len(unique), func(i int) bool { return unique[i] > cursorUserID })
|
|
if start >= len(unique) {
|
|
return nil, cursorUserID, true, nil
|
|
}
|
|
end := min(start+pageSize, len(unique))
|
|
page := append([]int64(nil), unique[start:end]...)
|
|
return page, page[len(page)-1], end >= len(unique), nil
|
|
}
|
|
|
|
func normalizeFanoutWorkerOptions(options FanoutWorkerOptions, defaultWorkerID string) FanoutWorkerOptions {
|
|
options.WorkerID = strings.TrimSpace(options.WorkerID)
|
|
if options.WorkerID == "" {
|
|
options.WorkerID = firstNonEmpty(defaultWorkerID, "activity-fanout-worker")
|
|
}
|
|
if options.PollInterval <= 0 {
|
|
options.PollInterval = defaultFanoutPollInterval
|
|
}
|
|
if options.LockTTL <= 0 {
|
|
options.LockTTL = defaultFanoutLockTTL
|
|
}
|
|
if options.BatchSize <= 0 {
|
|
options.BatchSize = defaultBatchSize
|
|
}
|
|
if options.MaxRetry <= 0 {
|
|
options.MaxRetry = defaultFanoutMaxRetry
|
|
}
|
|
return options
|
|
}
|
|
|
|
func rawJSONToString(raw json.RawMessage) string {
|
|
raw = json.RawMessage([]byte(strings.TrimSpace(string(raw))))
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return ""
|
|
}
|
|
var text string
|
|
if err := json.Unmarshal(raw, &text); err == nil {
|
|
return strings.TrimSpace(text)
|
|
}
|
|
return string(raw)
|
|
}
|
|
|
|
func firstNonEmpty(values ...string) string {
|
|
for _, value := range values {
|
|
value = strings.TrimSpace(value)
|
|
if value != "" {
|
|
return value
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func firstPositiveInt32(values ...int32) int32 {
|
|
for _, value := range values {
|
|
if value > 0 {
|
|
return value
|
|
}
|
|
}
|
|
return 0
|
|
}
|