broad-announce/cmd/ingestd/main.go

// Command ingestd receives alerts via HTTP POST / WebSocket / MQTT / gRPC,
// validates, rate-limits, dedupes, and publishes to NATS JetStream.
//
// M0: HTTP POST endpoint only. Other transports land in M5 / M4 / M11.
package main

import (
	"context"
	"log/slog"
	"net"
	"os"
	"os/signal"
	"syscall"
	"time"

	"git3.techno-world.net/lrosales/broad-announce/internal/broker"
	"git3.techno-world.net/lrosales/broad-announce/internal/circuitbreaker"
	"git3.techno-world.net/lrosales/broad-announce/internal/concurrency"
	"git3.techno-world.net/lrosales/broad-announce/internal/config"
	"git3.techno-world.net/lrosales/broad-announce/internal/dedupe"
	"git3.techno-world.net/lrosales/broad-announce/internal/grpcserver"
	"git3.techno-world.net/lrosales/broad-announce/internal/httpserver"
	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
	"git3.techno-world.net/lrosales/broad-announce/internal/pipeline"
	"git3.techno-world.net/lrosales/broad-announce/internal/quarantine"
	"git3.techno-world.net/lrosales/broad-announce/internal/ratelimit"
	"git3.techno-world.net/lrosales/broad-announce/internal/store"
	"git3.techno-world.net/lrosales/broad-announce/internal/tailhub"
)

// grpcDeps implements grpcserver.IngestdDeps so internal/grpcserver doesn't
// import cmd/ingestd (which would create a cycle).
type grpcDeps struct {
	pipeline  pipeline.Deps
	logger   *slog.Logger
	metrics  *observability.IngestdMetrics
}

func (g *grpcDeps) GetPipeline() pipeline.Deps                  { return g.pipeline }
func (g *grpcDeps) GetLogger() interface{ Info(string, ...any) }                             { return g.logger }
func (g *grpcDeps) GetMetrics() *observability.IngestdMetrics   { return g.metrics }

