M1(4-5/8): routerd M1 broadcast + deliverd M1 fcm

- internal/routing: Resolver. M1 broadcast: a single SQL join
  returns every active fcm_token for every active individual in
  the company. M2 swaps for the rules engine from SPEC §6
  (subscriptions, quiet hours, routing rules).
- cmd/routerd: M1 main
  - connects to NATS + Postgres
  - subscribes to alerts.> via durable consumer 'routerd'
  - parses alert, extracts company_id from subject
  - calls Resolver.ResolveTokens
  - publishes one deliveries.fcm.<company_id> per token with
    envelope {alert, individual_id, fcm_token, locale}
  - logs routed/recipients/enqueued counts
  - graceful shutdown drains the consumer before exit
- cmd/deliverd: M1 main (deliverd-fcm)
  - subscribes to deliveries.fcm.>
  - builds the FCM HTTP v1 message body shape (M3 swap is a no-op
    at this layer — the shape matches real FCM)
  - android priority HIGH for critical/inminent_colapse
  - sound file: klaxon for inminent_colapse, siren_<category>
    for the rest (SPEC §4 + §8)
  - notification channel: alerts.{info|warning|critical|imminent}
  - posts to BA_FAKECMD_URL (default http://fakefcmd:8820)
  - writes a deliveries row (status: sent|failed, attempts=1)
  - ack on terminal status (no retry in M1; M3 adds exp backoff
    + DLQ from SPEC §9)

cmd/deliverd/main.go

@@ -3,21 +3,42 @@
 // (FCM, Telegram, SMS, email, Slack, Teams, webhook).
 //
 // M0: per-channel worker binary that connects to NATS, /health, /metrics.
-// Real delivery lands in M1+ per channel.
+// M1: deliverd-fcm — consumes deliveries.fcm.*, posts to fakefcmd,
+//     records a row in Postgres per attempt.
 package main
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
+	"time"
 
 	"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"
+	"github.com/nats-io/nats.go/jetstream"
 )
 
