stasis/application/composition/
runtime_composition.rs

1use std::sync::Arc;
2
3use surrealdb::engine::any::Any;
4use surrealdb::Surreal;
5
6use crate::application::runtime::in_memory_runtime::InMemoryRuntime;
7use crate::application::runtime::surreal_runtime::SurrealRuntime;
8use crate::domain::errors::{Result, StasisError};
9use crate::infrastructure::llm::genai_chat_client::GenaiChatClient;
10use crate::infrastructure::memory::locus_context_reader::LocusContextReader;
11use crate::infrastructure::memory::locus_context_writer::LocusContextWriter;
12use crate::infrastructure::memory::locus_memory_operations::LocusMemoryOperations;
13use crate::infrastructure::memory::locus_node_store_factory::LocusNodeStoreFactory;
14use crate::infrastructure::memory::surreal_identity_memory_store::SurrealIdentityMemoryStore;
15use crate::infrastructure::runtime::endpoint_routing_event_publisher::EndpointRoutingEventPublisher;
16use crate::infrastructure::runtime::grapheme_sdk_workflow_engine::GraphemeSdkWorkflowEngine;
17use crate::infrastructure::runtime::in_memory_cluster_node_store::InMemoryClusterNodeStore;
18use crate::infrastructure::runtime::in_memory_delivery_endpoint_store::InMemoryDeliveryEndpointStore;
19use crate::infrastructure::runtime::in_memory_endpoint_delivery_status_store::InMemoryEndpointDeliveryStatusStore;
20use crate::infrastructure::runtime::in_memory_thread_store::InMemoryThreadStore;
21use crate::infrastructure::runtime::surreal_cluster_node_store::SurrealClusterNodeStore;
22use crate::infrastructure::runtime::surreal_delivery_endpoint_store::SurrealDeliveryEndpointStore;
23use crate::infrastructure::runtime::surreal_endpoint_delivery_status_store::SurrealEndpointDeliveryStatusStore;
24use crate::infrastructure::runtime::surreal_thread_store::SurrealThreadStore;
25use crate::ports::outbound::ai_chat_client::AiChatClient;
26use crate::ports::outbound::memory::memory_context_reader::MemoryContextReader;
27use crate::ports::outbound::memory::memory_context_writer::MemoryContextWriter;
28use crate::ports::outbound::memory::memory_operations::MemoryOperations;
29use crate::ports::outbound::runtime::cluster_node_store::ClusterNodeStore;
30use crate::ports::outbound::runtime::delivery_endpoint_store::DeliveryEndpointStore;
31use crate::ports::outbound::runtime::endpoint_delivery_status_store::EndpointDeliveryStatusStore;
32use crate::ports::outbound::runtime::endpoint_routing_policy::EndpointRoutingPolicy;
33use crate::ports::outbound::runtime::endpoint_transport_publisher::EndpointTransportPublisher;
34use crate::ports::outbound::runtime::thread_store::ThreadStore;
35use crate::ports::outbound::runtime::workflow_engine::WorkflowEngine;
36
37#[derive(Clone, Debug)]
38pub enum RuntimeBackend {
39    InMemory,
40    SurrealMem { namespace: String, database: String },
41    SurrealWs {
42        endpoint: String,
43        namespace: String,
44        database: String,
45    },
46    SurrealKv {
47        path: String,
48        namespace: String,
49        database: String,
50    },
51}
52
53#[derive(Clone)]
54pub enum RuntimeComposition {
55    InMemory(InMemoryRuntime),
56    Surreal(SurrealRuntime),
57}
58
59pub struct RuntimeFactory;
60
61impl RuntimeFactory {
62    async fn connect_surreal_any(
63        endpoint: &str,
64        namespace: String,
65        database: String,
66    ) -> Result<RuntimeComposition> {
67        let db = Surreal::<Any>::init();
68        db.connect(endpoint)
69            .await
70            .map_err(|e| StasisError::PortFailure(format!("connect surreal db ({endpoint}): {e}")))?;
71
72        db.use_ns(namespace).use_db(database).await.map_err(|e| {
73            StasisError::PortFailure(format!("select surreal namespace/database: {e}"))
74        })?;
75
76        SurrealIdentityMemoryStore::ensure_schema_for_db(&db).await?;
77
78        Ok(RuntimeComposition::Surreal(SurrealRuntime::new(db)))
79    }
80
81    pub async fn build(config: RuntimeBackend) -> Result<RuntimeComposition> {
82        match config {
83            RuntimeBackend::InMemory => Ok(RuntimeComposition::InMemory(InMemoryRuntime::new())),
84            RuntimeBackend::SurrealMem {
85                namespace,
86                database,
87            } => Self::connect_surreal_any("mem://", namespace, database).await,
88            RuntimeBackend::SurrealWs {
89                endpoint,
90                namespace,
91                database,
92            } => Self::connect_surreal_any(&endpoint, namespace, database).await,
93            RuntimeBackend::SurrealKv {
94                path,
95                namespace,
96                database,
97            } => {
98                let endpoint = if path.starts_with("surrealkv://") {
99                    path
100                } else {
101                    format!("surrealkv://{path}")
102                };
103                Self::connect_surreal_any(&endpoint, namespace, database).await
104            }
105        }
106    }
107
108    pub fn from_db(db: Surreal<Any>) -> RuntimeComposition {
109        RuntimeComposition::Surreal(SurrealRuntime::new(db))
110    }
111
112    pub fn default_chat_client() -> Arc<dyn AiChatClient> {
113        Arc::new(GenaiChatClient::from_env())
114    }
115
116    pub fn default_workflow_engine() -> Arc<dyn WorkflowEngine> {
117        Arc::new(GraphemeSdkWorkflowEngine::new())
118    }
119
120    pub async fn ensure_locus_memory_adapters(
121        enable_locus_memory: bool,
122        mut memory_context_reader: Option<Arc<dyn MemoryContextReader>>,
123        mut memory_context_writer: Option<Arc<dyn MemoryContextWriter>>,
124        mut memory_operations: Option<Arc<dyn MemoryOperations>>,
125    ) -> Result<(
126        Option<Arc<dyn MemoryContextReader>>,
127        Option<Arc<dyn MemoryContextWriter>>,
128        Option<Arc<dyn MemoryOperations>>,
129    )> {
130        if enable_locus_memory
131            && (memory_context_reader.is_none()
132                || memory_context_writer.is_none()
133                || memory_operations.is_none())
134        {
135            let store = LocusNodeStoreFactory::in_memory().await?;
136            if memory_context_reader.is_none() {
137                memory_context_reader = Some(Arc::new(LocusContextReader::new(store.clone())));
138            }
139            if memory_context_writer.is_none() {
140                memory_context_writer = Some(Arc::new(LocusContextWriter::new(store.clone())));
141            }
142            if memory_operations.is_none() {
143                memory_operations = Some(Arc::new(LocusMemoryOperations::new(store, None)));
144            }
145        }
146
147        Ok((
148            memory_context_reader,
149            memory_context_writer,
150            memory_operations,
151        ))
152    }
153
154    pub fn resolve_thread_store(
155        runtime: &RuntimeComposition,
156        configured: Option<Arc<dyn ThreadStore>>,
157    ) -> Arc<dyn ThreadStore> {
158        if let Some(store) = configured {
159            return store;
160        }
161
162        match runtime {
163            RuntimeComposition::InMemory(_) => Arc::new(InMemoryThreadStore::default()),
164            RuntimeComposition::Surreal(rt) => Arc::new(SurrealThreadStore::new(rt.job_store.db())),
165        }
166    }
167
168    pub fn resolve_cluster_node_store(
169        runtime: &RuntimeComposition,
170        configured: Option<Arc<dyn ClusterNodeStore>>,
171    ) -> Arc<dyn ClusterNodeStore> {
172        if let Some(store) = configured {
173            return store;
174        }
175
176        match runtime {
177            RuntimeComposition::InMemory(_) => Arc::new(InMemoryClusterNodeStore::default()),
178            RuntimeComposition::Surreal(rt) => {
179                Arc::new(SurrealClusterNodeStore::new(rt.job_store.db()))
180            }
181        }
182    }
183
184    pub fn resolve_delivery_endpoint_store(
185        runtime: &RuntimeComposition,
186        configured: Option<Arc<dyn DeliveryEndpointStore>>,
187    ) -> Arc<dyn DeliveryEndpointStore> {
188        if let Some(store) = configured {
189            return store;
190        }
191
192        match runtime {
193            RuntimeComposition::InMemory(_) => Arc::new(InMemoryDeliveryEndpointStore::default()),
194            RuntimeComposition::Surreal(rt) => {
195                Arc::new(SurrealDeliveryEndpointStore::new(rt.job_store.db()))
196            }
197        }
198    }
199
200    pub fn resolve_endpoint_delivery_status_store(
201        runtime: &RuntimeComposition,
202        configured: Option<Arc<dyn EndpointDeliveryStatusStore>>,
203    ) -> Arc<dyn EndpointDeliveryStatusStore> {
204        if let Some(store) = configured {
205            return store;
206        }
207
208        match runtime {
209            RuntimeComposition::InMemory(_) => {
210                Arc::new(InMemoryEndpointDeliveryStatusStore::default())
211            }
212            RuntimeComposition::Surreal(rt) => {
213                Arc::new(SurrealEndpointDeliveryStatusStore::new(rt.job_store.db()))
214            }
215        }
216    }
217
218    pub fn build_endpoint_routing_publisher(
219        endpoint_store: Arc<dyn DeliveryEndpointStore>,
220        status_store: Arc<dyn EndpointDeliveryStatusStore>,
221        transports: &[Arc<dyn EndpointTransportPublisher>],
222        routing_policy: Option<Arc<dyn EndpointRoutingPolicy>>,
223    ) -> EndpointRoutingEventPublisher {
224        let mut routing_publisher =
225            EndpointRoutingEventPublisher::new(endpoint_store).fail_on_unsupported_protocol(false);
226
227        if transports.is_empty() {
228            routing_publisher = routing_publisher
229                .with_http_webhook_transport()
230                .with_tcp_socket_transport();
231        } else {
232            for transport in transports {
233                routing_publisher = routing_publisher.with_transport_arc(transport.clone());
234            }
235        }
236
237        if let Some(policy) = routing_policy {
238            routing_publisher = routing_publisher.with_routing_policy_arc(policy);
239        }
240
241        routing_publisher.with_status_store_arc(status_store)
242    }
243}
stasis/application/composition/runtime_composition.rs

stasis/application/composition/
runtime_composition.rs
