nodedb-lite 0.1.0

// SPDX-License-Identifier: Apache-2.0
//! CRDT version-history handlers: time-travel reads, delta export, restore, compaction.

use std::collections::HashMap;

use loro::VersionVector;
use nodedb_types::result::QueryResult;
use nodedb_types::value::Value;

use crate::error::LiteError;
use crate::query::engine::LiteQueryEngine;
use crate::storage::engine::StorageEngine;

/// Parse a JSON `{"<peer_hex>": counter}` object into a Loro `VersionVector`.
///
/// Peer IDs are encoded as 16-char lowercase hex strings. Counters are i64
/// on the wire (JSON number) and narrowed to i32 (Loro's `Counter` type).
fn parse_version_vector(json: &str) -> Result<VersionVector, LiteError> {
    let map: HashMap<String, i64> =
        sonic_rs::from_str(json).map_err(|e| LiteError::BadRequest {
            detail: format!("invalid version vector JSON: {e}"),
        })?;

    let pairs: Vec<(u64, i32)> = map
        .into_iter()
        .map(|(hex, counter)| {
            let peer = u64::from_str_radix(&hex, 16).map_err(|e| LiteError::BadRequest {
                detail: format!("peer ID '{hex}' is not valid hex: {e}"),
            })?;
            let counter_i32 = i32::try_from(counter).map_err(|_| LiteError::BadRequest {
                detail: format!("peer '{hex}' counter {counter} exceeds Loro's i32 Counter range"),
            })?;
            Ok((peer, counter_i32))
        })
        .collect::<Result<Vec<_>, LiteError>>()?;

    Ok(pairs.into_iter().collect())
}

/// Error for a version-history op naming a collection that has no CRDT
/// document.
///
/// Version history is per collection now that each one owns its own Loro
/// document, so "no document" cannot be answered with an empty success — that
/// would report an empty history for a typo'd collection name.
fn missing_collection(op: &str, collection: &str) -> LiteError {
    LiteError::BadRequest {
        detail: format!("{op}: CRDT collection '{collection}' does not exist"),
    }
}

/// Read a document's state at a historical version.
pub async fn handle_read_at_version<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
    collection: &str,
    document_id: &str,
    version_vector_json: &str,
) -> Result<QueryResult, LiteError> {
    let vv = parse_version_vector(version_vector_json)?;

    let crdt = engine.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
    let state = crdt
        .state(collection)
        .ok_or_else(|| missing_collection("ReadAtVersion", collection))?;

    let value = state
        .read_at_version(collection, document_id, &vv)
        .map_err(|e| LiteError::crdt("ReadAtVersion", e))?;

    let row = match value {
        Some(v) => {
            let json = sonic_rs::to_string(&v).map_err(|e| LiteError::Serialization {
                detail: format!("ReadAtVersion serialize: {e}"),
            })?;
            vec![vec![Value::String(json)]]
        }
        None => vec![],
    };

    Ok(QueryResult {
        columns: vec!["document".to_string()],
        rows: row,
        rows_affected: 0,
        command: None,
    })
}

/// Return the current oplog version vector as a JSON string.
pub async fn handle_get_version_vector<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
) -> Result<QueryResult, LiteError> {
    let crdt = engine.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
    let clock = crdt.export_vector_clock();
    let json = sonic_rs::to_string(&clock).map_err(|e| LiteError::Serialization {
        detail: format!("GetVersionVector serialize: {e}"),
    })?;
    Ok(QueryResult {
        columns: vec!["version_vector_json".to_string()],
        rows: vec![vec![Value::String(json)]],
        rows_affected: 0,
        command: None,
    })
}

