inventory-flo-debezium-nats/cmd/server/main.go

package main

import (
	"context"
	"embed"
	"encoding/json"
	"errors"
	"fmt"
	"html/template"
	"io/fs"
	"log"
	"net/http"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/nats-io/nats.go"
)

//go:embed internal/web/templates/index.html
var templateFS embed.FS

//go:embed internal/web/static
var staticFS embed.FS

type Config struct {
	HTTPAddr    string
	DatabaseURL string
	NATSURL     string
	NATSSubject string
}

const maxInventoryItems = 4

type InventoryItem struct {
	ID        int       `json:"id"`
	SKU       string    `json:"sku"`
	Name      string    `json:"name"`
	Quantity  int       `json:"quantity"`
	Location  string    `json:"location"`
	UpdatedAt time.Time `json:"updated_at"`
}

type AcceptedResponse struct {
	Status  string `json:"status"`
	CRUD    string `json:"crud"`
	Command string `json:"command"`
}

type CDCEvent struct {
	CRUD      string          `json:"crud"`
	Op        string          `json:"op"`
	Table     string          `json:"table"`
	Before    any             `json:"before"`
	After     any             `json:"after"`
	Raw       json.RawMessage `json:"raw"`
	Inventory []InventoryItem `json:"inventory"`
	Received  time.Time       `json:"received_at"`
	Source    string          `json:"source"`
}

type Server struct {
	db        *pgxpool.Pool
	broker    *Broker
	templates *template.Template
}

type Broker struct {
	mu      sync.Mutex
	clients map[chan CDCEvent]struct{}
	last    *CDCEvent
}

func main() {
	cfg := loadConfig()

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	db, err := connectDB(ctx, cfg.DatabaseURL)
	if err != nil {
		log.Fatalf("connect postgres: %v", err)
	}
	defer db.Close()

	tpl, err := template.ParseFS(templateFS, "internal/web/templates/index.html")
	if err != nil {
		log.Fatalf("parse templates: %v", err)
	}

	server := &Server{
		db:        db,
		broker:    NewBroker(),
		templates: tpl,
	}

	go consumeNATS(ctx, cfg, db, server.broker)

	mux := http.NewServeMux()
	server.routes(mux)

	httpServer := &http.Server{
		Addr:              cfg.HTTPAddr,
		Handler:           logRequests(mux),
		ReadHeaderTimeout: 5 * time.Second,
	}

	go func() {
		<-ctx.Done()
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = httpServer.Shutdown(shutdownCtx)
	}()

	log.Printf("starting live inventory CDC demo on %s", cfg.HTTPAddr)
	if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
		log.Fatalf("http server: %v", err)
	}
}

func loadConfig() Config {
	return Config{
		HTTPAddr:    env("HTTP_ADDR", ":8080"),
		DatabaseURL: env("DATABASE_URL", "postgres://postgres:postgres@localhost:5453/postgres?sslmode=disable"),
		NATSURL:     env("NATS_URL", "nats://localhost:4222"),
		NATSSubject: env("NATS_SUBJECT", "postgres.public.inventory"),
	}
}

func env(key, fallback string) string {
	if value := strings.TrimSpace(os.Getenv(key)); value != "" {
		return value
	}
	return fallback
}

func connectDB(ctx context.Context, databaseURL string) (*pgxpool.Pool, error) {
	var lastErr error
	for attempt := 1; attempt <= 30; attempt++ {
		db, err := pgxpool.New(ctx, databaseURL)
		if err == nil {
			pingCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
			err = db.Ping(pingCtx)
			cancel()
			if err == nil {
				return db, nil
			}
			db.Close()
		}
		lastErr = err
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-time.After(time.Duration(attempt) * 250 * time.Millisecond):
		}
	}
	return nil, lastErr
}

