umadb-dcb 0.5.6

//! API for Dynamic Consistency Boundaries (DCB) event store
//!
//! This module provides the core interfaces and data structures for working with
//! an event store that supports dynamic consistency boundaries.

use async_trait::async_trait;
use futures_core::Stream;
use futures_util::StreamExt;
use std::iter::Iterator;
use thiserror::Error;
use uuid::Uuid;

/// Non-async Rust interface for recording and retrieving events
pub trait DcbEventStoreSync {
    /// Reads events from the store based on the provided query and constraints
    ///
    /// Returns a `DcbReadResponseSync` that provides an iterator over all events,
    /// unless 'from' is given then only those with position greater than 'after',
    /// and unless any query items are given, then only those that match at least one
    /// query item. An event matches a query item if its type is in the item types or
    /// there are no item types, and if all the item tags are in the event tags.
    fn read(
        &self,
        query: Option<DcbQuery>,
        start: Option<u64>,
        backwards: bool,
        limit: Option<u32>,
        subscribe: bool, // Deprecated - remove in v1.0.
    ) -> DcbResult<Box<dyn DcbReadResponseSync + Send + 'static>>;

    /// Reads events from the store and returns them as a tuple of `(Vec<DcbSequencedEvent>, Option<u64>)`
    fn read_with_head(
        &self,
        query: Option<DcbQuery>,
        start: Option<u64>,
        backwards: bool,
        limit: Option<u32>,
    ) -> DcbResult<(Vec<DcbSequencedEvent>, Option<u64>)> {
        let mut response = self.read(query, start, backwards, limit, false)?;
        response.collect_with_head()
    }

    /// Returns the current head position of the event store, or None if empty
    ///
    /// Returns the value of `last_committed_position`, or `None` if `last_committed_position` is zero
    fn head(&self) -> DcbResult<Option<u64>>;

    /// Returns the greatest recorded upstream position for a tracking source, if any
    fn get_tracking_info(&self, source: &str) -> DcbResult<Option<u64>>;

    /// Appends given events to the event store, unless the condition fails
    ///
    /// Returns the position of the last appended event
    fn append(
        &self,
        events: Vec<DcbEvent>,
        condition: Option<DcbAppendCondition>,
        tracking_info: Option<TrackingInfo>,
    ) -> DcbResult<u64>;
}

/// Response from a read operation, providing an iterator over sequenced events
pub trait DcbReadResponseSync: Iterator<Item = DcbResult<DcbSequencedEvent>> + Send {
    /// Returns the current head position of the event store, or None if empty
    fn head(&mut self) -> DcbResult<Option<u64>>;
    /// Returns a vector of events with head
    fn collect_with_head(&mut self) -> DcbResult<(Vec<DcbSequencedEvent>, Option<u64>)>;
    /// Returns the next batch of events for this read. Implementations may buffer
    /// events per underlying transport message ("batch"). If there are no more
    /// events available, returns an empty Vec. The head() method should reflect
    /// the latest known head as reported by the underlying store.
    fn next_batch(&mut self) -> DcbResult<Vec<DcbSequencedEvent>>;
}

/// Response from a subscribe operation, providing an iterator over sequenced events
pub trait DcbSubscriptionSync: Iterator<Item = DcbResult<DcbSequencedEvent>> + Send {
    /// Returns the next batch of events for this read. Implementations may buffer
    /// events per underlying transport message ("batch"). If there are no more
    /// events available, returns an empty Vec.
    fn next_batch(&mut self) -> DcbResult<Vec<DcbSequencedEvent>>;
}

