broad-announce/cmd/archiverd/admin.go

// M13a W5: admin HTTP routes for archiverd. One route is
// registered when BA_AUTHD_JWT_SECRET is set:
//
//	POST /v1/admin/archiver/run — trigger an immediate archiver
//	  pass (instead of waiting for the next ticker fire). Destructive:
//	  takes the Postgres advisory lock and runs a full drain. Any
//	  authenticated user can fire it; the existing runLock
//	  guarantees no two pass can overlap.
//
// When BA_AUTHD_JWT_SECRET is unset, no admin routes are
// registered. The /health and /metrics endpoints are owned by
// the shared metricsHandler and continue to work as before.
package main

import (
	"encoding/json"
	"log/slog"
	"net/http"

	"git3.techno-world.net/lrosales/broad-announce/internal/authd"
)

// wireAdminRoutes wires the admin routes onto mux. The triggerCh
// is a buffered channel of size 1; a non-blocking send coalesces
// concurrent trigger requests (a run that's already in progress
// will complete, and any queued trigger is dropped — the next
// ticker will fire normally).
//
// When the JWT gate is disabled (env not set), the function logs
// a warning and returns without registering anything — backward
// compat for non-M13 deploys.
func wireAdminRoutes(mux *http.ServeMux, triggerCh chan<- struct{}, logger *slog.Logger) {
	if !authd.EnvEnabled() {
		logger.Warn("admin routes are DISABLED (set BA_AUTHD_JWT_SECRET to enable)")
		return
	}
	ad, err := authd.NewFromEnv()
	if err != nil {
		logger.Error("BA_AUTHD_JWT_SECRET set but authd init failed; admin routes DISABLED", "err", err)
		return
	}
	logger.Info("admin routes enabled with JWT gate")
	// Any authenticated user can fire an archiver run; it's not
	// state-destroying (the archiver is idempotent thanks to the
	// Postgres advisory lock + the SELECT/DELETE contract).
	mux.Handle("POST /v1/admin/archiver/run",
		ad.RequireAuth(http.HandlerFunc(handleRunNow(triggerCh, logger))))
}

// handleRunNow signals the runLoop to fire a pass. The send is
// non-blocking: if a trigger is already pending (a previous
// request fired and the loop hasn't picked it up yet), we report
// it back to the caller instead of queueing duplicates.
func handleRunNow(triggerCh chan<- struct{}, logger *slog.Logger) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		actor := ""
		if c := authd.ClaimsFromContext(r.Context()); c != nil {
			actor = c.UserID
		}
		select {
		case triggerCh <- struct{}{}:
			logger.Info("admin archiver run triggered", "actor", actor)
			w.Header().Set("Content-Type", "application/json")
			_ = json.NewEncoder(w).Encode(map[string]any{
				"triggered": true,
			})
		default:
			// Trigger already pending or run in progress.
			// Coalesce — return 202 Accepted with a hint.
			logger.Info("admin archiver run coalesced (already pending)", "actor", actor)
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(http.StatusAccepted)
			_ = json.NewEncoder(w).Encode(map[string]any{
				"triggered": false,
				"reason":    "already_pending",
			})
		}
	}
}