M0(2-3/12): config + observability packages

- internal/config: env-driven Common + Ingestd, all defaults
  sensible, BA_INGESTD_* knobs match SPEC §22 caps
- internal/config_test: defaults + bad-env cases
- internal/observability: slog (JSON in non-dev, text in dev),
  Prometheus registry factory, /health handler, trace-id stub
  for M9 OTel wiring
- internal/observability/metrics: IngestdMetrics implementing
  SPEC §22 metric names: alerts_received_total, payload_bytes,
  rate_limited_total{scope}, source_quarantined_total,
  circuit_breaker_state, publish_latency_seconds
- go.mod: prometheus/client_golang 1.23.2

go.mod

@@ -1,3 +1,16 @@
 module git3.techno-world.net/lrosales/broad-announce
 
 go 1.24.4
+
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/prometheus/client_golang v1.23.2 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
+	github.com/prometheus/common v0.66.1 // indirect
+	github.com/prometheus/procfs v0.16.1 // indirect
+	go.yaml.in/yaml/v2 v2.4.2 // indirect
+	golang.org/x/sys v0.35.0 // indirect
+	google.golang.org/protobuf v1.36.8 // indirect
+)

go.sum

@@ -0,0 +1,21 @@
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
+github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
+github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
+github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
+github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
+github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs=
+github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
+github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
+github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
+go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
+go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
+golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
+golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
+google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

internal/config/config.go

