evidentsource-client 1.0.0-rc1

//! Connection to an EvidentSource database with live metadata updates.

use std::sync::Arc;
use std::time::Duration;

use chrono::{DateTime, Utc};
use futures::stream::{self, StreamExt};
use futures::Stream;
use nonempty::NonEmpty;
use tokio::sync::{watch, Mutex};
use tokio::task::JoinHandle;

use evidentsource_core::domain::{
    AppendCondition, CommandRequest, DatabaseError, DatabaseName, ProspectiveEvent, Revision,
    StateChangeError, StateChangeName, StateChangeVersion, Transaction, TransactionError,
    TransactionSummary,
};
use evidentsource_core::{DatabaseConnection, DatabaseIdentity, DatabaseProvider};

use crate::com::evidentsource as proto;
use crate::conversions::{timestamp_to_datetime, ConversionError};
use crate::database::DatabaseAtRevisionImpl;
use crate::io::cloudevents::v1 as proto_ce;
use crate::EvidentSourceClient;

/// Metadata about the database state, updated via subscription.
#[derive(Clone, Debug)]
pub struct DatabaseMetadata {
    /// Database name.
    pub name: DatabaseName,
    /// When the database was created.
    pub created_at: DateTime<Utc>,
    /// Current revision number.
    pub revision: u64,
    /// When the current revision was committed.
    pub revision_timestamp: DateTime<Utc>,
}

impl DatabaseMetadata {
    /// Create metadata from a proto Database message.
    fn from_proto(proto: proto::Database) -> Result<Self, ConversionError> {
        let name = DatabaseName::new(&proto.name)?;
        let created_at = proto
            .created_at
            .ok_or_else(|| ConversionError::missing_field("Database", "created_at"))
            .and_then(timestamp_to_datetime)?;
        let revision_timestamp = proto
            .revision_timestamp
            .ok_or_else(|| ConversionError::missing_field("Database", "revision_timestamp"))
            .and_then(timestamp_to_datetime)?;

        Ok(Self {
            name,
            created_at,
            revision: proto.revision,
            revision_timestamp,
        })
    }
}

/// Shared state for the connection, held behind Arc for cloning.
struct ConnectionInner {
    /// The low-level gRPC client.
    client: EvidentSourceClient,
    /// Database name (stored explicitly for guaranteed availability).
    database_name: DatabaseName,
    /// Sender side of the watch channel (used by subscription task).
    metadata_tx: watch::Sender<DatabaseMetadata>,
    /// Handle to the subscription task for lifecycle management.
    subscription_handle: Mutex<Option<JoinHandle<()>>>,
    /// Shutdown signal sender.
    shutdown_tx: watch::Sender<bool>,
}

impl Drop for ConnectionInner {
    fn drop(&mut self) {
        // Signal shutdown to subscription task
        let _ = self.shutdown_tx.send(true);
    }
}

/// A connection to an EvidentSource database with live metadata updates.
///
/// The connection maintains a background subscription to receive database
/// updates whenever transactions are committed. This enables efficient access
/// to the latest database state without polling.
///
/// # Example
///
/// ```ignore
/// let es = EvidentSource::connect_to_server("http://localhost:50051").await?;
/// let conn = es.connect(&DatabaseName::new("my-db")?).await?;
///
/// // Get latest database state
/// let db = conn.latest_database().await?;
/// println!("Revision: {}", db.revision());
/// ```
#[derive(Clone)]
pub struct Connection {
    /// Shared inner state (clones share this).
    inner: Arc<ConnectionInner>,
    /// Receiver side of metadata watch channel.
    /// Each clone gets its own receiver but they all see the same updates.
    metadata_rx: watch::Receiver<DatabaseMetadata>,
    /// Receiver for shutdown signal.
    shutdown_rx: watch::Receiver<bool>,
}

impl Connection {
    /// Create a new connection to a database.
    ///
    /// This fetches the initial database state and starts a background
    /// subscription to receive updates.
    pub async fn new(
        client: EvidentSourceClient,
        database_name: DatabaseName,
    ) -> Result<Self, DatabaseError> {
        let mut client_clone = client.clone();

        // Fetch initial database state
        let initial_db = client_clone
            .fetch_latest_database(database_name.to_string())
            .await
            .map_err(|_| DatabaseError::NotFound(database_name.to_string()))?;

        let initial_metadata = DatabaseMetadata::from_proto(initial_db)
            .map_err(|e| DatabaseError::ServerError(format!("failed to parse metadata: {}", e)))?;

        // Create watch channel with initial state
        let (metadata_tx, metadata_rx) = watch::channel(initial_metadata);

        // Create shutdown signal
        let (shutdown_tx, shutdown_rx) = watch::channel(false);

        let inner = Arc::new(ConnectionInner {
            client,
            database_name: database_name.clone(),
            metadata_tx,
            subscription_handle: Mutex::new(None),
            shutdown_tx,
        });

        let connection = Connection {
            inner,
            metadata_rx,
            shutdown_rx,
        };

        // Spawn the subscription task
        connection.spawn_subscription_task().await;

        Ok(connection)
    }

