dbos 0.6.0

//! The client, against real databases.
//!
//! Two shapes of test here, and the split is the point of the surface. Most of these use a client
//! **alone**, with no application anywhere: what they assert is the row, because a row is all a
//! client can produce — it names a workflow nothing in this process has ever heard of. The rest
//! stand a real [`DBOS`] instance up beside the client, and assert the thing that makes the client
//! worth having: work handed over by one process and run by another.

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;

use dbos::sysdb::postgres::{PostgresSystemDatabase, Settings};
use dbos::sysdb::types::WorkflowFilter;
use dbos::sysdb::types::WorkflowStatus;
use dbos::sysdb::{Error as SysdbError, SystemDatabase};
use dbos::{
    Change, Children, Client, ClientConfig, Config, DBOS, DuplicationPolicy, EngineOnly,
    EnqueueOptions, Error, Fork, ForkOptions, PortableWorkflowArgs, QualifiedWorkflowName,
    QueueChange, QueueConflict, QueueOptions, QueueRequest, SendMessage, SendOptions,
    SerializationFormat, StartOptions, Timeout, WorkflowHandle, WorkflowIdReusePolicy,
};
use dbos_test_support::{TestDatabase, raw_database, skewed_database, test_database};

/// A client for `db`, acting for `app_name`.
async fn client(app_name: &str, db: &TestDatabase) -> Client {
    Client::connect(ClientConfig {
        app_name: Some(app_name.to_owned()),
        ..ClientConfig::new(db.url())
    })
    .await
    .expect("connect failed")
}

/// A client for `db` that speaks for no application.
async fn nameless(db: &TestDatabase) -> Client {
    Client::connect(ClientConfig::new(db.url()))
        .await
        .expect("connect failed")
}

/// The version an instance in this file launches with unless it says otherwise: DBOS computes
/// none, so a launch without one fails.
const APP_VERSION: &str = "1.0.0";

/// An application whose database is already migrated, so launching only connects.
fn config(app_name: &str, db: &TestDatabase) -> Config {
    Config {
        migrate: false,
        app_version: Some(APP_VERSION.to_owned()),
        ..Config::new(app_name, db.url())
    }
}

/// A second handle on the database, for reading what a call actually wrote.
async fn reader(db: &TestDatabase) -> PostgresSystemDatabase {
    PostgresSystemDatabase::from_pool(db.pool().await, &Settings::default())
}

/// A client enqueue whose timeout or delay the row cannot hold is refused before it writes one.
#[tokio::test]
async fn a_client_enqueue_with_an_unstorable_duration_is_refused() {
    let db = test_database().await;
    let client = client("client-unstorable", &db).await;

    let cases = [
        (
            "refused-timeout",
            EnqueueOptions {
                workflow_id: Some("refused-timeout"),
                timeout: Timeout::Explicit(Duration::MAX),
                ..EnqueueOptions::new("work")
            },
            "`timeout`",
        ),
        (
            "refused-delay",
            EnqueueOptions {
                workflow_id: Some("refused-delay"),
                delay: Some(Duration::MAX),
                ..EnqueueOptions::new("work")
            },
            "`delay`",
        ),
        (
            "zero-timeout",
            EnqueueOptions {
                workflow_id: Some("zero-timeout"),
                timeout: Timeout::Explicit(Duration::ZERO),
                ..EnqueueOptions::new("work")
            },
            "`timeout` must be a positive, non-zero duration",
        ),
        (
            "zero-delay",
            EnqueueOptions {
                workflow_id: Some("zero-delay"),
                delay: Some(Duration::ZERO),
                ..EnqueueOptions::new("work")
            },
            "`delay` must be a positive, non-zero duration",
        ),
    ];
    for (id, options, expected) in cases {
        let refused = client
            .enqueue_with::<_, (), EngineOnly>("never_registered", (), options)
            .await;
        assert!(
            matches!(&refused, Err(Error::InvalidArgument { operation, detail })
                if operation == "enqueue a workflow" && detail.contains(expected)),
            "{id}: expected a refusal mentioning {expected}, got {:?}",
            refused.map(|_| ())
        );
        assert!(
            reader(&db)
                .await
                .get_workflow(id)
                .await
                .expect("read failed")
                .is_none(),
            "{id}: a refused enqueue wrote a row"
        );
    }
}

/// The client hands work over and the application runs it — the whole point of the surface.
///
/// Nothing about the enqueue mentions a function: the client names the workflow by string, and the
/// executor that dequeues it is the only process that has the code. The handle is a polling one,
/// and it resolves when that other process finishes.
#[tokio::test]
async fn an_application_runs_what_a_client_enqueues() {
    let db = test_database().await;
    let dbos = DBOS::new(config("client-handover", &db));
    dbos.register_workflow("double", |n: u32| async move { Ok::<u32, Error>(n * 2) })
        .unwrap();
    dbos.launch().await.expect("launch failed");
    dbos.register_queue(
        "work",
        QueueOptions::default(),
        QueueConflict::UpdateIfLatestVersion,
    )
    .await
    .expect("registration failed");

    let client = client("client-handover", &db).await;
    let handle: WorkflowHandle<u32> = client
        .enqueue("double", "work", 21u32)
        .await
        .expect("enqueue failed");
    assert_eq!(handle.result().await.expect("the workflow failed"), 42);

    client.close().await;
    dbos.shutdown().await;
}

/// A client enqueues onto a queue nothing is draining, and the row is still exactly right.
///
/// This is the case a client is *usually* in: the application it is handing work to is somewhere
/// else, or not running yet. Nothing here can run `never_registered`, and the enqueue is no less
/// successful for it.
#[tokio::test]
async fn an_enqueue_writes_a_row_nothing_here_could_run() {
    let db = test_database().await;
    let client = client("client-rows", &db).await;

    let handle: WorkflowHandle<()> = client
        .enqueue_with(
            "never_registered",
            "payload",
            EnqueueOptions {
                workflow_id: Some("row-under-test"),
                ..EnqueueOptions::new("work")
            },
        )
        .await
        .expect("enqueue failed");
    assert_eq!(handle.workflow_id(), "row-under-test");

    let row = reader(&db)
        .await
        .get_workflow("row-under-test")
        .await
        .expect("read failed")
        .expect("the row is missing");
    assert_eq!(row.status, WorkflowStatus::Enqueued);
    assert_eq!(row.name.as_deref(), Some("never_registered"));
    assert_eq!(row.queue_name.as_deref(), Some("work"));
    assert_eq!(row.input.as_deref(), Some("\"payload\""));
    assert_eq!(
        row.app_name.as_deref(),
        Some("client-rows"),
        "a named client owns what it writes"
    );
    assert_eq!(
        row.executor_id, None,
        "a client claims nothing: the executor that dequeues it stamps its own id"
    );
    assert_eq!(
        row.app_version, None,
        "a client has no version of its own to pin the row to"
    );
    assert_eq!(row.priority, 0, "unprioritised is stored as the sentinel");

    client.close().await;
}