/// Async Rust interface for recording and retrieving events
#[async_trait]
pub trait DcbEventStoreAsync: Send + Sync {
    /// Reads events from the store based on the provided query and constraints
    ///
    /// Returns a `DcbReadResponseSync` that provides an iterator over all events,
    /// unless 'after' is given then only those with position greater than 'after',
    /// and unless any query items are given, then only those that match at least one
    /// query item. An event matches a query item if its type is in the item types or
    /// there are no item types, and if all the item tags are in the event tags.
    async fn read<'a>(
        &'a self,
        query: Option<DcbQuery>,
        start: Option<u64>,
        backwards: bool,
        limit: Option<u32>,
        subscribe: bool,
    ) -> DcbResult<Box<dyn DcbReadResponseAsync + Send + 'static>>;

    /// Reads events from the store and returns them as a tuple of `(Vec<DcbSequencedEvent>, Option<u64>)`
    async fn read_with_head<'a>(
        &'a self,
        query: Option<DcbQuery>,
        after: Option<u64>,
        backwards: bool,
        limit: Option<u32>,
    ) -> DcbResult<(Vec<DcbSequencedEvent>, Option<u64>)> {
        let mut response = self.read(query, after, backwards, limit, false).await?;
        response.collect_with_head().await
    }

    /// Returns the current head position of the event store, or None if empty
    ///
    /// Returns the value of last_committed_position, or None if last_committed_position is zero
    async fn head(&self) -> DcbResult<Option<u64>>;

    /// Returns the greatest recorded upstream position for a tracking source, if any
    async fn get_tracking_info(&self, source: &str) -> DcbResult<Option<u64>>;

    /// Appends given events to the event store, unless the condition fails
    ///
    /// Returns the position of the last appended event
    async fn append(
        &self,
        events: Vec<DcbEvent>,
        condition: Option<DcbAppendCondition>,
        tracking_info: Option<TrackingInfo>,
    ) -> DcbResult<u64>;
}

/// Asynchronous response from a read operation, providing a stream of sequenced events
#[async_trait]
pub trait DcbReadResponseAsync: Stream<Item = DcbResult<DcbSequencedEvent>> + Send + Unpin {
    async fn head(&mut self) -> DcbResult<Option<u64>>;

    async fn collect_with_head(&mut self) -> DcbResult<(Vec<DcbSequencedEvent>, Option<u64>)> {
        let mut events = Vec::new();
        while let Some(result) = self.next().await {
            events.push(result?); // propagate error from stream
        }

        let head = self.head().await?;
        Ok((events, head))
    }

    async fn next_batch(&mut self) -> DcbResult<Vec<DcbSequencedEvent>>;
}

/// Asynchronous response from a subscribe operation, providing a stream of sequenced events
#[async_trait]
pub trait DcbSubscriptionAsync: Stream<Item = DcbResult<DcbSequencedEvent>> + Send + Unpin {
    async fn next_batch(&mut self) -> DcbResult<Vec<DcbSequencedEvent>>;
}

/// Represents a query item for filtering events
#[derive(Debug, Clone, Default)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct DcbQueryItem {
    /// Event types to match
    pub types: Vec<String>,
    /// Tags that must all be present in the event
    pub tags: Vec<String>,
}

impl DcbQueryItem {
    /// Creates a new query item
    pub fn new() -> Self {
        Self {
            types: vec![],
            tags: vec![],
        }
    }

    /// Sets the types for this query item
    pub fn types<I, S>(mut self, types: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.types = types.into_iter().map(|s| s.into()).collect();
        self
    }

    /// Sets the tags for this query item
    pub fn tags<I, S>(mut self, tags: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.tags = tags.into_iter().map(|s| s.into()).collect();
        self
    }
}

/// A query composed of multiple query items
#[derive(Debug, Clone, Default)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct DcbQuery {
    /// List of query items, where events matching any item are included in results
    pub items: Vec<DcbQueryItem>,
}

impl DcbQuery {
    /// Creates a new empty query
    pub fn new() -> Self {
        Self { items: Vec::new() }
    }

    /// Creates a query with the specified items
    pub fn with_items<I>(items: I) -> Self
    where
        I: IntoIterator<Item = DcbQueryItem>,
    {
        Self {
            items: items.into_iter().collect(),
        }
    }

    /// Adds a query item to this query
    pub fn item(mut self, item: DcbQueryItem) -> Self {
        self.items.push(item);
        self
    }

