//! SQL Server to NATS bridge implementation.
//!
//! This module provides the outbox-style bridge that reliably publishes
//! SQL Server entity change events to NATS `JetStream` using the tiberius driver.
//!
//! # Architecture
//!
//! ```text
//! SQL Server (tb_entity_change_log)
//!     │
//!     ↓ cursor-based polling
//! MSSQLNatsBridge
//!     │
//!     ↓ publish + ACK
//! NATS JetStream
//! ```
//!
//! # Differences from PostgreSQL/MySQL Bridges
//!
//! - **Uses tiberius crate**: Not sqlx (sqlx doesn't support MSSQL)
//! - **UNIQUEIDENTIFIER type**: Native UUID support
//! - **DATETIME2 timestamps**: Higher precision than MySQL
//! - **MERGE for upserts**: T-SQL specific syntax
//! - **No LISTEN/NOTIFY**: Uses pure polling (same as MySQL)
//!
//! # Guarantees
//!
//! Same as other bridges:
//! - **Zero data loss**: `tb_entity_change_log` is the durable source of truth
//! - **At-least-once delivery**: Events may be published multiple times
//! - **Crash recovery**: Checkpoint-based resumption from last processed ID

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

use async_trait::async_trait;
#[cfg(feature = "mssql")]
use bb8::Pool;
#[cfg(feature = "mssql")]
use bb8_tiberius::ConnectionManager;
use chrono::{DateTime, Utc};
use serde_json::Value;
#[cfg(feature = "mssql")]
use tiberius::{Query, Row};
use tracing::{debug, error, info, warn};
use uuid::Uuid;

use super::CheckpointStore;
#[cfg(all(feature = "mssql", feature = "nats"))]
use super::{EventTransport, NatsTransport};
use crate::{
    error::{ObserverError, Result},
    event::EntityEvent,
};

/// Type alias for SQL Server connection pool.
#[cfg(feature = "mssql")]
pub type MSSQLPool = Pool<ConnectionManager>;

// ============================================================================
// MSSQLCheckpointStore Implementation
// ============================================================================

/// SQL Server-backed checkpoint store.
///
/// Stores checkpoints in `tb_transport_checkpoint` table with
/// MERGE semantics for crash-safe persistence.
#[derive(Clone)]
#[cfg(feature = "mssql")]
pub struct MSSQLCheckpointStore {
    pool: MSSQLPool,
}

#[cfg(feature = "mssql")]
impl MSSQLCheckpointStore {
    /// Create a new SQL Server checkpoint store.
    #[must_use]
    pub const fn new(pool: MSSQLPool) -> Self {
        Self { pool }
    }
}

#[cfg(feature = "mssql")]
// Reason: CheckpointStore is defined with #[async_trait]; all implementations must match
// its transformed method signatures to satisfy the trait contract
// async_trait: dyn-dispatch required; remove when RTN + Send is stable (RFC 3425)
#[async_trait]
impl CheckpointStore for MSSQLCheckpointStore {
    async fn get_checkpoint(&self, transport_name: &str) -> Result<Option<i64>> {
        let mut conn = self.pool.get().await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL pool get failed: {e}"),
        })?;

        let mut query =
            Query::new("SELECT last_pk FROM tb_transport_checkpoint WHERE transport_name = @P1");
        query.bind(transport_name);

        let stream = query.query(&mut *conn).await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL checkpoint query failed: {e}"),
        })?;

        let row = stream.into_row().await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL checkpoint fetch failed: {e}"),
        })?;

        match row {
            Some(r) => {
                let last_pk: i64 = r.get(0).ok_or_else(|| ObserverError::DatabaseError {
                    reason: "MSSQL checkpoint: missing last_pk column".to_string(),
                })?;
                Ok(Some(last_pk))
            },
            None => Ok(None),
        }
    }

    async fn save_checkpoint(&self, transport_name: &str, cursor: i64) -> Result<()> {
        let mut conn = self.pool.get().await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL pool get failed: {e}"),
        })?;

        // Use stored procedure for atomic upsert
        let mut query = Query::new("EXEC sp_upsert_checkpoint @P1, @P2");
        query.bind(transport_name);
        query.bind(cursor);

        query.execute(&mut *conn).await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL checkpoint save failed: {e}"),
        })?;

        Ok(())
    }

    async fn delete_checkpoint(&self, transport_name: &str) -> Result<()> {
        let mut conn = self.pool.get().await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL pool get failed: {e}"),
        })?;

        let mut query =
            Query::new("DELETE FROM tb_transport_checkpoint WHERE transport_name = @P1");
        query.bind(transport_name);

        query.execute(&mut *conn).await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL checkpoint delete failed: {e}"),
        })?;

        Ok(())
    }
}

