nodedb_lite/query/document_ops/writes/
insert.rs

1// SPDX-License-Identifier: Apache-2.0
2use super::super::is_strict;
3use super::super::reads::msgpack_bytes_to_crdt_fields;
4use super::super::write_helpers::affected;
5use super::super::write_helpers::{decode_strict_fields, fields_to_values, strict_schema};
6use crate::error::LiteError;
7use crate::query::engine::LiteQueryEngine;
8use crate::query::text_index::{index_strict_rows, reindex_documents, reindex_strict_rows};
9use crate::storage::engine::StorageEngine;
10use nodedb_types::result::QueryResult;
11use nodedb_types::value::Value;
12use std::collections::HashMap;
13
14/// PointPut: unconditional overwrite (upsert semantics).
15pub async fn point_put<S: StorageEngine>(
16    engine: &LiteQueryEngine<S>,
17    collection: &str,
18    document_id: &str,
19    value_bytes: &[u8],
20) -> Result<QueryResult, LiteError> {
21    point_put_coordinated(engine, None, collection, document_id, value_bytes).await
22}
23
24pub(crate) async fn point_put_coordinated<S: StorageEngine>(
25    engine: &LiteQueryEngine<S>,
26    permit: Option<&crate::engine::fts::coordinator::TextMutationPermit>,
27    collection: &str,
28    document_id: &str,
29    value_bytes: &[u8],
30) -> Result<QueryResult, LiteError> {
31    if let Some(permit) = permit {
32        return point_put_admitted(engine, permit, collection, document_id, value_bytes).await;
33    }
34    let guard = engine.fts_state.admit_mutation().await;
35    let result =
36        point_put_admitted(engine, guard.permit(), collection, document_id, value_bytes).await;
37    guard.finish(result)
38}
39
40pub(crate) async fn point_put_admitted<S: StorageEngine>(
41    engine: &LiteQueryEngine<S>,
42    permit: &crate::engine::fts::coordinator::TextMutationPermit,
43    collection: &str,
44    document_id: &str,
45    value_bytes: &[u8],
46) -> Result<QueryResult, LiteError> {
47    let _permit = permit;
48    if is_strict(engine, collection) {
49        let fields = decode_strict_fields(value_bytes)?;
50        let existing_pk = Value::String(document_id.to_string());
51        if engine.strict.get(collection, &existing_pk).await?.is_some() {
52            let updates: HashMap<String, Value> = fields.into_iter().collect();
53            engine
54                .strict
55                .update(collection, &existing_pk, &updates)
56                .await?;
57        } else {
58            let schema = strict_schema(engine, collection)?;
59            let values = fields_to_values(&fields, &schema.columns);
60            engine.strict.insert(collection, &values).await?;
61        }
62        reindex_strict_rows(engine, collection, &[existing_pk]).await?;
63    } else {
64        let crdt_fields = msgpack_bytes_to_crdt_fields(value_bytes)?;
65        let loro_fields: Vec<(&str, loro::LoroValue)> = crdt_fields
66            .iter()
67            .map(|(k, v)| (k.as_str(), v.clone()))
68            .collect();
69        engine
70            .crdt
71            .lock()
72            .map_err(|_| LiteError::LockPoisoned)?
73            .upsert(collection, document_id, &loro_fields)?;
74        reindex_documents(engine, collection, [document_id])?;
75    }
76    Ok(affected(1, "INSERT"))
77}
78
79/// PointInsert: insert-only, fail on duplicate PK (or skip if `if_absent`).
80pub async fn point_insert<S: StorageEngine>(
81    engine: &LiteQueryEngine<S>,
82    collection: &str,
83    document_id: &str,
84    value_bytes: &[u8],
85    if_absent: bool,
86) -> Result<QueryResult, LiteError> {
87    point_insert_coordinated(
88        engine,
89        None,
90        collection,
91        document_id,
92        value_bytes,
93        if_absent,
94    )
95    .await
96}
97
98pub(crate) async fn point_insert_coordinated<S: StorageEngine>(
99    engine: &LiteQueryEngine<S>,
100    permit: Option<&crate::engine::fts::coordinator::TextMutationPermit>,
101    collection: &str,
102    document_id: &str,
103    value_bytes: &[u8],
104    if_absent: bool,
105) -> Result<QueryResult, LiteError> {
106    if let Some(permit) = permit {
107        return point_insert_admitted(
108            engine,
109            permit,
110            collection,
111            document_id,
112            value_bytes,
113            if_absent,
114        )
115        .await;
116    }
117    let guard = engine.fts_state.admit_mutation().await;
118    let result = point_insert_admitted(
119        engine,
120        guard.permit(),
121        collection,
122        document_id,
123        value_bytes,
124        if_absent,
125    )
126    .await;
127    guard.finish(result)
128}
129
130pub(crate) async fn point_insert_admitted<S: StorageEngine>(
131    engine: &LiteQueryEngine<S>,
132    permit: &crate::engine::fts::coordinator::TextMutationPermit,
133    collection: &str,
134    document_id: &str,
135    value_bytes: &[u8],
136    if_absent: bool,
137) -> Result<QueryResult, LiteError> {
138    let _permit = permit;
139    if is_strict(engine, collection) {
140        let pk = Value::String(document_id.to_string());
141        if engine.strict.get(collection, &pk).await?.is_some() {
142            if if_absent {
143                return Ok(affected(0, "INSERT"));
144            }
145            return Err(LiteError::BadRequest {
146                detail: format!(
147                    "duplicate key value violates unique constraint on '{collection}' (id = '{document_id}')"
148                ),
149            });
150        }
151        let fields = decode_strict_fields(value_bytes)?;
152        let schema = strict_schema(engine, collection)?;
153        let values = fields_to_values(&fields, &schema.columns);
154        engine.strict.insert(collection, &values).await?;
155        index_strict_rows(engine, collection, [values.as_slice()])?;
156    } else {
157        let crdt_fields = msgpack_bytes_to_crdt_fields(value_bytes)?;
158        let loro_fields: Vec<(&str, loro::LoroValue)> = crdt_fields
159            .iter()
160            .map(|(k, v)| (k.as_str(), v.clone()))
161            .collect();
162        let mut crdt = engine.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
163        if crdt.exists(collection, document_id) {
164            if if_absent {
165                return Ok(affected(0, "INSERT"));
166            }
167            return Err(LiteError::BadRequest {
168                detail: format!(
169                    "duplicate key value violates unique constraint on '{collection}' (id = '{document_id}')"
170                ),
171            });
172        }
173        crdt.upsert(collection, document_id, &loro_fields)?;
174        drop(crdt);
175        reindex_documents(engine, collection, [document_id])?;
176    }
177    Ok(affected(1, "INSERT"))
178}
nodedb_lite/query/document_ops/writes/insert.rs

nodedb_lite/query/document_ops/writes/
insert.rs
