// Command admind is the admin HTTP API + (later) UI host. Tenant CRUD, // DLQ inspection, replay, audit log. // // M0: /health, /metrics, /v1/ping. // M8: + DLQ surface — // GET /v1/dlq — list/filter DLQ rows // GET /v1/dlq/{id} — single row (with payload) // POST /v1/dlq/{id}/replay — re-INSERT into the deliverd NATS subject // POST /v1/dlq/{id}/discard — mark discarded (hidden from default list) // GET /dlq — minimal HTML UI // // Auth: M8 ships without auth (LAN-only). The M11 work // gates /v1/dlq* behind an operator JWT. package main import ( "context" "embed" "encoding/json" "errors" "fmt" "html" "html/template" "log/slog" "net/http" "os" "os/signal" "strconv" "strings" "syscall" "time" "git3.techno-world.net/lrosales/broad-announce/internal/authd" "git3.techno-world.net/lrosales/broad-announce/internal/broker" "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" "git3.techno-world.net/lrosales/broad-announce/internal/postgres" "github.com/jackc/pgx/v5" "github.com/nats-io/nats.go" ) //go:embed ui/* var uiFS embed.FS // dlqRow is the wire shape returned by /v1/dlq and // rendered by the HTML UI. The payload field is omitted // from the list endpoint (operators can fetch the full // row via /v1/dlq/{id}). type dlqRow 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"` OriginalSubject string `json:"original_subject"` Attempts int `json:"attempts"` LastError string `json:"last_error"` Discarded bool `json:"discarded"` DiscardedAt *time.Time `json:"discarded_at,omitempty"` DiscardedBy string `json:"discarded_by,omitempty"` CreatedAt time.Time `json:"created_at"` Payload json.RawMessage `json:"payload,omitempty"` } func main() { cfg, err := config.LoadCommon("admind") if err != nil { os.Stderr.WriteString("config: " + err.Error() + "\n") os.Exit(1) } logger := observability.Init(cfg.Env, cfg.LogLevel, "admind") logger.Info("starting", "env", cfg.Env, "addr", cfg.HTTPAddr) ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() br, err := broker.Connect(ctx, cfg.NATSURL) if err != nil { logger.Error("nats connect", "err", err) os.Exit(1) } defer br.Close() pool, err := postgres.Connect(ctx, cfg.PostgresDSN) if err != nil { logger.Error("postgres connect", "err", err) os.Exit(1) } defer pool.Close() reg, _ := observability.NewRegistry("admind") srv := httpserver.New(httpserver.Config{ Addr: cfg.HTTPAddr, ServiceName: "admind", ShutdownGrace: cfg.ShutdownGrace, }, logger, observability.MetricsHandler(reg)) mux := srv.Mux() mux.HandleFunc("GET /v1/ping", handlePing) mux.HandleFunc("GET /dlq", handleDLQUI(pool, logger)) // M13a W3: JWT-gate the /v1/dlq* routes when BA_AUTHD_JWT_SECRET // is set. When unset, the routes stay unauthenticated (the // pre-M13 behavior) so the LAN-only deploy path keeps working. wireDLQRoutes(mux, br, pool, 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") } func handlePing(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ "pong": true, "service": "admind", "timestamp": time.Now().UTC().Format(time.RFC3339Nano), }) } // listFilters is the parsed query string for /v1/dlq. type listFilters struct { CompanyID string Channel string AlertID string Include string // "all" to show discarded; default hides them Limit int Offset int } func parseListFilters(r *http.Request) listFilters { q := r.URL.Query() limit, _ := strconv.Atoi(q.Get("limit")) if limit <= 0 || limit > 500 { limit = 100 } offset, _ := strconv.Atoi(q.Get("offset")) if offset < 0 { offset = 0 } return listFilters{ CompanyID: strings.TrimSpace(q.Get("company_id")), Channel: strings.TrimSpace(q.Get("channel")), AlertID: strings.TrimSpace(q.Get("alert_id")), Include: strings.TrimSpace(q.Get("include")), Limit: limit, Offset: offset, } } // handleListDLQ returns a paginated list of DLQ rows. // The list excludes discarded rows by default; pass // include=all to see them. func handleListDLQ(pool *postgres.Pool, logger *slog.Logger) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { f := parseListFilters(r) rows, err := queryDLQ(r.Context(), pool, dlqQuery{ CompanyID: f.CompanyID, Channel: f.Channel, AlertID: f.AlertID, IncludeDiscarded: f.Include == "all", Limit: f.Limit, Offset: f.Offset, }) if err != nil { logger.Error("dlq list", "err", err) http.Error(w, "dlq list: "+err.Error(), http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ "rows": rows, "limit": f.Limit, "offset": f.Offset, }) } } // handleGetDLQ returns one row by id, with the payload // included. Used by the UI's "view payload" expand. func handleGetDLQ(pool *postgres.Pool, logger *slog.Logger) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { id, ok := parseID(r) if !ok { http.Error(w, "bad id", http.StatusBadRequest) return } row, err := getDLQ(r.Context(), pool, id) if err != nil { logger.Error("dlq get", "err", err, "id", id) http.Error(w, "dlq get: "+err.Error(), http.StatusInternalServerError) return } if row == nil { http.Error(w, "not found", http.StatusNotFound) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(row) } } // handleReplayDLQ re-publishes the original NATS envelope // (stored in deliveries_dlq.payload) onto the original // subject, then marks the DLQ row discarded=true (with // discarded_by='replay') and writes a fresh audit row in // the live deliveries table. // // Idempotency: a single replay succeeds or fails. If it // fails, the DLQ row is NOT marked discarded so the // operator can retry. There is no race because the // discarded=true UPDATE happens in the same handler call // as the Publish; a second concurrent replay would // publish twice (the second one will hit dedupe at the // ingestd layer if the alert_id has been seen recently, // but for the DLQ replay path we don't dedupe — the // operator explicitly asked for a re-send). func handleReplayDLQ(br *broker.Client, pool *postgres.Pool, logger *slog.Logger) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { id, ok := parseID(r) if !ok { http.Error(w, "bad id", http.StatusBadRequest) return } row, err := getDLQ(r.Context(), pool, id) if err != nil { logger.Error("dlq replay get", "err", err, "id", id) http.Error(w, "dlq replay: "+err.Error(), http.StatusInternalServerError) return } if row == nil { http.Error(w, "not found", http.StatusNotFound) return } if row.Discarded { http.Error(w, "already discarded", http.StatusConflict) return } // Re-publish the original payload (json.RawMessage // holds the NATS envelope bytes verbatim). nc := br.NC() if err := nc.Publish(row.OriginalSubject, row.Payload); err != nil { logger.Error("dlq replay publish", "err", err, "id", id, "subject", row.OriginalSubject) http.Error(w, "publish: "+err.Error(), http.StatusBadGateway) return } // Flush so the message is on the wire before we // mark the row discarded. If the worker can't // deliver it, it will go back into the DLQ with // a fresh row, and the operator will see both. if err := nc.Flush(); err != nil { logger.Warn("dlq replay flush", "err", err, "id", id) } // Mark the DLQ row discarded. We do this AFTER // the publish so a publish failure leaves the // DLQ row in place for retry. now := time.Now().UTC() _, dbErr := pool.Exec(r.Context(), ` UPDATE deliveries_dlq SET discarded = true, discarded_at = $1, discarded_by = $2 WHERE id = $3 AND created_at IN ( SELECT created_at FROM deliveries_dlq WHERE id = $3 LIMIT 1 ) `, now, "replay", id) if dbErr != nil { // Publish succeeded but the DB update // didn't. The replay still happened — // log loud, return 200, and let the // operator handle the duplicate UI row // if they re-replay. logger.Error("dlq replay: publish ok but update failed", "err", dbErr, "id", id) } logger.Info("dlq replay ok", "id", id, "subject", row.OriginalSubject, "alert_id", row.AlertID) w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ "replayed": true, "id": id, "subject": row.OriginalSubject, "alert_id": row.AlertID, "company_id": row.CompanyID, "original_attempts": row.Attempts, }) } } // handleDiscardDLQ marks a DLQ row as discarded. After // discard, the row is hidden from /v1/dlq by default and // from the HTML UI. func handleDiscardDLQ(pool *postgres.Pool, logger *slog.Logger) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { id, ok := parseID(r) if !ok { http.Error(w, "bad id", http.StatusBadRequest) return } now := time.Now().UTC() // The (id, created_at) PK means the UPDATE // has to match the row's created_at too. We // fetch first, then UPDATE with both keys. row, err := getDLQ(r.Context(), pool, id) if err != nil { logger.Error("dlq discard get", "err", err, "id", id) http.Error(w, "dlq discard: "+err.Error(), http.StatusInternalServerError) return } if row == nil { http.Error(w, "not found", http.StatusNotFound) return } if row.Discarded { // Idempotent: a second discard is a no-op. w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{"discarded": true, "id": id, "already": true}) return } _, dbErr := pool.Exec(r.Context(), ` UPDATE deliveries_dlq SET discarded = true, discarded_at = $1, discarded_by = $2 WHERE id = $3 AND created_at = $4 `, now, "operator", id, row.CreatedAt) if dbErr != nil { logger.Error("dlq discard", "err", dbErr, "id", id) http.Error(w, "discard: "+dbErr.Error(), http.StatusInternalServerError) return } logger.Info("dlq discard ok", "id", id, "alert_id", row.AlertID) w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ "discarded": true, "id": id, "discarded_at": now, }) } } // dlqQuery is the structured query for the list endpoint. type dlqQuery struct { CompanyID string Channel string AlertID string IncludeDiscarded bool Limit int Offset int } // queryDLQ returns the rows that match the filters. // Excludes the payload from the list view (operators // fetch /v1/dlq/{id} for the full row). func queryDLQ(ctx context.Context, pool *postgres.Pool, q dlqQuery) ([]dlqRow, error) { // Build the WHERE clause dynamically. We always // filter on created_at > now - 30d for the list // view (operators rarely need ancient rows; the // archiver ships them to CH). conds := []string{"created_at > now() - INTERVAL '30 days'"} args := []any{} if q.CompanyID != "" { args = append(args, q.CompanyID) conds = append(conds, fmt.Sprintf("company_id = $%d", len(args))) } if q.Channel != "" { args = append(args, q.Channel) conds = append(conds, fmt.Sprintf("channel = $%d", len(args))) } if q.AlertID != "" { args = append(args, q.AlertID) conds = append(conds, fmt.Sprintf("alert_id = $%d", len(args))) } if !q.IncludeDiscarded { conds = append(conds, "discarded = false") } where := strings.Join(conds, " AND ") args = append(args, q.Limit, q.Offset) q1 := fmt.Sprintf(` SELECT id, alert_id, company_id, individual_id, channel, target, original_subject, attempts, last_error, discarded, discarded_at, discarded_by, created_at FROM deliveries_dlq WHERE %s ORDER BY created_at DESC LIMIT $%d OFFSET $%d `, where, len(args)-1, len(args)) rows, err := pool.Query(ctx, q1, args...) if err != nil { return nil, err } defer rows.Close() var out []dlqRow for rows.Next() { var r dlqRow var discardedAt *time.Time var discardedBy *string if err := rows.Scan( &r.ID, &r.AlertID, &r.CompanyID, &r.IndividualID, &r.Channel, &r.Target, &r.OriginalSubject, &r.Attempts, &r.LastError, &r.Discarded, &discardedAt, &discardedBy, &r.CreatedAt, ); err != nil { return nil, err } r.DiscardedAt = discardedAt if discardedBy != nil { r.DiscardedBy = *discardedBy } out = append(out, r) } return out, rows.Err() } // getDLQ returns one row by id (with payload). Returns // nil, nil if the row doesn't exist. func getDLQ(ctx context.Context, pool *postgres.Pool, id int64) (*dlqRow, error) { row := pool.QueryRow(ctx, ` SELECT id, alert_id, company_id, individual_id, channel, target, original_subject, attempts, last_error, payload, discarded, discarded_at, discarded_by, created_at FROM deliveries_dlq WHERE id = $1 LIMIT 1 `, id) var r dlqRow var payload []byte var discardedAt *time.Time var discardedBy *string if err := row.Scan( &r.ID, &r.AlertID, &r.CompanyID, &r.IndividualID, &r.Channel, &r.Target, &r.OriginalSubject, &r.Attempts, &r.LastError, &payload, &r.Discarded, &discardedAt, &discardedBy, &r.CreatedAt, ); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, nil } return nil, err } r.Payload = json.RawMessage(payload) r.DiscardedAt = discardedAt if discardedBy != nil { r.DiscardedBy = *discardedBy } return &r, nil } func parseID(r *http.Request) (int64, bool) { raw := r.PathValue("id") id, err := strconv.ParseInt(raw, 10, 64) if err != nil || id <= 0 { return 0, false } return id, true } // ── HTML UI ────────────────────────────────────────────────────── // uiData is the template data for the /dlq HTML page. type uiData struct { Rows []dlqRow Filters listFilters ShowAll bool HasFilters bool } // handleDLQUI serves a minimal HTML page with the DLQ // list and inline replay/discard buttons. POSTs go to // the same /v1/dlq/{id}/{replay,discard} endpoints; the // JS does a fetch and reloads the page. func handleDLQUI(pool *postgres.Pool, logger *slog.Logger) http.HandlerFunc { tpl := template.Must(template.New("dlq").Funcs(template.FuncMap{ "safeHTML": func(s string) template.HTML { return template.HTML(html.EscapeString(s)) }, "shortErr": func(s string) string { if len(s) > 80 { return s[:80] + "…" } return s }, "ago": func(t time.Time) string { d := time.Since(t) switch { case d < time.Minute: return fmt.Sprintf("%ds ago", int(d.Seconds())) case d < time.Hour: return fmt.Sprintf("%dm ago", int(d.Minutes())) case d < 24*time.Hour: return fmt.Sprintf("%dh ago", int(d.Hours())) default: return fmt.Sprintf("%dd ago", int(d.Hours()/24)) } }, }).ParseFS(uiFS, "ui/*.html")) return func(w http.ResponseWriter, r *http.Request) { f := parseListFilters(r) filters := dlqQuery{ CompanyID: f.CompanyID, Channel: f.Channel, AlertID: f.AlertID, IncludeDiscarded: f.Include == "all", Limit: f.Limit, Offset: f.Offset, } rows, err := queryDLQ(r.Context(), pool, filters) if err != nil { logger.Error("dlq ui list", "err", err) http.Error(w, "dlq ui: "+err.Error(), http.StatusInternalServerError) return } data := uiData{ Rows: rows, Filters: f, ShowAll: f.Include == "all", HasFilters: f.CompanyID != "" || f.Channel != "" || f.AlertID != "", } w.Header().Set("Content-Type", "text/html; charset=utf-8") if err := tpl.ExecuteTemplate(w, "dlq.html", data); err != nil { logger.Error("dlq ui render", "err", err) } } } // broker re-export so the handler signatures stay clean. // We import broker just for the Connect + Conn() pair. var _ = nats.Conn{} // wireDLQRoutes decides whether the /v1/dlq* routes go behind the // JWT gate or stay open, based on BA_AUTHD_JWT_SECRET. Extracted so // main() stays linear (no goto, no early returns from main). func wireDLQRoutes(mux *http.ServeMux, br *broker.Client, pool *postgres.Pool, logger *slog.Logger) { if !authd.EnvEnabled() { logger.Warn("dlq routes are UNAUTHENTICATED (set BA_AUTHD_JWT_SECRET to enable JWT gate)") mux.HandleFunc("GET /v1/dlq", handleListDLQ(pool, logger)) mux.HandleFunc("GET /v1/dlq/{id}", handleGetDLQ(pool, logger)) mux.HandleFunc("POST /v1/dlq/{id}/replay", handleReplayDLQ(br, pool, logger)) mux.HandleFunc("POST /v1/dlq/{id}/discard", handleDiscardDLQ(pool, logger)) return } ad, err := authd.NewFromEnv() if err != nil { logger.Error("BA_AUTHD_JWT_SECRET set but authd init failed; falling back to UNAUTHENTICATED routes", "err", err) mux.HandleFunc("GET /v1/dlq", handleListDLQ(pool, logger)) mux.HandleFunc("GET /v1/dlq/{id}", handleGetDLQ(pool, logger)) mux.HandleFunc("POST /v1/dlq/{id}/replay", handleReplayDLQ(br, pool, logger)) mux.HandleFunc("POST /v1/dlq/{id}/discard", handleDiscardDLQ(pool, logger)) return } logger.Info("dlq routes enabled with JWT gate") // replay and discard are destructive — require admin role. // list and get are read-only — any authenticated user. mux.Handle("GET /v1/dlq", ad.RequireAuth(http.HandlerFunc(handleListDLQ(pool, logger)))) mux.Handle("GET /v1/dlq/{id}", ad.RequireAuth(http.HandlerFunc(handleGetDLQ(pool, logger)))) mux.Handle("POST /v1/dlq/{id}/replay", ad.RequireRole("super_admin", "tenant_admin")(http.HandlerFunc(handleReplayDLQ(br, pool, logger)))) mux.Handle("POST /v1/dlq/{id}/discard", ad.RequireRole("super_admin", "tenant_admin")(http.HandlerFunc(handleDiscardDLQ(pool, logger)))) }