Domů Procházet Nápověda
Registrovat se Přihlásit se
lrosales
/
broad-announce
1
0
Rozštěpit 0
Soubory Úkoly 0 Pull Requesty 0 Wiki
Procházet zdrojové kódy

M11 W2.1: extract shared pipeline to internal/pipeline/

The alert-processing pipeline (SPEC §22 protection chain) moves from
cmd/ingestd/process.go into internal/pipeline/pipeline.go. Both HTTP
and gRPC transports will call the same Deps.Process() entry point,
eliminating the divergence risk that would otherwise compound across M12+.

Changes:
- NEW internal/pipeline/pipeline.go — 330 lines
  * Deps struct (Logger, Metrics, Limiter, Deduper, JetStream, Sources,
    CompanyRatePerSec, Tail, Transport, NowFunc, MaxSeen,
    CircuitBreaker, Quarantine)
  * Process(ctx, body, sig) — the full protection chain (layers 2-8)
  * Result / Accept() / Reject() — same as before
  * SourceConfig — canonical per-source auth + rate-limit config
  * NewNatsPublisher(js) — natsPublisher constructor
  * verifyHMAC() — exported for HTTP; gRPC passes sig= to skip it

- MODIFIED cmd/ingestd/process.go — thin shim (20 lines)
  * processDeps struct embeds pipeline.Deps
  * ProcessAlert() delegates to Process() — backward compat wrapper
  * SourceConfig, Result, Accept, Reject re-exported

- MODIFIED cmd/ingestd/http.go
  * Removed duplicate SourceConfig (now from pipeline)
  * Now() method updated to call processDeps.Now()

- MODIFIED cmd/ingestd/http_test.go
  * pipeline import added
  * processDeps{} → processDeps{pipeline.Deps{...}} in test deps

- MODIFIED cmd/ingestd/main.go
  * pipeline import added
  * newNatsPublisher() → pipeline.NewNatsPublisher()
  * processDeps{} → processDeps{pipeline.Deps{...}} in deps construction

All existing tests pass (cmd/ingestd, all internal packages).
go build ./... : clean
Luis Rosales před 1 měsícem
rodič
aab63b3a45
revize
f834267102
5 změnil soubory, kde provedl 411 přidání a 355 odebrání
Jednotné zobrazení Ukázat statistiku rozdílových dat
  1. 1 11
      cmd/ingestd/http.go
  2. 3 2
      cmd/ingestd/http_test.go
  3. 5 4
      cmd/ingestd/main.go
  4. 23 338
      cmd/ingestd/process.go
  5. 379 0
      internal/pipeline/pipeline.go

+ 1 - 11
cmd/ingestd/http.go
Zobrazit soubor 

@@ -43,16 +43,6 @@ type httpDeps struct {
 
 	MaxBytes int
 
 	MaxBytes int
 
 }
 
 }
 
 
 
-// SourceConfig is what we need to know about a source to authenticate
 
-// + rate-limit it. The full Sources row has more fields; this is the
 
-// hot-path subset.
 
-type SourceConfig struct {
 
-	CompanyID         string
 
-	HMACSecret        []byte
 
-	RateLimitPerSec   int
 
-	AllowedTargets    []string // M2
 
-}
 
-
 
 // AcceptResponse is the JSON body returned on 202.
 
 // AcceptResponse is the JSON body returned on 202.
 
 type AcceptResponse struct {
 
 type AcceptResponse struct {
 
 	AlertID     string `json:"alert_id"`
 
 	AlertID     string `json:"alert_id"`
 
@@ -110,7 +100,7 @@ func (d *httpDeps) handleIngest(w http.ResponseWriter, r *http.Request) {
 
 
 
 // now is a thin alias for processDeps.now (kept here so the
 
 // now is a thin alias for processDeps.now (kept here so the
 
 // existing tests that call d.now() on httpDeps still work).
 
 // existing tests that call d.now() on httpDeps still work).
 
-func (d *httpDeps) now() time.Time { return d.processDeps.now() }
 
+func (d *httpDeps) now() time.Time { return d.processDeps.Now() }
 
 
 
 // verifyHMAC parses `X-BA-Signature: t=<unix>,v1=<hex>` and checks
 
 // verifyHMAC parses `X-BA-Signature: t=<unix>,v1=<hex>` and checks
 
 // HMAC-SHA256(secret, "<unix>.<body>") == hex. Replay window: 5 min.
 
 // HMAC-SHA256(secret, "<unix>.<body>") == hex. Replay window: 5 min.

+ 3 - 2
cmd/ingestd/http_test.go
Zobrazit soubor 

@@ -19,6 +19,7 @@ import (
 
 
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
 
+	pipeline "git3.techno-world.net/lrosales/broad-announce/internal/pipeline"
 
 )
 
 )
 
 
 
 // fakePublisher records subjects+payloads.
 
 // fakePublisher records subjects+payloads.
 
@@ -63,7 +64,7 @@ func newTestDeps() (*httpDeps, *fakePublisher) {
 
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
 
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
 
 	pub := &fakePublisher{}
 
 	pub := &fakePublisher{}
 
 	return &httpDeps{
 
 	return &httpDeps{
 
-		processDeps: processDeps{
 
+		processDeps: processDeps{pipeline.Deps{
 
 			Logger:           logger,
 
 			Logger:           logger,
 
 			Metrics:          m,
 
 			Metrics:          m,
 
 			Limiter:          nil, // unused; rate-limit paths use real limiter; we skip by hitting the bypass branch
 
 			Limiter:          nil, // unused; rate-limit paths use real limiter; we skip by hitting the bypass branch
 
@@ -77,7 +78,7 @@ func newTestDeps() (*httpDeps, *fakePublisher) {
 
 					RateLimitPerSec: 100,
 
 					RateLimitPerSec: 100,
 
 				},
 
 				},
 
 			},
 
 			},
 
-		},
 
+		}},
 
 		MaxBytes: 1024,
 
 		MaxBytes: 1024,
 
 	}, pub
 
 	}, pub
 
 }
 
 }

