file-server/src/FileServer.jl

module FileServer

using Logging
using Random
using UUIDs
using HTTP
using JSON3
using Oxygen
using Lux
using JLD2
using Languages

include("multipart.jl") # streaming multipart reader (defines UPLOAD_CHUNK_BYTES, used by config.jl)
include("config.jl")
include("job.jl")
include("queue.jl")
include("stats.jl")     # per-stage counters behind GET /stats (needs Config/Job/JobQueue)
include("spool.jl")
include("model.jl")     # build_model() + read_features(); shared with bin/train.jl
include("classify.jl")  # Classifier + load_classifier/classify (needs model.jl)
include("metadata.jl")  # exiftool extraction + sidecar enrichment (stage 2)
include("content.jl")   # binary-vs-text triage for unknown files (stage 3)
include("language.jl")  # natural + programming language enrichment for text (stage 4)
include("cluster.jl")   # unknown-format discovery by header clustering (stage 5, science)
include("catalog.jl")   # durable single-owner format catalog (stage 5, phase B; needs cluster.jl + metadata.jl fsync)
include("worker.jl")

# Globals the HTTP handlers read at request time. Set once in `run`, before the
# server starts accepting connections. Declared after the includes above so the
# `Config`/`ChannelQueue` types exist.
const CONFIG = Ref{Config}()
const QUEUE  = Ref{ChannelQueue}()          # stage-1 (classification) queue; HTTP intake enqueues here
const KNOWN_QUEUE = Ref{ChannelQueue}()     # stage-2 (enrichment) queue; stage-1 workers enqueue here
const UNKNOWN_QUEUE = Ref{ChannelQueue}()   # stage-3 (content triage) queue; stage-1 workers enqueue here
const TEXT_QUEUE = Ref{ChannelQueue}()      # stage-4 (language enrichment) queue; stage-3 workers enqueue here
const CLASSIFIER = Ref{Classifier}()  # loaded once at startup, shared read-only across workers
const DETECTOR = Ref{LanguageDetector}()  # natural-language detector; built once at startup, shared read-only

include("server.jl")   # registers routes (references CONFIG/QUEUE at call time)

export run

# When stderr is a live terminal Julia flushes each write, but when it's
# redirected to a file or pipe (a log file, `tee`, journald, a container log
# driver) Julia block-buffers it — so a long-running server's logs sit unseen in
# the buffer until it fills or the process exits, making it look like nothing is
# happening. This wrapper delegates to a normal logger and flushes after every
# message so output appears immediately wherever stderr is pointed.
struct FlushLogger{L<:AbstractLogger} <: AbstractLogger
    inner::L
end
Logging.min_enabled_level(f::FlushLogger) = Logging.min_enabled_level(f.inner)
Logging.shouldlog(f::FlushLogger, args...) = Logging.shouldlog(f.inner, args...)
Logging.catch_exceptions(f::FlushLogger) = Logging.catch_exceptions(f.inner)
function Logging.handle_message(f::FlushLogger, args...; kwargs...)
    Logging.handle_message(f.inner, args...; kwargs...)
    flush(f.inner.stream)
    return nothing
end

