stateset_embedded/events/

store.rs

//! Event store implementations for persisting events

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use stateset_core::{CommerceEvent, EventStore, Result};
use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use uuid::Uuid;

/// Stored event with metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(super) struct StoredEvent {
    /// Sequence number (monotonically increasing)
    pub sequence: u64,
    /// Unique event ID
    pub id: Uuid,
    /// Event type string
    pub event_type: String,
    /// Aggregate type (e.g., "order", "customer")
    pub aggregate_type: Option<String>,
    /// Aggregate ID
    pub aggregate_id: Option<String>,
    /// Serialized event data
    pub data: String,
    /// When the event was stored
    pub stored_at: DateTime<Utc>,
}

/// In-memory event store for testing
pub struct InMemoryEventStore {
    events: Arc<RwLock<VecDeque<StoredEvent>>>,
    sequence: AtomicU64,
    max_events: usize,
}

impl std::fmt::Debug for InMemoryEventStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("InMemoryEventStore")
            .field("max_events", &self.max_events)
            .finish_non_exhaustive()
    }
}

impl InMemoryEventStore {
    /// Create a new in-memory event store
    #[must_use]
    pub fn new(max_events: usize) -> Self {
        Self {
            events: Arc::new(RwLock::new(VecDeque::with_capacity(max_events))),
            sequence: AtomicU64::new(0),
            max_events,
        }
    }