    /// Adds multiple query items to this query
    pub fn items<I>(mut self, items: I) -> Self
    where
        I: IntoIterator<Item = DcbQueryItem>,
    {
        self.items.extend(items);
        self
    }
}

/// Conditions that must be satisfied for an append operation to succeed
#[derive(Debug, Clone, Default)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct DcbAppendCondition {
    /// Query that, if matching any events, will cause the append to fail
    pub fail_if_events_match: DcbQuery,
    /// Position after which to append; if None, append at the end
    pub after: Option<u64>,
}

impl DcbAppendCondition {
    /// Creates a new empty append condition
    pub fn new(fail_if_events_match: DcbQuery) -> Self {
        Self {
            fail_if_events_match,
            after: None,
        }
    }

    pub fn after(mut self, after: Option<u64>) -> Self {
        self.after = after;
        self
    }
}

/// Represents an event in the event store
#[derive(Debug, Clone)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct DcbEvent {
    /// Type of the event
    pub event_type: String,
    /// Tags associated with the event
    pub tags: Vec<String>,
    /// Binary data associated with the event
    pub data: Vec<u8>,
    /// Unique event ID
    pub uuid: Option<Uuid>,
}

impl Default for DcbEvent {
    fn default() -> Self {
        Self::new()
    }
}

impl DcbEvent {
    /// Creates a new event
    pub fn new() -> Self {
        Self {
            event_type: "".to_string(),
            data: Vec::new(),
            tags: Vec::new(),
            uuid: None,
        }
    }

    /// Sets the type for this event
    pub fn event_type<S: Into<String>>(mut self, event_type: S) -> Self {
        self.event_type = event_type.into();
        self
    }

    /// Sets the data for this event
    pub fn data<D: Into<Vec<u8>>>(mut self, data: D) -> Self {
        self.data = data.into();
        self
    }

    /// Sets the tags for this event
    pub fn tags<I, S>(mut self, tags: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.tags = tags.into_iter().map(|s| s.into()).collect();
        self
    }

    /// Sets the UUID for this event
    pub fn uuid(mut self, uuid: Uuid) -> Self {
        self.uuid = Some(uuid);
        self
    }
}

#[derive(Debug, Clone)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct TrackingInfo {
    pub source: String,
    pub position: u64,
}

/// An event with its position in the event sequence
#[derive(Debug, Clone)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct DcbSequencedEvent {
    /// Position of the event in the sequence
    pub position: u64,
    /// The event
    pub event: DcbEvent,
}

// Error types
#[derive(Error, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub enum DcbError {
    // Generic/system errors
    #[error("io error: {0}")]
    #[cfg_attr(feature = "serde", serde(with = "serde_io_error"))]
    Io(#[from] std::io::Error),

    // DCB domain errors
    #[error("integrity error: condition failed: {0}")]
    IntegrityError(String),
    #[error("corruption detected: {0}")]
    Corruption(String),
    /// Invalid input argument provided by the caller
    #[error("invalid argument: {0}")]
    InvalidArgument(String),

    // Storage errors (unified into DCBError)
    #[error("initialization error: {0}")]
    InitializationError(String),
    #[error("page not found: {0}")]
    PageNotFound(u64),
    #[error("dirty page not found: {0}")]
    DirtyPageNotFound(u64),
    #[error("root ID mismatched: old {0} new {1}")]
    RootIDMismatch(u64, u64),
    #[error("database corrupted: {0}")]
    DatabaseCorrupted(String),
    #[error("internal error: {0}")]
    InternalError(String),
    #[error("serialization error: {0}")]
    SerializationError(String),
    #[error("deserialization error: {0}")]
    DeserializationError(String),
    #[error("page already freed: {0}")]
    PageAlreadyFreed(u64),
    #[error("page already dirty: {0}")]
    PageAlreadyDirty(u64),
    #[error("transport error: {0}")]
    TransportError(String),
    #[error("cancelled by user")]
    CancelledByUser(),

    // Authentication error
    #[error("authentication error: {0}")]
    AuthenticationError(String),
}

pub type DcbResult<T> = Result<T, DcbError>;

