liminal_sdk/remote/participant/
replay_apply.rs

1use liminal_protocol::client::{
2    ApplyAttachDecision, ApplyDetachOutcomeDecision, ApplyLeaveDecision, ClientResponseCorrelation,
3    DetachReplayOutcome, DetachReplayRefusalReason, apply_attach, apply_detach_outcome,
4    apply_leave_durable,
5};
6use liminal_protocol::wire::{AttachBound, LeaveCommitted};
7
8use super::{
9    ParticipantResumeStore, RemoteParticipantError, RemoteParticipantHandle,
10    RemoteReplayApplyOutcome, persist, take_aggregate,
11};
12
13impl<S: ParticipantResumeStore> RemoteParticipantHandle<S> {
14    /// Delegates an attach supersession input with the held one-use correlation.
15    ///
16    /// # Errors
17    ///
18    /// Returns when no response authority exists or persistence fails.
19    pub fn apply_attach(
20        &self,
21        attach: AttachBound,
22    ) -> Result<RemoteReplayApplyOutcome<AttachBound>, RemoteParticipantError> {
23        self.apply_replay_input(attach, |aggregate, input, correlation| {
24            match apply_attach(aggregate, input, correlation) {
25                ApplyAttachDecision::Superseded(applied) => (applied.into_aggregate(), None),
26                ApplyAttachDecision::Refused(refusal) => {
27                    let reason = refusal.reason();
28                    let (aggregate, (input, correlation)) = refusal.into_parts();
29                    (aggregate, Some((input, correlation, reason)))
30                }
31            }
32        })
33    }
34
35    /// Delegates a durable Leave supersession input with held correlation.
36    ///
37    /// # Errors
38    ///
39    /// Returns when no response authority exists or persistence fails.
40    pub fn apply_leave_durable(
41        &self,
42        leave: LeaveCommitted,
43    ) -> Result<RemoteReplayApplyOutcome<LeaveCommitted>, RemoteParticipantError> {
44        self.apply_replay_input(
45            leave,
46            |aggregate, input, correlation| match apply_leave_durable(aggregate, input, correlation)
47            {
48                ApplyLeaveDecision::Superseded(applied) => (applied.into_aggregate(), None),
49                ApplyLeaveDecision::Refused(refusal) => {
50                    let reason = refusal.reason();
51                    let (aggregate, (input, correlation)) = refusal.into_parts();
52                    (aggregate, Some((input, correlation, reason)))
53                }
54            },
55        )
56    }
57
58    /// Delegates one typed terminal detach outcome with held correlation.
59    ///
60    /// # Errors
61    ///
62    /// Returns when no response authority exists or persistence fails.
63    pub fn apply_detach_outcome(
64        &self,
65        outcome: DetachReplayOutcome,
66    ) -> Result<RemoteReplayApplyOutcome<DetachReplayOutcome>, RemoteParticipantError> {
67        self.apply_replay_input(
68            outcome,
69            |aggregate, input, correlation| match apply_detach_outcome(
70                aggregate,
71                input,
72                correlation,
73            ) {
74                ApplyDetachOutcomeDecision::Terminal(applied) => (applied.into_aggregate(), None),
75                ApplyDetachOutcomeDecision::Refused(refusal) => {
76                    let reason = refusal.reason();
77                    let (aggregate, (input, correlation)) = refusal.into_parts();
78                    (aggregate, Some((input, correlation, reason)))
79                }
80            },
81        )
82    }
83
84    fn apply_replay_input<T>(
85        &self,
86        input: T,
87        decide: impl FnOnce(
88            liminal_protocol::client::ClientParticipantAggregate,
89            T,
90            ClientResponseCorrelation,
91        ) -> (
92            liminal_protocol::client::ClientParticipantAggregate,
93            Option<(T, ClientResponseCorrelation, DetachReplayRefusalReason)>,
94        ),
95    ) -> Result<RemoteReplayApplyOutcome<T>, RemoteParticipantError> {
96        let mut state = self.state.lock();
97        let correlation = state
98            .correlation
99            .take()
100            .ok_or(RemoteParticipantError::ResponseAuthorityUnavailable)?;
101        let aggregate = take_aggregate(&mut state)?;
102        let (aggregate, refusal) = decide(aggregate, input, correlation);
103        if let Some((input, correlation, reason)) = refusal {
104            state.aggregate = Some(aggregate);
105            state.correlation = Some(correlation);
106            Ok(RemoteReplayApplyOutcome::Refused { input, reason })
107        } else {
108            persist(&mut state.store, &aggregate)?;
109            state.aggregate = Some(aggregate);
110            Ok(RemoteReplayApplyOutcome::Applied)
111        }
112    }
113}
liminal_sdk/remote/participant/replay_apply.rs

liminal_sdk/remote/participant/
replay_apply.rs
