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

cufflink_types/
db.rs