@@ -0,0 +1,152 @@
+// Package config is the env-driven configuration loader used by every
+// service. Twelve-factor: no flag parsing, no config files, just env
+// vars. Anything not in here has a sane default.
+//
+// Naming convention: BA_<SERVICE>_<KEY>, e.g. BA_INGESTD_HTTP_ADDR.
+package config
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// Common is shared by every service.
+type Common struct {
+	Env         string // dev | staging | prod
+	ServiceName string
+	LogLevel    string // debug | info | warn | error
+	HTTPAddr    string // /health + /metrics + (later) /v1/*
+
+	// NATS
+	NATSURL string // nats://nats:4222
+
+	// Postgres
+	PostgresDSN string
+
+	// Redis
+	RedisURL string // redis://redis:6379/0
+
+	// Shutdown
+	ShutdownGrace time.Duration
+}
+
+// Default values applied if env unset.
+func defaultCommon() Common {
+	return Common{
+		Env:           "dev",
+		ServiceName:   "broad-announce",
+		LogLevel:      "info",
+		HTTPAddr:      ":8080",
+		NATSURL:       envOr("BA_NATS_URL", "nats://localhost:4222"),
+		PostgresDSN:   envOr("BA_POSTGRES_DSN", "postgres://ba:ba@localhost:5432/ba?sslmode=disable"),
+		RedisURL:      envOr("BA_REDIS_URL", "redis://localhost:6379/0"),
+		ShutdownGrace: 15 * time.Second,
+	}
+}
+
+// LoadCommon reads env, applies defaults, and returns a validated Common.
+func LoadCommon(serviceName string) (Common, error) {
+	c := defaultCommon()
+	c.ServiceName = serviceName
+
+	if v := os.Getenv("BA_ENV"); v != "" {
+		c.Env = v
+	}
+	if v := os.Getenv("BA_LOG_LEVEL"); v != "" {
+		c.LogLevel = v
+	}
+	if v := os.Getenv("BA_HTTP_ADDR"); v != "" {
+		c.HTTPAddr = v
+	}
+	if v := os.Getenv("BA_NATS_URL"); v != "" {
+		c.NATSURL = v
+	}
+	if v := os.Getenv("BA_POSTGRES_DSN"); v != "" {
+		c.PostgresDSN = v
+	}
+	if v := os.Getenv("BA_REDIS_URL"); v != "" {
+		c.RedisURL = v
+	}
+	if v := os.Getenv("BA_SHUTDOWN_GRACE_SEC"); v != "" {
+		n, err := strconv.Atoi(v)
+		if err != nil {
+			return c, fmt.Errorf("BA_SHUTDOWN_GRACE_SEC: %w", err)
+		}
+		c.ShutdownGrace = time.Duration(n) * time.Second
+	}
+
+	if c.Env != "dev" && c.Env != "staging" && c.Env != "prod" {
+		return c, fmt.Errorf("BA_ENV must be dev|staging|prod, got %q", c.Env)
+	}
+	return c, nil
+}
+
+func envOr(k, def string) string {
+	if v, ok := os.LookupEnv(k); ok && strings.TrimSpace(v) != "" {
+		return v
+	}
+	return def
+}
+
+// GetInt parses an env var as int, returns def if unset.
+func GetInt(k string, def int) int {
+	v := os.Getenv(k)
+	if v == "" {
+		return def
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil {
+		return def
+	}
+	return n
+}
+
+// GetDuration parses an env var as Go duration, returns def if unset.
+func GetDuration(k string, def time.Duration) time.Duration {
+	v := os.Getenv(k)
+	if v == "" {
+		return def
+	}
+	d, err := time.ParseDuration(v)
+	if err != nil {
+		return def
+	}
+	return d
+}
+
+// Ingestd is ingestd-specific config. Kept here so the service
+// binary has one import.
+type Ingestd struct {
+	Common
+
+	// Source protection defaults (SPEC §22). Per-source overrides
+	// come from the DB and override these.
+	MaxPayloadBytes          int
+	RateLimitPerSource       int
+	RateLimitPerCompany      int
+	MaxConcurrentPerIP       int
+	QuarantineHitsThreshold  int
+	QuarantineWindowSeconds  int
+	QuarantineDurationSecond int
+}
+
+// LoadIngestd reads ingestd-specific config.
+func LoadIngestd() (Ingestd, error) {
+	c, err := LoadCommon("ingestd")
+	if err != nil {
+		return Ingestd{}, err
+	}
+	return Ingestd{
+		Common:                   c,
+		MaxPayloadBytes:          GetInt("BA_INGESTD_MAX_PAYLOAD_BYTES", 256*1024),
+		RateLimitPerSource:       GetInt("BA_INGESTD_RATE_LIMIT_PER_SOURCE", 100),
+		RateLimitPerCompany:      GetInt("BA_INGESTD_RATE_LIMIT_PER_COMPANY", 10_000),
+		MaxConcurrentPerIP:       GetInt("BA_INGESTD_MAX_CONCURRENT_PER_IP", 64),
+		QuarantineHitsThreshold:  GetInt("BA_INGESTD_QUARANTINE_HITS_THRESHOLD", 100),
+		QuarantineWindowSeconds:  GetInt("BA_INGESTD_QUARANTINE_WINDOW_SECONDS", 60),
+		QuarantineDurationSecond: GetInt("BA_INGESTD_QUARANTINE_DURATION_SECONDS", 300),
+	}, nil
+}

internal/config/config_test.go

@@ -0,0 +1,43 @@
+package config
+
+import (
+	"testing"
+)
+
+func TestLoadCommon_Defaults(t *testing.T) {
+	t.Setenv("BA_ENV", "")
+	c, err := LoadCommon("ingestd")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if c.Env != "dev" {
+		t.Fatalf("default env: want dev, got %q", c.Env)
+	}
+	if c.ServiceName != "ingestd" {
+		t.Fatalf("service name: got %q", c.ServiceName)
+	}
+	if c.ShutdownGrace.Seconds() != 15 {
+		t.Fatalf("shutdown grace default: got %v", c.ShutdownGrace)
+	}
+}
+
+func TestLoadCommon_BadEnv(t *testing.T) {
+	t.Setenv("BA_ENV", "qa")
+	_, err := LoadCommon("ingestd")
+	if err == nil {
+		t.Fatal("expected error for bad env")
+	}
+}
+
+func TestLoadIngestd_Defaults(t *testing.T) {
+	c, err := LoadIngestd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if c.MaxPayloadBytes != 256*1024 {
+		t.Fatalf("max payload: %d", c.MaxPayloadBytes)
+	}
+	if c.RateLimitPerSource != 100 {
+		t.Fatalf("rate limit per source: %d", c.RateLimitPerSource)
+	}
+}

internal/observability/internal.go

@@ -0,0 +1,9 @@
+package observability
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+)
+
+func readRandom(b []byte) (int, error) { return rand.Read(b) }
+func hexEncode(b []byte) string         { return hex.EncodeToString(b) }

