kmp_application/commands/
command_application_service.rs

1use std::sync::Arc;
2
3#[cfg(test)]
4#[path = "command_application_service_tests.rs"]
5mod tests;
6
7use kmp_domain::{ContextEventStore, ProjectionWriter};
8
9use crate::ApplicationError;
10use crate::commands::{
11    NoopProjectionWriter, UpdateContextCommand, UpdateContextOutcome, UpdateContextUseCase,
12};
13
14#[derive(Debug)]
15pub struct CommandApplicationService<E, W = NoopProjectionWriter> {
16    update_context: Arc<UpdateContextUseCase<E, W>>,
17    // Local ordering for split-store adapters. SQLite additionally validates
18    // the read revisions and projects atomically across independent engines.
19    projection_access: tokio::sync::RwLock<()>,
20}
21
22impl<E, W> CommandApplicationService<E, W>
23where
24    E: ContextEventStore + Send + Sync,
25    W: ProjectionWriter + Send + Sync,
26{
27    pub fn new(update_context: Arc<UpdateContextUseCase<E, W>>) -> Self {
28        Self {
29            update_context,
30            projection_access: tokio::sync::RwLock::new(()),
31        }
32    }
33
34    pub async fn update_context(
35        &self,
36        command: UpdateContextCommand,
37    ) -> Result<UpdateContextOutcome, ApplicationError> {
38        self.update_context_after_read(command, &std::collections::BTreeMap::new())
39            .await
40    }
41
42    pub(crate) async fn projection_read(&self) -> tokio::sync::RwLockReadGuard<'_, ()> {
43        self.projection_access.read().await
44    }
45
46    pub(crate) async fn memory_revision(&self, about: &str) -> Result<u64, ApplicationError> {
47        self.update_context.memory_revision(about).await
48    }
49
50    /// Recheck every explicitly read about while excluding concurrent writes,
51    /// including writes to a foreign equivalence endpoint. A stale view applies
52    /// nothing. Existing idempotent outcomes take precedence over fresh context.
53    pub(crate) async fn update_context_after_read(
54        &self,
55        command: UpdateContextCommand,
56        revisions: &std::collections::BTreeMap<String, u64>,
57    ) -> Result<UpdateContextOutcome, ApplicationError> {
58        let _guard = self.projection_access.write().await;
59        let accepted = match command.idempotency_key.as_deref() {
60            Some(key) => self.accepted_outcome(key).await?.is_some(),
61            None => false,
62        };
63        if !accepted {
64            for (about, revision) in revisions {
65                if self.memory_revision(about).await? != *revision {
66                    return Err(ApplicationError::RetryableConflict(
67                        "write neighborhood changed during commit; retry the same logical write to refresh it".into(),
68                    ));
69                }
70            }
71        }
72        self.update_context
73            .execute_after_read(command, revisions)
74            .await
75    }
76
77    /// What an idempotency key was already accepted with, if anything.
78    pub async fn accepted_outcome(
79        &self,
80        idempotency_key: &str,
81    ) -> Result<Option<kmp_domain::IdempotentOutcome>, ApplicationError> {
82        self.update_context.accepted_outcome(idempotency_key).await
83    }
84}
kmp_application/commands/command_application_service.rs

kmp_application/commands/
command_application_service.rs