// ============================================================================
// MSSQLChangeLogEntry (Row from tb_entity_change_log)
// ============================================================================

/// Row from SQL Server `tb_entity_change_log` table.
///
/// Represents a single entity change event stored in SQL Server.
#[derive(Debug, Clone)]
#[cfg(feature = "mssql")]
pub struct MSSQLChangeLogEntry {
    /// Primary key (used as cursor)
    pub pk_entity_change_log: i64,

    /// UUID identifier for the change
    pub id: Uuid,

    /// Customer organization ID (tenant)
    pub fk_customer_org: Option<i64>,

    /// Contact ID (user who made the change)
    pub fk_contact: Option<i64>,

    /// Entity type (e.g., "Order", "User")
    pub object_type: String,

    /// Entity ID (UUID of the changed entity)
    pub object_id: Uuid,

    /// Modification type: INSERT, UPDATE, DELETE
    pub modification_type: String,

    /// Change status (e.g., "pending", "processed")
    pub change_status: Option<String>,

    /// Entity data as JSON
    pub object_data: Option<Value>,

    /// Extra metadata as JSON
    pub extra_metadata: Option<Value>,

    /// When the change was created
    pub created_at: DateTime<Utc>,

    /// Public-facing tenant UUID (Change-Spine `tenant_id`, UNIQUEIDENTIFIER);
    /// distinct from `fk_customer_org`. `None` when the column is `NULL`.
    pub tenant_id: Option<Uuid>,

    /// Wall-clock duration of the originating mutation in milliseconds
    /// (Change-Spine perf column `duration_ms`); `None` when unstamped.
    pub duration_ms: Option<i32>,

    /// Monotonic Change-Spine sequence (`seq`) for durable ordering / dedup;
    /// `None` when the source row carried no sequence.
    pub seq: Option<i64>,

    /// The request's actor classification (#390 envelope column `actor_type`);
    /// `None` when unstamped. Recorded for fan-out, never an authz input.
    pub actor_type: Option<String>,

    /// For a delegated-agent request, the delegated human's public-facing UUID
    /// (#390 envelope column `acting_for`, UNIQUEIDENTIFIER); `None` otherwise.
    pub acting_for: Option<Uuid>,

    /// The producer's application schema version (#377 envelope column
    /// `schema_version`, NVARCHAR); `None` when unstamped.
    pub schema_version: Option<String>,

    /// When the change was published to NATS (None = not published)
    pub nats_published_at: Option<DateTime<Utc>>,

    /// NATS event ID (for deduplication)
    pub nats_event_id: Option<Uuid>,
}

#[cfg(feature = "mssql")]
impl MSSQLChangeLogEntry {
    /// Convert to `EntityEvent` for publishing.
    ///
    /// # Errors
    ///
    /// Returns [`ObserverError::InvalidConfig`] if `modification_type` is not one
    /// of `INSERT`, `UPDATE`, or `DELETE`.
    pub fn to_entity_event(&self) -> Result<EntityEvent> {
        use crate::event::EventKind;

        let event_type = match self.modification_type.to_uppercase().as_str() {
            "INSERT" => EventKind::Created,
            "UPDATE" => EventKind::Updated,
            "DELETE" => EventKind::Deleted,
            other => {
                return Err(ObserverError::InvalidConfig {
                    message: format!("Unknown modification type: {other}"),
                });
            },
        };

        let event_id = self.nats_event_id.unwrap_or_else(Uuid::new_v4);

        let mut event = EntityEvent::new(
            event_type,
            self.object_type.clone(),
            self.object_id,
            self.object_data.clone().unwrap_or(Value::Null),
        );

        // Override the auto-generated ID with the change log's ID
        event.id = event_id;

        // Set user ID from contact if available
        if let Some(contact_id) = self.fk_contact {
            event.user_id = Some(contact_id.to_string());
        }

        // Surface the full Change-Spine envelope (parity with the listener / PG
        // bridge). UUID columns project to their string form (mirrors tenant_id).
        event.tenant_id = self.tenant_id.map(|t| t.to_string());
        event.duration_ms = self.duration_ms;
        event.seq = self.seq;
        event.actor_type.clone_from(&self.actor_type);
        event.acting_for = self.acting_for.map(|u| u.to_string());
        event.schema_version.clone_from(&self.schema_version);

        Ok(event)
    }