    /// Spawn (or respawn) the subscription task.
    async fn spawn_subscription_task(&self) {
        let inner = self.inner.clone();
        let db_name = self.inner.database_name.to_string();
        let mut shutdown_rx = self.shutdown_rx.clone();

        let handle = tokio::spawn(async move {
            let mut backoff = Duration::from_millis(100);
            let max_backoff = Duration::from_secs(30);

            loop {
                // Try to establish subscription
                let stream_result = {
                    let mut client = inner.client.clone();
                    client.subscribe_database_updates(db_name.clone()).await
                };

                match stream_result {
                    Ok(mut stream) => {
                        backoff = Duration::from_millis(100); // Reset backoff on success

                        // Process updates until error or shutdown
                        loop {
                            tokio::select! {
                                // Check for shutdown signal
                                _ = shutdown_rx.changed() => {
                                    if *shutdown_rx.borrow() {
                                        tracing::debug!("Subscription task shutting down");
                                        return;
                                    }
                                }

                                // Process next update
                                update = stream.next() => {
                                    match update {
                                        Some(Ok(reply)) => {
                                            if let Some(db) = reply.database {
                                                match DatabaseMetadata::from_proto(db) {
                                                    Ok(metadata) => {
                                                        // Send update - ignore error if all receivers dropped
                                                        let _ = inner.metadata_tx.send(metadata);
                                                    }
                                                    Err(e) => {
                                                        tracing::warn!("Failed to parse metadata: {}", e);
                                                    }
                                                }
                                            }
                                        }
                                        Some(Err(status)) => {
                                            tracing::warn!("Stream error: {}", status);
                                            break; // Will reconnect
                                        }
                                        None => {
                                            tracing::debug!("Stream ended");
                                            break; // Will reconnect
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        tracing::warn!("Failed to subscribe: {}", e);
                    }
                }

                // Check shutdown before reconnecting
                if *shutdown_rx.borrow() {
                    return;
                }

                // Exponential backoff before retry
                tracing::debug!("Reconnecting in {:?}", backoff);
                tokio::time::sleep(backoff).await;
                backoff = (backoff * 2).min(max_backoff);
            }
        });

        // Store the handle
        let mut guard = self.inner.subscription_handle.lock().await;
        *guard = Some(handle);
    }

    /// Explicitly close the connection and wait for cleanup.
    pub async fn close(self) -> Result<(), DatabaseError> {
        // Signal shutdown
        let _ = self.inner.shutdown_tx.send(true);

        // Wait for subscription task to complete
        if let Some(handle) = self.inner.subscription_handle.lock().await.take() {
            // Give it a reasonable timeout
            match tokio::time::timeout(Duration::from_secs(5), handle).await {
                Ok(Ok(())) => Ok(()),
                Ok(Err(e)) => Err(DatabaseError::ServerError(format!(
                    "subscription task panicked: {}",
                    e
                ))),
                Err(_) => Err(DatabaseError::Timeout),
            }
        } else {
            Ok(())
        }
    }

    /// Get the current cached metadata.
    fn current_metadata(&self) -> DatabaseMetadata {
        self.metadata_rx.borrow().clone()
    }

    /// Create a database snapshot from the current metadata.
    fn snapshot_from_metadata(&self, metadata: &DatabaseMetadata) -> DatabaseAtRevisionImpl {
        DatabaseAtRevisionImpl::at_revision_with_metadata(
            self.inner.client.clone(),
            metadata.name.clone(),
            metadata.created_at,
            metadata.revision,
            metadata.revision_timestamp,
        )
    }
}

impl std::fmt::Debug for Connection {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Connection")
            .field("database_name", &self.inner.database_name)
            .field("current_revision", &self.metadata_rx.borrow().revision)
            .finish()
    }
}

impl DatabaseIdentity for Connection {
    fn name(&self) -> &DatabaseName {
        &self.inner.database_name
    }

    fn created_at(&self) -> DateTime<Utc> {
        self.metadata_rx.borrow().created_at
    }
}

impl DatabaseProvider for Connection {
    type AtRevision = DatabaseAtRevisionImpl;

    fn local_database(&self) -> Self::AtRevision {
        let metadata = self.current_metadata();
        self.snapshot_from_metadata(&metadata)
    }

    fn latest_database(
        &self,
    ) -> impl std::future::Future<Output = Result<Self::AtRevision, DatabaseError>> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();

        async move {
            let proto_db = client
                .fetch_latest_database(database_name.clone())
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &database_name)
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            DatabaseAtRevisionImpl::new(client, proto_db)
                .map_err(|e| DatabaseError::ServerError(format!("failed to parse database: {}", e)))
        }
    }

    fn database_at_revision(
        &self,
        revision: u64,
    ) -> impl std::future::Future<Output = Result<Self::AtRevision, DatabaseError>> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();

        async move {
            let proto_db = client
                .await_database(database_name.clone(), revision)
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &database_name)
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            DatabaseAtRevisionImpl::new(client, proto_db)
                .map_err(|e| DatabaseError::ServerError(format!("failed to parse database: {}", e)))
        }
    }

