broad-announce/internal/dedupe/dedupe_test.go

package dedupe

import (
	"context"
	"testing"
	"time"

	"github.com/redis/go-redis/v9"
	"git3.techno-world.net/lrosales/broad-announce/internal/store"
)

// These tests need a real Redis. They use a short window and a
// test-prefix so they don't collide with prod data.
//
// In CI: run with `go test -tags=redis ./...` (the integration
// build tag) and ensure Redis is on localhost:6379.
//
// In the M0 docker-compose, `make test-redis` runs the same suite
// against the dev container.

func newRedis(t *testing.T) *store.Redis {
	t.Helper()
	r, err := store.ConnectRedis(context.Background(), "redis://localhost:6379/0")
	if err != nil {
		t.Skipf("redis not available, skipping: %v", err)
	}
	return r
}

func TestCheck_FirstAndDuplicate(t *testing.T) {
	r := newRedis(t)
	d := New(r.Client, 2*time.Second)

	isNew, n, err := d.Check(context.Background(), "src-A", "key-1")
	if err != nil {
		t.Fatal(err)
	}
	if !isNew || n != 1 {
		t.Fatalf("first: want new/1, got new=%v n=%d", isNew, n)
	}
	isNew, n, err = d.Check(context.Background(), "src-A", "key-1")
	if err != nil {
		t.Fatal(err)
	}
	if isNew || n != 2 {
		t.Fatalf("second: want dup/2, got new=%v n=%d", isNew, n)
	}
}

func TestCheck_DifferentSourcesDoNotCollide(t *testing.T) {
	r := newRedis(t)
	d := New(r.Client, 2*time.Second)

	if isNew, _, _ := d.Check(context.Background(), "src-A", "shared"); !isNew {
		t.Fatal("src-A first should be new")
	}
	if isNew, _, _ := d.Check(context.Background(), "src-B", "shared"); !isNew {
		t.Fatal("src-B first should be new (different source)")
	}
}

func TestCheck_EmptyDedupeKeyIsNeverDeduped(t *testing.T) {
	r := newRedis(t)
	d := New(r.Client, 2*time.Second)

	for i := 0; i < 5; i++ {
		isNew, n, err := d.Check(context.Background(), "src-A", "")
		if err != nil {
			t.Fatal(err)
		}
		if !isNew || n != 1 {
			t.Fatalf("empty dedupe_key should always be new, got new=%v n=%d", isNew, n)
		}
	}
}

func TestCheck_WindowExpires(t *testing.T) {
	r := newRedis(t)
	d := New(r.Client, 500*time.Millisecond)
	// Use a unique key per run so we don't see state from a
	// previous test that used a longer default window.
	key := "expiring-" + time.Now().Format("150405.000000000")

	if isNew, _, _ := d.Check(context.Background(), "src-A", key); !isNew {
		t.Fatal("first should be new")
	}
	// The Lua script rounds sub-second windows up to 1s via
	// math.ceil, so we need to wait > 1s for the TTL to
	// actually elapse.
	time.Sleep(1200 * time.Millisecond)
	isNew, n, err := d.Check(context.Background(), "src-A", key)
	if err != nil {
		t.Fatal(err)
	}
	if !isNew || n != 1 {
		t.Fatalf("after window: want new/1, got new=%v n=%d", isNew, n)
	}
}

// M6: sliding window keeps the key alive when duplicates keep
// arriving. With a 500ms window, if a duplicate hits at 300ms
// (within the window) the TTL is refreshed from that point,
// and a third hit at 700ms is still a duplicate (window was
// pushed out to 1.2s by the second hit). This is the M6
// contract that lets operators see (×N) for arbitrarily long
// alert storms.
func TestCheck_SlidingWindowKeepsAlive(t *testing.T) {
	r := newRedis(t)
	d := New(r.Client, 500*time.Millisecond)
	key := "storm-" + time.Now().Format("150405.000000000")

	if isNew, _, _ := d.Check(context.Background(), "src-A", key); !isNew {
		t.Fatal("first should be new")
	}
	// Three duplicates, 200ms apart — every one within the
	// 500ms window from the *previous* observation.
	for i := 0; i < 3; i++ {
		time.Sleep(200 * time.Millisecond)
		isNew, n, err := d.Check(context.Background(), "src-A", key)
		if err != nil {
			t.Fatal(err)
		}
		if isNew {
			t.Fatalf("dupe %d should be dup, got new", i+1)
		}
		want := uint32(i + 2) // 2, 3, 4
		if n != want {
			t.Fatalf("dupe %d: want n=%d, got n=%d", i+1, want, n)
		}
	}
	// After the stream stops, wait for the window to expire
	// and the next arrival should be a fresh "new".
	// The Lua script rounds sub-second windows up to 1s.
	time.Sleep(1300 * time.Millisecond)
	isNew, n, err := d.Check(context.Background(), "src-A", key)
	if err != nil {
		t.Fatal(err)
	}
	if !isNew || n != 1 {
		t.Fatalf("after stream ends: want new/1, got new=%v n=%d", isNew, n)
	}
}

// M6: a high-volume burst (1000 hits) all share the same
// key. The first is new, the next 999 are dupes with
// monotonically increasing counts. The TTL refreshes on
// every hit, so the key stays alive throughout.
func TestCheck_BurstOfThousand(t *testing.T) {
	r := newRedis(t)
	d := New(r.Client, 2*time.Second)
	key := "burst-" + time.Now().Format("150405.000000000")

	isNew, n, _ := d.Check(context.Background(), "src-A", key)
	if !isNew || n != 1 {
		t.Fatalf("first: want new/1, got new=%v n=%d", isNew, n)
	}
	for i := 2; i <= 1000; i++ {
		isNew, n, err := d.Check(context.Background(), "src-A", key)
		if err != nil {
			t.Fatalf("hit %d err: %v", i, err)
		}
		if isNew {
			t.Fatalf("hit %d: want dup, got new", i)
		}
		if n != uint32(i) {
			t.Fatalf("hit %d: want n=%d, got n=%d", i, i, n)
		}
	}
}

// Sanity: ensure the package compiles when redis isn't around.
var _ = redis.Nil