felhom.eu/hub/internal/intent/hub.go

// Package intent is the hub's in-memory operator-intent notifier — the server side of the
// Direction-2 immediate-sync arc (SPIKE-immediate-sync-transport-2026-07-16.md, option b).
//
// Every operator action that changes a customer's desired state (config save/delete, claim
// resend, offsite re-issue/freeze, floor change, block/unblock, log-pull request) Bumps that
// customer's generation. A box holds a hanging GET /api/v1/wait against the hub carrying its
// last-seen generation; Wait returns the instant the generation moves (or on timeout/shutdown).
// The box then fires its ordinary out-of-cycle report — the ACK delivers everything through the
// UNCHANGED report machinery. The wait signal is CONTENTLESS: it carries a generation number
// and nothing else (no config, no escrow, no claim).
//
// The counter is IN-MEMORY BY DESIGN — a restart resets all generations to zero, which the box
// treats as "a different generation" (it compares with != , not >), firing exactly one harmless
// full-state report before settling. No persistence, no store schema (Part 2.1 / §5).
//
// Semantics mirror the wgsync reconciler / controller report.Trigger: a per-customer counter,
// coalescing (N bumps during one hold → one completion carrying the LATEST generation), never a
// per-bump queue. Every Wait exit path deregisters its channel, so there is no waiter leak.
package intent

import (
	"context"
	"sync"
	"time"
)

// Hub is the in-memory per-customer generation counter + waiter registry. The zero value is not
// usable; construct with New. All methods are safe for concurrent use.
type Hub struct {
	mu       sync.Mutex
	gen      map[string]uint64          // customerID -> current generation (0 = never bumped)
	waiters  map[string][]chan struct{} // customerID -> registered wake channels (buffered, cap 1)
	closed   chan struct{}              // closed by Close() to wake every waiter
	isClosed bool

	// R-30 box presence (presence.go): per-customer wait-request bookkeeping, the hub's own start
	// time (a hub younger than PresenceConnectedWindow answers "unknown"), a clock seam and an
	// optional debug logger.
	presence map[string]*presenceRec
	born     time.Time
	now      func() time.Time
	logf     func(format string, args ...any)
}

// New builds an empty intent hub.
func New() *Hub {
	return &Hub{
		gen:      make(map[string]uint64),
		waiters:  make(map[string][]chan struct{}),
		closed:   make(chan struct{}),
		presence: make(map[string]*presenceRec),
		born:     time.Now(),
	}
}

// Bump advances the customer's generation and wakes every currently-registered waiter for that
// customer, then clears them (each waiter re-reads the LATEST generation under the lock on wake,
// so a burst of bumps collapses into one completion carrying the final value — a counter, not a
// queue). Safe from any goroutine; never blocks. An empty customerID is a no-op, as is any bump
// after Close.
func (h *Hub) Bump(customerID string) {
	if customerID == "" {
		return
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.isClosed {
		return
	}
	h.gen[customerID]++
	for _, ch := range h.waiters[customerID] {
		// Non-blocking send into a cap-1 channel: the waiter reads the latest gen under the lock,
		// so a single wake is all it needs; a second bump before the waiter runs is harmless.
		select {
		case ch <- struct{}{}:
		default:
		}
	}
	delete(h.waiters, customerID)
}

// Wait blocks until the customer's generation differs from lastSeen, ctx is cancelled, maxHold
// elapses, or Close is called — whichever first — and returns the customer's CURRENT generation
// to hand back to the caller. If the generation already differs from lastSeen at entry (a bump
// landed before this wait registered), it returns immediately: the race gap between a bump and a
// (re)connect is closed by this pre-register check, so a bump is never lost.
func (h *Hub) Wait(ctx context.Context, customerID string, lastSeen uint64, maxHold time.Duration) uint64 {
	h.mu.Lock()
	if h.isClosed {
		cur := h.gen[customerID]
		h.mu.Unlock()
		return cur
	}
	if cur := h.gen[customerID]; cur != lastSeen {
		h.mu.Unlock()
		return cur // race-closer: intent already moved; no need to hold
	}
	ch := make(chan struct{}, 1)
	h.waiters[customerID] = append(h.waiters[customerID], ch)
	h.mu.Unlock()

	timer := time.NewTimer(maxHold)
	defer timer.Stop()

	select {
	case <-ch:
	case <-ctx.Done():
	case <-timer.C:
	case <-h.closed:
	}
	return h.finish(customerID, ch)
}

// finish deregisters ch (if a Bump/Close hasn't already cleared it) and returns the customer's
// current — i.e. latest — generation. Called on every Wait exit path, so waiters never leak.
func (h *Hub) finish(customerID string, ch chan struct{}) uint64 {
	h.mu.Lock()
	defer h.mu.Unlock()
	ws := h.waiters[customerID]
	for i, c := range ws {
		if c == ch {
			h.waiters[customerID] = append(ws[:i], ws[i+1:]...)
			break
		}
	}
	if len(h.waiters[customerID]) == 0 {
		delete(h.waiters, customerID)
	}
	return h.gen[customerID]
}

// Close wakes every registered waiter (each returns its customer's current generation) and makes
// all subsequent Wait calls return immediately. Idempotent. Wired into the hub's graceful
// shutdown BEFORE server.Shutdown so held waits complete instantly instead of eating the grace
// window.
func (h *Hub) Close() {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.isClosed {
		return
	}
	h.isClosed = true
	close(h.closed)
	h.waiters = make(map[string][]chan struct{})
}

// Generation returns the customer's current generation (0 if never bumped). Test/observability
// helper — the wait path uses the value returned by Wait directly.
func (h *Hub) Generation(customerID string) uint64 {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.gen[customerID]
}