broad-announce/internal/dedupe/collapser.go

// Package dedupe provides the M6.5 router-level dedupe Collapser.
//
// The Collapser is a per-source debounce with a max-wait. It lets
// the router turn a burst of 100 identical alerts into 1 delivery,
// with the final dedupe_count attached to that single message.
//
// Behavior:
//
//   - Observe(sourceID, dedupeKey, alert) returns one of:
//       * Collapse: the alert is being held; the router should NOT
//         publish deliveries yet. The Collapser will publish a
//         single alert via the onFlush callback when the max-wait
//         elapses, with the latest dedupe_count.
//       * Passthrough: the dedupe_key was empty. The router should
//         publish deliveries immediately, same as the M2 path.
//
//   - When Observe is called with a (source, key) we already hold:
//       * Stored alert's dedupe_count is updated to the max of the
//         stored value and the incoming value (defensive — Redis
//         is the canonical counter).
//       * The flush timer is reset to the full max-wait.
//
//   - The flush loop wakes every `flushMs / 4` (or 100ms, whichever
//     is smaller) and publishes any pending entries where
//     firstSeen + flushMs <= now.
//
//   - State is per (sourceID, dedupeKey). Two sources with the same
//     key are isolated (matches M6's per-source dedupe state).
//
// Concurrency:
//
//   - All public methods are safe to call from multiple goroutines.
//   - The pending map is guarded by a single mutex. The collision
//     rate in production is bounded (a handful of dedupe keys
//     active at once per source), so a coarse mutex is fine.
//   - The flush loop runs in its own goroutine via Run(ctx).
package dedupe

