// A redis-like server in Hew v0.5 — command parsing, key/value storage, and
// MULTI/EXEC transactions, behind actor message-passing.
//
// WHAT IS REAL HERE:
//   * line-oriented command parsing (split + dispatch on the verb via `match`)
//   * a stateful connection protocol: a Ready/Queuing state machine that queues
//     commands issued inside MULTI and REPLAYS them on EXEC (or drops on DISCARD)
//   * HashMap-backed storage, all serialized through the actor mailbox
//   Commands: PING SET GET DEL EXISTS DBSIZE MULTI EXEC DISCARD
//
// WHAT IS NOT HERE YET — the actual TCP listen/accept/stream-read/frame/write
// loop. On trunk that is blocked by three substrate gaps the v0.5 gauntlet
// confirmed:
//   * std::net is un-importable (std::fs tag-aware-drop / D10 codegen gap)
//   * the active-mode conn.attach surface has a reactor use-after-free (fix in flight)
//   * Vec<LocalPid<RedisConn>> (to hold one connection actor per client) hits the
//     Vec-of-actor-handle layout ABI gap
// So the socket is stood in for by feeding the parser a scripted client session.
// Swapping that for a real connection is the remaining work, not the logic below.
enum ConnState {
    Ready;
    Queuing;
}

// Apply one data command against the store; returns a RESP-style reply.
// Shared by the immediate path and the EXEC replay loop.
fn run_cmd(store: HashMap<string, string>, line: string) -> string {
    let parts = line.split(" ");
    let verb = parts.get(0);
    match verb {
        "PING" => "+PONG",
        "DBSIZE" => f":{store.len()}",
        "SET" => {
            store.insert(parts.get(1), parts.get(2));
            "+OK"
        },
        "GET" => match store.get(parts.get(1)) {
            Some(v) => f"+{v}",
            None => "-NOT_FOUND",
        },
        "DEL" => {
            let key = parts.get(1);
            if store.contains_key(key) {
                store.remove(key);
                "+OK"
            } else {
                "-NOT_FOUND"
            }
        },
        "EXISTS" => if store.contains_key(parts.get(1)) {
            ":1"
        } else {
            ":0"
        },
        _ => "-ERR unknown command",
    }
}

actor RedisConn {
    let data: HashMap<string, string>;
    let state: ConnState;
    let queue: Vec<string>;
    receive fn feed(line: string) -> string {
        let cmd = line.trim();
        let verb = cmd.split(" ").get(0);
        match verb {
            "MULTI" => match state {
                ConnState::Ready => {
                    state = ConnState::Queuing;
                    "+OK"
                },
                ConnState::Queuing => "-ERR MULTI calls can not be nested",
            },
            "DISCARD" => match state {
                ConnState::Queuing => {
                    state = ConnState::Ready;
                    queue = Vec::new();
                    "+OK"
                },
                ConnState::Ready => "-ERR DISCARD without MULTI",
            },
            "EXEC" => match state {
                ConnState::Ready => "-ERR EXEC without MULTI",
                ConnState::Queuing => {
                    state = ConnState::Ready;
                    var applied = 0;
                    for i in 0 .. queue.len() {
                        let _ = run_cmd(data, queue.get(i));
                        applied = applied + 1;
                    }
                    queue = Vec::new();
                    f"+EXEC applied {applied} queued command(s)"
                },
            },
            _ => match state {
                ConnState::Queuing => {
                    queue.push(cmd);
                    "+QUEUED"
                },
                ConnState::Ready => run_cmd(data, cmd),
            },
        }
    }
}

fn step(conn: LocalPid<RedisConn>, line: string) {
    let reply = match await conn.feed(line.clone()) {
        Ok(r) => r,
        Err(_) => "-ASK_FAILED",
    };
    println(f"{line}  ->  {reply}");
}

fn main() {
    let conn = spawn RedisConn(data: HashMap::new(), state: ConnState::Ready, queue: Vec::new());
    // A scripted client session — stands in for the not-yet-wired TCP stream.
    step(conn, "PING");
    step(conn, "SET greeting hello");
    step(conn, "GET greeting");
    step(conn, "EXISTS greeting");
    step(conn, "DEL greeting");
    step(conn, "GET greeting");

    // A transaction: queued under MULTI, applied atomically on EXEC.
    step(conn, "MULTI");
    step(conn, "SET a 1");
    step(conn, "SET b 2");
    step(conn, "EXEC");
    step(conn, "DBSIZE");
    step(conn, "GET a");

    // A discarded transaction leaves no trace.
    step(conn, "MULTI");
    step(conn, "SET c 3");
    step(conn, "DISCARD");
    step(conn, "EXISTS c");
}