laminar_connectors/postgres/cdc/
decoder.rs

1//! `PostgreSQL` `pgoutput` logical replication protocol decoder.
2//!
3//! Implements a binary protocol parser for the `pgoutput` output plugin
4//! used by `PostgreSQL` logical replication (PG 10+). Parses WAL stream
5//! bytes into structured [`WalMessage`] variants.
6//!
7//! # Protocol Reference
8//!
9//! See `PostgreSQL` docs: "Logical Replication Message Formats"
10//! (<https://www.postgresql.org/docs/current/protocol-logicalrep-message-formats.html>)
11
12use super::lsn::Lsn;
13use super::types::PgColumn;
14use bytes::Bytes;
15
16/// Offset from `PostgreSQL` epoch (2000-01-01) to Unix epoch (1970-01-01)
17/// in microseconds.
18const PG_EPOCH_OFFSET_US: i64 = 946_684_800_000_000;
19const MAX_POSTGRES_COLUMNS: usize = 1_600;
20
21/// A decoded WAL message from the `pgoutput` protocol.
22#[derive(Debug, Clone, PartialEq)]
23pub enum WalMessage {
24    /// Transaction begin.
25    Begin(BeginMessage),
26    /// Transaction commit.
27    Commit(CommitMessage),
28    /// Relation (table) metadata.
29    Relation(RelationMessage),
30    /// Row inserted.
31    Insert(InsertMessage),
32    /// Row updated.
33    Update(UpdateMessage),
34    /// Row deleted.
35    Delete(DeleteMessage),
36    /// Table(s) truncated.
37    Truncate(TruncateMessage),
38    /// Origin information.
39    Origin(OriginMessage),
40    /// Custom type definition.
41    Type(TypeMessage),
42}
43
44/// Transaction begin message.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct BeginMessage {
47    /// LSN of the final record of the transaction.
48    pub final_lsn: Lsn,
49    /// Commit timestamp in milliseconds since Unix epoch.
50    pub commit_ts_ms: i64,
51    /// Transaction ID (XID).
52    pub xid: u32,
53}
54
55/// Transaction commit message.
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct CommitMessage {
58    /// Flags (currently unused by `PostgreSQL`).
59    pub flags: u8,
60    /// LSN of the commit record.
61    pub commit_lsn: Lsn,
62    /// End LSN of the transaction.
63    pub end_lsn: Lsn,
64    /// Commit timestamp in milliseconds since Unix epoch.
65    pub commit_ts_ms: i64,
66}
67
68/// Relation (table schema) message.
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub struct RelationMessage {
71    /// Relation OID.
72    pub relation_id: u32,
73    /// Schema (namespace) name.
74    pub namespace: String,
75    /// Table name.
76    pub name: String,
77    /// Replica identity setting: 'd', 'n', 'f', or 'i'.
78    pub replica_identity: u8,
79    /// Column descriptors.
80    pub columns: Vec<PgColumn>,
81}
82
83/// Row insert message.
84#[derive(Debug, Clone, PartialEq)]
85pub struct InsertMessage {
86    /// Relation OID of the target table.
87    pub relation_id: u32,
88    /// The new row data.
89    pub new_tuple: TupleData,
90}
91
92/// Old tuple representation identified by the pgoutput wire tag.
93#[derive(Debug, Clone, PartialEq)]
94pub enum OldTuple {
95    /// Replica-identity key (`K`); non-key column positions are unavailable.
96    Key(TupleData),
97    /// Old tuple (`O`) emitted for `REPLICA IDENTITY FULL`.
98    Full(TupleData),
99}
100
101/// Row update message.
102#[derive(Debug, Clone, PartialEq)]
103pub struct UpdateMessage {
104    /// Relation OID of the target table.
105    pub relation_id: u32,
106    /// Old identity/full-row data when pgoutput includes a `K` or `O` tuple.
107    pub old_tuple: Option<OldTuple>,
108    /// The new row data.
109    pub new_tuple: TupleData,
110}
111
112/// Row delete message.
113#[derive(Debug, Clone, PartialEq)]
114pub struct DeleteMessage {
115    /// Relation OID of the target table.
116    pub relation_id: u32,
117    /// The old identity or full-row tuple.
118    pub old_tuple: OldTuple,
119}
120
121/// Table truncate message.
122#[derive(Debug, Clone, PartialEq, Eq)]
123pub struct TruncateMessage {
124    /// Relation OIDs of truncated tables.
125    pub relation_ids: Vec<u32>,
126    /// Option flags: bit 0 = CASCADE, bit 1 = RESTART IDENTITY.
127    pub options: u8,
128}
129
130/// Origin message (for replication from a downstream).
131#[derive(Debug, Clone, PartialEq, Eq)]
132pub struct OriginMessage {
133    /// Origin LSN.
134    pub origin_lsn: Lsn,
135    /// Origin name.
136    pub name: String,
137}
138
139/// Custom type definition message.
140#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct TypeMessage {
142    /// Type OID.
143    pub type_id: u32,
144    /// Schema (namespace) name.
145    pub namespace: String,
146    /// Type name.
147    pub name: String,
148}
149
150/// Tuple data containing column values.
151#[derive(Debug, Clone, PartialEq)]
152pub struct TupleData {
153    /// Column values in ordinal order.
154    pub columns: Vec<ColumnValue>,
155}
156
157/// A single column value in a tuple.
158#[derive(Debug, Clone, PartialEq)]
159pub enum ColumnValue {
160    /// NULL value.
161    Null,
162    /// Unchanged TOAST value (not sent by server).
163    Unchanged,
164    /// Text-format value.
165    Text(Bytes),
166}
167
168impl ColumnValue {
169    /// Returns the text value if present.
170    #[must_use]
171    pub fn as_text(&self) -> Option<&str> {
172        match self {
173            ColumnValue::Text(bytes) => std::str::from_utf8(bytes).ok(),
174            _ => None,
175        }
176    }
177
178    /// Returns `true` if the value is NULL.
179    #[must_use]
180    pub fn is_null(&self) -> bool {
181        matches!(self, ColumnValue::Null)
182    }
183}
184
185/// Errors from the `pgoutput` protocol decoder.
186#[derive(Debug, Clone, thiserror::Error)]
187pub enum DecoderError {
188    /// Not enough bytes to read the expected value.
189    #[error("unexpected end of data at offset {offset}, need {needed} bytes")]
190    UnexpectedEof {
191        /// Current position in the buffer.
192        offset: usize,
193        /// Number of bytes needed.
194        needed: usize,
195    },
196
197    /// Invalid message type byte.
198    #[error("unknown message type: 0x{0:02X}")]
199    UnknownMessageType(u8),
200
201    /// Invalid or corrupted data.
202    #[error("invalid data: {0}")]
203    InvalidData(String),
204
205    /// Invalid UTF-8 in a string field.
206    #[error("invalid UTF-8 at offset {0}")]
207    InvalidUtf8(usize),
208}
209
210/// A cursor for reading binary data from a byte buffer.
211struct Cursor {
212    data: Bytes,
213    pos: usize,
214}
215
216impl Cursor {
217    fn new(data: Bytes) -> Self {
218        Self { data, pos: 0 }
219    }
220
221    fn remaining(&self) -> usize {
222        self.data.len().saturating_sub(self.pos)
223    }
224
225    fn read_u8(&mut self) -> Result<u8, DecoderError> {
226        if self.pos >= self.data.len() {
227            return Err(DecoderError::UnexpectedEof {
228                offset: self.pos,
229                needed: 1,
230            });
231        }
232        let val = self.data[self.pos];
233        self.pos += 1;
234        Ok(val)
235    }
236
237    fn read_i16(&mut self) -> Result<i16, DecoderError> {
238        self.check_remaining(2)?;
239        let val = i16::from_be_bytes([self.data[self.pos], self.data[self.pos + 1]]);
240        self.pos += 2;
241        Ok(val)
242    }
243
244    fn read_i32(&mut self) -> Result<i32, DecoderError> {
245        self.check_remaining(4)?;
246        let bytes: [u8; 4] = self.data[self.pos..self.pos + 4].try_into().map_err(|_| {
247            DecoderError::UnexpectedEof {
248                offset: self.pos,
249                needed: 4,
250            }
251        })?;
252        let val = i32::from_be_bytes(bytes);
253        self.pos += 4;
254        Ok(val)
255    }
256
257    fn read_u32(&mut self) -> Result<u32, DecoderError> {
258        self.check_remaining(4)?;
259        let bytes: [u8; 4] = self.data[self.pos..self.pos + 4].try_into().map_err(|_| {
260            DecoderError::UnexpectedEof {
261                offset: self.pos,
262                needed: 4,
263            }
264        })?;
265        let val = u32::from_be_bytes(bytes);
266        self.pos += 4;
267        Ok(val)
268    }
269
270    fn read_i64(&mut self) -> Result<i64, DecoderError> {
271        self.check_remaining(8)?;
272        let bytes: [u8; 8] = self.data[self.pos..self.pos + 8].try_into().map_err(|_| {
273            DecoderError::UnexpectedEof {
274                offset: self.pos,
275                needed: 8,
276            }
277        })?;
278        let val = i64::from_be_bytes(bytes);
279        self.pos += 8;
280        Ok(val)
281    }
282
283    fn read_u64(&mut self) -> Result<u64, DecoderError> {
284        self.check_remaining(8)?;
285        let bytes: [u8; 8] = self.data[self.pos..self.pos + 8].try_into().map_err(|_| {
286            DecoderError::UnexpectedEof {
287                offset: self.pos,
288                needed: 8,
289            }
290        })?;
291        let val = u64::from_be_bytes(bytes);
292        self.pos += 8;
293        Ok(val)
294    }
295
296    /// Reads a null-terminated string.
297    fn read_cstring(&mut self) -> Result<String, DecoderError> {
298        let start = self.pos;
299        let nul_pos = self.data[self.pos..]
300            .iter()
301            .position(|&b| b == 0)
302            .ok_or(DecoderError::InvalidData("unterminated string".to_string()))?;
303
304        let s = std::str::from_utf8(&self.data[self.pos..self.pos + nul_pos])
305            .map_err(|_| DecoderError::InvalidUtf8(start))?;
306
307        self.pos += nul_pos + 1; // skip the NUL byte
308        Ok(s.to_string())
309    }
310
311    fn read_bytes(&mut self, len: usize) -> Result<Bytes, DecoderError> {
312        self.check_remaining(len)?;
313        let slice = self.data.slice(self.pos..self.pos + len);
314        self.pos += len;
315        Ok(slice)
316    }
317
318    fn check_remaining(&self, needed: usize) -> Result<(), DecoderError> {
319        if self.remaining() < needed {
320            return Err(DecoderError::UnexpectedEof {
321                offset: self.pos,
322                needed,
323            });
324        }
325        Ok(())
326    }
327}
328
329/// Converts a `PostgreSQL` timestamp (microseconds since 2000-01-01) to
330/// milliseconds since Unix epoch (1970-01-01).
331///
332/// # Errors
333///
334/// Returns [`DecoderError`] when converting to the Unix epoch overflows.
335pub(super) fn pg_timestamp_to_unix_ms(pg_us: i64) -> Result<i64, DecoderError> {
336    pg_us
337        .checked_add(PG_EPOCH_OFFSET_US)
338        .map(|unix_us| unix_us.div_euclid(1000))
339        .ok_or_else(|| DecoderError::InvalidData("PostgreSQL timestamp overflow".into()))
340}
341
342/// Decodes a single `pgoutput` WAL message from raw bytes.
343///
344/// # Errors
345///
346/// Returns [`DecoderError`] if the data is truncated, malformed, or
347/// contains an unknown message type.
348pub(super) fn decode_message(data: Bytes) -> Result<WalMessage, DecoderError> {
349    if data.is_empty() {
350        return Err(DecoderError::InvalidData("empty message".to_string()));
351    }
352
353    let mut cur = Cursor::new(data);
354    let msg_type = cur.read_u8()?;
355
356    let message = match msg_type {
357        b'B' => decode_begin(&mut cur),
358        b'C' => decode_commit(&mut cur),
359        b'R' => decode_relation(&mut cur),
360        b'I' => decode_insert(&mut cur),
361        b'U' => decode_update(&mut cur),
362        b'D' => decode_delete(&mut cur),
363        b'T' => decode_truncate(&mut cur),
364        b'O' => decode_origin(&mut cur),
365        b'Y' => decode_type(&mut cur),
366        _ => Err(DecoderError::UnknownMessageType(msg_type)),
367    }?;
368    if cur.remaining() != 0 {
369        return Err(DecoderError::InvalidData(format!(
370            "trailing bytes after pgoutput message: {}",
371            cur.remaining()
372        )));
373    }
374    Ok(message)
375}
376
377fn decode_begin(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
378    let final_lsn = Lsn::new(cur.read_u64()?);
379    let commit_ts_us = cur.read_i64()?;
380    let xid = cur.read_u32()?;
381    Ok(WalMessage::Begin(BeginMessage {
382        final_lsn,
383        commit_ts_ms: pg_timestamp_to_unix_ms(commit_ts_us)?,
384        xid,
385    }))
386}
387
388fn decode_commit(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
389    let flags = cur.read_u8()?;
390    if flags != 0 {
391        return Err(DecoderError::InvalidData(format!(
392            "unsupported COMMIT flags: 0x{flags:02X}"
393        )));
394    }
395    let commit_lsn = Lsn::new(cur.read_u64()?);
396    let end_lsn = Lsn::new(cur.read_u64()?);
397    let commit_ts_us = cur.read_i64()?;
398    Ok(WalMessage::Commit(CommitMessage {
399        flags,
400        commit_lsn,
401        end_lsn,
402        commit_ts_ms: pg_timestamp_to_unix_ms(commit_ts_us)?,
403    }))
404}
405
406fn decode_relation(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
407    let relation_id = cur.read_u32()?;
408    let namespace = cur.read_cstring()?;
409    let name = cur.read_cstring()?;
410    let replica_identity = cur.read_u8()?;
411    let n_cols_raw = cur.read_i16()?;
412    let n_cols = usize::try_from(n_cols_raw)
413        .map_err(|_| DecoderError::InvalidData(format!("negative column count: {n_cols_raw}")))?;
414    validate_column_count(n_cols)?;
415
416    let mut columns = Vec::with_capacity(n_cols);
417    for _ in 0..n_cols {
418        let flags = cur.read_u8()?;
419        let col_name = cur.read_cstring()?;
420        let type_oid = cur.read_u32()?;
421        let type_modifier = cur.read_i32()?;
422        columns.push(PgColumn::new(
423            col_name,
424            type_oid,
425            type_modifier,
426            flags & 1 != 0,
427        ));
428    }
429
430    Ok(WalMessage::Relation(RelationMessage {
431        relation_id,
432        namespace,
433        name,
434        replica_identity,
435        columns,
436    }))
437}
438
439fn decode_insert(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
440    let relation_id = cur.read_u32()?;
441    let tag = cur.read_u8()?;
442    if tag != b'N' {
443        return Err(DecoderError::InvalidData(format!(
444            "expected 'N' tag in INSERT, got 0x{tag:02X}"
445        )));
446    }
447    let new_tuple = decode_tuple_data(cur)?;
448    Ok(WalMessage::Insert(InsertMessage {
449        relation_id,
450        new_tuple,
451    }))
452}
453
454fn decode_update(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
455    let relation_id = cur.read_u32()?;
456    let tag = cur.read_u8()?;
457
458    let (old_tuple, new_tuple) = match tag {
459        // No old tuple, just new
460        b'N' => (None, decode_tuple_data(cur)?),
461        // Old replica-identity key or full tuple followed by new.
462        b'K' | b'O' => {
463            let old_data = decode_tuple_data(cur)?;
464            let old = if tag == b'K' {
465                OldTuple::Key(old_data)
466            } else {
467                OldTuple::Full(old_data)
468            };
469            let new_tag = cur.read_u8()?;
470            if new_tag != b'N' {
471                return Err(DecoderError::InvalidData(format!(
472                    "expected 'N' tag after old tuple in UPDATE, got 0x{new_tag:02X}"
473                )));
474            }
475            let new = decode_tuple_data(cur)?;
476            (Some(old), new)
477        }
478        _ => {
479            return Err(DecoderError::InvalidData(format!(
480                "unexpected tag in UPDATE: 0x{tag:02X}"
481            )));
482        }
483    };
484
485    Ok(WalMessage::Update(UpdateMessage {
486        relation_id,
487        old_tuple,
488        new_tuple,
489    }))
490}
491
492fn decode_delete(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
493    let relation_id = cur.read_u32()?;
494    let tag = cur.read_u8()?;
495    if tag != b'K' && tag != b'O' {
496        return Err(DecoderError::InvalidData(format!(
497            "expected 'K' or 'O' tag in DELETE, got 0x{tag:02X}"
498        )));
499    }
500    let old_data = decode_tuple_data(cur)?;
501    let old_tuple = if tag == b'K' {
502        OldTuple::Key(old_data)
503    } else {
504        OldTuple::Full(old_data)
505    };
506    Ok(WalMessage::Delete(DeleteMessage {
507        relation_id,
508        old_tuple,
509    }))
510}
511
512fn decode_truncate(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
513    let n_relations_raw = cur.read_u32()?;
514    let n_relations = usize::try_from(n_relations_raw).map_err(|_| {
515        DecoderError::InvalidData(format!(
516            "TRUNCATE relation count {n_relations_raw} does not fit this platform"
517        ))
518    })?;
519    let options = cur.read_u8()?;
520    let relation_bytes = n_relations
521        .checked_mul(std::mem::size_of::<u32>())
522        .ok_or_else(|| DecoderError::InvalidData("TRUNCATE relation count overflow".into()))?;
523    if relation_bytes != cur.remaining() {
524        return Err(DecoderError::InvalidData(format!(
525            "TRUNCATE relation count {n_relations} requires {relation_bytes} bytes, but {} remain",
526            cur.remaining()
527        )));
528    }
529    let mut relation_ids = Vec::with_capacity(n_relations);
530    for _ in 0..n_relations {
531        relation_ids.push(cur.read_u32()?);
532    }
533    Ok(WalMessage::Truncate(TruncateMessage {
534        relation_ids,
535        options,
536    }))
537}
538
539fn decode_origin(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
540    let origin_lsn = Lsn::new(cur.read_u64()?);
541    let name = cur.read_cstring()?;
542    Ok(WalMessage::Origin(OriginMessage { origin_lsn, name }))
543}
544
545fn decode_type(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
546    let type_id = cur.read_u32()?;
547    let namespace = cur.read_cstring()?;
548    let name = cur.read_cstring()?;
549    Ok(WalMessage::Type(TypeMessage {
550        type_id,
551        namespace,
552        name,
553    }))
554}
555
556fn decode_tuple_data(cur: &mut Cursor) -> Result<TupleData, DecoderError> {
557    let n_cols_raw = cur.read_i16()?;
558    let n_cols = usize::try_from(n_cols_raw)
559        .map_err(|_| DecoderError::InvalidData(format!("negative column count: {n_cols_raw}")))?;
560    validate_column_count(n_cols)?;
561    let mut columns = Vec::with_capacity(n_cols);
562
563    for _ in 0..n_cols {
564        let col_type = cur.read_u8()?;
565        match col_type {
566            b'n' => columns.push(ColumnValue::Null),
567            b'u' => columns.push(ColumnValue::Unchanged),
568            b't' => {
569                let len_raw = cur.read_i32()?;
570                let len = usize::try_from(len_raw).map_err(|_| {
571                    DecoderError::InvalidData(format!("negative text length: {len_raw}"))
572                })?;
573                let data = cur.read_bytes(len)?;
574                std::str::from_utf8(&data).map_err(|_| DecoderError::InvalidUtf8(cur.pos - len))?;
575                columns.push(ColumnValue::Text(data));
576            }
577            _ => {
578                return Err(DecoderError::InvalidData(format!(
579                    "unknown column type: 0x{col_type:02X}"
580                )));
581            }
582        }
583    }
584
585    Ok(TupleData { columns })
586}
587
588fn validate_column_count(count: usize) -> Result<(), DecoderError> {
589    if count > MAX_POSTGRES_COLUMNS {
590        return Err(DecoderError::InvalidData(format!(
591            "column count {count} exceeds PostgreSQL maximum {MAX_POSTGRES_COLUMNS}"
592        )));
593    }
594    Ok(())
595}
596
597#[cfg(test)]
598mod tests {
599    use super::*;
600
601    fn decode_message(data: &[u8]) -> Result<WalMessage, DecoderError> {
602        super::decode_message(Bytes::copy_from_slice(data))
603    }
604
605    // ── Test helpers: build binary pgoutput messages ──
606
607    /// Helper to build binary messages for testing.
608    struct MessageBuilder {
609        buf: Vec<u8>,
610    }
611
612    impl MessageBuilder {
613        fn new(msg_type: u8) -> Self {
614            Self {
615                buf: vec![msg_type],
616            }
617        }
618
619        fn u8(mut self, v: u8) -> Self {
620            self.buf.push(v);
621            self
622        }
623
624        fn i16(mut self, v: i16) -> Self {
625            self.buf.extend_from_slice(&v.to_be_bytes());
626            self
627        }
628
629        fn i32(mut self, v: i32) -> Self {
630            self.buf.extend_from_slice(&v.to_be_bytes());
631            self
632        }
633
634        fn u32(mut self, v: u32) -> Self {
635            self.buf.extend_from_slice(&v.to_be_bytes());
636            self
637        }
638
639        fn i64(mut self, v: i64) -> Self {
640            self.buf.extend_from_slice(&v.to_be_bytes());
641            self
642        }
643
644        fn u64(mut self, v: u64) -> Self {
645            self.buf.extend_from_slice(&v.to_be_bytes());
646            self
647        }
648
649        fn cstring(mut self, s: &str) -> Self {
650            self.buf.extend_from_slice(s.as_bytes());
651            self.buf.push(0);
652            self
653        }
654
655        fn text_col(mut self, s: &str) -> Self {
656            self.buf.push(b't');
657            self.buf.extend_from_slice(&(s.len() as i32).to_be_bytes());
658            self.buf.extend_from_slice(s.as_bytes());
659            self
660        }
661
662        fn null_col(mut self) -> Self {
663            self.buf.push(b'n');
664            self
665        }
666
667        fn unchanged_col(mut self) -> Self {
668            self.buf.push(b'u');
669            self
670        }
671
672        fn build(self) -> Vec<u8> {
673            self.buf
674        }
675    }
676
677    // ── Begin ──
678
679    #[test]
680    fn test_decode_begin() {
681        // Timestamp: 2024-01-01 00:00:00 UTC in PG microseconds
682        // PG epoch = 2000-01-01, so 24 years = ~757382400 seconds
683        let pg_ts_us: i64 = 757_382_400_000_000;
684        let data = MessageBuilder::new(b'B')
685            .u64(0x1234_ABCD) // final_lsn
686            .i64(pg_ts_us) // commit_ts (PG epoch)
687            .u32(42) // xid
688            .build();
689
690        let msg = decode_message(&data).unwrap();
691        match msg {
692            WalMessage::Begin(b) => {
693                assert_eq!(b.final_lsn.as_u64(), 0x1234_ABCD);
694                assert_eq!(b.xid, 42);
695                assert_eq!(b.commit_ts_ms, (pg_ts_us + PG_EPOCH_OFFSET_US) / 1000);
696            }
697            _ => panic!("expected Begin"),
698        }
699    }
700
701    // ── Commit ──
702
703    #[test]
704    fn test_decode_commit() {
705        let pg_ts_us: i64 = 757_382_400_000_000;
706        let data = MessageBuilder::new(b'C')
707            .u8(0) // flags
708            .u64(0x100) // commit_lsn
709            .u64(0x200) // end_lsn
710            .i64(pg_ts_us) // commit_ts
711            .build();
712
713        let msg = decode_message(&data).unwrap();
714        match msg {
715            WalMessage::Commit(c) => {
716                assert_eq!(c.flags, 0);
717                assert_eq!(c.commit_lsn.as_u64(), 0x100);
718                assert_eq!(c.end_lsn.as_u64(), 0x200);
719            }
720            _ => panic!("expected Commit"),
721        }
722    }
723
724    // ── Relation ──
725
726    #[test]
727    fn test_decode_relation() {
728        let data = MessageBuilder::new(b'R')
729            .u32(16384) // relation_id
730            .cstring("public") // namespace
731            .cstring("users") // name
732            .u8(b'd') // replica_identity = default
733            .i16(2) // n_cols
734            // Column 1: id (key)
735            .u8(1) // flags = key
736            .cstring("id")
737            .u32(20) // int8 OID
738            .i32(-1) // type_modifier
739            // Column 2: name (not key)
740            .u8(0) // flags
741            .cstring("name")
742            .u32(25) // text OID
743            .i32(-1)
744            .build();
745
746        let msg = decode_message(&data).unwrap();
747        match msg {
748            WalMessage::Relation(r) => {
749                assert_eq!(r.relation_id, 16384);
750                assert_eq!(r.namespace, "public");
751                assert_eq!(r.name, "users");
752                assert_eq!(r.replica_identity, b'd');
753                assert_eq!(r.columns.len(), 2);
754                assert_eq!(r.columns[0].name, "id");
755                assert!(r.columns[0].is_key);
756                assert_eq!(r.columns[0].type_oid, 20);
757                assert_eq!(r.columns[1].name, "name");
758                assert!(!r.columns[1].is_key);
759            }
760            _ => panic!("expected Relation"),
761        }
762    }
763
764    // ── Insert ──
765
766    #[test]
767    fn test_decode_insert() {
768        let data = MessageBuilder::new(b'I')
769            .u32(16384) // relation_id
770            .u8(b'N') // new tuple tag
771            .i16(3) // n_cols
772            .text_col("42") // id
773            .text_col("Alice") // name
774            .null_col() // nullable field
775            .build();
776
777        let msg = decode_message(&data).unwrap();
778        match msg {
779            WalMessage::Insert(ins) => {
780                assert_eq!(ins.relation_id, 16384);
781                assert_eq!(ins.new_tuple.columns.len(), 3);
782                assert_eq!(ins.new_tuple.columns[0].as_text(), Some("42"));
783                assert_eq!(ins.new_tuple.columns[1].as_text(), Some("Alice"));
784                assert!(ins.new_tuple.columns[2].is_null());
785            }
786            _ => panic!("expected Insert"),
787        }
788    }
789
790    // ── Update (no old tuple) ──
791
792    #[test]
793    fn test_decode_update_no_old() {
794        let data = MessageBuilder::new(b'U')
795            .u32(16384)
796            .u8(b'N') // new tuple directly (no old)
797            .i16(2)
798            .text_col("42")
799            .text_col("Bob")
800            .build();
801
802        let msg = decode_message(&data).unwrap();
803        match msg {
804            WalMessage::Update(upd) => {
805                assert!(upd.old_tuple.is_none());
806                assert_eq!(upd.new_tuple.columns[1].as_text(), Some("Bob"));
807            }
808            _ => panic!("expected Update"),
809        }
810    }
811
812    // ── Update (with old tuple, REPLICA IDENTITY FULL) ──
813
814    #[test]
815    fn test_decode_update_with_old() {
816        let data = MessageBuilder::new(b'U')
817            .u32(16384)
818            .u8(b'O') // old tuple (FULL identity)
819            .i16(2) // old: 2 cols
820            .text_col("42")
821            .text_col("Alice")
822            .u8(b'N') // new tuple tag
823            .i16(2) // new: 2 cols
824            .text_col("42")
825            .text_col("Bob")
826            .build();
827
828        let msg = decode_message(&data).unwrap();
829        match msg {
830            WalMessage::Update(upd) => {
831                let Some(OldTuple::Full(old)) = upd.old_tuple else {
832                    panic!("expected full old tuple");
833                };
834                assert_eq!(old.columns[1].as_text(), Some("Alice"));
835                assert_eq!(upd.new_tuple.columns[1].as_text(), Some("Bob"));
836            }
837            _ => panic!("expected Update"),
838        }
839    }
840
841    // ── Delete ──
842
843    #[test]
844    fn test_decode_delete_key() {
845        let data = MessageBuilder::new(b'D')
846            .u32(16384)
847            .u8(b'K') // key columns only
848            .i16(1)
849            .text_col("42")
850            .build();
851
852        let msg = decode_message(&data).unwrap();
853        match msg {
854            WalMessage::Delete(del) => {
855                assert_eq!(del.relation_id, 16384);
856                let OldTuple::Key(old) = del.old_tuple else {
857                    panic!("expected key old tuple");
858                };
859                assert_eq!(old.columns[0].as_text(), Some("42"));
860            }
861            _ => panic!("expected Delete"),
862        }
863    }
864
865    #[test]
866    fn test_decode_delete_full() {
867        let data = MessageBuilder::new(b'D')
868            .u32(16384)
869            .u8(b'O') // full old row
870            .i16(2)
871            .text_col("42")
872            .text_col("Alice")
873            .build();
874
875        let msg = decode_message(&data).unwrap();
876        match msg {
877            WalMessage::Delete(del) => {
878                let OldTuple::Full(old) = del.old_tuple else {
879                    panic!("expected full old tuple");
880                };
881                assert_eq!(old.columns.len(), 2);
882            }
883            _ => panic!("expected Delete"),
884        }
885    }
886
887    // ── Truncate ──
888
889    #[test]
890    fn test_decode_truncate() {
891        let data = MessageBuilder::new(b'T')
892            .u32(2) // 2 relations
893            .u8(1) // CASCADE
894            .u32(16384) // first relation
895            .u32(16385) // second relation
896            .build();
897
898        let msg = decode_message(&data).unwrap();
899        match msg {
900            WalMessage::Truncate(t) => {
901                assert_eq!(t.relation_ids, vec![16384, 16385]);
902                assert_eq!(t.options, 1);
903            }
904            _ => panic!("expected Truncate"),
905        }
906    }
907
908    // ── Origin ──
909
910    #[test]
911    fn test_decode_origin() {
912        let data = MessageBuilder::new(b'O')
913            .u64(0xABCD)
914            .cstring("upstream")
915            .build();
916
917        let msg = decode_message(&data).unwrap();
918        match msg {
919            WalMessage::Origin(o) => {
920                assert_eq!(o.origin_lsn.as_u64(), 0xABCD);
921                assert_eq!(o.name, "upstream");
922            }
923            _ => panic!("expected Origin"),
924        }
925    }
926
927    // ── Type ──
928
929    #[test]
930    fn test_decode_type() {
931        let data = MessageBuilder::new(b'Y')
932            .u32(12345)
933            .cstring("public")
934            .cstring("my_enum")
935            .build();
936
937        let msg = decode_message(&data).unwrap();
938        match msg {
939            WalMessage::Type(t) => {
940                assert_eq!(t.type_id, 12345);
941                assert_eq!(t.namespace, "public");
942                assert_eq!(t.name, "my_enum");
943            }
944            _ => panic!("expected Type"),
945        }
946    }
947
948    // ── Tuple data with unchanged TOAST column ──
949
950    #[test]
951    fn test_decode_insert_with_unchanged() {
952        let data = MessageBuilder::new(b'I')
953            .u32(16384)
954            .u8(b'N')
955            .i16(2)
956            .text_col("42")
957            .unchanged_col()
958            .build();
959
960        let msg = decode_message(&data).unwrap();
961        match msg {
962            WalMessage::Insert(ins) => {
963                assert_eq!(ins.new_tuple.columns[0].as_text(), Some("42"));
964                assert!(matches!(ins.new_tuple.columns[1], ColumnValue::Unchanged));
965            }
966            _ => panic!("expected Insert"),
967        }
968    }
969
970    // ── Error cases ──
971
972    #[test]
973    fn test_decode_empty_data() {
974        assert!(decode_message(&[]).is_err());
975    }
976
977    #[test]
978    fn test_decode_unknown_type() {
979        let err = decode_message(&[0xFF]).unwrap_err();
980        assert!(matches!(err, DecoderError::UnknownMessageType(0xFF)));
981    }
982
983    #[test]
984    fn test_decode_truncated_begin() {
985        // Begin needs 20 bytes after type, only give 4
986        let data = MessageBuilder::new(b'B').u32(0).build();
987        assert!(decode_message(&data).is_err());
988    }
989
990    #[test]
991    fn test_decode_invalid_insert_tag() {
992        let data = MessageBuilder::new(b'I')
993            .u32(16384)
994            .u8(b'X') // invalid tag
995            .build();
996        assert!(decode_message(&data).is_err());
997    }
998
999    #[test]
1000    fn nonzero_commit_flags_are_rejected() {
1001        let data = MessageBuilder::new(b'C')
1002            .u8(1)
1003            .u64(0x100)
1004            .u64(0x200)
1005            .i64(0)
1006            .build();
1007        let error = decode_message(&data).unwrap_err();
1008        assert!(error.to_string().contains("COMMIT flags"), "{error}");
1009    }
1010
1011    #[test]
1012    fn trailing_bytes_are_rejected() {
1013        let mut data = MessageBuilder::new(b'B').u64(1).i64(0).u32(1).build();
1014        data.push(0xff);
1015        let error = decode_message(&data).unwrap_err();
1016        assert!(error.to_string().contains("trailing bytes"), "{error}");
1017    }
1018
1019    // ── Timestamp conversion ──
1020
1021    #[test]
1022    fn test_pg_timestamp_to_unix_ms() {
1023        // 2000-01-01 00:00:00 UTC in PG epoch = 0
1024        // In Unix epoch = 946684800 seconds = 946684800000 ms
1025        assert_eq!(pg_timestamp_to_unix_ms(0).unwrap(), 946_684_800_000);
1026
1027        // 2024-01-01 00:00:00 UTC
1028        // PG epoch: 757382400 seconds = 757382400000000 us
1029        let pg_us: i64 = 757_382_400_000_000;
1030        let expected_unix_ms = (pg_us + PG_EPOCH_OFFSET_US).div_euclid(1000);
1031        assert_eq!(pg_timestamp_to_unix_ms(pg_us).unwrap(), expected_unix_ms);
1032
1033        // Sub-millisecond instants before Unix epoch round toward negative infinity.
1034        assert_eq!(
1035            pg_timestamp_to_unix_ms(-PG_EPOCH_OFFSET_US - 1).unwrap(),
1036            -1
1037        );
1038        assert_eq!(
1039            pg_timestamp_to_unix_ms(-PG_EPOCH_OFFSET_US - 1_001).unwrap(),
1040            -2
1041        );
1042        assert!(pg_timestamp_to_unix_ms(i64::MAX).is_err());
1043    }
1044
1045    // ── ColumnValue methods ──
1046
1047    #[test]
1048    fn test_column_value_as_text() {
1049        let text = ColumnValue::Text(Bytes::from_static(b"hello"));
1050        assert_eq!(text.as_text(), Some("hello"));
1051        assert!(!text.is_null());
1052
1053        let null = ColumnValue::Null;
1054        assert_eq!(null.as_text(), None);
1055        assert!(null.is_null());
1056
1057        let unchanged = ColumnValue::Unchanged;
1058        assert_eq!(unchanged.as_text(), None);
1059        assert!(!unchanged.is_null());
1060    }
1061
1062    // ── Update with key identity ──
1063
1064    #[test]
1065    fn test_decode_update_with_key_identity() {
1066        let data = MessageBuilder::new(b'U')
1067            .u32(16384)
1068            .u8(b'K') // key identity
1069            .i16(2) // old tuple retains the relation's published-column cardinality
1070            .text_col("42")
1071            .null_col() // non-key position is unavailable
1072            .u8(b'N') // new tuple
1073            .i16(2) // new: 2 cols
1074            .text_col("42")
1075            .text_col("Updated")
1076            .build();
1077
1078        let msg = decode_message(&data).unwrap();
1079        match msg {
1080            WalMessage::Update(upd) => {
1081                let Some(OldTuple::Key(old)) = upd.old_tuple else {
1082                    panic!("expected key old tuple");
1083                };
1084                assert_eq!(old.columns.len(), 2);
1085                assert!(old.columns[1].is_null());
1086                assert_eq!(upd.new_tuple.columns.len(), 2);
1087            }
1088            _ => panic!("expected Update"),
1089        }
1090    }
1091
1092    #[test]
1093    fn tuple_text_is_a_zero_copy_slice_of_the_wal_frame() {
1094        let frame = Bytes::from(
1095            MessageBuilder::new(b'I')
1096                .u32(16_384)
1097                .u8(b'N')
1098                .i16(1)
1099                .text_col("allocation-free")
1100                .build(),
1101        );
1102        let start = frame.as_ptr() as usize;
1103        let end = start + frame.len();
1104
1105        let message = super::decode_message(frame).unwrap();
1106        let WalMessage::Insert(insert) = message else {
1107            panic!("expected insert");
1108        };
1109        let ColumnValue::Text(value) = &insert.new_tuple.columns[0] else {
1110            panic!("expected text");
1111        };
1112        let value_ptr = value.as_ptr() as usize;
1113        assert!(value_ptr >= start && value_ptr + value.len() <= end);
1114        assert_eq!(
1115            insert.new_tuple.columns[0].as_text(),
1116            Some("allocation-free")
1117        );
1118    }
1119
1120    #[test]
1121    fn tuple_column_count_above_postgres_limit_is_rejected_before_values() {
1122        let data = MessageBuilder::new(b'I')
1123            .u32(16_384)
1124            .u8(b'N')
1125            .i16((MAX_POSTGRES_COLUMNS + 1) as i16)
1126            .build();
1127        let error = decode_message(&data).unwrap_err();
1128        assert!(error.to_string().contains("PostgreSQL maximum"), "{error}");
1129    }
1130
1131    #[test]
1132    fn maximum_postgres_tuple_column_count_is_accepted() {
1133        let mut data = MessageBuilder::new(b'I')
1134            .u32(16_384)
1135            .u8(b'N')
1136            .i16(MAX_POSTGRES_COLUMNS as i16)
1137            .build();
1138        data.extend(std::iter::repeat_n(b'n', MAX_POSTGRES_COLUMNS));
1139
1140        let message = decode_message(&data).unwrap();
1141        let WalMessage::Insert(insert) = message else {
1142            panic!("expected insert");
1143        };
1144        assert_eq!(insert.new_tuple.columns.len(), MAX_POSTGRES_COLUMNS);
1145    }
1146
1147    #[test]
1148    fn invalid_tuple_utf8_is_rejected() {
1149        let mut data = MessageBuilder::new(b'I')
1150            .u32(16_384)
1151            .u8(b'N')
1152            .i16(1)
1153            .build();
1154        data.push(b't');
1155        data.extend_from_slice(&1_i32.to_be_bytes());
1156        data.push(0xff);
1157
1158        assert!(matches!(
1159            decode_message(&data),
1160            Err(DecoderError::InvalidUtf8(_))
1161        ));
1162    }
1163
1164    #[test]
1165    fn truncate_count_is_validated_before_allocation() {
1166        let data = MessageBuilder::new(b'T').u32(u32::MAX).u8(0).build();
1167
1168        let error = decode_message(&data).unwrap_err();
1169        assert!(error.to_string().contains("requires"), "{error}");
1170    }
1171}
laminar_connectors/postgres/cdc/decoder.rs

laminar_connectors/postgres/cdc/
decoder.rs
