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