felhom.eu/hub/internal/api/wait.go

package api

import (
	"fmt"
	"net/http"
	"strconv"
	"time"
)

// waitMaxHold / waitHeartbeat are effectively constants — vars only so tests can shrink them.
var (
	// waitMaxHold is how long the hub holds a wait before completing it with the (unchanged)
	// generation. Kept well under the ingress/NAT idle ceilings the transport spike measured
	// (SPIKE-immediate-sync-transport-2026-07-16.md): the nginx 60 s proxy_read_timeout is defeated
	// by the heartbeat below, and the operator NAT held an idle hold > 600 s. 240 s → a box
	// reconnects roughly every 4 min in steady state (negligible fleet load).
	waitMaxHold = 240 * time.Second
	// waitHeartbeat is the spacing of keepalive newline bytes written while holding. nginx's
	// proxy_read_timeout is measured BETWEEN successive reads from the upstream, so a byte every
	// 25 s keeps the default 60 s from ever firing — no ingress annotation / manifest change is
	// needed (§13 proves this live; the annotation is the documented fallback if it doesn't hold).
	// 25 s mirrors the WG PersistentKeepalive cadence and doubles as NAT keepalive.
	waitHeartbeat = 25 * time.Second
)

// handleWait is the Direction-2 immediate-sync long-poll (option b of the transport spike). A box
// holds this authenticated GET carrying its last-seen generation (?gen=N); the hub completes it the
// instant any operator intent lands for the box's customer (intent.Hub.Bump from the web handlers),
// or after waitMaxHold, whichever comes first. The response is CONTENTLESS — a single {"gen":N}
// line preceded by heartbeat newlines. The box reacts by firing its ordinary out-of-cycle report;
// the report ACK then delivers config/escrow/claim/floor through the UNCHANGED machinery. The hub
// never connects inbound and never rides state on this channel (ground truth #1).
func (h *Handler) handleWait(w http.ResponseWriter, r *http.Request) {
	customerID, isGlobal, ok := h.checkAuthCustomer(r)
	if !ok {
		http.Error(w, "Unauthorized", http.StatusUnauthorized)
		return
	}
	// Per-customer only: the key RESOLVES the customer and no customer_id parameter is accepted,
	// so customer A can never observe B's generation. A global operator key has no single customer
	// to wait on → 400 (the controller always holds a per-customer key).
	if isGlobal {
		http.Error(w, "wait is per-customer (present a customer key)", http.StatusBadRequest)
		return
	}
	if h.intentHub == nil {
		http.Error(w, "wait channel unavailable", http.StatusServiceUnavailable)
		return
	}

	var lastSeen uint64
	if s := r.URL.Query().Get("gen"); s != "" {
		if v, err := strconv.ParseUint(s, 10, 64); err == nil {
			lastSeen = v
		}
	}

	// Lift THIS connection's write deadline past the hold. The global http.Server.WriteTimeout is
	// 60 s (main.go — deliberately untouched); http.NewResponseController overrides it per-connection
	// (Go 1.20+). Without this the held response is cut at 60 s regardless of the ingress.
	rc := http.NewResponseController(w)
	_ = rc.SetWriteDeadline(time.Now().Add(waitMaxHold + 30*time.Second))

	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
	w.Header().Set("X-Accel-Buffering", "no") // ask nginx not to buffer the streamed hold
	w.Header().Set("Cache-Control", "no-store")
	w.WriteHeader(http.StatusOK)
	if err := rc.Flush(); err != nil {
		return // client already gone
	}

	deadline := time.Now().Add(waitMaxHold)
	for {
		remaining := time.Until(deadline)
		if remaining <= 0 {
			// Hold elapsed with no change: complete with the current generation (Scenario B).
			h.writeWaitGen(w, rc, h.intentHub.Generation(customerID))
			return
		}
		sub := remaining
		if sub > waitHeartbeat {
			sub = waitHeartbeat
		}
		gen := h.intentHub.Wait(r.Context(), customerID, lastSeen, sub)
		if gen != lastSeen {
			h.writeWaitGen(w, rc, gen) // intent moved → complete immediately
			return
		}
		// No change within this heartbeat window: emit a keepalive byte so the ingress read-timeout
		// never fires. A write/flush failure means the client is gone — return quietly (no log
		// noise, no goroutine leak; the intent.Wait above already deregistered).
		if _, err := w.Write([]byte("\n")); err != nil {
			return
		}
		if err := rc.Flush(); err != nil {
			return
		}
	}
}

// writeWaitGen emits the single contentless completion line and flushes.
func (h *Handler) writeWaitGen(w http.ResponseWriter, rc *http.ResponseController, gen uint64) {
	fmt.Fprintf(w, "{\"gen\":%d}\n", gen)
	_ = rc.Flush()
}