evento_sql/
sql.rs

1//! Core SQL implementation for event sourcing.
2
3use std::ops::{Deref, DerefMut};
4
5#[cfg(feature = "mysql")]
6use sea_query::MysqlQueryBuilder;
7#[cfg(feature = "postgres")]
8use sea_query::PostgresQueryBuilder;
9#[cfg(feature = "sqlite")]
10use sea_query::SqliteQueryBuilder;
11use sea_query::{
12    Cond, Expr, ExprTrait, Func, Iden, IntoColumnRef, OnConflict, Query, SelectStatement,
13};
14use sea_query_sqlx::SqlxBinder;
15use sqlx::{Database, Pool};
16use ulid::Ulid;
17
18use evento_core::{
19    cursor::{self, Args, Cursor, Edge, PageInfo, ReadResult, Value},
20    EventFilter, Executor, WriteError,
21};
22
23/// Column identifiers for the `event` table.
24///
25/// Used with sea-query for type-safe SQL query construction.
26///
27/// # Columns
28///
29/// - `Id` - Event identifier (ULID format, VARCHAR(26))
30/// - `Name` - Event type name (VARCHAR(50))
31/// - `AggregatorType` - Aggregate root type (VARCHAR(50))
32/// - `AggregatorId` - Aggregate root instance ID (VARCHAR(26))
33/// - `Version` - Event sequence number within the aggregate
34/// - `Data` - Serialized event payload (BLOB, bitcode format)
35/// - `Metadata` - Serialized event metadata (BLOB, bitcode format)
36/// - `RoutingKey` - Optional routing key for partitioning (VARCHAR(50))
37/// - `Timestamp` - Event timestamp in seconds (BIGINT)
38/// - `TimestampSubsec` - Sub-second precision (BIGINT)
39#[derive(Iden, Clone)]
40pub enum Event {
41    /// The table name: `event`
42    Table,
43    /// Event ID column (ULID)
44    Id,
45    /// Event type name
46    Name,
47    /// Aggregate root type
48    AggregatorType,
49    /// Aggregate root instance ID
50    AggregatorId,
51    /// Event version/sequence number
52    Version,
53    /// Serialized event data
54    Data,
55    /// Serialized event metadata
56    Metadata,
57    /// Optional routing key
58    RoutingKey,
59    /// Timestamp in seconds
60    Timestamp,
61    /// Sub-second precision
62    TimestampSubsec,
63}
64
65/// Column identifiers for the `snapshot` table.
66///
67/// Used with sea-query for type-safe SQL query construction. The snapshot table
68/// backs projection snapshots via [`Executor::get_snapshot`], [`Executor::save_snapshot`],
69/// and [`Executor::delete_snapshot`].
70#[derive(Iden)]
71pub enum Snapshot {
72    /// The table name: `snapshot`
73    Table,
74    /// Snapshot ID
75    Id,
76    /// Snapshot type
77    Type,
78    /// Event stream cursor position
79    Cursor,
80    /// Revision identifier
81    Revision,
82    /// Serialized snapshot data
83    Data,
84    /// Creation timestamp
85    CreatedAt,
86    /// Last update timestamp
87    UpdatedAt,
88}
89
90/// Column identifiers for the `subscriber` table.
91///
92/// Used with sea-query for type-safe SQL query construction.
93///
94/// # Columns
95///
96/// - `Key` - Subscriber identifier (primary key)
97/// - `WorkerId` - ULID of the current worker processing events
98/// - `Cursor` - Current position in the event stream
99/// - `Lag` - Number of events behind the latest
100/// - `Enabled` - Whether the subscription is active
101/// - `CreatedAt` / `UpdatedAt` - Timestamps
102#[derive(Iden)]
103pub enum Subscriber {
104    /// The table name: `subscriber`
105    Table,
106    /// Subscriber key (primary key)
107    Key,
108    /// Current worker ID (ULID)
109    WorkerId,
110    /// Current cursor position
111    Cursor,
112    /// Event lag counter
113    Lag,
114    /// Whether subscription is enabled
115    Enabled,
116    /// Creation timestamp
117    CreatedAt,
118    /// Last update timestamp
119    UpdatedAt,
120}
121
122/// Type alias for MySQL executor.
123///
124/// Equivalent to `Sql<sqlx::MySql>`.
125#[cfg(feature = "mysql")]
126pub type MySql = Sql<sqlx::MySql>;
127
128/// Read-write executor pair for MySQL.
129///
130/// Used in CQRS patterns where you may have separate read and write connections.
131#[cfg(feature = "mysql")]
132pub type RwMySql = evento_core::Rw<MySql, MySql>;
133
134/// Type alias for PostgreSQL executor.
135///
136/// Equivalent to `Sql<sqlx::Postgres>`.
137#[cfg(feature = "postgres")]
138pub type Postgres = Sql<sqlx::Postgres>;
139
140/// Read-write executor pair for PostgreSQL.
141///
142/// Used in CQRS patterns where you may have separate read and write connections.
143#[cfg(feature = "postgres")]
144pub type RwPostgres = evento_core::Rw<Postgres, Postgres>;
145
146/// Type alias for SQLite executor.
147///
148/// Equivalent to `Sql<sqlx::Sqlite>`.
149#[cfg(feature = "sqlite")]
150pub type Sqlite = Sql<sqlx::Sqlite>;
151
152/// Read-write executor pair for SQLite.
153///
154/// Used in CQRS patterns where you may have separate read and write connections.
155#[cfg(feature = "sqlite")]
156pub type RwSqlite = evento_core::Rw<Sqlite, Sqlite>;
157
158/// SQL database executor for event sourcing operations.
159///
160/// A generic wrapper around a SQLx connection pool that implements the
161/// [`Executor`](evento_core::Executor) trait for storing and querying events.
162///
163/// # Type Parameters
164///
165/// - `DB` - The SQLx database type (e.g., `sqlx::Sqlite`, `sqlx::MySql`, `sqlx::Postgres`)
166///
167/// # Example
168///
169/// ```rust,ignore
170/// use evento_sql::Sql;
171/// use sqlx::sqlite::SqlitePoolOptions;
172///
173/// // Create a connection pool
174/// let pool = SqlitePoolOptions::new()
175///     .connect(":memory:")
176///     .await?;
177///
178/// // Convert to Sql executor
179/// let executor: Sql<sqlx::Sqlite> = pool.into();
180///
181/// // Or use the type alias
182/// let executor: evento_sql::Sqlite = pool.into();
183/// ```
184///
185/// # Executor Implementation
186///
187/// The `Sql` type implements [`Executor`](evento_core::Executor) with the following operations:
188///
189/// - **`read`** - Query events with filtering and cursor-based pagination
190/// - **`write`** - Persist events with optimistic concurrency control
191/// - **`get_subscriber_cursor`** - Get the current cursor position for a subscriber
192/// - **`is_subscriber_running`** - Check if a subscriber is active with a specific worker
193/// - **`upsert_subscriber`** - Create or update a subscriber record
194/// - **`acknowledge`** - Update subscriber cursor after processing events
195pub struct Sql<DB: Database>(Pool<DB>);
196
197impl<DB: Database> Sql<DB> {
198    fn build_sqlx<S: SqlxBinder>(statement: S) -> (String, sea_query_sqlx::SqlxValues) {
199        match DB::NAME {
200            #[cfg(feature = "sqlite")]
201            "SQLite" => statement.build_sqlx(SqliteQueryBuilder),
202            #[cfg(feature = "mysql")]
203            "MySQL" => statement.build_sqlx(MysqlQueryBuilder),
204            #[cfg(feature = "postgres")]
205            "PostgreSQL" => statement.build_sqlx(PostgresQueryBuilder),
206            name => panic!("'{name}' not supported, consider using SQLite, PostgreSQL or MySQL"),
207        }
208    }
209}
210
211#[async_trait::async_trait]
212impl<DB> Executor for Sql<DB>
213where
214    DB: Database,
215    for<'c> &'c mut DB::Connection: sqlx::Executor<'c, Database = DB>,
216    sea_query_sqlx::SqlxValues: sqlx::IntoArguments<DB>,
217    String: for<'r> sqlx::Decode<'r, DB> + sqlx::Type<DB>,
218    bool: for<'r> sqlx::Decode<'r, DB> + sqlx::Type<DB>,
219    Vec<u8>: for<'r> sqlx::Decode<'r, DB> + sqlx::Type<DB>,
220    i64: for<'r> sqlx::Decode<'r, DB> + sqlx::Type<DB>,
221    usize: sqlx::ColumnIndex<DB::Row>,
222    SqlEvent: for<'r> sqlx::FromRow<'r, DB::Row>,
223{
224    async fn read(
225        &self,
226        aggregators: Option<Vec<EventFilter>>,
227        routing_key: Option<evento_core::RoutingKey>,
228        args: Args,
229    ) -> anyhow::Result<ReadResult<evento_core::Event>> {
230        let statement = Query::select()
231            .columns([
232                Event::Id,
233                Event::Name,
234                Event::AggregatorType,
235                Event::AggregatorId,
236                Event::Version,
237                Event::Data,
238                Event::Metadata,
239                Event::RoutingKey,
240                Event::Timestamp,
241                Event::TimestampSubsec,
242            ])
243            .from(Event::Table)
244            .conditions(
245                aggregators.is_some(),
246                |q| {
247                    let Some(aggregators) = aggregators else {
248                        return;
249                    };
250
251                    let mut cond = Cond::any();
252
253                    for aggregator in aggregators {
254                        let mut aggregator_cond = Cond::all()
255                            .add(Expr::col(Event::AggregatorType).eq(aggregator.aggregate_type));
256
257                        if let Some(id) = aggregator.aggregate_id {
258                            aggregator_cond =
259                                aggregator_cond.add(Expr::col(Event::AggregatorId).eq(id));
260                        }
261
262                        if let Some(name) = aggregator.name {
263                            aggregator_cond = aggregator_cond.add(Expr::col(Event::Name).eq(name));
264                        }
265
266                        cond = cond.add(aggregator_cond);
267                    }
268
269                    q.and_where(cond.into());
270                },
271                |_| {},
272            )
273            .conditions(
274                matches!(routing_key, Some(evento_core::RoutingKey::Value(_))),
275                |q| {
276                    if let Some(evento_core::RoutingKey::Value(Some(ref routing_key))) = routing_key
277                    {
278                        q.and_where(Expr::col(Event::RoutingKey).eq(routing_key));
279                    }
280
281                    if let Some(evento_core::RoutingKey::Value(None)) = routing_key {
282                        q.and_where(Expr::col(Event::RoutingKey).is_null());
283                    }
284                },
285                |_q| {},
286            )
287            .to_owned();
288
289        Ok(Reader::new(statement)
290            .args(args)
291            .execute::<_, SqlEvent, _>(&self.0)
292            .await?
293            .map(|e| e.0))
294    }
295
296    async fn latest_timestamp(
297        &self,
298        aggregators: Option<Vec<EventFilter>>,
299        routing_key: Option<evento_core::RoutingKey>,
300    ) -> anyhow::Result<u64> {
301        let statement = Query::select()
302            .expr(Func::max(Expr::col(Event::Timestamp)))
303            .from(Event::Table)
304            .conditions(
305                aggregators.is_some(),
306                |q| {
307                    let Some(aggregators) = aggregators else {
308                        return;
309                    };
310
311                    let mut cond = Cond::any();
312
313                    for aggregator in aggregators {
314                        let mut aggregator_cond = Cond::all()
315                            .add(Expr::col(Event::AggregatorType).eq(aggregator.aggregate_type));
316
317                        if let Some(id) = aggregator.aggregate_id {
318                            aggregator_cond =
319                                aggregator_cond.add(Expr::col(Event::AggregatorId).eq(id));
320                        }
321
322                        if let Some(name) = aggregator.name {
323                            aggregator_cond = aggregator_cond.add(Expr::col(Event::Name).eq(name));
324                        }
325
326                        cond = cond.add(aggregator_cond);
327                    }
328
329                    q.and_where(cond.into());
330                },
331                |_| {},
332            )
333            .conditions(
334                matches!(routing_key, Some(evento_core::RoutingKey::Value(_))),
335                |q| {
336                    if let Some(evento_core::RoutingKey::Value(Some(ref routing_key))) = routing_key
337                    {
338                        q.and_where(Expr::col(Event::RoutingKey).eq(routing_key));
339                    }
340
341                    if let Some(evento_core::RoutingKey::Value(None)) = routing_key {
342                        q.and_where(Expr::col(Event::RoutingKey).is_null());
343                    }
344                },
345                |_q| {},
346            )
347            .to_owned();
348
349        let (sql, values) = Self::build_sqlx(statement);
350
351        let (ts,): (Option<i64>,) =
352            sqlx::query_as_with::<DB, (Option<i64>,), _>(sqlx::AssertSqlSafe(sql.as_str()), values)
353                .fetch_one(&self.0)
354                .await?;
355
356        Ok(ts.map(|v| if v < 0 { 0 } else { v as u64 }).unwrap_or(0))
357    }
358
359    async fn get_subscriber_cursor(&self, key: String) -> anyhow::Result<Option<Value>> {
360        let statement = Query::select()
361            .columns([Subscriber::Cursor])
362            .from(Subscriber::Table)
363            .and_where(Expr::col(Subscriber::Key).eq(Expr::value(key)))
364            .limit(1)
365            .to_owned();
366
367        let (sql, values) = Self::build_sqlx(statement);
368
369        let Some((cursor,)) = sqlx::query_as_with::<DB, (Option<String>,), _>(
370            sqlx::AssertSqlSafe(sql.as_str()),
371            values,
372        )
373        .fetch_optional(&self.0)
374        .await?
375        else {
376            return Ok(None);
377        };
378
379        Ok(cursor.map(|c| c.into()))
380    }
381
382    async fn is_subscriber_running(&self, key: String, worker_id: Ulid) -> anyhow::Result<bool> {
383        let statement = Query::select()
384            .columns([Subscriber::WorkerId, Subscriber::Enabled])
385            .from(Subscriber::Table)
386            .and_where(Expr::col(Subscriber::Key).eq(Expr::value(key)))
387            .limit(1)
388            .to_owned();
389
390        let (sql, values) = Self::build_sqlx(statement);
391
392        let (id, enabled) =
393            sqlx::query_as_with::<DB, (String, bool), _>(sqlx::AssertSqlSafe(sql.as_str()), values)
394                .fetch_one(&self.0)
395                .await?;
396
397        Ok(worker_id.to_string() == id && enabled)
398    }
399
400    async fn upsert_subscriber(&self, key: String, worker_id: Ulid) -> anyhow::Result<()> {
401        let statement = Query::insert()
402            .into_table(Subscriber::Table)
403            .columns([Subscriber::Key, Subscriber::WorkerId, Subscriber::Lag])
404            .values_panic([key.into(), worker_id.to_string().into(), 0.into()])
405            .on_conflict(
406                OnConflict::column(Subscriber::Key)
407                    .update_columns([Subscriber::WorkerId])
408                    .value(Subscriber::UpdatedAt, Expr::current_timestamp())
409                    .to_owned(),
410            )
411            .to_owned();
412
413        let (sql, values) = Self::build_sqlx(statement);
414
415        sqlx::query_with::<DB, _>(sqlx::AssertSqlSafe(sql.as_str()), values)
416            .execute(&self.0)
417            .await?;
418
419        Ok(())
420    }
421
422    async fn write(&self, events: Vec<evento_core::Event>) -> Result<(), WriteError> {
423        let mut statement = Query::insert()
424            .into_table(Event::Table)
425            .columns([
426                Event::Id,
427                Event::Name,
428                Event::Data,
429                Event::Metadata,
430                Event::AggregatorType,
431                Event::AggregatorId,
432                Event::Version,
433                Event::RoutingKey,
434                Event::Timestamp,
435                Event::TimestampSubsec,
436            ])
437            .to_owned();
438
439        for event in events {
440            let metadata = bitcode::encode(&event.metadata);
441            statement.values_panic([
442                event.id.to_string().into(),
443                event.name.into(),
444                event.data.into(),
445                metadata.into(),
446                event.aggregate_type.into(),
447                event.aggregate_id.into(),
448                event.version.into(),
449                event.routing_key.into(),
450                event.timestamp.into(),
451                event.timestamp_subsec.into(),
452            ]);
453        }
454
455        let (sql, values) = Self::build_sqlx(statement);
456
457        sqlx::query_with::<DB, _>(sqlx::AssertSqlSafe(sql.as_str()), values)
458            .execute(&self.0)
459            .await
460            .map_err(|err| {
461                let err_str = err.to_string();
462                if err_str.contains("(code: 2067)") {
463                    return WriteError::InvalidOriginalVersion;
464                }
465                if err_str.contains("1062 (23000): Duplicate entry") {
466                    return WriteError::InvalidOriginalVersion;
467                }
468                if err_str.contains("duplicate key value violates unique constraint") {
469                    return WriteError::InvalidOriginalVersion;
470                }
471                WriteError::Unknown(err.into())
472            })?;
473
474        Ok(())
475    }
476
477    async fn acknowledge(&self, key: String, cursor: Value, lag: u64) -> anyhow::Result<()> {
478        let statement = Query::update()
479            .table(Subscriber::Table)
480            .values([
481                (Subscriber::Cursor, cursor.0.into()),
482                (Subscriber::Lag, lag.into()),
483                (Subscriber::UpdatedAt, Expr::current_timestamp()),
484            ])
485            .and_where(Expr::col(Subscriber::Key).eq(key))
486            .to_owned();
487
488        let (sql, values) = Self::build_sqlx(statement);
489
490        sqlx::query_with::<DB, _>(sqlx::AssertSqlSafe(sql.as_str()), values)
491            .execute(&self.0)
492            .await?;
493
494        Ok(())
495    }
496
497    async fn get_snapshot(
498        &self,
499        aggregate_type: String,
500        aggregate_revision: String,
501        id: String,
502    ) -> anyhow::Result<Option<(Vec<u8>, Value)>> {
503        let statement = Query::select()
504            .columns([Snapshot::Data, Snapshot::Cursor])
505            .from(Snapshot::Table)
506            .and_where(Expr::col(Snapshot::Type).eq(Expr::value(aggregate_type)))
507            .and_where(Expr::col(Snapshot::Id).eq(Expr::value(id)))
508            .and_where(Expr::col(Snapshot::Revision).eq(Expr::value(aggregate_revision)))
509            .limit(1)
510            .to_owned();
511
512        let (sql, values) = Self::build_sqlx(statement);
513
514        Ok(sqlx::query_as_with::<DB, (Vec<u8>, String), _>(
515            sqlx::AssertSqlSafe(sql.as_str()),
516            values,
517        )
518        .fetch_optional(&self.0)
519        .await
520        .map(|res| res.map(|(data, cursor)| (data, cursor.into())))?)
521    }
522
523    async fn save_snapshot(
524        &self,
525        aggregate_type: String,
526        aggregate_revision: String,
527        id: String,
528        data: Vec<u8>,
529        cursor: Value,
530    ) -> anyhow::Result<()> {
531        let statement = Query::insert()
532            .into_table(Snapshot::Table)
533            .columns([
534                Snapshot::Type,
535                Snapshot::Id,
536                Snapshot::Cursor,
537                Snapshot::Revision,
538                Snapshot::Data,
539            ])
540            .values_panic([
541                aggregate_type.into(),
542                id.to_string().into(),
543                cursor.to_string().into(),
544                aggregate_revision.into(),
545                data.into(),
546            ])
547            .on_conflict(
548                OnConflict::columns([Snapshot::Type, Snapshot::Id])
549                    .update_columns([Snapshot::Data, Snapshot::Cursor, Snapshot::Revision])
550                    .value(Snapshot::UpdatedAt, Expr::current_timestamp())
551                    .to_owned(),
552            )
553            .to_owned();
554
555        let (sql, values) = Self::build_sqlx(statement);
556
557        sqlx::query_with::<DB, _>(sqlx::AssertSqlSafe(sql.as_str()), values)
558            .execute(&self.0)
559            .await?;
560
561        Ok(())
562    }
563
564    async fn delete_snapshot(&self, aggregate_type: String, id: String) -> anyhow::Result<()> {
565        let statement = Query::delete()
566            .from_table(Snapshot::Table)
567            .and_where(Expr::col(Snapshot::Type).eq(Expr::value(aggregate_type)))
568            .and_where(Expr::col(Snapshot::Id).eq(Expr::value(id)))
569            .to_owned();
570
571        let (sql, values) = Self::build_sqlx(statement);
572
573        sqlx::query_with::<DB, _>(sqlx::AssertSqlSafe(sql.as_str()), values)
574            .execute(&self.0)
575            .await?;
576
577        Ok(())
578    }
579}
580
581impl<D: Database> Clone for Sql<D> {
582    fn clone(&self) -> Self {
583        Self(self.0.clone())
584    }
585}
586
587impl<D: Database> From<Pool<D>> for Sql<D> {
588    fn from(value: Pool<D>) -> Self {
589        Self(value)
590    }
591}
592
593/// Query builder for reading events with cursor-based pagination.
594///
595/// `Reader` wraps a sea-query [`SelectStatement`] and adds support for:
596/// - Forward pagination (first N after cursor)
597/// - Backward pagination (last N before cursor)
598/// - Ascending/descending order
599///
600/// # Example
601///
602/// ```rust,ignore
603/// use evento_sql::{Reader, Event};
604/// use sea_query::Query;
605///
606/// let statement = Query::select()
607///     .columns([Event::Id, Event::Name, Event::Data])
608///     .from(Event::Table)
609///     .to_owned();
610///
611/// let result = Reader::new(statement)
612///     .forward(10, None)  // First 10 events
613///     .execute::<_, MyEvent, _>(&pool)
614///     .await?;
615///
616/// for edge in result.edges {
617///     println!("Event: {:?}, Cursor: {:?}", edge.node, edge.cursor);
618/// }
619///
620/// // Continue with next page
621/// if result.page_info.has_next_page {
622///     let next_result = Reader::new(statement)
623///         .forward(10, result.page_info.end_cursor)
624///         .execute::<_, MyEvent, _>(&pool)
625///         .await?;
626/// }
627/// ```
628///
629/// # Deref
630///
631/// `Reader` implements `Deref` and `DerefMut` to the underlying `SelectStatement`,
632/// allowing direct access to sea-query builder methods.
633pub struct Reader {
634    statement: SelectStatement,
635    args: Args,
636    order: cursor::Order,
637}
638
639impl Reader {
640    /// Creates a new reader from a sea-query select statement.
641    pub fn new(statement: SelectStatement) -> Self {
642        Self {
643            statement,
644            args: Args::default(),
645            order: cursor::Order::Asc,
646        }
647    }
648
649    /// Sets the sort order for results.
650    pub fn order(&mut self, order: cursor::Order) -> &mut Self {
651        self.order = order;
652
653        self
654    }
655
656    /// Sets descending sort order.
657    pub fn desc(&mut self) -> &mut Self {
658        self.order(cursor::Order::Desc)
659    }
660
661    /// Sets pagination arguments directly.
662    pub fn args(&mut self, args: Args) -> &mut Self {
663        self.args = args;
664
665        self
666    }
667
668    /// Configures backward pagination (last N before cursor).
669    ///
670    /// # Arguments
671    ///
672    /// - `last` - Number of items to return
673    /// - `before` - Optional cursor to paginate before
674    pub fn backward(&mut self, last: u16, before: Option<Value>) -> &mut Self {
675        self.args(Args {
676            last: Some(last),
677            before,
678            ..Default::default()
679        })
680    }
681
682    /// Configures forward pagination (first N after cursor).
683    ///
684    /// # Arguments
685    ///
686    /// - `first` - Number of items to return
687    /// - `after` - Optional cursor to paginate after
688    pub fn forward(&mut self, first: u16, after: Option<Value>) -> &mut Self {
689        self.args(Args {
690            first: Some(first),
691            after,
692            ..Default::default()
693        })
694    }
695
696    /// Executes the query and returns paginated results.
697    ///
698    /// # Type Parameters
699    ///
700    /// - `DB` - The SQLx database type
701    /// - `O` - The output row type (must implement `FromRow`, `Cursor`, and `Bind`)
702    /// - `E` - The executor type
703    ///
704    /// # Returns
705    ///
706    /// A [`ReadResult`](evento_core::cursor::ReadResult) containing edges with nodes and cursors,
707    /// plus pagination info.
708    pub async fn execute<'e, 'c: 'e, DB, O, E>(
709        &mut self,
710        executor: E,
711    ) -> anyhow::Result<ReadResult<O>>
712    where
713        DB: Database,
714        E: 'e + sqlx::Executor<'c, Database = DB>,
715        O: for<'r> sqlx::FromRow<'r, DB::Row>,
716        O: Cursor,
717        O: Send + Unpin,
718        O: Bind<Cursor = O>,
719        <<O as Bind>::I as IntoIterator>::IntoIter: DoubleEndedIterator,
720        <<O as Bind>::V as IntoIterator>::IntoIter: DoubleEndedIterator,
721        sea_query_sqlx::SqlxValues: sqlx::IntoArguments<DB>,
722    {
723        let limit = self.build_reader::<O, O>()?;
724
725        let (sql, values) = match DB::NAME {
726            #[cfg(feature = "sqlite")]
727            "SQLite" => self.statement.build_sqlx(SqliteQueryBuilder),
728            #[cfg(feature = "mysql")]
729            "MySQL" => self.build_sqlx(MysqlQueryBuilder),
730            #[cfg(feature = "postgres")]
731            "PostgreSQL" => self.build_sqlx(PostgresQueryBuilder),
732            name => panic!("'{name}' not supported, consider using SQLite, PostgreSQL or MySQL"),
733        };
734
735        let mut rows = sqlx::query_as_with::<DB, O, _>(sqlx::AssertSqlSafe(sql.as_str()), values)
736            .fetch_all(executor)
737            .await?;
738
739        let has_more = rows.len() > limit as usize;
740        if has_more {
741            rows.pop();
742        }
743
744        let mut edges = vec![];
745        for node in rows.into_iter() {
746            edges.push(Edge {
747                cursor: node.serialize_cursor()?,
748                node,
749            });
750        }
751
752        if self.args.is_backward() {
753            edges = edges.into_iter().rev().collect();
754        }
755
756        let page_info = if self.args.is_backward() {
757            let start_cursor = edges.first().map(|e| e.cursor.clone());
758
759            PageInfo {
760                has_previous_page: has_more,
761                has_next_page: false,
762                start_cursor,
763                end_cursor: None,
764            }
765        } else {
766            let end_cursor = edges.last().map(|e| e.cursor.clone());
767            PageInfo {
768                has_previous_page: false,
769                has_next_page: has_more,
770                start_cursor: None,
771                end_cursor,
772            }
773        };
774
775        Ok(ReadResult { edges, page_info })
776    }
777
778    fn build_reader<O: Cursor, B: Bind<Cursor = O>>(&mut self) -> Result<u16, cursor::CursorError>
779    where
780        B::T: Clone,
781        <<B as Bind>::I as IntoIterator>::IntoIter: DoubleEndedIterator,
782        <<B as Bind>::V as IntoIterator>::IntoIter: DoubleEndedIterator,
783    {
784        let (limit, cursor) = self.args.get_info();
785
786        if let Some(cursor) = cursor.as_ref() {
787            self.build_reader_where::<O, B>(cursor)?;
788        }
789
790        self.build_reader_order::<B>();
791        self.limit((limit + 1).into());
792
793        Ok(limit)
794    }
795
796    fn build_reader_where<O, B>(&mut self, cursor: &Value) -> Result<(), cursor::CursorError>
797    where
798        O: Cursor,
799        B: Bind<Cursor = O>,
800        B::T: Clone,
801        <<B as Bind>::I as IntoIterator>::IntoIter: DoubleEndedIterator,
802        <<B as Bind>::V as IntoIterator>::IntoIter: DoubleEndedIterator,
803    {
804        let is_order_desc = self.is_order_desc();
805        let cursor = O::deserialize_cursor(cursor)?;
806        let columns = B::columns().into_iter().rev();
807        let values = B::values(cursor).into_iter().rev();
808
809        let mut expr = None::<Expr>;
810        for (col, value) in columns.zip(values) {
811            let current_expr = if is_order_desc {
812                Expr::col(col.clone()).lt(value.clone())
813            } else {
814                Expr::col(col.clone()).gt(value.clone())
815            };
816
817            let Some(ref prev_expr) = expr else {
818                expr = Some(current_expr.clone());
819                continue;
820            };
821
822            expr = Some(current_expr.or(Expr::col(col).eq(value).and(prev_expr.clone())));
823        }
824
825        self.and_where(expr.unwrap());
826
827        Ok(())
828    }
829
830    fn build_reader_order<O: Bind>(&mut self) {
831        let order = if self.is_order_desc() {
832            sea_query::Order::Desc
833        } else {
834            sea_query::Order::Asc
835        };
836
837        let columns = O::columns();
838        for col in columns {
839            self.order_by(col, order.clone());
840        }
841    }
842
843    fn is_order_desc(&self) -> bool {
844        matches!(
845            (&self.order, self.args.is_backward()),
846            (cursor::Order::Asc, true) | (cursor::Order::Desc, false)
847        )
848    }
849}
850
851impl Deref for Reader {
852    type Target = SelectStatement;
853
854    fn deref(&self) -> &Self::Target {
855        &self.statement
856    }
857}
858
859impl DerefMut for Reader {
860    fn deref_mut(&mut self) -> &mut Self::Target {
861        &mut self.statement
862    }
863}
864
865/// Trait for binding cursor values in paginated queries.
866///
867/// This trait defines how to serialize cursor data for keyset pagination.
868/// It specifies which columns are used for ordering and how to extract
869/// their values from a cursor.
870///
871/// # Implementation
872///
873/// The trait is implemented for [`evento_core::Event`] to enable pagination
874/// over the event table using timestamp, version, and ID columns.
875///
876/// # Associated Types
877///
878/// - `T` - Column reference type
879/// - `I` - Iterator over column references
880/// - `V` - Iterator over value expressions
881/// - `Cursor` - The cursor type that provides pagination data
882pub trait Bind {
883    /// Column reference type (e.g., `Event` enum variant).
884    type T: IntoColumnRef + Clone;
885    /// Iterator type for columns.
886    type I: IntoIterator<Item = Self::T>;
887    /// Iterator type for values.
888    type V: IntoIterator<Item = Expr>;
889    /// The cursor type used for pagination.
890    type Cursor: Cursor;
891
892    /// Returns the columns used for cursor-based ordering.
893    fn columns() -> Self::I;
894    /// Extracts values from a cursor for WHERE clause construction.
895    fn values(cursor: <<Self as Bind>::Cursor as Cursor>::T) -> Self::V;
896}
897
898impl evento_core::cursor::Cursor for SqlEvent {
899    type T = evento_core::EventCursor;
900
901    fn serialize(&self) -> Self::T {
902        evento_core::EventCursor {
903            i: self.0.id.to_string(),
904            v: self.0.version,
905            t: self.0.timestamp,
906            s: self.0.timestamp_subsec,
907        }
908    }
909}
910
911impl Bind for SqlEvent {
912    type T = Event;
913    type I = [Self::T; 4];
914    type V = [Expr; 4];
915    type Cursor = Self;
916
917    fn columns() -> Self::I {
918        [
919            Event::Timestamp,
920            Event::TimestampSubsec,
921            Event::Version,
922            Event::Id,
923        ]
924    }
925
926    fn values(cursor: <<Self as Bind>::Cursor as Cursor>::T) -> Self::V {
927        [
928            cursor.t.into(),
929            cursor.s.into(),
930            cursor.v.into(),
931            cursor.i.into(),
932        ]
933    }
934}
935
936#[cfg(feature = "sqlite")]
937impl From<Sqlite> for evento_core::Evento {
938    fn from(value: Sqlite) -> Self {
939        evento_core::Evento::new(value)
940    }
941}
942
943#[cfg(feature = "sqlite")]
944impl From<&Sqlite> for evento_core::Evento {
945    fn from(value: &Sqlite) -> Self {
946        evento_core::Evento::new(value.clone())
947    }
948}
949
950#[cfg(feature = "mysql")]
951impl From<MySql> for evento_core::Evento {
952    fn from(value: MySql) -> Self {
953        evento_core::Evento::new(value)
954    }
955}
956
957#[cfg(feature = "mysql")]
958impl From<&MySql> for evento_core::Evento {
959    fn from(value: &MySql) -> Self {
960        evento_core::Evento::new(value.clone())
961    }
962}
963
964#[cfg(feature = "postgres")]
965impl From<Postgres> for evento_core::Evento {
966    fn from(value: Postgres) -> Self {
967        evento_core::Evento::new(value)
968    }
969}
970
971#[cfg(feature = "postgres")]
972impl From<&Postgres> for evento_core::Evento {
973    fn from(value: &Postgres) -> Self {
974        evento_core::Evento::new(value.clone())
975    }
976}
977
978#[derive(Debug, Clone, PartialEq, Default)]
979pub struct SqlEvent(pub evento_core::Event);
980
981impl<R: sqlx::Row> sqlx::FromRow<'_, R> for SqlEvent
982where
983    i32: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
984    Vec<u8>: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
985    String: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
986    i64: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
987    for<'r> &'r str: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
988    for<'r> &'r str: sqlx::ColumnIndex<R>,
989{
990    fn from_row(row: &R) -> Result<Self, sqlx::Error> {
991        let timestamp: i64 = sqlx::Row::try_get(row, "timestamp")?;
992        let timestamp_subsec: i64 = sqlx::Row::try_get(row, "timestamp_subsec")?;
993        let version: i32 = sqlx::Row::try_get(row, "version")?;
994        let metadata: Vec<u8> = sqlx::Row::try_get(row, "metadata")?;
995        let metadata: evento_core::metadata::Metadata =
996            bitcode::decode(&metadata).map_err(|e| sqlx::Error::Decode(e.into()))?;
997
998        Ok(SqlEvent(evento_core::Event {
999            id: Ulid::from_string(sqlx::Row::try_get(row, "id")?)
1000                .map_err(|err| sqlx::Error::InvalidArgument(err.to_string()))?,
1001            aggregate_id: sqlx::Row::try_get(row, "aggregator_id")?,
1002            aggregate_type: sqlx::Row::try_get(row, "aggregator_type")?,
1003            version: version as u16,
1004            name: sqlx::Row::try_get(row, "name")?,
1005            routing_key: sqlx::Row::try_get(row, "routing_key")?,
1006            data: sqlx::Row::try_get(row, "data")?,
1007            timestamp: timestamp as u64,
1008            timestamp_subsec: timestamp_subsec as u32,
1009            metadata,
1010        }))
1011    }
1012}
evento_sql/sql.rs

evento_sql/
sql.rs
