# Per-stage pipeline metrics.
#
# End-to-end throughput says how fast the pipeline is; it does not say which
# stage is the reason. The external harness can't answer that either, and now
# cannot even approximate it: every in-flight file sits in `spool/` regardless of
# how far it has got, because stages route by enqueueing a reference rather than
# by moving bytes (src/worker.jl header). A file's stage is a property of the
# queue holding it, so only the pipeline itself can see it. There is no
# directory to poll.
#
# So each stage keeps four counters, all incremented in `worker_loop` (the one
# place every stage's work passes through) and read by `GET /stats`:
#
#   completed / failed   jobs finished, jobs quarantined  → throughput
#   bytes                job bytes processed              → per-stage MiB/s
#   busy_ns              summed handler wall time         → service time
#   blocked_ns           of that, time parked on a full downstream queue
#   in_flight            handlers running right now       → saturation
#
# Rates fall out of a pair of scrapes taken Δt apart:
#
#   throughput  = Δcompleted / Δt
#   utilization = (Δbusy_ns - Δblocked_ns) / (Δt * workers)   ~1.0 ⇒ bottleneck
#
# Utilization is the useful one. Throughput alone can't distinguish a stage that
# is saturated from one that is merely starved by the stage ahead of it: both
# show the same files/s. Utilization separates them: the saturated stage
# sits near 1.0 with its queue backing up, the starved stage sits near 0.
#
# `blocked_ns` is what keeps that true downstream. Stages 1 and 3 apply blocking
# backpressure: when the next queue is full the handler parks and retries rather
# than dropping the file (see ROUTE_ENQUEUE_RETRY_SECONDS). That parked time is
# inside the handler, so counting it as busy would show stage 1 pinned at 1.0
# whenever stage 2 is the real bottleneck, and every stage upstream of the jam would
# look like the jam. Subtracting it leaves utilization meaning "doing its own
# work", and the blocked share becomes its own signal: a stage blocked 90% of the
# time is naming its successor as the bottleneck.
#
# Everything here is monotonic since process start (or since `reset_metrics!`),
# in the Prometheus style: counters, never rates. Rates are the reader's job, so
# a scrape is stateless and two readers can't disturb each other.
#
# Cost is a handful of atomic adds per file, against handlers that spawn
# exiftool: unmeasurable in practice, and the counters are never read on the
# hot path.

# Stage identity lives here, in declaration order, so the report, the JSON and
# the worker wiring can't drift apart: adding stage 5 to the live path means
# adding it here and passing the new `StageStats` to its `worker_loop`.
const STAGE_KEYS = (:classify, :enrich, :triage, :language)
const STAGE_TITLES = (classify = "classify", enrich = "enrich",
                      triage = "triage", language = "language")

"""
Counters for one pipeline stage. All fields are atomic and monotonic: workers
only ever add, readers only ever read, so no lock is needed between them.

`in_flight` is the exception to monotonic. It goes up and down, and is the one
counter that is a *level* rather than a total.
"""
struct StageStats
    completed::Threads.Atomic{Int}
    failed::Threads.Atomic{Int}
    bytes::Threads.Atomic{Int}
    busy_ns::Threads.Atomic{Int}
    blocked_ns::Threads.Atomic{Int}
    in_flight::Threads.Atomic{Int}
end

StageStats() = StageStats(Threads.Atomic{Int}(0), Threads.Atomic{Int}(0),
                          Threads.Atomic{Int}(0), Threads.Atomic{Int}(0),
                          Threads.Atomic{Int}(0), Threads.Atomic{Int}(0))

"""
Counters for HTTP intake, which has no worker loop to hang them off.

`files` counts what was actually spooled *and* queued, the same thing the 202
body reports as `accepted`, so it lines up with stage 1's arrivals. Files
dropped because the queue was full are counted separately in `rejected`, since a
run where intake outruns the pipeline should be visible as backpressure rather
than as slow intake.
"""
struct IntakeStats
    requests::Threads.Atomic{Int}
    files::Threads.Atomic{Int}
    bytes::Threads.Atomic{Int}
    rejected::Threads.Atomic{Int}
end

IntakeStats() = IntakeStats(Threads.Atomic{Int}(0), Threads.Atomic{Int}(0),
                            Threads.Atomic{Int}(0), Threads.Atomic{Int}(0))

"""
Every counter in the process, plus the wall-clock origin the totals are measured
from.

`since` matters to a reader computing rates from a single scrape: without it,
"1000 files completed" has no denominator. Readers that scrape twice should use
their own Δt instead, which excludes the time before they started watching.
"""
struct Metrics
    intake::IntakeStats
    stages::NamedTuple{STAGE_KEYS,NTuple{4,StageStats}}
    since::Base.RefValue{Float64}
