topfans/backend/gateway/pkg/starcache/starcache_test.go

package starcache

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"dubbo.apache.org/dubbo-go/v3/client"
	"github.com/topfans/backend/pkg/logger"
	pb "github.com/topfans/backend/pkg/proto/user"
	"go.uber.org/zap"
)

// TestMain 注入 no-op logger,避免 logger.Logger 为 nil 时 GetStar 中的 WARN
// 日志调用造成 panic。
func TestMain(m *testing.M) {
	if logger.Logger == nil {
		logger.Logger = zap.NewNop()
	}
	logger.Sugar = logger.Logger.Sugar()
	m.Run()
}

// fakeUserClient 是 pb.UserSocialService 的最小存根,只实现 GetFanIdentities。
// 其他方法通过嵌入 nil 接口占位,GetStar 路径只触达 GetFanIdentities。
type fakeUserClient struct {
	pb.UserSocialService // 嵌入 nil 接口,未 stub 的方法调用时 panic(测试不会触发)

	getFanIdentitiesCalls int32 // atomic
	getFanIdentitiesErr   error
	stars                 []*pb.Star

	// refreshBlock 用来在并发测试里阻塞 GetFanIdentities,便于验证 singleflight
	refreshBlock chan struct{}
}

func (f *fakeUserClient) GetFanIdentities(ctx context.Context, req *pb.GetFanIdentitiesRequest, opts ...client.CallOption) (*pb.GetFanIdentitiesResponse, error) {
	atomic.AddInt32(&f.getFanIdentitiesCalls, 1)
	if f.refreshBlock != nil {
		<-f.refreshBlock
	}
	if f.getFanIdentitiesErr != nil {
		return nil, f.getFanIdentitiesErr
	}
	return &pb.GetFanIdentitiesResponse{Stars: f.stars}, nil
}

// ensure interface compliance at compile time
var _ pb.UserSocialService = (*fakeUserClient)(nil)

// TestGetStar_CacheMissThenHit 验证首次 miss 触发 RPC,再次请求命中缓存不再 RPC。
func TestGetStar_CacheMissThenHit(t *testing.T) {
	star := &pb.Star{StarId: 42, IdentityId: "id-42", Name: "Star 42"}
	cli := &fakeUserClient{stars: []*pb.Star{star}}
	c := New(cli)

	// 1st call: miss -> RPC
	got, err := c.GetStar(context.Background(), 42)
	if err != nil {
		t.Fatalf("1st GetStar: unexpected err: %v", err)
	}
	if got == nil || got.StarId != 42 {
		t.Fatalf("1st GetStar: expected star 42, got %+v", got)
	}
	if atomic.LoadInt32(&cli.getFanIdentitiesCalls) != 1 {
		t.Fatalf("expected 1 RPC after 1st miss, got %d", cli.getFanIdentitiesCalls)
	}

	// 2nd call: hit -> no extra RPC
	got, err = c.GetStar(context.Background(), 42)
	if err != nil || got == nil || got.StarId != 42 {
		t.Fatalf("2nd GetStar: unexpected result err=%v got=%+v", err, got)
	}
	if atomic.LoadInt32(&cli.getFanIdentitiesCalls) != 1 {
		t.Fatalf("expected 1 RPC after cached hit, got %d", cli.getFanIdentitiesCalls)
	}
}

// TestGetStar_ExpiryTriggersRefresh 验证 TTL 过期后再次请求会刷新。
// 我们把 CacheTTL 临时缩短为 10ms(通过构造时直接修改 map),避免 60s 真实等待。
func TestGetStar_ExpiryTriggersRefresh(t *testing.T) {
	star := &pb.Star{StarId: 7, IdentityId: "id-7"}
	cli := &fakeUserClient{stars: []*pb.Star{star}}
	c := New(cli)

	if _, err := c.GetStar(context.Background(), 7); err != nil {
		t.Fatalf("prime: %v", err)
	}

	// 手工把 expiry 拨到过去,模拟 TTL 过期。
	c.mu.Lock()
	c.expiry[7] = time.Now().Add(-time.Second)
	c.mu.Unlock()

	// 换一个新名字以验证数据被刷新
	cli.stars = []*pb.Star{{StarId: 7, IdentityId: "id-7", Name: "updated"}}

	got, err := c.GetStar(context.Background(), 7)
	if err != nil {
		t.Fatalf("post-expiry: %v", err)
	}
	if got.Name != "updated" {
		t.Fatalf("expected refreshed star Name=updated, got %+v", got)
	}
	if atomic.LoadInt32(&cli.getFanIdentitiesCalls) != 2 {
		t.Fatalf("expected 2 RPCs after expiry, got %d", cli.getFanIdentitiesCalls)
	}
}

