admin.go 3.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091
  1. // M13a W5: admin HTTP routes for routerd. Two routes are
  2. // registered when BA_AUTHD_JWT_SECRET is set:
  3. //
  4. // GET /v1/admin/dedupe/state — number of pending collapse
  5. // windows. Read-only. Any authenticated user.
  6. // POST /v1/admin/dedupe/flush — force-flush every pending
  7. // collapse immediately (calls Collapser.FlushAll). Destructive:
  8. // re-publishes all held alerts to NATS. super_admin or
  9. // tenant_admin only.
  10. //
  11. // When BA_AUTHD_JWT_SECRET is unset, no admin routes are
  12. // registered and the existing /health + /metrics keep working
  13. // unchanged (LAN deploy path).
  14. package main
  15. import (
  16. "encoding/json"
  17. "log/slog"
  18. "net/http"
  19. "git3.techno-world.net/lrosales/broad-announce/internal/authd"
  20. "git3.techno-world.net/lrosales/broad-announce/internal/dedupe"
  21. )
  22. // wireAdminRoutes wires the admin routes onto mux. When the JWT
  23. // gate is disabled (env not set), the function logs a warning and
  24. // returns without registering anything — backward compat for
  25. // non-M13 deploys.
  26. func wireAdminRoutes(
  27. mux *http.ServeMux,
  28. collapser *dedupe.Collapser,
  29. state *fanoutState,
  30. logger *slog.Logger,
  31. ) {
  32. if !authd.EnvEnabled() {
  33. logger.Warn("admin routes are DISABLED (set BA_AUTHD_JWT_SECRET to enable)")
  34. return
  35. }
  36. ad, err := authd.NewFromEnv()
  37. if err != nil {
  38. logger.Error("BA_AUTHD_JWT_SECRET set but authd init failed; admin routes DISABLED", "err", err)
  39. return
  40. }
  41. logger.Info("admin routes enabled with JWT gate")
  42. mux.Handle("GET /v1/admin/dedupe/state",
  43. ad.RequireAuth(http.HandlerFunc(handleDedupeState(collapser, state, logger))))
  44. mux.Handle("POST /v1/admin/dedupe/flush",
  45. ad.RequireRole("super_admin", "tenant_admin")(
  46. http.HandlerFunc(handleDedupeFlush(collapser, state, logger))))
  47. }
  48. // handleDedupeState returns the count of pending collapse windows.
  49. // `collapser.Pending()` is the source of truth (it owns the
  50. // per-(source,key) map); `state.len()` mirrors it but tracks
  51. // the cached targets, so we report both for visibility.
  52. func handleDedupeState(collapser *dedupe.Collapser, state *fanoutState, logger *slog.Logger) http.HandlerFunc {
  53. return func(w http.ResponseWriter, r *http.Request) {
  54. pending := collapser.Pending()
  55. cached := state.len()
  56. w.Header().Set("Content-Type", "application/json")
  57. _ = json.NewEncoder(w).Encode(map[string]any{
  58. "pending_collapses": pending,
  59. "cached_target_lists": cached,
  60. })
  61. }
  62. }
  63. // handleDedupeFlush force-flushes every pending collapse. The
  64. // Collapser's Run loop also reads its own pending map and fires
  65. // onFlush, so this is just a no-wait path to the same outcome.
  66. // Safe to call when the map is empty (no-op).
  67. func handleDedupeFlush(collapser *dedupe.Collapser, state *fanoutState, logger *slog.Logger) http.HandlerFunc {
  68. return func(w http.ResponseWriter, r *http.Request) {
  69. before := collapser.Pending()
  70. collapser.FlushAll()
  71. actor := ""
  72. if c := authd.ClaimsFromContext(r.Context()); c != nil {
  73. actor = c.UserID
  74. }
  75. logger.Info("admin dedupe flush",
  76. "pending_before", before,
  77. "actor", actor,
  78. )
  79. w.Header().Set("Content-Type", "application/json")
  80. _ = json.NewEncoder(w).Encode(map[string]any{
  81. "flushed": true,
  82. "pending_before": before,
  83. })
  84. }
  85. }