broad-announce/internal/dedupe/collapser_test.go

package dedupe

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

	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
)

// flushMs used by most tests; small enough to be fast.
const testFlushMs = 50

// mkAlert is a small constructor so the tests stay readable.
func mkAlert(severity, sourceID, key string, count uint32) alert.Alert {
	return alert.Alert{
		ID:          "test-" + sourceID + "-" + key,
		CompanyID:   "co",
		SourceID:    sourceID,
		Severity:    alert.Severity(severity),
		Category:    "test",
		Title:       "t",
		Body:        "b",
		DedupeKey:   key,
		DedupeCount: count,
		ReceivedAt:  time.Now(),
	}
}

// TestObserve_Passthrough_EmptyKey: an empty dedupe_key is
// delivered as Passthrough, the collapser holds nothing.
func TestObserve_Passthrough_EmptyKey(t *testing.T) {
	var pass int32
	c := NewCollapser(testFlushMs*time.Millisecond, func(_, _ string, _ alert.Alert) {
		t.Fatal("onFlush should not be called for empty key")
	})
	c.OnPassthrough = func() { atomic.AddInt32(&pass, 1) }

	for i := 0; i < 5; i++ {
		a := mkAlert("warning", "prom-prod", "", 1)
		if d := c.Observe("prom-prod", "", a); d != Passthrough {
			t.Fatalf("expected Passthrough, got %v", d)
		}
	}
	if got := atomic.LoadInt32(&pass); got != 5 {
		t.Errorf("OnPassthrough fired %d times, want 5", got)
	}
	if c.Pending() != 0 {
		t.Errorf("Pending() = %d, want 0", c.Pending())
	}
}

// TestObserve_Collapse_FirstThenFlush: 100 observes with the
// same (source, key) collapse into 1 flush, dedupe_count=100.
func TestObserve_Collapse_FirstThenFlush(t *testing.T) {
	var (
		mu       sync.Mutex
		flushed  []alert.Alert
		flushSrc []string
		flushKey []string
	)
	c := NewCollapser(testFlushMs*time.Millisecond, func(src, key string, a alert.Alert) {
		mu.Lock()
		flushed = append(flushed, a)
		flushSrc = append(flushSrc, src)
		flushKey = append(flushKey, key)
		mu.Unlock()
	})

	for i := uint32(1); i <= 100; i++ {
		a := mkAlert("warning", "prom-prod", "burst-1", i)
		var d Decision
		if i == 1 {
			d = c.Observe("prom-prod", "burst-1", a)
			if d != CollapseNew {
				t.Fatalf("first Observe: got %v, want CollapseNew", d)
			}
		} else {
			d = c.Observe("prom-prod", "burst-1", a)
			if d != CollapseDupe {
				t.Fatalf("Observe #%d: got %v, want CollapseDupe", i, d)
			}
		}
	}
	if c.Pending() != 1 {
		t.Errorf("Pending() during burst = %d, want 1", c.Pending())
	}

	runCtx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { c.Run(runCtx); close(done) }()

	// Wait up to 1s for the flush. With testFlushMs=50, it
	// should fire on the next tick (≤25ms).
	deadline := time.Now().Add(1 * time.Second)
	for time.Now().Before(deadline) {
		mu.Lock()
		n := len(flushed)
		mu.Unlock()
		if n > 0 {
			break
		}
		time.Sleep(2 * time.Millisecond)
	}

	cancel()
	<-done
	c.FlushAll()

	mu.Lock()
	defer mu.Unlock()
	if len(flushed) != 1 {
		t.Fatalf("flushed %d alerts, want 1", len(flushed))
	}
	if flushed[0].DedupeCount != 100 {
		t.Errorf("flushed dedupe_count = %d, want 100", flushed[0].DedupeCount)
	}
	if flushSrc[0] != "prom-prod" || flushKey[0] != "burst-1" {
		t.Errorf("flushed (%q, %q), want (prom-prod, burst-1)", flushSrc[0], flushKey[0])
	}
}

// TestObserve_PerSourceIsolation: same key from two sources
// produces two flushes.
func TestObserve_PerSourceIsolation(t *testing.T) {
	var mu sync.Mutex
	flushed := make(map[string]alert.Alert) // key = "source|dedupeKey"
	c := NewCollapser(testFlushMs*time.Millisecond, func(src, key string, a alert.Alert) {
		mu.Lock()
		flushed[src+"|"+key] = a
		mu.Unlock()
	})

	for i := uint32(1); i <= 5; i++ {
		c.Observe("prom-prod", "shared", mkAlert("warning", "prom-prod", "shared", i))
		c.Observe("grafana", "shared", mkAlert("warning", "grafana", "shared", i))
	}
	if c.Pending() != 2 {
		t.Errorf("Pending() = %d, want 2", c.Pending())
	}

	runCtx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { c.Run(runCtx); close(done) }()

	// Wait for 2 flushes.
	deadline := time.Now().Add(1 * time.Second)
	for time.Now().Before(deadline) {
		mu.Lock()
		n := len(flushed)
		mu.Unlock()
		if n == 2 {
			break
		}
		time.Sleep(2 * time.Millisecond)
	}
	cancel()
	<-done
	c.FlushAll()

	mu.Lock()
	defer mu.Unlock()
	if len(flushed) != 2 {
		t.Fatalf("flushed %d, want 2", len(flushed))
	}
	if flushed["prom-prod|shared"].DedupeCount != 5 {
		t.Errorf("prom-prod count = %d, want 5", flushed["prom-prod|shared"].DedupeCount)
	}
	if flushed["grafana|shared"].DedupeCount != 5 {
		t.Errorf("grafana count = %d, want 5", flushed["grafana|shared"].DedupeCount)
	}
}

