durare 0.3.0

#![cfg(feature = "postgres")]
//! Postgres backend tests. Skipped unless `DATABASE_URL` points at a reachable
//! Postgres instance (ideally an empty database — `init` runs the migrations).
//!
//!   createdb durare_test && DATABASE_URL=postgres://localhost/durare_test cargo test --test postgres

use durare::{
    DurableContext, DurableEngine, Error, ErrorCode, ListFilter, PortableWorkflowError,
    PostgresProvider, Result, ScheduledInput, Serializer, StateProvider, TransactionOptions,
    WorkflowOptions, WorkflowQueue, WorkflowStatus, STATUS_PENDING,
};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

fn database_url() -> Option<String> {
    std::env::var("DATABASE_URL").ok().filter(|s| !s.is_empty())
}

async fn engine_with(url: &str, fmt: Serializer) -> Result<DurableEngine> {
    let provider = PostgresProvider::connect(url).await?.with_serializer(fmt);
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("greet", |ctx: DurableContext, name: String| async move {
        let msg = ctx
            .step("build", || async { Ok::<_, Error>(format!("hi {name}")) })
            .await?;
        Ok::<_, Error>(msg)
    });
    Ok(engine)
}

/// Round-trip a workflow's input/step-output/result through Postgres, and prove
/// a provider in a different serialization format still decodes them.
#[tokio::test]
async fn pg_serialization_cross_format() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_serialization_cross_format: DATABASE_URL unset");
        return Ok(());
    };
    let id = format!("wf-ser-{}", uuid::Uuid::new_v4());

    {
        let engine = engine_with(&url, Serializer::Portable).await?;
        let out: String = engine
            .start::<_, String>("greet", "ada".to_string(), WorkflowOptions::with_id(&id))
            .await?
            .result()
            .await?;
        assert_eq!(out, "hi ada");
    }
    {
        let engine = engine_with(&url, Serializer::Json).await?;
        let handle = engine.retrieve_workflow::<String>(&id).await?;
        let status = handle.get_status().await?;
        assert_eq!(status.input, serde_json::json!("ada"));
        assert_eq!(status.output, Some(serde_json::json!("hi ada")));
        assert_eq!(handle.result().await?, "hi ada");
    }
    Ok(())
}

/// A workflow that fails under `portable_json` stores its error as the
/// cross-language envelope, so a reader recovers the structured `error_info`
/// (name/message); under the default format the error stays a bare string.
#[tokio::test]
async fn pg_portable_error_envelope_round_trip() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_portable_error_envelope_round_trip: DATABASE_URL unset");
        return Ok(());
    };

    let run = |fmt: Serializer, id: String| {
        let url = url.clone();
        async move {
            let provider = PostgresProvider::connect(&url).await?.with_serializer(fmt);
            let mut engine = DurableEngine::new(Arc::new(provider)).await?;
            engine.register("boom", |_ctx: DurableContext, _: ()| async move {
                Err::<(), _>(Error::app("kaboom"))
            });
            let outcome = engine
                .start::<_, ()>("boom", (), WorkflowOptions::with_id(&id))
                .await?
                .result()
                .await;
            assert!(outcome.is_err());
            let handle = engine.retrieve_workflow::<()>(&id).await?;
            handle.get_status().await
        }
    };

    // Portable: the structured envelope is persisted and read back.
    let portable = run(
        Serializer::Portable,
        format!("wf-errp-{}", uuid::Uuid::new_v4()),
    )
    .await?;
    assert_eq!(portable.error.as_deref(), Some("kaboom"));
    let info = portable
        .error_info
        .expect("a portable error carries structured info");
    assert_eq!(info.name, "Portable Error");
    assert_eq!(info.message, "kaboom");

    // Default: bare string, no structured info.
    let default = run(
        Serializer::Json,
        format!("wf-errd-{}", uuid::Uuid::new_v4()),
    )
    .await?;
    assert_eq!(default.error.as_deref(), Some("kaboom"));
    assert!(default.error_info.is_none());

    // A typed Error::Portable keeps its own name/code through storage, and a
    // separate reader reconstructs it from result.
    let typed_id = format!("wf-errt-{}", uuid::Uuid::new_v4());
    {
        let provider = PostgresProvider::connect(&url)
            .await?
            .with_serializer(Serializer::Portable);
        let mut engine = DurableEngine::new(Arc::new(provider)).await?;
        engine.register("validate", |_ctx: DurableContext, _: ()| async move {
            Err::<(), _>(Error::Portable(Box::new(PortableWorkflowError {
                name: "ValidationError".to_string(),
                message: "bad email".to_string(),
                code: Some(serde_json::json!(400)),
                data: None,
            })))
        });
        let _ = engine
            .start::<_, ()>("validate", (), WorkflowOptions::with_id(&typed_id))
            .await?
            .result()
            .await;
    }
    {
        let provider = PostgresProvider::connect(&url)
            .await?
            .with_serializer(Serializer::Portable);
        let engine = DurableEngine::new(Arc::new(provider)).await?;
        let handle = engine.retrieve_workflow::<()>(&typed_id).await?;
        let info = handle
            .get_status()
            .await?
            .error_info
            .expect("typed error survives storage");
        assert_eq!(info.name, "ValidationError");
        assert_eq!(info.code, Some(serde_json::json!(400)));
        assert!(
            matches!(handle.result().await, Err(Error::Portable(pe)) if pe.name == "ValidationError")
        );
    }

    Ok(())
}

/// A caught step failure is checkpointed across a real restart: a fresh engine
/// over the same database replays the recorded error without re-running the step.
#[tokio::test]
async fn pg_checkpoints_a_caught_step_failure() -> Result<()> {
    use std::sync::atomic::{AtomicUsize, Ordering};
    static RUNS: AtomicUsize = AtomicUsize::new(0);
    let Some(url) = database_url() else {
        eprintln!("skipping pg_checkpoints_a_caught_step_failure: DATABASE_URL unset");
        return Ok(());
    };
    RUNS.store(0, Ordering::SeqCst);
    let id = format!("wf-step-err-{}", uuid::Uuid::new_v4());

    let register = |engine: &mut DurableEngine| {
        engine.register("flaky_caught", |ctx: DurableContext, _: ()| async move {
            let r: Result<i64> = ctx
                .step("maybe", || async {
                    let n = RUNS.fetch_add(1, Ordering::SeqCst);
                    if n == 0 {
                        Err(Error::app("transient"))
                    } else {
                        Ok(7)
                    }
                })
                .await;
            Ok::<_, Error>(if r.is_ok() { "ok" } else { "caught-error" }.to_string())
        });
    };

    {
        let mut engine =
            DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;
        register(&mut engine);
        let out: String = engine
            .start("flaky_caught", (), WorkflowOptions::with_id(&id))
            .await?
            .result()
            .await?;
        assert_eq!(out, "caught-error");
        let steps = engine.get_workflow_steps(&id).await?;
        let maybe = steps
            .iter()
            .find(|s| s.name == "maybe")
            .expect("step recorded");
        assert_eq!(maybe.error.as_deref(), Some("transient"));
    }
    {
        let mut engine =
            DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;
        register(&mut engine);
        let out: String = engine
            .start("flaky_caught", (), WorkflowOptions::with_id(&id))
            .await?
            .result()
            .await?;
        assert_eq!(out, "caught-error", "replay observes the recorded error");
    }
    assert_eq!(
        RUNS.load(Ordering::SeqCst),
        1,
        "a checkpointed failed step is not re-run on replay"
    );
    Ok(())
}

