eventcore_postgres/projections/
projector.rs

1use std::convert::Infallible;
2use std::error::Error;
3use std::future::Future;
4
5use eventcore_types::{AttemptNumber, DeliveryPosition, PersistedEventEnvelope, ProjectorName};
6use serde::de::DeserializeOwned;
7use sqlx::{Postgres, Transaction};
8
9use super::error::BoxedProjectionError;
10
11/// Work that is safe to perform only after the read-model transaction commits.
12pub trait AfterCommit: Send + 'static {
13    /// Error returned by the after-commit action.
14    type Error: Error + Send + Sync + 'static;
15
16    /// Executes the action after the transaction's commit was acknowledged.
17    fn execute(self) -> impl Future<Output = Result<(), Self::Error>> + Send;
18}
19
20/// An after-commit action that has no work to perform.
21#[derive(Debug, Clone, Copy, Default)]
22pub struct NoopAfterCommit;
23
24impl AfterCommit for NoopAfterCommit {
25    type Error = Infallible;
26
27    async fn execute(self) -> Result<(), Self::Error> {
28        Ok(())
29    }
30}
31
32/// Decision made by an application after its event effect failed.
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34#[non_exhaustive]
35pub enum ProjectionFailureDecision {
36    /// Roll back and retry the event, subject to the configured retry policy.
37    Retry,
38    /// Roll back the failed effect and explicitly advance past the event.
39    Skip,
40    /// Roll back and end this run without advancing progress.
41    Stop,
42    /// Roll back and return the application error.
43    Fatal,
44}
45
46/// Public failure information passed to [`PostgresProjector::on_error`].
47#[derive(Debug)]
48pub struct ProjectionFailureContext<'a, E> {
49    position: DeliveryPosition,
50    error: &'a E,
51    attempt: AttemptNumber,
52}
53
54impl<'a, E> ProjectionFailureContext<'a, E> {
55    /// Creates failure context for a projector invocation.
56    pub fn new(position: DeliveryPosition, error: &'a E, attempt: AttemptNumber) -> Self {
57        Self {
58            position,
59            error,
60            attempt,
61        }
62    }
63
64    /// Returns the delivery position that failed.
65    pub fn position(&self) -> DeliveryPosition {
66        self.position
67    }
68
69    /// Returns the application error.
70    pub fn error(&self) -> &'a E {
71        self.error
72    }
73
74    /// Returns the one-based attempt number.
75    pub fn attempt(&self) -> AttemptNumber {
76        self.attempt
77    }
78}
79
80/// Applies decoded events through the transaction owned by the PostgreSQL runner.
81pub trait PostgresProjector: Send {
82    /// Decoded event contract accepted by this projection.
83    type Event: DeserializeOwned + Send + Sync;
84    /// Application error returned while applying an event.
85    type Error: Error + Send + Sync + 'static;
86    /// Work deferred until the event transaction commits.
87    type AfterCommit: AfterCommit;
88
89    /// Returns this projection's stable, application-supplied name.
90    fn name(&self) -> &ProjectorName;
91
92    /// Decodes a persisted envelope into this projection's application event.
93    ///
94    /// Applications own this boundary and may inspect the envelope's persisted
95    /// event type, metadata, stream identity, or other fields before decoding.
96    /// The default preserves payload-only JSON decoding for existing projectors.
97    /// Any error returned here is terminal and is reported as
98    /// [`super::TransactionalProjectionError::Decode`].
99    fn decode(
100        &self,
101        envelope: &PersistedEventEnvelope,
102    ) -> Result<Self::Event, BoxedProjectionError> {
103        serde_json::from_str(envelope.payload().get())
104            .map_err(|error| -> BoxedProjectionError { Box::new(error) })
105    }
106
107    /// Applies one event through the runner-owned read-model transaction.
108    fn apply<'a, 'c>(
109        &'a mut self,
110        event: &'a Self::Event,
111        position: DeliveryPosition,
112        tx: &'a mut Transaction<'c, Postgres>,
113    ) -> impl Future<Output = Result<Self::AfterCommit, Self::Error>> + Send + 'a
114    where
115        'c: 'a;
116
117    /// Chooses how the runner handles an application failure.
118    fn on_error(
119        &mut self,
120        failure: ProjectionFailureContext<'_, Self::Error>,
121    ) -> ProjectionFailureDecision {
122        let _ = failure;
123        ProjectionFailureDecision::Fatal
124    }
125}
126
127/// Application-owned read-model reinitialization performed in the leader transaction.
128pub trait PostgresProjectionReset: Send {
129    /// Error returned by application reset code.
130    type Error: Error + Send + Sync + 'static;
131
132    /// Reinitializes the read model using the runner-owned transaction.
133    fn reset<'a, 'c>(
134        &'a mut self,
135        tx: &'a mut Transaction<'c, Postgres>,
136    ) -> impl Future<Output = Result<(), Self::Error>> + Send + 'a
137    where
138        'c: 'a;
139}
eventcore_postgres/projections/projector.rs

eventcore_postgres/projections/
projector.rs
