mire 0.1.1

use sqlx::{PgPool, Postgres, Row, Transaction};

use crate::{
    Aggregate, AggregateRoot, Event, EventData, EventStoreError, RecordedEvent,
    snapshot::{SNAPSHOT_CATEGORY, Snapshot, SnapshotEnvelope},
    stream::{ExpectedVersion, ReadDirection, StreamQuery},
};

/// Whether a transactional load should take a `FOR UPDATE` row-lock on the
/// stream entry. See [`TransactionScope::load_for_update`].
enum LoadLock {
    None,
    ForUpdate,
}

/// Transaction-scoped advisory-lock key guarding schema migration. Arbitrary
/// but stable: the ASCII bytes of "mire".
const MIGRATE_LOCK_KEY: i64 = 0x6d_69_72_65;

/// Map a queried row into a [`RecordedEvent`]. `transaction_id` is selected as
/// text (`xid8::text`) because sqlx has no native xid8 decoder; it always holds
/// a valid u64.
fn map_event(row: &sqlx::postgres::PgRow) -> RecordedEvent {
    let transaction_id: String = row.get("transaction_id");
    RecordedEvent {
        global_position: row.get("global_position"),
        stream_id: row.get("stream_id"),
        stream_version: row.get("stream_version"),
        event_type: row.get("event_type"),
        data: row.get("data"),
        metadata: row.get("metadata"),
        transaction_id: transaction_id.parse().unwrap_or(0),
        created_at: row.get("created_at"),
    }
}

#[derive(Clone)]
pub struct EventStore {
    pool: PgPool,
}

