radixdb_api/
orm.rs

1//! Embedded ORM execution extensions.
2//!
3//! The facade borrows the existing [`Database`] or [`Transaction`]. It never
4//! creates a second executor, transaction, connection, or catalog snapshot.
5
6use std::collections::BTreeMap;
7
8use base64::Engine as _;
9use chrono::{DateTime, NaiveDate, Utc};
10use radixdb_orm::{
11    CatalogOperation, ColumnDescriptor, ConstraintDescriptor, DataTypeDescriptor,
12    DatabaseDescriptor, DescriptorEnvelope, DescriptorKind, IndexDescriptor, IrDocument,
13    TableDescriptor, TypedValue,
14};
15
16use crate::{Database, FromValue, Rows, Transaction};
17use radixdb_core::{DataType, Error, Value};
18
19#[derive(Debug, thiserror::Error)]
20pub enum OrmError {
21    #[error(transparent)]
22    Database(#[from] Error),
23    #[error(transparent)]
24    Render(#[from] radixdb_orm::RenderError),
25    #[error(transparent)]
26    Descriptor(#[from] radixdb_orm::DescriptorError),
27    #[error("invalid typed ORM value: {0}")]
28    InvalidValue(String),
29    #[error("unexpected ORM result: {0}")]
30    UnexpectedResult(&'static str),
31    #[error("ORM record was not found")]
32    NotFound,
33    #[error(transparent)]
34    Record(#[from] radixdb_orm::RecordError),
35    #[error(transparent)]
36    Build(#[from] radixdb_orm::BuilderError),
37    #[error(transparent)]
38    Generated(#[from] radixdb_orm::GeneratedRecordError),
39}
40
41pub type OrmResult<T> = std::result::Result<T, OrmError>;
42
43impl Database {
44    /// Compile and execute command-like ORM IR on this exact embedded session.
45    pub fn execute_orm(&self, document: &IrDocument) -> OrmResult<i64> {
46        let compiled = document.to_sql()?;
47        Ok(self.execute(&compiled.sql, typed_values_to_core(&compiled.parameters)?)?)
48    }
49
50    /// Compile and execute row-producing ORM IR on this exact embedded session.
51    pub fn query_orm(&self, document: &IrDocument) -> OrmResult<Rows> {
52        let compiled = document.to_sql()?;
53        Ok(self.query(&compiled.sql, typed_values_to_core(&compiled.parameters)?)?)
54    }
55
56    /// Borrow this same embedded session for catalog operations.
57    pub fn schema(&self) -> EmbeddedSchemaClient<'_> {
58        EmbeddedSchemaClient {
59            session: EmbeddedSession::Database(self),
60        }
61    }
62
63    pub fn entity(&self, table: impl Into<String>) -> OrmResult<radixdb_orm::DynamicEntity> {
64        Ok(radixdb_orm::DynamicEntity::new(
65            self.schema().table(table).describe().fetch()?,
66        ))
67    }
68}
69
70impl Transaction {
71    /// Compile and execute command-like ORM IR inside this transaction.
72    pub fn execute_orm(&mut self, document: &IrDocument) -> OrmResult<i64> {
73        let compiled = document.to_sql()?;
74        Ok(self.execute(&compiled.sql, typed_values_to_core(&compiled.parameters)?)?)
75    }
76
77    /// Compile and execute row-producing ORM IR inside this transaction.
78    pub fn query_orm(&mut self, document: &IrDocument) -> OrmResult<Rows> {
79        let compiled = document.to_sql()?;
80        Ok(self.query(&compiled.sql, typed_values_to_core(&compiled.parameters)?)?)
81    }
82
83    /// Borrow this transaction for catalog operations without leaving it.
84    pub fn schema(&mut self) -> EmbeddedSchemaClient<'_> {
85        EmbeddedSchemaClient {
86            session: EmbeddedSession::Transaction(self),
87        }
88    }
89
90    pub fn entity(&mut self, table: impl Into<String>) -> OrmResult<radixdb_orm::DynamicEntity> {
91        Ok(radixdb_orm::DynamicEntity::new(
92            self.schema().table(table).describe().fetch()?,
93        ))
94    }
95}
96
97enum EmbeddedSession<'a> {
98    Database(&'a Database),
99    Transaction(&'a mut Transaction),
100}
101
102impl EmbeddedSession<'_> {
103    fn execute_orm(&mut self, document: &IrDocument) -> OrmResult<i64> {
104        match self {
105            Self::Database(database) => database.execute_orm(document),
106            Self::Transaction(transaction) => transaction.execute_orm(document),
107        }
108    }
109
110    fn query_orm(&mut self, document: &IrDocument) -> OrmResult<Rows> {
111        match self {
112            Self::Database(database) => database.query_orm(document),
113            Self::Transaction(transaction) => transaction.query_orm(document),
114        }
115    }
116}
117
118pub struct EmbeddedSchemaClient<'a> {
119    session: EmbeddedSession<'a>,
120}
121
122impl<'a> EmbeddedSchemaClient<'a> {
123    pub fn tables(self) -> EmbeddedListTablesRequest<'a> {
124        EmbeddedListTablesRequest {
125            session: self.session,
126        }
127    }
128
129    pub fn table(self, name: impl Into<String>) -> EmbeddedTableSchemaClient<'a> {
130        EmbeddedTableSchemaClient {
131            session: self.session,
132            table: name.into(),
133        }
134    }
135
136    pub fn describe_database(self) -> EmbeddedDescribeDatabaseRequest<'a> {
137        EmbeddedDescribeDatabaseRequest {
138            session: self.session,
139        }
140    }
141
142    pub fn create_table(self, table: impl Into<String>) -> EmbeddedCreateTableRequest<'a> {
143        let table = table.into();
144        EmbeddedCreateTableRequest {
145            session: self.session,
146            builder: radixdb_orm::DdlBuilder::create_table(table.clone()),
147            table,
148        }
149    }
150
151    pub fn alter_table(self, table: impl Into<String>) -> EmbeddedAlterTableRequest<'a> {
152        let table = table.into();
153        EmbeddedAlterTableRequest {
154            session: self.session,
155            builder: radixdb_orm::DdlBuilder::alter_table(table.clone()),
156            result_table: table,
157        }
158    }
159
160    pub fn create_table_as(
161        self,
162        table: impl Into<String>,
163        query: radixdb_orm::QueryBuilder,
164    ) -> EmbeddedCreateTableAsRequest<'a> {
165        let table = table.into();
166        EmbeddedCreateTableAsRequest {
167            session: self.session,
168            builder: radixdb_orm::DdlBuilder::create_table_as(table.clone(), query),
169            table,
170        }
171    }
172
173    pub fn drop_table(self, table: impl Into<String>) -> EmbeddedDdlRequest<'a> {
174        EmbeddedDdlRequest {
175            session: self.session,
176            builder: radixdb_orm::DdlBuilder::drop_table(table),
177        }
178    }
179
180    pub fn truncate_table(self, table: impl Into<String>) -> EmbeddedDdlRequest<'a> {
181        EmbeddedDdlRequest {
182            session: self.session,
183            builder: radixdb_orm::DdlBuilder::truncate_table(table),
184        }
185    }
186
187    pub fn create_index(self, index: radixdb_orm::IndexDefinition) -> EmbeddedDdlRequest<'a> {
188        EmbeddedDdlRequest {
189            session: self.session,
190            builder: radixdb_orm::DdlBuilder::create_index(index),
191        }
192    }
193
194    pub fn drop_index(
195        self,
196        table: impl Into<String>,
197        index: impl Into<String>,
198        if_exists: bool,
199    ) -> EmbeddedDdlRequest<'a> {
200        EmbeddedDdlRequest {
201            session: self.session,
202            builder: radixdb_orm::DdlBuilder::drop_index(table, index, if_exists),
203        }
204    }
205
206    pub fn alter_index(
207        self,
208        index: impl Into<String>,
209        new_name: impl Into<String>,
210    ) -> EmbeddedDdlRequest<'a> {
211        EmbeddedDdlRequest {
212            session: self.session,
213            builder: radixdb_orm::DdlBuilder::alter_index(index, new_name),
214        }
215    }
216}
217
218pub struct EmbeddedCreateTableAsRequest<'a> {
219    session: EmbeddedSession<'a>,
220    builder: radixdb_orm::CreateTableAsBuilder,
221    table: String,
222}
223
224impl EmbeddedCreateTableAsRequest<'_> {
225    pub fn if_not_exists(mut self, value: bool) -> Self {
226        self.builder = self.builder.if_not_exists(value);
227        self
228    }
229    pub fn to_json(&self) -> Result<String, radixdb_orm::BuilderJsonError> {
230        radixdb_orm::OrmBuilder::to_json(&self.builder)
231    }
232    pub fn to_sql(&self) -> Result<radixdb_orm::CompiledStatement, radixdb_orm::BuilderSqlError> {
233        radixdb_orm::OrmBuilder::to_sql(&self.builder)
234    }
235    pub fn execute(mut self) -> OrmResult<TableDescriptor> {
236        let document = radixdb_orm::OrmBuilder::document(&self.builder)
237            .map_err(|error| OrmError::InvalidValue(error.to_string()))?;
238        self.session.execute_orm(&document)?;
239        describe_table_on_session(&mut self.session, self.table)
240    }
241}
242
243pub struct EmbeddedCreateTableRequest<'a> {
244    session: EmbeddedSession<'a>,
245    builder: radixdb_orm::CreateTableBuilder,
246    table: String,
247}
248
249impl EmbeddedCreateTableRequest<'_> {
250    pub fn if_not_exists(mut self, value: bool) -> Self {
251        self.builder = self.builder.if_not_exists(value);
252        self
253    }
254
255    pub fn column(mut self, column: radixdb_orm::Column) -> Self {
256        self.builder = self.builder.column(column);
257        self
258    }
259
260    pub fn constraint(mut self, constraint: radixdb_orm::ConstraintDefinitionIr) -> Self {
261        self.builder = self.builder.constraint(constraint);
262        self
263    }
264
265    pub fn to_json(&self) -> Result<String, radixdb_orm::BuilderJsonError> {
266        radixdb_orm::OrmBuilder::to_json(&self.builder)
267    }
268
269    pub fn to_sql(&self) -> Result<radixdb_orm::CompiledStatement, radixdb_orm::BuilderSqlError> {
270        radixdb_orm::OrmBuilder::to_sql(&self.builder)
271    }
272
273    pub fn execute(mut self) -> OrmResult<TableDescriptor> {
274        let document = radixdb_orm::OrmBuilder::document(&self.builder)
275            .map_err(|error| OrmError::InvalidValue(error.to_string()))?;
276        self.session.execute_orm(&document)?;
277        describe_table_on_session(&mut self.session, self.table)
278    }
279}
280
281pub struct EmbeddedAlterTableRequest<'a> {
282    session: EmbeddedSession<'a>,
283    builder: radixdb_orm::AlterTableBuilder,
284    result_table: String,
285}
286
287impl EmbeddedAlterTableRequest<'_> {
288    pub fn add_column(mut self, column: radixdb_orm::Column) -> Self {
289        self.builder = self.builder.add_column(column);
290        self
291    }
292    pub fn modify_column(mut self, column: radixdb_orm::Column) -> Self {
293        self.builder = self.builder.modify_column(column);
294        self
295    }
296    pub fn drop_column(mut self, column: impl Into<String>) -> Self {
297        self.builder = self.builder.drop_column(column);
298        self
299    }
300    pub fn rename_column(mut self, from: impl Into<String>, to: impl Into<String>) -> Self {
301        self.builder = self.builder.rename_column(from, to);
302        self
303    }
304    pub fn rename_table(mut self, to: impl Into<String>) -> Self {
305        let to = to.into();
306        self.result_table = to.clone();
307        self.builder = self.builder.rename_table(to);
308        self
309    }
310    pub fn add_constraint(mut self, constraint: radixdb_orm::ConstraintDefinitionIr) -> Self {
311        self.builder = self.builder.add_constraint(constraint);
312        self
313    }
314    pub fn drop_constraint(mut self, name: impl Into<String>, if_exists: bool) -> Self {
315        self.builder = self.builder.drop_constraint(name, if_exists);
316        self
317    }
318    pub fn execute(mut self) -> OrmResult<TableDescriptor> {
319        let document = radixdb_orm::OrmBuilder::document(&self.builder)
320            .map_err(|error| OrmError::InvalidValue(error.to_string()))?;
321        self.session.execute_orm(&document)?;
322        describe_table_on_session(&mut self.session, self.result_table)
323    }
324}
325
326pub struct EmbeddedDdlRequest<'a> {
327    session: EmbeddedSession<'a>,
328    builder: radixdb_orm::DdlBuilder,
329}
330
331impl EmbeddedDdlRequest<'_> {
332    pub fn if_exists(mut self, value: bool) -> Self {
333        self.builder = self.builder.if_exists(value);
334        self
335    }
336    pub fn to_json(&self) -> Result<String, radixdb_orm::BuilderJsonError> {
337        radixdb_orm::OrmBuilder::to_json(&self.builder)
338    }
339    pub fn to_sql(&self) -> Result<radixdb_orm::CompiledStatement, radixdb_orm::BuilderSqlError> {
340        radixdb_orm::OrmBuilder::to_sql(&self.builder)
341    }
342    pub fn execute(mut self) -> OrmResult<i64> {
343        let document = radixdb_orm::OrmBuilder::document(&self.builder)
344            .map_err(|error| OrmError::InvalidValue(error.to_string()))?;
345        self.session.execute_orm(&document)
346    }
347}
348
349fn describe_table_on_session(
350    session: &mut EmbeddedSession<'_>,
351    table: String,
352) -> OrmResult<TableDescriptor> {
353    let document = catalog_document(CatalogOperation::DescribeTable { table });
354    let json = fetch_one_text(session.query_orm(&document)?)?;
355    Ok(DescriptorEnvelope::<TableDescriptor>::from_json(&json, DescriptorKind::Table)?.payload)
356}
357
358pub struct EmbeddedListTablesRequest<'a> {
359    session: EmbeddedSession<'a>,
360}
361
362impl EmbeddedListTablesRequest<'_> {
363    pub fn fetch(mut self) -> OrmResult<Vec<String>> {
364        let document = catalog_document(CatalogOperation::ListTables);
365        fetch_single_text_column(self.session.query_orm(&document)?)
366    }
367}
368
369pub struct EmbeddedTableSchemaClient<'a> {
370    session: EmbeddedSession<'a>,
371    table: String,
372}
373
374impl<'a> EmbeddedTableSchemaClient<'a> {
375    pub fn describe(self) -> EmbeddedDescribeTableRequest<'a> {
376        EmbeddedDescribeTableRequest {
377            session: self.session,
378            table: self.table,
379        }
380    }
381
382    pub fn columns(self) -> EmbeddedTableColumnsRequest<'a> {
383        EmbeddedTableColumnsRequest {
384            request: self.describe(),
385        }
386    }
387
388    pub fn indexes(self) -> EmbeddedTableIndexesRequest<'a> {
389        EmbeddedTableIndexesRequest {
390            request: self.describe(),
391        }
392    }
393
394    pub fn constraints(self) -> EmbeddedTableConstraintsRequest<'a> {
395        EmbeddedTableConstraintsRequest {
396            request: self.describe(),
397        }
398    }
399}
400
401pub struct EmbeddedDescribeTableRequest<'a> {
402    session: EmbeddedSession<'a>,
403    table: String,
404}
405
406impl EmbeddedDescribeTableRequest<'_> {
407    pub fn fetch(mut self) -> OrmResult<TableDescriptor> {
408        let document = catalog_document(CatalogOperation::DescribeTable { table: self.table });
409        let json = fetch_one_text(self.session.query_orm(&document)?)?;
410        Ok(DescriptorEnvelope::<TableDescriptor>::from_json(&json, DescriptorKind::Table)?.payload)
411    }
412}
413
414pub struct EmbeddedTableColumnsRequest<'a> {
415    request: EmbeddedDescribeTableRequest<'a>,
416}
417
418impl EmbeddedTableColumnsRequest<'_> {
419    pub fn fetch(self) -> OrmResult<Vec<ColumnDescriptor>> {
420        Ok(self.request.fetch()?.columns)
421    }
422}
423
424pub struct EmbeddedTableIndexesRequest<'a> {
425    request: EmbeddedDescribeTableRequest<'a>,
426}
427
428impl EmbeddedTableIndexesRequest<'_> {
429    pub fn fetch(self) -> OrmResult<Vec<IndexDescriptor>> {
430        Ok(self.request.fetch()?.indexes)
431    }
432}
433
434pub struct EmbeddedTableConstraintsRequest<'a> {
435    request: EmbeddedDescribeTableRequest<'a>,
436}
437
438impl EmbeddedTableConstraintsRequest<'_> {
439    pub fn fetch(self) -> OrmResult<Vec<ConstraintDescriptor>> {
440        Ok(self.request.fetch()?.constraints)
441    }
442}
443
444pub struct EmbeddedDescribeDatabaseRequest<'a> {
445    session: EmbeddedSession<'a>,
446}
447
448impl EmbeddedDescribeDatabaseRequest<'_> {
449    pub fn fetch(mut self) -> OrmResult<DatabaseDescriptor> {
450        let document = catalog_document(CatalogOperation::DescribeDatabase);
451        let json = fetch_one_text(self.session.query_orm(&document)?)?;
452        Ok(
453            DescriptorEnvelope::<DatabaseDescriptor>::from_json(&json, DescriptorKind::Database)?
454                .payload,
455        )
456    }
457}
458
459fn catalog_document(operation: CatalogOperation) -> IrDocument {
460    IrDocument::new(radixdb_orm::Operation::Catalog { operation })
461}
462
463fn fetch_one_text(rows: Rows) -> OrmResult<String> {
464    let values = fetch_single_text_column(rows)?;
465    match values.as_slice() {
466        [value] => Ok(value.clone()),
467        _ => Err(OrmError::UnexpectedResult("expected exactly one text row")),
468    }
469}
470
471fn fetch_single_text_column(rows: Rows) -> OrmResult<Vec<String>> {
472    rows.map(|row| {
473        let row = row?;
474        String::from_value(
475            row.get_value(0)
476                .ok_or_else(|| Error::invalid_argument("expected one text result column"))?,
477        )
478        .map_err(OrmError::from)
479    })
480    .collect()
481}
482
483pub(crate) fn typed_values_to_core(values: &[TypedValue]) -> OrmResult<Vec<Value>> {
484    values.iter().map(typed_value_to_core).collect()
485}
486
487fn typed_value_to_core(value: &TypedValue) -> OrmResult<Value> {
488    Ok(match value {
489        TypedValue::Null(data_type) => Value::null(descriptor_data_type(data_type)),
490        TypedValue::Integer(value) => Value::integer(*value),
491        TypedValue::Float(value) => Value::float(value.as_f64()),
492        TypedValue::Text(value) => Value::text(value.clone()),
493        TypedValue::Boolean(value) => Value::boolean(*value),
494        TypedValue::Timestamp(value) => {
495            let timestamp = DateTime::parse_from_rfc3339(value).map_err(|error| {
496                OrmError::InvalidValue(format!("invalid RFC3339 timestamp: {error}"))
497            })?;
498            Value::timestamp(timestamp.with_timezone(&Utc))
499        }
500        TypedValue::Date(value) => {
501            let date = NaiveDate::parse_from_str(value, "%Y-%m-%d")
502                .map_err(|error| OrmError::InvalidValue(format!("invalid ISO date: {error}")))?;
503            let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).expect("valid Unix epoch");
504            let days =
505                i32::try_from(date.signed_duration_since(epoch).num_days()).map_err(|_| {
506                    OrmError::InvalidValue("date is outside i32 day domain".to_string())
507                })?;
508            Value::date(days)
509        }
510        TypedValue::Json(value) => Value::try_json(
511            serde_json::to_string(value)
512                .map_err(|error| OrmError::InvalidValue(format!("invalid JSON value: {error}")))?,
513        )?,
514        TypedValue::Uuid(value) => Value::uuid(
515            *uuid::Uuid::parse_str(value)
516                .map_err(|error| OrmError::InvalidValue(format!("invalid UUID: {error}")))?
517                .as_bytes(),
518        ),
519        TypedValue::Bytes(value) => Value::bytes(
520            base64::engine::general_purpose::STANDARD
521                .decode(value)
522                .map_err(|error| {
523                    OrmError::InvalidValue(format!("invalid base64 BYTES: {error}"))
524                })?,
525        ),
526        TypedValue::Decimal(value) => {
527            let (unscaled, precision, scale) = radixdb_orm::parse_decimal_literal(value)
528                .map_err(|error| OrmError::InvalidValue(error.to_string()))?;
529            Value::try_decimal(unscaled, precision, scale)?
530        }
531        TypedValue::Vector(values) => Value::vector(values.clone()),
532    })
533}
534
535fn descriptor_data_type(data_type: &DataTypeDescriptor) -> DataType {
536    match data_type {
537        DataTypeDescriptor::Null => DataType::Null,
538        DataTypeDescriptor::Integer => DataType::Integer,
539        DataTypeDescriptor::Float => DataType::Float,
540        DataTypeDescriptor::Text => DataType::Text,
541        DataTypeDescriptor::Boolean => DataType::Boolean,
542        DataTypeDescriptor::Timestamp => DataType::Timestamp,
543        DataTypeDescriptor::Date => DataType::Date,
544        DataTypeDescriptor::Json => DataType::Json,
545        DataTypeDescriptor::Uuid => DataType::Uuid,
546        DataTypeDescriptor::Bytes => DataType::Bytes,
547        DataTypeDescriptor::Decimal { .. } => DataType::Decimal,
548        DataTypeDescriptor::Vector { .. } => DataType::Vector,
549    }
550}
551
552impl radixdb_orm::OrmSession for &Database {
553    type CommandOutput = i64;
554    type QueryOutput = Rows;
555    type Error = OrmError;
556
557    fn execute_document(self, document: &IrDocument) -> OrmResult<Self::CommandOutput> {
558        self.execute_orm(document)
559    }
560
561    fn query_document(self, document: &IrDocument) -> OrmResult<Self::QueryOutput> {
562        self.query_orm(document)
563    }
564}
565
566impl radixdb_orm::OrmSession for &mut Transaction {
567    type CommandOutput = i64;
568    type QueryOutput = Rows;
569    type Error = OrmError;
570
571    fn execute_document(self, document: &IrDocument) -> OrmResult<Self::CommandOutput> {
572        self.execute_orm(document)
573    }
574
575    fn query_document(self, document: &IrDocument) -> OrmResult<Self::QueryOutput> {
576        self.query_orm(document)
577    }
578}
579
580impl radixdb_orm::OrmRecordSession for &Database {
581    type Error = OrmError;
582
583    fn mutate_record(
584        self,
585        record: &mut radixdb_orm::DynamicRecord,
586        mutation: radixdb_orm::RecordMutation,
587    ) -> OrmResult<()> {
588        if mutation == radixdb_orm::RecordMutation::Delete {
589            let document = record.delete_document()?;
590            if self.execute_orm(&document)? == 0 {
591                return Err(OrmError::NotFound);
592            }
593            return Ok(());
594        }
595        let document = record_mutation_document(record, mutation)?;
596        let values = one_embedded_record(self.query_orm(&document)?, record.descriptor())?;
597        record.apply_returning(values)?;
598        Ok(())
599    }
600}
601
602impl radixdb_orm::OrmRecordSession for &mut Transaction {
603    type Error = OrmError;
604
605    fn mutate_record(
606        self,
607        record: &mut radixdb_orm::DynamicRecord,
608        mutation: radixdb_orm::RecordMutation,
609    ) -> OrmResult<()> {
610        if mutation == radixdb_orm::RecordMutation::Delete {
611            let document = record.delete_document()?;
612            if self.execute_orm(&document)? == 0 {
613                return Err(OrmError::NotFound);
614            }
615            return Ok(());
616        }
617        let document = record_mutation_document(record, mutation)?;
618        let values = one_embedded_record(self.query_orm(&document)?, record.descriptor())?;
619        record.apply_returning(values)?;
620        Ok(())
621    }
622}
623
624impl radixdb_orm::OrmGeneratedRecordSession for &Database {
625    type Error = OrmError;
626
627    fn mutate_generated_record<R: radixdb_orm::GeneratedRecord>(
628        self,
629        record: &mut R,
630        mutation: radixdb_orm::RecordMutation,
631    ) -> OrmResult<()> {
632        let descriptor = self
633            .schema()
634            .table(<R::Entity as radixdb_orm::GeneratedEntity>::TABLE)
635            .describe()
636            .fetch()?;
637        let mut dynamic = record.to_dynamic(&descriptor)?;
638        radixdb_orm::OrmRecordSession::mutate_record(self, &mut dynamic, mutation)?;
639        record.apply_dynamic(&dynamic)?;
640        Ok(())
641    }
642}
643
644impl radixdb_orm::OrmGeneratedRecordSession for &mut Transaction {
645    type Error = OrmError;
646
647    fn mutate_generated_record<R: radixdb_orm::GeneratedRecord>(
648        self,
649        record: &mut R,
650        mutation: radixdb_orm::RecordMutation,
651    ) -> OrmResult<()> {
652        let descriptor = self
653            .schema()
654            .table(<R::Entity as radixdb_orm::GeneratedEntity>::TABLE)
655            .describe()
656            .fetch()?;
657        let mut dynamic = record.to_dynamic(&descriptor)?;
658        radixdb_orm::OrmRecordSession::mutate_record(&mut *self, &mut dynamic, mutation)?;
659        record.apply_dynamic(&dynamic)?;
660        Ok(())
661    }
662}
663
664impl radixdb_orm::OrmGeneratedQuerySession for &Database {
665    type Error = OrmError;
666
667    fn query_generated_records<R: radixdb_orm::GeneratedRecord + Default>(
668        self,
669        document: &IrDocument,
670    ) -> OrmResult<Vec<R>> {
671        let descriptor = self
672            .schema()
673            .table(<R::Entity as radixdb_orm::GeneratedEntity>::TABLE)
674            .describe()
675            .fetch()?;
676        generated_records_from_embedded_rows::<R>(self.query_orm(document)?, descriptor)
677    }
678}
679
680impl radixdb_orm::OrmGeneratedQuerySession for &mut Transaction {
681    type Error = OrmError;
682
683    fn query_generated_records<R: radixdb_orm::GeneratedRecord + Default>(
684        self,
685        document: &IrDocument,
686    ) -> OrmResult<Vec<R>> {
687        let descriptor = self
688            .schema()
689            .table(<R::Entity as radixdb_orm::GeneratedEntity>::TABLE)
690            .describe()
691            .fetch()?;
692        generated_records_from_embedded_rows::<R>(self.query_orm(document)?, descriptor)
693    }
694}
695
696fn generated_records_from_embedded_rows<R: radixdb_orm::GeneratedRecord + Default>(
697    rows: Rows,
698    descriptor: TableDescriptor,
699) -> OrmResult<Vec<R>> {
700    radixdb_orm::ensure_schema_fingerprint(
701        <R::Entity as radixdb_orm::GeneratedEntity>::SCHEMA_FINGERPRINT,
702        &descriptor.fingerprint,
703    )
704    .map_err(radixdb_orm::GeneratedRecordError::from)?;
705    rows.map(|row| {
706        let row = row?;
707        if row.len() != descriptor.columns.len() {
708            return Err(OrmError::UnexpectedResult(
709                "generated SELECT row width differs from descriptor",
710            ));
711        }
712        let values = descriptor
713            .columns
714            .iter()
715            .enumerate()
716            .map(|(index, column)| {
717                let value = row.get_value(index).ok_or(OrmError::UnexpectedResult(
718                    "generated SELECT row width differs from descriptor",
719                ))?;
720                Ok((
721                    column.name.clone(),
722                    core_value_to_typed(value, &column.data_type)?,
723                ))
724            })
725            .collect::<OrmResult<BTreeMap<_, _>>>()?;
726        let mut dynamic = radixdb_orm::DynamicRecord::new(descriptor.clone());
727        dynamic.apply_returning(values)?;
728        let mut generated = R::default();
729        generated.apply_dynamic(&dynamic)?;
730        Ok(generated)
731    })
732    .collect()
733}
734
735fn record_mutation_document(
736    record: &radixdb_orm::DynamicRecord,
737    mutation: radixdb_orm::RecordMutation,
738) -> OrmResult<IrDocument> {
739    Ok(match mutation {
740        radixdb_orm::RecordMutation::Insert => record.insert_document()?,
741        radixdb_orm::RecordMutation::Save => record.save_document()?,
742        radixdb_orm::RecordMutation::Update => record.update_document()?,
743        radixdb_orm::RecordMutation::Delete => record.delete_document()?,
744    })
745}
746
747fn one_embedded_record(
748    mut rows: Rows,
749    descriptor: &TableDescriptor,
750) -> OrmResult<BTreeMap<String, TypedValue>> {
751    let row = rows.next().ok_or(OrmError::NotFound)??;
752    if rows.next().is_some() {
753        return Err(OrmError::UnexpectedResult(
754            "RETURNING produced multiple rows",
755        ));
756    }
757    descriptor
758        .columns
759        .iter()
760        .enumerate()
761        .map(|(index, column)| {
762            let value = row.get_value(index).ok_or(OrmError::UnexpectedResult(
763                "RETURNING row width differs from descriptor",
764            ))?;
765            Ok((
766                column.name.clone(),
767                core_value_to_typed(value, &column.data_type)?,
768            ))
769        })
770        .collect()
771}
772
773fn core_value_to_typed(value: &Value, declared: &DataTypeDescriptor) -> OrmResult<TypedValue> {
774    if value.is_null() {
775        return Ok(TypedValue::Null(declared.clone()));
776    }
777    Ok(match value {
778        Value::Integer(value) => TypedValue::Integer(*value),
779        Value::Float(value) => TypedValue::Float((*value).into()),
780        Value::Text(value) => TypedValue::Text(value.to_string()),
781        Value::Boolean(value) => TypedValue::Boolean(*value),
782        Value::Timestamp(value) => {
783            TypedValue::Timestamp(value.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true))
784        }
785        Value::Extension(_) => match declared {
786            DataTypeDescriptor::Json => TypedValue::Json(
787                serde_json::from_str(value.as_json().ok_or_else(|| {
788                    OrmError::InvalidValue("invalid JSON result payload".to_string())
789                })?)
790                .map_err(|error| OrmError::InvalidValue(error.to_string()))?,
791            ),
792            DataTypeDescriptor::Uuid => TypedValue::Uuid(
793                uuid::Uuid::from_bytes(value.as_uuid_bytes().ok_or_else(|| {
794                    OrmError::InvalidValue("invalid UUID result payload".to_string())
795                })?)
796                .to_string(),
797            ),
798            DataTypeDescriptor::Decimal { .. } => {
799                let (unscaled, _, scale) = value.as_decimal_parts().ok_or_else(|| {
800                    OrmError::InvalidValue("invalid DECIMAL result payload".to_string())
801                })?;
802                TypedValue::Decimal(radixdb_core::value::format_decimal_parts(unscaled, scale))
803            }
804            DataTypeDescriptor::Date => {
805                let days = value.as_date_days().ok_or_else(|| {
806                    OrmError::InvalidValue("invalid DATE result payload".to_string())
807                })?;
808                let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).expect("valid epoch");
809                TypedValue::Date(
810                    (epoch + chrono::Duration::days(i64::from(days)))
811                        .format("%Y-%m-%d")
812                        .to_string(),
813                )
814            }
815            DataTypeDescriptor::Bytes => {
816                TypedValue::Bytes(base64::engine::general_purpose::STANDARD.encode(
817                    value.as_bytes_value().ok_or_else(|| {
818                        OrmError::InvalidValue("invalid BYTES result payload".to_string())
819                    })?,
820                ))
821            }
822            DataTypeDescriptor::Vector { .. } => {
823                TypedValue::Vector(value.as_vector_f32().ok_or_else(|| {
824                    OrmError::InvalidValue("invalid VECTOR result payload".to_string())
825                })?)
826            }
827            other => {
828                return Err(OrmError::InvalidValue(format!(
829                    "extension result does not match {other:?}"
830                )))
831            }
832        },
833        Value::Null(_) => unreachable!("handled above"),
834    })
835}
836
837#[cfg(test)]
838mod tests {
839    use radixdb_orm::FieldValue;
840    use radixdb_orm::{
841        BinaryOperator, Expression, Insert, Operation, Projection, Relation, Select, TypedValue,
842    };
843
844    use super::*;
845
846    fn users_select() -> IrDocument {
847        IrDocument::new(Operation::Select {
848            query: Select {
849                projection: vec![Projection {
850                    expression: Expression::column("name"),
851                    alias: None,
852                }],
853                from: Some(Relation::Table {
854                    name: "orm_users".to_string(),
855                    alias: None,
856                }),
857                filter: Some(Expression::Binary {
858                    left: Box::new(Expression::column("id")),
859                    operator: BinaryOperator::Eq,
860                    right: Box::new(Expression::literal(TypedValue::Integer(1))),
861                }),
862                ..Select::default()
863            },
864        })
865    }
866
867    #[test]
868    fn raw_orm_raw_share_one_embedded_transaction() {
869        let db = Database::open_in_memory().unwrap();
870        db.execute(
871            "CREATE TABLE orm_users (id INTEGER PRIMARY KEY, name TEXT)",
872            (),
873        )
874        .unwrap();
875
876        let mut transaction = db.begin().unwrap();
877        transaction
878            .execute("INSERT INTO orm_users VALUES (1, 'raw-before')", ())
879            .unwrap();
880        let selected: String = transaction
881            .query_orm(&users_select())
882            .unwrap()
883            .next()
884            .unwrap()
885            .unwrap()
886            .get(0)
887            .unwrap();
888        assert_eq!(selected, "raw-before");
889
890        let insert = IrDocument::new(Operation::Insert {
891            statement: Insert {
892                table: "orm_users".to_string(),
893                columns: vec!["id".to_string(), "name".to_string()],
894                rows: vec![vec![
895                    Expression::literal(TypedValue::Integer(2)),
896                    Expression::literal(TypedValue::Text("orm-middle".to_string())),
897                ]],
898                source: None,
899                returning: Vec::new(),
900            },
901        });
902        assert_eq!(transaction.execute_orm(&insert).unwrap(), 1);
903        assert_eq!(
904            transaction
905                .query_one::<i64, _>("SELECT COUNT(*) FROM orm_users", ())
906                .unwrap(),
907            2
908        );
909        assert_eq!(
910            transaction.schema().tables().fetch().unwrap(),
911            vec!["orm_users"]
912        );
913        transaction.rollback().unwrap();
914
915        assert_eq!(
916            db.query_one::<i64, _>("SELECT COUNT(*) FROM orm_users", ())
917                .unwrap(),
918            0
919        );
920    }
921
922    #[test]
923    fn dynamic_record_crud_hydrates_only_after_server_success() {
924        let db = Database::open_in_memory().unwrap();
925        db.execute(
926            "CREATE TABLE orm_records (id INTEGER PRIMARY KEY, name TEXT DEFAULT 'server')",
927            (),
928        )
929        .unwrap();
930        let descriptor = db.schema().table("orm_records").describe().fetch().unwrap();
931        let mut record = radixdb_orm::DynamicRecord::new(descriptor);
932        record.set("id", TypedValue::Integer(7)).unwrap();
933        record.insert(&db).unwrap();
934        assert!(!record.is_dirty());
935        assert!(matches!(
936            record.field("name").unwrap().value(),
937            FieldValue::Value { value: TypedValue::Text(value) } if value == "server"
938        ));
939
940        record
941            .set("name", TypedValue::Text("updated".to_string()))
942            .unwrap();
943        record.update(&db).unwrap();
944        assert!(!record.is_dirty());
945        assert_eq!(
946            db.query_one::<String, _>("SELECT name FROM orm_records WHERE id = 7", ())
947                .unwrap(),
948            "updated"
949        );
950
951        record
952            .set("name", TypedValue::Text("saved".to_string()))
953            .unwrap();
954        record.save(&db).unwrap();
955        assert!(!record.is_dirty());
956        record.delete(&db).unwrap();
957        assert!(matches!(record.delete(&db), Err(OrmError::NotFound)));
958
959        record
960            .set("name", TypedValue::Text("missing-row".to_string()))
961            .unwrap();
962        assert!(matches!(record.update(&db), Err(OrmError::NotFound)));
963        assert!(record.is_dirty(), "zero-row UPDATE must retain dirty state");
964
965        let descriptor = db.schema().table("orm_records").describe().fetch().unwrap();
966        let mut duplicate = radixdb_orm::DynamicRecord::new(descriptor);
967        duplicate.set("id", TypedValue::Integer(8)).unwrap();
968        duplicate
969            .set("name", TypedValue::Text("first".to_string()))
970            .unwrap();
971        duplicate.insert(&db).unwrap();
972        let mut rejected = duplicate.clone();
973        rejected
974            .set("name", TypedValue::Text("duplicate".to_string()))
975            .unwrap();
976        assert!(rejected.insert(&db).is_err());
977        assert!(rejected.is_dirty(), "failed INSERT must retain dirty state");
978    }
979
980    #[test]
981    fn bound_schema_ddl_builders_execute_and_return_current_descriptor() {
982        let db = Database::open_in_memory().unwrap();
983        let created = db
984            .schema()
985            .create_table("orm_ddl")
986            .column(
987                radixdb_orm::Column::integer("id")
988                    .primary_key(true)
989                    .check(radixdb_orm::Expr::column("id").gt(0_i64)),
990            )
991            .column(radixdb_orm::Column::text("name"))
992            .column(
993                radixdb_orm::Column::new("active", radixdb_orm::DataTypeDescriptor::Boolean)
994                    .not_null(true)
995                    .default(true),
996            )
997            .column(radixdb_orm::Column::vector("embedding", 3))
998            .execute()
999            .unwrap();
1000        assert_eq!(created.name, "orm_ddl");
1001        assert_eq!(created.columns.len(), 4);
1002        db.execute(
1003            "INSERT INTO orm_ddl(id, name, embedding) VALUES (1, 'bound', $1)",
1004            vec![Value::vector(vec![1.0, 2.0, 3.0])],
1005        )
1006        .unwrap();
1007        assert!(db
1008            .query_one::<bool, _>("SELECT active FROM orm_ddl WHERE id = 1", ())
1009            .unwrap());
1010
1011        let altered = db
1012            .schema()
1013            .alter_table("orm_ddl")
1014            .add_column(radixdb_orm::Column::text("note"))
1015            .execute()
1016            .unwrap();
1017        assert!(altered.columns.iter().any(|column| column.name == "note"));
1018        db.schema()
1019            .create_index(
1020                radixdb_orm::IndexDefinition::new(
1021                    "idx_orm_ddl_embedding",
1022                    "orm_ddl",
1023                    ["embedding"],
1024                )
1025                .method(radixdb_orm::IndexMethod::Hnsw)
1026                .option("m", TypedValue::Integer(8))
1027                .option("metric", TypedValue::Text("cosine".to_string())),
1028            )
1029            .execute()
1030            .unwrap();
1031        db.schema()
1032            .create_index(
1033                radixdb_orm::IndexDefinition::new("idx_orm_ddl_active", "orm_ddl", ["active"])
1034                    .if_not_exists(true)
1035                    .where_(radixdb_orm::Expr::column("active").eq(true)),
1036            )
1037            .execute()
1038            .unwrap();
1039        let indexes = db.schema().table("orm_ddl").indexes().fetch().unwrap();
1040        assert!(indexes.iter().any(|index| {
1041            index.name == "idx_orm_ddl_embedding"
1042                && index.options.get("m") == Some(&serde_json::Value::from(8))
1043                && index.options.get("distance_metric") == Some(&serde_json::Value::from("cosine"))
1044        }));
1045        assert!(
1046            indexes.iter().any(|index| {
1047                index.name == "idx_orm_ddl_active"
1048                    && index.predicate.as_deref() == Some("(\"active\" = TRUE)")
1049            }),
1050            "unexpected index descriptors: {indexes:?}"
1051        );
1052        db.schema()
1053            .alter_index("idx_orm_ddl_active", "idx_orm_ddl_active_renamed")
1054            .execute()
1055            .unwrap();
1056        assert!(db
1057            .schema()
1058            .table("orm_ddl")
1059            .indexes()
1060            .fetch()
1061            .unwrap()
1062            .iter()
1063            .any(|index| index.name == "idx_orm_ddl_active_renamed"));
1064        db.schema()
1065            .drop_index("orm_ddl", "idx_orm_ddl_active_renamed", false)
1066            .execute()
1067            .unwrap();
1068        assert!(!db
1069            .schema()
1070            .table("orm_ddl")
1071            .indexes()
1072            .fetch()
1073            .unwrap()
1074            .iter()
1075            .any(|index| index.name == "idx_orm_ddl_active_renamed"));
1076
1077        let copied = db
1078            .schema()
1079            .create_table_as(
1080                "orm_ddl_copy",
1081                radixdb_orm::QueryBuilder::from_relation(radixdb_orm::table("orm_ddl")).select([
1082                    radixdb_orm::Expr::column("id"),
1083                    radixdb_orm::Expr::column("name"),
1084                ]),
1085            )
1086            .execute()
1087            .unwrap();
1088        assert_eq!(copied.columns.len(), 2);
1089        assert_eq!(
1090            db.query_one::<i64, _>("SELECT COUNT(*) FROM orm_ddl_copy", ())
1091                .unwrap(),
1092            1
1093        );
1094        db.schema()
1095            .drop_table("orm_ddl_copy")
1096            .if_exists(true)
1097            .execute()
1098            .unwrap();
1099        db.schema()
1100            .drop_table("orm_ddl_copy")
1101            .if_exists(true)
1102            .execute()
1103            .unwrap();
1104        db.schema().truncate_table("orm_ddl").execute().unwrap();
1105        db.schema().drop_table("orm_ddl").execute().unwrap();
1106        assert!(db.schema().table("orm_ddl").describe().fetch().is_err());
1107    }
1108}
radixdb_api/orm.rs

radixdb_api/
orm.rs
