mire 0.2.0

use std::time::Duration;

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;

/// Page size for full-stream hydration reads. Hydration pages until a short
/// page so a long stream is never silently truncated (review C1/SNAP-1).
const HYDRATE_PAGE_SIZE: i64 = 1000;

/// Map a queried row into a [`RecordedEvent`]. `transaction_id` is selected as
/// text (`xid8::text`) because sqlx has no native xid8 decoder.
///
/// Fallible: a missing column or type mismatch (`try_get`) surfaces as
/// `Database` rather than panicking in library code (CORE-16), and a
/// `transaction_id` that fails to parse surfaces as `StreamCorruption`
/// rather than silently defaulting to `0` — which would rewind a
/// subscription cursor to the epoch and re-deliver the entire log (CORE-15).
fn map_event(row: &sqlx::postgres::PgRow) -> Result<RecordedEvent, EventStoreError> {
    let to_db = |e: sqlx::Error| EventStoreError::Database(e);
    let transaction_id_text: String = row.try_get("transaction_id").map_err(to_db)?;
    let stream_id: String = row.try_get("stream_id").map_err(to_db)?;
    let global_position: i64 = row.try_get("global_position").map_err(to_db)?;
    let transaction_id = transaction_id_text.parse().map_err(|_| {
        EventStoreError::StreamCorruption {
            stream_id: stream_id.clone(),
            recorded_version: -1,
            read_count: -1,
            detail: format!("event at position {global_position} has unparseable transaction_id '{transaction_id_text}'"),
        }
    })?;
    Ok(RecordedEvent {
        global_position,
        stream_id,
        stream_version: row.try_get("stream_version").map_err(to_db)?,
        event_type: row.try_get("event_type").map_err(to_db)?,
        data: row.try_get("data").map_err(to_db)?,
        metadata: row.try_get("metadata").map_err(to_db)?,
        transaction_id,
        created_at: row.try_get("created_at").map_err(to_db)?,
    })
}

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

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

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

    /// Apply the schema migrations. Idempotent — safe to call at every boot.
    ///
    /// **Run this once at startup, before serving traffic.** The migration
    /// runs `CREATE INDEX` / `ALTER TABLE`, which take table-level locks
    /// (`ShareLock` / `AccessExclusiveLock`) that conflict with in-flight
    /// writers — even on the already-applied no-op path, because Postgres
    /// grabs the lock *before* the `IF NOT EXISTS` check. So a migrate that
    /// races a live writer to the same table would otherwise block until
    /// that writer's transaction ends. To keep that from becoming an
    /// invisible, unbounded hang (a paused writer connection shows as
    /// `idle in transaction`, so Postgres's deadlock detector never breaks
    /// the cycle), the DDL runs under a bounded `lock_timeout`: if a lock
    /// can't be taken in time, migrate fails fast with a retryable
    /// [`DbErrorKind::LockTimeout`] error instead of wedging.
    ///
    /// The advisory lock that serialises concurrent migrators is acquired
    /// *before* the timeout is armed, so multiple replicas booting at once
    /// still queue cleanly rather than timing each other out.
    pub async fn migrate(&self) -> Result<(), EventStoreError> {
        self.migrate_with_lock_timeout(Duration::from_secs(15)).await
    }

    /// [`migrate`](Self::migrate) with an explicit DDL `lock_timeout`. Tests
    /// use a short value so a contended migrate surfaces quickly instead of
    /// stalling the suite; production uses the 15s default.
    pub async fn migrate_with_lock_timeout(
        &self,
        lock_timeout: Duration,
    ) -> 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.
        //
        // The advisory lock is acquired with NO `lock_timeout` (default 0 =
        // wait forever) so genuine migrator serialisation is never spuriously
        // aborted; the bounded timeout is armed only for the DDL that follows.
        let mut tx = self.pool.begin().await?;
        sqlx::query("SELECT pg_advisory_xact_lock($1)")
            .bind(MIGRATE_LOCK_KEY)
            .execute(&mut *tx)
            .await?;
        // Arm a bounded lock_timeout for the DDL below (transaction-local).
        // A `CREATE INDEX` / `ALTER TABLE` that can't take its table lock —
        // because a live writer holds a conflicting row lock — now errors
        // with SQLSTATE 55P03 (→ DbErrorKind::LockTimeout, retryable) rather
        // than blocking forever behind a connection Postgres sees as idle.
        // `set_config(name, value, is_local=true)` is the parameterised
        // equivalent of `SET LOCAL lock_timeout` — no dynamic SQL string.
        sqlx::query("SELECT set_config('lock_timeout', $1, true)")
            .bind(lock_timeout.as_millis().max(1).to_string())
            .execute(&mut *tx)
            .await?;
        sqlx::raw_sql(include_str!("../migrations/001_event_store.sql"))
            .execute(&mut *tx)
            .await?;
        // 002: replica-coordination schema (subscription fence + leases),
        // append-only with idempotent ALTER so pre-lease databases upgrade
        // correctly (review MIG-1). Applied under the same advisory lock.
        sqlx::raw_sql(include_str!("../migrations/002_projection_leases.sql"))
            .execute(&mut *tx)
            .await?;
        // 003: denormalize stream_category onto es_events (plan §P4.5). DDL
        // only here — ADD COLUMN (nullable) + the composite cursor index.
        sqlx::raw_sql(include_str!("../migrations/003_event_category.sql"))
            .execute(&mut *tx)
            .await?;
        // Backfill stream_category on pre-existing rows in bounded batches,
        // still under the advisory lock so concurrent migrators don't double
        // it. Blocking on first boot after upgrade is acceptable (pre-1.0,
        // small installs — plan §P4.5 "bias to the simpler blocking
        // variant"). Batched (not one giant UPDATE) to bound lock/WAL per
        // statement; resumable because each pass targets only NULL rows, so a
        // crash mid-backfill just resumes on the next migrate. New writes from
        // this version already stamp the column, so once this drains there are
        // no NULLs and read_category_after can trust the column without a join.
        loop {
            let updated = sqlx::query(
                "UPDATE es_events e
                    SET stream_category = s.stream_category
                   FROM es_streams s
                  WHERE e.stream_id = s.stream_id
                    AND e.global_position IN (
                        SELECT global_position FROM es_events
                         WHERE stream_category IS NULL
                         ORDER BY global_position
                         LIMIT 5000
                    )",
            )
            .execute(&mut *tx)
            .await?
            .rows_affected();
            if updated == 0 {
                break;
            }
        }
        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_all(&stream_id, 0).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)
        };

        // Pending events are only dropped once the append *succeeds*. On any
        // failure (transient DB error, conflict, cancellation) they are
        // returned to the root so a retry — which the error type explicitly
        // advertises via `is_retryable()` — re-attempts the same write
        // rather than silently succeeding as a no-op (review C8/CORE-3).
        let metadata = match serde_json::to_value(&root.metadata) {
            Ok(m) => m,
            Err(e) => {
                root.restore_pending(events);
                return Err(e.into());
            }
        };
        let wrapped: Vec<Event<A::Event>> = events
            .into_iter()
            .map(|e| Event::new(e).with_metadata(metadata.clone()))
            .collect();
        match self
            .append(&root.stream_id, A::stream_category(), expected, &wrapped)
            .await
        {
            Ok(new_version) => {
                root.version = new_version;
                Ok(())
            }
            Err(e) => {
                root.restore_pending(wrapped.into_iter().map(|w| w.event).collect());
                Err(e)
            }
        }
    }

    /// Append `events` to `stream_id` in a **single atomic statement** —
    /// no client-driven `BEGIN`/`COMMIT`.
    ///
    /// The version fence (OCC on `stream_version`) is the whole reason we
    /// don't need a pessimistic transaction here: the CAS and the event
    /// inserts run as one data-modifying CTE, which Postgres executes in an
    /// implicit transaction. That is atomic, costs one round trip instead
    /// of three, holds the `es_streams` row lock only for the statement's
    /// own execution, and — critically — never leaves the connection
    /// `idle in transaction`. A multi-statement `BEGIN/…/COMMIT` parks the
    /// connection between statements holding locks, where Postgres's
    /// deadlock detector is blind to it; a single statement is always
    /// `active` and part of the lock wait-graph, so any contention is
    /// detected and broken by Postgres rather than hanging (review
    /// CORE-5/CORE-12, and the migrate-vs-writer wedge that motivated this).
    ///
    /// Composable transactional writes (read-before-write via
    /// [`TransactionScope`], or the saga runner appending events + routing
    /// rows together) still use [`append_in_tx`](Self::append_in_tx),
    /// which layers onto a caller-owned transaction.
    ///
    /// Semantic note (breaking, pre-1.0): `Exact(0)` on a missing stream
    /// returns `ConcurrencyConflict { expected: 0, actual: 0 }`. Callers
    /// wanting "create stream and write at version 0" must use `NoStream`.
    pub async fn append<E: EventData>(
        &self,
        stream_id: &str,
        category: &str,
        expected: ExpectedVersion,
        events: &[Event<E>],
    ) -> Result<i64, EventStoreError> {
        if events.is_empty() {
            // An empty append still asserts the expected version: callers
            // use it as a fence/precondition (review CORE-14).
            let actual = Self::read_current_version_on(&self.pool, stream_id).await?;
            match expected {
                ExpectedVersion::Exact(v) if v != actual => {
                    return Err(EventStoreError::ConcurrencyConflict {
                        stream_id: stream_id.to_string(),
                        expected: v,
                        actual,
                    });
                }
                ExpectedVersion::NoStream if actual != 0 => {
                    return Err(EventStoreError::ConcurrencyConflict {
                        stream_id: stream_id.to_string(),
                        expected: 0,
                        actual,
                    });
                }
                _ => {}
            }
            return Ok(actual);
        }

        let n_events = events.len() as i64;

        // Serialize all payloads up front, before issuing the statement —
        // no work happens while the row lock is held.
        let mut event_types: Vec<String> = Vec::with_capacity(events.len());
        let mut datas: Vec<serde_json::Value> = Vec::with_capacity(events.len());
        let mut metadatas: Vec<serde_json::Value> = Vec::with_capacity(events.len());
        for event in events {
            event_types.push(event.event.event_type().to_string());
            datas.push(serde_json::to_value(&event.event)?);
            metadatas.push(event.metadata.clone());
        }

        // One data-modifying CTE per expected-version mode. Each:
        //   1. advances `es_streams` (the version CAS / create), RETURNING
        //      the new high-water version, then
        //   2. inserts the N events at the allocated `(base+ord)` slots,
        // and the final SELECT returns the new version — or NULL when the
        // CAS matched no row (a conflict). Postgres runs every
        // data-modifying CTE to completion regardless of whether the final
        // SELECT reads it, so the inserts always happen.
        //
        // `WITH ORDINALITY` numbers the unnested events 1..N; the
        // `CROSS JOIN` against the (0-or-1-row) version CTE means a failed
        // CAS inserts zero events.
        let new_version: Option<i64> = match expected {
            // $1=expected version, $2=n, $3=stream_id, $4=types, $5=data, $6=meta, $7=category
            ExpectedVersion::Exact(v) => {
                sqlx::query_scalar(
                    "WITH bump AS (
                         UPDATE es_streams
                            SET stream_version = $1 + $2, updated_at = now()
                          WHERE stream_id = $3 AND stream_version = $1
                      RETURNING stream_version
                     ),
                     ins AS (
                         INSERT INTO es_events
                             (stream_id, stream_version, event_type, data, metadata, stream_category)
                         SELECT $3, $1 + t.ord, t.etype, t.data, t.meta, $7
                           FROM bump
                           CROSS JOIN unnest($4::text[], $5::jsonb[], $6::jsonb[])
                                WITH ORDINALITY AS t(etype, data, meta, ord)
                      RETURNING 1
                     )
                     SELECT (SELECT stream_version FROM bump) AS new_version",
                )
                .bind(v)
                .bind(n_events)
                .bind(stream_id)
                .bind(&event_types)
                .bind(&datas)
                .bind(&metadatas)
                .bind(category)
                .fetch_one(&self.pool)
                .await?
            }
            // $1=n, $2=stream_id, $3=types, $4=data, $5=meta, $6=category
            ExpectedVersion::NoStream => {
                sqlx::query_scalar(
                    "WITH created AS (
                         INSERT INTO es_streams (stream_id, stream_category, stream_version)
                         VALUES ($2, $6, $1)
                         ON CONFLICT (stream_id) DO NOTHING
                      RETURNING stream_version
                     ),
                     ins AS (
                         INSERT INTO es_events
                             (stream_id, stream_version, event_type, data, metadata, stream_category)
                         SELECT $2, t.ord, t.etype, t.data, t.meta, $6
                           FROM created
                           CROSS JOIN unnest($3::text[], $4::jsonb[], $5::jsonb[])
                                WITH ORDINALITY AS t(etype, data, meta, ord)
                      RETURNING 1
                     )
                     SELECT (SELECT stream_version FROM created) AS new_version",
                )
                .bind(n_events)
                .bind(stream_id)
                .bind(&event_types)
                .bind(&datas)
                .bind(&metadatas)
                .bind(category)
                .fetch_one(&self.pool)
                .await?
            }
            // $1=n, $2=stream_id, $3=types, $4=data, $5=meta, $6=category.
            // `ON CONFLICT DO UPDATE` is the create-race fix in one shot:
            // insert the stream at version N, or atomically increment an
            // existing row by N. Either way the upsert returns the new
            // high-water version; events land at `(new - N + ord)`.
            ExpectedVersion::Any => {
                sqlx::query_scalar(
                    "WITH upsert AS (
                         INSERT INTO es_streams (stream_id, stream_category, stream_version)
                         VALUES ($2, $6, $1)
                         ON CONFLICT (stream_id) DO UPDATE
                            SET stream_version = es_streams.stream_version + $1,
                                updated_at     = now()
                      RETURNING stream_version
                     ),
                     ins AS (
                         INSERT INTO es_events
                             (stream_id, stream_version, event_type, data, metadata, stream_category)
                         SELECT $2, (SELECT stream_version FROM upsert) - $1 + t.ord,
                                t.etype, t.data, t.meta, $6
                           FROM unnest($3::text[], $4::jsonb[], $5::jsonb[])
                                WITH ORDINALITY AS t(etype, data, meta, ord)
                      RETURNING 1
                     )
                     SELECT (SELECT stream_version FROM upsert) AS new_version",
                )
                .bind(n_events)
                .bind(stream_id)
                .bind(&event_types)
                .bind(&datas)
                .bind(&metadatas)
                .bind(category)
                .fetch_one(&self.pool)
                .await?
            }
        };

        match new_version {
            Some(v) => Ok(v),
            None => {
                // The CAS matched no row: read the committed version to
                // build a precise conflict (expected 0 for NoStream).
                let actual = Self::read_current_version_on(&self.pool, stream_id).await?;
                let expected_v = match expected {
                    ExpectedVersion::Exact(v) => v,
                    _ => 0,
                };
                Err(EventStoreError::ConcurrencyConflict {
                    stream_id: stream_id.to_string(),
                    expected: expected_v,
                    actual,
                })
            }
        }
    }

    /// Read the current `stream_version` (0 if the stream doesn't exist)
    /// on any executor — pool or connection.
    async fn read_current_version_on<'e, X>(
        executor: X,
        stream_id: &str,
    ) -> Result<i64, EventStoreError>
    where
        X: sqlx::Executor<'e, Database = Postgres>,
    {
        let v: Option<i64> =
            sqlx::query_scalar("SELECT stream_version FROM es_streams WHERE stream_id = $1")
                .bind(stream_id)
                .fetch_optional(executor)
                .await?;
        Ok(v.unwrap_or(0))
    }

    /// Append `events` to `stream_id`, advancing the stream version
    /// atomically.
    ///
    /// **No explicit `SELECT ... FOR UPDATE`** — the version check is folded
    /// into the conditional `UPDATE` (or, for fresh streams, the
    /// `INSERT ... ON CONFLICT DO NOTHING`), saving one round trip. Note
    /// this does **not** make concurrent writers lock-free: Postgres holds
    /// a row lock on the `es_streams` row from the moment this `UPDATE`
    /// executes until the transaction commits or rolls back, so a second
    /// writer to the *same* stream blocks until the first finishes, then
    /// re-evaluates the `WHERE` against the committed row (EvalPlanQual at
    /// READ COMMITTED) and surfaces `ConcurrencyConflict` on a mismatch.
    /// Writers to *different* streams never contend. The
    /// `UNIQUE (stream_id, stream_version)` invariant on `es_events` is the
    /// backup guarantee against any residual schedule anomaly.
    ///
    /// The lock window currently spans payload serialization, all per-event
    /// inserts, and the commit fsync; shortening it (serialize-first, batch
    /// insert) is tracked as review CORE-5/CORE-12.
    ///
    /// 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: &[Event<E>],
    ) -> Result<i64, EventStoreError> {
        if events.is_empty() {
            // An empty append still asserts the expected version: callers
            // use it as a fence/precondition. Read the current version and
            // verify it matches `expected` rather than blindly returning it
            // (review CORE-14).
            let actual = Self::read_current_version_in_tx(tx, stream_id).await?;
            match expected {
                ExpectedVersion::Exact(v) if v != actual => {
                    return Err(EventStoreError::ConcurrencyConflict {
                        stream_id: stream_id.to_string(),
                        expected: v,
                        actual,
                    });
                }
                ExpectedVersion::NoStream if actual != 0 => {
                    return Err(EventStoreError::ConcurrencyConflict {
                        stream_id: stream_id.to_string(),
                        expected: 0,
                        actual,
                    });
                }
                _ => {}
            }
            return Ok(actual);
        }

        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 all events into the allocated range in ONE statement
        // (`unnest … WITH ORDINALITY`) rather than N per-event inserts.
        // Inside a caller-owned transaction the `es_streams` row lock is
        // held from the CAS above until the caller commits; batching the
        // inserts shortens that window from N round trips to one (review
        // CORE-5/CORE-12). Payloads are serialised before the statement so
        // no encoding work happens while the lock is held. The base slot is
        // `new_version - n_events`, so event `ord` (1-based) lands at
        // `base + ord`. The UNIQUE (stream_id, stream_version) constraint
        // remains the backstop against any CAS-missed slot collision.
        let base = new_version - n_events;
        let mut event_types: Vec<String> = Vec::with_capacity(events.len());
        let mut datas: Vec<serde_json::Value> = Vec::with_capacity(events.len());
        let mut metadatas: Vec<serde_json::Value> = Vec::with_capacity(events.len());
        for event in events {
            event_types.push(event.event.event_type().to_string());
            datas.push(serde_json::to_value(&event.event)?);
            metadatas.push(event.metadata.clone());
        }
        sqlx::query(
            "INSERT INTO es_events (stream_id, stream_version, event_type, data, metadata, stream_category)
             SELECT $1, $2 + t.ord, t.etype, t.data, t.meta, $6
               FROM unnest($3::text[], $4::jsonb[], $5::jsonb[])
                    WITH ORDINALITY AS t(etype, data, meta, ord)",
        )
        .bind(stream_id)
        .bind(base)
        .bind(&event_types)
        .bind(&datas)
        .bind(&metadatas)
        .bind(category)
        .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?;

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

    /// Read the entire forward history of `stream_id` from `from_version`
    /// upward, paging in [`HYDRATE_PAGE_SIZE`] chunks until exhausted.
    ///
    /// Hydration must never use a single row-limited read: a stream longer
    /// than the limit would fold to truncated state while still reporting
    /// the full `stream_version`, and the resulting bad command decision
    /// would pass the optimistic-concurrency check on save (review
    /// C1/SNAP-1). The user-facing [`read_stream`](Self::read_stream) keeps
    /// its explicit limit for paging APIs; every hydration path uses this.
    pub async fn read_stream_all(
        &self,
        stream_id: &str,
        from_version: i64,
    ) -> Result<Vec<RecordedEvent>, EventStoreError> {
        let mut all = Vec::new();
        let mut next = from_version;
        loop {
            let page = Self::read_stream_with(
                &self.pool,
                stream_id,
                StreamQuery {
                    direction: ReadDirection::Forward,
                    from_version: next,
                    limit: HYDRATE_PAGE_SIZE,
                },
            )
            .await?;
            let len = page.len() as i64;
            if let Some(last) = page.last() {
                next = last.stream_version + 1;
            }
            all.extend(page);
            if len < HYDRATE_PAGE_SIZE {
                break;
            }
        }
        Ok(all)
    }

    /// Transaction-scoped counterpart to [`read_stream_all`](Self::read_stream_all),
    /// paging within the caller's transaction.
    async fn read_stream_all_in_tx(
        tx: &mut Transaction<'_, Postgres>,
        stream_id: &str,
        from_version: i64,
    ) -> Result<Vec<RecordedEvent>, EventStoreError> {
        let mut all = Vec::new();
        let mut next = from_version;
        loop {
            let page = Self::read_stream_with(
                &mut **tx,
                stream_id,
                StreamQuery {
                    direction: ReadDirection::Forward,
                    from_version: next,
                    limit: HYDRATE_PAGE_SIZE,
                },
            )
            .await?;
            let len = page.len() as i64;
            if let Some(last) = page.last() {
                next = last.stream_version + 1;
            }
            all.extend(page);
            if len < HYDRATE_PAGE_SIZE {
                break;
            }
        }
        Ok(all)
    }

    /// 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?;

        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(
            // No es_streams join: stream_category is denormalized onto
            // es_events (plan §P4.5), so this is a single-table range scan on
            // idx_es_events_category_cursor. Backfilled + stamped on write, so
            // the column is non-null for every row by the time reads run.
            "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
             WHERE e.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?;

        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_all(&stream_id, envelope.version + 1)
                    .await?;
                Self::apply_tail(&mut root, &tail)?;
            }
            root.version = current_version;
            return Ok(Some(root));
        }

        // Fall back to a full replay.
        let events = self.read_stream_all(&stream_id, 0).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,
            &[Event::new(envelope)],
        )
        .await?;
        Ok(())
    }

    /// Deserialize and fold `tail` events onto a snapshot-seeded root.
    /// Shared by the snapshot load paths (pool and transaction).
    fn apply_tail<A: Aggregate>(
        root: &mut AggregateRoot<A>,
        tail: &[RecordedEvent],
    ) -> Result<(), EventStoreError>
    where
        A::Event: serde::de::DeserializeOwned,
    {
        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);
        }
        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_all_in_tx(&mut self.tx, &stream_id, 0).await?;
        Ok(Some(AggregateRoot::hydrate(stream_id, &events, version)?))
    }

    /// Snapshot-aware load within the transaction. Mirrors
    /// [`EventStore::load_snapshotted`], reading the snapshot and tail
    /// through the scope's transaction.
    ///
    /// **Consistency caveat:** a default sqlx transaction runs at READ
    /// COMMITTED, where *each statement* takes a fresh snapshot — so the
    /// stream-version, snapshot, and tail reads here do **not** observe a
    /// single point-in-time view, and a concurrent writer between them can
    /// yield a `(state, version)` pair where `state` reflects more events
    /// than `version`. The optimistic-concurrency check at save time still
    /// prevents log corruption, but a cross-stream decision made on the
    /// returned pair can be inconsistent. For a true point-in-time read,
    /// take `load_for_update` (locks the stream row) or run the scope at
    /// REPEATABLE READ. (Review CORE-9/SNAP-3.)
    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_all_in_tx(
                    &mut self.tx,
                    &stream_id,
                    envelope.version + 1,
                )
                .await?;
                EventStore::apply_tail(&mut root, &tail)?;
            }
            root.version = current_version;
            return Ok(Some(root));
        }

        let events = EventStore::read_stream_all_in_tx(&mut self.tx, &stream_id, 0).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)
        };

        // Restore pending on failure so the root stays retryable; the events
        // are only consumed once the in-transaction append succeeds (C8).
        let metadata = match serde_json::to_value(&root.metadata) {
            Ok(m) => m,
            Err(e) => {
                root.restore_pending(events);
                return Err(e.into());
            }
        };
        let wrapped: Vec<Event<A::Event>> = events
            .into_iter()
            .map(|e| Event::new(e).with_metadata(metadata.clone()))
            .collect();
        let new_version = match EventStore::append_in_tx(
            &mut self.tx,
            &root.stream_id,
            A::stream_category(),
            expected,
            &wrapped,
        )
        .await
        {
            Ok(v) => v,
            Err(e) => {
                root.restore_pending(wrapped.into_iter().map(|w| w.event).collect());
                return Err(e);
            }
        };
        root.version = new_version;

        let recorded =
            EventStore::read_stream_all_in_tx(&mut self.tx, &root.stream_id, old_version + 1)
                .await?;
        self.committed_events.extend(recorded);

        Ok(())
    }

    pub async fn append<E: EventData>(
        &mut self,
        stream_id: &str,
        category: &str,
        expected: ExpectedVersion,
        events: &[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()
    }
}