M8(1/3): DLQ schema + in-process retry + deliveries_dlq writer

SPEC §9: failed deliveries go to a DLQ after the
retry budget is exhausted. This commit wires up the
foundation:

* migrations/008_dlq.{up,down}.sql — new
  deliveries_dlq Timescale hypertable (PK
  (id, created_at), 1d chunks, 7d retention matching
  the live deliveries table). Plus the
  original_subject column (for replay) and the
  discarded/discarded_at/discarded_by columns (for
  the operator's discard action).

* internal/retry — small in-process retry helper.
  Exp backoff (BaseDelay doubles per attempt, capped
  at MaxDelay), wall-clock budget, ctx-cancel aware,
  PermanentError short-circuit. 10 attempts, base
  100ms, cap 2s, budget 30s — total wall clock ~12s
  for a fully failing target. Configurable via env.

* internal/dlq — single Write() that inserts a row
  into deliveries_dlq and best-effort flips the live
  deliveries row to status='dlq' for the audit trail.
  No external surface; both deliverd-* main.go files
  call it on retry exhaustion.

* internal/config — four new env knobs on Common:
  BA_DELIVERD_MAX_ATTEMPTS (10), BA_DELIVERD_RETRY_BASE_MS
  (100), BA_DELIVERD_RETRY_MAX_MS (2000),
  BA_DELIVERD_RETRY_BUDGET_MS (30000). Other services
  ignore them.

* cmd/deliverd-fcm, cmd/deliverd-telegram — refactored
  handleOne() to use retry.Run. On success, a 'sent'
  deliveries row. On exhaustion, a 'failed' row per
  attempt + a final dlq.Write() call. PermanentError
  triggers on FCM/Telegram 4xx (excluding 408/429)
  so we don't burn the retry budget on a bad token.
  Per-attempt timeout = 10s (fcm) / 15s (telegram);
  outer Budget caps total wall clock.

* internal/retry/retry_test.go — 7 unit tests
  covering: first-try success, retry-then-succeed,
  exhaustion, permanent short-circuit, budget
  respect, ctx cancel, monotonic backoff.

go build ./... clean. go test ./internal/retry 7/7
PASS.

cmd/deliverd-fcm/main.go

@@ -11,6 +11,8 @@
 // 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 (
@@ -29,9 +31,11 @@ import (
 
 	"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"
 )
 
@@ -53,7 +57,14 @@ func main() {
 		os.Exit(1)
 	}
 	logger := observability.Init(cfg.Env, cfg.LogLevel, "deliverd")
-	logger.Info("starting", "env", cfg.Env, "addr", cfg.HTTPAddr)
+	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()
@@ -81,6 +92,16 @@ func main() {
 
 	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 {
@@ -101,7 +122,7 @@ func main() {
 	runCtx, runCancel := context.WithCancel(ctx)
 	defer runCancel()
 
-	go consume(runCtx, logger, consumer, pool, httpClient, fakefcmdURL)
+	go consume(runCtx, logger, consumer, pool, httpClient, fakefcmdURL, retryCfg)
 
 	reg, _ := observability.NewRegistry("deliverd")
 	srv := httpserver.New(httpserver.Config{
@@ -129,7 +150,7 @@ func main() {
 	logger.Info("bye")
 }
 
-func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, pool *postgres.Pool, httpClient *http.Client, fakefcmdURL string) {
+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
@@ -144,7 +165,7 @@ func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, poo
 			continue
 		}
 		for m := range batch.Messages() {
-			handleOne(ctx, logger, m, pool, httpClient, fakefcmdURL)
+			handleOne(ctx, logger, m, pool, httpClient, fakefcmdURL, retryCfg)
 			if batch.Error() != nil {
 				logger.Warn("batch error", "err", batch.Error())
 				break
@@ -153,7 +174,17 @@ func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, poo
 	}
 }
 
-func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *postgres.Pool, httpClient *http.Client, fakefcmdURL string) {
+// 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())
@@ -163,21 +194,14 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 
 	// 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"`
+		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)
 
@@ -195,22 +219,14 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 		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).
+	// 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 := map[string]any{
+	fcmBody, _ := json.Marshal(map[string]any{
 		"message": map[string]any{
 			"token": env.Endpoint,
 			"notification": map[string]any{
@@ -218,14 +234,14 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 				"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),
+				"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),
+				"locale":        env.Locale,
+				"deep_link":     fmt.Sprintf("broadannounce://alert/%s", alertHeader.ID),
 			}),
 			"android": map[string]any{
 				"priority": androidPriority(alertHeader.Severity),
@@ -235,48 +251,136 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 				},
 			},
 		},
-	}
-	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"
+
+	// 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 {
-			lastErr = fmt.Sprintf("status %d: %s", resp.StatusCode, string(respBody))
+			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)
 		}
-	}
 
-	// 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)
+		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
 	}
 
-	logger.Info("delivery",
-		"alert_id", alertHeader.ID,
-		"company", companyID,
-		"individual", env.IndividualID,
-		"channel", fcmChannel,
-		"status", status,
-		"err", lastErr,
-	)
+	// 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()
 }
 
