299 lines
13 KiB
Go
299 lines
13 KiB
Go
package cpnotice
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"testing"
|
||
"time"
|
||
|
||
"hyapp/pkg/tencentim"
|
||
"hyapp/pkg/usermq"
|
||
)
|
||
|
||
func TestProcessApplicationCreatedPublishesC2CAndBothRoomIMs(t *testing.T) {
|
||
repo := &fakeRepository{}
|
||
publisher := &fakePublisher{}
|
||
service := New(Config{NodeID: "notice-test"}, repo, publisher)
|
||
|
||
handled, err := service.ProcessUserOutboxMessage(context.Background(), usermq.UserOutboxMessage{
|
||
AppCode: "lalu",
|
||
EventID: "cp_application:created:1002:1",
|
||
EventType: eventApplicationCreated,
|
||
AggregateType: "cp_application",
|
||
AggregateID: 1002,
|
||
PayloadJSON: applicationPayloadJSON(),
|
||
OccurredAtMS: 1700000000000,
|
||
}, CPNoticeWorkerOptions{PublishTimeout: time.Second})
|
||
if err != nil || !handled {
|
||
t.Fatalf("ProcessUserOutboxMessage failed handled=%v err=%v", handled, err)
|
||
}
|
||
if len(publisher.userMessages) != 1 || publisher.userMessages[0].FromAccount != "1001" || publisher.userMessages[0].ToAccount != "1002" {
|
||
t.Fatalf("expected target c2c message, got %+v", publisher.userMessages)
|
||
}
|
||
if len(publisher.groupMessages) != 2 || publisher.groupMessages[0].GroupID != "room-100" || publisher.groupMessages[1].GroupID != "room-100" {
|
||
t.Fatalf("expected target and requester room group messages, got %+v", publisher.groupMessages)
|
||
}
|
||
targetRoomPayload := mustObject(string(publisher.groupMessages[0].PayloadJSON))
|
||
if targetRoomPayload["recipient_user_id"] != "1002" {
|
||
t.Fatalf("target room payload mismatch: %+v", targetRoomPayload)
|
||
}
|
||
requesterRoomPayload := mustObject(string(publisher.groupMessages[1].PayloadJSON))
|
||
if requesterRoomPayload["recipient_user_id"] != "1001" {
|
||
t.Fatalf("requester room payload mismatch: %+v", requesterRoomPayload)
|
||
}
|
||
if len(repo.delivered) != 3 || repo.delivered[0].Channel != channelTencentIMC2C || repo.delivered[1].Channel != channelTencentIMRoom || repo.delivered[2].Channel != channelTencentIMRoomSender {
|
||
t.Fatalf("delivery channels mismatch: %+v", repo.delivered)
|
||
}
|
||
}
|
||
|
||
func TestProcessApplicationCreatedContinuesAfterAlreadyFinalChannel(t *testing.T) {
|
||
// 模拟同一事实的 C2C 已在上一次消费中成功，而两个房间 channel 尚未处理；重放必须
|
||
// 依赖 channel 级幂等跳过 C2C，并继续补齐后续副作用，不能把首个终态误判为整条完成。
|
||
repo := &fakeRepository{notClaimedChannels: map[string]bool{channelTencentIMC2C: true}}
|
||
publisher := &fakePublisher{}
|
||
service := New(Config{NodeID: "notice-test"}, repo, publisher)
|
||
|
||
handled, err := service.ProcessUserOutboxMessage(context.Background(), usermq.UserOutboxMessage{
|
||
AppCode: "lalu",
|
||
EventID: "cp_application:created:1002:partial-replay",
|
||
EventType: eventApplicationCreated,
|
||
AggregateType: "cp_application",
|
||
AggregateID: 1002,
|
||
PayloadJSON: applicationPayloadJSON(),
|
||
OccurredAtMS: 1700000000000,
|
||
}, CPNoticeWorkerOptions{PublishTimeout: time.Second})
|
||
if err != nil || !handled {
|
||
t.Fatalf("ProcessUserOutboxMessage failed handled=%v err=%v", handled, err)
|
||
}
|
||
if len(repo.claimedChannels) != 3 {
|
||
t.Fatalf("expected every channel to be inspected, got %+v", repo.claimedChannels)
|
||
}
|
||
if len(publisher.userMessages) != 0 {
|
||
t.Fatalf("already delivered c2c must not be published again: %+v", publisher.userMessages)
|
||
}
|
||
if len(publisher.groupMessages) != 2 || publisher.groupMessages[0].GroupID != "room-100" || publisher.groupMessages[1].GroupID != "room-100" {
|
||
t.Fatalf("expected both remaining room deliveries, got %+v", publisher.groupMessages)
|
||
}
|
||
if len(repo.delivered) != 2 || repo.delivered[0].Channel != channelTencentIMRoom || repo.delivered[1].Channel != channelTencentIMRoomSender {
|
||
t.Fatalf("expected only remaining channels to become delivered, got %+v", repo.delivered)
|
||
}
|
||
}
|
||
|
||
func TestProcessApplicationAcceptedPublishesMutualC2CAndRoomIM(t *testing.T) {
|
||
repo := &fakeRepository{}
|
||
publisher := &fakePublisher{}
|
||
service := New(Config{NodeID: "notice-test"}, repo, publisher)
|
||
|
||
handled, err := service.ProcessUserOutboxMessage(context.Background(), usermq.UserOutboxMessage{
|
||
AppCode: "lalu",
|
||
EventID: "cp_application:accepted:1002:1",
|
||
EventType: eventApplicationAccepted,
|
||
AggregateType: "cp_application",
|
||
AggregateID: 1002,
|
||
PayloadJSON: applicationPayloadJSON(),
|
||
OccurredAtMS: 1700000000001,
|
||
}, CPNoticeWorkerOptions{PublishTimeout: time.Second})
|
||
if err != nil || !handled {
|
||
t.Fatalf("ProcessUserOutboxMessage failed handled=%v err=%v", handled, err)
|
||
}
|
||
if len(publisher.userMessages) != 2 {
|
||
t.Fatalf("expected mutual c2c messages, got %+v", publisher.userMessages)
|
||
}
|
||
if publisher.userMessages[0].FromAccount != "1002" || publisher.userMessages[0].ToAccount != "1001" {
|
||
t.Fatalf("expected acceptor -> requester c2c message, got %+v", publisher.userMessages[0])
|
||
}
|
||
requesterPayload := mustObject(string(publisher.userMessages[0].PayloadJSON))
|
||
if requesterPayload["c2c_notice_role"] != "accepted_to_requester" || requesterPayload["message_text"] != "我同意了你" {
|
||
t.Fatalf("requester c2c payload mismatch: %+v", requesterPayload)
|
||
}
|
||
if publisher.userMessages[1].FromAccount != "1001" || publisher.userMessages[1].ToAccount != "1002" {
|
||
t.Fatalf("expected requester -> acceptor c2c message, got %+v", publisher.userMessages[1])
|
||
}
|
||
acceptorPayload := mustObject(string(publisher.userMessages[1].PayloadJSON))
|
||
if acceptorPayload["c2c_notice_role"] != "accepted_to_acceptor" || acceptorPayload["message_text"] != "对方已同意你的申请" {
|
||
t.Fatalf("acceptor c2c payload mismatch: %+v", acceptorPayload)
|
||
}
|
||
if len(publisher.groupMessages) != 1 || publisher.groupMessages[0].GroupID != "room-100" {
|
||
t.Fatalf("expected requester room group message, got %+v", publisher.groupMessages)
|
||
}
|
||
if len(repo.delivered) != 3 || repo.delivered[0].Channel != channelTencentIMC2C || repo.delivered[1].Channel != channelTencentIMC2CAcceptor || repo.delivered[2].Channel != channelTencentIMRoom {
|
||
t.Fatalf("delivery channels mismatch: %+v", repo.delivered)
|
||
}
|
||
}
|
||
|
||
func TestProcessApplicationRejectedPublishesRequesterC2CAndRoomIM(t *testing.T) {
|
||
repo := &fakeRepository{}
|
||
publisher := &fakePublisher{}
|
||
service := New(Config{NodeID: "notice-test"}, repo, publisher)
|
||
|
||
body, _ := json.Marshal(map[string]any{
|
||
"application": mustObject(applicationPayloadJSON()),
|
||
"reason": "not_now",
|
||
})
|
||
handled, err := service.ProcessUserOutboxMessage(context.Background(), usermq.UserOutboxMessage{
|
||
AppCode: "lalu",
|
||
EventID: "cp_application:rejected:1002:1",
|
||
EventType: eventApplicationRejected,
|
||
AggregateType: "cp_application",
|
||
AggregateID: 1002,
|
||
PayloadJSON: string(body),
|
||
OccurredAtMS: 1700000000001,
|
||
}, CPNoticeWorkerOptions{PublishTimeout: time.Second})
|
||
if err != nil || !handled {
|
||
t.Fatalf("ProcessUserOutboxMessage failed handled=%v err=%v", handled, err)
|
||
}
|
||
if len(publisher.userMessages) != 1 || publisher.userMessages[0].FromAccount != "1002" || publisher.userMessages[0].ToAccount != "1001" {
|
||
t.Fatalf("expected requester c2c message, got %+v", publisher.userMessages)
|
||
}
|
||
if len(publisher.groupMessages) != 1 || publisher.groupMessages[0].GroupID != "room-100" {
|
||
t.Fatalf("expected requester room group message, got %+v", publisher.groupMessages)
|
||
}
|
||
if len(repo.delivered) != 2 || repo.delivered[0].NoticeType != noticeTypeApplicationRejected || repo.delivered[1].Channel != channelTencentIMRoom {
|
||
t.Fatalf("delivery channels mismatch: %+v", repo.delivered)
|
||
}
|
||
}
|
||
|
||
func TestProcessRelationshipCreatedPublishesRegionIM(t *testing.T) {
|
||
repo := &fakeRepository{}
|
||
publisher := &fakePublisher{}
|
||
service := New(Config{NodeID: "notice-test", GroupIDPrefix: "test_"}, repo, publisher)
|
||
|
||
payload := map[string]any{
|
||
"application": mustObject(applicationPayloadJSON()),
|
||
"relationship": map[string]any{
|
||
"RelationshipID": "cp_rel:1001:1002",
|
||
},
|
||
}
|
||
body, _ := json.Marshal(payload)
|
||
handled, err := service.ProcessUserOutboxMessage(context.Background(), usermq.UserOutboxMessage{
|
||
AppCode: "lalu",
|
||
EventID: "cp_relationship:created:1002:1",
|
||
EventType: eventRelationshipCreated,
|
||
AggregateType: "cp_relationship",
|
||
AggregateID: 1002,
|
||
PayloadJSON: string(body),
|
||
OccurredAtMS: 1700000000002,
|
||
}, CPNoticeWorkerOptions{PublishTimeout: time.Second})
|
||
if err != nil || !handled {
|
||
t.Fatalf("ProcessUserOutboxMessage failed handled=%v err=%v", handled, err)
|
||
}
|
||
if len(publisher.ensuredGroups) != 1 || publisher.ensuredGroups[0].GroupID != "test_hy_lalu_bc_r_v2_86" {
|
||
t.Fatalf("expected region group ensure, got %+v", publisher.ensuredGroups)
|
||
}
|
||
// v2 区域群是 AVChatRoom(不带 ApplyJoinOption),不再受 ChatRoom 套餐成员上限约束。
|
||
if publisher.ensuredGroups[0].Type != tencentim.BroadcastGroupType || publisher.ensuredGroups[0].ApplyJoinOption != "" {
|
||
t.Fatalf("v2 region group must be AVChatRoom: %+v", publisher.ensuredGroups[0])
|
||
}
|
||
// v2 群投递成功后过渡期向 v1 旧群补发一条 _legacy 副本。
|
||
if len(publisher.groupMessages) != 2 || publisher.groupMessages[0].GroupID != "test_hy_lalu_bc_r_v2_86" {
|
||
t.Fatalf("expected region group message, got %+v", publisher.groupMessages)
|
||
}
|
||
if publisher.groupMessages[1].GroupID != "test_hy_lalu_bc_r_86" || publisher.groupMessages[1].EventID != "cp_relationship:created:1002:1_legacy" {
|
||
t.Fatalf("expected legacy region copy, got %+v", publisher.groupMessages[1])
|
||
}
|
||
if len(publisher.userMessages) != 0 {
|
||
t.Fatalf("relationship created should not publish c2c: %+v", publisher.userMessages)
|
||
}
|
||
}
|
||
|
||
func TestPublishFailureMarksRetryable(t *testing.T) {
|
||
repo := &fakeRepository{}
|
||
publisher := &fakePublisher{err: errors.New("im timeout")}
|
||
service := New(Config{NodeID: "notice-test"}, repo, publisher)
|
||
|
||
_, err := service.ProcessUserOutboxMessage(context.Background(), usermq.UserOutboxMessage{
|
||
AppCode: "lalu",
|
||
EventID: "cp_application:created:1002:retry",
|
||
EventType: eventApplicationCreated,
|
||
AggregateType: "cp_application",
|
||
AggregateID: 1002,
|
||
PayloadJSON: applicationPayloadJSON(),
|
||
OccurredAtMS: 1700000000003,
|
||
}, CPNoticeWorkerOptions{PublishTimeout: time.Second, MaxRetryCount: 3})
|
||
if err != nil {
|
||
t.Fatalf("retryable publish should be recorded without returning poison error: %v", err)
|
||
}
|
||
if len(repo.retryable) == 0 || repo.retryable[0].RetryCount != 1 {
|
||
t.Fatalf("expected retryable delivery, got %+v", repo.retryable)
|
||
}
|
||
}
|
||
|
||
type fakeRepository struct {
|
||
notClaimedChannels map[string]bool
|
||
claimedChannels []string
|
||
delivered []CPDelivery
|
||
retryable []CPDelivery
|
||
failed []CPDelivery
|
||
}
|
||
|
||
func (r *fakeRepository) ClaimCPDelivery(_ context.Context, _ string, delivery CPDelivery, _ time.Duration) (CPDelivery, bool, error) {
|
||
r.claimedChannels = append(r.claimedChannels, delivery.Channel)
|
||
if r.notClaimedChannels[delivery.Channel] {
|
||
return CPDelivery{}, false, nil
|
||
}
|
||
return delivery, true, nil
|
||
}
|
||
|
||
func (r *fakeRepository) MarkCPDelivered(_ context.Context, delivery CPDelivery, _ []byte, _ int64) error {
|
||
r.delivered = append(r.delivered, delivery)
|
||
return nil
|
||
}
|
||
|
||
func (r *fakeRepository) MarkCPRetryable(_ context.Context, delivery CPDelivery, retryCount int, _ int64, _ string, _ int64) error {
|
||
delivery.RetryCount = retryCount
|
||
r.retryable = append(r.retryable, delivery)
|
||
return nil
|
||
}
|
||
|
||
func (r *fakeRepository) MarkCPFailed(_ context.Context, delivery CPDelivery, retryCount int, _ string, _ int64) error {
|
||
delivery.RetryCount = retryCount
|
||
r.failed = append(r.failed, delivery)
|
||
return nil
|
||
}
|
||
|
||
type fakePublisher struct {
|
||
userMessages []tencentim.CustomUserMessage
|
||
groupMessages []tencentim.CustomGroupMessage
|
||
ensuredGroups []tencentim.GroupSpec
|
||
err error
|
||
}
|
||
|
||
func (p *fakePublisher) PublishUserCustomMessage(_ context.Context, message tencentim.CustomUserMessage) error {
|
||
p.userMessages = append(p.userMessages, message)
|
||
return p.err
|
||
}
|
||
|
||
func (p *fakePublisher) PublishGroupCustomMessage(_ context.Context, message tencentim.CustomGroupMessage) error {
|
||
p.groupMessages = append(p.groupMessages, message)
|
||
return p.err
|
||
}
|
||
|
||
func (p *fakePublisher) EnsureGroup(_ context.Context, spec tencentim.GroupSpec) error {
|
||
p.ensuredGroups = append(p.ensuredGroups, spec)
|
||
return p.err
|
||
}
|
||
|
||
func applicationPayloadJSON() string {
|
||
return `{
|
||
"ApplicationID":"cp_app:event-1",
|
||
"RelationType":"cp",
|
||
"Status":"pending",
|
||
"Requester":{"UserID":1001,"DisplayUserID":"163001","Username":"Alice","Avatar":"https://cdn.example/a.png"},
|
||
"Target":{"UserID":1002,"DisplayUserID":"163002","Username":"Bob","Avatar":"https://cdn.example/b.png"},
|
||
"RoomID":"room-100",
|
||
"RoomRegionID":86,
|
||
"Gift":{"GiftID":"cp_ring","GiftName":"CP Ring","GiftIconURL":"https://cdn.example/gift.png","GiftAnimationURL":"","GiftCount":1,"GiftValue":100,"BillingReceiptID":"tx-1"},
|
||
"ExpiresAtMS":1700003600000,
|
||
"DecidedAtMS":0
|
||
}`
|
||
}
|
||
|
||
func mustObject(value string) map[string]any {
|
||
var out map[string]any
|
||
if err := json.Unmarshal([]byte(value), &out); err != nil {
|
||
panic(err)
|
||
}
|
||
return out
|
||
}
|