drasi_postgres_common/
value.rs

1// Copyright 2025 The Drasi Authors.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Canonical PostgreSQL value representation and ElementValue conversion.
16
17use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
18use drasi_core::models::ElementValue;
19use ordered_float::OrderedFloat;
20use rust_decimal::prelude::ToPrimitive;
21use rust_decimal::Decimal;
22use serde_json::Value as JsonValue;
23use std::sync::Arc;
24use uuid::Uuid;
25
26/// Intermediate typed value shared by bootstrap and CDC paths.
27#[derive(Debug, Clone)]
28pub enum PostgresValue {
29    Null,
30    Bool(bool),
31    Int2(i16),
32    Int4(i32),
33    Int8(i64),
34    Float4(f32),
35    Float8(f64),
36    Numeric(Decimal),
37    Text(String),
38    Varchar(String),
39    Char(String),
40    Uuid(Uuid),
41    Timestamp(NaiveDateTime),
42    TimestampTz(DateTime<Utc>),
43    Date(NaiveDate),
44    Time(NaiveTime),
45    Json(JsonValue),
46    Jsonb(JsonValue),
47    Array(Vec<PostgresValue>),
48    Bytea(Vec<u8>),
49    /// Column present in a pgoutput tuple but value was not sent (unchanged TOAST).
50    /// Must not be mapped to [`ElementValue::Null`]; property should be omitted so
51    /// callers can preserve the previously stored value.
52    UnchangedToast,
53}
54
55impl PostgresValue {
56    /// Convert to JSON (lossy for temporals). Prefer [`to_element_value`] for Drasi paths.
57    pub fn to_json(&self) -> JsonValue {
58        match self {
59            PostgresValue::Null => JsonValue::Null,
60            // Not a SQL NULL — omit at property-map layer; JSON null only if forced.
61            PostgresValue::UnchangedToast => JsonValue::Null,
62            PostgresValue::Bool(b) => JsonValue::Bool(*b),
63            PostgresValue::Int2(i) => JsonValue::Number((*i).into()),
64            PostgresValue::Int4(i) => JsonValue::Number((*i).into()),
65            PostgresValue::Int8(i) => JsonValue::Number((*i).into()),
66            PostgresValue::Float4(f) => serde_json::Number::from_f64(*f as f64)
67                .map(JsonValue::Number)
68                .unwrap_or(JsonValue::Null),
69            PostgresValue::Float8(f) => serde_json::Number::from_f64(*f)
70                .map(JsonValue::Number)
71                .unwrap_or(JsonValue::Null),
72            PostgresValue::Numeric(d) => d
73                .to_string()
74                .parse::<serde_json::Number>()
75                .map(JsonValue::Number)
76                .unwrap_or(JsonValue::Null),
77            PostgresValue::Text(s) | PostgresValue::Varchar(s) | PostgresValue::Char(s) => {
78                JsonValue::String(s.clone())
79            }
80            PostgresValue::Uuid(u) => JsonValue::String(u.to_string()),
81            PostgresValue::Timestamp(ts) => JsonValue::String(ts.to_string()),
82            PostgresValue::TimestampTz(ts) => JsonValue::String(ts.to_rfc3339()),
83            PostgresValue::Date(d) => JsonValue::String(d.to_string()),
84            PostgresValue::Time(t) => JsonValue::String(t.to_string()),
85            PostgresValue::Json(j) | PostgresValue::Jsonb(j) => j.clone(),
86            PostgresValue::Array(arr) => {
87                JsonValue::Array(arr.iter().map(|v| v.to_json()).collect())
88            }
89            PostgresValue::Bytea(bytes) => JsonValue::String(encode_base64(bytes)),
90        }
91    }
92
93    /// Canonical conversion used by both bootstrap and CDC.
94    ///
95    /// Mapping:
96    /// - `Timestamp` → `LocalDateTime`, `TimestampTz` → `ZonedDateTime`
97    /// - `Numeric` → `Float` (always, including whole numbers)
98    /// - `Date`, `Time`, `Uuid` → `String`
99    /// - `Json` / `Jsonb` → `String` (compact JSON text)
100    /// - `Bytea` → `String` (base64 of raw bytes)
101    /// - `Char` is expected to already be trimmed by the decoder
102    pub fn to_element_value(&self) -> ElementValue {
103        match self {
104            PostgresValue::Null => ElementValue::Null,
105            // Callers must omit UnchangedToast properties; this arm is defensive only.
106            PostgresValue::UnchangedToast => ElementValue::Null,
107            PostgresValue::Bool(b) => ElementValue::Bool(*b),
108            PostgresValue::Int2(i) => ElementValue::Integer(*i as i64),
109            PostgresValue::Int4(i) => ElementValue::Integer(*i as i64),
110            PostgresValue::Int8(i) => ElementValue::Integer(*i),
111            PostgresValue::Float4(f) => ElementValue::Float(OrderedFloat(*f as f64)),
112            PostgresValue::Float8(f) => ElementValue::Float(OrderedFloat(*f)),
113            PostgresValue::Numeric(d) => {
114                ElementValue::Float(OrderedFloat(d.to_f64().unwrap_or(f64::NAN)))
115            }
116            PostgresValue::Text(s) | PostgresValue::Varchar(s) | PostgresValue::Char(s) => {
117                ElementValue::String(Arc::from(s.as_str()))
118            }
119            PostgresValue::Uuid(u) => ElementValue::String(Arc::from(u.to_string())),
120            PostgresValue::Timestamp(ts) => ElementValue::LocalDateTime(*ts),
121            PostgresValue::TimestampTz(ts) => ElementValue::ZonedDateTime(ts.fixed_offset()),
122            PostgresValue::Date(d) => ElementValue::String(Arc::from(d.to_string())),
123            PostgresValue::Time(t) => ElementValue::String(Arc::from(t.to_string())),
124            PostgresValue::Json(j) | PostgresValue::Jsonb(j) => {
125                ElementValue::String(Arc::from(j.to_string()))
126            }
127            PostgresValue::Array(arr) => {
128                ElementValue::List(arr.iter().map(|v| v.to_element_value()).collect())
129            }
130            PostgresValue::Bytea(bytes) => ElementValue::String(Arc::from(encode_base64(bytes))),
131        }
132    }
133
134    /// Returns `true` if this value is [`PostgresValue::Null`].
135    pub fn is_null(&self) -> bool {
136        matches!(self, PostgresValue::Null)
137    }
138
139    /// Returns `true` if this is an unchanged TOAST placeholder from pgoutput (`b'u'`).
140    pub fn is_unchanged_toast(&self) -> bool {
141        matches!(self, PostgresValue::UnchangedToast)
142    }
143
144    /// Stable string form for element ID key parts.
145    pub fn to_key_string(&self) -> Option<String> {
146        match self {
147            PostgresValue::Null | PostgresValue::UnchangedToast => None,
148            PostgresValue::Bool(b) => Some(b.to_string()),
149            PostgresValue::Int2(i) => Some(i.to_string()),
150            PostgresValue::Int4(i) => Some(i.to_string()),
151            PostgresValue::Int8(i) => Some(i.to_string()),
152            PostgresValue::Float4(f) => Some(f.to_string()),
153            PostgresValue::Float8(f) => Some(f.to_string()),
154            PostgresValue::Numeric(d) => Some(d.to_string()),
155            PostgresValue::Text(s) | PostgresValue::Varchar(s) | PostgresValue::Char(s) => {
156                Some(s.clone())
157            }
158            PostgresValue::Uuid(u) => Some(u.to_string()),
159            PostgresValue::Timestamp(ts) => Some(ts.to_string()),
160            PostgresValue::TimestampTz(ts) => Some(ts.to_rfc3339()),
161            PostgresValue::Date(d) => Some(d.to_string()),
162            PostgresValue::Time(t) => Some(t.to_string()),
163            PostgresValue::Json(j) | PostgresValue::Jsonb(j) => Some(j.to_string()),
164            PostgresValue::Bytea(bytes) => Some(encode_base64(bytes)),
165            PostgresValue::Array(_) => Some(self.to_json().to_string()),
166        }
167    }
168}
169
170/// Base64-encode bytes (standard alphabet with padding).
171pub fn encode_base64(input: &[u8]) -> String {
172    use base64::Engine;
173    base64::engine::general_purpose::STANDARD.encode(input)
174}
175
176/// Parse PostgreSQL text-format bytea (`\xdeadbeef` or escaped).
177pub fn parse_bytea_text(text: &str) -> anyhow::Result<Vec<u8>> {
178    let trimmed = text.trim();
179    if let Some(hex) = trimmed
180        .strip_prefix("\\x")
181        .or_else(|| trimmed.strip_prefix(r"\x"))
182    {
183        if hex.len() % 2 != 0 {
184            anyhow::bail!("Odd-length bytea hex string");
185        }
186        let mut out = Vec::with_capacity(hex.len() / 2);
187        for i in (0..hex.len()).step_by(2) {
188            let byte = u8::from_str_radix(&hex[i..i + 2], 16)
189                .map_err(|e| anyhow::anyhow!("Invalid bytea hex: {e}"))?;
190            out.push(byte);
191        }
192        return Ok(out);
193    }
194    // Fallback: treat as raw UTF-8 bytes of the text representation
195    Ok(trimmed.as_bytes().to_vec())
196}
197
198#[cfg(test)]
199mod tests {
200    use super::*;
201    use chrono::{NaiveDate, TimeZone, Utc};
202    use rust_decimal::Decimal;
203    use std::str::FromStr;
204
205    #[test]
206    fn timestamp_to_element_value_is_local_datetime() {
207        let ts = NaiveDate::from_ymd_opt(2024, 6, 15)
208            .unwrap()
209            .and_hms_opt(10, 30, 45)
210            .unwrap();
211        assert_eq!(
212            PostgresValue::Timestamp(ts).to_element_value(),
213            ElementValue::LocalDateTime(ts)
214        );
215    }
216
217    #[test]
218    fn timestamptz_to_element_value_is_zoned_datetime() {
219        let ts = Utc.with_ymd_and_hms(2024, 6, 15, 10, 30, 45).unwrap();
220        assert_eq!(
221            PostgresValue::TimestampTz(ts).to_element_value(),
222            ElementValue::ZonedDateTime(ts.fixed_offset())
223        );
224    }
225
226    #[test]
227    fn numeric_whole_number_is_float() {
228        let dec = Decimal::from_str("4200").unwrap();
229        match PostgresValue::Numeric(dec).to_element_value() {
230            ElementValue::Float(f) => assert_eq!(f.into_inner(), 4200.0),
231            other => panic!("expected Float, got {other:?}"),
232        }
233    }
234
235    #[test]
236    fn bytea_is_base64_of_raw_bytes() {
237        let bytes = vec![0xde, 0xad, 0xbe, 0xef];
238        let ev = PostgresValue::Bytea(bytes.clone()).to_element_value();
239        assert_eq!(ev, ElementValue::String(Arc::from(encode_base64(&bytes))));
240        // Must not be JSON-quoted
241        if let ElementValue::String(s) = ev {
242            assert!(!s.starts_with('"'));
243        }
244    }
245
246    #[test]
247    fn parse_bytea_hex_text() {
248        assert_eq!(
249            parse_bytea_text(r"\xdeadbeef").unwrap(),
250            vec![0xde, 0xad, 0xbe, 0xef]
251        );
252    }
253
254    #[test]
255    fn null_to_element_value() {
256        assert_eq!(PostgresValue::Null.to_element_value(), ElementValue::Null);
257        assert!(PostgresValue::Null.is_null());
258    }
259
260    #[test]
261    fn uuid_to_element_value_is_string() {
262        let uuid = Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap();
263        assert_eq!(
264            PostgresValue::Uuid(uuid).to_element_value(),
265            ElementValue::String(Arc::from(uuid.to_string()))
266        );
267    }
268
269    #[test]
270    fn json_to_element_value_is_string() {
271        let j = serde_json::json!({"k": 1});
272        let ev = PostgresValue::Json(j.clone()).to_element_value();
273        assert_eq!(ev, ElementValue::String(Arc::from(j.to_string())));
274        let evb = PostgresValue::Jsonb(j.clone()).to_element_value();
275        assert_eq!(evb, ElementValue::String(Arc::from(j.to_string())));
276    }
277
278    #[test]
279    fn array_to_element_value_is_list() {
280        let arr = PostgresValue::Array(vec![
281            PostgresValue::Int4(1),
282            PostgresValue::Null,
283            PostgresValue::Text("x".into()),
284        ]);
285        match arr.to_element_value() {
286            ElementValue::List(items) => {
287                assert_eq!(items.len(), 3);
288                assert_eq!(items[0], ElementValue::Integer(1));
289                assert_eq!(items[1], ElementValue::Null);
290                assert_eq!(items[2], ElementValue::String(Arc::from("x")));
291            }
292            other => panic!("expected List, got {other:?}"),
293        }
294    }
295
296    #[test]
297    fn unchanged_toast_is_not_sql_null() {
298        assert!(!PostgresValue::UnchangedToast.is_null());
299        assert!(PostgresValue::UnchangedToast.is_unchanged_toast());
300        assert_eq!(PostgresValue::UnchangedToast.to_key_string(), None);
301    }
302}
drasi_postgres_common/value.rs

drasi_postgres_common/
value.rs
