broad-announce/internal/broker/broker.go

// 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.
//
// Retention: tuned for the M11 NATS resource-limit finding. The
// 24h MaxAge on ALERTS allowed 6+ GiB of test data to accumulate
// and exceed the server-level max_storage cap. With a 1h MaxAge
// plus a 1 GiB MaxBytes safety cap, the stream self-trims long
// before the server cap is hit. routerd is the only consumer and
// processes in real time, so 1h is a generous safety window.
//
// See M11_NATS_INVESTIGATION.md for the full analysis.
func (c *Client) EnsureStreams(ctx context.Context) error {
	streams := []struct {
		name     string
		subjects []string
		age      time.Duration
		maxBytes int64
	}{
		{"ALERTS", []string{"alerts.>"}, 1 * time.Hour, 1 << 30},       // 1h, 1 GiB
		{"DELIVERIES", []string{"deliveries.>"}, 1 * time.Hour, 100 << 20}, // 1h, 100 MiB
		{"DLQ", []string{"dlq.>"}, 1 * time.Hour, 10 << 20},          // 1h, 10 MiB
	}
	for _, s := range streams {
		_, err := c.js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
			Name:     s.name,
			Subjects: s.subjects,
			MaxAge:   s.age,
			MaxBytes: s.maxBytes,
			Discard:  jetstream.DiscardOld,
			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
}