stasis/sdk/
runtime_sdk.rs

1use chrono::Utc;
2
3use crate::application::runtime::runtime_factory::{RuntimeBackend, SurrealAuth};
4use crate::application::runtime::stasis_runtime_builder::StasisRuntimeBuilder;
5use crate::application::runtime::runtime_factory::RuntimeComposition;
6use crate::domain::errors::Result;
7use crate::domain::runtime::job::{JobState, NewJob};
8use crate::domain::runtime::recurring::RecurringDefinition;
9use crate::ports::outbound::runtime::job_store::JobStore;
10use crate::ports::outbound::runtime::outbox_store::OutboxStore;
11use crate::ports::outbound::runtime::recurring_store::RecurringStore;
12
13/// Snapshot of high-level runtime queue, outbox, and recurring workload counts.
14#[derive(Clone, Debug, Default)]
15pub struct RuntimeStatsSnapshot {
16    pub enqueued_jobs: usize,
17    pub running_jobs: usize,
18    pub succeeded_jobs: usize,
19    pub failed_jobs: usize,
20    pub dead_letter_jobs: usize,
21    pub pending_outbox_events: usize,
22    pub recurring_definitions: usize,
23}
24
25/// Backend-agnostic facade for runtime queue, outbox, and recurring operations.
26#[derive(Clone)]
27pub struct RuntimeSdk {
28    runtime: RuntimeComposition,
29}
30
31/// Preferred public runtime naming.
32pub type StasisRuntime = RuntimeSdk;
33
34impl RuntimeSdk {
35    /// Creates a new facade over a pre-built runtime composition.
36    pub fn new(runtime: RuntimeComposition) -> Self {
37        Self { runtime }
38    }
39
40    /// Builds an in-memory runtime facade with default wiring.
41    pub async fn in_memory() -> Result<Self> {
42        Self::from_builder(StasisRuntimeBuilder::new(RuntimeBackend::InMemory)).await
43    }
44
45    /// Builds a surreal-mem runtime facade with default wiring.
46    pub async fn surreal_mem(
47        namespace: impl Into<String>,
48        database: impl Into<String>,
49    ) -> Result<Self> {
50        Self::from_builder(StasisRuntimeBuilder::new(RuntimeBackend::surreal_mem(
51            namespace,
52            database,
53        )))
54        .await
55    }
56
57    /// Builds a remote websocket surreal runtime facade with default wiring.
58    pub async fn surreal_ws(
59        endpoint: impl Into<String>,
60        namespace: impl Into<String>,
61        database: impl Into<String>,
62    ) -> Result<Self> {
63        Self::surreal_ws_with_auth(endpoint, namespace, database, None).await
64    }
65
66    /// Builds a remote websocket surreal runtime facade with optional root credentials.
67    pub async fn surreal_ws_with_auth(
68        endpoint: impl Into<String>,
69        namespace: impl Into<String>,
70        database: impl Into<String>,
71        auth: Option<SurrealAuth>,
72    ) -> Result<Self> {
73        let mut backend = RuntimeBackend::surreal_ws(endpoint, namespace, database);
74        if let Some(auth) = auth {
75            backend = backend.with_surreal_auth(auth);
76        }
77        Self::from_builder(StasisRuntimeBuilder::new(backend)).await
78    }
79
80    /// Builds an embedded surreal-kv runtime facade with default wiring.
81    pub async fn surreal_kv(
82        path: impl Into<String>,
83        namespace: impl Into<String>,
84        database: impl Into<String>,
85    ) -> Result<Self> {
86        Self::surreal_kv_with_auth(path, namespace, database, None).await
87    }
88
89    /// Builds an embedded surreal-kv runtime facade with optional root credentials.
90    pub async fn surreal_kv_with_auth(
91        path: impl Into<String>,
92        namespace: impl Into<String>,
93        database: impl Into<String>,
94        auth: Option<SurrealAuth>,
95    ) -> Result<Self> {
96        let mut backend = RuntimeBackend::surreal_kv(path, namespace, database);
97        if let Some(auth) = auth {
98            backend = backend.with_surreal_auth(auth);
99        }
100        Self::from_builder(StasisRuntimeBuilder::new(backend)).await
101    }
102
103    /// Builds a surreal-mem runtime facade with optional root credentials.
104    pub async fn surreal_mem_with_auth(
105        namespace: impl Into<String>,
106        database: impl Into<String>,
107        auth: Option<SurrealAuth>,
108    ) -> Result<Self> {
109        let mut backend = RuntimeBackend::surreal_mem(namespace, database);
110        if let Some(auth) = auth {
111            backend = backend.with_surreal_auth(auth);
112        }
113        Self::from_builder(StasisRuntimeBuilder::new(backend)).await
114    }
115
116    /// Builds a runtime facade from a fully configured runtime builder.
117    pub async fn from_builder(builder: StasisRuntimeBuilder) -> Result<Self> {
118        let runtime = builder.build().await?;
119        Ok(Self::new(runtime))
120    }
121
122    /// Returns a shared reference to the underlying runtime composition.
123    pub fn runtime(&self) -> &RuntimeComposition {
124        &self.runtime
125    }
126
127    /// Consumes this facade and returns the owned runtime composition.
128    pub fn into_runtime(self) -> RuntimeComposition {
129        self.runtime
130    }
131
132    /// Enqueues a single runtime job.
133    pub async fn enqueue(&self, job: NewJob) -> Result<()> {
134        match &self.runtime {
135            RuntimeComposition::InMemory(rt) => rt.enqueue(job).await,
136            RuntimeComposition::Surreal(rt) => rt.enqueue(job).await,
137        }
138    }
139
140    /// Registers a recurring job definition.
141    pub async fn register_recurring(&self, definition: RecurringDefinition) -> Result<()> {
142        match &self.runtime {
143            RuntimeComposition::InMemory(rt) => rt.register_recurring(definition).await,
144            RuntimeComposition::Surreal(rt) => rt.register_recurring(definition).await,
145        }
146    }
147
148    /// Attempts to process one job from a queue using the provided worker id.
149    pub async fn process_once(&self, queue: &str, worker_id: &str) -> Result<Option<String>> {
150        let now = Utc::now();
151        match &self.runtime {
152            RuntimeComposition::InMemory(rt) => rt.process_once(queue, worker_id, now).await,
153            RuntimeComposition::Surreal(rt) => rt.process_once(queue, worker_id, now).await,
154        }
155    }
156
157    /// Publishes pending outbox events up to `limit`.
158    pub async fn publish_pending_events(&self, limit: usize) -> Result<usize> {
159        let now = Utc::now();
160        match &self.runtime {
161            RuntimeComposition::InMemory(rt) => rt.publish_pending_events(limit, now).await,
162            RuntimeComposition::Surreal(rt) => rt.publish_pending_events(limit, now).await,
163        }
164    }
165
166    /// Materializes any due recurring jobs at the current wall-clock time.
167    pub async fn materialize_recurring_now(&self, scheduler_id: &str) -> Result<usize> {
168        match &self.runtime {
169            RuntimeComposition::InMemory(rt) => rt.materialize_recurring_now(scheduler_id).await,
170            RuntimeComposition::Surreal(rt) => rt.materialize_recurring_now(scheduler_id).await,
171        }
172    }
173
174    /// Aggregates common runtime counts into a single snapshot.
175    pub async fn stats_snapshot(&self, pending_limit: usize) -> Result<RuntimeStatsSnapshot> {
176        Ok(RuntimeStatsSnapshot {
177            enqueued_jobs: self.job_count_by_state(JobState::Enqueued).await?,
178            running_jobs: self.job_count_by_state(JobState::Running).await?,
179            succeeded_jobs: self.job_count_by_state(JobState::Succeeded).await?,
180            failed_jobs: self.job_count_by_state(JobState::Failed).await?,
181            dead_letter_jobs: self.job_count_by_state(JobState::DeadLetter).await?,
182            pending_outbox_events: self.pending_outbox_count(pending_limit).await?,
183            recurring_definitions: self.recurring_count().await?,
184        })
185    }
186
187    /// Counts jobs currently in the specified state.
188    pub async fn job_count_by_state(&self, state: JobState) -> Result<usize> {
189        let jobs = match &self.runtime {
190            RuntimeComposition::InMemory(rt) => rt.job_store.list_by_state(state).await?,
191            RuntimeComposition::Surreal(rt) => rt.job_store.list_by_state(state).await?,
192        };
193        Ok(jobs.len())
194    }
195
196    /// Counts pending outbox events, bounded by `limit`.
197    pub async fn pending_outbox_count(&self, limit: usize) -> Result<usize> {
198        let pending = match &self.runtime {
199            RuntimeComposition::InMemory(rt) => rt.outbox_store.list_pending(limit).await?,
200            RuntimeComposition::Surreal(rt) => rt.outbox_store.list_pending(limit).await?,
201        };
202        Ok(pending.len())
203    }
204
205    /// Counts registered recurring definitions.
206    pub async fn recurring_count(&self) -> Result<usize> {
207        let definitions = match &self.runtime {
208            RuntimeComposition::InMemory(rt) => rt.recurring_store.list().await?,
209            RuntimeComposition::Surreal(rt) => rt.recurring_store.list().await?,
210        };
211        Ok(definitions.len())
212    }
213}
214
215#[cfg(test)]
216mod tests {
217    use std::env;
218    use std::fs;
219    use std::time::{SystemTime, UNIX_EPOCH};
220
221    use crate::application::runtime::runtime_factory::RuntimeComposition;
222
223    use super::RuntimeSdk;
224
225    #[tokio::test]
226    async fn runtime_sdk_in_memory_constructor_builds() {
227        let runtime = RuntimeSdk::in_memory()
228            .await
229            .expect("in-memory runtime should build");
230        let stats = runtime
231            .stats_snapshot(10)
232            .await
233            .expect("stats snapshot should succeed");
234        assert_eq!(stats.enqueued_jobs, 0);
235    }
236
237    #[tokio::test]
238    async fn runtime_sdk_surreal_mem_constructor_builds() {
239        let runtime = RuntimeSdk::surreal_mem("stasis", "runtime")
240            .await
241            .expect("surreal-mem runtime should build");
242        assert!(matches!(runtime.runtime(), RuntimeComposition::Surreal(_)));
243    }
244
245    #[tokio::test]
246    async fn runtime_sdk_surreal_ws_constructor_rejects_invalid_endpoint() {
247        let result = RuntimeSdk::surreal_ws("not-a-valid-endpoint", "stasis", "runtime").await;
248        assert!(result.is_err(), "invalid websocket endpoint should fail");
249        let err = result.err().expect("result should contain an error");
250        assert!(err.to_string().contains("connect surreal db"));
251    }
252
253    #[tokio::test]
254    async fn runtime_sdk_surreal_kv_constructor_builds() {
255        let nanos = SystemTime::now()
256            .duration_since(UNIX_EPOCH)
257            .expect("system clock should be after epoch")
258            .as_nanos();
259        let path = env::temp_dir().join(format!("stasis-surrealkv-{nanos}"));
260        let path_str = path.to_string_lossy().into_owned();
261
262        let runtime = RuntimeSdk::surreal_kv(path_str, "stasis", "runtime")
263            .await
264            .expect("surreal-kv runtime should build");
265        assert!(matches!(runtime.runtime(), RuntimeComposition::Surreal(_)));
266
267        drop(runtime);
268        let _ = fs::remove_dir_all(path);
269    }
270}
stasis/sdk/runtime_sdk.rs

stasis/sdk/
runtime_sdk.rs
