evidentsource-client 1.0.0-rc1

//! DatabaseAtRevision implementation backed by gRPC.

use std::sync::Arc;

use chrono::{DateTime, Utc};
use futures::stream::{self, StreamExt};
use futures::Stream;
use nonempty::NonEmpty;

use evidentsource_core::domain::{
    DatabaseName, Event, EventAttribute, EventSelector, ProspectiveEvent, QueryOptions, Revision,
    StateView, StateViewError, StateViewName, StateViewVersion,
};
use evidentsource_core::{DatabaseAtRevision, DatabaseIdentity};

use crate::com::evidentsource as proto;
use crate::conversions::ConversionError;
use crate::EvidentSourceClient;

use super::effective_timestamp::EffectiveTimestampViewImpl;
use super::speculative::SpeculativeDatabaseImpl;

/// Inner state shared via Arc for Clone-ability.
struct DatabaseAtRevisionInner {
    /// Reference to the gRPC client for making queries.
    client: EvidentSourceClient,
    /// Database name.
    name: DatabaseName,
    /// When the database was created.
    created_at: DateTime<Utc>,
    /// This view's revision number.
    revision: Revision,
    /// When this revision was committed.
    revision_timestamp: DateTime<Utc>,
}

/// A database view at a specific revision, backed by gRPC calls.
///
/// This implements `DatabaseAtRevision` trait from the core crate, allowing
/// queries for events and state views at a specific point-in-time revision.
#[derive(Clone)]
pub struct DatabaseAtRevisionImpl {
    inner: Arc<DatabaseAtRevisionInner>,
}

impl DatabaseAtRevisionImpl {
    /// Create a new database view from a proto Database message.
    pub fn new(
        client: EvidentSourceClient,
        proto_db: proto::Database,
    ) -> Result<Self, ConversionError> {
        use crate::conversions::timestamp_to_datetime;

        let name = DatabaseName::new(&proto_db.name)?;
        let created_at = proto_db
            .created_at
            .ok_or_else(|| ConversionError::missing_field("Database", "created_at"))
            .and_then(timestamp_to_datetime)?;
        let revision_timestamp = proto_db
            .revision_timestamp
            .ok_or_else(|| ConversionError::missing_field("Database", "revision_timestamp"))
            .and_then(timestamp_to_datetime)?;

        Ok(Self {
            inner: Arc::new(DatabaseAtRevisionInner {
                client,
                name,
                created_at,
                revision: proto_db.revision,
                revision_timestamp,
            }),
        })
    }

    /// Create a database view at a specific revision using existing metadata.
    pub(crate) fn at_revision_with_metadata(
        client: EvidentSourceClient,
        name: DatabaseName,
        created_at: DateTime<Utc>,
        revision: Revision,
        revision_timestamp: DateTime<Utc>,
    ) -> Self {
        Self {
            inner: Arc::new(DatabaseAtRevisionInner {
                client,
                name,
                created_at,
                revision,
                revision_timestamp,
            }),
        }
    }

    /// Get a clone of the underlying client.
    pub(crate) fn client(&self) -> EvidentSourceClient {
        self.inner.client.clone()
    }
}

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

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

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

impl DatabaseAtRevision for DatabaseAtRevisionImpl {
    type EffectiveTimestampView = EffectiveTimestampViewImpl;
    type Speculative = SpeculativeDatabaseImpl;

    fn revision(&self) -> Revision {
        self.inner.revision
    }

    fn revision_timestamp(&self) -> DateTime<Utc> {
        self.inner.revision_timestamp
    }

    fn at_effective_timestamp(
        &self,
        effective_timestamp: DateTime<Utc>,
    ) -> Self::EffectiveTimestampView {
        EffectiveTimestampViewImpl::new(self.clone(), effective_timestamp)
    }

    fn speculate_with_transaction(
        &self,
        transaction: NonEmpty<ProspectiveEvent>,
    ) -> Self::Speculative {
        SpeculativeDatabaseImpl::new(self.clone(), transaction)
    }

    fn at_revision(&self, revision: Revision) -> impl std::future::Future<Output = Self> {
        let mut client = self.inner.client.clone();
        let database_name = self.inner.name.to_string();
        let name = self.inner.name.clone();
        let created_at = self.inner.created_at;

        async move {
            // Fetch the database at the requested revision
            let proto_db = client.await_database(database_name, revision).await;

            match proto_db {
                Ok(db) => {
                    use crate::conversions::timestamp_to_datetime;
                    let revision_timestamp = db
                        .revision_timestamp
                        .and_then(|ts| timestamp_to_datetime(ts).ok())
                        .unwrap_or_else(chrono::Utc::now);

                    Self::at_revision_with_metadata(
                        client,
                        name,
                        created_at,
                        db.revision,
                        revision_timestamp,
                    )
                }
                Err(_) => {
                    // On error, return self (best effort)
                    Self::at_revision_with_metadata(
                        client,
                        name,
                        created_at,
                        revision,
                        chrono::Utc::now(),
                    )
                }
            }
        }
    }