    /// Get aggregate info from event
    fn extract_aggregate(event: &CommerceEvent) -> (Option<String>, Option<String>) {
        match event {
            CommerceEvent::OrderCreated { order_id, .. }
            | CommerceEvent::OrderStatusChanged { order_id, .. }
            | CommerceEvent::OrderPaymentStatusChanged { order_id, .. }
            | CommerceEvent::OrderFulfillmentStatusChanged { order_id, .. }
            | CommerceEvent::OrderCancelled { order_id, .. }
            | CommerceEvent::OrderItemAdded { order_id, .. }
            | CommerceEvent::OrderItemRemoved { order_id, .. } => {
                (Some("order".to_string()), Some(order_id.to_string()))
            }
            CommerceEvent::CustomerCreated { customer_id, .. }
            | CommerceEvent::CustomerUpdated { customer_id, .. }
            | CommerceEvent::CustomerStatusChanged { customer_id, .. }
            | CommerceEvent::CustomerAddressAdded { customer_id, .. } => {
                (Some("customer".to_string()), Some(customer_id.to_string()))
            }
            CommerceEvent::ProductCreated { product_id, .. }
            | CommerceEvent::ProductUpdated { product_id, .. }
            | CommerceEvent::ProductStatusChanged { product_id, .. } => {
                (Some("product".to_string()), Some(product_id.to_string()))
            }
            CommerceEvent::ProductVariantAdded { variant_id, .. }
            | CommerceEvent::ProductVariantUpdated { variant_id, .. } => {
                (Some("variant".to_string()), Some(variant_id.to_string()))
            }
            CommerceEvent::CustomObjectTypeCreated { type_id, .. }
            | CommerceEvent::CustomObjectTypeUpdated { type_id, .. }
            | CommerceEvent::CustomObjectTypeDeleted { type_id, .. } => {
                (Some("custom_object_type".to_string()), Some(type_id.to_string()))
            }
            CommerceEvent::CustomObjectCreated { object_id, .. }
            | CommerceEvent::CustomObjectUpdated { object_id, .. }
            | CommerceEvent::CustomObjectDeleted { object_id, .. } => {
                (Some("custom_object".to_string()), Some(object_id.to_string()))
            }
            CommerceEvent::InventoryItemCreated { item_id, .. }
            | CommerceEvent::InventoryAdjusted { item_id, .. } => {
                (Some("inventory".to_string()), Some(item_id.to_string()))
            }
            CommerceEvent::InventoryReserved { reservation_id, .. }
            | CommerceEvent::InventoryReservationReleased { reservation_id, .. }
            | CommerceEvent::InventoryReservationConfirmed { reservation_id, .. } => {
                (Some("reservation".to_string()), Some(reservation_id.to_string()))
            }
            CommerceEvent::LowStockAlert { sku, .. } => {
                (Some("inventory".to_string()), Some(sku.clone()))
            }
            CommerceEvent::ReturnRequested { return_id, .. }
            | CommerceEvent::ReturnStatusChanged { return_id, .. }
            | CommerceEvent::ReturnApproved { return_id, .. }
            | CommerceEvent::ReturnRejected { return_id, .. }
            | CommerceEvent::ReturnCompleted { return_id, .. }
            | CommerceEvent::RefundIssued { return_id, .. } => {
                (Some("return".to_string()), Some(return_id.to_string()))
            }
            // x402 Payment Intent events
            CommerceEvent::X402IntentCreated { intent_id, .. }
            | CommerceEvent::X402IntentSigned { intent_id, .. }
            | CommerceEvent::X402IntentSequenced { intent_id, .. }
            | CommerceEvent::X402IntentSettled { intent_id, .. }
            | CommerceEvent::X402IntentFailed { intent_id, .. }
            | CommerceEvent::X402IntentExpired { intent_id, .. } => {
                (Some("x402_intent".to_string()), Some(intent_id.to_string()))
            }
            // Agent Card events
            CommerceEvent::AgentCardCreated { agent_id, .. }
            | CommerceEvent::AgentCardVerified { agent_id, .. }
            | CommerceEvent::AgentCardSuspended { agent_id, .. }
            | CommerceEvent::AgentCardReactivated { agent_id, .. } => {
                (Some("agent_card".to_string()), Some(agent_id.to_string()))
            }
            // A2A Commerce events
            CommerceEvent::A2AQuoteRequested { quote_id, .. }
            | CommerceEvent::A2AQuoteAccepted { quote_id, .. }
            | CommerceEvent::A2AQuoteRejected { quote_id, .. } => {
                (Some("a2a_quote".to_string()), Some(quote_id.to_string()))
            }
            CommerceEvent::A2APurchaseInitiated { purchase_id, .. }
            | CommerceEvent::A2APurchasePaid { purchase_id, .. }
            | CommerceEvent::A2ADeliveryConfirmed { purchase_id, .. } => {
                (Some("a2a_purchase".to_string()), Some(purchase_id.to_string()))
            }
            // Cart events
            CommerceEvent::CartCreated { cart_id, .. }
            | CommerceEvent::CartItemAdded { cart_id, .. }
            | CommerceEvent::CartStatusChanged { cart_id, .. }
            | CommerceEvent::CartCheckoutCompleted { cart_id, .. } => {
                (Some("cart".to_string()), Some(cart_id.to_string()))
            }
            // Payment events
            CommerceEvent::PaymentCreated { payment_id, .. }
            | CommerceEvent::PaymentStatusChanged { payment_id, .. }
            | CommerceEvent::PaymentCompleted { payment_id, .. } => {
                (Some("payment".to_string()), Some(payment_id.to_string()))
            }
            // Shipment events
            CommerceEvent::ShipmentCreated { shipment_id, .. }
            | CommerceEvent::ShipmentDelivered { shipment_id, .. } => {
                (Some("shipment".to_string()), Some(shipment_id.to_string()))
            }
            // Invoice events
            CommerceEvent::InvoiceCreated { invoice_id, .. }
            | CommerceEvent::InvoiceStatusChanged { invoice_id, .. } => {
                (Some("invoice".to_string()), Some(invoice_id.to_string()))
            }
            // Subscription events
            CommerceEvent::SubscriptionCreated { subscription_id, .. }
            | CommerceEvent::SubscriptionStatusChanged { subscription_id, .. }
            | CommerceEvent::SubscriptionRenewed { subscription_id, .. }
            | CommerceEvent::SubscriptionCancelled { subscription_id, .. } => {
                (Some("subscription".to_string()), Some(subscription_id.to_string()))
            }
            // Gift card events
            CommerceEvent::GiftCardCreated { gift_card_id, .. }
            | CommerceEvent::GiftCardRedeemed { gift_card_id, .. } => {
                (Some("gift_card".to_string()), Some(gift_card_id.to_string()))
            }
            // Store credit events
            CommerceEvent::StoreCreditIssued { store_credit_id, .. }
            | CommerceEvent::StoreCreditApplied { store_credit_id, .. } => {
                (Some("store_credit".to_string()), Some(store_credit_id.to_string()))
            }
            // Loyalty events
            CommerceEvent::LoyaltyPointsEarned { program_id, .. }
            | CommerceEvent::LoyaltyPointsRedeemed { program_id, .. } => {
                (Some("loyalty".to_string()), Some(program_id.to_string()))
            }
            // Fixed asset events
            CommerceEvent::FixedAssetPlacedInService { asset_id, .. }
            | CommerceEvent::FixedAssetDisposed { asset_id, .. }
            | CommerceEvent::FixedAssetWrittenOff { asset_id, .. }
            | CommerceEvent::DepreciationPosted { asset_id, .. } => {
                (Some("fixed_asset".to_string()), Some(asset_id.to_string()))
            }
            // Revenue recognition events
            CommerceEvent::RevenueRecognized { obligation_id, .. } => {
                (Some("performance_obligation".to_string()), Some(obligation_id.to_string()))
            }
            CommerceEvent::RevenueContractCompleted { contract_id, .. } => {
                (Some("revenue_contract".to_string()), Some(contract_id.to_string()))
            }
            // Warehouse events
            CommerceEvent::CycleCountCompleted { cycle_count_id, .. } => {
                (Some("cycle_count".to_string()), Some(cycle_count_id.to_string()))
            }
            // Accounts payable events
            CommerceEvent::ThreeWayMatchVarianceDetected { bill_id, .. } => {
                (Some("bill".to_string()), Some(bill_id.to_string()))
            }
            // General ledger events
            CommerceEvent::FxRevaluationPosted { as_of_date, .. } => {
                (Some("fx_revaluation".to_string()), Some(as_of_date.to_string()))
            }
            CommerceEvent::MonthEndCloseCompleted { period_id, .. } => {
                (Some("gl_period".to_string()), Some(period_id.to_string()))
            }
        }
    }
}

