package shortrecommend
import (
"context"
"errors"
"fmt"
"reflect"
"strings"
"sync"
"testing"
"time"
"91porn-server/common/redis"
"91porn-server/models/v/vidmod"
"go.mongodb.org/mongo-driver/bson/primitive"
)
type fakePublishClient struct {
queue []string
pipelineWindowSizes []int
pipelineBatchSizes []int
pipelineExpirations []time.Duration
pipelineCalls int
pipelineErrorAt int
pipelineDuplicateAt int
cancelAfterPipelineAt int
cancel context.CancelFunc
duplicateFirstPush bool
firstPushTTL int64
metaBuildingTTL int64
publishedTTL int64
metaLength int
swapAttempts int
swapped bool
deleteCalls int
queueLengthBeforeDelete int
swapScript string
swapKeys []string
swapArgs []interface{}
}
func (f *fakePublishClient) DelContext(_ context.Context, _ ...string) (int64, error) {
f.deleteCalls++
f.queueLengthBeforeDelete = len(f.queue)
f.queue = nil
return 1, nil
}
func (f *fakePublishClient) EvalContext(
_ context.Context,
script string,
keys []string,
args ...interface{},
) (interface{}, error) {
switch {
case strings.Contains(script, "for i = 1, #ARGV - 1"):
if len(args) < 2 {
return nil, fmt.Errorf("invalid first push args")
}
ids := make([]string, len(args)-1)
for i := range ids {
id, ok := args[i].(string)
if !ok {
return nil, fmt.Errorf("first push id %d is not a string", i)
}
ids[i] = id
}
ttl, ok := args[len(args)-1].(int64)
if !ok {
return nil, fmt.Errorf("first push ttl is not int64")
}
f.firstPushTTL = ttl
f.queue = append(f.queue, ids...)
if f.duplicateFirstPush {
f.queue = append(f.queue, ids...)
}
return int64(1), nil
case strings.Contains(script, "redis.call('HSET', KEYS[1]"):
length, ok := args[0].(int)
if !ok {
return nil, fmt.Errorf("meta length is not int")
}
f.metaLength = length
ttl, ok := args[3].(int64)
if !ok {
return nil, fmt.Errorf("meta ttl is not int64")
}
f.metaBuildingTTL = ttl
return int64(1), nil
case strings.Contains(script, "redis.call('RENAME'"):
f.swapAttempts++
f.swapScript = script
f.swapKeys = append([]string(nil), keys...)
f.swapArgs = append([]interface{}(nil), args...)
ttl, ok := args[1].(int64)
if !ok {
return nil, fmt.Errorf("published ttl is not int64")
}
f.publishedTTL = ttl
expected, ok := args[4].(int)
if !ok {
return nil, fmt.Errorf("expected queue length is not int")
}
if len(f.queue) != expected {
return int64(-2), nil
}
f.swapped = true
return int64(1), nil
default:
return nil, fmt.Errorf("unexpected script")
}
}
type fakeScriptClient struct {
result interface{}
err error
run func(*fakeScriptClient, context.Context, *redis.Script, []string, ...interface{}) (interface{}, error)
calls int
ctx context.Context
script *redis.Script
keys []string
args []interface{}
}
func (f *fakeScriptClient) RunScriptContext(
ctx context.Context,
script *redis.Script,
keys []string,
args ...interface{},
) (interface{}, error) {
f.calls++
f.ctx = ctx
f.script = script
f.keys = append([]string(nil), keys...)
f.args = append([]interface{}(nil), args...)
if f.run != nil {
return f.run(f, ctx, script, keys, args...)
}
return f.result, f.err
}
func (f *fakePublishClient) RPushPipelineContext(
_ context.Context,
_ string,
values []string,
batchSize int,
expirations ...time.Duration,
) (int64, error) {
f.pipelineCalls++
f.pipelineWindowSizes = append(f.pipelineWindowSizes, len(values))
f.pipelineBatchSizes = append(f.pipelineBatchSizes, batchSize)
if len(expirations) > 0 {
f.pipelineExpirations = append(f.pipelineExpirations, expirations[0])
}
if f.pipelineCalls == f.pipelineErrorAt {
f.queue = append(f.queue, values[:len(values)/2]...)
return 0, fmt.Errorf("injected pipeline error")
}
f.queue = append(f.queue, values...)
if f.pipelineCalls == f.pipelineDuplicateAt {
f.queue = append(f.queue, values...)
}
if f.pipelineCalls == f.cancelAfterPipelineAt && f.cancel != nil {
f.cancel()
}
return int64(len(f.queue)), nil
}
func candidate(n int, score int64, reviewAt time.Time) vidmod.RecommendCandidate {
return vidmod.RecommendCandidate{
ID: primitive.NewObjectIDFromTimestamp(time.Unix(int64(n+1), int64(n))),
RecommendScore: score,
ReviewAt: reviewAt,
}
}
func TestAssembleSeventeenHighThreeNew(t *testing.T) {
now := time.Date(2026, 7, 24, 8, 0, 0, 0, time.UTC)
input := make([]vidmod.RecommendCandidate, 0, 40)
for i := 0; i < 34; i++ {
input = append(input, candidate(i, int64(1000-i), now.Add(-48*time.Hour)))
}
for i := 0; i < 6; i++ {
input = append(input, candidate(100+i, 0, now.Add(-time.Duration(i)*time.Hour)))
}
got := Assemble(input, now)
if len(got) != 40 {
t.Fatalf("length=%d want=40", len(got))
}
newIDs := map[string]bool{}
for i := 0; i < 6; i++ {
newIDs[input[34+i].ID.Hex()] = true
}
for block := 0; block < 2; block++ {
count := 0
for _, id := range got[block*20 : block*20+20] {
if newIDs[id] {
count++
}
}
if count != 3 {
t.Fatalf("block %d new count=%d want=3", block, count)
}
}
assertUnique(t, got)
}
func TestAssembleWithStats(t *testing.T) {
now := time.Date(2026, 7, 24, 8, 0, 0, 0, time.UTC)
input := make([]vidmod.RecommendCandidate, 0, 23)
for i := 0; i < 20; i++ {
input = append(input, candidate(i, int64(1000-i), now.Add(-48*time.Hour)))
}
for i := 0; i < 3; i++ {
input = append(input, candidate(100+i, 0, now.Add(-time.Duration(i)*time.Hour)))
}
ids, stats := AssembleWithStats(input, now)
if len(ids) != len(input) {
t.Fatalf("len(ids) = %d, want %d", len(ids), len(input))
}
if stats.TotalCount != 23 || stats.HighCount != 20 ||
stats.NewCount != 3 || stats.BlockCount != 2 {
t.Fatalf("stats = %+v", stats)
}
}
func TestAssembleNewShortageFilledByHigh(t *testing.T) {
now := time.Now()
for newCount := 0; newCount <= 3; newCount++ {
t.Run(fmt.Sprintf("new_%d", newCount), func(t *testing.T) {
input := make([]vidmod.RecommendCandidate, 0, 20)
for i := 0; i < 20-newCount; i++ {
input = append(input, candidate(i, int64(100-i), now.Add(-48*time.Hour)))
}
for i := 0; i < newCount; i++ {
input = append(input, candidate(100+i, 0, now.Add(-time.Hour)))
}
got := Assemble(input, now)
if len(got) != 20 {
t.Fatalf("length=%d want=20", len(got))
}
assertUnique(t, got)
})
}
}
func TestAssembleTailAndDeterminism(t *testing.T) {
now := time.Date(2026, 7, 24, 8, 0, 0, 0, time.UTC)
input := make([]vidmod.RecommendCandidate, 0, 9)
for i := 0; i < 7; i++ {
input = append(input, candidate(i, int64(i), now.Add(-48*time.Hour)))
}
input = append(input, candidate(100, 0, now.Add(-time.Hour)), candidate(101, 0, now.Add(-2*time.Hour)))
first, second := Assemble(input, now), Assemble(input, now)
if fmt.Sprint(first) != fmt.Sprint(second) {
t.Fatalf("same version must be deterministic")
}
if len(first) != 9 {
t.Fatalf("tail length=%d want=9", len(first))
}
assertUnique(t, first)
}
func TestAssembleReviewAtBoundariesExcludesFuture(t *testing.T) {
now := time.Date(2026, 7, 24, 8, 0, 0, 0, time.UTC)
input := []vidmod.RecommendCandidate{
candidate(1, 100, now.Add(-24*time.Hour)),
candidate(2, 99, now.Add(-24*time.Hour-time.Nanosecond)),
candidate(3, 98, now.Add(time.Nanosecond)),
}
got, stats := AssembleWithStats(input, now)
if len(got) != 2 {
t.Fatalf("length=%d want=2", len(got))
}
assertUnique(t, got)
if stats.TotalCount != 2 || stats.NewCount != 1 || stats.HighCount != 1 {
t.Fatalf("stats=%+v want total=2 new=1 high=1", stats)
}
futureID := input[2].ID.Hex()
for _, id := range got {
if id == futureID {
t.Fatalf("future reviewAt video %s must not enter recommendation pools", futureID)
}
}
}
func TestTakeRejectsNonPositiveSizeBeforeRedis(t *testing.T) {
for _, size := range []int{0, -1} {
result, err := Take(nil, 123, size)
if err != nil {
t.Fatalf("Take(size=%d) error = %v", size, err)
}
if result.Length != 0 || len(result.IDs) != 0 {
t.Fatalf("Take(size=%d) result = %+v, want empty", size, result)
}
}
}
func TestTakeVersionContextUsesReusableScriptAndExpectedVersion(t *testing.T) {
ctx := context.WithValue(context.Background(), struct{}{}, "request")
client := &fakeScriptClient{
result: []interface{}{"20260731", "4", "2", "video-3", "video-4"},
}
result, err := takeContext(ctx, client, 123, 2, "20260731", "")
if err != nil {
t.Fatalf("TakeVersionContext() error = %v", err)
}
if client.calls != 1 || client.ctx != ctx {
t.Fatalf("script calls = %d context propagated = %v", client.calls, client.ctx == ctx)
}
if client.script != takeRedisScript || client.script.Hash() == "" {
t.Fatal("TakeVersionContext() did not reuse the cached Redis script")
}
if got := client.args[5]; got != "20260731" {
t.Fatalf("expected version arg = %v, want 20260731", got)
}
if result.Version != "20260731" || result.Length != 4 || result.Offset != 2 ||
!reflect.DeepEqual(result.IDs, []string{"video-3", "video-4"}) {
t.Fatalf("TakeVersionContext() result = %+v", result)
}
}
func TestTakeVersionContextReportsVersionChangeWithoutIDs(t *testing.T) {
client := &fakeScriptClient{
result: []interface{}{"20260801", "-1", "0"},
}
result, err := takeContext(
context.Background(), client, 123, 20, "20260731", "",
)
if !errors.Is(err, ErrVersionChanged) {
t.Fatalf("TakeVersionContext() error = %v, want ErrVersionChanged", err)
}
if result.Version != "20260801" || len(result.IDs) != 0 {
t.Fatalf("TakeVersionContext() result = %+v", result)
}
}
func TestTakeVersionContextReportsUnhealthyQueue(t *testing.T) {
client := &fakeScriptClient{
result: []interface{}{"20260731", "-2", "0"},
}
_, err := takeContext(context.Background(), client, 123, 20, "20260731", "")
if !errors.Is(err, ErrQueueUnhealthy) {
t.Fatalf("TakeVersionContext() error = %v, want ErrQueueUnhealthy", err)
}
}
func TestTakeContextStopsBeforeRedisWhenCancelled(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
cancel()
client := &fakeScriptClient{}
_, err := takeContext(ctx, client, 123, 20, "20260731", "")
if !errors.Is(err, context.Canceled) {
t.Fatalf("TakeVersionContext() error = %v, want context.Canceled", err)
}
if client.calls != 0 {
t.Fatalf("script calls = %d, want 0", client.calls)
}
}
func TestTakeIdempotentContextBuildsStableScopedCacheKey(t *testing.T) {
client := &fakeScriptClient{
result: []interface{}{"20260731", "4", "0", "video-1", "video-2"},
}
_, err := takeIdempotentContext(
context.Background(), client, 123, 2, "20260731", " request-1 ", 3,
)
if err != nil {
t.Fatalf("takeIdempotentContext() error = %v", err)
}
idempotencyKey, ok := client.args[6].(string)
if !ok || !strings.HasPrefix(
idempotencyKey,
"recommend:short:take-cache:123:",
) || !strings.HasSuffix(idempotencyKey, ":3") {
t.Fatalf("idempotency key = %#v", client.args[6])
}
if strings.Contains(idempotencyKey, "request-1") {
t.Fatal("raw request ID must not be copied into Redis keys")
}
if client.args[7] != int64(takeIdempotencyTTL/time.Second) {
t.Fatalf("idempotency ttl = %v", client.args[7])
}
}
func TestTakeIdempotentContextAllowsFirstBatchWithoutExpectedVersion(t *testing.T) {
client := &fakeScriptClient{
result: []interface{}{"20260731", "4", "0", "video-1"},
}
result, err := takeIdempotentContext(
context.Background(), client, 123, 1, "", "request-1", 0,
)
if err != nil {
t.Fatalf("takeIdempotentContext() error = %v", err)
}
if result.Version != "20260731" || client.args[5] != "" {
t.Fatalf("result = %+v expected version arg = %#v", result, client.args[5])
}
}
func TestTakeCurrentIdempotencyCacheIsScopedByLuaCurrentVersion(t *testing.T) {
if !strings.Contains(
takeScriptSource,
"idempotencyKey = ARGV[7] .. ':' .. version",
) {
t.Fatal("take script must scope its idempotency cache by the atomic current version")
}
first := &fakeScriptClient{
result: []interface{}{"20260731", "4", "0", "video-1"},
}
second := &fakeScriptClient{
result: []interface{}{"20260801", "4", "0", "video-2"},
}
if _, err := takeIdempotentContext(
context.Background(), first, 123, 1, "", "request-1", 0,
); err != nil {
t.Fatalf("first take error = %v", err)
}
if _, err := takeIdempotentContext(
context.Background(), second, 123, 1, "", "request-1", 0,
); err != nil {
t.Fatalf("second take error = %v", err)
}
if first.args[6] != second.args[6] {
t.Fatalf("cache base differs: first=%v second=%v", first.args[6], second.args[6])
}
// Lua appends the version it read atomically, yielding different physical
// keys ":20260731" and ":20260801".
}
func TestTakeIdempotencyCacheStoresReservationInsteadOfVideoIDs(t *testing.T) {
for _, want := range []string{
"local function readBatch",
"local cachedWanted = tonumber(cached[4])",
"version, tostring(length), tostring(offset), tostring(wanted)",
} {
if !strings.Contains(takeScriptSource, want) {
t.Fatalf("take script missing compact reservation logic %q", want)
}
}
if strings.Contains(takeScriptSource, "unpack(result)") {
t.Fatal("take script must not duplicate all video IDs in idempotency cache")
}
}
func TestTakeIdempotentContextValidatesScope(t *testing.T) {
longRequestID := strings.Repeat("x", maxTakeRequestIDLength+1)
tests := []struct {
name string
expectedVersion string
requestID string
batchIndex int
}{
{name: "blank request", expectedVersion: "20260731", requestID: " ", batchIndex: 0},
{name: "long request", expectedVersion: "20260731", requestID: longRequestID, batchIndex: 0},
{name: "negative batch", expectedVersion: "20260731", requestID: "request", batchIndex: -1},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
client := &fakeScriptClient{}
_, err := takeIdempotentContext(
context.Background(), client, 1, 20,
tt.expectedVersion, tt.requestID, tt.batchIndex,
)
if err == nil {
t.Fatal("takeIdempotentContext() should reject invalid scope")
}
if client.calls != 0 {
t.Fatalf("script calls = %d, want 0", client.calls)
}
})
}
}
func TestTakeScriptUsesFixedAbsoluteOffsetExpiration(t *testing.T) {
if !strings.Contains(takeScriptSource, "redis.call('PEXPIREAT', offsetKey, expiresAtMs)") {
t.Fatal("take script must align offset to the queue absolute expiration")
}
if !strings.Contains(takeScriptSource, "if offsetKeyTTL <= 0 then") {
t.Fatal("take script must set offset expiration only when the hash has no TTL")
}
if strings.Contains(takeScriptSource, "redis.call('EXPIRE', offsetKey") {
t.Fatal("take script must not renew a sliding offset TTL")
}
}
func TestReserveContextUsesStableReceiptAndRandomLeaseWithoutAdvancingOffset(t *testing.T) {
newClient := func() *fakeScriptClient {
return &fakeScriptClient{
run: func(
f *fakeScriptClient,
_ context.Context,
_ *redis.Script,
_ []string,
args ...interface{},
) (interface{}, error) {
token := fmt.Sprint(args[8])
return []interface{}{
"20260731-build", "RESERVED", "4", "3", "2", token,
"video-4", "video-1",
}, nil
},
}
}
firstClient, secondClient := newClient(), newClient()
first, err := reserveContext(
context.Background(), firstClient, 123, 2, "", "request-1",
)
if err != nil {
t.Fatalf("first reserve error = %v", err)
}
second, err := reserveContext(
context.Background(), secondClient, 123, 2, "", "request-1",
)
if err != nil {
t.Fatalf("second reserve error = %v", err)
}
if first.Version != "20260731-build" || first.Offset != 3 ||
first.Reserved != 2 ||
!reflect.DeepEqual(first.IDs, []string{"video-4", "video-1"}) {
t.Fatalf("first reservation = %+v", first)
}
if first.ReceiptID == "" || first.ReceiptID != second.ReceiptID {
t.Fatalf("receipt IDs first=%q second=%q", first.ReceiptID, second.ReceiptID)
}
if first.LeaseToken == "" || first.LeaseToken == second.LeaseToken {
t.Fatalf("lease tokens first=%q second=%q", first.LeaseToken, second.LeaseToken)
}
if firstClient.script != reserveRedisScript ||
firstClient.args[10] != int64(reservationTTL/time.Millisecond) {
t.Fatal("reserve did not use the reusable script and bounded lease TTL")
}
if strings.Contains(reserveScriptSource, "HSET', offsetKey") {
t.Fatal("Reserve must never advance or create the user offset hash")
}
for _, want := range []string{
"'HMGET', leaseKey",
"redis.call('PEXPIRE', leaseKey",
"'COMMITTED'",
"'BUSY'",
} {
if !strings.Contains(reserveScriptSource, want) {
t.Fatalf("reserve script missing %q", want)
}
}
}
func TestReserveContextParsesCommittedReceiptAndErrors(t *testing.T) {
committedClient := &fakeScriptClient{
result: []interface{}{
"20260731-old", "COMMITTED", "4", "1", "2", "lease",
"video-2", "video-3",
},
}
committed, err := reserveContext(
context.Background(), committedClient, 123, 2, "", "request",
)
if err != nil {
t.Fatalf("committed reserve error = %v", err)
}
if !committed.AlreadyCommitted || committed.Reserved != 2 {
t.Fatalf("committed reservation = %+v", committed)
}
tests := []struct {
status string
want error
}{
{status: "VERSION_CHANGED", want: ErrVersionChanged},
{status: "UNHEALTHY", want: ErrQueueUnhealthy},
{status: "BUSY", want: ErrReservationBusy},
}
for _, tt := range tests {
t.Run(tt.status, func(t *testing.T) {
client := &fakeScriptClient{
result: []interface{}{"20260731", tt.status, "0", "0", "0", ""},
}
result, err := reserveContext(
context.Background(), client, 1, 20, "", "request",
)
if !errors.Is(err, tt.want) || result.Version != "20260731" {
t.Fatalf("result=%+v error=%v want=%v", result, err, tt.want)
}
})
}
}
func TestReserveContextRetriesUncertainResponseWithSameLease(t *testing.T) {
wantErr := errors.New("connection reset after write")
client := &fakeScriptClient{}
var firstArgs []interface{}
client.run = func(
f *fakeScriptClient,
_ context.Context,
_ *redis.Script,
_ []string,
args ...interface{},
) (interface{}, error) {
if f.calls == 1 {
firstArgs = append([]interface{}(nil), args...)
return nil, wantErr
}
if !reflect.DeepEqual(args, firstArgs) {
t.Fatalf("retry args changed:\nfirst=%#v\nretry=%#v", firstArgs, args)
}
return []interface{}{
"20260731-build", "RESERVED", "2", "0", "2",
fmt.Sprint(args[8]), "video-1", "video-2",
}, nil
}
reservation, err := reserveContext(
context.Background(), client, 123, 2, "", "request",
)
if err != nil {
t.Fatalf("reserveContext() error = %v", err)
}
if client.calls != 2 || reservation.LeaseToken != fmt.Sprint(firstArgs[8]) ||
reservation.ReceiptID != fmt.Sprint(firstArgs[9]) {
t.Fatalf("calls=%d reservation=%+v args=%#v", client.calls, reservation, firstArgs)
}
}
func TestCommitReservationContextUsesCASReceiptAndAbsoluteTTL(t *testing.T) {
reservation := Reservation{
Version: "20260731-build",
Length: 10,
Offset: 8,
Reserved: 5,
LeaseToken: "lease",
ReceiptID: "receipt",
}
for _, raw := range []interface{}{int64(1), int64(2)} {
client := &fakeScriptClient{result: raw}
if err := commitReservationContext(
context.Background(), client, 123, reservation, 3,
); err != nil {
t.Fatalf("commit result %v error = %v", raw, err)
}
if client.script != commitReservationRedisScript {
t.Fatal("commit did not reuse cached Redis script")
}
if client.args[11] != 3 ||
client.args[12] != int64(takeIdempotencyTTL/time.Second) ||
client.args[13] != "recommend:short:commit-receipt:" {
t.Fatalf("commit args = %#v", client.args)
}
}
for _, want := range []string{
"currentOffset ~= tonumber(lease[4])",
"(currentOffset + consumed) % length",
"redis.call('PEXPIREAT', offsetKey, expiresAtMs)",
"'RPUSH', receiptKey",
} {
if !strings.Contains(commitReservationScriptSource, want) {
t.Fatalf("commit script missing %q", want)
}
}
if strings.Index(commitReservationScriptSource, "local receipt =") >
strings.Index(commitReservationScriptSource, "local version =") {
t.Fatal("commit must check its receipt before current for uncertain-response retries")
}
}
func TestCommitReservationContextMapsRejectedStates(t *testing.T) {
reservation := Reservation{
Version: "20260731-build",
Length: 10,
Offset: 0,
Reserved: 5,
LeaseToken: "lease",
ReceiptID: "receipt",
}
tests := []struct {
raw int64
want error
}{
{raw: -1, want: ErrVersionChanged},
{raw: -2, want: ErrQueueUnhealthy},
{raw: -3, want: ErrReservationExpired},
{raw: -4, want: ErrReservationConflict},
{raw: -5, want: ErrReservationConflict},
}
for _, tt := range tests {
client := &fakeScriptClient{result: tt.raw}
err := commitReservationContext(
context.Background(), client, 123, reservation, 3,
)
if !errors.Is(err, tt.want) {
t.Fatalf("raw=%d error=%v want=%v", tt.raw, err, tt.want)
}
}
}
func TestAbortReservationContextIsTokenScoped(t *testing.T) {
client := &fakeScriptClient{result: int64(1)}
reservation := Reservation{
Version: "20260731-build",
LeaseToken: "lease",
ReceiptID: "receipt",
}
if err := abortReservationContext(
context.Background(), client, 123, reservation,
); err != nil {
t.Fatal(err)
}
if client.script != abortReservationRedisScript ||
!reflect.DeepEqual(client.args, []interface{}{
"recommend:short:reservation:",
"20260731-build",
"123",
"lease",
"receipt",
}) {
t.Fatalf("abort script=%p args=%#v", client.script, client.args)
}
if !strings.Contains(abortReservationScriptSource, "lease[1] ~= ARGV[4]") ||
!strings.Contains(abortReservationScriptSource, "lease[2] ~= ARGV[5]") {
t.Fatal("abort must not delete a newer lease after TTL takeover")
}
}
func TestReservationAPIsRejectInvalidInputsBeforeRedis(t *testing.T) {
client := &fakeScriptClient{}
if _, err := reserveContext(
context.Background(), client, 0, 20, "", "",
); err == nil {
t.Fatal("reserve should reject uid 0")
}
invalid := Reservation{
Version: "20260731",
Length: 10,
Offset: 0,
Reserved: 2,
LeaseToken: "lease",
ReceiptID: "receipt",
}
if err := commitReservationContext(
context.Background(), client, 1, invalid, 3,
); err == nil {
t.Fatal("commit should reject consumed > reserved")
}
if client.calls != 0 {
t.Fatalf("Redis calls = %d, want 0", client.calls)
}
}
func TestHealthContextParsesAtomicHealthResult(t *testing.T) {
client := &fakeScriptClient{
result: []interface{}{"OK", "20260731", "154290", "154290", "7200000", "7199999"},
}
health, err := healthContext(context.Background(), client, "20260731")
if err != nil {
t.Fatalf("HealthContext() error = %v", err)
}
if !health.Healthy || health.Status != "OK" || health.Version != "20260731" ||
health.QueueLength != 154290 || health.MetadataLength != 154290 ||
health.QueueTTL != 2*time.Hour ||
health.MetadataTTL != 2*time.Hour-time.Millisecond {
t.Fatalf("HealthContext() = %+v", health)
}
if client.script != healthRedisScript || client.args[0] != "20260731" {
t.Fatal("HealthContext() did not use the expected cached script/version")
}
}
func TestHealthContextReturnsUnhealthyStatusWithoutRedisError(t *testing.T) {
client := &fakeScriptClient{
result: []interface{}{"LENGTH_MISMATCH", "20260731", "20", "19", "1000", "1000"},
}
health, err := healthContext(context.Background(), client, "20260731")
if err != nil {
t.Fatalf("HealthContext() error = %v", err)
}
if health.Healthy || health.Status != "LENGTH_MISMATCH" {
t.Fatalf("HealthContext() = %+v", health)
}
}
func TestTakeAndHealthRejectMalformedScriptResults(t *testing.T) {
takeClient := &fakeScriptClient{
result: []interface{}{"20260731", "not-a-length", "0"},
}
if _, err := takeContext(
context.Background(), takeClient, 1, 20, "", "",
); err == nil {
t.Fatal("takeContext() should reject malformed length")
}
healthClient := &fakeScriptClient{
result: []interface{}{"OK", "20260731", "20", "20", "not-a-ttl", "1000"},
}
if _, err := healthContext(context.Background(), healthClient, "20260731"); err == nil {
t.Fatal("HealthContext() should reject malformed TTL")
}
}
func TestSetKeyTTLHoursValidatesAndPublishesConfiguredTTL(t *testing.T) {
originalHours := int(KeyTTL() / time.Hour)
defer func() {
if err := SetKeyTTLHours(originalHours); err != nil {
t.Fatalf("restore key TTL: %v", err)
}
}()
for _, hours := range []int{minKeyTTLHours - 1, maxKeyTTLHours + 1} {
if err := SetKeyTTLHours(hours); err == nil {
t.Fatalf("SetKeyTTLHours(%d) should fail", hours)
}
}
const configuredHours = 96
if err := SetKeyTTLHours(configuredHours); err != nil {
t.Fatalf("SetKeyTTLHours() error = %v", err)
}
client := &fakePublishClient{}
if err := publish(
context.Background(), client, publishTestMeta(1), []string{"video-1"}, "lock", "token",
); err != nil {
t.Fatalf("publish() error = %v", err)
}
if client.publishedTTL != int64(96*time.Hour/time.Second) {
t.Fatalf("published ttl = %d, want %d", client.publishedTTL, int64(96*time.Hour/time.Second))
}
}
func TestPublishRejectsVersionMismatchBeforeRedis(t *testing.T) {
meta := QueueMeta{
Version: "20260730",
GeneratedAt: time.Date(2026, 7, 31, 0, 0, 0, 0, time.UTC),
}
err := Publish(context.Background(), nil, meta, []string{"video-id"}, "lock", "token")
if err == nil {
t.Fatal("Publish() should reject a version that does not match generatedAt")
}
}
func TestVersionWithRevisionKeepsDateAndSeparatesSameDayBuilds(t *testing.T) {
generatedAt := time.Date(2026, 7, 31, 2, 0, 0, 0, time.UTC)
first := VersionWithRevision(generatedAt, "build-a")
second := VersionWithRevision(generatedAt, "build-b")
if first != "20260731-build-a" || second != "20260731-build-b" {
t.Fatalf("versions = %q, %q", first, second)
}
if first == second {
t.Fatal("same-day builds must use different queue versions")
}
for _, version := range []string{"20260731", first, second} {
if !versionMatchesDate(version, "20260731") {
t.Fatalf("version %q should belong to date 20260731", version)
}
}
for _, version := range []string{
"20260730-build-a",
"202607310-build-a",
"20260731-",
} {
if versionMatchesDate(version, "20260731") {
t.Fatalf("version %q must not belong to date 20260731", version)
}
}
}
func TestPublishAcceptsSameDayRevisionVersion(t *testing.T) {
meta := publishTestMeta(1)
meta.Version = VersionWithRevision(meta.GeneratedAt, "build-a")
client := &fakePublishClient{}
if err := publish(
context.Background(), client, meta, []string{"video-1"}, "lock", "token",
); err != nil {
t.Fatalf("publish() error = %v", err)
}
if !client.swapped || client.swapArgs[0] != meta.Version {
t.Fatalf("same-day revision version was not published: %#v", client.swapArgs)
}
}
func TestHealthScriptAcceptsPureDateAndSameDayRevisionOnly(t *testing.T) {
for _, want := range []string{
"version == expectedDate",
"string.len(version) > string.len(expectedDate) + 1",
"expectedDate .. '-'",
} {
if !strings.Contains(healthScriptSource, want) {
t.Fatalf("health script missing date-family check %q", want)
}
}
}
func TestPublishUsesBoundedPipelineAndPreservesOrder(t *testing.T) {
total := queuePushBatchSize + queuePushBatchSize*queuePipelineBatchCount + 57
ids := make([]string, total)
for i := range ids {
ids[i] = fmt.Sprintf("video-%06d", i)
}
meta := publishTestMeta(total)
client := &fakePublishClient{}
if err := publish(context.Background(), client, meta, ids, "lock", "token"); err != nil {
t.Fatalf("publish() error = %v", err)
}
if !client.swapped || client.swapAttempts != 1 {
t.Fatalf("swap state = %v, attempts = %d, want successful single swap", client.swapped, client.swapAttempts)
}
if !reflect.DeepEqual(client.queue, ids) {
t.Fatal("published queue order differs from input")
}
wantWindows := []int{queuePushBatchSize * queuePipelineBatchCount, 57}
if !reflect.DeepEqual(client.pipelineWindowSizes, wantWindows) {
t.Fatalf("pipeline windows = %v, want %v", client.pipelineWindowSizes, wantWindows)
}
for _, batchSize := range client.pipelineBatchSizes {
if batchSize != queuePushBatchSize {
t.Fatalf("pipeline batch size = %d, want %d", batchSize, queuePushBatchSize)
}
}
if len(client.pipelineExpirations) != len(wantWindows) {
t.Fatalf("pipeline expirations = %v, want one per window", client.pipelineExpirations)
}
for _, expiration := range client.pipelineExpirations {
if expiration != buildingKeyTTL {
t.Fatalf("pipeline expiration = %v, want %v", expiration, buildingKeyTTL)
}
}
if client.firstPushTTL != int64(buildingKeyTTL/time.Second) {
t.Fatalf("building ttl = %d, want %d", client.firstPushTTL, int64(buildingKeyTTL/time.Second))
}
if client.metaBuildingTTL != int64(buildingKeyTTL/time.Second) {
t.Fatalf("meta building ttl = %d, want %d", client.metaBuildingTTL, int64(buildingKeyTTL/time.Second))
}
if client.publishedTTL != int64(KeyTTL()/time.Second) {
t.Fatalf("published ttl = %d, want %d", client.publishedTTL, int64(KeyTTL()/time.Second))
}
if client.metaLength != total {
t.Fatalf("meta length = %d, want %d", client.metaLength, total)
}
if !strings.Contains(client.swapScript, "redis.call('UNLINK'") {
t.Fatal("swap script must asynchronously remove current and previous offset hashes")
}
if !strings.Contains(client.swapScript, "redis.call('PEXPIREAT'") {
t.Fatal("swap script must set one absolute expiration for queue and metadata")
}
if len(client.swapArgs) != 6 || client.swapArgs[5] != "recommend:short:offset:" {
t.Fatalf("swap args = %#v, want previous offset prefix", client.swapArgs)
}
}
func TestPublishPipelineFailureCleansBuildingQueueWithoutSwap(t *testing.T) {
ids := make([]string, queuePushBatchSize+101)
for i := range ids {
ids[i] = fmt.Sprintf("video-%d", i)
}
client := &fakePublishClient{pipelineErrorAt: 1}
err := publish(context.Background(), client, publishTestMeta(len(ids)), ids, "lock", "token")
if err == nil || !strings.Contains(err.Error(), "injected pipeline error") {
t.Fatalf("publish() error = %v, want injected pipeline error", err)
}
if client.swapped || client.swapAttempts != 0 {
t.Fatalf("swap state = %v, attempts = %d, want no swap", client.swapped, client.swapAttempts)
}
if client.deleteCalls == 0 || client.queueLengthBeforeDelete <= queuePushBatchSize {
t.Fatalf(
"cleanup calls = %d, pre-delete length = %d, want partial queue cleanup",
client.deleteCalls, client.queueLengthBeforeDelete,
)
}
}
func TestPublishRejectsPipelineDuplicateBeforeSwap(t *testing.T) {
ids := make([]string, queuePushBatchSize+10)
for i := range ids {
ids[i] = fmt.Sprintf("video-%d", i)
}
client := &fakePublishClient{pipelineDuplicateAt: 1}
err := publish(context.Background(), client, publishTestMeta(len(ids)), ids, "lock", "token")
if err == nil || !strings.Contains(err.Error(), "queue length mismatch after pipeline") {
t.Fatalf("publish() error = %v, want pipeline length mismatch", err)
}
if client.swapped || client.swapAttempts != 0 {
t.Fatalf("swap state = %v, attempts = %d, want no swap", client.swapped, client.swapAttempts)
}
if client.deleteCalls == 0 {
t.Fatal("duplicate pipeline queue was not cleaned")
}
}
func TestPublishFinalLengthGateRejectsFirstPushDuplicate(t *testing.T) {
ids := []string{"video-1", "video-2"}
client := &fakePublishClient{duplicateFirstPush: true}
err := publish(context.Background(), client, publishTestMeta(len(ids)), ids, "lock", "token")
if err == nil || !strings.Contains(err.Error(), "queue length mismatch before queue swap") {
t.Fatalf("publish() error = %v, want final queue length mismatch", err)
}
if client.swapped || client.swapAttempts != 1 {
t.Fatalf("swap state = %v, attempts = %d, want rejected swap", client.swapped, client.swapAttempts)
}
if client.deleteCalls == 0 {
t.Fatal("invalid first push queue was not cleaned")
}
}
func TestPublishCancellationStopsBeforeNextPipelineWindow(t *testing.T) {
total := queuePushBatchSize + queuePushBatchSize*queuePipelineBatchCount + 1
ids := make([]string, total)
for i := range ids {
ids[i] = fmt.Sprintf("video-%d", i)
}
ctx, cancel := context.WithCancel(context.Background())
client := &fakePublishClient{cancelAfterPipelineAt: 1, cancel: cancel}
err := publish(ctx, client, publishTestMeta(len(ids)), ids, "lock", "token")
if !errors.Is(err, context.Canceled) {
t.Fatalf("publish() error = %v, want context.Canceled", err)
}
if client.pipelineCalls != 1 {
t.Fatalf("pipeline calls = %d, want 1", client.pipelineCalls)
}
if client.swapped || client.swapAttempts != 0 {
t.Fatalf("swap state = %v, attempts = %d, want no swap", client.swapped, client.swapAttempts)
}
if client.deleteCalls == 0 {
t.Fatal("cancelled building queue was not cleaned")
}
}
func TestPublishRejectsMetadataLengthMismatchBeforeRedis(t *testing.T) {
meta := publishTestMeta(2)
meta.Length = 1
err := publish(context.Background(), nil, meta, []string{"video-1", "video-2"}, "lock", "token")
if err == nil || !strings.Contains(err.Error(), "metadata length") {
t.Fatalf("publish() error = %v, want metadata length mismatch", err)
}
}
func publishTestMeta(length int) QueueMeta {
generatedAt := time.Date(2026, 7, 31, 2, 0, 0, 0, time.UTC)
return QueueMeta{
Version: VersionAt(generatedAt),
Length: length,
GeneratedAt: generatedAt,
StartOffset: 0,
}
}
func TestVersionValidUntilUsesCSTDayBoundary(t *testing.T) {
meta := QueueMeta{
Version: "20260731",
GeneratedAt: time.Date(2026, 7, 31, 15, 59, 59, 0, time.UTC),
}
want := time.Date(2026, 7, 31, 16, 0, 0, 0, time.UTC)
if got := versionValidUntil(meta); !got.Equal(want) {
t.Fatalf("versionValidUntil() = %s, want %s", got, want)
}
}
func TestAssembleConcurrentDeterministic(t *testing.T) {
now := time.Date(2026, 7, 24, 8, 0, 0, 0, time.UTC)
input := make([]vidmod.RecommendCandidate, 0, 100)
for i := 0; i < 100; i++ {
reviewAt := now.Add(-48 * time.Hour)
if i >= 90 {
reviewAt = now.Add(-time.Duration(i-90) * time.Hour)
}
input = append(input, candidate(i, int64(1000-i), reviewAt))
}
want := fmt.Sprint(Assemble(input, now))
var wg sync.WaitGroup
errs := make(chan string, 64)
for i := 0; i < 64; i++ {
wg.Add(1)
go func() {
defer wg.Done()
if got := fmt.Sprint(Assemble(input, now)); got != want {
errs <- got
}
}()
}
wg.Wait()
close(errs)
if got, ok := <-errs; ok {
t.Fatalf("concurrent result differs: %s", got)
}
}
func assertUnique(t *testing.T, ids []string) {
t.Helper()
seen := map[string]bool{}
for _, id := range ids {
if seen[id] {
t.Fatalf("duplicate id %s", id)
}
seen[id] = true
}
}