    fn database_at_timestamp(
        &self,
        revision_timestamp: DateTime<Utc>,
    ) -> impl std::future::Future<Output = Result<Self::AtRevision, DatabaseError>> {
        use crate::conversions::datetime_to_timestamp;

        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();
        let ts = datetime_to_timestamp(revision_timestamp);

        async move {
            let proto_db = client
                .database_effective_at_timestamp(database_name.clone(), ts)
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &database_name)
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            DatabaseAtRevisionImpl::new(client, proto_db)
                .map_err(|e| DatabaseError::ServerError(format!("failed to parse database: {}", e)))
        }
    }
}

impl DatabaseConnection for Connection {
    fn transact(
        &self,
        events: NonEmpty<ProspectiveEvent>,
        conditions: Vec<AppendCondition>,
    ) -> impl std::future::Future<Output = Result<Self::AtRevision, DatabaseError>> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();

        // Convert domain types to proto
        let proto_events: Vec<proto_ce::CloudEvent> =
            events.into_iter().map(|pe| pe.into()).collect();
        let proto_conditions: Vec<proto::AppendCondition> =
            conditions.into_iter().map(|c| c.into()).collect();
        let transaction_id = uuid::Uuid::new_v4().to_string();

        async move {
            let result = client
                .transact(
                    transaction_id,
                    database_name.clone(),
                    proto_events,
                    proto_conditions,
                )
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &database_name)
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            // Get the new revision from the result
            let new_revision = result
                .transaction_summary
                .map(|s| s.revision)
                .unwrap_or_default();

