broad-announce/cmd/routerd/collapse.go

// 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"
	"time"

	"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 == "" {
		t0 := time.Now()
		targets, err := resolver.ResolveTargets(ctx, &a)
		if routerdMetrics != nil {
			routerdMetrics.RecipientExpansionLatency.Observe(time.Since(t0).Seconds())
		}
		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.
		t0 := time.Now()
		targets, err := resolver.ResolveTargets(ctx, &a)
		if routerdMetrics != nil {
			routerdMetrics.RecipientExpansionLatency.Observe(time.Since(t0).Seconds())
		}
		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"`
}