// Package agentsock is the agent's own listening socket — the ONE channel
// between a sidecar and the platform, and it points inward: children dial
// the agent, the agent never dials a child. That is what lets a sidecar
// bind nothing, and what lets the identical artifact run off-platform
// behind NAT with no protocol difference.
//
// Every exec child runs as the agent's user, so the socket's file mode is
// not an identity boundary. Identity is the per-wake token the activator
// mints and passes in the child's environment (CARLOS_SIDECAR_TOKEN),
// scoped to one host for one wake: a token for A cannot speak about B, and
// a token from a previous wake matches nothing.
//
// # What the socket trusts
//
// Nothing the caller sends, except as material for the one check. The
// Carlos-Sidecar-Host header names a host, but naming it proves nothing —
// only holding that host's live token does, and every effect is then applied
// to the host the TOKEN authorized. No body carries a host field, and none
// ever should: that is the confused-deputy shape this design exists to
// exclude. Bodies are read under a 4KiB cap, parsed strictly, and refused on
// anything unexpected; a refused call — for any reason, at any stage — has no
// effect at all.
//
// Refusals are deliberately uniform. Wrong token, absent token, unknown host,
// and a host whose worker is asleep all produce one identical 401, so the
// socket cannot be used to enumerate hosts or probe which ones are awake.
package agentsock

import (
	"context"
	"crypto/rand"
	"crypto/subtle"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os"
	"strings"
	"time"
)

// Workers is the activator's half of this conversation. It is declared here,
// not imported, so the socket stays testable and the direction stays plain:
// this package is told about workers, and never reaches for one.
//
// WorkerToken returns "" for a host that is unknown, asleep, or stopped. That
// is a REFUSAL, never "no token required" — see authorize.
type Workers interface {
	WorkerToken(host string) string
	NoteWorkerContact(host string)
	SetWorkerNextDue(host string, at int64) error
	SleepWorker(host string)
	NoteWorkerReady(host, version string)
}

const (
	// hostHeader names the host a caller claims to be. Untrusted on its own.
	hostHeader = "Carlos-Sidecar-Host"

	// maxBody bounds every body we read, following the next-due decode in
	// activator/backend_exec.go. These messages are two small fields; a
	// caller sending more is malformed, not generous.
	maxBody = 1 << 12

	// maxVersion bounds the build string a child reports. It is displayed to
	// members, so it is length- and character-checked before it is recorded.
	maxVersion = 200

	// socketMode keeps the socket off the world. It is defence in depth, not
	// the boundary: exec children share the agent's uid, so the mode cannot
	// tell one child from another. The token does that.
	socketMode = 0o660

	// Local calls, all of them. Nothing here should ever be slow, and a
	// child that stalls mid-request must not pin a connection.
	readHeaderTimeout = 2 * time.Second
	readTimeout       = 5 * time.Second
	writeTimeout      = 5 * time.Second
	idleTimeout       = 30 * time.Second
	shutdownGrace     = 3 * time.Second

	// dialProbe is how long we wait to learn whether an existing socket file
	// still has somebody listening on it.
	dialProbe = 250 * time.Millisecond
)

// Server serves the agent socket. Zero per-request state: one instance
// answers every child on the box concurrently.
type Server struct {
	w Workers
}

// New builds the socket server over w.
func New(w Workers) *Server { return &Server{w: w} }

// Handler is the routing table, exported so tests can exercise it without
// binding anything. Every verb is POST-only: the method pattern means a GET
// is turned away by the router, before authorize and before any effect.
func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()
	mux.Handle("POST /v1/ready", s.guard(s.ready))
	mux.Handle("POST /v1/next-due", s.guard(s.nextDue))
	mux.Handle("POST /v1/idle", s.guard(s.idle))
	mux.Handle("POST /v1/heartbeat", s.guard(s.heartbeat))
	return mux
}

// guard is the single gate. Nothing reaches a verb without passing it, and a
// verb never sees the request's claimed host — only the authorized one, so
// there is no way for a handler to act on a host it was not entitled to.
func (s *Server) guard(fn func(http.ResponseWriter, *http.Request, string)) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		host, ok := s.authorize(r)
		if !ok {
			refuse(w)
			return
		}
		fn(w, r, host)
	})
}