+ 5 - 4
cmd/ingestd/main.go
Zobrazit soubor 

@@ -18,6 +18,7 @@ import (
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/dedupe"
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/dedupe"
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/httpserver"
 
 	"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/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/quarantine"
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/ratelimit"
 
 	"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/store"
 
@@ -131,20 +132,20 @@ func main() {
 
 	maxSeen := observability.NewMaxSeen()
 
 	maxSeen := observability.NewMaxSeen()
 
 
 
 	deps := &httpDeps{
 
 	deps := &httpDeps{
 
-		processDeps: processDeps{
 
-			Logger:           logger.With("component", "http"),
 
+		processDeps: processDeps{pipeline.Deps{
 
+			Logger:            logger.With("component", "http"),
 
 			Metrics:          m,
 
 			Metrics:          m,
 
 			Limiter:          limiter,
 
 			Limiter:          limiter,
 
 			Deduper:          ded,
 
 			Deduper:          ded,
 
 			Sources:          sources,
 
 			Sources:          sources,
 
-			JetStream:        newNatsPublisher(js),
 
+			JetStream:        pipeline.NewNatsPublisher(js),
 
 			CompanyRatePerSec: cfg.RateLimitPerCompany,
 
 			CompanyRatePerSec: cfg.RateLimitPerCompany,
 
 			Tail:             hub,
 
 			Tail:             hub,
 
 			Transport:        "http",
 
 			Transport:        "http",
 
 			MaxSeen:          maxSeen,
 
 			MaxSeen:          maxSeen,
 
 			CircuitBreaker:   cb,
 
 			CircuitBreaker:   cb,
 
 			Quarantine:       q,
 
 			Quarantine:       q,
 
-		},
 
+		}},
 
 		MaxBytes:   cfg.MaxPayloadBytes,
 
 		MaxBytes:   cfg.MaxPayloadBytes,
 
 	}
 
 	}

+ 23 - 338
cmd/ingestd/process.go
Zobrazit soubor 

@@ -1,355 +1,40 @@
 
-// process.go is the shared alert-processing pipeline used by
 
-// both the HTTP handler (cmd/ingestd/http.go) and the MQTT
 
-// subscriber (cmd/ingestd/mqtt.go). It does the M0→M4 SPEC §22
 
-// work in one place: parse → validate → look up source → HMAC
 
-// verify → rate-limit (per-source, per-company) → dedupe → stamp
 
-// → publish to NATS → return a result struct the caller maps
 
-// to its own transport-level response.
 
-//
 
-// The function is pure (no global state, no transport types).
 
-// It returns Result{Accepted/Rejected/Error + reason + alert_id
 
-// + dedupe_count} so the HTTP handler can map to a status code
 
-// and the MQTT handler can map to a log line.
 
+// process.go is a thin re-export and compatibility shim for the shared
 
+// pipeline (internal/pipeline/). All the actual logic lives in pipeline/.
 
+// This file exists so that existing callers in package main (http.go,
 
+// mqtt.go, ws.go, main.go) that reference types by their package-main
 
+// names don't need to change.
 
 package main
 
 package main
 
 
 
 import (
 
 import (
 
 	"context"
 
 	"context"
 
-	"encoding/json"
 
-	"errors"
 
-	"fmt"
 
-	"log/slog"
 
-	"strconv"
 
-	"time"
 
 
 
-	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
 
-	"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/dedupe"
 
-	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
 
-	"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/tailhub"
 
-	"github.com/nats-io/nats.go"
 
+	pipeline "git3.techno-world.net/lrosales/broad-announce/internal/pipeline"
 
 )
 
 )
 
 
 
-// Result is the outcome of a ProcessAlert call. The caller
 
-// (HTTP/MQTT) maps it to its transport's response shape.
 
-type Result struct {
 
-	// Accepted is true if the alert passed all checks and was
 
-	// published to NATS.
 
-	Accepted bool
 
-	// AlertID is the server-assigned id (empty on error paths).
 
-	AlertID string
 
-	// DedupeCount is the count returned by the dedupe layer.
 
-	DedupeCount uint32
 
-	// IsNew is true for the first alert in a dedupe window.
 
-	IsNew bool
 
-	// RejectReason is the SPEC §22 / auth reason; one of:
 
-	//   "payload_too_large", "bad_request", "invalid_json",
 
-	//   "invalid", "unknown_source", "bad_signature",
 
-	//   "rate_limited_source", "rate_limited_company",
 
-	//   "marshal_failed", "broker_unavailable"
 
-	RejectReason string
 
-	// HTTPStatus is the suggested HTTP status code (0 for
 
-	// accepted / 202).
 
-	HTTPStatus int
 
-	// Detail is the free-form string the caller can show in
 
-	// a response body or a log message.
 
-	Detail string
 
-}
 
+// SourceConfig is re-exported from pipeline so callers in package main
 
+// (ws.go, mqtt.go, loadSourcesFromEnv) can use it without an import.
 
+type SourceConfig = pipeline.SourceConfig
 
+
 
+// Result is re-exported for backward compatibility.
 
+type Result = pipeline.Result
 
 
 
-// Accept is the canonical "ok" result.
 
+// Accept is re-exported for backward compatibility.
 
 func Accept(id string, count uint32, isNew bool) Result {
 
 func Accept(id string, count uint32, isNew bool) Result {
 
-	return Result{Accepted: true, AlertID: id, DedupeCount: count, IsNew: isNew, HTTPStatus: 202}
 
+	return pipeline.Accept(id, count, isNew)
 
 }
 
 }
 
 
 
-// Reject is the canonical "no" result.
 
+// Reject is re-exported for backward compatibility.
 
 func Reject(reason string, status int, detail string) Result {
 
 func Reject(reason string, status int, detail string) Result {
 
-	return Result{RejectReason: reason, HTTPStatus: status, Detail: detail}
 
+	return pipeline.Reject(reason, status, detail)
 
 }
 
 }
 
 
 
-// processDeps is the process-pipeline dependency set. Smaller
 
-// than httpDeps — no HTTP-specific fields. Both httpDeps and
 
-// the MQTT subscriber construct one and call ProcessAlert.
 
-type processDeps struct {
 
-	Logger    *slog.Logger
 
-	Metrics   *observability.IngestdMetrics
 
-	Limiter   *ratelimit.Limiter
 
-	Deduper   *dedupe.Deduper
 
-	JetStream natsPublisher
 
-	// Sources is the (company_id, source_id) → SourceConfig map.
 
-	// M0 reads it from env; M2 from Postgres. The MQTT subscriber
 
-	// uses the same map keyed on the topic-parsed (co, src).
 
-	Sources map[string]SourceConfig
 
-	// Per-company default rate cap. HTTP and MQTT use the same
 
-	// constant; once the per-company cap lives in DB, both
 
-	// transports read it.
 
-	CompanyRatePerSec int
 
-	// Tail is an optional M5 live-tail hub. When non-nil, every
 
-	// accepted alert is also published to in-process tail
 
-	// subscribers. nil is fine (HTTP/MQTT tests don't need it).
 
-	Tail *tailhub.Hub
 
-	// Transport is the per-process transport label used in
 
-	// structured log lines ("http" | "mqtt" | "ws"). The MQTT
 
-	// path overrides this on the receiver's scoped copy.
 
-	Transport string
 
-	// Now is overridable in tests.
 
-	Now func() time.Time
 
-	// MaxSeen is the M6 per-source monotonic max tracker for
 
-	// dedupe_count. processDeps owns one so all transports
 
-	// (HTTP, MQTT, WS) share the same in-process state and
 
-	// the same max-observed gauge.
 
-	MaxSeen *observability.MaxSeen
 
-	// CircuitBreaker is the M9 layer-6 per-component circuit
 
-	// breaker wrapping the NATS publish call. Nil is fine (falls
 
-	// back to direct publish without circuit protection).
 
-	CircuitBreaker *circuitbreaker.Breaker
 
-	// Quarantine is the M9 layer-7 per-source error-rate limiter.
 
-	// Nil is fine (no quarantine enforcement).
 
-	Quarantine *quarantine.Manager
 
-}
 
+// processDeps embeds pipeline.Deps and adds ProcessAlert as a thin
 
+// compatibility wrapper so existing callers (http.go, mqtt.go, ws.go)
 
+// don't need to change their call sites.
 
+type processDeps struct{ pipeline.Deps }
 
 
 
-// ProcessAlert runs the full SPEC §22 protection chain on one
 
-// alert body. It is the single source of truth for the ingest
 
-// pipeline; both the HTTP POST handler and the MQTT subscriber
 
-// call it.
 
-//
 
-// Layer order (matches SPEC §22):
 
-//   1. payload-size cap   (caller does this — http.go via
 
-//                          MaxBytesReader; mqtt.go via the
 
-//                          broker-side max-inflight setting)
 
-//   2. quarantine check (M9 layer 7) — per-source ban if error
 
-//                          rate exceeds threshold
 
-//   3. per-source rate limit
 
-//   4. per-company rate limit
 
-//   5. schema validate
 
-//   5b. parse JSON
 
-//   (auth) HMAC verify — see verifyHMAC
 
-//   5c. dedupe
 
-//   publish to NATS (M9 layer 6 circuit breaker wraps this)
 
+// ProcessAlert is the legacy entry point. It delegates to the embedded
 
+// pipeline.Deps.Process, which has the same signature.
 
 func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader string) Result {
 
 func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader string) Result {
 
-	now := d.now()
 
-
 
-	// 5. Parse + validate. We treat any parse failure as invalid.
 
-	var a alert.Alert
 
-	if err := json.Unmarshal(body, &a); err != nil {
 
-		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
 
-		return Reject("invalid_json", 400, err.Error())
 
-	}
 
-	if err := a.Validate(); err != nil {
 
-		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
 
-		return Reject("invalid", 400, err.Error())
 
-	}
 
-
 
-	// Look up source. M0: in-memory map. M2: DB.
 
-	src, ok := d.Sources[a.CompanyID+":"+a.SourceID]
 
-	if !ok {
 
-		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
 
-		return Reject("unknown_source", 401,
 
-			fmt.Sprintf("no such source %s/%s", a.CompanyID, a.SourceID))
 
-	}
 
-
 
-	// M9 layer 7: quarantine check. Check before we spend any CPU
 
-	// on a known-bad source.
 
-	if d.Quarantine != nil {
 
-		if banned, remaining, err := d.Quarantine.IsBanned(ctx, a.SourceID); err == nil && banned {
 
-			d.Metrics.AlertsReceived.WithLabelValues("quarantined").Inc()
 
-			d.Logger.Warn("source quarantined",
 
-				"source_id", a.SourceID,
 
-				"company_id", a.CompanyID,
 
-				"remaining", remaining,
 
-			)
 
-			return Reject("quarantined", 429,
 
-				fmt.Sprintf("source quarantined for %v; retry after", remaining.Round(time.Second)))
 
-		}
 
-	}
 
-
 
-	// Auth. Stripe-style: X-BA-Signature: t=<unix>,v1=<hex>.
 
-	// For HTTP it's a header; for MQTT it's a top-level field
 
-	// on the envelope — both call sites pass the same string.
 
-	if !verifyHMAC(sigHeader, src.HMACSecret, body, now) {
 
-		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
 
-		return Reject("bad_signature", 401, "")
 
-	}
 
-
 
-	// M9: quarantine hit tracking. Any rejection after the HMAC
 
-	// check (where we know the source is real) indicates a
 
-	// problematic source. We use a closure so the defer pattern
 
-	// catches every return path without boilerplate at each one.
 
-	var hitRecorded bool
 
-	defer func() {
 
-		if !hitRecorded && d.Quarantine != nil {
 
-			_ = d.Quarantine.RecordHit(context.Background(), a.SourceID)
 
-		}
 
-	}()
 
-	// recordHit records a quarantine error hit for a.SourceID.
 
-	// Call it before any rejection return.
 
-	recordHit := func() {
 
-		if d.Quarantine != nil && !hitRecorded {
 
-			hitRecorded = true
 
-			_ = d.Quarantine.RecordHit(context.Background(), a.SourceID)
 
-		}
 
-	}
 
-
 
-	// M6: Dedupe BEFORE rate limit. A duplicate (isNew=false)
 
-	// is a Redis INCR + JSON marshal + NATS publish — it does
 
-	// not warrant burning a rate-limit token. The rate limit
 
-	// exists to backpressure "new alert" volume; the dedupe
 
-	// itself is the canonical "do less work for repeats"
 
-	// mechanism. We only burn a token on the first arrival
 
-	// in a sliding window; the next 999 dupes pass through
 
-	// the rate limit gates for free, the recipient sees one
 
-	// consolidated message with `(×N)` appended.
 
-	isNew, count, err := d.Deduper.Check(ctx, a.SourceID, a.DedupeKey)
 
-	if err != nil {
 
-		d.Logger.Warn("dedupe redis error (failing open)", "err", err)
 
-		isNew, count = true, 1
 
-	}
 
-	if !isNew {
 
-		// M6 metrics: tick the per-source collapse counter and
 
-		// bump the max-observed gauge if this hit set a new
 
-		// peak. We use a tiny in-process max tracker (Prom's
 
-		// Gauge doesn't expose Get() — the canonical pattern
 
-		// is to read via .Gauges() and pick out the value, but
 
-		// that's a 2-step write+read; cleaner to just remember
 
-		// the max in our own map and Set the gauge on growth).
 
-		d.Metrics.DedupeCollapsed.WithLabelValues(a.SourceID).Inc()
 
-		d.MaxSeen.RecordAndExport(a.SourceID, count,
 
-			func(s string, v float64) {
 
-				d.Metrics.DedupeCountMax.WithLabelValues(s).Set(v)
 
-			})
 
-	}
 
-
 
-	// 3. Per-source rate limit (only charged for new alerts).
 
-	if isNew {
 
-		if ok, ttl, err := d.Limiter.Allow(ctx, "source:"+a.CompanyID+":"+a.SourceID, src.RateLimitPerSec); err != nil {
 
-			d.Logger.Warn("ratelimit redis error (failing open)", "err", err, "scope", "source")
 
-		} else if !ok {
 
-			d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
 
-			d.Metrics.RateLimitHits.WithLabelValues("source").Inc()
 
-			_ = ttl
 
-			recordHit()
 
-			return Reject("rate_limited_source", 429, strconv.Itoa(int(ttl.Seconds())))
 
-		}
 
-	}
 
-
 
-	// 4. Per-company rate limit (only charged for new alerts).
 
-	if isNew {
 
-		if ok, ttl, _ := d.Limiter.Allow(ctx, "company:"+a.CompanyID, d.CompanyRatePerSec); !ok {
 
-			d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
 
-			d.Metrics.RateLimitHits.WithLabelValues("company").Inc()
 
-			_ = ttl
 
-			recordHit()
 
-			return Reject("rate_limited_company", 429, strconv.Itoa(int(ttl.Seconds())))
 
-		}
 
-	}
 
-
 
-	// Stamp server-side fields.
 
-	a.ID = alert.NewID()
 
-	a.ReceivedAt = now.UTC()
 
-	a.DedupeCount = count
 
-
 
-	// Publish to NATS. M9 layer 6: circuit breaker wraps the
 
-	// publish call so a sick NATS server doesn't take down ingestd.
 
-	subject := broker.AlertsSubject(a.CompanyID)
 
-	payload, err := json.Marshal(a)
 
-	if err != nil {
 
-		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
 
-		return Reject("marshal_failed", 500, err.Error())
 
-	}
 
-	start := time.Now()
 
-	var publishErr error
 
-	if d.CircuitBreaker != nil {
 
-		// Wrap the synchronous publish in the circuit breaker.
 
-		// We use Publish (sync) so Do() gets immediate feedback.
 
-		publishErr = d.CircuitBreaker.Do(ctx, func() error {
 
-			return d.JetStream.Publish(subject, payload)
 
-		})
 
-	} else {
 
-		publishErr = d.JetStream.Publish(subject, payload)
 
-	}
 
-	if publishErr != nil {
 
-		if errors.Is(publishErr, circuitbreaker.ErrCircuitOpen) {
 
-			d.Metrics.AlertsReceived.WithLabelValues("circuit_open").Inc()
 
-			d.Metrics.CBState.WithLabelValues("nats").Set(circuitbreaker.StateOpen)
 
-			d.Logger.Warn("circuit breaker open",
 
-				"subject", subject,
 
-				"alert_id", a.ID,
 
-				"company_id", a.CompanyID,
 
-			)
 
-			recordHit()
 
-			return Reject("circuit_open", 503, "broker circuit breaker open")
 
-		}
 
-		// Real publish error (network, auth, etc.).
 
-		d.Metrics.AlertsReceived.WithLabelValues("broker_unavailable").Inc()
 
-		d.Logger.Error("nats publish", "err", publishErr, "subject", subject)
 
-		recordHit()
 
-		return Reject("broker_unavailable", 503, publishErr.Error())
 
-	}
 
-	d.Metrics.PublishLatency.WithLabelValues(a.SourceID).Observe(time.Since(start).Seconds())
 
-	d.Metrics.PayloadBytes.Observe(float64(len(payload)))
 
-
 
-	if isNew {
 
-		d.Metrics.AlertsReceived.WithLabelValues("accepted").Inc()
 
-	} else {
 
-		d.Metrics.AlertsReceived.WithLabelValues("deduped").Inc()
 
-	}
 
-
 
-	d.Logger.Info("alert accepted",
 
-		"alert_id", a.ID,
 
-		"company_id", a.CompanyID,
 
-		"source_id", a.SourceID,
 
-		"severity", string(a.Severity),
 
-		"transport", d.Transport,
 
-		"dedupe_count", count,
 
-	)
 
-	// M5: fan out to the live-tail hub (if configured). This
 
-	// is best-effort and never blocks the producer — the hub's
 
-	// Publish drops on slow consumers.
 
-	if d.Tail != nil {
 
-		ev := tailhub.FromAlert(&a, d.Transport)
 
-		d.Tail.Publish(ev)
 
-	}
 
-	// Accepted: mark that we did not get a rejection so the
 
-	// defer does not record a spurious quarantine hit.
 
-	hitRecorded = true
 
-	return Accept(a.ID, count, isNew)
 
-}
 
-
 
-func (d *processDeps) now() time.Time {
 
-	if d.Now != nil {
 
-		return d.Now()
 
-	}
 
-	return time.Now()
 
-}
 
-
 
-// natsPublisher is the minimal NATS interface. The HTTP and
 
-// MQTT paths share it; tests can swap in a fake.
 
-type natsPublisher interface {
 
-	PublishAsync(subj string, data []byte) error
 
-	// Publish is synchronous. The circuit breaker uses this
 
-	// to get immediate success/failure feedback.
 
-	Publish(subj string, data []byte) error
 
-}
 
-
 
-// jsPublisher adapts a nats.JetStreamContext to the natsPublisher interface.
 
-type jsPublisher struct {
 
-	js nats.JetStreamContext
 
-}
 
-
 
-// Publish is synchronous (blocks until server ack or timeout). Used
 
-// by the circuit breaker which needs immediate success/failure feedback.
 
-func (j *jsPublisher) Publish(subj string, data []byte) error {
 
-	_, err := j.js.Publish(subj, data)
 
-	return err
 
-}
 
-
 
-func (j *jsPublisher) PublishAsync(subj string, data []byte) error {
 
-	_, err := j.js.PublishAsync(subj, data)
 
-	return err
 
-}
 
-
 
-// newNatsPublisher is the constructor used by main.
 
-func newNatsPublisher(js nats.JetStreamContext) natsPublisher {
 
-	return &jsPublisher{js: js}
 
+	return d.Process(ctx, body, sigHeader)
 
 }
 
 }