func (s *Server) routes(mux *http.ServeMux) {
	static, err := fs.Sub(staticFS, "internal/web/static")
	if err != nil {
		panic(err)
	}
	mux.Handle("GET /static/", http.StripPrefix("/static/", http.FileServerFS(static)))
	mux.HandleFunc("GET /", s.handleIndex)
	mux.HandleFunc("GET /health", s.handleHealth)
	mux.HandleFunc("GET /events", s.handleEvents)
	mux.HandleFunc("GET /api/inventory", s.handleListInventory)
	mux.HandleFunc("POST /api/inventory", s.handleCreateInventory)
	mux.HandleFunc("PATCH /api/inventory/{id}", s.handleUpdateInventory)
	mux.HandleFunc("PATCH /api/inventory/{id}/sell", s.handleSellInventory)
	mux.HandleFunc("PATCH /api/inventory/{id}/restock", s.handleRestockInventory)
	mux.HandleFunc("DELETE /api/inventory/{id}", s.handleDeleteInventory)
}

func (s *Server) handleIndex(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	data := map[string]string{"Version": "dev"}
	if err := s.templates.ExecuteTemplate(w, "index.html", data); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
	}
}

func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}

func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")

	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}

	client := s.broker.Subscribe()
	defer s.broker.Unsubscribe(client)

	if last := s.broker.Last(); last != nil {
		writeSSE(w, *last)
		flusher.Flush()
	}

	ticker := time.NewTicker(20 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-r.Context().Done():
			return
		case event := <-client:
			writeSSE(w, event)
			flusher.Flush()
		case <-ticker.C:
			_, _ = fmt.Fprint(w, ": keepalive\n\n")
			flusher.Flush()
		}
	}
}

func (s *Server) handleListInventory(w http.ResponseWriter, r *http.Request) {
	items, err := listInventory(r.Context(), s.db)
	if err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}
	writeJSON(w, http.StatusOK, items)
}

