backbone-integrations 0.6.0

//! Event Store
//!
//! Generated by metaphor-schema. Do not edit manually.
//!
//! Provides persistence for domain events.

use std::pin::Pin;
use std::sync::Arc;

use anyhow::Result;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use futures::Stream;
use sqlx::PgPool;
use uuid::Uuid;

use super::event_envelope::{EventEnvelope, StoredEvent};

// ============================================================================
// EVENT STORE TRAIT
// ============================================================================

/// Trait for event store implementations
#[async_trait]
pub trait EventStore: Send + Sync {
    /// Append events to an aggregate stream
    async fn append(
        &self,
        aggregate_id: Uuid,
        aggregate_type: &str,
        events: Vec<StoredEvent>,
        expected_version: Option<i64>,
    ) -> Result<i64>;

    /// Load all events for an aggregate
    async fn load(&self, aggregate_id: Uuid) -> Result<Vec<StoredEvent>>;

    /// Load events from a specific version
    async fn load_from(&self, aggregate_id: Uuid, from_version: i64) -> Result<Vec<StoredEvent>>;

    /// Get the current version of an aggregate
    async fn get_version(&self, aggregate_id: Uuid) -> Result<i64>;

    /// Stream all events (for projections)
    async fn stream_all(&self, from_position: i64) -> Result<Pin<Box<dyn Stream<Item = StoredEvent> + Send>>>;

    /// Stream events by type
    async fn stream_by_type(&self, event_type: &str, from_position: i64) -> Result<Pin<Box<dyn Stream<Item = StoredEvent> + Send>>>;

    /// Stream events by aggregate type (for projectors)
    async fn stream_by_aggregate_type(&self, aggregate_type: &str, from_position: i64) -> Result<Pin<Box<dyn Stream<Item = StoredEvent> + Send>>>;

    /// Get the last processed sequence for a projector
    async fn get_projector_position(&self, projector_name: &str) -> Result<i64>;

    /// Save the last processed sequence for a projector
    async fn save_projector_position(&self, projector_name: &str, position: i64) -> Result<()>;
}

// ============================================================================
// POSTGRESQL EVENT STORE
// ============================================================================

/// PostgreSQL-based event store implementation
pub struct PostgresEventStore {
    pool: PgPool,
    table_name: String,
}

impl PostgresEventStore {
    /// Create a new PostgreSQL event store
    pub fn new(pool: PgPool) -> Self {
        Self {
            pool,
            table_name: "integrations.domain_events".to_string(),
        }
    }

    /// Create with custom table name
    pub fn with_table_name(pool: PgPool, table_name: impl Into<String>) -> Self {
        Self {
            pool,
            table_name: table_name.into(),
        }
    }

    /// Get the database pool
    pub fn pool(&self) -> &PgPool {
        &self.pool
    }
}

#[async_trait]
impl EventStore for PostgresEventStore {
    async fn append(
        &self,
        aggregate_id: Uuid,
        aggregate_type: &str,
        events: Vec<StoredEvent>,
        expected_version: Option<i64>,
    ) -> Result<i64> {
        // Check optimistic concurrency if expected_version provided
        if let Some(expected) = expected_version {
            let current = self.get_version(aggregate_id).await?;
            if current != expected {
                anyhow::bail!("Concurrency conflict: expected version {}, got {}", expected, current);
            }
        }

        let mut tx = self.pool.begin().await?;
        let mut last_sequence = 0i64;

        for event in events {
            let query = format!(
                "INSERT INTO {} (event_id, aggregate_id, aggregate_type, sequence, event_type, event_version, payload, metadata, occurred_at, stored_at) 
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NOW()) RETURNING sequence",
                self.table_name
            );

            last_sequence = sqlx::query_scalar::<_, i64>(&query)
                .bind(event.event_id)
                .bind(aggregate_id)
                .bind(aggregate_type)
                .bind(event.sequence)
                .bind(&event.event_type)
                .bind(event.event_version)
                .bind(&event.payload)
                .bind(&event.metadata)
                .bind(event.occurred_at)
                .fetch_one(&mut *tx)
                .await?;
        }