#[cfg(feature = "serde")]
mod serde_io_error {
    use std::{borrow::Cow, io};

    use serde::{Deserialize, Serialize};

    #[derive(Serialize, Deserialize)]
    struct IoError {
        kind: Option<Cow<'static, str>>,
        message: Option<String>,
    }

    pub fn serialize<S>(err: &io::Error, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        let kind = match err.kind() {
            io::ErrorKind::NotFound => Some(Cow::Borrowed("NotFound")),
            io::ErrorKind::PermissionDenied => Some(Cow::Borrowed("PermissionDenied")),
            io::ErrorKind::ConnectionRefused => Some(Cow::Borrowed("ConnectionRefused")),
            io::ErrorKind::ConnectionReset => Some(Cow::Borrowed("ConnectionReset")),
            io::ErrorKind::HostUnreachable => Some(Cow::Borrowed("HostUnreachable")),
            io::ErrorKind::NetworkUnreachable => Some(Cow::Borrowed("NetworkUnreachable")),
            io::ErrorKind::ConnectionAborted => Some(Cow::Borrowed("ConnectionAborted")),
            io::ErrorKind::NotConnected => Some(Cow::Borrowed("NotConnected")),
            io::ErrorKind::AddrInUse => Some(Cow::Borrowed("AddrInUse")),
            io::ErrorKind::AddrNotAvailable => Some(Cow::Borrowed("AddrNotAvailable")),
            io::ErrorKind::NetworkDown => Some(Cow::Borrowed("NetworkDown")),
            io::ErrorKind::BrokenPipe => Some(Cow::Borrowed("BrokenPipe")),
            io::ErrorKind::AlreadyExists => Some(Cow::Borrowed("AlreadyExists")),
            io::ErrorKind::WouldBlock => Some(Cow::Borrowed("WouldBlock")),
            io::ErrorKind::NotADirectory => Some(Cow::Borrowed("NotADirectory")),
            io::ErrorKind::IsADirectory => Some(Cow::Borrowed("IsADirectory")),
            io::ErrorKind::DirectoryNotEmpty => Some(Cow::Borrowed("DirectoryNotEmpty")),
            io::ErrorKind::ReadOnlyFilesystem => Some(Cow::Borrowed("ReadOnlyFilesystem")),
            io::ErrorKind::StaleNetworkFileHandle => Some(Cow::Borrowed("StaleNetworkFileHandle")),
            io::ErrorKind::InvalidInput => Some(Cow::Borrowed("InvalidInput")),
            io::ErrorKind::InvalidData => Some(Cow::Borrowed("InvalidData")),
            io::ErrorKind::TimedOut => Some(Cow::Borrowed("TimedOut")),
            io::ErrorKind::WriteZero => Some(Cow::Borrowed("WriteZero")),
            io::ErrorKind::StorageFull => Some(Cow::Borrowed("StorageFull")),
            io::ErrorKind::NotSeekable => Some(Cow::Borrowed("NotSeekable")),
            io::ErrorKind::QuotaExceeded => Some(Cow::Borrowed("QuotaExceeded")),
            io::ErrorKind::FileTooLarge => Some(Cow::Borrowed("FileTooLarge")),
            io::ErrorKind::ResourceBusy => Some(Cow::Borrowed("ResourceBusy")),
            io::ErrorKind::ExecutableFileBusy => Some(Cow::Borrowed("ExecutableFileBusy")),
            io::ErrorKind::Deadlock => Some(Cow::Borrowed("Deadlock")),
            io::ErrorKind::CrossesDevices => Some(Cow::Borrowed("CrossesDevices")),
            io::ErrorKind::TooManyLinks => Some(Cow::Borrowed("TooManyLinks")),
            io::ErrorKind::InvalidFilename => Some(Cow::Borrowed("InvalidFilename")),
            io::ErrorKind::ArgumentListTooLong => Some(Cow::Borrowed("ArgumentListTooLong")),
            io::ErrorKind::Interrupted => Some(Cow::Borrowed("Interrupted")),
            io::ErrorKind::Unsupported => Some(Cow::Borrowed("Unsupported")),
            io::ErrorKind::UnexpectedEof => Some(Cow::Borrowed("UnexpectedEof")),
            io::ErrorKind::OutOfMemory => Some(Cow::Borrowed("OutOfMemory")),
            io::ErrorKind::Other => Some(Cow::Borrowed("Other")),
            _ => None,
        };

        IoError {
            kind,
            message: err.get_ref().map(|err| err.to_string()),
        }
        .serialize(serializer)
    }

