broad-announce/cmd/archiverd/main.go

// 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)

	// M13a W5: trigger channel for the admin route. Buffered
	// to size 1 — concurrent trigger requests coalesce.
	triggerCh := make(chan struct{}, 1)

	// 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, triggerCh)

	srv := httpserver.New(httpserver.Config{
		Addr:          cfg.HTTPAddr,
		ServiceName:   "archiverd",
		ShutdownGrace: cfg.ShutdownGrace,
	}, logger, metricsHandler(reg, &lastRunUnix, cfg.RunEverySeconds*2))

	// M13a W5: admin routes (JWT-gated). When BA_AUTHD_JWT_SECRET
	// is unset, wireAdminRoutes is a no-op so the LAN deploy
	// path keeps working unchanged.
	wireAdminRoutes(srv.Mux(), triggerCh, logger)

	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. The
// M13a W5 admin route can also fire a run via triggerCh.
func runLoop(
	ctx context.Context,
	logger *slog.Logger,
	opts archiver.RunOptions,
	rowsArchived *prometheus.CounterVec,
	lastRunUnix *atomic.Int64,
	runDuration prometheus.Gauge,
	triggerCh <-chan struct{},
) {
	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:
			// scheduled tick
		case <-triggerCh:
			// M13a W5: admin trigger. Falls through to
			// the loop body, which calls oneRun again.
			logger.Info("archiver run triggered by admin route")
		}
	}
}

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