    fn query_events(&self, selector: &EventSelector) -> impl Stream<Item = Event> {
        let database_name = self.inner.name.to_string();
        let revision = self.inner.revision;
        let mut client = self.inner.client.clone();
        let proto_selector: proto::EventSelector = selector.clone().into();

        let query = proto::DatabaseQuery {
            selector: Some(proto_selector),
            range: Some(proto::QueryRange {
                range: Some(proto::query_range::Range::Revision(
                    proto::query_range::RevisionRange { start_at: Some(0) },
                )),
            }),
            direction: proto::QueryDirection::Forward as i32,
            limit: None,
        };

        stream::once(async move {
            let result = client
                .query_events(database_name, revision, true, query)
                .await;

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

    fn query_events_with_options(
        &self,
        selector: &EventSelector,
        options: QueryOptions,
    ) -> impl Stream<Item = Event> {
        use evidentsource_core::domain::QueryDirection;

        let database_name = self.inner.name.to_string();
        let revision = self.inner.revision;
        let mut client = self.inner.client.clone();
        let proto_selector: proto::EventSelector = selector.clone().into();

        let direction = match options.get_direction() {
            QueryDirection::Forward => proto::QueryDirection::Forward as i32,
            QueryDirection::Reverse => proto::QueryDirection::Reverse as i32,
        };

        let query = proto::DatabaseQuery {
            selector: Some(proto_selector),
            range: Some(proto::QueryRange {
                range: Some(proto::query_range::Range::Revision(
                    proto::query_range::RevisionRange { start_at: Some(0) },
                )),
            }),
            direction,
            limit: options.get_limit().map(|l| l as u32),
        };

        stream::once(async move {
            let result = client
                .query_events(database_name, revision, true, query)
                .await;

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

    fn view_state(
        &self,
        name: &StateViewName,
        version: StateViewVersion,
    ) -> impl std::future::Future<Output = Result<StateView, StateViewError>> {
        let state_view_name = name.to_string();
        let identity = proto::StateViewIdentity {
            database_name: self.inner.name.to_string(),
            state_view_name: state_view_name.clone(),
            state_view_version: version,
        };
        let revision = self.inner.revision;
        let mut client = self.inner.client.clone();

        async move {
            let proto_view = client
                .fetch_state_view_at_revision(Some(identity), revision, None, None)
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_state_view_error(
                            status,
                            &state_view_name,
                            version,
                        )
                    }
                    _ => StateViewError::ServerError(e.to_string()),
                })?;

            StateView::try_from(proto_view).map_err(|e| {
                StateViewError::ServerError(format!("failed to parse state view: {}", e))
            })
        }
    }

    fn view_state_with_params(
        &self,
        name: &StateViewName,
        version: StateViewVersion,
        params: &[(String, EventAttribute)],
    ) -> impl std::future::Future<Output = Result<StateView, StateViewError>> {
        let state_view_name = name.to_string();
        let identity = proto::StateViewIdentity {
            database_name: self.inner.name.to_string(),
            state_view_name: state_view_name.clone(),
            state_view_version: version,
        };
        let revision = self.inner.revision;
        let mut client = self.inner.client.clone();

        // Convert params to proto ParameterBindings
        let param_bindings = if params.is_empty() {
            None
        } else {
            Some(proto::ParameterBindings {
                bindings: params
                    .iter()
                    .map(|(k, v)| (k.clone(), v.clone().into()))
                    .collect(),
            })
        };

        async move {
            let proto_view = client
                .fetch_state_view_at_revision(Some(identity), revision, param_bindings, None)
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_state_view_error(
                            status,
                            &state_view_name,
                            version,
                        )
                    }
                    _ => StateViewError::ServerError(e.to_string()),
                })?;

            StateView::try_from(proto_view).map_err(|e| {
                StateViewError::ServerError(format!("failed to parse state view: {}", e))
            })
        }
    }
}

// =============================================================================
// DatabaseAtRevision Extensions
// =============================================================================