            // Await the database at the new revision
            let proto_db = client
                .await_database(database_name.clone(), new_revision)
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &database_name)
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            DatabaseAtRevisionImpl::new(client, proto_db)
                .map_err(|e| DatabaseError::ServerError(format!("failed to parse database: {}", e)))
        }
    }

    fn execute_state_change(
        &self,
        name: &StateChangeName,
        version: StateChangeVersion,
        request: CommandRequest,
    ) -> impl std::future::Future<Output = Result<Self::AtRevision, StateChangeError>> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();
        let state_change_name = name.to_string();
        let current_revision = self.current_metadata().revision;

        // Convert CommandRequest to proto
        let proto_request = proto::CommandRequest {
            headers: request
                .headers
                .into_iter()
                .map(|(k, v)| proto::Header { key: k, value: v })
                .collect(),
            body: request.body,
            content_type: request.content_type,
            content_schema: request.content_schema,
        };

        async move {
            let result = client
                .execute_state_change(
                    database_name.clone(),
                    state_change_name.clone(),
                    version,
                    Some(current_revision),
                    proto_request,
                    None,
                )
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_state_change_error(
                            status,
                            &state_change_name,
                            version,
                        )
                    }
                    _ => StateChangeError::ServerError(e.to_string()),
                })?;

            // Get the new revision from the result
            let new_revision = result
                .transaction_summary
                .map(|s| s.revision)
                .unwrap_or_default();

            // Await the database at the new revision
            let proto_db = client
                .await_database(database_name.clone(), new_revision)
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => StateChangeError::Database(
                        crate::status_mapping::to_database_error(status, &database_name),
                    ),
                    _ => StateChangeError::Database(DatabaseError::ServerError(e.to_string())),
                })?;

            DatabaseAtRevisionImpl::new(client, proto_db).map_err(|e| {
                StateChangeError::Database(DatabaseError::ServerError(format!(
                    "failed to parse database: {}",
                    e
                )))
            })
        }
    }

    fn log(&self) -> impl Stream<Item = TransactionSummary> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();

        stream::once(async move {
            let result = client.scan_database_log(database_name, 0, false).await;

            match result {
                Ok(response_stream) => response_stream
                    .filter_map(|result| async move {
                        match result {
                            Ok(reply) => {
                                if let Some(proto::database_log_reply::Transaction::Summary(
                                    summary,
                                )) = reply.transaction
                                {
                                    Some(TransactionSummary::from(summary))
                                } else {
                                    None
                                }
                            }
                            Err(_) => None,
                        }
                    })
                    .boxed(),
                Err(_) => stream::empty().boxed(),
            }
        })
        .flatten()
    }

    fn log_detail(&self) -> impl Stream<Item = Transaction> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();

        stream::once(async move {
            let result = client.scan_database_log(database_name, 0, true).await;

            match result {
                Ok(response_stream) => response_stream
                    .filter_map(|result| async move {
                        match result {
                            Ok(reply) => {
                                if let Some(proto::database_log_reply::Transaction::Detail(txn)) =
                                    reply.transaction
                                {
                                    Transaction::try_from(txn).ok()
                                } else {
                                    None
                                }
                            }
                            Err(_) => None,
                        }
                    })
                    .boxed(),
                Err(_) => stream::empty().boxed(),
            }
        })
        .flatten()
    }

    fn transact_with_id(
        &self,
        transaction_id: &str,
        events: NonEmpty<ProspectiveEvent>,
        conditions: Vec<AppendCondition>,
    ) -> impl std::future::Future<Output = Result<Self::AtRevision, DatabaseError>> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();
        let transaction_id = transaction_id.to_string();

        // Convert domain types to proto
        let proto_events: Vec<proto_ce::CloudEvent> =
            events.into_iter().map(|pe| pe.into()).collect();
        let proto_conditions: Vec<proto::AppendCondition> =
            conditions.into_iter().map(|c| c.into()).collect();

        async move {
            let result = client
                .transact(
                    transaction_id,
                    database_name.clone(),
                    proto_events,
                    proto_conditions,
                )
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &database_name)
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            // Get the new revision from the result
            let new_revision = result
                .transaction_summary
                .map(|s| s.revision)
                .unwrap_or_default();

            // Await the database at the new revision
            let proto_db = client
                .await_database(database_name.clone(), new_revision)
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &database_name)
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            DatabaseAtRevisionImpl::new(client, proto_db)
                .map_err(|e| DatabaseError::ServerError(format!("failed to parse database: {}", e)))
        }
    }

    fn log_from(&self, from_revision: Revision) -> impl Stream<Item = TransactionSummary> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();

        stream::once(async move {
            let result = client
                .scan_database_log(database_name, from_revision, false)
                .await;

            match result {
                Ok(response_stream) => response_stream
                    .filter_map(|result| async move {
                        match result {
                            Ok(reply) => {
                                if let Some(proto::database_log_reply::Transaction::Summary(
                                    summary,
                                )) = reply.transaction
                                {
                                    Some(TransactionSummary::from(summary))
                                } else {
                                    None
                                }
                            }
                            Err(_) => None,
                        }
                    })
                    .boxed(),
                Err(_) => stream::empty().boxed(),
            }
        })
        .flatten()
    }

    fn log_detail_from(&self, from_revision: Revision) -> impl Stream<Item = Transaction> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();

        stream::once(async move {
            let result = client
                .scan_database_log(database_name, from_revision, true)
                .await;

            match result {
                Ok(response_stream) => response_stream
                    .filter_map(|result| async move {
                        match result {
                            Ok(reply) => {
                                if let Some(proto::database_log_reply::Transaction::Detail(txn)) =
                                    reply.transaction
                                {
                                    Transaction::try_from(txn).ok()
                                } else {
                                    None
                                }
                            }
                            Err(_) => None,
                        }
                    })
                    .boxed(),
                Err(_) => stream::empty().boxed(),
            }
        })
        .flatten()
    }
}

// =============================================================================
// Connection Additional Methods
// =============================================================================

impl Connection {
    /// List all state change definitions registered with this database.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use futures::StreamExt;
    ///
    /// let mut state_changes = conn.list_state_changes();
    /// while let Some(sc) = state_changes.next().await {
    ///     println!("State change: {} v{}", sc.name, sc.version);
    /// }
    /// ```
    pub fn list_state_changes(
        &self,
    ) -> impl Stream<Item = evidentsource_core::domain::StateChangeDefinitionSummary> {
        use evidentsource_core::domain::{StateChangeDefinitionSummary, StateChangeName};

        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();

        stream::once(async move {
            let result = client.list_state_changes(database_name).await;

            match result {
                Ok(response_stream) => response_stream
                    .filter_map(|result| async move {
                        match result {
                            Ok(reply) => StateChangeName::new(&reply.name).ok().map(|name| {
                                StateChangeDefinitionSummary {
                                    name,
                                    version: reply.version,
                                }
                            }),
                            Err(_) => None,
                        }
                    })
                    .boxed(),
                Err(_) => stream::empty().boxed(),
            }
        })
        .flatten()
    }

