krow_backend/go-api/internal/httpserver/server.go

// Package httpserver holds the HTTP surface.
//
// It serves /health, the sign-in endpoints, the entity endpoints described in
// docs/api-contract.md, the current-user endpoints, the agent and skill
// definition endpoints, and the Owliver panel's suggestion endpoint.
//
// Authentication replaced the development identity: every request outside the
// small public allowlist in auth.go must carry a session cookie; the middleware
// resolves it to a user row and puts that user, and their organization, on the
// request context. Nothing downstream changed — every service and repository
// already took the organization as a parameter, which is what devOrgMiddleware
// existed to make true.
//
// Authorization is here, in Server.authorize: it reads the role off the
// authenticated identity, consults the deny-by-default policy table in
// internal/domain/policy.go, and answers 403 before any query runs. Row
// visibility — organization scope, and ownership for talent callers — is a SQL
// predicate in internal/repo instead, so an invisible row answers 404 rather
// than 403. The definition endpoints are the exception: they are not
// domain.Resource values, so their role checks are written inline in
// internal/service/definitions.go rather than in the policy table.
package httpserver

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"strconv"
	"time"

	"github.com/krow/krow-backend/go-api/internal/auth"
	"github.com/krow/krow-backend/go-api/internal/config"
	"github.com/krow/krow-backend/go-api/internal/db"
	"github.com/krow/krow-backend/go-api/internal/definition"
	"github.com/krow/krow-backend/go-api/internal/knowledge"
	"github.com/krow/krow-backend/go-api/internal/memory"
	"github.com/krow/krow-backend/go-api/internal/ratelimit"
	"github.com/krow/krow-backend/go-api/internal/runtime"
	"github.com/krow/krow-backend/go-api/internal/service"
	"github.com/krow/krow-backend/go-api/internal/tools"
)

// Server binds the router, the pool, authentication and the lifecycle together.
type Server struct {
	cfg         *config.Config
	db          *db.DB
	api         *service.Registry
	definitions *service.DefinitionsService
	workflows   *service.WorkflowService
	suggestions *service.SuggestionsService

	// The agent runtime. Nil when no model credential is configured — the run
	// routes are then not registered at all, so the deployment answers 404
	// ("this deployment does not serve agents") rather than 500 ("this
	// deployment is broken"). Only one of those is true.
	agents  *runtime.Engine
	runs    *runtime.RunReader
	version string

	// memories is the long-term memory store, for the scheduled sweep that
	// enforces its retention promise. The runtime builds its own; this one is
	// here so maintenance can prune without reaching through the engine.
	memories *memory.Store

	// toolCatalogue is the tool set an agent author may choose from.
	//
	// Built whether or not a model credential exists: the catalogue describes
	// what the tools ARE, and a deployment that cannot currently run agents can
	// still be one where somebody is authoring them.
	toolCatalogue []tools.ToolInfo

	log       *slog.Logger
	http      *http.Server
	started   time.Time
	endpoints int

	// The authentication surface. sessions owns the lifecycle, users is the
	// read side of the users table, credentials verifies a password against it,
	// and the two limiters bound how often that may be attempted.
	//
	// Two limiters, not one, because the budgets are different sizes on
	// purpose: an email is one account and gets a tight budget, while an
	// address may be a whole office behind NAT and gets a loose one. Sharing a
	// limiter would force the office to live within one person's budget.
	sessions     *auth.Manager
	users        auth.UserStore
	credentials  *auth.Credentials
	loginByEmail *attemptLimiter

	// limiter bounds the OAuth and MCP routes, shared across instances via
	// Postgres. Nil when those routes are not registered — see mcplimit.go,
	// where a nil limiter means the middleware is not installed at all rather
	// than installed and permissive.
	limiter     *ratelimit.Limiter
	loginByAddr *attemptLimiter

	// trust resolves a request to the address its per-address limits are keyed
	// by, reading a forwarded address only from a configured proxy. Wired once
	// here so no handler can be given a different notion of who called.
	trust proxyTrust

	// now is injectable so tests can drive expiry without sleeping.
	now func() time.Time
}

// Option adjusts the server before it is wired. Production passes none.
type Option func(*serverOptions)

