Function handle_scan

nodedb/src/control/server/resp/handler.rs:255–347

(cmd: &RespCommand, session: &RespSession, state: &SharedState)

// ---------------------------------------------------------------------------

async fn handle_scan(cmd: &RespCommand, session: &RespSession, state: &SharedState) -> RespValue {
    let cursor_str = cmd.arg_str(0).unwrap_or("0");
    let cursor = if cursor_str == "0" {
        Vec::new()
    } else {
        base64::Engine::decode(&base64::engine::general_purpose::STANDARD, cursor_str)
            .unwrap_or_default()
    };

    // Parse MATCH, COUNT, and FILTER options.
    let mut match_pattern: Option<String> = None;
    let mut count: usize = 10;
    let mut filter_bytes: Vec<u8> = Vec::new();
    let mut i = 1;
    while i < cmd.argc() {
        match cmd.arg_str(i).map(|s| s.to_uppercase()) {
            Some(ref flag) if flag == "MATCH" => {
                match_pattern = cmd.arg_str(i + 1).map(|s| s.to_string());
                i += 2;
            }
            Some(ref flag) if flag == "COUNT" => {
                count = cmd.arg_i64(i + 1).unwrap_or(10) as usize;
                i += 2;
            }
            // NodeDB extension: SCAN 0 FILTER <field> = <value>
            Some(ref flag) if flag == "FILTER" && i + 4 <= cmd.argc() => {
                // Parse simple "field = value" predicate (needs 4 args: FILTER field = value).
                let field = cmd.arg_str(i + 1).unwrap_or("");
                let _op = cmd.arg_str(i + 2).unwrap_or(""); // "=" expected
                let value = cmd.arg_str(i + 3).unwrap_or("");
                let scan_filter = serde_json::json!([{
                    "field": field,
                    "op": "eq",
                    "value": value,
                }]);
                match nodedb_types::json_to_msgpack(&scan_filter) {
                    Ok(bytes) => filter_bytes = bytes,
                    Err(_) => {
                        return RespValue::err("ERR filter serialization failed");
                    }
                }
                i += 4;
            }
            _ => {
                i += 1;
            }
        }
    }

    let plan = PhysicalPlan::Kv(KvOp::Scan {
        collection: session.collection.clone(),
        cursor,
        count,
        filters: filter_bytes,
        match_pattern,
        sort_keys: Vec::new(),
        surrogate_ceiling: None,
    });