/// Everything an enqueue may say reaches the row it writes.
///
/// One workflow rather than one per option, because what is being checked is the mapping — every
/// part of the `QualifiedWorkflowName` and every field of `EnqueueOptions` and `QueueRequest`
/// landing in the column it names.
#[tokio::test]
async fn every_option_reaches_the_row() {
    let db = test_database().await;
    let client = client("client-options", &db).await;
    let mut attributes = serde_json::Map::new();
    attributes.insert("tenant".to_owned(), serde_json::json!("acme"));
    let billing = QualifiedWorkflowName::instance("charge", "Billing", "eu");

    let handle: WorkflowHandle<()> = client
        .enqueue_with(
            &billing,
            (),
            EnqueueOptions {
                workflow_id: Some("fully-specified"),
                workflow_id_reuse_policy: WorkflowIdReusePolicy::Reject,
                app_name: Some("some-other-app"),
                app_version: Some("v1.2.3"),
                timeout: Timeout::Explicit(Duration::from_secs(90)),
                attributes: Some(&attributes),
                // Every field named, which is the point of this test -- so there is nothing left
                // for a functional update to fill in.
                queue_name: "billing",
                deduplication_id: None,
                duplication_policy: DuplicationPolicy::Reject,
                priority: Some(5),
                partition_key: Some("acme"),
                delay: None,
                serialization: SerializationFormat::Default,
            },
        )
        .await
        .expect("enqueue failed");
    assert_eq!(handle.workflow_id(), "fully-specified");

    let row = reader(&db)
        .await
        .get_workflow("fully-specified")
        .await
        .expect("read failed")
        .expect("the row is missing");
    assert_eq!(row.name.as_deref(), Some("charge"));
    assert_eq!(row.class_name.as_deref(), Some("Billing"));
    assert_eq!(row.config_name.as_deref(), Some("eu"));
    assert_eq!(row.qualified_name(), Some(billing));
    assert_eq!(
        row.app_name.as_deref(),
        Some("some-other-app"),
        "an enqueue may hand work to an application this client is not"
    );
    assert_eq!(row.app_version.as_deref(), Some("v1.2.3"));
    assert_eq!(row.timeout, Some(Duration::from_secs(90)));
    assert_eq!(
        row.deadline, None,
        "a queued workflow's deadline is computed on dequeue: the wait is not part of the budget"
    );
    assert_eq!(row.priority, 5);
    assert_eq!(row.queue_partition_key.as_deref(), Some("acme"));
    assert!(
        row.attributes
            .as_deref()
            .is_some_and(|json| json.contains("acme")),
        "{:?}",
        row.attributes
    );

    client.close().await;
}

/// A nameless client writes rows no application owns.
///
/// Unclaimed is not a failure state: every application matches such a row, and the first executor
/// to dequeue it claims it. It is what a cross-application tool wants, and what a client sharing a
/// database with several applications must not do by accident — which is why the bare constructor
/// is the nameless one and a name has to be asked for.
#[tokio::test]
async fn a_nameless_client_writes_unclaimed_rows() {
    let db = test_database().await;
    let client = nameless(&db).await;
    assert_eq!(client.app_name(), None);

    let handle: WorkflowHandle<()> = client
        .enqueue("anything", "work", ())
        .await
        .expect("enqueue failed");

    let row = reader(&db)
        .await
        .get_workflow(handle.workflow_id())
        .await
        .expect("read failed")
        .expect("the row is missing");
    assert_eq!(row.app_name, None);

    client.close().await;
}

/// A delay leaves the workflow `DELAYED`, not `ENQUEUED`.
///
/// The supervisor moves it across when the delay expires, so this is the one status a fresh
/// enqueue can have that is not `ENQUEUED`.
#[tokio::test]
async fn a_delayed_enqueue_is_held_back() {
    let db = test_database().await;
    let client = client("client-delay", &db).await;

    let handle: WorkflowHandle<()> = client
        .enqueue_with(
            "later",
            (),
            EnqueueOptions {
                delay: Some(Duration::from_secs(3600)),
                ..EnqueueOptions::new("work")
            },
        )
        .await
        .expect("enqueue failed");

    let row = reader(&db)
        .await
        .get_workflow(handle.workflow_id())
        .await
        .expect("read failed")
        .expect("the row is missing");
    assert_eq!(row.status, WorkflowStatus::Delayed);
    assert!(row.delay_until.is_some(), "the database stamped the moment");

    client.close().await;
}

/// The same id twice is one workflow, which is what makes a retried request safe.
#[tokio::test]
async fn the_workflow_id_is_an_idempotency_key() {
    let db = test_database().await;
    let client = client("client-idempotent", &db).await;
    let options = || EnqueueOptions {
        workflow_id: Some("request-7"),
        ..EnqueueOptions::new("work")
    };

    let first: WorkflowHandle<()> = client
        .enqueue_with("handle_request", (), options())
        .await
        .expect("enqueue failed");
    let second: WorkflowHandle<()> = client
        .enqueue_with("handle_request", (), options())
        .await
        .expect("the second enqueue should join the first, not fail");
    assert_eq!(first.workflow_id(), second.workflow_id());

    let rows = reader(&db)
        .await
        .list_workflows(&Default::default(), None)
        .await
        .expect("list failed");
    assert_eq!(rows.len(), 1, "two enqueues, one workflow");

    client.close().await;
}

/// A deduplication key is refused by default and joined when the caller asks for it.
///
/// The two halves of the same collision: `Reject` reports it, `ReturnExisting` hands back a handle
/// to the workflow already holding the key. The second is idempotent enqueue — the first caller's
/// workflow is the one that runs, and everyone else waits on it.
#[tokio::test]
async fn a_held_deduplication_key_is_refused_or_joined() {
    let db = test_database().await;
    let client = client("client-dedup", &db).await;
    let options = |duplication_policy| EnqueueOptions {
        deduplication_id: Some("order-42"),
        duplication_policy,
        ..EnqueueOptions::new("work")
    };

    let first: WorkflowHandle<()> = client
        .enqueue_with("process_order", (), options(DuplicationPolicy::Reject))
        .await
        .expect("enqueue failed");

    let error = client
        .enqueue_with::<_, (), Error>("process_order", (), options(DuplicationPolicy::Reject))
        .await
        .expect_err("the key is held, and rejecting is the default");
    assert!(
        matches!(
            error,
            Error::SystemDatabase(SysdbError::QueueDeduplicated { .. })
        ),
        "{error}"
    );

    let joined: WorkflowHandle<()> = client
        .enqueue_with(
            "process_order",
            (),
            options(DuplicationPolicy::ReturnExisting),
        )
        .await
        .expect("returning the existing workflow should not fail");
    assert_eq!(
        joined.workflow_id(),
        first.workflow_id(),
        "the handle is to the workflow already holding the key"
    );

    client.close().await;
}

