// Package-internal reverse-channel hub: remote hosts dial OUT to the edge's
// existing 443 listener on a reserved hostname, authenticate with a bearer
// token bound to their channel name, and the connection inverts — the edge
// becomes an HTTP/2 client over the hijacked conn, the host an HTTP/2 server.
// Nothing ever connects TO a remote host (issue #53's locked-down property);
// this file is the edge half of that bargain. Transport judgment call
// (spec 2026-08-04): reverse h2 rather than a websocket port of titogo's
// tunnel — x/net/http2 already supplies mux, flow control, and pings.
package edge

import (
	"context"
	"crypto/sha256"
	"crypto/subtle"
	"encoding/hex"
	"fmt"
	"log"
	"net"
	"net/http"
	"slices"
	"strings"
	"sync"
	"time"

	"golang.org/x/net/http/httpguts"
	"golang.org/x/net/http2"
)

const (
	tunnelRegisterPath = "/register"
	// tunnelUpgradeProto is the wire contract the host dialer codes against:
	//
	//	GET /register
	//	Authorization: Bearer <token>
	//	Carlos-Tunnel-Name: <name>
	//	Connection: Upgrade
	//	Upgrade: carlos-tunnel/1
	//
	// The dialer MUST also pin NextProtos to {"http/1.1"} on its TLS config.
	// The edge's 443 listener advertises h2, and a dialer that lets ALPN pick
	// it authenticates fine and then dies on "hijack unsupported": HTTP/2
	// response writers are not http.Hijacker, and hijacking is how this
	// handshake gets the raw conn. The h2 session begins only AFTER the 101,
	// on that raw conn, with the roles inverted.
	tunnelUpgradeProto = "carlos-tunnel/1"
	// tunnelPingEvery / tunnelPingTimeout bound how long a half-dead channel
	// can look connected: a failed or timed-out ping tears it down, the host
	// notices its conn die and re-dials (seconds — the bound Task 10 asserts).
	tunnelPingEvery   = 15 * time.Second
	tunnelPingTimeout = 10 * time.Second
	// Fleet (dynamic) channels ping tighter (fleets-console spec D7): a
	// customer fleet's failover is "requests fall back to a surviving box once
	// the dead channel is reaped", so the reap bound IS the user-visible
	// outage window — ≤10s (one 5s sleep + one 5s deadline) instead of the
	// static channels' ~25s. Static platform channels (tito hq) keep the wider
	// cadence: their routes have no fallback box, so a tighter ping only adds
	// churn risk for no availability gain.
	fleetPingEvery   = 5 * time.Second
	fleetPingTimeout = 5 * time.Second
)

// pingCadence picks a channel's liveness cadence from which credential source
// admitted it. Split out from pingLoop so the choice is testable without
// waiting on wall-clock pings.
func pingCadence(dynamic bool) (every, timeout time.Duration) {
	if dynamic {
		return fleetPingEvery, fleetPingTimeout
	}
	return tunnelPingEvery, tunnelPingTimeout
}

type tunnelConn struct {
	cc  *http2.ClientConn
	raw net.Conn
	// dynamic records that this conn authenticated against the registry-backed
	// (fleet) credential source rather than the static env map — the one fact
	// ServeHTTP knows and pingLoop needs for pingCadence.
	dynamic bool
}

type TunnelHub struct {
	host    string            // reserved hostname this hub answers on
	clients map[string]string // channel name -> bearer token (static env config, plaintext)
	mu      sync.Mutex
	conns   map[string]*tunnelConn
	// dynamicAuth is the second, registry-backed credential source: channel
	// "<fleet>/<label>" -> sha256 hex of its bearer token. Guarded by mu
	// alongside conns since SetDynamicAuth's rotation path (below) touches
	// both. Static clients keep authenticating by plaintext compare,
	// unchanged; this is consulted only when a name misses h.clients, which
	// is always true for a dynamic name — ValidTunnelName-checked static
	// names never contain "/".
	dynamicAuth map[string]string
}

func NewTunnelHub(host string, clients map[string]string) *TunnelHub {
	return &TunnelHub{host: host, clients: clients, conns: make(map[string]*tunnelConn)}
}

func (h *TunnelHub) Host() string { return h.host }

