Initial file-ingestion service

REST endpoint (Oxygen.jl POST /upload, multipart) that spools uploaded
files to disk, enqueues lightweight references onto a bounded thread-safe
work queue, and hands off immediately (202 + job IDs; 503 when full). A
configurable pool of worker threads pulls jobs off the queue, logs the
received filename (placeholder for real processing), and moves files to
done/ on success or failed/ on error.

- Queue behind an enqueue!/dequeue!/close! seam for a future RabbitMQ swap
- Startup recovery: re-enqueues leftover files in spool/
- Graceful drain on SIGINT and SIGTERM (via atexit)
- Env-var config; filenames sanitized + UUID-prefixed on disk

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/FileServer.jl

@@ -0,0 +1,92 @@
+module FileServer
+
+using Logging
+using UUIDs
+using HTTP
+using JSON3
+using Oxygen
+
+include("config.jl")
+include("job.jl")
+include("queue.jl")
+include("spool.jl")
+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}()
+
+include("server.jl")   # registers routes (references CONFIG/QUEUE at call time)
+
+export run
+
+"""
+    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)
+
+    cfg = config_from_env(; overrides...)
+    ensure_dirs(cfg)
+
+    if cfg.worker_count > Threads.nthreads()
+        @warn "worker_count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" worker_count=cfg.worker_count nthreads=Threads.nthreads()
+    end
+
+    queue = ChannelQueue(cfg.queue_capacity)
+    CONFIG[] = cfg
+    QUEUE[]  = queue
+
+    recovered = recover_spool!(cfg, queue)
+    @info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count capacity=cfg.queue_capacity recovered=recovered
+
+    workers = [Threads.@spawn worker_loop(i, cfg, queue) for i in 1:cfg.worker_count]
+
+    register_routes()
+    serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false)
+
+    # 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 queue and stopping workers"
+        terminate()               # stop accepting new HTTP requests
+        close!(queue)             # let workers drain buffered jobs, then exit
+        foreach(wait, workers)
+        @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

src/config.jl

@@ -0,0 +1,46 @@
+# Runtime configuration. Built once at startup, then treated as immutable.
+# Every knob has a default so the service runs with zero configuration, and
+# every knob can be overridden by an environment variable (see `config_from_env`).
+
+Base.@kwdef struct Config
+    host::String        = "127.0.0.1"
+    port::Int           = 8080
+    worker_count::Int   = Threads.nthreads()
+    queue_capacity::Int = 1000
+    spool_dir::String   = "data/spool"   # files land here on intake (pending)
+    done_dir::String    = "data/done"    # files move here after successful processing
+    failed_dir::String  = "data/failed"  # files move here if a worker throws
+end
+
+"""
+    config_from_env(; overrides...)
+
+Build a `Config` from environment variables, falling back to the struct
+defaults. Any keyword `overrides` win over the environment (useful for tests
+and for `FileServer.run(; port=...)`).
+
+Recognised variables:
+  FS_HOST, FS_PORT, FS_WORKERS, FS_QUEUE_CAPACITY,
+  FS_SPOOL_DIR, FS_DONE_DIR, FS_FAILED_DIR
+"""
+function config_from_env(; host=nothing, port=nothing, worker_count=nothing,
+                          queue_capacity=nothing, spool_dir=nothing,
+                          done_dir=nothing, failed_dir=nothing)
+    Config(
+        host           = something(host,           get(ENV, "FS_HOST", "127.0.0.1")),
+        port           = something(port,           parse(Int, get(ENV, "FS_PORT", "8080"))),
+        worker_count   = something(worker_count,   parse(Int, get(ENV, "FS_WORKERS", string(Threads.nthreads())))),
+        queue_capacity = something(queue_capacity, parse(Int, get(ENV, "FS_QUEUE_CAPACITY", "1000"))),
+        spool_dir      = something(spool_dir,      get(ENV, "FS_SPOOL_DIR", "data/spool")),
+        done_dir       = something(done_dir,       get(ENV, "FS_DONE_DIR", "data/done")),
+        failed_dir     = something(failed_dir,     get(ENV, "FS_FAILED_DIR", "data/failed")),
+    )
+end
+
+"Create the spool/done/failed directories if they don't already exist."
+function ensure_dirs(cfg::Config)
+    for d in (cfg.spool_dir, cfg.done_dir, cfg.failed_dir)
+        mkpath(d)
+    end
+    return nothing
+end

src/job.jl

@@ -0,0 +1,13 @@
+# A unit of work on the queue. Deliberately lightweight: the file *bytes* live
+# on disk in the spool directory, and only this small reference travels through
+# the queue. This is what keeps intake fast and memory flat regardless of file
+# size, and it's the shape you'd publish to RabbitMQ later (the "claim check"
+# pattern — enqueue a reference, not the payload).
+
+struct Job
+    id::String            # server-minted UUID; also the on-disk filename prefix
+    original_name::String # client-supplied name, kept for display/logging only
+    path::String          # absolute-ish path to the spooled file
+    size::Int             # bytes
+    received_at::Float64  # time() at intake
+end

src/queue.jl

@@ -0,0 +1,92 @@
+# The queue seam.
+#
+# The rest of the app only ever calls `enqueue!`, `dequeue!`, and `close!`.
+# Today those are backed by an in-process, bounded, thread-safe buffer
+# (the Go-channel / Julia-`Channel` model). To move to RabbitMQ (or any broker)
+# later, implement a new `JobQueue` subtype with these three methods and swap
+# the construction in `run` — no HTTP handler or worker code needs to change.
+
+abstract type JobQueue end
+
+"""
+    enqueue!(q, job) -> Bool
+
+Non-blocking. Returns `true` if the job was accepted, `false` if the queue is
+full (the HTTP layer turns `false` into a 503) or closed. Never blocks the
+calling request thread.
+"""
+function enqueue! end
+
+"""
+    dequeue!(q) -> Union{Job,Nothing}
+
+Blocks until a job is available and returns it. Returns `nothing` only when the
+queue has been closed *and* fully drained — the signal for a worker to exit.
+"""
+function dequeue! end
+
+"""
+    close!(q)
+
+Mark the queue closed and wake all waiting workers. Buffered jobs are still
+handed out (drain-then-exit); no new jobs are accepted.
+"""
+function close! end
+
+# --- In-process bounded implementation --------------------------------------
+
+mutable struct ChannelQueue <: JobQueue
+    const buffer::Vector{Job}
+    const capacity::Int
+    const cond::Threads.Condition   # its internal lock guards `buffer` + `closed`
+    closed::Bool
+end
+
+ChannelQueue(capacity::Integer) =
+    ChannelQueue(Job[], Int(capacity), Threads.Condition(), false)
+
+function enqueue!(q::ChannelQueue, job::Job)::Bool
+    lock(q.cond)
+    try
+        (q.closed || length(q.buffer) >= q.capacity) && return false
+        push!(q.buffer, job)
+        notify(q.cond)   # wake a waiting worker (notify wakes all by default)
+        return true
+    finally
+        unlock(q.cond)
+    end
+end
+
+function dequeue!(q::ChannelQueue)::Union{Job,Nothing}
+    lock(q.cond)
+    try
+        while isempty(q.buffer)
+            q.closed && return nothing   # closed and drained → tell worker to stop
+            wait(q.cond)                 # releases lock while parked
+        end
+        return popfirst!(q.buffer)
+    finally
+        unlock(q.cond)
+    end
+end
+
+function close!(q::ChannelQueue)
+    lock(q.cond)
+    try
+        q.closed = true
+        notify(q.cond)   # wake every parked worker so they can drain/exit
+    finally
+        unlock(q.cond)
+    end
+    return nothing
+end
+
+"Number of jobs currently buffered (for logging/introspection)."
+function Base.length(q::ChannelQueue)
+    lock(q.cond)
+    try
+        return length(q.buffer)
+    finally
+        unlock(q.cond)
+    end
+end

src/server.jl

@@ -0,0 +1,62 @@
+# HTTP layer: a single multipart upload endpoint plus a health check.
+#
+# The handler's whole job is to get files onto the queue fast and get out of the
+# way: spool each uploaded file to disk, enqueue a reference, respond 202. It
+# never does real processing — that's the workers' job.
+#
+# NOTE: routes are registered at runtime via `register_routes()` (called from
+# `run`), NOT with top-level macros. In a precompiled package, top-level
+# `@get`/`@post` would execute during precompilation and be lost before serving.
+
+jsonresp(status::Int, data) =
+    HTTP.Response(status, ["Content-Type" => "application/json"], JSON3.write(data))
+
+function health_handler(_::HTTP.Request)
+    return jsonresp(200, (; status = "ok"))
+end
+
+function upload_handler(req::HTTP.Request)
+    cfg   = CONFIG[]
+    queue = QUEUE[]
+
+    parts = try
+        HTTP.parse_multipart_form(req)
+    catch
+        nothing
+    end
+    parts === nothing &&
+        return jsonresp(400, (; error = "expected multipart/form-data"))
+
+    files = filter(p -> p.filename !== nothing && !isempty(p.filename), parts)
+    isempty(files) &&
+        return jsonresp(400, (; error = "no files found in request"))
+
+    accepted = NamedTuple{(:id, :name),Tuple{String,String}}[]
+    for p in files
+        bytes = read(p.data)
+
+        job = try
+            spool_file(cfg, p.filename, bytes)
+        catch e
+            @error "spool failed" name=p.filename exception=(e, catch_backtrace())
+            return jsonresp(500, (; error = "failed to store file", accepted))
+        end
+
+        if !enqueue!(queue, job)
+            rm(job.path; force = true)   # never queued → don't leave it in spool
+            return jsonresp(503, (; error = "queue full, retry later", accepted))
+        end
+
+        @info "accepted" id=job.id name=job.original_name size=job.size
+        push!(accepted, (; id = job.id, name = job.original_name))
+    end
+
+    return jsonresp(202, (; accepted))
+end
+
+"Register HTTP routes on the Oxygen instance. Must run at runtime, before serve."
+function register_routes()
+    @get("/health", health_handler)
+    @post("/upload", upload_handler)
+    return nothing
+end

src/spool.jl

@@ -0,0 +1,69 @@
+# Disk lifecycle: sanitize names, spool bytes to disk, move processed files,
+# and recover leftover files on startup.
+
+const MAX_NAME_LEN = 100
+
+"""
+    sanitize_filename(name) -> String
+
+Turn an untrusted, client-supplied filename into something safe to place in a
+path. Strips directory components, replaces anything outside a conservative
+charset, removes leading dots (so `..` and dotfiles can't sneak through), caps
+the length, and falls back to `"unnamed"` if nothing usable remains.
+"""
+function sanitize_filename(name::AbstractString)::String
+    base = basename(String(name))                     # drop any path components
+    base = replace(base, r"[^A-Za-z0-9._-]" => "_")   # ASCII-only safe charset
+    base = lstrip(base, '.')                          # kill "..", ".hidden", etc.
+    isempty(base) && (base = "unnamed")
+    return first(base, MAX_NAME_LEN)
+end
+
+"Write `bytes` to the spool dir under `<uuid>-<sanitized>` and return the Job."
+function spool_file(cfg::Config, original_name::AbstractString, bytes::Vector{UInt8})::Job
+    id   = string(uuid4())
+    safe = sanitize_filename(original_name)
+    path = joinpath(cfg.spool_dir, string(id, "-", safe))
+    open(path, "w") do io
+        write(io, bytes)
+    end
+    return Job(id, String(original_name), path, length(bytes), time())
+end
+
+"Move a spooled file into `dir` (done/ or failed/), returning the destination."
+function move_to(dir::AbstractString, job::Job)::String
+    dest = joinpath(dir, basename(job.path))
+    mv(job.path, dest; force=true)
+    return dest
+end
+
+# The UUID string produced by `uuid4()` is always 36 chars, followed by '-',
+# then the sanitized name. That fixed width lets us split reliably on recovery.
+const UUID_LEN = 36
+
+"""
+    recover_spool!(cfg, queue) -> Int
+
+Re-enqueue any files already sitting in the spool directory (left by a crash,
+a hard shutdown, or an intake that never got processed). This is the payoff of
+spooling to disk: a restart resumes work instead of stranding it. Returns the
+number of files recovered.
+"""
+function recover_spool!(cfg::Config, queue::JobQueue)::Int
+    n = 0
+    for path in sort(readdir(cfg.spool_dir; join=true))
+        isfile(path) || continue
+        fname = basename(path)
+        if length(fname) > UUID_LEN + 1
+            id   = fname[1:UUID_LEN]
+            name = fname[(UUID_LEN + 2):end]   # skip the '-'
+        else
+            id   = string(uuid4())             # unexpected name; give it an id
+            name = fname
+        end
+        job = Job(id, name, path, filesize(path), time())
+        enqueue!(queue, job) || @warn "queue full during recovery; leaving file for next start" path
+        n += 1
+    end
+    return n
+end

src/worker.jl

@@ -0,0 +1,46 @@
+# Worker task: pull jobs off the queue and process them. One of these runs per
+# configured worker, each as its own `Threads.@spawn`'d task.
+
+"""
+    handle_job(job, cfg, worker_id)
+
+Do the work for a single job, then move the file to `done/`.
+
+For now the "work" is just logging the received filename to prove the flow —
+this is the seam where real heavy-lifting will go later.
+"""
+function handle_job(job::Job, cfg::Config, worker_id::Int)
+    # --- placeholder for real heavy-lifting work -------------------------
+    @info "received file" worker=worker_id id=job.id name=job.original_name size=job.size
+    # --------------------------------------------------------------------
+    dest = move_to(cfg.done_dir, job)
+    @info "completed" worker=worker_id id=job.id dest=dest
+    return nothing
+end
+
+"""
+    worker_loop(worker_id, cfg, queue)
+
+Consume jobs until the queue is closed and drained. A failure on one job is
+logged and the file is quarantined in `failed/` — it must never kill the
+worker, or the pool would silently shrink.
+"""
+function worker_loop(worker_id::Int, cfg::Config, queue::JobQueue)
+    @info "worker started" worker=worker_id
+    while true
+        job = dequeue!(queue)
+        job === nothing && break   # queue closed and drained → exit
+        try
+            handle_job(job, cfg, worker_id)
+        catch e
+            @error "processing failed" worker=worker_id id=job.id name=job.original_name exception=(e, catch_backtrace())
+            try
+                move_to(cfg.failed_dir, job)
+            catch e2
+                @error "could not quarantine failed file" worker=worker_id id=job.id path=job.path exception=(e2, catch_backtrace())
+            end
+        end
+    end
+    @info "worker stopped" worker=worker_id
+    return nothing
+end