sub2api/backend/internal/repository/user_msg_queue_cache_integration_test.go

//go:build integration

package repository

import (
	"errors"
	"testing"
	"time"

	"github.com/redis/go-redis/v9"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
)

type UserMsgQueueCacheSuite struct {
	IntegrationRedisSuite
	cache *userMsgQueueCache
}

func TestUserMsgQueueCacheSuite(t *testing.T) {
	suite.Run(t, new(UserMsgQueueCacheSuite))
}

func (s *UserMsgQueueCacheSuite) SetupTest() {
	s.IntegrationRedisSuite.SetupTest()
	s.cache = NewUserMsgQueueCache(s.rdb).(*userMsgQueueCache)
}

func (s *UserMsgQueueCacheSuite) TestAcquireLockWritesIndexAndReleaseRemovesIt() {
	accountID := int64(701)
	nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
	require.NoError(s.T(), err)

	acquired, err := s.cache.AcquireLock(s.ctx, accountID, "req-701", 10_000)
	require.NoError(s.T(), err)
	require.True(s.T(), acquired)

	score, err := s.rdb.ZScore(s.ctx, umqLockIndexKey, "701").Result()
	require.NoError(s.T(), err)
	require.Greater(s.T(), int64(score), nowMs)

	released, err := s.cache.ReleaseLock(s.ctx, accountID, "req-701")
	require.NoError(s.T(), err)
	require.True(s.T(), released)

	_, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "701").Result()
	require.ErrorIs(s.T(), err, redis.Nil)
}

func (s *UserMsgQueueCacheSuite) TestReconcileExpiredLockCandidatesRemovesNaturallyExpiredLockIndex() {
	accountID := int64(702)
	acquired, err := s.cache.AcquireLock(s.ctx, accountID, "req-702", 20)
	require.NoError(s.T(), err)
	require.True(s.T(), acquired)

	_, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "702").Result()
	require.NoError(s.T(), err)
	require.Eventually(s.T(), func() bool {
		_, err := s.rdb.Get(s.ctx, umqLockKey(accountID)).Result()
		return errors.Is(err, redis.Nil)
	}, time.Second, 10*time.Millisecond)

	cleaned, err := s.cache.ReconcileExpiredLockCandidates(s.ctx, 1000)
	require.NoError(s.T(), err)
	require.Equal(s.T(), 0, cleaned)

	_, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "702").Result()
	require.ErrorIs(s.T(), err, redis.Nil)
}

func (s *UserMsgQueueCacheSuite) TestReconcileExpiredLockCandidatesRefreshesLiveLockIndex() {
	accountID := int64(703)
	nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
	require.NoError(s.T(), err)
	require.NoError(s.T(), s.rdb.Set(s.ctx, umqLockKey(accountID), "req-703", time.Minute).Err())
	require.NoError(s.T(), s.rdb.ZAdd(s.ctx, umqLockIndexKey, redis.Z{
		Score:  float64(nowMs - 1),
		Member: "703",
	}).Err())

	cleaned, err := s.cache.ReconcileExpiredLockCandidates(s.ctx, 1000)
	require.NoError(s.T(), err)
	require.Equal(s.T(), 0, cleaned)

	score, err := s.rdb.ZScore(s.ctx, umqLockIndexKey, "703").Result()
	require.NoError(s.T(), err)
	require.Greater(s.T(), int64(score), nowMs)
	exists, err := s.rdb.Exists(s.ctx, umqLockKey(accountID)).Result()
	require.NoError(s.T(), err)
	require.EqualValues(s.T(), 1, exists)
}

