distributed/repository/
async_repository.rs

1use std::future::Future;
2
3use crate::entity::{Entity, EventRecord};
4use crate::outbox::OutboxMessage;
5use crate::read_model::{
6    ReadModelAdapterCapabilities, ReadModelCommitOutcome, ReadModelError, ReadModelLoadGraph,
7    ReadModelLoadRequest, ReadModelQueryCapabilities, ReadModelWritePlan,
8};
9use crate::snapshot::SnapshotRecord;
10
11use super::inbox::InboxReceipt;
12use super::{RepositoryError, StreamIdentity};
13
14/// One aggregate event stream staged for an async transactional commit.
15pub struct AsyncStreamWrite<'a> {
16    pub identity: StreamIdentity,
17    pub entity: &'a mut Entity,
18}
19
20impl<'a> AsyncStreamWrite<'a> {
21    pub fn new(identity: StreamIdentity, entity: &'a mut Entity) -> Self {
22        Self { identity, entity }
23    }
24}
25
26/// Snapshot writes staged in an async transactional commit.
27#[derive(Clone, Debug)]
28pub enum AsyncSnapshotWrite {
29    Save {
30        identity: StreamIdentity,
31        record: SnapshotRecord,
32    },
33}
34
35/// A structured async write batch that must commit under one backend transaction.
36pub struct AsyncCommitBatch<'a> {
37    pub streams: Vec<AsyncStreamWrite<'a>>,
38    pub outbox_messages: Vec<OutboxMessage>,
39    pub read_model_plans: Vec<ReadModelWritePlan>,
40    pub snapshots: Vec<AsyncSnapshotWrite>,
41    /// Consumer inbox receipts to record in the same transaction (the optional
42    /// effectively-once effect fence). Empty for the default idempotent path.
43    pub inbox_receipts: Vec<InboxReceipt>,
44}
45
46impl<'a> AsyncCommitBatch<'a> {
47    pub fn new(streams: Vec<AsyncStreamWrite<'a>>) -> Self {
48        Self {
49            streams,
50            outbox_messages: Vec::new(),
51            read_model_plans: Vec::new(),
52            snapshots: Vec::new(),
53            inbox_receipts: Vec::new(),
54        }
55    }
56
57    pub fn empty() -> Self {
58        Self::new(Vec::new())
59    }
60}
61
62/// Owned append data prepared from a borrowed stream write before async I/O.
63#[derive(Clone, Debug)]
64pub struct PreparedEventAppend {
65    pub identity: StreamIdentity,
66    pub expected_version: u64,
67    pub events: Vec<EventRecord>,
68}
69
70impl PreparedEventAppend {
71    pub fn from_stream_write(write: &AsyncStreamWrite<'_>) -> Self {
72        Self {
73            identity: write.identity.clone(),
74            expected_version: write.entity.committed_version(),
75            events: write.entity.new_events().to_vec(),
76        }
77    }
78}
79
80/// Async stream-aware aggregate loading.
81pub trait AsyncGetStream: Send + Sync {
82    fn get_stream<'a>(
83        &'a self,
84        identity: &'a StreamIdentity,
85    ) -> impl Future<Output = Result<Option<Entity>, RepositoryError>> + Send + 'a;
86
87    fn get_streams<'a>(
88        &'a self,
89        identities: &'a [StreamIdentity],
90    ) -> impl Future<Output = Result<Vec<Entity>, RepositoryError>> + Send + 'a;
91}
92
93/// Async transactional commit capability for durable persistence backends.
94pub trait AsyncTransactionalCommit: Send + Sync {
95    fn commit_batch_async<'a>(
96        &'a self,
97        batch: AsyncCommitBatch<'a>,
98    ) -> impl Future<Output = Result<(), RepositoryError>> + Send + 'a;
99}
100
101/// Consumer inbox read capability: check whether a `(consumer, message_id)`
102/// receipt has already been recorded.
103///
104/// The pre-check lets a consumer skip re-running a handler for an already-processed
105/// message (and ack the redelivery) before opening a transaction. The
106/// authoritative dedupe is still the receipt's `(consumer, message_id)` primary
107/// key written in [`commit_batch_async`](AsyncTransactionalCommit::commit_batch_async),
108/// which fences the race where two deliveries both pass the pre-check.
109pub trait AsyncInboxStore: Send + Sync {
110    fn inbox_contains_async<'a>(
111        &'a self,
112        consumer: &'a str,
113        message_id: &'a str,
114    ) -> impl Future<Output = Result<bool, RepositoryError>> + Send + 'a;
115}
116
117/// Repository trait for types that implement async stream reads and commits.
118pub trait AsyncRepository: AsyncGetStream + AsyncTransactionalCommit {}
119
120impl<T> AsyncRepository for T where T: AsyncGetStream + AsyncTransactionalCommit {}
121
122/// Async adapter contract for committing read-model write plans.
123pub trait AsyncReadModelWritePlanStore: Send + Sync {
124    fn read_model_capabilities_async(&self) -> ReadModelAdapterCapabilities;
125
126    fn commit_write_plan_async(
127        &self,
128        plan: ReadModelWritePlan,
129    ) -> impl Future<Output = Result<ReadModelCommitOutcome, ReadModelError>> + Send + '_;
130}
131
132/// Async primary-key relational read-model query contract.
133pub trait AsyncRelationalReadModelQueryStore: Send + Sync {
134    fn read_model_query_capabilities_async(&self) -> ReadModelQueryCapabilities;
135
136    fn load_graph_async(
137        &self,
138        request: ReadModelLoadRequest,
139    ) -> impl Future<Output = Result<ReadModelLoadGraph, ReadModelError>> + Send + '_;
140}
141
142/// Async snapshot persistence keyed by full stream identity.
143pub trait AsyncSnapshotStore: Send + Sync {
144    fn get_snapshot_async<'a>(
145        &'a self,
146        identity: &'a StreamIdentity,
147    ) -> impl Future<Output = Result<Option<SnapshotRecord>, RepositoryError>> + Send + 'a;
148
149    fn save_snapshot_async<'a>(
150        &'a self,
151        identity: &'a StreamIdentity,
152        record: SnapshotRecord,
153    ) -> impl Future<Output = Result<(), RepositoryError>> + Send + 'a;
154
155    fn delete_snapshot_async<'a>(
156        &'a self,
157        identity: &'a StreamIdentity,
158    ) -> impl Future<Output = Result<bool, RepositoryError>> + Send + 'a;
159}
distributed/repository/async_repository.rs

distributed/repository/
async_repository.rs