    /// Fetch a transaction by its ID.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let txn = conn.fetch_transaction_by_id("my-txn-id").await?;
    /// println!("Transaction revision: {}", txn.summary.revision);
    /// ```
    pub async fn fetch_transaction_by_id(
        &self,
        transaction_id: &str,
    ) -> Result<Transaction, DatabaseError> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();

        let reply = client
            .fetch_transaction_by_id(database_name.clone(), transaction_id.to_string())
            .await
            .map_err(|e| match e {
                crate::Error::GrpcStatus(ref status) => {
                    crate::status_mapping::to_database_error(status, &database_name)
                }
                _ => DatabaseError::ServerError(e.to_string()),
            })?;

        reply
            .transaction
            .ok_or_else(|| {
                DatabaseError::ServerError("missing transaction in response".to_string())
            })
            .and_then(|txn| {
                Transaction::try_from(txn)
                    .map_err(|e| DatabaseError::ServerError(format!("invalid transaction: {}", e)))
            })
    }

    /// Get a reference to the underlying gRPC client.
    ///
    /// This provides access to low-level operations not exposed through
    /// the high-level API.
    pub fn client(&self) -> &EvidentSourceClient {
        &self.inner.client
    }

    /// Create a transaction builder for constructing transactions.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let db = conn
    ///     .transaction()
    ///     .event(my_event)
    ///     .require_not_exists(EventSelector::stream_equals("my-stream"))
    ///     .commit()
    ///     .await?;
    /// ```
    pub fn transaction(&self) -> TransactionBuilder {
        TransactionBuilder::new(self.clone())
    }

    /// Create a state change builder for executing state changes.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let db = conn
    ///     .state_change("create-account", 1)
    ///     .json(&CreateAccountRequest { name: "Alice" })?
    ///     .execute()
    ///     .await?;
    /// ```
    pub fn state_change(&self, name: &str, version: StateChangeVersion) -> StateChangeBuilder {
        StateChangeBuilder::new(self.clone(), name.to_string(), version)
    }
}

// =============================================================================
// Async Command Extension Trait
// =============================================================================

/// A correlation ID returned from async commands.
///
/// This can be used to track the result of an async operation via Kafka
/// or other message brokers.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct CorrelationId(pub String);

impl std::fmt::Display for CorrelationId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl From<String> for CorrelationId {
    fn from(s: String) -> Self {
        Self(s)
    }
}

/// Extension trait for async database operations.
///
/// Async operations return immediately with a correlation ID that can be used
/// to track the result via Kafka or other message brokers.
pub trait DatabaseConnectionAsync {
    /// Transact asynchronously.
    ///
    /// Returns a correlation ID that can be used to track the result.
    fn transact_async(
        &self,
        events: NonEmpty<ProspectiveEvent>,
        conditions: Vec<AppendCondition>,
    ) -> impl std::future::Future<Output = Result<CorrelationId, DatabaseError>>;

    /// Transact asynchronously with a specific transaction ID.
    fn transact_async_with_id(
        &self,
        transaction_id: &str,
        events: NonEmpty<ProspectiveEvent>,
        conditions: Vec<AppendCondition>,
    ) -> impl std::future::Future<Output = Result<CorrelationId, DatabaseError>>;

    /// Execute a state change asynchronously.
    ///
    /// Returns a correlation ID that can be used to track the result.
    fn execute_state_change_async(
        &self,
        name: &StateChangeName,
        version: StateChangeVersion,
        request: CommandRequest,
    ) -> impl std::future::Future<Output = Result<CorrelationId, StateChangeError>>;
}

impl DatabaseConnectionAsync for Connection {
    fn transact_async(
        &self,
        events: NonEmpty<ProspectiveEvent>,
        conditions: Vec<AppendCondition>,
    ) -> impl std::future::Future<Output = Result<CorrelationId, DatabaseError>> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();
        let transaction_id = uuid::Uuid::new_v4().to_string();

        // Convert domain types to proto
        let proto_events: Vec<proto_ce::CloudEvent> =
            events.into_iter().map(|pe| pe.into()).collect();
        let proto_conditions: Vec<proto::AppendCondition> =
            conditions.into_iter().map(|c| c.into()).collect();

