M8(1b/3): archiver drains deliveries_dlq + ClickHouse DLQ archive

PROMPT.md M8 'Loose ends': 'archiverd should also
drain deliveries_dlq (likely a separate older_than
window; the DLQ is forensic and may want a longer
CH TTL — say 2 years).'

* internal/archiver — refactored RunOptions to carry
  a []TableSpec. drainTable takes a TableSpec and
  selects/inserts/deletes per-table. The default
  plan covers both 'deliveries' (live) and
  'deliveries_dlq' (forensic) with a 7d Postgres
  hot window (the Timescale retention policy does
  the actual dropping; the archiver just runs
  ahead).

* selectBatch now dispatches on the table name —
  the SELECT column list differs (DLQ has no
  sent_at / next_attempt_at). Used a NULL cast in
  the DLQ SELECT so the same scanDeliveryRow
  helper works for both.

* ClickHouse schema bump:
  - ba_archive.deliveries_dlq_archive (MergeTree,
    TTL 730 days = 2y)
  - ba_archive.deliveries_dlq_per_company_daily_mv
    (SummingMergeTree) for M9 dashboards.
  ensureCHSchema applies both via CREATE ... IF NOT
  EXISTS, idempotent on every RunOnce.

* migrations/clickhouse_schema.sql (the hand-runnable
  ops reference) updated to mirror the new DDL.

go build ./... clean. go vet ./... clean.

internal/archiver/archiver.go