import (
	"context"
	"sync"
	"time"

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

// Decision is the result of an Observe call.
type Decision int

const (
	// Passthrough means the alert had no dedupe_key and should
	// be delivered immediately, unchanged.
	Passthrough Decision = iota
	// CollapseNew is the first arrival of a (source, key)
	// pair. The router should do any expensive per-alert work
	// (e.g. resolve recipients) NOW and cache it. The Collapser
	// will publish a single alert via the onFlush callback
	// when the max-wait elapses.
	CollapseNew
	// CollapseDupe is a subsequent arrival of the same
	// (source, key) within the active window. The router
	// should NOT redo expensive work; the stored alert's
	// dedupe_count has already been updated in-place.
	CollapseDupe
)

// FlushFunc is called when a pending alert is ready to publish.
// The Collapser hands back the final dedupe_count it observed for
// that (source, key) pair. The router is responsible for the
// fanout to deliveries.<channel>.<company_id> subjects.
type FlushFunc func(sourceID, dedupeKey string, a alert.Alert)

// pendingEntry is one in-flight collapse.
type pendingEntry struct {
	sourceID  string
	key       string
	alert     alert.Alert
	firstSeen time.Time
	lastSeen  time.Time
}

// Collapser is a per-(source, key) debounce with max-wait.
// Zero-value is NOT usable; use NewCollapser.
type Collapser struct {
	flushMs    time.Duration
	tickPeriod time.Duration
	onFlush    FlushFunc

	mu      sync.Mutex
	pending map[collapseKey]*pendingEntry

	// OnPassthrough fires when an alert with empty dedupe_key
	// passes through. Useful for tests/observability; nil OK.
	OnPassthrough func()

	// onCollapse fires when a new (source, key) collapse starts.
	// Useful for tests/observability; nil OK.
	onCollapse func()
}

// collapseKey is the dedupe state key. We use a single struct
// value to avoid string allocation on every Observe.
type collapseKey struct {
	sourceID  string
	dedupeKey string
}

// NewCollapser constructs a Collapser that flushes after `flushMs`.
//
//   - flushMs > 0: required, the max time a held alert waits
//     before being flushed (debounce window).
//   - onFlush: required, called for each pending entry at flush
//     time. Must not block; the flush loop runs in its own
//     goroutine and onFlush serializes with itself via the loop.
func NewCollapser(flushMs time.Duration, onFlush FlushFunc) *Collapser {
	if flushMs <= 0 {
		flushMs = 2000 * time.Millisecond
	}
	// Tick at most every 100ms, or 1/4 of the flush window
	// (whichever is smaller). 100ms is a good upper bound for
	// responsiveness when flushMs is large.
	tick := flushMs / 4
	if tick > 100*time.Millisecond {
		tick = 100 * time.Millisecond
	}
	if tick < 5*time.Millisecond {
		tick = 5 * time.Millisecond
	}
	return &Collapser{
		flushMs:    flushMs,
		tickPeriod: tick,
		onFlush:    onFlush,
		pending:    make(map[collapseKey]*pendingEntry),
	}
}

// Observe records an alert and returns the action the router
// should take.
//
//   - If dedupeKey is empty: returns Passthrough. The router
//     delivers immediately (no collapse, no state).
//   - If (source, key) is new: stores the alert, arms the
//     max-wait timer, returns CollapseNew. The router should
//     resolve recipients NOW.
//   - If (source, key) is held: updates the stored
//     dedupe_count to max(stored, incoming), updates
//     lastSeen (timer is reset at flush time), returns
//     CollapseDupe. The router should NOT re-resolve.
func (c *Collapser) Observe(sourceID, dedupeKey string, a alert.Alert) Decision {
	if dedupeKey == "" {
		if c.OnPassthrough != nil {
			c.OnPassthrough()
		}
		return Passthrough
	}
	k := collapseKey{sourceID: sourceID, dedupeKey: dedupeKey}
	c.mu.Lock()
	defer c.mu.Unlock()
	if e, ok := c.pending[k]; ok {
		if a.DedupeCount > e.alert.DedupeCount {
			e.alert.DedupeCount = a.DedupeCount
		}
		// Refresh lastSeen so the flush window slides for
		// continuous bursts (matches M6 sliding-window TTL
		// semantics for the in-router collapse).
		e.lastSeen = time.Now()
		return CollapseDupe
	}
	c.pending[k] = &pendingEntry{
		sourceID:  sourceID,
		key:       dedupeKey,
		alert:     a,
		firstSeen: time.Now(),
		lastSeen:  time.Now(),
	}
	if c.onCollapse != nil {
		c.onCollapse()
	}
	return CollapseNew
}

// Pending returns the number of held entries. Used by tests
// and the routerd /metrics endpoint.
func (c *Collapser) Pending() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return len(c.pending)
}

// Run is the flush loop. It exits when ctx is canceled. Call
// it in a goroutine:
//
//	go collapser.Run(ctx)
func (c *Collapser) Run(ctx context.Context) {
	t := time.NewTicker(c.tickPeriod)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case now := <-t.C:
			c.flushReady(now)
		}
	}
}

// flushReady publishes any entries whose firstSeen+flushMs <= now.
// The sliding window is the simple "first arrival" version: once
// the burst has been quiet for flushMs, flush. (The Observe path
// updates lastSeen for metric/observability purposes only; the
// flush uses firstSeen to keep the contract simple and testable.)
func (c *Collapser) flushReady(now time.Time) {
	threshold := c.flushMs
	var ready []pendingEntry
	c.mu.Lock()
	for k, e := range c.pending {
		if now.Sub(e.firstSeen) >= threshold {
			ready = append(ready, *e)
			delete(c.pending, k)
		}
	}
	c.mu.Unlock()
	for _, e := range ready {
		c.onFlush(e.sourceID, e.key, e.alert)
	}
}

// FlushAll is for graceful shutdown — publishes every pending
// entry immediately. Call from a defer after runCancel.
func (c *Collapser) FlushAll() {
	c.mu.Lock()
	ready := make([]pendingEntry, 0, len(c.pending))
	for k, e := range c.pending {
		ready = append(ready, *e)
		delete(c.pending, k)
	}
	c.mu.Unlock()
	for _, e := range ready {
		c.onFlush(e.sourceID, e.key, e.alert)
	}
}