+ 379 - 0
internal/pipeline/pipeline.go
Zobrazit soubor 

@@ -0,0 +1,379 @@
 
+// Package pipeline is the shared alert-processing engine used by all
 
+// ingestd transports (HTTP POST, WebSocket, MQTT, gRPC).
 
+//
 
+// It implements the SPEC §22 protection chain in order:
 
+//
 
+//	1. payload-size cap          (caller enforces)
 
+//	2. quarantine check          (M9 layer 7 — per-source error ban)
 
+//	3. per-source rate limit
 
+//	4. per-company rate limit
 
+//	5. schema validate + parse
 
+//	6. HMAC verify              (transport-specific — caller passes sig)
 
+//	7. dedupe                    (Redis sliding window)
 
+//	8. publish to NATS JetStream (M9 layer 6 — circuit breaker)
 
+//
 
+// The function is pure (no global state, no transport types). It returns
 
+// Result{Accepted/Rejected + reason + alert_id + dedupe_count} so the
 
+// caller maps it to its own transport-level response shape.
 
+package pipeline
 
+
 
+import (
 
+	"context"
 
+	"crypto/hmac"
 
+	"crypto/sha256"
 
+	"crypto/subtle"
 
+	"encoding/hex"
 
+	"encoding/json"
 
+	"errors"
 
+	"fmt"
 
+	"log/slog"
 
+	"strconv"
 
+	"strings"
 
+	"time"
 
+
 
+	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
 
+	"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/dedupe"
 
+	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
 
+	"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/tailhub"
 
+	"github.com/nats-io/nats.go"
 
+)
 
