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.
// M8: in-process retry with exp backoff (10 attempts, default)
//     and a DLQ insert on terminal failure. See SPEC §9.
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/dlq"
	"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/retry"
	"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,
		"max_attempts", cfg.DeliverdMaxAttempts,
		"retry_base_ms", cfg.DeliverdRetryBaseMs,
		"retry_max_ms", cfg.DeliverdRetryMaxMs,
		"retry_budget_ms", cfg.DeliverdRetryBudgetMs,
	)

	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}

	// M8: build the retry config from env. Defaults match
	// internal/retry.Default() so the .env.example can stay
	// sparse; ops can override per-deploy.
	retryCfg := retry.Config{
		MaxAttempts: cfg.DeliverdMaxAttempts,
		BaseDelay:   time.Duration(cfg.DeliverdRetryBaseMs) * time.Millisecond,
		MaxDelay:    time.Duration(cfg.DeliverdRetryMaxMs) * time.Millisecond,
		Budget:      time.Duration(cfg.DeliverdRetryBudgetMs) * time.Millisecond,
	}

	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, retryCfg)

	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, retryCfg retry.Config) {
	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, retryCfg)
			if batch.Error() != nil {
				logger.Warn("batch error", "err", batch.Error())
				break
			}
		}
	}
}

// handleOne processes a single delivery. M8 split the
// work into three steps:
//   1. Parse the envelope; if it's malformed, Ack and
//      bail (no DLQ for un-parseable data — there's no
//      payload to replay anyway).
//   2. Run the retry loop. Each attempt POSTs to the
//      target and writes a `deliveries` audit row. On
//      the final failure, write a `deliveries_dlq` row.
//   3. Ack the NATS message; the work is done one way
//      or another (sent, or durably parked in the DLQ).
func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *postgres.Pool, httpClient *http.Client, fakefcmdURL string, retryCfg retry.Config) {
	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"`
		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 once. Same shape
	// across attempts. M6 dedupe_count flow-through is
	// unchanged.
	notificationTitle := alertHeader.Title
	if alertHeader.DedupeCount > 1 {
		notificationTitle = fmt.Sprintf("%s (×%d)", alertHeader.Title, alertHeader.DedupeCount)
	}
	fcmBody, _ := json.Marshal(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),
				},
			},
		},
	})

	url := fakefcmdURL + "/v1/projects/fakefcmd/messages:send"

	// M8: retry loop. Each attempt: POST, persist a
	// deliveries row with status sent/failed and the
	// attempt counter, return the error to the helper.
	// On success, the loop exits early and we write one
	// final 'sent' row. On exhaustion, we write a DLQ
	// row and a final 'dlq' audit row.
	//
	// The ctx is the per-message context; we want a
	// per-attempt timeout, so we use a child ctx.
	var lastErr error
	var sent bool
	res := retry.Run(ctx, retryCfg, func(attemptCtx context.Context, attempt int) error {
		// Per-attempt timeout: 10s. The retry.Config
		// budget is the outer wall-clock cap.
		attemptCtx, cancel := context.WithTimeout(attemptCtx, 10*time.Second)
		defer cancel()

		req, _ := http.NewRequestWithContext(attemptCtx, "POST", url, bytes.NewReader(fcmBody))
		req.Header.Set("Content-Type", "application/json")
		resp, err := httpClient.Do(req)
		status := "failed"
		lastErrStr := ""
		if err != nil {
			lastErrStr = err.Error()
		} else {
			respBody, _ := io.ReadAll(resp.Body)
			_ = resp.Body.Close()
			if resp.StatusCode/100 == 2 {
				status = "sent"
				sent = true
			} else {
				lastErrStr = fmt.Sprintf("status %d: %s", resp.StatusCode, truncate(string(respBody), 200))
			}
		}

		// Per-attempt deliveries row. M1 pattern: one row
		// per attempt for the audit trail. The M8 status
		// enum is unchanged (pending|sent|failed|dlq).
		_, dbErr := pool.Exec(attemptCtx, `
			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, $7, NULLIF($8, ''), $9,
			    CASE WHEN $6 = 'sent' THEN now() ELSE NULL END)
		`, alertHeader.ID, companyID, env.IndividualID, fcmChannel, env.Endpoint, status, attempt, lastErrStr, json.RawMessage(m.Data()))
		if dbErr != nil {
			logger.Warn("delivery row insert", "err", dbErr, "attempt", attempt)
		}

		if status == "sent" {
			logger.Info("delivery sent",
				"alert_id", alertHeader.ID,
				"company", companyID,
				"individual", env.IndividualID,
				"channel", fcmChannel,
				"attempt", attempt,
			)
			return nil
		}
		// M8: surface the last error to the retry helper
		// so it logs and (on exhaustion) gets written to
		// the DLQ row. We use a PermanentError for 4xx
		// (other than 408/429) so the retry helper
		// short-circuits — FCM's 4xx means the token is
		// bad, retrying won't help.
		if resp != nil && resp.StatusCode >= 400 && resp.StatusCode < 500 && resp.StatusCode != 408 && resp.StatusCode != 429 {
			logger.Warn("permanent fcm error",
				"alert_id", alertHeader.ID,
				"status", resp.StatusCode,
				"err", lastErrStr,
				"attempt", attempt,
			)
			lastErr = &retry.PermanentError{Err: fmt.Errorf("fcm permanent %s", lastErrStr)}
			return lastErr
		}
		logger.Warn("delivery attempt failed",
			"alert_id", alertHeader.ID,
			"company", companyID,
			"channel", fcmChannel,
			"attempt", attempt,
			"err", lastErrStr,
		)
		lastErr = fmt.Errorf("%s", lastErrStr)
		return lastErr
	})

	if sent {
		_ = m.Ack()
		return
	}

	// All attempts exhausted (or permanent error). Write
	// a DLQ row so an operator can replay. Use the
	// original NATS subject so replay doesn't need to
	// re-derive it from the payload.
	dlqID, dlqErr := dlq.Write(ctx, pool, dlq.Entry{
		AlertID:         alertHeader.ID,
		CompanyID:       companyID,
		IndividualID:    env.IndividualID,
		Channel:         fcmChannel,
		Target:          env.Endpoint,
		OriginalSubject: m.Subject(),
		Attempts:        res.Attempts,
		LastError:       errString(lastErr),
		Payload:         json.RawMessage(m.Data()),
	})
	if dlqErr != nil {
		// The DLQ write itself failed. Log loud; we
		// still Ack the NATS message (the original
		// delivery is lost, but the live `deliveries`
		// audit row has the failure recorded). An
		// operator alert can be wired in M9.
		logger.Error("dlq insert FAILED",
			"err", dlqErr,
			"alert_id", alertHeader.ID,
			"company", companyID,
		)
	} else {
		logger.Warn("delivery → DLQ",
			"dlq_id", dlqID,
			"alert_id", alertHeader.ID,
			"company", companyID,
			"channel", fcmChannel,
			"attempts", res.Attempts,
			"err", errString(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"
	}
}

func errString(err error) string {
	if err == nil {
		return ""
	}
	return err.Error()
}

// truncate keeps a status-line response body to N bytes
// for the deliveries.last_error column. We never want a
// multi-MB error body in the audit row.
func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n] + "…"
}