1use std::sync::Arc;
2
3use chrono::{Duration, Utc};
4
5use crate::application::composition::surreal_backend_config::{
6 resolve_surreal_auth_from_env, resolve_surreal_database_from_env, resolve_surreal_namespace_from_env,
7};
8use crate::application::config::env::{required, truthy, with_default};
9use crate::application::dto::{
10 HeartbeatClusterNodeRequest, RegisterClusterNodeRequest, RegisterDeliveryEndpointRequest,
11};
12use crate::application::runtime::in_memory_runtime::{
13 InMemoryRuntime, JobExecutionOutcome, JobHandler,
14};
15use crate::application::runtime::runtime_factory::{
16 RuntimeBackend, RuntimeComposition, SurrealAuth,
17};
18use crate::application::runtime::stasis_runtime_builder::StasisRuntimeBuilder;
19use crate::dashboard::service::RuntimeDashboardQueryService;
20use crate::domain::errors::Result;
21use crate::domain::runtime::cluster_node::ClusterNodeRole;
22use crate::domain::runtime::delivery_endpoint::DeliveryProtocol;
23use crate::domain::runtime::job::{BackoffPolicy, Job, NewJob};
24use crate::infrastructure::runtime::composite_control_plane_store::CompositeControlPlaneStore;
25use crate::infrastructure::runtime::in_memory_cluster_node_store::InMemoryClusterNodeStore;
26use crate::infrastructure::runtime::in_memory_delivery_endpoint_store::InMemoryDeliveryEndpointStore;
27use crate::infrastructure::runtime::in_memory_endpoint_delivery_status_store::InMemoryEndpointDeliveryStatusStore;
28use crate::ports::outbound::runtime::endpoint_delivery_status_store::EndpointDeliveryStatusStore;
29use crate::sdk::control_plane_sdk::ControlPlaneSdk;
30
31type InMemoryControlPlane = ControlPlaneSdk<
32 CompositeControlPlaneStore<InMemoryDeliveryEndpointStore, InMemoryClusterNodeStore>,
33>;
34
35#[derive(Clone, Debug)]
36pub struct DashboardBootstrapOptions {
37 pub seed_demo: bool,
38}
39
40impl Default for DashboardBootstrapOptions {
41 fn default() -> Self {
42 Self {
43 seed_demo: truthy("STASIS_DASHBOARD_DEMO_SEED"),
44 }
45 }
46}
47
48#[derive(Clone)]
49struct DemoSuccessHandler;
50
51#[async_trait::async_trait]
52impl JobHandler for DemoSuccessHandler {
53 fn job_type(&self) -> &'static str {
54 "demo.success"
55 }
56
57 async fn execute(&self, _job: &Job) -> Result<JobExecutionOutcome> {
58 Ok(JobExecutionOutcome::Success {
59 sttp_output_node_id: "sttp:out:demo-success".to_string(),
60 execution_id: Some("exec-demo-success".to_string()),
61 diagnostics: None,
62 })
63 }
64}
65
66#[derive(Clone)]
67struct DemoFatalHandler;
68
69#[async_trait::async_trait]
70impl JobHandler for DemoFatalHandler {
71 fn job_type(&self) -> &'static str {
72 "demo.fatal"
73 }
74
75 async fn execute(&self, _job: &Job) -> Result<JobExecutionOutcome> {
76 Ok(JobExecutionOutcome::FatalFailure {
77 message: "demo fatal crash".to_string(),
78 execution_id: Some("exec-demo-fatal".to_string()),
79 diagnostics: Some("{\"guardrail_code\":\"DEMO_FATAL\"}".to_string()),
80 })
81 }
82}
83
84pub fn resolve_dashboard_runtime_backend() -> RuntimeBackend {
85 let backend = with_default("STASIS_DASHBOARD_RUNTIME_BACKEND", "in-memory")
86 .trim()
87 .to_ascii_lowercase();
88
89 match backend.as_str() {
90 "in-memory" | "inmemory" => RuntimeBackend::InMemory,
91 "surreal-mem" | "mem" => apply_surreal_auth(RuntimeBackend::surreal_mem(
92 dashboard_surreal_namespace(),
93 dashboard_surreal_database(),
94 )),
95 "surreal-ws" | "ws" => apply_surreal_auth(RuntimeBackend::surreal_ws(
96 required("STASIS_DASHBOARD_SURREAL_ENDPOINT").unwrap_or_else(|_| {
97 panic!(
98 "STASIS_DASHBOARD_SURREAL_ENDPOINT is required when STASIS_DASHBOARD_RUNTIME_BACKEND=surreal-ws"
99 )
100 }),
101 dashboard_surreal_namespace(),
102 dashboard_surreal_database(),
103 )),
104 "surreal-kv" | "kv" => apply_surreal_auth(RuntimeBackend::surreal_kv(
105 required("STASIS_DASHBOARD_SURREAL_KV_PATH").unwrap_or_else(|_| {
106 panic!(
107 "STASIS_DASHBOARD_SURREAL_KV_PATH is required when STASIS_DASHBOARD_RUNTIME_BACKEND=surreal-kv"
108 )
109 }),
110 dashboard_surreal_namespace(),
111 dashboard_surreal_database(),
112 )),
113 other => {
114 eprintln!(
115 "unknown STASIS_DASHBOARD_RUNTIME_BACKEND='{}', falling back to in-memory",
116 other
117 );
118 RuntimeBackend::InMemory
119 }
120 }
121}
122
123pub async fn build_dashboard_query_service(
124 options: DashboardBootstrapOptions,
125) -> Result<Arc<RuntimeDashboardQueryService>> {
126 let backend = resolve_dashboard_runtime_backend();
127
128 match backend {
129 RuntimeBackend::InMemory => {
130 build_in_memory_dashboard_query_service(options, RuntimeBackend::InMemory).await
131 }
132 other => build_surreal_dashboard_query_service(options, other).await,
133 }
134}
135
136async fn build_in_memory_dashboard_query_service(
137 options: DashboardBootstrapOptions,
138 backend: RuntimeBackend,
139) -> Result<Arc<RuntimeDashboardQueryService>> {
140 let endpoint_store = InMemoryDeliveryEndpointStore::default();
141 let cluster_store = InMemoryClusterNodeStore::default();
142 let endpoint_status_store = Arc::new(InMemoryEndpointDeliveryStatusStore::default());
143
144 let mut builder = StasisRuntimeBuilder::new(backend)
145 .with_cluster_node_store(Arc::new(cluster_store.clone()))
146 .with_delivery_endpoint_store(Arc::new(endpoint_store.clone()))
147 .with_endpoint_delivery_status_store(endpoint_status_store.clone());
148
149 builder = apply_dashboard_builder_options(builder, &options)?;
150
151 let runtime = builder.build().await?;
152 let RuntimeComposition::InMemory(runtime) = runtime else {
153 return Err(crate::domain::errors::StasisError::PortFailure(
154 "expected in-memory runtime composition".to_string(),
155 ));
156 };
157
158 if options.seed_demo {
159 seed_demo_jobs(&runtime).await;
160 }
161
162 let control_store = CompositeControlPlaneStore::new(endpoint_store, cluster_store);
163 let control_plane =
164 ControlPlaneSdk::new_with_status_store(control_store, endpoint_status_store.clone());
165
166 if options.seed_demo {
167 seed_control_plane_data(&control_plane, endpoint_status_store).await;
168 }
169
170 Ok(Arc::new(RuntimeDashboardQueryService::from_in_memory_composition(
171 runtime,
172 control_plane,
173 )))
174}
175
176async fn build_surreal_dashboard_query_service(
177 options: DashboardBootstrapOptions,
178 backend: RuntimeBackend,
179) -> Result<Arc<RuntimeDashboardQueryService>> {
180 if options.seed_demo {
181 eprintln!("dashboard demo seed mode is ignored for surreal runtime backends");
182 }
183
184 let builder = apply_dashboard_builder_options(StasisRuntimeBuilder::new(backend), &options)?;
185 let runtime = builder.build().await?;
186
187 Ok(Arc::new(RuntimeDashboardQueryService::from_runtime_composition(
188 runtime,
189 )))
190}
191
192fn apply_dashboard_builder_options(
193 mut builder: StasisRuntimeBuilder,
194 options: &DashboardBootstrapOptions,
195) -> Result<StasisRuntimeBuilder> {
196 if dashboard_locus_memory_enabled() {
197 builder = builder.with_locus_memory();
198 }
199
200 if truthy("STASIS_DASHBOARD_LOGGING_CHAT") {
201 builder = builder.with_logging_chat_middleware();
202 }
203
204 if options.seed_demo {
205 builder = builder
206 .with_extra_handler(DemoSuccessHandler)
207 .with_extra_handler(DemoFatalHandler);
208 }
209
210 #[cfg(feature = "otel")]
211 {
212 use crate::infrastructure::telemetry::otel::otel_enabled;
213
214 if otel_enabled() {
215 builder = builder.with_otel_from_env()?;
216 }
217 }
218
219 Ok(builder)
220}
221
222fn dashboard_locus_memory_enabled() -> bool {
223 truthy("STASIS_DASHBOARD_LOCUS_MEMORY")
224}
225
226fn dashboard_surreal_namespace() -> String {
227 resolve_surreal_namespace_from_env("STASIS_DASHBOARD_SURREAL_NAMESPACE", None, "stasis")
228}
229
230fn dashboard_surreal_database() -> String {
231 resolve_surreal_database_from_env("STASIS_DASHBOARD_SURREAL_DATABASE", None, "runtime")
232}
233
234fn dashboard_surreal_auth() -> Option<SurrealAuth> {
235 resolve_surreal_auth_from_env(
236 "STASIS_DASHBOARD_SURREAL_USERNAME",
237 "STASIS_DASHBOARD_SURREAL_PASSWORD",
238 None,
239 None,
240 )
241}
242
243fn apply_surreal_auth(backend: RuntimeBackend) -> RuntimeBackend {
244 match dashboard_surreal_auth() {
245 Some(auth) => backend.with_surreal_auth(auth),
246 None => backend,
247 }
248}
249
250async fn seed_demo_jobs(runtime: &InMemoryRuntime) {
251 let now = Utc::now();
252 let backoff = BackoffPolicy {
253 base_delay_seconds: 1,
254 max_delay_seconds: 4,
255 };
256
257 runtime
258 .enqueue(NewJob {
259 id: "job-demo-success-1".to_string(),
260 queue: "default".to_string(),
261 job_type: "demo.success".to_string(),
262 payload_ref: "sttp:in:demo-1".to_string(),
263 priority: 100,
264 max_attempts: 2,
265 idempotency_key: "idem-demo-success-1".to_string(),
266 correlation_id: "corr-demo-success-1".to_string(),
267 causation_id: "cause-demo-success-1".to_string(),
268 trace_id: "trace-demo-success-1".to_string(),
269 sttp_input_node_id: "sttp:in:demo-1".to_string(),
270 scheduled_at: now,
271 backoff_policy: backoff.clone(),
272 })
273 .await
274 .expect("enqueue success demo job");
275
276 runtime
277 .enqueue(NewJob {
278 id: "job-demo-fatal-1".to_string(),
279 queue: "default".to_string(),
280 job_type: "demo.fatal".to_string(),
281 payload_ref: "sttp:in:demo-2".to_string(),
282 priority: 90,
283 max_attempts: 1,
284 idempotency_key: "idem-demo-fatal-1".to_string(),
285 correlation_id: "corr-demo-fatal-1".to_string(),
286 causation_id: "cause-demo-fatal-1".to_string(),
287 trace_id: "trace-demo-fatal-1".to_string(),
288 sttp_input_node_id: "sttp:in:demo-2".to_string(),
289 scheduled_at: now,
290 backoff_policy: backoff.clone(),
291 })
292 .await
293 .expect("enqueue fatal demo job");
294
295 runtime
296 .enqueue(NewJob {
297 id: "job-demo-pending-1".to_string(),
298 queue: "default".to_string(),
299 job_type: "demo.success".to_string(),
300 payload_ref: "sttp:in:demo-3".to_string(),
301 priority: 80,
302 max_attempts: 3,
303 idempotency_key: "idem-demo-pending-1".to_string(),
304 correlation_id: "corr-demo-pending-1".to_string(),
305 causation_id: "cause-demo-pending-1".to_string(),
306 trace_id: "trace-demo-pending-1".to_string(),
307 sttp_input_node_id: "sttp:in:demo-3".to_string(),
308 scheduled_at: now + Duration::minutes(5),
309 backoff_policy: backoff,
310 })
311 .await
312 .expect("enqueue pending demo job");
313
314 runtime
315 .process_once("default", "worker-demo-a", now)
316 .await
317 .expect("process first demo job");
318 runtime
319 .process_once("default", "worker-demo-b", now + Duration::seconds(1))
320 .await
321 .expect("process second demo job");
322}
323
324async fn seed_control_plane_data(
325 control_plane: &InMemoryControlPlane,
326 endpoint_status_store: Arc<InMemoryEndpointDeliveryStatusStore>,
327) {
328 let now = Utc::now();
329
330 control_plane
331 .register_delivery_endpoint(RegisterDeliveryEndpointRequest {
332 endpoint_id: "endpoint.webhook.ops".to_string(),
333 name: "Ops Webhook".to_string(),
334 protocol: DeliveryProtocol::HttpWebhook,
335 target: "https://ops.example/hook".to_string(),
336 metadata: None,
337 })
338 .await
339 .expect("register webhook endpoint");
340
341 control_plane
342 .register_delivery_endpoint(RegisterDeliveryEndpointRequest {
343 endpoint_id: "endpoint.kafka.audit".to_string(),
344 name: "Audit Kafka".to_string(),
345 protocol: DeliveryProtocol::Kafka,
346 target: "kafka://broker:9092/audit".to_string(),
347 metadata: None,
348 })
349 .await
350 .expect("register kafka endpoint");
351
352 endpoint_status_store
353 .record_success("endpoint.webhook.ops", "evt-demo-1", now)
354 .await
355 .expect("record endpoint success");
356 endpoint_status_store
357 .record_failure(
358 "endpoint.kafka.audit",
359 "evt-demo-2",
360 "delivery timeout",
361 now - Duration::seconds(10),
362 )
363 .await
364 .expect("record endpoint failure");
365
366 control_plane
367 .register_cluster_node(RegisterClusterNodeRequest {
368 node_id: "worker-12".to_string(),
369 role: ClusterNodeRole::Worker,
370 region: "eu-west-1".to_string(),
371 queue_ownership: vec!["default".to_string(), "billing".to_string()],
372 capability_tags: vec!["cpu".to_string()],
373 heartbeat_at: now,
374 lease_ttl_seconds: 45,
375 queue_ownership_mode: None,
376 metadata: Some("v1.0.0".to_string()),
377 })
378 .await
379 .expect("register worker node");
380
381 control_plane
382 .register_cluster_node(RegisterClusterNodeRequest {
383 node_id: "scheduler-2".to_string(),
384 role: ClusterNodeRole::Scheduler,
385 region: "eu-west-1".to_string(),
386 queue_ownership: vec!["priority".to_string()],
387 capability_tags: vec!["orchestration".to_string()],
388 heartbeat_at: now - Duration::seconds(40),
389 lease_ttl_seconds: 60,
390 queue_ownership_mode: None,
391 metadata: Some("rolling".to_string()),
392 })
393 .await
394 .expect("register scheduler node");
395
396 control_plane
397 .heartbeat_cluster_node(HeartbeatClusterNodeRequest {
398 node_id: "worker-12".to_string(),
399 heartbeat_at: now,
400 lease_ttl_seconds: 45,
401 queue_ownership_mode: None,
402 queue_ownership: None,
403 capability_tags: None,
404 metadata: None,
405 })
406 .await
407 .expect("heartbeat worker node");
408}