"""
    run(; overrides...)

Start the file server: build config, ensure directories, recover any leftover
spooled files, spawn the worker pool, then serve HTTP until interrupted
(Ctrl-C / SIGINT or SIGTERM). On shutdown it stops accepting uploads, drains the
queue, waits for workers to finish in-flight files, and exits cleanly.

Keyword `overrides` (e.g. `port=9000`) take precedence over environment vars.
"""
function run(; overrides...)
    # By default a Julia script exits immediately on SIGINT (Ctrl-C). Disable
    # that so the interrupt surfaces as a catchable InterruptException, letting
    # us drain the queue gracefully below.
    Base.exit_on_sigint(false)

    # Flush every log message so output is visible in real time even when stderr
    # is redirected to a file/pipe (see FlushLogger). Set before anything logs.
    global_logger(FlushLogger(ConsoleLogger(stderr)))

    cfg = config_from_env(; overrides...)
    ensure_dirs(cfg)

    # All pools draw from the same OS threads. Warn on the *combined* size (still
    # allowed): oversubscription just means tasks share threads, not a failure.
    total_workers = cfg.worker_count + cfg.known_worker_count + cfg.unknown_worker_count + cfg.text_worker_count
    if total_workers > Threads.nthreads()
        @warn "combined worker count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" classify_workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count text_workers=cfg.text_worker_count total=total_workers nthreads=Threads.nthreads()
    end

    # exiftool is a hard prerequisite for stage-2 enrichment. Fail fast at
    # startup rather than discover it missing on the first known file.
    assert_exiftool()

    # github-linguist powers stage-4 *programming*-language detection, but it's
    # best-effort (natural-language enrichment stands on its own), so a missing
    # binary is a warning, not a fatal error — per-file lookups degrade to none.
    linguist_available() || @warn "github-linguist not found on PATH; stage-4 text files will have no programming language (install it to enable)"

    queue         = ChannelQueue(cfg.queue_capacity)
    known_queue   = ChannelQueue(cfg.known_queue_capacity)
    unknown_queue = ChannelQueue(cfg.unknown_queue_capacity)
    text_queue    = ChannelQueue(cfg.text_queue_capacity)
    CONFIG[]        = cfg
    QUEUE[]         = queue
    KNOWN_QUEUE[]   = known_queue
    UNKNOWN_QUEUE[] = unknown_queue
    TEXT_QUEUE[]    = text_queue

    # Load the classifier before serving. Fail fast: a server that silently
    # doesn't classify is a worse surprise than a clear startup error.
    CLASSIFIER[] = load_classifier(cfg.model_path)
    @info "loaded classifier" path=cfg.model_path

    # Build the natural-language detector once (it loads the whatlang n-gram
    # model) and share it read-only across the stage-4 pool, like the classifier.
    DETECTOR[] = LanguageDetector()
    @info "loaded language detector"

    # Stage-aware recovery: re-drive each stage's leftovers onto its own queue so
    # files resume where they were, not from scratch. spool/ → stage-1,
    # known/ → stage-2, unknown/ → stage-3.
    recovered         = recover_dir!(cfg.spool_dir, queue)
    recovered_known   = recover_dir!(cfg.known_dir, known_queue)
    recovered_unknown = recover_dir!(cfg.unknown_dir, unknown_queue)
    recovered_text    = recover_dir!(cfg.text_dir, text_queue)
    @info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count text_workers=cfg.text_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity text_capacity=cfg.text_queue_capacity recovered=recovered recovered_known=recovered_known recovered_unknown=recovered_unknown recovered_text=recovered_text

    # Zero the counters here, not at module load: `since` should mean "serving
    # since", so a scrape's totals cover the run, not the minutes spent loading
    # the classifier. Nothing has been processed yet — recovery only enqueues.
    reset_metrics!()

    # Each pool gets its stage's counters (src/stats.jl); `worker_loop` records
    # into them, `GET /stats` reads them out. Stage 1 and 3 also hand theirs to
    # their handler, which charges time parked on a full downstream queue to
    # `blocked_ns` so it isn't mistaken for work.
    st = METRICS.stages
    workers = [Threads.@spawn worker_loop(i, cfg, queue,
                   (job, c, wid) -> handle_classify_job(job, c, wid, known_queue, unknown_queue, st.classify),
                   st.classify)
               for i in 1:cfg.worker_count]
    known_workers = [Threads.@spawn worker_loop(i, cfg, known_queue, handle_known_job, st.enrich)
                     for i in 1:cfg.known_worker_count]
    unknown_workers = [Threads.@spawn worker_loop(i, cfg, unknown_queue,
                   (job, c, wid) -> handle_unknown_job(job, c, wid, text_queue, st.triage),
                   st.triage)
                       for i in 1:cfg.unknown_worker_count]
    text_workers = [Threads.@spawn worker_loop(i, cfg, text_queue,
                   (job, c, wid) -> handle_text_job(job, c, wid, DETECTOR[]),
                   st.language)
                    for i in 1:cfg.text_worker_count]

    register_routes()
    # `handler` replaces Oxygen's root stream handler so POST /upload can read its
    # body incrementally instead of having it buffered into memory first; every
    # other route still goes through Oxygen (see `root_stream_handler`).
    serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false,
            handler = root_stream_handler)

    # Idempotent graceful drain: stop accepting uploads, let workers finish the
    # buffered jobs, then exit. Called from two places:
    #   * the `finally` below, for SIGINT (Ctrl-C) and normal return, and
    #   * an `atexit` hook, for SIGTERM (systemd/Docker/k8s `stop`).
    # We can't intercept SIGTERM directly — Julia blocks it on worker threads and
    # handles it in its own runtime, so a user signal() handler never fires. But
    # Julia's SIGTERM path runs `atexit` hooks, which gives us a reliable seam.
    drained = Threads.Atomic{Bool}(false)
    function drain()
        Threads.atomic_xchg!(drained, true) && return   # run at most once
        @info "draining queues and stopping workers"
        terminate()                    # 1. stop accepting new HTTP requests
        close!(queue)                  # 2. no new classify jobs; stage-1 drains buffered
        foreach(wait, workers)         # 3. wait out stage-1 — the ONLY producer of BOTH the
                                       #    known and unknown queues — so nothing else enqueues
        close!(known_queue)            # 4. now safe to close the queues stage-1 fed
        close!(unknown_queue)
        foreach(wait, known_workers)   # 5. wait out stage-2 (terminal) and stage-3 — stage-3 is
        foreach(wait, unknown_workers) #    the ONLY producer of the text queue
        close!(text_queue)             # 6. now safe to close the queue stage-3 fed
        foreach(wait, text_workers)    # 7. wait out stage-4
        @info "shutdown complete"
    end
    atexit(drain)

    try
        while true
            sleep(0.5)            # interruptible; SIGINT throws in here
        end
    catch e
        e isa InterruptException || rethrow(e)
        @info "shutdown requested (SIGINT)"
    finally
        drain()
    end

    return nothing
end

end # module