datum-cdc 0.10.3

use serde::{Deserialize, Serialize};

use crate::PgLsn;

/// Metadata identifying the PostgreSQL replication source that produced an
/// event.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SourceMetadata {
    pub database: String,
    pub slot: String,
    pub publication: String,
}

/// PostgreSQL replica identity for a relation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ReplicaIdentity {
    Default,
    Nothing,
    Full,
    Index,
    Unknown(u8),
}

impl ReplicaIdentity {
    #[must_use]
    pub fn from_wire(value: u8) -> Self {
        match value {
            b'd' => Self::Default,
            b'n' => Self::Nothing,
            b'f' => Self::Full,
            b'i' => Self::Index,
            other => Self::Unknown(other),
        }
    }
}

/// Column metadata as sent by a pgoutput `Relation` message.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ColumnMetadata {
    pub name: String,
    pub type_oid: u32,
    pub type_modifier: i32,
    pub key: bool,
}

/// Relation metadata cached by relation OID.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RelationMetadata {
    pub oid: u32,
    pub schema: String,
    pub table: String,
    pub replica_identity: ReplicaIdentity,
    pub columns: Vec<ColumnMetadata>,
    pub revision: u64,
}

impl RelationMetadata {
    #[must_use]
    pub fn column_index(&self, name: &str) -> Option<usize> {
        self.columns.iter().position(|column| column.name == name)
    }
}

/// A text-mode tuple value from pgoutput.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum ColumnValue {
    Null,
    ToastUnchanged,
    Text(String),
    Binary(Vec<u8>),
}

impl ColumnValue {
    #[must_use]
    pub fn as_text(&self) -> Option<&str> {
        match self {
            Self::Text(value) => Some(value),
            Self::Null | Self::ToastUnchanged | Self::Binary(_) => None,
        }
    }
}

/// Row tuple values in relation-column order.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RowData {
    pub values: Vec<ColumnValue>,
}

impl RowData {
    #[must_use]
    pub fn get<'a>(
        &'a self,
        relation: &'a RelationMetadata,
        column: &str,
    ) -> Option<&'a ColumnValue> {
        relation
            .column_index(column)
            .and_then(|index| self.values.get(index))
    }

    #[must_use]
    pub fn get_text<'a>(&'a self, relation: &'a RelationMetadata, column: &str) -> Option<&'a str> {
        self.get(relation, column).and_then(ColumnValue::as_text)
    }
}

/// CDC operation kind.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ChangeOperation {
    Insert,
    Update,
    Delete,
    Truncate,
}

impl ChangeOperation {
    #[must_use]
    pub const fn debezium_letter(self) -> &'static str {
        match self {
            Self::Insert => "c",
            Self::Update => "u",
            Self::Delete => "d",
            Self::Truncate => "t",
        }
    }
}

/// Options attached to a pgoutput truncate message.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
pub struct TruncateOptions {
    pub cascade: bool,
    pub restart_identity: bool,
}

/// Stable offset for downstream checkpointing and de-duplication.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct CdcOffset {
    pub slot: String,
    pub tx_end_lsn: PgLsn,
    pub commit_lsn: PgLsn,
    pub xid: u32,
    pub event_index: u32,
    pub event_count: u32,
}

/// Transaction metadata attached after the commit boundary is known.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TransactionMeta {
    pub xid: u32,
    /// Microseconds since PostgreSQL's epoch (`2000-01-01 00:00:00 UTC`).
    pub commit_time_micros: i64,
    pub event_index: u32,
    pub event_count: u32,
}

/// One row/table change from PostgreSQL logical replication.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChangeEvent {
    pub source: SourceMetadata,
    pub schema: String,
    pub table: String,
    pub op: ChangeOperation,
    pub before: Option<RowData>,
    pub after: Option<RowData>,
    pub truncate: Option<TruncateOptions>,
    pub lsn: CdcOffset,
    pub tx: TransactionMeta,
    pub relation: RelationMetadata,
}