+// M1: only the FCM channel. M3+ adds telegram, sms, etc.
+const fcmChannel = "fcm"
+
+type deliveryEnvelope struct {
+	Alert        json.RawMessage `json:"alert"`
+	IndividualID string          `json:"individual_id"`
+	FCMToken     string          `json:"fcm_token"`
+	Locale       string          `json:"locale,omitempty"`
+}
+
 func main() {
 	cfg, err := config.LoadCommon("deliverd")
 	if err != nil {
@@ -36,10 +57,46 @@ func main() {
 		os.Exit(1)
 	}
 	defer br.Close()
-	logger.Info("nats connected")
 
-	reg, _ := observability.NewRegistry("deliverd")
+	pool, err := postgres.Connect(ctx, cfg.PostgresDSN)
+	if err != nil {
+		logger.Error("postgres connect", "err", err)
+		os.Exit(1)
+	}
+	defer pool.Close()
+
+	// M1 only: fakefcmd URL. M3+ uses the real FCM HTTP v1 endpoint.
+	fakefcmdURL := os.Getenv("BA_FAKECMD_URL")
+	if fakefcmdURL == "" {
+		fakefcmdURL = "http://fakefcmd:8820"
+	}
+	logger.Info("fcm target", "url", fakefcmdURL)
+
+	httpClient := &http.Client{Timeout: 10 * time.Second}
+
+	js := br.JS()
+	stream, err := js.Stream(ctx, "DELIVERIES")
+	if err != nil {
+		logger.Error("nats stream DELIVERIES", "err", err)
+		os.Exit(1)
+	}
+	consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
+		Name:          "deliverd-fcm",
+		Durable:       "deliverd-fcm",
+		FilterSubjects: []string{"deliveries.fcm.>"},
+		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, pool, httpClient, fakefcmdURL)
 
+	reg, _ := observability.NewRegistry("deliverd")
 	srv := httpserver.New(httpserver.Config{
 		Addr:          cfg.HTTPAddr,
 		ServiceName:   "deliverd",
@@ -57,8 +114,184 @@ func main() {
 			os.Exit(1)
 		}
 	}
+	runCancel()
+	time.Sleep(500 * time.Millisecond)
 	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, pool *postgres.Pool, httpClient *http.Client, fakefcmdURL string) {
+	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, pool, httpClient, fakefcmdURL)
+			if batch.Error() != nil {
+				logger.Warn("batch error", "err", batch.Error())
+				break
+			}
+		}
+	}
+}
+
+func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *postgres.Pool, httpClient *http.Client, fakefcmdURL string) {
+	var env deliveryEnvelope
+	if err := json.Unmarshal(m.Data(), &env); err != nil {
+		logger.Warn("malformed delivery envelope", "err", err, "subject", m.Subject())
+		_ = m.Ack() // poison message; we don't have a DLQ for malformed yet
+		return
+	}
+
+	// Parse out the alert_id and company_id from the inner alert JSON.
+	var alertHeader struct {
+		ID        string `json:"id"`
+		CompanyID string `json:"company_id"`
+		Title     string `json:"title"`
+		Body      string `json:"body"`
+		Data      map[string]string `json:"data"`
+		Category  string `json:"category"`
+		Severity  string `json:"severity"`
+	}
+	_ = json.Unmarshal(env.Alert, &alertHeader)
+
+	companyID := alertHeader.CompanyID
+	if companyID == "" {
+		// subject is "deliveries.fcm.<company_id>"
+		parts := strings.SplitN(m.Subject(), ".", 3)
+		if len(parts) == 3 {
+			companyID = parts[2]
+		}
+	}
+	if companyID == "" || env.FCMToken == "" || alertHeader.ID == "" {
+		logger.Warn("delivery envelope missing fields", "subject", m.Subject(), "company", companyID, "token", env.FCMToken != "", "alert_id", alertHeader.ID)
+		_ = m.Ack()
+		return
+	}
+
+	// Build the FCM HTTP v1 message body. The shape matches what
+	// real FCM expects, so the M3 swap is a no-op at this layer.
+	fcmBody := map[string]any{
+		"message": map[string]any{
+			"token": env.FCMToken,
+			"notification": map[string]any{
+				"title": alertHeader.Title,
+				"body":  alertHeader.Body,
+			},
+			"data": mergeData(alertHeader.Data, map[string]string{
+				"company_id":   companyID,
+				"alert_id":     alertHeader.ID,
+				"severity":     alertHeader.Severity,
+				"category":     alertHeader.Category,
+				"individual_id": env.IndividualID,
+				"locale":       env.Locale,
+				"deep_link":    fmt.Sprintf("broadannounce://alert/%s", alertHeader.ID),
+			}),
+			"android": map[string]any{
+				"priority": androidPriority(alertHeader.Severity),
+				"notification": map[string]any{
+					"sound":      androidSound(alertHeader.Category, alertHeader.Severity),
+					"channel_id": "alerts." + channelForSeverity(alertHeader.Severity),
+				},
+			},
+		},
+	}
+	body, _ := json.Marshal(fcmBody)
+
+	// POST to fakefcmd. M3: real FCM endpoint.
+	url := fakefcmdURL + "/v1/projects/fakefcmd/messages:send"
+	req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := httpClient.Do(req)
+	status := "failed"
+	lastErr := ""
+	if err != nil {
+		lastErr = err.Error()
+	} else {
+		defer resp.Body.Close()
+		respBody, _ := io.ReadAll(resp.Body)
+		if resp.StatusCode/100 == 2 {
+			status = "sent"
+		} else {
+			lastErr = fmt.Sprintf("status %d: %s", resp.StatusCode, string(respBody))
+		}
+	}
+
+	// Persist delivery row. M1: just insert. M3+: per-channel
+	// retry with exp backoff and DLQ on terminal failure.
+	_, dbErr := pool.Exec(ctx, `
+		INSERT INTO deliveries
+		    (alert_id, company_id, individual_id, channel, target, status, attempts, last_error, payload, sent_at)
+		VALUES ($1, $2, $3, $4, $5, $6, 1, NULLIF($7, ''), $8,
+		    CASE WHEN $6 = 'sent' THEN now() ELSE NULL END)
+	`, alertHeader.ID, companyID, env.IndividualID, fcmChannel, env.FCMToken, status, lastErr, json.RawMessage(m.Data()))
+	if dbErr != nil {
+		logger.Warn("delivery row insert", "err", dbErr)
+	}
+
+	logger.Info("delivery",
+		"alert_id", alertHeader.ID,
+		"company", companyID,
+		"individual", env.IndividualID,
+		"channel", fcmChannel,
+		"status", status,
+		"err", lastErr,
+	)
+	_ = m.Ack()
+}
+
+func mergeData(base, add map[string]string) map[string]string {
+	if base == nil {
+		base = map[string]string{}
+	}
+	for k, v := range add {
+		if _, ok := base[k]; !ok {
+			base[k] = v
+		}
+	}
+	return base
+}
+
+func androidPriority(sev string) string {
+	if sev == "critical" || sev == "inminent_colapse" {
+		return "HIGH"
+	}
+	return "NORMAL"
+}
+
+// androidSound picks a sound file. SPEC §4 says the Android app
+// uses data.category to pick siren_<category>.ogg; the per-severity
+// override for inminent_colapse wins.
+func androidSound(category, sev string) string {
+	if sev == "inminent_colapse" {
+		return "klaxon"
+	}
+	if category == "" {
+		return "default"
+	}
+	return "siren_" + category
+}
+
+func channelForSeverity(sev string) string {
+	switch sev {
+	case "critical":
+		return "critical"
+	case "inminent_colapse":
+		return "imminent"
+	case "warning":
+		return "warning"
+	default:
+		return "info"
+	}
+}

