broad-announce/internal/config/config.go

// 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/*
	GRPCAddr    string // M11: gRPC ingestd service (default :9090)
	GRPCMaxInflight int    // M11: max in-flight per gRPC stream (default 256)

	// NATS
	NATSURL string // nats://nats:4222

	// Postgres
	PostgresDSN string

	// Redis
	RedisURL string // redis://redis:6379/0

	// Shutdown
	ShutdownGrace time.Duration

	// M8: retry knobs shared by every deliverd-* binary.
	// Ingestd, routerd, archiverd, and admind ignore them
	// (they don't call retry.Run). Defaults match
	// internal/retry.Default() so the .env.example can
	// stay sparse.
	DeliverdMaxAttempts  int
	DeliverdRetryBaseMs  int
	DeliverdRetryMaxMs   int
	DeliverdRetryBudgetMs int
}

// Default values applied if env unset.
func defaultCommon() Common {
	return Common{
		Env:           "dev",
		ServiceName:   "broad-announce",
		LogLevel:      "info",
		HTTPAddr:      ":8800",
		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
	}

	// M8 retry knobs. Read on every LoadCommon so the
	// deliverd-* binaries don't need a separate config
	// struct just for these four values. Non-deliverd
	// services simply never call retry.Run.
	c.DeliverdMaxAttempts = GetInt("BA_DELIVERD_MAX_ATTEMPTS", 10)
	c.DeliverdRetryBaseMs = GetInt("BA_DELIVERD_RETRY_BASE_MS", 100)
	c.DeliverdRetryMaxMs = GetInt("BA_DELIVERD_RETRY_MAX_MS", 2000)
	c.DeliverdRetryBudgetMs = GetInt("BA_DELIVERD_RETRY_BUDGET_MS", 30000)

	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
	// Circuit breaker (SPEC §22 layer 6). Trips when
	// CircuitFailureThreshold failures accumulate in
	// CircuitFailureWindowSeconds. Stays open for
	// CircuitOpenDurationSeconds, then admits up to
	// CircuitMaxHalfOpen test calls.
	CircuitFailureThreshold    int
	CircuitFailureWindowSecs   int
	CircuitOpenDurationSecs    int
	CircuitMaxHalfOpen         int
	// DedupeTTLSeconds is the M6 sliding-window TTL for a
	// dedupe entry. Refreshed on every duplicate observation,
	// so a steady stream of duplicates keeps the window alive.
	// Default 300s (5 min) — up from 60s in M0–M5 to give
	// operators a longer window to see `×N` rollups.
	DedupeTTLSeconds int

	// M11: gRPC ingestd server
	GRPCAddr        string // default :9090
	GRPCMaxInflight int    // max in-flight per gRPC stream; default 256
}

// 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
}

// Archiverd is archiverd-specific config (M7).
type Archiverd struct {
	Common
	// RunEverySeconds is the cadence between archiverd
	// passes. Default 3600 (1 hour).
	RunEverySeconds int
	// OlderThanHours is the retention threshold; rows
	// older than now() - OlderThanHours are eligible
	// for archive. Default 168 (7 days). The Timescale
	// retention policy does the same at 7 days; the
	// archiver just runs ahead so CH has the data
	// before TS drops it.
	OlderThanHours int
	// BatchSize is the cap per SELECT/INSERT. Default
	// 10000. Each pass drains until the SELECT returns
	// < BatchSize rows, with a 100-cycle safety cap.
	BatchSize int
	// ClickHouseURL is the HTTP base URL for the CH
	// server (no trailing slash). Default
	// http://clickhouse:8123.
	ClickHouseURL string
}

// 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
}

// LoadArchiverd reads archiverd-specific config.
func LoadArchiverd() (Archiverd, error) {
	c, err := LoadCommon("archiverd")
	if err != nil {
		return Archiverd{}, err
	}
	return Archiverd{
		Common:          c,
		RunEverySeconds: GetInt("BA_ARCHIVERD_RUN_EVERY_SECONDS", 3600),
		OlderThanHours:  GetInt("BA_ARCHIVERD_OLDER_THAN_HOURS", 168),
		BatchSize:       GetInt("BA_ARCHIVERD_BATCH_SIZE", 10000),
		ClickHouseURL:  envOr("BA_ARCHIVERD_CLICKHOUSE_URL", "http://clickhouse:8123"),
	}, nil
}

// 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),
		DedupeTTLSeconds:         GetInt("BA_INGESTD_DEDUPE_TTL_SECONDS", 300),
		QuarantineHitsThreshold:   GetInt("BA_INGESTD_QUARANTINE_HITS_THRESHOLD", 100),
		QuarantineWindowSeconds:   GetInt("BA_INGESTD_QUARANTINE_WINDOW_SECONDS", 60),
		QuarantineDurationSecond:  GetInt("BA_INGESTD_QUARANTINE_DURATION_SECONDS", 300),
		CircuitFailureThreshold:  GetInt("BA_INGESTD_CB_FAILURE_THRESHOLD", 5),
		CircuitFailureWindowSecs:  GetInt("BA_INGESTD_CB_FAILURE_WINDOW_SECS", 10),
		CircuitOpenDurationSecs:   GetInt("BA_INGESTD_CB_OPEN_DURATION_SECS", 30),
		CircuitMaxHalfOpen:        GetInt("BA_INGESTD_CB_MAX_HALF_OPEN", 1),
		GRPCAddr:                 envOr("BA_INGESTD_GRPC_ADDR", ":9090"),
		GRPCMaxInflight:          GetInt("BA_INGESTD_GRPC_MAX_INFLIGHT", 256),
	}, nil
}