broad-announce/cmd/routerd/main.go

// Command routerd consumes alerts from NATS JetStream, resolves
// recipients (companies → groups → individuals ∩ subscriptions),
// and enqueues one delivery per (individual, channel) to
// deliveries.<channel>.<company_id> subjects.
//
// M0: connects to NATS, /health, /metrics. No business logic.
// M1: broadcast mode (BA_ROUTERD_M1_BROADCAST=true is implicit).
//     Resolves every active FCM token for the company, publishes
//     one deliveries.fcm.<company_id> per token.
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"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/config"
	"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/postgres"
	"git3.techno-world.net/lrosales/broad-announce/internal/routing"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	cfg, err := config.LoadCommon("routerd")
	if err != nil {
		os.Stderr.WriteString("config: " + err.Error() + "\n")
		os.Exit(1)
	}
	logger := observability.Init(cfg.Env, cfg.LogLevel, "routerd")
	logger.Info("starting", "env", cfg.Env, "addr", cfg.HTTPAddr)

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	br, err := broker.Connect(ctx, cfg.NATSURL)
	if err != nil {
		logger.Error("nats connect", "err", err)
		os.Exit(1)
	}
	defer br.Close()

	pool, err := postgres.Connect(ctx, cfg.PostgresDSN)
	if err != nil {
		logger.Error("postgres connect", "err", err)
		os.Exit(1)
	}
	defer pool.Close()

	resolver := routing.New(pool)

	// Subscribe to all alerts.* subjects.
	js := br.JS()
	stream, err := js.Stream(ctx, "ALERTS")
	if err != nil {
		logger.Error("nats stream ALERTS", "err", err)
		os.Exit(1)
	}
	consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		Name:    "routerd",
		Durable: "routerd",
		FilterSubjects: []string{"alerts.>"},
		AckPolicy: jetstream.AckExplicitPolicy,
	})
	if err != nil {
		logger.Error("nats consumer", "err", err)
		os.Exit(1)
	}

	runCtx, runCancel := context.WithCancel(ctx)
	defer runCancel()

	go consume(runCtx, logger, consumer, br, resolver)

	reg, _ := observability.NewRegistry("routerd")
	srv := httpserver.New(httpserver.Config{
		Addr:          cfg.HTTPAddr,
		ServiceName:   "routerd",
		ShutdownGrace: cfg.ShutdownGrace,
	}, logger, observability.MetricsHandler(reg))

	errCh := make(chan error, 1)
	go func() { errCh <- srv.Start() }()
	select {
	case <-ctx.Done():
		logger.Info("shutdown signal received")
	case err := <-errCh:
		if err != nil {
			logger.Error("http server", "err", err)
			os.Exit(1)
		}
	}
	runCancel()
	time.Sleep(500 * time.Millisecond) // let consumer drain
	if err := srv.Shutdown(ctx); err != nil {
		logger.Warn("graceful shutdown", "err", err)
	}
	logger.Info("bye")
}

func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, br *broker.Client, r *routing.Resolver) {
	for {
		if ctx.Err() != nil {
			return
		}
		batch, err := c.Fetch(16, jetstream.FetchMaxWait(2*time.Second))
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			logger.Warn("nats fetch", "err", err)
			time.Sleep(500 * time.Millisecond)
			continue
		}
		for m := range batch.Messages() {
			handleOne(ctx, logger, m, br, r)
			if batch.Error() != nil {
				logger.Warn("batch error", "err", batch.Error())
				break
			}
		}
	}
}

func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, br *broker.Client, r *routing.Resolver) {
	var a alert.Alert
	if err := json.Unmarshal(m.Data(), &a); err != nil {
		logger.Warn("malformed alert payload", "err", err, "subject", m.Subject())
		_ = m.Ack()
		return
	}

	// M1: extract company_id from subject "alerts.<company_id>".
	parts := strings.SplitN(m.Subject(), ".", 2)
	if len(parts) != 2 {
		logger.Warn("bad subject", "subject", m.Subject())
		_ = m.Ack()
		return
	}
	companyID := parts[1]
	if a.CompanyID != "" && a.CompanyID != companyID {
		logger.Warn("company_id mismatch", "subject", companyID, "body", a.CompanyID)
	}
	a.CompanyID = companyID

	tokens, err := r.ResolveTokens(ctx, companyID)
	if err != nil {
		logger.Error("resolve tokens", "err", err, "company", companyID)
		// Nack so the message is redelivered. In M9 we add the
		// circuit breaker; for M1 we just retry.
		_ = m.Nak()
		return
	}

	if len(tokens) == 0 {
		logger.Info("no recipients", "alert_id", a.ID, "company", companyID)
		_ = m.Ack()
		return
	}

	// Enqueue one deliveries.fcm.<company_id> per token.
	js, err := br.NC().JetStream()
	if err != nil {
		logger.Error("js ctx", "err", err)
		_ = m.Nak()
		return
	}
	delivered := 0
	for _, t := range tokens {
		envelope := deliveryEnvelope{
			Alert:        a,
			IndividualID: t.IndividualID,
			FCMToken:     t.FCMToken,
			Locale:       t.Locale,
		}
		body, err := json.Marshal(envelope)
		if err != nil {
			logger.Warn("marshal envelope", "err", err)
			continue
		}
		subject := broker.DeliveriesSubject("fcm", companyID)
		if _, err := js.PublishAsync(subject, body); err != nil {
			logger.Warn("publish delivery", "err", err, "subject", subject)
			continue
		}
		delivered++
	}
	logger.Info("routed",
		"alert_id", a.ID,
		"company", companyID,
		"recipients", len(tokens),
		"enqueued", delivered,
	)
	_ = m.Ack()
}

// deliveryEnvelope is the wire shape published on
// deliveries.fcm.<company_id>. M3+ will swap to per-channel shapes.
type deliveryEnvelope struct {
	Alert        alert.Alert `json:"alert"`
	IndividualID string      `json:"individual_id"`
	FCMToken     string      `json:"fcm_token"`
	Locale       string      `json:"locale,omitempty"`
}

var _ = fmt.Sprintf // keep import