oikos/internal/httpapi/server.go

// Package httpapi implements the Oikos REST API. The contract is
// api/openapi.yaml (contract-first, ADR-0004); handlers implement the
// oapi-codegen strict-server interface in gen/. Errors map to RFC 9457
// problem+json via domain sentinels.
package httpapi

import (
	"context"
	"crypto/rsa"
	"crypto/subtle"
	"crypto/tls"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"log/slog"
	"math/big"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/dtoro/oikos/internal/config"
	"github.com/dtoro/oikos/internal/db"
	"github.com/dtoro/oikos/internal/httpapi/gen"
	mcphandler "github.com/dtoro/oikos/internal/mcp"
	"github.com/dtoro/oikos/internal/safego"
	"github.com/go-chi/chi/v5"
	"github.com/go-chi/chi/v5/middleware"
	"github.com/go-chi/cors"
	"github.com/golang-jwt/jwt/v5"
	"github.com/google/uuid"
)

// contextKey for storing actor identity in request context.
type contextKey string

const actorKey contextKey = "oikos_actor"

// actor holds the resolved identity of the API caller.
type actor struct {
	Type      string // "operator", "agent", "system"
	Label     string // human-readable label
	ID        string // OIDC sub or static token identifier
	TokenType string // "static" or "oidc"
}

// Server implements gen.StrictServerInterface over the DB layer.
type Server struct {
	pool           *db.Pool
	cfg            config.Config
	secretsManager secretsBackend
	sseBroker      *sseBroker
	sseSubs        map[*sseSubscriber]struct{}
	sseMu          sync.Mutex
}

// secretsBackend is a minimal interface for secrets operations used by the
// HTTP API (enrollment key storage, listing). Compatible with internal/secrets.
type secretsBackend interface {
	Set(ctx context.Context, key string, value string) error
	List(ctx context.Context) ([]string, error)
}