func (s *Server) handleCreateInventory(w http.ResponseWriter, r *http.Request) {
	var req struct {
		SKU      string `json:"sku"`
		Name     string `json:"name"`
		Quantity int    `json:"quantity"`
		Location string `json:"location"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, err)
		return
	}
	req.SKU = strings.TrimSpace(req.SKU)
	req.Name = strings.TrimSpace(req.Name)
	req.Location = strings.TrimSpace(req.Location)
	if req.SKU == "" || req.Name == "" || req.Location == "" {
		writeError(w, http.StatusBadRequest, errors.New("sku, name, and location are required"))
		return
	}

	tx, err := s.db.BeginTx(r.Context(), pgx.TxOptions{IsoLevel: pgx.Serializable})
	if err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}
	defer tx.Rollback(r.Context())

	if _, err := tx.Exec(r.Context(), `lock table inventory in exclusive mode`); err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}

	var count int
	if err := tx.QueryRow(r.Context(), `select count(*) from inventory`).Scan(&count); err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}
	if count >= maxInventoryItems {
		writeError(w, http.StatusConflict, fmt.Errorf("inventory is limited to %d products", maxInventoryItems))
		return
	}

	_, err = tx.Exec(r.Context(), `
		insert into inventory (sku, name, quantity, location, updated_at)
		values ($1, $2, $3, $4, now())`,
		req.SKU, req.Name, req.Quantity, req.Location,
	)
	if err != nil {
		writeError(w, http.StatusBadRequest, err)
		return
	}
	if err := tx.Commit(r.Context()); err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}

	writeJSON(w, http.StatusAccepted, AcceptedResponse{
		Status:  "command accepted",
		CRUD:    "CREATE",
		Command: fmt.Sprintf("CREATE inventory %s", req.SKU),
	})
}

func (s *Server) handleSellInventory(w http.ResponseWriter, r *http.Request) {
	id, ok := pathID(w, r)
	if !ok {
		return
	}
	_, err := s.db.Exec(r.Context(), `
		update inventory
		set quantity = greatest(quantity - 1, 0), updated_at = now()
		where id = $1`,
		id,
	)
	if err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}
	writeJSON(w, http.StatusAccepted, AcceptedResponse{
		Status:  "command accepted",
		CRUD:    "UPDATE",
		Command: "UPDATE inventory quantity -1",
	})
}

func (s *Server) handleRestockInventory(w http.ResponseWriter, r *http.Request) {
	id, ok := pathID(w, r)
	if !ok {
		return
	}
	var req struct {
		Amount int `json:"amount"`
	}
	_ = json.NewDecoder(r.Body).Decode(&req)
	if req.Amount <= 0 {
		req.Amount = 5
	}
	_, err := s.db.Exec(r.Context(), `
		update inventory
		set quantity = quantity + $2, updated_at = now()
		where id = $1`,
		id, req.Amount,
	)
	if err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}
	writeJSON(w, http.StatusAccepted, AcceptedResponse{
		Status:  "command accepted",
		CRUD:    "UPDATE",
		Command: fmt.Sprintf("UPDATE inventory quantity +%d", req.Amount),
	})
}

func (s *Server) handleUpdateInventory(w http.ResponseWriter, r *http.Request) {
	id, ok := pathID(w, r)
	if !ok {
		return
	}
	var req struct {
		Name     *string `json:"name"`
		Location *string `json:"location"`
		Quantity *int    `json:"quantity"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, err)
		return
	}

	tx, err := s.db.BeginTx(r.Context(), pgx.TxOptions{})
	if err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}
	defer tx.Rollback(r.Context())

	current := InventoryItem{}
	err = tx.QueryRow(r.Context(), `
		select id, sku, name, quantity, location, updated_at
		from inventory
		where id = $1
		for update`, id,
	).Scan(&current.ID, &current.SKU, &current.Name, &current.Quantity, &current.Location, &current.UpdatedAt)
	if err != nil {
		writeError(w, http.StatusNotFound, err)
		return
	}

	if req.Name != nil && strings.TrimSpace(*req.Name) != "" {
		current.Name = strings.TrimSpace(*req.Name)
	}
	if req.Location != nil && strings.TrimSpace(*req.Location) != "" {
		current.Location = strings.TrimSpace(*req.Location)
	}
	if req.Quantity != nil {
		current.Quantity = *req.Quantity
	}

	_, err = tx.Exec(r.Context(), `
		update inventory
		set name = $2, quantity = $3, location = $4, updated_at = now()
		where id = $1`,
		current.ID, current.Name, current.Quantity, current.Location,
	)
	if err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}
	if err := tx.Commit(r.Context()); err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}

	writeJSON(w, http.StatusAccepted, AcceptedResponse{
		Status:  "command accepted",
		CRUD:    "UPDATE",
		Command: fmt.Sprintf("UPDATE inventory %s", current.SKU),
	})
}

func (s *Server) handleDeleteInventory(w http.ResponseWriter, r *http.Request) {
	id, ok := pathID(w, r)
	if !ok {
		return
	}
	_, err := s.db.Exec(r.Context(), `delete from inventory where id = $1`, id)
	if err != nil {
		writeError(w, http.StatusInternalServerError, err)
		return
	}
	writeJSON(w, http.StatusAccepted, AcceptedResponse{
		Status:  "command accepted",
		CRUD:    "DELETE",
		Command: fmt.Sprintf("DELETE inventory id=%d", id),
	})
}

func pathID(w http.ResponseWriter, r *http.Request) (int, bool) {
	id, err := strconv.Atoi(r.PathValue("id"))
	if err != nil || id <= 0 {
		writeError(w, http.StatusBadRequest, errors.New("valid numeric id is required"))
		return 0, false
	}
	return id, true
}