/// The run identity round-trips through Postgres: the three auth columns are
/// written on insert (roles as a JSON array in the text column), read back into
/// the status, threaded into the workflow context, and copied onto a fork.
#[tokio::test]
async fn pg_auth_context_round_trip() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_auth_context_round_trip: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("whoami", |ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(format!(
            "{}/{}/{}",
            ctx.authenticated_user().unwrap_or("-"),
            ctx.assumed_role().unwrap_or("-"),
            ctx.authenticated_roles().join(","),
        ))
    });

    let id = format!("wf-auth-{tag}");
    let opts = WorkflowOptions::with_id(&id)
        .authenticated_user("alice")
        .assumed_role("admin")
        .authenticated_roles(["admin", "user"]);
    let handle = engine.start::<_, String>("whoami", (), opts).await?;
    assert_eq!(handle.result().await?, "alice/admin/admin,user");

    let status = handle.get_status().await?;
    assert_eq!(status.authenticated_user.as_deref(), Some("alice"));
    assert_eq!(status.assumed_role.as_deref(), Some("admin"));
    assert_eq!(status.authenticated_roles, vec!["admin", "user"]);

    let fork_id = format!("wf-auth-fork-{tag}");
    let forked = engine
        .fork_workflow::<String>(&id, 0, WorkflowOptions::with_id(&fork_id))
        .await?;
    let fstatus = forked.get_status().await?;
    assert_eq!(fstatus.authenticated_user.as_deref(), Some("alice"));
    assert_eq!(fstatus.authenticated_roles, vec!["admin", "user"]);

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// Child workflows round-trip through Postgres: a parent starts a child, the
/// child is linked via `parent_workflow_id` and the `child_workflow_id`
/// checkpoint, and inherits the parent's identity.
#[tokio::test]
async fn pg_child_workflow() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_child_workflow: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("child", |ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(format!("{n}:{}", ctx.authenticated_user().unwrap_or("-")))
    });
    engine.register("parent", |ctx: DurableContext, n: i64| async move {
        let child = ctx
            .start_workflow::<_, String>("child", n, WorkflowOptions::default())
            .await?;
        child.result().await
    });

    let parent_id = format!("parent-{tag}");
    let opts = WorkflowOptions::with_id(&parent_id).authenticated_user("alice");
    let handle = engine.start::<_, String>("parent", 5_i64, opts).await?;
    assert_eq!(handle.result().await?, "5:alice");

    let child = engine
        .retrieve_workflow::<String>(&format!("{parent_id}-0"))
        .await?;
    let status = child.get_status().await?;
    assert_eq!(
        status.parent_workflow_id.as_deref(),
        Some(parent_id.as_str())
    );
    assert_eq!(status.authenticated_user.as_deref(), Some("alice"));

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// Step introspection round-trips through Postgres: `get_workflow_steps` returns
/// recorded steps (output decoded per serialization) and the child link, ordered.
#[tokio::test]
async fn pg_workflow_steps() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_workflow_steps: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("kid", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n)
    });
    engine.register("worker", |ctx: DurableContext, _: ()| async move {
        let v = ctx
            .step("compute", || async { Ok::<_, Error>(42_i64) })
            .await?;
        let child = ctx
            .start_workflow::<_, i64>("kid", v, WorkflowOptions::default())
            .await?;
        child.result().await
    });

    let id = format!("steps-{tag}");
    let _: i64 = engine
        .start::<_, i64>("worker", (), WorkflowOptions::with_id(&id))
        .await?
        .result()
        .await?;

    let steps = engine.get_workflow_steps(&id).await?;
    assert_eq!(steps.len(), 2);
    assert_eq!(steps[0].step_id, 0);
    assert_eq!(steps[0].name, "compute");
    assert_eq!(steps[0].output, Some(serde_json::json!(42)));
    assert_eq!(steps[1].name, "kid");
    assert_eq!(
        steps[1].child_workflow_id.as_deref(),
        Some(format!("{id}-1").as_str())
    );

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// Code patching round-trips through Postgres: a new workflow takes the patched
/// path and the marker is recorded; a pre-patch workflow takes the old path.
#[tokio::test]
async fn pg_patch() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_patch: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("wf", |ctx: DurableContext, _: ()| async move {
        ctx.patch("feature").await
    });

    // A brand-new workflow takes the new path and records the marker.
    let fresh = format!("patch-new-{tag}");
    let patched: bool = engine
        .start::<_, bool>("wf", (), WorkflowOptions::with_id(&fresh))
        .await?
        .result()
        .await?;
    assert!(patched);
    let steps = engine.get_workflow_steps(&fresh).await?;
    assert_eq!(steps[0].name, "DBOS.patch-feature");

    // A workflow with a different step already at seq 0 takes the old path.
    let old = format!("patch-old-{tag}");
    let provider2 = PostgresProvider::connect(&url).await?;
    provider2
        .insert_workflow_status(WorkflowStatus::new(
            &old,
            "wf",
            serde_json::Value::Null,
            STATUS_PENDING,
            "",
            engine.app_version(),
        ))
        .await?;
    provider2
        .record_step_result(&old, 0, "legacy_step", serde_json::json!(1), None, None)
        .await?;
    let patched: bool = engine
        .start::<_, bool>("wf", (), WorkflowOptions::with_id(&old))
        .await?
        .result()
        .await?;
    assert!(!patched, "a pre-patch workflow stays on the old path");

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// A durable `select` round-trips through Postgres: the winning index and value
/// are recorded as a `DBOS.select` step.
#[tokio::test]
async fn pg_select() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_select: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("racer", |ctx: DurableContext, _: ()| async move {
        let branches: Vec<Pin<Box<dyn Future<Output = i64> + Send>>> = vec![
            Box::pin(async { 7_i64 }),
            Box::pin(async {
                tokio::time::sleep(Duration::from_millis(50)).await;
                8_i64
            }),
        ];
        ctx.select(branches).await
    });

    let id = format!("select-{tag}");
    let (index, value): (usize, i64) = engine
        .start::<_, (usize, i64)>("racer", (), WorkflowOptions::with_id(&id))
        .await?
        .result()
        .await?;
    assert_eq!((index, value), (0, 7));

    let steps = engine.get_workflow_steps(&id).await?;
    assert_eq!(steps.len(), 1);
    assert_eq!(steps[0].name, "DBOS.select");

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// A durable stream round-trips through Postgres: a producer writes values and
/// closes the stream, and an external reader drains them in order and sees the
/// close.
#[tokio::test]
async fn pg_stream_round_trip() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_stream_round_trip: DATABASE_URL unset");
        return Ok(());
    };
    let id = format!("stream-{}", uuid::Uuid::new_v4());

    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("producer", |ctx: DurableContext, _: ()| async move {
        for i in 0..3_i64 {
            ctx.write_stream("nums", i).await?;
        }
        ctx.close_stream("nums").await?;
        Ok::<_, Error>(())
    });

    engine
        .start::<_, ()>("producer", (), WorkflowOptions::with_id(&id))
        .await?
        .result()
        .await?;

    let (values, closed): (Vec<i64>, bool) = engine.read_stream(&id, "nums").await?;
    assert_eq!(values, vec![0, 1, 2]);
    assert!(closed);

    // A snapshot read from an offset returns only the tail, without blocking.
    let (tail, closed): (Vec<i64>, bool) = engine.read_stream_snapshot(&id, "nums", 2).await?;
    assert_eq!(tail, vec![2]);
    assert!(closed);

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// Postgres surfaces the dedup unique-violation and the destination FK violation
/// as typed, classifiable errors (verifies the sqlx Postgres driver mapping).
#[tokio::test]
async fn pg_typed_db_errors() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_typed_db_errors: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("noop", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    engine.register_queue(WorkflowQueue::new(format!("q-{tag}")));

    let dedup = format!("once-{tag}");
    let mut opts = WorkflowOptions::with_id(format!("wf-a-{tag}"));
    opts.dedup_id = Some(dedup.clone());
    engine
        .start::<_, ()>("noop", (), opts.queue(format!("q-{tag}")))
        .await?;

    let mut opts = WorkflowOptions::with_id(format!("wf-b-{tag}"));
    opts.dedup_id = Some(dedup);
    let err = match engine
        .start::<_, ()>("noop", (), opts.queue(format!("q-{tag}")))
        .await
    {
        Ok(_) => panic!("dedup reuse must be rejected"),
        Err(e) => e,
    };
    assert_eq!(err.code(), ErrorCode::QueueDeduplicated);

    let err = engine
        .send(&format!("ghost-{tag}"), 1_i64, "topic")
        .await
        .expect_err("send to nonexistent workflow must fail");
    assert_eq!(err.code(), ErrorCode::NonExistentWorkflow);

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// Global concurrency caps how many of a queue's workflows run at once, even
/// under the snapshot-isolation (`REPEATABLE READ` + `FOR UPDATE NOWAIT`) dequeue
/// path it triggers. With a cap of 1, the observed peak must stay 1.
#[tokio::test]
async fn pg_global_concurrency_caps_running() -> Result<()> {
    use std::sync::atomic::{AtomicUsize, Ordering};
    static CURRENT: AtomicUsize = AtomicUsize::new(0);
    static PEAK: AtomicUsize = AtomicUsize::new(0);

    let Some(url) = database_url() else {
        eprintln!("skipping pg_global_concurrency_caps_running: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();

    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("track", |ctx: DurableContext, _: ()| async move {
        let now = CURRENT.fetch_add(1, Ordering::SeqCst) + 1;
        PEAK.fetch_max(now, Ordering::SeqCst);
        ctx.sleep(Duration::from_millis(60)).await?;
        CURRENT.fetch_sub(1, Ordering::SeqCst);
        Ok::<_, Error>(())
    });
    engine.register_queue(
        WorkflowQueue::new("gc")
            .global_concurrency(1)
            .base_polling_interval(Duration::from_millis(10)),
    );
    engine.launch().await?;

    let mut handles = Vec::new();
    for i in 0..4 {
        handles.push(
            engine
                .start::<_, ()>(
                    "track",
                    (),
                    WorkflowOptions::with_id(format!("gc-{tag}-{i}")).queue("gc"),
                )
                .await?,
        );
    }
    for h in handles {
        h.result().await?;
    }
    assert_eq!(
        PEAK.load(Ordering::SeqCst),
        1,
        "global_concurrency(1) must keep at most one workflow running"
    );

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// Bulk cancel / resume / delete through Postgres: exercises the `ANY($1)`
/// statements, the `RETURNING` resume, and the recursive-CTE child delete (with
/// FK cascade).
#[tokio::test]
async fn pg_bulk_ops() -> Result<()> {
    use durare::{STATUS_CANCELLED, STATUS_SUCCESS};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_bulk_ops: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let id = |s: &str| format!("{s}-{tag}");
    // A unique app version isolates this test's internal-queue work from other
    // parallel tests' engines sharing the database (the dispatch version gate).
    let ver = format!("v-{tag}");

    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new_with_version(Arc::new(provider), &ver).await?;
    engine.register("noop", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    // Resume re-queues work for a dispatcher, so the engine must be live.
    engine.launch().await?;
    let provider = PostgresProvider::connect(&url).await?;

    let seed = |wid: String, status: &str, parent: Option<String>| {
        let mut s = WorkflowStatus::new(wid, "noop", serde_json::Value::Null, status, "", &ver);
        s.parent_workflow_id = parent;
        s
    };

    for s in ["wf-1", "wf-2", "wf-3"] {
        provider
            .insert_workflow_status(seed(id(s), STATUS_PENDING, None))
            .await?;
    }

    // Bulk cancel a subset + a missing id (skipped, no error).
    engine
        .cancel_workflows(&[id("wf-1"), id("wf-2"), id("ghost")])
        .await?;
    assert_eq!(
        provider
            .get_workflow_status(&id("wf-1"))
            .await?
            .unwrap()
            .status,
        STATUS_CANCELLED
    );
    assert_eq!(
        provider
            .get_workflow_status(&id("wf-3"))
            .await?
            .unwrap()
            .status,
        STATUS_PENDING
    );

    // Bulk resume returns a handle per transitioned id.
    let handles = engine
        .resume_workflows::<()>(&[id("wf-1"), id("wf-2")])
        .await?;
    assert_eq!(handles.len(), 2);
    for h in handles {
        h.result().await?;
    }
    assert_eq!(
        provider
            .get_workflow_status(&id("wf-1"))
            .await?
            .unwrap()
            .status,
        STATUS_SUCCESS
    );

    // Recursive delete: parent → child → grandchild.
    provider
        .insert_workflow_status(seed(id("p"), STATUS_SUCCESS, None))
        .await?;
    provider
        .insert_workflow_status(seed(id("c"), STATUS_SUCCESS, Some(id("p"))))
        .await?;
    provider
        .insert_workflow_status(seed(id("gc"), STATUS_SUCCESS, Some(id("c"))))
        .await?;
    engine.delete_workflows(&[id("p")], true).await?;
    assert!(provider.get_workflow_status(&id("p")).await?.is_none());
    assert!(provider.get_workflow_status(&id("c")).await?.is_none());
    assert!(provider.get_workflow_status(&id("gc")).await?.is_none());

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// A partitioned queue persists the partition key and dispatches each partition
/// independently through Postgres.
#[tokio::test]
async fn pg_partitioned_queue_dispatch() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_partitioned_queue_dispatch: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();

    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("echo", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n)
    });
    engine.register_queue(
        WorkflowQueue::new("pq")
            .partitioned()
            .base_polling_interval(Duration::from_millis(10)),
    );
    engine.launch().await?;

    let east = format!("east-{tag}");
    let west = format!("west-{tag}");
    let a = engine
        .start::<_, i64>(
            "echo",
            1_i64,
            WorkflowOptions::with_id(&east)
                .partition_key("east")
                .queue("pq"),
        )
        .await?;
    let b = engine
        .start::<_, i64>(
            "echo",
            2_i64,
            WorkflowOptions::with_id(&west)
                .partition_key("west")
                .queue("pq"),
        )
        .await?;
    assert_eq!(a.result().await?, 1);
    assert_eq!(b.result().await?, 2);
    assert_eq!(
        a.get_status().await?.queue_partition_key.as_deref(),
        Some("east")
    );

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// set_workflow_delay reschedules a DELAYED workflow through Postgres: a
/// far-future delay is shortened so the dispatcher runs it promptly; a
/// non-DELAYED row is a no-op.
#[tokio::test]
async fn pg_set_workflow_delay() -> Result<()> {
    use durare::STATUS_DELAYED;
    let Some(url) = database_url() else {
        eprintln!("skipping pg_set_workflow_delay: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let wid = format!("wf-delay-{tag}");

    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("echo", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n)
    });
    engine
        .register_queue(WorkflowQueue::new("dq").base_polling_interval(Duration::from_millis(10)));
    engine.launch().await?;

    let mut opts = WorkflowOptions::with_id(&wid);
    opts.delay = Some(Duration::from_secs(60));
    let handle = engine
        .start::<_, i64>("echo", 8_i64, opts.queue("dq"))
        .await?;
    assert_eq!(handle.get_status().await?.status, STATUS_DELAYED);

    let started = std::time::Instant::now();
    assert!(
        engine
            .set_workflow_delay(&wid, Duration::from_millis(20))
            .await?
    );
    assert_eq!(handle.result().await?, 8);
    assert!(
        started.elapsed() < Duration::from_secs(5),
        "must run on the shortened delay, not the original 60s"
    );

    // Completed → no longer DELAYED → silent no-op.
    assert!(!engine.set_workflow_delay(&wid, Duration::ZERO).await?);

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// The extended list filters work through Postgres: has_parent and the
/// load_input/load_output column substitution.
#[tokio::test]
async fn pg_list_filters_extended() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_list_filters_extended: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();

    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("child", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n * 10)
    });
    engine.register("parent", |ctx: DurableContext, _: ()| async move {
        let h = ctx
            .start_workflow::<i64, i64>("child", 5_i64, WorkflowOptions::default())
            .await?;
        h.result().await
    });
    let pid = format!("parent-{tag}");
    let out: i64 = engine
        .start("parent", (), WorkflowOptions::with_id(&pid))
        .await?
        .result()
        .await?;
    assert_eq!(out, 50);

    let child_id = format!("{pid}-0");
    let lean = engine
        .list_workflows(&ListFilter {
            workflow_ids: vec![child_id.clone()],
            has_parent: Some(true),
            load_input: false,
            load_output: false,
            ..Default::default()
        })
        .await?;
    assert_eq!(lean.len(), 1);
    assert_eq!(lean[0].id, child_id);
    assert_eq!(lean[0].input, serde_json::Value::Null);
    assert!(lean[0].output.is_none());

    let full = engine
        .list_workflows(&ListFilter {
            workflow_ids: vec![child_id],
            ..Default::default()
        })
        .await?;
    assert_eq!(full[0].output, Some(serde_json::json!(50)));

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// get_workflow_aggregates groups through Postgres (by status, with a time
/// bucket).
#[tokio::test]
async fn pg_workflow_aggregates() -> Result<()> {
    use durare::{WorkflowAggregateQuery, STATUS_SUCCESS};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_workflow_aggregates: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();

    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("agg_ok", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    let prefix = format!("agg-{tag}-");
    for i in 0..3 {
        engine
            .start::<_, ()>(
                "agg_ok",
                (),
                WorkflowOptions::with_id(format!("{prefix}{i}")),
            )
            .await?
            .result()
            .await?;
    }

    // Group by status, scoped to this run's id prefix; select count + latency.
    let by_status = engine
        .get_workflow_aggregates(&WorkflowAggregateQuery {
            by_status: true,
            select_count: true,
            select_min_created_at: true,
            select_max_total_latency_ms: true,
            workflow_id_prefix: Some(prefix.clone()),
            ..Default::default()
        })
        .await?;
    assert_eq!(by_status.len(), 1);
    assert_eq!(
        by_status[0].group.get("status"),
        Some(&Some(STATUS_SUCCESS.to_string()))
    );
    assert_eq!(by_status[0].count, Some(3));
    assert!(by_status[0].min_created_at.is_some());
    assert!(by_status[0].max_total_latency_ms.is_some_and(|l| l >= 0));

    // A wide time bucket collapses them into one group with a time_bucket key.
    let bucketed = engine
        .get_workflow_aggregates(&WorkflowAggregateQuery {
            select_count: true,
            time_bucket_ms: Some(3_600_000),
            workflow_id_prefix: Some(prefix),
            ..Default::default()
        })
        .await?;
    assert_eq!(bucketed.len(), 1);
    assert_eq!(bucketed[0].count, Some(3));
    assert!(bucketed[0].group.contains_key("time_bucket"));

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// Step aggregates through Postgres: count + max duration grouped by function
/// name, scoped to this run via the id prefix.
#[tokio::test]
async fn pg_step_aggregates() -> Result<()> {
    use durare::StepAggregateQuery;
    let Some(url) = database_url() else {
        eprintln!("skipping pg_step_aggregates: DATABASE_URL unset");
        return Ok(());
    };
    let id = format!("stepagg-{}", uuid::Uuid::new_v4());

    let provider = PostgresProvider::connect(&url).await?;
    let mut engine = DurableEngine::new(Arc::new(provider)).await?;
    engine.register("work", |ctx: DurableContext, _: ()| async move {
        ctx.step("a", || async { Ok::<_, Error>(1_i64) }).await?;
        ctx.step("b", || async {
            tokio::time::sleep(Duration::from_millis(20)).await;
            Ok::<_, Error>(2_i64)
        })
        .await?;
        ctx.step("a", || async { Ok::<_, Error>(3_i64) }).await?;
        Ok::<_, Error>(())
    });
    engine
        .start::<_, ()>("work", (), WorkflowOptions::with_id(&id))
        .await?
        .result()
        .await?;

    let by_fn = engine
        .get_step_aggregates(&StepAggregateQuery {
            by_function_name: true,
            select_count: true,
            select_max_duration_ms: true,
            workflow_id_prefix: Some(id.clone()),
            ..Default::default()
        })
        .await?;
    let a = by_fn
        .iter()
        .find(|r| r.group.get("function_name") == Some(&Some("a".to_string())))
        .expect("group a");
    let b = by_fn
        .iter()
        .find(|r| r.group.get("function_name") == Some(&Some("b".to_string())))
        .expect("group b");
    assert_eq!(a.count, Some(2));
    assert_eq!(b.count, Some(1));
    assert!(b.max_duration_ms.unwrap_or(0) >= 10);

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// Schedule CRUD round-trips through Postgres: create with context + queue, read
/// it back, pause (reflected in a status filter), and delete. A uuid-suffixed
/// name keeps it isolated from other tests sharing the database.
#[tokio::test]
async fn pg_schedule_crud() -> Result<()> {
    use durare::{ScheduleFilter, ScheduleOptions, ScheduleStatus};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_schedule_crud: DATABASE_URL unset");
        return Ok(());
    };

    let mut engine = DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;
    engine.register(
        "nightly_job",
        |_ctx: DurableContext, _: ScheduledInput| async move { Ok::<_, Error>(()) },
    );

    let name = format!("nightly-{}", uuid::Uuid::new_v4());
    engine
        .create_schedule(
            &name,
            "nightly_job",
            "0 0 0 * * *",
            ScheduleOptions::new()
                .context(&serde_json::json!({"region": "us"}))
                .queue_name("internal")
                .automatic_backfill(true)
                .cron_timezone("America/New_York"),
        )
        .await?;

    let got = engine.get_schedule(&name).await?.expect("persisted");
    assert_eq!(got.workflow_name, "nightly_job");
    assert_eq!(got.status, ScheduleStatus::Active);
    assert_eq!(got.queue_name.as_deref(), Some("internal"));
    assert_eq!(
        got.context.as_ref().and_then(|v| v.get("region")),
        Some(&serde_json::json!("us"))
    );
    // Every configured option round-trips through get_schedule.
    assert!(got.automatic_backfill, "automatic_backfill round-trips");
    assert_eq!(got.cron_timezone.as_deref(), Some("America/New_York"));

    // A duplicate name is rejected.
    assert!(engine
        .create_schedule(&name, "nightly_job", "0 0 0 * * *", ScheduleOptions::new())
        .await
        .is_err());

    assert!(engine.pause_schedule(&name).await?);
    let active = engine
        .list_schedules(&ScheduleFilter {
            statuses: vec![ScheduleStatus::Active],
            name_prefixes: vec![name.clone()],
            ..Default::default()
        })
        .await?;
    assert!(
        active.is_empty(),
        "paused schedule drops out of the active filter"
    );

    assert!(engine.delete_schedule(&name).await?);
    assert!(engine.get_schedule(&name).await?.is_none());
    Ok(())
}

/// Through Postgres: `apply_schedules` creates/replaces a schedule,
/// `backfill_schedule` persists one run per past tick (idempotently), and
/// `trigger_schedule` runs it once now. A uuid-suffixed name isolates the test.
#[tokio::test]
async fn pg_schedule_backfill_apply_trigger() -> Result<()> {
    use chrono::{TimeZone, Utc};
    use durare::{ApplySchedule, ScheduleOptions, WorkflowHandle};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_schedule_backfill_apply_trigger: DATABASE_URL unset");
        return Ok(());
    };

    let mut engine = DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;
    engine.register(
        "nightly_job",
        |_ctx: DurableContext, _: ScheduledInput| async move { Ok::<_, Error>(()) },
    );

    let name = format!("bf-{}", uuid::Uuid::new_v4());
    engine
        .apply_schedules(vec![ApplySchedule::new(
            &name,
            "nightly_job",
            "0 0 12 * * *",
        )])
        .await?;
    let created = engine.get_schedule(&name).await?.expect("applied");
    assert_eq!(created.schedule, "0 0 12 * * *");

    // Re-apply with a new spec: replaced with a fresh schedule_id.
    engine
        .apply_schedules(vec![ApplySchedule::new(
            &name,
            "nightly_job",
            "0 0 6 * * *",
        )
        .options(ScheduleOptions::new())])
        .await?;
    let replaced = engine.get_schedule(&name).await?.expect("replaced");
    assert_ne!(replaced.schedule_id, created.schedule_id);
    assert_eq!(replaced.schedule, "0 0 6 * * *");

    // Backfill three past ticks (06:00 daily).
    let start = Utc.with_ymd_and_hms(2026, 3, 1, 0, 0, 0).unwrap();
    let end = Utc.with_ymd_and_hms(2026, 3, 4, 0, 0, 0).unwrap();
    let ids = engine.backfill_schedule(&name, start, end).await?;
    assert_eq!(ids.len(), 3, "one tick per day");

    let prefix = format!("sched-{name}-");
    let filter = ListFilter {
        workflow_id_prefix: vec![prefix.clone()],
        ..Default::default()
    };
    for _ in 0..100 {
        let rows = engine.list_workflows(&filter).await?;
        if rows.len() == 3 && rows.iter().all(|r| r.status == "SUCCESS") {
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    let rows = engine.list_workflows(&filter).await?;
    assert_eq!(rows.len(), 3, "one persisted row per backfilled tick");

    // Re-backfill is idempotent.
    assert_eq!(engine.backfill_schedule(&name, start, end).await?, ids);
    tokio::time::sleep(Duration::from_millis(100)).await;
    assert_eq!(engine.list_workflows(&filter).await?.len(), 3);

    // Trigger runs once now, under a distinct -trigger- id.
    let handle: WorkflowHandle<()> = engine.trigger_schedule(&name).await?;
    assert!(handle.id().starts_with(&format!("sched-{name}-trigger-")));
    handle.result().await?;

    engine.delete_schedule(&name).await?;
    Ok(())
}

/// Application version registry through Postgres. The DB is shared and tests run
/// in parallel, so versions are uuid-suffixed and only the *relative* order of
/// this test's own two versions is asserted (global "latest" would be racy).
#[tokio::test]
async fn pg_application_versions() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_application_versions: DATABASE_URL unset");
        return Ok(());
    };
    let v_a = format!("va-{}", uuid::Uuid::new_v4());
    let v_b = format!("vb-{}", uuid::Uuid::new_v4());

    // Index of a version_name within the (newest-first) list.
    async fn index_of(engine: &DurableEngine, name: &str) -> Result<usize> {
        let versions = engine.list_application_versions().await?;
        Ok(versions
            .iter()
            .position(|v| v.version_name == name)
            .expect("version present"))
    }

    // Register v_a, then v_b later, so v_b sorts ahead of v_a.
    let a = DurableEngine::new_with_version(
        Arc::new(PostgresProvider::connect(&url).await?),
        v_a.clone(),
    )
    .await?;
    a.launch().await?;
    a.shutdown(Duration::from_secs(1)).await?;
    tokio::time::sleep(Duration::from_millis(5)).await;
    let b = DurableEngine::new_with_version(
        Arc::new(PostgresProvider::connect(&url).await?),
        v_b.clone(),
    )
    .await?;
    b.launch().await?;
    b.shutdown(Duration::from_secs(1)).await?;

    assert!(
        index_of(&a, &v_b).await? < index_of(&a, &v_a).await?,
        "newer version sorts first"
    );

    // Promote v_a: it now sorts ahead of v_b.
    tokio::time::sleep(Duration::from_millis(5)).await;
    assert!(a.set_latest_application_version(&v_a).await?);
    assert!(
        index_of(&a, &v_a).await? < index_of(&a, &v_b).await?,
        "promoted version sorts first"
    );

    // Unknown version is a no-op.
    assert!(
        !a.set_latest_application_version(&format!("nope-{}", uuid::Uuid::new_v4()))
            .await?
    );

    // Restore this binary's real version as latest: unversioned-row dequeue
    // gating is latest-only, so leaving a uuid version promoted would starve
    // every other test's client-enqueued ('') claims on the shared database.
    // Register the version first (idempotent): registration happens at
    // `launch()`, which this engine never does, and on a fresh database (CI)
    // no other test may have registered it yet.
    let restore_provider = Arc::new(PostgresProvider::connect(&url).await?);
    let real = DurableEngine::new(restore_provider.clone()).await?;
    restore_provider
        .create_application_version(real.app_version())
        .await?;
    tokio::time::sleep(Duration::from_millis(5)).await;
    assert!(
        real.set_latest_application_version(real.app_version())
            .await?
    );
    Ok(())
}

/// An out-of-process `Client` enqueues work over Postgres that a separate engine
/// (sharing the database) claims and runs; the client observes the result. The
/// queue and workflow names are uuid-suffixed so parallel tests don't cross-claim.
#[tokio::test]
async fn pg_client_enqueues_work_an_engine_runs() -> Result<()> {
    use durare::{Client, WorkflowHandle, WorkflowQueue};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_client_enqueues_work_an_engine_runs: DATABASE_URL unset");
        return Ok(());
    };
    let wf = format!("double-{}", uuid::Uuid::new_v4());
    let queue = format!("q-{}", uuid::Uuid::new_v4());
    let job = format!("job-{}", uuid::Uuid::new_v4());

    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new(provider.clone()).await?;
    engine.register(&wf, |ctx: DurableContext, n: i64| async move {
        ctx.step("mul", || async { Ok::<_, Error>(n * 2) }).await
    });
    engine.register_queue(WorkflowQueue::new(&queue));
    engine.launch().await?;

    let client = Client::new(provider.clone());
    let opts = WorkflowOptions {
        workflow_id: Some(job.clone()),
        ..Default::default()
    };
    let handle = client.enqueue::<_, i64>(&queue, &wf, 21i64, opts).await?;
    assert_eq!(handle.result().await?, 42);

    let steps = client.get_workflow_steps(&job).await?;
    assert!(steps.iter().any(|s| s.name == "mul"));
    let again: WorkflowHandle<i64> = client.retrieve_workflow(&job).await?;
    assert_eq!(again.result().await?, 42);

    engine.shutdown(Duration::from_secs(2)).await?;
    Ok(())
}

/// An out-of-process `Client` debounces over a real Postgres database: three
/// rapid `Client.debouncer(...).debounce` calls (sharing the DB with a launched
/// engine that runs the collector and target) coalesce to a single run with the
/// latest input, all handles pointing at the same run.
#[tokio::test]
async fn pg_client_debounces() -> Result<()> {
    use durare::{Client, WorkflowHandle};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_client_debounces: DATABASE_URL unset");
        return Ok(());
    };
    let wf = format!("notify-{}", uuid::Uuid::new_v4());
    let key = format!("k-{}", uuid::Uuid::new_v4());
    // Pin a unique app version on both the engine and the client. The debouncer
    // rides the shared internal queue, and dequeue gates on application version
    // (`= mine OR ''`): a client-enqueued collector stamped '' can be claimed by
    // ANY concurrently-launched test engine, which has `_dbos_debouncer` but not
    // this test's uuid-scoped target — the collector errors and the test hangs.
    // Version-pinning scopes the collector to this test's engine alone.
    let ver = format!("v-{}", uuid::Uuid::new_v4());

    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new_with_version(provider.clone(), &ver).await?;
    engine.register(&wf, |_ctx: DurableContext, msg: String| async move {
        Ok::<_, Error>(msg)
    });
    engine.launch().await?;

    let client = Client::new(provider.clone()).with_app_version(&ver);
    let delay = Duration::from_millis(300);
    let h1: WorkflowHandle<String> = client
        .debouncer(&wf)
        .debounce(&key, delay, "a".to_string())
        .await?;
    tokio::time::sleep(Duration::from_millis(50)).await;
    let h2: WorkflowHandle<String> = client
        .debouncer(&wf)
        .debounce(&key, delay, "b".to_string())
        .await?;
    tokio::time::sleep(Duration::from_millis(50)).await;
    let h3: WorkflowHandle<String> = client
        .debouncer(&wf)
        .debounce(&key, delay, "c".to_string())
        .await?;

    assert_eq!(h1.id(), h2.id());
    assert_eq!(h2.id(), h3.id());
    let out = h3.result().await?;
    assert_eq!(out, "c", "the debounced run used the latest input");

    engine.shutdown(Duration::from_secs(2)).await?;

    // The pinned engine registered `ver` at launch, making it the latest
    // application version. Unversioned-row dequeue gating is latest-only, so
    // restore this binary's real version as latest or other tests' ('')
    // client-enqueued claims on the shared database would starve. Register it
    // first (idempotent): registration happens at `launch()`, which this
    // engine never does, and on a fresh database (CI) no other test may have
    // registered it yet.
    let real = DurableEngine::new(provider.clone()).await?;
    provider
        .create_application_version(real.app_version())
        .await?;
    tokio::time::sleep(Duration::from_millis(5)).await;
    assert!(
        real.set_latest_application_version(real.app_version())
            .await?
    );
    Ok(())
}

/// Dequeue version gating over live Postgres: an executor claims only rows of
/// its exact version; unversioned ('') rows are claimable only by the executor
/// running the latest registered application version. Uses a uuid-scoped queue
/// and the database's ambient latest (this binary's version) — no global state
/// is mutated.
#[tokio::test]
async fn pg_dequeue_gates_by_version() -> Result<()> {
    use durare::DequeueRequest;
    let Some(url) = database_url() else {
        eprintln!("skipping pg_dequeue_gates_by_version: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let queue = format!("q-{tag}");
    let (v_foo, v_bar) = (format!("vf-{tag}"), format!("vb-{tag}"));

    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    provider.init().await?;
    // This binary's version is (or is about to be) the latest — the same engine
    // registration every other test performs.
    let engine = DurableEngine::new(provider.clone()).await?;
    engine.launch().await?;
    engine
        .set_latest_application_version(engine.app_version())
        .await?;

    let mk = |id: String, ver: &str| {
        let mut s = WorkflowStatus::new(
            id,
            "wf",
            serde_json::Value::Null,
            durare::STATUS_ENQUEUED,
            "",
            ver,
        );
        s.queue_name = Some(queue.clone());
        s
    };
    let req = |ver: &str| DequeueRequest {
        queue_name: queue.clone(),
        executor_id: "exec".into(),
        app_version: ver.into(),
        partition_key: None,
        max_tasks: 10,
        global_concurrency: None,
        rate_limit_max: None,
        rate_limit_period_ms: None,
    };
    let id_foo = format!("wf-foo-{tag}");
    let id_bare = format!("wf-bare-{tag}");
    provider
        .insert_workflow_status(mk(id_foo.clone(), &v_foo))
        .await?;
    provider
        .insert_workflow_status(mk(id_bare.clone(), ""))
        .await?;

    // A non-latest, non-matching executor claims nothing.
    assert!(
        provider.dequeue_workflows(&req(&v_bar)).await?.is_empty(),
        "wrong version + not latest claims neither row"
    );
    // Exact version match claims its own row only (v_foo is not latest).
    let got: Vec<String> = provider
        .dequeue_workflows(&req(&v_foo))
        .await?
        .into_iter()
        .map(|w| w.id)
        .collect();
    assert_eq!(got, vec![id_foo.clone()]);
    // The latest-version executor claims the unversioned row.
    let got: Vec<String> = provider
        .dequeue_workflows(&req(engine.app_version()))
        .await?
        .into_iter()
        .map(|w| w.id)
        .collect();
    assert_eq!(got, vec![id_bare.clone()]);

    engine.shutdown(Duration::from_secs(1)).await?;
    engine.delete_workflows(&[id_foo, id_bare], false).await?;
    Ok(())
}

/// Config-name routing over live Postgres: a dispatcher reads the persisted
/// `config_name` from the claimed row and routes to the matching instance;
/// `class_name`/`config_name` round-trip on the row.
#[tokio::test]
async fn pg_config_name_routes_on_queue_dispatch() -> Result<()> {
    use durare::WorkflowQueue;
    let Some(url) = database_url() else {
        eprintln!("skipping pg_config_name_routes_on_queue_dispatch: DATABASE_URL unset");
        return Ok(());
    };
    let wf = format!("greet-{}", uuid::Uuid::new_v4());
    let queue = format!("q-{}", uuid::Uuid::new_v4());
    let id_fr = format!("wf-fr-{}", uuid::Uuid::new_v4());
    let id_en = format!("wf-en-{}", uuid::Uuid::new_v4());

    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new(provider.clone()).await?;
    engine.register_configured(&wf, "en", |_ctx: DurableContext, who: String| async move {
        Ok::<_, Error>(format!("Hello, {who}"))
    });
    engine.register_configured(&wf, "fr", |_ctx: DurableContext, who: String| async move {
        Ok::<_, Error>(format!("Bonjour, {who}"))
    });
    engine.register_queue(
        WorkflowQueue::new(&queue).base_polling_interval(Duration::from_millis(10)),
    );
    engine.launch().await?;

    let fr = engine
        .start::<_, String>(
            &wf,
            "Sam".to_string(),
            WorkflowOptions::with_id(&id_fr)
                .config_name("fr")
                .class_name("Greeter")
                .queue(&queue),
        )
        .await?;
    let en = engine
        .start::<_, String>(
            &wf,
            "Sam".to_string(),
            WorkflowOptions::with_id(&id_en)
                .config_name("en")
                .queue(&queue),
        )
        .await?;
    assert_eq!(fr.result().await?, "Bonjour, Sam");
    assert_eq!(en.result().await?, "Hello, Sam");

    let status = engine
        .retrieve_workflow::<String>(&id_fr)
        .await?
        .get_status()
        .await?;
    assert_eq!(status.config_name.as_deref(), Some("fr"));
    assert_eq!(status.class_name.as_deref(), Some("Greeter"));

    engine.shutdown(Duration::from_secs(2)).await?;
    Ok(())
}

/// Client management over Postgres: reschedule a DELAYED workflow so the engine
/// runs it, and cancel + delete another. Names are uuid-scoped for isolation.
#[tokio::test]
async fn pg_client_manages_workflows() -> Result<()> {
    use durare::WorkflowQueue;
    let Some(url) = database_url() else {
        eprintln!("skipping pg_client_manages_workflows: DATABASE_URL unset");
        return Ok(());
    };
    let wf = format!("ping-{}", uuid::Uuid::new_v4());
    let queue = format!("q-{}", uuid::Uuid::new_v4());
    let run_id = format!("run-{}", uuid::Uuid::new_v4());
    let cancel_id = format!("cancel-{}", uuid::Uuid::new_v4());

    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new(provider.clone()).await?;
    engine.register(&wf, |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    engine.register_queue(WorkflowQueue::new(&queue));
    engine.launch().await?;
    let client = durare::Client::new(provider.clone());

    // Enqueue far in the future, then pull it in: the engine runs it.
    let run = client
        .enqueue::<_, ()>(
            &queue,
            &wf,
            (),
            WorkflowOptions {
                workflow_id: Some(run_id.clone()),
                delay: Some(Duration::from_secs(60)),
                ..Default::default()
            },
        )
        .await?;
    assert!(
        client
            .set_workflow_delay(&run_id, Duration::from_millis(10))
            .await?
    );
    run.result().await?;

    // Enqueue another, cancel it, then delete it.
    client
        .enqueue::<_, ()>(
            &queue,
            &wf,
            (),
            WorkflowOptions {
                workflow_id: Some(cancel_id.clone()),
                delay: Some(Duration::from_secs(60)),
                ..Default::default()
            },
        )
        .await?;
    client.cancel_workflow(&cancel_id).await?;
    let one = client
        .list_workflows(&ListFilter {
            workflow_id_prefix: vec![cancel_id.clone()],
            ..Default::default()
        })
        .await?;
    assert_eq!(one.len(), 1);
    assert_eq!(one[0].status, "CANCELLED");
    client
        .delete_workflows(std::slice::from_ref(&cancel_id), false)
        .await?;
    assert!(client
        .list_workflows(&ListFilter {
            workflow_id_prefix: vec![cancel_id],
            ..Default::default()
        })
        .await?
        .is_empty());

    engine.shutdown(Duration::from_secs(2)).await?;
    Ok(())
}

/// Client schedule management persists through Postgres: create with context +
/// queue, read back, pause (drops from the active filter), apply (replace), and
/// delete. A uuid-suffixed name isolates it; no engine runs, so nothing fires.
#[tokio::test]
async fn pg_client_manages_schedules() -> Result<()> {
    use durare::{ApplySchedule, Client, ScheduleFilter, ScheduleOptions, ScheduleStatus};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_client_manages_schedules: DATABASE_URL unset");
        return Ok(());
    };
    let name = format!("nightly-{}", uuid::Uuid::new_v4());
    // A Client requires an initialized database (it never migrates); on a fresh
    // schema that's the application's (here: the test's) job.
    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    provider.init().await?;
    let client = Client::new(provider);

    client
        .create_schedule(
            &name,
            "report",
            "0 0 0 * * *",
            ScheduleOptions::new()
                .context(&serde_json::json!({"region": "us"}))
                .queue_name("internal"),
        )
        .await?;
    let got = client.get_schedule(&name).await?.expect("created");
    assert_eq!(got.workflow_name, "report");
    assert_eq!(got.queue_name.as_deref(), Some("internal"));

    // Duplicate rejected.
    assert!(client
        .create_schedule(&name, "report", "0 0 0 * * *", ScheduleOptions::new())
        .await
        .is_err());

    assert!(client.pause_schedule(&name).await?);
    assert!(client
        .list_schedules(&ScheduleFilter {
            statuses: vec![ScheduleStatus::Active],
            name_prefixes: vec![name.clone()],
            ..Default::default()
        })
        .await?
        .is_empty());

    // apply replaces it with a fresh id.
    let before = got.schedule_id;
    client
        .apply_schedules(vec![ApplySchedule::new(&name, "report", "0 0 1 * * *")])
        .await?;
    let after = client.get_schedule(&name).await?.expect("replaced");
    assert_ne!(after.schedule_id, before);
    assert_eq!(after.schedule, "0 0 1 * * *");

    assert!(client.delete_schedule(&name).await?);
    assert!(client.get_schedule(&name).await?.is_none());
    Ok(())
}

/// A client backfills and triggers a direct schedule on Postgres: backfilled
/// ticks and the trigger land ENQUEUED on the internal queue under their
/// deterministic ids, and re-backfilling the same window is idempotent. The
/// client pins a unique application version so the dequeue version gate keeps
/// any other test's running engine from claiming these ticks off the shared
/// internal queue — the rows stay enqueued, as asserted.
#[tokio::test]
async fn pg_client_backfills_and_triggers_a_schedule() -> Result<()> {
    use chrono::{TimeZone, Utc};
    use durare::{Client, ScheduleOptions};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_client_backfills_and_triggers_a_schedule: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let name = format!("bf-{tag}");
    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    provider.init().await?;
    let client = Client::new(provider).with_app_version(format!("v-{tag}"));

    client
        .create_schedule(&name, "report", "0 0 12 * * *", ScheduleOptions::new())
        .await?;

    let start = Utc.with_ymd_and_hms(2026, 3, 1, 0, 0, 0).unwrap();
    let end = Utc.with_ymd_and_hms(2026, 3, 4, 0, 0, 0).unwrap();
    let ids = client.backfill_schedule(&name, start, end).await?;
    assert_eq!(ids.len(), 3, "one tick per day");

    let backfilled = ListFilter {
        workflow_id_prefix: vec![format!("sched-{name}-")],
        ..Default::default()
    };
    let rows = client.list_workflows(&backfilled).await?;
    assert_eq!(rows.len(), 3);
    assert!(rows.iter().all(|r| r.status == "ENQUEUED"));
    assert!(rows
        .iter()
        .all(|r| r.queue_name.as_deref() == Some("_dbos_internal_queue")));

    // Idempotent re-backfill: same ids, no new rows.
    assert_eq!(client.backfill_schedule(&name, start, end).await?, ids);

    // Trigger enqueues one more run under a distinct `-trigger-` id.
    let h: durare::WorkflowHandle<()> = client.trigger_schedule(&name).await?;
    assert!(h.id().starts_with(&format!("sched-{name}-trigger-")));
    let trig = client
        .list_workflows(&ListFilter {
            workflow_id_prefix: vec![format!("sched-{name}-trigger-")],
            ..Default::default()
        })
        .await?;
    assert_eq!(trig.len(), 1, "trigger enqueued exactly one run");

    client.delete_schedule(&name).await?;
    Ok(())
}

/// Portable mode stores a workflow's input in the cross-language args envelope
/// `{"positionalArgs":[…],"namedArgs":{}}` (so a DBOS app in another language can
/// run it), while the output stays a bare portable value; the input reads back
/// unwrapped to what the workflow receives.
#[tokio::test]
async fn pg_portable_input_envelope() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_portable_input_envelope: DATABASE_URL unset");
        return Ok(());
    };
    let id = format!("wf-env-{}", uuid::Uuid::new_v4());

    let engine = engine_with(&url, Serializer::Portable).await?;
    let out: String = engine
        .start::<_, String>("greet", "ada".to_string(), WorkflowOptions::with_id(&id))
        .await?
        .result()
        .await?;
    assert_eq!(out, "hi ada");

    // Read the raw columns: input is the args envelope, output is a bare value.
    let pool = sqlx::PgPool::connect(&url).await?;
    let raw_inputs: String =
        sqlx::query_scalar("SELECT inputs FROM dbos.workflow_status WHERE workflow_uuid = $1")
            .bind(&id)
            .fetch_one(&pool)
            .await?;
    assert_eq!(raw_inputs, r#"{"positionalArgs":["ada"],"namedArgs":{}}"#);
    let raw_output: String =
        sqlx::query_scalar("SELECT output FROM dbos.workflow_status WHERE workflow_uuid = $1")
            .bind(&id)
            .fetch_one(&pool)
            .await?;
    assert_eq!(
        raw_output, r#""hi ada""#,
        "output is a bare portable value, not enveloped"
    );

    // Through the provider, the envelope is unwrapped to the bare input.
    let status = engine
        .retrieve_workflow::<String>(&id)
        .await?
        .get_status()
        .await?;
    assert_eq!(status.input, serde_json::json!("ada"));
    Ok(())
}

/// On Postgres, the client honors a per-enqueue application-version override and
/// the `ReturnExisting` dedup policy (a colliding dedup id returns the holder).
#[tokio::test]
async fn pg_enqueue_dedup_and_app_version() -> Result<()> {
    use durare::{Client, DeduplicationPolicy, WorkflowHandle};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_enqueue_dedup_and_app_version: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let queue = format!("dq-{tag}");
    let dedup = format!("once-{tag}");
    let client = Client::new(Arc::new(PostgresProvider::connect(&url).await?));

    // Per-enqueue application-version override.
    let ver_id = format!("wf-ver-{tag}");
    let _: WorkflowHandle<i64> = client
        .enqueue(
            &queue,
            "wf",
            1i64,
            WorkflowOptions::with_id(&ver_id).app_version("v-special"),
        )
        .await?;
    let rows = client
        .list_workflows(&ListFilter {
            workflow_id_prefix: vec![ver_id.clone()],
            ..Default::default()
        })
        .await?;
    assert_eq!(rows[0].app_version, "v-special");

    // ReturnExisting returns the holder; the default rejects.
    let first: WorkflowHandle<i64> = client
        .enqueue(
            &queue,
            "wf",
            1i64,
            WorkflowOptions::with_id(format!("d1-{tag}")).dedup_id(&dedup),
        )
        .await?;
    assert!(client
        .enqueue::<_, i64>(
            &queue,
            "wf",
            2i64,
            WorkflowOptions::with_id(format!("d2-{tag}")).dedup_id(&dedup),
        )
        .await
        .is_err());
    let again: WorkflowHandle<i64> = client
        .enqueue(
            &queue,
            "wf",
            3i64,
            WorkflowOptions::with_id(format!("d3-{tag}"))
                .dedup_id(&dedup)
                .dedup_policy(DeduplicationPolicy::ReturnExisting),
        )
        .await?;
    assert_eq!(again.id(), first.id());

    client
        .delete_workflows(&[first.id().to_string(), ver_id], false)
        .await?;
    Ok(())
}

/// A deduplication id is released once its holder reaches a terminal state, so
/// the same id can be enqueued again afterward.
#[tokio::test]
async fn pg_dedup_slot_frees_on_completion() -> Result<()> {
    use durare::{Client, WorkflowHandle, STATUS_SUCCESS};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_dedup_slot_frees_on_completion: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let queue = format!("dq-{tag}");
    let dedup = format!("once-{tag}");
    let d1 = format!("d1-{tag}");
    let d3 = format!("d3-{tag}");
    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let client = Client::new(provider.clone());

    let first: WorkflowHandle<i64> = client
        .enqueue(
            &queue,
            "wf",
            1i64,
            WorkflowOptions::with_id(&d1).dedup_id(&dedup),
        )
        .await?;
    // The partial unique index rejects a colliding dedup id while d1 is active.
    assert!(client
        .enqueue::<_, i64>(
            &queue,
            "wf",
            2i64,
            WorkflowOptions::with_id(format!("d2-{tag}")).dedup_id(&dedup),
        )
        .await
        .is_err());

    // Completing d1 nulls its deduplication_id, freeing the slot.
    provider
        .set_workflow_status(first.id(), STATUS_SUCCESS, None, None)
        .await?;

    let third: WorkflowHandle<i64> = client
        .enqueue(
            &queue,
            "wf",
            3i64,
            WorkflowOptions::with_id(&d3).dedup_id(&dedup),
        )
        .await?;
    assert_eq!(third.id(), d3);

    client.delete_workflows(&[d1, d3], false).await?;
    Ok(())
}

/// A workflow cancelled during its final step must stay cancelled: a late
/// SUCCESS/ERROR completion is rejected and does not overwrite the status.
#[tokio::test]
async fn pg_completion_cannot_overwrite_cancelled() -> Result<()> {
    use durare::{Client, WorkflowHandle, STATUS_CANCELLED, STATUS_SUCCESS};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_completion_cannot_overwrite_cancelled: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let id = format!("c1-{tag}");
    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let client = Client::new(provider.clone());

    let h: WorkflowHandle<i64> = client
        .enqueue(
            &format!("q-{tag}"),
            "wf",
            1i64,
            WorkflowOptions::with_id(&id),
        )
        .await?;
    provider
        .set_workflow_status(h.id(), STATUS_CANCELLED, None, Some("cancelled"))
        .await?;

    // A completion racing the cancellation must error, not flip it to SUCCESS.
    let late = provider
        .set_workflow_status(h.id(), STATUS_SUCCESS, None, None)
        .await;
    assert!(late.is_err(), "completing a cancelled workflow must error");

    let row = provider.get_workflow_status(h.id()).await?.unwrap();
    assert_eq!(row.status, STATUS_CANCELLED);

    client.delete_workflows(&[id], false).await?;
    Ok(())
}

/// Through Postgres: `apply_schedules` is atomic — a mid-batch failure rolls the
/// whole batch back, leaving any schedule it would have replaced untouched.
#[tokio::test]
async fn pg_apply_schedules_is_atomic() -> Result<()> {
    use durare::{ScheduleFilter, ScheduleStatus, WorkflowSchedule};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_apply_schedules_is_atomic: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let prefix = format!("sch-{tag}-");
    let keep = format!("{prefix}keep");
    let other = format!("{prefix}other");
    let dup = format!("dup-{tag}");
    let provider = PostgresProvider::connect(&url).await?;
    // This test drives the provider directly (no engine), so run migrations here —
    // every engine-backed test gets the schema via `DurableEngine::new`.
    provider.init().await?;

    let make = |id: String, name: String, cron: &str| WorkflowSchedule {
        schedule_id: id,
        schedule_name: name,
        workflow_name: "wf".to_string(),
        schedule: cron.to_string(),
        status: ScheduleStatus::Active,
        context: None,
        last_fired_at: None,
        automatic_backfill: false,
        cron_timezone: None,
        queue_name: None,
    };

    provider
        .apply_schedules(&[make(format!("idkeep-{tag}"), keep.clone(), "0 0 1 * * *")])
        .await?;

    // The second insert reuses schedule_id `dup` → PRIMARY KEY violation, so the
    // whole batch (including the replacement of `keep`) must roll back.
    let bad = vec![
        make(dup.clone(), keep.clone(), "0 0 2 * * *"),
        make(dup.clone(), other.clone(), "0 0 3 * * *"),
    ];
    assert!(
        provider.apply_schedules(&bad).await.is_err(),
        "a duplicate schedule_id must fail the batch"
    );

    let mine = provider
        .list_schedules(&ScheduleFilter {
            name_prefixes: vec![prefix],
            ..Default::default()
        })
        .await?;
    assert_eq!(mine.len(), 1, "rollback left exactly the original schedule");
    assert_eq!(mine[0].schedule_name, keep);
    assert_eq!(mine[0].schedule_id, format!("idkeep-{tag}"), "original id");
    assert_eq!(mine[0].schedule, "0 0 1 * * *", "original cron preserved");

    provider.delete_schedule(&keep).await?;
    provider.delete_schedule(&other).await?;
    Ok(())
}

/// Through Postgres: a transactional step commits the user's writes and the step
/// checkpoint together. Re-running the same workflow id replays the debit from
/// its checkpoint instead of reapplying it (exactly-once).
#[tokio::test]
async fn pg_transaction_step() -> Result<()> {
    use durare::params;
    let Some(url) = database_url() else {
        eprintln!("skipping pg_transaction_step: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4().simple().to_string();
    let table = format!("acct_{tag}");
    let wf = format!("wf-txn-{tag}");
    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new(provider.clone()).await?;

    engine.register("acct", |ctx: DurableContext, table: String| async move {
        let t = table.clone();
        ctx.transaction::<(), _>("setup", move |tx| {
            let t = t.clone();
            Box::pin(async move {
                tx.execute(
                    &format!("CREATE TABLE IF NOT EXISTS {t} (id INT PRIMARY KEY, bal BIGINT)"),
                    &params![],
                )
                .await?;
                tx.execute(
                    &format!(
                        "INSERT INTO {t} (id, bal) VALUES (1, 100) ON CONFLICT (id) DO NOTHING"
                    ),
                    &params![],
                )
                .await?;
                Ok(())
            })
        })
        .await?;
        let t = table.clone();
        let bal: i64 = ctx
            .transaction("debit", move |tx| {
                let t = t.clone();
                Box::pin(async move {
                    tx.execute(
                        &format!("UPDATE {t} SET bal = bal - ? WHERE id = ?"),
                        &params![10_i64, 1_i64],
                    )
                    .await?;
                    let row = tx
                        .query_one(
                            &format!("SELECT bal FROM {t} WHERE id = ?"),
                            &params![1_i64],
                        )
                        .await?;
                    Ok(row.get::<i64>("bal"))
                })
            })
            .await?;
        Ok::<_, Error>(bal)
    });

    // Register the cleanup workflow up front too: the shared runtime is built on
    // the first run, so a later registration would not be seen.
    engine.register("drop", |ctx: DurableContext, table: String| async move {
        ctx.transaction::<(), _>("drop", move |tx| {
            let table = table.clone();
            Box::pin(async move {
                tx.execute(&format!("DROP TABLE IF EXISTS {table}"), &params![])
                    .await?;
                Ok(())
            })
        })
        .await?;
        Ok::<_, Error>(())
    });

    let bal1: i64 = engine
        .start("acct", table.clone(), WorkflowOptions::with_id(&wf))
        .await?
        .result()
        .await?;
    assert_eq!(bal1, 90);
    // Re-run the same id: the debit replays from its checkpoint, not reapplied.
    let bal2: i64 = engine
        .start("acct", table.clone(), WorkflowOptions::with_id(&wf))
        .await?
        .result()
        .await?;
    assert_eq!(bal2, 90, "exactly-once: re-running does not debit again");

    // Clean up the user table and the workflow rows.
    let drop_wf = format!("wf-drop-{tag}");
    let _: () = engine
        .start("drop", table.clone(), WorkflowOptions::with_id(&drop_wf))
        .await?
        .result()
        .await?;
    provider.delete_workflows(&[wf, drop_wf], false).await?;
    Ok(())
}

/// A transactional step whose body writes a row and then returns an application
/// error rolls the write back on Postgres: a later step reads the original value.
/// This is the shared-system-DB-pool path — where Go's `runAsTxn` instead commits
/// the partial write (its `runastxn_partial_write` repro documents that as a bug).
/// Rust records the error *outside* the body's transaction, so the body's tx
/// aborts cleanly and nothing it wrote persists.
#[tokio::test]
async fn pg_transaction_step_rolls_back_on_error() -> Result<()> {
    use durare::params;
    let Some(url) = database_url() else {
        eprintln!("skipping pg_transaction_step_rolls_back_on_error: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4().simple().to_string();
    let table = format!("kv_{tag}");
    let wf = format!("wf-txnrb-{tag}");
    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new(provider.clone()).await?;

    engine.register("rb", |ctx: DurableContext, table: String| async move {
        let t = table.clone();
        ctx.transaction::<(), _>("setup", move |tx| {
            let t = t.clone();
            Box::pin(async move {
                tx.execute(
                    &format!("CREATE TABLE IF NOT EXISTS {t} (id INT PRIMARY KEY, v BIGINT)"),
                    &params![],
                )
                .await?;
                tx.execute(
                    &format!("INSERT INTO {t} (id, v) VALUES (1, 0) ON CONFLICT (id) DO NOTHING"),
                    &params![],
                )
                .await?;
                Ok(())
            })
        })
        .await?;

        // Write, then return an application error: the body's write must roll back
        // with its transaction — the error is recorded separately, not committed
        // inside the aborted body tx.
        let t = table.clone();
        let failed = ctx
            .transaction::<(), _>("bad", move |tx| {
                let t = t.clone();
                Box::pin(async move {
                    tx.execute(&format!("UPDATE {t} SET v = 999 WHERE id = 1"), &params![])
                        .await?;
                    Err(Error::app("business rule violated"))
                })
            })
            .await
            .is_err();

        let t = table.clone();
        let v: i64 = ctx
            .transaction("read", move |tx| {
                let t = t.clone();
                Box::pin(async move {
                    let row = tx
                        .query_one(&format!("SELECT v FROM {t} WHERE id = 1"), &params![])
                        .await?;
                    Ok(row.get::<i64>("v"))
                })
            })
            .await?;
        Ok::<_, Error>((failed, v))
    });

    engine.register("drop", |ctx: DurableContext, table: String| async move {
        ctx.transaction::<(), _>("drop", move |tx| {
            let table = table.clone();
            Box::pin(async move {
                tx.execute(&format!("DROP TABLE IF EXISTS {table}"), &params![])
                    .await?;
                Ok(())
            })
        })
        .await?;
        Ok::<_, Error>(())
    });

    let (failed, v): (bool, i64) = engine
        .start("rb", table.clone(), WorkflowOptions::with_id(&wf))
        .await?
        .result()
        .await?;
    assert!(failed, "the failing transaction returned its error");
    assert_eq!(
        v, 0,
        "the failed transaction's write rolled back on Postgres"
    );

    let drop_wf = format!("wf-drop-{tag}");
    let _: () = engine
        .start("drop", table.clone(), WorkflowOptions::with_id(&drop_wf))
        .await?
        .result()
        .await?;
    provider.delete_workflows(&[wf, drop_wf], false).await?;
    Ok(())
}

/// Two concurrent SERIALIZABLE transactional steps read-then-write the same row.
/// One hits a serialization conflict and retries on a fresh transaction, so
/// neither increment is lost (the counter ends at 2, results are {1, 2}).
#[tokio::test]
async fn pg_transaction_serializable_retries_on_conflict() -> Result<()> {
    use durare::{params, IsolationLevel, TransactionOptions};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_transaction_serializable_retries_on_conflict: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4().simple().to_string();
    let table = format!("ctr_{tag}");
    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new(provider.clone()).await?;

    // All workflows registered up front (the shared runtime is built on first run).
    engine.register("seed", |ctx: DurableContext, table: String| async move {
        ctx.transaction::<(), _>("seed", move |tx| {
            let table = table.clone();
            Box::pin(async move {
                tx.execute(
                    &format!("CREATE TABLE IF NOT EXISTS {table} (id INT PRIMARY KEY, v BIGINT)"),
                    &params![],
                )
                .await?;
                tx.execute(
                    &format!(
                        "INSERT INTO {table} (id, v) VALUES (1, 0) ON CONFLICT (id) DO NOTHING"
                    ),
                    &params![],
                )
                .await?;
                Ok(())
            })
        })
        .await?;
        Ok::<_, Error>(())
    });
    engine.register("incr", |ctx: DurableContext, table: String| async move {
        let t = table.clone();
        let v: i64 = ctx
            .transaction_with(
                TransactionOptions::new("incr").isolation(IsolationLevel::Serializable),
                move |tx| {
                    let t = t.clone();
                    Box::pin(async move {
                        let row = tx
                            .query_one(&format!("SELECT v FROM {t} WHERE id = 1"), &params![])
                            .await?;
                        let cur: i64 = row.get("v");
                        // Widen the read-write window so the two overlap.
                        tokio::time::sleep(std::time::Duration::from_millis(60)).await;
                        tx.execute(
                            &format!("UPDATE {t} SET v = ? WHERE id = 1"),
                            &params![cur + 1],
                        )
                        .await?;
                        Ok(cur + 1)
                    })
                },
            )
            .await?;
        Ok::<_, Error>(v)
    });
    engine.register("drop", |ctx: DurableContext, table: String| async move {
        ctx.transaction::<(), _>("drop", move |tx| {
            let table = table.clone();
            Box::pin(async move {
                tx.execute(&format!("DROP TABLE IF EXISTS {table}"), &params![])
                    .await?;
                Ok(())
            })
        })
        .await?;
        Ok::<_, Error>(())
    });

    let _: () = engine
        .start(
            "seed",
            table.clone(),
            WorkflowOptions::with_id(format!("seed-{tag}")),
        )
        .await?
        .result()
        .await?;
    let (id_a, id_b) = (format!("a-{tag}"), format!("b-{tag}"));
    let ha = engine
        .start::<_, i64>("incr", table.clone(), WorkflowOptions::with_id(&id_a))
        .await?;
    let hb = engine
        .start::<_, i64>("incr", table.clone(), WorkflowOptions::with_id(&id_b))
        .await?;
    let (a, b) = tokio::join!(ha.result(), hb.result());
    let mut results = [a?, b?];
    results.sort();
    assert_eq!(
        results,
        [1, 2],
        "both increments applied — the serialization conflict was retried"
    );

    let _: () = engine
        .start(
            "drop",
            table.clone(),
            WorkflowOptions::with_id(format!("drop-{tag}")),
        )
        .await?
        .result()
        .await?;
    provider
        .delete_workflows(
            &[
                format!("seed-{tag}"),
                format!("a-{tag}"),
                format!("b-{tag}"),
                format!("drop-{tag}"),
            ],
            false,
        )
        .await?;
    Ok(())
}

/// A transactional step's failure is checkpointed across a real restart: a fresh
/// engine over the same database replays the recorded error without re-running
/// the body.
#[tokio::test]
async fn pg_checkpoints_a_caught_transaction_failure() -> Result<()> {
    use std::sync::atomic::{AtomicUsize, Ordering};
    static TX_RUNS: AtomicUsize = AtomicUsize::new(0);
    let Some(url) = database_url() else {
        eprintln!("skipping pg_checkpoints_a_caught_transaction_failure: DATABASE_URL unset");
        return Ok(());
    };
    TX_RUNS.store(0, Ordering::SeqCst);
    let id = format!("wf-txn-err-{}", uuid::Uuid::new_v4());

    let register = |engine: &mut DurableEngine| {
        engine.register("txn_flaky", |ctx: DurableContext, _: ()| async move {
            let r: Result<i64> = ctx
                .transaction::<i64, _>("maybe", |_tx| {
                    Box::pin(async move {
                        let n = TX_RUNS.fetch_add(1, Ordering::SeqCst);
                        if n == 0 {
                            Err(Error::app("transient"))
                        } else {
                            Ok(7)
                        }
                    })
                })
                .await;
            Ok::<_, Error>(if r.is_ok() { "ok" } else { "caught-error" }.to_string())
        });
    };

    {
        let mut engine =
            DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;
        register(&mut engine);
        let out: String = engine
            .start("txn_flaky", (), WorkflowOptions::with_id(&id))
            .await?
            .result()
            .await?;
        assert_eq!(out, "caught-error");
        let steps = engine.get_workflow_steps(&id).await?;
        let maybe = steps
            .iter()
            .find(|s| s.name == "maybe")
            .expect("step recorded");
        assert_eq!(maybe.error.as_deref(), Some("transient"));
    }
    {
        let mut engine =
            DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;
        register(&mut engine);
        let out: String = engine
            .start("txn_flaky", (), WorkflowOptions::with_id(&id))
            .await?
            .result()
            .await?;
        assert_eq!(out, "caught-error", "replay observes the recorded error");
    }
    assert_eq!(
        TX_RUNS.load(Ordering::SeqCst),
        1,
        "a checkpointed failed transaction step is not re-run on replay"
    );
    Ok(())
}

/// `TransactionOptions::max_retries` re-runs the whole body on an application
/// error until it succeeds; a `retry_if` predicate that rejects the error fails
/// fast without burning the budget. Both run against a real Postgres transaction.
#[tokio::test]
async fn pg_transaction_body_user_retry() -> Result<()> {
    use durare::params;
    use std::sync::atomic::{AtomicUsize, Ordering};
    static RETRY_RUNS: AtomicUsize = AtomicUsize::new(0);
    static FAST_RUNS: AtomicUsize = AtomicUsize::new(0);
    let Some(url) = database_url() else {
        eprintln!("skipping pg_transaction_body_user_retry: DATABASE_URL unset");
        return Ok(());
    };
    RETRY_RUNS.store(0, Ordering::SeqCst);
    FAST_RUNS.store(0, Ordering::SeqCst);
    let mut engine = DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;

    // Body fails twice then succeeds; max_retries(3) re-runs it to success.
    engine.register("retry", |ctx: DurableContext, _: ()| async move {
        let opts = TransactionOptions::new("flaky")
            .max_retries(3)
            .base_interval(Duration::from_millis(1));
        let n: i64 = ctx
            .transaction_with(opts, |tx| {
                Box::pin(async move {
                    tx.execute("SELECT 1", &params![]).await?;
                    let run = RETRY_RUNS.fetch_add(1, Ordering::SeqCst);
                    if run < 2 {
                        Err(Error::app("transient"))
                    } else {
                        Ok(42_i64)
                    }
                })
            })
            .await?;
        Ok::<_, Error>(n)
    });

    // A rejected error stops immediately despite max_retries(5).
    engine.register("nofast", |ctx: DurableContext, _: ()| async move {
        let opts = TransactionOptions::new("permanent")
            .max_retries(5)
            .base_interval(Duration::from_millis(1))
            .retry_if(|e: &Error| e.is_retryable());
        let r: Result<i64> = ctx
            .transaction_with(opts, |tx| {
                Box::pin(async move {
                    tx.execute("SELECT 1", &params![]).await?;
                    FAST_RUNS.fetch_add(1, Ordering::SeqCst);
                    Err(Error::app("permanent failure"))
                })
            })
            .await;
        Ok::<_, Error>(r.is_err())
    });

    let out: i64 = engine
        .start(
            "retry",
            (),
            WorkflowOptions::with_id(format!("wf-txn-retry-{}", uuid::Uuid::new_v4())),
        )
        .await?
        .result()
        .await?;
    assert_eq!(out, 42, "the body eventually succeeds");
    assert_eq!(
        RETRY_RUNS.load(Ordering::SeqCst),
        3,
        "the body re-runs twice then succeeds (1 + 2 retries)"
    );

    let failed: bool = engine
        .start(
            "nofast",
            (),
            WorkflowOptions::with_id(format!("wf-txn-nofast-{}", uuid::Uuid::new_v4())),
        )
        .await?
        .result()
        .await?;
    assert!(failed, "the error is surfaced to the caller");
    assert_eq!(
        FAST_RUNS.load(Ordering::SeqCst),
        1,
        "a rejected error is not retried despite max_retries(5)"
    );
    Ok(())
}

/// A custom system-tables schema isolates tenants sharing one database: each
/// `connect_with_schema` provider creates and migrates its own schema, a
/// workflow run in one is invisible to the other, and an invalid schema name is
/// rejected up front.
#[tokio::test]
async fn pg_custom_schema_isolates_tenants() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_custom_schema_isolates_tenants: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4().simple().to_string();
    let (schema_a, schema_b) = (format!("t_a_{tag}"), format!("t_b_{tag}"));
    let id = format!("wf-schema-{tag}");

    let provider_a = Arc::new(PostgresProvider::connect_with_schema(&url, &schema_a).await?);
    let provider_b = Arc::new(PostgresProvider::connect_with_schema(&url, &schema_b).await?);

    let mut engine_a = DurableEngine::new(provider_a.clone()).await?;
    engine_a.register("hello", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n * 2)
    });
    let out: i64 = engine_a
        .start::<_, i64>("hello", 21_i64, WorkflowOptions::with_id(&id))
        .await?
        .result()
        .await?;
    assert_eq!(out, 42);

    // Tenant A sees its run; tenant B — same database, different schema — does not.
    assert!(provider_a.get_workflow_status(&id).await?.is_some());
    let engine_b = DurableEngine::new(provider_b.clone()).await?; // migrates schema_b
    assert!(
        provider_b.get_workflow_status(&id).await?.is_none(),
        "a run in schema {schema_a} must be invisible from schema {schema_b}"
    );

    // Both schemas hold their own copy of the system tables.
    let count: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM information_schema.tables
         WHERE table_schema IN ($1, $2) AND table_name = 'workflow_status'",
    )
    .bind(&schema_a)
    .bind(&schema_b)
    .fetch_one(provider_a.pool())
    .await
    .map_err(Error::from)?;
    assert_eq!(count, 2, "each tenant schema has its own workflow_status");

    // A schema name that isn't a plain identifier is rejected before any SQL.
    assert!(
        PostgresProvider::connect_with_schema(&url, "bad-name; drop table x")
            .await
            .is_err(),
        "non-identifier schema names are rejected"
    );

    engine_a.shutdown(Duration::from_secs(1)).await?;
    engine_b.shutdown(Duration::from_secs(1)).await?;
    // Clean up the tenant schemas so repeated runs don't accumulate.
    for schema in [&schema_a, &schema_b] {
        sqlx::query(&format!("DROP SCHEMA {schema} CASCADE"))
            .execute(provider_a.pool())
            .await
            .map_err(Error::from)?;
    }
    Ok(())
}

/// export_workflow / import_workflow round-trip a workflow's full durable state
/// (status, steps, events, streams) through Postgres after deletion.
#[tokio::test]
async fn pg_export_import_round_trip() -> Result<()> {
    use durare::STATUS_SUCCESS;
    let Some(url) = database_url() else {
        eprintln!("skipping pg_export_import_round_trip: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let id = format!("wf-export-{tag}");

    let mut engine = DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;
    engine.register("expo", |ctx: DurableContext, n: i64| async move {
        let doubled = ctx
            .step("double", || async { Ok::<_, Error>(n * 2) })
            .await?;
        ctx.set_event("k", "v").await?;
        ctx.write_stream("s", 1_i64).await?;
        ctx.write_stream("s", 2_i64).await?;
        ctx.close_stream("s").await?;
        Ok::<_, Error>(doubled)
    });

    let out: i64 = engine
        .start::<_, i64>("expo", 21_i64, WorkflowOptions::with_id(&id))
        .await?
        .result()
        .await?;
    assert_eq!(out, 42);

    let exported = engine.export_workflow(&id, false).await?;
    assert_eq!(exported.len(), 1);
    engine
        .delete_workflows(std::slice::from_ref(&id), false)
        .await?;
    assert!(engine
        .list_workflows(&ListFilter {
            workflow_ids: vec![id.clone()],
            ..Default::default()
        })
        .await?
        .is_empty());

    engine.import_workflow(&exported).await?;
    let rows = engine
        .list_workflows(&ListFilter {
            workflow_ids: vec![id.clone()],
            load_output: true,
            ..Default::default()
        })
        .await?;
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].status, STATUS_SUCCESS);
    assert_eq!(rows[0].output, Some(serde_json::json!(42)));

    let steps = engine.get_workflow_steps(&id).await?;
    let double = steps.iter().find(|s| s.name == "double").expect("step");
    assert_eq!(double.output, Some(serde_json::json!(42)));

    let events = engine.list_workflow_events(&id).await?;
    assert_eq!(events, vec![("k".to_string(), serde_json::json!("v"))]);

    let streams = engine.list_workflow_streams(&id).await?;
    assert_eq!(
        streams,
        vec![(
            "s".to_string(),
            vec![serde_json::json!(1), serde_json::json!(2)]
        )]
    );

    // Importing again fails: import never overwrites an existing workflow.
    assert!(engine.import_workflow(&exported).await.is_err());

    engine
        .delete_workflows(std::slice::from_ref(&id), false)
        .await?;
    Ok(())
}

/// A blocked recv wakes via LISTEN/NOTIFY: an external send returns the message
/// well within the long backstop interval (proving the push path, not polling).
#[tokio::test]
async fn pg_recv_wakes_via_listen_notify() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_recv_wakes_via_listen_notify: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let id = format!("wf-recv-{tag}");

    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    assert!(provider.supports_listen_notify());
    let mut engine = DurableEngine::new(provider).await?;
    engine.register("waiter", |ctx: DurableContext, _: ()| async move {
        let msg: Option<String> = ctx.recv("topic", Duration::from_secs(10)).await?;
        Ok::<_, Error>(msg.unwrap_or_default())
    });

    let handle = engine
        .start::<_, String>("waiter", (), WorkflowOptions::with_id(&id))
        .await?;
    // Let the workflow reach recv and subscribe to the channel.
    tokio::time::sleep(Duration::from_millis(300)).await;

    let started = std::time::Instant::now();
    engine.send(&id, "hello".to_string(), "topic").await?;
    assert_eq!(handle.result().await?, "hello");
    let elapsed = started.elapsed();
    assert!(
        elapsed < Duration::from_secs(2),
        "recv should wake via NOTIFY, not the backstop (took {elapsed:?})"
    );

    engine
        .delete_workflows(std::slice::from_ref(&id), false)
        .await?;
    Ok(())
}

/// On Postgres, an idempotent send is deduplicated by the `notifications`
/// primary key: two sends sharing a key insert one row (the retry hits
/// `ON CONFLICT DO NOTHING`), while a distinct key inserts its own.
#[tokio::test]
async fn pg_send_with_idempotency_key_delivers_once() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_send_with_idempotency_key_delivers_once: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let id = format!("wf-idem-{tag}");

    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new(provider.clone()).await?;
    engine.register("sink", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    // A completed destination that can still receive messages.
    engine
        .start::<_, ()>("sink", (), WorkflowOptions::with_id(&id))
        .await?
        .result()
        .await?;

    engine
        .send_with_idempotency_key(&id, "a".to_string(), "t", "k1")
        .await?;
    engine
        .send_with_idempotency_key(&id, "a-again".to_string(), "t", "k1")
        .await?;
    engine
        .send_with_idempotency_key(&id, "b".to_string(), "t", "k2")
        .await?;

    // Exactly two messages persisted, in send order; the duplicate was dropped.
    let m1 = provider.consume_notification(&id, "t", 0, "probe").await?;
    let m2 = provider.consume_notification(&id, "t", 1, "probe").await?;
    let m3 = provider.consume_notification(&id, "t", 2, "probe").await?;
    assert_eq!(m1, Some(serde_json::json!("a")));
    assert_eq!(m2, Some(serde_json::json!("b")));
    assert_eq!(
        m3, None,
        "the duplicate keyed send was dropped by ON CONFLICT"
    );

    engine
        .delete_workflows(std::slice::from_ref(&id), false)
        .await?;
    Ok(())
}

/// A blocked get_event wakes via LISTEN/NOTIFY when another workflow sets the
/// event, returning well within the backstop interval.
#[tokio::test]
async fn pg_get_event_wakes_via_listen_notify() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_get_event_wakes_via_listen_notify: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let reader_id = format!("reader-{tag}");
    let setter_id = format!("setter-{tag}");

    let mut engine = DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;
    engine.register("reader", |ctx: DurableContext, target: String| async move {
        let v: Option<String> = ctx.get_event(&target, "k", Duration::from_secs(10)).await?;
        Ok::<_, Error>(v.unwrap_or_default())
    });
    engine.register("setter", |ctx: DurableContext, _: ()| async move {
        ctx.set_event("k", "v").await?;
        Ok::<_, Error>(String::new())
    });

    let reader = engine
        .start::<_, String>(
            "reader",
            setter_id.clone(),
            WorkflowOptions::with_id(&reader_id),
        )
        .await?;
    // Let the reader reach get_event and subscribe.
    tokio::time::sleep(Duration::from_millis(300)).await;

    let started = std::time::Instant::now();
    engine
        .start::<_, String>("setter", (), WorkflowOptions::with_id(&setter_id))
        .await?
        .result()
        .await?;
    assert_eq!(reader.result().await?, "v");
    let elapsed = started.elapsed();
    assert!(
        elapsed < Duration::from_secs(2),
        "get_event should wake via NOTIFY, not the backstop (took {elapsed:?})"
    );

    engine
        .delete_workflows(&[reader_id, setter_id], false)
        .await?;
    Ok(())
}

/// Multi-value list filters go through the SQL WHERE builder (`IN`/`= ANY`,
/// `LIKE ANY`) and `was_forked_from` on Postgres: OR-matching several names,
/// prefixes, and authenticated users, and isolating forks.
#[tokio::test]
async fn pg_list_filters_multi_value() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_list_filters_multi_value: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4().to_string();
    let mut engine = DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;
    engine.register("mvf_a", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n)
    });
    engine.register("mvf_b", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n)
    });
    engine.register("mvf_c", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n)
    });
    engine.launch().await?;

    let a_id = format!("mvf-a-{tag}");
    let b_id = format!("mvf-b-{tag}");
    let c_id = format!("mvf-c-{tag}");
    engine
        .start::<_, i64>(
            "mvf_a",
            1i64,
            WorkflowOptions::with_id(&a_id).authenticated_user("alice"),
        )
        .await?
        .result()
        .await?;
    engine
        .start::<_, i64>(
            "mvf_b",
            2i64,
            WorkflowOptions::with_id(&b_id).authenticated_user("bob"),
        )
        .await?
        .result()
        .await?;
    engine
        .start::<_, i64>(
            "mvf_c",
            3i64,
            WorkflowOptions::with_id(&c_id).authenticated_user("carol"),
        )
        .await?
        .result()
        .await?;

    // OR across names (mvf_a, mvf_b) scoped to this run's ids.
    let by_name = engine
        .list_workflows(&ListFilter {
            name: vec!["mvf_a".to_string(), "mvf_b".to_string()],
            workflow_ids: vec![a_id.clone(), b_id.clone(), c_id.clone()],
            ..Default::default()
        })
        .await?;
    assert_eq!(by_name.len(), 2, "IN-match on name");

    // OR across id prefixes (LIKE ANY).
    let by_prefix = engine
        .list_workflows(&ListFilter {
            workflow_id_prefix: vec![format!("mvf-a-{tag}"), format!("mvf-c-{tag}")],
            ..Default::default()
        })
        .await?;
    let mut got: Vec<&str> = by_prefix.iter().map(|w| w.id.as_str()).collect();
    got.sort();
    assert_eq!(
        got,
        vec![a_id.as_str(), c_id.as_str()],
        "LIKE ANY on prefix"
    );

    // OR across authenticated users.
    let by_user = engine
        .list_workflows(&ListFilter {
            authenticated_users: vec!["alice".to_string(), "carol".to_string()],
            workflow_ids: vec![a_id.clone(), b_id.clone(), c_id.clone()],
            ..Default::default()
        })
        .await?;
    assert_eq!(by_user.len(), 2, "= ANY on authenticated_user");

    // Fork a-id: the ORIGINAL a-id becomes was_forked_from=true (a fork was
    // taken from it); the new fork and the untouched b/c are not.
    let fork_id = format!("mvf-fork-{tag}");
    engine
        .fork_workflow::<i64>(&a_id, 0, WorkflowOptions::with_id(&fork_id))
        .await?;
    let sources = engine
        .list_workflows(&ListFilter {
            was_forked_from: Some(true),
            workflow_ids: vec![a_id.clone(), b_id.clone(), c_id.clone(), fork_id.clone()],
            ..Default::default()
        })
        .await?;
    assert_eq!(
        sources.len(),
        1,
        "only the fork's source is was_forked_from"
    );
    assert_eq!(sources[0].id, a_id);
    // The fork itself and the untouched b/c are not sources.
    let not_sources = engine
        .list_workflows(&ListFilter {
            was_forked_from: Some(false),
            workflow_ids: vec![b_id.clone(), c_id.clone(), fork_id.clone()],
            ..Default::default()
        })
        .await?;
    assert_eq!(
        not_sources.len(),
        3,
        "non-source rows are was_forked_from=false"
    );

    engine
        .delete_workflows(&[a_id, b_id, c_id, fork_id], false)
        .await?;
    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// `was_forked_from` survives an export/import round trip on Postgres with its
