// Summarises a web server's access log: how many requests, how many failed,
// and the hits and bytes sent per path, busiest first.
//
//     nikaia run -- tests/five-requests.in/access.log
//     5 requests, 1 failed
//     /index.html 2 10240
//     /api/order 1 512
//     …
//
// Each line of the log is `<ip> <method> <path> <status> <bytes>`, such as
// `203.0.113.7 GET /index.html 200 5120`.
//
// Shows: a `grammar` for lines of several fields, a parallel `par_fold`, a
// parse error caught and printed, and a two-key sort done as two stable sorts.
//
// A package: `nikaia run` in this directory runs it, and `nikaia test` checks
// what it prints against `tests/` (ADR-269 D13).

use std::fs
use std::cli
use std::collections

grammar Log {

    // Up to the next blank or the end of the line. Without `frame_end`, a line
    // missing a field would run on into the next line; the compiler rejects
    // a scan like that inside a frame.
    rule FIELD -> String = s:until(" " | frame_end) -> { s }

    // Exactly three digits: a status of "20" is a parse error here.
    rule STATUS -> i32 = n:dec[i32](digit{3}) -> { n }

    rule SIZE -> i64 = n:dec[i64](digit+) -> { n }

    // `@frame` says a request can be found from any offset by scanning to the
    // next newline, so the file can be cut into pieces and parsed in parallel.
    // The rule is uppercase, so *lexical*: no blanks are skipped for it, and
    // each `" "` between fields is written out. `=>` is a commit point: once the
    // first field and its blank are read, the line must be a request.
    @frame(boundary: "\n")
    rule ENTRY -> Hit =
        _ip:FIELD " " => method:FIELD " " path:FIELD " " status:STATUS " " bytes:SIZE frame_end
        -> { Hit { path, status, bytes } }

    // A fold hands each entry to an accumulator instead of collecting them
    // all, as `ENTRY*` would. `par_fold` also says how two partial reports
    // merge, so each core can fold its own piece of the file.
    entry rule file -> Report =
        par_fold(ENTRY, Report, fn(acc, e) { acc.record(e) }, Report::merge)
}

// `path` is a slice of the mapped file, not a copy.
pub struct Hit {
    path: String,
    status: i32,
    bytes: i64,
}

pub struct Counts {
    hits: i64,
    bytes: i64,
}

impl Counts {
    pub fn(bytes: i64) -> Counts sync {
        return Counts { hits: 1, bytes: bytes }
    }

    fn add(ref mut self, bytes: i64) sync {
        self.hits += 1
        self.bytes += bytes
    }

    fn merge(ref mut self, other: ref Counts) sync {
        self.hits += other.hits
        self.bytes += other.bytes
    }
}

pub struct Report {
    paths: collections::HashMap[String, Counts],
    requests: i64,
    failures: i64,
}

impl Report {
    pub fn() -> Report sync {
        return Report { paths: collections::HashMap(), requests: 0, failures: 0 }
    }

    // `sync`: this function cannot pause, and the compiler checks it.
    fn record(ref mut self, e: Hit) sync {
        self.requests += 1
        if e.status >= 400 { self.failures += 1 }
        self.paths
            .entry(e.path)
            .and_modify fn (counts) { counts.add(e.bytes) }
            .or_insert_with fn { Counts(e.bytes) }
    }

    fn merge(ref mut self, other: Report) sync {
        self.requests += other.requests
        self.failures += other.failures
        for (path, counts) in other.paths.drain() {
            self.paths
                .entry(path)
                .and_modify fn (into) { into.merge(ref counts) }
                .or_insert(counts)
        }
    }
}

fn main() throws {
    let path = cli::args().nth(1) ?? "access.log"

    let data = fs::map(ref path, fs::Root::Anywhere) catch {
        eprintln(f"cannot open {path}: {error}")
        return
    }

    // A malformed line is an error that `catch` prints: its line and column
    // in the whole file, and what was expected there. No summary is printed.
    let report = Log::file(data) catch {
        eprintln(f"{path}: {error}")
        return
    }

    println(f"{report.requests} requests, {report.failures} failed")

    // A map keeps no order, so the report chooses one.
    let mut names = report.paths.keys().collect()
    names.sort()

    // One row per path, as a tuple: the name and the two numbers.
    let mut rows = names.map fn(path) {
        // A lookup gives a `Counts?` that may be absent; `??` handles that
        // case, which cannot happen for the map's own keys.
        let counts = report.paths[path] ?? panic(f"{path} was a key a moment ago")
        (path, counts.hits, counts.bytes)
    }

    // `sort_by_key` is stable: rows with the same number of hits keep the name
    // order from the first sort.
    rows.sort_by_key fn(row) { -row.1 }

    let body = rows
        .map fn(row) { f"{row.0} {row.1} {row.2}" }
        .join("\n")

    println(f"{body}")
}