boson_runtime/

boson.rs

//! Main Boson runtime type.

use std::sync::Arc;

use boson_core::{
    ActorJsonPolicy, BosonError, EnqueueTrust, IdempotencyMode, Job, JobEnqueueDisposition,
    JobStatus, QueueBackend, RejectExternalSystemActor, Result, Run, TaskConfig, TaskRunStats,
};
use chrono::{DateTime, Utc};

use crate::registry::TaskRegistry;
use crate::worker::WorkerSettings;

/// Boson work engine — enqueue, admin reads, and worker orchestration.
#[derive(Clone)]
pub struct Boson {
    pub(crate) backend: Arc<dyn QueueBackend>,
    pub(crate) registry: Arc<TaskRegistry>,
    worker: WorkerSettings,
    /// Runtime default when [`TaskConfig::idempotency_mode`] is unset.
    idempotency_mode: IdempotencyMode,
    /// Optional actor JSON policy (defaults to [`RejectExternalSystemActor`]).
    actor_policy: Option<Arc<dyn ActorJsonPolicy>>,
}

impl Boson {
    /// Construct from injected parts (used by builder and tests).
    pub fn from_parts(
        backend: Arc<dyn QueueBackend>,
        registry: Arc<TaskRegistry>,
        worker: WorkerSettings,
    ) -> Self {
        Self::from_parts_with_idempotency(backend, registry, worker, IdempotencyMode::Lwt)
    }

    /// Construct with an explicit default idempotency mode.
    pub fn from_parts_with_idempotency(
        backend: Arc<dyn QueueBackend>,
        registry: Arc<TaskRegistry>,
        worker: WorkerSettings,
        idempotency_mode: IdempotencyMode,
    ) -> Self {
        Self {
            backend,
            registry,
            worker,
            idempotency_mode,
            actor_policy: Some(Arc::new(RejectExternalSystemActor)),
        }
    }

    /// Construct with idempotency mode and actor policy.
    pub fn from_parts_full(
        backend: Arc<dyn QueueBackend>,
        registry: Arc<TaskRegistry>,
        worker: WorkerSettings,
        idempotency_mode: IdempotencyMode,
        actor_policy: Option<Arc<dyn ActorJsonPolicy>>,
    ) -> Self {
        Self {
            backend,
            registry,
            worker,
            idempotency_mode,
            actor_policy,
        }
    }

    /// Runtime default idempotency mode (builder / task override may change per enqueue).
    #[must_use]
    pub const fn idempotency_mode(&self) -> IdempotencyMode {
        self.idempotency_mode
    }

    /// Worker settings this instance was built with.
    #[must_use]
    pub const fn worker_settings(&self) -> &WorkerSettings {
        &self.worker
    }

    /// Telemetry/runtime label (topology slug or `embedded`).
    #[must_use]
    pub fn runtime_label(&self) -> &str {
        &self.worker.runtime_label
    }

    /// Queue backend handle.
    #[must_use]
    pub fn queue_backend(&self) -> Arc<dyn QueueBackend> {
        Arc::clone(&self.backend)
    }

    /// Task registry.
    #[must_use]
    pub fn registry(&self) -> &TaskRegistry {
        &self.registry
    }

    /// Resolve task config from backend or registry defaults.
    ///
    /// Precedence: persisted backend config, else descriptor policy defaults, then runtime
    /// idempotency fallback when the mode is unset.
    ///
    /// # Errors
    ///
    /// Returns an error if the task is unknown or the backend fails.
    pub async fn resolve_task_config(&self, task_name: &str) -> Result<TaskConfig> {
        let config = if let Some(c) = self.backend.get_task_config(task_name).await? {
            c
        } else {
            self.registry.get_or_err(task_name)?.to_task_config()
        };
        Ok(config.with_runtime_idempotency_fallback(self.idempotency_mode()))
    }

    /// Resolve priority and pool for enqueue.
    ///
    /// # Errors
    ///
    /// Returns an error if the task is unknown or the backend fails.
    pub async fn resolve_priority_pool(&self, task_name: &str) -> Result<(i32, String)> {
        let config = self.resolve_task_config(task_name).await?;
        Ok((config.priority, config.pool))
    }

    /// Enqueue helper used internally and by admin APIs.
    pub(crate) async fn enqueue_internal(
        &self,
        task_name: &str,
        actor_json: serde_json::Value,
        params_json: serde_json::Value,
        idempotency_key: Option<String>,
    ) -> Result<String> {
        let descriptor = self.registry.get_or_err(task_name)?;
        let task_config = self.resolve_task_config(task_name).await?;
        let priority = task_config.priority;
        let pool = task_config.pool.clone();
        let job = Job::new(
            task_name,
            actor_json,
            params_json,
            priority,
            &pool,
            descriptor.signature_hash,
            idempotency_key,
        );
        let (job_id, disposition) = self
            .backend
            .enqueue_with_policies(job, &task_config)
            .await
            .map_err(|e| {
                if matches!(e, BosonError::RateLimited(_)) {
                    crate::telemetry::record_task_failed(task_name, "", "", &e.to_string(), false);
                }
                e
            })?;
        if disposition == JobEnqueueDisposition::InsertedNew {
            crate::telemetry::record_task_enqueued(task_name, self.runtime_label());
        }
        Ok(job_id)
    }

