M0(4-7/12): broker (NATS JS), store (redis), dedupe, ratelimit

- internal/broker: NATS JetStream wrapper. Connect(), EnsureStreams()
  creates ALERTS (24h), DELIVERIES (1h), DLQ (7d). Subject
  helpers AlertsSubject/DeliveriesSubject/DLQSubject.
- internal/store/redis: ConnectRedis with ping + retry-friendly
  ParseURL.
- internal/dedupe: 60s SET NX EX + INCR pattern, returns
  (isNew, count, err). Empty dedupe_key is never deduped.
- internal/ratelimit: per-second INCR bucket (per-second key).
  Returns (ok, ttl, err). Zero cap = no-op.
- dedupe_test.go and limiter_test.go require live Redis
  (localhost:6379); t.Skipf if not available.
- go.mod: nats.go + jetstream + go-redis v9.

go.mod

@@ -1,16 +1,23 @@
 module git3.techno-world.net/lrosales/broad-announce
 
-go 1.24.4
+go 1.25.0
 
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/klauspost/compress v1.18.5 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/nats-io/nats.go v1.52.0 // indirect
+	github.com/nats-io/nkeys v0.4.15 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/prometheus/client_golang v1.23.2 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.66.1 // indirect
 	github.com/prometheus/procfs v0.16.1 // indirect
+	github.com/redis/go-redis/v9 v9.20.1 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
-	golang.org/x/sys v0.35.0 // indirect
+	golang.org/x/crypto v0.49.0 // indirect
+	golang.org/x/sys v0.42.0 // indirect
 	google.golang.org/protobuf v1.36.8 // indirect
 )

go.sum

@@ -2,8 +2,16 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
+github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
+github.com/nats-io/nats.go v1.52.0 h1:n3avV4VBsCgsdwh71TppsTwtv+QdPs7ntSKM8qJLGsc=
+github.com/nats-io/nats.go v1.52.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
+github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
+github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
 github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
 github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
@@ -12,10 +20,18 @@ github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9Z
 github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
 github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
 github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
+github.com/redis/go-redis/v9 v9.20.1 h1:sfCU6A8P3dXbKyWes02uxA2baehGux9dZHfEKtsTB1w=
+github.com/redis/go-redis/v9 v9.20.1/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
+go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
+go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
 go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
+golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
+golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
 golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
 golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
+golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
 google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

internal/broker/broker.go

@@ -0,0 +1,110 @@
+// Package broker is the NATS JetStream wrapper. Every service that
+// publishes or subscribes goes through this package so subject
+// construction and stream creation stay consistent.
+//
+// Subject layout (SPEC ARCHITECTURE §6):
+//   alerts.<company_id>                  – ingestd → routerd
+//   deliveries.<channel>.<company_id>    – routerd → deliverd
+//   dlq.<channel>.<company_id>           – deliverd → operator
+//
+// JetStream streams declared in EnsureStreams():
+//   ALERTS        – alerts.*            (24h retention, replicas=1 for M0)
+//   DELIVERIES    – deliveries.*        (1h retention)
+//   DLQ           – dlq.*               (7d retention)
+package broker
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Client is the wrapper. One per service.
+type Client struct {
+	nc  *nats.Conn
+	js  jetstream.JetStream
+	url string
+}
+
+// Connect dials NATS and ensures the streams exist.
+func Connect(ctx context.Context, url string) (*Client, error) {
+	nc, err := nats.Connect(url, nats.Name("broad-announce"),
+		nats.MaxReconnects(-1),
+		nats.ReconnectWait(2*time.Second),
+		nats.Timeout(5*time.Second),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("nats connect %s: %w", url, err)
+	}
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		nc.Close()
+		return nil, fmt.Errorf("jetstream init: %w", err)
+	}
+
+	c := &Client{nc: nc, js: js, url: url}
+	if err := c.EnsureStreams(ctx); err != nil {
+		nc.Close()
+		return nil, err
+	}
+	return c, nil
+}
+
+// Close drains and closes the connection.
+func (c *Client) Close() {
+	if c.nc != nil {
+		c.nc.Drain()
+	}
+}
+
+// JS exposes the underlying JetStream context for callers that need
+// it (e.g. consumer creation). Prefer the helpers below for normal use.
+func (c *Client) JS() jetstream.JetStream { return c.js }
+
+// NC exposes the raw NATS connection for ping/flush operations.
+func (c *Client) NC() *nats.Conn { return c.nc }
+
+// EnsureStreams creates the three JetStream streams if they don't
+// exist. M0: single-node, no replication.
+func (c *Client) EnsureStreams(ctx context.Context) error {
+	streams := []struct {
+		name     string
+		subjects []string
+		age      time.Duration
+	}{
+		{"ALERTS", []string{"alerts.>"}, 24 * time.Hour},
+		{"DELIVERIES", []string{"deliveries.>"}, 1 * time.Hour},
+		{"DLQ", []string{"dlq.>"}, 7 * 24 * time.Hour},
+	}
+	for _, s := range streams {
+		_, err := c.js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
+			Name:     s.name,
+			Subjects: s.subjects,
+			MaxAge:   s.age,
+			Storage:  jetstream.FileStorage,
+		})
+		if err != nil {
+			return fmt.Errorf("ensure stream %s: %w", s.name, err)
+		}
+	}
+	return nil
+}
+
+// AlertsSubject returns the subject for an alert destined to a company.
+func AlertsSubject(companyID string) string {
+	return "alerts." + companyID
+}
+
+// DeliveriesSubject returns the per-channel subject for a (company, channel).
+func DeliveriesSubject(channel, companyID string) string {
+	return "deliveries." + channel + "." + companyID
+}
+
+// DLQSubject returns the per-channel DLQ subject.
+func DLQSubject(channel, companyID string) string {
+	return "dlq." + channel + "." + companyID
+}