        async move {
            let response = client
                .transact_async(
                    transaction_id,
                    database_name.clone(),
                    proto_events,
                    proto_conditions,
                )
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &database_name)
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            Ok(CorrelationId(response.correlation_id))
        }
    }

    fn transact_async_with_id(
        &self,
        transaction_id: &str,
        events: NonEmpty<ProspectiveEvent>,
        conditions: Vec<AppendCondition>,
    ) -> impl std::future::Future<Output = Result<CorrelationId, DatabaseError>> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();
        let transaction_id = transaction_id.to_string();

        // Convert domain types to proto
        let proto_events: Vec<proto_ce::CloudEvent> =
            events.into_iter().map(|pe| pe.into()).collect();
        let proto_conditions: Vec<proto::AppendCondition> =
            conditions.into_iter().map(|c| c.into()).collect();

        async move {
            let response = client
                .transact_async(
                    transaction_id,
                    database_name.clone(),
                    proto_events,
                    proto_conditions,
                )
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &database_name)
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            Ok(CorrelationId(response.correlation_id))
        }
    }

    fn execute_state_change_async(
        &self,
        name: &StateChangeName,
        version: StateChangeVersion,
        request: CommandRequest,
    ) -> impl std::future::Future<Output = Result<CorrelationId, StateChangeError>> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.database_name.to_string();
        let state_change_name = name.to_string();
        let current_revision = self.current_metadata().revision;

        // Convert CommandRequest to proto
        let proto_request = proto::CommandRequest {
            headers: request
                .headers
                .into_iter()
                .map(|(k, v)| proto::Header { key: k, value: v })
                .collect(),
            body: request.body,
            content_type: request.content_type,
            content_schema: request.content_schema,
        };

        async move {
            let response = client
                .execute_state_change_async(
                    database_name,
                    state_change_name.clone(),
                    version,
                    Some(current_revision),
                    proto_request,
                    None,
                )
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_state_change_error(
                            status,
                            &state_change_name,
                            version,
                        )
                    }
                    _ => StateChangeError::ServerError(e.to_string()),
                })?;

            Ok(CorrelationId(response.correlation_id))
        }
    }
}

// =============================================================================
// Transaction Builder
// =============================================================================

/// Builder for constructing transactions.
///
/// # Example
///
/// ```ignore
/// let db = conn
///     .transaction()
///     .event(event1)
///     .event(event2)
///     .condition(AppendCondition::must_not_exist(selector))
///     .with_correlation_id("order-12345")
///     .commit()
///     .await?;
/// ```
pub struct TransactionBuilder {
    connection: Connection,
    events: Vec<ProspectiveEvent>,
    conditions: Vec<AppendCondition>,
    transaction_id: Option<String>,
    correlation_id: Option<String>,
    causation_id: Option<String>,
}

impl TransactionBuilder {
    /// Create a new transaction builder.
    pub fn new(connection: Connection) -> Self {
        Self {
            connection,
            events: Vec::new(),
            conditions: Vec::new(),
            transaction_id: None,
            correlation_id: None,
            causation_id: None,
        }
    }

    /// Add an event to the transaction.
    pub fn event(mut self, event: ProspectiveEvent) -> Self {
        self.events.push(event);
        self
    }

    /// Add multiple events to the transaction.
    pub fn events(mut self, events: impl IntoIterator<Item = ProspectiveEvent>) -> Self {
        self.events.extend(events);
        self
    }

    /// Add a condition to the transaction.
    pub fn condition(mut self, condition: AppendCondition) -> Self {
        self.conditions.push(condition);
        self
    }

    /// Add multiple conditions to the transaction.
    pub fn conditions(mut self, conditions: impl IntoIterator<Item = AppendCondition>) -> Self {
        self.conditions.extend(conditions);
        self
    }

    /// Require that no events matching the selector exist.
    pub fn require_not_exists(self, selector: evidentsource_core::domain::EventSelector) -> Self {
        self.condition(AppendCondition::must_not_exist(selector))
    }

    /// Require that at least one event matching the selector exists.
    pub fn require_exists(self, selector: evidentsource_core::domain::EventSelector) -> Self {
        self.condition(AppendCondition::must_exist(selector))
    }

    /// Set a specific transaction ID for idempotency.
    pub fn with_transaction_id(mut self, transaction_id: impl Into<String>) -> Self {
        self.transaction_id = Some(transaction_id.into());
        self
    }

    /// Set the correlation ID for grouping related events in a business flow.
    ///
    /// See: <https://github.com/cloudevents/spec/blob/main/cloudevents/extensions/correlation.md>
    pub fn with_correlation_id(mut self, correlation_id: impl Into<String>) -> Self {
        self.correlation_id = Some(correlation_id.into());
        self
    }

    /// Set the causation ID for parent-child event relationships.
    ///
    /// See: <https://github.com/cloudevents/spec/blob/main/cloudevents/extensions/correlation.md>
    pub fn with_causation_id(mut self, causation_id: impl Into<String>) -> Self {
        self.causation_id = Some(causation_id.into());
        self
    }

