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