type serverOptions struct {
	policy      auth.Policy
	now         func() time.Time
	perEmail    int
	perAddress  int
	loginWindow time.Duration

	// agents replaces the engine New would otherwise build from configuration.
	//
	// For tests, and only for tests: production wires a real gateway from a
	// real key, and an option that let a deployment substitute the runtime
	// would be a way to run agents against something nobody configured.
	agents *runtime.Engine

	// version is the build identifier, stamped into the binary at link time.
	// Not configuration: it describes the artefact, not the deployment, and an
	// environment variable could disagree with the code it claims to describe.
	version string

	// curatedAgents replaces the set New would otherwise read from disk.
	// nil means "read the configured directory"; an empty non-nil set means
	// "protect nothing", which is a thing a test needs to be able to say.
	curatedAgents map[string]bool
}

// WithBuildVersion records which build this is.
//
// Unlike the options above this one is for production. Without it there is no
// way to answer "did my deploy land?" — the symptom is pushing an image,
// redeploying, and having nobody, including the operator, able to tell whether
// the running process is the new one.
func WithBuildVersion(v string) Option {
	return func(o *serverOptions) {
		if v != "" {
			o.version = v
		}
	}
}

// WithSessionPolicy overrides the session lifetimes. For tests that need to
// reach an expiry without waiting twelve hours for it.
func WithSessionPolicy(p auth.Policy) Option {
	return func(o *serverOptions) { o.policy = p }
}

// WithClock replaces the clock used for session expiry and last_login_at.
func WithClock(now func() time.Time) Option {
	return func(o *serverOptions) {
		if now != nil {
			o.now = now
		}
	}
}

// WithLoginRateLimit overrides the failed-attempt budgets and their window.
//
// perEmail bounds attempts against one account; perAddress bounds attempts from
// one client address across all accounts. Both are consulted on every attempt.
// WithAgentEngine substitutes the agent runtime.
//
// The seam that lets the HTTP layer be tested without a model credential —
// which matters more than it sounds, because the alternative is that the run
// endpoint is the one part of this service no test can reach until somebody
// pays for a key.
//
// It does not weaken anything: the engine still loads agents through the same
// loader, still runs them under the same budgets, and still authorizes through
// the same principal. Only the model behind it changes.
// WithCuratedAgents names the delete-protected agent ids directly.
//
// Production loads these from disk; this exists so a test can state its own
// protected set without a directory, exactly as WithAgentEngine lets one
// supply an engine without a model credential.
func WithCuratedAgents(ids ...string) Option {
	return func(o *serverOptions) {
		set := make(map[string]bool, len(ids))
		for _, id := range ids {
			set[id] = true
		}
		o.curatedAgents = set
	}
}

func WithAgentEngine(e *runtime.Engine) Option {
	return func(o *serverOptions) { o.agents = e }
}

func WithLoginRateLimit(perEmail, perAddress int, window time.Duration) Option {
	return func(o *serverOptions) {
		o.perEmail, o.perAddress, o.loginWindow = perEmail, perAddress, window
	}
}