// NewHandler builds the full HTTP handler: /healthz (unauthenticated,
// SG18) + the OpenAPI surface under /api/v1 behind bearer auth.
//
// ctx governs the lifetime of the background SSE listener goroutine, which
// holds a dedicated pooled connection for LISTEN. Callers MUST cancel ctx
// before closing the pool — otherwise the held connection never releases
// and pool.Close() deadlocks.
func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config) http.Handler {
	s := &Server{
		pool:      pool,
		cfg:       cfg,
		sseBroker: newSSEBroker(10000),
		sseSubs:   make(map[*sseSubscriber]struct{}),
	}

	// Start background SSE listener, tied to ctx for clean shutdown.
	// handleNotification (called per-message inside sseListener's loop) has
	// its own recover for the common case; this outer one covers the
	// connection-setup/reconnect code around it.
	safego.Go("httpapi:sse-listener", func() { s.sseListener(ctx) })

	r := chi.NewRouter()
	r.Use(middleware.Recoverer)
	r.Use(middleware.RequestID)
	r.Use(requestLogger)
	r.Use(cors.Handler(cors.Options{
		AllowedOrigins: []string{cfg.CORSAllowedOrigin},
		AllowedMethods: []string{"GET", "POST", "PATCH", "DELETE", "OPTIONS"},
		AllowedHeaders: []string{"Authorization", "Content-Type", "If-Match"},
		MaxAge:         86400,
	}))

	// Liveness — no auth, no audit (plan SG18). Not exposed via Caddy.
	r.Get("/healthz", func(w http.ResponseWriter, req *http.Request) {
		ctx, cancel := context.WithTimeout(req.Context(), 2*time.Second)
		defer cancel()
		if err := pool.Ping(ctx); err != nil {
			writeProblem(w, req, http.StatusServiceUnavailable, "database unreachable", "")
			return
		}
		w.Header().Set("Content-Type", "application/json")
		w.Write([]byte(`{"status":"ok"}`))
	})

	// Client enrollment — unauthenticated (IP-gated in handler).
	// Registered on base router BEFORE HandlerWithOptions so it
	// bypasses the combinedAuth middleware applied to all /api/v1/*.
	r.Post("/api/v1/clients/enroll", func(w http.ResponseWriter, req *http.Request) {
		var body gen.EnrollRequest
		if err := json.NewDecoder(req.Body).Decode(&body); err != nil {
			writeProblem(w, req, http.StatusBadRequest, "invalid request body", err.Error())
			return
		}
		resp, err := s.EnrollClient(req.Context(), gen.EnrollClientRequestObject{Body: &body})
		if err != nil {
			writeProblemFromErr(w, req, err)
			return
		}
		if err := resp.VisitEnrollClientResponse(w); err != nil {
			writeProblem(w, req, http.StatusInternalServerError, "response encoding failed", err.Error())
		}
	})

	strict := gen.NewStrictHandlerWithOptions(s, nil, gen.StrictHTTPServerOptions{
		RequestErrorHandlerFunc: func(w http.ResponseWriter, req *http.Request, err error) {
			writeProblem(w, req, http.StatusBadRequest, "bad request", err.Error())
		},
		ResponseErrorHandlerFunc: writeProblemFromErr,
	})

	gen.HandlerWithOptions(strict, gen.ChiServerOptions{
		BaseURL:     "/api/v1",
		BaseRouter:  r,
		Middlewares: []gen.MiddlewareFunc{combinedAuth(cfg, false)},
		ErrorHandlerFunc: func(w http.ResponseWriter, req *http.Request, err error) {
			writeProblem(w, req, http.StatusBadRequest, "bad request", err.Error())
		},
	})

	// SSE stream: override the generated /events/stream route with a raw
	// flushing handler (registered AFTER HandlerWithOptions so chi's last
	// registration wins). The strict-server path can't Flush() per event;
	// this one uses the real ResponseWriter for real-time delivery. It
	// inherits the router's base middleware and applies auth via With().
	// allowQueryToken=true: EventSource can't set custom headers, so the
	// SPA passes the token as ?token=... instead of Authorization.
	r.With(combinedAuth(cfg, true)).Get("/api/v1/events/stream", s.serveSSE)

	// Custom (non-OpenAPI) route: recency-ordered knowledge + stats for the
	// Knowledge page's "what the system has learned" view. Registered after
	// HandlerWithOptions so it wins over any generated catch-all.
	r.With(combinedAuth(cfg, false)).Get("/api/v1/knowledge/recent", s.serveRecentKnowledge)

	// Custom (non-OpenAPI) route: full markdown content for a knowledge
	// entity (document/investigation/runbook) by its own id or slug — the
	// generated /api/v1/knowledge/{id} route (GetEntityKnowledge) answers a
	// different question (knowledge referencing this entity), not this one.
	r.With(combinedAuth(cfg, false)).Get("/api/v1/knowledge/content/{id}", s.serveKnowledgeContent)

	// Custom (non-OpenAPI) routes: the global activity feed (recency-ordered,
	// unlike ListExecutions which sorts by target for pagination) and the
	// per-session "what did this session do" digest.
	r.With(combinedAuth(cfg, false)).Get("/api/v1/activity/recent", s.serveRecentActivity)
	r.With(combinedAuth(cfg, false)).Get("/api/v1/activity/session/{id}", s.serveSessionDigest)

	// Learning view: capability timeline + success trend, both derived from
	// executions (real, growing data) rather than the patterns/skills tables,
	// which are correctly modeled but have no writers anywhere yet.
	r.With(combinedAuth(cfg, false)).Get("/api/v1/learning/timeline", s.serveLearningTimeline)
	r.With(combinedAuth(cfg, false)).Get("/api/v1/learning/trend", s.serveLearningTrend)

	// Mount MCP at /mcp (plan R3-10)
	nomosAgentID := uuid.Nil
	if cfg.NomosAgentID != "" {
		if id, err := uuid.Parse(cfg.NomosAgentID); err == nil {
			nomosAgentID = id
		}
	}
	if nomosAgentID == uuid.Nil && cfg.NomosAgentSlug != "" {
		_ = pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", cfg.NomosAgentSlug).Scan(&nomosAgentID)
	}
	r.With(combinedAuth(cfg, false)).Handle("/mcp", mcphandler.NewHandler(pool, cfg.MCPBearerToken, nomosAgentID))

	if nomosURL := os.Getenv("NOMOS_PROXY_URL"); nomosURL != "" {
		target, _ := url.Parse(nomosURL)
		proxy := httputil.NewSingleHostReverseProxy(target)
		// Was unauthenticated (pre-existing gap, predates the client/server
		// split — this mount was never wrapped in combinedAuth, unlike every
		// other custom route below). Harmless while dev-open was in effect;
		// a real hole now that every route needs a real credential.
		r.Mount("/agent", combinedAuth(cfg, false)(http.StripPrefix("/agent", proxy)))
	}

	return r
}

