broad-announce/internal/observability/metrics.go

package observability

import (
	"github.com/prometheus/client_golang/prometheus"
)

// NewRegistry returns a fresh Prometheus registry. Each service gets
// its own so the metric labels are scoped correctly.
func NewRegistry(serviceName string) (*prometheus.Registry, *IngestdMetrics) {
	reg := prometheus.NewRegistry()
	reg.MustRegister(
		prometheus.NewGoCollector(),
		prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}),
	)
	return reg, NewIngestdMetrics(reg, serviceName)
}

// IngestdMetrics groups the counters/histograms declared in SPEC §22
// for the ingest tier. Other tiers get their own metric groups.
type IngestdMetrics struct {
	AlertsReceived *prometheus.CounterVec   // result=accepted|invalid|rate_limited|payload_too_large|quarantined|circuit_open
	PayloadBytes   prometheus.Histogram
	RateLimitHits  *prometheus.CounterVec   // scope=source|company
	Quarantines    *prometheus.CounterVec
	CBState        *prometheus.GaugeVec
	PublishLatency *prometheus.HistogramVec // labels: source_id
	// MQTTMessages is the M4 per-message counter; labels mirror
	// the same result taxonomy as AlertsReceived (accepted,
	// deduped, bad_topic, bad_signature, unknown_source,
	// rate_limited_source, rate_limited_company, invalid,
	// invalid_json, broker_unavailable) plus a "received" label
	// for every message that survived parseIncomingTopic.
	MQTTMessages *prometheus.CounterVec
	// WSMessages is the M5 per-message counter; same result
	// taxonomy as MQTTMessages plus a "received" label for every
	// message that survived the post-auth frame read.
	WSMessages *prometheus.CounterVec
	// WSConnections tracks WS endpoint lifecycle. state ∈
	// {open, closed_clean, closed_protocol_error, closed_unauth,
	// closed_rate_limited, closed_per_ip_cap}.
	WSConnections *prometheus.CounterVec
	// ConnectionRejected tracks per-IP concurrency cap rejections.
	// transport ∈ {http, ws}.
	ConnectionRejected *prometheus.CounterVec
	// TailSubscribers is the current number of /v1/tail/ws
	// clients (gauge, not counter).
	TailSubscribers prometheus.Gauge
	// TailDropped tracks tail events dropped because a
	// subscriber's channel was full. The hub increments this.
	TailDropped *prometheus.CounterVec
	// DedupeCollapsed is the M6 per-message counter for
	// duplicate observations (isNew=false). Incremented on
	// every dedupe hit that finds an existing key. Lets
	// operators answer "how loud is the dupe noise?" without
	// parsing logs.
	DedupeCollapsed *prometheus.CounterVec
	// DedupeCountMax is the M6 per-source gauge of the
	// highest dedupe_count ever observed since process start.
	// source label is the source_id. Helps dashboards alert
	// on multi-hundred duplicates (e.g. a misconfigured
	// Prometheus rule that loops every 100ms).
	DedupeCountMax *prometheus.GaugeVec

	// --- gRPC transport (M11) ---
	StreamsActive  prometheus.Gauge       // ba_ingestd_grpc_streams_active
	GRPCInflight   *prometheus.HistogramVec // ba_ingestd_grpc_inflight_per_stream{source_id}
	GRPCRateLimited *prometheus.CounterVec  // ba_ingestd_grpc_rate_limited_total{source_id}
	GRPCAckLatency *prometheus.HistogramVec // ba_ingestd_grpc_ack_latency_seconds{source_id}

	// --- F2: NATS publish outcome (M11 NATS investigation) ---
	// Receivers of ba_ingestd_alerts_received_total cannot tell whether
	// a received alert was successfully published to NATS. The receive
	// path (gRPC server) and publish path (pipeline) are decoupled, and
	// the M11 2026-06-16 finding showed a broken publish path that
	// looked healthy on the receive metric. This counter tags every
	// publish attempt as ok / error so the smoke can assert that the
	// publish path is healthy too.
	// See M11_NATS_INVESTIGATION.md, "Medium-term (prevent recurrence)".
	NATSPublishTotal *prometheus.CounterVec // ba_ingestd_nats_publish_total{result}
}

