// Who is using the network: routers export flows here, in NetFlow v5, v9 or IPFIX, and a collector
// on a thread decodes them, each exporter's templates remembered. A route drains what came and
// shows the top talkers; lost packets, data that came before its template, and damaged packets
// are counted, not fatal.
//
//   NETFLOW_PORT=2055 quillo run netflow.qwe      then point a router at this machine, UDP 2055,
//   and open http://localhost:8080/flows and /stats
//
// An `every` tick works as well as a route for draining: `every(1000, () => keep(netflow_receive("edge", 1000, 0)?))`.

use "netflow.qwer"

let PORT = int(env("NETFLOW_PORT", "2055"))
let EXPORTER = env("NETFLOW_EXPORTER", "")
netflow_collect("edge", PORT, { host: "127.0.0.1" })

let seen = []

fn gather(want) {
    if EXPORTER != "" { udp.send("127.0.0.1", int(EXPORTER), "go") }
    let tries = 0
    while len(seen) < want && tries < 30 {
        seen = seen + netflow_receive("edge", 1000, 100)?
        tries = tries + 1
    }
    seen
}

route GET /v5 {
    let f = (gather(9) | filter((x) => x.version == 5))
    let a = f[0]
    { count: len(f), first: [a.src_addr, a.src_port, a.dst_addr, a.dst_port, a.protocol_name, a.flags, a.packets, a.bytes, a.src_as, a.src_mask], wall_clock: a.start_ms, lasted: a.duration_ms, exporter: a.exporter }
}

route GET /v9 {
    let f = (gather(9) | filter((x) => x.version == 9 && !x.options))
    { count: len(f), flows: f | map((x) => [x.src_addr, x.dst_port, x.protocol_name, x.bytes]), sampling: (seen | filter((x) => x.options))[0] }
}

route GET /ipfix {
    let f = (gather(9) | filter((x) => x.version == 10))
    { count: len(f), v6: [f[0].src_addr, f[0].dst_addr, f[0].start_ms, f[0].duration_ms, f[0].bytes], vendor: f[1].e9_12345, text_field: f[1].field_351 }
}

route GET /stats {
    gather(9)
    sleep(200)
    let s = netflow_stats("edge")?
    let e = s.exporters[0]
    {
        packets_ok: s.packets >= 7, flows: s.flows, errors: s.errors, last_error: s.last_error, lost: s.lost, held: s.held_sets, versions: s.versions,
        exporter: e.address, templates: netflow_collector_templates("edge", e.address)? | map((t) => [t.id, t.options, len(t.fields)]),
    }
}

route GET /top = netflow_top(gather(9), "src_addr", "bytes", 2)

// One packet decoded by hand: a v5 header with no flows, a wrong version, and a cut-off one.
route GET /decode = {
    empty: netflow_decode("000500000000000000000000000000000000000000000000", "test")?,
    wrong: netflow_decode("0007000000000000", "test").error,
    cut: netflow_decode("00050002000000000000000000000000000000000000", "test").error,
    bad_hex: netflow_decode("xyz", "test").error,
    too_many: netflow_decode("0005" + "001f" + repeat("00", 20), "test").error,
    no_collector: netflow_receive("nobody", 1, 0).error,
    stopped: netflow_stop("nobody")?,
}