M7(1/3): Timescale 7d + ClickHouse archive + archiverd (code, smoke unverified on remote)

- migrations/006_timescale: deliveries -> hypertable on created_at, 1d chunks, 7-day retention (PK rebased to (id, created_at))
- migrations/007_clickhouse: ba_archive.deliveries_archive MergeTree (TTL 365d) + per-company daily MV
- internal/archiver: RunOnce(ctx) with pg_try_advisory_lock + FOR UPDATE SKIP LOCKED + CH HTTP INSERT
- cmd/archiverd: hourly loop, /health, /metrics
- internal/config: Archiverd struct + LoadArchiverd
- docker-compose: archiverd service on host port 8804
- Dockerfile: archiverd in build chain
- .env.example: BA_ARCHIVERD_RUN_EVERY_SECONDS, BA_ARCHIVERD_OLDER_THAN_HOURS, BA_ARCHIVERD_BATCH_SIZE
- scripts/m7_smoke.sh: 4-step, 9-check smoke

Verification doc + smoke log + SPEC bump deferred to M7(2/3)/(3/3) after remote smoke is green.

.env.example

@@ -28,6 +28,14 @@ BA_INGESTD_QUARANTINE_HITS_THRESHOLD=100
 # alerts is held for up to this many ms, then a single
 # delivery is fanned out with the final dedupe_count.
 BA_ROUTERD_DEDUPE_FLUSH_MS=2000
+
+# M7: archiver cadence + retention cutoff. The Timescale
+# retention policy does the same at 7d; the archiver just
+# runs ahead so ClickHouse has the data before TS drops it.
+BA_ARCHIVERD_RUN_EVERY_SECONDS=3600
+BA_ARCHIVERD_OLDER_THAN_HOURS=168
+BA_ARCHIVERD_BATCH_SIZE=10000
+BA_ARCHIVERD_CLICKHOUSE_URL=http://clickhouse:8123
 BA_INGESTD_QUARANTINE_WINDOW_SECONDS=60
 BA_INGESTD_QUARANTINE_DURATION_SECONDS=300
 

Dockerfile

@@ -25,6 +25,8 @@ RUN --mount=type=cache,target=/root/.cache/go-build \
       -o /out/telegramd ./cmd/telegramd && \
     CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
       -o /out/admind   ./cmd/admind && \
+    CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
+      -o /out/archiverd ./cmd/archiverd && \
     CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
       -o /out/seed     ./cmd/seed && \
     cd loadgen && \

cmd/archiverd/main.go

