// A redis-style key/value server: a Store actor owns the data, a Session
// machine drives MULTI/EXEC transactions on the client side.
//
// Invariants:
//   * All key/value state lives in the Store actor and is mutated only inside
//     its receive fn, so every command is serialized through one mailbox.
//   * A Session is always in exactly one state. MULTI moves Ready -> Buffering;
//     each command then appends to the buffer; EXEC replays the buffer in order
//     and returns to Ready; DISCARD drops the buffer and returns to Ready.
//   * exec applies exactly one command and returns its RESP-style reply.

machine Session {
    events {
        Begin;
        Enqueue { cmd: string; }
        Commit;
        Abort;
    }

    state Ready;
    state Buffering { pending: Vec<string>; }

    on Begin: Ready => Buffering {
        Buffering { pending: Vec::new() }
    }
    on Enqueue(cmd): Buffering => Buffering reenter {
        self.pending.push(cmd);
        Buffering { pending: self.pending }
    }
    on Commit: Buffering => Ready { Ready }
    on Abort: Buffering => Ready { Ready }

    default { state }
}

actor Store {
    let data: HashMap<string, string>;

    receive fn exec(line: string) -> string {
        let parts = line.trim().split(" ");
        let verb = parts.get(0);
        match verb {
            "PING" => "+PONG",
            "DBSIZE" => f":{data.len()}",
            "SET" => {
                data.insert(parts.get(1), parts.get(2));
                "+OK"
            },
            "GET" => match data.get(parts.get(1)) {
                Some(value) => f"+{value}",
                None => "-NOT_FOUND",
            },
            "DEL" => {
                let key = parts.get(1);
                if data.contains_key(key) {
                    data.remove(key);
                    "+OK"
                } else {
                    "-NOT_FOUND"
                }
            },
            "EXISTS" => if data.contains_key(parts.get(1)) { ":1" } else { ":0" },
            _ => "-ERR unknown command",
        }
    }
}

fn apply(store: LocalPid<Store>, line: string) {
    let reply = match await store.exec(line.clone()) {
        Ok(r) => r,
        Err(_) => "-ERR no reply",
    };
    println(f"{line}  ->  {reply}");
}

fn main() {
    let store = spawn Store(data: HashMap::new());

    apply(store, "PING");
    apply(store, "SET greeting hello");
    apply(store, "GET greeting");
    apply(store, "EXISTS greeting");

    // A transaction: MULTI buffers, EXEC replays the buffer in order.
    var session = Ready;
    session.step(Begin);
    session.step(Enqueue { cmd: "SET a 1" });
    session.step(Enqueue { cmd: "SET b 2" });
    match session {
        Buffering { pending } => {
            for i in 0 .. pending.len() {
                apply(store, pending.get(i));
            }
        },
        Ready => {},
    }
    session.step(Commit);

    apply(store, "DBSIZE");
    apply(store, "GET a");
}