stasis-rs 0.9.3

use chrono::Utc;

use crate::application::runtime::in_memory_runtime::JobHandler;
use crate::application::runtime::job_lifecycle::StaleRecoverReport;
use crate::application::runtime::runtime_factory::RuntimeComposition;
use crate::application::runtime::runtime_factory::{RuntimeBackend, SurrealAuth};
use crate::application::runtime::stasis_runtime_builder::StasisRuntimeBuilder;
use crate::application::runtime::typed_job::{JobConsumer, TypedEnqueueBuilder};
use crate::domain::errors::Result;
use crate::domain::runtime::job::{JobState, NewJob};
use crate::domain::runtime::recurring::RecurringDefinition;
use crate::domain::runtime::resource_lease::{FencingToken, ResourceLease};
use crate::domain::runtime::typed_contract::{StasisEvent, StasisJob};
use crate::ports::outbound::runtime::job_store::JobStore;
use crate::ports::outbound::runtime::outbox_store::OutboxStore;
use crate::ports::outbound::runtime::recurring_store::RecurringStore;

/// Snapshot of high-level runtime queue, outbox, and recurring workload counts.
#[derive(Clone, Debug, Default)]
pub struct RuntimeStatsSnapshot {
    pub enqueued_jobs: usize,
    pub running_jobs: usize,
    pub succeeded_jobs: usize,
    pub failed_jobs: usize,
    pub dead_letter_jobs: usize,
    pub pending_outbox_events: usize,
    pub recurring_definitions: usize,
}

/// Backend-agnostic facade for runtime queue, outbox, and recurring operations.
#[derive(Clone)]
pub struct RuntimeSdk {
    runtime: RuntimeComposition,
}

/// Preferred public runtime naming.
pub type StasisRuntime = RuntimeSdk;

impl RuntimeSdk {
    /// Creates a new facade over a pre-built runtime composition.
    pub fn new(runtime: RuntimeComposition) -> Self {
        Self { runtime }
    }

    /// Builds an in-memory runtime facade with default wiring.
    pub async fn in_memory() -> Result<Self> {
        Self::from_builder(StasisRuntimeBuilder::new(RuntimeBackend::InMemory)).await
    }

    /// Builds a surreal-mem runtime facade with default wiring.
    pub async fn surreal_mem(
        namespace: impl Into<String>,
        database: impl Into<String>,
    ) -> Result<Self> {
        Self::from_builder(StasisRuntimeBuilder::new(RuntimeBackend::surreal_mem(
            namespace, database,
        )))
        .await
    }

    /// Builds a remote websocket surreal runtime facade with default wiring.
    pub async fn surreal_ws(
        endpoint: impl Into<String>,
        namespace: impl Into<String>,
        database: impl Into<String>,
    ) -> Result<Self> {
        Self::surreal_ws_with_auth(endpoint, namespace, database, None).await
    }

    /// Builds a remote websocket surreal runtime facade with optional root credentials.
    pub async fn surreal_ws_with_auth(
        endpoint: impl Into<String>,
        namespace: impl Into<String>,
        database: impl Into<String>,
        auth: Option<SurrealAuth>,
    ) -> Result<Self> {
        let mut backend = RuntimeBackend::surreal_ws(endpoint, namespace, database);
        if let Some(auth) = auth {
            backend = backend.with_surreal_auth(auth);
        }
        Self::from_builder(StasisRuntimeBuilder::new(backend)).await
    }

    /// Builds an embedded surreal-kv runtime facade with default wiring.
    pub async fn surreal_kv(
        path: impl Into<String>,
        namespace: impl Into<String>,
        database: impl Into<String>,
    ) -> Result<Self> {
        Self::surreal_kv_with_auth(path, namespace, database, None).await
    }

    /// Builds an embedded surreal-kv runtime facade with optional root credentials.
    pub async fn surreal_kv_with_auth(
        path: impl Into<String>,
        namespace: impl Into<String>,
        database: impl Into<String>,
        auth: Option<SurrealAuth>,
    ) -> Result<Self> {
        let mut backend = RuntimeBackend::surreal_kv(path, namespace, database);
        if let Some(auth) = auth {
            backend = backend.with_surreal_auth(auth);
        }
        Self::from_builder(StasisRuntimeBuilder::new(backend)).await
    }

