// TaskForge server: users, sessions and the shared project, stored in SQLite (WAL mode).
// Also serves the web client. Meant to sit behind Caddy (or nginx) for HTTPS.
//
//   taskforge-server [--port 8080] [--listen 127.0.0.1] [--db taskforge.db] [--web ./web] [--threads 8] [--demo]
//   taskforge-server [--db taskforge.db] add-user <username> "<Display Name>" [--admin]     (password read from stdin)
//   taskforge-server [--db taskforge.db] set-password <username>                           (password read from stdin)
//   taskforge-server [--db taskforge.db] list-users
//
// Sync model: the project is a set of records (one text line per entity, see src/storage.jai), each with a key
// like "task 42" or "val 42 7". Every write bumps a global revision; clients send their changed records and
// receive everything changed since the revision they last saw. Last writer wins per record.

Config :: struct {
    port:      u16 = 8080;
    listen_on := "127.0.0.1";
    db_path   := "taskforge.db";
    web_dir   := "web";
    threads:   s32 = 8;
    demo:      bool;
}

config: Config;
listen_socket: Socket;

PBKDF2_ITERATIONS :: 310000;
SESSION_DAYS      :: 30;
ID_BLOCK          :: 1000;
MAX_BODY          :: 16 * 1024 * 1024;

main :: () {
    args := get_command_line_arguments();
    command := "serve";
    positional: [..] string;
    admin_flag := false;
    i := 1;
    while i < args.count {
        a := args[i];
        has_value := i + 1 < args.count;
        if a == "--port" && has_value          { config.port = cast(u16) string_to_int(args[i + 1]); i += 2; continue; }
        if a == "--listen" && has_value        { config.listen_on = args[i + 1]; i += 2; continue; }
        if a == "--db" && has_value            { config.db_path = args[i + 1]; i += 2; continue; }
        if a == "--web" && has_value           { config.web_dir = args[i + 1]; i += 2; continue; }
        if a == "--threads" && has_value       { config.threads = cast(s32) string_to_int(args[i + 1]); i += 2; continue; }
        if a == "--demo"                       { config.demo = true; i += 1; continue; }
        if a == "--admin"                      { admin_flag = true; i += 1; continue; }
        if a == "--help" || a == "-h"          { print_usage(); return; }
        if !positional.count && (a == "serve" || a == "add-user" || a == "set-password" || a == "list-users") {
            command = a;
            array_add(*positional, a);
        } else {
            array_add(*positional, a);
        }
        i += 1;
    }

    if !load_sqlite() exit(1);
    db, ok := db_open(config.db_path);
    if !ok exit(1);
    if !migrate(db) exit(1);

    if command == {
        case "add-user";
            if positional.count < 3 { print_usage(); exit(1); }
            password := read_password();
            error := add_user(db, positional[1], positional[2], password, admin_flag);
            if error { log_error("%", error); exit(1); }
            print("Added user %.\n", positional[1]);
        case "set-password";
            if positional.count < 2 { print_usage(); exit(1); }
            id := find_user_id(db, positional[1]);
            if !id { log_error("No user named %.", positional[1]); exit(1); }
            password := read_password();
            error := set_password(db, id, password);
            if error { log_error("%", error); exit(1); }
            print("Password updated.\n");
        case "list-users";
            s := prepare(db, "SELECT id, username, display_name, is_admin, active FROM users ORDER BY id");
            while step(s) print("% % \"%\"% %\n", column_int(s, 0), column_text(s, 1), column_text(s, 2),
                ifx column_int(s, 3) then " admin" else "", ifx column_int(s, 4) then "" else "(deactivated)");
            finalize(s);
        case;
            serve();
    }
}

print_usage :: () {
    print(#string TXT
Usage:
  taskforge-server [--port 8080] [--listen 127.0.0.1] [--db taskforge.db] [--web ./web] [--threads 8] [--demo]
  taskforge-server [--db taskforge.db] add-user <username> "<Display Name>" [--admin]   (reads password from stdin)
  taskforge-server [--db taskforge.db] set-password <username>                         (reads password from stdin)
  taskforge-server [--db taskforge.db] list-users
TXT);
}

//
// Database
//