@@ -0,0 +1,219 @@
+// Command archiverd is the M7 periodic data-tier job. It
+// moves deliveries older than 7 days from TimescaleDB to
+// ClickHouse, then drops the source rows. The hot-path
+// services are unaffected — archiverd only reads and
+// deletes, never publishes or queues.
+//
+// Cadence: BA_ARCHIVERD_RUN_EVERY_SECONDS (default 3600).
+// Health: /health returns 200 if the last successful run
+// was less than 2x the cadence ago. /metrics exposes
+// ba_archiverd_rows_archived_total and
+// ba_archiverd_last_run_timestamp_seconds.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/signal"
+	"strconv"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/archiver"
+	"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"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+func main() {
+	cfg, err := config.LoadArchiverd()
+	if err != nil {
+		os.Stderr.WriteString("config: " + err.Error() + "\n")
+		os.Exit(1)
+	}
+	logger := observability.Init(cfg.Env, cfg.LogLevel, "archiverd")
+	logger.Info("starting",
+		"env", cfg.Env,
+		"addr", cfg.HTTPAddr,
+		"run_every_seconds", cfg.RunEverySeconds,
+		"older_than_hours", cfg.OlderThanHours,
+		"batch_size", cfg.BatchSize,
+	)
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	// Metrics. We use a tiny custom registry here rather
+	// than the shared IngestdMetrics struct because the
+	// archiver has different counters (rows_archived by
+	// table) and doesn't need the full ingestd suite.
+	reg := prometheus.NewRegistry()
+	reg.MustRegister(prometheus.NewGoCollector(), prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))
+	rowsArchived := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "ba",
+		Subsystem: "archiverd",
+		Name:      "rows_archived_total",
+		Help:      "M7: number of rows moved from Timescale to ClickHouse by the archiver.",
+	}, []string{"table"})
+	lastRunTs := prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: "ba",
+		Subsystem: "archiverd",
+		Name:      "last_run_timestamp_seconds",
+		Help:      "M7: unix timestamp of the last successful archiverd run.",
+	})
+	runDuration := prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: "ba",
+		Subsystem: "archiverd",
+		Name:      "last_run_duration_seconds",
+		Help:      "M7: duration of the last archiverd run, in seconds.",
+	})
+	reg.MustRegister(rowsArchived, lastRunTs, runDuration)
+
+	// Shared atomic for the /health check.
+	var lastRunUnix atomic.Int64
+	lastRunUnix.Store(0)
+
+	// Periodic loop. RunOnce is called in a goroutine so
+	// the /health endpoint stays responsive between runs.
+	go runLoop(ctx, logger, archiver.RunOptions{
+		PostgresDSN:   cfg.PostgresDSN,
+		ClickHouseURL: cfg.ClickHouseURL,
+		OlderThan:     time.Duration(cfg.OlderThanHours) * time.Hour,
+		BatchSize:     cfg.BatchSize,
+		Logger:        logger,
+	}, rowsArchived, &lastRunUnix, runDuration)
+
+	srv := httpserver.New(httpserver.Config{
+		Addr:          cfg.HTTPAddr,
+		ServiceName:   "archiverd",
+		ShutdownGrace: cfg.ShutdownGrace,
+	}, logger, metricsHandler(reg, &lastRunUnix, cfg.RunEverySeconds*2))
+
+	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)
+		}
+	}
+	if err := srv.Shutdown(ctx); err != nil {
+		logger.Warn("graceful shutdown", "err", err)
+	}
+	logger.Info("bye")
+}
+
+// runLoop drives the periodic execution. The first run
+// fires immediately on startup (so a fresh deploy catches
+// up on backlog), then every `RunEvery` seconds.
+func runLoop(
+	ctx context.Context,
+	logger *slog.Logger,
+	opts archiver.RunOptions,
+	rowsArchived *prometheus.CounterVec,
+	lastRunUnix *atomic.Int64,
+	runDuration prometheus.Gauge,
+) {
+	tick := time.NewTicker(time.Duration(optsRunEverySeconds()) * time.Second)
+	defer tick.Stop()
+	for {
+		// Fire immediately on first iteration.
+		oneRun(ctx, logger, opts, rowsArchived, lastRunUnix, runDuration)
+		select {
+		case <-ctx.Done():
+			return
+		case <-tick.C:
+		}
+	}
+}
+
+func oneRun(
+	ctx context.Context,
+	logger *slog.Logger,
+	opts archiver.RunOptions,
+	rowsArchived *prometheus.CounterVec,
+	lastRunUnix *atomic.Int64,
+	runDuration prometheus.Gauge,
+) {
+	runCtx, cancel := context.WithTimeout(ctx, 5*time.Minute)
+	defer cancel()
+	rep, err := archiver.RunOnce(runCtx, opts)
+	if err != nil {
+		logger.Error("archiver run failed", "err", err)
+		return
+	}
+	lastRunUnix.Store(rep.FinishedAt.Unix())
+	for _, t := range rep.Tables {
+		rowsArchived.WithLabelValues(t.Name).Add(float64(t.Rows))
+	}
+	runDuration.Set(rep.Duration.Seconds())
+	logger.Info("archiver run ok",
+		"duration_seconds", rep.Duration.Seconds(),
+		"tables", rep.Tables,
+	)
+}
+
+// metricsHandler is a small wrapper that exposes /health
+// (returns 200 if the last run was within 2x the cadence)
+// alongside the /metrics endpoint.
+func metricsHandler(reg *prometheus.Registry, lastRun *atomic.Int64, healthyAfterSeconds int) *http.ServeMux {
+	mux := http.NewServeMux()
+	mux.Handle("GET /metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
+	mux.HandleFunc("GET /health", func(w http.ResponseWriter, r *http.Request) {
+		last := lastRun.Load()
+		if last == 0 {
+			// No run yet — still healthy if we're
+			// within the first 2x cadence of
+			// startup.
+			w.WriteHeader(http.StatusOK)
+			_ = json.NewEncoder(w).Encode(map[string]any{
+				"status": "starting",
+				"note":   "no run has completed yet",
+			})
+			return
+		}
+		age := time.Now().Unix() - last
+		if age > int64(healthyAfterSeconds) {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			_ = json.NewEncoder(w).Encode(map[string]any{
+				"status":   "stale",
+				"age_secs": age,
+			})
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"status":   "ok",
+			"age_secs": age,
+		})
+	})
+	return mux
+}
+
+// optsRunEverySeconds is read from the env at startup.
+// We keep it in a func rather than the config struct
+// because the loop closure captures it.
+func optsRunEverySeconds() int {
+	v := os.Getenv("BA_ARCHIVERD_RUN_EVERY_SECONDS")
+	if v == "" {
+		return 3600
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil || n <= 0 {
+		return 3600
+	}
+	return n
+}
+
+// keep fmt import
+var _ = fmt.Sprintf

docker-compose.yml

@@ -232,6 +232,26 @@ services:
       nats:     { condition: service_healthy }
       postgres: { condition: service_healthy }
 
+  archiverd:
+    build: .
+    command: ["/app/archiverd"]
+    environment:
+      BA_ENV: dev
+      BA_HTTP_ADDR: ":8804"
+      BA_POSTGRES_DSN: postgres://ba:ba@postgres:5432/ba?sslmode=disable
+      # M7: archiver cadence + retention cutoff. The
+      # Timescale retention policy does the same at 7d;
+      # the archiver just runs ahead so ClickHouse has
+      # the data before TS drops it.
+      BA_ARCHIVERD_RUN_EVERY_SECONDS: "3600"
+      BA_ARCHIVERD_OLDER_THAN_HOURS: "168"
+      BA_ARCHIVERD_BATCH_SIZE: "10000"
+      BA_ARCHIVERD_CLICKHOUSE_URL: "http://clickhouse:8123"
+    ports: ["8804:8804"]
+    depends_on:
+      postgres:    { condition: service_healthy }
+      clickhouse:  { condition: service_started }
+
   # ── Observability ────────────────────────────────────────────────
   prometheus:
     image: prom/prometheus:latest

internal/archiver/archiver.go

@@ -0,0 +1,436 @@
+// Package archiver is the M7 periodic data-tier job. It
+// moves old rows from TimescaleDB `deliveries` to ClickHouse
+// `deliveries_archive` in batches. The hot-path services
+// (ingestd, routerd, deliverd-*) are unaffected; they
+// INSERT into `deliveries` just as before.
+//
+// The archiver is fail-stop: a partial batch is rolled back
+// (no DELETE happens if the ClickHouse INSERT fails).
+// Container restart retries the next hour.
+//
+// Concurrency: only one archiverd should run at a time per
+// environment. We use a Postgres advisory lock so two
+// replicas can't double-archive. `SKIP LOCKED` in the
+// SELECT makes the lock acquisition non-blocking.
+package archiver
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/url"
+	"time"
+
+	_ "github.com/jackc/pgx/v5/stdlib"
+)
+
+// RunOnce executes one full pass: for each table in scope,
+// drain rows older than `olderThan` in batches of
+// `batchSize`. Returns the total rows archived, total
+// batches, and any error encountered (caller decides
+// whether to retry the next interval).
+//
+// The function is safe to call repeatedly; each call
+// acquires the advisory lock, processes whatever's old,
+// releases the lock, and returns.
+func RunOnce(ctx context.Context, opts RunOptions) (Report, error) {
+	rep := Report{StartedAt: time.Now().UTC()}
+
+	conn, err := sql.Open("pgx", opts.PostgresDSN)
+	if err != nil {
+		return rep, fmt.Errorf("postgres open: %w", err)
+	}
+	defer conn.Close()
+	if err := conn.PingContext(ctx); err != nil {
+		return rep, fmt.Errorf("postgres ping: %w", err)
+	}
+
+	if err := ensureCHSchema(ctx, opts.ClickHouseURL, opts.Logger); err != nil {
+		return rep, fmt.Errorf("clickhouse ensure-schema: %w", err)
+	}
+
+	// Per-table drain loop. We process the same table
+	// repeatedly until the SELECT returns fewer than
+	// `batchSize` rows.
+	for _, t := range []string{"deliveries"} {
+		n, batches, err := drainTable(ctx, conn, opts, t)
+		if err != nil {
+			return rep, fmt.Errorf("%s: %w", t, err)
+		}
+		rep.Tables = append(rep.Tables, TableReport{Name: t, Rows: n, Batches: batches})
+	}
+
+	rep.FinishedAt = time.Now().UTC()
+	rep.Duration = rep.FinishedAt.Sub(rep.StartedAt)
+	return rep, nil
+}
+
+// RunOptions is the per-call configuration.
+type RunOptions struct {
+	// PostgresDSN is the libpq-style DSN (e.g. from
+	// config.Common.PostgresDSN).
+	PostgresDSN string
+	// ClickHouseURL is the HTTP base URL for the
+	// 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 time.Duration
+	// BatchSize is the cap per SELECT/INSERT. The
+	// per-run cap is BatchSize * ~10 cycles (the
+	// drain loop runs until a SELECT returns <BatchSize
+	// rows; the safety cap is 100 cycles).
+	BatchSize int
+	// Logger is the slog handle.
+	Logger *slog.Logger
+}
+
+// Report is the result of one RunOnce call.
+type Report struct {
+	StartedAt  time.Time
+	FinishedAt time.Time
+	Duration   time.Duration
+	Tables     []TableReport
+}
+
+// TableReport is the per-table result.
+type TableReport struct {
+	Name    string
+	Rows    int
+	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.
+//
+// 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
+	if olderThan == 0 {
+		olderThan = 7 * 24 * time.Hour
+	}
+	batchSize := opts.BatchSize
+	if batchSize == 0 {
+		batchSize = 10000
+	}
+
+	// Acquire the per-table advisory lock. If another
+	// archiverd holds it, we return early (no error —
+	// this is normal in a multi-replica deploy).
+	lockKey := int64(0xBA21B0DA) // arbitrary stable key
+	conn2, err := conn.Conn(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+	defer conn2.Close()
+	var gotLock bool
+	if err := conn2.QueryRowContext(ctx,
+		`SELECT pg_try_advisory_lock($1)`, lockKey,
+	).Scan(&gotLock); err != nil {
+		return 0, 0, fmt.Errorf("advisory lock: %w", err)
+	}
+	if !gotLock {
+		opts.Logger.Info("another archiverd holds the lock; skipping this run")
+		return 0, 0, nil
+	}
+	defer func() {
+		_, _ = conn2.ExecContext(ctx, `SELECT pg_advisory_unlock($1)`, lockKey)
+	}()
+
+	totalRows := 0
+	batches := 0
+	safetyCycles := 100
+	for cycle := 0; cycle < safetyCycles; cycle++ {
+		rows, err := selectBatch(ctx, conn2, table, olderThan, batchSize)
+		if err != nil {
+			return totalRows, batches, err
+		}
+		if len(rows) == 0 {
+			break
+		}
+		if err := insertCH(ctx, opts.ClickHouseURL, table, rows); err != nil {
+			return totalRows, batches, fmt.Errorf("clickhouse insert: %w", err)
+		}
+		if err := deleteBatch(ctx, conn2, table, rows); err != nil {
+			return totalRows, batches, fmt.Errorf("postgres delete: %w", err)
+		}
+		totalRows += len(rows)
+		batches++
+		opts.Logger.Info("archived batch",
+			"table", table,
+			"batch", batches,
+			"rows", len(rows),
+			"total_rows", totalRows,
+		)
+		if len(rows) < batchSize {
+			break // drained
+		}
+	}
+	return totalRows, batches, nil
+}
+
+// deliveryRow is the wire shape we read from Postgres
+// and INSERT into ClickHouse. JSON tags match the CH
+// column names (case-insensitive in CH).
+//
+// Time fields are stored as strings (CH wire format) —
+// not as time.Time. We do the conversion in selectBatch
+// after scanning. This sidesteps the default time.Time
+// JSON encoding (RFC3339Nano with 'T' + 'Z') that CH
+// can't parse for DateTime64 columns.
+type deliveryRow struct {
+	ID            int64           `json:"id"`
+	AlertID       string          `json:"alert_id"`
+	CompanyID     string          `json:"company_id"`
+	IndividualID  string          `json:"individual_id"`
+	Channel       string          `json:"channel"`
+	Target        string          `json:"target"`
+	Status        string          `json:"status"`
+	Attempts      int32           `json:"attempts"`
+	LastError     string          `json:"last_error"`
+	Payload       json.RawMessage `json:"payload"`
+	CreatedAt     string          `json:"created_at"`
+	SentAt        string          `json:"sent_at,omitempty"`
+	NextAttemptAt string          `json:"next_attempt_at,omitempty"`
+}
+
+// chTime formats a time as ClickHouse's preferred
+// DateTime64(3, 'UTC') wire format: "2026-06-06 21:11:52.036".
+// Postgres returns RFC3339Nano with 'T' separator and a 'Z'
+// suffix; CH doesn't parse those directly. We use 3-decimal
+// precision to match the column type.
+func chTime(t time.Time) string {
+	return t.UTC().Format("2006-01-02 15:04:05.000")
+}
+
+// chTimeOrEmpty renders a nullable time as "" for CH's
+// Nullable(DateTime64) column. CH accepts an empty string
+// for nullable datetime columns in JSONEachRow.
+func chTimeOrEmpty(t *time.Time) string {
+	if t == nil {
+		return ""
+	}
+	return chTime(*t)
+}
+
+// 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) {
+	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
+	rows, err := conn.QueryContext(ctx, q, cutoff, limit)
+	if err != nil {
+		return nil, err
+	}
+	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 {
+			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()
+}
+
+// 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
+
+	// Build a JSONEachRow payload. CH accepts one JSON
+	// object per line.
+	var buf bytes.Buffer
+	for _, r := range rows {
+		// CH's DateTime64(3, 'UTC') expects RFC3339Nano
+		// format. time.Time's default MarshalJSON gives
+		// RFC3339Nano.
+		j, err := json.Marshal(r)
+		if err != nil {
+			return fmt.Errorf("marshal row %d: %w", r.ID, err)
+		}
+		buf.Write(j)
+		buf.WriteByte('\n')
+	}
+
+	u := chURL + "/?" + url.Values{
+		"query": {"INSERT INTO " + chTable + " FORMAT JSONEachRow"},
+	}.Encode()
+	req, err := http.NewRequestWithContext(ctx, "POST", u, &buf)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/x-ndjson")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != 200 {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("clickhouse HTTP %d: %s", resp.StatusCode, string(body))
+	}
+	return nil
+}
+
+// deleteBatch removes the just-archived rows from
+// Postgres. The `created_at < cutoff` is the same
+// 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
+	ids := make([]int64, 0, len(rows))
+	for _, r := range rows {
+		ids = append(ids, r.ID)
+	}
+	// Cap chunk size at 1000 ids per query to stay
+	// within pgx's parameter limit and not blow up
+	// statement parsing on large batches.
+	const chunk = 1000
+	for i := 0; i < len(ids); i += chunk {
+		end := i + chunk
+		if end > len(ids) {
+			end = len(ids)
+		}
+		q := `DELETE FROM deliveries WHERE id = ANY($1::bigint[])`
+		_, err := conn.ExecContext(ctx, q, ids[i:end])
+		if err != nil {
+			return err
+		}
+	}
+	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.
+func ensureCHSchema(ctx context.Context, chURL string, logger *slog.Logger) error {
+	stmts := []string{
+		`CREATE DATABASE IF NOT EXISTS ba_archive`,
+		`CREATE TABLE IF NOT EXISTS ba_archive.deliveries_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'),
+			sent_at         Nullable(DateTime64(3, 'UTC')),
+			next_attempt_at Nullable(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 365 DAY`,
+		`CREATE MATERIALIZED VIEW IF NOT EXISTS
+			ba_archive.deliveries_per_company_daily_mv
+		ENGINE = SummingMergeTree
+		  PARTITION BY toYYYYMM(day)
+		  ORDER BY (company_id, day, channel, status)
+		AS
+		SELECT
+			company_id,
+			toDate(created_at) AS day,
+			channel,
+			status,
+			count()           AS n
+		FROM ba_archive.deliveries_archive
+		GROUP BY company_id, day, channel, status`,
+	}
+	for _, s := range stmts {
+		req, err := http.NewRequestWithContext(ctx, "POST", chURL+"/", bytes.NewBufferString(s))
+		if err != nil {
+			return err
+		}
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return fmt.Errorf("ch stmt: %w", err)
+		}
+		if resp.StatusCode != 200 {
+			body, _ := io.ReadAll(resp.Body)
+			resp.Body.Close()
+			return fmt.Errorf("ch stmt HTTP %d: %s", resp.StatusCode, string(body))
+		}
+		resp.Body.Close()
+		logger.Debug("clickhouse schema ok", "stmt_prefix", firstLine(s))
+	}
+	return nil
+}
+
+// firstLine returns up to 80 chars of the first line of
+// a SQL statement, for log readability.
+func firstLine(s string) string {
+	for i, c := range s {
+		if c == '\n' {
+			if i > 80 {
+				return s[:80] + "..."
+			}
+			return s[:i]
+		}
+	}
+	if len(s) > 80 {
+		return s[:80] + "..."
+	}
+	return s
+}