    /// Builds a surreal-mem runtime facade with optional root credentials.
    pub async fn surreal_mem_with_auth(
        namespace: impl Into<String>,
        database: impl Into<String>,
        auth: Option<SurrealAuth>,
    ) -> Result<Self> {
        let mut backend = RuntimeBackend::surreal_mem(namespace, database);
        if let Some(auth) = auth {
            backend = backend.with_surreal_auth(auth);
        }
        Self::from_builder(StasisRuntimeBuilder::new(backend)).await
    }

    /// Builds a runtime facade from a fully configured runtime builder.
    pub async fn from_builder(builder: StasisRuntimeBuilder) -> Result<Self> {
        let runtime = builder.build().await?;
        Ok(Self::new(runtime))
    }

    /// Builds a runtime facade and returns composition-root bridge handles (MCP/agent).
    pub async fn from_builder_with_handles(
        builder: StasisRuntimeBuilder,
    ) -> Result<(
        Self,
        crate::application::runtime::stasis_runtime_builder::McpBridgeHandles,
    )> {
        let (runtime, handles) = builder.build_with_handles().await?;
        Ok((Self::new(runtime), handles))
    }

    /// Returns a shared reference to the underlying runtime composition.
    pub fn runtime(&self) -> &RuntimeComposition {
        &self.runtime
    }

    /// Consumes this facade and returns the owned runtime composition.
    pub fn into_runtime(self) -> RuntimeComposition {
        self.runtime
    }