impl EventStore {
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }

    pub fn pool(&self) -> &PgPool {
        &self.pool
    }

    pub async fn migrate(&self) -> Result<(), EventStoreError> {
        // Serialize concurrent migrations (parallel tests sharing a database,
        // or several service replicas booting at once). `CREATE TABLE IF NOT
        // EXISTS` is not concurrency-safe — simultaneous runs race on the
        // pg_type catalog and one fails with a duplicate-key error — so we take
        // a transaction-scoped advisory lock before applying the schema.
        let mut tx = self.pool.begin().await?;
        sqlx::query("SELECT pg_advisory_xact_lock($1)")
            .bind(MIGRATE_LOCK_KEY)
            .execute(&mut *tx)
            .await?;
        sqlx::raw_sql(include_str!("../migrations/001_event_store.sql"))
            .execute(&mut *tx)
            .await?;
        tx.commit().await?;
        Ok(())
    }

    pub async fn load<A: Aggregate>(
        &self,
        id: &str,
    ) -> Result<Option<AggregateRoot<A>>, EventStoreError> {
        let stream_id = format!("{}-{}", A::stream_category(), id);

        let stream_row = sqlx::query("SELECT stream_version FROM es_streams WHERE stream_id = $1")
            .bind(&stream_id)
            .fetch_optional(&self.pool)
            .await?;

        let Some(stream_row) = stream_row else {
            return Ok(None);
        };

        let version: i64 = stream_row.get("stream_version");
        let events = self.read_stream(&stream_id, StreamQuery::default()).await?;

        Ok(Some(AggregateRoot::hydrate(stream_id, &events, version)?))
    }

    pub async fn load_or_default<A: Aggregate>(
        &self,
        id: &str,
    ) -> Result<AggregateRoot<A>, EventStoreError> {
        match self.load::<A>(id).await? {
            Some(root) => Ok(root),
            None => Ok(AggregateRoot::new(id)),
        }
    }

    /// Cheap presence check: returns `true` iff a stream row exists for
    /// `A::stream_category()-id`. Single round-trip — does not fetch
    /// events. Prefer this over discarding the result of [`load`] when
    /// you just need to know whether the aggregate exists.
    pub async fn stream_exists<A: Aggregate>(&self, id: &str) -> Result<bool, EventStoreError> {
        let stream_id = format!("{}-{}", A::stream_category(), id);
        let exists: bool =
            sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM es_streams WHERE stream_id = $1)")
                .bind(&stream_id)
                .fetch_one(&self.pool)
                .await?;
        Ok(exists)
    }

    pub async fn save<A: Aggregate>(
        &self,
        root: &mut AggregateRoot<A>,
    ) -> Result<(), EventStoreError> {
        if !root.has_pending() {
            return Ok(());
        }

        let events = root.take_pending();
        let expected = if root.version == 0 {
            ExpectedVersion::NoStream
        } else {
            ExpectedVersion::Exact(root.version)
        };

        let metadata = serde_json::to_value(&root.metadata)?;
        let wrapped: Vec<Event<A::Event>> = events
            .into_iter()
            .map(|e| Event::new(e).with_metadata(metadata.clone()))
            .collect();
        let new_version = self
            .append(&root.stream_id, A::stream_category(), expected, wrapped)
            .await?;
        root.version = new_version;

        Ok(())
    }

    pub async fn append<E: EventData>(
        &self,
        stream_id: &str,
        category: &str,
        expected: ExpectedVersion,
        events: Vec<Event<E>>,
    ) -> Result<i64, EventStoreError> {
        let mut tx = self.pool.begin().await?;
        let version = Self::append_in_tx(&mut tx, stream_id, category, expected, events).await?;
        tx.commit().await?;
        Ok(version)
    }

    /// Append `events` to `stream_id`, advancing the stream version
    /// atomically.
    ///
    /// **No `SELECT ... FOR UPDATE`** — the version check is performed by
    /// the conditional `UPDATE` (or, for fresh streams, by the
    /// `INSERT ... ON CONFLICT DO NOTHING`). Concurrent writers race
    /// rather than queue; the loser sees 0 rows affected and surfaces
    /// `ConcurrencyConflict`. The `UNIQUE (stream_id, stream_version)`
    /// invariant on `es_events` is the backup guarantee against
    /// hypothetical schedule anomalies.
    ///
    /// Semantic note (breaking, pre-1.0): `Exact(0)` on a missing stream
    /// now returns `ConcurrencyConflict { expected: 0, actual: 0 }`.
    /// Callers wanting "create stream and write at version 0" must use
    /// `NoStream` (which is what they should have been doing).
    pub async fn append_in_tx<E: EventData>(
        tx: &mut Transaction<'_, Postgres>,
        stream_id: &str,
        category: &str,
        expected: ExpectedVersion,
        events: Vec<Event<E>>,
    ) -> Result<i64, EventStoreError> {
        if events.is_empty() {
            let row = sqlx::query("SELECT stream_version FROM es_streams WHERE stream_id = $1")
                .bind(stream_id)
                .fetch_optional(&mut **tx)
                .await?;
            return Ok(row.map(|r| r.get("stream_version")).unwrap_or(0));
        }

        let n_events = events.len() as i64;

        // Allocate the version range via CAS. `new_version` is the
        // highest version slot this writer owns; the allocated range is
        // `(new_version - n_events .. new_version]`.
        let new_version = match expected {
            ExpectedVersion::Exact(v) => {
                let updated: Option<i64> = sqlx::query_scalar(
                    "UPDATE es_streams
                        SET stream_version = $1 + $2,
                            updated_at     = now()
                      WHERE stream_id      = $3
                        AND stream_version = $1
                  RETURNING stream_version",
                )
                .bind(v)
                .bind(n_events)
                .bind(stream_id)
                .fetch_optional(&mut **tx)
                .await?;
                match updated {
                    Some(new_v) => new_v,
                    None => {
                        let actual = Self::read_current_version_in_tx(tx, stream_id).await?;
                        return Err(EventStoreError::ConcurrencyConflict {
                            stream_id: stream_id.to_string(),
                            expected: v,
                            actual,
                        });
                    }
                }
            }
            ExpectedVersion::Any => {
                // Try a blind increment first; fall through to INSERT if
                // the row doesn't exist yet.
                let bumped: Option<i64> = sqlx::query_scalar(
                    "UPDATE es_streams
                        SET stream_version = stream_version + $1,
                            updated_at     = now()
                      WHERE stream_id      = $2
                  RETURNING stream_version",
                )
                .bind(n_events)
                .bind(stream_id)
                .fetch_optional(&mut **tx)
                .await?;
                match bumped {
                    Some(v) => v,
                    None => {
                        // No row — try to create. ON CONFLICT covers the
                        // race where a peer creates between our UPDATE
                        // and our INSERT.
                        let inserted: Option<i64> = sqlx::query_scalar(
                            "INSERT INTO es_streams (stream_id, stream_category, stream_version)
                             VALUES ($1, $2, $3)
                             ON CONFLICT (stream_id) DO NOTHING
                          RETURNING stream_version",
                        )
                        .bind(stream_id)
                        .bind(category)
                        .bind(n_events)
                        .fetch_optional(&mut **tx)
                        .await?;
                        match inserted {
                            Some(v) => v,
                            None => {
                                // Lost the create race — retry the
                                // blind increment, guaranteed to find a
                                // row this time.
                                sqlx::query_scalar(
                                    "UPDATE es_streams
                                        SET stream_version = stream_version + $1,
                                            updated_at     = now()
                                      WHERE stream_id      = $2
                                  RETURNING stream_version",
                                )
                                .bind(n_events)
                                .bind(stream_id)
                                .fetch_one(&mut **tx)
                                .await?
                            }
                        }
                    }
                }
            }
            ExpectedVersion::NoStream => {
                let inserted: Option<i64> = sqlx::query_scalar(
                    "INSERT INTO es_streams (stream_id, stream_category, stream_version)
                     VALUES ($1, $2, $3)
                     ON CONFLICT (stream_id) DO NOTHING
                  RETURNING stream_version",
                )
                .bind(stream_id)
                .bind(category)
                .bind(n_events)
                .fetch_optional(&mut **tx)
                .await?;
                match inserted {
                    Some(v) => v,
                    None => {
                        let actual = Self::read_current_version_in_tx(tx, stream_id).await?;
                        return Err(EventStoreError::ConcurrencyConflict {
                            stream_id: stream_id.to_string(),
                            expected: 0,
                            actual,
                        });
                    }
                }
            }
        };

        // Insert events into the allocated range. The UNIQUE
        // (stream_id, stream_version) constraint on es_events catches
        // any version-slot collision that the CAS above failed to
        // prevent (defensive — should be unreachable).
        let mut version = new_version - n_events;
        for event in &events {
            version += 1;
            let data = serde_json::to_value(&event.event)?;
            let event_type = event.event.event_type();

            sqlx::query(
                "INSERT INTO es_events (stream_id, stream_version, event_type, data, metadata)
                 VALUES ($1, $2, $3, $4, $5)",
            )
            .bind(stream_id)
            .bind(version)
            .bind(event_type)
            .bind(&data)
            .bind(&event.metadata)
            .execute(&mut **tx)
            .await?;
        }

        Ok(new_version)
    }

    async fn read_current_version_in_tx(
        tx: &mut Transaction<'_, Postgres>,
        stream_id: &str,
    ) -> Result<i64, EventStoreError> {
        let row = sqlx::query("SELECT stream_version FROM es_streams WHERE stream_id = $1")
            .bind(stream_id)
            .fetch_optional(&mut **tx)
            .await?;
        Ok(row.map(|r| r.get("stream_version")).unwrap_or(0))
    }

    pub async fn read_stream(
        &self,
        stream_id: &str,
        query: StreamQuery,
    ) -> Result<Vec<RecordedEvent>, EventStoreError> {
        Self::read_stream_with(&self.pool, stream_id, query).await
    }

    pub async fn read_stream_with<'e, E>(
        executor: E,
        stream_id: &str,
        query: StreamQuery,
    ) -> Result<Vec<RecordedEvent>, EventStoreError>
    where
        E: sqlx::Executor<'e, Database = Postgres>,
    {
        // Two static query strings (sqlx 0.9 requires `&'static str`): the only
        // difference is the sort direction, which is never user input.
        let sql: &'static str = match query.direction {
            ReadDirection::Forward => {
                "SELECT global_position, stream_id, stream_version, event_type, data, metadata, transaction_id::text AS transaction_id, created_at
                 FROM es_events
                 WHERE stream_id = $1 AND stream_version >= $2
                 ORDER BY stream_version ASC
                 LIMIT $3"
            }
            ReadDirection::Backward => {
                "SELECT global_position, stream_id, stream_version, event_type, data, metadata, transaction_id::text AS transaction_id, created_at
                 FROM es_events
                 WHERE stream_id = $1 AND stream_version >= $2
                 ORDER BY stream_version DESC
                 LIMIT $3"
            }
        };

        let rows = sqlx::query(sql)
            .bind(stream_id)
            .bind(query.from_version)
            .bind(query.limit)
            .fetch_all(executor)
            .await?;

        Ok(rows.iter().map(map_event).collect())
    }

    /// Read committed events across all streams after the `(transaction_id,
    /// position)` cursor, in transaction-then-position order. Events from
    /// still-in-flight transactions are excluded, so the result is gapless and
    /// safe to checkpoint past. This backs catch-up subscriptions.
    pub async fn read_all_after(
        &self,
        after_transaction_id: u64,
        after_position: i64,
        limit: i64,
    ) -> Result<Vec<RecordedEvent>, EventStoreError> {
        let rows = sqlx::query(
            "SELECT global_position, stream_id, stream_version, event_type, data, metadata, transaction_id::text AS transaction_id, created_at
             FROM es_events
             WHERE (
                 (transaction_id = $1::text::xid8 AND global_position > $2)
                 OR transaction_id > $1::text::xid8
             )
             AND transaction_id < pg_snapshot_xmin(pg_current_snapshot())
             ORDER BY transaction_id ASC, global_position ASC
             LIMIT $3",
        )
        .bind(after_transaction_id.to_string())
        .bind(after_position)
        .bind(limit)
        .fetch_all(&self.pool)
        .await?;

        Ok(rows.iter().map(map_event).collect())
    }

    /// Like [`read_all_after`](Self::read_all_after) but restricted to one
    /// stream category.
    pub async fn read_category_after(
        &self,
        category: &str,
        after_transaction_id: u64,
        after_position: i64,
        limit: i64,
    ) -> Result<Vec<RecordedEvent>, EventStoreError> {
        let rows = sqlx::query(
            "SELECT e.global_position, e.stream_id, e.stream_version, e.event_type, e.data, e.metadata, e.transaction_id::text AS transaction_id, e.created_at
             FROM es_events e
             JOIN es_streams s ON e.stream_id = s.stream_id
             WHERE s.stream_category = $1
             AND (
                 (e.transaction_id = $2::text::xid8 AND e.global_position > $3)
                 OR e.transaction_id > $2::text::xid8
             )
             AND e.transaction_id < pg_snapshot_xmin(pg_current_snapshot())
             ORDER BY e.transaction_id ASC, e.global_position ASC
             LIMIT $4",
        )
        .bind(category)
        .bind(after_transaction_id.to_string())
        .bind(after_position)
        .bind(limit)
        .fetch_all(&self.pool)
        .await?;

        Ok(rows.iter().map(map_event).collect())
    }

    /// A coarse health signal for one subscription: the newest event's global
    /// position minus the subscription's last checkpoint. A growing value means
    /// the projection is falling behind.
    ///
    /// Note this compares against the *global* max position, so a
    /// category-scoped subscription will report non-zero lag whenever other
    /// categories receive events — treat it as a trend, not an exact backlog.
    pub async fn projection_lag(&self, subscription_id: &str) -> Result<i64, EventStoreError> {
        let max_position: i64 =
            sqlx::query_scalar("SELECT COALESCE(MAX(global_position), 0) FROM es_events")
                .fetch_one(&self.pool)
                .await?;
        let last_position: i64 = sqlx::query_scalar(
            "SELECT COALESCE(
                 (SELECT last_position FROM es_subscriptions WHERE subscription_id = $1),
                 0
             )",
        )
        .bind(subscription_id)
        .fetch_one(&self.pool)
        .await?;
        Ok(max_position - last_position)
    }

    /// Load an aggregate, seeding from its latest snapshot when one is present
    /// and current. Falls back to a full replay if there is no snapshot, the
    /// snapshot's version no longer matches, or it fails to deserialize —
    /// snapshots are a disposable optimisation, never the source of truth.
    /// Returns `None` if the stream does not exist.
    pub async fn load_snapshotted<A: Snapshot>(
        &self,
        id: &str,
    ) -> Result<Option<AggregateRoot<A>>, EventStoreError> {
        let stream_id = format!("{}-{}", A::stream_category(), id);

        let stream_row = sqlx::query("SELECT stream_version FROM es_streams WHERE stream_id = $1")
            .bind(&stream_id)
            .fetch_optional(&self.pool)
            .await?;
        let Some(stream_row) = stream_row else {
            return Ok(None);
        };
        let current_version: i64 = stream_row.get("stream_version");

        // Latest snapshot for this stream (highest version, read backwards).
        let snapshot_stream = format!("{stream_id}-snapshot");
        let latest = self
            .read_stream(
                &snapshot_stream,
                StreamQuery {
                    direction: ReadDirection::Backward,
                    from_version: 0,
                    limit: 1,
                },
            )
            .await?;

        if let Some(recorded) = latest.first()
            && let Ok(envelope) = serde_json::from_value::<SnapshotEnvelope>(recorded.data.clone())
            && envelope.snapshot_version == A::SNAPSHOT_VERSION
            && let Ok(state) = serde_json::from_value::<A>(envelope.state)
        {
            let mut root = AggregateRoot::from_snapshot(stream_id.clone(), state, envelope.version);
            if envelope.version < current_version {
                let tail = self
                    .read_stream(
                        &stream_id,
                        StreamQuery {
                            from_version: envelope.version + 1,
                            ..Default::default()
                        },
                    )
                    .await?;
                for tail_event in &tail {
                    let event = serde_json::from_value::<A::Event>(tail_event.data.clone())
                        .map_err(|source| EventStoreError::Deserialization {
                            stream_id: tail_event.stream_id.clone(),
                            global_position: tail_event.global_position,
                            event_type: tail_event.event_type.clone(),
                            source,
                        })?;
                    root.state.apply(&event);
                }
            }
            root.version = current_version;
            return Ok(Some(root));
        }

        // Fall back to a full replay.
        let events = self.read_stream(&stream_id, StreamQuery::default()).await?;
        Ok(Some(AggregateRoot::hydrate(
            stream_id,
            &events,
            current_version,
        )?))
    }

    /// Snapshot-aware counterpart to [`load_or_default`]: seeds from the
    /// latest snapshot if one is current, replaying only the tail; falls
    /// back to a full replay otherwise, or returns a fresh `AggregateRoot`
    /// if the stream doesn't exist.
    pub async fn load_or_default_snapshotted<A: Snapshot>(
        &self,
        id: &str,
    ) -> Result<AggregateRoot<A>, EventStoreError> {
        match self.load_snapshotted::<A>(id).await? {
            Some(root) => Ok(root),
            None => Ok(AggregateRoot::new(id)),
        }
    }

    /// Save pending events like [`save`](Self::save), then write a snapshot if
    /// the new version crosses a [`SNAPSHOT_FREQUENCY`](Snapshot::SNAPSHOT_FREQUENCY)
    /// boundary. This is the snapshot-aware save path.
    pub async fn save_snapshotting<A: Snapshot>(
        &self,
        root: &mut AggregateRoot<A>,
    ) -> Result<(), EventStoreError> {
        let old_version = root.version;
        self.save(root).await?;

        let freq = A::SNAPSHOT_FREQUENCY;
        if freq > 0 && root.version / freq > old_version / freq {
            self.save_snapshot(root).await?;
        }
        Ok(())
    }

    /// Explicitly write a snapshot of `root`'s current state into its snapshot
    /// stream. Idempotent in effect: only the latest snapshot is ever read.
    pub async fn save_snapshot<A: Snapshot>(
        &self,
        root: &AggregateRoot<A>,
    ) -> Result<(), EventStoreError> {
        let envelope = SnapshotEnvelope {
            snapshot_version: A::SNAPSHOT_VERSION,
            version: root.version,
            state: serde_json::to_value(&root.state)?,
        };
        let snapshot_stream = format!("{}-snapshot", root.stream_id);
        self.append::<SnapshotEnvelope>(
            &snapshot_stream,
            SNAPSHOT_CATEGORY,
            ExpectedVersion::Any,
            vec![Event::new(envelope)],
        )
        .await?;
        Ok(())
    }

    pub async fn begin_transaction(&self) -> Result<TransactionScope<'_>, EventStoreError> {
        let tx = self.pool.begin().await?;
        Ok(TransactionScope {
            store: self,
            tx,
            committed_events: Vec::new(),
        })
    }
}