// NewIngestdMetrics registers and returns the ingestd metrics.
func NewIngestdMetrics(reg prometheus.Registerer, serviceName string) *IngestdMetrics {
	m := &IngestdMetrics{
		AlertsReceived: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "alerts_received_total",
			Help:      "Number of inbound alerts by result and transport.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"transport", "result"}),
		PayloadBytes: prometheus.NewHistogram(prometheus.HistogramOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "payload_bytes",
			Help:      "Accepted alert payload size in bytes.",
			Buckets:   prometheus.ExponentialBuckets(64, 4, 8), // 64..1MB
			ConstLabels: prometheus.Labels{"service": serviceName},
		}),
		RateLimitHits: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "rate_limited_total",
			Help:      "Rate-limit rejections by scope.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"scope"}),
		Quarantines: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "source_quarantined_total",
			Help:      "Source quarantines triggered.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"source_id", "company_id"}),
		CBState: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "circuit_breaker_state",
			Help:      "0=closed, 1=half_open, 2=open.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"component"}),
		PublishLatency: prometheus.NewHistogramVec(prometheus.HistogramOpts{
			Namespace:   "ba",
			Subsystem:    "ingestd",
			Name:         "publish_latency_seconds",
			Help:         "Time to publish an accepted alert to NATS.",
			Buckets:      prometheus.DefBuckets,
			ConstLabels:  prometheus.Labels{"service": serviceName},
		}, []string{"source_id"}),
		NATSPublishTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "nats_publish_total",
			Help:      "NATS JetStream publish attempts by result (ok/error). F2: catches publish-path failures that the receive metric misses.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"result"}),
		MQTTMessages: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "mqtt_messages_total",
			Help:      "Inbound MQTT messages by result (M4).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"result"}),
		WSMessages: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "ws_messages_total",
			Help:      "Inbound WebSocket messages by result (M5).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"result"}),
		WSConnections: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "ws_connections_total",
			Help:      "WebSocket connection lifecycle events (M5).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"state"}),
		ConnectionRejected: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "connection_rejected_total",
			Help:      "Per-IP concurrency cap rejections (M5, SPEC §22 layer 2).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"transport"}),
		TailSubscribers: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "tail_subscribers",
			Help:      "Current number of /v1/tail/ws clients.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}),
		TailDropped: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "tail_dropped_total",
			Help:      "Tail events dropped because a subscriber was too slow.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"reason"}),
		DedupeCollapsed: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "dedupe_collapsed_total",
			Help:      "M6: alert messages that hit an existing dedupe key (isNew=false).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"source"}),
		DedupeCountMax: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "dedupe_count_max_observed",
			Help:      "M6: highest dedupe_count ever observed since process start, per source.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"source"}),
		// gRPC transport (M11)
		StreamsActive: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace:   "ba",
			Subsystem:  "ingestd",
			Name:       "grpc_streams_active",
			Help:       "Number of currently open gRPC StreamAlerts streams.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}),
		GRPCInflight: prometheus.NewHistogramVec(prometheus.HistogramOpts{
			Namespace:   "ba",
			Subsystem:  "ingestd",
			Name:       "grpc_inflight_per_stream",
			Help:       "Messages currently being processed per gRPC stream.",
			Buckets:    []float64{1, 8, 16, 32, 64, 128, 256, 512},
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"source_id"}),
		GRPCRateLimited: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace:   "ba",
			Subsystem:  "ingestd",
			Name:       "grpc_rate_limited_total",
			Help:       "RATE_LIMITED Acks sent to gRPC streams.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"source_id"}),
		GRPCAckLatency: prometheus.NewHistogramVec(prometheus.HistogramOpts{
			Namespace:   "ba",
			Subsystem:  "ingestd",
			Name:       "grpc_ack_latency_seconds",
			Help:       "Server-side Ack latency for gRPC StreamAlerts (seconds).",
			Buckets:    []float64{0.001, 0.005, 0.010, 0.025, 0.050, 0.100, 0.250, 0.500, 1.0},
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"source_id"}),
	}
	reg.MustRegister(
		m.AlertsReceived,
		m.PayloadBytes,
		m.RateLimitHits,
		m.Quarantines,
		m.CBState,
		m.PublishLatency,
		m.MQTTMessages,
		m.WSMessages,
		m.WSConnections,
		m.ConnectionRejected,
		m.TailSubscribers,
		m.TailDropped,
		m.DedupeCollapsed,
		m.DedupeCountMax,
		// gRPC transport (M11)
		m.StreamsActive,
		m.GRPCInflight,
		m.GRPCRateLimited,
		m.GRPCAckLatency,
		// F2: NATS publish outcome (M11 NATS investigation)
		m.NATSPublishTotal,
	)
	m.AlertsReceived.WithLabelValues("internal", "accepted")
	return m
}