// New wires the routes and returns a server that has not yet been started.
//
// Authentication is built here rather than passed in, so there is exactly one
// construction of the session manager and no way to start a server with the
// middleware wired to a different store than the login handler.
func New(cfg *config.Config, database *db.DB, log *slog.Logger, opts ...Option) (*Server, error) {
	o := serverOptions{
		policy:      auth.DefaultPolicy,
		now:         time.Now,
		perEmail:    loginAttemptLimit,
		perAddress:  loginAddressLimit,
		loginWindow: loginAttemptWindow,
		version:     "unknown",
	}
	for _, opt := range opts {
		opt(&o)
	}

	sessions, err := auth.NewManager(auth.NewPGStore(database.Pool), o.policy)
	if err != nil {
		return nil, fmt.Errorf("build session manager: %w", err)
	}
	sessions.WithClock(o.now)

	users := auth.NewPGUserStore(database.Pool)
	s := &Server{
		cfg: cfg, db: database, log: log,
		version:      o.version,
		api:          service.NewRegistry(database.Pool),
		definitions:  service.NewDefinitions(database.Pool),
		workflows:    service.NewWorkflows(database.Pool).WithClock(o.now),
		suggestions:  service.NewSuggestions(database.Pool),
		started:      o.now(),
		sessions:     sessions,
		users:        users,
		credentials:  auth.NewCredentials(users),
		loginByEmail: newAttemptLimiter(o.perEmail, o.loginWindow, o.now),
		loginByAddr:  newAttemptLimiter(o.perAddress, o.loginWindow, o.now),
		trust:        newProxyTrust(cfg.HTTP.TrustedProxies),
		now:          o.now,
	}

	// The agent runtime, wired only when there is a model to reach.
	//
	// Registering the routes without a credential would accept runs and fail
	// every one of them at the gateway — an outage shaped like a feature. A
	// deployment without a key is a deployment that does not serve agents, and
	// saying so at boot is kinder than saying it once per request.
	switch {
	case o.agents != nil:
		s.agents = o.agents
		s.runs = runtime.NewRunReader(database.Pool)
	case cfg.Model.APIKey != "":
		s.agents = runtime.NewModelEngine(database.Pool, *cfg)
		s.runs = runtime.NewRunReader(database.Pool)
	}

	// Memories are swept wherever the database is, agents or not: a deployment
	// that stops serving agents still holds what earlier ones remembered, and
	// retention is a promise about the table rather than about the feature.
	s.memories = memory.New(database.Pool, runtime.NewEmbedder(*cfg))

	// Built the same way the runtime builds its own, so the list an author is
	// offered is the list their agent will actually have.
	toolRegistry := runtime.DefaultTools(
		database.Pool,
		knowledge.NewRetriever(database.Pool, runtime.NewEmbedder(*cfg)),
	)
	s.toolCatalogue = toolRegistry.Catalogue()

	// So a definition naming a tool that does not exist is refused at publish
	// rather than becoming an agent that silently cannot do what it claims.
	s.definitions = s.definitions.WithToolCheck(toolRegistry.Known)

	// The agents this deployment ships specs for, so DELETE refuses them at the
	// endpoint rather than only in the list that renders the button.
	//
	// Read from the directory `importagents` publishes from, so the protected
	// set is the published set by construction. A deployment without that
	// directory protects nothing and says so here, once, at boot: silence would
	// leave an operator believing in a guard that is not running.
	curated := o.curatedAgents
	if curated == nil {
		loaded, err := definition.CuratedIDs(cfg.Agents.CuratedPath)
		if err != nil {
			return nil, fmt.Errorf("load curated agents: %w", err)
		}
		curated = loaded
	}
	if len(curated) == 0 {
		log.Warn("no curated agent specs found; built-in agents are not delete-protected",
			"path", cfg.Agents.CuratedPath)
	} else {
		log.Info("curated agents are delete-protected",
			"count", len(curated), "ids", definition.SortedIDs(curated))
	}
	s.definitions = s.definitions.WithCuratedAgents(curated)

	// The shared limiter, built only when the routes that use it exist. The
	// existing in-process login limiter is untouched: it guards a different
	// thing (failed password attempts) with a different model (count failures,
	// reset on success), and replacing it is not this change's business.
	if cfg.OAuth.Enabled() {
		s.limiter = ratelimit.New(database.Pool)
	}

	mux := http.NewServeMux()
	mux.HandleFunc("GET /health", s.handleHealth)
	s.endpoints = s.routeAuth(mux) + s.routeResources(mux) + s.routeMe(mux) +
		s.routeDefinitions(mux) + s.routeWorkflows(mux) + s.routeOwliver(mux) +
		s.routeRuns(mux) + s.routeVersion(mux) + s.routeTools(mux) +
		// The MCP surface and the OAuth server behind it. Both return 0 and
		// register nothing when OAUTH_ISSUER and MCP_RESOURCE are unset, which
		// is every deployment that has not asked for them.
		s.routeOAuth(mux) + s.routeMCP(mux)

	handler := jsonErrors(mux)
	// Authentication sits where devOrgMiddleware used to, so every route below
	// it — including the mux's own 404 — is behind the allowlist.
	handler = s.authenticate(handler)
	handler = recoverer(log)(handler)
	// CORS sits outside the recoverer so a preflight is answered without
	// touching the router, and inside the logger so refused origins are still
	// visible in the log. With no allowlist configured it is not installed at
	// all, which is the same-origin default.
	if len(cfg.HTTP.CORSOrigins) > 0 {
		handler = cors(cfg.HTTP.CORSOrigins)(handler)
	}
	handler = requestLogger(log)(handler)

	s.http = &http.Server{
		Addr:         net.JoinHostPort(cfg.HTTP.Host, strconv.Itoa(cfg.HTTP.Port)),
		Handler:      handler,
		ReadTimeout:  cfg.HTTP.ReadTimeout,
		WriteTimeout: cfg.HTTP.WriteTimeout,
		IdleTimeout:  cfg.HTTP.IdleTimeout,
	}
	return s, nil
}

// Sessions exposes the session manager, so the process can sweep expired rows
// and tests can drive the clock.
func (s *Server) Sessions() *auth.Manager { return s.sessions }

// Handler exposes the routed handler so tests can drive it without a listener.
func (s *Server) Handler() http.Handler { return s.http.Handler }

// Endpoints is how many routes were registered.
func (s *Server) Endpoints() int { return s.endpoints }