/// Asking to join a holder with no key to be held is refused rather than ignored.
#[tokio::test]
async fn returning_the_existing_workflow_needs_a_key() {
    let db = test_database().await;
    let client = client("client-dedup-guard", &db).await;

    let error = client
        .enqueue_with::<_, (), Error>(
            "process_order",
            (),
            EnqueueOptions {
                duplication_policy: DuplicationPolicy::ReturnExisting,
                ..EnqueueOptions::new("work")
            },
        )
        .await
        .expect_err("there is no collision to resolve without a deduplication id");
    assert!(
        matches!(&error, Error::InvalidArgument { operation, .. } if operation == "enqueue a workflow"),
        "{error}"
    );
    assert!(error.to_string().contains("deduplication_id"), "{error}");

    client.close().await;
}

/// A value only the insert checks is refused as the enqueue's argument, named as the API spells
/// it rather than by its column.
#[tokio::test]
async fn a_client_enqueue_with_an_empty_value_is_refused_by_its_api_name() {
    let db = test_database().await;
    let client = client("client-empty-values", &db).await;

    let cases = [
        (
            EnqueueOptions {
                deduplication_id: Some(""),
                ..EnqueueOptions::new("work")
            },
            "queue `work`: `deduplication_id` must be absent rather than empty",
        ),
        (
            EnqueueOptions {
                partition_key: Some(""),
                ..EnqueueOptions::new("work")
            },
            "queue `work`: `partition_key` must be absent rather than empty",
        ),
        (
            EnqueueOptions {
                app_version: Some(""),
                ..EnqueueOptions::new("work")
            },
            "queue `work`: `app_version` must be absent rather than empty",
        ),
        (EnqueueOptions::new(""), "`queue_name` must not be empty"),
    ];
    for (options, expected) in cases {
        let refused = client
            .enqueue_with::<_, (), EngineOnly>("never_registered", (), options)
            .await;
        assert!(
            matches!(&refused, Err(Error::InvalidArgument { operation, detail })
                if operation == "enqueue a workflow" && detail == expected),
            "expected {expected:?}, got {:?}",
            refused.map(|_| ())
        );
    }

    // A workflow with no name, which nothing could dequeue, from either enqueue.
    let expected = "queue `work`: `workflow` must not be empty";
    let refused = client
        .enqueue_with::<_, (), EngineOnly>("", (), EnqueueOptions::new("work"))
        .await;
    assert!(
        matches!(&refused, Err(Error::InvalidArgument { detail, .. }) if detail == expected),
        "{:?}",
        refused.map(|_| ())
    );
    let refused = client
        .enqueue_portable::<(), EngineOnly>(
            QualifiedWorkflowName::method("", "Billing"),
            PortableWorkflowArgs::default(),
            EnqueueOptions::new("work"),
        )
        .await;
    assert!(
        matches!(&refused, Err(Error::InvalidArgument { detail, .. }) if detail == expected),
        "{:?}",
        refused.map(|_| ())
    );
    let rows = reader(&db)
        .await
        .list_workflows(&Default::default(), None)
        .await
        .expect("read failed");
    assert!(rows.is_empty(), "nothing was written");

    client.close().await;
}

/// A key's empty class or config is absent, and is written as `NULL`.
#[tokio::test]
async fn an_enqueued_key_writes_an_empty_class_or_config_as_null() {
    let db = test_database().await;
    let client = client("client-keys", &db).await;

    let handle: WorkflowHandle<()> = client
        .enqueue_with(
            QualifiedWorkflowName::instance("never_registered", "", ""),
            (),
            EnqueueOptions {
                workflow_id: Some("blank-key"),
                ..EnqueueOptions::new("work")
            },
        )
        .await
        .unwrap();
    let row = reader(&db)
        .await
        .get_workflow(handle.workflow_id())
        .await
        .expect("read failed")
        .expect("the row is missing");
    assert_eq!(row.class_name, None);
    assert_eq!(row.config_name, None);

    client.close().await;
}

/// A client enqueue with `WorkflowIdReusePolicy::Reject` refuses an id another enqueue took, and
/// leaves that workflow as it was; the default joins it.
#[tokio::test]
async fn a_rejecting_client_enqueue_refuses_a_taken_id() {
    let db = test_database().await;
    let client = client("client-reuse-policy", &db).await;
    let id = "client-taken-id";
    let options = |policy| EnqueueOptions {
        workflow_id: Some(id),
        workflow_id_reuse_policy: policy,
        ..EnqueueOptions::new("work")
    };

    client
        .enqueue_with::<_, (), EngineOnly>(
            "never_registered",
            1,
            options(WorkflowIdReusePolicy::Reject),
        )
        .await
        .expect("a fresh id is not in use");
    let before = reader(&db)
        .await
        .get_workflow(id)
        .await
        .expect("read failed")
        .expect("the row is missing");

    let refused = client
        .enqueue_with::<_, (), EngineOnly>(
            "never_registered",
            2,
            options(WorkflowIdReusePolicy::Reject),
        )
        .await;
    assert!(
        matches!(
            &refused,
            Err(Error::SystemDatabase(SysdbError::WorkflowIdInUse {
                workflow_id,
                status: WorkflowStatus::Enqueued,
                name: Some(name),
            })) if workflow_id == id && name == "never_registered"
        ),
        "{:?}",
        refused.map(|_| ())
    );
    let joined = client
        .enqueue_with::<_, (), EngineOnly>(
            "never_registered",
            3,
            options(WorkflowIdReusePolicy::ReturnExisting),
        )
        .await
        .expect("the default joins");
    assert_eq!(joined.workflow_id(), id);

    let after = reader(&db)
        .await
        .get_workflow(id)
        .await
        .expect("read failed")
        .expect("the row is missing");
    assert_eq!(after.input, before.input, "the first enqueue's input");
    assert_eq!(after.updated_at, before.updated_at, "nothing was written");

    // `Reject` is about an id the caller chose, and is refused without one.
    let refused = client
        .enqueue_with::<_, (), EngineOnly>(
            "never_registered",
            4,
            EnqueueOptions {
                workflow_id_reuse_policy: WorkflowIdReusePolicy::Reject,
                ..EnqueueOptions::new("work")
            },
        )
        .await;
    assert!(
        matches!(&refused, Err(Error::InvalidArgument { detail, .. })
            if detail == "queue `work`: `WorkflowIdReusePolicy::Reject` needs a `workflow_id`: \
                          the policy is about an id the caller chose"),
        "{:?}",
        refused.map(|_| ())
    );

    client.close().await;
}