// DeliverdMetrics groups the Prometheus counters/histograms for
// the deliverd tier (SPEC §22 L3: delivery attempts + DLQ).
// Both deliverd-fcm and deliverd-telegram share this type.
type DeliverdMetrics struct {
	// DeliveryAttempts records each per-attempt delivery row.
	// channel=fcm|telegram, status=sent|failed.
	DeliveryAttempts *prometheus.CounterVec
	// DLQTotal records each time an alert is parked in the DLQ.
	// channel=fcm|telegram.
	DLQTotal *prometheus.CounterVec
	// DLQLatency records how long the retry budget lasted before
	// the alert hit the DLQ (wall-clock time from first attempt
	// to DLQ insert).
	DLQLatency prometheus.Histogram
	// RetryAttempts is the total number of retry loop iterations
	// across all alerts (sum of the attempts column on deliveries
	// rows that ended in DLQ).
	RetryAttempts *prometheus.CounterVec
}

// NewDeliverdMetrics registers and returns deliverd metrics.
func NewDeliverdMetrics(reg prometheus.Registerer, serviceName string) *DeliverdMetrics {
	m := &DeliverdMetrics{
		DeliveryAttempts: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "deliverd",
			Name:      "delivery_attempts_total",
			Help:      "Per-channel delivery attempt rows (one row per attempt).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"channel", "status"}),
		DLQTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "deliverd",
			Name:      "dlq_total",
			Help:      "Alerts parked in the DLQ (one per alert that exhausted retries).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"channel"}),
		DLQLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
			Namespace: "ba",
			Subsystem: "deliverd",
			Name:      "dlq_latency_seconds",
			Help:      "Wall-clock time from first delivery attempt to DLQ insert.",
			ConstLabels: prometheus.Labels{"service": serviceName},
			Buckets:   prometheus.ExponentialBuckets(0.1, 2, 10), // 100ms → ~100s
		}),
		RetryAttempts: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "deliverd",
			Name:      "retry_attempts_total",
			Help:      "Total retry loop iterations across all DLQ'd alerts.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"channel"}),
	}
	reg.MustRegister(m.DeliveryAttempts, m.DLQTotal, m.DLQLatency, m.RetryAttempts)
	return m
}

// RouterdMetrics groups the Prometheus histogram for the routerd tier
// (SPEC §22 L1: recipient resolution latency).
type RouterdMetrics struct {
	// RecipientExpansionLatency is the wall-clock time for
	// routing.Resolver.ResolveTargets to complete (DB call).
	RecipientExpansionLatency prometheus.Histogram
}

// NewRouterdMetrics registers and returns routerd metrics.
func NewRouterdMetrics(reg prometheus.Registerer, serviceName string) *RouterdMetrics {
	m := &RouterdMetrics{
		RecipientExpansionLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
			Namespace: "ba",
			Subsystem: "routerd",
			Name:      "recipient_expansion_seconds",
			Help:      "Time to resolve recipients for an alert (DB call).",
			ConstLabels: prometheus.Labels{"service": serviceName},
			Buckets:   prometheus.DefBuckets,
		}),
	}
	reg.MustRegister(m.RecipientExpansionLatency)
	return m
}