fleetd/bridged/src/main/java/dev/ltms/bridged/metrics/BridgedMetrics.java

package dev.ltms.bridged.metrics;

import dev.ltms.bridged.msg.ReplyInbox;
import dev.ltms.bridged.session.SessionManager;
import dev.ltms.bridged.session.MemberSession;

import java.util.LinkedHashMap;
import java.util.Map;

/**
 * The daemon's metric definitions (CB-502) — one place where every series is named, described, and
 * (for gauges) bound to live state.
 *
 * <p>The set is deliberately small: each series maps to a failure mode this project has actually
 * hit, not to whatever was easy to count. The two worth watching in practice are
 * {@code bridged_sends_total{outcome="completion_fallback"}} — a rising share means turn detection
 * is degrading, the CB-115/116/118 failure family — and
 * {@code bridged_push_nudges_total{outcome="exhausted"}}, which means the primary stopped draining
 * its inbox and CB-307's active push gave up.
 */
public final class BridgedMetrics {

    /** Counter: delegated sends by terminal outcome. */
    public static final String SENDS = "bridged_sends_total";
    /** Counter: worker replies by the path that carried them (rendezvous vs stranded-to-inbox). */
    public static final String REPLIES = "bridged_replies_total";
    /** Counter: push-loop nudges to the primary, by outcome. */
    public static final String PUSH_NUDGES = "bridged_push_nudges_total";
    /** Counter: idle-lead heartbeat nudges to the lead, by outcome (CB-551). */
    public static final String HEARTBEAT_NUDGES = "bridged_lead_heartbeat_nudges_total";
    /** Counter: spawn attempts by peer kind and outcome. */
    public static final String SPAWNS = "bridged_spawns_total";
    /** Counter: herdr socket calls by method and outcome. */
    public static final String HERDR_CALLS = "bridged_herdr_calls_total";
    /** Counter: rejected requests by reason (CB-501). */
    public static final String AUTH_FAILURES = "bridged_auth_failures_total";
    /** Gauge: session census by lifecycle state. */
    public static final String SESSIONS = "bridged_sessions";
    /** Gauge: undrained replies held per target. */
    public static final String INBOX_DEPTH = "bridged_inbox_depth";

    private BridgedMetrics() {
    }

    /**
     * Build the registry with its help text and live gauges bound.
     *
     * @param sessions the authoritative session registry (census gauge)
     * @param inbox    the reply inbox; only used for a depth gauge when it can be inspected
     */
    public static Metrics create(SessionManager sessions, ReplyInbox inbox) {
        Metrics m = new Metrics();

        m.describe(SENDS, "counter",
                "Delegated sends by terminal outcome (replied|completion_fallback|timeout|failed).");
        m.describe(REPLIES, "counter",
                "Worker replies by delivery path (rendezvous=resolved an open send, inbox=stranded and held).");
        m.describe(PUSH_NUDGES, "counter",
                "CB-307 push-loop nudges to the primary (delivered|exhausted).");
        m.describe(HEARTBEAT_NUDGES, "counter",
                "CB-551 idle-lead heartbeat nudges (delivered|failed|exhausted). Quiet-cap exhaustion "
                        + "means the lead idled with nothing pending and was told to stand down.");
        m.describe(SPAWNS, "counter",
                "Worker spawn attempts by peer kind and outcome (ready|timeout|guard_rejected).");
        m.describe(HERDR_CALLS, "counter",
                "herdr socket calls by method and outcome — the dependency everything else rests on.");
        m.describe(AUTH_FAILURES, "counter",
                "Requests refused by CB-501/505 (unauthenticated|forbidden).");
        m.describe(SESSIONS, "gauge",
                "Registered worker sessions by lifecycle state.");
        m.describe(INBOX_DEPTH, "gauge",
                "Replies held for a target that the primary has not drained. Steady state is 0; "
                        + "a target stuck above 0 means CB-307 delivery is not completing.");

        // One gauge per state so a scrape shows the whole census even when a state is empty —
        // an absent series and a zero series read very differently on a dashboard.
        for (MemberSession.State state : MemberSession.State.values()) {
            String label = state.name().toLowerCase();
            m.gauge(SESSIONS, () -> countIn(sessions, state), "state", label);
        }

        // Depth is per live session, so the label set is only known at scrape time. peek() is the
        // port's non-destructive read — scraping metrics must never ack a reply out of the inbox.
        m.collector(INBOX_DEPTH, "target", () -> {
            Map<String, Number> depths = new LinkedHashMap<>();
            for (MemberSession s : sessions.roster()) {
                String target = s.terminalId();
                if (target == null) {
                    continue;
                }
                depths.put(target, inbox.peek(target).size());
            }
            return depths;
        });
        return m;
    }

    private static long countIn(SessionManager sessions, MemberSession.State state) {
        return sessions.roster().stream().filter(s -> s.state() == state).count();
    }
}