// combinedAuth tries OIDC JWT validation first (if configured), then falls
// back to static bearer token validation. Every request needs a valid
// credential — there is no dev-open bypass (closed as part of the
// client/server split, plans/2026-07-12-wails-desktop-app.md 0.4: once the
// SPA is a separate client, a dev-open API is reachable from any origin).
// When allowQueryToken is set, a missing Authorization header falls back to
// a `?token=` query param — only used for the SSE route, since EventSource
// can't set custom headers.
func combinedAuth(cfg config.Config, allowQueryToken bool) func(http.Handler) http.Handler {
	hasOIDC := cfg.OIDCIssuer != "" && cfg.OIDCClientID != ""
	hasStatic := cfg.APIToken != "" || cfg.MCPBearerToken != ""

	// Cache JWKS for OIDC
	var jwksURL string
	var jwksCache []jwtVerificationKey
	var jwksMu sync.RWMutex
	if hasOIDC {
		// Fetch JWKS URI from OIDC discovery
		jwksURL = discoverJWKSURI(cfg.OIDCIssuer)
		if jwksURL != "" {
			keys, err := fetchJWKS(jwksURL)
			if err != nil {
				slog.Warn("oidc initial jwks fetch failed, will retry on demand", "error", err)
			} else {
				jwksCache = keys
			}
		}
	}

	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			auth := r.Header.Get("Authorization")
			raw, ok := strings.CutPrefix(auth, "Bearer ")
			if (!ok || raw == "") && allowQueryToken {
				raw = r.URL.Query().Get("token")
				ok = raw != ""
			}
			if !ok || raw == "" {
				writeProblem(w, r, http.StatusUnauthorized, "unauthorized",
					"missing bearer token")
				return
			}

			// Try OIDC first if configured
			if hasOIDC {
				jwksMu.RLock()
				keys := jwksCache
				jwksMu.RUnlock()

				// If cache is empty, try to refresh
				if len(keys) == 0 && jwksURL != "" {
					if freshKeys, err := fetchJWKS(jwksURL); err == nil {
						jwksMu.Lock()
						jwksCache = freshKeys
						keys = freshKeys
						jwksMu.Unlock()
					}
				}

				if act, err := validateOIDCToken(raw, cfg, keys); err == nil {
					ctx := context.WithValue(r.Context(), actorKey, act)
					next.ServeHTTP(w, r.WithContext(ctx))
					return
				} else {
					slog.Debug("oidc validation failed", "error", err)
				}
			}

			// Fall back to static tokens
			if hasStatic {
				if act, ok := staticTokenActor(cfg, raw); ok {
					ctx := context.WithValue(r.Context(), actorKey, act)
					next.ServeHTTP(w, r.WithContext(ctx))
					return
				}
			}

			writeProblem(w, r, http.StatusUnauthorized, "unauthorized",
				"invalid or expired bearer token")
		})
	}
}

// staticTokenActor validates raw against the configured static bearer
// tokens (API token, MCP token) in constant time and returns the resolved
// actor. Shared between combinedAuth's header-based check and serveSSE's
// query-param check (EventSource can't set custom headers, so the SSE
// stream takes the token as ?token=...).
func staticTokenActor(cfg config.Config, raw string) (actor, bool) {
	if raw == "" {
		return actor{}, false
	}
	idPrefix := raw
	if len(idPrefix) > 8 {
		idPrefix = idPrefix[:8]
	}
	if cfg.MCPBearerToken != "" && subtle.ConstantTimeCompare([]byte(raw), []byte(cfg.MCPBearerToken)) == 1 {
		return actor{Type: "agent", Label: "agent:mcp", ID: idPrefix + "...", TokenType: "static"}, true
	}
	if cfg.APIToken != "" && subtle.ConstantTimeCompare([]byte(raw), []byte(cfg.APIToken)) == 1 {
		return actor{Type: "operator", Label: "operator:api", ID: idPrefix + "...", TokenType: "static"}, true
	}
	return actor{}, false
}

