durable/

lib.rs

pub mod ctx;
pub mod error;
pub mod executor;

pub use ctx::Ctx;
pub use ctx::RetryPolicy;
pub use ctx::{TaskQuery, TaskSort, TaskSummary};
pub use durable_db::entity::sea_orm_active_enums::TaskStatus;
pub use durable_macros::{step, workflow};
pub use error::DurableError;
pub use executor::{Executor, HeartbeatConfig, RecoveredTask};
pub use sea_orm::DatabaseTransaction;

use sea_orm::{ConnectOptions, Database, DatabaseConnection};
use sea_orm_migration::MigratorTrait;

/// Handle returned by [`init`] / [`init_with_config`].
///
/// Wraps a database connection, the executor identity, and any tasks
/// recovered from prior crashes. Implements `Deref<Target = DatabaseConnection>`
/// so it can be passed anywhere a `&DatabaseConnection` is expected.
///
/// ```ignore
/// let durable = durable::init("postgres://localhost/mydb").await?;
///
/// // Use as &DatabaseConnection (via Deref)
/// let ctx = Ctx::start(&durable, "my_wf", None).await?;
///
/// // Or start with executor tracking for proper crash recovery
/// let ctx = durable.start_workflow("my_wf", None).await?;
///
/// // Inspect tasks recovered on startup
/// for task in durable.recovered() {
///     println!("recovered: {} ({})", task.name, task.id);
/// }
/// ```
pub struct DurableInstance {
    db: DatabaseConnection,
    executor_id: String,
    recovered: Vec<RecoveredTask>,
}

impl std::ops::Deref for DurableInstance {
    type Target = DatabaseConnection;
    fn deref(&self) -> &DatabaseConnection {
        &self.db
    }
}

impl DurableInstance {
    /// The underlying database connection.
    pub fn db(&self) -> &DatabaseConnection {
        &self.db
    }

    /// The unique executor ID for this process.
    pub fn executor_id(&self) -> &str {
        &self.executor_id
    }

    /// Tasks recovered from prior crashes during [`init`].
    ///
    /// Each entry was reset from RUNNING to PENDING. To resume a recovered
    /// workflow, call [`Ctx::from_id`] with the task's `id`.
    pub fn recovered(&self) -> &[RecoveredTask] {
        &self.recovered
    }

    /// Start a new root workflow, tagging it with this instance's executor_id
    /// so heartbeat-based recovery works if this process crashes.
    pub async fn start_workflow(
        &self,
        name: &str,
        input: Option<serde_json::Value>,
    ) -> Result<Ctx, DurableError> {
        Ctx::start_with_executor(&self.db, name, input, Some(self.executor_id.clone())).await
    }

    /// Start a new root workflow with a timeout, tagging it with this
    /// instance's executor_id.
    pub async fn start_workflow_with_timeout(
        &self,
        name: &str,
        input: Option<serde_json::Value>,
        timeout_ms: i64,
    ) -> Result<Ctx, DurableError> {
        Ctx::start_with_timeout_and_executor(
            &self.db,
            name,
            input,
            timeout_ms,
            Some(self.executor_id.clone()),
        )
        .await
    }
}

/// Initialize durable: connect to Postgres, run migrations, start heartbeat,
/// and recover stale tasks from prior crashes.
///
/// Returns a [`DurableInstance`] which derefs to `DatabaseConnection` for
/// backwards compatibility.
///
/// Uses [`HeartbeatConfig::default()`] (60 s heartbeat, 180 s staleness).
/// For custom intervals use [`init_with_config`].
///
/// ```ignore
/// let durable = durable::init("postgres://localhost/mydb").await?;
/// ```
pub async fn init(database_url: &str) -> Result<DurableInstance, DurableError> {
    init_with_config(database_url, HeartbeatConfig::default()).await
}

/// Like [`init`] but with a custom [`HeartbeatConfig`].
pub async fn init_with_config(
    database_url: &str,
    config: HeartbeatConfig,
) -> Result<DurableInstance, DurableError> {
    let mut opt = ConnectOptions::new(database_url);
    // Apply search_path at the pool level so every connection resolves
    // the `task_status` enum in the `durable` schema.
    opt.set_schema_search_path("public,durable");
    let db = Database::connect(opt).await?;

    // Run migrations
    durable_db::Migrator::up(&db, None).await?;

    // Create executor with a unique ID for this process
    let executor_id = format!("exec-{}-{}", std::process::id(), uuid::Uuid::new_v4());
    let executor = Executor::new(db.clone(), executor_id.clone());

    // Write initial heartbeat so other workers know we're alive
    executor.heartbeat().await?;

    let mut all_recovered = Vec::new();

    // Recover stale tasks: timeout/deadline-based
    let recovered = executor.recover().await?;
    if !recovered.is_empty() {
        tracing::info!(
            "recovered {} stale tasks (timeout/deadline)",
            recovered.len()
        );
    }
    all_recovered.extend(recovered);

    // Recover stale tasks: heartbeat-based (dead workers, unknown executors, orphaned NULL)
    let recovered = executor
        .recover_stale_tasks(config.staleness_threshold)
        .await?;
    if !recovered.is_empty() {
        tracing::info!(
            "recovered {} stale tasks from dead/unknown workers",
            recovered.len()
        );
    }
    all_recovered.extend(recovered);

    // Start background heartbeat + recovery loops
    executor.start_heartbeat(&config);
    executor.start_recovery_loop(&config);

    tracing::info!("durable initialized (executor={})", executor.executor_id());
    Ok(DurableInstance {
        db,
        executor_id,
        recovered: all_recovered,
    })
}

/// Initialize durable: connect to Postgres and run migrations only.
///
/// Does **not** start heartbeat or recovery loops. Use this for tests,
/// migrations-only scripts, or when you manage the [`Executor`] yourself.
///
/// ```ignore
/// let db = durable::init_db("postgres://localhost/mydb").await?;
/// ```
pub async fn init_db(database_url: &str) -> Result<DatabaseConnection, DurableError> {
    let mut opt = ConnectOptions::new(database_url);
    opt.set_schema_search_path("public,durable");
    let db = Database::connect(opt).await?;
    durable_db::Migrator::up(&db, None).await?;
    tracing::info!("durable initialized (db only)");
    Ok(db)
}