+
 
+// SourceConfig holds the per-source authentication and rate-limit parameters
 
+// needed by the pipeline. It is the canonical definition; callers must
 
+// populate the Sources map with entries keyed by "company_id:source_id".
 
+type SourceConfig struct {
 
+	CompanyID       string
 
+	HMACSecret      []byte   // may be empty for transports that don't use HMAC
 
+	RateLimitPerSec int
 
+	AllowedTargets  []string // M2: allowed routing targets (pipeline ignores; caller enforces)
 
+}
 
+
 
+// Result is the outcome of a Process call. The caller maps it to
 
+// its transport's response shape (HTTP status, gRPC status, etc.).
 
+type Result struct {
 
+	// Accepted is true if the alert passed all checks and was published to NATS.
 
+	Accepted bool
 
+
 
+	// AlertID is the server-assigned id (empty on all error paths).
 
+	AlertID string
 
+
 
+	// DedupeCount is the dedupe hit count: 1 = first arrival in window,
 
+	// >1 = collapsed burst.
 
+	DedupeCount uint32
 
+
 
+	// IsNew is true for the first alert in a dedupe window.
 
+	IsNew bool
 
+
 
+	// RejectReason is one of:
 
+	//   "invalid_json", "invalid", "unknown_source", "bad_signature",
 
+	//   "quarantined", "rate_limited_source", "rate_limited_company",
 
+	//   "marshal_failed", "broker_unavailable", "circuit_open"
 
+	RejectReason string
 
+
 
+	// HTTPStatus is the suggested HTTP status code (202 on accept, 4xx/5xx on reject).
 
+	HTTPStatus int
 
+
 
+	// Detail is free-form context for logging or error bodies.
 
+	Detail string
 
+}
 