func consumeNATS(ctx context.Context, cfg Config, db *pgxpool.Pool, broker *Broker) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}

		nc, err := nats.Connect(
			cfg.NATSURL,
			nats.Name("live-inventory-cdc-demo"),
			nats.RetryOnFailedConnect(true),
			nats.MaxReconnects(-1),
			nats.ReconnectWait(2*time.Second),
		)
		if err != nil {
			log.Printf("connect nats: %v", err)
			sleepOrDone(ctx, 2*time.Second)
			continue
		}

		sub, err := nc.Subscribe(cfg.NATSSubject, func(msg *nats.Msg) {
			event, err := normalizeCDC(ctx, db, msg.Data)
			if err != nil {
				log.Printf("normalize cdc event: %v", err)
				return
			}
			broker.Publish(event)
		})
		if err != nil {
			log.Printf("subscribe nats: %v", err)
			nc.Close()
			sleepOrDone(ctx, 2*time.Second)
			continue
		}
		log.Printf("subscribed to nats subject %s", cfg.NATSSubject)

		<-ctx.Done()
		_ = sub.Unsubscribe()
		nc.Close()
		return
	}
}

func normalizeCDC(ctx context.Context, db *pgxpool.Pool, payload []byte) (CDCEvent, error) {
	var raw map[string]any
	if err := json.Unmarshal(payload, &raw); err != nil {
		return CDCEvent{}, err
	}

	op, _ := raw["op"].(string)
	table := ""
	if source, ok := raw["source"].(map[string]any); ok {
		table, _ = source["table"].(string)
	}
	if table == "" {
		table = "inventory"
	}

	items, err := listInventory(ctx, db)
	if err != nil {
		return CDCEvent{}, err
	}

	return CDCEvent{
		CRUD:      crudLabel(op),
		Op:        op,
		Table:     table,
		Before:    raw["before"],
		After:     raw["after"],
		Raw:       append(json.RawMessage(nil), payload...),
		Inventory: items,
		Received:  time.Now().UTC(),
		Source:    "Debezium -> NATS JetStream -> SSE",
	}, nil
}

func crudLabel(op string) string {
	switch op {
	case "c":
		return "CREATE"
	case "r":
		return "SNAPSHOT"
	case "u":
		return "UPDATE"
	case "d":
		return "DELETE"
	default:
		return strings.ToUpper(op)
	}
}

func listInventory(ctx context.Context, db *pgxpool.Pool) ([]InventoryItem, error) {
	rows, err := db.Query(ctx, `
		select id, sku, name, quantity, location, updated_at
		from inventory
		order by id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]InventoryItem, 0)
	for rows.Next() {
		var item InventoryItem
		if err := rows.Scan(&item.ID, &item.SKU, &item.Name, &item.Quantity, &item.Location, &item.UpdatedAt); err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func NewBroker() *Broker {
	return &Broker{clients: make(map[chan CDCEvent]struct{})}
}

func (b *Broker) Subscribe() chan CDCEvent {
	ch := make(chan CDCEvent, 8)
	b.mu.Lock()
	b.clients[ch] = struct{}{}
	b.mu.Unlock()
	return ch
}

func (b *Broker) Unsubscribe(ch chan CDCEvent) {
	b.mu.Lock()
	delete(b.clients, ch)
	close(ch)
	b.mu.Unlock()
}

func (b *Broker) Publish(event CDCEvent) {
	b.mu.Lock()
	b.last = &event
	for ch := range b.clients {
		select {
		case ch <- event:
		default:
		}
	}
	b.mu.Unlock()
}

func (b *Broker) Last() *CDCEvent {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.last == nil {
		return nil
	}
	copy := *b.last
	return &copy
}

func writeSSE(w http.ResponseWriter, event CDCEvent) {
	data, err := json.Marshal(event)
	if err != nil {
		return
	}
	_, _ = fmt.Fprintf(w, "event: cdc\n")
	_, _ = fmt.Fprintf(w, "data: %s\n\n", data)
}

func writeJSON(w http.ResponseWriter, status int, payload any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(payload)
}

func writeError(w http.ResponseWriter, status int, err error) {
	writeJSON(w, status, map[string]string{"error": err.Error()})
}

func sleepOrDone(ctx context.Context, d time.Duration) {
	select {
	case <-ctx.Done():
	case <-time.After(d):
	}
}

func logRequests(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		start := time.Now()
		next.ServeHTTP(w, r)
		log.Printf("%s %s %s", r.Method, r.URL.Path, time.Since(start).Round(time.Millisecond))
	})
}