SCHEMA :: #string SQL
CREATE TABLE IF NOT EXISTS meta (
    key   TEXT PRIMARY KEY,
    value INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS records (
    key     TEXT PRIMARY KEY,
    line    TEXT NOT NULL,
    rev     INTEGER NOT NULL,
    deleted INTEGER NOT NULL DEFAULT 0,
    user_id INTEGER
);
CREATE INDEX IF NOT EXISTS records_by_rev ON records(rev);
CREATE TABLE IF NOT EXISTS users (
    id           INTEGER PRIMARY KEY,
    username     TEXT NOT NULL UNIQUE COLLATE NOCASE,
    display_name TEXT NOT NULL,
    pw_hash      BLOB NOT NULL,
    pw_salt      BLOB NOT NULL,
    pw_iter      INTEGER NOT NULL,
    is_admin     INTEGER NOT NULL DEFAULT 0,
    active       INTEGER NOT NULL DEFAULT 1,
    created      INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS sessions (
    token_hash BLOB PRIMARY KEY,
    user_id    INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    expires    INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS sessions_by_user ON sessions(user_id);
CREATE TABLE IF NOT EXISTS login_failures (
    username TEXT NOT NULL COLLATE NOCASE,
    at       INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS login_failures_by_name ON login_failures(username, at);
SQL

migrate :: (db: Db) -> bool {
    if !db_exec(db, SCHEMA) return false;
    if get_meta(db, "seeded") return true;

    // First start: seed the project.
    p := starter_project(config.demo);
    lines: [..] string;
    project_records(p, *lines);
    db_exec(db, "BEGIN IMMEDIATE;");
    s := prepare(db, "INSERT OR REPLACE INTO records(key, line, rev, deleted) VALUES (?, ?, 1, 0)");
    for lines {
        bind(s, 1, record_key(it));
        bind(s, 2, it);
        run(s);
        reset(s);
    }
    finalize(s);
    set_meta(db, "rev", 1);
    set_meta(db, "next_id", p.next_id);
    set_meta(db, "users_rev", 1);
    set_meta(db, "seeded", 1);
    db_exec(db, "COMMIT;");
    log("Created a new % project in %.", ifx config.demo then "demo" else "starter", config.db_path);
    return true;
}

get_meta :: (db: Db, key: string) -> s64 {
    s := prepare(db, "SELECT value FROM meta WHERE key = ?");
    bind(s, 1, key);
    value: s64;
    if step(s) value = column_int(s, 0);
    finalize(s);
    return value;
}

set_meta :: (db: Db, key: string, value: s64) {
    s := prepare(db, "INSERT INTO meta(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value");
    bind(s, 1, key);
    bind(s, 2, value);
    run(s);
    finalize(s);
}

NOW_SQL :: "CAST(strftime('%s', 'now') AS INTEGER)";

//
// Users
//

valid_username :: (name: string) -> bool {
    if name.count < 1 || name.count > 32 return false;
    for i: 0..name.count-1 {
        c := name[i];
        ok := (c >= #char "a" && c <= #char "z") || (c >= #char "A" && c <= #char "Z") || (c >= #char "0" && c <= #char "9") ||
              c == #char "." || c == #char "_" || c == #char "-";
        if !ok return false;
    }
    return true;
}

hash_password :: (password: string, salt: string, iterations: int) -> string {
    digest := pbkdf2_sha256(password, salt, iterations);
    result := alloc_string(32,, temp);
    memcpy(result.data, digest.data, 32);
    return result;
}

new_salt :: () -> string {
    salt := alloc_string(16,, temp);
    random_bytes(salt.data, 16);
    return salt;
}

add_user :: (db: Db, username: string, display_name: string, password: string, admin: bool) -> error: string {
    if !valid_username(username) return "Usernames are 1-32 characters: letters, digits, '.', '_' or '-'.";
    if !display_name.count return "Display name can't be empty.";
    if password.count < 8 return "Passwords need at least 8 characters.";
    if find_user_id(db, username) return "That username is taken.";
    salt := new_salt();
    s := prepare(db, tprint("INSERT INTO users(username, display_name, pw_hash, pw_salt, pw_iter, is_admin, active, created) VALUES (?, ?, ?, ?, ?, ?, 1, %)", NOW_SQL));
    bind(s, 1, username);
    bind(s, 2, display_name);
    bind_blob(s, 3, hash_password(password, salt, PBKDF2_ITERATIONS));
    bind_blob(s, 4, salt);
    bind(s, 5, PBKDF2_ITERATIONS);
    bind(s, 6, cast(s64) admin);
    ok := run(s);
    finalize(s);
    if !ok return "Could not add the user.";
    bump_users_rev(db);
    return "";
}

set_password :: (db: Db, user_id: s64, password: string, keep_session := "") -> error: string {
    if password.count < 8 return "Passwords need at least 8 characters.";
    salt := new_salt();
    s := prepare(db, "UPDATE users SET pw_hash = ?, pw_salt = ?, pw_iter = ? WHERE id = ?");
    bind_blob(s, 1, hash_password(password, salt, PBKDF2_ITERATIONS));
    bind_blob(s, 2, salt);
    bind(s, 3, PBKDF2_ITERATIONS);
    bind(s, 4, user_id);
    run(s);
    finalize(s);
    // Changing a password signs out every other session of that user.
    s = prepare(db, "DELETE FROM sessions WHERE user_id = ? AND token_hash != ?");
    bind(s, 1, user_id);
    bind_blob(s, 2, keep_session);
    run(s);
    finalize(s);
    return "";
}

find_user_id :: (db: Db, username: string) -> s64 {
    s := prepare(db, "SELECT id FROM users WHERE username = ?");
    bind(s, 1, username);
    id: s64;
    if step(s) id = column_int(s, 0);
    finalize(s);
    return id;
}

check_password :: (db: Db, user_id: s64, password: string) -> bool {
    s := prepare(db, "SELECT pw_hash, pw_salt, pw_iter FROM users WHERE id = ?");
    bind(s, 1, user_id);
    ok := false;
    if step(s) ok = constant_time_equal(hash_password(password, column_text(s, 1), column_int(s, 2)), column_text(s, 0));
    finalize(s);
    return ok;
}

bump_users_rev :: (db: Db) {
    set_meta(db, "users_rev", get_meta(db, "users_rev") + 1);
}

read_password :: () -> string {
    print("Password: ");
    line := read_stdin_line();
    // PowerShell prefixes piped input with a UTF-8 byte order mark.
    if begins_with(line, "\xEF\xBB\xBF") advance(*line, 3);
    while line.count && (line[line.count - 1] == #char "\n" || line[line.count - 1] == #char "\r") line.count -= 1;
    return line;
}

#if OS == .WINDOWS {
    read_stdin_line :: () -> string {
        tf_k32 :: #system_library "kernel32";
        GetStdHandle_ :: (which: s32) -> *void #foreign tf_k32 "GetStdHandle";
        ReadFile_ :: (file: *void, buffer: *u8, to_read: u32, did_read: *u32, overlapped: *void) -> s32 #foreign tf_k32 "ReadFile";
        handle := GetStdHandle_(-10);
        b: String_Builder;
        c: u8;
        n: u32;
        while ReadFile_(handle, *c, 1, *n, null) && n == 1 {
            if c == #char "\n" break;
            append(*b, c);
        }
        return builder_to_string(*b);
    }
} else {
    read_stdin_line :: () -> string {
        b: String_Builder;
        c: u8;
        while read(0, *c, 1) == 1 {
            if c == #char "\n" break;
            append(*b, c);
        }
        return builder_to_string(*b);
    }
}

//
// HTTP
//

serve :: () {
    if !socket_init() exit(1);
    listen_socket = socket(AF_INET, .STREAM, .TCP);
    if listen_socket == INVALID_SOCKET { log_error("socket() failed"); exit(1); }
    yes: s32 = 1;
    setsockopt(listen_socket, SOL_SOCKET, SO_REUSEADDR, xx *yes, size_of(s32));

    addr: sockaddr_in;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(config.port);
    ip := inet_addr(temp_c_string(config.listen_on));
    memcpy(*addr.sin_addr, *ip, 4);
    if bind(listen_socket, cast(*sockaddr) *addr, size_of(sockaddr_in)) != 0 {
        log_error("Could not bind %:% (already in use?)", config.listen_on, config.port);
        exit(1);
    }
    if listen(listen_socket, 128) != 0 { log_error("listen() failed"); exit(1); }

    log("TaskForge server listening on http://%:% (db: %, web: %, % threads, SQLite %)",
        config.listen_on, config.port, config.db_path, config.web_dir, config.threads, to_string(sql.sqlite3_libversion()));

    // Every worker blocks in accept() on the shared socket and owns its own SQLite connection.
    threads := NewArray(max(config.threads, 1) - 1, Thread);
    for * threads {
        thread_init(it, (t: *Thread) -> s64 { worker(); return 0; });
        thread_start(it);
    }
    worker();
}

worker :: () {
    db, ok := db_open(config.db_path);
    if !ok return;
    while true {
        client := accept(listen_socket, null, null);
        if client == INVALID_SOCKET continue;
        set_timeouts(client, 15);
        handle_connection(db, client);
        #if OS == .WINDOWS closesocket(client);
        else               close(client);
        reset_temporary_storage();
    }
}

set_timeouts :: (s: Socket, seconds: s32) {
    #if OS == .WINDOWS {
        ms: u32 = cast(u32) seconds * 1000;
        setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, xx *ms, size_of(u32));
        setsockopt(s, SOL_SOCKET, SO_SNDTIMEO, xx *ms, size_of(u32));
    } else {
        tv: timeval;
        tv.tv_sec = seconds;
        setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, *tv, size_of(timeval));
        setsockopt(s, SOL_SOCKET, SO_SNDTIMEO, *tv, size_of(timeval));
    }
}

Request :: struct {
    method: string;
    path:   string;
    authorization: string;
    body:   string;
}

handle_connection :: (db: Db, s: Socket) {
    request, ok := read_request(s);
    if !ok {
        respond(s, 400, "text/plain", "Bad request\n");
        return;
    }

    if request.method == "GET" {
        if request.path == "/api/info" {
            respond(s, 200, "text/plain", "taskforge-server 1\n");
            return;
        }
        serve_static(s, request.path);
        return;
    }
    if request.method != "POST" {
        respond(s, 405, "text/plain", "Method not allowed\n");
        return;
    }

    if request.path == {
        case "/api/login";  handle_login(db, s, request);
        case "/api/logout"; handle_logout(db, s, request);
        case "/api/sync";   handle_sync(db, s, request);
        case "/api/users";  handle_users(db, s, request);
        case;               respond(s, 404, "text/plain", "Not found\n");
    }
}

read_request :: (s: Socket) -> Request, bool {
    request: Request;
    buffer: [..] u8;
    buffer.allocator = temp;
    header_end := -1;
    chunk: [16384] u8;
    while header_end < 0 {
        n := recv(s, chunk.data, xx chunk.count, 0);
        if n <= 0 return request, false;
        old := buffer.count;
        array_resize(*buffer, old + n, initialize = false);
        memcpy(buffer.data + old, chunk.data, n);
        // Look for the blank line ending the headers.
        start := max(0, old - 3);
        for i: start..buffer.count-4 {
            if buffer[i] == #char "\r" && buffer[i+1] == #char "\n" && buffer[i+2] == #char "\r" && buffer[i+3] == #char "\n" {
                header_end = i + 4;
                break;
            }
        }
        if header_end < 0 && buffer.count > 65536 return request, false;
    }

    head: string;
    head.data  = buffer.data;
    head.count = header_end;
    lines := split(head, "\r\n",, temp);
    if !lines.count return request, false;
    parts := split(lines[0], " ",, temp);
    if parts.count < 2 return request, false;
    request.method = parts[0];
    request.path   = parts[1];
    query := find_index_from_left(request.path, #char "?");
    if query >= 0 request.path.count = query;

    content_length := 0;
    for line: lines {
        colon := find_index_from_left(line, #char ":");
        if colon <= 0 continue;
        name  := to_lower_copy(substring(line, 0, colon),, temp);
        value := trim(substring(line, colon + 1, line.count));
        if name == "content-length" content_length = string_to_int(value);
        if name == "authorization"  request.authorization = value;
    }
    if content_length < 0 || content_length > MAX_BODY return request, false;

    body := alloc_string(content_length,, temp);
    have := min(buffer.count - header_end, content_length);
    memcpy(body.data, buffer.data + header_end, have);
    while have < content_length {
        n := recv(s, body.data + have, xx min(content_length - have, 1 << 20), 0);
        if n <= 0 return request, false;
        have += n;
    }
    request.body = body;
    return request, true;
}

respond :: (s: Socket, status: int, content_type: string, body: string, extra_headers := "") {
    reason := "OK";
    if status == {
        case 400; reason = "Bad Request";
        case 401; reason = "Unauthorized";
        case 403; reason = "Forbidden";
        case 404; reason = "Not Found";
        case 405; reason = "Method Not Allowed";
        case 429; reason = "Too Many Requests";
        case 500; reason = "Internal Server Error";
    }
    header := tprint("HTTP/1.1 % %\r\nContent-Type: %\r\nContent-Length: %\r\nConnection: close\r\nX-Content-Type-Options: nosniff\r\nCache-Control: no-cache\r\n%\r\n",
        status, reason, content_type, body.count, extra_headers);
    send_all(s, header);
    send_all(s, body);
}

send_all :: (s: Socket, data: string) {
    sent := 0;
    while sent < data.count {
        n := send(s, data.data + sent, xx min(data.count - sent, 1 << 20), 0);
        if n <= 0 return;
        sent += n;
    }
}

serve_static :: (s: Socket, path: string) {
    name := ifx path == "/" then "index.html" else substring(path, 1, path.count);
    // Only plain file names from the web directory.
    for i: 0..name.count-1 {
        c := name[i];
        ok := (c >= #char "a" && c <= #char "z") || (c >= #char "A" && c <= #char "Z") || (c >= #char "0" && c <= #char "9") ||
              c == #char "." || c == #char "_" || c == #char "-";
        if !ok || (c == #char "." && i == 0) {
            respond(s, 404, "text/plain", "Not found\n");
            return;
        }
    }
    data, ok := read_entire_file(tprint("%/%", config.web_dir, name), log_errors = false);
    if !ok {
        respond(s, 404, "text/plain", "Not found\n");
        return;
    }
    type := "application/octet-stream";
    if      ends_with(name, ".html") type = "text/html; charset=utf-8";
    else if ends_with(name, ".js")   type = "text/javascript; charset=utf-8";
    else if ends_with(name, ".wasm") type = "application/wasm";
    else if ends_with(name, ".css")  type = "text/css; charset=utf-8";
    respond(s, 200, type, data);
    free(data);
}

//
// API
//

Session :: struct {
    user_id:  s64;
    is_admin: bool;
    token_hash: string;
}

authenticate :: (db: Db, request: Request) -> Session, bool {
    session: Session;
    prefix :: "Bearer ";
    if !begins_with(request.authorization, prefix) return session, false;
    token := substring(request.authorization, prefix.count, request.authorization.count);
    token_hash := sha256(token);
    s := prepare(db, tprint("SELECT u.id, u.is_admin, s.expires - % FROM sessions s JOIN users u ON u.id = s.user_id WHERE s.token_hash = ? AND s.expires > % AND u.active = 1", NOW_SQL, NOW_SQL));
    bind_blob(s, 1, as_string(*token_hash));
    ok := step(s);
    remaining: s64;
    if ok {
        session.user_id  = column_int(s, 0);
        session.is_admin = column_int(s, 1) != 0;
        session.token_hash = copy_string(as_string(*token_hash),, temp);
        remaining = column_int(s, 2);
    }
    finalize(s);
    // Sliding expiry, refreshed at most about once a day.
    if ok && remaining < (SESSION_DAYS - 1) * 86400 {
        u := prepare(db, tprint("UPDATE sessions SET expires = % + % WHERE token_hash = ?", NOW_SQL, SESSION_DAYS * 86400));
        bind_blob(u, 1, as_string(*token_hash));
        run(u);
        finalize(u);
    }
    return session, ok;
}

handle_login :: (db: Db, s: Socket, request: Request) {
    r := Line_Reader.{line = request.body};
    username := next_token(*r);
    password := next_token(*r);

    // Throttle password guessing: 10 failures within 10 minutes locks the username for a while.
    MAX_FAILURES   :: 10;
    WINDOW_SECONDS :: 600;
    f := prepare(db, tprint("SELECT COUNT(*) FROM login_failures WHERE username = ? AND at > % - %", NOW_SQL, WINDOW_SECONDS));
    bind(f, 1, username);
    failures: s64;
    if step(f) failures = column_int(f, 0);
    finalize(f);
    if failures >= MAX_FAILURES {
        respond(s, 429, "text/plain", "error \"Too many failed attempts. Try again in a few minutes.\"\n");
        return;
    }

    q := prepare(db, "SELECT id, pw_hash, pw_salt, pw_iter, active FROM users WHERE username = ?");
    bind(q, 1, username);
    found := step(q);
    user_id: s64;
    ok := false;
    if found {
        user_id = column_int(q, 0);
        hash := hash_password(password, column_text(q, 2), column_int(q, 3));
        ok = constant_time_equal(hash, column_text(q, 1)) && column_int(q, 4) != 0;
    } else {
        hash_password(password, "timing-equalizer", PBKDF2_ITERATIONS);  // Don't reveal which usernames exist.
    }
    finalize(q);
    if !ok {
        record := prepare(db, tprint("INSERT INTO login_failures(username, at) VALUES (?, %)", NOW_SQL));
        bind(record, 1, username);
        run(record);
        finalize(record);
        db_exec(db, tprint("DELETE FROM login_failures WHERE at < % - 86400;", NOW_SQL));
        respond(s, 401, "text/plain", "error \"Wrong username or password.\"\n");
        return;
    }
    clear := prepare(db, "DELETE FROM login_failures WHERE username = ?");
    bind(clear, 1, username);
    run(clear);
    finalize(clear);

    token_bytes: [32] u8;
    random_bytes(token_bytes.data, 32);
    token := to_hex(token_bytes.data, 32);
    token_hash := sha256(token);
    db_exec(db, tprint("DELETE FROM sessions WHERE expires < %;", NOW_SQL));
    ins := prepare(db, tprint("INSERT INTO sessions(token_hash, user_id, expires) VALUES (?, ?, % + %)", NOW_SQL, SESSION_DAYS * 86400));
    bind_blob(ins, 1, as_string(*token_hash));
    bind(ins, 2, user_id);
    run(ins);
    finalize(ins);
    respond(s, 200, "text/plain", tprint("token %\n", token));
    free(token);
}

handle_logout :: (db: Db, s: Socket, request: Request) {
    prefix :: "Bearer ";
    if begins_with(request.authorization, prefix) {
        token := substring(request.authorization, prefix.count, request.authorization.count);
        token_hash := sha256(token);
        d := prepare(db, "DELETE FROM sessions WHERE token_hash = ?");
        bind_blob(d, 1, as_string(*token_hash));
        run(d);
        finalize(d);
    }
    respond(s, 200, "text/plain", "ok\n");
}

handle_sync :: (db: Db, s: Socket, request: Request) {
    session, authed := authenticate(db, request);
    if !authed {
        respond(s, 401, "text/plain", "error \"Please sign in again.\"\n");
        return;
    }

    since: s64;
    client_users_rev: s64;
    want_ids := false;
    puts: [..] string;
    puts.allocator = temp;
    dels: [..] string;
    dels.allocator = temp;

    body := request.body;
    while body.count {
        end := find_index_from_left(body, #char "\n");
        if end < 0 end = body.count;
        line := substring(body, 0, end);
        advance(*body, min(end + 1, body.count));
        if begins_with(line, "put ")        array_add(*puts, substring(line, 4, line.count));
        else if begins_with(line, "del ")   array_add(*dels, substring(line, 4, line.count));
        else if begins_with(line, "since ") since = string_to_int(substring(line, 6, line.count));
        else if begins_with(line, "users_rev ") client_users_rev = string_to_int(substring(line, 10, line.count));
        else if line == "ids"               want_ids = true;
    }

    out: String_Builder;
    out.allocator = temp;

    db_exec(db, "BEGIN IMMEDIATE;");
    rev := get_meta(db, "rev");
    if puts.count || dels.count {
        rev += 1;
        up := prepare(db, "INSERT INTO records(key, line, rev, deleted, user_id) VALUES (?, ?, ?, 0, ?) ON CONFLICT(key) DO UPDATE SET line = excluded.line, rev = excluded.rev, deleted = 0, user_id = excluded.user_id");
        for puts {
            key := record_key(it);
            if !key.count continue;
            bind(up, 1, key);
            bind(up, 2, it);
            bind(up, 3, rev);
            bind(up, 4, session.user_id);
            run(up);
            reset(up);
        }
        finalize(up);
        del := prepare(db, "UPDATE records SET deleted = 1, line = '', rev = ?, user_id = ? WHERE key = ?");
        for dels {
            bind(del, 1, rev);
            bind(del, 2, session.user_id);
            bind(del, 3, it);
            run(del);
            reset(del);
        }
        finalize(del);
        set_meta(db, "rev", rev);
    }
    if want_ids {
        start := get_meta(db, "next_id");
        set_meta(db, "next_id", start + ID_BLOCK);
        print_to_builder(*out, "ids % %\n", start, ID_BLOCK);
    }

    print_to_builder(*out, "rev %\n", rev);
    print_to_builder(*out, "me % %\n", session.user_id, cast(int) session.is_admin);
    users_rev := get_meta(db, "users_rev");
    if users_rev != client_users_rev {
        print_to_builder(*out, "users_rev %\n", users_rev);
        u := prepare(db, "SELECT id, username, display_name, is_admin, active FROM users ORDER BY display_name COLLATE NOCASE");
        while step(u) {
            print_to_builder(*out, "user % % % % %\n", column_int(u, 0), quoted(column_text(u, 1)), quoted(column_text(u, 2)), column_int(u, 3), column_int(u, 4));
        }
        finalize(u);
    }

    q := prepare(db, ifx since == 0 then "SELECT key, line, deleted FROM records WHERE rev > ? AND deleted = 0" else "SELECT key, line, deleted FROM records WHERE rev > ? ORDER BY rev");
    bind(q, 1, since);
    while step(q) {
        if column_int(q, 2) print_to_builder(*out, "del %\n", column_text(q, 0));
        else                print_to_builder(*out, "put %\n", column_text(q, 1));
    }
    finalize(q);
    db_exec(db, "COMMIT;");

    respond(s, 200, "text/plain; charset=utf-8", builder_to_string(*out,, temp));
}

// Commands (one per request):
//   add "username" "Display Name" "password" <admin>       admin only
//   update <id> "Display Name" <admin> <active>           admin only
//   password <id> "new password" "current password"        admin, or yourself with your current password
handle_users :: (db: Db, s: Socket, request: Request) {
    session, authed := authenticate(db, request);
    if !authed {
        respond(s, 401, "text/plain", "error \"Please sign in again.\"\n");
        return;
    }
    error := users_command(db, session, request.body, session.token_hash);
    if error respond(s, 400, "text/plain", tprint("error %\n", quoted(error)));
    else     respond(s, 200, "text/plain", "ok\n");
}

users_command :: (db: Db, session: Session, body: string, token_hash: string) -> error: string {
    r := Line_Reader.{line = body};
    command := next_token(*r);
    if command == {
        case "add";
            if !session.is_admin { return "Only admins can add users."; }
            username := next_token(*r);
            name     := next_token(*r);
            password := next_token(*r);
            admin    := next_u32(*r) != 0;
            return add_user(db, username, name, password, admin);
        case "update";
            if !session.is_admin { return "Only admins can change users."; }
            id     := cast(s64) next_u32(*r);
            name   := next_token(*r);
            admin  := next_u32(*r) != 0;
            active := next_u32(*r) != 0;
            if id == session.user_id && (!admin || !active) { return "You can't remove your own admin rights or deactivate yourself."; }
            if !name.count { return "Display name can't be empty."; }
            u := prepare(db, "UPDATE users SET display_name = ?, is_admin = ?, active = ? WHERE id = ?");
            bind(u, 1, name);
            bind(u, 2, cast(s64) admin);
            bind(u, 3, cast(s64) active);
            bind(u, 4, id);
            run(u);
            finalize(u);
            if !active {
                d := prepare(db, "DELETE FROM sessions WHERE user_id = ?");
                bind(d, 1, id);
                run(d);
                finalize(d);
            }
            bump_users_rev(db);
        case "password";
            id       := cast(s64) next_u32(*r);
            password := next_token(*r);
            current  := next_token(*r);
            if id != session.user_id && !session.is_admin { return "You can only change your own password."; }
            if id == session.user_id && !check_password(db, id, current) { return "Your current password is wrong."; }
            return set_password(db, id, password, ifx id == session.user_id then token_hash else "");
        case;
            return "Unknown command.";
    }
    return "";
}

#load "sqlite.jai";
#load "crypto.jai";
#load "../src/common.jai";
#load "../src/model.jai";
#load "../src/storage.jai";

#import "Basic";
#import "String";
#import "Math";
#import "File";
#import "Socket";
#import "Thread";
#if OS != .WINDOWS {
    #import "POSIX";
}