Dashboard redesign: React + shadcn/ui, live charts, history API, DLQ inspector

CLAUDE.md

@@ -142,7 +142,7 @@ internal/
     dlq/               # Dead-letter queue (Redis store, parking decorator, triage)
     server/            # HTTP server: health/readiness/metrics, APIs, dashboard
     metrics/           # Prometheus metrics
-ui/                    # Svelte 5 + Tailwind dashboard; dist/ committed + embedded
+ui/                    # React 19 + Tailwind dashboard; dist/ committed + embedded
 ```
 
 ## Adding New Sinks

CONTRIBUTING.md

@@ -43,7 +43,8 @@ docker compose -f docker-compose-local.yml exec redis \
 
 ## Working on the dashboard
 
-The dashboard is a Svelte 5 + Tailwind SPA in `ui/`, embedded into the
+The dashboard is a React 19 + TypeScript + Tailwind SPA in `ui/` (shadcn/ui,
+TanStack Table/Query, react-hook-form + zod), embedded into the
 binary PocketBase-style: the built `ui/dist/` is **committed**, so plain
 `go build` needs no Node toolchain.
 

README.md

@@ -58,14 +58,16 @@ PostgreSQL ──logical replication──▶ GTC ──▶ Redis Streams   (eve
   forever, then inspected, retried, or discarded over HTTP. Sink *outages*
   still stall (a down sink is not a poison event) — nothing is ever
   silently dropped.
-- **Embedded live dashboard** — a Svelte dashboard served from the binary at
+- **Embedded live dashboard** — a React dashboard served from the binary at
   `/` (PocketBase-style, no separate deployment): live throughput, WAL lag,
   sink health and breaker states, per-table activity, backfill progress, and
   DLQ triage with one-click retry/discard. Light and dark themes.
 - **Observability** — Prometheus metrics (events, per-sink latency, errors,
   retries, breaker state, WAL lag in bytes), health and readiness endpoints,
   structured JSON logs.
 
+![GTC dashboard — pipeline overview](docs/dashboard.png)
+
 ## Quick start
 
 ### Docker Compose
@@ -401,7 +403,7 @@ Set `CDC_DLQ_ENABLED=false` to opt out and always stall on failure.
 Open `http://localhost:8080/` for the embedded dashboard: throughput with a
 live sparkline, WAL lag, in-flight events, sink health with circuit-breaker
 states, per-table operation counts, backfill progress (with a "sync all"
-trigger), and dead-letter triage with retry/discard buttons. It's a Svelte
+trigger), and dead-letter triage with retry/discard buttons. It's a React
 SPA compiled into the binary — nothing extra to deploy, and it polls the
 same public API you can script against.
 
@@ -453,7 +455,7 @@ internal/
     config/            # Env + sink YAML loading
     resilience/        # Retry + circuit breaker wrapper for sinks
     dlq/               # Dead-letter queue: parking decorator + triage API backend