/// **Read-before-write escape hatch.** In a strict event-sourced design,
/// commands read from projections (CQRS) and write to aggregates. The
/// `TransactionScope` exists for edge cases where that separation is
/// impractical: cross-aggregate invariants enforced in a single deployment,
/// recovery tooling, one-shot migrations. **Prefer a projection-driven
/// design** wherever you can — a long-lived `TransactionScope` on a hot
/// stream blocks other writers and couples your read path to the event
/// store's transaction lifecycle.
///
/// If you find yourself reaching for [`load_for_update`](Self::load_for_update)
/// in steady-state command code, that's a signal the projection boundary is
/// in the wrong place.
pub struct TransactionScope<'a> {
    #[allow(dead_code)]
    store: &'a EventStore,
    tx: Transaction<'static, Postgres>,
    committed_events: Vec<RecordedEvent>,
}

impl<'a> TransactionScope<'a> {
    /// Load `A`'s aggregate within this transaction. Returns `None` if the
    /// stream does not exist. Does **not** lock the stream row — concurrent
    /// writers to this stream are still possible until [`save`](Self::save)
    /// takes its own `FOR UPDATE`.
    pub async fn load<A: Aggregate>(
        &mut self,
        id: &str,
    ) -> Result<Option<AggregateRoot<A>>, EventStoreError> {
        self.load_with::<A>(id, LoadLock::None).await
    }