// TestRun_MaxWaitReFlush: a continuous stream re-flushes every
// max-wait, each flush carrying the running count.
func TestRun_MaxWaitReFlush(t *testing.T) {
	flushMs := 30
	var (
		mu      sync.Mutex
		flushed []uint32
	)
	c := NewCollapser(time.Duration(flushMs)*time.Millisecond, func(_, _ string, a alert.Alert) {
		mu.Lock()
		flushed = append(flushed, a.DedupeCount)
		mu.Unlock()
	})

	runCtx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { c.Run(runCtx); close(done) }()

	// Send 90 alerts over ~100ms (1ms each). With flushMs=30ms
	// the first flush should fire around 30ms in. After the
	// flush, the (source, key) is removed from pending, so the
	// next Observe starts a new collapse window. Over 100ms we
	// should see 3+ flushes.
	for i := uint32(1); i <= 90; i++ {
		c.Observe("prom-prod", "continuous", mkAlert("warning", "prom-prod", "continuous", i))
		time.Sleep(1 * time.Millisecond)
	}

	// Wait a tick for any final flush.
	time.Sleep(2 * time.Duration(flushMs) * time.Millisecond)
	cancel()
	<-done
	c.FlushAll()

	mu.Lock()
	defer mu.Unlock()
	if len(flushed) < 2 {
		t.Errorf("got %d flushes, want ≥2 (continuous burst should re-flush)", len(flushed))
	}
	// Each flush count must be ≥1 and ≤90.
	for i, n := range flushed {
		if n < 1 || n > 90 {
			t.Errorf("flushed[%d] = %d, want 1..90", i, n)
		}
	}
	// Counts should be monotonically non-decreasing per window
	// (each new collapse window starts fresh; counts are local
	// to that window, so this is a weak assertion — just check
	// the last is > 0).
	if flushed[len(flushed)-1] == 0 {
		t.Errorf("final flush had count 0")
	}
}

// TestObserve_EmptyKeyDoesNotBlockRealKey: an empty-key alert
// during a non-empty-key burst doesn't reset the burst's timer.
func TestObserve_EmptyKeyDoesNotBlockRealKey(t *testing.T) {
	var mu sync.Mutex
	flushed := 0
	c := NewCollapser(testFlushMs*time.Millisecond, func(_, _ string, _ alert.Alert) {
		mu.Lock()
		flushed++
		mu.Unlock()
	})

	// Real key burst.
	c.Observe("prom-prod", "real", mkAlert("warning", "prom-prod", "real", 1))
	// Empty keys interspersed.
	for i := 0; i < 5; i++ {
		c.Observe("prom-prod", "", mkAlert("warning", "prom-prod", "", 1))
	}
	if c.Pending() != 1 {
		t.Errorf("Pending() = %d, want 1", c.Pending())
	}

	runCtx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { c.Run(runCtx); close(done) }()
	time.Sleep(2 * testFlushMs * time.Millisecond)
	cancel()
	<-done
	c.FlushAll()

	mu.Lock()
	defer mu.Unlock()
	if flushed != 1 {
		t.Errorf("flushed = %d, want 1 (empty keys must not interfere)", flushed)
	}
}

// TestObserve_ConcurrentSameKey: 8 goroutines all hitting the
// same (source, key) — final dedupe_count is exactly the max
// observed.
func TestObserve_ConcurrentSameKey(t *testing.T) {
	var mu sync.Mutex
	flushed := alert.Alert{}
	c := NewCollapser(testFlushMs*time.Millisecond, func(_, _ string, a alert.Alert) {
		mu.Lock()
		flushed = a
		mu.Unlock()
	})

	const goroutines = 8
	const perG = 500
	var wg sync.WaitGroup
	wg.Add(goroutines)
	counter := uint32(0)
	for g := 0; g < goroutines; g++ {
		go func() {
			defer wg.Done()
			for i := 0; i < perG; i++ {
				n := atomic.AddUint32(&counter, 1)
				c.Observe("prom-prod", "concurrent", mkAlert("warning", "prom-prod", "concurrent", n))
			}
		}()
	}
	wg.Wait()

	runCtx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { c.Run(runCtx); close(done) }()

	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		mu.Lock()
		ok := flushed.DedupeCount > 0
		mu.Unlock()
		if ok {
			break
		}
		time.Sleep(2 * time.Millisecond)
	}
	cancel()
	<-done
	c.FlushAll()

	mu.Lock()
	defer mu.Unlock()
	want := uint32(goroutines * perG)
	if flushed.DedupeCount != want {
		t.Errorf("flushed dedupe_count = %d, want %d", flushed.DedupeCount, want)
	}
}

// TestFlushAll_DrainsPending: FlushAll publishes every held
// entry immediately, regardless of timer state.
func TestFlushAll_DrainsPending(t *testing.T) {
	var mu sync.Mutex
	flushed := []string{}
	c := NewCollapser(10*time.Second, func(src, key string, _ alert.Alert) {
		mu.Lock()
		flushed = append(flushed, src+"|"+key)
		mu.Unlock()
	})
	c.Observe("s1", "k1", mkAlert("warning", "s1", "k1", 1))
	c.Observe("s2", "k2", mkAlert("warning", "s2", "k2", 1))
	if c.Pending() != 2 {
		t.Fatalf("Pending() = %d, want 2", c.Pending())
	}
	c.FlushAll()
	if c.Pending() != 0 {
		t.Errorf("Pending() after FlushAll = %d, want 0", c.Pending())
	}
	mu.Lock()
	defer mu.Unlock()
	if len(flushed) != 2 {
		t.Errorf("flushed %d, want 2", len(flushed))
	}
}