// jwtVerificationKey holds a parsed RSA public key or HMAC secret for JWT
// verification, identified by its key ID (kid).
type jwtVerificationKey struct {
	Kid     string
	Alg     string
	Key     any // *rsa.PublicKey or []byte for HMAC
	IsHMAC  bool
}

// discoverJWKSURI fetches the OIDC discovery document and extracts the
// jwks_uri field.
func discoverJWKSURI(issuerURL string) string {
	discURL := strings.TrimRight(issuerURL, "/") + "/.well-known/openid-configuration"
	client := &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{
		TLSClientConfig: &tls.Config{InsecureSkipVerify: false},
	}}
	resp, err := client.Get(discURL)
	if err != nil {
		slog.Warn("oidc discovery failed", "url", discURL, "error", err)
		return ""
	}
	defer resp.Body.Close()

	var disc struct {
		JWKSURI string `json:"jwks_uri"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&disc); err != nil {
		slog.Warn("oidc discovery decode failed", "error", err)
		return ""
	}
	if disc.JWKSURI == "" {
		slog.Warn("oidc discovery missing jwks_uri")
		return ""
	}
	return disc.JWKSURI
}

// fetchJWKS retrieves the JWK set from a URL and returns the parsed keys.
func fetchJWKS(jwksURL string) ([]jwtVerificationKey, error) {
	client := &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{
		TLSClientConfig: &tls.Config{InsecureSkipVerify: false},
	}}
	resp, err := client.Get(jwksURL)
	if err != nil {
		return nil, fmt.Errorf("fetch jwks: %w", err)
	}
	defer resp.Body.Close()

	var jwks struct {
		Keys []json.RawMessage `json:"keys"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&jwks); err != nil {
		return nil, fmt.Errorf("decode jwks: %w", err)
	}

	var keys []jwtVerificationKey
	for _, raw := range jwks.Keys {
		var header struct {
			Kty string `json:"kty"`
			Kid string `json:"kid,omitempty"`
			Alg string `json:"alg,omitempty"`
			Use string `json:"use,omitempty"`
		}
		if err := json.Unmarshal(raw, &header); err != nil {
			continue
		}
		// Skip keys not intended for signature verification
		if header.Use != "" && header.Use != "sig" {
			continue
		}

		var key jwtVerificationKey
		key.Kid = header.Kid
		key.Alg = header.Alg

		switch header.Kty {
		case "RSA":
			var rsaKey struct {
				N string `json:"n"`
				E string `json:"e"`
			}
			if err := json.Unmarshal(raw, &rsaKey); err != nil {
				continue
			}
			pubKey, err := parseRSAPublicKey(rsaKey.N, rsaKey.E)
			if err != nil {
				slog.Debug("oidc parse rsa key failed", "kid", header.Kid, "error", err)
				continue
			}
			key.Key = pubKey
		case "oct":
			// HMAC keys not expected for OIDC but handle gracefully
			key.IsHMAC = true
		default:
			continue
		}

		if key.Key != nil || key.IsHMAC {
			keys = append(keys, key)
		}
	}

	if len(keys) == 0 {
		return nil, fmt.Errorf("no usable keys in jwks")
	}
	return keys, nil
}

// parseRSAPublicKey decodes a base64url-encoded RSA modulus and exponent into
// an *rsa.PublicKey.
func parseRSAPublicKey(nB64, eB64 string) (any, error) {
	// Decode base64url modulus
	nBytes, err := base64.RawURLEncoding.DecodeString(nB64)
	if err != nil {
		return nil, fmt.Errorf("decode modulus: %w", err)
	}
	// Decode base64url exponent
	eBytes, err := base64.RawURLEncoding.DecodeString(eB64)
	if err != nil {
		return nil, fmt.Errorf("decode exponent: %w", err)
	}

	// Build RSA public key
	n := new(big.Int).SetBytes(nBytes)
	e := 0
	for _, b := range eBytes {
		e = (e << 8) | int(b)
	}
	return &rsa.PublicKey{N: n, E: e}, nil
}