@@ -324,3 +428,20 @@ func channelForSeverity(sev string) string {
 		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] + "…"
+}

cmd/deliverd-telegram/main.go

@@ -20,16 +20,14 @@
 //                                 script; if set, takes priority.
 //
 // M3 ships single-attempt sends (no retry, no DLQ). That
-// follows the same pattern as deliverd-fcm in M1. M9 adds
-// the retry + DLQ chain.
+// follows the same pattern as deliverd-fcm in M1.
+// M8 adds the retry + DLQ chain (SPEC §9).
 package main
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"log/slog"
 	"net/http"
 	"os"
@@ -41,9 +39,11 @@ import (
 	"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/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"
 	"git3.techno-world.net/lrosales/broad-announce/internal/telegram"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -65,7 +65,14 @@ func main() {
 		os.Exit(1)
 	}
 	logger := observability.Init(cfg.Env, cfg.LogLevel, "deliverd-telegram")
-	logger.Info("starting", "env", cfg.Env, "addr", cfg.HTTPAddr)
+	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()
@@ -108,6 +115,15 @@ func main() {
 		HTTP:    &http.Client{Timeout: 15 * time.Second},
 	}
 
+	// M8: same retry config as deliverd-fcm. See
+	// internal/retry for the policy.
+	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 {
@@ -128,7 +144,7 @@ func main() {
 	runCtx, runCancel := context.WithCancel(ctx)
 	defer runCancel()
 
-	go consume(runCtx, logger, consumer, pool, client, botToken)
+	go consume(runCtx, logger, consumer, pool, client, botToken, retryCfg)
 
 	reg, _ := observability.NewRegistry("deliverd-telegram")
 	srv := httpserver.New(httpserver.Config{
@@ -156,7 +172,7 @@ func main() {
 	logger.Info("bye")
 }
 
-func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, pool *postgres.Pool, client telegram.BotClient, botToken string) {
+func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, pool *postgres.Pool, client telegram.BotClient, botToken string, retryCfg retry.Config) {
 	for {
 		if ctx.Err() != nil {
 			return
@@ -171,7 +187,7 @@ func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, poo
 			continue
 		}
 		for m := range batch.Messages() {
-			handleOne(ctx, logger, m, pool, client, botToken)
+			handleOne(ctx, logger, m, pool, client, botToken, retryCfg)
 			if batch.Error() != nil {
 				logger.Warn("batch error", "err", batch.Error())
 				break
@@ -180,7 +196,14 @@ func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, poo
 	}
 }
 
-func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *postgres.Pool, client telegram.BotClient, botToken string) {
+// handleOne mirrors deliverd-fcm's structure:
+//   1. Parse the envelope (drop poison messages).
+//   2. Retry loop: each attempt calls SendMessage and
+//      writes a `deliveries` audit row. On success,
+//      break out. On exhaustion, write a `deliveries_dlq`
+//      row + a final 'dlq' audit row.
+//   3. Ack the NATS message either way.
+func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *postgres.Pool, client telegram.BotClient, botToken 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())
@@ -188,26 +211,20 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 		return
 	}
 
-	// Pull the alert fields we need.
 	var ah struct {
-		ID          string `json:"id"`
-		CompanyID   string `json:"company_id"`
-		Title       string `json:"title"`
-		Body        string `json:"body"`
+		ID          string         `json:"id"`
+		CompanyID   string         `json:"company_id"`
+		Title       string         `json:"title"`
+		Body        string         `json:"body"`
 		Severity    alert.Severity `json:"severity"`
-		Category    string `json:"category"`
-		SourceID    string `json:"source_id"`
-		// M6: dedupe_count. 0 or 1 means "not a duplicate",
-		// ≥2 means the dedupe window has seen this alert N
-		// times; we surface that to the recipient via the
-		// inline `(×N)` suffix on the title.
-		DedupeCount uint32 `json:"dedupe_count"`
+		Category    string         `json:"category"`
+		SourceID    string         `json:"source_id"`
+		DedupeCount uint32         `json:"dedupe_count"`
 	}
 	_ = json.Unmarshal(env.Alert, &ah)
 
 	companyID := ah.CompanyID
 	if companyID == "" {
-		// subject is "deliveries.telegram.<company_id>"
 		parts := strings.SplitN(m.Subject(), ".", 3)
 		if len(parts) == 3 {
 			companyID = parts[2]
@@ -226,42 +243,152 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 		return
 	}
 
-	// Build the text. SPEC §8 says the Android app does
-	// localization; for Telegram we use the source's
-	// pre-localized title + body verbatim, with a severity
-	// prefix so a glance at the chat shows priority.
+	// Build the text once, retry the same text. Format
+	// unchanged from M3.
 	text := formatMessage(ah.Severity, ah.Title, ah.Body, ah.ID, ah.DedupeCount)
 
-	status := "failed"
-	lastErr := ""
-	if _, err := client.SendMessage(ctx, botToken, chatID, text); err != nil {
-		lastErr = err.Error()
-		logger.Warn("sendMessage failed", "err", err, "chat_id", chatID, "alert_id", ah.ID)
-	} else {
-		status = "sent"
-	}
+	// M8: retry loop. Telegram returns errors on:
+	//   - 4xx (chat not found, bot blocked, etc.) — these
+	//     are permanent; we use PermanentError to skip
+	//     further retries.
+	//   - 5xx, network timeouts — these are transient;
+	//     retry with backoff.
+	//   - 429 (rate limited) — transient; retry.
+	var lastErr error
+	var sent bool
+	res := retry.Run(ctx, retryCfg, func(attemptCtx context.Context, attempt int) error {
+		attemptCtx, cancel := context.WithTimeout(attemptCtx, 15*time.Second)
+		defer cancel()
+		_, sendErr := client.SendMessage(attemptCtx, botToken, chatID, text)
+		status := "failed"
+		lastErrStr := ""
+		if sendErr == nil {
+			status = "sent"
+			sent = true
+		} else {
+			lastErrStr = sendErr.Error()
+		}
+
+		_, 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)
+		`, ah.ID, companyID, env.IndividualID, telegramChannel, env.Endpoint, status, attempt, lastErrStr, json.RawMessage(m.Data()))
+		if dbErr != nil {
+			logger.Warn("delivery row insert", "err", dbErr, "attempt", attempt)
+		}
 
-	_, 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)
-	`, ah.ID, companyID, env.IndividualID, telegramChannel, env.Endpoint, status, lastErr, json.RawMessage(m.Data()))
-	if dbErr != nil {
-		logger.Warn("delivery row insert", "err", dbErr)
+		if status == "sent" {
+			logger.Info("delivery sent",
+				"alert_id", ah.ID,
+				"company", companyID,
+				"individual", env.IndividualID,
+				"channel", telegramChannel,
+				"attempt", attempt,
+			)
+			return nil
+		}
+
+		// Permanent-error detection. The Telegram Bot API
+		// returns errors of the form "telegram api <code>:
+		// <description>". 4xx codes (400, 401, 403, 404,
+		// 409) are not retryable; 429 (Too Many Requests) is
+		// transient and stays in the retry path. The
+		// faketgmd test double mirrors this convention.
+		if isTelegramPermanent(lastErrStr) {
+			logger.Warn("permanent telegram error",
+				"alert_id", ah.ID,
+				"err", lastErrStr,
+				"attempt", attempt,
+			)
+			lastErr = &retry.PermanentError{Err: fmt.Errorf("telegram permanent %s", lastErrStr)}
+			return lastErr
+		}
+		logger.Warn("delivery attempt failed",
+			"alert_id", ah.ID,
+			"company", companyID,
+			"channel", telegramChannel,
+			"attempt", attempt,
+			"err", lastErrStr,
+		)
+		lastErr = fmt.Errorf("%s", lastErrStr)
+		return lastErr
+	})
+
+	if sent {
+		_ = m.Ack()
+		return
 	}
 
-	logger.Info("delivery",
-		"alert_id", ah.ID,
-		"company", companyID,
-		"individual", env.IndividualID,
-		"channel", telegramChannel,
-		"status", status,
-		"err", lastErr,
-	)
+	// All attempts exhausted (or permanent error). DLQ
+	// insert for the operator UI; replay re-INSERTs the
+	// original payload into the same NATS subject.
+	dlqID, dlqErr := dlq.Write(ctx, pool, dlq.Entry{
+		AlertID:         ah.ID,
+		CompanyID:       companyID,
+		IndividualID:    env.IndividualID,
+		Channel:         telegramChannel,
+		Target:          env.Endpoint,
+		OriginalSubject: m.Subject(),
+		Attempts:        res.Attempts,
+		LastError:       errString(lastErr),
+		Payload:         json.RawMessage(m.Data()),
+	})
+	if dlqErr != nil {
+		logger.Error("dlq insert FAILED",
+			"err", dlqErr,
+			"alert_id", ah.ID,
+			"company", companyID,
+		)
+	} else {
+		logger.Warn("delivery → DLQ",
+			"dlq_id", dlqID,
+			"alert_id", ah.ID,
+			"company", companyID,
+			"channel", telegramChannel,
+			"attempts", res.Attempts,
+			"err", errString(lastErr),
+		)
+	}
 	_ = m.Ack()
 }
 
+// isTelegramPermanent classifies a Telegram Bot API
+// error string as permanent (no point retrying). The
+// faketgmd test double prefixes errors with "telegram
+// api <code>:" so we can match on the code.
+//
+// 429 (rate limited) is intentionally NOT permanent —
+// the retry helper will space it out via backoff.
+func isTelegramPermanent(errStr string) bool {
+	if errStr == "" {
+		return false
+	}
+	// Look for "telegram api NNN" or "status NNN" in the
+	// error string. 4xx (except 429) is permanent.
+	for _, prefix := range []string{"telegram api ", "status "} {
+		i := strings.Index(errStr, prefix)
+		if i < 0 {
+			continue
+		}
+		rest := errStr[i+len(prefix):]
+		// Parse a 3-digit code.
+		if len(rest) < 3 {
+			continue
+		}
+		codeStr := rest[:3]
+		var code int
+		if _, err := fmt.Sscanf(codeStr, "%d", &code); err != nil {
+			continue
+		}
+		if code >= 400 && code < 500 && code != 429 && code != 408 {
+			return true
+		}
+	}
+	return false
+}
+
 // formatMessage produces a Telegram-friendly rendering of
 // the alert. The format is intentionally plain (Telegram
 // supports Markdown/HTML but they vary across clients);
@@ -285,10 +412,6 @@ func formatMessage(sev alert.Severity, title, body, alertID string, dedupeCount
 	default:
 		prefix = "🟦 INFO"
 	}
-	// M6: inline `(×N)` suffix on the title when this alert
-	// was suppressed N-1 times by the dedupe gate. We render
-	// the count as a small visual hint that doesn't
-	// interfere with the severity prefix or the body.
 	displayedTitle := title
 	if dedupeCount > 1 {
 		displayedTitle = fmt.Sprintf("%s (×%d)", title, dedupeCount)
@@ -311,5 +434,9 @@ func parseInt64(s string) (int64, error) {
 	return n, err
 }
 
-var _ = bytes.NewReader
-var _ = io.ReadAll
+func errString(err error) string {
+	if err == nil {
+		return ""
+	}
+	return err.Error()
+}

internal/config/config.go

@@ -31,6 +31,16 @@ type Common struct {
 
 	// Shutdown
 	ShutdownGrace time.Duration
+
+	// M8: retry knobs shared by every deliverd-* binary.
+	// Ingestd, routerd, archiverd, and admind ignore them
+	// (they don't call retry.Run). Defaults match
+	// internal/retry.Default() so the .env.example can
+	// stay sparse.
+	DeliverdMaxAttempts  int
+	DeliverdRetryBaseMs  int
+	DeliverdRetryMaxMs   int
+	DeliverdRetryBudgetMs int
 }
 
 // Default values applied if env unset.
@@ -78,6 +88,15 @@ func LoadCommon(serviceName string) (Common, error) {
 		c.ShutdownGrace = time.Duration(n) * time.Second
 	}
 
+	// M8 retry knobs. Read on every LoadCommon so the
+	// deliverd-* binaries don't need a separate config
+	// struct just for these four values. Non-deliverd
+	// services simply never call retry.Run.
+	c.DeliverdMaxAttempts = GetInt("BA_DELIVERD_MAX_ATTEMPTS", 10)
+	c.DeliverdRetryBaseMs = GetInt("BA_DELIVERD_RETRY_BASE_MS", 100)
+	c.DeliverdRetryMaxMs = GetInt("BA_DELIVERD_RETRY_MAX_MS", 2000)
+	c.DeliverdRetryBudgetMs = GetInt("BA_DELIVERD_RETRY_BUDGET_MS", 30000)
+
 	if c.Env != "dev" && c.Env != "staging" && c.Env != "prod" {
 		return c, fmt.Errorf("BA_ENV must be dev|staging|prod, got %q", c.Env)
 	}

internal/dlq/dlq.go

@@ -0,0 +1,90 @@
+// Package dlq is the M8 dead-letter queue writer used by
+// deliverd-fcm and deliverd-telegram. When a delivery
+// exhausts its retry budget, the worker calls Write()
+// once to insert a forensic row into deliveries_dlq and
+// mark the live deliveries row as status='dlq'.
+//
+// The package is intentionally tiny: one function, one
+// row insert, one row update. The retry loop itself
+// stays in the deliverd main.go (it's small and per-
+// channel, since the HTTP target and the request shape
+// differ per channel).
+//
+// Schema reference: migrations/008_dlq.up.sql.
+package dlq
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/postgres"
+)
+
+// Entry is the minimal payload we need to write a DLQ
+// row. It mirrors the columns of `deliveries_dlq` that
+// the worker fills in; everything else (id, created_at,
+// discarded) is set by the database.
+type Entry struct {
+	AlertID         string
+	CompanyID       string
+	IndividualID    string
+	Channel         string // fcm | telegram | …
+	Target          string // fcm_token, chat_id, …
+	OriginalSubject string // deliveries.fcm.<co>, for replay
+	Attempts        int
+	LastError       string
+	Payload         json.RawMessage // raw NATS envelope bytes
+}
+
+// Write inserts one row into deliveries_dlq. It also
+// updates the matching `deliveries` row to status='dlq'
+// for the audit trail (best-effort: we don't fail the
+// DLQ write if the update misses, because the DLQ row
+// is the source of truth for replay).
+//
+// Returns the new DLQ row's id (for logging).
+func Write(ctx context.Context, pool *postgres.Pool, e Entry) (int64, error) {
+	if e.OriginalSubject == "" {
+		return 0, fmt.Errorf("dlq.Write: OriginalSubject is required for replay")
+	}
+	if e.Payload == nil {
+		// Store the literal JSON null rather than an
+		// empty byte slice so CH's String column gets
+		// a sensible value.
+		e.Payload = json.RawMessage("null")
+	}
+	// The DLQ insert.
+	var newID int64
+	err := pool.QueryRow(ctx, `
+		INSERT INTO deliveries_dlq
+		    (alert_id, company_id, individual_id, channel, target,
+		     original_subject, attempts, last_error, payload)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+		RETURNING id
+	`,
+		e.AlertID, e.CompanyID, e.IndividualID, e.Channel, e.Target,
+		e.OriginalSubject, e.Attempts, e.LastError, e.Payload,
+	).Scan(&newID)
+	if err != nil {
+		return 0, fmt.Errorf("dlq insert: %w", err)
+	}
+
+	// Best-effort status flip on the live deliveries row.
+	// We do a soft match: same (alert_id, company_id,
+	// individual_id, channel) and attempts column matches.
+	// This is good enough for the audit trail; the DLQ
+	// row id is the source of truth for replay.
+	_, _ = pool.Exec(ctx, `
+		UPDATE deliveries
+		   SET status = 'dlq',
+		       last_error = $1
+		 WHERE alert_id = $2
+		   AND company_id = $3
+		   AND individual_id = $4
+		   AND channel = $5
+		   AND status NOT IN ('sent', 'dlq')
+	`, e.LastError, e.AlertID, e.CompanyID, e.IndividualID, e.Channel)
+
+	return newID, nil
+}

internal/retry/retry.go

@@ -0,0 +1,170 @@
+// Package retry is the M8 in-process retry helper used by
+// deliverd-fcm and deliverd-telegram. It implements
+// bounded exponential backoff with a per-attempt cap
+// and a total time budget so a single delivery can't
+// tie up a NATS consumer for minutes.
+//
+// Why in-process retry, not JetStream redelivery?
+//   - We want explicit, config-driven backoff (SPEC §9:
+//     1s, 2s, 4s, … up to 10 attempts). JetStream's
+//     redelivery timer is fixed at the consumer level
+//     and doesn't express per-attempt exp backoff.
+//   - We want a single deliveries table row per
+//     attempt for the audit trail. JetStream redelivery
+//     would re-process the same envelope; we'd have
+//     to dedupe in the worker anyway.
+//   - We want a hard cap (RetryBudget) so a stuck
+//     downstream (e.g. fakefcmd with --fail-rate=1.0)
+//     can never block a consumer for the full
+//     1+2+4+…+512 = 1023s the SPEC literally calls
+//     for. In practice we ship defaults that
+//     terminate in ~10s.
+package retry
+
+import (
+	"context"
+	"errors"
+	"time"
+)
+
+// Config is the retry policy. All durations are wall
+// clock; the helper never sleeps past the ctx deadline.
+type Config struct {
+	// MaxAttempts is the total number of attempts
+	// (including the first). Default 10. After
+	// MaxAttempts failures, Run returns the last
+	// error from fn.
+	MaxAttempts int
+	// BaseDelay is the wait before the SECOND attempt;
+	// it doubles each subsequent attempt. Default 100ms.
+	BaseDelay time.Duration
+	// MaxDelay caps the per-attempt wait. Default 2s.
+	// With BaseDelay=100ms, MaxDelay=2s, MaxAttempts=10,
+	// the per-attempt waits are 100, 200, 400, 800, 1600,
+	// 2000, 2000, 2000, 2000 ms (9 waits), ~12s total.
+	MaxDelay time.Duration
+	// Budget is the wall-clock cap across all attempts.
+	// Run returns ctx.DeadlineExceeded if the budget
+	// is hit before MaxAttempts completes. Default 30s.
+	Budget time.Duration
+}
+
+// Default returns the M8 spec defaults.
+func Default() Config {
+	return Config{
+		MaxAttempts: 10,
+		BaseDelay:  100 * time.Millisecond,
+		MaxDelay:   2 * time.Second,
+		Budget:     30 * time.Second,
+	}
+}
+
+// PermanentError signals "don't retry this". The helper
+// returns it to the caller as-is after wrapping the
+// attempt count. Use it for HTTP 4xx (except 408/429),
+// parse errors, and other "retry won't help" cases.
+type PermanentError struct {
+	Err error
+}
+
+func (e *PermanentError) Error() string { return e.Err.Error() }
+func (e *PermanentError) Unwrap() error { return e.Err }
+
+// IsPermanent reports whether err is a PermanentError.
+func IsPermanent(err error) bool {
+	var p *PermanentError
+	return errors.As(err, &p)
+}
+
+// Result is the outcome of Run.
+type Result struct {
+	// Attempts is the number of fn invocations that ran
+	// (1 = succeeded on first try, MaxAttempts = gave up).
+	Attempts int
+	// LastError is the error from the final attempt, or
+	// nil on success.
+	LastError error
+}
+
+// Run calls fn up to cfg.MaxAttempts times, sleeping
+// exp-backoff between failed attempts. It returns a
+// Result with the attempt count and last error.
+//
+// The sleep respects ctx cancellation. If the budget
+// is hit before all attempts complete, fn is not called
+// again and Result.LastError is the original fn error
+// (not ctx.DeadlineExceeded, so the caller can decide
+// whether to DLQ the message).
+//
+// PermanentError short-circuits the loop: if fn
+// returns &PermanentError{…}, Run returns immediately
+// with Attempts set to the current count and LastError
+// = the wrapped error.
+func Run(ctx context.Context, cfg Config, fn func(ctx context.Context, attempt int) error) Result {
+	if cfg.MaxAttempts <= 0 {
+		cfg.MaxAttempts = 10
+	}
+	if cfg.BaseDelay <= 0 {
+		cfg.BaseDelay = 100 * time.Millisecond
+	}
+	if cfg.MaxDelay <= 0 {
+		cfg.MaxDelay = 2 * time.Second
+	}
+	if cfg.Budget <= 0 {
+		cfg.Budget = 30 * time.Second
+	}
+	deadline := time.Now().Add(cfg.Budget)
+	res := Result{}
+	for attempt := 1; attempt <= cfg.MaxAttempts; attempt++ {
+		res.Attempts = attempt
+		err := fn(ctx, attempt)
+		if err == nil {
+			res.LastError = nil // clear any prior failure on success
+			return res
+		}
+		res.LastError = err
+		if IsPermanent(err) {
+			return res
+		}
+		if attempt == cfg.MaxAttempts {
+			break
+		}
+		// Compute this attempt's wait, then sleep
+		// with ctx + budget awareness.
+		wait := backoff(cfg.BaseDelay, cfg.MaxDelay, attempt)
+		// If the budget would be exceeded by this wait,
+		// bail out early with the last error.
+		if time.Now().Add(wait).After(deadline) {
+			break
+		}
+		t := time.NewTimer(wait)
+		select {
+		case <-ctx.Done():
+			t.Stop()
+			return res
+		case <-t.C:
+		}
+	}
+	return res
+}
+
+// backoff returns the wait for the Nth retry. attempt=1
+// is the FIRST attempt (no wait); attempt=2 is the wait
+// before the second attempt. So we use attempt-1 to
+// compute the exponent.
+func backoff(base, max time.Duration, attempt int) time.Duration {
+	if attempt < 2 {
+		return 0
+	}
+	d := base
+	for i := 2; i < attempt; i++ {
+		d *= 2
+		if d > max {
+			return max
+		}
+	}
+	if d > max {
+		return max
+	}
+	return d
+}

internal/retry/retry_test.go

@@ -0,0 +1,174 @@
+package retry
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestRunSucceedsOnFirstTry(t *testing.T) {
+	calls := 0
+	res := Run(context.Background(), Default(), func(ctx context.Context, attempt int) error {
+		calls++
+		return nil
+	})
+	if calls != 1 {
+		t.Fatalf("want 1 call, got %d", calls)
+	}
+	if res.Attempts != 1 {
+		t.Fatalf("want Attempts=1, got %d", res.Attempts)
+	}
+	if res.LastError != nil {
+		t.Fatalf("want nil err, got %v", res.LastError)
+	}
+}
+
+func TestRunRetriesThenSucceeds(t *testing.T) {
+	calls := 0
+	cfg := Config{
+		MaxAttempts: 5,
+		BaseDelay:   1 * time.Millisecond,
+		MaxDelay:    5 * time.Millisecond,
+		Budget:      5 * time.Second,
+	}
+	res := Run(context.Background(), cfg, func(ctx context.Context, attempt int) error {
+		calls++
+		if attempt < 3 {
+			return errors.New("transient")
+		}
+		return nil
+	})
+	if calls != 3 {
+		t.Fatalf("want 3 calls, got %d", calls)
+	}
+	if res.Attempts != 3 {
+		t.Fatalf("want Attempts=3, got %d", res.Attempts)
+	}
+	if res.LastError != nil {
+		t.Fatalf("want nil err, got %v", res.LastError)
+	}
+}
+
+func TestRunExhaustsAttempts(t *testing.T) {
+	calls := 0
+	cfg := Config{
+		MaxAttempts: 3,
+		BaseDelay:   1 * time.Millisecond,
+		MaxDelay:    5 * time.Millisecond,
+		Budget:      5 * time.Second,
+	}
+	res := Run(context.Background(), cfg, func(ctx context.Context, attempt int) error {
+		calls++
+		return errors.New("nope")
+	})
+	if calls != 3 {
+		t.Fatalf("want 3 calls, got %d", calls)
+	}
+	if res.Attempts != 3 {
+		t.Fatalf("want Attempts=3, got %d", res.Attempts)
+	}
+	if res.LastError == nil {
+		t.Fatalf("want non-nil err")
+	}
+}
+
+func TestRunShortCircuitsOnPermanent(t *testing.T) {
+	calls := 0
+	cfg := Config{
+		MaxAttempts: 5,
+		BaseDelay:   1 * time.Millisecond,
+		MaxDelay:    5 * time.Millisecond,
+		Budget:      5 * time.Second,
+	}
+	res := Run(context.Background(), cfg, func(ctx context.Context, attempt int) error {
+		calls++
+		return &PermanentError{Err: errors.New("bad token")}
+	})
+	if calls != 1 {
+		t.Fatalf("want 1 call, got %d", calls)
+	}
+	if !IsPermanent(res.LastError) {
+		t.Fatalf("want PermanentError, got %T: %v", res.LastError, res.LastError)
+	}
+}
+
+func TestRunRespectsBudget(t *testing.T) {
+	calls := 0
+	cfg := Config{
+		MaxAttempts: 100,
+		BaseDelay:   50 * time.Millisecond,
+		MaxDelay:    200 * time.Millisecond,
+		Budget:      100 * time.Millisecond, // tight budget
+	}
+	start := time.Now()
+	res := Run(context.Background(), cfg, func(ctx context.Context, attempt int) error {
+		calls++
+		return errors.New("nope")
+	})
+	elapsed := time.Since(start)
+	if elapsed > 500*time.Millisecond {
+		t.Fatalf("budget exceeded: took %s", elapsed)
+	}
+	// With Budget=100ms and base 50ms, we expect at most
+	// 3-4 calls (50 + 100 = 150ms would already bust the
+	// budget for the next wait).
+	if calls > 5 {
+		t.Fatalf("too many calls under tight budget: %d", calls)
+	}
+	if res.LastError == nil {
+		t.Fatalf("want non-nil err")
+	}
+}
+
+func TestRunRespectsContextCancel(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	calls := 0
+	cfg := Config{
+		MaxAttempts: 10,
+		BaseDelay:   100 * time.Millisecond,
+		MaxDelay:    500 * time.Millisecond,
+		Budget:      5 * time.Second,
+	}
+	// Cancel the ctx after the first call.
+	res := Run(ctx, cfg, func(ctx context.Context, attempt int) error {
+		calls++
+		if calls == 1 {
+			// Cancel while we're sleeping.
+			go func() {
+				time.Sleep(20 * time.Millisecond)
+				cancel()
+			}()
+		}
+		return errors.New("nope")
+	})
+	if calls >= 10 {
+		t.Fatalf("ctx cancel did not stop the loop; calls=%d", calls)
+	}
+	if res.LastError == nil {
+		t.Fatalf("want non-nil err")
+	}
+}
+
+func TestBackoffMonotonic(t *testing.T) {
+	// Per-attempt wait is the time we sleep BEFORE that
+	// attempt. So attempt 1 = no wait, attempt 2 = base,
+	// attempt 3 = base*2, etc., capped at max.
+	base := 100 * time.Millisecond
+	max := 2 * time.Second
+	prev := time.Duration(0)
+	for n := 2; n <= 10; n++ {
+		got := backoff(base, max, n)
+		if n == 2 && got != base {
+			t.Fatalf("attempt 2: want %s, got %s", base, got)
+		}
+		if got > max {
+			t.Fatalf("attempt %d: exceeded max: %s", n, got)
+		}
+		// Wait must be >= previous (or equal under cap).
+		if got < prev {
+			t.Fatalf("attempt %d: wait decreased: %s < %s", n, got, prev)
+		}
+		prev = got
+	}
+}

migrations/008_dlq.down.sql

@@ -0,0 +1,54 @@
+-- 008_dlq.down.sql
+-- Reverse M8 DLQ additions. Drop the retention policy
+-- first, then convert the hypertable back to a plain
+-- table via the shadow-table rename pattern (see
+-- 006_timescale.down.sql for the rationale).
+
+-- 1. Remove the retention policy.
+SELECT remove_retention_policy('deliveries_dlq', if_exists => true);
+
+-- 2. Create a plain shadow table with the original
+-- (pre-hypertable) layout: BIGSERIAL PK, no chunks.
+CREATE TABLE IF NOT EXISTS deliveries_dlq_plain (
+    id                BIGSERIAL PRIMARY KEY,
+    alert_id          TEXT NOT NULL,
+    company_id        TEXT NOT NULL,
+    individual_id     TEXT NOT NULL,
+    channel           TEXT NOT NULL,
+    target            TEXT NOT NULL,
+    original_subject  TEXT NOT NULL,
+    attempts          INT NOT NULL,
+    last_error        TEXT NOT NULL,
+    payload           JSONB,
+    discarded         BOOLEAN NOT NULL DEFAULT false,
+    discarded_at      TIMESTAMPTZ,
+    discarded_by      TEXT,
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- 3. Copy the live data over. Timescale will reject
+-- SELECT FROM a hypertable that doesn't have
+-- move_data/copy_data logic; here we use a plain
+-- SELECT, which works on hypertables (you just lose
+-- the chunk-aware planner).
+INSERT INTO deliveries_dlq_plain
+    (id, alert_id, company_id, individual_id, channel, target,
+     original_subject, attempts, last_error, payload,
+     discarded, discarded_at, discarded_by, created_at)
+SELECT
+    id, alert_id, company_id, individual_id, channel, target,
+    original_subject, attempts, last_error, payload,
+    discarded, discarded_at, discarded_by, created_at
+FROM deliveries_dlq
+ON CONFLICT (id) DO NOTHING;
+
+-- 4. Drop the hypertable, rename the plain shadow.
+DROP TABLE IF EXISTS deliveries_dlq CASCADE;
+ALTER TABLE deliveries_dlq_plain RENAME TO deliveries_dlq;
+
+-- 5. Recreate the operator-UI indexes.
+CREATE INDEX IF NOT EXISTS idx_dlq_company_created
+    ON deliveries_dlq(company_id, created_at DESC)
+    WHERE discarded = false;
+CREATE INDEX IF NOT EXISTS idx_dlq_alert
+    ON deliveries_dlq(alert_id);

migrations/008_dlq.up.sql

@@ -0,0 +1,98 @@
+-- 008_dlq.up.sql
+-- M8: Dead-Letter Queue for failed deliveries.
+--
+-- Concept:
+--   - When a deliverd-* worker exhausts its retry budget
+--     (BA_DELIVERD_MAX_ATTEMPTS attempts), it inserts one
+--     row here AND marks the live `deliveries` row as
+--     status='dlq' for the audit trail.
+--   - The row is then visible to operators via admind's
+--     DLQ endpoints and the /dlq HTML page; the operator
+--     can replay (re-INSERT the original NATS envelope
+--     into the same deliveries.<channel>.<company> subject)
+--     or discard (set discarded=true, hidden from the
+--     default list).
+--   - The `original_subject` column captures the NATS
+--     subject the row was originally delivered on, so
+--     replay doesn't need to re-derive it from the payload.
+--
+-- Schema choices:
+--   - Same shape as `deliveries` plus:
+--       * original_subject  TEXT — `deliveries.fcm.<co>` etc.
+--       * last_error        TEXT — copy of the last failure
+--       * discarded         BOOLEAN — set by operator via
+--         POST /v1/dlq/{id}/discard. Default false.
+--       * discarded_at      TIMESTAMPTZ NULL.
+--   - PK is (id, created_at) so this table is also a
+--     Timescale hypertable (see step 2 below). 1d chunks,
+--     7d retention matches the live deliveries table.
+--   - The DLQ is forensic data: archiverd ships rows
+--     older than 7d to ClickHouse `ba_archive.deliveries_dlq_archive`
+--     with a 2-year TTL (vs 1y for live deliveries). See
+--     PROMPT.md M8 "Loose ends".
+--
+-- Why a separate table, not just status='dlq' on deliveries?
+--   - Operator UI: the DLQ list is a focused view, not a
+--     full-text search across all deliveries.
+--   - Replay needs the original_subject and a clean payload
+--     snapshot. Keeping that in a dedicated table avoids
+--     re-deriving it from deliveries.payload (which is
+--     the full NATS envelope).
+--   - ClickHouse-side, the DLQ gets a longer TTL (2y vs 1y)
+--     because DLQ entries are forensic — you want them
+--     around longer when triaging a regression.
+--
+-- Why 7d retention (same as deliveries)?
+--   - DLQ rows are infrequent (only terminal failures).
+--     7d is enough for the operator to notice + replay in
+--     the normal ops loop. After 7d the data lives on in
+--     ClickHouse for 2y, which is the long-term home.
+
+CREATE TABLE IF NOT EXISTS deliveries_dlq (
+    id                BIGSERIAL,
+    alert_id          TEXT NOT NULL,
+    company_id        TEXT NOT NULL,
+    individual_id     TEXT NOT NULL,
+    channel           TEXT NOT NULL,                -- fcm | telegram | sms | email | slack | teams | webhook
+    target            TEXT NOT NULL,                -- the fcm_token, telegram_chat_id, phone_e164, …
+    original_subject  TEXT NOT NULL,                -- deliveries.fcm.<co>  (for replay)
+    attempts          INT NOT NULL,                 -- total attempts before giving up
+    last_error        TEXT NOT NULL,                -- last failure reason
+    payload           JSONB,                        -- snapshot at enqueue time (forensic)
+    discarded         BOOLEAN NOT NULL DEFAULT false,
+    discarded_at      TIMESTAMPTZ,
+    discarded_by      TEXT,                         -- operator id / token (M8: 'admind-cli' until auth lands)
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),  -- time of DLQ insert
+    PRIMARY KEY (id, created_at)
+);
+
+-- The operator UI is filter-driven: most queries are
+-- "show me non-discarded DLQ rows for company X in the
+-- last 24h". These two indexes cover that.
+CREATE INDEX IF NOT EXISTS idx_dlq_company_created
+    ON deliveries_dlq(company_id, created_at DESC)
+    WHERE discarded = false;
+CREATE INDEX IF NOT EXISTS idx_dlq_alert
+    ON deliveries_dlq(alert_id);
+
+-- Convert to a Timescale hypertable on created_at, with
+-- 1-day chunks and the same 7-day retention policy as
+-- the live `deliveries` table. The composite PK
+-- (id, created_at) is already in place above, satisfying
+-- Timescale's "partition column must be in UNIQUE/PK".
+SELECT create_hypertable(
+    'deliveries_dlq',
+    'created_at',
+    chunk_time_interval => INTERVAL '1 day',
+    migrate_data => true,
+    if_not_exists => true
+);
+
+-- 7-day hot retention in Postgres. The archiver ships
+-- rows older than that to ClickHouse before Timescale
+-- drops the chunk. M8 ships BA_ARCHIVERD_DLQ_OLDER_THAN_HOURS=168.
+SELECT add_retention_policy(
+    'deliveries_dlq',
+    INTERVAL '7 days',
+    if_not_exists => true
+);