    pub fn deserialize<'de, D>(deserializer: D) -> Result<io::Error, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let io_err: IoError = <IoError as Deserialize>::deserialize(deserializer)?;
        let kind = match io_err.kind.as_deref() {
            Some("NotFound") => io::ErrorKind::NotFound,
            Some("PermissionDenied") => io::ErrorKind::PermissionDenied,
            Some("ConnectionRefused") => io::ErrorKind::ConnectionRefused,
            Some("ConnectionReset") => io::ErrorKind::ConnectionReset,
            Some("HostUnreachable") => io::ErrorKind::HostUnreachable,
            Some("NetworkUnreachable") => io::ErrorKind::NetworkUnreachable,
            Some("ConnectionAborted") => io::ErrorKind::ConnectionAborted,
            Some("NotConnected") => io::ErrorKind::NotConnected,
            Some("AddrInUse") => io::ErrorKind::AddrInUse,
            Some("AddrNotAvailable") => io::ErrorKind::AddrNotAvailable,
            Some("NetworkDown") => io::ErrorKind::NetworkDown,
            Some("BrokenPipe") => io::ErrorKind::BrokenPipe,
            Some("AlreadyExists") => io::ErrorKind::AlreadyExists,
            Some("WouldBlock") => io::ErrorKind::WouldBlock,
            Some("NotADirectory") => io::ErrorKind::NotADirectory,
            Some("IsADirectory") => io::ErrorKind::IsADirectory,
            Some("DirectoryNotEmpty") => io::ErrorKind::DirectoryNotEmpty,
            Some("ReadOnlyFilesystem") => io::ErrorKind::ReadOnlyFilesystem,
            Some("StaleNetworkFileHandle") => io::ErrorKind::StaleNetworkFileHandle,
            Some("InvalidInput") => io::ErrorKind::InvalidInput,
            Some("InvalidData") => io::ErrorKind::InvalidData,
            Some("TimedOut") => io::ErrorKind::TimedOut,
            Some("WriteZero") => io::ErrorKind::WriteZero,
            Some("StorageFull") => io::ErrorKind::StorageFull,
            Some("NotSeekable") => io::ErrorKind::NotSeekable,
            Some("QuotaExceeded") => io::ErrorKind::QuotaExceeded,
            Some("FileTooLarge") => io::ErrorKind::FileTooLarge,
            Some("ResourceBusy") => io::ErrorKind::ResourceBusy,
            Some("ExecutableFileBusy") => io::ErrorKind::ExecutableFileBusy,
            Some("Deadlock") => io::ErrorKind::Deadlock,
            Some("CrossesDevices") => io::ErrorKind::CrossesDevices,
            Some("TooManyLinks") => io::ErrorKind::TooManyLinks,
            Some("InvalidFilename") => io::ErrorKind::InvalidFilename,
            Some("ArgumentListTooLong") => io::ErrorKind::ArgumentListTooLong,
            Some("Interrupted") => io::ErrorKind::Interrupted,
            Some("Unsupported") => io::ErrorKind::Unsupported,
            Some("UnexpectedEof") => io::ErrorKind::UnexpectedEof,
            Some("OutOfMemory") => io::ErrorKind::OutOfMemory,
            Some("Other") => io::ErrorKind::Other,
            _ => io::ErrorKind::Other,
        };