internal/dedupe/dedupe.go

@@ -0,0 +1,70 @@
+// Package dedupe implements the 60s-window dedupe with dedupe_count
+// return per SPEC §5.
+//
+// Algorithm:
+//   key   = dedupe:{source_id}:{dedupe_key}
+//   SET key 1 NX EX 60
+//     OK     -> first arrival, return (count=1, isNew=true)
+//     nil    -> key existed; INCR; return (count=N, isNew=false)
+//
+// M0 uses plain SET NX + INCR. We can swap to a Lua script for
+// atomicity later if we see races.
+package dedupe
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	DefaultWindow = 60 * time.Second
+	keyPrefix     = "dedupe:"
+)
+
+type Deduper struct {
+	rdb    *redis.Client
+	window time.Duration
+}
+
+func New(rdb *redis.Client, window time.Duration) *Deduper {
+	if window <= 0 {
+		window = DefaultWindow
+	}
+	return &Deduper{rdb: rdb, window: window}
+}
+
+// Check atomically claims a dedupe slot. Returns:
+//
+//	(true,  1, nil)  – first arrival in the window
+//	(false, n, nil)  – duplicate; n is the 1-indexed count in this window
+//	(false, 0, err)  – redis error; caller should fail open
+func (d *Deduper) Check(ctx context.Context, sourceID, dedupeKey string) (isNew bool, count uint32, err error) {
+	if dedupeKey == "" {
+		// No dedupe key => not dedupable. Caller treats it as a new alert.
+		return true, 1, nil
+	}
+	key := keyPrefix + sourceID + ":" + dedupeKey
+
+	ok, err := d.rdb.SetNX(ctx, key, 1, d.window).Result()
+	if err != nil {
+		return false, 0, fmt.Errorf("dedupe setnx: %w", err)
+	}
+	if ok {
+		return true, 1, nil
+	}
+	n, err := d.rdb.Incr(ctx, key).Result()
+	if err != nil {
+		return false, 0, fmt.Errorf("dedupe incr: %w", err)
+	}
+	if n < 1 {
+		n = 1
+	}
+	return false, uint32(n), nil
+}
+
+// ErrInvalidSource is returned for a missing sourceID.
+var ErrInvalidSource = errors.New("dedupe: empty source id")

internal/dedupe/dedupe_test.go