func (s *UserMsgQueueCacheSuite) TestReconcileExpiredLockCandidatesDeletesNoTTLLock() {
	accountID := int64(704)
	nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
	require.NoError(s.T(), err)
	require.NoError(s.T(), s.rdb.Set(s.ctx, umqLockKey(accountID), "req-704", 0).Err())
	require.NoError(s.T(), s.rdb.ZAdd(s.ctx, umqLockIndexKey, redis.Z{
		Score:  float64(nowMs),
		Member: "704",
	}).Err())

	cleaned, err := s.cache.ReconcileExpiredLockCandidates(s.ctx, 1000)
	require.NoError(s.T(), err)
	require.Equal(s.T(), 1, cleaned)

	exists, err := s.rdb.Exists(s.ctx, umqLockKey(accountID)).Result()
	require.NoError(s.T(), err)
	require.EqualValues(s.T(), 0, exists)
	_, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "704").Result()
	require.ErrorIs(s.T(), err, redis.Nil)
}

func (s *UserMsgQueueCacheSuite) TestReconcileExpiredLockCandidatesRemovesInvalidMember() {
	nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
	require.NoError(s.T(), err)
	require.NoError(s.T(), s.rdb.ZAdd(s.ctx, umqLockIndexKey, redis.Z{
		Score:  float64(nowMs),
		Member: "not-an-account-id",
	}).Err())

	cleaned, err := s.cache.ReconcileExpiredLockCandidates(s.ctx, 1000)
	require.NoError(s.T(), err)
	require.Equal(s.T(), 0, cleaned)

	_, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "not-an-account-id").Result()
	require.True(s.T(), errors.Is(err, redis.Nil))
}

func (s *UserMsgQueueCacheSuite) TestAcquireLockBusyPathReindexesUnindexedLiveLock() {
	// 模拟索引丢失的存量锁（升级窗口/索引写失败/释放竞态误删）：
	// 锁存在且有 TTL，但索引里没有对应 member。
	accountID := int64(705)
	nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
	require.NoError(s.T(), err)
	require.NoError(s.T(), s.rdb.Set(s.ctx, umqLockKey(accountID), "holder-705", time.Minute).Err())

	// 另一个请求争锁失败，应顺手把观测到的持有者锁回填进索引。
	acquired, err := s.cache.AcquireLock(s.ctx, accountID, "contender-705", 10_000)
	require.NoError(s.T(), err)
	require.False(s.T(), acquired)

	score, err := s.rdb.ZScore(s.ctx, umqLockIndexKey, "705").Result()
	require.NoError(s.T(), err, "busy acquire should re-index the observed live lock")
	require.Greater(s.T(), int64(score), nowMs)
	// 锁本身不应被争锁方改动。
	val, err := s.rdb.Get(s.ctx, umqLockKey(accountID)).Result()
	require.NoError(s.T(), err)
	require.Equal(s.T(), "holder-705", val)
}

func (s *UserMsgQueueCacheSuite) TestAcquireLockBusyPathMakesNoTTLLockReconcilable() {
	// PTTL == -1 的异常锁若不在索引中，永远不会被 reconcile 发现；
	// 争锁失败路径必须以“已到期候选”的 score 回填它，形成自愈闭环。
	accountID := int64(706)
	require.NoError(s.T(), s.rdb.Set(s.ctx, umqLockKey(accountID), "holder-706", 0).Err())

	acquired, err := s.cache.AcquireLock(s.ctx, accountID, "contender-706", 10_000)
	require.NoError(s.T(), err)
	require.False(s.T(), acquired)

	nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
	require.NoError(s.T(), err)
	score, err := s.rdb.ZScore(s.ctx, umqLockIndexKey, "706").Result()
	require.NoError(s.T(), err, "busy acquire should index the anomalous lock")
	require.LessOrEqual(s.T(), int64(score), nowMs, "anomalous lock should be an immediately-expired candidate")

	cleaned, err := s.cache.ReconcileExpiredLockCandidates(s.ctx, 1000)
	require.NoError(s.T(), err)
	require.Equal(s.T(), 1, cleaned, "reconcile should delete the no-TTL lock")

	exists, err := s.rdb.Exists(s.ctx, umqLockKey(accountID)).Result()
	require.NoError(s.T(), err)
	require.EqualValues(s.T(), 0, exists, "queue is unblocked after reconcile")
	_, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "706").Result()
	require.ErrorIs(s.T(), err, redis.Nil)
}