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}
464
465impl DbErrorKind {
466    /// The kind of a Postgres SQLSTATE: the exact codes first, then the
467    /// data-exception and connection classes.
468    pub fn from_sqlstate(code: &str) -> Self {
469        match code {
470            "23505" => Self::UniqueViolation,
471            "23503" => Self::ForeignKeyViolation,
472            "23514" => Self::CheckViolation,
473            "23502" => Self::NotNullViolation,
474            "23P01" => Self::ExclusionViolation,
475            "40001" => Self::SerializationFailure,
476            "40P01" => Self::DeadlockDetected,
477            "57014" => Self::QueryCanceled,
478            "55P03" => Self::LockNotAvailable,
479            "22P02" => Self::InvalidTextRepresentation,
480            "22003" => Self::NumericOutOfRange,
481            "22008" => Self::DatetimeFieldOverflow,
482            "42601" => Self::SyntaxError,
483            "42P01" => Self::UndefinedTable,
484            "42703" => Self::UndefinedColumn,
485            "42501" => Self::InsufficientPrivilege,
486            "P0001" => Self::RaiseException,
487            "25P02" => Self::InFailedTransaction,
488            _ if code.starts_with("22") => Self::DataException,
489            _ if code.starts_with("08") => Self::Connection,
490            _ => Self::Other,
491        }
492    }
493}
494
495impl fmt::Display for DbErrorKind {
496    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
497        f.write_str(self.as_str())
498    }
499}
500
501/// JSON has no NaN or infinities; they travel as the strings Postgres uses.
502mod float {
503    use serde::{Deserialize, Deserializer, Serialize, Serializer};
504
505    #[derive(Serialize, Deserialize)]
506    #[serde(untagged)]
507    pub(super) enum Repr {
508        Number(f64),
509        Special(String),
510    }
511
512    impl Repr {
513        pub(super) fn from_f64(value: f64) -> Self {
514            match value {
515                v if v.is_finite() => Repr::Number(v),
516                v if v.is_nan() => Repr::Special("NaN".into()),
517                v if v > 0.0 => Repr::Special("Infinity".into()),
518                _ => Repr::Special("-Infinity".into()),
519            }
520        }
521
522        pub(super) fn into_f64<E: serde::de::Error>(self) -> Result<f64, E> {
523            match self {
524                Repr::Number(v) => Ok(v),
525                Repr::Special(s) => match s.as_str() {
526                    "NaN" => Ok(f64::NAN),
527                    "Infinity" => Ok(f64::INFINITY),
528                    "-Infinity" => Ok(f64::NEG_INFINITY),
529                    other => Err(E::custom(format!("{other:?} is not a number"))),
530                },
531            }
532        }
533    }
534
535    pub(super) fn serialize<T: Copy + Into<f64>, S: Serializer>(
536        value: &T,
537        serializer: S,
538    ) -> Result<S::Ok, S::Error> {
539        Repr::from_f64((*value).into()).serialize(serializer)
540    }
541
542    pub(super) fn deserialize<'de, T: FromF64, D: Deserializer<'de>>(
543        deserializer: D,
544    ) -> Result<T, D::Error> {
545        Repr::deserialize(deserializer)?.into_f64().map(T::from_f64)
546    }
547
548    pub(super) trait FromF64 {
549        fn from_f64(value: f64) -> Self;
550    }
551
552    impl FromF64 for f64 {
553        fn from_f64(value: f64) -> Self {
554            value
555        }
556    }
557
558    impl FromF64 for f32 {
559        fn from_f64(value: f64) -> Self {
560            value as f32
561        }
562    }
563}
564
565mod float_array {
566    use super::float::Repr;
567    use serde::{Deserialize, Deserializer, Serialize, Serializer};
568
569    pub(super) fn serialize<S: Serializer>(
570        values: &[Option<f64>],
571        serializer: S,
572    ) -> Result<S::Ok, S::Error> {
573        values
574            .iter()
575            .map(|v| v.map(Repr::from_f64))
576            .collect::<Vec<_>>()
577            .serialize(serializer)
578    }
579
580    pub(super) fn deserialize<'de, D: Deserializer<'de>>(
581        deserializer: D,
582    ) -> Result<Vec<Option<f64>>, D::Error> {
583        Vec::<Option<Repr>>::deserialize(deserializer)?
584            .into_iter()
585            .map(|v| v.map(Repr::into_f64).transpose())
586            .collect()
587    }
588}
589
590#[cfg(test)]
591mod tests {
592    use super::*;
593    use serde_json::json;
594
595    fn round_trip<T: Serialize + for<'de> Deserialize<'de> + PartialEq + fmt::Debug>(
596        value: &T,
597        wire: Value,
598    ) {
599        assert_eq!(serde_json::to_value(value).expect("serialize"), wire);
600        assert_eq!(
601            &serde_json::from_value::<T>(wire).expect("deserialize"),
602            value
603        );
604    }
605
606    #[test]
607    fn params_carry_their_tag_and_value() {
608        for (param, wire) in [
609            (Param::Null, json!({"t": "null"})),
610            (Param::Bool(true), json!({"t": "bool", "v": true})),
611            (Param::I16(-3), json!({"t": "i16", "v": -3})),
612            (Param::I64(i64::MAX), json!({"t": "i64", "v": i64::MAX})),
613            (Param::F64(1.5), json!({"t": "f64", "v": 1.5})),
614            (Param::F32(0.5), json!({"t": "f32", "v": 0.5})),
615            (
616                Param::F64(f64::INFINITY),
617                json!({"t": "f64", "v": "Infinity"}),
618            ),
619            (
620                Param::F64(f64::NEG_INFINITY),
621                json!({"t": "f64", "v": "-Infinity"}),
622            ),
623            (
624                Param::Text("it's".into()),
625                json!({"t": "text", "v": "it's"}),
626            ),
627            (Param::Uuid("a".into()), json!({"t": "uuid", "v": "a"})),
628            (
629                Param::Timestamp("2026-01-01T00:00:00Z".into()),
630                json!({"t": "timestamp", "v": "2026-01-01T00:00:00Z"}),
631            ),
632            (
633                Param::NaiveTimestamp("2026-01-01 00:00:00".into()),
634                json!({"t": "naive_timestamp", "v": "2026-01-01 00:00:00"}),
635            ),
636            (
637                Param::Date("2026-01-01".into()),
638                json!({"t": "date", "v": "2026-01-01"}),
639            ),
640            (
641                Param::Time("12:00:00".into()),
642                json!({"t": "time", "v": "12:00:00"}),
643            ),
644            (
645                Param::Json(json!({"a": [1]})),
646                json!({"t": "json", "v": {"a": [1]}}),
647            ),
648            (
649                Param::Numeric("1.50".into()),
650                json!({"t": "numeric", "v": "1.50"}),
651            ),
652            (
653                Param::Bytes(vec![0, 255]),
654                json!({"t": "bytes", "v": "AP8="}),
655            ),
656            (
657                Param::TextArray(vec![Some("a".into()), None]),
658                json!({"t": "text_array", "v": ["a", null]}),
659            ),
660            (
661                Param::UuidArray(vec![None]),
662                json!({"t": "uuid_array", "v": [null]}),
663            ),
664            (
665                Param::I32Array(vec![Some(1)]),
666                json!({"t": "i32_array", "v": [1]}),
667            ),
668            (Param::I64Array(vec![]), json!({"t": "i64_array", "v": []})),
669            (
670                Param::BoolArray(vec![Some(false), None]),
671                json!({"t": "bool_array", "v": [false, null]}),
672            ),
673            (
674                Param::F64Array(vec![Some(2.5), None, Some(f64::INFINITY)]),
675                json!({"t": "f64_array", "v": [2.5, null, "Infinity"]}),
676            ),
677            (
678                Param::NumericArray(vec![Some("0.1".into())]),
679                json!({"t": "numeric_array", "v": ["0.1"]}),
680            ),
681        ] {
682            assert_eq!(wire["t"], param.tag());
683            round_trip(&param, wire);
684        }
685    }
686
687    #[test]
688    fn nan_travels_as_a_string() {
689        let wire = serde_json::to_value(Param::F64(f64::NAN)).expect("serialize");
690        assert_eq!(wire, json!({"t": "f64", "v": "NaN"}));
691        match serde_json::from_value::<Param>(wire).expect("deserialize") {
692            Param::F64(v) => assert!(v.is_nan()),
693            other => panic!("{other:?}"),
694        }
695        assert!(serde_json::from_value::<Param>(json!({"t": "f64", "v": "many"})).is_err());
696        assert!(serde_json::from_value::<Param>(json!({"t": "bytes", "v": "%%"})).is_err());
697    }
698
699    #[test]
700    fn params_convert_from_rust_values() {
701        assert_eq!(Param::from("x"), Param::Text("x".into()));
702        assert_eq!(Param::from(7i64), Param::I64(7));
703        assert_eq!(Param::from(None::<i32>), Param::Null);
704        assert_eq!(Param::from(Some(true)), Param::Bool(true));
705        assert_eq!(Param::from(&b"ab"[..]), Param::Bytes(b"ab".to_vec()));
706        assert_eq!(
707            Param::from(vec!["a", "b"]),
708            Param::TextArray(vec![Some("a".into()), Some("b".into())])
709        );
710        assert_eq!(Param::from(vec![1i32]), Param::I32Array(vec![Some(1)]));
711        assert_eq!(Param::from(json!(1)), Param::Json(json!(1)));
712    }
713
714    #[test]
715    fn requests_are_tagged_by_op() {
716        for (request, wire) in [
717            (
718                DbRequest::Query {
719                    sql: "SELECT $1".into(),
720                    params: vec![Param::I32(1)],
721                    tx: None,
722                },
723                json!({"op": "query", "sql": "SELECT $1", "params": [{"t": "i32", "v": 1}], "tx": null}),
724            ),
725            (
726                DbRequest::Execute {
727                    sql: "DELETE FROM t".into(),
728                    params: vec![],
729                    tx: Some(1),
730                },
731                json!({"op": "execute", "sql": "DELETE FROM t", "params": [], "tx": 1}),
732            ),
733            (
734                DbRequest::Begin {
735                    isolation: Isolation::Serializable,
736                    read_only: true,
737                },
738                json!({"op": "begin", "isolation": "serializable", "read_only": true}),
739            ),
740            (
741                DbRequest::Commit { tx: 1 },
742                json!({"op": "commit", "tx": 1}),
743            ),
744            (
745                DbRequest::Rollback { tx: 2 },
746                json!({"op": "rollback", "tx": 2}),
747            ),
748            (
749                DbRequest::Savepoint {
750                    tx: 1,
751                    name: "sp_0".into(),
752                },
753                json!({"op": "savepoint", "tx": 1, "name": "sp_0"}),
754            ),
755            (
756                DbRequest::Release {
757                    tx: 1,
758                    name: "sp_0".into(),
759                },
760                json!({"op": "release", "tx": 1, "name": "sp_0"}),
761            ),
762            (
763                DbRequest::RollbackTo {
764                    tx: 1,
765                    name: "sp_0".into(),
766                },
767                json!({"op": "rollback_to", "tx": 1, "name": "sp_0"}),
768            ),
769        ] {
770            assert_eq!(wire["op"], request.op());
771            round_trip(&request, wire);
772        }
773        assert_eq!(
774            serde_json::from_value::<DbRequest>(json!({"op": "query", "sql": "SELECT 1"}))
775                .expect("defaults"),
776            DbRequest::Query {
777                sql: "SELECT 1".into(),
778                params: vec![],
779                tx: None
780            }
781        );
782        assert_eq!(
783            serde_json::from_value::<DbRequest>(json!({"op": "begin"})).expect("defaults"),
784            DbRequest::Begin {
785                isolation: Isolation::ReadCommitted,
786                read_only: false
787            }
788        );
789        assert!(serde_json::from_value::<DbRequest>(json!({"op": "vacuum"})).is_err());
790    }
791
792    #[test]
793    fn responses_match_the_wire_shapes() {
794        let unique = DbError::from(DbErrorFields {
795            code: Some("23505".into()),
796            constraint: Some("leads_email_key".into()),
797            table: Some("leads".into()),
798            schema: Some("leads_service".into()),
799            detail: Some("Key (email)=(a) already exists.".into()),
800            ..DbErrorFields::new(DbErrorKind::UniqueViolation, "duplicate key")
801        });
802        for (response, wire) in [
803            (
804                DbResponse::Rows {
805                    rows: vec![json!({"id": 1}).as_object().cloned().expect("object")],
806                    columns: vec![Column {
807                        name: "id".into(),
808                        type_name: "int4".into(),
809                    }],
810                },
811                json!({"ok": true, "rows": [{"id": 1}], "columns": [{"name": "id", "type": "int4"}]}),
812            ),
813            (DbResponse::Affected(3), json!({"ok": true, "affected": 3})),
814            (DbResponse::Begun(1), json!({"ok": true, "tx": 1})),
815            (DbResponse::Done, json!({"ok": true})),
816            (
817                DbResponse::Failed(unique),
818                json!({"ok": false, "error": {
819                    "code": "23505", "kind": "unique_violation", "message": "duplicate key",
820                    "detail": "Key (email)=(a) already exists.", "hint": null,
821                    "constraint": "leads_email_key", "table": "leads", "column": null,
822                    "schema": "leads_service", "position": null
823                }}),
824            ),
825        ] {
826            round_trip(&response, wire);
827        }
828        assert_eq!(
829            serde_json::from_value::<DbResponse>(json!({"ok": true, "rows": []}))
830                .expect("columns default"),
831            DbResponse::Rows {
832                rows: vec![],
833                columns: vec![]
834            }
835        );
836        assert!(serde_json::from_value::<DbResponse>(json!({"ok": false})).is_err());
837    }
838
839    #[test]
840    fn unknown_error_kinds_decode_as_other() {
841        let error: DbError = serde_json::from_value(json!({
842            "code": null, "kind": "quantum_decoherence", "message": "m", "detail": null,
843            "hint": null, "constraint": null, "table": null, "column": null,
844            "schema": null, "position": null
845        }))
846        .expect("decode");
847        assert_eq!(error.kind, DbErrorKind::Other);
848    }
849
850    #[test]
851    fn kind_labels_match_their_serde_names() {
852        for kind in [
853            DbErrorKind::UniqueViolation,
854            DbErrorKind::RaiseException,
855            DbErrorKind::InFailedTransaction,
856            DbErrorKind::PoolTimeout,
857            DbErrorKind::TransactionLimit,
858            DbErrorKind::Unavailable,
859            DbErrorKind::Other,
860        ] {
861            assert_eq!(
862                serde_json::to_value(kind).expect("kind"),
863                json!(kind.as_str())
864            );
865        }
866    }
867
868    #[test]
869    fn sqlstates_map_to_kinds() {
870        for (code, kind) in [
871            ("23505", DbErrorKind::UniqueViolation),
872            ("23503", DbErrorKind::ForeignKeyViolation),
873            ("23514", DbErrorKind::CheckViolation),
874            ("23502", DbErrorKind::NotNullViolation),
875            ("23P01", DbErrorKind::ExclusionViolation),
876            ("40001", DbErrorKind::SerializationFailure),
877            ("40P01", DbErrorKind::DeadlockDetected),
878            ("57014", DbErrorKind::QueryCanceled),
879            ("55P03", DbErrorKind::LockNotAvailable),
880            ("22P02", DbErrorKind::InvalidTextRepresentation),
881            ("22003", DbErrorKind::NumericOutOfRange),
882            ("22008", DbErrorKind::DatetimeFieldOverflow),
883            ("22012", DbErrorKind::DataException),
884            ("22001", DbErrorKind::DataException),
885            ("42601", DbErrorKind::SyntaxError),
886            ("42P01", DbErrorKind::UndefinedTable),
887            ("42703", DbErrorKind::UndefinedColumn),
888            ("42501", DbErrorKind::InsufficientPrivilege),
889            ("P0001", DbErrorKind::RaiseException),
890            ("25P02", DbErrorKind::InFailedTransaction),
891            ("08006", DbErrorKind::Connection),
892            ("08P01", DbErrorKind::Connection),
893            ("23000", DbErrorKind::Other),
894            ("XX000", DbErrorKind::Other),
895            ("", DbErrorKind::Other),
896        ] {
897            assert_eq!(DbErrorKind::from_sqlstate(code), kind, "{code}");
898        }
899    }
900
901    #[test]
902    fn errors_classify_and_display() {
903        let error = |kind| DbError::new(kind, "m");
904        assert!(error(DbErrorKind::SerializationFailure).is_retryable());
905        assert!(error(DbErrorKind::DeadlockDetected).is_retryable());
906        assert!(!error(DbErrorKind::UniqueViolation).is_retryable());
907        for kind in [
908            DbErrorKind::UniqueViolation,
909            DbErrorKind::ForeignKeyViolation,
910            DbErrorKind::CheckViolation,
911            DbErrorKind::NotNullViolation,
912            DbErrorKind::ExclusionViolation,
913        ] {
914            assert!(error(kind).is_constraint_violation(), "{kind}");
915        }
916        assert!(!error(DbErrorKind::RaiseException).is_constraint_violation());
917        assert_eq!(
918            DbError::from(DbErrorFields {
919                code: Some("57014".into()),
920                ..DbErrorFields::new(DbErrorKind::QueryCanceled, "canceling statement")
921            })
922            .to_string(),
923            "query_canceled: canceling statement (SQLSTATE 57014)"
924        );
925        assert_eq!(DbError::unavailable().kind, DbErrorKind::Unavailable);
926    }
927}
cufflink_types/db.rs

cufflink_types/
db.rs
