mnesis_store_testing/
row.rs

1//! Shared test-data row, id type, and drive/drain helpers used by every
2//! conformance module.
3
4use core::fmt;
5
6use bytes::Bytes;
7use futures::StreamExt;
8use futures::pin_mut;
9use mnesis::Version;
10use mnesis_store::envelope::{PendingEnvelope, PersistedEnvelope, pending_envelope};
11use mnesis_store::store::RawEventStore;
12use mnesis_store::value::SchemaVersion;
13use mnesis_store::{PendingBatch, StreamKey};
14
15/// One row of test data fed into an adapter for the conformance suite to
16/// observe back out. All fields must round-trip byte-for-byte.
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub struct ConformanceRow {
19    pub version: u64,
20    pub event_type: String,
21    pub schema_version: u32,
22    pub payload: Vec<u8>,
23    pub metadata: Option<Vec<u8>>,
24}
25
26impl ConformanceRow {
27    /// Convenience constructor: `schema_version = 1`, no metadata.
28    #[must_use]
29    pub fn new(version: u64, event_type: &str, payload: Vec<u8>) -> Self {
30        Self {
31            version,
32            event_type: event_type.to_owned(),
33            schema_version: 1,
34            payload,
35            metadata: None,
36        }
37    }
38
39    /// Set the schema version (defaults to 1).
40    #[must_use]
41    pub const fn with_schema_version(mut self, schema_version: u32) -> Self {
42        self.schema_version = schema_version;
43        self
44    }
45
46    /// Attach metadata (defaults to absent).
47    ///
48    /// `metadata` must be non-empty — empty metadata is unrepresentable
49    /// (`ValueError::MetadataEmpty`); use absent instead.
50    #[must_use]
51    pub fn with_metadata(mut self, metadata: Vec<u8>) -> Self {
52        self.metadata = Some(metadata);
53        self
54    }
55}
56
57/// Subscription/snapshot id: satisfies the `Id` blanket bounds.
58#[derive(Debug, Clone, Hash, PartialEq, Eq)]
59pub struct SubId(String);
60
61impl SubId {
62    #[must_use]
63    pub fn new(s: &str) -> Self {
64        Self(s.to_owned())
65    }
66
67    /// The `StreamKey` carrying the same bytes — for driving `append` on the
68    /// stream this id subscribes to.
69    #[must_use]
70    pub fn key(&self) -> StreamKey {
71        StreamKey::from_slice(self.0.as_bytes())
72    }
73}
74
75impl fmt::Display for SubId {
76    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
77        f.write_str(&self.0)
78    }
79}
80
81impl AsRef<[u8]> for SubId {
82    fn as_ref(&self) -> &[u8] {
83        self.0.as_bytes()
84    }
85}
86
87/// Build the `PendingEnvelope` a row describes. Panics on invalid rows — the
88/// suite only constructs valid ones.
89#[must_use]
90pub fn envelope_for(row: &ConformanceRow) -> PendingEnvelope {
91    let version = Version::new(row.version).expect("row version must be >= 1");
92    let mut staged = pending_envelope(version)
93        .event_type_bytes(Bytes::from(row.event_type.clone().into_bytes()))
94        .expect("valid event type")
95        .payload(row.payload.clone())
96        .schema_version(
97            SchemaVersion::from_u32(row.schema_version).expect("schema_version must be >= 1"),
98        );
99    if let Some(m) = &row.metadata {
100        staged = staged.metadata(m.clone());
101    }
102    staged.build().expect("valid envelope")
103}
104
105/// Read a `PersistedEnvelope` back into row form.
106#[must_use]
107pub fn row_of(env: &PersistedEnvelope) -> ConformanceRow {
108    ConformanceRow {
109        version: env.version().as_u64(),
110        event_type: env.event_type().to_owned(),
111        schema_version: env.schema_version(),
112        payload: env.payload().to_vec(),
113        metadata: env.metadata().map(<[u8]>::to_vec),
114    }
115}
116
117/// Append `rows` to `id` as one batch on a fresh stream (`expected = None`).
118pub async fn append_rows<S: RawEventStore>(store: &S, id: &StreamKey, rows: &[ConformanceRow]) {
119    if rows.is_empty() {
120        return;
121    }
122    let envs: Vec<PendingEnvelope> = rows.iter().map(envelope_for).collect();
123    store
124        .append(
125            id,
126            None,
127            PendingBatch::new(&envs).expect("kit batches are non-empty"),
128        )
129        .await
130        .unwrap_or_else(|e| panic!("append of {} rows failed: {e:?}", rows.len()));
131}
132
133/// Append one event at `version` with the matching optimistic expectation
134/// (`None` for version 1). Panics on failure — callers drive clean sequences.
135pub async fn append_event<S: RawEventStore>(
136    store: &S,
137    id: &StreamKey,
138    version: u64,
139    payload: &[u8],
140) {
141    append_event_at(store, id, version, payload).await;
142}
143
144/// Like [`append_event`] but returns the `$all` position the append assigned —
145/// the read-your-writes token (#330). The version-only callers use
146/// [`append_event`]; position checks use this.
147pub async fn append_event_at<S: RawEventStore>(
148    store: &S,
149    id: &StreamKey,
150    version: u64,
151    payload: &[u8],
152) -> S::AllPosition {
153    let expected = Version::new(version.saturating_sub(1));
154    let env = envelope_for(&ConformanceRow::new(version, "E", payload.to_vec()));
155    store
156        .append(id, expected, PendingBatch::of(&env))
157        .await
158        .unwrap_or_else(|e| panic!("append v{version} failed: {e:?}"))
159}
160
161/// Drain `read_stream(id, from)` fully into rows.
162pub async fn drain_stream<S: RawEventStore>(
163    store: &S,
164    id: &StreamKey,
165    from: Version,
166) -> Vec<ConformanceRow> {
167    let stream = store
168        .read_stream(id, from)
169        .await
170        .unwrap_or_else(|e| panic!("read_stream failed: {e:?}"));
171    pin_mut!(stream);
172    let mut out = Vec::new();
173    while let Some(item) = stream.next().await {
174        let env = item.unwrap_or_else(|e| panic!("read_stream item errored: {e:?}"));
175        out.push(row_of(&env));
176    }
177    out
178}
179
180/// Drain `read_all(from)` fully into `(position, payload)` pairs.
181pub async fn drain_all<S: RawEventStore>(
182    store: &S,
183    from: Option<S::AllPosition>,
184) -> Vec<(S::AllPosition, Vec<u8>)> {
185    let stream = store
186        .read_all(from)
187        .await
188        .unwrap_or_else(|e| panic!("read_all failed: {e:?}"));
189    pin_mut!(stream);
190    let mut out = Vec::new();
191    while let Some(item) = stream.next().await {
192        let (pos, _key, env) = item.unwrap_or_else(|e| panic!("read_all item errored: {e:?}"));
193        out.push((pos, env.payload().to_vec()));
194    }
195    out
196}
197
198/// Drain `read_all(from)` into `(position, stream-key bytes, payload)` triples —
199/// the attribution-aware form (`drain_all` drops the key). Used by the #333
200/// stream-attribution check.
201pub async fn drain_all_attributed<S: RawEventStore>(
202    store: &S,
203    from: Option<S::AllPosition>,
204) -> Vec<(S::AllPosition, Vec<u8>, Vec<u8>)> {
205    let stream = store
206        .read_all(from)
207        .await
208        .unwrap_or_else(|e| panic!("read_all failed: {e:?}"));
209    pin_mut!(stream);
210    let mut out = Vec::new();
211    while let Some(item) = stream.next().await {
212        let (pos, key, env) = item.unwrap_or_else(|e| panic!("read_all item errored: {e:?}"));
213        out.push((pos, key.as_bytes().to_vec(), env.payload().to_vec()));
214    }
215    out
216}
217
218/// Drain `read_all(from)` into `(position, metadata, payload)` triples.
219pub async fn drain_all_with_metadata<S: RawEventStore>(
220    store: &S,
221    from: Option<S::AllPosition>,
222) -> Vec<(S::AllPosition, Option<Vec<u8>>, Vec<u8>)> {
223    let stream = store
224        .read_all(from)
225        .await
226        .unwrap_or_else(|e| panic!("read_all failed: {e:?}"));
227    pin_mut!(stream);
228    let mut out = Vec::new();
229    while let Some(item) = stream.next().await {
230        let (pos, _key, env) = item.unwrap_or_else(|e| panic!("read_all item errored: {e:?}"));
231        out.push((
232            pos,
233            env.metadata().map(<[u8]>::to_vec),
234            env.payload().to_vec(),
235        ));
236    }
237    out
238}
239
240/// Assert positions strictly increase (monotonic, no duplicate).
241pub fn assert_strictly_increasing<P: Copy + Ord + fmt::Debug>(positions: &[(P, Vec<u8>)]) {
242    for w in positions.windows(2) {
243        assert!(
244            w[1].0 > w[0].0,
245            "$all positions must be strictly increasing: {:?} then {:?}",
246            w[0].0,
247            w[1].0,
248        );
249    }
250}
mnesis_store_testing/row.rs

mnesis_store_testing/
row.rs