internal/config/config.go

@@ -150,6 +150,29 @@ type Routerd struct {
 	DedupeFlushMs int
 }
 
+// Archiverd is archiverd-specific config (M7).
+type Archiverd struct {
+	Common
+	// RunEverySeconds is the cadence between archiverd
+	// passes. Default 3600 (1 hour).
+	RunEverySeconds int
+	// OlderThanHours is the retention threshold; rows
+	// older than now() - OlderThanHours are eligible
+	// for archive. Default 168 (7 days). The Timescale
+	// retention policy does the same at 7 days; the
+	// archiver just runs ahead so CH has the data
+	// before TS drops it.
+	OlderThanHours int
+	// BatchSize is the cap per SELECT/INSERT. Default
+	// 10000. Each pass drains until the SELECT returns
+	// < BatchSize rows, with a 100-cycle safety cap.
+	BatchSize int
+	// ClickHouseURL is the HTTP base URL for the CH
+	// server (no trailing slash). Default
+	// http://clickhouse:8123.
+	ClickHouseURL string
+}
+
 // LoadRouterd reads routerd-specific config.
 func LoadRouterd() (Routerd, error) {
 	c, err := LoadCommon("routerd")
@@ -162,6 +185,21 @@ func LoadRouterd() (Routerd, error) {
 	}, nil
 }
 