impl EventStore for InMemoryEventStore {
    fn append(&self, event: &CommerceEvent) -> Result<u64> {
        let (aggregate_type, aggregate_id) = Self::extract_aggregate(event);
        let data = event.to_json().map_err(|e| {
            stateset_core::CommerceError::Internal(format!("Failed to serialize event: {e}"))
        })?;

        let seq = self.sequence.fetch_add(1, Ordering::Relaxed) + 1;

        let stored = StoredEvent {
            sequence: seq,
            id: Uuid::new_v4(),
            event_type: event.event_type().to_string(),
            aggregate_type,
            aggregate_id,
            data,
            stored_at: Utc::now(),
        };

        let mut events = self.events.write().map_err(|_| {
            stateset_core::CommerceError::Internal(
                "InMemoryEventStore events lock poisoned".to_string(),
            )
        })?;
        if events.len() >= self.max_events {
            events.pop_front();
        }
        events.push_back(stored);

        Ok(seq)
    }

    fn get_events_since(&self, sequence: u64, limit: u32) -> Result<Vec<(u64, CommerceEvent)>> {
        let events = self.events.read().map_err(|_| {
            stateset_core::CommerceError::Internal(
                "InMemoryEventStore events lock poisoned".to_string(),
            )
        })?;
        let result = events
            .iter()
            .filter(|e| e.sequence > sequence)
            .take(limit as usize)
            .map(|e| {
                CommerceEvent::from_json(&e.data).map(|event| (e.sequence, event)).map_err(|err| {
                    stateset_core::CommerceError::Internal(format!(
                        "Failed to deserialize event {}: {}",
                        e.id, err
                    ))
                })
            })
            .collect::<Result<Vec<_>>>()?;
        Ok(result)
    }

    fn get_events_for_aggregate(
        &self,
        aggregate_type: &str,
        aggregate_id: &str,
    ) -> Result<Vec<CommerceEvent>> {
        let events = self.events.read().map_err(|_| {
            stateset_core::CommerceError::Internal(
                "InMemoryEventStore events lock poisoned".to_string(),
            )
        })?;
        let result = events
            .iter()
            .filter(|e| {
                e.aggregate_type.as_deref() == Some(aggregate_type)
                    && e.aggregate_id.as_deref() == Some(aggregate_id)
            })
            .map(|e| {
                CommerceEvent::from_json(&e.data).map_err(|err| {
                    stateset_core::CommerceError::Internal(format!(
                        "Failed to deserialize event {}: {}",
                        e.id, err
                    ))
                })
            })
            .collect::<Result<Vec<_>>>()?;
        Ok(result)
    }

    fn latest_sequence(&self) -> Result<u64> {
        Ok(self.sequence.load(Ordering::Relaxed))
    }
}

