hyapp-server/services/room-service/internal/room/service/gift_operation_internal_test.go

package service

import (
	"context"
	"errors"
	"testing"
	"time"

	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
	roomv1 "hyapp.local/api/proto/room/v1"
	"hyapp/pkg/appcode"
	"hyapp/pkg/xerr"
	"hyapp/services/room-service/internal/room/command"
)

type compensatedGiftOperationRepository struct {
	Repository
	operation       GiftOperation
	reactivateCalls int
}

func (r *compensatedGiftOperationRepository) EnsureGiftOperation(_ context.Context, _ GiftOperation) (GiftOperation, bool, error) {
	return r.operation, false, nil
}

func (r *compensatedGiftOperationRepository) ReactivateCompensatedGiftOperation(_ context.Context, _, _ string, updatedAtMS int64) (GiftOperation, bool, error) {
	r.reactivateCalls++
	if r.operation.Status == GiftOperationStatusCompensated {
		r.operation.Status = GiftOperationStatusPending
		r.operation.LastError = ""
		r.operation.UpdatedAtMS = updatedAtMS
		return r.operation, true, nil
	}
	return r.operation, false, nil
}

func TestGiftOperationRecoveryAttemptFitsInsideClaimLease(t *testing.T) {
	// mutateRoom 在钱包可能已经扣费后允许一个独立的持久提交窗口，失败处理也会脱离过期请求落 retry；
	// claim 租约必须覆盖三段最坏边界，否则第二实例会在第一实例仍收尾时重复施压。
	maxBoundedWork := giftOperationAttemptTimeout + roomMutationCommitTimeout + giftOperationPersistTimeout
	if maxBoundedWork >= giftOperationClaimTTL {
		t.Fatalf("recovery bounded work %s must be shorter than claim lease %s", maxBoundedWork, giftOperationClaimTTL)
	}
}

func TestGiftOperationRecoveryDeadlineKeepsClaimUntilCellCommitBudgetEnds(t *testing.T) {
	for _, err := range []error{
		context.DeadlineExceeded,
		context.Canceled,
		status.Error(codes.DeadlineExceeded, "wallet timed out"),
		status.Error(codes.Canceled, "attempt canceled"),
	} {
		if !giftOperationFailureMustKeepClaim(true, err) {
			t.Fatalf("recovery timeout must preserve its claim: %v", err)
		}
	}
	if giftOperationFailureMustKeepClaim(false, context.DeadlineExceeded) {
		t.Fatal("foreground request has no recovery claim to preserve")
	}
	if giftOperationFailureMustKeepClaim(true, errors.New("wallet unavailable")) {
		t.Fatal("completed retryable failures may release the claim for the next attempt")
	}
}

