cufflink_types/
db.rs

1//! The `db_call` wire contract shared by the guest SDK and the platform:
2//! requests, responses, bound parameters and structured database errors.
3
4use serde::{Deserialize, Serialize};
5use serde_json::{Map, Value};
6use std::fmt;
7
8/// A bound parameter. Serializes as `{"t": <tag>, "v": <value>}`; `Null`
9/// carries no `v`, `Bytes` is base64 and non-finite floats are the strings
10/// `"NaN"`, `"Infinity"` and `"-Infinity"`.
11#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
12#[serde(tag = "t", content = "v", rename_all = "snake_case")]
13pub enum Param {
14    Null,
15    Bool(bool),
16    I16(i16),
17    I32(i32),
18    I64(i64),
19    F32(#[serde(with = "float")] f32),
20    F64(#[serde(with = "float")] f64),
21    Text(String),
22    Uuid(String),
23    /// RFC 3339 with an offset.
24    Timestamp(String),
25    NaiveTimestamp(String),
26    Date(String),
27    Time(String),
28    Json(Value),
29    /// Exact decimal text.
30    Numeric(String),
31    Bytes(#[serde(with = "crate::serde_util::base64_bytes")] Vec<u8>),
32    TextArray(Vec<Option<String>>),
33    UuidArray(Vec<Option<String>>),
34    I32Array(Vec<Option<i32>>),
35    I64Array(Vec<Option<i64>>),
36    BoolArray(Vec<Option<bool>>),
37    F64Array(#[serde(with = "float_array")] Vec<Option<f64>>),
38    NumericArray(Vec<Option<String>>),
39}
40
41impl Param {
42    /// The wire tag, which names the parameter's kind in errors.
43    pub fn tag(&self) -> &'static str {
44        match self {
45            Param::Null => "null",
46            Param::Bool(_) => "bool",
47            Param::I16(_) => "i16",
48            Param::I32(_) => "i32",
49            Param::I64(_) => "i64",
50            Param::F32(_) => "f32",
51            Param::F64(_) => "f64",
52            Param::Text(_) => "text",
53            Param::Uuid(_) => "uuid",
54            Param::Timestamp(_) => "timestamp",
55            Param::NaiveTimestamp(_) => "naive_timestamp",
56            Param::Date(_) => "date",
57            Param::Time(_) => "time",
58            Param::Json(_) => "json",
59            Param::Numeric(_) => "numeric",
60            Param::Bytes(_) => "bytes",
61            Param::TextArray(_) => "text_array",
62            Param::UuidArray(_) => "uuid_array",
63            Param::I32Array(_) => "i32_array",
64            Param::I64Array(_) => "i64_array",
65            Param::BoolArray(_) => "bool_array",
66            Param::F64Array(_) => "f64_array",
67            Param::NumericArray(_) => "numeric_array",
68        }
69    }
70}
71
72macro_rules! param_from {
73    ($($ty:ty => $variant:ident),* $(,)?) => {
74        $(impl From<$ty> for Param {
75            fn from(value: $ty) -> Self {
76                Param::$variant(value.into())
77            }
78        })*
79    };
80}
81
82param_from! {
83    bool => Bool,
84    i16 => I16,
85    i32 => I32,
86    i64 => I64,
87    f32 => F32,
88    f64 => F64,
89    String => Text,
90    &str => Text,
91    Value => Json,
92    Vec<u8> => Bytes,
93    &[u8] => Bytes,
94}
95
96impl From<Vec<String>> for Param {
97    fn from(values: Vec<String>) -> Self {
98        Param::TextArray(values.into_iter().map(Some).collect())
99    }
100}
101
102impl From<Vec<&str>> for Param {
103    fn from(values: Vec<&str>) -> Self {
104        Param::TextArray(values.into_iter().map(|v| Some(v.to_string())).collect())
105    }
106}
107
108impl From<Vec<i32>> for Param {
109    fn from(values: Vec<i32>) -> Self {
110        Param::I32Array(values.into_iter().map(Some).collect())
111    }
112}
113
114impl From<Vec<i64>> for Param {
115    fn from(values: Vec<i64>) -> Self {
116        Param::I64Array(values.into_iter().map(Some).collect())
117    }
118}
119
120impl<T: Into<Param>> From<Option<T>> for Param {
121    fn from(value: Option<T>) -> Self {
122        value.map_or(Param::Null, Into::into)
123    }
124}
125
126#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
127#[serde(rename_all = "snake_case")]
128pub enum Isolation {
129    #[default]
130    ReadCommitted,
131    RepeatableRead,
132    Serializable,
133}
134
135/// One `db_call` request.
136#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
137#[serde(tag = "op", rename_all = "snake_case")]
138pub enum DbRequest {
139    Query {
140        sql: String,
141        #[serde(default)]
142        params: Vec<Param>,
143        #[serde(default)]
144        tx: Option<u32>,
145    },
146    Execute {
147        sql: String,
148        #[serde(default)]
149        params: Vec<Param>,
150        #[serde(default)]
151        tx: Option<u32>,
152    },
153    Begin {
154        #[serde(default)]
155        isolation: Isolation,
156        #[serde(default)]
157        read_only: bool,
158    },
159    Commit {
160        tx: u32,
161    },
162    Rollback {
163        tx: u32,
164    },
165    Savepoint {
166        tx: u32,
167        name: String,
168    },
169    Release {
170        tx: u32,
171        name: String,
172    },
173    RollbackTo {
174        tx: u32,
175        name: String,
176    },
177}
178
179impl DbRequest {
180    /// The `op` tag, which is also the metric label.
181    pub fn op(&self) -> &'static str {
182        match self {
183            DbRequest::Query { .. } => "query",
184            DbRequest::Execute { .. } => "execute",
185            DbRequest::Begin { .. } => "begin",
186            DbRequest::Commit { .. } => "commit",
187            DbRequest::Rollback { .. } => "rollback",
188            DbRequest::Savepoint { .. } => "savepoint",
189            DbRequest::Release { .. } => "release",
190            DbRequest::RollbackTo { .. } => "rollback_to",
191        }
192    }
193}
194
195/// A result column: its name and SQL type name.
196#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
197pub struct Column {
198    pub name: String,
199    #[serde(rename = "type")]
200    pub type_name: String,
201}
202
203pub type RowMap = Map<String, Value>;
204
205/// One `db_call` response.
206#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
207#[serde(into = "WireResponse", try_from = "WireResponse")]
208pub enum DbResponse {
209    Rows {
210        rows: Vec<RowMap>,
211        columns: Vec<Column>,
212    },
213    Affected(u64),
214    Begun(u32),
215    Done,
216    Failed(DbError),
217}
218
219#[derive(Serialize, Deserialize)]
220struct WireResponse {
221    ok: bool,
222    #[serde(default, skip_serializing_if = "Option::is_none")]
223    rows: Option<Vec<RowMap>>,
224    #[serde(default, skip_serializing_if = "Option::is_none")]
225    columns: Option<Vec<Column>>,
226    #[serde(default, skip_serializing_if = "Option::is_none")]
227    affected: Option<u64>,
228    #[serde(default, skip_serializing_if = "Option::is_none")]
229    tx: Option<u32>,
230    #[serde(default, skip_serializing_if = "Option::is_none")]
231    error: Option<DbError>,
232}
233
234impl WireResponse {
235    fn ok() -> Self {
236        Self {
237            ok: true,
238            rows: None,
239            columns: None,
240            affected: None,
241            tx: None,
242            error: None,
243        }
244    }
245}
246
247impl From<DbResponse> for WireResponse {
248    fn from(response: DbResponse) -> Self {
249        match response {
250            DbResponse::Rows { rows, columns } => Self {
251                rows: Some(rows),
252                columns: Some(columns),
253                ..Self::ok()
254            },
255            DbResponse::Affected(affected) => Self {
256                affected: Some(affected),
257                ..Self::ok()
258            },
259            DbResponse::Begun(tx) => Self {
260                tx: Some(tx),
261                ..Self::ok()
262            },
263            DbResponse::Done => Self::ok(),
264            DbResponse::Failed(error) => Self {
265                ok: false,
266                error: Some(error),
267                ..Self::ok()
268            },
269        }
270    }
271}
272
273impl TryFrom<WireResponse> for DbResponse {
274    type Error = String;
275
276    fn try_from(wire: WireResponse) -> Result<Self, String> {
277        match wire {
278            WireResponse {
279                ok: false,
280                error: Some(error),
281                ..
282            } => Ok(DbResponse::Failed(error)),
283            WireResponse { ok: false, .. } => Err("a failed response carries no error".into()),
284            WireResponse {
285                rows: Some(rows),
286                columns,
287                ..
288            } => Ok(DbResponse::Rows {
289                rows,
290                columns: columns.unwrap_or_default(),
291            }),
292            WireResponse {
293                affected: Some(affected),
294                ..
295            } => Ok(DbResponse::Affected(affected)),
296            WireResponse { tx: Some(tx), .. } => Ok(DbResponse::Begun(tx)),
297            WireResponse { .. } => Ok(DbResponse::Done),
298        }
299    }
300}
301
302/// A database failure as the guest sees it: the Postgres error fields when
303/// the server reported one, and a [`DbErrorKind`] either way. Boxed so a
304/// `Result<_, DbError>` stays a pointer wide; the fields read through
305/// `Deref`.
306#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
307#[serde(transparent)]
308pub struct DbError(Box<DbErrorFields>);
309
310#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
311pub struct DbErrorFields {
312    pub code: Option<String>,
313    pub kind: DbErrorKind,
314    pub message: String,
315    pub detail: Option<String>,
316    pub hint: Option<String>,
317    pub constraint: Option<String>,
318    pub table: Option<String>,
319    pub column: Option<String>,
320    pub schema: Option<String>,
321    pub position: Option<u32>,
322}
323
324impl DbErrorFields {
325    /// An error raised outside the server, with no Postgres fields.
326    pub fn new(kind: DbErrorKind, message: impl Into<String>) -> Self {
327        Self {
328            code: None,
329            kind,
330            message: message.into(),
331            detail: None,
332            hint: None,
333            constraint: None,
334            table: None,
335            column: None,
336            schema: None,
337            position: None,
338        }
339    }
340}
341
342impl From<DbErrorFields> for DbError {
343    fn from(fields: DbErrorFields) -> Self {
344        Self(Box::new(fields))
345    }
346}
347
348impl std::ops::Deref for DbError {
349    type Target = DbErrorFields;
350
351    fn deref(&self) -> &DbErrorFields {
352        &self.0
353    }
354}
355
356impl std::ops::DerefMut for DbError {
357    fn deref_mut(&mut self) -> &mut DbErrorFields {
358        &mut self.0
359    }
360}
361
362impl DbError {
363    /// An error raised outside the server, with no Postgres fields.
364    pub fn new(kind: DbErrorKind, message: impl Into<String>) -> Self {
365        DbErrorFields::new(kind, message).into()
366    }
367
368    /// What every database call returns where there is no platform to run it.
369    pub fn unavailable() -> Self {
370        Self::new(
371            DbErrorKind::Unavailable,
372            "the database is only available inside the Cufflink platform",
373        )
374    }
375
376    /// A serialization failure or deadlock: the transaction may succeed if
377    /// run again.
378    pub fn is_retryable(&self) -> bool {
379        matches!(
380            self.kind,
381            DbErrorKind::SerializationFailure | DbErrorKind::DeadlockDetected
382        )
383    }
384
385    pub fn is_constraint_violation(&self) -> bool {
386        matches!(
387            self.kind,
388            DbErrorKind::UniqueViolation
389                | DbErrorKind::ForeignKeyViolation
390                | DbErrorKind::CheckViolation
391                | DbErrorKind::NotNullViolation
392                | DbErrorKind::ExclusionViolation
393        )
394    }
395}
396
397impl fmt::Display for DbError {
398    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
399        write!(f, "{}: {}", self.kind.as_str(), self.message)?;
400        match &self.code {
401            Some(code) => write!(f, " (SQLSTATE {code})"),
402            None => Ok(()),
403        }
404    }
405}
406
407impl std::error::Error for DbError {}
408
409macro_rules! error_kinds {
410    ($($variant:ident => $label:literal),* $(,)?) => {
411        #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
412        #[serde(rename_all = "snake_case")]
413        pub enum DbErrorKind {
414            $($variant,)*
415            /// A kind this SDK does not know, from a newer platform.
416            #[serde(other)]
417            Other,
418        }
419
420        impl DbErrorKind {
421            pub fn as_str(self) -> &'static str {
422                match self {
423                    $(DbErrorKind::$variant => $label,)*
424                    DbErrorKind::Other => "other",
425                }
426            }
427        }
428    };
429}
430
431error_kinds! {
432    UniqueViolation => "unique_violation",
433    ForeignKeyViolation => "foreign_key_violation",
434    CheckViolation => "check_violation",
435    NotNullViolation => "not_null_violation",
436    ExclusionViolation => "exclusion_violation",
437    SerializationFailure => "serialization_failure",
438    DeadlockDetected => "deadlock_detected",
439    QueryCanceled => "query_canceled",
440    LockNotAvailable => "lock_not_available",
441    InvalidTextRepresentation => "invalid_text_representation",
442    NumericOutOfRange => "numeric_out_of_range",
443    DatetimeFieldOverflow => "datetime_field_overflow",
444    DataException => "data_exception",
445    SyntaxError => "syntax_error",
446    UndefinedTable => "undefined_table",
447    UndefinedColumn => "undefined_column",
448    InsufficientPrivilege => "insufficient_privilege",
449    RaiseException => "raise_exception",
450    InFailedTransaction => "in_failed_transaction",
451    Connection => "connection",
452    PoolTimeout => "pool_timeout",
453    InvalidParameter => "invalid_parameter",
454    UnsupportedType => "unsupported_type",
455    TooManyRows => "too_many_rows",
456    ColumnNotFound => "column_not_found",
457    Decode => "decode",
458    TransactionNotFound => "transaction_not_found",
459    TransactionLimit => "transaction_limit",
460    NotSupported => "not_supported",
461    Protocol => "protocol",
462    Unavailable => "unavailable",
463    TooLarge => "too_large",
464}
465
466impl DbErrorKind {
467    /// The kind of a Postgres SQLSTATE: the exact codes first, then the
468    /// data-exception and connection classes.
469    pub fn from_sqlstate(code: &str) -> Self {
470        match code {
471            "23505" => Self::UniqueViolation,
472            "23503" => Self::ForeignKeyViolation,
473            "23514" => Self::CheckViolation,
474            "23502" => Self::NotNullViolation,
475            "23P01" => Self::ExclusionViolation,
476            "40001" => Self::SerializationFailure,
477            "40P01" => Self::DeadlockDetected,
478            "57014" => Self::QueryCanceled,
479            "55P03" => Self::LockNotAvailable,
480            "22P02" => Self::InvalidTextRepresentation,
481            "22003" => Self::NumericOutOfRange,
482            "22008" => Self::DatetimeFieldOverflow,
483            "42601" => Self::SyntaxError,
484            "42P01" => Self::UndefinedTable,
485            "42703" => Self::UndefinedColumn,
486            "42501" => Self::InsufficientPrivilege,
487            "P0001" => Self::RaiseException,
488            "25P02" => Self::InFailedTransaction,
489            _ if code.starts_with("22") => Self::DataException,
490            _ if code.starts_with("08") => Self::Connection,
491            _ => Self::Other,
492        }
493    }
494}
495
496impl fmt::Display for DbErrorKind {
497    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
498        f.write_str(self.as_str())
499    }
500}
501
502/// JSON has no NaN or infinities; they travel as the strings Postgres uses.
503mod float {
504    use serde::{Deserialize, Deserializer, Serialize, Serializer};
505
506    #[derive(Serialize, Deserialize)]
507    #[serde(untagged)]
508    pub(super) enum Repr {
509        Number(f64),
510        Special(String),
511    }
512
513    impl Repr {
514        pub(super) fn from_f64(value: f64) -> Self {
515            match value {
516                v if v.is_finite() => Repr::Number(v),
517                v if v.is_nan() => Repr::Special("NaN".into()),
518                v if v > 0.0 => Repr::Special("Infinity".into()),
519                _ => Repr::Special("-Infinity".into()),
520            }
521        }
522
523        pub(super) fn into_f64<E: serde::de::Error>(self) -> Result<f64, E> {
524            match self {
525                Repr::Number(v) => Ok(v),
526                Repr::Special(s) => match s.as_str() {
527                    "NaN" => Ok(f64::NAN),
528                    "Infinity" => Ok(f64::INFINITY),
529                    "-Infinity" => Ok(f64::NEG_INFINITY),
530                    other => Err(E::custom(format!("{other:?} is not a number"))),
531                },
532            }
533        }
534    }
535
536    pub(super) fn serialize<T: Copy + Into<f64>, S: Serializer>(
537        value: &T,
538        serializer: S,
539    ) -> Result<S::Ok, S::Error> {
540        Repr::from_f64((*value).into()).serialize(serializer)
541    }
542
543    pub(super) fn deserialize<'de, T: FromF64, D: Deserializer<'de>>(
544        deserializer: D,
545    ) -> Result<T, D::Error> {
546        Repr::deserialize(deserializer)?.into_f64().map(T::from_f64)
547    }
548
549    pub(super) trait FromF64 {
550        fn from_f64(value: f64) -> Self;
551    }
552
553    impl FromF64 for f64 {
554        fn from_f64(value: f64) -> Self {
555            value
556        }
557    }
558
559    impl FromF64 for f32 {
560        fn from_f64(value: f64) -> Self {
561            value as f32
562        }
563    }
564}
565
566mod float_array {
567    use super::float::Repr;
568    use serde::{Deserialize, Deserializer, Serialize, Serializer};
569
570    pub(super) fn serialize<S: Serializer>(
571        values: &[Option<f64>],
572        serializer: S,
573    ) -> Result<S::Ok, S::Error> {
574        values
575            .iter()
576            .map(|v| v.map(Repr::from_f64))
577            .collect::<Vec<_>>()
578            .serialize(serializer)
579    }
580
581    pub(super) fn deserialize<'de, D: Deserializer<'de>>(
582        deserializer: D,
583    ) -> Result<Vec<Option<f64>>, D::Error> {
584        Vec::<Option<Repr>>::deserialize(deserializer)?
585            .into_iter()
586            .map(|v| v.map(Repr::into_f64).transpose())
587            .collect()
588    }
589}
590
591#[cfg(test)]
592mod tests {
593    use super::*;
594    use serde_json::json;
595
596    fn round_trip<T: Serialize + for<'de> Deserialize<'de> + PartialEq + fmt::Debug>(
597        value: &T,
598        wire: Value,
599    ) {
600        assert_eq!(serde_json::to_value(value).expect("serialize"), wire);
601        assert_eq!(
602            &serde_json::from_value::<T>(wire).expect("deserialize"),
603            value
604        );
605    }
606
607    #[test]
608    fn params_carry_their_tag_and_value() {
609        for (param, wire) in [
610            (Param::Null, json!({"t": "null"})),
611            (Param::Bool(true), json!({"t": "bool", "v": true})),
612            (Param::I16(-3), json!({"t": "i16", "v": -3})),
613            (Param::I64(i64::MAX), json!({"t": "i64", "v": i64::MAX})),
614            (Param::F64(1.5), json!({"t": "f64", "v": 1.5})),
615            (Param::F32(0.5), json!({"t": "f32", "v": 0.5})),
616            (
617                Param::F64(f64::INFINITY),
618                json!({"t": "f64", "v": "Infinity"}),
619            ),
620            (
621                Param::F64(f64::NEG_INFINITY),
622                json!({"t": "f64", "v": "-Infinity"}),
623            ),
624            (
625                Param::Text("it's".into()),
626                json!({"t": "text", "v": "it's"}),
627            ),
628            (Param::Uuid("a".into()), json!({"t": "uuid", "v": "a"})),
629            (
630                Param::Timestamp("2026-01-01T00:00:00Z".into()),
631                json!({"t": "timestamp", "v": "2026-01-01T00:00:00Z"}),
632            ),
633            (
634                Param::NaiveTimestamp("2026-01-01 00:00:00".into()),
635                json!({"t": "naive_timestamp", "v": "2026-01-01 00:00:00"}),
636            ),
637            (
638                Param::Date("2026-01-01".into()),
639                json!({"t": "date", "v": "2026-01-01"}),
640            ),
641            (
642                Param::Time("12:00:00".into()),
643                json!({"t": "time", "v": "12:00:00"}),
644            ),
645            (
646                Param::Json(json!({"a": [1]})),
647                json!({"t": "json", "v": {"a": [1]}}),
648            ),
649            (
650                Param::Numeric("1.50".into()),
651                json!({"t": "numeric", "v": "1.50"}),
652            ),
653            (
654                Param::Bytes(vec![0, 255]),
655                json!({"t": "bytes", "v": "AP8="}),
656            ),
657            (
658                Param::TextArray(vec![Some("a".into()), None]),
659                json!({"t": "text_array", "v": ["a", null]}),
660            ),
661            (
662                Param::UuidArray(vec![None]),
663                json!({"t": "uuid_array", "v": [null]}),
664            ),
665            (
666                Param::I32Array(vec![Some(1)]),
667                json!({"t": "i32_array", "v": [1]}),
668            ),
669            (Param::I64Array(vec![]), json!({"t": "i64_array", "v": []})),
670            (
671                Param::BoolArray(vec![Some(false), None]),
672                json!({"t": "bool_array", "v": [false, null]}),
673            ),
674            (
675                Param::F64Array(vec![Some(2.5), None, Some(f64::INFINITY)]),
676                json!({"t": "f64_array", "v": [2.5, null, "Infinity"]}),
677            ),
678            (
679                Param::NumericArray(vec![Some("0.1".into())]),
680                json!({"t": "numeric_array", "v": ["0.1"]}),
681            ),
682        ] {
683            assert_eq!(wire["t"], param.tag());
684            round_trip(&param, wire);
685        }
686    }
687
688    #[test]
689    fn nan_travels_as_a_string() {
690        let wire = serde_json::to_value(Param::F64(f64::NAN)).expect("serialize");
691        assert_eq!(wire, json!({"t": "f64", "v": "NaN"}));
692        match serde_json::from_value::<Param>(wire).expect("deserialize") {
693            Param::F64(v) => assert!(v.is_nan()),
694            other => panic!("{other:?}"),
695        }
696        assert!(serde_json::from_value::<Param>(json!({"t": "f64", "v": "many"})).is_err());
697        assert!(serde_json::from_value::<Param>(json!({"t": "bytes", "v": "%%"})).is_err());
698    }
699
700    #[test]
701    fn params_convert_from_rust_values() {
702        assert_eq!(Param::from("x"), Param::Text("x".into()));
703        assert_eq!(Param::from(7i64), Param::I64(7));
704        assert_eq!(Param::from(None::<i32>), Param::Null);
705        assert_eq!(Param::from(Some(true)), Param::Bool(true));
706        assert_eq!(Param::from(&b"ab"[..]), Param::Bytes(b"ab".to_vec()));
707        assert_eq!(
708            Param::from(vec!["a", "b"]),
709            Param::TextArray(vec![Some("a".into()), Some("b".into())])
710        );
711        assert_eq!(Param::from(vec![1i32]), Param::I32Array(vec![Some(1)]));
712        assert_eq!(Param::from(json!(1)), Param::Json(json!(1)));
713    }
714
715    #[test]
716    fn requests_are_tagged_by_op() {
717        for (request, wire) in [
718            (
719                DbRequest::Query {
720                    sql: "SELECT $1".into(),
721                    params: vec![Param::I32(1)],
722                    tx: None,
723                },
724                json!({"op": "query", "sql": "SELECT $1", "params": [{"t": "i32", "v": 1}], "tx": null}),
725            ),
726            (
727                DbRequest::Execute {
728                    sql: "DELETE FROM t".into(),
729                    params: vec![],
730                    tx: Some(1),
731                },
732                json!({"op": "execute", "sql": "DELETE FROM t", "params": [], "tx": 1}),
733            ),
734            (
735                DbRequest::Begin {
736                    isolation: Isolation::Serializable,
737                    read_only: true,
738                },
739                json!({"op": "begin", "isolation": "serializable", "read_only": true}),
740            ),
741            (
742                DbRequest::Commit { tx: 1 },
743                json!({"op": "commit", "tx": 1}),
744            ),
745            (
746                DbRequest::Rollback { tx: 2 },
747                json!({"op": "rollback", "tx": 2}),
748            ),
749            (
750                DbRequest::Savepoint {
751                    tx: 1,
752                    name: "sp_0".into(),
753                },
754                json!({"op": "savepoint", "tx": 1, "name": "sp_0"}),
755            ),
756            (
757                DbRequest::Release {
758                    tx: 1,
759                    name: "sp_0".into(),
760                },
761                json!({"op": "release", "tx": 1, "name": "sp_0"}),
762            ),
763            (
764                DbRequest::RollbackTo {
765                    tx: 1,
766                    name: "sp_0".into(),
767                },
768                json!({"op": "rollback_to", "tx": 1, "name": "sp_0"}),
769            ),
770        ] {
771            assert_eq!(wire["op"], request.op());
772            round_trip(&request, wire);
773        }
774        assert_eq!(
775            serde_json::from_value::<DbRequest>(json!({"op": "query", "sql": "SELECT 1"}))
776                .expect("defaults"),
777            DbRequest::Query {
778                sql: "SELECT 1".into(),
779                params: vec![],
780                tx: None
781            }
782        );
783        assert_eq!(
784            serde_json::from_value::<DbRequest>(json!({"op": "begin"})).expect("defaults"),
785            DbRequest::Begin {
786                isolation: Isolation::ReadCommitted,
787                read_only: false
788            }
789        );
790        assert!(serde_json::from_value::<DbRequest>(json!({"op": "vacuum"})).is_err());
791    }
792
793    #[test]
794    fn responses_match_the_wire_shapes() {
795        let unique = DbError::from(DbErrorFields {
796            code: Some("23505".into()),
797            constraint: Some("books_email_key".into()),
798            table: Some("books".into()),
799            schema: Some("catalog_service".into()),
800            detail: Some("Key (email)=(a) already exists.".into()),
801            ..DbErrorFields::new(DbErrorKind::UniqueViolation, "duplicate key")
802        });
803        for (response, wire) in [
804            (
805                DbResponse::Rows {
806                    rows: vec![json!({"id": 1}).as_object().cloned().expect("object")],
807                    columns: vec![Column {
808                        name: "id".into(),
809                        type_name: "int4".into(),
810                    }],
811                },
812                json!({"ok": true, "rows": [{"id": 1}], "columns": [{"name": "id", "type": "int4"}]}),
813            ),
814            (DbResponse::Affected(3), json!({"ok": true, "affected": 3})),
815            (DbResponse::Begun(1), json!({"ok": true, "tx": 1})),
816            (DbResponse::Done, json!({"ok": true})),
817            (
818                DbResponse::Failed(unique),
819                json!({"ok": false, "error": {
820                    "code": "23505", "kind": "unique_violation", "message": "duplicate key",
821                    "detail": "Key (email)=(a) already exists.", "hint": null,
822                    "constraint": "books_email_key", "table": "books", "column": null,
823                    "schema": "catalog_service", "position": null
824                }}),
825            ),
826        ] {
827            round_trip(&response, wire);
828        }
829        assert_eq!(
830            serde_json::from_value::<DbResponse>(json!({"ok": true, "rows": []}))
831                .expect("columns default"),
832            DbResponse::Rows {
833                rows: vec![],
834                columns: vec![]
835            }
836        );
837        assert!(serde_json::from_value::<DbResponse>(json!({"ok": false})).is_err());
838    }
839
840    #[test]
841    fn unknown_error_kinds_decode_as_other() {
842        let error: DbError = serde_json::from_value(json!({
843            "code": null, "kind": "quantum_decoherence", "message": "m", "detail": null,
844            "hint": null, "constraint": null, "table": null, "column": null,
845            "schema": null, "position": null
846        }))
847        .expect("decode");
848        assert_eq!(error.kind, DbErrorKind::Other);
849    }
850
851    #[test]
852    fn kind_labels_match_their_serde_names() {
853        for kind in [
854            DbErrorKind::UniqueViolation,
855            DbErrorKind::RaiseException,
856            DbErrorKind::InFailedTransaction,
857            DbErrorKind::PoolTimeout,
858            DbErrorKind::TransactionLimit,
859            DbErrorKind::Unavailable,
860            DbErrorKind::TooLarge,
861            DbErrorKind::Other,
862        ] {
863            assert_eq!(
864                serde_json::to_value(kind).expect("kind"),
865                json!(kind.as_str())
866            );
867        }
868    }
869
870    #[test]
871    fn sqlstates_map_to_kinds() {
872        for (code, kind) in [
873            ("23505", DbErrorKind::UniqueViolation),
874            ("23503", DbErrorKind::ForeignKeyViolation),
875            ("23514", DbErrorKind::CheckViolation),
876            ("23502", DbErrorKind::NotNullViolation),
877            ("23P01", DbErrorKind::ExclusionViolation),
878            ("40001", DbErrorKind::SerializationFailure),
879            ("40P01", DbErrorKind::DeadlockDetected),
880            ("57014", DbErrorKind::QueryCanceled),
881            ("55P03", DbErrorKind::LockNotAvailable),
882            ("22P02", DbErrorKind::InvalidTextRepresentation),
883            ("22003", DbErrorKind::NumericOutOfRange),
884            ("22008", DbErrorKind::DatetimeFieldOverflow),
885            ("22012", DbErrorKind::DataException),
886            ("22001", DbErrorKind::DataException),
887            ("42601", DbErrorKind::SyntaxError),
888            ("42P01", DbErrorKind::UndefinedTable),
889            ("42703", DbErrorKind::UndefinedColumn),
890            ("42501", DbErrorKind::InsufficientPrivilege),
891            ("P0001", DbErrorKind::RaiseException),
892            ("25P02", DbErrorKind::InFailedTransaction),
893            ("08006", DbErrorKind::Connection),
894            ("08P01", DbErrorKind::Connection),
895            ("23000", DbErrorKind::Other),
896            ("XX000", DbErrorKind::Other),
897            ("", DbErrorKind::Other),
898        ] {
899            assert_eq!(DbErrorKind::from_sqlstate(code), kind, "{code}");
900        }
901    }
902
903    #[test]
904    fn errors_classify_and_display() {
905        let error = |kind| DbError::new(kind, "m");
906        assert!(error(DbErrorKind::SerializationFailure).is_retryable());
907        assert!(error(DbErrorKind::DeadlockDetected).is_retryable());
908        assert!(!error(DbErrorKind::UniqueViolation).is_retryable());
909        for kind in [
910            DbErrorKind::UniqueViolation,
911            DbErrorKind::ForeignKeyViolation,
912            DbErrorKind::CheckViolation,
913            DbErrorKind::NotNullViolation,
914            DbErrorKind::ExclusionViolation,
915        ] {
916            assert!(error(kind).is_constraint_violation(), "{kind}");
917        }
918        assert!(!error(DbErrorKind::RaiseException).is_constraint_violation());
919        assert_eq!(
920            DbError::from(DbErrorFields {
921                code: Some("57014".into()),
922                ..DbErrorFields::new(DbErrorKind::QueryCanceled, "canceling statement")
923            })
924            .to_string(),
925            "query_canceled: canceling statement (SQLSTATE 57014)"
926        );
927        assert_eq!(DbError::unavailable().kind, DbErrorKind::Unavailable);
928    }
929}
cufflink_types/db.rs

cufflink_types/
db.rs
