package confirmnotice import ( "context" "errors" "testing" "time" "hyapp/pkg/messagemq" "hyapp/pkg/tencentim" ) func TestProcessActionOutboxMessagePublishesConfirmC2C(t *testing.T) { payloadJSON := `{"type":"im_confirm","confirm_id":"cfm_1","sender_user_id":"1001","receiver_user_id":"2002"}` repository := &confirmRepositoryFake{claimOK: true} publisher := &confirmPublisherFake{} service := New(Config{NodeID: "notice-node-1"}, repository, publisher) handled, err := service.ProcessActionOutboxMessage(context.Background(), messagemq.ActionOutboxMessage{ AppCode: " lalu ", EventID: "evt-confirm-1", EventType: "ConfirmMessageCreated", ConfirmID: "cfm_1", PayloadJSON: payloadJSON, OccurredAtMS: 1710000000123, }, WorkerOptions{PublishTimeout: time.Second}) if err != nil { t.Fatalf("ProcessActionOutboxMessage failed: %v", err) } if !handled { t.Fatal("ConfirmMessageCreated must be handled") } if repository.claimed.EventID != "evt-confirm-1" || repository.claimed.ConfirmID != "cfm_1" { t.Fatalf("claimed delivery lost outbox identity: %+v", repository.claimed) } if repository.claimed.AppCode != "lalu" || repository.claimed.SenderUserID != 1001 || repository.claimed.TargetUserID != 2002 { t.Fatalf("claimed delivery has incorrect routing fields: %+v", repository.claimed) } if len(publisher.messages) != 1 { t.Fatalf("published messages=%d, want 1", len(publisher.messages)) } message := publisher.messages[0] // 邀请者必须作为发送方、被邀请者必须作为接收方;交换两者会把确认按钮投递给无权处理邀请的人。 if message.FromAccount != "1001" || message.ToAccount != "2002" { t.Fatalf("unexpected C2C direction: from=%q to=%q", message.FromAccount, message.ToAccount) } if message.EventID != "evt-confirm-1" || message.Desc != "im_confirm" || message.Ext != "im_confirm" { t.Fatalf("unexpected C2C envelope: %+v", message) } if string(message.PayloadJSON) != payloadJSON { t.Fatalf("payload changed during delivery: got=%s want=%s", message.PayloadJSON, payloadJSON) } if repository.delivered.EventID != "evt-confirm-1" || string(repository.deliveredPayload) != payloadJSON { t.Fatalf("delivery was not durably marked: delivery=%+v payload=%s", repository.delivered, repository.deliveredPayload) } } func TestProcessActionOutboxMessageIgnoresUnrelatedEvents(t *testing.T) { repository := &confirmRepositoryFake{claimOK: true} publisher := &confirmPublisherFake{} service := New(Config{NodeID: "notice-node-1"}, repository, publisher) handled, err := service.ProcessActionOutboxMessage(context.Background(), messagemq.ActionOutboxMessage{ EventType: "ActivityNoticeCreated", }, WorkerOptions{}) if err != nil { t.Fatalf("unrelated event returned error: %v", err) } if handled { t.Fatal("unrelated event must not be claimed by confirm notice") } if repository.claimCalls != 0 || len(publisher.messages) != 0 { t.Fatalf("unrelated event caused side effects: claims=%d publishes=%d", repository.claimCalls, len(publisher.messages)) } } func TestProcessActionOutboxMessageRecordsPublishFailureState(t *testing.T) { tests := []struct { name string claimedRetry int maxRetry int wantErr bool wantRetry int wantFailed int }{ {name: "retryable", claimedRetry: 1, maxRetry: 3, wantErr: true, wantRetry: 2}, {name: "terminal", claimedRetry: 2, maxRetry: 3, wantFailed: 3}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { repository := &confirmRepositoryFake{claimOK: true, claimedRetryCount: tt.claimedRetry} publisher := &confirmPublisherFake{err: errors.New("tencent im timeout")} service := New(Config{NodeID: "notice-node-1"}, repository, publisher) handled, err := service.ProcessActionOutboxMessage(context.Background(), confirmActionMessage(), WorkerOptions{ PublishTimeout: time.Second, MaxRetryCount: tt.maxRetry, InitialBackoff: time.Second, MaxBackoff: time.Second, }) if !handled { t.Fatal("confirm event must remain owned by confirm notice on publish failure") } if (err != nil) != tt.wantErr { t.Fatalf("error=%v, wantErr=%v", err, tt.wantErr) } // 可重试错误必须返回给 MQ 触发重投;达到上限后落 failed 并 ACK,避免同一毒消息无限阻塞消费。 if repository.retryCount != tt.wantRetry || repository.failedRetryCount != tt.wantFailed { t.Fatalf("failure state mismatch: retry=%d failed=%d", repository.retryCount, repository.failedRetryCount) } }) } } func confirmActionMessage() messagemq.ActionOutboxMessage { return messagemq.ActionOutboxMessage{ AppCode: "lalu", EventID: "evt-confirm-failure", EventType: "ConfirmMessageCreated", ConfirmID: "cfm_failure", PayloadJSON: `{"sender_user_id":"1001","receiver_user_id":"2002"}`, OccurredAtMS: 1710000000123, } } type confirmRepositoryFake struct { claimOK bool claimedRetryCount int claimCalls int claimed Delivery delivered Delivery deliveredPayload []byte retryCount int failedRetryCount int } func (r *confirmRepositoryFake) ClaimConfirmDelivery(_ context.Context, _ string, delivery Delivery, _ time.Duration) (Delivery, bool, error) { r.claimCalls++ r.claimed = delivery delivery.RetryCount = r.claimedRetryCount return delivery, r.claimOK, nil } func (r *confirmRepositoryFake) MarkConfirmDelivered(_ context.Context, delivery Delivery, payload []byte, _ int64) error { r.delivered = delivery r.deliveredPayload = append([]byte(nil), payload...) return nil } func (r *confirmRepositoryFake) MarkConfirmRetryable(_ context.Context, _ Delivery, retryCount int, _ int64, _ string, _ int64) error { r.retryCount = retryCount return nil } func (r *confirmRepositoryFake) MarkConfirmFailed(_ context.Context, _ Delivery, retryCount int, _ string, _ int64) error { r.failedRetryCount = retryCount return nil } type confirmPublisherFake struct { messages []tencentim.CustomUserMessage err error } func (p *confirmPublisherFake) PublishUserCustomMessage(_ context.Context, message tencentim.CustomUserMessage) error { p.messages = append(p.messages, message) return p.err }