/// correct semantics (marks the fork *source*, not the fork). The exported payload
/// carries the flag (Python-style), so a source re-imported alone keeps it.
#[tokio::test]
async fn pg_was_forked_from_survives_import() -> Result<()> {
    let Some(url) = database_url() else {
        eprintln!("skipping pg_was_forked_from_survives_import: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4().to_string();
    let src = format!("wff-src-{tag}");
    let fork = format!("wff-fork-{tag}");
    let mut engine = DurableEngine::new(Arc::new(PostgresProvider::connect(&url).await?)).await?;
    engine.register("wff", |ctx: DurableContext, n: i64| async move {
        ctx.step("s", || async { Ok::<_, Error>(n) }).await
    });
    engine.launch().await?;

    engine
        .start::<_, i64>("wff", 1i64, WorkflowOptions::with_id(&src))
        .await?
        .result()
        .await?;
    // fork_workflow stamps the source and creates the fork row synchronously.
    engine
        .fork_workflow::<i64>(&src, 0, WorkflowOptions::with_id(&fork))
        .await?;

    // Export each alone, delete both.
    let src_export = engine.export_workflow(&src, false).await?;
    let fork_export = engine.export_workflow(&fork, false).await?;
    engine
        .delete_workflows(&[src.clone(), fork.clone()], false)
        .await?;

    // Import the source ALONE: carry-over restores was_forked_from=true from the
    // payload even with no fork in the batch to reconstruct from.
    engine.import_workflow(&src_export).await?;
    let sources = engine
        .list_workflows(&ListFilter {
            was_forked_from: Some(true),
            workflow_ids: vec![src.clone()],
            ..Default::default()
        })
        .await?;
    assert_eq!(
        sources.len(),
        1,
        "the source's was_forked_from carried over"
    );
    assert_eq!(sources[0].id, src);

    // The fork imports as not-a-source.
    engine.import_workflow(&fork_export).await?;
    let not_sources = engine
        .list_workflows(&ListFilter {
            was_forked_from: Some(false),
            workflow_ids: vec![fork.clone()],
            ..Default::default()
        })
        .await?;
    assert_eq!(not_sources.len(), 1, "the fork is not a source");
    assert_eq!(not_sources[0].id, fork);

    engine.delete_workflows(&[src, fork], false).await?;
    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// A fork routes to the queue named in the fork options and completes there,
/// inheriting the original's application version. A `Client` fork (which has
/// no engine version at hand) also inherits the original's — rather than
/// stamping '' — so the fleet that could run the original can run the fork.
#[tokio::test]
async fn pg_fork_routes_to_named_queue() -> Result<()> {
    use durare::{Client, WorkflowHandle, WorkflowQueue};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_fork_routes_to_named_queue: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let wf = format!("fork-wf-{tag}");
    let queue = format!("fork-q-{tag}");

    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new(provider.clone()).await?;
    engine.register(&wf, |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n * 2)
    });
    engine.register_queue(
        WorkflowQueue::new(&queue).base_polling_interval(Duration::from_millis(50)),
    );
    engine.launch().await?;

    let orig_id = format!("fork-orig-{tag}");
    let orig: i64 = engine
        .start(&wf, 21_i64, WorkflowOptions::with_id(&orig_id))
        .await?
        .result()
        .await?;
    assert_eq!(orig, 42);

    // Engine fork onto the named queue: its dispatcher completes it, and the
    // row records the queue and the inherited version.
    let fork_id = format!("fork-new-{tag}");
    let mut opts = WorkflowOptions::with_id(&fork_id);
    opts.queue = Some(queue.clone());
    let forked = engine.fork_workflow::<i64>(&orig_id, 0, opts).await?;
    assert_eq!(forked.result().await?, 42);
    let row = provider.get_workflow_status(&fork_id).await?.unwrap();
    assert_eq!(row.queue_name.as_deref(), Some(queue.as_str()));
    assert_eq!(row.app_version, engine.app_version());

    // Client fork (internal queue): the inherited version matches this engine,
    // so its dispatcher claims and completes the fork.
    let client_fork_id = format!("fork-client-{tag}");
    let client = Client::new(provider.clone());
    let cf: WorkflowHandle<i64> = client
        .fork_workflow(&orig_id, 0, WorkflowOptions::with_id(&client_fork_id))
        .await?;
    assert_eq!(cf.result().await?, 42);
    let crow = provider
        .get_workflow_status(&client_fork_id)
        .await?
        .unwrap();
    assert_eq!(
        crow.app_version,
        engine.app_version(),
        "a client fork inherits the original's version instead of stamping ''"
    );

    engine.shutdown(Duration::from_secs(2)).await?;
    Ok(())
}

/// A replay that finds a different step recorded at the current position fails
/// the workflow with an `UnexpectedStep` (non-deterministic workflow) error
/// instead of silently returning the wrong step's checkpoint.
#[tokio::test]
async fn pg_renamed_step_fails_as_unexpected_step() -> Result<()> {
    use durare::{StateProvider, WorkflowStatus, STATUS_PENDING};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_renamed_step_fails_as_unexpected_step: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let wf = format!("evolved-{tag}");
    let id = format!("wf-evolved-{tag}");

    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new(provider.clone()).await?;
    engine.register(&wf, |ctx: DurableContext, _: ()| async move {
        ctx.step("renamed", || async { Ok::<_, Error>(1_i64) })
            .await
    });
    engine.launch().await?;

    // Seed a PENDING run whose step 0 was checkpointed under the OLD name.
    provider
        .insert_workflow_status(WorkflowStatus::new(
            &id,
            &wf,
            serde_json::Value::Null,
            STATUS_PENDING,
            "",
            engine.app_version(),
        ))
        .await?;
    provider
        .record_step_result(&id, 0, "first", serde_json::json!(1), None, None)
        .await?;

    let h = engine.resume_workflow::<i64>(&id).await?;
    let err = h.result().await.expect_err("replay must fail");
    let msg = err.to_string();
    assert!(
        msg.contains("non-deterministic") && msg.contains("renamed") && msg.contains("first"),
        "expected an UnexpectedStep failure, got: {msg}"
    );
    engine.shutdown(Duration::from_secs(2)).await?;
    Ok(())
}

/// Resume onto a named queue over live Postgres: the resumed run is claimed by
/// that queue's dispatcher and the row records the queue.
#[tokio::test]
async fn pg_resume_routes_to_named_queue() -> Result<()> {
    use durare::{StateProvider, WorkflowQueue, WorkflowStatus, STATUS_PENDING};
    let Some(url) = database_url() else {
        eprintln!("skipping pg_resume_routes_to_named_queue: DATABASE_URL unset");
        return Ok(());
    };
    let tag = uuid::Uuid::new_v4();
    let wf = format!("bounce-{tag}");
    let queue = format!("resume-q-{tag}");
    let id = format!("wf-bounce-{tag}");

    let provider = Arc::new(PostgresProvider::connect(&url).await?);
    let mut engine = DurableEngine::new(provider.clone()).await?;
    engine.register(&wf, |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n + 1)
    });
    engine.register_queue(
        WorkflowQueue::new(&queue).base_polling_interval(Duration::from_millis(50)),
    );
    engine.launch().await?;

    provider
        .insert_workflow_status(WorkflowStatus::new(
            &id,
            &wf,
            serde_json::json!(41),
            STATUS_PENDING,
            "",
            engine.app_version(),
        ))
        .await?;
    engine.cancel_workflow(&id).await?;

    let h = engine.resume_workflow_on::<i64>(&id, &queue).await?;
    assert_eq!(h.result().await?, 42);
    let row = provider.get_workflow_status(&id).await?.unwrap();
    assert_eq!(row.queue_name.as_deref(), Some(queue.as_str()));
    engine.shutdown(Duration::from_secs(2)).await?;
    Ok(())
}