    /// Parse a SQL Server row into a `MSSQLChangeLogEntry`.
    ///
    /// # Errors
    ///
    /// Returns [`ObserverError::DatabaseError`] if any required column is missing
    /// or cannot be parsed from the row.
    pub fn from_row(row: &Row) -> Result<Self> {
        // Helper to extract values with proper error handling
        let pk: i64 = row.get(0).ok_or_else(|| ObserverError::DatabaseError {
            reason: "Missing pk_entity_change_log".to_string(),
        })?;

        let id: Uuid = row.get(1).ok_or_else(|| ObserverError::DatabaseError {
            reason: "Missing id".to_string(),
        })?;

        let fk_customer_org: Option<i64> = row.get(2);
        let fk_contact: Option<i64> = row.get(3);

        let object_type: &str = row.get(4).ok_or_else(|| ObserverError::DatabaseError {
            reason: "Missing object_type".to_string(),
        })?;

        let object_id: Uuid = row.get(5).ok_or_else(|| ObserverError::DatabaseError {
            reason: "Missing object_id".to_string(),
        })?;

        let modification_type: &str = row.get(6).ok_or_else(|| ObserverError::DatabaseError {
            reason: "Missing modification_type".to_string(),
        })?;

        let change_status: Option<&str> = row.get(7);

        // JSON columns are NVARCHAR(MAX), parse as needed
        let object_data_str: Option<&str> = row.get(8);
        let object_data = object_data_str.map(serde_json::from_str).transpose().map_err(|e| {
            ObserverError::DatabaseError {
                reason: format!("Invalid object_data JSON: {e}"),
            }
        })?;

        let extra_metadata_str: Option<&str> = row.get(9);
        let extra_metadata =
            extra_metadata_str.map(serde_json::from_str).transpose().map_err(|e| {
                ObserverError::DatabaseError {
                    reason: format!("Invalid extra_metadata JSON: {e}"),
                }
            })?;

        // Tiberius returns NaiveDateTime, convert to DateTime<Utc>
        let created_at_naive: chrono::NaiveDateTime =
            row.get(10).ok_or_else(|| ObserverError::DatabaseError {
                reason: "Missing created_at".to_string(),
            })?;
        let created_at = DateTime::<Utc>::from_naive_utc_and_offset(created_at_naive, Utc);

        let nats_published_at_naive: Option<chrono::NaiveDateTime> = row.get(11);
        let nats_published_at =
            nats_published_at_naive.map(|n| DateTime::<Utc>::from_naive_utc_and_offset(n, Utc));

        let nats_event_id: Option<Uuid> = row.get(12);

        // Change-Spine envelope columns, appended to the SELECT (indices 13-18)
        // so the existing positional indices above stay stable.
        let tenant_id: Option<Uuid> = row.get(13);
        let duration_ms: Option<i32> = row.get(14);
        let seq: Option<i64> = row.get(15);
        let actor_type: Option<&str> = row.get(16);
        let acting_for: Option<Uuid> = row.get(17);
        let schema_version: Option<&str> = row.get(18);

        Ok(Self {
            pk_entity_change_log: pk,
            id,
            fk_customer_org,
            fk_contact,
            object_type: object_type.to_string(),
            object_id,
            modification_type: modification_type.to_string(),
            change_status: change_status.map(ToString::to_string),
            object_data,
            extra_metadata,
            created_at,
            tenant_id,
            duration_ms,
            seq,
            actor_type: actor_type.map(ToString::to_string),
            acting_for,
            schema_version: schema_version.map(ToString::to_string),
            nats_published_at,
            nats_event_id,
        })
    }
}

