krabka_connect_postgres/
pgoutput.rs

1use std::collections::{BTreeMap, HashMap};
2
3use crate::{
4    PgLsn, PostgresConnectError,
5    ids::{CommitLsn, EndLsn, RelationId, TransactionId},
6    model::{
7        ColumnSchema, ColumnValue, EntityDifference, EntityKey, Operation, ScalarValue, TableSchema,
8    },
9};
10
11#[derive(Debug, Clone, PartialEq, Eq)]
12pub struct RelationEvent {
13    pub relation_id: RelationId,
14    pub schema: String,
15    pub table: String,
16    pub columns: Vec<ColumnSchema>,
17}
18
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
20pub enum RowTupleKind {
21    Full,
22    Key,
23}
24
25#[derive(Debug, Clone, PartialEq, Eq, Hash)]
26pub enum RowEventKind {
27    Insert,
28    Update {
29        old: Vec<ColumnValue>,
30        old_tuple_kind: RowTupleKind,
31    },
32    Delete {
33        tuple_kind: RowTupleKind,
34    },
35}
36
37#[derive(Debug, Clone, PartialEq, Eq, Hash)]
38pub struct RowEvent {
39    pub relation_id: RelationId,
40    pub lsn: PgLsn,
41    pub commit_lsn: Option<PgLsn>,
42    pub txid: Option<TransactionId>,
43    pub commit_timestamp_ms: Option<i64>,
44    pub kind: RowEventKind,
45    pub values: Vec<ColumnValue>,
46}
47
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub enum DecodedMessage {
50    Begin {
51        final_lsn: PgLsn,
52        xid: TransactionId,
53    },
54    Commit {
55        commit_lsn: CommitLsn,
56        end_lsn: EndLsn,
57        commit_timestamp_ms: i64,
58    },
59    Relation(RelationEvent),
60    Row(RowEvent),
61    Keepalive,
62}
63
64/// # Errors
65/// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
66pub fn decode_pgoutput_message(
67    bytes: &[u8],
68    lsn: PgLsn,
69    txid: Option<TransactionId>,
70) -> Result<DecodedMessage, PostgresConnectError> {
71    let mut reader = PgOutputReader::new(bytes);
72    let tag = reader.read_u8("message tag")?;
73
74    let message = match tag {
75        b'B' => {
76            let final_lsn = PgLsn(reader.read_u64("begin final_lsn")?);
77            let _commit_time = reader.read_i64("begin commit_time")?;
78            let xid = TransactionId(i64::from(reader.read_u32("begin xid")?));
79            DecodedMessage::Begin { final_lsn, xid }
80        }
81        b'C' => {
82            let _flags = reader.read_u8("commit flags")?;
83            let commit_lsn = CommitLsn(PgLsn(reader.read_u64("commit commit_lsn")?));
84            let end_lsn = EndLsn(PgLsn(reader.read_u64("commit end_lsn")?));
85            let commit_timestamp_ms =
86                postgres_timestamp_micros_to_unix_ms(reader.read_i64("commit commit_time")?);
87            DecodedMessage::Commit {
88                commit_lsn,
89                end_lsn,
90                commit_timestamp_ms,
91            }
92        }
93        b'R' => DecodedMessage::Relation(decode_relation(&mut reader)?),
94        b'I' => DecodedMessage::Row(decode_insert(&mut reader, lsn, txid)?),
95        b'U' => DecodedMessage::Row(decode_update(&mut reader, lsn, txid)?),
96        b'D' => DecodedMessage::Row(decode_delete(&mut reader, lsn, txid)?),
97        tag => {
98            return Err(PostgresConnectError::Backend(format!(
99                "unsupported pgoutput message tag {:?}",
100                char::from(tag)
101            )));
102        }
103    };
104
105    reader.finish()?;
106    Ok(message)
107}
108
109fn decode_relation(reader: &mut PgOutputReader<'_>) -> Result<RelationEvent, PostgresConnectError> {
110    let relation_id = RelationId(reader.read_u32("relation id")?);
111    let schema = reader.read_cstr("relation namespace")?;
112    let table = reader.read_cstr("relation name")?;
113    let _replica_identity = reader.read_u8("relation replica identity")?;
114    let column_count = reader.read_count("relation column count")?;
115    let mut columns = Vec::with_capacity(column_count);
116
117    for index in 0..column_count {
118        let flags = reader.read_u8("relation column flags")?;
119        let name = reader.read_cstr("relation column name")?;
120        let type_oid = reader.read_i32("relation column type oid")?;
121        let _atttypmod = reader.read_i32("relation column atttypmod")?;
122        columns.push(ColumnSchema {
123            name,
124            type_name: type_name(type_oid).to_owned(),
125            key: flags & 1 == 1,
126        });
127
128        if columns[index].name.is_empty() {
129            return Err(PostgresConnectError::Backend(format!(
130                "invalid pgoutput relation column {index}: empty name"
131            )));
132        }
133    }
134
135    Ok(RelationEvent {
136        relation_id,
137        schema,
138        table,
139        columns,
140    })
141}
142
143fn decode_insert(
144    reader: &mut PgOutputReader<'_>,
145    lsn: PgLsn,
146    txid: Option<TransactionId>,
147) -> Result<RowEvent, PostgresConnectError> {
148    let relation_id = RelationId(reader.read_u32("insert relation id")?);
149    reader.expect_u8(b'N', "insert tuple tag N")?;
150    let values = decode_tuple(reader)?;
151
152    Ok(RowEvent {
153        relation_id,
154        lsn,
155        commit_lsn: None,
156        txid,
157        commit_timestamp_ms: None,
158        kind: RowEventKind::Insert,
159        values,
160    })
161}
162
163fn decode_update(
164    reader: &mut PgOutputReader<'_>,
165    lsn: PgLsn,
166    txid: Option<TransactionId>,
167) -> Result<RowEvent, PostgresConnectError> {
168    let relation_id = RelationId(reader.read_u32("update relation id")?);
169    let tag = reader.read_u8("update tuple tag")?;
170    let (old, old_tuple_kind, new_tag) = match tag {
171        b'K' | b'O' => (
172            decode_tuple(reader)?,
173            tuple_kind(tag),
174            reader.read_u8("update new tuple tag")?,
175        ),
176        b'N' => (Vec::new(), RowTupleKind::Full, b'N'),
177        tag => {
178            return Err(PostgresConnectError::Backend(format!(
179                "invalid pgoutput update tuple tag {:?}",
180                char::from(tag)
181            )));
182        }
183    };
184
185    if new_tag != b'N' {
186        return Err(PostgresConnectError::Backend(format!(
187            "invalid pgoutput update new tuple tag {:?}",
188            char::from(new_tag)
189        )));
190    }
191
192    let values = decode_tuple(reader)?;
193    Ok(RowEvent {
194        relation_id,
195        lsn,
196        commit_lsn: None,
197        txid,
198        commit_timestamp_ms: None,
199        kind: RowEventKind::Update {
200            old,
201            old_tuple_kind,
202        },
203        values,
204    })
205}
206
207fn decode_delete(
208    reader: &mut PgOutputReader<'_>,
209    lsn: PgLsn,
210    txid: Option<TransactionId>,
211) -> Result<RowEvent, PostgresConnectError> {
212    let relation_id = RelationId(reader.read_u32("delete relation id")?);
213    let tag = reader.read_u8("delete tuple tag")?;
214    if !matches!(tag, b'K' | b'O') {
215        return Err(PostgresConnectError::Backend(format!(
216            "invalid pgoutput delete tuple tag {:?}",
217            char::from(tag)
218        )));
219    }
220    let values = decode_tuple(reader)?;
221
222    Ok(RowEvent {
223        relation_id,
224        lsn,
225        commit_lsn: None,
226        txid,
227        commit_timestamp_ms: None,
228        kind: RowEventKind::Delete {
229            tuple_kind: tuple_kind(tag),
230        },
231        values,
232    })
233}
234
235fn tuple_kind(tag: u8) -> RowTupleKind {
236    if tag == b'K' {
237        RowTupleKind::Key
238    } else {
239        RowTupleKind::Full
240    }
241}
242
243fn decode_tuple(reader: &mut PgOutputReader<'_>) -> Result<Vec<ColumnValue>, PostgresConnectError> {
244    let value_count = reader.read_count("tuple value count")?;
245    let mut values = Vec::with_capacity(value_count);
246
247    for index in 0..value_count {
248        let value_tag = reader.read_u8("tuple value tag")?;
249        let value = match value_tag {
250            b'n' => ScalarValue::Null,
251            b'u' => ScalarValue::UnchangedToast,
252            b't' => {
253                let bytes = reader.read_len_bytes("text tuple value")?;
254                let value = std::str::from_utf8(bytes).map_err(|error| {
255                    PostgresConnectError::Backend(format!(
256                        "invalid utf8 in pgoutput text tuple value: {error}"
257                    ))
258                })?;
259                ScalarValue::Text(value.to_owned())
260            }
261            b'b' => ScalarValue::Bytes(reader.read_len_bytes("binary tuple value")?.to_vec()),
262            tag => {
263                return Err(PostgresConnectError::Backend(format!(
264                    "unsupported pgoutput tuple value tag {:?}",
265                    char::from(tag)
266                )));
267            }
268        };
269
270        values.push(ColumnValue {
271            name: format!("col{index}"),
272            value,
273        });
274    }
275
276    Ok(values)
277}
278
279fn type_name(type_oid: i32) -> &'static str {
280    match type_oid {
281        16 => "bool",
282        17 => "bytea",
283        20 => "int8",
284        21 => "int2",
285        23 => "int4",
286        25 => "text",
287        700 => "float4",
288        701 => "float8",
289        1043 => "varchar",
290        1700 => "numeric",
291        _ => "unknown",
292    }
293}
294
295fn postgres_timestamp_micros_to_unix_ms(value: i64) -> i64 {
296    const POSTGRES_EPOCH_UNIX_MS: i64 = 946_684_800_000;
297    POSTGRES_EPOCH_UNIX_MS + value.div_euclid(1_000)
298}
299
300struct PgOutputReader<'a> {
301    bytes: &'a [u8],
302    position: usize,
303}
304
305impl<'a> PgOutputReader<'a> {
306    fn new(bytes: &'a [u8]) -> Self {
307        Self { bytes, position: 0 }
308    }
309
310    fn finish(&self) -> Result<(), PostgresConnectError> {
311        if self.position == self.bytes.len() {
312            Ok(())
313        } else {
314            Err(PostgresConnectError::Backend(format!(
315                "invalid pgoutput message: {} trailing bytes",
316                self.bytes.len() - self.position
317            )))
318        }
319    }
320
321    fn read_u8(&mut self, field: &str) -> Result<u8, PostgresConnectError> {
322        let bytes = self.read_exact(1, field)?;
323        Ok(bytes[0])
324    }
325
326    fn expect_u8(&mut self, expected: u8, field: &str) -> Result<(), PostgresConnectError> {
327        let actual = self.read_u8(field)?;
328        if actual == expected {
329            Ok(())
330        } else {
331            Err(PostgresConnectError::Backend(format!(
332                "invalid pgoutput {field}: expected {:?}, got {:?}",
333                char::from(expected),
334                char::from(actual)
335            )))
336        }
337    }
338
339    fn read_i16(&mut self, field: &str) -> Result<i16, PostgresConnectError> {
340        let bytes = self.read_exact(2, field)?;
341        Ok(i16::from_be_bytes([bytes[0], bytes[1]]))
342    }
343
344    fn read_count(&mut self, field: &str) -> Result<usize, PostgresConnectError> {
345        let count = self.read_i16(field)?;
346        usize::try_from(count).map_err(|_| {
347            PostgresConnectError::Backend(format!("invalid pgoutput {field}: negative count"))
348        })
349    }
350
351    fn read_i32(&mut self, field: &str) -> Result<i32, PostgresConnectError> {
352        let bytes = self.read_exact(4, field)?;
353        Ok(i32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
354    }
355
356    fn read_u32(&mut self, field: &str) -> Result<u32, PostgresConnectError> {
357        let bytes = self.read_exact(4, field)?;
358        Ok(u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
359    }
360
361    fn read_i64(&mut self, field: &str) -> Result<i64, PostgresConnectError> {
362        let bytes = self.read_exact(8, field)?;
363        Ok(i64::from_be_bytes([
364            bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
365        ]))
366    }
367
368    fn read_u64(&mut self, field: &str) -> Result<u64, PostgresConnectError> {
369        let bytes = self.read_exact(8, field)?;
370        Ok(u64::from_be_bytes([
371            bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
372        ]))
373    }
374
375    fn read_len_bytes(&mut self, field: &str) -> Result<&'a [u8], PostgresConnectError> {
376        let length = self.read_i32(field)?;
377        let length = usize::try_from(length).map_err(|_| {
378            PostgresConnectError::Backend(format!("invalid pgoutput {field}: negative length"))
379        })?;
380        self.read_exact(length, field)
381    }
382
383    fn read_cstr(&mut self, field: &str) -> Result<String, PostgresConnectError> {
384        let remaining = &self.bytes[self.position..];
385        let Some(length) = remaining.iter().position(|byte| *byte == 0) else {
386            return Err(PostgresConnectError::Backend(format!(
387                "truncated pgoutput {field}: missing null terminator"
388            )));
389        };
390        let bytes = &remaining[..length];
391        self.position += length + 1;
392        let value = std::str::from_utf8(bytes).map_err(|error| {
393            PostgresConnectError::Backend(format!("invalid utf8 in pgoutput {field}: {error}"))
394        })?;
395        Ok(value.to_owned())
396    }
397
398    fn read_exact(&mut self, length: usize, field: &str) -> Result<&'a [u8], PostgresConnectError> {
399        let end = self.position.checked_add(length).ok_or_else(|| {
400            PostgresConnectError::Backend(format!("invalid pgoutput {field}: length overflow"))
401        })?;
402        if end > self.bytes.len() {
403            return Err(PostgresConnectError::Backend(format!(
404                "truncated pgoutput {field}: needed {length} bytes, have {}",
405                self.bytes.len().saturating_sub(self.position)
406            )));
407        }
408
409        let bytes = &self.bytes[self.position..end];
410        self.position = end;
411        Ok(bytes)
412    }
413}
414
415#[derive(Debug, Clone, Default)]
416pub struct RelationCache {
417    relations: BTreeMap<RelationId, CachedRelation>,
418}
419
420#[derive(Debug, Clone)]
421struct CachedRelation {
422    schema: TableSchema,
423    /// Preserve the original named lookup's first-match duplicate semantics.
424    column_indices: HashMap<String, usize>,
425}
426
427impl RelationCache {
428    pub fn apply_relation(&mut self, event: RelationEvent) {
429        let mut column_indices = HashMap::with_capacity(event.columns.len());
430        for (index, column) in event.columns.iter().enumerate() {
431            column_indices.entry(column.name.clone()).or_insert(index);
432        }
433        self.relations.insert(
434            event.relation_id,
435            CachedRelation {
436                schema: TableSchema {
437                    schema: event.schema,
438                    table: event.table,
439                    columns: event.columns,
440                },
441                column_indices,
442            },
443        );
444    }
445
446    /// # Errors
447    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
448    pub fn translate(&self, event: RowEvent) -> Result<EntityDifference, PostgresConnectError> {
449        let relation = self.relations.get(&event.relation_id).ok_or_else(|| {
450            PostgresConnectError::Backend(format!(
451                "missing relation metadata for relation id {}",
452                event.relation_id
453            ))
454        })?;
455        let schema = &relation.schema;
456        let table = format!("{}.{}", schema.schema, schema.table);
457        let lsn = event.commit_lsn.unwrap_or(event.lsn);
458
459        let (op, before, after, key_columns) = match event.kind {
460            RowEventKind::Insert => {
461                let values = normalize_values(relation, event.values, "row values")?;
462                let key_columns = extract_key_columns(schema, &values)?;
463                (Operation::Insert, Vec::new(), values, key_columns)
464            }
465            RowEventKind::Update { old, .. } => {
466                let values = normalize_values(relation, event.values, "row values")?;
467                let key_columns = extract_key_columns(schema, &values)?;
468                let old = if old.is_empty() {
469                    old
470                } else {
471                    normalize_values(relation, old, "old row values")?
472                };
473                if has_any_key_column(schema, &old) {
474                    let old_key_columns = extract_key_columns(schema, &old)?;
475                    if old_key_columns != key_columns {
476                        return Err(PostgresConnectError::Backend(
477                            "key-changing updates are not supported".to_owned(),
478                        ));
479                    }
480                }
481
482                (Operation::Update, old, values, key_columns)
483            }
484            RowEventKind::Delete { .. } => {
485                let values = normalize_values(relation, event.values, "row values")?;
486                let key_columns = extract_key_columns(schema, &values)?;
487                (Operation::Delete, values, Vec::new(), key_columns)
488            }
489        };
490        let key = EntityKey {
491            table: table.clone(),
492            columns: key_columns,
493        };
494
495        Ok(EntityDifference {
496            table,
497            key,
498            op,
499            before,
500            after,
501            lsn,
502            txid: event.txid,
503            commit_timestamp_ms: event.commit_timestamp_ms,
504            schema: schema.clone(),
505        })
506    }
507}
508
509fn normalize_values(
510    relation: &CachedRelation,
511    values: Vec<ColumnValue>,
512    label: &str,
513) -> Result<Vec<ColumnValue>, PostgresConnectError> {
514    // pgoutput TupleData always follows Relation column arity and order. Key
515    // tuples retain non-key positions as null markers; key extraction filters
516    // those columns after names have been restored here.
517    let schema = &relation.schema;
518
519    if values.len() != schema.columns.len() {
520        return Err(PostgresConnectError::Backend(format!(
521            "column count mismatch for {}.{} {label}: decoded {} values but expected {} columns",
522            schema.schema,
523            schema.table,
524            values.len(),
525            schema.columns.len()
526        )));
527    }
528
529    // Decide the whole-row fallback before coercing anything. A single missing
530    // (or differently cased) name makes every value positional, as before.
531    if values
532        .iter()
533        .all(|value| relation.column_indices.contains_key(&value.name))
534    {
535        return values
536            .into_iter()
537            .map(|value| {
538                let column = &schema.columns[relation.column_indices[&value.name]];
539                Ok(ColumnValue {
540                    name: value.name,
541                    value: coerce_value(value.value, &column.type_name)?,
542                })
543            })
544            .collect();
545    }
546
547    values
548        .into_iter()
549        .zip(&schema.columns)
550        .map(|(value, column)| {
551            Ok(ColumnValue {
552                name: column.name.clone(),
553                value: coerce_value(value.value, &column.type_name)?,
554            })
555        })
556        .collect::<Result<Vec<_>, _>>()
557}
558
559fn coerce_value(value: ScalarValue, type_name: &str) -> Result<ScalarValue, PostgresConnectError> {
560    let ScalarValue::Text(text) = value else {
561        return Ok(value);
562    };
563
564    match type_name {
565        "bool" => coerce_bool(&text),
566        "int2" | "int4" | "int8" => text.parse::<i64>().map(ScalarValue::Int).map_err(|error| {
567            PostgresConnectError::Backend(format!(
568                "invalid {type_name} pgoutput text value {text:?}: {error}"
569            ))
570        }),
571        "float4" | "float8" | "numeric" => Ok(ScalarValue::Float(text)),
572        "bytea" => Ok(ScalarValue::Bytes(coerce_bytea(&text)?)),
573        _ => Ok(ScalarValue::Text(text)),
574    }
575}
576
577fn coerce_bool(text: &str) -> Result<ScalarValue, PostgresConnectError> {
578    match text {
579        "t" | "true" | "1" => Ok(ScalarValue::Bool(true)),
580        "f" | "false" | "0" => Ok(ScalarValue::Bool(false)),
581        _ => Err(PostgresConnectError::Backend(format!(
582            "invalid bool pgoutput text value {text:?}"
583        ))),
584    }
585}
586
587fn coerce_bytea(text: &str) -> Result<Vec<u8>, PostgresConnectError> {
588    let Some(hex) = text.strip_prefix("\\x") else {
589        return Ok(text.as_bytes().to_vec());
590    };
591    if hex.len() % 2 != 0 {
592        return Err(PostgresConnectError::Backend(format!(
593            "invalid bytea pgoutput text value {text:?}: odd hex length"
594        )));
595    }
596
597    hex.as_bytes()
598        .as_chunks::<2>()
599        .0
600        .iter()
601        .map(|chunk| {
602            let pair = std::str::from_utf8(chunk).map_err(|error| {
603                PostgresConnectError::Backend(format!(
604                    "invalid bytea pgoutput text value {text:?}: {error}"
605                ))
606            })?;
607            u8::from_str_radix(pair, 16).map_err(|error| {
608                PostgresConnectError::Backend(format!(
609                    "invalid bytea pgoutput text value {text:?}: {error}"
610                ))
611            })
612        })
613        .collect()
614}
615
616fn extract_key_columns(
617    schema: &TableSchema,
618    values: &[ColumnValue],
619) -> Result<Vec<ColumnValue>, PostgresConnectError> {
620    schema
621        .columns
622        .iter()
623        .filter(|column| column.key)
624        .map(|column| {
625            values
626                .iter()
627                .find(|value| value.name == column.name)
628                .cloned()
629                .ok_or_else(|| {
630                    PostgresConnectError::Backend(format!(
631                        "missing key column {:?} for {}.{}",
632                        column.name, schema.schema, schema.table
633                    ))
634                })
635        })
636        .collect()
637}
638
639fn has_any_key_column(schema: &TableSchema, values: &[ColumnValue]) -> bool {
640    schema
641        .columns
642        .iter()
643        .filter(|column| column.key)
644        .any(|column| values.iter().any(|value| value.name == column.name))
645}
646
647#[cfg(test)]
648mod tests {
649    use assert2::check;
650
651    use super::{
652        RelationCache, RelationEvent, RowEvent, RowEventKind, RowTupleKind, coerce_bytea,
653        coerce_value,
654    };
655    use crate::{
656        PgLsn, PostgresConnectError,
657        ids::{RelationId, TransactionId},
658        model::{
659            ColumnSchema, ColumnValue, EntityDifference, EntityKey, Operation, ScalarValue,
660            TableSchema,
661        },
662    };
663
664    fn orders_relation(type_name: &str) -> RelationEvent {
665        RelationEvent {
666            relation_id: RelationId(7),
667            schema: "public".to_owned(),
668            table: "orders".to_owned(),
669            columns: vec![
670                ColumnSchema {
671                    name: "id".to_owned(),
672                    type_name: "int8".to_owned(),
673                    key: true,
674                },
675                ColumnSchema {
676                    name: "status".to_owned(),
677                    type_name: type_name.to_owned(),
678                    key: false,
679                },
680            ],
681        }
682    }
683
684    fn orders_schema(type_name: &str) -> TableSchema {
685        let relation = orders_relation(type_name);
686        TableSchema {
687            schema: relation.schema,
688            table: relation.table,
689            columns: relation.columns,
690        }
691    }
692
693    fn id(value: i64) -> ColumnValue {
694        ColumnValue {
695            name: "id".to_owned(),
696            value: ScalarValue::Int(value),
697        }
698    }
699
700    fn status(value: &str) -> ColumnValue {
701        ColumnValue {
702            name: "status".to_owned(),
703            value: ScalarValue::Text(value.to_owned()),
704        }
705    }
706
707    fn tenant_id(value: i64) -> ColumnValue {
708        ColumnValue {
709            name: "tenant_id".to_owned(),
710            value: ScalarValue::Int(value),
711        }
712    }
713
714    #[test]
715    fn named_normalization_preserves_order_duplicates_and_whole_row_fallback() {
716        let mut cache = RelationCache::default();
717        cache.apply_relation(orders_relation("text"));
718        let relation = &cache.relations[&RelationId(7)];
719        let text = |name: &str, value: &str| ColumnValue {
720            name: name.into(),
721            value: ScalarValue::Text(value.into()),
722        };
723        let cases = [
724            (
725                vec![text("status", "paid"), text("id", "42")],
726                vec![status("paid"), id(42)],
727            ),
728            (vec![text("id", "7"), text("id", "8")], vec![id(7), id(8)]),
729            // One missing name makes even the recognized status value
730            // positional; a per-value fallback would attempt the wrong type.
731            (
732                vec![text("status", "42"), text("missing", "paid")],
733                vec![id(42), status("paid")],
734            ),
735            (
736                vec![text("status", "42"), text("ID", "paid")],
737                vec![id(42), status("paid")],
738            ),
739            (
740                vec![
741                    ColumnValue {
742                        name: "status".into(),
743                        value: ScalarValue::UnchangedToast,
744                    },
745                    ColumnValue {
746                        name: "id".into(),
747                        value: ScalarValue::Null,
748                    },
749                ],
750                vec![
751                    ColumnValue {
752                        name: "status".into(),
753                        value: ScalarValue::UnchangedToast,
754                    },
755                    ColumnValue {
756                        name: "id".into(),
757                        value: ScalarValue::Null,
758                    },
759                ],
760            ),
761        ];
762        for (values, expected) in cases {
763            assert2::assert!(
764                super::normalize_values(relation, values, "row values").unwrap() == expected
765            );
766        }
767        let error = super::normalize_values(relation, vec![id(42)], "old row values").unwrap_err();
768        assert2::assert!(
769            error.to_string()
770                == "postgres backend error: column count mismatch for public.orders old row values: decoded 1 values but expected 2 columns"
771        );
772        let error = super::normalize_values(
773            relation,
774            vec![text("id", "invalid"), text("status", "paid")],
775            "row values",
776        )
777        .unwrap_err();
778        assert2::assert!(
779            error
780                .to_string()
781                .contains("invalid int8 pgoutput text value")
782        );
783    }
784
785    #[test]
786    fn relation_refresh_rebuilds_first_match_column_lookup() {
787        let mut cache = RelationCache::default();
788        let mut relation = orders_relation("text");
789        relation.columns[1].name = "id".into();
790        cache.apply_relation(relation.clone());
791        let values = vec![
792            ColumnValue {
793                name: "id".into(),
794                value: ScalarValue::Text("7".into()),
795            },
796            ColumnValue {
797                name: "id".into(),
798                value: ScalarValue::Text("8".into()),
799            },
800        ];
801        assert2::assert!(
802            super::normalize_values(
803                &cache.relations[&RelationId(7)],
804                values.clone(),
805                "row values"
806            )
807            .unwrap()
808                == vec![id(7), id(8)]
809        );
810        relation.columns.swap(0, 1);
811        cache.apply_relation(relation);
812        assert2::assert!(
813            super::normalize_values(
814                &cache.relations[&RelationId(7)],
815                values.clone(),
816                "row values"
817            )
818            .unwrap()
819                == values
820        );
821        cache.apply_relation(orders_relation("bool"));
822        assert2::assert!(
823            super::normalize_values(
824                &cache.relations[&RelationId(7)],
825                vec![
826                    id(7),
827                    ColumnValue {
828                        name: "status".into(),
829                        value: ScalarValue::Text("t".into())
830                    }
831                ],
832                "row values",
833            )
834            .unwrap()
835                == vec![
836                    id(7),
837                    ColumnValue {
838                        name: "status".into(),
839                        value: ScalarValue::Bool(true)
840                    }
841                ]
842        );
843    }
844
845    fn composite_key_orders_relation() -> RelationEvent {
846        RelationEvent {
847            relation_id: RelationId(7),
848            schema: "public".to_owned(),
849            table: "orders".to_owned(),
850            columns: vec![
851                ColumnSchema {
852                    name: "tenant_id".to_owned(),
853                    type_name: "int8".to_owned(),
854                    key: true,
855                },
856                ColumnSchema {
857                    name: "id".to_owned(),
858                    type_name: "int8".to_owned(),
859                    key: true,
860                },
861                ColumnSchema {
862                    name: "status".to_owned(),
863                    type_name: "text".to_owned(),
864                    key: false,
865                },
866            ],
867        }
868    }
869
870    #[test]
871    fn insert_translates_to_entity_difference_with_key() {
872        let mut cache = RelationCache::default();
873        cache.apply_relation(orders_relation("text"));
874
875        let values = vec![id(42), status("paid")];
876        let difference = cache
877            .translate(RowEvent {
878                relation_id: RelationId(7),
879                lsn: PgLsn(0x16_b374_d848),
880                commit_lsn: None,
881                txid: Some(TransactionId(99)),
882                commit_timestamp_ms: Some(1_700_000_000_000),
883                kind: RowEventKind::Insert,
884                values: values.clone(),
885            })
886            .expect("relation should translate");
887
888        check!(
889            difference
890                == EntityDifference {
891                    table: "public.orders".to_owned(),
892                    key: EntityKey {
893                        table: "public.orders".to_owned(),
894                        columns: vec![id(42)],
895                    },
896                    op: Operation::Insert,
897                    before: Vec::new(),
898                    after: values,
899                    lsn: PgLsn(0x16_b374_d848),
900                    txid: Some(TransactionId(99)),
901                    commit_timestamp_ms: Some(1_700_000_000_000),
902                    schema: orders_schema("text"),
903                }
904        );
905    }
906
907    #[test]
908    fn delete_translates_to_before_only_difference() {
909        let mut cache = RelationCache::default();
910        cache.apply_relation(orders_relation("text"));
911
912        let values = vec![id(42), status("cancelled")];
913        let difference = cache
914            .translate(RowEvent {
915                relation_id: RelationId(7),
916                lsn: PgLsn(0x2a),
917                commit_lsn: None,
918                txid: None,
919                commit_timestamp_ms: None,
920                kind: RowEventKind::Delete {
921                    tuple_kind: RowTupleKind::Full,
922                },
923                values: values.clone(),
924            })
925            .expect("relation should translate");
926
927        check!(
928            difference
929                == EntityDifference {
930                    table: "public.orders".to_owned(),
931                    key: EntityKey {
932                        table: "public.orders".to_owned(),
933                        columns: vec![id(42)],
934                    },
935                    op: Operation::Delete,
936                    before: values,
937                    after: Vec::new(),
938                    lsn: PgLsn(0x2a),
939                    txid: None,
940                    commit_timestamp_ms: None,
941                    schema: orders_schema("text"),
942                }
943        );
944    }
945
946    #[test]
947    fn relation_refresh_changes_table_schema() {
948        let mut cache = RelationCache::default();
949        cache.apply_relation(orders_relation("text"));
950        cache.apply_relation(orders_relation("varchar"));
951
952        let difference = cache
953            .translate(RowEvent {
954                relation_id: RelationId(7),
955                lsn: PgLsn(0x2b),
956                commit_lsn: None,
957                txid: None,
958                commit_timestamp_ms: None,
959                kind: RowEventKind::Insert,
960                values: vec![id(7), status("new")],
961            })
962            .expect("relation should translate");
963
964        check!(difference.schema.columns[1].type_name == "varchar");
965    }
966
967    #[test]
968    fn composite_key_output_follows_schema_order_even_if_row_values_are_out_of_order() {
969        let mut cache = RelationCache::default();
970        cache.apply_relation(composite_key_orders_relation());
971
972        let difference = cache
973            .translate(RowEvent {
974                relation_id: RelationId(7),
975                lsn: PgLsn(0x2c),
976                commit_lsn: None,
977                txid: None,
978                commit_timestamp_ms: None,
979                kind: RowEventKind::Insert,
980                values: vec![status("new"), id(7), tenant_id(3)],
981            })
982            .expect("relation should translate");
983
984        check!(difference.key.columns == vec![tenant_id(3), id(7)]);
985    }
986
987    #[test]
988    fn missing_key_column_is_an_error() {
989        let mut cache = RelationCache::default();
990        cache.apply_relation(orders_relation("text"));
991
992        let result = cache.translate(RowEvent {
993            relation_id: RelationId(7),
994            lsn: PgLsn(0x2d),
995            commit_lsn: None,
996            txid: None,
997            commit_timestamp_ms: None,
998            kind: RowEventKind::Insert,
999            values: vec![status("new"), status("paid")],
1000        });
1001
1002        let error = result.expect_err("missing key should fail");
1003        match error {
1004            PostgresConnectError::Backend(message) => {
1005                check!(message.contains("missing key column"));
1006                check!(message.contains("id"));
1007            }
1008            error => panic!("expected backend error, got {error:?}"),
1009        }
1010    }
1011
1012    #[test]
1013    fn non_key_update_translates_with_before_after_and_stable_key() {
1014        let mut cache = RelationCache::default();
1015        cache.apply_relation(orders_relation("text"));
1016
1017        let old = vec![id(42), status("new")];
1018        let new = vec![id(42), status("paid")];
1019        let difference = cache
1020            .translate(RowEvent {
1021                relation_id: RelationId(7),
1022                lsn: PgLsn(0x2e),
1023                commit_lsn: None,
1024                txid: Some(TransactionId(100)),
1025                commit_timestamp_ms: Some(1_700_000_000_001),
1026                kind: RowEventKind::Update {
1027                    old: old.clone(),
1028                    old_tuple_kind: RowTupleKind::Full,
1029                },
1030                values: new.clone(),
1031            })
1032            .expect("relation should translate");
1033
1034        check!(
1035            difference
1036                == EntityDifference {
1037                    table: "public.orders".to_owned(),
1038                    key: EntityKey {
1039                        table: "public.orders".to_owned(),
1040                        columns: vec![id(42)],
1041                    },
1042                    op: Operation::Update,
1043                    before: old,
1044                    after: new,
1045                    lsn: PgLsn(0x2e),
1046                    txid: Some(TransactionId(100)),
1047                    commit_timestamp_ms: Some(1_700_000_000_001),
1048                    schema: orders_schema("text"),
1049                }
1050        );
1051    }
1052
1053    #[test]
1054    fn key_changing_update_is_an_error() {
1055        let mut cache = RelationCache::default();
1056        cache.apply_relation(orders_relation("text"));
1057
1058        let result = cache.translate(RowEvent {
1059            relation_id: RelationId(7),
1060            lsn: PgLsn(0x2f),
1061            commit_lsn: None,
1062            txid: None,
1063            commit_timestamp_ms: None,
1064            kind: RowEventKind::Update {
1065                old: vec![id(41), status("new")],
1066                old_tuple_kind: RowTupleKind::Full,
1067            },
1068            values: vec![id(42), status("paid")],
1069        });
1070
1071        let error = result.expect_err("key-changing update should fail");
1072        match error {
1073            PostgresConnectError::Backend(message) => {
1074                check!(message == "key-changing updates are not supported");
1075            }
1076            error => panic!("expected backend error, got {error:?}"),
1077        }
1078    }
1079
1080    #[test]
1081    fn decoded_placeholder_values_bind_to_schema_names_before_key_extraction() {
1082        let mut cache = RelationCache::default();
1083        cache.apply_relation(orders_relation("text"));
1084
1085        let difference = cache
1086            .translate(RowEvent {
1087                relation_id: RelationId(7),
1088                lsn: PgLsn(0x30),
1089                commit_lsn: None,
1090                txid: None,
1091                commit_timestamp_ms: None,
1092                kind: RowEventKind::Insert,
1093                values: vec![
1094                    ColumnValue {
1095                        name: "col0".to_owned(),
1096                        value: ScalarValue::Text("42".to_owned()),
1097                    },
1098                    ColumnValue {
1099                        name: "col1".to_owned(),
1100                        value: ScalarValue::Text("paid".to_owned()),
1101                    },
1102                ],
1103            })
1104            .expect("decoded placeholders should bind to relation schema");
1105
1106        check!(
1107            (difference.key.columns, difference.after)
1108                == (vec![id(42)], vec![id(42), status("paid")],)
1109        );
1110    }
1111
1112    #[test]
1113    fn decoded_placeholder_count_mismatch_is_an_error() {
1114        let mut cache = RelationCache::default();
1115        cache.apply_relation(orders_relation("text"));
1116
1117        let error = cache
1118            .translate(RowEvent {
1119                relation_id: RelationId(7),
1120                lsn: PgLsn(0x31),
1121                commit_lsn: None,
1122                txid: None,
1123                commit_timestamp_ms: None,
1124                kind: RowEventKind::Insert,
1125                values: vec![ColumnValue {
1126                    name: "col0".to_owned(),
1127                    value: ScalarValue::Text("42".to_owned()),
1128                }],
1129            })
1130            .expect_err("decoded placeholder count should match relation schema");
1131
1132        match error {
1133            PostgresConnectError::Backend(message) => {
1134                check!(message.contains("column count mismatch"));
1135                check!(message.contains("public.orders"));
1136            }
1137            error => panic!("expected backend error, got {error:?}"),
1138        }
1139    }
1140
1141    #[test]
1142    fn named_partial_full_row_values_are_a_count_mismatch() {
1143        let mut cache = RelationCache::default();
1144        cache.apply_relation(orders_relation("text"));
1145
1146        let error = cache
1147            .translate(RowEvent {
1148                relation_id: RelationId(7),
1149                lsn: PgLsn(0x34),
1150                commit_lsn: None,
1151                txid: None,
1152                commit_timestamp_ms: None,
1153                kind: RowEventKind::Insert,
1154                values: vec![id(42)],
1155            })
1156            .expect_err("partial full row should fail count validation");
1157
1158        match error {
1159            PostgresConnectError::Backend(message) => {
1160                check!(message.contains("column count mismatch"));
1161                check!(message.contains("public.orders"));
1162            }
1163            error => panic!("expected backend error, got {error:?}"),
1164        }
1165    }
1166
1167    #[test]
1168    fn pgoutput_text_scalars_are_coerced_to_declared_column_types() {
1169        for (name, text, ty, expected) in [
1170            ("true-bool", "t", "bool", ScalarValue::Bool(true)),
1171            ("false-bool", "f", "bool", ScalarValue::Bool(false)),
1172            (
1173                "numeric",
1174                "12.50",
1175                "numeric",
1176                ScalarValue::Float("12.50".to_owned()),
1177            ),
1178            (
1179                "bytea",
1180                "\\xDEad",
1181                "bytea",
1182                ScalarValue::Bytes(vec![0xde, 0xad]),
1183            ),
1184        ] {
1185            check!(
1186                coerce_value(ScalarValue::Text(text.to_owned()), ty).expect("value coerces")
1187                    == expected,
1188                "case {name}"
1189            );
1190        }
1191    }
1192
1193    #[test]
1194    fn invalid_bool_text_is_rejected() {
1195        let error = coerce_value(ScalarValue::Text("yes".to_owned()), "bool")
1196            .expect_err("invalid bool should fail");
1197
1198        match error {
1199            PostgresConnectError::Backend(message) => {
1200                check!(message.contains("invalid bool pgoutput text value"));
1201                check!(message.contains("yes"));
1202            }
1203            error => panic!("expected backend error, got {error:?}"),
1204        }
1205    }
1206
1207    #[test]
1208    fn bytea_text_supports_escape_and_raw_forms() {
1209        for (name, input, expected) in [
1210            ("hex", "\\x0001ff", vec![0, 1, 255]),
1211            ("raw", "raw", b"raw".to_vec()),
1212        ] {
1213            check!(
1214                coerce_bytea(input).expect("bytea coerces") == expected,
1215                "case {name}"
1216            );
1217        }
1218    }
1219
1220    #[test]
1221    fn bytea_hex_rejects_odd_length_and_invalid_pairs() {
1222        let odd = coerce_bytea("\\x0").expect_err("odd-length bytea should fail");
1223        let invalid = coerce_bytea("\\xzz").expect_err("invalid bytea should fail");
1224
1225        match odd {
1226            PostgresConnectError::Backend(message) => {
1227                check!(message.contains("odd hex length"));
1228            }
1229            error => panic!("expected backend error, got {error:?}"),
1230        }
1231        match invalid {
1232            PostgresConnectError::Backend(message) => {
1233                check!(message.contains("invalid bytea pgoutput text value"));
1234                check!(message.contains("zz"));
1235            }
1236            error => panic!("expected backend error, got {error:?}"),
1237        }
1238    }
1239
1240    #[test]
1241    fn update_without_old_tuple_does_not_trigger_key_change_check() {
1242        let mut cache = RelationCache::default();
1243        cache.apply_relation(orders_relation("text"));
1244
1245        let difference = cache
1246            .translate(RowEvent {
1247                relation_id: RelationId(7),
1248                lsn: PgLsn(0x35),
1249                commit_lsn: None,
1250                txid: None,
1251                commit_timestamp_ms: None,
1252                kind: RowEventKind::Update {
1253                    old: Vec::new(),
1254                    old_tuple_kind: RowTupleKind::Full,
1255                },
1256                values: vec![id(42), status("paid")],
1257            })
1258            .expect("missing old tuple should translate");
1259
1260        check!((difference.key.columns, difference.before) == (vec![id(42)], Vec::new()));
1261    }
1262
1263    #[test]
1264    fn key_column_presence_ignores_non_key_columns() {
1265        let schema = TableSchema {
1266            schema: "public".to_owned(),
1267            table: "orders".to_owned(),
1268            columns: orders_relation("text").columns,
1269        };
1270
1271        check!(super::has_any_key_column(&schema, &[id(42)]));
1272        check!(!super::has_any_key_column(&schema, &[status("paid")]));
1273        check!(!super::has_any_key_column(&schema, &[]));
1274    }
1275
1276    #[test]
1277    fn type_oids_map_to_supported_scalar_names() {
1278        for (oid, name) in [
1279            (16, "bool"),
1280            (17, "bytea"),
1281            (20, "int8"),
1282            (21, "int2"),
1283            (23, "int4"),
1284            (25, "text"),
1285            (700, "float4"),
1286            (701, "float8"),
1287            (1043, "varchar"),
1288            (1700, "numeric"),
1289            (999_999, "unknown"),
1290        ] {
1291            check!(super::type_name(oid) == name);
1292        }
1293    }
1294}
1295
1296#[cfg(test)]
1297mod decode_tests {
1298    use assert2::check;
1299
1300    use super::{
1301        DecodedMessage, RelationEvent, RowEvent, RowEventKind, RowTupleKind,
1302        decode_pgoutput_message,
1303    };
1304    use crate::{
1305        PgLsn, PostgresConnectError,
1306        ids::{CommitLsn, EndLsn, RelationId, TransactionId},
1307        model::{
1308            ColumnSchema, ColumnValue, EntityDifference, EntityKey, Operation, ScalarValue,
1309            TableSchema,
1310        },
1311    };
1312
1313    fn put_i16(bytes: &mut Vec<u8>, value: i16) {
1314        bytes.extend_from_slice(&value.to_be_bytes());
1315    }
1316
1317    fn put_i32(bytes: &mut Vec<u8>, value: i32) {
1318        bytes.extend_from_slice(&value.to_be_bytes());
1319    }
1320
1321    fn put_u32(bytes: &mut Vec<u8>, value: u32) {
1322        bytes.extend_from_slice(&value.to_be_bytes());
1323    }
1324
1325    fn put_i64(bytes: &mut Vec<u8>, value: i64) {
1326        bytes.extend_from_slice(&value.to_be_bytes());
1327    }
1328
1329    fn put_cstr(bytes: &mut Vec<u8>, value: &str) {
1330        bytes.extend_from_slice(value.as_bytes());
1331        bytes.push(0);
1332    }
1333
1334    fn put_text_value(bytes: &mut Vec<u8>, value: &str) {
1335        bytes.push(b't');
1336        let length = i32::try_from(value.len()).expect("test value length should fit i32");
1337        put_i32(bytes, length);
1338        bytes.extend_from_slice(value.as_bytes());
1339    }
1340
1341    fn orders_relation_message() -> super::RelationEvent {
1342        super::RelationEvent {
1343            relation_id: RelationId(7),
1344            schema: "public".to_owned(),
1345            table: "orders".to_owned(),
1346            columns: vec![
1347                ColumnSchema {
1348                    name: "id".to_owned(),
1349                    type_name: "int8".to_owned(),
1350                    key: true,
1351                },
1352                ColumnSchema {
1353                    name: "status".to_owned(),
1354                    type_name: "text".to_owned(),
1355                    key: false,
1356                },
1357            ],
1358        }
1359    }
1360
1361    fn wide_relation_message(columns: i16) -> Vec<u8> {
1362        let mut bytes = vec![b'R'];
1363        put_i32(&mut bytes, 7);
1364        put_cstr(&mut bytes, "public");
1365        put_cstr(&mut bytes, "wide");
1366        bytes.push(b'd');
1367        put_i16(&mut bytes, columns);
1368        for index in 0..columns {
1369            bytes.push(u8::from(index == 0));
1370            put_cstr(&mut bytes, &format!("col{index}"));
1371            put_i32(&mut bytes, 20);
1372            put_i32(&mut bytes, -1);
1373        }
1374        bytes
1375    }
1376
1377    fn put_wide_tuple(bytes: &mut Vec<u8>, columns: i16) {
1378        put_i16(bytes, columns);
1379        for index in 0..columns {
1380            if index == columns - 2 {
1381                bytes.push(b'n');
1382            } else if index == columns - 1 {
1383                bytes.push(b'u');
1384            } else {
1385                put_text_value(bytes, &index.to_string());
1386            }
1387        }
1388    }
1389
1390    #[test]
1391    fn matching_col_names_decode_and_translate_wide_insert_update_and_delete() {
1392        for columns in [100, 500, 1_000, 1_600] {
1393            let DecodedMessage::Relation(relation) =
1394                decode_pgoutput_message(&wide_relation_message(columns), PgLsn(0), None).unwrap()
1395            else {
1396                panic!("expected relation")
1397            };
1398            let schema = TableSchema {
1399                schema: relation.schema.clone(),
1400                table: relation.table.clone(),
1401                columns: relation.columns.clone(),
1402            };
1403            let mut cache = super::RelationCache::default();
1404            cache.apply_relation(relation);
1405            let expected_values: Vec<_> = (0..columns)
1406                .map(|index| ColumnValue {
1407                    name: format!("col{index}"),
1408                    value: if index == columns - 2 {
1409                        ScalarValue::Null
1410                    } else if index == columns - 1 {
1411                        ScalarValue::UnchangedToast
1412                    } else {
1413                        ScalarValue::Int(i64::from(index))
1414                    },
1415                })
1416                .collect();
1417            for (tag, op) in [
1418                (b'I', Operation::Insert),
1419                (b'U', Operation::Update),
1420                (b'D', Operation::Delete),
1421            ] {
1422                let mut bytes = vec![tag];
1423                put_i32(&mut bytes, 7);
1424                if tag == b'U' {
1425                    bytes.push(b'O');
1426                    put_wide_tuple(&mut bytes, columns);
1427                }
1428                bytes.push(if tag == b'D' { b'O' } else { b'N' });
1429                put_wide_tuple(&mut bytes, columns);
1430                let DecodedMessage::Row(row) =
1431                    decode_pgoutput_message(&bytes, PgLsn(50), Some(TransactionId(101))).unwrap()
1432                else {
1433                    panic!("expected row")
1434                };
1435                let expected = EntityDifference {
1436                    table: "public.wide".into(),
1437                    key: EntityKey {
1438                        table: "public.wide".into(),
1439                        columns: vec![expected_values[0].clone()],
1440                    },
1441                    op,
1442                    before: if op == Operation::Insert {
1443                        Vec::new()
1444                    } else {
1445                        expected_values.clone()
1446                    },
1447                    after: if op == Operation::Delete {
1448                        Vec::new()
1449                    } else {
1450                        expected_values.clone()
1451                    },
1452                    lsn: PgLsn(50),
1453                    txid: Some(TransactionId(101)),
1454                    commit_timestamp_ms: None,
1455                    schema: schema.clone(),
1456                };
1457                assert2::assert!(cache.translate(row).unwrap() == expected);
1458            }
1459        }
1460    }
1461
1462    #[test]
1463    fn decodes_relation_message() {
1464        let mut bytes = vec![b'R'];
1465        put_i32(&mut bytes, 7);
1466        put_cstr(&mut bytes, "public");
1467        put_cstr(&mut bytes, "orders");
1468        bytes.push(b'd');
1469        put_i16(&mut bytes, 2);
1470        bytes.push(1);
1471        put_cstr(&mut bytes, "id");
1472        put_i32(&mut bytes, 20);
1473        put_i32(&mut bytes, -1);
1474        bytes.push(0);
1475        put_cstr(&mut bytes, "status");
1476        put_i32(&mut bytes, 25);
1477        put_i32(&mut bytes, -1);
1478
1479        let decoded = decode_pgoutput_message(&bytes, PgLsn(0x2a), Some(TransactionId(99)))
1480            .expect("relation message should decode");
1481
1482        check!(
1483            decoded
1484                == DecodedMessage::Relation(RelationEvent {
1485                    relation_id: RelationId(7),
1486                    schema: "public".to_owned(),
1487                    table: "orders".to_owned(),
1488                    columns: vec![
1489                        ColumnSchema {
1490                            name: "id".to_owned(),
1491                            type_name: "int8".to_owned(),
1492                            key: true,
1493                        },
1494                        ColumnSchema {
1495                            name: "status".to_owned(),
1496                            type_name: "text".to_owned(),
1497                            key: false,
1498                        },
1499                    ],
1500                })
1501        );
1502    }
1503
1504    #[test]
1505    fn decodes_insert_message_to_row_event() {
1506        let mut bytes = vec![b'I'];
1507        put_i32(&mut bytes, 7);
1508        bytes.push(b'N');
1509        put_i16(&mut bytes, 2);
1510        put_text_value(&mut bytes, "42");
1511        put_text_value(&mut bytes, "paid");
1512
1513        let decoded = decode_pgoutput_message(&bytes, PgLsn(0x2a), Some(TransactionId(99)))
1514            .expect("insert message should decode");
1515
1516        check!(
1517            decoded
1518                == DecodedMessage::Row(RowEvent {
1519                    relation_id: RelationId(7),
1520                    lsn: PgLsn(0x2a),
1521                    commit_lsn: None,
1522                    txid: Some(TransactionId(99)),
1523                    commit_timestamp_ms: None,
1524                    kind: RowEventKind::Insert,
1525                    values: vec![
1526                        ColumnValue {
1527                            name: "col0".to_owned(),
1528                            value: ScalarValue::Text("42".to_owned()),
1529                        },
1530                        ColumnValue {
1531                            name: "col1".to_owned(),
1532                            value: ScalarValue::Text("paid".to_owned()),
1533                        },
1534                    ],
1535                })
1536        );
1537    }
1538
1539    #[test]
1540    fn decoded_relation_and_insert_translate_with_schema_column_names() {
1541        let mut relation_bytes = vec![b'R'];
1542        put_i32(&mut relation_bytes, 7);
1543        put_cstr(&mut relation_bytes, "public");
1544        put_cstr(&mut relation_bytes, "orders");
1545        relation_bytes.push(b'd');
1546        put_i16(&mut relation_bytes, 2);
1547        relation_bytes.push(1);
1548        put_cstr(&mut relation_bytes, "id");
1549        put_i32(&mut relation_bytes, 20);
1550        put_i32(&mut relation_bytes, -1);
1551        relation_bytes.push(0);
1552        put_cstr(&mut relation_bytes, "status");
1553        put_i32(&mut relation_bytes, 25);
1554        put_i32(&mut relation_bytes, -1);
1555
1556        let mut cache = super::RelationCache::default();
1557        let DecodedMessage::Relation(relation) =
1558            decode_pgoutput_message(&relation_bytes, PgLsn(0), None)
1559                .expect("relation should decode")
1560        else {
1561            panic!("expected relation");
1562        };
1563        cache.apply_relation(relation);
1564
1565        let mut insert_bytes = vec![b'I'];
1566        put_i32(&mut insert_bytes, 7);
1567        insert_bytes.push(b'N');
1568        put_i16(&mut insert_bytes, 2);
1569        put_text_value(&mut insert_bytes, "42");
1570        put_text_value(&mut insert_bytes, "paid");
1571
1572        let DecodedMessage::Row(row) =
1573            decode_pgoutput_message(&insert_bytes, PgLsn(0x32), Some(TransactionId(101)))
1574                .expect("insert should decode")
1575        else {
1576            panic!("expected row");
1577        };
1578        let difference = cache
1579            .translate(row)
1580            .expect("decoded insert should translate");
1581
1582        check!(
1583            (difference.key.columns, difference.after)
1584                == (
1585                    vec![ColumnValue {
1586                        name: "id".to_owned(),
1587                        value: ScalarValue::Int(42),
1588                    }],
1589                    vec![
1590                        ColumnValue {
1591                            name: "id".to_owned(),
1592                            value: ScalarValue::Int(42),
1593                        },
1594                        ColumnValue {
1595                            name: "status".to_owned(),
1596                            value: ScalarValue::Text("paid".to_owned()),
1597                        },
1598                    ],
1599                )
1600        );
1601    }
1602
1603    #[test]
1604    fn decoded_int8_text_value_translates_to_int_scalar_for_key() {
1605        let mut cache = super::RelationCache::default();
1606        cache.apply_relation(orders_relation_message());
1607
1608        let mut insert_bytes = vec![b'I'];
1609        put_i32(&mut insert_bytes, 7);
1610        insert_bytes.push(b'N');
1611        put_i16(&mut insert_bytes, 2);
1612        put_text_value(&mut insert_bytes, "42");
1613        put_text_value(&mut insert_bytes, "paid");
1614
1615        let DecodedMessage::Row(row) =
1616            decode_pgoutput_message(&insert_bytes, PgLsn(0x32), Some(TransactionId(101)))
1617                .expect("insert should decode")
1618        else {
1619            panic!("expected row");
1620        };
1621        let difference = cache
1622            .translate(row)
1623            .expect("decoded insert should translate");
1624
1625        check!(
1626            difference.key.columns
1627                == vec![ColumnValue {
1628                    name: "id".to_owned(),
1629                    value: ScalarValue::Int(42),
1630                }]
1631        );
1632    }
1633
1634    #[test]
1635    fn decoded_delete_key_tuple_translates_with_key_column_name() {
1636        let mut cache = super::RelationCache::default();
1637        cache.apply_relation(orders_relation_message());
1638
1639        let mut delete_bytes = vec![b'D'];
1640        put_i32(&mut delete_bytes, 7);
1641        delete_bytes.push(b'K');
1642        put_i16(&mut delete_bytes, 2);
1643        put_text_value(&mut delete_bytes, "42");
1644        delete_bytes.push(b'n');
1645
1646        let DecodedMessage::Row(row) =
1647            decode_pgoutput_message(&delete_bytes, PgLsn(0x35), Some(TransactionId(102)))
1648                .expect("delete should decode")
1649        else {
1650            panic!("expected row");
1651        };
1652        let difference = cache
1653            .translate(row)
1654            .expect("decoded delete should translate");
1655
1656        let expected = vec![ColumnValue {
1657            name: "id".to_owned(),
1658            value: ScalarValue::Int(42),
1659        }];
1660        check!(
1661            (difference.key.columns, difference.before)
1662                == (
1663                    expected,
1664                    vec![
1665                        ColumnValue {
1666                            name: "id".to_owned(),
1667                            value: ScalarValue::Int(42),
1668                        },
1669                        ColumnValue {
1670                            name: "status".to_owned(),
1671                            value: ScalarValue::Null,
1672                        },
1673                    ],
1674                )
1675        );
1676    }
1677
1678    #[test]
1679    fn decoded_update_key_old_tuple_translates_for_non_key_update() {
1680        let mut cache = super::RelationCache::default();
1681        cache.apply_relation(orders_relation_message());
1682
1683        let mut update_bytes = vec![b'U'];
1684        put_i32(&mut update_bytes, 7);
1685        update_bytes.push(b'K');
1686        put_i16(&mut update_bytes, 2);
1687        put_text_value(&mut update_bytes, "42");
1688        update_bytes.push(b'n');
1689        update_bytes.push(b'N');
1690        put_i16(&mut update_bytes, 2);
1691        put_text_value(&mut update_bytes, "42");
1692        put_text_value(&mut update_bytes, "paid");
1693
1694        let DecodedMessage::Row(row) =
1695            decode_pgoutput_message(&update_bytes, PgLsn(0x36), Some(TransactionId(103)))
1696                .expect("update should decode")
1697        else {
1698            panic!("expected row");
1699        };
1700        let difference = cache
1701            .translate(row)
1702            .expect("decoded update should translate");
1703
1704        check!(
1705            (difference.before, difference.after)
1706                == (
1707                    vec![
1708                        ColumnValue {
1709                            name: "id".to_owned(),
1710                            value: ScalarValue::Int(42),
1711                        },
1712                        ColumnValue {
1713                            name: "status".to_owned(),
1714                            value: ScalarValue::Null,
1715                        }
1716                    ],
1717                    vec![
1718                        ColumnValue {
1719                            name: "id".to_owned(),
1720                            value: ScalarValue::Int(42),
1721                        },
1722                        ColumnValue {
1723                            name: "status".to_owned(),
1724                            value: ScalarValue::Text("paid".to_owned()),
1725                        },
1726                    ],
1727                )
1728        );
1729    }
1730
1731    #[test]
1732    fn decodes_delete_message_to_row_event() {
1733        let mut bytes = vec![b'D'];
1734        put_i32(&mut bytes, 7);
1735        bytes.push(b'K');
1736        put_i16(&mut bytes, 2);
1737        put_text_value(&mut bytes, "42");
1738        bytes.push(b'n');
1739
1740        let decoded = decode_pgoutput_message(&bytes, PgLsn(0x2b), None)
1741            .expect("delete message should decode");
1742
1743        check!(
1744            decoded
1745                == DecodedMessage::Row(RowEvent {
1746                    relation_id: RelationId(7),
1747                    lsn: PgLsn(0x2b),
1748                    commit_lsn: None,
1749                    txid: None,
1750                    commit_timestamp_ms: None,
1751                    kind: RowEventKind::Delete {
1752                        tuple_kind: RowTupleKind::Key,
1753                    },
1754                    values: vec![
1755                        ColumnValue {
1756                            name: "col0".to_owned(),
1757                            value: ScalarValue::Text("42".to_owned()),
1758                        },
1759                        ColumnValue {
1760                            name: "col1".to_owned(),
1761                            value: ScalarValue::Null,
1762                        },
1763                    ],
1764                })
1765        );
1766    }
1767
1768    #[test]
1769    fn decodes_update_message_to_row_event() {
1770        let mut bytes = vec![b'U'];
1771        put_i32(&mut bytes, 7);
1772        bytes.push(b'K');
1773        put_i16(&mut bytes, 2);
1774        put_text_value(&mut bytes, "41");
1775        bytes.push(b'n');
1776        bytes.push(b'N');
1777        put_i16(&mut bytes, 2);
1778        put_text_value(&mut bytes, "42");
1779        put_text_value(&mut bytes, "paid");
1780
1781        let decoded = decode_pgoutput_message(&bytes, PgLsn(0x2c), Some(TransactionId(100)))
1782            .expect("update message should decode");
1783
1784        check!(
1785            decoded
1786                == DecodedMessage::Row(RowEvent {
1787                    relation_id: RelationId(7),
1788                    lsn: PgLsn(0x2c),
1789                    commit_lsn: None,
1790                    txid: Some(TransactionId(100)),
1791                    commit_timestamp_ms: None,
1792                    kind: RowEventKind::Update {
1793                        old: vec![
1794                            ColumnValue {
1795                                name: "col0".to_owned(),
1796                                value: ScalarValue::Text("41".to_owned()),
1797                            },
1798                            ColumnValue {
1799                                name: "col1".to_owned(),
1800                                value: ScalarValue::Null,
1801                            },
1802                        ],
1803                        old_tuple_kind: RowTupleKind::Key,
1804                    },
1805                    values: vec![
1806                        ColumnValue {
1807                            name: "col0".to_owned(),
1808                            value: ScalarValue::Text("42".to_owned()),
1809                        },
1810                        ColumnValue {
1811                            name: "col1".to_owned(),
1812                            value: ScalarValue::Text("paid".to_owned()),
1813                        },
1814                    ],
1815                })
1816        );
1817    }
1818
1819    #[test]
1820    fn decodes_unchanged_toast_as_distinct_scalar_value() {
1821        let mut bytes = vec![b'U'];
1822        put_i32(&mut bytes, 7);
1823        bytes.push(b'N');
1824        put_i16(&mut bytes, 2);
1825        put_text_value(&mut bytes, "42");
1826        bytes.push(b'u');
1827
1828        let decoded = decode_pgoutput_message(&bytes, PgLsn(0x33), Some(TransactionId(100)))
1829            .expect("update message should decode");
1830
1831        let DecodedMessage::Row(row) = decoded else {
1832            panic!("expected row message");
1833        };
1834        check!(row.values[1].value == ScalarValue::UnchangedToast);
1835    }
1836
1837    #[test]
1838    fn malformed_message_returns_backend_error() {
1839        let error = decode_pgoutput_message(&[b'R', 0, 0], PgLsn(0), None)
1840            .expect_err("truncated message should fail");
1841
1842        match error {
1843            PostgresConnectError::Backend(message) => {
1844                check!(message.contains("truncated"));
1845            }
1846            error => panic!("expected backend error, got {error:?}"),
1847        }
1848    }
1849
1850    #[test]
1851    fn trailing_message_bytes_are_rejected() {
1852        let mut bytes = vec![b'I'];
1853        put_i32(&mut bytes, 7);
1854        bytes.push(b'N');
1855        put_i16(&mut bytes, 2);
1856        put_text_value(&mut bytes, "42");
1857        put_text_value(&mut bytes, "paid");
1858        bytes.push(0xff);
1859
1860        let error = decode_pgoutput_message(&bytes, PgLsn(0), None)
1861            .expect_err("trailing bytes should fail");
1862
1863        match error {
1864            PostgresConnectError::Backend(message) => {
1865                check!(message.contains("trailing bytes"));
1866            }
1867            error => panic!("expected backend error, got {error:?}"),
1868        }
1869    }
1870
1871    #[test]
1872    fn decodes_begin_and_commit_messages() {
1873        let mut begin = vec![b'B'];
1874        put_i64(&mut begin, 0x0102_0304);
1875        put_i64(&mut begin, 1_700_000_000);
1876        put_i32(&mut begin, 123);
1877
1878        let decoded = decode_pgoutput_message(&begin, PgLsn(0), None).expect("begin should decode");
1879        check!(
1880            decoded
1881                == DecodedMessage::Begin {
1882                    final_lsn: PgLsn(0x0102_0304),
1883                    xid: TransactionId(123),
1884                }
1885        );
1886
1887        let mut commit = vec![b'C', 0];
1888        put_i64(&mut commit, 0x0102_0305);
1889        put_i64(&mut commit, 0x0102_0306);
1890        put_i64(&mut commit, 1_000);
1891
1892        let decoded =
1893            decode_pgoutput_message(&commit, PgLsn(0), None).expect("commit should decode");
1894        check!(
1895            decoded
1896                == DecodedMessage::Commit {
1897                    commit_lsn: CommitLsn(PgLsn(0x0102_0305)),
1898                    end_lsn: EndLsn(PgLsn(0x0102_0306)),
1899                    commit_timestamp_ms: 946_684_800_001,
1900                }
1901        );
1902    }
1903
1904    #[test]
1905    fn decodes_begin_xid_as_unsigned_u32() {
1906        let mut begin = vec![b'B'];
1907        put_i64(&mut begin, 0x0102_0304);
1908        put_i64(&mut begin, 1_700_000_000);
1909        put_u32(&mut begin, 0x8000_0000);
1910
1911        let decoded = decode_pgoutput_message(&begin, PgLsn(0), None).expect("begin should decode");
1912        check!(
1913            decoded
1914                == DecodedMessage::Begin {
1915                    final_lsn: PgLsn(0x0102_0304),
1916                    xid: TransactionId(2_147_483_648),
1917                }
1918        );
1919    }
1920}
krabka_connect_postgres/pgoutput.rs

krabka_connect_postgres/
pgoutput.rs
