554 lines
20 KiB
Go
554 lines
20 KiB
Go
// Package message implements the App system/activity message inbox owner.
|
||
package message
|
||
|
||
import (
|
||
"context"
|
||
"encoding/base64"
|
||
"encoding/json"
|
||
"strings"
|
||
"time"
|
||
"unicode/utf8"
|
||
|
||
"hyapp/pkg/idgen"
|
||
"hyapp/pkg/xerr"
|
||
messagedomain "hyapp/services/activity-service/internal/domain/message"
|
||
)
|
||
|
||
const (
|
||
defaultPageSize = 20
|
||
maxPageSize = 100
|
||
defaultBatchSize = 500
|
||
maxBatchSize = 5000
|
||
)
|
||
|
||
// Repository is the storage boundary for MySQL-backed inbox facts.
|
||
type Repository interface {
|
||
GetActiveMessageTemplate(ctx context.Context, templateID string, templateVersion string, messageType string) (messagedomain.MessageTemplate, bool, error)
|
||
SaveInboxMessage(ctx context.Context, message messagedomain.InboxMessage) (bool, messagedomain.InboxMessage, error)
|
||
CountUnreadInboxMessages(ctx context.Context, userID int64, section string, nowMS int64) (int64, error)
|
||
ListInboxMessages(ctx context.Context, query messagedomain.ListQuery) ([]messagedomain.InboxMessage, error)
|
||
MarkInboxMessageRead(ctx context.Context, userID int64, messageID string, nowMS int64) (messagedomain.InboxMessage, error)
|
||
MarkInboxSectionRead(ctx context.Context, userID int64, section string, nowMS int64) (int64, error)
|
||
DeleteInboxMessage(ctx context.Context, userID int64, messageID string, nowMS int64) error
|
||
SaveFanoutJob(ctx context.Context, job messagedomain.FanoutJob) (bool, messagedomain.FanoutJob, error)
|
||
ClaimFanoutJob(ctx context.Context, workerID string, nowMS int64, lockUntilMS int64, maxRetry int) (messagedomain.FanoutJob, bool, error)
|
||
FinishFanoutJobBatch(ctx context.Context, jobID string, workerID string, cursorUserID int64, totalDelta int64, successDelta int64, failureDelta int64, done bool, nowMS int64, errorMessage string) error
|
||
FailFanoutJob(ctx context.Context, jobID string, workerID string, nowMS int64, nextRunAtMS int64, maxRetry int, errorMessage string) error
|
||
}
|
||
|
||
// Config keeps service-level knobs that do not belong in transport code.
|
||
type Config struct {
|
||
NodeID string
|
||
}
|
||
|
||
// TargetQuery describes the user-service cursor query needed by fanout worker scopes.
|
||
type TargetQuery struct {
|
||
TargetScope string
|
||
CursorUserID int64
|
||
PageSize int
|
||
RegionID int64
|
||
Country string
|
||
}
|
||
|
||
// TargetSource hides user-service gRPC behind a minimal page query.
|
||
type TargetSource interface {
|
||
ListFanoutUserIDs(ctx context.Context, query TargetQuery) ([]int64, int64, bool, error)
|
||
}
|
||
|
||
// Option adjusts message service dependencies.
|
||
type Option func(*Service)
|
||
|
||
// Service owns system/activity inbox business rules.
|
||
type Service struct {
|
||
cfg Config
|
||
repository Repository
|
||
targetSource TargetSource
|
||
now func() time.Time
|
||
}
|
||
|
||
// New creates a message inbox service using MySQL as source of truth.
|
||
func New(cfg Config, repository Repository, options ...Option) *Service {
|
||
service := &Service{cfg: cfg, repository: repository, now: time.Now}
|
||
for _, option := range options {
|
||
option(service)
|
||
}
|
||
return service
|
||
}
|
||
|
||
// WithTargetSource connects fanout worker scopes to user-service without coupling to its storage.
|
||
func WithTargetSource(source TargetSource) Option {
|
||
return func(s *Service) {
|
||
s.targetSource = source
|
||
}
|
||
}
|
||
|
||
// SetClock injects deterministic time for tests.
|
||
func (s *Service) SetClock(now func() time.Time) {
|
||
if now != nil {
|
||
s.now = now
|
||
}
|
||
}
|
||
|
||
// ListMessageTabs returns the fixed App message tab summary.
|
||
func (s *Service) ListMessageTabs(ctx context.Context, userID int64) ([]messagedomain.TabSection, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return nil, err
|
||
}
|
||
if userID <= 0 {
|
||
return nil, xerr.New(xerr.InvalidArgument, "user_id is required")
|
||
}
|
||
nowMS := s.now().UnixMilli()
|
||
systemUnread, err := s.repository.CountUnreadInboxMessages(ctx, userID, messagedomain.SectionSystem, nowMS)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
activityUnread, err := s.repository.CountUnreadInboxMessages(ctx, userID, messagedomain.SectionActivity, nowMS)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return []messagedomain.TabSection{
|
||
{Section: messagedomain.SectionUser, Title: "用户", UnreadCount: 0, Source: messagedomain.SourceTencentIMSDK},
|
||
{Section: messagedomain.SectionSystem, Title: "系统", UnreadCount: systemUnread, Source: messagedomain.SourceBackend},
|
||
{Section: messagedomain.SectionActivity, Title: "活动", UnreadCount: activityUnread, Source: messagedomain.SourceBackend},
|
||
}, nil
|
||
}
|
||
|
||
// ListInboxMessages reads one cursor page for a backend-owned section.
|
||
func (s *Service) ListInboxMessages(ctx context.Context, userID int64, section string, pageSize int32, pageToken string) ([]messagedomain.InboxMessage, string, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return nil, "", err
|
||
}
|
||
if userID <= 0 {
|
||
return nil, "", xerr.New(xerr.InvalidArgument, "user_id is required")
|
||
}
|
||
section = normalizeSection(section)
|
||
if !backendSection(section) {
|
||
return nil, "", xerr.New(xerr.InvalidSection, "section is invalid")
|
||
}
|
||
limit := normalizePageSize(pageSize)
|
||
cursor, err := decodeCursor(section, pageToken)
|
||
if err != nil {
|
||
return nil, "", err
|
||
}
|
||
|
||
messages, err := s.repository.ListInboxMessages(ctx, messagedomain.ListQuery{
|
||
UserID: userID,
|
||
Section: section,
|
||
Limit: limit + 1,
|
||
CursorSentAtMS: cursor.SentAtMS,
|
||
CursorMessageID: cursor.MessageID,
|
||
NowMS: s.now().UnixMilli(),
|
||
})
|
||
if err != nil {
|
||
return nil, "", err
|
||
}
|
||
if len(messages) <= limit {
|
||
return messages, "", nil
|
||
}
|
||
page := messages[:limit]
|
||
next := encodeCursor(section, page[len(page)-1])
|
||
return page, next, nil
|
||
}
|
||
|
||
// MarkInboxMessageRead marks a single visible message read while preserving first read time.
|
||
func (s *Service) MarkInboxMessageRead(ctx context.Context, userID int64, messageID string) (messagedomain.InboxMessage, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return messagedomain.InboxMessage{}, err
|
||
}
|
||
messageID = strings.TrimSpace(messageID)
|
||
if userID <= 0 || messageID == "" {
|
||
return messagedomain.InboxMessage{}, xerr.New(xerr.InvalidArgument, "user_id and message_id are required")
|
||
}
|
||
return s.repository.MarkInboxMessageRead(ctx, userID, messageID, s.now().UnixMilli())
|
||
}
|
||
|
||
// MarkInboxSectionRead marks all visible unread messages in a backend section.
|
||
func (s *Service) MarkInboxSectionRead(ctx context.Context, userID int64, section string) (int64, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return 0, err
|
||
}
|
||
section = normalizeSection(section)
|
||
if userID <= 0 || !backendSection(section) {
|
||
return 0, xerr.New(xerr.InvalidSection, "section is invalid")
|
||
}
|
||
return s.repository.MarkInboxSectionRead(ctx, userID, section, s.now().UnixMilli())
|
||
}
|
||
|
||
// DeleteInboxMessage hides a visible message for the current user.
|
||
func (s *Service) DeleteInboxMessage(ctx context.Context, userID int64, messageID string) error {
|
||
if err := s.requireRepository(); err != nil {
|
||
return err
|
||
}
|
||
messageID = strings.TrimSpace(messageID)
|
||
if userID <= 0 || messageID == "" {
|
||
return xerr.New(xerr.InvalidArgument, "user_id and message_id are required")
|
||
}
|
||
return s.repository.DeleteInboxMessage(ctx, userID, messageID, s.now().UnixMilli())
|
||
}
|
||
|
||
// CreateInboxMessage materializes one producer event into a user inbox row.
|
||
func (s *Service) CreateInboxMessage(ctx context.Context, input messagedomain.CreateInput) (messagedomain.InboxMessage, bool, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return messagedomain.InboxMessage{}, false, err
|
||
}
|
||
input = normalizeCreateInput(input)
|
||
if err := s.fillTemplateSnapshot(ctx, &input); err != nil {
|
||
return messagedomain.InboxMessage{}, false, err
|
||
}
|
||
if err := validateCreateInput(input); err != nil {
|
||
return messagedomain.InboxMessage{}, false, err
|
||
}
|
||
|
||
nowMS := s.now().UnixMilli()
|
||
message := messagedomain.InboxMessage{
|
||
MessageID: idgen.New("msg"),
|
||
UserID: input.TargetUserID,
|
||
MessageType: input.MessageType,
|
||
Producer: input.Producer,
|
||
ProducerEventID: input.ProducerEventID,
|
||
ProducerEventType: input.ProducerEventType,
|
||
AggregateType: input.AggregateType,
|
||
AggregateID: input.AggregateID,
|
||
TemplateID: input.TemplateID,
|
||
TemplateVersion: input.TemplateVersion,
|
||
Title: input.Title,
|
||
Summary: input.Summary,
|
||
Body: input.Body,
|
||
IconURL: input.IconURL,
|
||
ImageURL: input.ImageURL,
|
||
ActionType: input.ActionType,
|
||
ActionParam: input.ActionParam,
|
||
Priority: input.Priority,
|
||
Status: messagedomain.StatusVisible,
|
||
SentAtMS: firstPositive(input.SentAtMS, nowMS),
|
||
ExpireAtMS: input.ExpireAtMS,
|
||
MetadataJSON: input.MetadataJSON,
|
||
CreatedAtMS: nowMS,
|
||
UpdatedAtMS: nowMS,
|
||
}
|
||
created, existing, err := s.repository.SaveInboxMessage(ctx, message)
|
||
if err != nil {
|
||
return messagedomain.InboxMessage{}, false, err
|
||
}
|
||
if !created {
|
||
if !sameInboxPayload(existing, message) {
|
||
return messagedomain.InboxMessage{}, false, xerr.New(xerr.ProducerEventConflict, "producer event payload conflicts with existing inbox message")
|
||
}
|
||
return existing, false, nil
|
||
}
|
||
return existing, true, nil
|
||
}
|
||
|
||
// CreateFanoutJob records an async fanout job and keeps large broadcasts off the request path.
|
||
func (s *Service) CreateFanoutJob(ctx context.Context, input messagedomain.CreateFanoutInput) (messagedomain.FanoutJob, bool, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return messagedomain.FanoutJob{}, false, err
|
||
}
|
||
input = normalizeFanoutInput(input)
|
||
if err := validateFanoutInput(input); err != nil {
|
||
return messagedomain.FanoutJob{}, false, err
|
||
}
|
||
nowMS := s.now().UnixMilli()
|
||
job := messagedomain.FanoutJob{
|
||
JobID: idgen.New("mfan"),
|
||
CommandID: input.CommandID,
|
||
MessageType: input.MessageType,
|
||
TargetScope: input.TargetScope,
|
||
TargetFilterJSON: input.TargetFilterJSON,
|
||
TemplateSnapshotJSON: input.TemplateSnapshotJSON,
|
||
Status: messagedomain.FanoutStatusPending,
|
||
BatchSize: normalizeBatchSize(input.BatchSize),
|
||
CreatedBy: input.CreatedBy,
|
||
CreatedAtMS: nowMS,
|
||
UpdatedAtMS: nowMS,
|
||
}
|
||
created, existing, err := s.repository.SaveFanoutJob(ctx, job)
|
||
if err != nil {
|
||
return messagedomain.FanoutJob{}, false, err
|
||
}
|
||
if !created {
|
||
if !sameFanoutPayload(existing, job) {
|
||
return messagedomain.FanoutJob{}, false, xerr.New(xerr.RequestConflict, "fanout command payload conflicts with existing job")
|
||
}
|
||
return existing, false, nil
|
||
}
|
||
return existing, true, nil
|
||
}
|
||
|
||
func (s *Service) requireRepository() error {
|
||
if s == nil || s.repository == nil {
|
||
return xerr.New(xerr.Unavailable, "message repository is not configured")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *Service) fillTemplateSnapshot(ctx context.Context, input *messagedomain.CreateInput) error {
|
||
if input.Title != "" || input.Summary != "" {
|
||
return nil
|
||
}
|
||
if input.TemplateID == "" || input.TemplateVersion == "" {
|
||
return nil
|
||
}
|
||
template, exists, err := s.repository.GetActiveMessageTemplate(ctx, input.TemplateID, input.TemplateVersion, input.MessageType)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !exists {
|
||
return xerr.New(xerr.NotFound, "message template not found")
|
||
}
|
||
input.Title = template.Title
|
||
input.Summary = template.Summary
|
||
input.Body = template.Body
|
||
input.IconURL = template.IconURL
|
||
input.ImageURL = template.ImageURL
|
||
input.ActionType = template.ActionType
|
||
input.ActionParam = template.ActionParam
|
||
return nil
|
||
}
|
||
|
||
type pageCursor struct {
|
||
Section string `json:"section"`
|
||
SentAtMS int64 `json:"sent_at_ms"`
|
||
MessageID string `json:"message_id"`
|
||
}
|
||
|
||
func encodeCursor(section string, message messagedomain.InboxMessage) string {
|
||
payload, err := json.Marshal(pageCursor{Section: section, SentAtMS: message.SentAtMS, MessageID: message.MessageID})
|
||
if err != nil {
|
||
return ""
|
||
}
|
||
return base64.RawURLEncoding.EncodeToString(payload)
|
||
}
|
||
|
||
func decodeCursor(section string, encoded string) (pageCursor, error) {
|
||
encoded = strings.TrimSpace(encoded)
|
||
if encoded == "" {
|
||
return pageCursor{Section: section}, nil
|
||
}
|
||
raw, err := base64.RawURLEncoding.DecodeString(encoded)
|
||
if err != nil {
|
||
return pageCursor{}, xerr.New(xerr.PageTokenInvalid, "page token is invalid")
|
||
}
|
||
var cursor pageCursor
|
||
if err := json.Unmarshal(raw, &cursor); err != nil {
|
||
return pageCursor{}, xerr.New(xerr.PageTokenInvalid, "page token is invalid")
|
||
}
|
||
if cursor.Section != section || cursor.SentAtMS <= 0 || strings.TrimSpace(cursor.MessageID) == "" {
|
||
return pageCursor{}, xerr.New(xerr.PageTokenInvalid, "page token is invalid")
|
||
}
|
||
return cursor, nil
|
||
}
|
||
|
||
func normalizeCreateInput(input messagedomain.CreateInput) messagedomain.CreateInput {
|
||
input.Producer = strings.TrimSpace(input.Producer)
|
||
input.ProducerEventID = strings.TrimSpace(input.ProducerEventID)
|
||
input.ProducerEventType = strings.TrimSpace(input.ProducerEventType)
|
||
input.MessageType = normalizeSection(input.MessageType)
|
||
input.AggregateType = strings.TrimSpace(input.AggregateType)
|
||
input.AggregateID = strings.TrimSpace(input.AggregateID)
|
||
input.TemplateID = strings.TrimSpace(input.TemplateID)
|
||
input.TemplateVersion = strings.TrimSpace(input.TemplateVersion)
|
||
input.Title = strings.TrimSpace(input.Title)
|
||
input.Summary = strings.TrimSpace(input.Summary)
|
||
input.Body = strings.TrimSpace(input.Body)
|
||
input.IconURL = strings.TrimSpace(input.IconURL)
|
||
input.ImageURL = strings.TrimSpace(input.ImageURL)
|
||
input.ActionType = strings.TrimSpace(input.ActionType)
|
||
input.ActionParam = strings.TrimSpace(input.ActionParam)
|
||
input.MetadataJSON = strings.TrimSpace(input.MetadataJSON)
|
||
return input
|
||
}
|
||
|
||
func validateCreateInput(input messagedomain.CreateInput) error {
|
||
if input.TargetUserID <= 0 || input.Producer == "" || input.ProducerEventID == "" {
|
||
return xerr.New(xerr.InvalidArgument, "target_user_id, producer and producer_event_id are required")
|
||
}
|
||
if !backendSection(input.MessageType) {
|
||
return xerr.New(xerr.InvalidSection, "message_type is invalid")
|
||
}
|
||
for _, field := range []struct {
|
||
name string
|
||
value string
|
||
max int
|
||
}{
|
||
{"producer", input.Producer, 64},
|
||
{"producer_event_id", input.ProducerEventID, 128},
|
||
{"producer_event_type", input.ProducerEventType, 96},
|
||
{"aggregate_type", input.AggregateType, 64},
|
||
{"aggregate_id", input.AggregateID, 128},
|
||
{"template_id", input.TemplateID, 96},
|
||
{"template_version", input.TemplateVersion, 64},
|
||
{"title", input.Title, 160},
|
||
{"summary", input.Summary, 512},
|
||
{"body", input.Body, 2048},
|
||
{"icon_url", input.IconURL, 512},
|
||
{"image_url", input.ImageURL, 512},
|
||
{"action_type", input.ActionType, 64},
|
||
{"action_param", input.ActionParam, 1024},
|
||
} {
|
||
if utf8.RuneCountInString(field.value) > field.max {
|
||
return xerr.New(xerr.InvalidArgument, field.name+" is too long")
|
||
}
|
||
}
|
||
if input.Title == "" || input.Summary == "" {
|
||
return xerr.New(xerr.InvalidArgument, "title and summary are required")
|
||
}
|
||
if !allowedActionType(input.ActionType) {
|
||
return xerr.New(xerr.InvalidArgument, "action_type is invalid")
|
||
}
|
||
if input.ExpireAtMS > 0 && input.SentAtMS > 0 && input.ExpireAtMS <= input.SentAtMS {
|
||
return xerr.New(xerr.InvalidArgument, "expire_at_ms must be after sent_at_ms")
|
||
}
|
||
if input.MetadataJSON != "" && !json.Valid([]byte(input.MetadataJSON)) {
|
||
return xerr.New(xerr.InvalidArgument, "metadata_json is invalid")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func normalizeFanoutInput(input messagedomain.CreateFanoutInput) messagedomain.CreateFanoutInput {
|
||
input.CommandID = strings.TrimSpace(input.CommandID)
|
||
input.MessageType = normalizeSection(input.MessageType)
|
||
input.TargetScope = strings.TrimSpace(input.TargetScope)
|
||
input.TargetFilterJSON = strings.TrimSpace(input.TargetFilterJSON)
|
||
input.TemplateSnapshotJSON = strings.TrimSpace(input.TemplateSnapshotJSON)
|
||
input.CreatedBy = strings.TrimSpace(input.CreatedBy)
|
||
return input
|
||
}
|
||
|
||
func validateFanoutInput(input messagedomain.CreateFanoutInput) error {
|
||
if input.CommandID == "" || input.TargetScope == "" || input.CreatedBy == "" {
|
||
return xerr.New(xerr.InvalidArgument, "command_id, target_scope and created_by are required")
|
||
}
|
||
if !backendSection(input.MessageType) {
|
||
return xerr.New(xerr.InvalidSection, "message_type is invalid")
|
||
}
|
||
if !allowedFanoutScope(input.TargetScope) {
|
||
return xerr.New(xerr.InvalidArgument, "target_scope is invalid")
|
||
}
|
||
if input.TargetFilterJSON == "" || !json.Valid([]byte(input.TargetFilterJSON)) {
|
||
return xerr.New(xerr.InvalidArgument, "target_filter_json is invalid")
|
||
}
|
||
if input.TemplateSnapshotJSON == "" || !json.Valid([]byte(input.TemplateSnapshotJSON)) {
|
||
return xerr.New(xerr.InvalidArgument, "template_snapshot_json is invalid")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func normalizeSection(section string) string {
|
||
return strings.ToLower(strings.TrimSpace(section))
|
||
}
|
||
|
||
func backendSection(section string) bool {
|
||
return section == messagedomain.SectionSystem || section == messagedomain.SectionActivity
|
||
}
|
||
|
||
func normalizePageSize(value int32) int {
|
||
if value <= 0 {
|
||
return defaultPageSize
|
||
}
|
||
if value > maxPageSize {
|
||
return maxPageSize
|
||
}
|
||
return int(value)
|
||
}
|
||
|
||
func normalizeBatchSize(value int32) int32 {
|
||
if value <= 0 {
|
||
return defaultBatchSize
|
||
}
|
||
if value > maxBatchSize {
|
||
return maxBatchSize
|
||
}
|
||
return value
|
||
}
|
||
|
||
func allowedActionType(actionType string) bool {
|
||
switch actionType {
|
||
case "", "wallet_withdraw_detail", "host_application_detail", "activity_detail", "room_detail", "user_profile", "app_h5", "coin_seller_transfer_detail":
|
||
return true
|
||
case "vip_coin_rebate_claim":
|
||
// VIP 返现通知只携带 wallet 生成的 rebate_id JSON，客户端仍需调用受鉴权的
|
||
// claim 接口才能入账；它不是任意外链或脚本动作，可以进入 system inbox。
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func allowedFanoutScope(scope string) bool {
|
||
switch scope {
|
||
case messagedomain.TargetScopeSingleUser, messagedomain.TargetScopeUserIDs, messagedomain.TargetScopeRegion, messagedomain.TargetScopeCountry, messagedomain.TargetScopeAllActiveUsers, messagedomain.TargetScopeAllRegistered:
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func firstPositive(values ...int64) int64 {
|
||
for _, value := range values {
|
||
if value > 0 {
|
||
return value
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func sameInboxPayload(existing messagedomain.InboxMessage, next messagedomain.InboxMessage) bool {
|
||
return existing.UserID == next.UserID &&
|
||
existing.MessageType == next.MessageType &&
|
||
existing.Producer == next.Producer &&
|
||
existing.ProducerEventID == next.ProducerEventID &&
|
||
existing.ProducerEventType == next.ProducerEventType &&
|
||
existing.AggregateType == next.AggregateType &&
|
||
existing.AggregateID == next.AggregateID &&
|
||
existing.TemplateID == next.TemplateID &&
|
||
existing.TemplateVersion == next.TemplateVersion &&
|
||
existing.Title == next.Title &&
|
||
existing.Summary == next.Summary &&
|
||
existing.Body == next.Body &&
|
||
existing.IconURL == next.IconURL &&
|
||
existing.ImageURL == next.ImageURL &&
|
||
existing.ActionType == next.ActionType &&
|
||
existing.ActionParam == next.ActionParam &&
|
||
existing.Priority == next.Priority &&
|
||
existing.ExpireAtMS == next.ExpireAtMS &&
|
||
sameJSONText(existing.MetadataJSON, next.MetadataJSON)
|
||
}
|
||
|
||
func sameFanoutPayload(existing messagedomain.FanoutJob, next messagedomain.FanoutJob) bool {
|
||
return existing.CommandID == next.CommandID &&
|
||
existing.MessageType == next.MessageType &&
|
||
existing.TargetScope == next.TargetScope &&
|
||
sameJSONText(existing.TargetFilterJSON, next.TargetFilterJSON) &&
|
||
sameJSONText(existing.TemplateSnapshotJSON, next.TemplateSnapshotJSON) &&
|
||
existing.BatchSize == next.BatchSize &&
|
||
existing.CreatedBy == next.CreatedBy
|
||
}
|
||
|
||
func sameJSONText(left string, right string) bool {
|
||
left = strings.TrimSpace(left)
|
||
right = strings.TrimSpace(right)
|
||
if left == "" || left == "null" {
|
||
return right == "" || right == "null"
|
||
}
|
||
leftCanonical, leftOK := canonicalJSON(left)
|
||
rightCanonical, rightOK := canonicalJSON(right)
|
||
if leftOK && rightOK {
|
||
return leftCanonical == rightCanonical
|
||
}
|
||
return left == right
|
||
}
|
||
|
||
func canonicalJSON(value string) (string, bool) {
|
||
var parsed any
|
||
if err := json.Unmarshal([]byte(value), &parsed); err != nil {
|
||
return "", false
|
||
}
|
||
encoded, err := json.Marshal(parsed)
|
||
if err != nil {
|
||
return "", false
|
||
}
|
||
return string(encoded), true
|
||
}
|