stasis/application/runtime/
memory_aggregate_job_handler.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use serde_json::json;
5
6use crate::application::orchestration::runtime_job_payloads::MemoryAggregateJobPayload;
7use crate::application::runtime::in_memory_runtime::{JobExecutionOutcome, JobHandler};
8use crate::application::runtime::memory_operation_job_outcome_helpers::{
9    operation_failure, operation_success, policy_violation_failure,
10};
11use crate::application::runtime::memory_job_request_helpers::memory_scope_from_fields;
12use crate::domain::errors::Result;
13use crate::domain::runtime::job::Job;
14use crate::ports::outbound::memory::memory_models::MemoryAggregateRequest;
15use crate::ports::outbound::memory::memory_operations::MemoryOperations;
16
17pub struct MemoryAggregateJobHandler {
18    operations: Arc<dyn MemoryOperations>,
19}
20
21impl MemoryAggregateJobHandler {
22    pub fn new(operations: Arc<dyn MemoryOperations>) -> Self {
23        Self { operations }
24    }
25
26    fn parse_payload(raw: &str) -> std::result::Result<MemoryAggregateJobPayload, String> {
27        serde_json::from_str(raw).map_err(|err| {
28            format!("policy violation: invalid memory-aggregate payload json: {err}")
29        })
30    }
31}
32
33#[async_trait]
34impl JobHandler for MemoryAggregateJobHandler {
35    fn job_type(&self) -> &'static str {
36        "workflow.stasis.memory.aggregate"
37    }
38
39    async fn execute(&self, job: &Job) -> Result<JobExecutionOutcome> {
40        let payload = match Self::parse_payload(&job.payload_ref) {
41            Ok(payload) => payload,
42            Err(message) => return Ok(policy_violation_failure("stasis-memory-aggregate", message)),
43        };
44
45        let request = MemoryAggregateRequest {
46            scope: memory_scope_from_fields(
47                None,
48                payload.session_ids,
49                payload.tiers,
50                payload.from_utc,
51                payload.to_utc,
52            ),
53            max_groups: payload.max_groups.unwrap_or(30),
54            max_nodes: payload.max_nodes.unwrap_or(5000),
55        };
56
57        match self.operations.aggregate(&request).await {
58            Ok(result) => Ok(operation_success(
59                "stasis-memory-aggregate",
60                "memory-aggregate",
61                &job.id,
62                json!({
63                    "total_groups": result.total_groups,
64                    "scanned_nodes": result.scanned_nodes,
65                }),
66            )),
67            Err(err) => Ok(operation_failure("stasis-memory-aggregate", err.to_string())),
68        }
69    }
70}
stasis/application/runtime/memory_aggregate_job_handler.rs

stasis/application/runtime/
memory_aggregate_job_handler.rs