impl DatabaseAtRevisionImpl {
    /// Create an event query builder for ergonomic event queries.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use futures::StreamExt;
    ///
    /// // Simple query
    /// let events: Vec<Event> = db.events(EventSelector::all()).collect().await;
    ///
    /// // With options
    /// let events: Vec<Event> = db
    ///     .events(EventSelector::stream_equals("my-stream"))
    ///     .reverse()
    ///     .limit(10)
    ///     .collect()
    ///     .await;
    /// ```
    pub fn events(&self, selector: EventSelector) -> EventQueryBuilder {
        EventQueryBuilder::new(self.clone(), selector)
    }

    /// List all stream names in the database at this revision.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use futures::StreamExt;
    ///
    /// let streams: Vec<StreamName> = db.list_streams().collect().await;
    /// for stream in streams {
    ///     println!("Stream: {}", stream);
    /// }
    /// ```
    pub fn list_streams(&self) -> impl Stream<Item = evidentsource_core::domain::StreamName> {
        use crate::com::evidentsource::index_key_scan_request::IndexKeyType;
        use evidentsource_core::domain::StreamName;

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

        stream::once(async move {
            let result = client
                .scan_index_keys(database_name, revision, IndexKeyType::Stream)
                .await;

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

    /// List all subjects in the database at this revision.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use futures::StreamExt;
    ///
    /// let subjects: Vec<EventSubject> = db.list_subjects().collect().await;
    /// ```
    pub fn list_subjects(&self) -> impl Stream<Item = evidentsource_core::domain::EventSubject> {
        use crate::com::evidentsource::index_key_scan_request::IndexKeyType;
        use evidentsource_core::domain::EventSubject;

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

        stream::once(async move {
            let result = client
                .scan_index_keys(database_name, revision, IndexKeyType::Subject)
                .await;

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

    /// List all event types in the database at this revision.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use futures::StreamExt;
    ///
    /// let types: Vec<EventType> = db.list_event_types().collect().await;
    /// ```
    pub fn list_event_types(&self) -> impl Stream<Item = evidentsource_core::domain::EventType> {
        use crate::com::evidentsource::index_key_scan_request::IndexKeyType;
        use evidentsource_core::domain::EventType;

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

        stream::once(async move {
            let result = client
                .scan_index_keys(database_name, revision, IndexKeyType::EventType)
                .await;

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

// =============================================================================
// Event Query Builder
// =============================================================================

/// Builder for constructing event queries with options.
///
/// # Example
///
/// ```ignore
/// // Get all events as a stream
/// let events = db.events(EventSelector::all()).stream();
///
/// // Get first 10 events in reverse order, collected to a Vec
/// let events: Vec<Event> = db
///     .events(EventSelector::stream_equals("my-stream"))
///     .reverse()
///     .limit(10)
///     .collect()
///     .await;
///
/// // Get just the first event
/// let first = db.events(EventSelector::all()).first().await;
/// ```
pub struct EventQueryBuilder {
    database: DatabaseAtRevisionImpl,
    selector: EventSelector,
    options: QueryOptions,
}

impl EventQueryBuilder {
    /// Create a new event query builder.
    pub fn new(database: DatabaseAtRevisionImpl, selector: EventSelector) -> Self {
        Self {
            database,
            selector,
            options: QueryOptions::default(),
        }
    }

    /// Set the query direction.
    pub fn direction(mut self, direction: evidentsource_core::domain::QueryDirection) -> Self {
        self.options = self.options.direction(direction);
        self
    }

    /// Set the query to reverse order (newest first).
    pub fn reverse(mut self) -> Self {
        self.options = self.options.reverse();
        self
    }

    /// Set the query to forward order (oldest first).
    pub fn forward(mut self) -> Self {
        self.options = self.options.forward();
        self
    }

    /// Set a limit on the number of events returned.
    pub fn limit(mut self, limit: u64) -> Self {
        self.options = self.options.limit(limit);
        self
    }

    /// Execute the query and collect all events into a Vec.
    pub async fn collect(self) -> Vec<Event> {
        use futures::StreamExt as _;
        self.database
            .query_events_with_options(&self.selector, self.options)
            .collect()
            .await
    }

    /// Execute the query and return just the first event, if any.
    pub async fn first(mut self) -> Option<Event> {
        use futures::StreamExt as _;
        self.options = self.options.limit(1);
        self.database
            .query_events_with_options(&self.selector, self.options)
            .boxed()
            .next()
            .await
    }

    /// Create a typed event query that filters and converts events.
    ///
    /// The type `T` must implement `TryFrom<&Event>`. Events that fail to
    /// convert are silently skipped.
    ///
    /// # Example
    ///
    /// ```ignore
    /// #[derive(Debug)]
    /// struct AccountCreated { name: String }
    ///
    /// impl TryFrom<&Event> for AccountCreated {
    ///     type Error = Box<dyn std::error::Error>;
    ///     fn try_from(event: &Event) -> Result<Self, Self::Error> {
    ///         // Parse the event data...
    ///     }
    /// }
    ///
    /// let accounts: Vec<AccountCreated> = db
    ///     .events(EventSelector::type_equals("AccountCreated"))
    ///     .typed::<AccountCreated>()
    ///     .collect()
    ///     .await;
    /// ```
    pub fn typed<T>(self) -> TypedEventQuery<T>
    where
        T: for<'a> TryFrom<&'a Event> + 'static,
    {
        TypedEventQuery::new(self)
    }
}

/// A typed event query that converts events to a specific type.
pub struct TypedEventQuery<T> {
    builder: EventQueryBuilder,
    _phantom: std::marker::PhantomData<T>,
}

impl<T> TypedEventQuery<T>
where
    T: for<'a> TryFrom<&'a Event> + 'static,
{
    fn new(builder: EventQueryBuilder) -> Self {
        Self {
            builder,
            _phantom: std::marker::PhantomData,
        }
    }

    /// Execute the query and collect all typed events into a Vec.
    ///
    /// Events that fail to convert are silently skipped.
    pub async fn collect(self) -> Vec<T> {
        let events: Vec<Event> = self.builder.collect().await;
        events
            .iter()
            .filter_map(|event| T::try_from(event).ok())
            .collect()
    }

    /// Execute the query and return just the first typed event, if any.
    pub async fn first(self) -> Option<T> {
        let event = self.builder.first().await?;
        T::try_from(&event).ok()
    }
}

// =============================================================================
// Typed State View Extension Trait
// =============================================================================

/// Extension trait for typed state view access.
///
/// This trait provides convenience methods for fetching state views and
/// automatically deserializing them to typed Rust structs.
pub trait DatabaseAtRevisionTyped: DatabaseAtRevision {
    /// Fetch a state view and deserialize it to the specified type.
    ///
    /// # Example
    ///
    /// ```ignore
    /// #[derive(Deserialize)]
    /// struct AccountSummary {
    ///     balance: i64,
    ///     transaction_count: u32,
    /// }
    ///
    /// let summary: AccountSummary = db
    ///     .view_state_as::<AccountSummary>(&StateViewName::new("account-summary")?, 1)
    ///     .await?;
    /// ```
    fn view_state_as<T>(
        &self,
        name: &StateViewName,
        version: StateViewVersion,
    ) -> impl std::future::Future<Output = Result<T, StateViewError>>
    where
        T: serde::de::DeserializeOwned;

    /// Fetch a state view and deserialize it, returning None if not found.
    ///
    /// This is useful when the state view might not exist yet (e.g., before
    /// any events have been recorded for an entity).
    fn view_state_opt<T>(
        &self,
        name: &StateViewName,
        version: StateViewVersion,
    ) -> impl std::future::Future<Output = Result<Option<T>, StateViewError>>
    where
        T: serde::de::DeserializeOwned;
}

impl DatabaseAtRevisionTyped for DatabaseAtRevisionImpl {
    fn view_state_as<T>(
        &self,
        name: &StateViewName,
        version: StateViewVersion,
    ) -> impl std::future::Future<Output = Result<T, StateViewError>>
    where
        T: serde::de::DeserializeOwned,
    {
        let view_future = self.view_state(name, version);

        async move {
            let state_view = view_future.await?;

            // Get the content bytes from the state view
            let content = state_view.content_bytes().ok_or_else(|| {
                StateViewError::ServerError("state view has no content".to_string())
            })?;

            // Deserialize from JSON
            serde_json::from_slice(content).map_err(|e| {
                StateViewError::ServerError(format!("failed to deserialize state view: {}", e))
            })
        }
    }

    fn view_state_opt<T>(
        &self,
        name: &StateViewName,
        version: StateViewVersion,
    ) -> impl std::future::Future<Output = Result<Option<T>, StateViewError>>
    where
        T: serde::de::DeserializeOwned,
    {
        let view_future = self.view_state(name, version);

        async move {
            match view_future.await {
                Ok(state_view) => {
                    // Get the content bytes from the state view
                    match state_view.content_bytes() {
                        Some(content) => {
                            let value = serde_json::from_slice(content).map_err(|e| {
                                StateViewError::ServerError(format!(
                                    "failed to deserialize state view: {}",
                                    e
                                ))
                            })?;
                            Ok(Some(value))
                        }
                        None => Ok(None),
                    }
                }
                Err(StateViewError::NotFound { .. }) => Ok(None),
                Err(e) => Err(e),
            }
        }
    }
}