    /// Enqueues a single runtime job.
    pub async fn enqueue(&self, job: NewJob) -> Result<()> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.enqueue(job).await,
            RuntimeComposition::Surreal(rt) => rt.enqueue(job).await,
        }
    }

    /// Enqueues a typed job payload. Call `.queue(...)`, `.retry(...)`, then `.send().await`.
    pub fn enqueue_job<T: StasisJob>(&self, payload: T) -> TypedEnqueueBuilder<T> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.enqueue_job(payload),
            RuntimeComposition::Surreal(rt) => rt.enqueue_job(payload),
        }
    }

    /// Registers a raw job handler.
    pub fn register_handler<H: JobHandler + 'static>(&self, handler: H) -> Result<()> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.register_handler(handler),
            RuntimeComposition::Surreal(rt) => rt.register_handler(handler),
        }
    }

    /// Registers a typed job consumer.
    pub fn register_consumer<T, H>(&self, handler: H) -> Result<()>
    where
        T: StasisJob,
        H: JobConsumer<T> + 'static,
    {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.register_consumer(handler),
            RuntimeComposition::Surreal(rt) => rt.register_consumer(handler),
        }
    }

    /// Cancels a non-terminal job, completes pending waits, and fires the cancel hook.
    pub async fn cancel(&self, job_id: &str) -> Result<bool> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.cancel(job_id).await,
            RuntimeComposition::Surreal(rt) => rt.cancel(job_id).await,
        }
    }

    /// Recovers expired `Leased`/`Running` jobs as retryable failures.
    pub async fn recover_stale(&self) -> Result<StaleRecoverReport> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.recover_stale_now().await,
            RuntimeComposition::Surreal(rt) => rt.recover_stale_now().await,
        }
    }

    /// Replays a dead-lettered job back to `Enqueued`.
    pub async fn replay_dead_letter(&self, job_id: &str) -> Result<bool> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.replay_dead_letter_now(job_id).await,
            RuntimeComposition::Surreal(rt) => rt.replay_dead_letter_now(job_id).await,
        }
    }

    /// Operator abort: mark a non-terminal job dead-lettered.
    pub async fn fail(&self, job_id: &str) -> Result<bool> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.fail(job_id).await,
            RuntimeComposition::Surreal(rt) => rt.fail(job_id).await,
        }
    }

    /// Deletes a terminal job. Refuse in-flight jobs; cancel or fail first.
    pub async fn delete(&self, job_id: &str) -> Result<bool> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.delete(job_id).await,
            RuntimeComposition::Surreal(rt) => rt.delete(job_id).await,
        }
    }

    /// Publishes a typed signal. Duplicate `(type, key, payload)` ids are ignored.
    pub async fn signal<E: StasisEvent>(
        &self,
        correlation_key: impl Into<String>,
        event: E,
    ) -> Result<bool> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.signal(correlation_key, event).await,
            RuntimeComposition::Surreal(rt) => rt.signal(correlation_key, event).await,
        }
    }

    pub async fn acquire_lease(
        &self,
        resource: impl Into<String>,
        owner: impl Into<String>,
        ttl: std::time::Duration,
    ) -> Result<ResourceLease> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.acquire_lease(resource, owner, ttl).await,
            RuntimeComposition::Surreal(rt) => rt.acquire_lease(resource, owner, ttl).await,
        }
    }

    pub async fn force_acquire_lease(
        &self,
        resource: impl Into<String>,
        owner: impl Into<String>,
        ttl: std::time::Duration,
    ) -> Result<ResourceLease> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.force_acquire_lease(resource, owner, ttl).await,
            RuntimeComposition::Surreal(rt) => rt.force_acquire_lease(resource, owner, ttl).await,
        }
    }

    pub async fn renew_lease(
        &self,
        resource: impl Into<String>,
        owner: impl Into<String>,
        fencing_token: FencingToken,
        ttl: std::time::Duration,
    ) -> Result<ResourceLease> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => {
                rt.renew_lease(resource, owner, fencing_token, ttl).await
            }
            RuntimeComposition::Surreal(rt) => {
                rt.renew_lease(resource, owner, fencing_token, ttl).await
            }
        }
    }

    pub async fn release_lease(
        &self,
        resource: impl Into<String>,
        owner: impl Into<String>,
        fencing_token: FencingToken,
    ) -> Result<bool> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => {
                rt.release_lease(resource, owner, fencing_token).await
            }
            RuntimeComposition::Surreal(rt) => {
                rt.release_lease(resource, owner, fencing_token).await
            }
        }
    }

    pub async fn transfer_lease(
        &self,
        resource: impl Into<String>,
        from: impl Into<String>,
        to: impl Into<String>,
        fencing_token: FencingToken,
        ttl: std::time::Duration,
    ) -> Result<ResourceLease> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => {
                rt.transfer_lease(resource, from, to, fencing_token, ttl)
                    .await
            }
            RuntimeComposition::Surreal(rt) => {
                rt.transfer_lease(resource, from, to, fencing_token, ttl)
                    .await
            }
        }
    }

    pub async fn validate_fence(
        &self,
        resource: impl Into<String>,
        fencing_token: FencingToken,
    ) -> Result<bool> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.validate_fence(resource, fencing_token).await,
            RuntimeComposition::Surreal(rt) => rt.validate_fence(resource, fencing_token).await,
        }
    }

    pub async fn watch_lease(&self, resource: impl Into<String>) -> Result<Option<ResourceLease>> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.watch_lease(resource).await,
            RuntimeComposition::Surreal(rt) => rt.watch_lease(resource).await,
        }
    }

    /// Registers a recurring job definition.
    pub async fn register_recurring(&self, definition: RecurringDefinition) -> Result<()> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.register_recurring(definition).await,
            RuntimeComposition::Surreal(rt) => rt.register_recurring(definition).await,
        }
    }

    /// Lists all recurring definitions currently registered in the runtime store.
    pub async fn list_recurring(&self) -> Result<Vec<RecurringDefinition>> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.recurring_store.list().await,
            RuntimeComposition::Surreal(rt) => rt.recurring_store.list().await,
        }
    }

    /// Upserts a recurring definition (used by declarative reconcile updates).
    pub async fn save_recurring(&self, definition: RecurringDefinition) -> Result<()> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.recurring_store.save(definition).await,
            RuntimeComposition::Surreal(rt) => rt.recurring_store.save(definition).await,
        }
    }

    /// Attempts to process one job from a queue using the provided worker id.
    pub async fn process_once(&self, queue: &str, worker_id: &str) -> Result<Option<String>> {
        let now = Utc::now();
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.process_once(queue, worker_id, now).await,
            RuntimeComposition::Surreal(rt) => rt.process_once(queue, worker_id, now).await,
        }
    }

    /// Publishes pending outbox events up to `limit`.
    pub async fn publish_pending_events(&self, limit: usize) -> Result<usize> {
        let now = Utc::now();
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.publish_pending_events(limit, now).await,
            RuntimeComposition::Surreal(rt) => rt.publish_pending_events(limit, now).await,
        }
    }

    /// Materializes any due recurring jobs at the current wall-clock time.
    pub async fn materialize_recurring_now(&self, scheduler_id: &str) -> Result<usize> {
        match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.materialize_recurring_now(scheduler_id).await,
            RuntimeComposition::Surreal(rt) => rt.materialize_recurring_now(scheduler_id).await,
        }
    }

    /// Aggregates common runtime counts into a single snapshot.
    pub async fn stats_snapshot(&self, pending_limit: usize) -> Result<RuntimeStatsSnapshot> {
        Ok(RuntimeStatsSnapshot {
            enqueued_jobs: self.job_count_by_state(JobState::Enqueued).await?,
            running_jobs: self.job_count_by_state(JobState::Running).await?,
            succeeded_jobs: self.job_count_by_state(JobState::Succeeded).await?,
            failed_jobs: self.job_count_by_state(JobState::Failed).await?,
            dead_letter_jobs: self.job_count_by_state(JobState::DeadLetter).await?,
            pending_outbox_events: self.pending_outbox_count(pending_limit).await?,
            recurring_definitions: self.recurring_count().await?,
        })
    }

    /// Counts jobs currently in the specified state.
    pub async fn job_count_by_state(&self, state: JobState) -> Result<usize> {
        let jobs = match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.job_store.list_by_state(state).await?,
            RuntimeComposition::Surreal(rt) => rt.job_store.list_by_state(state).await?,
        };
        Ok(jobs.len())
    }

    /// Counts pending outbox events, bounded by `limit`.
    pub async fn pending_outbox_count(&self, limit: usize) -> Result<usize> {
        let pending = match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.outbox_store.list_pending(limit).await?,
            RuntimeComposition::Surreal(rt) => rt.outbox_store.list_pending(limit).await?,
        };
        Ok(pending.len())
    }

    /// Counts registered recurring definitions.
    pub async fn recurring_count(&self) -> Result<usize> {
        let definitions = match &self.runtime {
            RuntimeComposition::InMemory(rt) => rt.recurring_store.list().await?,
            RuntimeComposition::Surreal(rt) => rt.recurring_store.list().await?,
        };
        Ok(definitions.len())
    }
}