/// An enqueue no queue could honour costs a round trip, not a row.
#[tokio::test]
async fn an_impossible_enqueue_is_refused_before_it_is_written() {
    let db = test_database().await;
    let client = client("client-validate", &db).await;

    let error = client
        .enqueue_with::<_, (), Error>(
            "anything",
            (),
            EnqueueOptions {
                workflow_id: Some("never-written"),
                priority: Some(0),
                ..EnqueueOptions::new("work")
            },
        )
        .await
        .expect_err("a priority of 0 was accepted");
    assert!(
        matches!(&error, Error::InvalidArgument { operation, .. } if operation == "enqueue a workflow"),
        "{error}"
    );

    assert!(
        reader(&db)
            .await
            .get_workflow("never-written")
            .await
            .expect("read failed")
            .is_none(),
        "nothing should have been written"
    );

    client.close().await;
}

/// **A client's deduplication id is held across partitions**, as an in-process enqueue's is: a key
/// held on one partition refuses the same key on another, and `ReturnExisting` joins the holder.
#[tokio::test]
async fn a_clients_deduplication_id_is_held_across_partitions() {
    let db = test_database().await;
    let client = client("client-dedup-partition", &db).await;
    let enqueue = |id, partition, policy| {
        client.enqueue_with::<_, (), Error>(
            "report",
            (),
            EnqueueOptions {
                workflow_id: Some(id),
                deduplication_id: Some("nightly-report"),
                partition_key: Some(partition),
                duplication_policy: policy,
                ..EnqueueOptions::new("tenants")
            },
        )
    };

    enqueue("tenant-a-report", "tenant-a", DuplicationPolicy::Reject)
        .await
        .expect("the first enqueue failed");
    let error = enqueue("tenant-b-report", "tenant-b", DuplicationPolicy::Reject)
        .await
        .expect_err("another partition took the held key");
    assert!(
        format!("{error}").contains("nightly-report"),
        "the refusal must name the key, got {error:?}"
    );
    let joined = enqueue(
        "tenant-c-report",
        "tenant-c",
        DuplicationPolicy::ReturnExisting,
    )
    .await
    .expect("return-existing was refused");
    assert_eq!(joined.workflow_id(), "tenant-a-report");

    client.close().await;
}

/// A message sent from a client waits in the database for its destination to read it.
///
/// A client's send is not a step — there is no workflow to checkpoint it against — so what makes a
/// resend safe is the idempotency key, and that is what the second half asserts: the same key twice
/// is one message.
#[tokio::test]
async fn a_message_waits_for_its_destination() {
    let db = test_database().await;
    let client = client("client-send", &db).await;
    let destination: WorkflowHandle<()> = client
        .enqueue("receiver", "work", ())
        .await
        .expect("enqueue failed");
    let id = destination.workflow_id().to_owned();

    client
        .send_with(
            SendMessage {
                topic: Some("approvals"),
                ..SendMessage::new(&id, &"approved")
            },
            SendOptions::default(),
        )
        .await
        .expect("send failed");
    let once = SendMessage {
        idempotency_key: Some("once"),
        ..SendMessage::new(&id, &"only once")
    };
    client
        .send_with(once, SendOptions::default())
        .await
        .expect("send failed");
    client
        .send_with(once, SendOptions::default())
        .await
        .expect("a resend under a held key is a no-op, not an error");

    let notifications = reader(&db)
        .await
        .get_all_notifications(&id)
        .await
        .expect("read failed");
    assert_eq!(
        notifications.len(),
        2,
        "three sends, two messages: the key deduplicated the resend"
    );
    let approval = notifications
        .iter()
        .find(|notification| notification.topic.as_deref() == Some("approvals"))
        .expect("the topicked message is missing");
    assert_eq!(approval.message, "\"approved\"");
    assert!(!approval.consumed, "nothing has received it");

    client.close().await;
}

/// A batch is one transaction: either every message is delivered or none is.
#[tokio::test]
async fn a_batch_of_messages_lands_together() {
    let db = test_database().await;
    let client = client("client-send-bulk", &db).await;
    let first: WorkflowHandle<()> = client
        .enqueue("receiver", "work", ())
        .await
        .expect("enqueue failed");
    let second: WorkflowHandle<()> = client
        .enqueue("receiver", "work", ())
        .await
        .expect("enqueue failed");
    let (first, second) = (
        first.workflow_id().to_owned(),
        second.workflow_id().to_owned(),
    );

    client
        .send_bulk(&[
            SendMessage::new(&first, &"one"),
            SendMessage::new(&second, &"two"),
            SendMessage::new(&first, &"three"),
        ])
        .await
        .expect("send failed");

    let reader = reader(&db).await;
    assert_eq!(
        reader
            .get_all_notifications(&first)
            .await
            .expect("read failed")
            .len(),
        2
    );
    assert_eq!(
        reader
            .get_all_notifications(&second)
            .await
            .expect("read failed")
            .len(),
        1
    );

    // Nothing is delivered when one destination does not exist: the foreign key fails the whole
    // insert, which is the guarantee a batch buys over a loop of sends.
    let error = client
        .send_bulk(&[
            SendMessage::new(&first, &"four"),
            SendMessage::new("no-such-workflow", &"five"),
        ])
        .await
        .expect_err("a message addressed to nothing should fail");
    assert!(matches!(error, Error::SystemDatabase(_)), "{error}");
    assert_eq!(
        reader
            .get_all_notifications(&first)
            .await
            .expect("read failed")
            .len(),
        2,
        "the failed batch delivered nothing, not a prefix"
    );

    client.close().await;
}

