Function handle_zadd

nodedb/src/control/server/resp/handler_sorted.rs:29–86

ZADD key score member [score member ...] Adds members with scores to a sorted index. The "key" is the sorted index name. Each score/member pair is inserted by writing to the underlying KV collection (which triggers sorted index auto-maintenance). For RESP compatibility, ZADD dispatches a KV PUT with the score embedded in the value, which auto-updates the sorted index.

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

/// For RESP compatibility, ZADD dispatches a KV PUT with the score embedded
/// in the value, which auto-updates the sorted index.
pub(super) async fn handle_zadd(
    cmd: &RespCommand,
    session: &RespSession,
    state: &SharedState,
) -> RespValue {
    // ZADD needs at least: key score member
    if cmd.argc() < 3 || !cmd.argc().is_multiple_of(2) {
        return RespValue::err("ERR wrong number of arguments for 'zadd' command");
    }

    // In RESP mode, the sorted index name = session.collection.
    // The args are: score1 member1 [score2 member2 ...]
    let index_name = session.collection.clone();
    let mut added = 0i64;

    let mut i = 0;
    while i + 1 < cmd.argc() {
        let score_str = match cmd.arg_str(i) {
            Some(s) => s,
            None => return RespValue::err("ERR value is not a valid float"),
        };
        let score: f64 = match score_str.parse() {
            Ok(v) => v,
            Err(_) => return RespValue::err("ERR value is not a valid float"),
        };
        let member = cmd.args[i + 1].clone();

        // Write to the underlying KV collection as a MessagePack document
        // containing the score and member. The sorted index auto-maintenance
        // in KvEngine::put will update the order-statistic tree.
        let value = nodedb_types::json_to_msgpack(&serde_json::json!({
            "score": score,
            "member": String::from_utf8_lossy(&member),
        }))
        .unwrap_or_default();

        let surrogate = match state.surrogate_assigner.assign(&index_name, &member) {
            Ok(s) => s,
            Err(e) => return RespValue::err(format!("ERR {e}")),
        };
        let plan = PhysicalPlan::Kv(KvOp::Put {
            collection: index_name.clone(),
            key: member,
            value,
            ttl_ms: 0,
            surrogate,
        });

        match dispatch_kv_write(state, session, plan).await {
            Ok(_) => added += 1,
            Err(e) => return RespValue::err(format!("ERR {e}")),
        }

        i += 2;
    }

    RespValue::integer(added)
}