// authorize resolves the caller to the one host it may speak about.
//
// The dangerous shape here is WorkerToken returning "" — unknown host, asleep,
// stopped — being read as "this host needs no token", which would be a total
// bypass for every host that is not currently awake. So the empty case is not
// handled by a guard that a later edit could delete: an absent token is
// replaced with fresh random bytes, and the comparison then simply cannot
// succeed. Refusal is a property of the compare, not of a branch above it.
//
// A caller that names no host at all is refused here, not left to WorkerToken.
// Asking a collaborator what "" means is the same mistake in a different
// place: this package would be safe only for as long as the activator's map
// happens to hold no "" key, which is not a property this package can see or
// keep. An unnamed caller has nothing to authorize, so it never reaches the
// lookup — and gets the same 401 as every other refusal.
//
// The compare itself is constant-time, and both failure paths — no live token,
// wrong live token — run the same code, so the structure carries no timing
// signal beyond the length leak inherent to any comparison.
func (s *Server) authorize(r *http.Request) (string, bool) {
	host := r.Header.Get(hostHeader)
	presented := bearer(r.Header.Get("Authorization"))

	want := ""
	if host != "" {
		want = s.w.WorkerToken(host)
	}
	if want == "" {
		want = unmatchable()
	}
	return host, subtle.ConstantTimeCompare([]byte(presented), []byte(want)) == 1
}

// unmatchable returns a value no caller can present. crypto/rand.Read never
// fails (it panics on a broken system rather than returning short), so this
// has no error path that could degrade into a guessable constant.
func unmatchable() string {
	var b [32]byte
	rand.Read(b[:])
	return string(b[:])
}

// bearer extracts the credential from an Authorization header, returning ""
// for anything that is not exactly a non-empty Bearer credential. "Bearer "
// with nothing after it yields "", which authorize then cannot match.
func bearer(h string) string {
	const scheme = "Bearer "
	if len(h) > len(scheme) && strings.EqualFold(h[:len(scheme)], scheme) {
		return h[len(scheme):]
	}
	return ""
}

// refuse is the only 401 in the package. One status, one body, for every
// reason — a caller learns that it was refused and nothing else.
func refuse(w http.ResponseWriter) {
	http.Error(w, "unauthorized", http.StatusUnauthorized)
}

// bad is the only 400 in the package, for the same reason.
func bad(w http.ResponseWriter) {
	http.Error(w, "bad request", http.StatusBadRequest)
}

// ready ends the settle window and records the build the child says it is
// running. The version is the child's own claim — §9.2's honesty rule — so it
// is bounded and character-checked, never trusted. The field must be PRESENT
// (a typo'd body is a loud 400, not a silent "unknown"), but may be empty: a
// child that genuinely does not know its build says so, and is still ready.
func (s *Server) ready(w http.ResponseWriter, r *http.Request, host string) {
	var body struct {
		Version *string `json:"version"`
	}
	if !decode(r, &body) {
		bad(w)
		return
	}
	if body.Version == nil || len(*body.Version) > maxVersion || !displayable(*body.Version) {
		bad(w)
		return
	}
	s.w.NoteWorkerReady(host, *body.Version)
	w.WriteHeader(http.StatusNoContent)
}

