orca/internal/daemon/server.go

// Package daemon implements the orca HTTP daemon.
//
// The daemon exposes health endpoints (/healthz, /readyz), a status endpoint
// (/v1/status), and a v1 resource API for jobs, nodes, and tasks. All handlers
// follow the project conventions:
//
//   - context.Context propagated to all I/O
//   - errors wrapped with %w
//   - structured JSON via writeJSON
//   - no secrets in logs
//   - input validation on path/query/body
package daemon

import (
	"context"
	"database/sql"
	"errors"
	"log/slog"
	"net/http"
	"sync/atomic"
	"time"
)

// Server is the orca HTTP daemon. It holds shared dependencies and lifecycle
// state. Construct it with NewServer, then call Start/Shutdown.
type Server struct {
	db    *sql.DB
	log   *slog.Logger
	addr  string
	ready atomic.Bool

	httpServer  *http.Server
	pprofServer *http.Server

	// mtls is non-nil after StartMTLS has been called; nil otherwise.
	// Plaintext HTTP and mTLS are mutually exclusive — a Server is
	// either in plaintext mode (default, v0.1 compat) or mTLS mode
	// (v0.2 P01 forward).
	mtls *MTLSState

	// dispatch is the orca.v1.Dispatch service mounted on
	// /orca.v1.Dispatch/* (P02). Optional — nil if no Dispatcher
	// was registered. P02 wires this via RegisterDispatch.
	dispatch *DispatchHandlers
}

// Options configures a new Server.
type Options struct {
	DB    *sql.DB
	Log   *slog.Logger
	Addr  string
	Actor string // used for audit logging from API requests

	// PprofAddr enables the pprof endpoint on a separate listener
	// when non-empty (e.g. "127.0.0.1:6060"). Default "" disables it.
	// The pprof listener is unauthenticated and operator-only; never
	// expose it publicly (AD-024).
	PprofAddr string
}

// NewServer constructs a Server with the default mux and route table.
func NewServer(opts Options) *Server {
	if opts.Log == nil {
		opts.Log = slog.Default()
	}
	if opts.Addr == "" {
		opts.Addr = ":8080"
	}
	if opts.Actor == "" {
		opts.Actor = "api"
	}
	s := &Server{
		db:   opts.DB,
		log:  opts.Log,
		addr: opts.Addr,
	}
	s.httpServer = &http.Server{
		Addr:              opts.Addr,
		Handler:           s.mux(),
		ReadHeaderTimeout: 5 * time.Second,
		ReadTimeout:       15 * time.Second,
		WriteTimeout:      30 * time.Second,
		IdleTimeout:       60 * time.Second,
	}
	if opts.PprofAddr != "" {
		ps, perr := StartPprof(opts.PprofAddr, opts.Log)
		if perr != nil {
			s.log.Error("pprof start failed", slog.String("component", "daemon"), slog.Any("err", perr))
		} else {
			s.pprofServer = ps
		}
	}
	return s
}

// Addr returns the configured listen address.
func (s *Server) Addr() string { return s.addr }

// MarkReady flips the readiness flag to true. The /readyz endpoint returns
// 200 only when this flag is set AND the database is reachable.
func (s *Server) MarkReady() { s.ready.Store(true) }

// MarkNotReady flips the readiness flag to false. Called at shutdown start
// so load balancers stop routing traffic.
func (s *Server) MarkNotReady() { s.ready.Store(false) }

// Ready reports the current readiness flag.
func (s *Server) Ready() bool { return s.ready.Load() }

// mux builds the route table. Handlers are split across files:
//   - health.go          /healthz, /readyz, /v1/status
//   - jobs_handler.go    /v1/jobs/*
//   - nodes_handler.go   /v1/nodes/*
//   - tasks_handler.go   /v1/tasks/*
//   - dispatch_handler.go  /orca.v1.Dispatch/* (P02; mounted only if
//     RegisterDispatch was called)
func (s *Server) mux() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("/healthz", s.handleHealthz)
	mux.HandleFunc("/readyz", s.handleReadyz)
	mux.HandleFunc("/v1/status", s.handleStatus)
	mux.HandleFunc("/v1/jobs", s.handleJobsCollection)
	mux.HandleFunc("/v1/jobs/", s.handleJobsItem)
	mux.HandleFunc("/v1/nodes", s.handleNodesCollection)
	mux.HandleFunc("/v1/tasks", s.handleTasksCollection)
	if s.dispatch != nil {
		s.dispatch.Mount(mux)
	}
	return loggingMiddleware(s.log, mux)
}

// RegisterDispatch attaches the orca.v1.Dispatch service to the
// daemon. Call before Start(). The dispatch routes are mounted at
// /orca.v1.Dispatch/Submit and /orca.v1.Dispatch/Status.
func (s *Server) RegisterDispatch(h *DispatchHandlers) {
	if h == nil {
		return
	}
	s.dispatch = h
	s.log.Info("dispatch handlers registered",
		slog.String("component", "daemon"),
		slog.String("submit", "/orca.v1.Dispatch/Submit"),
		slog.String("status", "/orca.v1.Dispatch/Status"),
	)
}

// Start runs the HTTP server. Returns http.ErrServerClosed on clean shutdown.
func (s *Server) Start() error {
	s.log.Info("daemon starting",
		slog.String("addr", s.addr),
		slog.String("component", "daemon"))
	return s.httpServer.ListenAndServe()
}

// Shutdown gracefully stops the server, bounded by ctx. It also flips the
// readiness flag to false so /readyz returns 503 immediately.
func (s *Server) Shutdown(ctx context.Context) error {
	s.MarkNotReady()
	s.log.Info("daemon shutting down", slog.String("component", "daemon"))
	if s.pprofServer != nil {
		if perr := s.pprofServer.Shutdown(ctx); perr != nil {
			s.log.Error("pprof shutdown failed", slog.String("component", "daemon"), slog.Any("err", perr))
		}
	}
	return s.httpServer.Shutdown(ctx)
}

// IsShutdownErr reports whether err is the expected error from a stopped server.
func IsShutdownErr(err error) bool {
	return errors.Is(err, http.ErrServerClosed)
}