end

Metrics() = Metrics(IntakeStats(),
                    NamedTuple{STAGE_KEYS}(ntuple(_ -> StageStats(), 4)),
                    Ref(time()))

# Process-global, like the logger: workers on every thread add to it and the
# HTTP layer reads it, with no way to thread a handle through both paths that
# wouldn't just be this by another name.
const METRICS = Metrics()

"""
    reset_metrics!()

Zero every counter and restart the measurement window. For tests, and for a
benchmark that wants totals covering only its own run, though a harness that
scrapes before and after can subtract instead, which is safer against a server
that is also serving someone else.
"""
function reset_metrics!(m::Metrics = METRICS)
    for a in (m.intake.requests, m.intake.files, m.intake.bytes, m.intake.rejected)
        a[] = 0
    end
    for s in m.stages
        s.completed[] = 0
        s.failed[] = 0
        s.bytes[] = 0
        s.busy_ns[] = 0
        s.blocked_ns[] = 0
        s.in_flight[] = 0
    end
    m.since[] = time()
    return nothing
end

"Record one finished job: `ok` distinguishes a completion from a quarantine."
function record_job!(s::StageStats, ok::Bool, bytes::Int, elapsed_ns::Int)
    Threads.atomic_add!(ok ? s.completed : s.failed, 1)
    Threads.atomic_add!(s.bytes, bytes)
    Threads.atomic_add!(s.busy_ns, elapsed_ns)
    return nothing
end

"""
    enqueue_blocking!(queue, job, stats; retry_seconds) -> nothing

Hand `job` to a downstream queue, parking and retrying until it fits, and charge
the parked time to `stats.blocked_ns`.

This is the routing half of stages 1 and 3: a classified file is never dropped,
so a full downstream queue means waiting, not failing. Wrapping the retry loop
here, rather than repeating `while !enqueue! sleep end` at each call site, is
what makes that wait measurable at all, and keeps the three routing paths from
drifting into three different backoff behaviours.

`stats` is `nothing` for callers that are not a pipeline stage. Startup
recovery (`recover_dir!`) uses the same never-drop retry, but it runs outside any
handler, so it has no `busy_ns` to charge against. Billing its wait to a stage's
`blocked_ns` would make `/stats` utilization, `(Δbusy_ns - Δblocked_ns) / …`,
go *negative* after a long recovery. Skipping the charge keeps `blocked_ns`
meaning exactly what its docs say: time a handler spent parked on its successor.
"""
function enqueue_blocking!(queue::JobQueue, job::Job, stats::Union{StageStats,Nothing};
                           retry_seconds::Real)
    enqueue!(queue, job) && return nothing
    t0 = time_ns()
    while !enqueue!(queue, job)
        sleep(retry_seconds)
    end
    stats === nothing || Threads.atomic_add!(stats.blocked_ns, Int(time_ns() - t0))
    return nothing
end

# --- snapshot ---------------------------------------------------------------

"""
    stats_snapshot(cfg, queues, m = METRICS) -> NamedTuple

Read every counter into a plain value tree for `GET /stats`.

`queues` is a NamedTuple of `JobQueue`s keyed by `STAGE_KEYS`, supplying each
stage's live depth and capacity. The counters above are about work *done*, and
a stage's depth is what says whether work is piling up in front of it.

The read is not atomic across stages: counters keep moving while we walk them,
so a snapshot can show a job counted as complete by stage 1 but not yet arrived
at stage 2. Over a benchmark window that skew is a job or two and does not move
a rate; a consistent snapshot would mean stopping the pipeline to read it.
"""
function stats_snapshot(cfg::Config, queues, m::Metrics = METRICS)
    workers = (classify = cfg.worker_count, enrich = cfg.known_worker_count,
               triage = cfg.unknown_worker_count, language = cfg.text_worker_count)
    now = time()
    stages = map(STAGE_KEYS, ntuple(identity, 4)) do key, i
        s, q = m.stages[key], queues[key]
        (; stage = i,
           name = STAGE_TITLES[key],
           workers = workers[key],
           queue_depth = length(q),
           queue_capacity = capacity(q),
           completed = s.completed[],
           failed = s.failed[],
           bytes = s.bytes[],
           busy_seconds = s.busy_ns[] / 1e9,
           blocked_seconds = s.blocked_ns[] / 1e9,
           in_flight = s.in_flight[])
    end
    return (; now,
              since = m.since[],
              uptime_seconds = now - m.since[],
              intake = (; requests = m.intake.requests[],
                          files = m.intake.files[],
                          bytes = m.intake.bytes[],
                          rejected = m.intake.rejected[]),
              stages = collect(stages))
end