// parseTunnelClients parses CARLOS_TUNNEL_CLIENTS ("name=token[,name=token]").
func parseTunnelClients(s string) (map[string]string, error) {
	out := make(map[string]string)
	for _, pair := range strings.Split(s, ",") {
		pair = strings.TrimSpace(pair)
		if pair == "" {
			continue
		}
		name, token, ok := strings.Cut(pair, "=")
		if !ok || name == "" || token == "" {
			return nil, fmt.Errorf("tunnel clients: malformed entry %q (want name=token)", pair)
		}
		out[name] = token
	}
	return out, nil
}

// ServeHTTP handles ONLY the registration handshake; the Edge routes requests
// here when r.Host equals the reserved hostname, so ordinary app traffic can
// never reach it and this handler never sees ordinary traffic.
func (h *TunnelHub) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.URL.Path != tunnelRegisterPath {
		http.NotFound(w, r)
		return
	}
	name := r.Header.Get("Carlos-Tunnel-Name")
	// CutPrefix's ok is load-bearing: TrimPrefix would return a scheme-less
	// header (e.g. "Authorization: sekrit") unchanged, letting a credential
	// that never claimed to be a bearer token compare equal to a configured
	// one. No "Bearer " prefix is an outright 401, not a token of "".
	token, hasBearer := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ")
	want, ok := h.clients[name]
	// Constant-time compare; the name lookup happening first is fine — the
	// name is not the secret, the token is. A token is bound to exactly one
	// channel name, so a leaked token cannot impersonate a different host.
	// An empty configured token never authenticates anything: parseTunnelClients
	// rejects one, and a hub built some other way must not turn a blank config
	// value into a channel that answers to "Authorization: Bearer ".
	authOK := name != "" && ok && want != "" && hasBearer &&
		subtle.ConstantTimeCompare([]byte(token), []byte(want)) == 1
	// Second credential source: a name that misses the static map (every
	// "<fleet>/<label>" dynamic channel does, since ValidTunnelName-checked
	// static names never contain "/") falls through to the registry-backed
	// set SetDynamicAuth maintains. The presented token is never compared in
	// the clear here — only its sha256 hex is, against the stored hash — so
	// a leaked auth log or memory dump of this table never yields a usable
	// bearer token. Folded into the SAME authOK the static path feeds, so
	// there is exactly one 401 below regardless of which source (or neither)
	// missed; a caller cannot distinguish "wrong static token" from "unknown
	// dynamic channel" from "wrong dynamic token" by response shape.
	// viaDynamic is that branch's other output: which source admitted this
	// conn decides its ping cadence (pingCadence), and this is the only place
	// that fact is known.
	viaDynamic := false
	if !authOK && name != "" && hasBearer {
		_, viaDynamic = h.VerifyDynamic(name, token)
		authOK = viaDynamic
	}
	if !authOK {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	// Wire-contract gate, deliberately AFTER auth: an unauthenticated caller
	// gets 401 whatever it sends, so probing cannot discover which protocols
	// this endpoint speaks. Checking it at all is what keeps a version skew
	// diagnosable — without it we would answer "101 carlos-tunnel/1" to a
	// client that never asked, and a future carlos-tunnel/2 host would see a
	// mystery h2-preface failure instead of a refusal naming the version.
	// Connection is matched token-wise and Upgrade case-insensitively, both
	// per RFC 7230; the version suffix is what must match exactly.
	if !httpguts.HeaderValuesContainsToken(r.Header["Connection"], "Upgrade") ||
		!strings.EqualFold(strings.TrimSpace(r.Header.Get("Upgrade")), tunnelUpgradeProto) {
		http.Error(w, "expected Connection: Upgrade with Upgrade: "+tunnelUpgradeProto, http.StatusBadRequest)
		return
	}
	hj, ok := w.(http.Hijacker)
	if !ok {
		http.Error(w, "hijack unsupported", http.StatusInternalServerError)
		return
	}
	conn, bufrw, err := hj.Hijack()
	if err != nil {
		log.Printf("edge: tunnel %s: hijack: %v", name, err)
		return
	}
	// The h2 session runs on the RAW conn, so anything the host pipelined
	// ahead of our 101 is sitting unreadable in bufrw and would corrupt the
	// handshake. Fail loudly here rather than hang on a broken preface: the
	// host-side dialer's contract is to write nothing until it reads the 101.
	if n := bufrw.Reader.Buffered(); n > 0 {
		log.Printf("edge: tunnel %s: %d bytes sent before 101 — dropping", name, n)
		conn.Close()
		return
	}
	// Manual 101 on the hijacked conn (the ResponseWriter is spent once hijacked).
	fmt.Fprintf(bufrw, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: %s\r\nConnection: Upgrade\r\n\r\n", tunnelUpgradeProto)
	if err := bufrw.Flush(); err != nil {
		conn.Close()
		return
	}
	tr := &http2.Transport{}
	cc, err := tr.NewClientConn(conn)
	if err != nil {
		log.Printf("edge: tunnel %s: h2 client conn: %v", name, err)
		conn.Close()
		return
	}
	h.mu.Lock()
	if old := h.conns[name]; old != nil {
		old.raw.Close() // a re-register replaces the previous channel
	}
	h.conns[name] = &tunnelConn{cc: cc, raw: conn, dynamic: viaDynamic}
	h.mu.Unlock()
	log.Printf("edge: tunnel %s: connected from %s", name, conn.RemoteAddr())
	go h.pingLoop(name, cc, conn, viaDynamic)
}

// pingLoop is the edge-side liveness check: a dead ping closes the raw conn
// and unregisters the channel, so requests fail fast to 502 instead of
// hanging on a black-holed TCP session (the venue-Wi-Fi failure mode).
// dynamic carries the credential source through from ServeHTTP: fleet
// channels reap on a tighter bound because that bound is their failover
// window (see pingCadence).
func (h *TunnelHub) pingLoop(name string, cc *http2.ClientConn, raw net.Conn, dynamic bool) {
	every, timeout := pingCadence(dynamic)
	for {
		time.Sleep(every)
		ctx, cancel := context.WithTimeout(context.Background(), timeout)
		err := cc.Ping(ctx)
		cancel()
		if err != nil {
			log.Printf("edge: tunnel %s: ping failed (%v) — dropping channel", name, err)
			raw.Close()
			// Only unregister if WE are still the live channel: a re-register
			// may already have replaced this entry (and closed our conn, which
			// is what failed the ping), and deleting then would blank out a
			// perfectly healthy successor.
			h.mu.Lock()
			if cur := h.conns[name]; cur != nil && cur.raw == raw {
				delete(h.conns, name)
			}
			h.mu.Unlock()
			return
		}
	}
}

// VerifyDynamic reports whether (name, token) authenticates as a dynamic
// (registry-backed fleet) channel, and which fleet owns it. It is the single
// answer to that question: ServeHTTP's dynamic branch is one caller, and
// surfaces with no ResponseWriter to 401 with are the others.
//
// The "/" requirement is a gate, not a parse convenience: static channel names
// are ValidTunnelName-checked and never contain "/", so refusing slash-less
// names here means a static client's plaintext token can never be admitted by
// this path, whatever h.clients (or a hand-built dynamicAuth) holds. An empty
// fleet segment ("/box-a") is refused for the same reason: it would otherwise
// authenticate into fleet "", a name no real fleet has, and every ownership
// check downstream compares against the fleet this returns.
//
// The presented token is never compared in the clear — only its sha256 hex is,
// against the stored hash — so a leaked dump of this table yields no usable
// bearer token, and the compare is constant-time so a wrong token leaks no
// prefix information by timing.
func (h *TunnelHub) VerifyDynamic(name, token string) (fleet string, ok bool) {
	fleet, _, hasSlash := strings.Cut(name, "/")
	if !hasSlash || fleet == "" {
		return "", false
	}
	h.mu.Lock()
	wantHash, known := h.dynamicAuth[name]
	h.mu.Unlock()
	if !known || wantHash == "" {
		return "", false
	}
	sum := sha256.Sum256([]byte(token))
	gotHash := hex.EncodeToString(sum[:])
	if subtle.ConstantTimeCompare([]byte(gotHash), []byte(wantHash)) != 1 {
		return "", false
	}
	return fleet, true
}

// SetDynamicAuth replaces the entire registry-backed credential set —
// channel "<fleet>/<label>" -> sha256 hex of its bearer token — with auth.
// The whole-map REPLACE (not an incremental merge) is what makes a NEW dial
// with a stale token get refused the instant this returns, for both
// rotation and detach: the next ServeHTTP call either hashes the presented
// token against a DIFFERENT stored hash (rotation: the name is still
// present, just under a new hash) or misses the map entirely (detach: the
// name is gone). Either way, a new dial with the old token fails now.
//
// ROTATE and DETACH are deliberately asymmetric on an ALREADY-LIVE session,
// and that asymmetry is load-bearing, not an oversight:
//
//   - Detach: the name vanishes from auth entirely, so it is in the OLD set
//     but missing from the new one — the loop below calls CloseChannel on
//     it, force-closing any live conn right now. Detach is the hard
//     revocation path; a box taken out of the fleet must not keep a
//     channel open on its old credential for even one more request.
//   - Rotate: the name STAYS present in auth, just under a new hash — the
//     loop below sees it as still-authorized (present in both old and new)
//     and does NOT close it. An already-established h2 session therefore
//     keeps carrying traffic on its OLD token's connection until the
//     legitimate box notices (its own ping/reconnect cycle) and re-dials
//     with the NEW token — not until this call returns. This is
//     deliberate: closing on every rotation would take the box's channel
//     down for the gap between the registry write and the box learning its
//     new token, and rotation (unlike detach) is not "this box must be cut
//     off now" — it is "this box's credential changed." A caller that wants
//     rotation to also force an immediate reconnect must close explicitly
//     (CloseChannel) rather than relying on SetDynamicAuth to do it.
func (h *TunnelHub) SetDynamicAuth(auth map[string]string) {
	h.mu.Lock()
	old := h.dynamicAuth
	h.dynamicAuth = auth
	h.mu.Unlock()
	for name := range old {
		if _, stillAuthorized := auth[name]; !stillAuthorized {
			h.CloseChannel(name)
		}
	}
}

// CloseChannel force-closes and unregisters name's live conn, if any — the
// detach/rotate path (SetDynamicAuth above), also usable directly. Mirrors
// pingLoop's delete-if-still-live guard: the raw conn is closed OUTSIDE the
// lock (Close can block briefly on the wire), then a second lock re-checks
// that this exact conn is STILL the one registered before deleting it — a
// re-register that has already replaced the entry in between must not have
// ITS fresh conn deleted out from under it.
func (h *TunnelHub) CloseChannel(name string) {
	h.mu.Lock()
	cur := h.conns[name]
	h.mu.Unlock()
	if cur == nil {
		return
	}
	cur.raw.Close()
	h.mu.Lock()
	if c := h.conns[name]; c != nil && c.raw == cur.raw {
		delete(h.conns, name)
	}
	h.mu.Unlock()
}

func (h *TunnelHub) conn(name string) *http2.ClientConn {
	h.mu.Lock()
	defer h.mu.Unlock()
	if c := h.conns[name]; c != nil {
		return c.cc
	}
	return nil
}

// State reports "connected" or "". The heartbeat's tunnel-state field comes
// from the host-side dial loop, not from here; today this is test-consumed
// only — no production caller reads it.
func (h *TunnelHub) State(name string) string {
	if h.conn(name) != nil {
		return "connected"
	}
	return ""
}

// ConnectedOfFleet returns every currently-connected channel name belonging
// to fleet — the "<fleet>/<label>" names whose prefix matches — sorted. This
// is what makes the edge's fleet-route fallback (edge.go's fleetProxy)
// deterministic: "any connected box of this fleet" always picks the same one
// for a given live set, first entry, rather than whatever order Go's map
// iteration happens to produce.
func (h *TunnelHub) ConnectedOfFleet(fleet string) []string {
	prefix := fleet + "/"
	h.mu.Lock()
	defer h.mu.Unlock()
	var out []string
	for name := range h.conns {
		if strings.HasPrefix(name, prefix) {
			out = append(out, name)
		}
	}
	slices.Sort(out)
	return out
}

type roundTripperFunc func(*http.Request) (*http.Response, error)

func (f roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }

// transportFor resolves the channel PER REQUEST, so a proxy built before a
// reconnect uses the fresh channel afterwards with no reload needed.
func (h *TunnelHub) transportFor(name string) http.RoundTripper {
	return roundTripperFunc(func(req *http.Request) (*http.Response, error) {
		cc := h.conn(name)
		if cc == nil {
			return nil, fmt.Errorf("tunnel %q not connected", name)
		}
		return cc.RoundTrip(req)
	})
}