-ui/                    # Svelte dashboard (built dist/ is committed and embedded)
+ui/                    # React dashboard (built dist/ is committed and embedded)
     server/            # HTTP health/readiness/metrics
     metrics/           # Prometheus collectors
 ```

cmd/gateway/modules.go

@@ -342,7 +342,12 @@ func newHTTPServer(
 		Backfill: backfillManager,
 		DLQ:      dlqPort,
 		WAL:      reader,
-		Logger:   logger,
+		Info: server.InstanceInfo{
+			Version:     version,
+			SlotName:    cfg.SlotName,
+			Publication: cfg.PublicationName,
+		},
+		Logger: logger,
 	}
 	if dist, ok := ui.DistFS(); ok {
 		params.UI = dist

internal/infrastructure/server/history.go

@@ -0,0 +1,106 @@
+package server
+
+import (
+	"encoding/json"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// sample is one point of dashboard history. Counters are cumulative; the UI
+// derives rates from deltas so the server stays stateless about windows.
+type sample struct {
+	T           int64   `json:"t"` // unix milliseconds
+	EventsTotal float64 `json:"events_total"`
+	WALLagBytes float64 `json:"wal_lag_bytes"`
+	Inflight    float64 `json:"inflight"`
+	DLQEntries  float64 `json:"dlq_entries"`
+}
+
+const (
+	samplerInterval = 5 * time.Second
+	samplerCapacity = 360 // 30 minutes at 5s
+)
+
+// sampler keeps a ring buffer of pipeline history so charts have real data
+// the moment the dashboard opens, surviving page reloads.
+type sampler struct {
+	mu      sync.Mutex
+	samples []sample
+	stop    chan struct{}
+	once    sync.Once
+}
+
+func newSampler() *sampler {
+	return &sampler{
+		samples: make([]sample, 0, samplerCapacity),
+		stop:    make(chan struct{}),
+	}
+}
+
+func (s *sampler) run() {
+	ticker := time.NewTicker(samplerInterval)
+	defer ticker.Stop()
+
+	s.collect()
+	for {
+		select {
+		case <-s.stop:
+			return
+		case <-ticker.C:
+			s.collect()
+		}
+	}
+}
+
+func (s *sampler) close() {
+	s.once.Do(func() { close(s.stop) })
+}
+
+func (s *sampler) collect() {
+	point := sample{T: time.Now().UnixMilli()}
+
+	families, err := prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		return
+	}
+	for _, fam := range families {
+		switch fam.GetName() {
+		case "cdc_events_received_total":
+			for _, m := range fam.GetMetric() {
+				point.EventsTotal += metricValue(m)
+			}
+		case "cdc_wal_lag_bytes":
+			point.WALLagBytes = singleValue(fam)
+		case "cdc_inflight_events":
+			point.Inflight = singleValue(fam)
+		case "cdc_dlq_entries":
+			point.DLQEntries = singleValue(fam)
+		}
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.samples = append(s.samples, point)
+	if len(s.samples) > samplerCapacity {
+		s.samples = s.samples[len(s.samples)-samplerCapacity:]
+	}
+}
+
+func (s *sampler) snapshot() []sample {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]sample, len(s.samples))
+	copy(out, s.samples)
+	return out
+}
+
+func (s *Server) handleHistory(w http.ResponseWriter, _ *http.Request) {
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(map[string]any{
+		"interval_seconds": samplerInterval.Seconds(),
+		"samples":          s.sampler.snapshot(),
+	})
+}

internal/infrastructure/server/server.go

@@ -27,6 +27,15 @@ type Server struct {
 	wal        WALInfo
 	ui         fs.FS
 	startedAt  time.Time
+	sampler    *sampler
+	info       InstanceInfo
+}
+
+// InstanceInfo is static build/config context surfaced to the dashboard.
+type InstanceInfo struct {
+	Version     string
+	SlotName    string
+	Publication string
 }
 
 type HealthChecker interface {
@@ -45,6 +54,7 @@ type ServerParams struct {
 	DLQ      ports.DLQManager      // nil when the DLQ is disabled
 	WAL      WALInfo               // reader state for /api/stats
 	UI       fs.FS                 // embedded dashboard; nil disables it
+	Info     InstanceInfo
 	Logger   *slog.Logger
 }
 
@@ -65,6 +75,8 @@ func New(p ServerParams) *Server {
 		wal:       p.WAL,
 		ui:        p.UI,
 		startedAt: time.Now(),
+		sampler:   newSampler(),
+		info:      p.Info,
 		httpServer: &http.Server{
 			Addr:         fmt.Sprintf(":%d", p.Config.Port),
 			Handler:      r,
@@ -88,6 +100,7 @@ func (s *Server) registerRoutes() {
 	s.router.Post("/dlq/retry", s.handleDLQRetry)
 	s.router.Post("/dlq/discard", s.handleDLQDiscard)
 	s.router.Get("/api/stats", s.handleStats)
+	s.router.Get("/api/history", s.handleHistory)
 
 	if s.ui != nil {
 		s.router.Get("/*", s.handleUI())
@@ -285,6 +298,7 @@ func (s *Server) handleReadiness(w http.ResponseWriter, r *http.Request) {
 
 func (s *Server) Start() error {
 	s.logger.Info("starting HTTP server", slog.String("addr", s.httpServer.Addr))
+	go s.sampler.run()
 	if err := s.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 		return err
 	}
@@ -293,6 +307,7 @@ func (s *Server) Start() error {
 
 func (s *Server) Stop(ctx context.Context) error {
 	s.logger.Info("stopping HTTP server")
+	s.sampler.close()
 	return s.httpServer.Shutdown(ctx)
 }
 

internal/infrastructure/server/stats.go

@@ -18,13 +18,14 @@ type WALInfo interface {
 }
 
 type sinkStats struct {
-	Name         string  `json:"name"`
-	Healthy      bool    `json:"healthy"`
-	BreakerState float64 `json:"breaker_state"` // 0 closed, 1 half-open, 2 open
-	Succeeded    float64 `json:"succeeded"`
-	Failed       float64 `json:"failed"`
-	Filtered     float64 `json:"filtered"`
-	Retries      float64 `json:"retries"`
+	Name         string             `json:"name"`
+	Healthy      bool               `json:"healthy"`
+	BreakerState float64            `json:"breaker_state"` // 0 closed, 1 half-open, 2 open
+	Succeeded    float64            `json:"succeeded"`
+	Failed       float64            `json:"failed"`
+	Filtered     float64            `json:"filtered"`
+	Retries      float64            `json:"retries"`
+	ErrorsByType map[string]float64 `json:"errors_by_type"`
 }
 
 type tableStats struct {
@@ -39,21 +40,27 @@ type tableStats struct {
 
 type statsResponse struct {
 	UptimeSeconds float64                      `json:"uptime_seconds"`
+	Version       string                       `json:"version"`
+	SlotName      string                       `json:"slot_name"`
+	Publication   string                       `json:"publication"`
 	Ready         bool                         `json:"ready"`
 	Streaming     bool                         `json:"streaming"`
 	CurrentLSN    string                       `json:"current_lsn"`
 	WALLagBytes   float64                      `json:"wal_lag_bytes"`
 	Inflight      float64                      `json:"inflight"`
 	EventsTotal   float64                      `json:"events_total"`
+	Operations    map[string]float64           `json:"operations"`
 	Sinks         []sinkStats                  `json:"sinks"`
 	Tables        []tableStats                 `json:"tables"`
 	Backfill      []domain.BackfillTableStatus `json:"backfill"`
 	DLQ           dlqStats                     `json:"dlq"`
 }
 
 type dlqStats struct {
-	Enabled bool  `json:"enabled"`
-	Entries int64 `json:"entries"`
+	Enabled      bool    `json:"enabled"`
+	Entries      int64   `json:"entries"`
+	ParkedTotal  float64 `json:"parked_total"`
+	RetriedTotal float64 `json:"retried_total"`
 }
 
 // handleStats shapes live state (reader, health, backfill, DLQ) plus the
@@ -63,6 +70,10 @@ func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) {
 
 	resp := statsResponse{
 		UptimeSeconds: time.Since(s.startedAt).Seconds(),
+		Version:       s.info.Version,
+		SlotName:      s.info.SlotName,
+		Publication:   s.info.Publication,
+		Operations:    map[string]float64{},
 		Sinks:         []sinkStats{},
 		Tables:        []tableStats{},
 		Backfill:      []domain.BackfillTableStatus{},
@@ -123,7 +134,9 @@ func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) {
 					}
 					v := metricValue(m)
 					ts.Total += v
-					switch labels["operation"] {
+					op := labels["operation"]
+					resp.Operations[op] += v
+					switch op {
 					case "INSERT":
 						ts.Insert += v
 					case "UPDATE":
@@ -159,6 +172,23 @@ func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) {
 				for _, m := range fam.GetMetric() {
 					sink(labelMap(m)["sink"]).BreakerState = metricValue(m)
 				}
+			case "cdc_sink_errors_total":
+				for _, m := range fam.GetMetric() {
+					labels := labelMap(m)
+					st := sink(labels["sink"])
+					if st.ErrorsByType == nil {
+						st.ErrorsByType = map[string]float64{}
+					}
+					st.ErrorsByType[labels["error_type"]] += metricValue(m)
+				}
+			case "cdc_dlq_parked_total":
+				for _, m := range fam.GetMetric() {
+					resp.DLQ.ParkedTotal += metricValue(m)
+				}
+			case "cdc_dlq_retried_total":
+				for _, m := range fam.GetMetric() {
+					resp.DLQ.RetriedTotal += metricValue(m)
+				}
 			}
 		}
 	}