修复CP通知多通道重放

services/notice-service/internal/modules/cpnotice/service.go

@@ -86,9 +86,15 @@ func (s *Service) ProcessUserOutboxMessage(ctx context.Context, message usermq.U
 	}
 	for _, delivery := range deliveries {
 		claimedDelivery, claimed, err := s.repository.ClaimCPDelivery(ctx, options.WorkerID, delivery, options.LockTTL)
-		if err != nil || !claimed {
+		if err != nil {
 			return true, err
 		}
+		if !claimed {
+			// 一条 user_outbox 事实会展开为多个按 channel 独立幂等的投递。当前 channel 已终态、
+			// 被其他 worker 持有或尚未到重试时间时，只能跳过这一项；提前返回会把整条 MQ
+			// 消息确认掉，导致其后尚未落库或仍需补发的 channel 永久漏发。
+			continue
+		}
 		if err := s.publishDelivery(ctx, claimedDelivery, options); err != nil {
 			return true, err
 		}

services/notice-service/internal/modules/cpnotice/service_test.go

@@ -47,6 +47,39 @@ func TestProcessApplicationCreatedPublishesC2CAndBothRoomIMs(t *testing.T) {
 	}
 }
 
+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{}
@@ -187,12 +220,18 @@ func TestPublishFailureMarksRetryable(t *testing.T) {
 }
 
 type fakeRepository struct {
-	delivered []CPDelivery
+	notClaimedChannels map[string]bool
-	retryable []CPDelivery
+	claimedChannels    []string
-	failed    []CPDelivery
+	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
 }