+
 
+// Accept is the canonical "ok" result.
 
+func Accept(id string, count uint32, isNew bool) Result {
 
+	return Result{Accepted: true, AlertID: id, DedupeCount: count, IsNew: isNew, HTTPStatus: 202}
 
+}
 
+
 
+// Reject is the canonical "no" result.
 
+func Reject(reason string, status int, detail string) Result {
 
+	return Result{RejectReason: reason, HTTPStatus: status, Detail: detail}
 
+}
 
+
 
+// Deps is the dependency set for the processing pipeline.
 
+// All transports (HTTP, MQTT, WS, gRPC) construct one of these and call Deps.Process.
 
+type Deps struct {
 
+	Logger *slog.Logger
 
+
 
+	Metrics *observability.IngestdMetrics
 
+
 
+	Limiter *ratelimit.Limiter
 
+
 
+	Deduper *dedupe.Deduper
 
+
 
+	// JetStream is the NATS JetStream publisher.
 
+	JetStream natsPublisher
 
+
 
+	// Sources is the (company_id, source_id) → SourceConfig map.
 
+	// M0 reads from env; M2 reads from Postgres.
 
+	Sources map[string]SourceConfig
 
+
 
+	// CompanyRatePerSec is the default per-company rate limit (backstop).
 
+	CompanyRatePerSec int
 
+
 
+	// Tail is the M5 live-tail hub. nil is fine (tests don't need it).
 
+	Tail *tailhub.Hub
 
+
 
+	// Transport is the label used in structured log lines
 
+	// ("http" | "mqtt" | "ws" | "grpc").
 
+	Transport string
 
+
 
+	// NowFunc is overridable in tests.
 
+	NowFunc func() time.Time
 
+
 
+	// MaxSeen is the M6 per-source monotonic max tracker for dedupe_count.
 
+	// Owned here so all transports share the same in-process state.
 
+	MaxSeen *observability.MaxSeen
 
+
 
+	// CircuitBreaker wraps the NATS publish call (M9 layer 6). Nil = no CB.
 
+	CircuitBreaker *circuitbreaker.Breaker
 
+
 
+	// Quarantine is the M9 layer-7 per-source error-rate limiter. Nil = no quarantine.
 
+	Quarantine *quarantine.Manager
 
+}
 