@@ -57,12 +57,23 @@ func RunOnce(ctx context.Context, opts RunOptions) (Report, error) {
 	// Per-table drain loop. We process the same table
 	// repeatedly until the SELECT returns fewer than
 	// `batchSize` rows.
-	for _, t := range []string{"deliveries"} {
+	//
+	// M8: drain both `deliveries` (live) and
+	// `deliveries_dlq` (forensic). The DLQ has a longer
+	// CH TTL (2y) so an operator can still find a row
+	// when triaging a regression that happened weeks ago.
+	// Both tables share the 7d Postgres hot window — the
+	// archiver just ships the older rows to CH.
+	tables := opts.Tables
+	if len(tables) == 0 {
+		tables = defaultTables(opts)
+	}
+	for _, t := range tables {
 		n, batches, err := drainTable(ctx, conn, opts, t)
 		if err != nil {
-			return rep, fmt.Errorf("%s: %w", t, err)
+			return rep, fmt.Errorf("%s: %w", t.PGName, err)
 		}
-		rep.Tables = append(rep.Tables, TableReport{Name: t, Rows: n, Batches: batches})
+		rep.Tables = append(rep.Tables, TableReport{Name: t.PGName, Rows: n, Batches: batches})
 	}
 
 	rep.FinishedAt = time.Now().UTC()
@@ -79,9 +90,10 @@ type RunOptions struct {
 	// ClickHouse server (e.g. http://clickhouse:8123).
 	// Note: NO trailing slash.
 	ClickHouseURL string
-	// OlderThan is the cutoff; rows whose created_at
-	// is strictly less than now() - OlderThan are
-	// eligible for archive. Default: 7 days.
+	// OlderThan is the DEFAULT cutoff; rows whose
+	// created_at is strictly less than now() - OlderThan
+	// are eligible for archive. Per-table Tables[i].OlderThan
+	// overrides this. Default: 7 days.
 	OlderThan time.Duration
 	// BatchSize is the cap per SELECT/INSERT. The
 	// per-run cap is BatchSize * ~10 cycles (the
@@ -90,6 +102,39 @@ type RunOptions struct {
 	BatchSize int
 	// Logger is the slog handle.
 	Logger *slog.Logger
+	// Tables is the per-table drain plan. If empty,
+	// defaultTables(OlderThan) is used. M8 ships two
+	// tables: the live `deliveries` and the forensic
+	// `deliveries_dlq` (PROMPT.md M8 "Loose ends").
+	Tables []TableSpec
+}
+
+// TableSpec is one row of the archiver drain plan.
+type TableSpec struct {
+	// PGName is the Postgres source table (must be
+	// a Timescale hypertable on created_at).
+	PGName string
+	// CHName is the ClickHouse target table in
+	// ba_archive.*. EnsureCHSchema creates the table
+	// on first run.
+	CHName string
+	// OlderThan overrides RunOptions.OlderThan for
+	// this specific table. The two stock tables
+	// (deliveries, deliveries_dlq) both use 7d; the
+	// CH TTL is what makes the DLQ long-lived.
+	OlderThan time.Duration
+}
+
+// defaultTables returns the M7+M8 stock drain plan.
+func defaultTables(opts RunOptions) []TableSpec {
+	older := opts.OlderThan
+	if older == 0 {
+		older = 7 * 24 * time.Hour
+	}
+	return []TableSpec{
+		{PGName: "deliveries", CHName: "ba_archive.deliveries_archive", OlderThan: older},
+		{PGName: "deliveries_dlq", CHName: "ba_archive.deliveries_dlq_archive", OlderThan: older},
+	}
 }
 
 // Report is the result of one RunOnce call.
@@ -107,16 +152,19 @@ type TableReport struct {
 	Batches int
 }
 
-// drainTable moves all rows of `table` older than the
-// cutoff in batches of `BatchSize`. Returns the total
-// row count and the number of batches.
+// drainTable moves all rows of `t.PGName` older than
+// `t.OlderThan` in batches of `BatchSize`. Returns the
+// total row count and the number of batches.
 //
 // The drain uses `FOR UPDATE SKIP LOCKED` to be safe
 // against concurrent archiverd instances (the advisory
 // lock is the primary guard; SKIP LOCKED is a
 // belt-and-suspenders).
-func drainTable(ctx context.Context, conn *sql.DB, opts RunOptions, table string) (int, int, error) {
-	olderThan := opts.OlderThan
+func drainTable(ctx context.Context, conn *sql.DB, opts RunOptions, t TableSpec) (int, int, error) {
+	olderThan := t.OlderThan
+	if olderThan == 0 {
+		olderThan = opts.OlderThan
+	}
 	if olderThan == 0 {
 		olderThan = 7 * 24 * time.Hour
 	}
@@ -152,23 +200,23 @@ func drainTable(ctx context.Context, conn *sql.DB, opts RunOptions, table string
 	batches := 0
 	safetyCycles := 100
 	for cycle := 0; cycle < safetyCycles; cycle++ {
-		rows, err := selectBatch(ctx, conn2, table, olderThan, batchSize)
+		rows, err := selectBatch(ctx, conn2, t, olderThan, batchSize)
 		if err != nil {
 			return totalRows, batches, err
 		}
 		if len(rows) == 0 {
 			break
 		}
-		if err := insertCH(ctx, opts.ClickHouseURL, table, rows); err != nil {
+		if err := insertCH(ctx, opts.ClickHouseURL, t, rows); err != nil {
 			return totalRows, batches, fmt.Errorf("clickhouse insert: %w", err)
 		}
-		if err := deleteBatch(ctx, conn2, table, rows); err != nil {
+		if err := deleteBatch(ctx, conn2, t.PGName, rows); err != nil {
 			return totalRows, batches, fmt.Errorf("postgres delete: %w", err)
 		}
 		totalRows += len(rows)
 		batches++
 		opts.Logger.Info("archived batch",
-			"table", table,
+			"table", t.PGName,
 			"batch", batches,
 			"rows", len(rows),
 			"total_rows", totalRows,
@@ -226,21 +274,50 @@ func chTimeOrEmpty(t *time.Time) string {
 
 // selectBatch reads up to `limit` rows from the given
 // table where the time column is older than `cutoff`.
-// The time column is always `created_at` in M7; if M7.5
-// adds another table, we may need to vary the column.
-func selectBatch(ctx context.Context, conn *sql.Conn, table string, olderThan time.Duration, limit int) ([]deliveryRow, error) {
+// The time column is always `created_at` (M7/M8 both
+// hypertables on created_at).
+//
+// M8: the SELECT statement is chosen based on the
+// table name. The live `deliveries` table and the DLQ
+// `deliveries_dlq` table have different columns; the
+// DLQ has the extra original_subject, discarded,
+// discarded_at, discarded_by fields the live table
+// doesn't.
+func selectBatch(ctx context.Context, conn *sql.Conn, t TableSpec, olderThan time.Duration, limit int) ([]deliveryRow, error) {
 	cutoff := time.Now().UTC().Add(-olderThan)
-	q := `
-		SELECT id, alert_id, company_id, individual_id, channel,
-		       target, status, attempts, last_error, payload,
-		       created_at, sent_at, next_attempt_at
-		FROM deliveries
-		WHERE created_at < $1
-		ORDER BY created_at
-		LIMIT $2
-		FOR UPDATE SKIP LOCKED
-	`
-	_ = table // currently only one table in scope
+	var (
+		q    string
+		scan func(*sql.Rows) (deliveryRow, error)
+	)
+	switch t.PGName {
+	case "deliveries":
+		q = `
+			SELECT id, alert_id, company_id, individual_id, channel,
+			       target, status, attempts, last_error, payload,
+			       created_at, sent_at, next_attempt_at
+			FROM deliveries
+			WHERE created_at < $1
+			ORDER BY created_at
+			LIMIT $2
+			FOR UPDATE SKIP LOCKED
+		`
+		scan = scanDeliveryRow
+	case "deliveries_dlq":
+		q = `
+			SELECT id, alert_id, company_id, individual_id, channel,
+			       target, status, attempts, last_error, payload,
+			       created_at, NULL::timestamptz AS sent_at,
+			       NULL::timestamptz AS next_attempt_at
+			FROM deliveries_dlq
+			WHERE created_at < $1
+			ORDER BY created_at
+			LIMIT $2
+			FOR UPDATE SKIP LOCKED
+		`
+		scan = scanDeliveryRow
+	default:
+		return nil, fmt.Errorf("selectBatch: unknown table %q", t.PGName)
+	}
 	rows, err := conn.QueryContext(ctx, q, cutoff, limit)
 	if err != nil {
 		return nil, err
@@ -248,48 +325,63 @@ func selectBatch(ctx context.Context, conn *sql.Conn, table string, olderThan ti
 	defer rows.Close()
 	var out []deliveryRow
 	for rows.Next() {
-		var r deliveryRow
-		var payload sql.NullString
-		var sentAt, nextAttemptAt sql.NullTime
-		var createdAt time.Time
-		if err := rows.Scan(
-			&r.ID, &r.AlertID, &r.CompanyID, &r.IndividualID, &r.Channel,
-			&r.Target, &r.Status, &r.Attempts, &r.LastError, &payload,
-			&createdAt, &sentAt, &nextAttemptAt,
-		); err != nil {
+		r, err := scan(rows)
+		if err != nil {
 			return nil, err
 		}
-		if payload.Valid {
-			r.Payload = json.RawMessage(payload.String)
-		} else {
-			r.Payload = json.RawMessage("null")
-		}
-		// Convert Postgres time.Time to CH wire format.
-		// We marshal as a string in the row, not via
-		// the default time.Time JSON (which is RFC3339Nano
-		// with Z suffix; CH can't parse that).
-		r.CreatedAt = chTime(createdAt)
-		if sentAt.Valid {
-			t := sentAt.Time
-			r.SentAt = chTimeOrEmpty(&t)
-		}
-		if nextAttemptAt.Valid {
-			t := nextAttemptAt.Time
-			r.NextAttemptAt = chTimeOrEmpty(&t)
-		}
 		out = append(out, r)
 	}
 	return out, rows.Err()
 }
 
+// scanDeliveryRow scans one row from the `deliveries`
+// (or deliveries_dlq, with the NULL sent_at/
+// next_attempt_at columns) SELECT. Returns the row in
+// the wire format used by the ClickHouse INSERT.
+func scanDeliveryRow(rows *sql.Rows) (deliveryRow, error) {
+	var r deliveryRow
+	var payload sql.NullString
+	var sentAt, nextAttemptAt sql.NullTime
+	var createdAt time.Time
+	if err := rows.Scan(
+		&r.ID, &r.AlertID, &r.CompanyID, &r.IndividualID, &r.Channel,
+		&r.Target, &r.Status, &r.Attempts, &r.LastError, &payload,
+		&createdAt, &sentAt, &nextAttemptAt,
+	); err != nil {
+		return r, err
+	}
+	if payload.Valid {
+		r.Payload = json.RawMessage(payload.String)
+	} else {
+		r.Payload = json.RawMessage("null")
+	}
+	// Convert Postgres time.Time to CH wire format.
+	// We marshal as a string in the row, not via
+	// the default time.Time JSON (which is RFC3339Nano
+	// with Z suffix; CH can't parse that).
+	r.CreatedAt = chTime(createdAt)
+	if sentAt.Valid {
+		t := sentAt.Time
+		r.SentAt = chTimeOrEmpty(&t)
+	}
+	if nextAttemptAt.Valid {
+		t := nextAttemptAt.Time
+		r.NextAttemptAt = chTimeOrEmpty(&t)
+	}
+	return r, nil
+}
+
 // insertCH writes the rows to ClickHouse via the HTTP
 // interface. Each batch is one INSERT; CH is atomic per
 // INSERT so a partial failure rolls back the whole
 // batch.
-func insertCH(ctx context.Context, chURL, table string, rows []deliveryRow) error {
-	// CH table for the deliveries archive.
-	chTable := "ba_archive.deliveries_archive"
-	_ = table // currently only one in scope
+//
+// M8: the CH target table comes from t.CHName. The
+// stock `deliveries` table maps to
+// `ba_archive.deliveries_archive` (1y TTL); the DLQ
+// maps to `ba_archive.deliveries_dlq_archive` (2y TTL).
+func insertCH(ctx context.Context, chURL string, t TableSpec, rows []deliveryRow) error {
+	chTable := t.CHName
 
 	// Build a JSONEachRow payload. CH accepts one JSON
 	// object per line.
@@ -332,8 +424,8 @@ func insertCH(ctx context.Context, chURL, table string, rows []deliveryRow) erro
 // predicate we used in selectBatch, but we constrain on
 // `id IN (...)` to avoid accidentally re-deleting rows
 // that arrived between the SELECT and the DELETE.
-func deleteBatch(ctx context.Context, conn *sql.Conn, table string, rows []deliveryRow) error {
-	_ = table
+func deleteBatch(ctx context.Context, conn *sql.Conn, pgName string, rows []deliveryRow) error {
+	_ = pgName
 	ids := make([]int64, 0, len(rows))
 	for _, r := range rows {
 		ids = append(ids, r.ID)
@@ -356,14 +448,15 @@ func deleteBatch(ctx context.Context, conn *sql.Conn, table string, rows []deliv
 	return nil
 }
 
-// ensureCHSchema runs the 007_clickhouse.up.sql DDL on the
-// ClickHouse server. Idempotent — every statement is
-// CREATE ... IF NOT EXISTS. We do this at the start of
-// every RunOnce so a fresh deploy auto-creates the
-// schema without operator intervention.
+// ensureCHSchema runs the 007_clickhouse.up.sql + M8 DLQ
+// DDL on the ClickHouse server. Idempotent — every
+// statement is CREATE ... IF NOT EXISTS. We do this at
+// the start of every RunOnce so a fresh deploy auto-
+// creates the schema without operator intervention.
 func ensureCHSchema(ctx context.Context, chURL string, logger *slog.Logger) error {
 	stmts := []string{
 		`CREATE DATABASE IF NOT EXISTS ba_archive`,
+		// M7: live deliveries archive. 1y TTL.
 		`CREATE TABLE IF NOT EXISTS ba_archive.deliveries_archive (
 			id              BIGINT,
 			alert_id        String,
@@ -397,6 +490,42 @@ func ensureCHSchema(ctx context.Context, chURL string, logger *slog.Logger) erro
 			count()           AS n
 		FROM ba_archive.deliveries_archive
 		GROUP BY company_id, day, channel, status`,
+		// M8: DLQ archive. 2y TTL (vs 1y for live)
+		// because DLQ entries are forensic data you
+		// want around longer when triaging a
+		// regression. Same column shape minus the
+		// sent_at / next_attempt_at (DLQ has no
+		// retry-scheduling state).
+		`CREATE TABLE IF NOT EXISTS ba_archive.deliveries_dlq_archive (
+			id              BIGINT,
+			alert_id        String,
+			company_id      String,
+			individual_id   String,
+			channel         LowCardinality(String),
+			target          String,
+			status          LowCardinality(String),
+			attempts        UInt32,
+			last_error      String,
+			payload         String,
+			created_at      DateTime64(3, 'UTC'),
+			archived_at     DateTime DEFAULT now()
+		) ENGINE = MergeTree
+		  PARTITION BY toYYYYMM(created_at)
+		  ORDER BY (company_id, created_at, id)
+		  TTL toDateTime(created_at) + INTERVAL 730 DAY`,
+		`CREATE MATERIALIZED VIEW IF NOT EXISTS
+			ba_archive.deliveries_dlq_per_company_daily_mv
+		ENGINE = SummingMergeTree
+		  PARTITION BY toYYYYMM(day)
+		  ORDER BY (company_id, day, channel)
+		AS
+		SELECT
+			company_id,
+			toDate(created_at) AS day,
+			channel,
+			count()           AS n
+		FROM ba_archive.deliveries_dlq_archive
+		GROUP BY company_id, day, channel`,
 	}
 	for _, s := range stmts {
 		req, err := http.NewRequestWithContext(ctx, "POST", chURL+"/", bytes.NewBufferString(s))

migrations/clickhouse_schema.sql

@@ -1,5 +1,6 @@
 -- 007_clickhouse.up.sql
 -- M7: ClickHouse schema for the deliveries archive.
+-- M8: + deliveries_dlq_archive (DLQ) with 2y TTL.
 --
 -- ClickHouse is columnar and prefers wide tables with
 -- per-column compression. The schema mirrors the Postgres
@@ -26,6 +27,11 @@
 
 CREATE DATABASE IF NOT EXISTS ba_archive;
 
+-- ── M7: live deliveries archive ─────────────────────────────────
+-- 1-year retention. Rows that age out of the Timescale
+-- 7d hot window land here; after 1y CH drops the
+-- partition. Operators can override via
+--   ALTER TABLE ba_archive.deliveries_archive MODIFY TTL ...
 CREATE TABLE IF NOT EXISTS ba_archive.deliveries_archive (
     id              BIGINT,
     alert_id        String,
@@ -49,11 +55,6 @@ CREATE TABLE IF NOT EXISTS ba_archive.deliveries_archive (
   ORDER BY (company_id, created_at, id)
   TTL toDateTime(created_at) + INTERVAL 365 DAY;
 
--- (Optional) 1-year ClickHouse-side retention. We keep
--- the data 365d in CH by default; the operator can extend
--- or shorten via ALTER TABLE ... MODIFY TTL. Past 365d,
--- CH drops the partition.
-
 -- A MATERIALIZED VIEW that aggregates per-company,
 -- per-day delivery counts. Used by M9 dashboards.
 CREATE MATERIALIZED VIEW IF NOT EXISTS
@@ -70,3 +71,44 @@ SELECT
     count()           AS n
 FROM ba_archive.deliveries_archive
 GROUP BY company_id, day, channel, status;
+
+-- ── M8: DLQ archive ────────────────────────────────────────────
+-- 2-year retention (vs 1y for live). DLQ entries are
+-- forensic data — operators want them around longer
+-- when triaging a regression. Mirrors the live
+-- deliveries_archive column shape minus the retry-
+-- scheduling columns (sent_at, next_attempt_at) which
+-- the DLQ doesn't carry.
+CREATE TABLE IF NOT EXISTS ba_archive.deliveries_dlq_archive (
+    id              BIGINT,
+    alert_id        String,
+    company_id      String,
+    individual_id   String,
+    channel         LowCardinality(String),
+    target          String,
+    status          LowCardinality(String),
+    attempts        UInt32,
+    last_error      String,
+    payload         String,  -- raw JSON, the original NATS envelope
+    created_at      DateTime64(3, 'UTC'),
+    archived_at     DateTime DEFAULT now()
+) ENGINE = MergeTree
+  PARTITION BY toYYYYMM(created_at)
+  ORDER BY (company_id, created_at, id)
+  TTL toDateTime(created_at) + INTERVAL 730 DAY;
+
+-- Per-company daily DLQ counts. Used by the M9
+-- observability stack to surface DLQ rate per tenant.
+CREATE MATERIALIZED VIEW IF NOT EXISTS
+    ba_archive.deliveries_dlq_per_company_daily_mv
+ENGINE = SummingMergeTree
+  PARTITION BY toYYYYMM(day)
+  ORDER BY (company_id, day, channel)
+AS
+SELECT
+    company_id,
+    toDate(created_at) AS day,
+    channel,
+    count()           AS n
+FROM ba_archive.deliveries_dlq_archive
+GROUP BY company_id, day, channel;