M6.5(1/3): Router-level dedupe Collapser with max-wait debounce

Closes the loop on the M6 deferred item: a burst of 100
identical alerts now produces 1 delivery to the recipient
(not 100 messages, the last tagged ×100).

New code:
- internal/dedupe/collapser.go: per-(source, dedupe_key)
  debounce with max-wait. Three decisions returned by Observe:
    * Passthrough    — empty dedupe_key, deliver immediately
    * CollapseNew    — first arrival of (source, key); caller
      does the expensive work (resolve recipients) and caches it
    * CollapseDupe   — subsequent arrival within the window;
      caller skips the expensive work
  The Collapser holds the alert and arms a max-wait timer;
  when the timer fires, onFlush is called with the final
  dedupe_count. Run(ctx) drives the flush loop; FlushAll()
  drains on shutdown.

- internal/dedupe/collapser_test.go: 7 unit tests, no Redis:
    1. Passthrough on empty key (no state held)
    2. Single collapse: 100 observes → 1 flush, count=100
    3. Per-source isolation: same key from 2 sources, 2 flushes
    4. Max-wait re-flush: continuous stream re-flushes every
       window, each flush carrying the running count
    5. Empty-key alerts don't interfere with a real-key burst
    6. 8-goroutine concurrent same-key storm → final
       dedupe_count is exactly the max observed
    7. FlushAll drains all pending on shutdown

- cmd/routerd/collapse.go: wires the Collapser into the
  routerd alert-processing path. The router holds a tiny
  fanoutState map (sourceID+key → resolved targets) so
  duplicates skip the DB-bound ResolveTargets call. The
  resolved targets are cached when the FIRST alert of a
  (source, key) pair arrives, and read when the Collapser
  flushes. Empty-key alerts bypass the Collapser entirely
  (M2 behavior preserved).

- cmd/routerd/main.go: refactored handleOne to route
  through observeAndFanout. The Collapser's Run loop is
  started in a goroutine in consume(); FlushAll is called
  on graceful shutdown. Config now loaded via
  config.LoadRouterd() (was LoadCommon).

- internal/config/config.go: new Routerd struct with
  DedupeFlushMs field, env var BA_ROUTERD_DEDUPE_FLUSH_MS,
  default 2000ms.

- docker-compose.yml: BA_ROUTERD_DEDUPE_FLUSH_MS=2000 on
  the routerd service.

- .env.example: same env var documented.

Live sanity check (5 alerts → 1 message):
  loadgen-ws --api-key acme-001:prom-prod:s3cret-acme \
    --count 5 --rate 5 --dedupe-key m65-sanity-1
  faketgmd /admin/sent → count=1, text='...LG M5 #0 (×5)...'

All packages green: go test ./...
7 new collapser tests + 6 existing dedupe tests.

.env.example

@@ -23,6 +23,11 @@ BA_INGESTD_RATE_LIMIT_PER_COMPANY=10000
 BA_INGESTD_MAX_CONCURRENT_PER_IP=64
 BA_INGESTD_DEDUPE_TTL_SECONDS=300
 BA_INGESTD_QUARANTINE_HITS_THRESHOLD=100
+
+# M6.5: router-level dedupe collapse. A burst of identical
+# alerts is held for up to this many ms, then a single
+# delivery is fanned out with the final dedupe_count.
+BA_ROUTERD_DEDUPE_FLUSH_MS=2000
 BA_INGESTD_QUARANTINE_WINDOW_SECONDS=60
 BA_INGESTD_QUARANTINE_DURATION_SECONDS=300
 

cmd/routerd/collapse.go