func main() {
	cfg, err := config.LoadIngestd()
	if err != nil {
		// Logger isn't up yet; stderr is the only thing we have.
		os.Stderr.WriteString("config: " + err.Error() + "\n")
		os.Exit(1)
	}
	logger := observability.Init(cfg.Env, cfg.LogLevel, "ingestd")
	logger.Info("starting", "env", cfg.Env, "addr", cfg.HTTPAddr)

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	// Redis
	r, err := store.ConnectRedis(ctx, cfg.RedisURL)
	if err != nil {
		logger.Error("redis connect", "err", err)
		os.Exit(1)
	}
	defer r.Close()
	logger.Info("redis connected")

	// NATS JetStream
	br, err := broker.Connect(ctx, cfg.NATSURL)
	if err != nil {
		logger.Error("nats connect", "err", err)
		os.Exit(1)
	}
	defer br.Close()
	logger.Info("nats connected", "url", cfg.NATSURL)

	// Metrics
	reg, m := observability.NewRegistry("ingestd")
	limiter := ratelimit.New(r.Client)
	// M6: sliding-window dedupe TTL from config. Default 300s
	// (5 min), overridable via BA_INGESTD_DEDUPE_TTL_SECONDS.
	dedTTL := time.Duration(cfg.DedupeTTLSeconds) * time.Second
	if dedTTL <= 0 {
		dedTTL = dedupe.DefaultWindow
	}
	ded := dedupe.New(r.Client, dedTTL)

	// M9: circuit breaker (layer 6). Wraps the NATS publish
	// call so a sick broker doesn't take down ingestd.
	cbCfg := circuitbreaker.Config{
		Name:             "nats-publish",
		FailureThreshold: cfg.CircuitFailureThreshold,
		FailureWindow:    time.Duration(cfg.CircuitFailureWindowSecs) * time.Second,
		OpenDuration:     time.Duration(cfg.CircuitOpenDurationSecs) * time.Second,
		MaxHalfOpen:      cfg.CircuitMaxHalfOpen,
	}
	cb := circuitbreaker.New(cbCfg)
	// Also report CB state changes to Prometheus.
	cb.Measure = func(state int, err error) {
		m.CBState.WithLabelValues("nats").Set(float64(state))
	}
	// Initialize gauge to CLOSED (0) so Prometheus sees the metric before first transition.
	m.CBState.WithLabelValues("nats").Set(0)
	logger.Info("circuit breaker configured",
		"failure_threshold", cfg.CircuitFailureThreshold,
		"failure_window_sec", cfg.CircuitFailureWindowSecs,
		"open_duration_sec", cfg.CircuitOpenDurationSecs,
		"max_half_open", cfg.CircuitMaxHalfOpen,
	)

	// M9: quarantine manager (layer 7). Per-source error-rate
	// limiter backed by Redis so the ban is shared across
	// multiple ingestd instances.
	quarantineCfg := quarantine.Config{
		HitsThreshold: cfg.QuarantineHitsThreshold,
		HitsWindow:    time.Duration(cfg.QuarantineWindowSeconds) * time.Second,
		BanDuration:   time.Duration(cfg.QuarantineDurationSecond) * time.Second,
	}
	q := quarantine.New(r.Client, quarantineCfg)
	logger.Info("quarantine configured",
		"hits_threshold", cfg.QuarantineHitsThreshold,
		"hits_window_sec", cfg.QuarantineWindowSeconds,
		"ban_duration_sec", cfg.QuarantineDurationSecond,
	)

	// M0 source registry: loaded from env. M2 replaces with DB.
	sources := loadSourcesFromEnv(logger)

	js, err := br.NC().JetStream()
	if err != nil {
		logger.Error("nats jetstream context", "err", err)
		os.Exit(1)
	}

	// M5: per-IP concurrency cap (SPEC §22 layer 2). The same
	// gate is used by the HTTP server (via the WS path; HTTP's
	// per-IP cap is M10's "M5-bump" deferred — see SPEC §22
	// milestone rollout) and the WebSocket ingest path.
	perIP := concurrency.NewPerIP(cfg.MaxConcurrentPerIP)
	defer perIP.Close()

	// M5: in-process tail hub. nil-safe; both the HTTP and
	// MQTT paths call Tail.Publish if non-nil.
	hub := tailhub.NewHub()

	// M6: per-source monotonic max tracker for dedupe_count.
	// Shared by all transports so the
	// ba_ingestd_dedupe_count_max_observed gauge reflects the
	// global peak across HTTP, MQTT, and WS.
	maxSeen := observability.NewMaxSeen()

	// grpcDeps adapts ingestd's concrete types to the grpcserver.IngestdDeps
	// interface so internal/grpcserver doesn't need to import cmd/ingestd.
	grpcDep := &grpcDeps{
		pipeline: pipeline.Deps{
			Logger:            logger.With("component", "grpc"),
			Metrics:          m,
			Limiter:          limiter,
			Deduper:          ded,
			Sources:          sources,
			JetStream:        pipeline.NewNatsPublisher(js),
			CompanyRatePerSec: cfg.RateLimitPerCompany,
			Tail:             hub,
			Transport:        "grpc",
			MaxSeen:          maxSeen,
			CircuitBreaker:   cb,
			Quarantine:       q,
		},
		logger:  logger,
		metrics: m,
	}

	deps := &httpDeps{
		processDeps: processDeps{pipeline.Deps{
			Logger:            logger.With("component", "http"),
			Metrics:          m,
			Limiter:          limiter,
			Deduper:          ded,
			Sources:          sources,
			JetStream:        pipeline.NewNatsPublisher(js),
			CompanyRatePerSec: cfg.RateLimitPerCompany,
			Tail:             hub,
			Transport:        "http",
			MaxSeen:          maxSeen,
			CircuitBreaker:   cb,
			Quarantine:       q,
		}},
		MaxBytes:   cfg.MaxPayloadBytes,
	}

	// M5: WS ingest + tail handler. The WS path uses a copy of
	// deps.processDeps with Transport="ws" so the structured
	// log lines and the tail event label are correct.
	wsDeps := &wsIngestDeps{
		processDeps:   deps.processDeps,
		PerIP:         perIP,
		MaxFrameBytes: int64(cfg.MaxPayloadBytes),
		ReadDeadline:  30 * time.Second,
		WriteDeadline: 10 * time.Second,
	}
	// Override the transport on the wsDeps copy. We have to
	// use a fresh processDeps because the embedded struct is a
	// value, not a pointer, in the struct literal above.
	wsDeps.processDeps.Logger = logger.With("component", "ws")
	wsDeps.processDeps.Transport = "ws"

	tailDeps := &wsTailDeps{
		Token:   os.Getenv("BA_INGESTD_TAIL_TOKEN"),
		Hub:     hub,
		Metrics: m,
	}

	// HTTP server
	srv := httpserver.New(httpserver.Config{
		Addr:          cfg.HTTPAddr,
		ServiceName:   "ingestd",
		ShutdownGrace: cfg.ShutdownGrace,
	}, logger, observability.MetricsHandler(reg))
	RegisterRoutes(srv.Mux(), deps)
	RegisterWSRoutes(srv.Mux(), wsDeps, tailDeps)

	// MQTT subscriber (M4). Disabled if BA_INGESTD_MQTT_BROKER is
	// empty. The subscriber shares the processDeps with the HTTP
	// handler so the dedupe window, rate limits, and metrics are
	// per-source exactly once across both transports.
	mqttCfg := loadMQTTConfig(logger)
	mqttDeps := deps.processDeps
	mqttDeps.Logger = logger.With("component", "mqtt")
	mqttDeps.Transport = "mqtt"
	mqttErrCh := make(chan error, 1)
	go func() {
		mqttErrCh <- startMQTT(ctx, mqttCfg, &mqttDeps, logger, m)
	}()

	grpcSrv, err := grpcserver.NewForIngestd(grpcserver.Config{
		Addr:        cfg.GRPCAddr,
		MaxInflight: cfg.GRPCMaxInflight,
	}, grpcDep)
	if err != nil {
		logger.Error("grpcserver.NewForIngestd", "err", err)
		os.Exit(1)
	}
	grpcLn, err := net.Listen("tcp", cfg.GRPCAddr)
	if err != nil {
		logger.Error("grpc listen", "addr", cfg.GRPCAddr, "err", err)
		os.Exit(1)
	}
	go func() { grpcSrv.Serve(grpcLn) }()
	logger.Info("grpc server started", "addr", cfg.GRPCAddr)

	// Run + graceful shutdown
	errCh := make(chan error, 1)
	go func() { errCh <- srv.Start() }()

	select {
	case <-ctx.Done():
		logger.Info("shutdown signal received")
	case err := <-errCh:
		if err != nil {
			logger.Error("http server", "err", err)
			os.Exit(1)
		}
	case err := <-mqttErrCh:
		if err != nil {
			logger.Error("mqtt subscriber", "err", err)
			os.Exit(1)
		}
	}

	if err := srv.Shutdown(ctx); err != nil {
		logger.Warn("graceful shutdown", "err", err)
	}
	logger.Info("bye")
}