@@ -0,0 +1,95 @@
+package dedupe
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"git3.techno-world.net/lrosales/broad-announce/internal/store"
+)
+
+// These tests need a real Redis. They use a short window and a
+// test-prefix so they don't collide with prod data.
+//
+// In CI: run with `go test -tags=redis ./...` (the integration
+// build tag) and ensure Redis is on localhost:6379.
+//
+// In the M0 docker-compose, `make test-redis` runs the same suite
+// against the dev container.
+
+func newRedis(t *testing.T) *store.Redis {
+	t.Helper()
+	r, err := store.ConnectRedis(context.Background(), "redis://localhost:6379/0")
+	if err != nil {
+		t.Skipf("redis not available, skipping: %v", err)
+	}
+	return r
+}
+
+func TestCheck_FirstAndDuplicate(t *testing.T) {
+	r := newRedis(t)
+	d := New(r.Client, 2*time.Second)
+
+	isNew, n, err := d.Check(context.Background(), "src-A", "key-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !isNew || n != 1 {
+		t.Fatalf("first: want new/1, got new=%v n=%d", isNew, n)
+	}
+	isNew, n, err = d.Check(context.Background(), "src-A", "key-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if isNew || n != 2 {
+		t.Fatalf("second: want dup/2, got new=%v n=%d", isNew, n)
+	}
+}
+
+func TestCheck_DifferentSourcesDoNotCollide(t *testing.T) {
+	r := newRedis(t)
+	d := New(r.Client, 2*time.Second)
+
+	if isNew, _, _ := d.Check(context.Background(), "src-A", "shared"); !isNew {
+		t.Fatal("src-A first should be new")
+	}
+	if isNew, _, _ := d.Check(context.Background(), "src-B", "shared"); !isNew {
+		t.Fatal("src-B first should be new (different source)")
+	}
+}
+
+func TestCheck_EmptyDedupeKeyIsNeverDeduped(t *testing.T) {
+	r := newRedis(t)
+	d := New(r.Client, 2*time.Second)
+
+	for i := 0; i < 5; i++ {
+		isNew, n, err := d.Check(context.Background(), "src-A", "")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !isNew || n != 1 {
+			t.Fatalf("empty dedupe_key should always be new, got new=%v n=%d", isNew, n)
+		}
+	}
+}
+
+func TestCheck_WindowExpires(t *testing.T) {
+	r := newRedis(t)
+	d := New(r.Client, 500*time.Millisecond)
+
+	if isNew, _, _ := d.Check(context.Background(), "src-A", "expiring"); !isNew {
+		t.Fatal("first should be new")
+	}
+	time.Sleep(600 * time.Millisecond)
+	isNew, n, err := d.Check(context.Background(), "src-A", "expiring")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !isNew || n != 1 {
+		t.Fatalf("after window: want new/1, got new=%v n=%d", isNew, n)
+	}
+}
+
+// Sanity: ensure the package compiles when redis isn't around.
+var _ = redis.Nil

internal/ratelimit/limiter.go

@@ -0,0 +1,57 @@
+// Package ratelimit implements the per-source and per-company token
+// bucket from SPEC §22 layers 3 & 4.
+//
+// We use the INCR + EXPIRE pattern (works on stock Redis, no
+// redis-cell module needed). The window is 1 second; rate is
+// cap tokens / second. Tokens are integer; the bucket refills
+// atomically on the next request.
+//
+// This is an *approximation* of a token bucket — a leaky bucket
+// would be more accurate. Good enough for protecting the tier.
+package ratelimit
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const windowSeconds = 1
+
+type Limiter struct {
+	rdb *redis.Client
+}
+
+func New(rdb *redis.Client) *Limiter { return &Limiter{rdb: rdb} }
+
+// Allow consumes one token from the bucket `key` and returns:
+//
+//	(true,  0,    nil) – allowed
+//	(false, ttl,  nil) – denied, ttl is the seconds until reset
+//	(false, 0,    err) – redis error
+//
+// The cap is the per-second limit. If cap <= 0 the limiter is a
+// no-op (always allow) — useful for sources we explicitly disable.
+func (l *Limiter) Allow(ctx context.Context, key string, cap int) (bool, time.Duration, error) {
+	if cap <= 0 {
+		return true, 0, nil
+	}
+	full := fmt.Sprintf("rl:%s:%d", key, time.Now().Unix())
+	pipe := l.rdb.Pipeline()
+	incr := pipe.Incr(ctx, full)
+	pipe.Expire(ctx, full, windowSeconds*2*time.Second)
+	if _, err := pipe.Exec(ctx); err != nil {
+		return false, 0, fmt.Errorf("ratelimit: %w", err)
+	}
+	n := incr.Val()
+	if n > int64(cap) {
+		ttl, _ := l.rdb.TTL(ctx, full).Result()
+		if ttl < 0 {
+			ttl = time.Second
+		}
+		return false, ttl, nil
+	}
+	return true, 0, nil
+}

