datum-cdc 0.10.3

use std::{
    sync::Arc,
    time::{Duration, Instant},
};

use pgwire_replication::{ReplicationConfig, TlsConfig};
use url::Url;

use crate::{CdcCheckpointStore, CdcError, CdcResult, PgLsn};

/// PostgreSQL connection settings for `datum-cdc`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PostgresCdcConfig {
    pub host: String,
    pub port: u16,
    pub user: String,
    pub password: String,
    pub database: String,
    pub tls: CdcTlsConfig,
    pub application_name: String,
}

impl PostgresCdcConfig {
    pub fn from_url(value: impl AsRef<str>) -> CdcResult<Self> {
        let url = Url::parse(value.as_ref())
            .map_err(|err| CdcError::Config(format!("invalid PostgreSQL URL: {err}")))?;
        if url.scheme() != "postgresql" && url.scheme() != "postgres" {
            return Err(CdcError::Config(format!(
                "unsupported PostgreSQL URL scheme {:?}",
                url.scheme()
            )));
        }
        let host = url
            .host_str()
            .ok_or_else(|| CdcError::Config("PostgreSQL URL is missing a host".into()))?
            .to_owned();
        let user = percent_decode(url.username());
        if user.is_empty() {
            return Err(CdcError::Config(
                "PostgreSQL URL is missing a username".into(),
            ));
        }
        let database = url.path().trim_start_matches('/').to_owned();
        if database.is_empty() {
            return Err(CdcError::Config(
                "PostgreSQL URL is missing a database name".into(),
            ));
        }
        let tls = url
            .query_pairs()
            .find_map(|(key, value)| (key == "sslmode").then(|| CdcTlsConfig::from_sslmode(&value)))
            .transpose()?
            .unwrap_or(CdcTlsConfig::Disable);
        Ok(Self {
            host,
            port: url.port().unwrap_or(5432),
            user,
            password: url.password().map(percent_decode).unwrap_or_default(),
            database,
            tls,
            application_name: "datum-cdc".to_owned(),
        })
    }

    #[must_use]
    pub fn replication_config(
        &self,
        slot: &str,
        publication: &str,
        start_lsn: PgLsn,
        status_interval: Duration,
        idle_wakeup_interval: Duration,
        buffer_events: usize,
    ) -> ReplicationConfig {
        ReplicationConfig {
            host: self.host.clone(),
            port: self.port,
            user: self.user.clone(),
            password: self.password.clone(),
            database: self.database.clone(),
            tls: self.tls.to_pgwire(),
            slot: slot.to_owned(),
            publication: publication.to_owned(),
            start_lsn: start_lsn.into(),
            stop_at_lsn: None,
            status_interval,
            idle_wakeup_interval,
            buffer_events,
        }
    }
}

fn percent_decode(value: &str) -> String {
    url::form_urlencoded::parse(value.as_bytes())
        .map(|(decoded, _)| decoded.into_owned())
        .next()
        .unwrap_or_else(|| value.to_owned())
}

/// TLS mode for the replication connection.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CdcTlsConfig {
    Disable,
    Require,
    VerifyCa(Option<std::path::PathBuf>),
    VerifyFull(Option<std::path::PathBuf>),
}

impl CdcTlsConfig {
    pub fn from_sslmode(value: &str) -> CdcResult<Self> {
        match value {
            "disable" => Ok(Self::Disable),
            "prefer" => Ok(Self::Disable),
            "require" => Ok(Self::Require),
            "verify-ca" => Ok(Self::VerifyCa(None)),
            "verify-full" => Ok(Self::VerifyFull(None)),
            other => Err(CdcError::Config(format!(
                "unsupported sslmode {other:?}; expected disable, prefer, require, verify-ca, or verify-full"
            ))),
        }
    }

    #[must_use]
    pub fn to_pgwire(&self) -> TlsConfig {
        match self {
            Self::Disable => TlsConfig::disabled(),
            Self::Require => TlsConfig::require(),
            Self::VerifyCa(path) => TlsConfig::verify_ca(path.clone()),
            Self::VerifyFull(path) => TlsConfig::verify_full(path.clone()),
        }
    }
}

/// Starting point for a materialized CDC source.
#[derive(Clone, Default)]
pub enum CdcStart {
    /// Load a durable checkpoint when configured; otherwise start from the
    /// slot's confirmed LSN.
    #[default]
    CheckpointOrSlot,
    /// Start from the slot's confirmed LSN.
    SlotConfirmed,
    /// Start from an explicit LSN.
    Lsn(PgLsn),
}

/// Feedback policy. The MVP intentionally exposes only checkpoint-gated
/// feedback because immediate feedback would violate at-least-once delivery.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum FeedbackMode {
    #[default]
    AfterCheckpoint,
}

/// Logical replication slot lifecycle policy.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum SlotLifecycle {
    #[default]
    Existing,
    CreateIfMissing,
    CreateOwned,
    Temporary,
}

impl SlotLifecycle {
    #[must_use]
    pub const fn may_create(self) -> bool {
        matches!(
            self,
            Self::CreateIfMissing | Self::CreateOwned | Self::Temporary
        )
    }

    #[must_use]
    pub const fn may_drop(self) -> bool {
        matches!(self, Self::CreateOwned | Self::Temporary)
    }

    #[must_use]
    pub const fn temporary(self) -> bool {
        matches!(self, Self::Temporary)
    }
}

#[derive(Clone)]
pub(crate) struct SourceSettings {
    pub(crate) postgres: PostgresCdcConfig,
    pub(crate) slot: String,
    pub(crate) publication: String,
    pub(crate) source: crate::SourceMetadata,
    pub(crate) start: CdcStart,
    pub(crate) feedback: FeedbackMode,
    pub(crate) slot_lifecycle: SlotLifecycle,
    pub(crate) buffer_capacity: usize,
    pub(crate) replication_buffer_events: usize,
    pub(crate) status_interval: Duration,
    pub(crate) idle_wakeup_interval: Duration,
    pub(crate) validate_publication: bool,
    pub(crate) checkpoint_store: Option<Arc<dyn CdcCheckpointStore>>,
    pub(crate) reconnect: Option<ReconnectSettings>,
}

impl SourceSettings {
    pub(crate) fn start_deadline(&self) -> Option<Instant> {
        self.reconnect
            .as_ref()
            .map(|settings| Instant::now() + settings.max_elapsed)
    }
}

/// Bounded reconnect policy used by the carrier task.
#[derive(Debug, Clone)]
pub struct ReconnectSettings {
    pub min_backoff: Duration,
    pub max_backoff: Duration,
    pub max_elapsed: Duration,
}

impl Default for ReconnectSettings {
    fn default() -> Self {
        Self {
            min_backoff: Duration::from_millis(200),
            max_backoff: Duration::from_secs(5),
            max_elapsed: Duration::from_secs(60),
        }
    }
}