@@ -0,0 +1,250 @@
+// collapse.go wires the M6.5 router-level dedupe Collapser into
+// the alert-processing pipeline.
+//
+// What it does:
+//   - On the first arrival of a (source, dedupe_key) pair, the
+//     router resolves recipients (DB call), then stores the
+//     targets alongside the alert. The Collapser holds the alert
+//     until the flush window elapses, then fires onFlush.
+//   - On subsequent arrivals of the same pair, the router skips
+//     recipient resolution and just updates the stored
+//     dedupe_count. This is the M6.5 win: a 100-alert burst
+//     becomes 1 resolve-recipients call instead of 100.
+//   - On flush, the router fans out exactly one delivery per
+//     (target, channel) pair, with the latest dedupe_count on
+//     the alert.
+//
+// What it does NOT do:
+//   - It does NOT touch alerts with empty dedupe_key — those
+//     pass through unchanged (M2 behavior preserved).
+//   - It does NOT touch alerts whose recipient resolution
+//     fails — those get Nack'd and retried by NATS, same as M2.
+//   - It does NOT collapse across sources — per-source state is
+//     enforced by the Collapser's keying.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"sync"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
+	"git3.techno-world.net/lrosales/broad-announce/internal/broker"
+	"git3.techno-world.net/lrosales/broad-announce/internal/dedupe"
+	"git3.techno-world.net/lrosales/broad-announce/internal/routing"
+)
+
+// fanoutState is the router's per-collapse-window state.
+// It holds the resolved recipients for each (source, key) pair
+// so duplicate alerts skip the expensive ResolveTargets call.
+type fanoutState struct {
+	mu      sync.Mutex
+	pending map[collapseKey][]routing.Target
+}
+
+// collapseKey mirrors dedupe.collapseKey without exporting it.
+// We re-declare the struct here to keep the public surface of
+// internal/dedupe minimal.
+type collapseKey struct {
+	sourceID  string
+	dedupeKey string
+}
+
+func newFanoutState() *fanoutState {
+	return &fanoutState{pending: make(map[collapseKey][]routing.Target)}
+}
+
+func (f *fanoutState) put(sourceID, key string, targets []routing.Target) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.pending[collapseKey{sourceID, key}] = targets
+}
+
+func (f *fanoutState) take(sourceID, key string) []routing.Target {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	t, ok := f.pending[collapseKey{sourceID, key}]
+	if ok {
+		delete(f.pending, collapseKey{sourceID, key})
+	}
+	return t
+}
+
+func (f *fanoutState) len() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.pending)
+}
+
+// observeAndFanout is the M6.5-aware alert handler. It is called
+// instead of the old "resolve then publish" path inside handleOne.
+//
+// Behavior:
+//   - If a.DedupeKey is empty: resolves targets and publishes
+//     immediately (M2 behavior).
+//   - On the first arrival of (a.SourceID, a.DedupeKey): resolves
+//     targets, stores them, returns. The Collapser's flush will
+//     later call onFlushWithKey, which reads the stored targets
+//     and publishes.
+//   - On subsequent arrivals: returns immediately. The Collapser
+//     already updated the stored dedupe_count.
+//
+// If recipient resolution fails on the first arrival, the alert
+// is Nack'd (M2 behavior). On subsequent arrivals we cannot
+// Nack — the first arrival was already acked.
+func observeAndFanout(
+	ctx context.Context,
+	logger *slog.Logger,
+	collapser *dedupe.Collapser,
+	state *fanoutState,
+	resolver *routing.Resolver,
+	br *broker.Client,
+	companyID string,
+	a alert.Alert,
+) (handled bool, shouldAck bool, shouldNak bool) {
+	// Empty dedupe_key → M2 path, no collapse.
+	if a.DedupeKey == "" {
+		targets, err := resolver.ResolveTargets(ctx, &a)
+		if err != nil {
+			logger.Error("resolve targets", "err", err, "alert_id", a.ID, "company", companyID)
+			return true, false, true
+		}
+		if len(targets) == 0 {
+			logger.Warn("zero recipients, dropping",
+				"alert_id", a.ID,
+				"company", companyID,
+				"source_id", a.SourceID,
+				"severity", string(a.Severity),
+				"category", a.Category,
+			)
+			return true, true, false
+		}
+		publishDeliveries(logger, br, companyID, a, targets)
+		return true, true, false
+	}
+
+	// Collapse path: hand the alert to the Collapser.
+	dec := collapser.Observe(a.SourceID, a.DedupeKey, a)
+	switch dec {
+	case dedupe.CollapseNew:
+		// First arrival: resolve targets NOW and cache them.
+		targets, err := resolver.ResolveTargets(ctx, &a)
+		if err != nil {
+			logger.Error("resolve targets (new collapse)",
+				"err", err,
+				"alert_id", a.ID,
+				"company", companyID,
+				"source_id", a.SourceID,
+				"dedupe_key", a.DedupeKey,
+			)
+			return true, false, true
+		}
+		if len(targets) == 0 {
+			logger.Warn("zero recipients, dropping (new collapse)",
+				"alert_id", a.ID,
+				"company", companyID,
+				"source_id", a.SourceID,
+				"dedupe_key", a.DedupeKey,
+			)
+			return true, true, false
+		}
+		state.put(a.SourceID, a.DedupeKey, targets)
+		return true, true, false // ack the original NATS msg; the flush will publish
+	case dedupe.CollapseDupe:
+		// Stored alert's dedupe_count already updated by
+		// the Collapser. Nothing to do; the flush will
+		// publish.
+		return true, true, false
+	case dedupe.Passthrough:
+		// Should not happen here — Observe with empty
+		// key returns Passthrough and we already handled
+		// that above. Defensive: fall through to no-op.
+		return true, true, false
+	}
+	return true, true, false
+}
+
+// onFlushWithKey is the Collapser's onFlush callback. It reads
+// the cached targets for the (source, key) pair, then publishes
+// one delivery per (target, channel) with the final dedupe_count
+// on the alert.
+func onFlushWithKey(
+	logger *slog.Logger,
+	state *fanoutState,
+	br *broker.Client,
+) func(sourceID, key string, a alert.Alert) {
+	return func(sourceID, key string, a alert.Alert) {
+		targets := state.take(sourceID, key)
+		if len(targets) == 0 {
+			// No targets cached — the original message
+			// must have hit the zero-recipients path and
+			// was dropped. Nothing to do.
+			logger.Debug("flush with no cached targets",
+				"source_id", sourceID,
+				"dedupe_key", key,
+				"alert_id", a.ID,
+			)
+			return
+		}
+		logger.Info("collapsed fanout",
+			"alert_id", a.ID,
+			"source_id", sourceID,
+			"dedupe_key", key,
+			"recipients", len(targets),
+			"dedupe_count", a.DedupeCount,
+		)
+		publishDeliveries(logger, br, a.CompanyID, a, targets)
+	}
+}
+
+// publishDeliveries is the existing "enqueue one delivery per
+// target" logic, lifted out of handleOne for reuse.
+func publishDeliveries(
+	logger *slog.Logger,
+	br *broker.Client,
+	companyID string,
+	a alert.Alert,
+	targets []routing.Target,
+) int {
+	js, err := br.NC().JetStream()
+	if err != nil {
+		logger.Error("js ctx", "err", err)
+		return 0
+	}
+	enqueued := 0
+	for _, t := range targets {
+		envelope := deliveryEnvelope{
+			Alert:        a,
+			IndividualID: t.IndividualID,
+			Channel:      t.Channel,
+			Endpoint:     t.Endpoint,
+			Locale:       t.Locale,
+		}
+		body, err := json.Marshal(envelope)
+		if err != nil {
+			logger.Warn("marshal envelope", "err", err)
+			continue
+		}
+		subject := broker.DeliveriesSubject(t.Channel, companyID)
+		if _, err := js.PublishAsync(subject, body); err != nil {
+			logger.Warn("publish delivery", "err", err, "subject", subject)
+			continue
+		}
+		enqueued++
+	}
+	return enqueued
+}
+
+// deliveryEnvelope is the wire shape published on
+// deliveries.<channel>.<company_id>. Same as the one in main.go;
+// we redeclare it here so this file is self-contained for the
+// collapse code path. The fields are identical and the JSON
+// tags match — both forms serialize to the same bytes.
+type deliveryEnvelope struct {
+	Alert        alert.Alert `json:"alert"`
+	IndividualID string      `json:"individual_id"`
+	Channel      string      `json:"channel"`
+	Endpoint     string      `json:"endpoint"`
+	Locale       string      `json:"locale,omitempty"`
+}