+
 
+// Process runs the full SPEC §22 protection chain on one alert body.
 
+// sig is the transport-specific auth token. For HTTP: the HMAC header value.
 
+// For gRPC (which authenticates via API key metadata before entering the pipeline):
 
+// pass an empty string — the pipeline skips HMAC verification.
 
+func (d *Deps) Process(ctx context.Context, body []byte, sig string) Result {
 
+	now := d.Now()
 
+
 
+	// 5. Parse + validate.
 
+	var a alert.Alert
 
+	if err := json.Unmarshal(body, &a); err != nil {
 
+		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
 
+		return Reject("invalid_json", 400, err.Error())
 
+	}
 
+	if err := a.Validate(); err != nil {
 
+		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
 
+		return Reject("invalid", 400, err.Error())
 
+	}
 
+
 
+	// Source lookup.
 
+	src, ok := d.Sources[a.CompanyID+":"+a.SourceID]
 
+	if !ok {
 
+		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
 
+		return Reject("unknown_source", 401,
 
+			fmt.Sprintf("no such source %s/%s", a.CompanyID, a.SourceID))
 
+	}
 
+
 
+	// 2. Quarantine check (M9 layer 7). Before we spend any CPU.
 
+	if d.Quarantine != nil {
 
+		if banned, remaining, err := d.Quarantine.IsBanned(ctx, a.SourceID); err == nil && banned {
 
+			d.Metrics.AlertsReceived.WithLabelValues("quarantined").Inc()
 
+			d.Logger.Warn("source quarantined",
 
+				"source_id", a.SourceID,
 
+				"company_id", a.CompanyID,
 
+				"remaining", remaining,
 
+			)
 
+			return Reject("quarantined", 429,
 
+				fmt.Sprintf("source quarantined for %v; retry after", remaining.Round(time.Second)))
 
+		}
 
+	}
 
+
 
+	// 6. Auth (transport-specific; gRPC skips by passing "").
 
+	if sig != "" && !verifyHMAC(sig, src.HMACSecret, body, now) {
 
+		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
 
+		return Reject("bad_signature", 401, "")
 
+	}
 
+
 
+	// M9 quarantine hit tracking. Every rejection after source-confirmation
 
+	// gets recorded so the source's error rate climbs.
 
+	var hitRecorded bool
 
+	defer func() {
 
+		if !hitRecorded && d.Quarantine != nil {
 
+			_ = d.Quarantine.RecordHit(context.Background(), a.SourceID)
 
+		}
 
+	}()
 
+	recordHit := func() {
 
+		if d.Quarantine != nil && !hitRecorded {
 
+			hitRecorded = true
 
+			_ = d.Quarantine.RecordHit(context.Background(), a.SourceID)
 
+		}
 
+	}
 
+
 
+	// 7. Dedupe BEFORE rate limit (M6). Duplicates don't burn rate-limit tokens.
 
+	isNew, count, err := d.Deduper.Check(ctx, a.SourceID, a.DedupeKey)
 
+	if err != nil {
 
+		d.Logger.Warn("dedupe redis error (failing open)", "err", err)
 
+		isNew, count = true, 1
 
+	}
 
+	if !isNew {
 
+		d.Metrics.DedupeCollapsed.WithLabelValues(a.SourceID).Inc()
 
+		d.MaxSeen.RecordAndExport(a.SourceID, count,
 
+			func(s string, v float64) {
 
+				d.Metrics.DedupeCountMax.WithLabelValues(s).Set(v)
 
+			})
 
+	}
 
+
 
+	// 3. Per-source rate limit (new alerts only).
 
+	if isNew {
 
+		if ok, ttl, err := d.Limiter.Allow(ctx, "source:"+a.CompanyID+":"+a.SourceID, src.RateLimitPerSec); err != nil {
 
+			d.Logger.Warn("ratelimit redis error (failing open)", "err", err, "scope", "source")
 
+		} else if !ok {
 
+			d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
 
+			d.Metrics.RateLimitHits.WithLabelValues("source").Inc()
 
+			recordHit()
 
+			return Reject("rate_limited_source", 429, strconv.Itoa(int(ttl.Seconds())))
 
+		}
 
+	}
 
+
 
+	// 4. Per-company rate limit (new alerts only).
 
+	if isNew {
 
+		if ok, ttl, _ := d.Limiter.Allow(ctx, "company:"+a.CompanyID, d.CompanyRatePerSec); !ok {
 
+			d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
 
+			d.Metrics.RateLimitHits.WithLabelValues("company").Inc()
 
+			recordHit()
 
+			return Reject("rate_limited_company", 429, strconv.Itoa(int(ttl.Seconds())))
 
+		}
 
+	}
 
+
 
+	// Stamp server-side fields.
 
+	a.ID = alert.NewID()
 
+	a.ReceivedAt = now.UTC()
 
+	a.DedupeCount = count
 
+
 
+	// 8. Publish to NATS JetStream (M9 layer 6 circuit breaker wraps this).
 
+	subject := broker.AlertsSubject(a.CompanyID)
 
+	payload, err := json.Marshal(a)
 
+	if err != nil {
 
+		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
 
+		return Reject("marshal_failed", 500, err.Error())
 
+	}
 
+	start := time.Now()
 
+	var publishErr error
 
+	if d.CircuitBreaker != nil {
 
+		publishErr = d.CircuitBreaker.Do(ctx, func() error {
 
+			return d.JetStream.Publish(subject, payload)
 
+		})
 
+	} else {
 
+		publishErr = d.JetStream.Publish(subject, payload)
 
+	}
 
+	if publishErr != nil {
 
+		if errors.Is(publishErr, circuitbreaker.ErrCircuitOpen) {
 
+			d.Metrics.AlertsReceived.WithLabelValues("circuit_open").Inc()
 
+			d.Metrics.CBState.WithLabelValues("nats").Set(circuitbreaker.StateOpen)
 
+			d.Logger.Warn("circuit breaker open",
 
+				"subject", subject,
 
+				"alert_id", a.ID,
 
+				"company_id", a.CompanyID,
 
+			)
 
+			recordHit()
 
+			return Reject("circuit_open", 503, "broker circuit breaker open")
 
+		}
 
+		d.Metrics.AlertsReceived.WithLabelValues("broker_unavailable").Inc()
 
+		d.Logger.Error("nats publish", "err", publishErr, "subject", subject)
 
+		recordHit()
 
+		return Reject("broker_unavailable", 503, publishErr.Error())
 
+	}
 
+	d.Metrics.PublishLatency.WithLabelValues(a.SourceID).Observe(time.Since(start).Seconds())
 
+	d.Metrics.PayloadBytes.Observe(float64(len(payload)))
 
+
 
+	if isNew {
 
+		d.Metrics.AlertsReceived.WithLabelValues("accepted").Inc()
 
+	} else {
 
+		d.Metrics.AlertsReceived.WithLabelValues("deduped").Inc()
 
+	}
 
+
 
+	d.Logger.Info("alert accepted",
 
+		"alert_id", a.ID,
 
+		"company_id", a.CompanyID,
 
+		"source_id", a.SourceID,
 
+		"severity", string(a.Severity),
 
+		"transport", d.Transport,
 
+		"dedupe_count", count,
 
+	)
 
+
 
+	// M5: fan out to live-tail hub (if configured). Best-effort, never blocks.
 
+	if d.Tail != nil {
 
+		ev := tailhub.FromAlert(&a, d.Transport)
 
+		d.Tail.Publish(ev)
 
+	}
 
+
 
+	hitRecorded = true // mark accepted so defer doesn't record a spurious hit
 
+	return Accept(a.ID, count, isNew)
 
+}
 