#[cfg(test)]
mod tests {
    use std::env;
    use std::fs;
    use std::time::{SystemTime, UNIX_EPOCH};

    use crate::application::runtime::runtime_factory::RuntimeComposition;

    use super::RuntimeSdk;

    #[tokio::test]
    async fn runtime_sdk_in_memory_constructor_builds() {
        let runtime = RuntimeSdk::in_memory()
            .await
            .expect("in-memory runtime should build");
        let stats = runtime
            .stats_snapshot(10)
            .await
            .expect("stats snapshot should succeed");
        assert_eq!(stats.enqueued_jobs, 0);
    }

    #[tokio::test]
    async fn runtime_sdk_surreal_mem_constructor_builds() {
        let runtime = RuntimeSdk::surreal_mem("stasis", "runtime")
            .await
            .expect("surreal-mem runtime should build");
        assert!(matches!(runtime.runtime(), RuntimeComposition::Surreal(_)));
    }

    #[tokio::test]
    async fn runtime_sdk_surreal_ws_constructor_rejects_invalid_endpoint() {
        let result = RuntimeSdk::surreal_ws("not-a-valid-endpoint", "stasis", "runtime").await;
        assert!(result.is_err(), "invalid websocket endpoint should fail");
        let err = result.err().expect("result should contain an error");
        assert!(err.to_string().contains("connect surreal db"));
    }

    #[tokio::test]
    async fn runtime_sdk_surreal_kv_constructor_builds() {
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system clock should be after epoch")
            .as_nanos();
        let path = env::temp_dir().join(format!("stasis-surrealkv-{nanos}"));
        let path_str = path.to_string_lossy().into_owned();

        let runtime = RuntimeSdk::surreal_kv(path_str, "stasis", "runtime")
            .await
            .expect("surreal-kv runtime should build");
        assert!(matches!(runtime.runtime(), RuntimeComposition::Surreal(_)));

        drop(runtime);
        let _ = fs::remove_dir_all(path);
    }
}