nodedb_types/
protocol.rs

1//! Native binary wire protocol types.
2//!
3//! Shared between the server (`nodedb`) and the client (`nodedb-client`).
4//! All types derive `Serialize`/`Deserialize` for MessagePack encoding.
5
6use serde::{Deserialize, Serialize};
7
8use crate::value::Value;
9
10// ─── Operation Codes ────────────────────────────────────────────────
11
12/// Operation codes for the native binary protocol.
13///
14/// Encoded as a single `u8` in the MessagePack request frame.
15/// Opcodes are grouped by functional area with 16-slot gaps to allow
16/// future additions without renumbering.
17#[repr(u8)]
18#[derive(
19    Debug,
20    Clone,
21    Copy,
22    PartialEq,
23    Eq,
24    Hash,
25    Serialize,
26    Deserialize,
27    zerompk::ToMessagePack,
28    zerompk::FromMessagePack,
29)]
30#[msgpack(c_enum)]
31pub enum OpCode {
32    // ── Auth & session ──────────────────────────────────────────
33    Auth = 0x01,
34    Ping = 0x02,
35
36    // ── Data operations (direct Data Plane dispatch) ────────────
37    PointGet = 0x10,
38    PointPut = 0x11,
39    PointDelete = 0x12,
40    VectorSearch = 0x13,
41    RangeScan = 0x14,
42    CrdtRead = 0x15,
43    CrdtApply = 0x16,
44    GraphRagFusion = 0x17,
45    AlterCollectionPolicy = 0x18,
46
47    // ── SQL & DDL ───────────────────────────────────────────────
48    Sql = 0x20,
49    Ddl = 0x21,
50    Explain = 0x22,
51    CopyFrom = 0x23,
52
53    // ── Session parameters ──────────────────────────────────────
54    Set = 0x30,
55    Show = 0x31,
56    Reset = 0x32,
57
58    // ── Transaction control ─────────────────────────────────────
59    Begin = 0x40,
60    Commit = 0x41,
61    Rollback = 0x42,
62
63    // ── Graph operations (direct Data Plane dispatch) ───────────
64    GraphHop = 0x50,
65    GraphNeighbors = 0x51,
66    GraphPath = 0x52,
67    GraphSubgraph = 0x53,
68    EdgePut = 0x54,
69    EdgeDelete = 0x55,
70    GraphAlgo = 0x56,
71    GraphMatch = 0x57,
72
73    // ── Spatial operations (direct Data Plane dispatch) ────────
74    SpatialScan = 0x19,
75
76    // ── Timeseries operations (direct Data Plane dispatch) ──────
77    TimeseriesScan = 0x1A,
78    TimeseriesIngest = 0x1B,
79
80    // ── Search operations (direct Data Plane dispatch) ──────────
81    TextSearch = 0x60,
82    HybridSearch = 0x61,
83
84    // ── Batch operations ────────────────────────────────────────
85    VectorBatchInsert = 0x70,
86    DocumentBatchInsert = 0x71,
87
88    // ── KV advanced operations ──────────────────────────────────
89    KvScan = 0x72,
90    KvExpire = 0x73,
91    KvPersist = 0x74,
92    KvGetTtl = 0x75,
93    KvBatchGet = 0x76,
94    KvBatchPut = 0x77,
95    KvFieldGet = 0x78,
96    KvFieldSet = 0x79,
97
98    // ── Document advanced operations ────────────────────────────
99    DocumentUpdate = 0x7A,
100    DocumentScan = 0x7B,
101    DocumentUpsert = 0x7C,
102    DocumentBulkUpdate = 0x7D,
103    DocumentBulkDelete = 0x7E,
104
105    // ── Vector advanced operations ──────────────────────────────
106    VectorInsert = 0x7F,
107    VectorMultiSearch = 0x80,
108    VectorDelete = 0x81,
109
110    // ── Columnar operations ─────────────────────────────────────
111    ColumnarScan = 0x82,
112    ColumnarInsert = 0x83,
113
114    // ── Query operations ────────────────────────────────────────
115    RecursiveScan = 0x84,
116
117    // ── Document DDL operations ─────────────────────────────────
118    DocumentTruncate = 0x85,
119    DocumentEstimateCount = 0x86,
120    DocumentInsertSelect = 0x87,
121    DocumentRegister = 0x88,
122    DocumentDropIndex = 0x89,
123
124    // ── KV DDL operations ───────────────────────────────────────
125    KvRegisterIndex = 0x8A,
126    KvDropIndex = 0x8B,
127    KvTruncate = 0x8C,
128
129    // ── Vector DDL operations ───────────────────────────────────
130    VectorSetParams = 0x8D,
131
132    // ── KV atomic operations ───────────────────────────────────
133    KvIncr = 0x8E,
134    KvIncrFloat = 0x8F,
135    KvCas = 0x90,
136    KvGetSet = 0x91,
137
138    // ── KV sorted index operations ─────────────────────────────
139    KvRegisterSortedIndex = 0x92,
140    KvDropSortedIndex = 0x93,
141    KvSortedIndexRank = 0x94,
142    KvSortedIndexTopK = 0x95,
143    KvSortedIndexRange = 0x96,
144    KvSortedIndexCount = 0x97,
145    KvSortedIndexScore = 0x98,
146}
147
148impl OpCode {
149    /// Returns true if this operation is a write that requires WAL append.
150    pub fn is_write(&self) -> bool {
151        matches!(
152            self,
153            OpCode::PointPut
154                | OpCode::PointDelete
155                | OpCode::CrdtApply
156                | OpCode::EdgePut
157                | OpCode::EdgeDelete
158                | OpCode::VectorBatchInsert
159                | OpCode::DocumentBatchInsert
160                | OpCode::AlterCollectionPolicy
161                | OpCode::TimeseriesIngest
162                | OpCode::KvExpire
163                | OpCode::KvPersist
164                | OpCode::KvBatchPut
165                | OpCode::KvFieldSet
166                | OpCode::DocumentUpdate
167                | OpCode::DocumentUpsert
168                | OpCode::DocumentBulkUpdate
169                | OpCode::DocumentBulkDelete
170                | OpCode::VectorInsert
171                | OpCode::VectorDelete
172                | OpCode::ColumnarInsert
173                | OpCode::DocumentTruncate
174                | OpCode::DocumentInsertSelect
175                | OpCode::DocumentRegister
176                | OpCode::DocumentDropIndex
177                | OpCode::KvRegisterIndex
178                | OpCode::KvDropIndex
179                | OpCode::KvTruncate
180                | OpCode::VectorSetParams
181                | OpCode::KvIncr
182                | OpCode::KvIncrFloat
183                | OpCode::KvCas
184                | OpCode::KvGetSet
185                | OpCode::KvRegisterSortedIndex
186                | OpCode::KvDropSortedIndex
187        )
188    }
189}
190
191// ─── Response Status ────────────────────────────────────────────────
192
193/// Status code in response frames.
194#[repr(u8)]
195#[derive(
196    Debug,
197    Clone,
198    Copy,
199    PartialEq,
200    Eq,
201    Serialize,
202    Deserialize,
203    zerompk::ToMessagePack,
204    zerompk::FromMessagePack,
205)]
206#[msgpack(c_enum)]
207pub enum ResponseStatus {
208    /// Request completed successfully.
209    Ok = 0,
210    /// Partial result — more chunks follow with the same `seq`.
211    Partial = 1,
212    /// Request failed — see `error` field.
213    Error = 2,
214}
215
216// ─── Auth ───────────────────────────────────────────────────────────
217
218/// Authentication method in an `Auth` request.
219#[derive(
220    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
221)]
222#[serde(tag = "method", rename_all = "snake_case")]
223pub enum AuthMethod {
224    Trust {
225        #[serde(default = "default_username")]
226        username: String,
227    },
228    Password {
229        username: String,
230        password: String,
231    },
232    ApiKey {
233        token: String,
234    },
235}
236
237fn default_username() -> String {
238    "admin".into()
239}
240
241/// Successful auth response payload.
242#[derive(
243    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
244)]
245pub struct AuthResponse {
246    pub username: String,
247    pub tenant_id: u32,
248}
249
250// ─── Request Frame ──────────────────────────────────────────────────
251
252/// A request sent from client to server over the native protocol.
253///
254/// Serialized as MessagePack. The `op` field selects the handler,
255/// `seq` correlates request to response.
256#[derive(Debug, Clone, Serialize, Deserialize)]
257pub struct NativeRequest {
258    /// Operation code.
259    pub op: OpCode,
260    /// Client-assigned sequence number for request/response correlation.
261    pub seq: u64,
262    /// Operation-specific fields (flattened into the same map).
263    #[serde(flatten)]
264    pub fields: RequestFields,
265}
266
267impl zerompk::ToMessagePack for NativeRequest {
268    fn write<W: zerompk::Write>(&self, writer: &mut W) -> zerompk::Result<()> {
269        writer.write_array_len(3)?;
270        self.op.write(writer)?;
271        writer.write_u64(self.seq)?;
272        self.fields.write(writer)
273    }
274}
275
276impl<'a> zerompk::FromMessagePack<'a> for NativeRequest {
277    fn read<R: zerompk::Read<'a>>(reader: &mut R) -> zerompk::Result<Self> {
278        let len = reader.read_array_len()?;
279        if len != 3 {
280            return Err(zerompk::Error::ArrayLengthMismatch {
281                expected: 3,
282                actual: len,
283            });
284        }
285        let op = OpCode::read(reader)?;
286        let seq = reader.read_u64()?;
287        let fields = RequestFields::read(reader)?;
288        Ok(Self { op, seq, fields })
289    }
290}
291
292/// Operation-specific request fields.
293///
294/// Each variant carries only the fields needed for that operation.
295/// Unknown fields are silently ignored during deserialization.
296#[derive(
297    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
298)]
299#[serde(untagged)]
300pub enum RequestFields {
301    /// Auth, SQL, DDL, Explain, Set, Show, Reset, Begin, Commit, Rollback,
302    /// CopyFrom, Ping — all use a subset of these text fields.
303    Text(TextFields),
304}
305
306/// Catch-all text fields used by most operations.
307///
308/// Each operation uses a subset; unused fields default to `None`/empty.
309#[derive(Debug, Clone, Default, Serialize, Deserialize)]
310pub struct TextFields {
311    // ── Auth ─────────────────────────────────────────────────
312    #[serde(skip_serializing_if = "Option::is_none")]
313    pub auth: Option<AuthMethod>,
314
315    // ── SQL/DDL/Explain/Set/Show/Reset/CopyFrom ─────────────
316    #[serde(skip_serializing_if = "Option::is_none")]
317    pub sql: Option<String>,
318    #[serde(skip_serializing_if = "Option::is_none")]
319    pub key: Option<String>,
320    #[serde(skip_serializing_if = "Option::is_none")]
321    pub value: Option<String>,
322
323    // ── Collection + document targeting ──────────────────────
324    #[serde(skip_serializing_if = "Option::is_none")]
325    pub collection: Option<String>,
326    #[serde(skip_serializing_if = "Option::is_none")]
327    pub document_id: Option<String>,
328    #[serde(skip_serializing_if = "Option::is_none")]
329    pub data: Option<Vec<u8>>,
330
331    // ── Vector search ────────────────────────────────────────
332    #[serde(skip_serializing_if = "Option::is_none")]
333    pub query_vector: Option<Vec<f32>>,
334    #[serde(skip_serializing_if = "Option::is_none")]
335    pub top_k: Option<u64>,
336
337    // ── Range scan ───────────────────────────────────────────
338    #[serde(skip_serializing_if = "Option::is_none")]
339    pub field: Option<String>,
340    #[serde(skip_serializing_if = "Option::is_none")]
341    pub limit: Option<u64>,
342
343    // ── CRDT ─────────────────────────────────────────────────
344    #[serde(skip_serializing_if = "Option::is_none")]
345    pub delta: Option<Vec<u8>>,
346    #[serde(skip_serializing_if = "Option::is_none")]
347    pub peer_id: Option<u64>,
348
349    // ── Graph RAG fusion ─────────────────────────────────────
350    #[serde(skip_serializing_if = "Option::is_none")]
351    pub vector_top_k: Option<u64>,
352    #[serde(skip_serializing_if = "Option::is_none")]
353    pub edge_label: Option<String>,
354    #[serde(skip_serializing_if = "Option::is_none")]
355    pub direction: Option<String>,
356    #[serde(skip_serializing_if = "Option::is_none")]
357    pub expansion_depth: Option<u64>,
358    #[serde(skip_serializing_if = "Option::is_none")]
359    pub final_top_k: Option<u64>,
360    #[serde(skip_serializing_if = "Option::is_none")]
361    pub vector_k: Option<f64>,
362    #[serde(skip_serializing_if = "Option::is_none")]
363    pub graph_k: Option<f64>,
364
365    // ── Graph ops ────────────────────────────────────────────
366    #[serde(skip_serializing_if = "Option::is_none")]
367    pub start_node: Option<String>,
368    #[serde(skip_serializing_if = "Option::is_none")]
369    pub end_node: Option<String>,
370    #[serde(skip_serializing_if = "Option::is_none")]
371    pub depth: Option<u64>,
372    #[serde(skip_serializing_if = "Option::is_none")]
373    pub from_node: Option<String>,
374    #[serde(skip_serializing_if = "Option::is_none")]
375    pub to_node: Option<String>,
376    #[serde(skip_serializing_if = "Option::is_none")]
377    pub edge_type: Option<String>,
378    #[serde(skip_serializing_if = "Option::is_none")]
379    pub properties: Option<serde_json::Value>,
380
381    // ── Text/Hybrid search ───────────────────────────────────
382    #[serde(skip_serializing_if = "Option::is_none")]
383    pub query_text: Option<String>,
384    #[serde(skip_serializing_if = "Option::is_none")]
385    pub vector_weight: Option<f64>,
386    #[serde(skip_serializing_if = "Option::is_none")]
387    pub fuzzy: Option<bool>,
388
389    // ── Vector search tuning ────────────────────────────────
390    #[serde(skip_serializing_if = "Option::is_none")]
391    pub ef_search: Option<u64>,
392    /// Named vector field (for multi-field vector collections).
393    #[serde(skip_serializing_if = "Option::is_none")]
394    pub field_name: Option<String>,
395
396    // ── Range scan bounds ───────────────────────────────────
397    #[serde(skip_serializing_if = "Option::is_none")]
398    pub lower_bound: Option<Vec<u8>>,
399    #[serde(skip_serializing_if = "Option::is_none")]
400    pub upper_bound: Option<Vec<u8>>,
401
402    // ── CRDT dedup ──────────────────────────────────────────
403    /// Monotonic mutation ID for CRDT delta deduplication.
404    #[serde(skip_serializing_if = "Option::is_none")]
405    pub mutation_id: Option<u64>,
406
407    // ── Batch operations ─────────────────────────────────────
408    #[serde(skip_serializing_if = "Option::is_none")]
409    pub vectors: Option<Vec<BatchVector>>,
410    #[serde(skip_serializing_if = "Option::is_none")]
411    pub documents: Option<Vec<BatchDocument>>,
412
413    // ── Spatial scan ───────────────────────────────────────────
414    /// Query geometry as GeoJSON bytes (for SpatialScan).
415    #[serde(skip_serializing_if = "Option::is_none")]
416    pub query_geometry: Option<Vec<u8>>,
417    /// Spatial predicate name: "dwithin", "contains", "intersects", "within".
418    #[serde(skip_serializing_if = "Option::is_none")]
419    pub spatial_predicate: Option<String>,
420    /// Distance threshold in meters (for ST_DWithin).
421    #[serde(skip_serializing_if = "Option::is_none")]
422    pub distance_meters: Option<f64>,
423
424    // ── Timeseries ───────────────────────────────────────────
425    /// ILP payload bytes (for TimeseriesIngest).
426    #[serde(skip_serializing_if = "Option::is_none")]
427    pub payload: Option<Vec<u8>>,
428    /// Ingest format (default: "ilp").
429    #[serde(skip_serializing_if = "Option::is_none")]
430    pub format: Option<String>,
431    /// Time range start (epoch ms, for TimeseriesScan).
432    #[serde(skip_serializing_if = "Option::is_none")]
433    pub time_range_start: Option<i64>,
434    /// Time range end (epoch ms, for TimeseriesScan).
435    #[serde(skip_serializing_if = "Option::is_none")]
436    pub time_range_end: Option<i64>,
437    /// Bucket interval string for time_bucket aggregation (e.g., "5m").
438    #[serde(skip_serializing_if = "Option::is_none")]
439    pub bucket_interval: Option<String>,
440
441    // ── KV advanced ─────────────────────────────────────────
442    /// TTL in milliseconds (for KvExpire, KvBatchPut).
443    #[serde(skip_serializing_if = "Option::is_none")]
444    pub ttl_ms: Option<u64>,
445    /// Cursor bytes for KvScan pagination.
446    #[serde(skip_serializing_if = "Option::is_none")]
447    pub cursor: Option<Vec<u8>>,
448    /// Glob pattern for KvScan key matching.
449    #[serde(skip_serializing_if = "Option::is_none")]
450    pub match_pattern: Option<String>,
451    /// Multiple keys for BatchGet / Delete.
452    #[serde(skip_serializing_if = "Option::is_none")]
453    pub keys: Option<Vec<Vec<u8>>>,
454    /// Key-value entries for BatchPut: [(key, value), ...].
455    #[serde(skip_serializing_if = "Option::is_none")]
456    pub entries: Option<Vec<(Vec<u8>, Vec<u8>)>>,
457    /// Field names for FieldGet.
458    #[serde(skip_serializing_if = "Option::is_none")]
459    pub fields: Option<Vec<String>>,
460
461    // ── KV atomic operations ────────────────────────────────
462    /// Integer delta for KvIncr.
463    #[serde(skip_serializing_if = "Option::is_none")]
464    pub incr_delta: Option<i64>,
465    /// Float delta for KvIncrFloat.
466    #[serde(skip_serializing_if = "Option::is_none")]
467    pub incr_float_delta: Option<f64>,
468    /// Expected value for KvCas.
469    #[serde(skip_serializing_if = "Option::is_none")]
470    pub expected: Option<Vec<u8>>,
471    /// New value for KvCas / KvGetSet.
472    #[serde(skip_serializing_if = "Option::is_none")]
473    pub new_value: Option<Vec<u8>>,
474
475    // ── KV sorted index operations ──────────────────────────
476    /// Sorted index name.
477    #[serde(skip_serializing_if = "Option::is_none")]
478    pub index_name: Option<String>,
479    /// Sort columns: [(column_name, direction), ...].
480    #[serde(skip_serializing_if = "Option::is_none")]
481    pub sort_columns: Option<Vec<(String, String)>>,
482    /// Primary key column for sorted index.
483    #[serde(skip_serializing_if = "Option::is_none")]
484    pub key_column: Option<String>,
485    /// Window type for sorted index: "none", "daily", "weekly", "monthly", "custom".
486    #[serde(skip_serializing_if = "Option::is_none")]
487    pub window_type: Option<String>,
488    /// Timestamp column for windowed sorted index.
489    #[serde(skip_serializing_if = "Option::is_none")]
490    pub window_timestamp_column: Option<String>,
491    /// Custom window start (ms since epoch).
492    #[serde(skip_serializing_if = "Option::is_none")]
493    pub window_start_ms: Option<u64>,
494    /// Custom window end (ms since epoch).
495    #[serde(skip_serializing_if = "Option::is_none")]
496    pub window_end_ms: Option<u64>,
497    /// Top-K value for SortedIndexTopK.
498    #[serde(skip_serializing_if = "Option::is_none")]
499    pub top_k_count: Option<u32>,
500    /// Score min for SortedIndexRange (encoded bytes).
501    #[serde(skip_serializing_if = "Option::is_none")]
502    pub score_min: Option<Vec<u8>>,
503    /// Score max for SortedIndexRange (encoded bytes).
504    #[serde(skip_serializing_if = "Option::is_none")]
505    pub score_max: Option<Vec<u8>>,
506
507    // ── Document advanced ───────────────────────────────────
508    /// Field-level updates: [(field_name, value_bytes), ...].
509    #[serde(skip_serializing_if = "Option::is_none")]
510    pub updates: Option<Vec<(String, Vec<u8>)>>,
511    /// Serialized filter predicates (MessagePack).
512    #[serde(skip_serializing_if = "Option::is_none")]
513    pub filters: Option<Vec<u8>>,
514
515    // ── Vector advanced ─────────────────────────────────────
516    /// Single vector embedding (for VectorInsert).
517    #[serde(skip_serializing_if = "Option::is_none")]
518    pub vector: Option<Vec<f32>>,
519    /// Vector ID for deletion.
520    #[serde(skip_serializing_if = "Option::is_none")]
521    pub vector_id: Option<u32>,
522
523    // ── Collection policy ────────────────────────────────────
524    #[serde(skip_serializing_if = "Option::is_none")]
525    pub policy: Option<serde_json::Value>,
526
527    // ── Graph algorithm/match ───────────────────────────────
528    /// Algorithm name for GraphAlgo (e.g., "pagerank", "wcc", "sssp").
529    #[serde(skip_serializing_if = "Option::is_none")]
530    pub algorithm: Option<String>,
531    /// Cypher-subset MATCH query string for GraphMatch.
532    #[serde(skip_serializing_if = "Option::is_none")]
533    pub match_query: Option<String>,
534    /// Algorithm-specific parameters (JSON object).
535    #[serde(skip_serializing_if = "Option::is_none")]
536    pub algo_params: Option<serde_json::Value>,
537
538    // ── Document DDL ────────────────────────────────────────
539    /// Index paths for DocumentRegister.
540    #[serde(skip_serializing_if = "Option::is_none")]
541    pub index_paths: Option<Vec<String>>,
542    /// Source collection for InsertSelect.
543    #[serde(skip_serializing_if = "Option::is_none")]
544    pub source_collection: Option<String>,
545
546    // ── KV DDL ──────────────────────────────────────────────
547    /// Field position in tuple for KvRegisterIndex.
548    #[serde(skip_serializing_if = "Option::is_none")]
549    pub field_position: Option<u64>,
550    /// Whether to backfill existing keys on index creation.
551    #[serde(skip_serializing_if = "Option::is_none")]
552    pub backfill: Option<bool>,
553
554    // ── Vector DDL ──────────────────────────────────────────
555    /// HNSW M parameter (max connections per layer).
556    #[serde(skip_serializing_if = "Option::is_none")]
557    pub m: Option<u64>,
558    /// HNSW ef_construction parameter.
559    #[serde(skip_serializing_if = "Option::is_none")]
560    pub ef_construction: Option<u64>,
561    /// Distance metric name ("cosine", "euclidean", "dot").
562    #[serde(skip_serializing_if = "Option::is_none")]
563    pub metric: Option<String>,
564    /// Index type ("hnsw", "ivfpq", "flat").
565    #[serde(skip_serializing_if = "Option::is_none")]
566    pub index_type: Option<String>,
567}
568
569impl zerompk::ToMessagePack for TextFields {
570    fn write<W: zerompk::Write>(&self, writer: &mut W) -> zerompk::Result<()> {
571        use crate::json_msgpack::JsonValue;
572        writer.write_array_len(81)?;
573        self.auth.write(writer)?;
574        self.sql.write(writer)?;
575        self.key.write(writer)?;
576        self.value.write(writer)?;
577        self.collection.write(writer)?;
578        self.document_id.write(writer)?;
579        self.data.write(writer)?;
580        self.query_vector.write(writer)?;
581        self.top_k.write(writer)?;
582        self.field.write(writer)?;
583        self.limit.write(writer)?;
584        self.delta.write(writer)?;
585        self.peer_id.write(writer)?;
586        self.vector_top_k.write(writer)?;
587        self.edge_label.write(writer)?;
588        self.direction.write(writer)?;
589        self.expansion_depth.write(writer)?;
590        self.final_top_k.write(writer)?;
591        self.vector_k.write(writer)?;
592        self.graph_k.write(writer)?;
593        self.start_node.write(writer)?;
594        self.end_node.write(writer)?;
595        self.depth.write(writer)?;
596        self.from_node.write(writer)?;
597        self.to_node.write(writer)?;
598        self.edge_type.write(writer)?;
599        self.properties
600            .as_ref()
601            .map(|v| JsonValue(v.clone()))
602            .write(writer)?;
603        self.query_text.write(writer)?;
604        self.vector_weight.write(writer)?;
605        self.fuzzy.write(writer)?;
606        self.ef_search.write(writer)?;
607        self.field_name.write(writer)?;
608        self.lower_bound.write(writer)?;
609        self.upper_bound.write(writer)?;
610        self.mutation_id.write(writer)?;
611        self.vectors.write(writer)?;
612        self.documents.write(writer)?;
613        self.query_geometry.write(writer)?;
614        self.spatial_predicate.write(writer)?;
615        self.distance_meters.write(writer)?;
616        self.payload.write(writer)?;
617        self.format.write(writer)?;
618        self.time_range_start.write(writer)?;
619        self.time_range_end.write(writer)?;
620        self.bucket_interval.write(writer)?;
621        self.ttl_ms.write(writer)?;
622        self.cursor.write(writer)?;
623        self.match_pattern.write(writer)?;
624        self.keys.write(writer)?;
625        self.entries.write(writer)?;
626        self.fields.write(writer)?;
627        self.incr_delta.write(writer)?;
628        self.incr_float_delta.write(writer)?;
629        self.expected.write(writer)?;
630        self.new_value.write(writer)?;
631        self.index_name.write(writer)?;
632        self.sort_columns.write(writer)?;
633        self.key_column.write(writer)?;
634        self.window_type.write(writer)?;
635        self.window_timestamp_column.write(writer)?;
636        self.window_start_ms.write(writer)?;
637        self.window_end_ms.write(writer)?;
638        self.top_k_count.write(writer)?;
639        self.score_min.write(writer)?;
640        self.score_max.write(writer)?;
641        self.updates.write(writer)?;
642        self.filters.write(writer)?;
643        self.vector.write(writer)?;
644        self.vector_id.write(writer)?;
645        self.policy
646            .as_ref()
647            .map(|v| JsonValue(v.clone()))
648            .write(writer)?;
649        self.algorithm.write(writer)?;
650        self.match_query.write(writer)?;
651        self.algo_params
652            .as_ref()
653            .map(|v| JsonValue(v.clone()))
654            .write(writer)?;
655        self.index_paths.write(writer)?;
656        self.source_collection.write(writer)?;
657        self.field_position.write(writer)?;
658        self.backfill.write(writer)?;
659        self.m.write(writer)?;
660        self.ef_construction.write(writer)?;
661        self.metric.write(writer)?;
662        self.index_type.write(writer)
663    }
664}
665
666impl<'a> zerompk::FromMessagePack<'a> for TextFields {
667    fn read<R: zerompk::Read<'a>>(reader: &mut R) -> zerompk::Result<Self> {
668        use crate::json_msgpack::JsonValue;
669        let len = reader.read_array_len()?;
670        if len != 81 {
671            return Err(zerompk::Error::ArrayLengthMismatch {
672                expected: 81,
673                actual: len,
674            });
675        }
676        Ok(Self {
677            auth: Option::<AuthMethod>::read(reader)?,
678            sql: Option::<String>::read(reader)?,
679            key: Option::<String>::read(reader)?,
680            value: Option::<String>::read(reader)?,
681            collection: Option::<String>::read(reader)?,
682            document_id: Option::<String>::read(reader)?,
683            data: Option::<Vec<u8>>::read(reader)?,
684            query_vector: Option::<Vec<f32>>::read(reader)?,
685            top_k: Option::<u64>::read(reader)?,
686            field: Option::<String>::read(reader)?,
687            limit: Option::<u64>::read(reader)?,
688            delta: Option::<Vec<u8>>::read(reader)?,
689            peer_id: Option::<u64>::read(reader)?,
690            vector_top_k: Option::<u64>::read(reader)?,
691            edge_label: Option::<String>::read(reader)?,
692            direction: Option::<String>::read(reader)?,
693            expansion_depth: Option::<u64>::read(reader)?,
694            final_top_k: Option::<u64>::read(reader)?,
695            vector_k: Option::<f64>::read(reader)?,
696            graph_k: Option::<f64>::read(reader)?,
697            start_node: Option::<String>::read(reader)?,
698            end_node: Option::<String>::read(reader)?,
699            depth: Option::<u64>::read(reader)?,
700            from_node: Option::<String>::read(reader)?,
701            to_node: Option::<String>::read(reader)?,
702            edge_type: Option::<String>::read(reader)?,
703            properties: Option::<JsonValue>::read(reader)?.map(|v| v.0),
704            query_text: Option::<String>::read(reader)?,
705            vector_weight: Option::<f64>::read(reader)?,
706            fuzzy: Option::<bool>::read(reader)?,
707            ef_search: Option::<u64>::read(reader)?,
708            field_name: Option::<String>::read(reader)?,
709            lower_bound: Option::<Vec<u8>>::read(reader)?,
710            upper_bound: Option::<Vec<u8>>::read(reader)?,
711            mutation_id: Option::<u64>::read(reader)?,
712            vectors: Option::<Vec<BatchVector>>::read(reader)?,
713            documents: Option::<Vec<BatchDocument>>::read(reader)?,
714            query_geometry: Option::<Vec<u8>>::read(reader)?,
715            spatial_predicate: Option::<String>::read(reader)?,
716            distance_meters: Option::<f64>::read(reader)?,
717            payload: Option::<Vec<u8>>::read(reader)?,
718            format: Option::<String>::read(reader)?,
719            time_range_start: Option::<i64>::read(reader)?,
720            time_range_end: Option::<i64>::read(reader)?,
721            bucket_interval: Option::<String>::read(reader)?,
722            ttl_ms: Option::<u64>::read(reader)?,
723            cursor: Option::<Vec<u8>>::read(reader)?,
724            match_pattern: Option::<String>::read(reader)?,
725            keys: Option::<Vec<Vec<u8>>>::read(reader)?,
726            entries: Option::<Vec<(Vec<u8>, Vec<u8>)>>::read(reader)?,
727            fields: Option::<Vec<String>>::read(reader)?,
728            incr_delta: Option::<i64>::read(reader)?,
729            incr_float_delta: Option::<f64>::read(reader)?,
730            expected: Option::<Vec<u8>>::read(reader)?,
731            new_value: Option::<Vec<u8>>::read(reader)?,
732            index_name: Option::<String>::read(reader)?,
733            sort_columns: Option::<Vec<(String, String)>>::read(reader)?,
734            key_column: Option::<String>::read(reader)?,
735            window_type: Option::<String>::read(reader)?,
736            window_timestamp_column: Option::<String>::read(reader)?,
737            window_start_ms: Option::<u64>::read(reader)?,
738            window_end_ms: Option::<u64>::read(reader)?,
739            top_k_count: Option::<u32>::read(reader)?,
740            score_min: Option::<Vec<u8>>::read(reader)?,
741            score_max: Option::<Vec<u8>>::read(reader)?,
742            updates: Option::<Vec<(String, Vec<u8>)>>::read(reader)?,
743            filters: Option::<Vec<u8>>::read(reader)?,
744            vector: Option::<Vec<f32>>::read(reader)?,
745            vector_id: Option::<u32>::read(reader)?,
746            policy: Option::<JsonValue>::read(reader)?.map(|v| v.0),
747            algorithm: Option::<String>::read(reader)?,
748            match_query: Option::<String>::read(reader)?,
749            algo_params: Option::<JsonValue>::read(reader)?.map(|v| v.0),
750            index_paths: Option::<Vec<String>>::read(reader)?,
751            source_collection: Option::<String>::read(reader)?,
752            field_position: Option::<u64>::read(reader)?,
753            backfill: Option::<bool>::read(reader)?,
754            m: Option::<u64>::read(reader)?,
755            ef_construction: Option::<u64>::read(reader)?,
756            metric: Option::<String>::read(reader)?,
757            index_type: Option::<String>::read(reader)?,
758        })
759    }
760}
761
762/// A single vector in a batch insert.
763#[derive(Debug, Clone, Serialize, Deserialize)]
764pub struct BatchVector {
765    pub id: String,
766    pub embedding: Vec<f32>,
767    #[serde(skip_serializing_if = "Option::is_none")]
768    pub metadata: Option<serde_json::Value>,
769}
770
771impl zerompk::ToMessagePack for BatchVector {
772    fn write<W: zerompk::Write>(&self, writer: &mut W) -> zerompk::Result<()> {
773        use crate::json_msgpack::JsonValue;
774        writer.write_array_len(3)?;
775        writer.write_string(&self.id)?;
776        self.embedding.write(writer)?;
777        self.metadata
778            .as_ref()
779            .map(|v| JsonValue(v.clone()))
780            .write(writer)
781    }
782}
783
784impl<'a> zerompk::FromMessagePack<'a> for BatchVector {
785    fn read<R: zerompk::Read<'a>>(reader: &mut R) -> zerompk::Result<Self> {
786        use crate::json_msgpack::JsonValue;
787        let len = reader.read_array_len()?;
788        if len != 3 {
789            return Err(zerompk::Error::ArrayLengthMismatch {
790                expected: 3,
791                actual: len,
792            });
793        }
794        let id = reader.read_string()?.into_owned();
795        let embedding = Vec::<f32>::read(reader)?;
796        let metadata = Option::<JsonValue>::read(reader)?.map(|v| v.0);
797        Ok(Self {
798            id,
799            embedding,
800            metadata,
801        })
802    }
803}
804
805/// A single document in a batch insert.
806#[derive(Debug, Clone, Serialize, Deserialize)]
807pub struct BatchDocument {
808    pub id: String,
809    pub fields: serde_json::Value,
810}
811
812impl zerompk::ToMessagePack for BatchDocument {
813    fn write<W: zerompk::Write>(&self, writer: &mut W) -> zerompk::Result<()> {
814        use crate::json_msgpack::JsonValue;
815        writer.write_array_len(2)?;
816        writer.write_string(&self.id)?;
817        JsonValue(self.fields.clone()).write(writer)
818    }
819}
820
821impl<'a> zerompk::FromMessagePack<'a> for BatchDocument {
822    fn read<R: zerompk::Read<'a>>(reader: &mut R) -> zerompk::Result<Self> {
823        use crate::json_msgpack::JsonValue;
824        let len = reader.read_array_len()?;
825        if len != 2 {
826            return Err(zerompk::Error::ArrayLengthMismatch {
827                expected: 2,
828                actual: len,
829            });
830        }
831        let id = reader.read_string()?.into_owned();
832        let fields = JsonValue::read(reader)?.0;
833        Ok(Self { id, fields })
834    }
835}
836
837// ─── Response Frame ─────────────────────────────────────────────────
838
839/// A response sent from server to client over the native protocol.
840#[derive(
841    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
842)]
843pub struct NativeResponse {
844    /// Echoed from the request for correlation.
845    pub seq: u64,
846    /// Execution outcome.
847    pub status: ResponseStatus,
848    /// Column names (for query results).
849    #[serde(skip_serializing_if = "Option::is_none")]
850    pub columns: Option<Vec<String>>,
851    /// Row data (for query results). Each row is a Vec of Values.
852    #[serde(skip_serializing_if = "Option::is_none")]
853    pub rows: Option<Vec<Vec<Value>>>,
854    /// Number of rows affected (for writes).
855    #[serde(skip_serializing_if = "Option::is_none")]
856    pub rows_affected: Option<u64>,
857    /// WAL LSN watermark at time of computation.
858    pub watermark_lsn: u64,
859    /// Error details (if status == Error).
860    #[serde(skip_serializing_if = "Option::is_none")]
861    pub error: Option<ErrorPayload>,
862    /// Auth response (if op == Auth and status == Ok).
863    #[serde(skip_serializing_if = "Option::is_none")]
864    pub auth: Option<AuthResponse>,
865}
866
867/// Error details in a response.
868#[derive(
869    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
870)]
871pub struct ErrorPayload {
872    /// SQLSTATE-style error code (e.g., "42P01" for undefined table).
873    pub code: String,
874    /// Human-readable error message.
875    pub message: String,
876}
877
878impl NativeResponse {
879    /// Create a successful response with no data.
880    pub fn ok(seq: u64) -> Self {
881        Self {
882            seq,
883            status: ResponseStatus::Ok,
884            columns: None,
885            rows: None,
886            rows_affected: None,
887            watermark_lsn: 0,
888            error: None,
889            auth: None,
890        }
891    }
892
893    /// Create a successful response from a `QueryResult`.
894    pub fn from_query_result(seq: u64, qr: crate::result::QueryResult, lsn: u64) -> Self {
895        Self {
896            seq,
897            status: ResponseStatus::Ok,
898            columns: Some(qr.columns),
899            rows: Some(qr.rows),
900            rows_affected: Some(qr.rows_affected),
901            watermark_lsn: lsn,
902            error: None,
903            auth: None,
904        }
905    }
906
907    /// Create an error response.
908    pub fn error(seq: u64, code: impl Into<String>, message: impl Into<String>) -> Self {
909        Self {
910            seq,
911            status: ResponseStatus::Error,
912            columns: None,
913            rows: None,
914            rows_affected: None,
915            watermark_lsn: 0,
916            error: Some(ErrorPayload {
917                code: code.into(),
918                message: message.into(),
919            }),
920            auth: None,
921        }
922    }
923
924    /// Create an auth success response.
925    pub fn auth_ok(seq: u64, username: String, tenant_id: u32) -> Self {
926        Self {
927            seq,
928            status: ResponseStatus::Ok,
929            columns: None,
930            rows: None,
931            rows_affected: None,
932            watermark_lsn: 0,
933            error: None,
934            auth: Some(AuthResponse {
935                username,
936                tenant_id,
937            }),
938        }
939    }
940
941    /// Create a response with a single "status" column and one row.
942    pub fn status_row(seq: u64, message: impl Into<String>) -> Self {
943        Self {
944            seq,
945            status: ResponseStatus::Ok,
946            columns: Some(vec!["status".into()]),
947            rows: Some(vec![vec![Value::String(message.into())]]),
948            rows_affected: Some(1),
949            watermark_lsn: 0,
950            error: None,
951            auth: None,
952        }
953    }
954}
955
956// ─── Protocol Constants ─────────────────────────────────────────────
957
958/// Maximum frame payload size (16 MiB).
959pub const MAX_FRAME_SIZE: u32 = 16 * 1024 * 1024;
960
961/// Length of the frame header (4-byte big-endian u32 payload length).
962pub const FRAME_HEADER_LEN: usize = 4;
963
964/// Default server port for the native protocol.
965pub const DEFAULT_NATIVE_PORT: u16 = 6433;
966
967#[cfg(test)]
968mod tests {
969    use super::*;
970
971    #[test]
972    fn opcode_repr() {
973        assert_eq!(OpCode::Auth as u8, 0x01);
974        assert_eq!(OpCode::Sql as u8, 0x20);
975        assert_eq!(OpCode::Begin as u8, 0x40);
976        assert_eq!(OpCode::GraphHop as u8, 0x50);
977        assert_eq!(OpCode::TextSearch as u8, 0x60);
978        assert_eq!(OpCode::VectorBatchInsert as u8, 0x70);
979    }
980
981    #[test]
982    fn opcode_is_write() {
983        assert!(OpCode::PointPut.is_write());
984        assert!(OpCode::PointDelete.is_write());
985        assert!(OpCode::CrdtApply.is_write());
986        assert!(OpCode::EdgePut.is_write());
987        assert!(!OpCode::PointGet.is_write());
988        assert!(!OpCode::Sql.is_write());
989        assert!(!OpCode::VectorSearch.is_write());
990        assert!(!OpCode::Ping.is_write());
991    }
992
993    #[test]
994    fn response_status_repr() {
995        assert_eq!(ResponseStatus::Ok as u8, 0);
996        assert_eq!(ResponseStatus::Partial as u8, 1);
997        assert_eq!(ResponseStatus::Error as u8, 2);
998    }
999
1000    #[test]
1001    fn native_response_ok() {
1002        let r = NativeResponse::ok(42);
1003        assert_eq!(r.seq, 42);
1004        assert_eq!(r.status, ResponseStatus::Ok);
1005        assert!(r.error.is_none());
1006    }
1007
1008    #[test]
1009    fn native_response_error() {
1010        let r = NativeResponse::error(1, "42P01", "collection not found");
1011        assert_eq!(r.status, ResponseStatus::Error);
1012        let e = r.error.unwrap();
1013        assert_eq!(e.code, "42P01");
1014        assert_eq!(e.message, "collection not found");
1015    }
1016
1017    #[test]
1018    fn native_response_from_query_result() {
1019        let qr = crate::result::QueryResult {
1020            columns: vec!["id".into(), "name".into()],
1021            rows: vec![vec![
1022                Value::String("u1".into()),
1023                Value::String("Alice".into()),
1024            ]],
1025            rows_affected: 0,
1026        };
1027        let r = NativeResponse::from_query_result(5, qr, 100);
1028        assert_eq!(r.seq, 5);
1029        assert_eq!(r.watermark_lsn, 100);
1030        assert_eq!(r.columns.as_ref().unwrap().len(), 2);
1031        assert_eq!(r.rows.as_ref().unwrap().len(), 1);
1032    }
1033
1034    #[test]
1035    fn native_response_status_row() {
1036        let r = NativeResponse::status_row(3, "OK");
1037        assert_eq!(r.columns.as_ref().unwrap(), &["status"]);
1038        assert_eq!(r.rows.as_ref().unwrap()[0][0].as_str(), Some("OK"));
1039    }
1040
1041    #[test]
1042    fn msgpack_roundtrip_request() {
1043        let req = NativeRequest {
1044            op: OpCode::Sql,
1045            seq: 1,
1046            fields: RequestFields::Text(TextFields {
1047                sql: Some("SELECT 1".into()),
1048                ..Default::default()
1049            }),
1050        };
1051        let bytes = zerompk::to_msgpack_vec(&req).unwrap();
1052        let decoded: NativeRequest = zerompk::from_msgpack(&bytes).unwrap();
1053        assert_eq!(decoded.op, OpCode::Sql);
1054        assert_eq!(decoded.seq, 1);
1055    }
1056
1057    #[test]
1058    fn msgpack_roundtrip_response() {
1059        let resp = NativeResponse::from_query_result(
1060            7,
1061            crate::result::QueryResult {
1062                columns: vec!["x".into()],
1063                rows: vec![vec![Value::Integer(42)]],
1064                rows_affected: 0,
1065            },
1066            99,
1067        );
1068        let bytes = zerompk::to_msgpack_vec(&resp).unwrap();
1069        let decoded: NativeResponse = zerompk::from_msgpack(&bytes).unwrap();
1070        assert_eq!(decoded.seq, 7);
1071        assert_eq!(decoded.watermark_lsn, 99);
1072        assert_eq!(decoded.rows.unwrap()[0][0].as_i64(), Some(42));
1073    }
1074
1075    #[test]
1076    fn auth_method_variants() {
1077        let trust = AuthMethod::Trust {
1078            username: "admin".into(),
1079        };
1080        let bytes = zerompk::to_msgpack_vec(&trust).unwrap();
1081        let decoded: AuthMethod = zerompk::from_msgpack(&bytes).unwrap();
1082        match decoded {
1083            AuthMethod::Trust { username } => assert_eq!(username, "admin"),
1084            _ => panic!("expected Trust variant"),
1085        }
1086
1087        let pw = AuthMethod::Password {
1088            username: "user".into(),
1089            password: "secret".into(),
1090        };
1091        let bytes = zerompk::to_msgpack_vec(&pw).unwrap();
1092        let decoded: AuthMethod = zerompk::from_msgpack(&bytes).unwrap();
1093        match decoded {
1094            AuthMethod::Password { username, password } => {
1095                assert_eq!(username, "user");
1096                assert_eq!(password, "secret");
1097            }
1098            _ => panic!("expected Password variant"),
1099        }
1100    }
1101}
nodedb_types/protocol.rs

nodedb_types/
protocol.rs