// Addr is the address the server listens on.
func (s *Server) Addr() string { return s.http.Addr }

// Start blocks until the server stops accepting connections.
func (s *Server) Start() error {
	err := s.http.ListenAndServe()
	if errors.Is(err, http.ErrServerClosed) {
		return nil
	}
	return err
}

// Shutdown drains in-flight requests, then gives up after the configured grace.
func (s *Server) Shutdown(ctx context.Context) error {
	ctx, cancel := context.WithTimeout(ctx, s.cfg.HTTP.ShutdownTimeout)
	defer cancel()
	return s.http.Shutdown(ctx)
}

// healthResponse is the entire public /health body: one field, deliberately.
//
// /health is unauthenticated and reachable by anyone who can reach the port,
// so it is treated as a public document rather than as an operator's console.
// Everything an unauthenticated caller legitimately needs is the answer to
// "should traffic be sent here", and that fits in a status string plus the
// HTTP status code.
//
// What used to be here and is now deliberately absent: the PostgreSQL version,
// the database name, the schema name, the applied migration version, the table
// count, the connection error text, the deployment environment and the process
// uptime. Individually each is small; together they are a free reconnaissance
// report — the server version to look up known CVEs against, the migration
// version to date the deployment, the table count and error text to infer
// shape and topology. None of it is diagnostic to anyone who could not already
// read it from the database directly.
//
// The check itself is unchanged. db.Check still runs on every request and
// still decides the answer; its full detail now goes to the server log, where
// the operator is, instead of into the response, where the internet is. See
// logHealth.
type healthResponse struct {
	Status string `json:"status"`
}

// routeTools lists the tools an agent author may choose from.
//
// The frontend's agent editor had no tools field at all, so an authored agent
// carried none and could talk without being able to look anything up. Serving
// the catalogue rather than hard-coding it in the UI keeps one list: a tool
// added or renamed here cannot leave a stale copy behind in a form.
func (s *Server) routeTools(mux *http.ServeMux) int {
	mux.HandleFunc("GET /api/v1/tools", func(w http.ResponseWriter, r *http.Request) {
		writeJSON(w, http.StatusOK, envelope{Data: s.toolCatalogue})
	})
	return 1
}

// routeVersion exposes the build identifier to an authenticated caller.
//
// Under /api/v1 rather than on /health deliberately. /health is public, and it
// already withholds its detail from the internet for the reason given above; a
// build identifier is exactly the kind of thing that tells an unauthenticated
// reader which source to go and read. An operator has a session, so this is
// where an operator can reach it and a stranger cannot.
func (s *Server) routeVersion(mux *http.ServeMux) int {
	mux.HandleFunc("GET /api/v1/version", func(w http.ResponseWriter, r *http.Request) {
		writeJSON(w, http.StatusOK, envelope{Data: map[string]any{
			"version":   s.version,
			"env":       s.cfg.AppEnv,
			"endpoints": s.endpoints,
		}})
	})
	return 1
}

// handleHealth reports whether this instance should be sent traffic.
//
//	200 "ok"           serving normally
//	200 "degraded"     the process is healthy, the schema is not: unmigrated,
//	                   or a migration left the version dirty. Still 200,
//	                   because the fault is the database's and taking the
//	                   instance out of rotation would not fix it.
//	503 "unavailable"  the database is unreachable, so a load balancer can act
//	                   on the status code alone without parsing the body.
//
// The three status words are a coarse operational signal, not infrastructure
// detail: they say what a caller should do, and nothing about what is running.
func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
	defer cancel()

	health := s.db.Check(ctx)

	status, code := "ok", http.StatusOK
	switch {
	case !health.Reachable:
		status, code = "unavailable", http.StatusServiceUnavailable
	case health.MigrationDirty, !health.SchemaPresent:
		status = "degraded"
	}

	s.logHealth(status, health)

	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	w.Header().Set("Cache-Control", "no-store")
	w.WriteHeader(code)
	enc := json.NewEncoder(w)
	enc.SetIndent("", "  ")
	_ = enc.Encode(healthResponse{Status: status})
}