        tx.commit().await?;
        Ok(last_sequence)
    }

    async fn load(&self, aggregate_id: Uuid) -> Result<Vec<StoredEvent>> {
        let query = format!(
            "SELECT * FROM {} WHERE aggregate_id = $1 ORDER BY sequence ASC",
            self.table_name
        );

        let events = sqlx::query_as::<_, StoredEvent>(&query)
            .bind(aggregate_id)
            .fetch_all(&self.pool)
            .await?;

        Ok(events)
    }

    async fn load_from(&self, aggregate_id: Uuid, from_version: i64) -> Result<Vec<StoredEvent>> {
        let query = format!(
            "SELECT * FROM {} WHERE aggregate_id = $1 AND sequence > $2 ORDER BY sequence ASC",
            self.table_name
        );

        let events = sqlx::query_as::<_, StoredEvent>(&query)
            .bind(aggregate_id)
            .bind(from_version)
            .fetch_all(&self.pool)
            .await?;

        Ok(events)
    }

    async fn get_version(&self, aggregate_id: Uuid) -> Result<i64> {
        let query = format!(
            "SELECT COALESCE(MAX(sequence), 0) FROM {} WHERE aggregate_id = $1",
            self.table_name
        );

        let version = sqlx::query_scalar::<_, i64>(&query)
            .bind(aggregate_id)
            .fetch_one(&self.pool)
            .await?;

        Ok(version)
    }

    async fn stream_all(&self, from_position: i64) -> Result<Pin<Box<dyn Stream<Item = StoredEvent> + Send>>> {
        let query = format!(
            "SELECT * FROM {} WHERE sequence > $1 ORDER BY sequence ASC",
            self.table_name
        );

        let rows = sqlx::query_as::<_, StoredEvent>(&query)
            .bind(from_position)
            .fetch_all(&self.pool)
            .await?;

        Ok(Box::pin(futures::stream::iter(rows)))
    }

    async fn stream_by_type(&self, event_type: &str, from_position: i64) -> Result<Pin<Box<dyn Stream<Item = StoredEvent> + Send>>> {
        let query = format!(
            "SELECT * FROM {} WHERE event_type = $1 AND sequence > $2 ORDER BY sequence ASC",
            self.table_name
        );

        let rows = sqlx::query_as::<_, StoredEvent>(&query)
            .bind(event_type)
            .bind(from_position)
            .fetch_all(&self.pool)
            .await?;

        Ok(Box::pin(futures::stream::iter(rows)))
    }

    async fn stream_by_aggregate_type(&self, aggregate_type: &str, from_position: i64) -> Result<Pin<Box<dyn Stream<Item = StoredEvent> + Send>>> {
        let query = format!(
            "SELECT * FROM {} WHERE aggregate_type = $1 AND sequence > $2 ORDER BY sequence ASC",
            self.table_name
        );

        let rows = sqlx::query_as::<_, StoredEvent>(&query)
            .bind(aggregate_type)
            .bind(from_position)
            .fetch_all(&self.pool)
            .await?;

        Ok(Box::pin(futures::stream::iter(rows)))
    }

    async fn get_projector_position(&self, projector_name: &str) -> Result<i64> {
        let query = "SELECT COALESCE(
            (SELECT last_position FROM integrations.projector_positions WHERE projector_name = $1),
            0
        )";

        let position = sqlx::query_scalar::<_, i64>(query)
            .bind(projector_name)
            .fetch_one(&self.pool)
            .await?;

        Ok(position)
    }

    async fn save_projector_position(&self, projector_name: &str, position: i64) -> Result<()> {
        let query = "INSERT INTO integrations.projector_positions (projector_name, last_position, updated_at)
            VALUES ($1, $2, NOW())
            ON CONFLICT (projector_name) DO UPDATE SET
            last_position = EXCLUDED.last_position,
            updated_at = EXCLUDED.updated_at";

        sqlx::query(query)
            .bind(projector_name)
            .bind(position)
            .execute(&self.pool)
            .await?;

        Ok(())
    }
}

// <<< CUSTOM EVENT STORE START >>>
// Add custom event store implementations here
// <<< CUSTOM EVENT STORE END >>>