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
//
// Deprecated: v0.9 re-architecture replaces this with SSH-push to bare
// servers (no orca binary on servers) per R-001. The orca daemon is
// repurposed to drain-and-stop in v0.10-P05 and scheduled for deletion
// in v0.10-P14. See .ciagent/PRD_v0.9.md R-001/R-006. The dual-write
// window (REQ-090/REQ-085) keeps this package compiling until v0.10-P14.
package daemon

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"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

	// acl is the access-control policy (P04, v0.13; C-44/C-45). When
	// nil, no ACL enforcement is applied (legacy/compat for tests
	// that construct a Server directly). Production wiring sets this
	// via NewServer (Options.ACLEnforce) so handlers can call
	// s.acl.Check before dispatching.
	acl *aclPolicy
}

// 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

	// ACLEnforce controls C-45 staged rollout. When false (the default
	// for the first run after P04 wiring), ACL denials are LOGGED but
	// NOT enforced — the request proceeds. When true, ACL denials
	// return 403. The operator switches to true after verifying the
	// bootstrap ACL grants the right identities.
	ACLEnforce bool
}

// maxBodyBytes is the limit for request bodies on JSON-decoding
// endpoints (REQ-124, F24). 1 MiB is generous for orca API calls.
const maxBodyBytes int64 = 1 << 20

// bodyLimitMiddleware wraps the handler with a MaxBytesReader so
// oversized request bodies are rejected before decoding (REQ-124).
func bodyLimitMiddleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes)
		next.ServeHTTP(w, r)
	})
}

// 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,
		acl:  NewACLPolicy(opts.ACLEnforce, opts.Log),
	}
	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() }

// ACL returns the daemon's ACL enforcement policy (P04). Returns nil
// if no policy is configured (legacy/compat). Handlers use this to
// call Check before dispatching; tests use it to assert enforcement
// mode.
func (s *Server) ACL() *aclPolicy { return s.acl }

// SetACLPolicy replaces the ACL policy. Used by tests to inject a
// policy without going through NewServer. Production code should use
// NewServer with Options.ACLEnforce.
func (s *Server) SetACLPolicy(p *aclPolicy) { s.acl = p }

// 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.mountWithACL(mux, s.acl)
	}
	return bodyLimitMiddleware(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.
// R-021 / REQ-123: the daemon MUST run in mTLS mode (no plaintext).
// If StartMTLS has not been called, Start refuses to run.
func (s *Server) Start() error {
	if s.mtls == nil {
		s.log.Error("daemon refuses to start in plaintext mode",
			slog.String("component", "daemon"),
			slog.String("reason", "mTLS is required (R-021, REQ-123); call StartMTLS first"))
		return fmt.Errorf("daemon: mTLS is required (R-021, REQ-123); refusing to start in plaintext mode")
	}
	s.log.Info("daemon starting (mTLS required)",
		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)
}