broad-announce/cmd/routerd/admin.go

// M13a W5: admin HTTP routes for routerd. Two routes are
// registered when BA_AUTHD_JWT_SECRET is set:
//
//	GET  /v1/admin/dedupe/state — number of pending collapse
//	  windows. Read-only. Any authenticated user.
//	POST /v1/admin/dedupe/flush — force-flush every pending
//	  collapse immediately (calls Collapser.FlushAll). Destructive:
//	  re-publishes all held alerts to NATS. super_admin or
//	  tenant_admin only.
//
// When BA_AUTHD_JWT_SECRET is unset, no admin routes are
// registered and the existing /health + /metrics keep working
// unchanged (LAN deploy path).
package main

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

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

// wireAdminRoutes wires the admin routes onto mux. 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,
	collapser *dedupe.Collapser,
	state *fanoutState,
	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")

	mux.Handle("GET /v1/admin/dedupe/state",
		ad.RequireAuth(http.HandlerFunc(handleDedupeState(collapser, state, logger))))
	mux.Handle("POST /v1/admin/dedupe/flush",
		ad.RequireRole("super_admin", "tenant_admin")(
			http.HandlerFunc(handleDedupeFlush(collapser, state, logger))))
}

// handleDedupeState returns the count of pending collapse windows.
// `collapser.Pending()` is the source of truth (it owns the
// per-(source,key) map); `state.len()` mirrors it but tracks
// the cached targets, so we report both for visibility.
func handleDedupeState(collapser *dedupe.Collapser, state *fanoutState, logger *slog.Logger) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		pending := collapser.Pending()
		cached := state.len()
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{
			"pending_collapses":    pending,
			"cached_target_lists":  cached,
		})
	}
}

// handleDedupeFlush force-flushes every pending collapse. The
// Collapser's Run loop also reads its own pending map and fires
// onFlush, so this is just a no-wait path to the same outcome.
// Safe to call when the map is empty (no-op).
func handleDedupeFlush(collapser *dedupe.Collapser, state *fanoutState, logger *slog.Logger) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		before := collapser.Pending()
		collapser.FlushAll()
		actor := ""
		if c := authd.ClaimsFromContext(r.Context()); c != nil {
			actor = c.UserID
		}
		logger.Info("admin dedupe flush",
			"pending_before", before,
			"actor", actor,
		)
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{
			"flushed":        true,
			"pending_before": before,
		})
	}
}