    /// Load `A`'s aggregate within this transaction and take a `FOR UPDATE`
    /// row-lock on its stream entry. Returns `None` if the stream does not
    /// exist; in that case no lock is acquired (there is no row to lock),
    /// and the standard optimistic-concurrency check at save time still
    /// guards the create path.
    ///
    /// Use this when a *different* stream's write depends on this
    /// aggregate's state and must not race a concurrent writer to *this*
    /// stream. Without the lock, you observe a snapshot that can be
    /// invalidated before commit.
    ///
    /// Hot-path warning: holds the row lock until the transaction commits
    /// or rolls back. Long-running scopes block other writers to the same
    /// stream.
    pub async fn load_for_update<A: Aggregate>(
        &mut self,
        id: &str,
    ) -> Result<Option<AggregateRoot<A>>, EventStoreError> {
        self.load_with::<A>(id, LoadLock::ForUpdate).await
    }

    pub async fn load_or_default<A: Aggregate>(
        &mut self,
        id: &str,
    ) -> Result<AggregateRoot<A>, EventStoreError> {
        match self.load::<A>(id).await? {
            Some(root) => Ok(root),
            None => Ok(AggregateRoot::new(id)),
        }
    }

    /// Like [`load_or_default`](Self::load_or_default) but takes
    /// `FOR UPDATE` on the stream row when one exists. See
    /// [`load_for_update`](Self::load_for_update) for caveats.
    pub async fn load_or_default_for_update<A: Aggregate>(
        &mut self,
        id: &str,
    ) -> Result<AggregateRoot<A>, EventStoreError> {
        match self.load_for_update::<A>(id).await? {
            Some(root) => Ok(root),
            None => Ok(AggregateRoot::new(id)),
        }
    }

