Nereus/apps/api/http.go

package main

import (
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"math"
	mathrand "math/rand/v2"
	"net/http"
	"strconv"
	"strings"
	"sync/atomic"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/jackc/pgx/v5"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
	"go.opentelemetry.io/otel/trace"
)

type api struct {
	store      store
	metrics    *metrics
	logger     *slog.Logger
	version    string
	chaosRate  float64
	migrated   *atomic.Bool
	prometheus prometheus.Gatherer
}

func (a *api) routes() http.Handler {
	r := chi.NewRouter()
	r.Use(a.versionHeader)
	r.Use(a.observeRequest)
	r.Get("/healthz", func(w http.ResponseWriter, _ *http.Request) {
		writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
	})
	r.Get("/readyz", a.ready)
	r.Handle("/metrics", promhttp.HandlerFor(a.prometheus, promhttp.HandlerOpts{}))
	r.Route("/api/v1", func(r chi.Router) {
		r.Use(a.chaos)
		r.Get("/buoys", a.listBuoys)
		r.Post("/buoys", a.createBuoy)
		r.Get("/buoys/{id}", a.getBuoy)
		r.Delete("/buoys/{id}", a.deleteBuoy)
		r.Get("/readings", a.listReadings)
		r.Post("/readings", a.createReading)
		r.Get("/readings/aggregate", a.aggregateReadings)
	})
	return otelhttp.NewHandler(r, "http.request")
}

func (a *api) versionHeader(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("X-Nereus-Version", a.version)
		next.ServeHTTP(w, r)
	})
}

type responseRecorder struct {
	http.ResponseWriter
	status int
}

func (w *responseRecorder) WriteHeader(status int) {
	w.status = status
	w.ResponseWriter.WriteHeader(status)
}

func (a *api) observeRequest(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		started := time.Now()
		recorder := &responseRecorder{ResponseWriter: w, status: http.StatusOK}
		next.ServeHTTP(recorder, r)
		path := chi.RouteContext(r.Context()).RoutePattern()
		if path == "" {
			path = "unmatched"
		}
		status := strconv.Itoa(recorder.status)
		a.metrics.requests.WithLabelValues(r.Method, path, status, a.version).Inc()
		a.metrics.duration.WithLabelValues(r.Method, path, a.version).Observe(time.Since(started).Seconds())
		span := trace.SpanFromContext(r.Context()).SpanContext()
		a.logger.InfoContext(r.Context(), "request", "method", r.Method, "path", path, "status", recorder.status, "duration_ms", time.Since(started).Milliseconds(), "trace_id", span.TraceID().String())
	})
}

func (a *api) chaos(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if a.chaosRate > 0 && mathrand.Float64() < a.chaosRate {
			writeError(w, http.StatusInternalServerError, "injected failure")
			return
		}
		next.ServeHTTP(w, r)
	})
}

func (a *api) ready(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := contextWithTimeout(r, 2*time.Second)
	defer cancel()
	if !a.migrated.Load() || a.store.Ping(ctx) != nil {
		writeError(w, http.StatusServiceUnavailable, "database unavailable")
		return
	}
	writeJSON(w, http.StatusOK, map[string]string{"status": "ready"})
}

func (a *api) listBuoys(w http.ResponseWriter, r *http.Request) {
	items, err := a.store.ListBuoys(r.Context())
	if err != nil {
		a.internalError(w, r, err)
		return
	}
	if items == nil {
		items = []buoy{}
	}
	a.metrics.active.Set(float64(len(items)))
	writeJSON(w, http.StatusOK, items)
}

func (a *api) createBuoy(w http.ResponseWriter, r *http.Request) {
	var input buoy
	if !decodeJSON(w, r, &input) {
		return
	}
	input.Name = strings.TrimSpace(input.Name)
	if input.Name == "" || len(input.Name) > 120 || input.Latitude < -90 || input.Latitude > 90 || input.Longitude < -180 || input.Longitude > 180 {
		writeError(w, http.StatusBadRequest, "invalid buoy")
		return
	}
	id, err := newUUID()
	if err != nil {
		a.internalError(w, r, err)
		return
	}
	input.ID = id
	created, err := a.store.CreateBuoy(r.Context(), input)
	if err != nil {
		a.internalError(w, r, err)
		return
	}
	a.metrics.active.Inc()
	writeJSON(w, http.StatusCreated, created)
}

func (a *api) getBuoy(w http.ResponseWriter, r *http.Request) {
	id := chi.URLParam(r, "id")
	if !validUUID(id) {
		writeError(w, http.StatusBadRequest, "invalid buoy ID")
		return
	}
	item, err := a.store.GetBuoy(r.Context(), id)
	if errors.Is(err, pgx.ErrNoRows) {
		writeError(w, http.StatusNotFound, "buoy not found")
		return
	}
	if err != nil {
		a.internalError(w, r, err)
		return
	}
	writeJSON(w, http.StatusOK, item)
}