// ============================================================================
// MSSQLBridgeConfig
// ============================================================================

/// SQL Server bridge configuration.
#[derive(Debug, Clone)]
#[cfg(feature = "mssql")]
pub struct MSSQLBridgeConfig {
    /// Transport name for checkpoint storage (e.g., `mssql_to_nats`)
    pub transport_name: String,

    /// Batch size for fetching entries
    pub batch_size: usize,

    /// Poll interval (seconds) - SQL Server has no LISTEN/NOTIFY
    pub poll_interval_secs: u64,
}

#[cfg(feature = "mssql")]
impl Default for MSSQLBridgeConfig {
    fn default() -> Self {
        Self {
            transport_name:     "mssql_to_nats".to_string(),
            batch_size:         100,
            poll_interval_secs: 1,
        }
    }
}

// ============================================================================
// MSSQLNatsBridge
// ============================================================================

/// SQL Server to NATS bridge.
///
/// Reliably publishes entity change events from SQL Server's
/// `tb_entity_change_log` to NATS `JetStream` using cursor-based polling.
///
/// # Key Differences from PostgreSQL Bridge
///
/// - **Uses tiberius crate**: Native SQL Server driver (not sqlx)
/// - **UNIQUEIDENTIFIER type**: Native UUID support in SQL Server
/// - **No LISTEN/NOTIFY**: Uses pure polling with configurable interval
///
/// # Design Properties
///
/// Same as PostgreSQL/MySQL bridges:
/// 1. CURSOR-based polling ensures no missed events
/// 2. Checkpoint persistence enables crash recovery
/// 3. Conditional `mark_published` prevents races
/// 4. At-least-once delivery (consumers must be idempotent)
#[cfg(all(feature = "mssql", feature = "nats"))]
pub struct MSSQLNatsBridge {
    pool:             MSSQLPool,
    nats_transport:   Arc<NatsTransport>,
    checkpoint_store: Arc<dyn CheckpointStore>,
    config:           MSSQLBridgeConfig,
}

/// Cursor-paged fetch for the MSSQL change log. `@P1` is the row cap (the
/// configured `batch_size`, bound — never a hardcoded literal, M-mssql-batch);
/// `@P2` is the exclusive cursor lower bound.
#[cfg(all(feature = "mssql", feature = "nats"))]
const FETCH_BATCH_SQL: &str = r"
            SELECT TOP (@P1)
                pk_entity_change_log, id, fk_customer_org, fk_contact,
                object_type, object_id, modification_type, change_status,
                object_data, extra_metadata, created_at,
                nats_published_at, nats_event_id,
                tenant_id, duration_ms, seq, actor_type, acting_for,
                schema_version
            FROM tb_entity_change_log
            WHERE pk_entity_change_log > @P2
            ORDER BY pk_entity_change_log ASC
            ";

#[cfg(all(feature = "mssql", feature = "nats"))]
impl MSSQLNatsBridge {
    /// Create a new MSSQL-to-NATS bridge.
    pub fn new(
        pool: MSSQLPool,
        nats_transport: Arc<NatsTransport>,
        checkpoint_store: Arc<dyn CheckpointStore>,
        config: MSSQLBridgeConfig,
    ) -> Self {
        Self {
            pool,
            nats_transport,
            checkpoint_store,
            config,
        }
    }

    /// Create with default configuration.
    pub fn with_defaults(
        pool: MSSQLPool,
        nats_transport: Arc<NatsTransport>,
        checkpoint_store: Arc<dyn CheckpointStore>,
    ) -> Self {
        Self::new(pool, nats_transport, checkpoint_store, MSSQLBridgeConfig::default())
    }

    /// Load last processed cursor from checkpoint store.
    async fn load_cursor(&self) -> Result<i64> {
        let cursor = self
            .checkpoint_store
            .get_checkpoint(&self.config.transport_name)
            .await?
            .unwrap_or(0);

        Ok(cursor)
    }