// TestGetStar_StarIDZero_NoOp 验证 starID==0 不调 RPC、返回 nil。
func TestGetStar_StarIDZero_NoOp(t *testing.T) {
	cli := &fakeUserClient{}
	c := New(cli)
	got, err := c.GetStar(context.Background(), 0)
	if err != nil || got != nil {
		t.Fatalf("expected (nil, nil) for starID=0, got (%+v, %v)", got, err)
	}
	if n := atomic.LoadInt32(&cli.getFanIdentitiesCalls); n != 0 {
		t.Fatalf("expected 0 RPCs for starID=0, got %d", n)
	}
}

// TestGetStar_RPCError 验证上游 RPC 失败时返回 error,不写缓存。
func TestGetStar_RPCError(t *testing.T) {
	cli := &fakeUserClient{getFanIdentitiesErr: errors.New("upstream boom")}
	c := New(cli)

	_, err := c.GetStar(context.Background(), 99)
	if err == nil {
		t.Fatal("expected error from upstream RPC")
	}
	if c.Len() != 0 {
		t.Fatalf("expected empty cache after RPC error, got %d entries", c.Len())
	}

	// 错误恢复后,下一次 GetStar 应该成功并填充缓存。
	cli.getFanIdentitiesErr = nil
	cli.stars = []*pb.Star{{StarId: 99, IdentityId: "id-99"}}
	got, err := c.GetStar(context.Background(), 99)
	if err != nil || got == nil || got.StarId != 99 {
		t.Fatalf("post-recovery: expected star 99, got (%+v, %v)", got, err)
	}
}

// TestGetStar_SingleflightCollapsesConcurrentMisses 验证 singleflight:100 个并发
// miss 应只触发 1 次 RPC,而不是 100 次。等价于"雪崩防护"。
func TestGetStar_SingleflightCollapsesConcurrentMisses(t *testing.T) {
	star := &pb.Star{StarId: 11, IdentityId: "id-11"}
	cli := &fakeUserClient{
		stars:       []*pb.Star{star},
		refreshBlock: make(chan struct{}),
	}
	c := New(cli)

	const N = 100
	var wg sync.WaitGroup
	wg.Add(N)
	results := make([]*pb.Star, N)
	for i := 0; i < N; i++ {
		i := i
		go func() {
			defer wg.Done()
			s, _ := c.GetStar(context.Background(), 11)
			results[i] = s
		}()
	}
	// 给 goroutine 一点时间开始"miss"
	time.Sleep(20 * time.Millisecond)
	// 放行单飞 leader 的 RPC
	close(cli.refreshBlock)
	wg.Wait()

	if got := atomic.LoadInt32(&cli.getFanIdentitiesCalls); got != 1 {
		t.Fatalf("expected 1 RPC under singleflight, got %d", got)
	}
	for i, s := range results {
		if s == nil || s.StarId != 11 {
			t.Fatalf("result[%d]: expected star 11, got %+v", i, s)
		}
	}
}

// TestInvalidate 验证 Invalidate 后下一次 GetStar 重新走 RPC。
func TestInvalidate(t *testing.T) {
	star := &pb.Star{StarId: 5, IdentityId: "id-5"}
	cli := &fakeUserClient{stars: []*pb.Star{star}}
	c := New(cli)

	if _, err := c.GetStar(context.Background(), 5); err != nil {
		t.Fatal(err)
	}
	// 缓存已命中。
	c.Invalidate(5)

	// 刷新上游数据,验证下次 GetStar 拿到新值。
	cli.stars = []*pb.Star{{StarId: 5, IdentityId: "id-5", Name: "after-invalidate"}}
	got, err := c.GetStar(context.Background(), 5)
	if err != nil {
		t.Fatal(err)
	}
	if got.Name != "after-invalidate" {
		t.Fatalf("expected name=after-invalidate, got %q", got.Name)
	}
}