    /// Enqueue a job with [`EnqueueTrust::Internal`] (in-process callers).
    ///
    /// Priority and pool come from persisted [`TaskConfig`](boson_core::TaskConfig) merged with
    /// [`TaskDescriptor`](crate::registry::TaskDescriptor) defaults. Optional `idempotency_key`
    /// deduplicates non-terminal jobs. Rate limits may return
    /// [`BosonError::RateLimited`](boson_core::BosonError::RateLimited).
    ///
    /// # Errors
    ///
    /// Returns an error if the task is unknown, rate limits apply, actor policy rejects, or the
    /// backend fails.
    pub async fn enqueue(
        &self,
        task_name: &str,
        actor_json: serde_json::Value,
        params_json: serde_json::Value,
        idempotency_key: Option<String>,
    ) -> Result<String> {
        self.enqueue_with_trust(
            task_name,
            actor_json,
            params_json,
            idempotency_key,
            EnqueueTrust::Internal,
        )
        .await
    }

    /// Enqueue with an explicit trust level for [`ActorJsonPolicy`].
    ///
    /// Use [`EnqueueTrust::External`] for HTTP admin / untrusted surfaces.
    ///
    /// # Errors
    ///
    /// Returns an error if policy rejects the actor, the task is unknown, rate limits apply, or
    /// the backend fails.
    pub async fn enqueue_with_trust(
        &self,
        task_name: &str,
        actor_json: serde_json::Value,
        params_json: serde_json::Value,
        idempotency_key: Option<String>,
        trust: EnqueueTrust,
    ) -> Result<String> {
        if let Some(ref policy) = self.actor_policy {
            policy.validate(trust, &actor_json)?;
        }
        self.enqueue_internal(task_name, actor_json, params_json, idempotency_key)
            .await
    }

    /// Get a job by id.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend fails.
    pub async fn get_job(&self, job_id: &str) -> Result<Option<Job>> {
        self.backend.get_job(job_id).await
    }

    /// List jobs with optional status filter.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend fails.
    pub async fn list_jobs(
        &self,
        status_filter: Option<JobStatus>,
        offset: usize,
        limit: usize,
    ) -> Result<Vec<Job>> {
        self.backend.list_jobs(status_filter, offset, limit).await
    }

    /// Cancel a job if still active.
    ///
    /// # Errors
    ///
    /// Returns an error if the job is not found or the backend fails.
    pub async fn cancel_job(&self, job_id: &str) -> Result<()> {
        self.backend.cancel_job_if_active(job_id).await
    }

    /// Get or default task config.
    ///
    /// # Errors
    ///
    /// Returns an error if the task is unknown or the backend fails.
    pub async fn get_task_config(&self, task_name: &str) -> Result<TaskConfig> {
        self.resolve_task_config(task_name).await
    }

    /// Upsert task config.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend fails.
    pub async fn upsert_task_config(&self, config: TaskConfig) -> Result<()> {
        self.backend.upsert_task_config(&config).await
    }

    /// List runs.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend fails.
    pub async fn list_runs(
        &self,
        job_id_filter: Option<&str>,
        offset: usize,
        limit: usize,
    ) -> Result<Vec<Run>> {
        self.backend.list_runs(job_id_filter, offset, limit).await
    }

    /// Get run by id.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend fails.
    pub async fn get_run(&self, run_id: &str) -> Result<Option<Run>> {
        self.backend.get_run(run_id).await
    }

    /// Count jobs.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend fails.
    pub async fn count_jobs(&self, status_filter: Option<JobStatus>) -> Result<u64> {
        self.backend.count_jobs(status_filter).await
    }

    /// Count runs.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend fails.
    pub async fn count_runs(&self, job_id_filter: Option<&str>) -> Result<u64> {
        self.backend.count_runs(job_id_filter).await
    }

    /// Count runs since timestamp.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend fails.
    pub async fn count_runs_since(&self, since: DateTime<Utc>) -> Result<u64> {
        self.backend.count_runs_since(since).await
    }

    /// Count jobs for one task.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend fails.
    pub async fn count_jobs_for_task(
        &self,
        task_name: &str,
        status: Option<JobStatus>,
    ) -> Result<u64> {
        self.backend.count_jobs_for_task(task_name, status).await
    }

    /// Aggregate run stats for one task.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend fails.
    pub async fn task_run_stats(&self, task_name: &str) -> Result<TaskRunStats> {
        self.backend.task_run_stats(task_name).await
    }
}