    async fn load_with<A: Aggregate>(
        &mut self,
        id: &str,
        lock: LoadLock,
    ) -> Result<Option<AggregateRoot<A>>, EventStoreError> {
        let stream_id = format!("{}-{}", A::stream_category(), id);

        let sql = match lock {
            LoadLock::None => "SELECT stream_version FROM es_streams WHERE stream_id = $1",
            LoadLock::ForUpdate => {
                "SELECT stream_version FROM es_streams WHERE stream_id = $1 FOR UPDATE"
            }
        };

        let stream_row = sqlx::query(sql)
            .bind(&stream_id)
            .fetch_optional(&mut *self.tx)
            .await?;

        let Some(row) = stream_row else {
            return Ok(None);
        };
        let version: i64 = row.get("stream_version");
        let events =
            EventStore::read_stream_with(&mut *self.tx, &stream_id, StreamQuery::default()).await?;
        Ok(Some(AggregateRoot::hydrate(stream_id, &events, version)?))
    }

    /// Snapshot-aware load within the transaction. Mirrors
    /// [`EventStore::load_snapshotted`] but using the scope's transaction
    /// so the snapshot and tail-event reads observe a single consistent
    /// view of committed data.
    pub async fn load_snapshotted<A: Snapshot>(
        &mut self,
        id: &str,
    ) -> Result<Option<AggregateRoot<A>>, EventStoreError> {
        let stream_id = format!("{}-{}", A::stream_category(), id);

        let stream_row = sqlx::query("SELECT stream_version FROM es_streams WHERE stream_id = $1")
            .bind(&stream_id)
            .fetch_optional(&mut *self.tx)
            .await?;
        let Some(stream_row) = stream_row else {
            return Ok(None);
        };
        let current_version: i64 = stream_row.get("stream_version");

        let snapshot_stream = format!("{stream_id}-snapshot");
        let latest = EventStore::read_stream_with(
            &mut *self.tx,
            &snapshot_stream,
            StreamQuery {
                direction: ReadDirection::Backward,
                from_version: 0,
                limit: 1,
            },
        )
        .await?;

        if let Some(recorded) = latest.first()
            && let Ok(envelope) = serde_json::from_value::<SnapshotEnvelope>(recorded.data.clone())
            && envelope.snapshot_version == A::SNAPSHOT_VERSION
            && let Ok(state) = serde_json::from_value::<A>(envelope.state)
        {
            let mut root = AggregateRoot::from_snapshot(stream_id.clone(), state, envelope.version);
            if envelope.version < current_version {
                let tail = EventStore::read_stream_with(
                    &mut *self.tx,
                    &stream_id,
                    StreamQuery {
                        from_version: envelope.version + 1,
                        ..Default::default()
                    },
                )
                .await?;
                for tail_event in &tail {
                    let event = serde_json::from_value::<A::Event>(tail_event.data.clone())
                        .map_err(|source| EventStoreError::Deserialization {
                            stream_id: tail_event.stream_id.clone(),
                            global_position: tail_event.global_position,
                            event_type: tail_event.event_type.clone(),
                            source,
                        })?;
                    root.state.apply(&event);
                }
            }
            root.version = current_version;
            return Ok(Some(root));
        }

        let events =
            EventStore::read_stream_with(&mut *self.tx, &stream_id, StreamQuery::default()).await?;
        Ok(Some(AggregateRoot::hydrate(
            stream_id,
            &events,
            current_version,
        )?))
    }

