// Greenhouse readings in InfluxDB: a line of line protocol built and read back, readings written to
// InfluxDB 2 and the latest read with Flux, request timings buffered and written in one gzipped
// batch, and the same readings in InfluxDB 3, read with SQL and a parameter.
//
//   INFLUX_URL=http://127.0.0.1:8086 INFLUX_TOKEN=... INFLUX_ORG=shop INFLUX_BUCKET=metrics quillo run influx.qwe      then open http://localhost:8080/record?temp=21.5
//   INFLUX3_URL=http://127.0.0.1:8181 INFLUX3_TOKEN=apiv3_... INFLUX3_DATABASE=shop      for /v3
//
// Needs InfluxDB 2 (influxd, then influx setup) and, for /v3, InfluxDB 3 (influxdb3 serve, then
// influxdb3 create token --admin).

use "influx.qwer"

let servers = [
    influx_connect("v2", env("INFLUX_URL", "http://127.0.0.1:8086"), {
        token: env("INFLUX_TOKEN", ""), org: env("INFLUX_ORG", "shop"), bucket: env("INFLUX_BUCKET", "metrics"), precision: "ms",
    }),
    influx_connect("v3", env("INFLUX3_URL", "http://127.0.0.1:8181"), { token: env("INFLUX3_TOKEN", ""), database: env("INFLUX3_DATABASE", "shop") }),
]

// What influx_add keeps goes out once a second (or as soon as 5000 lines wait).
every(1000, () => influx_flush("v2"))

route GET /line {
    let line = influx_line("cpu load", { host: "web 1", region: "eu,west", rack: "" }, { usage: 12.5, cores: 8, note: "said \"hi\"", ok: true }, 1761800000000)?
    { line, back: influx_parse(line)?[0] }
}

// Two readings written, then the latest of each field read back with Flux.
route GET /record ? { sensor: str = "north", temp: float = 21.5 } {
    let t = now()
    let written = influx_write("v2", [
        { measurement: "climate", tags: { sensor }, fields: { temp, humidity: 61 }, time: t - 1000 },
        { measurement: "climate", tags: { sensor }, fields: { temp: temp + 0.25, humidity: 60 }, time: t },
    ])?
    let rows = influx_query("v2", `from(bucket: "${env("INFLUX_BUCKET", "metrics")}")
        |> range(start: -1h)
        |> filter(fn: (r) => r._measurement == "climate" and r.sensor == "${sensor}")
        |> last()`, {})?
    { written: written.lines, latest: rows | map((r) => { field: r._field, value: r._value, sensor: r.sensor }) | sort_by((r) => r.field) }
}

// 2000 timings buffered, then written at once: the body is gzipped.
route GET /batch {
    let t = now()
    for i in range(0, 2000) {
        influx_add("v2", { measurement: "http", tags: { route: i % 2 == 0 ? "/home" : "/cart" }, fields: { ms: 3 + i % 17 }, time: t - i })?
    }
    let sent = influx_flush("v2")?
    let counted = influx_query("v2", `from(bucket: "${env("INFLUX_BUCKET", "metrics")}") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "http") |> count()`, {})?
    { sent, counted: counted | map((r) => { route: r.route, n: r._value }) | sort_by((r) => r.route) }
}

route GET /v3 ? { sensor: str = "south" } {
    let t = now()
    influx_write("v3", `climate,sensor=${sensor} temp=19.5,humidity=55i ${t - 1000}\nclimate,sensor=${sensor} temp=19.75,humidity=54i ${t}`)?
    let rows = influx_query("v3", "SELECT sensor, temp, humidity FROM climate WHERE sensor = $sensor ORDER BY time LIMIT 10", { params: { sensor } })?
    { rows, refused: influx_write("v3", "climate,sensor=x temp=oops").error }
}