    /// Commit the transaction synchronously.
    ///
    /// Returns the database at the new revision.
    pub async fn commit(self) -> Result<DatabaseAtRevisionImpl, DatabaseError> {
        let events = NonEmpty::from_vec(self.events)
            .ok_or(DatabaseError::Transaction(TransactionError::Empty))?;

        let mut client = self.connection.inner.client.clone();
        let database_name = self.connection.inner.database_name.to_string();
        let transaction_id = self
            .transaction_id
            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());

        // Convert domain types to proto
        let proto_events: Vec<proto_ce::CloudEvent> =
            events.into_iter().map(|pe| pe.into()).collect();
        let proto_conditions: Vec<proto::AppendCondition> =
            self.conditions.into_iter().map(|c| c.into()).collect();

        let result = client
            .transact_with_options(
                transaction_id,
                database_name.clone(),
                proto_events,
                proto_conditions,
                self.correlation_id,
                self.causation_id,
            )
            .await
            .map_err(|e| match e {
                crate::Error::GrpcStatus(ref status) => {
                    crate::status_mapping::to_database_error(status, &database_name)
                }
                _ => DatabaseError::ServerError(e.to_string()),
            })?;

        // Get the new revision from the result
        let new_revision = result
            .transaction_summary
            .map(|s| s.revision)
            .unwrap_or_default();

        // Await the database at the new revision
        let proto_db = client
            .await_database(database_name.clone(), new_revision)
            .await
            .map_err(|e| match e {
                crate::Error::GrpcStatus(ref status) => {
                    crate::status_mapping::to_database_error(status, &database_name)
                }
                _ => DatabaseError::ServerError(e.to_string()),
            })?;

        DatabaseAtRevisionImpl::new(client, proto_db)
            .map_err(|e| DatabaseError::ServerError(format!("failed to parse database: {}", e)))
    }

    /// Commit the transaction asynchronously.
    ///
    /// Returns a correlation ID for tracking.
    pub async fn commit_async(self) -> Result<CorrelationId, DatabaseError> {
        let events = NonEmpty::from_vec(self.events)
            .ok_or(DatabaseError::Transaction(TransactionError::Empty))?;

        let mut client = self.connection.inner.client.clone();
        let database_name = self.connection.inner.database_name.to_string();
        let transaction_id = self
            .transaction_id
            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());

        // Convert domain types to proto
        let proto_events: Vec<proto_ce::CloudEvent> =
            events.into_iter().map(|pe| pe.into()).collect();
        let proto_conditions: Vec<proto::AppendCondition> =
            self.conditions.into_iter().map(|c| c.into()).collect();

        let response = client
            .transact_async_with_options(
                transaction_id,
                database_name.clone(),
                proto_events,
                proto_conditions,
                self.correlation_id,
                self.causation_id,
            )
            .await
            .map_err(|e| match e {
                crate::Error::GrpcStatus(ref status) => {
                    crate::status_mapping::to_database_error(status, &database_name)
                }
                _ => DatabaseError::ServerError(e.to_string()),
            })?;

        Ok(CorrelationId(response.correlation_id))
    }
}

// =============================================================================
// State Change Builder
// =============================================================================

/// Builder for executing state changes.
///
/// # Example
///
/// ```ignore
/// let db = conn
///     .state_change("create-account", 1)
///     .json(&CreateAccountRequest { name: "Alice" })?
///     .with_correlation_id("order-12345")
///     .execute()
///     .await?;
/// ```
pub struct StateChangeBuilder {
    connection: Connection,
    name: String,
    version: StateChangeVersion,
    request: Option<CommandRequest>,
    transaction_id: Option<String>,
    correlation_id: Option<String>,
    causation_id: Option<String>,
}

impl StateChangeBuilder {
    /// Create a new state change builder.
    pub fn new(connection: Connection, name: String, version: StateChangeVersion) -> Self {
        Self {
            connection,
            name,
            version,
            request: None,
            transaction_id: None,
            correlation_id: None,
            causation_id: None,
        }
    }

    /// Set the request body as JSON.
    pub fn json<T: ::serde::Serialize>(
        mut self,
        value: &T,
    ) -> Result<Self, evidentsource_core::domain::CommandRequestError> {
        self.request = Some(CommandRequest::json(value)?);
        Ok(self)
    }

    /// Set the raw command request.
    pub fn request(mut self, request: CommandRequest) -> Self {
        self.request = Some(request);
        self
    }

    /// Set a specific transaction ID for idempotency.
    pub fn with_transaction_id(mut self, transaction_id: impl Into<String>) -> Self {
        self.transaction_id = Some(transaction_id.into());
        self
    }

    /// Set the correlation ID for grouping related events in a business flow.
    ///
    /// See: <https://github.com/cloudevents/spec/blob/main/cloudevents/extensions/correlation.md>
    pub fn with_correlation_id(mut self, correlation_id: impl Into<String>) -> Self {
        self.correlation_id = Some(correlation_id.into());
        self
    }