// validateOIDCToken parses and validates a JWT Bearer token against the OIDC
// configuration. Returns the resolved actor on success.
func validateOIDCToken(rawToken string, cfg config.Config, keys []jwtVerificationKey) (actor, error) {
	keyFunc := func(token *jwt.Token) (any, error) {
		kid, ok := token.Header["kid"].(string)
		if !ok {
			return nil, fmt.Errorf("no kid in token header")
		}

		// Find matching key
		for _, k := range keys {
			// If kid is empty in JWK, try algorithm match
			if k.Kid == kid || (k.Kid == "" && k.Alg == token.Header["alg"]) {
				return k.Key, nil
			}
		}
		// Fall back to any RSA key if no kid match (some providers don't set kid)
		if !ok {
			for _, k := range keys {
				if k.Key != nil {
					return k.Key, nil
				}
			}
		}
		return nil, fmt.Errorf("no matching key for kid: %s", kid)
	}

	token, err := jwt.Parse(rawToken, keyFunc,
		jwt.WithIssuer(cfg.OIDCIssuer),
		jwt.WithAudience(cfg.OIDCClientID),
		jwt.WithValidMethods([]string{"RS256", "RS384", "RS512", "ES256", "ES384", "ES512"}),
	)
	if err != nil {
		return actor{}, fmt.Errorf("jwt validation: %w", err)
	}

	claims, ok := token.Claims.(jwt.MapClaims)
	if !ok {
		return actor{}, fmt.Errorf("invalid claims")
	}

	sub, _ := claims.GetSubject()
	if sub == "" {
		// Try the Azure/Entra ID oid claim fallback
		if oid, ok := claims["oid"].(string); ok {
			sub = oid
		}
	}

	preferredUsername, _ := claims["preferred_username"].(string)
	email, _ := claims["email"].(string)

	label := sub
	if preferredUsername != "" {
		label = preferredUsername
	} else if email != "" {
		label = email
	}

	return actor{
		Type:      "operator",
		Label:     label,
		ID:        sub,
		TokenType: "oidc",
	}, nil
}

// GetActor extracts the actor identity from the context. Returns nil if not
// set (should not happen for authenticated routes).
func GetActor(ctx context.Context) *actor {
	a, ok := ctx.Value(actorKey).(actor)
	if !ok {
		return nil
	}
	return &a
}

// requestLogger logs one line per request with method, path, status,
// duration, and the chi request id.
func requestLogger(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		start := time.Now()
		ww := middleware.NewWrapResponseWriter(w, r.ProtoMajor)
		next.ServeHTTP(ww, r)
		slog.Info("http",
			"method", r.Method,
			"path", r.URL.Path,
			"status", ww.Status(),
			"duration_ms", time.Since(start).Milliseconds(),
			"request_id", middleware.GetReqID(r.Context()),
		)
	})
}

// ListenAndServe runs the API server with graceful shutdown on ctx cancel
// (SG4): stop accepting, drain in-flight for up to 30s, then exit.
func ListenAndServe(ctx context.Context, pool *db.Pool, cfg config.Config) error {
	srv := &http.Server{
		Addr:              cfg.APIListen,
		Handler:           NewHandler(ctx, pool, cfg),
		ReadHeaderTimeout: 10 * time.Second,
	}

	errCh := make(chan error, 1)
	go func() {
		// Recovers a panic in ListenAndServe (stdlib, so extremely unlikely,
		// but an unrecovered panic here would crash the whole process rather
		// than surfacing as a normal startup error) and reports it through
		// errCh instead — the select below would otherwise just hang waiting
		// for a value that never arrives.
		defer func() {
			if r := recover(); r != nil {
				errCh <- fmt.Errorf("panic in ListenAndServe: %v", r)
			}
		}()
		slog.Info("api listening", "addr", cfg.APIListen)
		errCh <- srv.ListenAndServe()
	}()

	select {
	case err := <-errCh:
		return err
	case <-ctx.Done():
		slog.Info("api shutting down")
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		return srv.Shutdown(shutdownCtx)
	}
}