/// SQLite event store implementation
#[cfg(feature = "sqlite-events")]
pub struct SqliteEventStore {
    pool: r2d2::Pool<r2d2_sqlite::SqliteConnectionManager>,
}

#[cfg(feature = "sqlite-events")]
impl std::fmt::Debug for SqliteEventStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SqliteEventStore").finish_non_exhaustive()
    }
}

#[cfg(feature = "sqlite-events")]
impl SqliteEventStore {
    /// Create a new SQLite event store
    pub fn new(pool: r2d2::Pool<r2d2_sqlite::SqliteConnectionManager>) -> Result<Self> {
        let store = Self { pool };
        store.create_table()?;
        Ok(store)
    }

    fn create_table(&self) -> Result<()> {
        let conn = self.pool.get().map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to get connection: {}", e))
        })?;

        conn.execute(
            r#"
            CREATE TABLE IF NOT EXISTS commerce_events (
                sequence INTEGER PRIMARY KEY AUTOINCREMENT,
                id TEXT NOT NULL UNIQUE,
                event_type TEXT NOT NULL,
                aggregate_type TEXT,
                aggregate_id TEXT,
                data TEXT NOT NULL,
                stored_at TEXT NOT NULL
            )
            "#,
            [],
        )
        .map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to create table: {}", e))
        })?;

        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_events_aggregate ON commerce_events(aggregate_type, aggregate_id)",
            [],
        ).map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to create index: {}", e))
        })?;

        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_events_type ON commerce_events(event_type)",
            [],
        )
        .map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to create index: {}", e))
        })?;

        Ok(())
    }
}

#[cfg(feature = "sqlite-events")]
impl EventStore for SqliteEventStore {
    fn append(&self, event: &CommerceEvent) -> Result<u64> {
        let conn = self.pool.get().map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to get connection: {}", e))
        })?;

        let (aggregate_type, aggregate_id) = InMemoryEventStore::extract_aggregate(event);
        let data = event.to_json().map_err(|e| {
            stateset_core::CommerceError::Internal(format!("Failed to serialize event: {}", e))
        })?;

        conn.execute(
            r#"
            INSERT INTO commerce_events (id, event_type, aggregate_type, aggregate_id, data, stored_at)
            VALUES (?1, ?2, ?3, ?4, ?5, ?6)
            "#,
            rusqlite::params![
                Uuid::new_v4().to_string(),
                event.event_type(),
                aggregate_type,
                aggregate_id,
                data,
                Utc::now().to_rfc3339(),
            ],
        )
        .map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to insert event: {}", e))
        })?;

        let sequence = conn.last_insert_rowid() as u64;
        Ok(sequence)
    }

    fn get_events_since(&self, sequence: u64, limit: u32) -> Result<Vec<(u64, CommerceEvent)>> {
        let conn = self.pool.get().map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to get connection: {}", e))
        })?;

        let mut stmt = conn
            .prepare(
                r#"
            SELECT sequence, data FROM commerce_events
            WHERE sequence > ?1
            ORDER BY sequence ASC
            LIMIT ?2
            "#,
            )
            .map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!(
                    "Failed to prepare statement: {}",
                    e
                ))
            })?;

        let rows = stmt
            .query_map(rusqlite::params![sequence as i64, limit], |row| {
                let seq: i64 = row.get(0)?;
                let data: String = row.get(1)?;
                Ok((seq as u64, data))
            })
            .map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!(
                    "Failed to query events: {}",
                    e
                ))
            })?;

        let mut events = Vec::new();
        for row in rows {
            let (seq, data) = row.map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!("Failed to read row: {}", e))
            })?;
            let event = CommerceEvent::from_json(&data).map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!(
                    "Failed to deserialize event at sequence {}: {}",
                    seq, e
                ))
            })?;
            events.push((seq, event));
        }

        Ok(events)
    }

    fn get_events_for_aggregate(
        &self,
        aggregate_type: &str,
        aggregate_id: &str,
    ) -> Result<Vec<CommerceEvent>> {
        let conn = self.pool.get().map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to get connection: {}", e))
        })?;

        let mut stmt = conn
            .prepare(
                r#"
            SELECT data FROM commerce_events
            WHERE aggregate_type = ?1 AND aggregate_id = ?2
            ORDER BY sequence ASC
            "#,
            )
            .map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!(
                    "Failed to prepare statement: {}",
                    e
                ))
            })?;

        let rows = stmt
            .query_map(rusqlite::params![aggregate_type, aggregate_id], |row| {
                let data: String = row.get(0)?;
                Ok(data)
            })
            .map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!(
                    "Failed to query events: {}",
                    e
                ))
            })?;

        let mut events = Vec::new();
        for row in rows {
            let data = row.map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!("Failed to read row: {}", e))
            })?;
            let event = CommerceEvent::from_json(&data).map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!(
                    "Failed to deserialize event for {}:{}: {}",
                    aggregate_type, aggregate_id, e
                ))
            })?;
            events.push(event);
        }

        Ok(events)
    }

    fn latest_sequence(&self) -> Result<u64> {
        let conn = self.pool.get().map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to get connection: {}", e))
        })?;

        let sequence: Option<i64> = conn
            .query_row("SELECT MAX(sequence) FROM commerce_events", [], |row| row.get(0))
            .map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!(
                    "Failed to get latest sequence: {}",
                    e
                ))
            })?;

        Ok(sequence.unwrap_or(0) as u64)
    }
}