func (a *api) deleteBuoy(w http.ResponseWriter, r *http.Request) {
	id := chi.URLParam(r, "id")
	if !validUUID(id) {
		writeError(w, http.StatusBadRequest, "invalid buoy ID")
		return
	}
	err := a.store.DeleteBuoy(r.Context(), id)
	if errors.Is(err, pgx.ErrNoRows) {
		writeError(w, http.StatusNotFound, "buoy not found")
		return
	}
	if err != nil {
		a.internalError(w, r, err)
		return
	}
	a.metrics.active.Dec()
	w.WriteHeader(http.StatusNoContent)
}

func (a *api) listReadings(w http.ResponseWriter, r *http.Request) {
	query := r.URL.Query()
	from, ok := optionalTime(w, query.Get("from"))
	if !ok {
		return
	}
	to, ok := optionalTime(w, query.Get("to"))
	if !ok {
		return
	}
	if from != nil && to != nil && from.After(*to) {
		writeError(w, http.StatusBadRequest, "from must not be after to")
		return
	}
	if id := query.Get("buoy_id"); id != "" && !validUUID(id) {
		writeError(w, http.StatusBadRequest, "invalid buoy ID")
		return
	}
	limit, ok := boundedInt(query.Get("limit"), 100, 1, 1000)
	if !ok {
		writeError(w, http.StatusBadRequest, "invalid limit")
		return
	}
	offset, ok := boundedInt(query.Get("offset"), 0, 0, math.MaxInt)
	if !ok {
		writeError(w, http.StatusBadRequest, "invalid offset")
		return
	}
	items, err := a.store.ListReadings(r.Context(), query.Get("buoy_id"), from, to, limit, offset)
	if err != nil {
		a.internalError(w, r, err)
		return
	}
	if items == nil {
		items = []reading{}
	}
	writeJSON(w, http.StatusOK, items)
}

func (a *api) createReading(w http.ResponseWriter, r *http.Request) {
	var input reading
	if !decodeJSON(w, r, &input) {
		return
	}
	if !validUUID(input.BuoyID) || input.WaveHeight < 0 || input.Salinity != nil && *input.Salinity < 0 {
		writeError(w, http.StatusBadRequest, "invalid reading")
		return
	}
	id, err := newUUID()
	if err != nil {
		a.internalError(w, r, err)
		return
	}
	input.ID = id
	created, err := a.store.CreateReading(r.Context(), input)
	if err != nil {
		a.internalError(w, r, err)
		return
	}
	a.metrics.ingested.Inc()
	writeJSON(w, http.StatusCreated, created)
}

func (a *api) aggregateReadings(w http.ResponseWriter, r *http.Request) {
	window, err := time.ParseDuration(r.URL.Query().Get("window"))
	if err != nil || window < time.Minute || window > 24*time.Hour {
		writeError(w, http.StatusBadRequest, "window must be between 1m and 24h")
		return
	}
	items, err := a.store.AggregateReadings(r.Context(), window)
	if err != nil {
		a.internalError(w, r, err)
		return
	}
	if items == nil {
		items = []aggregate{}
	}
	writeJSON(w, http.StatusOK, items)
}

func (a *api) internalError(w http.ResponseWriter, r *http.Request, err error) {
	a.logger.ErrorContext(r.Context(), "request failed", "error", err)
	writeError(w, http.StatusInternalServerError, "internal error")
}
func writeError(w http.ResponseWriter, status int, message string) {
	writeJSON(w, status, map[string]string{"error": message})
}
func writeJSON(w http.ResponseWriter, status int, value any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(value)
}

func decodeJSON(w http.ResponseWriter, r *http.Request, dst any) bool {
	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(dst); err != nil {
		writeError(w, http.StatusBadRequest, "invalid JSON")
		return false
	}
	if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
		writeError(w, http.StatusBadRequest, "request body must contain one JSON object")
		return false
	}
	return true
}

func optionalTime(w http.ResponseWriter, value string) (*time.Time, bool) {
	if value == "" {
		return nil, true
	}
	parsed, err := time.Parse(time.RFC3339, value)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid timestamp")
		return nil, false
	}
	return &parsed, true
}
func boundedInt(value string, fallback, min, max int) (int, bool) {
	if value == "" {
		return fallback, true
	}
	parsed, err := strconv.Atoi(value)
	return parsed, err == nil && parsed >= min && parsed <= max
}

func validUUID(value string) bool {
	if len(value) != 36 || value[8] != '-' || value[13] != '-' || value[18] != '-' || value[23] != '-' {
		return false
	}
	compact := strings.ReplaceAll(value, "-", "")
	_, err := hex.DecodeString(compact)
	return err == nil
}

func newUUID() (string, error) {
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		return "", fmt.Errorf("generate UUID: %w", err)
	}
	b[6] = (b[6] & 0x0f) | 0x40
	b[8] = (b[8] & 0x3f) | 0x80
	encoded := make([]byte, 36)
	hex.Encode(encoded[0:8], b[0:4])
	encoded[8] = '-'
	hex.Encode(encoded[9:13], b[4:6])
	encoded[13] = '-'
	hex.Encode(encoded[14:18], b[6:8])
	encoded[18] = '-'
	hex.Encode(encoded[19:23], b[8:10])
	encoded[23] = '-'
	hex.Encode(encoded[24:36], b[10:16])
	return string(encoded), nil
}