    pub async fn save<A: Aggregate>(
        &mut self,
        root: &mut AggregateRoot<A>,
    ) -> Result<(), EventStoreError> {
        if !root.has_pending() {
            return Ok(());
        }

        let events = root.take_pending();
        let old_version = root.version;
        let expected = if old_version == 0 {
            ExpectedVersion::NoStream
        } else {
            ExpectedVersion::Exact(old_version)
        };

        let metadata = serde_json::to_value(&root.metadata)?;
        let wrapped: Vec<Event<A::Event>> = events
            .into_iter()
            .map(|e| Event::new(e).with_metadata(metadata.clone()))
            .collect();
        let new_version = EventStore::append_in_tx(
            &mut self.tx,
            &root.stream_id,
            A::stream_category(),
            expected,
            wrapped,
        )
        .await?;
        root.version = new_version;

        let recorded = EventStore::read_stream_with(
            &mut *self.tx,
            &root.stream_id,
            StreamQuery {
                from_version: old_version + 1,
                ..Default::default()
            },
        )
        .await?;
        self.committed_events.extend(recorded);

        Ok(())
    }

    pub async fn append<E: EventData>(
        &mut self,
        stream_id: &str,
        category: &str,
        expected: ExpectedVersion,
        events: Vec<Event<E>>,
    ) -> Result<i64, EventStoreError> {
        EventStore::append_in_tx(&mut self.tx, stream_id, category, expected, events).await
    }

    pub fn tx(&mut self) -> &mut Transaction<'static, Postgres> {
        &mut self.tx
    }

    pub fn take_committed_events(&mut self) -> Vec<RecordedEvent> {
        std::mem::take(&mut self.committed_events)
    }

    pub async fn commit(self) -> Result<CommittedEvents, EventStoreError> {
        self.tx.commit().await?;
        Ok(CommittedEvents {
            events: self.committed_events,
        })
    }

    pub async fn rollback(self) -> Result<(), EventStoreError> {
        self.tx.rollback().await?;
        Ok(())
    }
}

pub struct CommittedEvents {
    pub events: Vec<RecordedEvent>,
}

impl CommittedEvents {
    pub fn is_empty(&self) -> bool {
        self.events.is_empty()
    }
}