+// LoadArchiverd reads archiverd-specific config.
+func LoadArchiverd() (Archiverd, error) {
+	c, err := LoadCommon("archiverd")
+	if err != nil {
+		return Archiverd{}, err
+	}
+	return Archiverd{
+		Common:          c,
+		RunEverySeconds: GetInt("BA_ARCHIVERD_RUN_EVERY_SECONDS", 3600),
+		OlderThanHours:  GetInt("BA_ARCHIVERD_OLDER_THAN_HOURS", 168),
+		BatchSize:       GetInt("BA_ARCHIVERD_BATCH_SIZE", 10000),
+		ClickHouseURL:  envOr("BA_ARCHIVERD_CLICKHOUSE_URL", "http://clickhouse:8123"),
+	}, nil
+}
+
 // LoadIngestd reads ingestd-specific config.
 func LoadIngestd() (Ingestd, error) {
 	c, err := LoadCommon("ingestd")

loadgen/cmd/http/main.go

@@ -92,7 +92,7 @@ func main() {
 				if err := limiter.Wait(ctx); err != nil {
 					return
 				}
-				a := mkAlert(*mode, *companyN, *dedupePct, *payloadB)
+				a := mkAlert(*mode, *companyN, *dedupePct, *dedupeKey, *payloadB)
 				if err := sendOne(ctx, client, *target, company, source, []byte(secret), a); err != nil {
 					failed.Add(1)
 					if isRateLimited(err) {
@@ -163,7 +163,7 @@ func main() {
 	)
 }
 
-func mkAlert(mode string, companies, dedupePct, payloadB int) alert.Alert {
+func mkAlert(mode string, companies, dedupePct int, dedupeKey string, payloadB int) alert.Alert {
 	// Cycle through a few company_ids; the source is fixed for the
 	// run (loadgen is one source).
 	companyID := fmt.Sprintf("acme-%03d", (rand.IntN(companies) + 1))
@@ -176,8 +176,8 @@ func mkAlert(mode string, companies, dedupePct, payloadB int) alert.Alert {
 	// M6: --dedupe-key forces a specific key on every alert,
 	// which is what the ×N smoke test needs (one key, N copies).
 	dk := ""
-	if *dedupeKey != "" {
-		dk = *dedupeKey
+	if dedupeKey != "" {
+		dk = dedupeKey
 	} else if rand.IntN(100) < dedupePct {
 		dk = fmt.Sprintf("burst:%s:probe", category)
 	}

migrations/006_timescale.down.sql

@@ -0,0 +1,60 @@
+-- 006_timescale.down.sql
+-- Reverse M7. Removes the retention policy, then converts
+-- the hypertable back to a plain table. The data is
+-- preserved; only the chunk metadata is dropped.
+--
+-- NOTE: Timescale's `detach_database` or `hypertable_to_table`
+-- is a no-op in modern Timescale — once a table is a
+-- hypertable, you can't truly "downgrade" it without a
+-- table swap. We use the documented pattern of creating a
+-- plain shadow table, copying data over, and renaming.
+-- This is destructive on the timeline dimension but
+-- preserves all rows.
+
+-- 1. Remove the retention policy. After this, Timescale
+-- will no longer auto-drop chunks.
+SELECT remove_retention_policy('deliveries', if_exists => true);
+
+-- 2. Timescale has no clean "downgrade a hypertable"
+-- primitive. The recommended approach for a rollback is:
+--   a) create a plain shadow table with the same schema
+--   b) INSERT ... SELECT all rows
+--   c) drop the hypertable, rename the shadow
+-- We do that here. Note this is non-atomic and can take
+-- a while on large tables.
+
+CREATE TABLE IF NOT EXISTS deliveries_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,
+    status          TEXT NOT NULL DEFAULT 'pending',
+    attempts        INT NOT NULL DEFAULT 0,
+    last_error      TEXT,
+    payload         JSONB,
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    sent_at         TIMESTAMPTZ,
+    next_attempt_at TIMESTAMPTZ
+);
+
+-- The hypertable has the same columns, so a plain
+-- INSERT ... SELECT is enough. We add the original
+-- indexes back too.
+CREATE INDEX IF NOT EXISTS idx_deliveries_plain_company_alert
+    ON deliveries_plain(company_id, alert_id);
+CREATE INDEX IF NOT EXISTS idx_deliveries_plain_status
+    ON deliveries_plain(status) WHERE status IN ('pending','failed');
+
+-- Drop the hypertable and rename. This breaks the FK
+-- references and recreates the original table layout.
+DROP TABLE IF EXISTS deliveries CASCADE;
+ALTER TABLE deliveries_plain RENAME TO deliveries;
+
+-- The original 002_deliveries.up.sql had these indexes
+-- too; recreate them.
+CREATE INDEX IF NOT EXISTS idx_deliveries_company_alert
+    ON deliveries(company_id, alert_id);
+CREATE INDEX IF NOT EXISTS idx_deliveries_status
+    ON deliveries(status) WHERE status IN ('pending','failed');

migrations/006_timescale.up.sql

@@ -0,0 +1,75 @@
+-- 006_timescale.up.sql
+-- M7: Convert deliveries to a TimescaleDB hypertable with 7d
+-- retention. Hot-path writes still go through the same
+-- deliveries.*<channel>.<company> NATS subjects; the
+-- deliverers INSERT rows just as before. The data-layer
+-- change is invisible to the services.
+--
+-- SPEC §23 calls for both alerts and deliveries as hypertables,
+-- but alerts in our codebase never live in Postgres — they
+-- flow through NATS JetStream with a 24h stream retention
+-- (see cmd/ingestd + cmd/routerd). The "alerts archive" path
+-- would be a separate M7.5 if we ever want to long-term-store
+-- alert bodies in ClickHouse. For M7, only deliveries is in
+-- scope.
+--
+-- What this migration does:
+--   1. Convert `deliveries` to a hypertable on `created_at`
+--      (1d chunks). Timescale will copy data behind the
+--      scenes if the table has rows; the existing
+--      `idx_deliveries_company_alert` and
+--      `idx_deliveries_status` indexes are kept.
+--   2. Add a 7d retention policy. Timescale will drop
+--      chunks older than 7 days automatically.
+--   3. Note: PRIMARY KEY (id) is preserved; Timescale allows
+--      a non-time column as PK on a hypertable. The
+--      `id BIGSERIAL` is sequence-based, not time-based, so
+--      it stays valid across chunk drops.
+--
+-- Why `created_at` not `sent_at`:
+--   - `sent_at` is NULL while a delivery is in-flight. We
+--     can't hypertable on a NULLable time column.
+--   - `created_at` is the enqueue time, which is when the
+--     delivery came into existence. For retention purposes
+--     (7d after enqueue), this is the right anchor.
+--   - The M7 archiver uses `created_at < now() - 7d` as the
+--     "old enough to move" predicate.
+
+-- 1. Create the hypertable. The `migrate_data => true` flag
+-- tells Timescale to copy existing rows (the dev DB has
+-- ~270 rows from the M5–M6.5 smoke runs).
+--
+-- Timescale requires the partitioning column to be part
+-- of any UNIQUE/PRIMARY KEY constraint. The original PK
+-- is `id BIGSERIAL` which is a sequence and never reused
+-- across chunks, so we extend it to a composite PK
+-- (id, created_at). No application code needs to change
+-- because the constraint is still keyed on id for
+-- uniqueness purposes (BIGSERIAL is monotonically
+-- increasing, so (id, created_at) is in practice the
+-- same as id alone).
+ALTER TABLE deliveries DROP CONSTRAINT IF EXISTS deliveries_pkey CASCADE;
+ALTER TABLE deliveries ADD PRIMARY KEY (id, created_at);
+
+SELECT create_hypertable(
+    'deliveries',
+    'created_at',
+    chunk_time_interval => INTERVAL '1 day',
+    migrate_data => true,
+    if_not_exists => true
+);
+
+-- 2. Add the 7-day retention policy. Timescale drops the
+-- whole chunk (not row-by-row) when the chunk's max time
+-- is older than the threshold. This is dramatically cheaper
+-- than a DELETE on a large table.
+SELECT add_retention_policy(
+    'deliveries',
+    INTERVAL '7 days',
+    if_not_exists => true
+);
+
+-- 3. Optional: enable compression on older chunks. This
+-- is a Timescale 2.x feature. We keep it off in v1 to
+-- avoid surprising the operator; can be enabled per-chunk
+-- age in M7.5.

migrations/007_clickhouse.down.sql

@@ -0,0 +1,7 @@
+-- 007_clickhouse.down.sql
+-- Reverse M7. Drops the archive database and everything in it.
+-- This is destructive — archive rows are lost. The M7
+-- archiver's next run will recreate the schema (it has
+-- CREATE IF NOT EXISTS in its init path).
+
+DROP DATABASE IF EXISTS ba_archive;

migrations/007_clickhouse.up.sql

@@ -0,0 +1,72 @@
+-- 007_clickhouse.up.sql
+-- M7: ClickHouse schema for the deliveries archive.
+--
+-- ClickHouse is columnar and prefers wide tables with
+-- per-column compression. The schema mirrors the Postgres
+-- `deliveries` table one-to-one for v1; a future M7.5 can
+-- strip unused fields (e.g. drop the JSONB payload blob
+-- if we never query into it in CH).
+--
+-- The DDL below is designed to be runnable from the
+-- archiverd's first-run setup hook. It's also what the
+-- M7 smoke executes by hand via curl http://clickhouse:8123/.
+--
+-- Note on types:
+--   - `id` is BIGINT in Postgres (BIGSERIAL). In CH we
+--     keep BIGINT — CH's UInt64 is the closest match,
+--     but for an archive that doesn't reference id
+--     anywhere else, BIGINT is fine.
+--   - `payload` is JSONB in Postgres. We store it as a
+--     CH String. The Go side marshals to JSON text
+--     before INSERT. CH's JSON type is still experimental
+--     in 24.x; String is the safe choice.
+--   - `status` is a small enum ('pending','sent','failed','dlq').
+--     In CH we use LowCardinality(String) — the column
+--     takes ~0.5 bytes per row for the dictionary.
+
+CREATE DATABASE IF NOT EXISTS ba_archive;
+
+CREATE TABLE IF NOT EXISTS ba_archive.deliveries_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
+    created_at      DateTime64(3, 'UTC'),
+    sent_at         Nullable(DateTime64(3, 'UTC')),
+    next_attempt_at Nullable(DateTime64(3, 'UTC')),
+    -- The source postgres row's last-update marker. We
+    -- set this to the wall-clock time at insert. Useful
+    -- for "show me everything archived in the last 24h".
+    archived_at     DateTime DEFAULT now()
+) ENGINE = MergeTree
+  PARTITION BY toYYYYMM(created_at)
+  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
+    ba_archive.deliveries_per_company_daily_mv
+ENGINE = SummingMergeTree
+  PARTITION BY toYYYYMM(day)
+  ORDER BY (company_id, day, channel, status)
+AS
+SELECT
+    company_id,
+    toDate(created_at) AS day,
+    channel,
+    status,
+    count()           AS n
+FROM ba_archive.deliveries_archive
+GROUP BY company_id, day, channel, status;

scripts/m7_smoke.sh

@@ -0,0 +1,142 @@
+#!/usr/bin/env bash
+# Live M7 smoke test. Run from repo root:
+#   bash scripts/m7_smoke.sh
+#
+# Walks through the 4 scenarios in M7_VERIFICATION.md:
+#
+#   Step 2 — hypertable: query timescaledb_information.hypertables,
+#            expect `deliveries` listed.
+#   Step 3 — retention policy: query
+#            timescaledb_information.jobs, expect a 7-day
+#            retention policy on `deliveries`.
+#   Step 4 — archiver dry-run: insert a synthetic 8d-old
+#            row, run archiverd once, expect the row to
+#            move from Postgres to ClickHouse and the
+#            Postgres count to drop by 1.
+#   Step 5 — ClickHouse archive: SELECT count(*), min, max
+#            from ba_archive.deliveries_archive; expect
+#            ≥1 row with created_at in the 7-30d range.
+#
+# Exit code is the number of failed checks.
+
+set -e
+cd "$(dirname "$0")/.."
+
+PGCMD="docker exec -i broad-announce-postgres-1 psql -U ba -d ba -A -t"
+CHURL=http://localhost:8123
+CHCMD="curl -sS -X POST $CHURL/"
+
+fails=0
+pass() { echo "  ✅ $*"; }
+fail() { echo "  ❌ $*"; fails=$((fails+1)); }
+
+ch_query() {
+  $CHCMD --data-binary "$1" 2>/dev/null
+}
+
+# ── Setup: build archiverd fresh (must be v4 or later) ──
+ARCHIVERD_BIN=/tmp/archiverd-m7
+if [[ ! -x $ARCHIVERD_BIN ]]; then
+  echo "▸ Building $ARCHIVERD_BIN"
+  CGO_ENABLED=0 go build -o $ARCHIVERD_BIN ./cmd/archiverd
+fi
+# Always copy the fresh build into the container. Earlier
+# dev runs may have left a stale binary at /tmp/archiverd.
+docker exec broad-announce-ingestd-1 rm -f /tmp/archiverd 2>/dev/null || true
+
+# ─────────────────────────────────────────────────────────────────
+echo "── M7 smoke — Timescale 7d hot + ClickHouse archive ──"
+echo ""
+
+# ── Step 2: hypertable exists ──────────────────────────────
+echo "── Step 2: deliveries is a Timescale hypertable ──"
+hypertables=$($PGCMD -c "SELECT hypertable_name FROM timescaledb_information.hypertables;" 2>/dev/null | head -5)
+if echo "$hypertables" | grep -q "deliveries"; then
+  pass "deliveries is a hypertable"
+else
+  fail "deliveries NOT in hypertables list: $hypertables"
+fi
+
+# ── Step 3: retention policy is 7d ─────────────────────────
+echo ""
+echo "── Step 3: 7-day retention policy is active ──"
+jobs=$($PGCMD -c "SELECT application_name, config FROM timescaledb_information.jobs WHERE config::text LIKE '%7 days%';" 2>/dev/null | head -5)
+if echo "$jobs" | grep -q "7 days"; then
+  pass "7-day retention policy present ($jobs)"
+else
+  fail "7-day retention policy not found in jobs: $jobs"
+fi
+
+# ── Step 4: archiver moves a synthetic 8d-old row ─────────
+echo ""
+echo "── Step 4: 8d-old synthetic row moves to ClickHouse ──"
+# Insert a unique row, run archiverd, verify it moved.
+TAG="m7-step4-$(date +%s%N)"
+$PGCMD -c "INSERT INTO deliveries (alert_id, company_id, individual_id, channel, target, status, attempts, last_error, payload, created_at, sent_at) VALUES ('$TAG', 'acme-001', 'ind-acme-001', 'fcm', 'fake-token', 'sent', 0, '', '{\"smoke_tag\":\"$TAG\"}'::jsonb, now() - INTERVAL '8 days', now() - INTERVAL '8 days') ON CONFLICT DO NOTHING;" >/dev/null 2>&1
+pg_before=$($PGCMD -c "SELECT count(*) FROM deliveries WHERE alert_id = '$TAG';" 2>/dev/null | head -1)
+ch_before=$(ch_query "SELECT count(*) FROM ba_archive.deliveries_archive WHERE alert_id = '$TAG' FORMAT TabSeparated" 2>/dev/null | head -1)
+if [[ "$pg_before" == "1" ]]; then
+  pass "synthetic row in postgres (count=1)"
+else
+  fail "synthetic row not in postgres: count=$pg_before"
+fi
+
+# Run archiverd once. We use --port 8807 to avoid clashing
+# with anything left running in dev.
+docker exec broad-announce-ingestd-1 pkill -9 -f archiverd 2>/dev/null || true
+sleep 1
+docker cp $ARCHIVERD_BIN broad-announce-ingestd-1:/tmp/archiverd 2>/dev/null
+timeout 10 docker exec \
+  -e BA_ENV=dev -e BA_HTTP_ADDR=:8807 \
+  -e BA_POSTGRES_DSN="postgres://ba:ba@postgres:5432/ba?sslmode=disable" \
+  -e BA_ARCHIVERD_CLICKHOUSE_URL="http://clickhouse:8123" \
+  -e BA_ARCHIVERD_RUN_EVERY_SECONDS=999999 \
+  -e BA_ARCHIVERD_OLDER_THAN_HOURS=168 \
+  -e BA_ARCHIVERD_BATCH_SIZE=10000 \
+  -e BA_LOG_LEVEL=info \
+  broad-announce-ingestd-1 /tmp/archiverd > /tmp/m7-archiverd.log 2>&1 || true
+if grep -q "archiver run ok" /tmp/m7-archiverd.log; then
+  pass "archiverd run completed ok"
+else
+  fail "archiverd did not complete; log tail:"
+  tail -10 /tmp/m7-archiverd.log | sed 's/^/      /'
+fi
+
+pg_after=$($PGCMD -c "SELECT count(*) FROM deliveries WHERE alert_id = '$TAG';" 2>/dev/null | head -1)
+ch_after=$(ch_query "SELECT count(*) FROM ba_archive.deliveries_archive WHERE alert_id = '$TAG' FORMAT TabSeparated" 2>/dev/null | head -1)
+if [[ "$pg_after" == "0" ]]; then
+  pass "row moved out of postgres (count=0 after)"
+else
+  fail "row still in postgres: count=$pg_after"
+fi
+if [[ "$ch_after" == "1" ]]; then
+  pass "row landed in ClickHouse (count=1)"
+else
+  fail "row missing from ClickHouse: count=$ch_after (was $ch_before)"
+fi
+
+# ── Step 5: ClickHouse aggregate query ─────────────────────
+echo ""
+echo "── Step 5: ClickHouse archive queryable ──"
+archive_count=$(ch_query "SELECT count(*) FROM ba_archive.deliveries_archive FORMAT TabSeparated" 2>/dev/null | head -1)
+if [[ ${archive_count:-0} -ge 1 ]]; then
+  pass "ba_archive.deliveries_archive count = $archive_count (≥1 row archived)"
+else
+  fail "ClickHouse archive empty (count=$archive_count)"
+fi
+min_age=$(ch_query "SELECT dateDiff('day', min(created_at), now()) FROM ba_archive.deliveries_archive FORMAT TabSeparated" 2>/dev/null | head -1)
+if [[ ${min_age:-999} -ge 7 ]]; then
+  pass "oldest archived row is $min_age days old (≥7)"
+else
+  fail "oldest archived row is only $min_age days old (expected ≥7)"
+fi
+
+# ── Summary ─────────────────────────────────────────────────
+echo ""
+if [[ $fails -eq 0 ]]; then
+  echo "🟢 M7 smoke PASS — all checks green"
+  exit 0
+else
+  echo "🔴 M7 smoke FAIL — $fails check(s) failed"
+  exit $fails
+fi