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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

// SPDX-License-Identifier: BUSL-1.1

//! Decode helpers for sync-engine `ReplicatedWrite` variants.
//!
//! Each function maps the destructured fields of one `ReplicatedWrite` variant
//! back to a `PhysicalPlan`, using the leader-assigned surrogates verbatim
//! rather than re-deriving identity through the local assigner. `wal_lsn` is
//! always `None` — followers allocate their own WAL LSN at apply time.

use crate::bridge::envelope::PhysicalPlan;
use nodedb_physical::physical_plan::{
    ColumnarInsertIntent, ColumnarOp, SpatialOp, TextOp, TimeseriesOp,
};
use nodedb_types::Surrogate;

/// Decode optional sync provenance from the wire bytes.
///
/// Provenance carries the producer/epoch/seq that the Data Plane idempotency
/// gate uses to deduplicate replayed writes. A corrupt encoding must fail loud
/// (propagate) — the same contract as `geometry` decoding in
/// [`spatial_insert`] — rather than silently dropping to `None`. A silent drop
/// would blind the gate and risk double-applying the write on a follower.
pub fn decode_provenance(
    prov_bytes: &Option<Vec<u8>>,
) -> crate::Result<Option<nodedb_types::sync::wire::SyncProvenance>> {
    match prov_bytes {
        Some(b) => zerompk::from_msgpack::<nodedb_types::sync::wire::SyncProvenance>(b)
            .map(Some)
            .map_err(|e| crate::Error::Internal {
                detail: format!("SyncProvenance decode failed: {e}"),
            }),
        None => Ok(None),
    }
}

pub fn columnar_ingest(
    collection: &str,
    payload: &[u8],
    schema_bytes: &[u8],
    surrogates: &[u32],
    prov_bytes: &Option<Vec<u8>>,
) -> crate::Result<PhysicalPlan> {
    let provenance = decode_provenance(prov_bytes)?;
    Ok(PhysicalPlan::Columnar(ColumnarOp::Insert {
        collection: collection.to_owned(),
        payload: payload.to_vec(),
        format: "msgpack".to_owned(),
        // The sync path always uses a plain insert with no conflict resolution.
        intent: ColumnarInsertIntent::Insert,
        on_conflict_updates: Vec::new(),
        surrogates: surrogates.iter().copied().map(Surrogate::new).collect(),
        schema_bytes: schema_bytes.to_vec(),
        provenance,
        wal_lsn: None,
    }))
}

pub fn timeseries_ingest(
    collection: &str,
    payload: &[u8],
    format: &str,
    surrogates: &[u32],
    prov_bytes: &Option<Vec<u8>>,
) -> crate::Result<PhysicalPlan> {
    let provenance = decode_provenance(prov_bytes)?;
    Ok(PhysicalPlan::Timeseries(TimeseriesOp::Ingest {
        collection: collection.to_owned(),
        payload: payload.to_vec(),
        format: format.to_owned(),
        wal_lsn: None,
        surrogates: surrogates.iter().copied().map(Surrogate::new).collect(),
        provenance,
    }))
}

pub fn fts_index(
    collection: &str,
    surrogate: u32,
    text: &str,
    prov_bytes: &Option<Vec<u8>>,
) -> crate::Result<PhysicalPlan> {
    let provenance = decode_provenance(prov_bytes)?;
    Ok(PhysicalPlan::Text(TextOp::FtsIndexDoc {
        collection: collection.to_owned(),
        surrogate: Surrogate::new(surrogate),
        text: text.to_owned(),
        provenance,
    }))
}

pub fn fts_delete(
    collection: &str,
    surrogate: u32,
    prov_bytes: &Option<Vec<u8>>,
) -> crate::Result<PhysicalPlan> {
    let provenance = decode_provenance(prov_bytes)?;
    Ok(PhysicalPlan::Text(TextOp::FtsDeleteDoc {
        collection: collection.to_owned(),
        surrogate: Surrogate::new(surrogate),
        provenance,
    }))
}

pub fn spatial_insert(
    collection: &str,
    field: &str,
    surrogate: u32,
    geometry_bytes: &[u8],
    prov_bytes: &Option<Vec<u8>>,
) -> crate::Result<PhysicalPlan> {
    let geometry = zerompk::from_msgpack::<nodedb_types::geometry::Geometry>(geometry_bytes)
        .map_err(|e| crate::Error::Internal {
            detail: format!("SpatialInsert geometry decode failed: {e}"),
        })?;
    let provenance = decode_provenance(prov_bytes)?;
    Ok(PhysicalPlan::Spatial(SpatialOp::Insert {
        collection: collection.to_owned(),
        field: field.to_owned(),
        surrogate: Surrogate::new(surrogate),
        geometry,
        provenance,
    }))
}

pub fn spatial_delete(
    collection: &str,
    field: &str,
    surrogate: u32,
    prov_bytes: &Option<Vec<u8>>,
) -> crate::Result<PhysicalPlan> {
    let provenance = decode_provenance(prov_bytes)?;
    Ok(PhysicalPlan::Spatial(SpatialOp::Delete {
        collection: collection.to_owned(),
        field: field.to_owned(),
        surrogate: Surrogate::new(surrogate),
        provenance,
    }))
}