broad-announce/internal/pipeline/pipeline.go

// 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
	SourceID        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(d.Transport, "invalid").Inc()
		return Reject("invalid_json", 400, err.Error())
	}
	if err := a.Validate(); err != nil {
		d.Metrics.AlertsReceived.WithLabelValues(d.Transport, "invalid").Inc()
		return Reject("invalid", 400, err.Error())
	}

	// Source lookup.
	src, ok := d.Sources[a.CompanyID+":"+a.SourceID]
	if !ok {
		d.Metrics.AlertsReceived.WithLabelValues(d.Transport, "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(d.Transport, "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(d.Transport, "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(d.Transport, "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(d.Transport, "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(d.Transport, "invalid").Inc()
		return Reject("marshal_failed", 500, err.Error())
	}
	start := time.Now()
	var publishErr error
	if d.CircuitBreaker != nil {
		// M11 fix: PublishAsync returns a future immediately. The CB wraps
		// the *submission* (not the ack) so a JetStream stall still surfaces
		// to the circuit breaker within the configured timeout. The actual
		// ack is observed in a fire-and-forget goroutine.
		var fut nats.PubAckFuture
		publishErr = d.CircuitBreaker.Do(ctx, func() error {
			f, err := d.JetStream.PublishAsync(subject, payload)
			if err != nil {
				return err
			}
			fut = f
			return nil
		})
		if publishErr == nil && fut != nil {
			go observeAsyncAck(fut, d, a.SourceID, subject, start)
		}
	} else {
		fut, err := d.JetStream.PublishAsync(subject, payload)
		if err != nil {
			publishErr = err
		} else if fut != nil {
			go observeAsyncAck(fut, d, a.SourceID, subject, start)
		}
	}
	if publishErr != nil {
		if errors.Is(publishErr, circuitbreaker.ErrCircuitOpen) {
			d.Metrics.AlertsReceived.WithLabelValues(d.Transport, "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(d.Transport, "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(d.Transport, "accepted").Inc()
	} else {
		d.Metrics.AlertsReceived.WithLabelValues(d.Transport, "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 submits to JetStream's internal queue and returns a
	// future that resolves when the broker acks persistence. Callers
	// observe the future asynchronously to avoid blocking the hot path.
	PublishAsync(subj string, data []byte) (nats.PubAckFuture, 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) (nats.PubAckFuture, error) {
	return j.js.PublishAsync(subj, data)
}

// NewNatsPublisher constructs a natsPublisher from a JetStream context.
func NewNatsPublisher(js nats.JetStreamContext) natsPublisher {
	return &jsPublisher{js: js}
}

// observeAsyncAck blocks on the PubAckFuture and records publish latency
// or a warn-level log on failure. Runs in its own goroutine so the hot path
// returns immediately. Latency is measured from start to broker ack.
func observeAsyncAck(fut nats.PubAckFuture, d *Deps, sourceID, subject string, sentAt time.Time) {
	if fut == nil {
		return
	}
	select {
	case <-fut.Ok():
		if d != nil && d.Metrics != nil {
			d.Metrics.PublishLatency.WithLabelValues(sourceID).Observe(time.Since(sentAt).Seconds())
		}
	case err := <-fut.Err():
		if d != nil && d.Logger != nil {
			d.Logger.Warn("async publish failed", "subject", subject, "source_id", sourceID, "err", err)
		}
	}
}

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