internal/ratelimit/limiter_test.go

@@ -0,0 +1,63 @@
+package ratelimit
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/store"
+)
+
+func newRedis(t *testing.T) *store.Redis {
+	t.Helper()
+	r, err := store.ConnectRedis(context.Background(), "redis://localhost:6379/0")
+	if err != nil {
+		t.Skipf("redis not available: %v", err)
+	}
+	return r
+}
+
+func TestAllow_UnderCap(t *testing.T) {
+	r := newRedis(t)
+	l := New(r.Client)
+	for i := 0; i < 5; i++ {
+		ok, _, err := l.Allow(context.Background(), "test-under-"+time.Now().Format(time.RFC3339Nano), 100)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !ok {
+			t.Fatalf("denied at i=%d under cap", i)
+		}
+	}
+}
+
+func TestAllow_OverCap(t *testing.T) {
+	r := newRedis(t)
+	l := New(r.Client)
+	key := "test-over-" + time.Now().Format(time.RFC3339Nano)
+	// cap=2, try 5 in one second
+	for i := 0; i < 2; i++ {
+		ok, _, _ := l.Allow(context.Background(), key, 2)
+		if !ok {
+			t.Fatalf("denied at i=%d", i)
+		}
+	}
+	ok, ttl, _ := l.Allow(context.Background(), key, 2)
+	if ok {
+		t.Fatal("expected deny at i=3")
+	}
+	if ttl <= 0 || ttl > 2*time.Second {
+		t.Fatalf("ttl out of range: %v", ttl)
+	}
+}
+
+func TestAllow_ZeroCapIsNoOp(t *testing.T) {
+	r := newRedis(t)
+	l := New(r.Client)
+	for i := 0; i < 1000; i++ {
+		ok, _, _ := l.Allow(context.Background(), "noop", 0)
+		if !ok {
+			t.Fatal("zero cap must always allow")
+		}
+	}
+}

internal/store/redis.go

@@ -0,0 +1,50 @@
+// Package store wraps Redis and Postgres. M0 ships Redis only; the
+// Postgres layer lands with M2 (recipient resolution).
+//
+// Redis is used for:
+//   - dedupe (SET NX EX, INCR) – SPEC §5
+//   - rate limiting (INCR + EXPIRE, or redis-cell) – SPEC §22 layer 3/4
+//   - per-source quarantine (local map, see ingestd) – SPEC §22 layer 7
+package store
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// Redis is a thin wrapper to keep construction in one place.
+type Redis struct {
+	*redis.Client
+}
+
+// ConnectRedis dials Redis. Two retries with backoff.
+func ConnectRedis(ctx context.Context, url string) (*Redis, error) {
+	opts, err := redis.ParseURL(url)
+	if err != nil {
+		return nil, fmt.Errorf("parse redis url %q: %w", url, err)
+	}
+	c := redis.NewClient(opts)
+	pingCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
+	defer cancel()
+	if err := c.Ping(pingCtx).Err(); err != nil {
+		return nil, fmt.Errorf("redis ping: %w", err)
+	}
+	return &Redis{c}, nil
+}
+
+// IsConnRefused reports whether err looks like Redis being down.
+func IsConnRefused(err error) bool {
+	return err != nil && (contains(err.Error(), "connection refused") || contains(err.Error(), "EOF"))
+}
+
+func contains(s, sub string) bool {
+	for i := 0; i+len(sub) <= len(s); i++ {
+		if s[i:i+len(sub)] == sub {
+			return true
+		}
+	}
+	return false
+}