    /// Set the causation ID for parent-child event relationships.
    ///
    /// See: <https://github.com/cloudevents/spec/blob/main/cloudevents/extensions/correlation.md>
    pub fn with_causation_id(mut self, causation_id: impl Into<String>) -> Self {
        self.causation_id = Some(causation_id.into());
        self
    }

    /// Set the content type for the request.
    ///
    /// If no request exists yet, creates a default request.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let db = conn
    ///     .state_change("process-xml", 1)
    ///     .request(CommandRequest::with_body(xml_bytes))
    ///     .content_type("application/xml")
    ///     .execute()
    ///     .await?;
    /// ```
    pub fn content_type(mut self, content_type: impl Into<String>) -> Self {
        let request = self.request.take().unwrap_or_default();
        self.request = Some(request.content_type(content_type));
        self
    }

    /// Set the content schema URL for the request.
    ///
    /// If no request exists yet, creates a default request.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let db = conn
    ///     .state_change("create-account", 1)
    ///     .json(&request)?
    ///     .content_schema("https://example.com/schemas/create-account.json")
    ///     .execute()
    ///     .await?;
    /// ```
    pub fn content_schema(mut self, schema_url: impl Into<String>) -> Self {
        let request = self.request.take().unwrap_or_default();
        self.request = Some(request.content_schema(schema_url));
        self
    }

    /// Execute the state change synchronously.
    pub async fn execute(self) -> Result<DatabaseAtRevisionImpl, StateChangeError> {
        let mut client = self.connection.inner.client.clone();
        let database_name = self.connection.inner.database_name.to_string();
        let state_change_name = self.name.clone();
        let current_revision = self.connection.current_metadata().revision;
        let request = self.request.unwrap_or_default();

        // Convert CommandRequest to proto
        let proto_request = proto::CommandRequest {
            headers: request
                .headers
                .into_iter()
                .map(|(k, v)| proto::Header { key: k, value: v })
                .collect(),
            body: request.body,
            content_type: request.content_type,
            content_schema: request.content_schema,
        };

        let result = client
            .execute_state_change_with_options(
                database_name.clone(),
                state_change_name.clone(),
                self.version,
                Some(current_revision),
                proto_request,
                self.transaction_id,
                self.correlation_id,
                self.causation_id,
            )
            .await
            .map_err(|e| match e {
                crate::Error::GrpcStatus(ref status) => {
                    crate::status_mapping::to_state_change_error(
                        status,
                        &state_change_name,
                        self.version,
                    )
                }
                _ => StateChangeError::ServerError(e.to_string()),
            })?;

        // Get the new revision from the result
        let new_revision = result
            .transaction_summary
            .map(|s| s.revision)
            .unwrap_or_default();

        // Await the database at the new revision
        let proto_db = client
            .await_database(database_name.clone(), new_revision)
            .await
            .map_err(|e| match e {
                crate::Error::GrpcStatus(ref status) => StateChangeError::Database(
                    crate::status_mapping::to_database_error(status, &database_name),
                ),
                _ => StateChangeError::Database(DatabaseError::ServerError(e.to_string())),
            })?;

        DatabaseAtRevisionImpl::new(client, proto_db).map_err(|e| {
            StateChangeError::Database(DatabaseError::ServerError(format!(
                "failed to parse database: {}",
                e
            )))
        })
    }

    /// Execute the state change asynchronously.
    pub async fn execute_async(self) -> Result<CorrelationId, StateChangeError> {
        let mut client = self.connection.inner.client.clone();
        let database_name = self.connection.inner.database_name.to_string();
        let state_change_name = self.name.clone();
        let current_revision = self.connection.current_metadata().revision;
        let request = self.request.unwrap_or_default();

        // Convert CommandRequest to proto
        let proto_request = proto::CommandRequest {
            headers: request
                .headers
                .into_iter()
                .map(|(k, v)| proto::Header { key: k, value: v })
                .collect(),
            body: request.body,
            content_type: request.content_type,
            content_schema: request.content_schema,
        };

        let response = client
            .execute_state_change_async_with_options(
                database_name,
                state_change_name.clone(),
                self.version,
                Some(current_revision),
                proto_request,
                self.transaction_id,
                self.correlation_id,
                self.causation_id,
            )
            .await
            .map_err(|e| match e {
                crate::Error::GrpcStatus(ref status) => {
                    crate::status_mapping::to_state_change_error(
                        status,
                        &state_change_name,
                        self.version,
                    )
                }
                _ => StateChangeError::ServerError(e.to_string()),
            })?;

        Ok(CorrelationId(response.correlation_id))
    }
}