nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

// SPDX-License-Identifier: BUSL-1.1

//! WAL append helpers for FTS and Spatial sync ingest paths.
//!
//! Each helper accepts a prebuilt payload struct, serializes it, and appends
//! to the WAL via `WalManager`.  The CP allocates the LSN here; the gate runs
//! Data-Plane-side at the apply handler.
//!
//! Callers are responsible for constructing the payload (which bundles
//! provenance + all operation fields) before calling these helpers.

use nodedb_types::Surrogate;
use nodedb_types::geometry::Geometry;
use nodedb_types::sync::wire::SyncProvenance;

use crate::types::{DatabaseId, TenantId, VShardId};
use crate::wal::manager::WalManager;

/// Build a `SpatialPutPayload` from raw op fields, msgpack-encoding the
/// geometry the same way the sync ingest path (`spatial_handler.rs`) does.
///
/// This is the ONE builder for the shape: both the sync `dispatch_insert`
/// autocommit WAL append and the transaction-resolve serializer call it so
/// producer and `replay_spatial_wal` never drift. `doc_id` is derived from
/// `surrogate` via `surrogate_to_doc_id`, matching the hex-encoded key both
/// the R-tree entry and the sparse document body are keyed by.
pub(crate) fn encode_spatial_put_payload(
    collection: &str,
    field: &str,
    surrogate: Surrogate,
    geometry: &Geometry,
    provenance: &SyncProvenance,
) -> crate::Result<nodedb_wal::record::SpatialPutPayload> {
    let doc_id = crate::engine::document::store::surrogate_to_doc_id(surrogate);
    let geometry_bytes =
        zerompk::to_msgpack_vec(geometry).map_err(|e| crate::Error::Serialization {
            format: "msgpack".into(),
            detail: format!("spatial put: encode geometry for WAL: {e}"),
        })?;
    Ok(nodedb_wal::record::SpatialPutPayload::new(
        provenance.clone(),
        collection,
        field,
        doc_id,
        geometry_bytes,
    ))
}

/// Build a `SpatialDeletePayload` from raw op fields. The ONE builder for the
/// shape, shared by the sync ingest autocommit path and transaction resolve.
pub(crate) fn encode_spatial_delete_payload(
    collection: &str,
    field: &str,
    surrogate: Surrogate,
    provenance: &SyncProvenance,
) -> nodedb_wal::record::SpatialDeletePayload {
    let doc_id = crate::engine::document::store::surrogate_to_doc_id(surrogate);
    nodedb_wal::record::SpatialDeletePayload::new(provenance.clone(), collection, field, doc_id)
}

/// Append an FTS index operation to the WAL and return the assigned LSN.
///
/// The `payload` already carries provenance so replay routes through
/// `execute_fts_index_doc` and the idempotency gate fires on replay.
pub fn wal_append_fts_index(
    wal: &WalManager,
    tenant_id: TenantId,
    vshard_id: VShardId,
    database_id: DatabaseId,
    payload: &nodedb_wal::record::FtsIndexPayload,
) -> crate::Result<nodedb_types::Lsn> {
    let bytes = payload.to_bytes().map_err(crate::Error::Wal)?;
    let lsn = wal.append_fts_index(tenant_id, vshard_id, database_id, &bytes)?;
    Ok(lsn)
}

/// Append an FTS delete operation to the WAL and return the assigned LSN.
///
/// The `payload` already carries provenance so replay routes through
/// `execute_fts_delete_doc` and the idempotency gate fires on replay.
pub fn wal_append_fts_delete(
    wal: &WalManager,
    tenant_id: TenantId,
    vshard_id: VShardId,
    database_id: DatabaseId,
    payload: &nodedb_wal::record::FtsDeletePayload,
) -> crate::Result<nodedb_types::Lsn> {
    let bytes = payload.to_bytes().map_err(crate::Error::Wal)?;
    let lsn = wal.append_fts_delete(tenant_id, vshard_id, database_id, &bytes)?;
    Ok(lsn)
}

/// Append a spatial put (insert) to the WAL and return the assigned LSN.
///
/// The `payload` carries provenance and the msgpack-encoded `Geometry`
/// (identical to what `SpatialInsertMsg.geometry_bytes` carries).
pub fn wal_append_spatial_put(
    wal: &WalManager,
    tenant_id: TenantId,
    vshard_id: VShardId,
    database_id: DatabaseId,
    payload: &nodedb_wal::record::SpatialPutPayload,
) -> crate::Result<nodedb_types::Lsn> {
    let bytes = payload.to_bytes().map_err(crate::Error::Wal)?;
    let lsn = wal.append_spatial_put(tenant_id, vshard_id, database_id, &bytes)?;
    Ok(lsn)
}

/// Append a spatial delete to the WAL and return the assigned LSN.
pub fn wal_append_spatial_delete(
    wal: &WalManager,
    tenant_id: TenantId,
    vshard_id: VShardId,
    database_id: DatabaseId,
    payload: &nodedb_wal::record::SpatialDeletePayload,
) -> crate::Result<nodedb_types::Lsn> {
    let bytes = payload.to_bytes().map_err(crate::Error::Wal)?;
    let lsn = wal.append_spatial_delete(tenant_id, vshard_id, database_id, &bytes)?;
    Ok(lsn)
}