broad-announce/cmd/admind/main.go

// 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/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 /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))
	mux.HandleFunc("GET /dlq", handleDLQUI(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{}