        Ok(io::Error::new(
            kind,
            io_err
                .message
                .unwrap_or_else(|| "unknown error".to_string()),
        ))
    }
}

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

    // A simple implementation of DCBReadResponseSync for testing
    struct TestReadResponse {
        events: Vec<DcbSequencedEvent>,
        current_index: usize,
        head_position: Option<u64>,
    }

    impl TestReadResponse {
        fn new(events: Vec<DcbSequencedEvent>, head_position: Option<u64>) -> Self {
            Self {
                events,
                current_index: 0,
                head_position,
            }
        }
    }

    impl Iterator for TestReadResponse {
        type Item = DcbResult<DcbSequencedEvent>;

        fn next(&mut self) -> Option<Self::Item> {
            if self.current_index < self.events.len() {
                let event = self.events[self.current_index].clone();
                self.current_index += 1;
                Some(Ok(event))
            } else {
                None
            }
        }
    }

    impl DcbReadResponseSync for TestReadResponse {
        fn head(&mut self) -> DcbResult<Option<u64>> {
            Ok(self.head_position)
        }

        fn collect_with_head(&mut self) -> DcbResult<(Vec<DcbSequencedEvent>, Option<u64>)> {
            todo!()
        }

        fn next_batch(&mut self) -> DcbResult<Vec<DcbSequencedEvent>> {
            let mut batch = Vec::new();
            while let Some(result) = self.next() {
                match result {
                    Ok(event) => batch.push(event),
                    Err(err) => {
                        panic!("{}", err);
                    }
                }
            }
            Ok(batch)
        }
    }

    #[test]
    fn test_dcb_read_response() {
        // Create some test events
        let event1 = DcbEvent {
            event_type: "test_event".to_string(),
            data: vec![1, 2, 3],
            tags: vec!["tag1".to_string(), "tag2".to_string()],
            uuid: None,
        };

        let event2 = DcbEvent {
            event_type: "another_event".to_string(),
            data: vec![4, 5, 6],
            tags: vec!["tag2".to_string(), "tag3".to_string()],
            uuid: None,
        };

        let seq_event1 = DcbSequencedEvent {
            event: event1,
            position: 1,
        };

        let seq_event2 = DcbSequencedEvent {
            event: event2,
            position: 2,
        };

        // Create a test response
        let mut response =
            TestReadResponse::new(vec![seq_event1.clone(), seq_event2.clone()], Some(2));

        // Test head position
        assert_eq!(response.head().unwrap(), Some(2));

        // Test iterator functionality
        assert_eq!(response.next().unwrap().unwrap().position, 1);
        assert_eq!(response.next().unwrap().unwrap().position, 2);
        assert!(response.next().is_none());
    }

    #[test]
    fn test_event_new() {
        let event1 = DcbEvent::default()
            .event_type("type1")
            .data(b"data1")
            .tags(["tagX"]);

        // println!("Event created with builder API:");
        // println!("  event_type: {}", event1.event_type);
        // println!("  data: {:?}", event1.data);
        // println!("  tags: {:?}", event1.tags);
        // println!("  uuid: {:?}", event1.uuid);

        // Verify the fields match expectations
        assert_eq!(event1.event_type, "type1");
        assert_eq!(event1.data, b"data1".to_vec());
        assert_eq!(event1.tags, vec!["tagX".to_string()]);
        assert_eq!(event1.uuid, None);

        // Test with multiple tags
        let event2 = DcbEvent::default()
            .event_type("type2")
            .data(b"data2")
            .tags(["tag1", "tag2", "tag3"]);
        assert_eq!(event2.tags.len(), 3);

        // Test without data or tags
        let event3 = DcbEvent::default().event_type("type3");
        assert_eq!(event3.data.len(), 0);
        assert_eq!(event3.tags.len(), 0);

        // Test DCBQueryItem builder
        let query_item = DcbQueryItem::new()
            .types(["type1", "type2"])
            .tags(["tagA", "tagB"]);
        assert_eq!(query_item.types.len(), 2);
        assert_eq!(query_item.tags.len(), 2);

        // Test DCBQuery builder
        let query = DcbQuery::new().item(query_item);
        assert_eq!(query.items.len(), 1);

        println!("\nAll builder API tests passed!");
    }
}