func TestCompensatedGiftOperationReactivatesOnlyAfterPayloadMatch(t *testing.T) {
	ctx := appcode.WithContext(context.Background(), appcode.Default)
	originalRequest := &roomv1.SendGiftRequest{
		Meta:       &roomv1.RequestMeta{AppCode: appcode.Default, RoomId: "room-reactivate", CommandId: "cmd-reactivate", ActorUserId: 101},
		TargetType: "user", TargetUserId: 202, GiftId: "rose", GiftCount: 1, SenderRegionId: 10,
		TargetIsHost: true, TargetHostRegionId: 20, TargetAgencyOwnerUserId: 303,
		SenderDisplayProfile: &roomv1.SendGiftDisplayProfile{UserId: 101, Username: "first sender"},
	}
	// Build the stored command through the same normalizer used by production so this test exercises
	// the semantic idempotency comparison instead of depending on hand-written JSON field defaults.
	originalFlow := newGiftFlow(&Service{}, ctx, originalRequest, giftSendOptions{})
	storedPayload, err := command.Serialize(originalFlow.cmd)
	if err != nil {
		t.Fatalf("serialize stored gift command: %v", err)
	}
	storedRequestPayload, err := proto.Marshal(originalRequest)
	if err != nil {
		t.Fatalf("serialize stored gift request: %v", err)
	}

	t.Run("conflicting payload stays compensated", func(t *testing.T) {
		repository := &compensatedGiftOperationRepository{operation: GiftOperation{
			AppCode: appcode.Default, RoomID: "room-reactivate", CommandID: "cmd-reactivate",
			CommandPayload: storedPayload, RequestPayload: storedRequestPayload,
			Status: GiftOperationStatusCompensated, LastError: "insufficient balance",
		}}
		service := &Service{repository: repository}
		conflictingRequest := &roomv1.SendGiftRequest{
			Meta:       &roomv1.RequestMeta{AppCode: appcode.Default, RoomId: "room-reactivate", CommandId: "cmd-reactivate", ActorUserId: 101},
			TargetType: "user", TargetUserId: 202, GiftId: "different-gift", GiftCount: 1,
		}
		flow := newGiftFlow(service, ctx, conflictingRequest, giftSendOptions{})
		if err := flow.ensureGiftOperation(time.UnixMilli(2_000)); !xerr.IsCode(err, xerr.Conflict) {
			t.Fatalf("conflicting command must fail before reactivation: %v", err)
		}
		if repository.reactivateCalls != 0 || repository.operation.Status != GiftOperationStatusCompensated {
			t.Fatalf("conflict exposed stored work to recovery: calls=%d operation=%+v", repository.reactivateCalls, repository.operation)
		}
	})

	t.Run("identical payload CAS reactivates", func(t *testing.T) {
		repository := &compensatedGiftOperationRepository{operation: GiftOperation{
			AppCode: appcode.Default, RoomID: "room-reactivate", CommandID: "cmd-reactivate",
			CommandPayload: storedPayload, RequestPayload: storedRequestPayload,
			Status: GiftOperationStatusCompensated, LastError: "insufficient balance",
		}}
		service := &Service{repository: repository}
		flow := newGiftFlow(service, ctx, originalRequest, giftSendOptions{})
		if err := flow.ensureGiftOperation(time.UnixMilli(2_000)); err != nil {
			t.Fatalf("identical command should reactivate: %v", err)
		}
		if repository.reactivateCalls != 1 || repository.operation.Status != GiftOperationStatusPending || !flow.operationStarted {
			t.Fatalf("identical command did not enter pending exactly once: calls=%d operation=%+v", repository.reactivateCalls, repository.operation)
		}
	})

	t.Run("load miss duplicate resumes first gateway snapshots", func(t *testing.T) {
		repository := &compensatedGiftOperationRepository{operation: GiftOperation{
			AppCode: appcode.Default, RoomID: "room-reactivate", CommandID: "cmd-reactivate",
			CommandPayload: storedPayload, RequestPayload: storedRequestPayload, Status: GiftOperationStatusPending,
		}}
		service := &Service{repository: repository}
		refreshedRequest := proto.Clone(originalRequest).(*roomv1.SendGiftRequest)
		refreshedRequest.SenderRegionId = 99
		refreshedRequest.TargetHostRegionId = 88
		refreshedRequest.TargetAgencyOwnerUserId = 404
		refreshedRequest.SenderDisplayProfile = &roomv1.SendGiftDisplayProfile{UserId: 101, Username: "renamed sender"}
		flow := newGiftFlow(service, ctx, refreshedRequest, giftSendOptions{})
		if err := flow.ensureGiftOperation(time.UnixMilli(2_000)); err != nil {
			t.Fatalf("idempotency-equivalent duplicate should resume stored operation: %v", err)
		}
		if !flow.recovering || flow.cmd.SenderRegionID != 10 || flow.req.GetSenderRegionId() != 10 || flow.cmd.TargetHostRegionID != 20 || flow.req.GetTargetHostRegionId() != 20 || giftDisplayName(flow.cmd.SenderDisplayProfile) != "first sender" {
			t.Fatalf("duplicate continued with refreshed gateway snapshots: command=%+v request=%+v", flow.cmd, flow.req)
		}
		if repository.reactivateCalls != 0 {
			t.Fatalf("already-pending duplicate must not reactivate: %d", repository.reactivateCalls)
		}
	})
}