mire 0.1.0

//! `mire` — a small, generic PostgreSQL event-sourcing library.
//!
//! Events are appended to per-aggregate streams in an append-only log, ordered
//! both within a stream (`stream_version`) and globally (`global_position`).
//! Aggregates are rebuilt by replaying their events; read models are built by
//! [`Subscription`]s that poll the log and checkpoint their position.
//!
//! The core building blocks are:
//! - [`Aggregate`] / [`AggregateRoot`] — your domain state + the events it records.
//! - [`EventStore`] — load, save, append and read events (Postgres-backed).
//! - [`Subscription`] — poll the log to drive projections / read models.
//!
//! See the `examples/` directory for runnable end-to-end usage.

mod error;
mod event;
/// Replica-coordination primitives. Surfaced publicly so concurrency
/// tests can drive them directly; production code should use the
/// [`ProjectionRunner`] which wraps these correctly.
#[doc(hidden)]
pub mod lease;
mod projection;
mod snapshot;
mod store;
mod stream;
mod subscription;

pub use lease::LeaseStatus;

pub use error::{DbErrorKind, EventStoreError};
pub use event::{Event, EventData, EventMetadata, RecordedEvent};

/// `#[derive(EventData)]` — see the crate-level docs and
/// [`EventData`] for the runtime contract. Enabled by default via
/// the `derive` feature; turn it off with
/// `mire = { ..., default-features = false }` if you don't want the
/// proc-macro dep.
#[cfg(feature = "derive")]
pub use mire_derive::EventData;
pub use projection::{EventHandler, HandledEvent, ProjectionRunner, ProjectionRunnerBuilder};
pub use snapshot::Snapshot;
pub use store::{CommittedEvents, EventStore, TransactionScope};
pub use stream::{ExpectedVersion, ReadDirection, StreamQuery};
pub use subscription::Subscription;

use serde::de::DeserializeOwned;

/// The domain model whose state is rebuilt by folding events.
///
/// **Why `Default`?** In event sourcing, an aggregate's state is *defined*
/// by its event history. `Default::default()` represents the state
/// *before any events have been applied* — a logically empty aggregate.
/// All real state arises through [`apply`](Self::apply). If your
/// aggregate has fields that look invalid in their default form (e.g.
/// `BankAccount { open: false, owner: "".into() }`), that's correct —
/// those values are *unreachable in practice* because the first event
/// (e.g. `Opened`) will set them. Mire constructs the empty aggregate,
/// then replays events onto it.
///
/// If your `apply` implementation guards against operating on a default
/// aggregate (e.g. early-returns when `!self.open`), the invariants
/// hold even if someone accidentally calls a method on a `Default::default()`
/// instance — the events that would unlock real behaviour simply
/// haven't been applied yet.
pub trait Aggregate: Default + Send + Sync {
    type Event: EventData;

    /// The category prefix for streams of this aggregate type. Streams
    /// are identified as `format!("{category}-{id}")`; the category is
    /// what `ProjectionRunner`s subscribe to and what
    /// `EventStore::read_category_after` filters on.
    fn stream_category() -> &'static str;

    /// Fold a single event into the aggregate's state. Must be
    /// deterministic and side-effect-free — it runs every time the
    /// aggregate is reloaded.
    fn apply(&mut self, event: &Self::Event);
}

pub struct AggregateRoot<A: Aggregate> {
    pub state: A,
    pub version: i64,
    pub stream_id: String,
    pending_events: Vec<A::Event>,
    pub metadata: EventMetadata,
}

impl<A: Aggregate> AggregateRoot<A> {
    pub fn new(id: &str) -> Self {
        let stream_id = format!("{}-{}", A::stream_category(), id);
        Self {
            state: A::default(),
            version: 0,
            stream_id,
            pending_events: Vec::new(),
            metadata: EventMetadata::default(),
        }
    }

    pub fn set_metadata(&mut self, metadata: EventMetadata) {
        self.metadata = metadata;
    }

    /// Build a root seeded from snapshotted `state` at `version`, with no
    /// pending events. The caller replays any events after `version` on top.
    pub fn from_snapshot(stream_id: String, state: A, version: i64) -> Self {
        Self {
            state,
            version,
            stream_id,
            pending_events: Vec::new(),
            metadata: EventMetadata::default(),
        }
    }

    pub fn hydrate(
        stream_id: String,
        events: &[RecordedEvent],
        version: i64,
    ) -> Result<Self, EventStoreError>
    where
        A::Event: DeserializeOwned,
    {
        let mut state = A::default();
        for recorded in events {
            let event =
                serde_json::from_value::<A::Event>(recorded.data.clone()).map_err(|source| {
                    EventStoreError::Deserialization {
                        stream_id: recorded.stream_id.clone(),
                        global_position: recorded.global_position,
                        event_type: recorded.event_type.clone(),
                        source,
                    }
                })?;
            state.apply(&event);
        }
        Ok(Self {
            state,
            version,
            stream_id,
            pending_events: Vec::new(),
            metadata: EventMetadata::default(),
        })
    }

    pub fn record(&mut self, event: A::Event) {
        self.state.apply(&event);
        self.pending_events.push(event);
    }

    /// Record many events in one call. Equivalent to calling
    /// [`record`](Self::record) in sequence; the convenience is that the
    /// subsequent [`EventStore::save`] writes all pending events in a
    /// single append transaction (amortising 8 round trips per save
    /// across N events). For high-throughput writers, this is the
    /// canonical batched-write recipe.
    pub fn record_many<I>(&mut self, events: I)
    where
        I: IntoIterator<Item = A::Event>,
    {
        for event in events {
            self.record(event);
        }
    }

    pub fn take_pending(&mut self) -> Vec<A::Event> {
        std::mem::take(&mut self.pending_events)
    }

    pub fn has_pending(&self) -> bool {
        !self.pending_events.is_empty()
    }

    pub fn pending_count(&self) -> usize {
        self.pending_events.len()
    }
}