cmd/routerd/main.go

@@ -3,19 +3,31 @@
 // and enqueues one delivery per (individual, channel) to
 // deliveries.<channel>.<company_id> subjects.
 //
-// M0: connects to NATS, /health, /metrics. No business logic yet.
+// 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() {
@@ -36,10 +48,40 @@ func main() {
 		os.Exit(1)
 	}
 	defer br.Close()
-	logger.Info("nats connected")
 
-	reg, _ := observability.NewRegistry("routerd")
+	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",
@@ -57,8 +99,117 @@ func main() {
 			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

internal/routing/routing.go

@@ -0,0 +1,64 @@
+// Package routing is the recipient resolution + delivery enqueue step.
+// M1: broadcast mode — every alert to a company goes to every active
+// fcm_token of every active individual in that company. M2 replaces
+// this with the rules engine from SPEC §6 (subscriptions, opt-in,
+// quiet hours, routing rules, …).
+//
+// The interface is small: ResolveTokens returns the set of
+// (individual_id, fcm_token, locale) tuples that should receive
+// the alert. deliverd-fcm is the only consumer in M1.
+package routing
+
+import (
+	"context"
+	"fmt"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/postgres"
+)
+
+// Target is one device to deliver to.
+type Target struct {
+	IndividualID string
+	FCMToken     string
+	Locale       string
+}
+
+// Resolver looks up recipients for a (company, alert) pair.
+type Resolver struct {
+	pool *postgres.Pool
+}
+
+// New constructs a Resolver.
+func New(pool *postgres.Pool) *Resolver { return &Resolver{pool: pool} }
+
+// ResolveTokens returns every active FCM token for every active
+// individual in the given company. This is the M1 broadcast path.
+// M2 swaps this for the rules engine.
+//
+// One DB round-trip via a single join. Add a per-company limit
+// here if a malicious company can ever register 1M tokens.
+func (r *Resolver) ResolveTokens(ctx context.Context, companyID string) ([]Target, error) {
+	rows, err := r.pool.Query(ctx, `
+		SELECT i.id, t.token, COALESCE(t.locale, i.locale, 'en')
+		FROM individuals i
+		JOIN fcm_tokens t ON t.individual_id = i.id
+		WHERE i.company_id = $1
+		  AND i.status = 'active'
+		  AND t.status = 'active'
+		ORDER BY i.id, t.id
+	`, companyID)
+	if err != nil {
+		return nil, fmt.Errorf("resolve tokens: %w", err)
+	}
+	defer rows.Close()
+
+	var out []Target
+	for rows.Next() {
+		var t Target
+		if err := rows.Scan(&t.IndividualID, &t.FCMToken, &t.Locale); err != nil {
+			return nil, err
+		}
+		out = append(out, t)
+	}
+	return out, rows.Err()
+}