broad-announce/cmd/deliverd-fcm/main.go

// Command deliverd-fcm consumes deliveries.fcm.<company_id> subjects
// and posts each alert to the FCM HTTP v1 endpoint. M3 splits
// this out of the original `cmd/deliverd` (which is now two
// per-channel binaries: deliverd-fcm and deliverd-telegram).
// That gives us the option to scale, restart, and deploy each
// channel independently — your Q2 answer.
//
// M0: per-channel worker binary that connects to NATS, /health, /metrics.
// M1: deliverd-fcm — consumes deliveries.fcm.*, posts to fakefcmd,
//     records a row in Postgres per attempt.
// M3: rename only. No behavior change. The HTTP target URL is
//     BA_FAKECMD_URL (dev) or BA_FCM_BASE_URL (prod); M11+ will
//     add real FCM auth.
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"
)

// M2: 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"`
	Channel      string          `json:"channel"`
	Endpoint     string          `json:"endpoint"`
	Locale       string          `json:"locale,omitempty"`
}

func main() {
	cfg, err := config.LoadCommon("deliverd")
	if err != nil {
		os.Stderr.WriteString("config: " + err.Error() + "\n")
		os.Exit(1)
	}
	logger := observability.Init(cfg.Env, cfg.LogLevel, "deliverd")
	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()

	// 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",
		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)
	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"`
		// M6: dedupe_count. 0 or 1 means "first arrival",
		// ≥2 means this alert has been seen N times in the
		// sliding dedupe window. Native FCM clients (the
		// Android app) can display it or hide it via the
		// data map; the notification body is also suffixed
		// with ` (×N)` for clients that render the body
		// verbatim.
		DedupeCount uint32 `json:"dedupe_count"`
	}
	_ = 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.Endpoint == "" || alertHeader.ID == "" {
		logger.Warn("delivery envelope missing fields", "subject", m.Subject(), "company", companyID, "endpoint_present", env.Endpoint != "", "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.
	//
	// M6: dedupe_count flows through three surfaces:
	//   1. notification.title is suffixed with ` (×N)` when N>1
	//   2. notification.body is the source's pre-localized body
	//      verbatim; the title is where the count goes so the
	//      body isn't double-formatted.
	//   3. data.dedupe_count is the raw count for clients that
	//      want to render it themselves (e.g. an Android app
	//      that shows "×5" in a corner badge).
	notificationTitle := alertHeader.Title
	if alertHeader.DedupeCount > 1 {
		notificationTitle = fmt.Sprintf("%s (×%d)", alertHeader.Title, alertHeader.DedupeCount)
	}
	fcmBody := map[string]any{
		"message": map[string]any{
			"token": env.Endpoint,
			"notification": map[string]any{
				"title": notificationTitle,
				"body":  alertHeader.Body,
			},
			"data": mergeData(alertHeader.Data, map[string]string{
				"company_id":   companyID,
				"alert_id":     alertHeader.ID,
				"severity":     alertHeader.Severity,
				"category":     alertHeader.Category,
				"dedupe_count": fmt.Sprintf("%d", alertHeader.DedupeCount),
				"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.Endpoint, 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"
	}
}