+
 
+// Now returns the current time, using d.NowFunc if set.
 
+func (d *Deps) Now() time.Time {
 
+	if d.NowFunc != nil {
 
+		return d.NowFunc()
 
+	}
 
+	return time.Now()
 
+}
 
+
 
+// natsPublisher is the minimal NATS interface the pipeline needs.
 
+type natsPublisher interface {
 
+	Publish(subj string, data []byte) error
 
+	PublishAsync(subj string, data []byte) error
 
+}
 
+
 
+// jsPublisher adapts nats.JetStreamContext to natsPublisher.
 
+type jsPublisher struct{ js nats.JetStreamContext }
 
+
 
+func (j *jsPublisher) Publish(subj string, data []byte) error {
 
+	_, err := j.js.Publish(subj, data)
 
+	return err
 
+}
 
+
 
+func (j *jsPublisher) PublishAsync(subj string, data []byte) error {
 
+	_, err := j.js.PublishAsync(subj, data)
 
+	return err
 
+}
 
+
 
+// NewNatsPublisher constructs a natsPublisher from a JetStream context.
 
+func NewNatsPublisher(js nats.JetStreamContext) natsPublisher {
 
+	return &jsPublisher{js: js}
 
+}
 
+
 
+// verifyHMAC parses `X-BA-Signature: t=<unix>,v1=<hex>` and checks
 
