nexql_tools/
cell_json.rs

1// SPDX-License-Identifier: GPL-3.0-only
2// Copyright (C) 2026 NexQL-OSS Team
3
4//! Typed Postgres cell → JSON conversion shared by read and write tools.
5
6use nexql_policy::{ObjectRef, PII_REDACTED, column_matches_pii_policy};
7use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
8use chrono::{DateTime, FixedOffset, NaiveDate, NaiveDateTime, NaiveTime};
9use rust_decimal::Decimal;
10use serde_json::{Value, json};
11use tokio_postgres::types::{FromSql, Kind, Type};
12use uuid::Uuid;
13
14/// Convert one query row to a JSON object keyed by column name.
15pub fn row_to_json(row: &tokio_postgres::Row) -> Value {
16    let mut map = serde_json::Map::new();
17    for (i, col) in row.columns().iter().enumerate() {
18        map.insert(col.name().to_string(), cell_to_json(row, i));
19    }
20    Value::Object(map)
21}
22
23/// Convert query rows to a JSON array of objects.
24pub fn rows_to_json_vec(rows: &[tokio_postgres::Row]) -> Vec<Value> {
25    rows.iter().map(row_to_json).collect()
26}
27
28/// Convert query rows to a JSON array value (read-tool shape).
29pub fn rows_to_json_array(rows: &[tokio_postgres::Row]) -> Value {
30    Value::Array(rows_to_json_vec(rows))
31}
32
33/// Redact configured PII columns in row objects. Returns redacted JSON and column names touched.
34pub fn redact_pii_in_rows(
35    rows: Vec<Value>,
36    pii_columns: &[String],
37    tables: &[ObjectRef],
38) -> (Vec<Value>, Vec<String>) {
39    if pii_columns.is_empty() || tables.is_empty() {
40        return (rows, Vec::new());
41    }
42    let mut redacted_cols = Vec::new();
43    let out = rows
44        .into_iter()
45        .map(|row| {
46            let mut obj = match row {
47                Value::Object(map) => map,
48                other => return other,
49            };
50            for (col, val) in obj.iter_mut() {
51                if column_matches_pii_policy(pii_columns, tables, col) {
52                    *val = Value::String(PII_REDACTED.into());
53                    if !redacted_cols.iter().any(|c| c == col) {
54                        redacted_cols.push(col.clone());
55                    }
56                }
57            }
58            Value::Object(obj)
59        })
60        .collect();
61    (out, redacted_cols)
62}
63
64/// Redact PII inside a structured read payload (`{ "rows": [...] }` or a bare array).
65pub fn redact_pii_in_payload(
66    mut payload: Value,
67    pii_columns: &[String],
68    tables: &[ObjectRef],
69) -> (Value, Vec<String>) {
70    if let Some(rows) = payload.get_mut("rows").and_then(|v| v.as_array_mut()) {
71        let taken = std::mem::take(rows);
72        let (redacted, cols) = redact_pii_in_rows(taken, pii_columns, tables);
73        *rows = redacted;
74        return (payload, cols);
75    }
76    if let Value::Array(rows) = &mut payload {
77        let taken = std::mem::take(rows);
78        let (redacted, cols) = redact_pii_in_rows(taken, pii_columns, tables);
79        *rows = redacted;
80        return (payload, cols);
81    }
82    (payload, Vec::new())
83}
84
85/// Detect SQL NULL for any column type without committing to a concrete `FromSql` type.
86enum SqlNullness {
87    Null,
88    Value,
89}
90
91impl<'a> FromSql<'a> for SqlNullness {
92    fn from_sql(_: &Type, _: &'a [u8]) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
93        Ok(SqlNullness::Value)
94    }
95
96    fn from_sql_null(_: &Type) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
97        Ok(SqlNullness::Null)
98    }
99
100    fn accepts(_: &Type) -> bool {
101        true
102    }
103}
104
105fn try_cell<T, F>(row: &tokio_postgres::Row, idx: usize, map: F) -> Option<Value>
106where
107    T: for<'a> FromSql<'a>,
108    F: FnOnce(T) -> Value,
109{
110    match row.try_get::<_, Option<T>>(idx) {
111        Ok(Some(v)) => Some(map(v)),
112        Ok(None) => Some(Value::Null),
113        Err(_) => None,
114    }
115}
116
117pub fn cell_to_json(row: &tokio_postgres::Row, idx: usize) -> Value {
118    let col_type = row.columns()[idx].type_();
119    if matches!(row.try_get::<_, SqlNullness>(idx), Ok(SqlNullness::Null)) {
120        return Value::Null;
121    }
122
123    if let Kind::Array(elem) = col_type.kind() {
124        return array_cell_to_json(row, idx, elem);
125    }
126
127    if let Some(v) = match *col_type {
128        Type::BOOL => try_cell::<bool, _>(row, idx, |b| json!(b)),
129        Type::INT2 => try_cell::<i16, _>(row, idx, |n| json!(n)),
130        Type::INT4 | Type::OID => try_cell::<i32, _>(row, idx, |n| json!(n)),
131        Type::INT8 => try_cell::<i64, _>(row, idx, |n| json!(n)),
132        Type::FLOAT4 => try_cell::<f32, _>(row, idx, |n| json!(n)),
133        Type::FLOAT8 => try_cell::<f64, _>(row, idx, |n| json!(n)),
134        Type::TEXT | Type::VARCHAR | Type::BPCHAR | Type::NAME => {
135            try_cell::<String, _>(row, idx, Value::String)
136        }
137        Type::TIMESTAMP => try_cell::<NaiveDateTime, _>(row, idx, |t| {
138            json!(t.format("%Y-%m-%dT%H:%M:%S%.f").to_string())
139        }),
140        Type::TIMESTAMPTZ => {
141            try_cell::<DateTime<FixedOffset>, _>(row, idx, |t| json!(t.to_rfc3339()))
142        }
143        Type::DATE => {
144            try_cell::<NaiveDate, _>(row, idx, |d| json!(d.format("%Y-%m-%d").to_string()))
145        }
146        Type::TIME => {
147            try_cell::<NaiveTime, _>(row, idx, |t| json!(t.format("%H:%M:%S%.f").to_string()))
148        }
149        Type::UUID => try_cell::<Uuid, _>(row, idx, |u| json!(u.to_string())),
150        Type::JSON | Type::JSONB => try_cell::<Value, _>(row, idx, |j| j),
151        Type::NUMERIC => try_cell::<Decimal, _>(row, idx, |d| json!(d.to_string())),
152        Type::MONEY => try_cell::<i64, _>(row, idx, |v| json!(money_to_string(v))),
153        Type::BYTEA => try_cell::<Vec<u8>, _>(row, idx, |b| json!(BASE64.encode(b))),
154        _ => None,
155    } {
156        return v;
157    }
158
159    cell_to_json_untyped(row, idx, col_type)
160}
161
162fn array_cell_to_json(row: &tokio_postgres::Row, idx: usize, elem: &Type) -> Value {
163    let try_array = |result: Result<Option<Vec<Value>>, tokio_postgres::Error>| -> Option<Value> {
164        match result {
165            Ok(Some(items)) => Some(Value::Array(items)),
166            Ok(None) => Some(Value::Null),
167            Err(_) => None,
168        }
169    };
170
171    match *elem {
172        Type::BOOL => {
173            if let Some(v) = try_array(
174                row.try_get::<_, Option<Vec<bool>>>(idx)
175                    .map(|v| v.map(|a| a.into_iter().map(|x| json!(x)).collect())),
176            ) {
177                return v;
178            }
179        }
180        Type::INT2 => {
181            if let Some(v) = try_array(
182                row.try_get::<_, Option<Vec<i16>>>(idx)
183                    .map(|v| v.map(|a| a.into_iter().map(|x| json!(x)).collect())),
184            ) {
185                return v;
186            }
187        }
188        Type::INT4 | Type::OID => {
189            if let Some(v) = try_array(
190                row.try_get::<_, Option<Vec<i32>>>(idx)
191                    .map(|v| v.map(|a| a.into_iter().map(|x| json!(x)).collect())),
192            ) {
193                return v;
194            }
195        }
196        Type::INT8 => {
197            if let Some(v) = try_array(
198                row.try_get::<_, Option<Vec<i64>>>(idx)
199                    .map(|v| v.map(|a| a.into_iter().map(|x| json!(x)).collect())),
200            ) {
201                return v;
202            }
203        }
204        Type::FLOAT4 => {
205            if let Some(v) = try_array(
206                row.try_get::<_, Option<Vec<f32>>>(idx)
207                    .map(|v| v.map(|a| a.into_iter().map(|n| json!(n)).collect())),
208            ) {
209                return v;
210            }
211        }
212        Type::FLOAT8 => {
213            if let Some(v) = try_array(
214                row.try_get::<_, Option<Vec<f64>>>(idx)
215                    .map(|v| v.map(|a| a.into_iter().map(|n| json!(n)).collect())),
216            ) {
217                return v;
218            }
219        }
220        Type::TEXT | Type::VARCHAR | Type::BPCHAR | Type::NAME => {
221            if let Some(v) = try_array(
222                row.try_get::<_, Option<Vec<String>>>(idx)
223                    .map(|v| v.map(|a| a.into_iter().map(Value::String).collect())),
224            ) {
225                return v;
226            }
227        }
228        Type::UUID => {
229            if let Some(v) = try_array(
230                row.try_get::<_, Option<Vec<Uuid>>>(idx)
231                    .map(|v| v.map(|a| a.into_iter().map(|u| json!(u.to_string())).collect())),
232            ) {
233                return v;
234            }
235        }
236        Type::TIMESTAMP => {
237            if let Some(v) = try_array(row.try_get::<_, Option<Vec<NaiveDateTime>>>(idx).map(|v| {
238                v.map(|a| {
239                    a.into_iter()
240                        .map(|t| json!(t.format("%Y-%m-%dT%H:%M:%S%.f").to_string()))
241                        .collect()
242                })
243            })) {
244                return v;
245            }
246        }
247        Type::TIMESTAMPTZ => {
248            if let Some(v) = try_array(
249                row.try_get::<_, Option<Vec<DateTime<FixedOffset>>>>(idx)
250                    .map(|v| v.map(|a| a.into_iter().map(|t| json!(t.to_rfc3339())).collect())),
251            ) {
252                return v;
253            }
254        }
255        Type::DATE => {
256            if let Some(v) = try_array(row.try_get::<_, Option<Vec<NaiveDate>>>(idx).map(|v| {
257                v.map(|a| {
258                    a.into_iter()
259                        .map(|d| json!(d.format("%Y-%m-%d").to_string()))
260                        .collect()
261                })
262            })) {
263                return v;
264            }
265        }
266        Type::JSON | Type::JSONB => {
267            if let Some(v) = try_array(row.try_get::<_, Option<Vec<Value>>>(idx)) {
268                return v;
269            }
270        }
271        Type::NUMERIC => {
272            if let Some(v) = try_array(
273                row.try_get::<_, Option<Vec<Decimal>>>(idx)
274                    .map(|v| v.map(|a| a.into_iter().map(|d| json!(d.to_string())).collect())),
275            ) {
276                return v;
277            }
278        }
279        Type::MONEY => {
280            if let Some(v) = try_array(
281                row.try_get::<_, Option<Vec<i64>>>(idx)
282                    .map(|v| v.map(|a| a.into_iter().map(|m| json!(money_to_string(m))).collect())),
283            ) {
284                return v;
285            }
286        }
287        Type::BYTEA => {
288            if let Some(v) = try_array(
289                row.try_get::<_, Option<Vec<Vec<u8>>>>(idx)
290                    .map(|v| v.map(|a| a.into_iter().map(|b| json!(BASE64.encode(b))).collect())),
291            ) {
292                return v;
293            }
294        }
295        _ => {}
296    }
297
298    cell_to_json_untyped(row, idx, row.columns()[idx].type_())
299}
300
301/// PostgreSQL `money` is int64 in ten-thousandths of the base currency unit.
302fn money_to_string(v: i64) -> String {
303    let sign = if v < 0 { "-" } else { "" };
304    let abs = v.unsigned_abs();
305    format!("{}{}.{:04}", sign, abs / 10_000, abs % 10_000)
306}
307
308/// Last-resort decoding for unknown or composite Postgres types — never silent null for non-null cells.
309fn cell_to_json_untyped(row: &tokio_postgres::Row, idx: usize, pg_type: &Type) -> Value {
310    if let Ok(Some(s)) = row.try_get::<_, Option<String>>(idx) {
311        return Value::String(s);
312    }
313    json!({
314        "__untyped": true,
315        "type": pg_type.name()
316    })
317}
318
319#[cfg(test)]
320mod tests {
321    use super::*;
322    use serde_json::json;
323
324    #[test]
325    fn redact_pii_replaces_matching_columns() {
326        let rows = vec![json!({"id": 1, "ssn": "123-45-6789"})];
327        let tables = vec![ObjectRef::new("public", "users")];
328        let pii = vec!["public.users.ssn".into()];
329        let (out, cols) = redact_pii_in_rows(rows, &pii, &tables);
330        assert_eq!(cols, vec!["ssn"]);
331        assert_eq!(out[0]["ssn"], json!(PII_REDACTED));
332        assert_eq!(out[0]["id"], json!(1));
333    }
334}
nexql_tools/cell_json.rs

nexql_tools/
cell_json.rs