/// PostgreSQL event store implementation
#[cfg(feature = "postgres")]
pub struct PostgresEventStore {
    pool: sqlx::PgPool,
}

#[cfg(feature = "postgres")]
impl std::fmt::Debug for PostgresEventStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PostgresEventStore").finish_non_exhaustive()
    }
}

#[cfg(feature = "postgres")]
impl PostgresEventStore {
    /// Create a new PostgreSQL event store
    pub async fn new(pool: sqlx::PgPool) -> Result<Self> {
        let store = Self { pool };
        store.create_table().await?;
        Ok(store)
    }

    async fn create_table(&self) -> Result<()> {
        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS commerce_events (
                sequence BIGSERIAL PRIMARY KEY,
                id UUID NOT NULL UNIQUE,
                event_type TEXT NOT NULL,
                aggregate_type TEXT,
                aggregate_id TEXT,
                data JSONB NOT NULL,
                stored_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
            )
            "#,
        )
        .execute(&self.pool)
        .await
        .map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to create table: {}", e))
        })?;

        sqlx::query(
            "CREATE INDEX IF NOT EXISTS idx_events_aggregate ON commerce_events(aggregate_type, aggregate_id)",
        )
        .execute(&self.pool)
        .await
        .map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to create index: {}", e))
        })?;

        sqlx::query("CREATE INDEX IF NOT EXISTS idx_events_type ON commerce_events(event_type)")
            .execute(&self.pool)
            .await
            .map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!(
                    "Failed to create index: {}",
                    e
                ))
            })?;

        Ok(())
    }

    /// Append an event (async version)
    pub async fn append_async(&self, event: &CommerceEvent) -> Result<u64> {
        let (aggregate_type, aggregate_id) = InMemoryEventStore::extract_aggregate(event);
        let data = serde_json::to_value(event).map_err(|e| {
            stateset_core::CommerceError::Internal(format!("Failed to serialize event: {}", e))
        })?;

        let row: (i64,) = sqlx::query_as(
            r#"
            INSERT INTO commerce_events (id, event_type, aggregate_type, aggregate_id, data)
            VALUES ($1, $2, $3, $4, $5)
            RETURNING sequence
            "#,
        )
        .bind(Uuid::new_v4())
        .bind(event.event_type())
        .bind(aggregate_type)
        .bind(aggregate_id)
        .bind(data)
        .fetch_one(&self.pool)
        .await
        .map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to insert event: {}", e))
        })?;

        Ok(row.0 as u64)
    }

    /// Get events since sequence (async version)
    pub async fn get_events_since_async(
        &self,
        sequence: u64,
        limit: u32,
    ) -> Result<Vec<(u64, CommerceEvent)>> {
        let rows: Vec<(i64, Uuid, serde_json::Value)> = sqlx::query_as(
            r#"
            SELECT sequence, id, data FROM commerce_events
            WHERE sequence > $1
            ORDER BY sequence ASC
            LIMIT $2
            "#,
        )
        .bind(sequence as i64)
        .bind(limit as i32)
        .fetch_all(&self.pool)
        .await
        .map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to query events: {}", e))
        })?;

        let mut events = Vec::new();
        for (seq, id, data) in rows {
            let event: CommerceEvent = serde_json::from_value(data).map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!(
                    "Failed to deserialize event {id} at sequence {seq}: {e}"
                ))
            })?;
            events.push((seq as u64, event));
        }

        Ok(events)
    }

    /// Get events for aggregate (async version)
    pub async fn get_events_for_aggregate_async(
        &self,
        aggregate_type: &str,
        aggregate_id: &str,
    ) -> Result<Vec<CommerceEvent>> {
        let rows: Vec<(i64, Uuid, serde_json::Value)> = sqlx::query_as(
            r#"
            SELECT sequence, id, data FROM commerce_events
            WHERE aggregate_type = $1 AND aggregate_id = $2
            ORDER BY sequence ASC
            "#,
        )
        .bind(aggregate_type)
        .bind(aggregate_id)
        .fetch_all(&self.pool)
        .await
        .map_err(|e| {
            stateset_core::CommerceError::DatabaseError(format!("Failed to query events: {}", e))
        })?;

        let mut events = Vec::new();
        for (seq, id, data) in rows {
            let event: CommerceEvent = serde_json::from_value(data).map_err(|e| {
                stateset_core::CommerceError::DatabaseError(format!(
                    "Failed to deserialize event {id} at sequence {seq}: {e}"
                ))
            })?;
            events.push(event);
        }

        Ok(events)
    }

    /// Get latest sequence (async version)
    pub async fn latest_sequence_async(&self) -> Result<u64> {
        let row: Option<(Option<i64>,)> =
            sqlx::query_as("SELECT MAX(sequence) FROM commerce_events")
                .fetch_optional(&self.pool)
                .await
                .map_err(|e| {
                    stateset_core::CommerceError::DatabaseError(format!(
                        "Failed to get latest sequence: {}",
                        e
                    ))
                })?;

        Ok(row.and_then(|(s,)| s).unwrap_or(0) as u64)
    }
}