/// Export a collection's oplog delta from a version to current state.
pub async fn handle_export_delta<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
    collection: &str,
    from_version_json: &str,
) -> Result<QueryResult, LiteError> {
    let vv = parse_version_vector(from_version_json)?;

    let crdt = engine.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
    if crdt.state(collection).is_none() {
        return Err(missing_collection("ExportDelta", collection));
    }
    let delta_bytes = crdt.export_delta_from(collection, &vv)?;

    Ok(QueryResult {
        columns: vec!["delta_bytes".to_string()],
        rows: vec![vec![Value::Bytes(delta_bytes)]],
        rows_affected: 0,
        command: None,
    })
}

/// Restore a document to a historical version via a forward mutation.
///
/// The restore delta joins the sync queue. Returns its bytes.
pub async fn handle_restore_to_version<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
    collection: &str,
    document_id: &str,
    target_version_json: &str,
) -> Result<QueryResult, LiteError> {
    handle_restore_to_version_coordinated(
        engine,
        None,
        collection,
        document_id,
        target_version_json,
    )
    .await
}

pub(crate) async fn handle_restore_to_version_coordinated<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
    permit: Option<&crate::engine::fts::coordinator::TextMutationPermit>,
    collection: &str,
    document_id: &str,
    target_version_json: &str,
) -> Result<QueryResult, LiteError> {
    if let Some(permit) = permit {
        return handle_restore_to_version_admitted(
            engine,
            permit,
            collection,
            document_id,
            target_version_json,
        )
        .await;
    }
    let guard = engine.fts_state.admit_mutation().await;
    let result = handle_restore_to_version_admitted(
        engine,
        guard.permit(),
        collection,
        document_id,
        target_version_json,
    )
    .await;
    guard.finish(result)
}

pub(crate) async fn handle_restore_to_version_admitted<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
    permit: &crate::engine::fts::coordinator::TextMutationPermit,
    collection: &str,
    document_id: &str,
    target_version_json: &str,
) -> Result<QueryResult, LiteError> {
    let _permit = permit;
    let vv = parse_version_vector(target_version_json)?;

    let delta_bytes = {
        let mut crdt = engine.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
        if crdt.state(collection).is_none() {
            return Err(missing_collection("RestoreToVersion", collection));
        }
        crdt.restore_to_version(collection, document_id, &vv)?
    };
    // The restore is a local write: the text index follows the restored row.
    crate::query::text_index::reindex_documents(engine, collection, [document_id])?;

    Ok(QueryResult {
        columns: vec!["delta_bytes".to_string()],
        rows: vec![vec![Value::Bytes(delta_bytes)]],
        rows_affected: 1,
        command: None,
    })
}

/// Compact a collection's CRDT oplog at a specific version, discarding history
/// before it.
pub async fn handle_compact_at_version<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
    collection: &str,
    target_version_json: &str,
) -> Result<QueryResult, LiteError> {
    handle_compact_at_version_coordinated(engine, None, collection, target_version_json).await
}

pub(crate) async fn handle_compact_at_version_coordinated<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
    permit: Option<&crate::engine::fts::coordinator::TextMutationPermit>,
    collection: &str,
    target_version_json: &str,
) -> Result<QueryResult, LiteError> {
    if let Some(permit) = permit {
        return handle_compact_at_version_admitted(engine, permit, collection, target_version_json)
            .await;
    }
    let guard = engine.fts_state.admit_mutation().await;
    let result =
        handle_compact_at_version_admitted(engine, guard.permit(), collection, target_version_json)
            .await;
    guard.finish(result)
}

pub(crate) async fn handle_compact_at_version_admitted<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
    permit: &crate::engine::fts::coordinator::TextMutationPermit,
    collection: &str,
    target_version_json: &str,
) -> Result<QueryResult, LiteError> {
    let _permit = permit;
    let vv = parse_version_vector(target_version_json)?;

    let mut crdt = engine.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
    if crdt.state(collection).is_none() {
        return Err(missing_collection("CompactAtVersion", collection));
    }
    crdt.compact_at_version(collection, &vv)?;

    Ok(QueryResult {
        columns: vec![],
        rows: vec![],
        rows_affected: 1,
        command: None,
    })
}