// nextDue records when the child expects to work again, while it is alive to
// say so. "at" must be present and non-negative: an absent field decoding to
// zero would silently reschedule the worker to the epoch.
func (s *Server) nextDue(w http.ResponseWriter, r *http.Request, host string) {
	var body struct {
		At *int64 `json:"at"`
	}
	if !decode(r, &body) {
		bad(w)
		return
	}
	if body.At == nil || *body.At < 0 {
		bad(w)
		return
	}
	if err := s.w.SetWorkerNextDue(host, *body.At); err != nil {
		// The child must not read a 204 for a schedule that did not land.
		log.Printf("agentsock: set next-due for %s: %v", host, err)
		http.Error(w, "internal error", http.StatusInternalServerError)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

// idle is "I have nothing to do". It carries no arguments, so nothing in the
// request can steer it: the authorized host is the whole message. Whether the
// request is honoured at all is the activator's call — an always-on worker's
// idle is refused there.
func (s *Server) idle(w http.ResponseWriter, _ *http.Request, host string) {
	s.w.SleepWorker(host)
	w.WriteHeader(http.StatusNoContent)
}

// heartbeat is a worker's only sign of life: nothing routes to a sidecar, so
// no request will ever stamp one. Argument-free for the same reason as idle.
func (s *Server) heartbeat(w http.ResponseWriter, _ *http.Request, host string) {
	s.w.NoteWorkerContact(host)
	w.WriteHeader(http.StatusNoContent)
}

// decode parses a bounded body into v. The cap is applied to the READER, so an
// oversized body is truncated into a parse error rather than accepted: there is
// no size at which a caller gets a partial message honoured.
func decode(r *http.Request, v any) bool {
	return json.NewDecoder(io.LimitReader(r.Body, maxBody)).Decode(v) == nil
}

// displayable rejects control characters in a string bound for member-facing
// output — a version line should not be able to carry a NUL, a newline, or a
// terminal escape into a log or a console page.
func displayable(s string) bool {
	for _, r := range s {
		if r < 0x20 || r == 0x7f {
			return false
		}
	}
	return true
}

// Serve binds path and answers on it until ctx is cancelled, then shuts down
// and leaves no socket file behind.
//
// A socket file that outlives its agent is the ordinary result of a hard kill,
// and the next start must take it over rather than die on "address already in
// use" — so a stale file is removed before binding. A file somebody is still
// LISTENING on is a different thing entirely: unlinking that would orphan a
// live agent, whose children would go on talking to a process nothing else can
// reach. So we probe first, and refuse rather than steal — and, for the same
// reason, this function unlinks nothing it did not create (see below).
func (s *Server) Serve(ctx context.Context, path string) error {
	if listening(path) {
		return fmt.Errorf("agentsock: %s is already served by a live listener", path)
	}
	if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
		return fmt.Errorf("agentsock: remove stale socket %s: %w", path, err)
	}
	ln, err := net.Listen("unix", path)
	if err != nil {
		return fmt.Errorf("agentsock: listen on %s: %w", path, err)
	}
	// Cleanup is the LISTENER's job and nothing else's. Go unlinks a unix
	// socket it created when the listener closes, and that unlink happens
	// inside Close — indivisibly, from the point of view of a successor
	// trying to bind the same path.
	//
	// A deferred os.Remove(path) here would not be. Between the listener's
	// own unlink and the deferred call, a successor agent can have bound the
	// path, and we would then unlink ITS socket: the successor serves on an
	// inode nothing can reach, its children never arrive, and the box has the
	// same split-brain the probe above exists to refuse to cause. Guarding
	// one door and leaving the other open is worse than guarding neither,
	// because it reads as guarded. So there is no unconditional remove, and
	// every return below closes the listener instead.
	if ul, ok := ln.(*net.UnixListener); ok {
		ul.SetUnlinkOnClose(true) // the default; pinned because it is load-bearing
	}

	if err := os.Chmod(path, socketMode); err != nil {
		ln.Close() // unlinks the socket we just created, and only that one
		return fmt.Errorf("agentsock: chmod %s: %w", path, err)
	}

	srv := &http.Server{
		Handler:           s.Handler(),
		ReadHeaderTimeout: readHeaderTimeout,
		ReadTimeout:       readTimeout,
		WriteTimeout:      writeTimeout,
		IdleTimeout:       idleTimeout,
		MaxHeaderBytes:    1 << 14,
	}

	errc := make(chan error, 1)
	go func() { errc <- srv.Serve(ln) }()

	select {
	case err := <-errc:
		if errors.Is(err, http.ErrServerClosed) {
			return nil
		}
		return err
	case <-ctx.Done():
		// ctx is already cancelled, so Shutdown gets its own deadline.
		sctx, cancel := context.WithTimeout(context.Background(), shutdownGrace)
		defer cancel()
		err := srv.Shutdown(sctx)
		<-errc
		return err
	}
}

// listening reports whether something is currently accepting on path. A stale
// socket file, a regular file, and a missing path all answer false.
func listening(path string) bool {
	c, err := net.DialTimeout("unix", path, dialProbe)
	if err != nil {
		return false
	}
	c.Close()
	return true
}