1use kmp_domain::{
2 NodeDetailProjection, NodeProjection, NodeRelationProjection, PortError, ProcessedEventStore,
3 ProjectionCheckpoint, ProjectionCheckpointStore, ProjectionEvent, ProjectionEventHandler,
4 ProjectionHandlingRequest, ProjectionHandlingResult, ProjectionMutation, ProjectionWriter,
5};
6
7#[derive(Debug)]
8pub struct ProjectionApplicationService<W, P, C> {
9 projection_writer: W,
10 processed_event_store: P,
11 checkpoint_store: C,
12}
13
14impl<W, P, C> ProjectionApplicationService<W, P, C>
15where
16 W: ProjectionWriter + Send + Sync,
17 P: ProcessedEventStore + Send + Sync,
18 C: ProjectionCheckpointStore + Send + Sync,
19{
20 pub fn new(projection_writer: W, processed_event_store: P, checkpoint_store: C) -> Self {
21 Self {
22 projection_writer,
23 processed_event_store,
24 checkpoint_store,
25 }
26 }
27
28 async fn mutations_for_event(
29 &self,
30 event: &ProjectionEvent,
31 ) -> Result<Vec<ProjectionMutation>, PortError> {
32 Ok(match event {
33 ProjectionEvent::GraphNodeMaterialized(event) => {
34 let provenance = event
35 .data
36 .source_kind
37 .as_deref()
38 .and_then(|sk| kmp_domain::SourceKind::parse(sk).ok())
39 .map(|sk| {
40 let mut p = kmp_domain::Provenance::new(sk);
41 if let Some(ref agent) = event.data.source_agent {
42 p = p.with_source_agent(agent.clone());
43 }
44 if let Some(ref observed) = event.data.observed_at {
45 p = p.with_observed_at(observed.clone());
46 }
47 p
48 });
49 let mut mutations = vec![ProjectionMutation::UpsertNode(NodeProjection {
50 node_id: event.data.node_id.clone(),
51 node_kind: event.data.node_kind.clone(),
52 title: event.data.title.clone(),
53 summary: event.data.summary.clone(),
54 status: event.data.status.clone(),
55 labels: event.data.labels.clone(),
56 properties: event.data.properties.clone(),
57 provenance,
58 })];
59 mutations.extend(
60 event
61 .data
62 .related_nodes
63 .iter()
64 .map(|reference| {
65 Ok(ProjectionMutation::UpsertNodeRelation(Box::new(
66 NodeRelationProjection {
67 source_node_id: event.data.node_id.clone(),
68 target_node_id: reference.node_id.clone(),
69 relation_type: reference.relation_type.clone(),
70 explanation: reference.explanation.clone().try_into().map_err(|error| {
71 PortError::InvalidState(format!(
72 "invalid related node explanation for `{}` -> `{}`: {error}",
73 event.data.node_id, reference.node_id
74 ))
75 })?,
76 },
77 )))
78 })
79 .collect::<Result<Vec<_>, PortError>>()?,
80 );
81 mutations
82 }
83 ProjectionEvent::GraphRelationMaterialized(event) => {
84 vec![ProjectionMutation::UpsertNodeRelation(Box::new(
85 NodeRelationProjection {
86 source_node_id: event.data.source_node_id.clone(),
87 target_node_id: event.data.target_node_id.clone(),
88 relation_type: event.data.relation_type.clone(),
89 explanation: event.data.explanation.clone().try_into().map_err(
90 |error| {
91 PortError::InvalidState(format!(
92 "invalid relation explanation for `{}` -> `{}`: {error}",
93 event.data.source_node_id, event.data.target_node_id
94 ))
95 },
96 )?,
97 },
98 ))]
99 }
100 ProjectionEvent::NodeDetailMaterialized(event) => {
101 vec![ProjectionMutation::UpsertNodeDetail(NodeDetailProjection {
102 node_id: event.data.node_id.clone(),
103 detail: event.data.detail.clone(),
104 content_hash: event.data.content_hash.clone(),
105 revision: event.data.revision,
106 })]
107 }
108 })
109 }
110}
111
112impl<W, P, C> ProjectionEventHandler for ProjectionApplicationService<W, P, C>
113where
114 W: ProjectionWriter + Send + Sync,
115 P: ProcessedEventStore + Send + Sync,
116 C: ProjectionCheckpointStore + Send + Sync,
117{
118 async fn handle_projection_event(
119 &self,
120 request: ProjectionHandlingRequest,
121 ) -> Result<ProjectionHandlingResult, PortError> {
122 let event_id = request.event.event_id().to_string();
123 if self
124 .processed_event_store
125 .has_processed(&request.consumer_name, &event_id)
126 .await?
127 {
128 return Ok(ProjectionHandlingResult {
129 event_id,
130 subject: request.subject,
131 duplicate: true,
132 applied_mutations: 0,
133 checkpoint: None,
134 });
135 }
136
137 let mutations = self.mutations_for_event(&request.event).await?;
138 self.projection_writer
139 .apply_mutations(mutations.clone())
140 .await?;
141 self.processed_event_store
142 .record_processed(&request.consumer_name, &event_id)
143 .await?;
144 let checkpoint = ProjectionCheckpoint {
145 consumer_name: request.consumer_name,
146 stream_name: request.stream_name,
147 last_subject: request.subject.clone(),
148 last_event_id: event_id.clone(),
149 last_correlation_id: request.event.envelope().correlation_id.clone(),
150 last_occurred_at: request.event.envelope().occurred_at.clone(),
151 processed_events: 1,
152 updated_at: std::time::SystemTime::now(),
153 };
154 self.checkpoint_store
155 .save_checkpoint(checkpoint.clone())
156 .await?;
157
158 Ok(ProjectionHandlingResult {
159 event_id,
160 subject: request.subject,
161 duplicate: false,
162 applied_mutations: mutations.len(),
163 checkpoint: Some(checkpoint),
164 })
165 }
166}
167
168#[cfg(test)]
169mod tests {
170 use std::sync::Arc;
171
172 use kmp_domain::{
173 GraphRelationMaterializedData, GraphRelationMaterializedEvent, PortError,
174 ProcessedEventStore, ProjectionCheckpoint, ProjectionCheckpointStore, ProjectionEnvelope,
175 ProjectionEvent, ProjectionEventHandler, ProjectionHandlingRequest, ProjectionMutation,
176 ProjectionWriter, RelatedNodeExplanationData, RelationSemanticClass,
177 };
178 use tokio::sync::Mutex;
179
180 use super::ProjectionApplicationService;
181
182 #[derive(Debug, Default, Clone)]
183 struct RecordingProjectionWriter {
184 mutations: Arc<Mutex<Vec<ProjectionMutation>>>,
185 }
186
187 impl RecordingProjectionWriter {
188 async fn mutations(&self) -> Vec<ProjectionMutation> {
189 self.mutations.lock().await.clone()
190 }
191 }
192
193 impl ProjectionWriter for RecordingProjectionWriter {
194 async fn apply_mutations(
195 &self,
196 mutations: Vec<ProjectionMutation>,
197 ) -> Result<(), PortError> {
198 self.mutations.lock().await.extend(mutations);
199 Ok(())
200 }
201 }
202
203 #[derive(Debug, Default, Clone)]
204 struct RecordingProcessedEventStore {
205 processed: Arc<Mutex<Vec<(String, String)>>>,
206 }
207
208 impl ProcessedEventStore for RecordingProcessedEventStore {
209 async fn has_processed(
210 &self,
211 consumer_name: &str,
212 event_id: &str,
213 ) -> Result<bool, PortError> {
214 Ok(self
215 .processed
216 .lock()
217 .await
218 .iter()
219 .any(|(consumer, event)| consumer == consumer_name && event == event_id))
220 }
221
222 async fn record_processed(
223 &self,
224 consumer_name: &str,
225 event_id: &str,
226 ) -> Result<(), PortError> {
227 self.processed
228 .lock()
229 .await
230 .push((consumer_name.to_string(), event_id.to_string()));
231 Ok(())
232 }
233 }
234
235 #[derive(Debug, Default, Clone)]
236 struct RecordingCheckpointStore {
237 checkpoints: Arc<Mutex<Vec<ProjectionCheckpoint>>>,
238 }
239
240 impl ProjectionCheckpointStore for RecordingCheckpointStore {
241 async fn save_checkpoint(&self, checkpoint: ProjectionCheckpoint) -> Result<(), PortError> {
242 self.checkpoints.lock().await.push(checkpoint);
243 Ok(())
244 }
245
246 async fn load_checkpoint(
247 &self,
248 _consumer_name: &str,
249 _stream_name: &str,
250 ) -> Result<Option<ProjectionCheckpoint>, PortError> {
251 Ok(self.checkpoints.lock().await.last().cloned())
252 }
253 }
254
255 #[tokio::test]
256 async fn relation_materialized_event_upserts_relation_directly() {
257 let writer = RecordingProjectionWriter::default();
258 let service = ProjectionApplicationService::new(
259 writer.clone(),
260 RecordingProcessedEventStore::default(),
261 RecordingCheckpointStore::default(),
262 );
263
264 let request = ProjectionHandlingRequest {
265 consumer_name: "projection-consumer".to_string(),
266 stream_name: "rehydration.events".to_string(),
267 subject: "graph.relation.materialized".to_string(),
268 event: ProjectionEvent::GraphRelationMaterialized(GraphRelationMaterializedEvent {
269 envelope: ProjectionEnvelope {
270 event_id: "evt-relation-1".to_string(),
271 correlation_id: "corr-1".to_string(),
272 causation_id: "cmd-1".to_string(),
273 occurred_at: "2026-04-14T18:45:00Z".to_string(),
274 aggregate_id: "relation:decision|addresses|finding".to_string(),
275 aggregate_type: "node_relation".to_string(),
276 schema_version: "v1beta1".to_string(),
277 },
278 data: GraphRelationMaterializedData {
279 source_node_id: "decision-1".to_string(),
280 target_node_id: "finding-1".to_string(),
281 relation_type: "addresses".to_string(),
282 explanation: RelatedNodeExplanationData {
283 semantic_class: RelationSemanticClass::Causal,
284 rationale: Some("decision addresses finding".to_string()),
285 motivation: None,
286 method: None,
287 decision_id: Some("decision-1".to_string()),
288 caused_by_node_id: None,
289 evidence: None,
290 confidence: Some("high".to_string()),
291 sequence: Some(2),
292 },
293 },
294 }),
295 };
296
297 let result = service
298 .handle_projection_event(request)
299 .await
300 .expect("relation event should apply");
301
302 assert_eq!(result.subject, "graph.relation.materialized");
303 assert_eq!(result.applied_mutations, 1);
304
305 let mutations = writer.mutations().await;
306 assert_eq!(mutations.len(), 1);
307 match &mutations[0] {
308 ProjectionMutation::UpsertNodeRelation(relation) => {
309 assert_eq!(relation.source_node_id, "decision-1");
310 assert_eq!(relation.target_node_id, "finding-1");
311 assert_eq!(relation.relation_type, "addresses");
312 }
313 mutation => panic!("unexpected mutation: {mutation:?}"),
314 }
315 }
316}