datum-cdc 0.9.2

use std::{fmt, str::FromStr};

use serde::{Deserialize, Serialize};

use crate::{CdcError, CdcResult};

/// PostgreSQL log sequence number.
///
/// PostgreSQL displays LSNs as `X/Y`, where `X` is the high 32 bits and `Y` is
/// the low 32 bits in hexadecimal.
#[derive(
    Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
)]
pub struct PgLsn(u64);

impl PgLsn {
    /// Zero LSN. Passing this to PostgreSQL asks the slot to start from its
    /// confirmed position.
    pub const ZERO: Self = Self(0);

    #[must_use]
    pub const fn from_u64(value: u64) -> Self {
        Self(value)
    }

    #[must_use]
    pub const fn as_u64(self) -> u64 {
        self.0
    }

    #[must_use]
    pub const fn is_zero(self) -> bool {
        self.0 == 0
    }

    pub fn parse(value: &str) -> CdcResult<Self> {
        let Some((high, low)) = value.split_once('/') else {
            return Err(CdcError::Config(format!(
                "invalid LSN {value:?}: missing '/'"
            )));
        };
        let high = u64::from_str_radix(high, 16)
            .map_err(|err| CdcError::Config(format!("invalid LSN high half {high:?}: {err}")))?;
        let low = u64::from_str_radix(low, 16)
            .map_err(|err| CdcError::Config(format!("invalid LSN low half {low:?}: {err}")))?;
        Ok(Self((high << 32) | low))
    }
}

impl fmt::Display for PgLsn {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let high = (self.0 >> 32) as u32;
        let low = (self.0 & 0xffff_ffff) as u32;
        write!(f, "{high:X}/{low:X}")
    }
}

impl From<u64> for PgLsn {
    fn from(value: u64) -> Self {
        Self(value)
    }
}

impl From<PgLsn> for u64 {
    fn from(value: PgLsn) -> Self {
        value.as_u64()
    }
}

impl From<pgwire_replication::Lsn> for PgLsn {
    fn from(value: pgwire_replication::Lsn) -> Self {
        Self(value.as_u64())
    }
}

impl From<PgLsn> for pgwire_replication::Lsn {
    fn from(value: PgLsn) -> Self {
        pgwire_replication::Lsn::from_u64(value.as_u64())
    }
}

impl FromStr for PgLsn {
    type Err = CdcError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        Self::parse(value)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn formats_lsn_like_postgres() {
        let lsn = PgLsn::from_u64(0x16_0000_0000 + 0xb374_d848);
        assert_eq!(lsn.to_string(), "16/B374D848");
        assert_eq!(PgLsn::parse("16/B374D848").unwrap(), lsn);
    }
}