mnesis_store/export.rs
1//! Export contract — generic over [`RawEventStore`], raw and box-agnostic.
2//!
3//! Export is a **generic store capability**, defined against
4//! [`RawEventStore`] only — it never touches a wire frame or an adapter
5//! partition, so it works for the in-memory store, fjall, and a future
6//! postgres store alike.
7//!
8//! Export does **no data manipulation**. `export_stream(id, from)` is a pass
9//! through to [`RawEventStore::read_stream`] — it yields the stored
10//! [`PersistedEnvelope`](crate::envelope::PersistedEnvelope)s verbatim. The events are not rewritten, the
11//! store-local `global_seq` is not stripped, and the stream id is not stamped
12//! onto each event. Two facts make that unnecessary:
13//!
14//! - **The caller supplies the id.** `export_stream(id, …)` is per-stream, so
15//! the caller already knows which stream the events belong to — exactly like
16//! a read. The stream id never has to ride on each event.
17//! - **Import re-appends.** On restore, import writes events through the
18//! normal append path, which stamps a fresh `global_seq` itself. The old
19//! store-local value simply rides along and is ignored — stripping it at
20//! export would be work import redoes for free.
21//!
22//! Two traits:
23//!
24//! - [`StreamLister`] — enumerate the stream ids a store holds. The one new
25//! store-layer capability export needs; an all-streams export is
26//! `list_streams` ∘ `export_stream`.
27//! - [`EventExporter`] — open a per-stream export (a raw read).
28//!
29//! The stream id *is* recorded once per stream — but in the **backup box**
30//! (the CBOR default, a later card), as a per-stream section heading, never on
31//! the events. A restore reads that heading to route the section back to the
32//! right stream; import then ignores each event's `global_seq`. See issue
33//! #145 §5.
34
35use futures::Stream;
36use mnesis::Version;
37
38use crate::store::{RawEventStore, Store};
39use crate::stream::EventStream;
40use crate::stream_id::StreamKey;
41
42/// Enumerate the stream ids present in a store.
43///
44/// The generic source of "which streams exist" — needed because a backup of
45/// an arbitrary store doesn't know its ids up front, and `export_stream`
46/// requires one. Yields the raw stream-id bytes (the form the store holds
47/// them in); the caller reconstitutes a typed [`Id`] if it needs one.
48///
49/// Lazy and async, mirroring [`RawEventStore::read_all`]: a store with many
50/// streams streams its ids rather than materializing them all.
51///
52/// Adapters back this with whatever index already tracks streams (fjall: its
53/// `streams` partition; in-memory: its map; postgres: `SELECT DISTINCT`).
54pub trait StreamLister: RawEventStore {
55 /// The stream of stream ids.
56 type StreamList: Stream<Item = Result<StreamKey, Self::Error>> + Send + 'static;
57
58 /// Open a one-shot stream over every stream id in the store, in no
59 /// guaranteed order, terminating when exhausted.
60 fn list_streams(
61 &self,
62 ) -> impl core::future::Future<Output = Result<Self::StreamList, Self::Error>> + Send;
63}
64
65/// Export a single stream's events — a raw pass-through read.
66///
67/// `export_stream(id, from)` reads stream `id` from `from` **inclusive** (the
68/// same semantics as [`RawEventStore::read_stream`]: `from = Version::INITIAL`
69/// yields the whole stream from v1) up to its current head, then terminates.
70/// Each yielded [`PersistedEnvelope`](crate::envelope::PersistedEnvelope) is the stored event **verbatim** — no
71/// rewrite, `global_seq` intact, no per-event stream id.
72///
73/// `from` is inclusive because the type forbids otherwise: [`Version`] is a
74/// `NonZeroU64` (minimum 1), so an exclusive `from` could never include v1 and
75/// a full export would be impossible. To **resume** after the last exported
76/// version `V`, pass `V.next()` (the caller's responsibility, mirroring how a
77/// subscription cursor resumes).
78///
79/// The stream is **pull-based**: it reads as polled, in bounded memory, so a
80/// consumer can write events to a file incrementally over any timespan.
81///
82/// `export_all` (`list_streams` ∘ `export_stream`) and continuous/live export
83/// (compose with the never-ending `subscribe` cursor) are consumer-side
84/// combinators, not part of this trait. The blanket impl below makes **every**
85/// [`RawEventStore`] an `EventExporter` for free.
86pub trait EventExporter: RawEventStore {
87 /// The stream of exported events. Identical to the read stream — export
88 /// performs no transform.
89 type ExportStream: EventStream<Error = Self::Error> + 'static;
90
91 /// Open a per-stream export of stream `id`, starting at `from` (inclusive).
92 fn export_stream(
93 &self,
94 id: &StreamKey,
95 from: Version,
96 ) -> impl core::future::Future<Output = Result<Self::ExportStream, Self::Error>> + Send;
97}
98
99/// Every [`RawEventStore`] is an [`EventExporter`] — export is just a read.
100///
101/// `export_stream` forwards to [`RawEventStore::read_stream`] unchanged, so the
102/// associated [`ExportStream`](EventExporter::ExportStream) is the adapter's
103/// own [`Stream`](RawEventStore::Stream) type: a concrete, monomorphized
104/// cursor with no boxing, no dynamic dispatch, and no per-event transform.
105impl<S: RawEventStore> EventExporter for S {
106 type ExportStream = S::Stream;
107
108 fn export_stream(
109 &self,
110 id: &StreamKey,
111 from: Version,
112 ) -> impl core::future::Future<Output = Result<Self::ExportStream, Self::Error>> + Send {
113 self.read_stream(id, from)
114 }
115}
116
117/// `Store<S>` forwards [`StreamLister`] to its inner backend (issue #247), so a
118/// handle holder can `store.list_streams()` without `.raw()`. `EventExporter`
119/// then applies to `Store<S>` via the blanket impl above (`Store<S>` is itself a
120/// [`RawEventStore`]).
121impl<S: StreamLister> StreamLister for Store<S> {
122 type StreamList = S::StreamList;
123
124 async fn list_streams(&self) -> Result<Self::StreamList, Self::Error> {
125 self.raw().list_streams().await
126 }
127}