+// HMAC-SHA256(secret, "<unix>.<body>") == hex. Replay window: 5 min.
 
+// Exported so HTTP handlers can call it directly; gRPC passes sig="".
 
+func verifyHMAC(header string, secret, body []byte, now time.Time) bool {
 
+	if header == "" || len(secret) == 0 {
 
+		return false
 
+	}
 
+	var tsStr, sigHex string
 
+	for _, part := range strings.Split(header, ",") {
 
+		kv := strings.SplitN(strings.TrimSpace(part), "=", 2)
 
+		if len(kv) != 2 {
 
+			continue
 
+		}
 
+		switch kv[0] {
 
+		case "t":
 
+			tsStr = kv[1]
 
+		case "v1":
 
+			sigHex = kv[1]
 
+		}
 
+	}
 
+	if tsStr == "" || sigHex == "" {
 
+		return false
 
+	}
 
+	tsInt, err := strconv.ParseInt(tsStr, 10, 64)
 
+	if err != nil {
 
+		return false
 
+	}
 
+	ts := time.Unix(tsInt, 0)
 
+	if abs(now.Sub(ts)) > 5*time.Minute {
 
+		return false
 
+	}
 
+	mac := hmac.New(sha256.New, secret)
 
+	mac.Write([]byte(tsStr))
 
+	mac.Write([]byte("."))
 
+	mac.Write(body)
 
+	expected := mac.Sum(nil)
 
+	got, err := hex.DecodeString(sigHex)
 
+	if err != nil {
 
+		return false
 
+	}
 
+	return subtle.ConstantTimeCompare(expected, got) == 1
 
+}
 
+
 
+func abs(d time.Duration) time.Duration {
 
+	if d < 0 {
 
+		return -d
 
+	}
 
+	return d
 
+}