teaql_data_service/
lib.rs

1#![allow(async_fn_in_trait)]
2
3use std::time::SystemTime;
4use teaql_core::{
5    CompactRow, DeleteCommand, EntitySnapshot, Expr, GeneratedValues, InsertCommand,
6    RecoverCommand, SelectQuery, TraceNode, UpdateCommand,
7};
8
9#[derive(Debug, Clone, Default)]
10pub struct DataServiceCapabilities {
11    pub query: bool,
12    pub mutation: bool,
13    pub transaction: bool,
14    pub schema: bool,
15    pub id_generation: bool,
16    pub batch_mutation: bool,
17    pub returning: bool,
18    /// A small number of per-parent indexed relation queries is preferable to
19    /// a partitioned batch query for this provider (for example embedded SQLite).
20    pub small_parent_relation_probes: bool,
21}
22
23#[derive(Debug, Clone)]
24pub struct QueryRequest {
25    pub query: SelectQuery,
26    pub trace_chain: Vec<TraceNode>,
27    pub comment: Option<String>,
28    /// Build a copy-paste representation with bind values interpolated.
29    /// Runtimes should disable this when query logging is disabled.
30    pub capture_debug_query: bool,
31    /// Retain timings, parameterized text, bind values, trace and comment in the result.
32    /// Disable only when the caller will discard execution metadata.
33    pub capture_execution_metadata: bool,
34}
35
36#[derive(Debug, Clone)]
37pub struct QueryResult {
38    pub rows: Vec<CompactRow>,
39    pub metadata: ExecutionMetadata,
40}
41
42#[derive(Debug, Clone)]
43pub enum MutationRequest {
44    Insert(InsertCommand),
45    Update(UpdateCommand),
46    Delete(DeleteCommand),
47    Recover(RecoverCommand),
48    Batch(Vec<MutationRequest>),
49}
50
51impl MutationRequest {
52    pub fn trace_chain(&self) -> &[teaql_core::TraceNode] {
53        match self {
54            MutationRequest::Insert(cmd) => &cmd.trace_chain,
55            MutationRequest::Update(cmd) => &cmd.trace_chain,
56            MutationRequest::Delete(cmd) => &cmd.trace_chain,
57            MutationRequest::Recover(cmd) => &cmd.trace_chain,
58            MutationRequest::Batch(_) => &[], // Batch traces are per-item
59        }
60    }
61
62    pub fn comment(&self) -> Option<&str> {
63        match self {
64            MutationRequest::Insert(cmd) => cmd.trace_chain.last().map(|n| n.comment.as_str()),
65            MutationRequest::Update(cmd) => cmd.trace_chain.last().map(|n| n.comment.as_str()),
66            MutationRequest::Delete(cmd) => cmd.trace_chain.last().map(|n| n.comment.as_str()),
67            MutationRequest::Recover(cmd) => cmd.trace_chain.last().map(|n| n.comment.as_str()),
68            MutationRequest::Batch(_) => None,
69        }
70    }
71}
72
73#[derive(Debug, Clone)]
74pub struct MutationResult {
75    pub affected_rows: u64,
76    pub generated_values: GeneratedValues,
77    pub persisted_snapshot: Option<EntitySnapshot>,
78    pub metadata: ExecutionMetadata,
79}
80
81#[derive(Debug, Clone, Copy, PartialEq, Eq)]
82pub enum DataServiceOperation {
83    Query,
84    Insert,
85    Update,
86    Delete,
87    Recover,
88    Batch,
89    Schema,
90}
91
92#[derive(Debug, Clone)]
93pub struct ExecutionMetadata {
94    pub backend: String,
95    pub operation: DataServiceOperation,
96    pub started_at: SystemTime,
97    pub ended_at: SystemTime,
98    pub affected_rows: Option<u64>,
99    pub result_count: Option<usize>,
100    pub trace_chain: Vec<TraceNode>,
101    pub comment: Option<String>,
102    pub backend_request_id: Option<String>,
103    /// Provider-native parameterized request text. For SQL this contains
104    /// placeholders and never interpolated bind values.
105    pub parameterized_query: Option<String>,
106    /// Structured bind values corresponding to `parameterized_query`.
107    pub params: Vec<teaql_core::Value>,
108    pub debug_query: Option<String>,
109}
110
111impl ExecutionMetadata {
112    pub fn unrecorded_query(result_count: usize) -> Self {
113        Self {
114            backend: String::new(),
115            operation: DataServiceOperation::Query,
116            started_at: SystemTime::UNIX_EPOCH,
117            ended_at: SystemTime::UNIX_EPOCH,
118            affected_rows: None,
119            result_count: Some(result_count),
120            trace_chain: Vec::new(),
121            comment: None,
122            backend_request_id: None,
123            parameterized_query: None,
124            params: Vec::new(),
125            debug_query: None,
126        }
127    }
128}
129
130pub trait DataServiceExecutor {
131    type Error: std::error::Error + Send + Sync + 'static;
132
133    fn capabilities(&self) -> DataServiceCapabilities;
134}
135
136pub trait QueryExecutor: DataServiceExecutor {
137    fn query(
138        &self,
139        request: QueryRequest,
140    ) -> impl std::future::Future<Output = Result<QueryResult, Self::Error>> + Send;
141}
142
143/// Result of a single streaming chunk.
144#[derive(Debug, Clone)]
145pub struct StreamChunk {
146    pub rows: Vec<CompactRow>,
147    pub chunk_index: usize,
148    pub is_last: bool,
149}
150
151pub type QueryStream<'a, E> =
152    std::pin::Pin<Box<dyn futures_core::Stream<Item = Result<StreamChunk, E>> + 'a>>;
153
154/// Streaming query executor. Returns rows in chunks rather than all at once.
155pub trait StreamQueryExecutor: DataServiceExecutor {
156    fn query_stream(
157        &self,
158        request: QueryRequest,
159        chunk_size: usize,
160    ) -> QueryStream<'_, Self::Error>;
161}
162
163pub trait MutationExecutor: DataServiceExecutor {
164    fn mutate(
165        &self,
166        request: MutationRequest,
167    ) -> impl std::future::Future<Output = Result<MutationResult, Self::Error>> + Send;
168}
169
170/// A mutation whose target must also satisfy a trusted, datastore-enforced guard.
171///
172/// The guard is deliberately kept outside [`MutationRequest`]: ordinary domain
173/// mutations do not own authorization policy, while boundary adapters such as a
174/// federated endpoint must be able to require one atomic mutation predicate.
175#[derive(Debug, Clone)]
176pub struct GuardedMutationRequest {
177    pub mutation: MutationRequest,
178    pub guard: Expr,
179}
180
181impl GuardedMutationRequest {
182    pub fn new(mutation: MutationRequest, guard: Expr) -> Self {
183        Self { mutation, guard }
184    }
185}
186
187/// Executes a mutation only when its target also satisfies a trusted guard.
188///
189/// Implementations must apply the guard atomically in the same datastore
190/// statement as the mutation. A separate read-before-write check does not
191/// satisfy this contract.
192pub trait GuardedMutationExecutor: MutationExecutor {
193    fn mutate_guarded(
194        &self,
195        request: GuardedMutationRequest,
196    ) -> impl std::future::Future<Output = Result<MutationResult, Self::Error>> + Send;
197}
198
199pub trait TransactionExecutor: DataServiceExecutor {
200    type Tx<'a>: QueryExecutor<Error = Self::Error>
201        + MutationExecutor<Error = Self::Error>
202        + Transaction<Error = Self::Error>
203    where
204        Self: 'a;
205
206    fn begin(&self) -> impl std::future::Future<Output = Result<Self::Tx<'_>, Self::Error>> + Send;
207}
208
209pub trait Transaction {
210    type Error: std::error::Error + Send + Sync + 'static;
211
212    fn commit(self) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
213    fn rollback(self) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
214}
215
216#[derive(Debug, Clone)]
217pub struct SchemaRequest {
218    pub entity_name: String,
219}
220
221#[derive(Debug, Clone)]
222pub struct SchemaResult {
223    pub changed: bool,
224}
225
226pub trait SchemaExecutor: DataServiceExecutor {
227    fn ensure_schema(
228        &self,
229        request: SchemaRequest,
230    ) -> impl std::future::Future<Output = Result<SchemaResult, Self::Error>> + Send;
231}
232
233pub trait IdGeneratorExecutor: DataServiceExecutor {
234    fn next_id(
235        &self,
236        entity: &str,
237    ) -> impl std::future::Future<Output = Result<u64, Self::Error>> + Send;
238}
239
240pub trait SchemaProvider: Send + Sync {
241    fn get_entity(&self, name: &str) -> Option<std::sync::Arc<teaql_core::EntityDescriptor>>;
242}
243
244#[cfg(test)]
245mod tests {
246    use super::*;
247
248    #[test]
249    fn test_mutation_request_trace_and_comment_accessors() {
250        let trace1 = TraceNode {
251            kind: teaql_core::TraceKind::Entity,
252            entity_type: "User".to_string(),
253            entity_id: Some(1),
254            comment: "Create User".to_string(),
255        };
256        let trace2 = TraceNode {
257            kind: teaql_core::TraceKind::Entity,
258            entity_type: "Profile".to_string(),
259            entity_id: None,
260            comment: "Create Profile".to_string(),
261        };
262        let trace_chain = vec![trace1.clone(), trace2.clone()];
263
264        // Test Insert
265        let insert_cmd = InsertCommand {
266            entity: "User".to_string(),
267            values: teaql_core::MutationValues::new(),
268            trace_chain: trace_chain.clone(),
269        };
270        let req_insert = MutationRequest::Insert(insert_cmd);
271        assert_eq!(req_insert.trace_chain().len(), 2);
272        assert_eq!(req_insert.trace_chain()[1], trace2);
273        assert_eq!(req_insert.comment(), Some("Create Profile"));
274
275        // Test Update
276        let update_cmd = UpdateCommand {
277            entity: "User".to_string(),
278            id: teaql_core::Value::I64(1),
279            values: teaql_core::MutationValues::new(),
280            expected_version: None,
281            old_values: None,
282            trace_chain: trace_chain.clone(),
283        };
284        let req_update = MutationRequest::Update(update_cmd);
285        assert_eq!(req_update.trace_chain().len(), 2);
286        assert_eq!(req_update.comment(), Some("Create Profile"));
287
288        // Test Delete
289        let delete_cmd = DeleteCommand {
290            entity: "User".to_string(),
291            id: teaql_core::Value::I64(1),
292            expected_version: None,
293            soft_delete: true,
294            trace_chain: trace_chain.clone(),
295        };
296        let req_delete = MutationRequest::Delete(delete_cmd);
297        assert_eq!(req_delete.trace_chain().len(), 2);
298        assert_eq!(req_delete.comment(), Some("Create Profile"));
299
300        // Test Recover
301        let recover_cmd = RecoverCommand {
302            entity: "User".to_string(),
303            id: teaql_core::Value::I64(1),
304            expected_version: 1,
305            trace_chain: trace_chain.clone(),
306        };
307        let req_recover = MutationRequest::Recover(recover_cmd);
308        assert_eq!(req_recover.trace_chain().len(), 2);
309        assert_eq!(req_recover.comment(), Some("Create Profile"));
310
311        // Test Batch
312        let req_batch = MutationRequest::Batch(vec![req_insert, req_update]);
313        assert_eq!(req_batch.trace_chain().len(), 0);
314        assert_eq!(req_batch.comment(), None);
315
316        // Test empty trace chain
317        let insert_empty = InsertCommand {
318            entity: "User".to_string(),
319            values: teaql_core::MutationValues::new(),
320            trace_chain: vec![],
321        };
322        let req_empty = MutationRequest::Insert(insert_empty);
323        assert_eq!(req_empty.trace_chain().len(), 0);
324        assert_eq!(req_empty.comment(), None);
325    }
326}
teaql_data_service/lib.rs

teaql_data_service/
lib.rs