    /// Save cursor checkpoint.
    async fn save_cursor(&self, cursor: i64) -> Result<()> {
        self.checkpoint_store.save_checkpoint(&self.config.transport_name, cursor).await
    }

    /// Fetch batch from cursor.
    async fn fetch_batch_from_cursor(&self, cursor: i64) -> Result<Vec<MSSQLChangeLogEntry>> {
        let mut conn = self.pool.get().await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL pool get failed: {e}"),
        })?;

        #[allow(clippy::cast_possible_wrap)]
        // Reason: value is non-negative; wrap cannot occur in practice
        let batch_size = self.config.batch_size as i64;

        // SQL Server accepts a parameter in `TOP (expression)`, so the configured
        // batch size is bound rather than hardcoded (M-mssql-batch). @P1 is the
        // row cap; @P2 is the cursor.
        let mut query = Query::new(FETCH_BATCH_SQL);
        query.bind(batch_size);
        query.bind(cursor);

        let stream = query.query(&mut *conn).await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL fetch batch query failed: {e}"),
        })?;

        let rows = stream.into_first_result().await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL fetch batch failed: {e}"),
        })?;

        let mut entries = Vec::with_capacity(rows.len());
        for row in rows {
            entries.push(MSSQLChangeLogEntry::from_row(&row)?);
        }

        Ok(entries)
    }

    /// Mark event as published (idempotent, safe against races).
    async fn mark_published(&self, pk_change_log: i64, event_id: Uuid) -> Result<bool> {
        let mut conn = self.pool.get().await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL pool get failed: {e}"),
        })?;

        let mut query = Query::new(
            r"
            UPDATE tb_entity_change_log
            SET nats_published_at = GETUTCDATE(),
                nats_event_id = @P1
            WHERE pk_entity_change_log = @P2
              AND nats_published_at IS NULL
            ",
        );
        query.bind(event_id);
        query.bind(pk_change_log);

        let result = query.execute(&mut *conn).await.map_err(|e| ObserverError::DatabaseError {
            reason: format!("MSSQL mark_published failed: {e}"),
        })?;

        // rows_affected() returns a slice of counts per statement
        let total_affected: u64 = result.rows_affected().iter().sum();
        Ok(total_affected == 1)
    }

    /// Main bridge loop.
    ///
    /// SQL Server has no LISTEN/NOTIFY, so this uses
    /// pure polling with configurable interval.
    ///
    /// # Errors
    ///
    /// Propagates SQL Server query errors or NATS publish errors encountered
    /// while fetching or forwarding change log entries.
    pub async fn run(&self) -> Result<()> {
        info!("Starting MSSQL → NATS bridge: {}", self.config.transport_name);

        let mut cursor = self.load_cursor().await?;
        info!("Bridge starting from cursor: {cursor}");

        loop {
            let entries = self.fetch_batch_from_cursor(cursor).await?;

            if entries.is_empty() {
                // No entries, wait for poll interval
                debug!("No new entries, sleeping for {}s", self.config.poll_interval_secs);
                tokio::time::sleep(Duration::from_secs(self.config.poll_interval_secs)).await;
            } else {
                debug!("Processing {} entries from cursor {}", entries.len(), cursor);

                for entry in entries {
                    // Check if already published (skip if so)
                    if entry.nats_published_at.is_some() {
                        debug!(
                            "Entry pk={} already published, skipping (advancing cursor)",
                            entry.pk_entity_change_log
                        );
                        cursor = entry.pk_entity_change_log;
                        continue;
                    }

                    // Convert to EntityEvent
                    let event = match entry.to_entity_event() {
                        Ok(e) => e,
                        Err(e) => {
                            warn!(
                                "Failed to convert entry pk={} to event: {}. Skipping.",
                                entry.pk_entity_change_log, e
                            );
                            cursor = entry.pk_entity_change_log;
                            continue;
                        },
                    };

                    // Publish to NATS JetStream
                    match self.nats_transport.publish(event.clone()).await {
                        Ok(()) => {
                            let was_first =
                                self.mark_published(entry.pk_entity_change_log, event.id).await?;

                            if was_first {
                                debug!(
                                    "Published event {} (cursor: {})",
                                    event.id, entry.pk_entity_change_log
                                );
                            } else {
                                debug!(
                                    "Event {} already published by another bridge, safe skip",
                                    event.id
                                );
                            }

                            cursor = entry.pk_entity_change_log;
                        },
                        Err(e) => {
                            error!(
                                "Failed to publish event {} to NATS: {}. Will retry.",
                                event.id, e
                            );
                            break;
                        },
                    }
                }

                self.save_cursor(cursor).await?;
            }
        }
    }

    /// Run with graceful shutdown support.
    ///
    /// # Errors
    ///
    /// Returns an error if the MSSQL polling connection fails, cursor
    /// loading/saving fails, or NATS publishing encounters an unrecoverable
    /// error.
    pub async fn run_with_shutdown(
        &self,
        mut shutdown: tokio::sync::broadcast::Receiver<()>,
    ) -> Result<()> {
        info!(
            "Starting MSSQL → NATS bridge with shutdown support: {}",
            self.config.transport_name
        );

        let mut cursor = self.load_cursor().await?;
        info!("Bridge starting from cursor: {cursor}");

        loop {
            // Check for shutdown signal
            if shutdown.try_recv().is_ok() {
                info!("Shutdown signal received, stopping bridge");
                self.save_cursor(cursor).await?;
                return Ok(());
            }

            let entries = self.fetch_batch_from_cursor(cursor).await?;

            if entries.is_empty() {
                // Poll interval with shutdown check
                let poll_duration = Duration::from_secs(self.config.poll_interval_secs);

                tokio::select! {
                    _ = shutdown.recv() => {
                        info!("Shutdown signal received during wait, stopping bridge");
                        self.save_cursor(cursor).await?;
                        return Ok(());
                    }
                    () = tokio::time::sleep(poll_duration) => {
                        debug!("Poll interval timeout");
                    }
                }
            } else {
                for entry in entries {
                    if entry.nats_published_at.is_some() {
                        cursor = entry.pk_entity_change_log;
                        continue;
                    }

                    let event = match entry.to_entity_event() {
                        Ok(e) => e,
                        Err(e) => {
                            warn!(
                                "Failed to convert entry pk={} to event: {}. Skipping.",
                                entry.pk_entity_change_log, e
                            );
                            cursor = entry.pk_entity_change_log;
                            continue;
                        },
                    };

                    match self.nats_transport.publish(event.clone()).await {
                        Ok(()) => {
                            let _ =
                                self.mark_published(entry.pk_entity_change_log, event.id).await?;
                            cursor = entry.pk_entity_change_log;
                        },
                        Err(e) => {
                            error!("Failed to publish event {}: {}. Retrying.", event.id, e);
                            break;
                        },
                    }
                }

                self.save_cursor(cursor).await?;
            }
        }
    }
}

// ============================================================================
// Connection Pool Helper
// ============================================================================

/// Create a SQL Server connection pool.
///
/// # Arguments
///
/// * `connection_string` - ADO.NET style connection string
///
/// # Errors
///
/// Returns [`ObserverError::TransportConnectionFailed`] if the connection string
/// is invalid or the pool cannot be built.
///
/// # Example
///
/// ```ignore
/// let pool = create_mssql_pool(
///     "Server=localhost;Database=mydb;User Id=sa;Password=secret;TrustServerCertificate=true"
/// ).await?;
/// ```
#[cfg(feature = "mssql")]
pub async fn create_mssql_pool(connection_string: &str) -> Result<MSSQLPool> {
    use tiberius::Config;

    let config = Config::from_ado_string(connection_string).map_err(|e| {
        ObserverError::TransportConnectionFailed {
            reason: format!("Invalid SQL Server connection string: {e}"),
        }
    })?;

    let manager = ConnectionManager::new(config);

    Pool::builder().max_size(10).build(manager).await.map_err(|e| {
        ObserverError::TransportConnectionFailed {
            reason: format!("Failed to create SQL Server connection pool: {e}"),
        }
    })
}

#[cfg(all(test, feature = "mssql", feature = "nats"))]
mod tests;