cmd/routerd/main.go

@@ -26,6 +26,7 @@ import (
 	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
 	"git3.techno-world.net/lrosales/broad-announce/internal/broker"
 	"git3.techno-world.net/lrosales/broad-announce/internal/config"
+	"git3.techno-world.net/lrosales/broad-announce/internal/dedupe"
 	"git3.techno-world.net/lrosales/broad-announce/internal/httpserver"
 	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
 	"git3.techno-world.net/lrosales/broad-announce/internal/postgres"
@@ -34,13 +35,17 @@ import (
 )
 
 func main() {
-	cfg, err := config.LoadCommon("routerd")
+	cfg, err := config.LoadRouterd()
 	if err != nil {
 		os.Stderr.WriteString("config: " + err.Error() + "\n")
 		os.Exit(1)
 	}
 	logger := observability.Init(cfg.Env, cfg.LogLevel, "routerd")
-	logger.Info("starting", "env", cfg.Env, "addr", cfg.HTTPAddr)
+	logger.Info("starting",
+		"env", cfg.Env,
+		"addr", cfg.HTTPAddr,
+		"dedupe_flush_ms", cfg.DedupeFlushMs,
+	)
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
@@ -61,6 +66,16 @@ func main() {
 
 	resolver := routing.New(pool, logger.With("subsystem", "routing"))
 
+	// M6.5: build the router-level dedupe Collapser. It collapses
+	// bursts of identical alerts into 1 delivery per (source,
+	// dedupe_key) pair, with a max-wait debounce so a continuous
+	// stream re-flushes every flush window.
+	collapseState := newFanoutState()
+	collapser := dedupe.NewCollapser(
+		time.Duration(cfg.DedupeFlushMs)*time.Millisecond,
+		onFlushWithKey(logger, collapseState, br),
+	)
+
 	// Subscribe to all alerts.* subjects.
 	js := br.JS()
 	stream, err := js.Stream(ctx, "ALERTS")
@@ -82,7 +97,7 @@ func main() {
 	runCtx, runCancel := context.WithCancel(ctx)
 	defer runCancel()
 
-	go consume(runCtx, logger, consumer, br, resolver)
+	go consume(runCtx, logger, consumer, br, resolver, collapser, collapseState)
 
 	reg, _ := observability.NewRegistry("routerd")
 	srv := httpserver.New(httpserver.Config{
@@ -104,13 +119,30 @@ func main() {
 	}
 	runCancel()
 	time.Sleep(500 * time.Millisecond) // let consumer drain
+	// M6.5: drain any pending collapses so we don't lose
+	// the last few alerts of a graceful shutdown.
+	collapser.FlushAll()
 	if err := srv.Shutdown(ctx); err != nil {
 		logger.Warn("graceful shutdown", "err", err)
 	}
 	logger.Info("bye")
 }
 
-func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, br *broker.Client, r *routing.Resolver) {
+func consume(
+	ctx context.Context,
+	logger *slog.Logger,
+	c jetstream.Consumer,
+	br *broker.Client,
+	r *routing.Resolver,
+	collapser *dedupe.Collapser,
+	state *fanoutState,
+) {
+	// M6.5: start the Collapser's flush loop. It exits
+	// when ctx is canceled.
+	collapseCtx, collapseCancel := context.WithCancel(ctx)
+	defer collapseCancel()
+	go collapser.Run(collapseCtx)
+
 	for {
 		if ctx.Err() != nil {
 			return
@@ -125,7 +157,7 @@ func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, br
 			continue
 		}
 		for m := range batch.Messages() {
-			handleOne(ctx, logger, m, br, r)
+			handleOne(ctx, logger, m, br, r, collapser, state)
 			if batch.Error() != nil {
 				logger.Warn("batch error", "err", batch.Error())
 				break
@@ -134,7 +166,15 @@ func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, br
 	}
 }
 
-func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, br *broker.Client, r *routing.Resolver) {
+func handleOne(
+	ctx context.Context,
+	logger *slog.Logger,
+	m jetstream.Msg,
+	br *broker.Client,
+	r *routing.Resolver,
+	collapser *dedupe.Collapser,
+	state *fanoutState,
+) {
 	var a alert.Alert
 	if err := json.Unmarshal(m.Data(), &a); err != nil {
 		logger.Warn("malformed alert payload", "err", err, "subject", m.Subject())
@@ -155,79 +195,17 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, br *br
 	}
 	a.CompanyID = companyID
 
-	targets, err := r.ResolveTargets(ctx, &a)
-	if err != nil {
-		logger.Error("resolve targets", "err", err, "alert_id", a.ID, "company", companyID)
-		// Nack so the message is redelivered. In M9 we add the
-		// circuit breaker; for M2 we just retry.
+	// M6.5: route through the Collapser-aware handler. The
+	// collapse decision is made here, the actual delivery
+	// publish may happen later (on flush) — but the NATS
+	// message is always Ack'd now.
+	_, shouldAck, shouldNak := observeAndFanout(ctx, logger, collapser, state, r, br, companyID, a)
+	switch {
+	case shouldNak:
 		_ = m.Nak()
-		return
-	}
-
-	if len(targets) == 0 {
-		// M2 hard-fail. The alert is acked (it can't be delivered,
-		// retrying won't help) and the warning is logged. M3+ will
-		// route these to dlq.no_recipients.
-		logger.Warn("zero recipients, dropping",
-			"alert_id", a.ID,
-			"company", companyID,
-			"source_id", a.SourceID,
-			"severity", string(a.Severity),
-			"category", a.Category,
-		)
+	case shouldAck:
 		_ = m.Ack()
-		return
-	}
-
-	// Enqueue one deliveries.<channel>.<company_id> per (target).
-	js, err := br.NC().JetStream()
-	if err != nil {
-		logger.Error("js ctx", "err", err)
-		_ = m.Nak()
-		return
-	}
-	enqueued := 0
-	for _, t := range targets {
-		envelope := deliveryEnvelope{
-			Alert:        a,
-			IndividualID: t.IndividualID,
-			Channel:      t.Channel,
-			Endpoint:     t.Endpoint,
-			Locale:       t.Locale,
-		}
-		body, err := json.Marshal(envelope)
-		if err != nil {
-			logger.Warn("marshal envelope", "err", err)
-			continue
-		}
-		subject := broker.DeliveriesSubject(t.Channel, companyID)
-		if _, err := js.PublishAsync(subject, body); err != nil {
-			logger.Warn("publish delivery", "err", err, "subject", subject)
-			continue
-		}
-		enqueued++
 	}
-	logger.Info("routed",
-		"alert_id", a.ID,
-		"company", companyID,
-		"source_id", a.SourceID,
-		"severity", string(a.Severity),
-		"recipients", len(targets),
-		"enqueued", enqueued,
-	)
-	_ = m.Ack()
-}
-
-// deliveryEnvelope is the wire shape published on
-// deliveries.<channel>.<company_id>. M2 swaps FCMToken+Locale for
-// the channel-agnostic Channel+Endpoint so the same shape works
-// for telegram / sms / email / etc. in M3+.
-type deliveryEnvelope struct {
-	Alert        alert.Alert `json:"alert"`
-	IndividualID string      `json:"individual_id"`
-	Channel      string      `json:"channel"`
-	Endpoint     string      `json:"endpoint"`
-	Locale       string      `json:"locale,omitempty"`
 }
 
 var _ = fmt.Sprintf // keep import

docker-compose.yml

@@ -136,6 +136,12 @@ services:
       BA_HTTP_ADDR: ":8801"
       BA_NATS_URL: nats://nats:4222
       BA_POSTGRES_DSN: postgres://ba:ba@postgres:5432/ba?sslmode=disable
+      # M6.5: router-level dedupe collapse window. A burst of
+      # identical alerts is held for up to this many ms, then
+      # a single delivery is fanned out with the final
+      # dedupe_count. A continuous stream re-flushes every
+      # DedupeFlushMs.
+      BA_ROUTERD_DEDUPE_FLUSH_MS: "2000"
     ports: ["8801:8801"]
     depends_on:
       nats:     { condition: service_healthy }

internal/config/config.go

@@ -139,6 +139,29 @@ type Ingestd struct {
 	DedupeTTLSeconds int
 }
 
+// Routerd is routerd-specific config.
+type Routerd struct {
+	Common
+	// DedupeFlushMs is the M6.5 router-level dedupe collapse
+	// window. A burst of identical alerts is held for up to
+	// this many ms, then a single delivery is fanned out
+	// with the final dedupe_count. A continuous stream
+	// re-flushes every DedupeFlushMs. Default 2000ms.
+	DedupeFlushMs int
+}
+
+// LoadRouterd reads routerd-specific config.
+func LoadRouterd() (Routerd, error) {
+	c, err := LoadCommon("routerd")
+	if err != nil {
+		return Routerd{}, err
+	}
+	return Routerd{
+		Common:        c,
+		DedupeFlushMs: GetInt("BA_ROUTERD_DEDUPE_FLUSH_MS", 2000),
+	}, nil
+}
+
 // LoadIngestd reads ingestd-specific config.
 func LoadIngestd() (Ingestd, error) {
 	c, err := LoadCommon("ingestd")

internal/dedupe/collapser.go

@@ -0,0 +1,241 @@
+// 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)
+	}
+}

internal/dedupe/collapser_test.go

@@ -0,0 +1,344 @@
+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))
+	}
+}