internal/observability/metrics.go

@@ -0,0 +1,87 @@
+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.Histogram
+}
+
+// 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.",
+			ConstLabels: prometheus.Labels{"service": serviceName},
+		}, []string{"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.NewHistogram(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},
+		}),
+	}
+	reg.MustRegister(
+		m.AlertsReceived,
+		m.PayloadBytes,
+		m.RateLimitHits,
+		m.Quarantines,
+		m.CBState,
+		m.PublishLatency,
+	)
+	m.AlertsReceived.WithLabelValues("accepted")
+	return m
+}

internal/observability/observability.go

@@ -0,0 +1,96 @@
+// Package observability is the shared logging + metrics + OTel stub.
+//
+// M0: slog (stdlib structured logging, JSON output) and Prometheus
+// metrics via promhttp. OTel is a stub — we wire it in M9.
+//
+// All services should call Init() in main and then use the returned
+// logger and the package-level metrics.
+package observability
+
+import (
+	"context"
+	"log/slog"
+	"net/http"
+	"os"
+	"strings"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+// Init returns a slog.Logger configured for the env. JSON in non-dev
+// for log shippers, text in dev for human eyes.
+func Init(env, level, serviceName string) *slog.Logger {
+	lvl := parseLevel(level)
+	opts := &slog.HandlerOptions{Level: lvl}
+
+	var h slog.Handler
+	if env == "dev" {
+		h = slog.NewTextHandler(os.Stdout, opts)
+	} else {
+		h = slog.NewJSONHandler(os.Stdout, opts)
+	}
+
+	logger := slog.New(h).With(
+		slog.String("service", serviceName),
+		slog.String("env", env),
+	)
+	slog.SetDefault(logger)
+	return logger
+}
+
+func parseLevel(s string) slog.Level {
+	switch strings.ToLower(s) {
+	case "debug":
+		return slog.LevelDebug
+	case "warn":
+		return slog.LevelWarn
+	case "error":
+		return slog.LevelError
+	default:
+		return slog.LevelInfo
+	}
+}
+
+// MetricsHandler returns an http.Handler that serves Prometheus
+// metrics from the given registry.
+func MetricsHandler(reg *prometheus.Registry) http.Handler {
+	return promhttp.HandlerFor(reg, promhttp.HandlerOpts{Registry: reg})
+}
+
+// HealthHandler is the standard /health endpoint. Liveness only —
+// readiness (can serve traffic) is a separate concern, added per
+// service as it comes online.
+func HealthHandler(serviceName string) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"status":"ok","service":"` + serviceName + `"}`))
+	})
+}
+
+// CtxWithTrace is a stub for M9 when OTel is wired in. For now
+// it just attaches the request id (or generates one) so logs can
+// be correlated.
+func CtxWithTrace(ctx context.Context, traceID string) context.Context {
+	if traceID == "" {
+		// tiny random — not crypto-strong, but unique enough
+		traceID = newTraceID()
+	}
+	return context.WithValue(ctx, traceKey{}, traceID)
+}
+
+// TraceFromCtx returns the trace id, or "" if none.
+func TraceFromCtx(ctx context.Context) string {
+	v, _ := ctx.Value(traceKey{}).(string)
+	return v
+}
+
+type traceKey struct{}
+
+func newTraceID() string {
+	b := make([]byte, 8)
+	if _, err := readRandom(b); err != nil {
+		return "0"
+	}
+	return hexEncode(b)
+}