/// A client reads the events a running workflow publishes.
///
/// The progress channel: the application sets a key as it goes, and whatever is watching from
/// outside reads it. A key that is not there when the deadline passes is `None` rather than an
/// error, and a zero timeout makes the read a poll.
#[tokio::test]
async fn a_client_reads_a_workflow_s_events() {
    let db = test_database().await;
    let dbos = DBOS::new(config("client-events", &db));
    let workflow = dbos
        .register_workflow("publish", |()| async move {
            dbos::set_event("progress", &"done").await?;
            Ok::<(), Error>(())
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let client = client("client-events", &db).await;
    let handle = workflow.start(()).await.expect("start failed");
    let id = handle.workflow_id().to_owned();
    handle.result().await.expect("the workflow failed");

    let published: Option<String> = client
        .get_event(&id, "progress", Duration::from_secs(5))
        .await
        .expect("read failed");
    assert_eq!(published.as_deref(), Some("done"));

    let missing: Option<String> = client
        .get_event(&id, "not-published", Duration::ZERO)
        .await
        .expect("read failed");
    assert_eq!(missing, None, "absence is a value, and zero is a poll");

    client.close().await;
    dbos.shutdown().await;
}

/// A client sees a workflow's status, and says so plainly when there is none.
#[tokio::test]
async fn a_client_reads_workflow_status() {
    let db = test_database().await;
    let client = client("client-status", &db).await;
    let handle: WorkflowHandle<()> = client
        .enqueue("waiting", "work", ())
        .await
        .expect("enqueue failed");

    assert_eq!(
        client
            .workflow_status(handle.workflow_id())
            .await
            .expect("read failed"),
        Some(WorkflowStatus::Enqueued)
    );
    assert_eq!(
        client
            .workflow_status("no-such-workflow")
            .await
            .expect("read failed"),
        None,
        "a client asking after an id it was given deserves a plain answer"
    );

    // A handle to a workflow that does not exist is not an error until it is used: nothing is read
    // when it is built.
    let dangling: WorkflowHandle<()> = client.retrieve_workflow("no-such-workflow");
    let error = dangling.status().await.expect_err("there is no such row");
    assert!(matches!(error, Error::WorkflowNotFound { .. }), "{error}");

    // The other half answers differently, and deliberately: a wait treats a missing row as one
    // that has not been enqueued yet and keeps polling for it, which is what a client -- the
    // caller most likely to hold an id before its owner has committed -- needs. Nothing here ever
    // creates the row, so the wait is still running when the timeout takes it.
    let dangling: WorkflowHandle<()> = client.retrieve_workflow("no-such-workflow");
    assert!(
        tokio::time::timeout(std::time::Duration::from_millis(750), dangling.result())
            .await
            .is_err(),
        "awaiting an id with no row should wait for it to appear, not report it"
    );

    client.close().await;
}

/// The queues a fleet drains can be registered, adjusted and removed by a client.
///
/// The reason this belongs on a client at all: the process that decides a queue should slow down is
/// rarely one of the processes draining it.
#[tokio::test]
async fn a_client_manages_queues() {
    let db = test_database().await;
    let client = client("client-queues", &db).await;

    let queue = client
        .register_queue(
            "fleet",
            QueueOptions {
                worker_concurrency: Some(3),
                ..QueueOptions::default()
            },
            QueueConflict::AlwaysUpdate,
        )
        .await
        .expect("registration failed");
    assert_eq!(queue.name(), "fleet");
    assert_eq!(queue.worker_concurrency(), Some(3));

    assert_eq!(
        client
            .queue("fleet")
            .await
            .expect("read failed")
            .expect("the queue is missing")
            .worker_concurrency(),
        Some(3)
    );
    assert_eq!(
        client
            .list_queues()
            .await
            .expect("list failed")
            .iter()
            .map(dbos::Queue::name)
            .collect::<Vec<_>>(),
        vec!["fleet"]
    );

    let updated = client
        .update_queue(
            "fleet",
            QueueChange {
                worker_concurrency: Change::Set(Some(1)),
                ..QueueChange::default()
            },
        )
        .await
        .expect("update failed");
    assert_eq!(updated.worker_concurrency(), Some(1));

    client.delete_queue("fleet").await.expect("delete failed");
    assert!(client.queue("fleet").await.expect("read failed").is_none());

    client.close().await;
}

/// A client that is dropped rather than closed lets go of its connections.
///
/// Dropping is the ordinary end of a client — it is a value in someone's application state, not
/// something with a lifecycle — and it is the case that leaks unless the drop aborts the listener:
/// that task holds its own clone of the pool, so the pool cannot close itself, and the listener's
/// only way out of its loop is that close. An abandoned client would mean a `LISTEN` connection and
/// two tasks for the life of the process. So the listener is turned on here; a client starts
/// none by default.
#[tokio::test]
async fn a_dropped_client_releases_its_connections() {
    // Tagged so this counts its own connections and not another test's, on a shared server.
    const TAG: &str = "dbos-client-drop-probe";

    let db = test_database().await;
    let client = Client::connect(ClientConfig {
        app_name: Some("client-drop".to_owned()),
        use_listen_notify: true,
        ..ClientConfig::new(format!("{}?application_name={TAG}", db.url()))
    })
    .await
    .expect("connect failed");

    // One call, so the pool has actually connected and the listener has had a reason to.
    assert!(
        client
            .queue("nothing")
            .await
            .expect("read failed")
            .is_none()
    );
    assert!(db.connection_count(TAG).await > 0);

    drop(client);

    // The server drops a session shortly after its client goes away, so this is a bounded wait
    // rather than a single look. A leak never converges.
    let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
    loop {
        let open = db.connection_count(TAG).await;
        if open == 0 {
            break;
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "{open} connections from a dropped client are still open"
        );
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
}

/// A client is refused the conflict policy that needs an application version.
///
/// It has none — it runs none of the application's code — so "update if I am the latest version"
/// has no answer. Both surfaces take the same `QueueConflict`, and both refuse this combination
/// rather than guessing at an answer.
#[tokio::test]
async fn a_client_cannot_register_a_queue_by_version() {
    let db = test_database().await;
    let client = client("client-queue-policy", &db).await;

    let error = client
        .register_queue(
            "fleet",
            QueueOptions::default(),
            QueueConflict::UpdateIfLatestVersion,
        )
        .await
        .expect_err("a client has no version to be the latest of");
    assert!(matches!(error, Error::Config(_)), "{error}");
    assert!(error.to_string().contains("AlwaysUpdate"), "{error}");

    client.close().await;
}

/// A client's `AlwaysUpdate` re-registration replaces what is stored.
///
/// The policy an application reaches for first — update only if I am the latest version — is one a
/// client is refused, so `AlwaysUpdate` is what an operator's registration says instead, and this
/// is what it has to mean: the limits named last are the ones in force.
#[tokio::test]
async fn a_clients_registration_replaces_the_stored_limits() {
    let db = test_database().await;
    let client = client("client-queue-default", &db).await;

    let queue = client
        .register_queue(
            "fleet",
            QueueOptions::default(),
            QueueConflict::AlwaysUpdate,
        )
        .await
        .expect("a client's default policy has to be one it can use");
    assert_eq!(queue.name(), "fleet");

    // Registered again with a limit this time: the policy is an update, so the stored row is the
    // second registration's, not the first's.
    let queue = client
        .register_queue(
            "fleet",
            QueueOptions {
                worker_concurrency: Some(2),
                ..QueueOptions::default()
            },
            QueueConflict::AlwaysUpdate,
        )
        .await
        .expect("re-registration failed");
    assert_eq!(queue.worker_concurrency(), Some(2));

    client.delete_queue("fleet").await.expect("delete failed");
    client.close().await;
}

/// A client reads the version registry an application writes at launch.
#[tokio::test]
async fn a_client_reads_application_versions() {
    let db = test_database().await;
    let dbos = DBOS::new(Config {
        app_version: Some("v9".to_owned()),
        ..config("client-versions", &db)
    });
    dbos.launch().await.expect("launch failed");

    let client = client("client-versions", &db).await;
    let versions = client.list_app_versions().await.expect("list failed");
    assert_eq!(
        versions
            .iter()
            .map(|version| version.version_name.as_str())
            .collect::<Vec<_>>(),
        vec!["v9"]
    );
    assert_eq!(
        client
            .latest_app_version()
            .await
            .expect("read failed")
            .expect("a version was registered at launch")
            .version_name,
        "v9"
    );

    client.close().await;
    dbos.shutdown().await;
}

/// **A promotion goes by the database's clock, not the promoting process's.** The database's clock
/// is an hour behind this process. A promotion stamped by this process's clock would land an hour
/// ahead of the database's, so a version deployed after it would register at the database's now,
/// sort below the promoted one, and not become the latest for an hour.
#[tokio::test]
async fn a_client_promotes_on_the_database_clock() {
    let Some(db) = skewed_database(-3600).await else {
        return;
    };
    let deploy = async |version: &str| {
        let dbos = DBOS::new(Config {
            migrate: true,
            app_version: Some(version.to_owned()),
            ..config("client-skew-promote", &db)
        });
        dbos.launch().await.expect("launch failed");
        dbos.shutdown().await;
    };
    deploy("v1").await;
    deploy("v2").await;

    let client = client("client-skew-promote", &db).await;
    let latest = async || {
        client
            .latest_app_version()
            .await
            .expect("read failed")
            .expect("versions were registered at launch")
            .version_name
    };
    client
        .set_latest_app_version("v1")
        .await
        .expect("promote failed");
    assert_eq!(latest().await, "v1", "promoting rolls the fleet back");

    deploy("v3").await;
    assert_eq!(
        latest().await,
        "v3",
        "a version deployed after the promotion sorted below it: the promotion was stamped ahead \
         of the database's clock"
    );

    client.close().await;
}

/// Promoting an older version rolls a deploy back.
///
/// The latest version is the one with the newest *timestamp*, not the one registered last, which
/// is the whole reason promotion is a write an operator can make: moving `v1`'s timestamp forward
/// puts the fleet back on `v1` without redeploying it.
#[tokio::test]
async fn a_client_promotes_an_older_version_to_roll_a_deploy_back() {
    let db = test_database().await;

    // Two deployments in order, so the registry's latest is the second.
    for version in ["v1", "v2"] {
        let dbos = DBOS::new(Config {
            app_version: Some(version.to_owned()),
            ..config("client-promote", &db)
        });
        dbos.launch().await.expect("launch failed");
        dbos.shutdown().await;
    }

    let client = client("client-promote", &db).await;
    let latest = async || {
        client
            .latest_app_version()
            .await
            .expect("read failed")
            .expect("a version was registered at launch")
            .version_name
    };
    assert_eq!(latest().await, "v2", "the newer deployment is the latest");

    client
        .set_latest_app_version("v1")
        .await
        .expect("promote failed");
    assert_eq!(latest().await, "v1", "promoting rolls the fleet back");

    // Promotion moves a timestamp; it registers nothing.
    let mut names = client
        .list_app_versions()
        .await
        .expect("list failed")
        .into_iter()
        .map(|version| version.version_name)
        .collect::<Vec<_>>();
    names.sort();
    assert_eq!(names, ["v1", "v2"]);

    client.close().await;
}

/// Promoting on behalf of a named application — which is the whole reason the setter takes a name.
///
/// One operator tool, pointed at a shared system database, rolls a peer application back without
/// connecting a second client for it. Ownership still holds: the bare form refuses the peer's
/// version, and naming the peer is what makes the same call legal.
#[tokio::test]
async fn a_client_promotes_a_version_for_a_named_application() {
    let db = test_database().await;

    for (app, versions) in [
        ("promote-own", ["v1", "v2"]),
        ("promote-peer", ["p1", "p2"]),
    ] {
        for version in versions {
            let dbos = DBOS::new(Config {
                app_version: Some(version.to_owned()),
                ..config(app, &db)
            });
            dbos.launch().await.expect("launch failed");
            dbos.shutdown().await;
        }
    }

    let own = client("promote-own", &db).await;
    let peer = client("promote-peer", &db).await;
    let latest = async |client: &dbos::Client| {
        client
            .latest_app_version()
            .await
            .expect("read failed")
            .expect("a version was registered at launch")
            .version_name
    };

    // Unnamed, the call is scoped to this client's own application, so a peer's version is not its
    // to move.
    assert!(
        own.set_latest_app_version("p1").await.is_err(),
        "promoting a peer's version without naming the peer should be refused"
    );
    assert_eq!(latest(&peer).await, "p2", "the refusal moved nothing");

    // Naming the peer is what makes it legal.
    own.set_latest_app_version_for("p1", "promote-peer")
        .await
        .expect("promote failed");
    assert_eq!(latest(&peer).await, "p1", "the peer rolled back");
    assert_eq!(
        latest(&own).await,
        "v2",
        "this client's own latest is untouched"
    );

    peer.close().await;
    own.close().await;
}

/// A client never migrates — and so refuses a database no application has created.
///
/// The refusal is the interesting half. Connecting with migrations off *verifies* instead of
/// skipping, so a schema that is missing or behind what this build's queries are written against
/// fails at connect, naming the version it found, rather than surfacing as whatever SQL error the
/// first real call happens to raise.
#[tokio::test]
async fn connecting_to_a_database_no_application_has_created_is_refused() {
    let db = raw_database().await;
    let error = Client::connect(ClientConfig::new(db.url()))
        .await
        .expect_err("there is no DBOS schema to work with");
    assert!(matches!(error, Error::SystemDatabase(_)), "{error}");
    assert!(
        error.to_string().contains("migration 0"),
        "the error should name what it found: {error}"
    );

    let tables: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM information_schema.tables WHERE table_schema = 'dbos'",
    )
    .fetch_one(&db.pool().await)
    .await
    .expect("failed to count tables");
    assert_eq!(tables, 0, "the client should have created nothing");
}

/// Closing is deliberate, and what it closes is the pool every clone shares.
#[tokio::test]
async fn closing_ends_the_connection_for_every_clone() {
    let db = test_database().await;
    let client = client("client-close", &db).await;
    let clone = client.clone();

    client.close().await;
    client.close().await; // idempotent

    let error = clone
        .list_queues()
        .await
        .expect_err("the pool the clone shares is closed");
    assert!(matches!(error, Error::SystemDatabase(_)), "{error}");
}

/// A configuration that cannot work is refused before anything is connected.
#[tokio::test]
async fn a_bad_configuration_is_refused_at_connect() {
    let refused = |config: ClientConfig| async move {
        Client::connect(config)
            .await
            .expect_err("the configuration should have been refused")
    };

    assert!(matches!(
        refused(ClientConfig::new("")).await,
        Error::Config(_)
    ));
    assert!(matches!(
        refused(ClientConfig {
            app_name: Some("No".to_owned()),
            ..ClientConfig::new("postgres://localhost/nothing")
        })
        .await,
        Error::Config(_)
    ));
    assert!(matches!(
        refused(ClientConfig {
            max_connections: 0,
            ..ClientConfig::new("postgres://localhost/nothing")
        })
        .await,
        Error::Config(_)
    ));
}

/// **The operator's case: a client stops work it did not start, and starts it again.**
///
/// The application enqueues onto a queue nothing polls, so the row sits still. A client — a
/// separate process, holding nothing but a connection — cancels it, and the application never runs
/// it. Resuming moves it onto the internal queue, which the application does poll, and it runs.
#[tokio::test]
async fn a_client_cancels_and_resumes_a_workflow_it_did_not_start() {
    let db = test_database().await;
    let dbos = DBOS::new(config("client-cancel", &db));
    let ran = Arc::new(AtomicU32::new(0));
    let workflow = dbos
        .register_workflow("interruptible", {
            let ran = Arc::clone(&ran);
            move |()| {
                let ran = Arc::clone(&ran);
                async move {
                    ran.fetch_add(1, Ordering::SeqCst);
                    Ok::<u32, Error>(7)
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "stopped-from-outside";
    workflow
        .start_with(
            (),
            StartOptions {
                workflow_id: Some(id),
                queue: Some(QueueRequest::new("nothing-polls-this")),
                ..StartOptions::default()
            },
        )
        .await
        .expect("enqueue failed");

    let client = client("client-cancel", &db).await;
    let cancelled = client
        .cancel_all(&[id], Children::Skip)
        .await
        .expect("cancel failed");
    assert_eq!(cancelled, [id], "the cancel did not move the row it named");
    assert_eq!(
        reader(&db)
            .await
            .get_workflow(id)
            .await
            .expect("read failed")
            .expect("the row is missing")
            .status,
        WorkflowStatus::Cancelled,
    );
    assert_eq!(ran.load(Ordering::SeqCst), 0, "a cancelled workflow ran");

    // And back again, from the same process that stopped it.
    let handle = client
        .resume::<u32, EngineOnly>(id)
        .await
        .expect("resume failed");
    assert_eq!(handle.workflow_id(), id, "resuming changed the id");
    assert_eq!(handle.result().await.expect("the workflow failed"), 7);
    assert_eq!(ran.load(Ordering::SeqCst), 1);

    client.close().await;
    dbos.shutdown().await;
}

/// A client forks a workflow, and the application dequeues the fork and runs it.
///
/// The job an operator reaches for a client to do: re-run finished work without the application's
/// code and without touching the original.
#[tokio::test]
async fn a_client_forks_a_workflow_the_application_then_runs() {
    let db = test_database().await;
    let dbos = DBOS::new(config("client-fork", &db));
    let ran = Arc::new(AtomicU32::new(0));
    let workflow = dbos
        .register_workflow("repeatable", {
            let ran = Arc::clone(&ran);
            move |()| {
                let ran = Arc::clone(&ran);
                async move {
                    ran.fetch_add(1, Ordering::SeqCst);
                    Ok::<u32, Error>(9)
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let source = "already-run";
    workflow
        .run_with(
            (),
            dbos::RunOptions {
                workflow_id: Some(source),
                ..dbos::RunOptions::default()
            },
        )
        .await
        .expect("the workflow failed");
    assert_eq!(ran.load(Ordering::SeqCst), 1);

    let client = client("client-fork", &db).await;
    let fork = client
        .fork_with::<u32, EngineOnly>(
            Fork::new(source, 0),
            ForkOptions {
                app_version: Some(APP_VERSION),
                ..ForkOptions::default()
            },
        )
        .await
        .expect("fork failed");
    assert_ne!(
        fork.workflow_id(),
        source,
        "the fork reused the source's id"
    );
    assert_eq!(fork.result().await.expect("the fork failed"), 9);
    assert_eq!(ran.load(Ordering::SeqCst), 2, "the fork did not run");

    // The bulk form names its forks per source.
    let named = client
        .fork_all::<u32, EngineOnly>(
            &[Fork {
                forked_id: Some("client-named-fork"),
                ..Fork::new(source, 0)
            }],
            ForkOptions {
                app_version: Some(APP_VERSION),
                ..ForkOptions::default()
            },
        )
        .await
        .expect("bulk fork failed");
    assert_eq!(
        named.iter().map(|h| h.workflow_id()).collect::<Vec<_>>(),
        ["client-named-fork"],
    );
    for fork in named {
        assert_eq!(fork.result().await.expect("the named fork failed"), 9);
    }
    assert_eq!(ran.load(Ordering::SeqCst), 3, "the named fork did not run");

    client.close().await;
    dbos.shutdown().await;
}

/// A negative step is refused before any I/O, on the single and the bulk form — the same refusal
/// `DBOS::fork_with` and `DBOS::fork_all` make, checked here because the two surfaces enforce it
/// independently.
#[tokio::test]
async fn a_clients_fork_refuses_a_negative_step() {
    let db = test_database().await;
    let client = client("client-fork-negative", &db).await;

    let error = client
        .fork::<u32, EngineOnly>("has-a-step", -1)
        .await
        .expect_err("a negative step was accepted");
    assert!(
        matches!(&error, Error::InvalidArgument { detail, .. } if detail.contains("start_step")),
        "{error:?}"
    );
    let error = client
        .fork_all::<u32, EngineOnly>(&[Fork::new("has-a-step", -1)], ForkOptions::default())
        .await
        .expect_err("a negative step was accepted in bulk");
    assert!(
        matches!(&error, Error::InvalidArgument { detail, .. } if detail.contains("start_step")),
        "{error:?}"
    );

    client.close().await;
}

/// A batch with two forks under one id, an empty forked id or an empty original id is refused
/// before any I/O — neither source exists, so reaching the database would fail differently.
#[tokio::test]
async fn a_clients_bulk_fork_refuses_invalid_forks() {
    let db = test_database().await;
    let client = client("client-fork-invalid", &db).await;

    let twice = Fork {
        forked_id: Some("chosen-twice"),
        ..Fork::new("a", 0)
    };
    let empty_forked_id = Fork {
        forked_id: Some(""),
        ..Fork::new("b", 0)
    };
    for forks in [
        [
            twice,
            Fork {
                original_id: "b",
                ..twice
            },
        ],
        [Fork::new("a", 0), empty_forked_id],
        [Fork::new("a", 0), Fork::new("", 0)],
    ] {
        let error = client
            .fork_all::<u32, EngineOnly>(&forks, ForkOptions::default())
            .await
            .expect_err("an invalid batch was accepted");
        assert!(
            matches!(error, Error::InvalidArgument { .. }),
            "{forks:?}: {error:?}"
        );
    }

    client.close().await;
}

/// A client deletes a workflow, and the row and its steps go with it.
#[tokio::test]
async fn a_client_deletes_a_workflow_it_did_not_start() {
    let db = test_database().await;
    let client = client("client-delete", &db).await;

    let handle: WorkflowHandle<()> = client
        .enqueue_with(
            "never_registered",
            (),
            EnqueueOptions {
                workflow_id: Some("to-be-deleted"),
                ..EnqueueOptions::new("work")
            },
        )
        .await
        .expect("enqueue failed");
    assert_eq!(handle.workflow_id(), "to-be-deleted");

    let deleted = client
        .delete_all(&["to-be-deleted"], Children::Skip)
        .await
        .expect("delete failed");
    assert_eq!(deleted, 1, "the delete removed the wrong number of rows");
    assert!(
        reader(&db)
            .await
            .get_workflow("to-be-deleted")
            .await
            .expect("read failed")
            .is_none(),
        "the row survived its delete"
    );

    client.close().await;
}

/// Listing, tagging and delaying: the reads and small writes an operator's tool makes.
///
/// One test because they share a row, and because what each asserts is narrow. The attributes are
/// a **replacement** rather than a merge, and `None` clears them, which is the half worth checking.
#[tokio::test]
async fn a_client_lists_tags_and_delays_workflows() {
    let db = test_database().await;
    let client = client("client-reads", &db).await;

    let id = "under-inspection";
    let _: WorkflowHandle<()> = client
        .enqueue_with(
            "never_registered",
            (),
            EnqueueOptions {
                workflow_id: Some(id),
                ..EnqueueOptions::new("work")
            },
        )
        .await
        .expect("enqueue failed");

    let listed = client
        .list_workflows(&WorkflowFilter {
            workflow_ids: vec![id],
            ..WorkflowFilter::default()
        })
        .await
        .expect("listing failed");
    assert_eq!(
        listed
            .iter()
            .map(|w| w.workflow_id.as_str())
            .collect::<Vec<_>>(),
        [id],
    );
    assert!(
        client
            .list_workflow_steps(id)
            .await
            .expect("step listing failed")
            .is_empty(),
        "a workflow nothing has run has no steps"
    );

    // Containment, not equality: one key out of two matches.
    let tags = serde_json::json!({ "tenant": "acme", "tier": "gold" });
    client
        .update_workflow_attributes(id, tags.as_object())
        .await
        .expect("update failed");
    let found = client
        .list_workflows(&WorkflowFilter {
            attributes: Some(r#"{"tenant":"acme"}"#),
            ..WorkflowFilter::default()
        })
        .await
        .expect("listing failed");
    assert_eq!(
        found
            .iter()
            .map(|w| w.workflow_id.as_str())
            .collect::<Vec<_>>(),
        [id],
    );

    // A merge keeps what is there.
    let tier = serde_json::json!({ "tier": "silver" });
    client
        .merge_workflow_attributes(id, tier.as_object().unwrap())
        .await
        .expect("merge failed");
    let merged: serde_json::Value = serde_json::from_str(
        &reader(&db)
            .await
            .get_workflow(id)
            .await
            .expect("read failed")
            .expect("the row is missing")
            .attributes
            .expect("the merge left no attributes"),
    )
    .unwrap();
    assert_eq!(
        merged,
        serde_json::json!({ "tenant": "acme", "tier": "silver" })
    );

    // `None` clears, rather than leaving what was there.
    client
        .update_workflow_attributes(id, None)
        .await
        .expect("clear failed");
    assert!(
        reader(&db)
            .await
            .get_workflow(id)
            .await
            .expect("read failed")
            .expect("the row is missing")
            .attributes
            .is_none(),
        "the attributes were not cleared"
    );

    // A delay only moves a DELAYED row, and this one is ENQUEUED -- so the call is accepted and
    // changes nothing, rather than being an error.
    client
        .set_workflow_delay(id, dbos::WorkflowDelay::For(Duration::ZERO))
        .await
        .expect("delay failed");

    // A delay too long to be milliseconds at all is refused, and reported as an argument
    // refusal. One that fits is resolved on the database clock and saturates, so it is accepted.
    let refused = client
        .set_workflow_delay(id, dbos::WorkflowDelay::For(Duration::MAX))
        .await;
    assert!(
        matches!(&refused, Err(Error::InvalidArgument { detail, .. }) if detail.contains("`delay`")),
        "a delay of Duration::MAX should be refused, got {refused:?}"
    );
    client
        .set_workflow_delay(
            id,
            dbos::WorkflowDelay::For(Duration::from_millis(i64::MAX as u64)),
        )
        .await
        .expect("a delay that fits in milliseconds is accepted");

    client.close().await;
}

/// **A client's management call inside a workflow is not a step, where the same call on `DBOS`
/// is.**
///
/// The line this crate already draws for a client's handle and its `get_event`, checked for the
/// operator surface: the step id would have to come from an ambient context belonging to an
/// instance the client is not. So the workflow records nothing, and a replay would make the call
/// again.
#[tokio::test]
async fn a_clients_management_call_inside_a_workflow_is_not_a_step() {
    let db = test_database().await;
    let dbos = DBOS::new(config("client-inside", &db));
    let client = client("client-inside", &db).await;
    let target = dbos
        .register_workflow("target", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();
    let operator = dbos
        .register_workflow("operator", {
            let client = client.clone();
            move |id: String| {
                let client = client.clone();
                async move {
                    client.cancel(&id).await?;
                    Ok::<(), Error>(())
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let target_id = "cancelled-through-a-client";
    target
        .start_with(
            (),
            StartOptions {
                workflow_id: Some(target_id),
                queue: Some(QueueRequest::new("nothing-polls-this")),
                ..StartOptions::default()
            },
        )
        .await
        .expect("enqueue failed");

    let operator_id = "the-client-operator";
    operator
        .run_with(
            target_id.to_owned(),
            dbos::RunOptions {
                workflow_id: Some(operator_id),
                ..Default::default()
            },
        )
        .await
        .expect("the operator workflow failed");

    let reader = reader(&db).await;
    assert_eq!(
        reader
            .get_workflow(target_id)
            .await
            .expect("read failed")
            .expect("the row is missing")
            .status,
        WorkflowStatus::Cancelled,
        "the client's cancel did not reach the row",
    );
    assert!(
        reader
            .list_workflow_steps(operator_id, true, None, None, None)
            .await
            .expect("read failed")
            .is_empty(),
        "a client's call was checkpointed into the calling workflow's step sequence",
    );

    client.close().await;
    dbos.shutdown().await;
}