teaql_runtime/
entity_save.rs

1use std::collections::BTreeSet;
2use std::future::Future;
3use std::marker::PhantomData;
4use std::pin::Pin;
5use std::sync::Arc;
6
7use teaql_core::{Entity, MutationValues, Value};
8
9use crate::{DataServiceError, GraphNode, GraphOperation, RuntimeError, UserContext};
10
11// ---------------------------------------------------------------------------
12// DynGraphSaver — type-erased graph save capability
13// ---------------------------------------------------------------------------
14
15/// Object-safe trait for saving a [`GraphNode`] tree to the database.
16///
17/// A concrete implementation is registered in [`UserContext`] during setup so
18/// that [`Audited::save`] can persist entities without exposing the underlying
19/// executor type to business code.
20pub(crate) trait DynGraphSaver: Send + Sync {
21    fn save_graph_dyn<'a>(
22        &'a self,
23        context: &'a UserContext,
24        node: GraphNode,
25    ) -> Pin<Box<dyn Future<Output = Result<GraphNode, RuntimeError>> + Send + 'a>>;
26
27    fn save_ledger_dyn<'a>(
28        &'a self,
29        context: &'a UserContext,
30        node: GraphNode,
31        root: crate::EntityRuntimeState,
32    ) -> Pin<Box<dyn Future<Output = Result<GraphNode, RuntimeError>> + Send + 'a>>;
33}
34
35/// Marker struct that implements [`DynGraphSaver`] for a specific executor type `E`.
36///
37/// `E` is the full executor type (e.g. `SqlDataServiceExecutor<SqliteDialect, …>`).
38/// The struct itself is zero-sized; the actual executor is retrieved from
39/// [`UserContext`] at call time.
40pub(crate) struct GraphSaverFor<E> {
41    _marker: PhantomData<fn() -> E>,
42}
43
44impl<E> GraphSaverFor<E> {
45    pub(crate) fn new() -> Self {
46        Self {
47            _marker: PhantomData,
48        }
49    }
50}
51
52impl<E> DynGraphSaver for GraphSaverFor<E>
53where
54    E: teaql_data_service::QueryExecutor
55        + teaql_data_service::MutationExecutor
56        + Send
57        + Sync
58        + 'static,
59{
60    fn save_graph_dyn<'a>(
61        &'a self,
62        context: &'a UserContext,
63        node: GraphNode,
64    ) -> Pin<Box<dyn Future<Output = Result<GraphNode, RuntimeError>> + Send + 'a>> {
65        Box::pin(async move {
66            let entity = node.entity.clone();
67            let eds = context
68                .entity_data_service::<E>(entity)
69                .map_err(|e| RuntimeError::Graph(e.to_string()))?;
70            eds.save_graph_internal(node).await.map_err(|e| match e {
71                DataServiceError::Runtime(r) => r,
72                other => RuntimeError::Graph(other.to_string()),
73            })
74        })
75    }
76
77    fn save_ledger_dyn<'a>(
78        &'a self,
79        context: &'a UserContext,
80        mut node: GraphNode,
81        root: crate::EntityRuntimeState,
82    ) -> Pin<Box<dyn Future<Output = Result<GraphNode, RuntimeError>> + Send + 'a>> {
83        Box::pin(async move {
84            let entity = node.entity.clone();
85            let eds = context
86                .entity_data_service::<E>(&entity)
87                .map_err(|e| RuntimeError::Graph(e.to_string()))?;
88            let generated_ids = eds
89                .execute_ledger_plan_internal(root.clone())
90                .await
91                .map_err(|e| match e {
92                    DataServiceError::Runtime(r) => r,
93                    other => RuntimeError::Graph(other.to_string()),
94                })?;
95
96            let descriptor = context.require_entity(&entity).unwrap();
97            if let Some(id_prop) = descriptor.id_property() {
98                let current_id = node
99                    .values
100                    .get(&id_prop.name)
101                    .cloned()
102                    .unwrap_or(Value::I64(0));
103                let root_key = crate::EntityKey::new(entity.clone(), current_id);
104                if let Some(new_id) = generated_ids.get(&root_key) {
105                    node.values.insert(id_prop.name.clone(), new_id.clone());
106                }
107                if let Some(changes) = root.current_change_set().changes().get(&root_key) {
108                    for (field, value) in changes {
109                        node.values.insert(field.clone(), value.clone());
110                    }
111                }
112            }
113            root.clear_committed();
114            Ok(node)
115        })
116    }
117}
118
119// ---------------------------------------------------------------------------
120// Standalone graph-node extraction (no executor needed)
121// ---------------------------------------------------------------------------
122
123/// Convert a typed entity into a [`GraphNode`] tree.
124///
125/// This only requires metadata (entity descriptors) from the [`UserContext`],
126/// **not** the database executor.  It is the standalone equivalent of
127/// [`EntityDataService::graph_node_from_entity`].
128pub fn graph_node_from_entity<T: Entity>(
129    context: &UserContext,
130    entity: T,
131) -> Result<GraphNode, RuntimeError> {
132    let descriptor = T::entity_descriptor();
133    let dirty_fields = entity.dirty_fields();
134    let original_values = entity.original_values();
135    let is_new = entity.is_new();
136    let is_deleted = entity.is_marked_as_delete();
137    let comment = entity.get_comment();
138    let mut node = graph_node_from_values(context, &descriptor.name, entity.into_values())?;
139    node.dirty_fields = dirty_fields;
140    node.original_values = original_values.map(Into::into);
141    if is_new {
142        node.operation = GraphOperation::Create;
143    }
144    if is_deleted {
145        node.operation = GraphOperation::Remove;
146        node.relations.clear();
147    }
148    if let Some(c) = comment {
149        node.set_comment(c);
150    }
151    Ok(node)
152}
153
154/// Recursively convert entity mutation values into a [`GraphNode`] tree.
155///
156/// Relations are resolved via the entity descriptors stored in `context`.
157fn graph_node_from_values(
158    context: &UserContext,
159    entity: &str,
160    values: MutationValues,
161) -> Result<GraphNode, RuntimeError> {
162    let descriptor = context.require_entity(entity)?;
163    let mut node = GraphNode::new(entity);
164
165    for (field, value) in values {
166        if field == "_comment" {
167            if let Value::Text(comment) = value {
168                node.set_comment(comment);
169            }
170            continue;
171        }
172        if field == "_dirty_fields" {
173            if let Value::List(fields) = value {
174                let mut dirty = BTreeSet::new();
175                for f in fields {
176                    if let Value::Text(t) = f {
177                        dirty.insert(t);
178                    }
179                }
180                node.dirty_fields = Some(dirty);
181            }
182            continue;
183        }
184        if field == "_original_values" {
185            if let Value::Object(orig) = value {
186                node.original_values = Some(orig.into());
187            }
188            continue;
189        }
190        if field == "_is_new" {
191            if matches!(value, Value::Bool(true)) {
192                node.operation = GraphOperation::Create;
193            }
194            continue;
195        }
196        if field == "_is_deleted" {
197            if matches!(value, Value::Bool(true)) {
198                node.operation = GraphOperation::Remove;
199            }
200            continue;
201        }
202        let Some(relation) = descriptor.relation_by_name(&field) else {
203            node.values.insert(field, value);
204            continue;
205        };
206
207        match value {
208            Value::Null => {
209                node.relations.entry(field).or_default();
210            }
211            Value::Object(record) => {
212                let child =
213                    graph_node_from_values(context, &relation.target_entity, record.into())?;
214                node.relations.entry(field).or_default().push(child);
215            }
216            Value::List(values) => {
217                let children = node.relations.entry(field.clone()).or_default();
218                for value in values {
219                    let Value::Object(record) = value else {
220                        return Err(RuntimeError::Graph(format!(
221                            "relation {}.{} expects object children, got {:?}",
222                            entity, field, value
223                        )));
224                    };
225                    children.push(graph_node_from_values(
226                        context,
227                        &relation.target_entity,
228                        record.into(),
229                    )?);
230                }
231            }
232            other => {
233                return Err(RuntimeError::Graph(format!(
234                    "relation {}.{} expects object/list/null, got {:?}",
235                    entity, field, other
236                )));
237            }
238        }
239    }
240
241    Ok(node)
242}
243
244fn merge_relation_mutations_into_root(
245    root: &crate::EntityRuntimeState,
246    node: &GraphNode,
247) -> Result<(), RuntimeError> {
248    for children in node.relations.values() {
249        for child in children {
250            let id = child.values.get("id").cloned().ok_or_else(|| {
251                RuntimeError::Graph(format!(
252                    "related mutation {} is missing its id",
253                    child.entity
254                ))
255            })?;
256            let key = crate::EntityKey::new(child.entity.clone(), id);
257
258            match child.operation {
259                GraphOperation::Create => {
260                    root.mark_as_new(key.clone());
261                    for (field, value) in &child.values {
262                        root.set(key.clone(), field, value.clone());
263                    }
264                }
265                GraphOperation::Upsert => {
266                    if let Some(fields) = &child.dirty_fields {
267                        for field in fields {
268                            if let Some(value) = child.values.get(field) {
269                                root.set(key.clone(), field, value.clone());
270                            }
271                        }
272                    }
273                }
274                GraphOperation::Remove => root.mark_as_delete(key.clone()),
275                GraphOperation::Reference => {}
276            }
277
278            if let Some(version) = child
279                .original_values
280                .as_ref()
281                .and_then(|values| values.get("version"))
282                .and_then(Value::try_i64)
283            {
284                root.set_original_version(key, version);
285            }
286            merge_relation_mutations_into_root(root, child)?;
287        }
288    }
289    Ok(())
290}
291
292// ---------------------------------------------------------------------------
293// AuditedSaveExt — the `.save(&context)` method on `Audited<T>`
294// ---------------------------------------------------------------------------
295
296/// Extension trait that provides the `.save(&context)` method on [`Audited<T>`](teaql_core::Audited).
297///
298/// # Example
299/// ```ignore
300/// use teaql_runtime::AuditedSaveExt;
301///
302/// school.audit_as("创建学校").save(&context).await?;
303/// ```
304pub trait AuditedSaveExt {
305    type Entity;
306
307    fn save<'a>(
308        self,
309        context: &'a UserContext,
310    ) -> Pin<Box<dyn Future<Output = Result<Self::Entity, RuntimeError>> + Send + 'a>>;
311}
312
313impl<T> AuditedSaveExt for teaql_core::Audited<T>
314where
315    T: Entity + Send + 'static,
316{
317    type Entity = T;
318
319    fn save<'a>(
320        self,
321        context: &'a UserContext,
322    ) -> Pin<Box<dyn Future<Output = Result<Self::Entity, RuntimeError>> + Send + 'a>> {
323        Box::pin(async move {
324            let entity_name = T::entity_descriptor().name;
325            let entity = self.into_entity(); // applies comment onto the entity
326            let node = graph_node_from_entity(context, entity)?;
327            let saver = context
328                .require_resource::<Arc<dyn DynGraphSaver>>()
329                .map_err(|e| {
330                    RuntimeError::Graph(format!(
331                        "no DynGraphSaver registered — did you call register_executor()? ({})",
332                        e
333                    ))
334                })?;
335            let saved = saver.save_graph_dyn(context, node).await?;
336            T::from_compact_row(teaql_core::CompactRow::from_map(saved.values.into()))
337                .map_err(|e| RuntimeError::Graph(e.to_string()))
338        })
339    }
340}
341
342/// Persist an audited generated entity, including pending ledger changes that
343/// may span multiple related entities sharing the same [`EntityRuntimeState`](crate::EntityRuntimeState).
344///
345/// Generated service crates use this as the implementation behind
346/// `entity.audit_as("why").save(&context)`. The audited wrapper is required by the
347/// function signature; no unaudited entity write entry point is exposed.
348#[doc(hidden)]
349pub async fn save_audited_ledger_entity<T>(
350    audited: teaql_core::Audited<T>,
351    context: &UserContext,
352) -> Result<T, RuntimeError>
353where
354    T: crate::LedgerEntity + Send + 'static,
355{
356    let entity_name = T::entity_descriptor().name;
357    let entity = audited.into_entity();
358    let root = entity.entity_runtime_state();
359    let node = graph_node_from_entity(context, entity)?;
360    let saver = context
361        .require_resource::<Arc<dyn DynGraphSaver>>()
362        .map_err(|e| {
363            RuntimeError::Graph(format!(
364                "no DynGraphSaver registered — did you call register_executor()? ({e})"
365            ))
366        })?;
367
368    if let Some(root) = root {
369        merge_relation_mutations_into_root(&root, &node)?;
370        let has_ledger_changes = !root.current_change_set().changes().is_empty()
371            || !root.deleted_keys().is_empty()
372            || !root.new_keys().is_empty();
373        if has_ledger_changes {
374            let saved = saver.save_ledger_dyn(context, node, root).await?;
375            return T::from_compact_row(teaql_core::CompactRow::from_map(saved.values.into()))
376                .map_err(|e| RuntimeError::Graph(e.to_string()));
377        }
378    }
379
380    let saved = saver.save_graph_dyn(context, node).await?;
381    T::from_compact_row(teaql_core::CompactRow::from_map(saved.values.into()))
382        .map_err(|e| RuntimeError::Graph(e.to_string()))
383}
teaql_runtime/entity_save.rs

teaql_runtime/
entity_save.rs