// logHealth writes the detail the response body used to carry.
//
// This is the "internally" half of the change: nothing was deleted from
// db.Check, and nothing it learns is thrown away — the audience moved from the
// response to the log, which is already authenticated by virtue of being on
// the host.
//
// A load balancer polls this endpoint every few seconds, so a healthy check
// logs at debug and a bad one at warn. Anything other than "ok" is worth
// seeing without turning debug on.
func (s *Server) logHealth(status string, h db.Health) {
	attrs := []any{
		"status", status,
		"env", s.cfg.AppEnv,
		"uptime_seconds", int64(time.Since(s.started).Seconds()),
		"reachable", h.Reachable,
		"schema", h.Schema,
		"schema_present", h.SchemaPresent,
		"table_count", h.TableCount,
		"migration_dirty", h.MigrationDirty,
		"latency_ms", h.LatencyMS,
	}
	if h.Database != "" {
		attrs = append(attrs, "database", h.Database, "postgres_version", h.Version)
	}
	if h.AppliedMigration != nil {
		attrs = append(attrs, "applied_migration", *h.AppliedMigration)
	}
	if h.Error != "" {
		attrs = append(attrs, "error", h.Error)
	}

	if status == "ok" {
		s.log.Debug("health", attrs...)
		return
	}
	s.log.Warn("health", attrs...)
}

type statusRecorder struct {
	http.ResponseWriter
	code int
}

func (r *statusRecorder) WriteHeader(code int) {
	r.code = code
	r.ResponseWriter.WriteHeader(code)
}

// Flush forwards to the writer underneath.
//
// A wrapper that embeds http.ResponseWriter inherits Write and WriteHeader and
// SILENTLY DROPS every optional interface the real writer implements — Flusher
// among them. Nothing errors: the handler simply asks "can this flush?", is
// told no, and takes whatever fallback it has.
//
// That is exactly how it presented. The SSE endpoint answered ordinary JSON,
// correctly and completely, with no error anywhere — because two middlewares
// deep the writer had stopped being a Flusher and the streaming path politely
// declined to stream.
func (r *statusRecorder) Flush() {
	if f, ok := r.ResponseWriter.(http.Flusher); ok {
		f.Flush()
	}
}

func requestLogger(log *slog.Logger) func(http.Handler) http.Handler {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			started := time.Now()
			rec := &statusRecorder{ResponseWriter: w, code: http.StatusOK}
			next.ServeHTTP(rec, r)
			log.Info("request",
				"method", r.Method, "path", r.URL.Path,
				"status", rec.code, "duration_ms", time.Since(started).Milliseconds())
		})
	}
}

// jsonErrors converts net/http's own plain-text 404 and 405 replies into the
// documented error envelope.
//
// ServeMux writes those itself, before any handler of ours runs, so a client
// that hit a wrong path or method would otherwise get "404 page not found" in
// text/plain while every other response is JSON. Only the mux's own replies are
// rewritten: anything that set a content type has already answered properly.
func jsonErrors(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		iw := &interceptor{ResponseWriter: w}
		next.ServeHTTP(iw, r)
		if !iw.rewritten || iw.wrote {
			return
		}
		errCode, message := "not_found", "resource not found"
		if iw.code == http.StatusMethodNotAllowed {
			errCode = "method_not_allowed"
			message = r.Method + " is not supported for this resource"
		}
		writeJSON(w, iw.code, errorEnvelope{Error: errorBody{
			Code: errCode, Message: message, Details: map[string]string{},
		}})
	})
}

// interceptor defers the mux's plain-text 404/405 body so it can be replaced.
type interceptor struct {
	http.ResponseWriter
	code      int
	rewritten bool // this is a mux-generated 404/405 we intend to replace
	wrote     bool // a body already went to the client
}

func (i *interceptor) WriteHeader(code int) {
	i.code = code
	if code == http.StatusNotFound || code == http.StatusMethodNotAllowed {
		if i.Header().Get("Content-Type") != "application/json; charset=utf-8" {
			i.rewritten = true
			return // hold the header back; jsonErrors writes its own
		}
	}
	i.ResponseWriter.WriteHeader(code)
}

func (i *interceptor) Write(b []byte) (int, error) {
	if i.rewritten {
		return len(b), nil // swallow the mux's plain-text body
	}
	i.wrote = true
	return i.ResponseWriter.Write(b)
}

// Flush forwards to the writer underneath. See statusRecorder.Flush.
func (i *interceptor) Flush() {
	if f, ok := i.ResponseWriter.(http.Flusher); ok {
		f.Flush()
	}
}

// recoverer turns a panic into a logged 500 rather than a dropped connection.
func recoverer(log *slog.Logger) func(http.Handler) http.Handler {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			defer func() {
				if v := recover(); v != nil {
					log.Error("panic", "value", v, "path", r.URL.Path)
					writeJSON(w, http.StatusInternalServerError, errorEnvelope{Error: errorBody{
						Code: "internal", Message: "internal error", Details: map[string]string{},
					}})
				}
			}()
			next.ServeHTTP(w, r)
		})
	}
}