#[cfg(feature = "postgres")]
fn block_on<F, T>(fut: F) -> Result<T>
where
    F: std::future::Future<Output = Result<T>>,
{
    if tokio::runtime::Handle::try_current().is_ok() {
        return Err(stateset_core::CommerceError::NotPermitted(
            "Blocking Postgres event store call inside an async runtime; use the async methods instead"
                .to_string(),
        ));
    }

    let rt = tokio::runtime::Runtime::new().map_err(|e| {
        stateset_core::CommerceError::Internal(format!("Failed to create runtime: {}", e))
    })?;
    rt.block_on(fut)
}

#[cfg(feature = "postgres")]
impl EventStore for PostgresEventStore {
    fn append(&self, event: &CommerceEvent) -> Result<u64> {
        block_on(self.append_async(event))
    }

    fn get_events_since(&self, sequence: u64, limit: u32) -> Result<Vec<(u64, CommerceEvent)>> {
        block_on(self.get_events_since_async(sequence, limit))
    }

    fn get_events_for_aggregate(
        &self,
        aggregate_type: &str,
        aggregate_id: &str,
    ) -> Result<Vec<CommerceEvent>> {
        block_on(self.get_events_for_aggregate_async(aggregate_type, aggregate_id))
    }

    fn latest_sequence(&self) -> Result<u64> {
        block_on(self.latest_sequence_async())
    }
}

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

    #[test]
    fn test_in_memory_store() {
        let store = InMemoryEventStore::new(100);

        let event = CommerceEvent::OrderCreated {
            order_id: stateset_core::OrderId::new(),
            customer_id: stateset_core::CustomerId::new(),
            total_amount: dec!(100.00),
            item_count: 2,
            timestamp: Utc::now(),
        };

        let seq = store.append(&event).unwrap();
        assert_eq!(seq, 1);

        let events = store.get_events_since(0, 10).unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].0, 1);
    }

    #[test]
    fn test_in_memory_store_max_events() {
        let store = InMemoryEventStore::new(2);

        for i in 0..5 {
            let event = CommerceEvent::CustomerCreated {
                customer_id: stateset_core::CustomerId::new(),
                email: format!("test{}@example.com", i),
                timestamp: Utc::now(),
            };
            store.append(&event).unwrap();
        }

        // Should only keep last 2 events
        let events = store.get_events_since(0, 10).unwrap();
        assert_eq!(events.len(), 2);
    }
}