use assert_matches::assert_matches;
use chrono::{DateTime, Utc};
use concepts::prefixed_ulid::{DEPLOYMENT_ID_DUMMY, DelayId, RunId};
use concepts::storage::DeploymentFileRecord;
use concepts::storage::{
    self, AppendEventsToExecution, AppendRequest, AppendResponseToExecution, BacktraceFilter,
    BacktraceInfo, CancelOutcome, CreateRequest, DbConnection, DbConnectionTest,
    ExecutionListPagination, ExecutionRequest, ExecutionStateFilter, ExpiredDelay, ExpiredLock,
    ExpiredTimer, FrameInfo, FrameSymbol, FunctionNameFilter, JoinSetRequest, JoinSetResponse,
    JoinSetResponseEventOuter, LockedBy, LockedExecution, Pagination, PendingState,
    PendingStateBlockedByJoinSet, PendingStateCancelling, PendingStateLocked, PendingStatePaused,
    PendingStatePendingAt, ResponseCursor, TimeoutOutcome, Unlocked, Version, VersionType,
    WasmBacktrace,
};
use concepts::storage::{
    DbErrorWrite, DbPoolCloseable, DeploymentRecord, DeploymentStatus, EnqueueOutcome,
};
use concepts::storage::{DbErrorWriteNonRetriable, HistoryEvent, ListExecutionsFilter};
use concepts::storage::{HistoryEventScheduleAt, JoinSetResponseEvent, LogFilter};
use concepts::storage::{LogCursor, LogEntry, LogInfoAppendRow};
use concepts::time::ClockFn;
use concepts::{ComponentId, Params, StrVariant, SupportedFunctionReturnValue};
use concepts::{ComponentRetryConfig, JoinSetId, JoinSetKind, SUPPORTED_RETURN_VALUE_OK_EMPTY};
use concepts::{ExecutionId, FunctionFqn, prefixed_ulid::ExecutorId};
use db_postgres::postgres_dao::{DbInitialzationOutcome, PostgresPool, ProvisionPolicy};
use db_sqlite::sqlite_dao::{SqliteConfig, SqlitePool};
use obeli_db_tests::{CANCELLABLE_FFQN, SOME_FFQN};
use obeli_db_tests::{Database, initialize_fresh_postgres_db};
use rstest::rstest;
use std::sync::Arc;
use std::time::Duration;
use tempfile::NamedTempFile;
use test_db_macro::expand_enum_database;
use test_utils::set_up;
use test_utils::sim_clock::SimClock;
use tracing::{debug, info};

#[tokio::test]
async fn initialize_sqlite_twice() {
    let file = NamedTempFile::new().unwrap();
    let path = file.path();

    let pool = SqlitePool::new(path, SqliteConfig::default())
        .await
        .unwrap();
    pool.close().await;
    let pool = SqlitePool::new(path, SqliteConfig::default())
        .await
        .unwrap();
    pool.close().await;
}

#[tokio::test]
async fn initialize_postgres_twice() {
    let pool = initialize_fresh_postgres_db().await;
    pool.close().await;
    let (pool, init_outcome) = PostgresPool::new_with_outcome(pool.config, ProvisionPolicy::Auto)
        .await
        .unwrap();
    assert_eq!(DbInitialzationOutcome::Existing, init_outcome);
    pool.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_expired_lock_should_be_found(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    expired_lock_should_be_found(db_connection.as_ref(), sim_clock).await;
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_append_batch_respond_to_parent(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    append_batch_respond_to_parent(db_connection.as_ref(), sim_clock).await;
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_lock_pending_should_sort_by_scheduled_at(
    database: Database,
    #[values(
        LockingStrategyLifecycle::ByFfqns,
        LockingStrategyLifecycle::ByComponentDigest
    )]
    locking_strategy: LockingStrategyLifecycle,
) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    lock_pending_should_sort_by_scheduled_at(db_connection.as_ref(), sim_clock, locking_strategy)
        .await;
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_lock(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    test_lock_inner(db_connection.as_ref(), sim_clock).await;
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_get_expired_lock(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    get_expired_lock(db_connection.as_ref(), sim_clock).await;
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_get_expired_delay(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    get_expired_delay(db_connection.as_ref(), sim_clock).await;
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn append_after_finish_should_not_be_possible(
    database: Database,
    #[values(
        LockingStrategyLifecycle::ByFfqns,
        LockingStrategyLifecycle::ByComponentDigest
    )]
    locking_strategy: LockingStrategyLifecycle,
) {
    set_up();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    let sim_clock = SimClock::default();

    let execution_id = ExecutionId::generate();
    let exec1 = ExecutorId::generate();
    let lock_expiry = Duration::from_millis(500);

    // Create
    let component_id = ComponentId::dummy_activity();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: component_id.clone(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    let version = lock_pending(
        &execution_id,
        sim_clock.now() + lock_expiry,
        exec1,
        RunId::generate(),
        &component_id,
        db_connection.as_ref(),
        &sim_clock,
        locking_strategy,
    )
    .await;
    let version = {
        let created_at = sim_clock.now();
        debug!(now = %created_at, "Finish execution");
        let req = AppendRequest {
            event: ExecutionRequest::Finished {
                retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                http_client_traces: None,
            },
            created_at,
        };
        db_connection
            .append(execution_id.clone(), version, req)
            .await
            .unwrap()
    };
    {
        let created_at = sim_clock.now();
        debug!(now = %created_at, "Append after finish should fail");
        let req = AppendRequest {
            event: ExecutionRequest::Finished {
                retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                http_client_traces: None,
            },
            created_at,
        };
        let err = db_connection
            .append(execution_id, version, req)
            .await
            .unwrap_err();

        assert_matches!(
            err,
            DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::AlreadyFinished)
        );
    }
    drop(db_connection);
    db_close.close().await;
}

#[expect(clippy::too_many_arguments)]
async fn lock_pending(
    execution_id: &ExecutionId,
    lock_expires_at: DateTime<Utc>,
    executor_id: ExecutorId,
    run_id: RunId,
    component_id: &ComponentId,
    db_connection: &dyn DbConnection,
    sim_clock: &SimClock,
    locking_strategy: LockingStrategyLifecycle,
) -> Version {
    let created_at = sim_clock.now();
    info!(now = %created_at, "LockPending");
    let locked_executions = match locking_strategy {
        LockingStrategyLifecycle::ByFfqns => db_connection
            .lock_pending_by_ffqns(
                1,
                created_at,
                Arc::from([SOME_FFQN]),
                created_at,
                component_id.clone(),
                DEPLOYMENT_ID_DUMMY,
                executor_id,
                lock_expires_at,
                run_id,
                ComponentRetryConfig {
                    retry_exp_backoff: Duration::ZERO,
                    max_retries: Some(0),
                },
            )
            .await
            .unwrap(),

        LockingStrategyLifecycle::ByComponentDigest => db_connection
            .lock_pending_by_component_digest(
                1,
                created_at,
                component_id,
                DEPLOYMENT_ID_DUMMY,
                created_at,
                executor_id,
                lock_expires_at,
                run_id,
                ComponentRetryConfig {
                    retry_exp_backoff: Duration::ZERO,
                    max_retries: Some(0),
                },
            )
            .await
            .unwrap(),
    };
    assert_eq!(1, locked_executions.len());
    let locked_execution = locked_executions.into_iter().next().unwrap();
    assert_eq!(*execution_id, locked_execution.execution_id);
    assert_eq!(Version::new(2), locked_execution.next_version);
    assert_eq!(0, locked_execution.params.len());
    assert_eq!(SOME_FFQN, locked_execution.ffqn);
    locked_execution.next_version
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn locking_in_unlock_backoff_should_not_be_possible(
    database: Database,
    #[values(
        LockingStrategyLifecycle::ByFfqns,
        LockingStrategyLifecycle::ByComponentDigest
    )]
    locking_strategy: LockingStrategyLifecycle,
) {
    set_up();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    let sim_clock = SimClock::default();

    let execution_id = ExecutionId::generate();
    let exec1 = ExecutorId::generate();
    let lock_expiry = Duration::from_millis(500);

    // Create
    let component_id = ComponentId::dummy_activity();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: component_id.clone(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    let version = lock_pending(
        &execution_id,
        sim_clock.now() + lock_expiry,
        exec1,
        RunId::generate(),
        &component_id,
        db_connection.as_ref(),
        &sim_clock,
        locking_strategy,
    )
    .await;

    // The Executor / Worker responds with a Unlock
    let backoff = Duration::from_millis(500);
    let unlocked_at = sim_clock.now() + backoff;
    let _version = {
        let created_at = sim_clock.now();
        info!(now = %created_at, "unlock");
        let req = AppendRequest {
            event: ExecutionRequest::Unlocked(Unlocked {
                unlocked_at,
                reason: StrVariant::Static("reason"),
            }),
            created_at,
        };
        db_connection
            .append(execution_id.clone(), version, req)
            .await
            .unwrap()
    };

    // Too soon
    sim_clock.move_time_forward(backoff.checked_sub(Duration::from_millis(100)).unwrap());
    {
        let created_at = sim_clock.now();
        info!(now = %created_at, "Attempt to lock while in timeout backoff");
        let locked_executions = db_connection
            .lock_pending_by_ffqns(
                1,
                created_at,
                Arc::from([SOME_FFQN]),
                created_at,
                component_id.clone(),
                DEPLOYMENT_ID_DUMMY,
                exec1,
                created_at + lock_expiry,
                RunId::generate(),
                ComponentRetryConfig {
                    retry_exp_backoff: Duration::ZERO,
                    max_retries: Some(0),
                },
            )
            .await
            .unwrap();
        assert!(locked_executions.is_empty());
    }

    sim_clock.move_time_to(unlocked_at);
    {
        let created_at = sim_clock.now();
        info!(now = %created_at, "Locking exactly at `backoff_expires_at` should succeed");
        let locked_executions = db_connection
            .lock_pending_by_ffqns(
                1,
                created_at,
                Arc::from([SOME_FFQN]),
                created_at,
                component_id.clone(),
                DEPLOYMENT_ID_DUMMY,
                exec1,
                created_at + lock_expiry,
                RunId::generate(),
                ComponentRetryConfig {
                    retry_exp_backoff: Duration::ZERO,
                    max_retries: Some(0),
                },
            )
            .await
            .unwrap();
        assert_eq!(1, locked_executions.len());
    }
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn lock_extended_with_the_same_executor_should_work(database: Database) {
    set_up();
    let (_guard, db_pool, db_close) = database.set_up().await;
    {
        let db_connection = db_pool.connection().await.unwrap();
        let lock_resp = lock_and_attept_to_extend(true, true, db_connection.as_ref())
            .await
            .unwrap();
        assert_eq!(Version::new(3), lock_resp.next_version);
    }
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn lock_extended_with_another_executor_should_fail(database: Database) {
    set_up();
    let (_guard, db_pool, db_close) = database.set_up().await;
    {
        let db_connection = db_pool.connection().await.unwrap();
        lock_and_attept_to_extend(false, true, db_connection.as_ref())
            .await
            .unwrap_err();
        lock_and_attept_to_extend(true, false, db_connection.as_ref())
            .await
            .unwrap_err();
        lock_and_attept_to_extend(false, false, db_connection.as_ref())
            .await
            .unwrap_err();
    }
    db_close.close().await;
}
async fn lock_and_attept_to_extend(
    same_executor_id: bool,
    same_run_id: bool,
    db_connection: &dyn DbConnection,
) -> Result<LockedExecution, DbErrorWrite> {
    let sim_clock = SimClock::epoch();
    let execution_id = ExecutionId::generate();
    let exec_pending = ExecutorId::generate();
    let run_pending = RunId::generate();
    let exec_extend = if same_executor_id {
        exec_pending
    } else {
        ExecutorId::generate()
    };
    let run_extend = if same_run_id {
        run_pending
    } else {
        RunId::generate()
    };

    // Create
    let component_id = ComponentId::dummy_activity();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: component_id.clone(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    let lock_expiry = Duration::from_millis(500);

    // Lock using lock_pending
    let version = {
        let created_at = sim_clock.now();
        info!(now = %created_at, "LockPending");
        let locked_executions = db_connection
            .lock_pending_by_ffqns(
                1,
                created_at,
                Arc::from([SOME_FFQN]),
                created_at,
                component_id.clone(),
                DEPLOYMENT_ID_DUMMY,
                exec_pending,
                created_at + lock_expiry,
                run_pending,
                ComponentRetryConfig {
                    retry_exp_backoff: Duration::ZERO,
                    max_retries: Some(0),
                },
            )
            .await
            .unwrap();
        assert_eq!(1, locked_executions.len());
        let locked_execution = locked_executions.into_iter().next().unwrap();
        assert_eq!(execution_id, locked_execution.execution_id);
        assert_eq!(Version::new(2), locked_execution.next_version);
        assert_eq!(0, locked_execution.params.len());
        assert_eq!(SOME_FFQN, locked_execution.ffqn);
        locked_execution.next_version
    };

    sim_clock.move_time_forward(lock_expiry); // does not matter, as long as the execution is still in Locked state.
    let created_at = sim_clock.now();
    info!(now = %created_at, "Attempt extend the lock");
    db_connection
        .lock_one(
            created_at,
            component_id.clone(),
            DEPLOYMENT_ID_DUMMY,
            &execution_id,
            run_extend,
            version,
            exec_extend,
            created_at + lock_expiry,
            ComponentRetryConfig {
                retry_exp_backoff: Duration::ZERO,
                max_retries: Some(0),
            },
        )
        .await
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn locking_in_timeout_backoff_should_not_be_possible(database: Database) {
    set_up();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    let sim_clock = SimClock::default();

    let execution_id = ExecutionId::generate();
    let exec1 = ExecutorId::generate();
    let lock_expiry = Duration::from_millis(500);

    // Create
    let component_id = ComponentId::dummy_activity();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: component_id.clone(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // LockPending
    let version = {
        let created_at = sim_clock.now();
        info!(now = %created_at, "LockPending");
        let locked_executions = db_connection
            .lock_pending_by_ffqns(
                1,
                created_at,
                Arc::from([SOME_FFQN]),
                created_at,
                component_id.clone(),
                DEPLOYMENT_ID_DUMMY,
                exec1,
                created_at + lock_expiry,
                RunId::generate(),
                ComponentRetryConfig {
                    retry_exp_backoff: Duration::ZERO,
                    max_retries: Some(0),
                },
            )
            .await
            .unwrap();
        assert_eq!(1, locked_executions.len());
        let locked_execution = locked_executions.into_iter().next().unwrap();
        assert_eq!(execution_id, locked_execution.execution_id);
        assert_eq!(Version::new(2), locked_execution.next_version);
        assert_eq!(0, locked_execution.params.len());
        assert_eq!(SOME_FFQN, locked_execution.ffqn);
        locked_execution.next_version
    };
    // The Executor / Worker responds with a timeout
    sim_clock.move_time_forward(lock_expiry);
    let backoff = Duration::from_millis(300);
    {
        let created_at = sim_clock.now();
        info!(now = %created_at, "Temporary timeout");
        let req = AppendRequest {
            created_at,
            event: ExecutionRequest::TemporarilyTimedOut {
                backoff_expires_at: created_at + backoff,
                http_client_traces: None,
            },
        };

        db_connection
            .append(execution_id.clone(), version, req)
            .await
            .unwrap();
    }
    sim_clock.move_time_forward(backoff.checked_sub(Duration::from_millis(100)).unwrap());
    {
        let created_at = sim_clock.now();
        info!(now = %created_at, "Attempt to lock while in timeout backoff");
        let locked_executions = db_connection
            .lock_pending_by_ffqns(
                1,
                created_at,
                Arc::from([SOME_FFQN]),
                created_at,
                component_id.clone(),
                DEPLOYMENT_ID_DUMMY,
                exec1,
                created_at + lock_expiry,
                RunId::generate(),
                ComponentRetryConfig {
                    retry_exp_backoff: Duration::ZERO,
                    max_retries: Some(0),
                },
            )
            .await
            .unwrap();
        assert!(locked_executions.is_empty());
    }
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn lock_pending_while_nothing_is_pending_should_be_noop(database: Database) {
    set_up();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    let sim_clock = SimClock::default();

    let exec1 = ExecutorId::generate();
    let lock_expiry = Duration::from_millis(500);

    assert!(
        db_connection
            .lock_pending_by_ffqns(
                1,
                sim_clock.now(),
                Arc::from([SOME_FFQN]),
                sim_clock.now(),
                ComponentId::dummy_activity(),
                DEPLOYMENT_ID_DUMMY,
                exec1,
                sim_clock.now() + lock_expiry,
                RunId::generate(),
                ComponentRetryConfig {
                    retry_exp_backoff: Duration::ZERO,
                    max_retries: Some(0),
                },
            )
            .await
            .unwrap()
            .is_empty()
    );
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn creating_execution_twice_should_fail(database: Database) {
    set_up();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    let sim_clock = SimClock::default();

    let execution_id = ExecutionId::generate();
    let exec1 = ExecutorId::generate();
    let lock_expiry = Duration::from_millis(500);

    assert!(
        db_connection
            .lock_pending_by_ffqns(
                1,
                sim_clock.now(),
                Arc::from([SOME_FFQN]),
                sim_clock.now(),
                ComponentId::dummy_activity(),
                DEPLOYMENT_ID_DUMMY,
                exec1,
                sim_clock.now() + lock_expiry,
                RunId::generate(),
                ComponentRetryConfig {
                    retry_exp_backoff: Duration::ZERO,
                    max_retries: Some(0),
                },
            )
            .await
            .unwrap()
            .is_empty()
    );
    // Create
    let component_id = ComponentId::dummy_activity();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: component_id.clone(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Create again should fail
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap_err();
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn lock_pending_while_expired_lock_should_return_nothing(database: Database) {
    set_up();

    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    lock_pending_while_expired_lock_should_return_nothing_inner(db_connection.as_ref()).await;
    drop(db_connection);
    db_close.close().await;
}

async fn lock_pending_while_expired_lock_should_return_nothing_inner(
    db_connection: &dyn DbConnection,
) {
    const LOCK_EXPIRY: Duration = Duration::from_millis(500);
    let sim_clock = SimClock::default();
    let execution_id = ExecutionId::generate();
    let exec1 = ExecutorId::generate();

    // Create
    let component_id = ComponentId::dummy_activity();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: component_id.clone(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // LockPending
    {
        let created_at = sim_clock.now();
        info!(now = %created_at, "LockPending");
        let mut locked_executions = db_connection
            .lock_pending_by_ffqns(
                1,
                created_at,
                Arc::from([SOME_FFQN]),
                created_at,
                component_id.clone(),
                DEPLOYMENT_ID_DUMMY,
                exec1,
                created_at + LOCK_EXPIRY,
                RunId::generate(),
                ComponentRetryConfig {
                    retry_exp_backoff: Duration::ZERO,
                    max_retries: Some(0),
                },
            )
            .await
            .unwrap();
        assert_eq!(1, locked_executions.len());
        let locked_execution = locked_executions.pop().unwrap();
        assert_eq!(execution_id, locked_execution.execution_id);
        assert_eq!(Version::new(2), locked_execution.next_version);
        assert_eq!(0, locked_execution.params.len());
        assert_eq!(SOME_FFQN, locked_execution.ffqn);
    }
    // 0 = Before lock expiry
    // 1 = Right at lock expiry
    // 2 = Some time after lock expiry
    for _ in 0..3 {
        assert!(
            db_connection
                .lock_pending_by_ffqns(
                    1,
                    sim_clock.now(),
                    Arc::from([SOME_FFQN]),
                    sim_clock.now(),
                    ComponentId::dummy_activity(),
                    DEPLOYMENT_ID_DUMMY,
                    exec1,
                    sim_clock.now() + LOCK_EXPIRY,
                    RunId::generate(),
                    ComponentRetryConfig {
                        retry_exp_backoff: Duration::ZERO,
                        max_retries: Some(0),
                    },
                )
                .await
                .unwrap()
                .is_empty()
        );
        sim_clock.move_time_forward(LOCK_EXPIRY);
    }
}

pub async fn expired_lock_should_be_found(db_connection: &dyn DbConnection, sim_clock: SimClock) {
    const MAX_RETRIES: u32 = 1;
    const RETRY_EXP_BACKOFF: Duration = Duration::from_millis(100);

    let execution_id = ExecutionId::generate();
    let exec1 = ExecutorId::generate();
    // Create
    {
        db_connection
            .create(CreateRequest {
                created_at: sim_clock.now(),
                execution_id: execution_id.clone(),
                ffqn: SOME_FFQN,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: sim_clock.now(),
                component_id: ComponentId::dummy_activity(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
    }
    // Lock pending
    let lock_duration = Duration::from_millis(500);
    {
        let mut locked_executions = db_connection
            .lock_pending_by_ffqns(
                1,
                sim_clock.now(),
                Arc::from([SOME_FFQN]),
                sim_clock.now(),
                ComponentId::dummy_activity(),
                DEPLOYMENT_ID_DUMMY,
                exec1,
                sim_clock.now() + lock_duration,
                RunId::generate(),
                ComponentRetryConfig {
                    retry_exp_backoff: RETRY_EXP_BACKOFF,
                    max_retries: Some(MAX_RETRIES),
                },
            )
            .await
            .unwrap();
        assert_eq!(1, locked_executions.len());
        let locked_execution = locked_executions.pop().unwrap();
        assert_eq!(execution_id, locked_execution.execution_id);
        assert_eq!(SOME_FFQN, locked_execution.ffqn);
        assert_eq!(Version::new(2), locked_execution.next_version);
    }
    // Calling `get_expired_timers` after lock expiry should return the expired execution.
    sim_clock.move_time_forward(lock_duration);
    {
        let expired_at = sim_clock.now();
        let expired = db_connection.get_expired_timers(expired_at).await.unwrap();
        assert_eq!(1, expired.len());
        let expired = &expired[0];
        let lock = assert_matches!(expired, ExpiredTimer::Lock(lock) => lock);
        assert_eq!(execution_id, lock.execution_id);
        assert_eq!(Version::new(1), lock.locked_at_version);
        assert_eq!(Version::new(2), lock.next_version);
        assert_eq!(0, lock.intermittent_event_count);
        assert_eq!(Some(MAX_RETRIES), lock.max_retries);
        assert_eq!(RETRY_EXP_BACKOFF, lock.retry_exp_backoff);
    }
}

async fn append_batch_respond_to_parent(db_connection: &dyn DbConnectionTest, sim_clock: SimClock) {
    let parent_id = ExecutionId::generate();

    // Create parent
    let version = db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: parent_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    // Create joinset
    let join_set_id = JoinSetId::new(concepts::JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    let mut version = db_connection
        .append(
            parent_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetCreate {
                        join_set_id: join_set_id.clone(),
                    },
                },
            },
        )
        .await
        .unwrap();

    let child_a = {
        let parent_log = db_connection.get(&parent_id).await.unwrap();
        assert_matches!(
            parent_log.pending_state,
            PendingState::PendingAt(PendingStatePendingAt {
                scheduled_at,
                last_lock: None,
            }) if scheduled_at == sim_clock.now()
        );
        // Create child 1
        let child_id = parent_id.next_level(&join_set_id);
        let child_version = db_connection
            .create(CreateRequest {
                created_at: sim_clock.now(),
                execution_id: ExecutionId::Derived(child_id.clone()),
                ffqn: SOME_FFQN,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: sim_clock.now(),
                component_id: ComponentId::dummy_activity(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
        assert_eq!(Version(1), child_version);
        // Append JoinNext before receiving the response should make the parent BlockedByJoinSet
        db_connection
            .append(
                parent_id.clone(),
                version,
                AppendRequest {
                    created_at: sim_clock.now(),
                    event: ExecutionRequest::HistoryEvent {
                        event: HistoryEvent::JoinNext {
                            join_set_id: join_set_id.clone(),
                            run_expires_at: sim_clock.now(),
                            closing: false,
                            requested_ffqn: Some(SOME_FFQN),
                        },
                    },
                },
            )
            .await
            .unwrap();
        let parent_exe = db_connection.get(&parent_id).await.unwrap();
        assert_matches!(
            parent_exe.pending_state,
            PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet {
                join_set_id: found_join_set_id,
                lock_expires_at,
                closing: false,
            })
            if found_join_set_id == join_set_id && lock_expires_at == sim_clock.now()
        );

        // Append child response to unblock the parent
        db_connection
            .append_batch_respond_to_parent(
                AppendEventsToExecution {
                    execution_id: ExecutionId::Derived(child_id.clone()),
                    version: child_version.clone(),
                    batch: vec![AppendRequest {
                        created_at: sim_clock.now(),
                        event: ExecutionRequest::Finished {
                            retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                            http_client_traces: None,
                        },
                    }],
                },
                AppendResponseToExecution {
                    parent_execution_id: parent_id.clone(),
                    created_at: sim_clock.now(),
                    join_set_id: join_set_id.clone(),
                    child_execution_id: child_id.clone(),
                    finished_version: child_version, // will remain at 1.
                    result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                },
                sim_clock.now(),
            )
            .await
            .unwrap();

        let parent_exe = db_connection.get(&parent_id).await.unwrap();
        assert_matches!(parent_exe.pending_state, PendingState::PendingAt(..));
        version = parent_exe.next_version;

        child_id
    };
    let child_b = {
        // Create child 2
        let child_id = child_a.get_incremented();
        let child_version = db_connection
            .create(CreateRequest {
                created_at: sim_clock.now(),
                execution_id: ExecutionId::Derived(child_id.clone()),
                ffqn: SOME_FFQN,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: sim_clock.now(),
                component_id: ComponentId::dummy_activity(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
        let batch = vec![AppendRequest {
            created_at: sim_clock.now(),
            event: ExecutionRequest::Finished {
                retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                http_client_traces: None,
            },
        }];

        db_connection
            .append_batch_respond_to_parent(
                AppendEventsToExecution {
                    execution_id: ExecutionId::Derived(child_id.clone()),
                    version: child_version.clone(),
                    batch,
                },
                AppendResponseToExecution {
                    parent_execution_id: parent_id.clone(),
                    created_at: sim_clock.now(),
                    join_set_id: join_set_id.clone(),
                    child_execution_id: child_id.clone(),
                    finished_version: child_version,
                    result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                },
                sim_clock.now(),
            )
            .await
            .unwrap();

        // Append JoinNext after receiving the response should make the parent PendingAt
        db_connection
            .append(
                parent_id.clone(),
                version,
                AppendRequest {
                    created_at: sim_clock.now(),
                    event: ExecutionRequest::HistoryEvent {
                        event: HistoryEvent::JoinNext {
                            join_set_id: join_set_id.clone(),
                            run_expires_at: sim_clock.now(),
                            closing: false,
                            requested_ffqn: Some(SOME_FFQN),
                        },
                    },
                },
            )
            .await
            .unwrap();
        let parent_exe = db_connection.get(&parent_id).await.unwrap();
        assert_matches!(parent_exe.pending_state, PendingState::PendingAt(..));

        child_id
    };
    let parent_exe = db_connection.get(&parent_id).await.unwrap();
    assert_eq!(2, parent_exe.responses.len());
    assert_eq!(
        parent_exe.responses.first().unwrap().event,
        JoinSetResponseEventOuter {
            created_at: sim_clock.now(),
            event: JoinSetResponseEvent {
                join_set_id: join_set_id.clone(),
                event: JoinSetResponse::ChildExecutionFinished {
                    child_execution_id: child_a,
                    finished_version: Version(1),
                    result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                },
            }
        }
    );
    assert_eq!(
        parent_exe.responses.get(1).unwrap().event,
        JoinSetResponseEventOuter {
            created_at: sim_clock.now(),
            event: JoinSetResponseEvent {
                join_set_id: join_set_id.clone(),
                event: JoinSetResponse::ChildExecutionFinished {
                    child_execution_id: child_b,
                    finished_version: Version(1),
                    result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                },
            }
        }
    );
    assert_matches!(parent_exe.pending_state, PendingState::PendingAt(..));
}

enum LockingStrategyLifecycle {
    ByFfqns,
    ByComponentDigest,
}

async fn lock_pending_should_sort_by_scheduled_at(
    db_connection: &dyn DbConnection,
    sim_clock: SimClock,
    locking_strategy: LockingStrategyLifecycle,
) {
    let created_at = sim_clock.now();
    let older_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at,
            execution_id: older_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    sim_clock.move_time_forward(Duration::from_nanos(1));
    let newer_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at,
            execution_id: newer_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    sim_clock.move_time_forward(Duration::from_nanos(999));
    let newest_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at,
            execution_id: newest_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    let locked_ids = match locking_strategy {
        LockingStrategyLifecycle::ByFfqns => {
            db_connection
                .lock_pending_by_ffqns(
                    3,
                    sim_clock.now(),
                    Arc::from([SOME_FFQN]),
                    sim_clock.now(),
                    ComponentId::dummy_activity(),
                    DEPLOYMENT_ID_DUMMY,
                    ExecutorId::generate(),
                    sim_clock.now() + Duration::from_secs(1),
                    RunId::generate(),
                    ComponentRetryConfig {
                        retry_exp_backoff: Duration::ZERO,
                        max_retries: Some(0),
                    },
                )
                .await
        }
        LockingStrategyLifecycle::ByComponentDigest => {
            db_connection
                .lock_pending_by_component_digest(
                    3,
                    sim_clock.now(),
                    &ComponentId::dummy_activity(),
                    DEPLOYMENT_ID_DUMMY,
                    sim_clock.now(),
                    ExecutorId::generate(),
                    sim_clock.now() + Duration::from_secs(1),
                    RunId::generate(),
                    ComponentRetryConfig {
                        retry_exp_backoff: Duration::ZERO,
                        max_retries: Some(0),
                    },
                )
                .await
        }
    }
    .unwrap()
    .into_iter()
    .map(|locked| locked.execution_id)
    .collect::<Vec<_>>();

    assert_eq!(vec![older_id, newer_id, newest_id], locked_ids);
}

async fn test_lock_inner(db_connection: &dyn DbConnection, sim_clock: SimClock) {
    let execution_id = ExecutionId::generate();
    let executor_id = ExecutorId::generate();
    // Create
    let version = db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    // Append an event that does not change Pending state but must update the version.
    {
        let join_set_id =
            JoinSetId::new(concepts::JoinSetKind::OneOff, StrVariant::empty()).unwrap();
        db_connection
            .append(
                execution_id.clone(),
                version,
                AppendRequest {
                    created_at: sim_clock.now(),
                    event: ExecutionRequest::HistoryEvent {
                        event: HistoryEvent::JoinSetRequest {
                            request: JoinSetRequest::DelayRequest {
                                delay_id: DelayId::new(&execution_id, &join_set_id),
                                expires_at: sim_clock.now(),
                                schedule_at: HistoryEventScheduleAt::Now,
                                paused: false,
                            },
                            join_set_id,
                        },
                    },
                },
            )
            .await
            .unwrap()
    };

    lock(
        db_connection,
        &execution_id,
        &sim_clock,
        executor_id,
        sim_clock.now() + Duration::from_millis(100), // lock expires at
        RunId::generate(),
    )
    .await;
}

async fn lock(
    db_connection: &dyn DbConnection,
    execution_id: &ExecutionId,
    sim_clock: &SimClock,
    executor_id: ExecutorId,
    lock_expires_at: DateTime<Utc>,
    run_id: RunId,
) -> Version {
    let lock_pending_res = db_connection
        .lock_pending_by_ffqns(
            1,
            sim_clock.now(), // pending at or sooner
            Arc::from([SOME_FFQN]),
            sim_clock.now(), // created at
            ComponentId::dummy_activity(),
            DEPLOYMENT_ID_DUMMY,
            executor_id,
            lock_expires_at,
            run_id,
            ComponentRetryConfig {
                retry_exp_backoff: Duration::ZERO,
                max_retries: Some(0),
            },
        )
        .await
        .unwrap();
    assert_eq!(1, lock_pending_res.len());
    let lock_pending_res = lock_pending_res.into_iter().next().unwrap();
    assert_eq!(*execution_id, lock_pending_res.execution_id);

    lock_pending_res.next_version
}

async fn get_expired_lock(db_connection: &dyn DbConnection, sim_clock: SimClock) {
    let execution_id = ExecutionId::generate();
    let executor_id = ExecutorId::generate();
    // Create
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    let lock_expiry = Duration::from_millis(100);
    let run_id = RunId::generate();
    let version = lock(
        db_connection,
        &execution_id,
        &sim_clock,
        executor_id,
        sim_clock.now() + lock_expiry,
        run_id,
    )
    .await;

    assert!(
        db_connection
            .get_expired_timers(sim_clock.now())
            .await
            .unwrap()
            .is_empty()
    );

    sim_clock.move_time_forward(lock_expiry);

    let mut actual = db_connection
        .get_expired_timers(sim_clock.now())
        .await
        .unwrap();
    assert_eq!(1, actual.len());
    let actual = actual.pop().unwrap();
    let expected = ExpiredTimer::Lock(ExpiredLock {
        execution_id,
        locked_at_version: Version::new(1),
        next_version: version,
        intermittent_event_count: 0,
        max_retries: Some(0),
        retry_exp_backoff: Duration::ZERO,
        locked_by: LockedBy {
            executor_id,
            run_id,
        },
    });
    assert_eq!(expected, actual);
}

async fn get_expired_delay(db_connection: &dyn DbConnection, sim_clock: SimClock) {
    let execution_id = ExecutionId::generate();
    let executor_id = ExecutorId::generate();
    // Create
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    let lock_expiry = Duration::from_millis(100);
    let version = lock(
        db_connection,
        &execution_id,
        &sim_clock,
        executor_id,
        sim_clock.now() + lock_expiry * 2, // lock expires at
        RunId::generate(),
    )
    .await;

    // Create joinset
    let join_set_id = JoinSetId::new(concepts::JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    let version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetCreate {
                        join_set_id: join_set_id.clone(),
                    },
                },
            },
        )
        .await
        .unwrap();

    let delay_id = DelayId::new(&execution_id, &join_set_id);
    db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request: JoinSetRequest::DelayRequest {
                            delay_id: delay_id.clone(),
                            expires_at: sim_clock.now() + lock_expiry,
                            schedule_at: HistoryEventScheduleAt::In(lock_expiry),
                            paused: false,
                        },
                    },
                },
            },
        )
        .await
        .unwrap();

    assert!(
        db_connection
            .get_expired_timers(sim_clock.now())
            .await
            .unwrap()
            .is_empty()
    );

    sim_clock.move_time_forward(lock_expiry);

    let mut actual = db_connection
        .get_expired_timers(sim_clock.now())
        .await
        .unwrap();
    assert_eq!(1, actual.len());
    let actual = actual.pop().unwrap();
    let expected = ExpiredTimer::Delay(ExpiredDelay {
        execution_id: execution_id.clone(),
        join_set_id,
        delay_id,
    });
    assert_eq!(expected, actual);
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn get_expired_times_with_execution_that_made_progress(
    database: Database,
    #[values(
        LockingStrategyLifecycle::ByFfqns,
        LockingStrategyLifecycle::ByComponentDigest
    )]
    locking_strategy: LockingStrategyLifecycle,
) {
    set_up();

    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    let sim_clock = SimClock::default();

    let execution_id = ExecutionId::generate();
    let exec1 = ExecutorId::generate();
    let lock_expiry = Duration::from_millis(500);

    // Create = Version(0)
    let component_id = ComponentId::dummy_activity();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: component_id.clone(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Lock = Version(1)
    let run_id = RunId::generate();
    lock_pending(
        &execution_id,
        sim_clock.now() + lock_expiry,
        exec1,
        run_id,
        &component_id,
        db_connection.as_ref(),
        &sim_clock,
        locking_strategy,
    )
    .await;

    // Create joinset = Version(2)
    let join_set_id = JoinSetId::new(concepts::JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    let next_version = db_connection
        .append(
            execution_id.clone(),
            Version::new(2),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetCreate {
                        join_set_id: join_set_id.clone(),
                    },
                },
            },
        )
        .await
        .unwrap();
    assert_eq!(Version::new(3), next_version);

    assert!(
        db_connection
            .get_expired_timers(sim_clock.now())
            .await
            .unwrap()
            .is_empty(),
    );
    sim_clock.move_time_forward(lock_expiry);
    let expired_timers = db_connection
        .get_expired_timers(sim_clock.now())
        .await
        .unwrap();
    assert_eq!(1, expired_timers.len());
    let expired_timer = expired_timers.into_iter().next().unwrap();
    let expired_lock = assert_matches!(expired_timer, ExpiredTimer::Lock(lock) => lock);
    let expected_lock = ExpiredLock {
        execution_id,
        locked_at_version: Version(1),
        next_version: Version::new(3),
        intermittent_event_count: 0,
        max_retries: Some(0),
        retry_exp_backoff: Duration::ZERO,
        locked_by: LockedBy {
            executor_id: exec1,
            run_id,
        },
    };
    assert_eq!(expected_lock, expired_lock);
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_append_same_delay_id_twice_should_fail(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    append_same_delay_id_twice_should_fail(db_connection.as_ref(), sim_clock).await;
    drop(db_connection);
    db_close.close().await;
}

async fn append_same_delay_id_twice_should_fail(
    db_connection: &dyn DbConnection,
    sim_clock: SimClock,
) {
    let execution_id = ExecutionId::generate();
    let executor_id = ExecutorId::generate();
    // Create
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    let lock_expiry = Duration::from_millis(100);
    let mut version = lock(
        db_connection,
        &execution_id,
        &sim_clock,
        executor_id,
        sim_clock.now() + lock_expiry, // lock expires at
        RunId::generate(),
    )
    .await;

    // Create joinset
    let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection,
    )
    .await;
    let delay_id = DelayId::new(&execution_id, &join_set_id);
    let version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request: JoinSetRequest::DelayRequest {
                            delay_id: delay_id.clone(),
                            expires_at: sim_clock.now() + lock_expiry,
                            schedule_at: HistoryEventScheduleAt::In(lock_expiry),
                            paused: false,
                        },
                    },
                },
            },
        )
        .await
        .unwrap();

    let err = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request: JoinSetRequest::DelayRequest {
                            delay_id: delay_id.clone(),
                            expires_at: sim_clock.now() + lock_expiry,
                            schedule_at: HistoryEventScheduleAt::In(lock_expiry),
                            paused: false,
                        },
                    },
                },
            },
        )
        .await
        .unwrap_err();
    assert_matches!(
        err,
        DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::Conflict)
    );
}

async fn create_join_set(
    execution_id: &ExecutionId,
    version: &mut Version,
    join_set_id: JoinSetId,
    created_at: DateTime<Utc>,
    db_connection: &dyn DbConnection,
) {
    let new_version = db_connection
        .append(
            execution_id.clone(),
            version.clone(),
            AppendRequest {
                created_at,
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetCreate { join_set_id },
                },
            },
        )
        .await
        .unwrap();
    *version = new_version;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cannot_append_unlocked_to_blocked_execution(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    let lock_expiry = Duration::from_millis(100);
    let mut version = lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + lock_expiry,
        RunId::generate(),
    )
    .await;

    let join_set_id = JoinSetId::new(JoinSetKind::Named, "blocked".into()).unwrap();
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection.as_ref(),
    )
    .await;

    version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinNext {
                        join_set_id,
                        run_expires_at: sim_clock.now() + lock_expiry,
                        closing: false,
                        requested_ffqn: Some(SOME_FFQN),
                    },
                },
            },
        )
        .await
        .unwrap();

    let err = db_connection
        .append(
            execution_id,
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Unlocked(Unlocked {
                    unlocked_at: sim_clock.now(),
                    reason: "should fail".into(),
                }),
            },
        )
        .await
        .unwrap_err();
    assert_matches!(
        err,
        DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::UnlockedCannotBeAppended(_))
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cannot_append_unlocked_to_paused_execution(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    let version = db_connection
        .append(
            execution_id.clone(),
            Version::new(1),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap();

    let err = db_connection
        .append(
            execution_id,
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Unlocked(Unlocked {
                    unlocked_at: sim_clock.now(),
                    reason: "should fail".into(),
                }),
            },
        )
        .await
        .unwrap_err();
    assert_matches!(
        err,
        DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::UnlockedCannotBeAppended(_))
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_append_response_with_same_delay_id_twice_should_fail(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let execution_id = ExecutionId::generate();
    let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    let delay_id = DelayId::new(&execution_id, &join_set_id);
    let request = JoinSetRequest::DelayRequest {
        delay_id: delay_id.clone(),
        expires_at: sim_clock.now(),
        schedule_at: HistoryEventScheduleAt::Now,
        paused: false,
    };
    let response = JoinSetResponse::DelayFinished {
        delay_id: delay_id.clone(),
        result: Ok(()),
    };
    append_response_with_same_id_twice_should_fail(
        db_connection.as_ref(),
        sim_clock,
        execution_id,
        join_set_id,
        request,
        response,
    )
    .await;
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_append_response_with_same_child_id_twice_should_fail(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let execution_id = ExecutionId::generate();
    let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    let child_execution_id = execution_id.next_level(&join_set_id);
    let request = JoinSetRequest::ChildExecutionRequest {
        child_execution_id: child_execution_id.clone(),
        target_ffqn: SOME_FFQN,
        params: Params::empty(),
        result: Ok(()),
    };
    let response = JoinSetResponse::ChildExecutionFinished {
        child_execution_id,
        finished_version: Version(1),
        result: concepts::SupportedFunctionReturnValue::Ok(None),
    };
    append_response_with_same_id_twice_should_fail(
        db_connection.as_ref(),
        sim_clock,
        execution_id,
        join_set_id,
        request,
        response,
    )
    .await;
    drop(db_connection);
    db_close.close().await;
}

async fn append_response_with_same_id_twice_should_fail(
    db_connection: &dyn DbConnectionTest,
    sim_clock: SimClock,
    execution_id: ExecutionId,
    join_set_id: JoinSetId,
    request: JoinSetRequest,
    response: JoinSetResponse,
) {
    let executor_id = ExecutorId::generate();
    // Create
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    let lock_expiry = Duration::from_millis(100);
    let mut version = lock(
        db_connection,
        &execution_id,
        &sim_clock,
        executor_id,
        sim_clock.now() + lock_expiry, // lock expires at
        RunId::generate(),
    )
    .await;

    // Create joinset
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection,
    )
    .await;

    db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request,
                    },
                },
            },
        )
        .await
        .unwrap();

    db_connection
        .append_response(
            sim_clock.now(),
            execution_id.clone(),
            JoinSetResponseEvent {
                join_set_id: join_set_id.clone(),
                event: response.clone(),
            },
        )
        .await
        .unwrap();
    let err = db_connection
        .append_response(
            sim_clock.now(),
            execution_id.clone(),
            JoinSetResponseEvent {
                join_set_id: join_set_id.clone(),
                event: response,
            },
        )
        .await
        .unwrap_err();
    assert_matches!(
        err,
        DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::Conflict)
    );
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn delay_cancellation_should_be_idempotent(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    // Create
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    let lock_expiry = Duration::from_millis(100);
    let mut version = lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + lock_expiry, // lock expires at
        RunId::generate(),
    )
    .await;

    // Create joinset
    let join_set_id = JoinSetId::new(JoinSetKind::Named, "x".into()).unwrap();
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection.as_ref(),
    )
    .await;

    let delay_id = DelayId::new(&execution_id, &join_set_id);
    db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request: JoinSetRequest::DelayRequest {
                            delay_id: delay_id.clone(),
                            expires_at: sim_clock.now() + lock_expiry,
                            schedule_at: HistoryEventScheduleAt::In(lock_expiry),
                            paused: false,
                        },
                    },
                },
            },
        )
        .await
        .unwrap();

    let res = storage::cancel_delay(db_connection.as_ref(), delay_id.clone(), sim_clock.now())
        .await
        .unwrap();
    assert_eq!(CancelOutcome::Cancelled, res);
    let res = storage::cancel_delay(db_connection.as_ref(), delay_id.clone(), sim_clock.now())
        .await
        .unwrap();
    assert_eq!(CancelOutcome::Cancelled, res);
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn pause_and_unpause_should_work(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    let original_scheduled_at = sim_clock.now() + Duration::from_mins(1);
    // Create with a future scheduled time
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: original_scheduled_at,
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Pause the execution
    let next_version = db_connection
        .append(
            execution_id.clone(),
            Version::new(1),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap();
    assert_eq!(Version(2), next_version);

    // Verify state is paused
    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(log.pending_state, PendingState::Paused(..));

    // Pausing again should return an error
    let err = db_connection
        .append(
            execution_id.clone(),
            Version::new(2),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap_err();
    let reason = assert_matches!(err, DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::IllegalState { reason, .. }) => reason);
    assert_eq!("cannot pause, execution is already paused", reason.as_ref());

    // Unpause the execution
    let next_version = db_connection
        .append(
            execution_id.clone(),
            Version::new(2),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Unpaused,
            },
        )
        .await
        .unwrap();
    assert_eq!(Version(3), next_version);

    // Verify state is pending at the original scheduled time
    let log = db_connection.get(&execution_id).await.unwrap();
    if let PendingState::PendingAt(PendingStatePendingAt { scheduled_at, .. }) = log.pending_state {
        assert_eq!(scheduled_at, original_scheduled_at);
    } else {
        panic!("Expected PendingAt state, got {:?}", log.pending_state);
    }

    // Unpausing again should return an error
    let err = db_connection
        .append(
            execution_id.clone(),
            Version::new(3),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Unpaused,
            },
        )
        .await
        .unwrap_err();

    let reason = assert_matches!(err, DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::IllegalState { reason, .. }) => reason);
    assert_eq!("cannot unpause, execution is not paused", reason.as_ref());

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn pause_finished_execution_should_fail(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    // Create
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Finish the execution
    db_connection
        .append(
            execution_id.clone(),
            Version::new(1),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Finished {
                    retval: SupportedFunctionReturnValue::Ok(None),
                    http_client_traces: None,
                },
            },
        )
        .await
        .unwrap();

    // Pausing should return error
    let err = db_connection
        .append(
            execution_id.clone(),
            Version::new(2),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap_err();
    assert_matches!(
        err,
        DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::AlreadyFinished)
    );

    drop(db_connection);
    db_close.close().await;
}

#[rstest]
#[case(Database::Sqlite)]
#[case(Database::Postgres)]
#[tokio::test]
async fn pause_and_unpause_locked_workflow_should_return_to_pending_at(#[case] database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    let db_external = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    // Create with current scheduled time (ready to run). A workflow, since pausing a running
    // activity is rejected; see `cannot_pause_running_activity`.
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_workflow(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Lock the execution
    let lock_expiry = Duration::from_secs(30);
    let lock_expires_at = sim_clock.now() + lock_expiry;
    lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        lock_expires_at,
        RunId::generate(),
    )
    .await;

    // Verify it's locked
    let log = db_connection.get(&execution_id).await.unwrap();
    let (found_locked_by, found_lock_expires_at) = assert_matches!(log.pending_state,
        PendingState::Locked(PendingStateLocked{ locked_by, lock_expires_at }) => (locked_by, lock_expires_at));
    assert_eq!(lock_expires_at, found_lock_expires_at);
    let paused_at = sim_clock.now();
    // Pause the locked execution through the public API.
    db_external
        .pause_execution(&execution_id, paused_at)
        .await
        .unwrap();

    // Verify the lock was released before the execution was paused.
    let log = db_connection.get(&execution_id).await.unwrap();
    let paused_pending_at = assert_matches!(log.pending_state,
        PendingState::Paused(PendingStatePaused::PendingAt(PendingStatePendingAt { scheduled_at, last_lock })) => (scheduled_at, last_lock));
    assert_eq!(paused_at, paused_pending_at.0);
    assert_eq!(Some(found_locked_by.clone()), paused_pending_at.1);
    let reason = assert_matches!(
        &log.events[2].event,
        ExecutionRequest::Unlocked(Unlocked { unlocked_at, reason }) => {
            assert_eq!(&paused_at, unlocked_at);
            reason
        }
    );
    assert_eq!("paused", reason.to_string());
    assert_matches!(log.events[3].event, ExecutionRequest::Paused);

    // Unpause and verify it returns to pending state instead of reviving the old lock.
    db_external
        .unpause_execution(&execution_id, sim_clock.now())
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    let pending_at = assert_matches!(log.pending_state,
        PendingState::PendingAt(PendingStatePendingAt { scheduled_at, last_lock }) => (scheduled_at, last_lock));
    assert_eq!(paused_at, pending_at.0);
    assert_eq!(Some(found_locked_by), pending_at.1);

    drop(db_connection);
    drop(db_external);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cannot_pause_running_activity(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    let db_external = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + Duration::from_secs(30),
        RunId::generate(),
    )
    .await;

    let err = db_external
        .pause_execution(&execution_id, sim_clock.now())
        .await
        .unwrap_err();
    let reason = assert_matches!(err, DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::IllegalState { reason, .. }) => reason);
    assert_eq!(
        "cannot pause a running activity; cancel it instead",
        reason.as_ref()
    );

    // The row is untouched: still locked, no Unlocked/Paused appended.
    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(log.pending_state, PendingState::Locked(_));

    drop(db_connection);
    drop(db_external);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cannot_append_paused_to_locked_execution(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    let lock_expires_at = sim_clock.now() + Duration::from_secs(30);
    lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        lock_expires_at,
        RunId::generate(),
    )
    .await;

    let err = db_connection
        .append(
            execution_id.clone(),
            Version::new(2),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap_err();
    let reason = assert_matches!(err, DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::IllegalState { reason, .. }) => reason);
    assert_eq!(
        "cannot append Paused event when execution is locked; use pause_execution",
        reason.as_ref()
    );

    drop(db_connection);
    db_close.close().await;
}

async fn create_at(
    db_connection: &dyn DbConnection,
    sim_clock: &SimClock,
    execution_id: &ExecutionId,
    scheduled_at: DateTime<Utc>,
) {
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at,
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancellation_requested_from_pending_at_should_keep_underlying_state(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    // Future scheduled time so the underlying state stays PendingAt.
    let scheduled_at = sim_clock.now() + Duration::from_mins(1);
    create_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        scheduled_at,
    )
    .await;

    let next_version = db_connection
        .append(
            execution_id.clone(),
            Version::new(1),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::CancellationRequested,
            },
        )
        .await
        .unwrap();
    assert_eq!(Version(2), next_version);

    let log = db_connection.get(&execution_id).await.unwrap();
    // The state is now first-class Cancelling, wrapping the unchanged underlying
    // PendingAt.
    let scheduled = assert_matches!(
        log.pending_state,
        PendingState::Cancelling(PendingStateCancelling::PendingAt(PendingStatePendingAt {
            scheduled_at,
            ..
        })) => scheduled_at
    );
    assert_eq!(scheduled_at, scheduled);
    assert_matches!(
        log.last_event().event,
        ExecutionRequest::CancellationRequested
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancellation_requested_on_locked_workflow_should_keep_underlying_state(
    database: Database,
) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_workflow(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    let lock_expires_at = sim_clock.now() + Duration::from_secs(30);
    lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        lock_expires_at,
        RunId::generate(),
    )
    .await;

    db_connection
        .append(
            execution_id.clone(),
            Version::new(2),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::CancellationRequested,
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    let locked = assert_matches!(
        &log.pending_state,
        PendingState::Cancelling(PendingStateCancelling::Locked(locked)) => locked
    );
    assert_eq!(lock_expires_at, locked.lock_expires_at);

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancel_activity_on_locked_execution_requests_cancellation(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    create_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        sim_clock.now(),
    )
    .await;
    let lock_expires_at = sim_clock.now() + Duration::from_secs(30);
    lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        lock_expires_at,
        RunId::generate(),
    )
    .await;

    let outcome = db_connection
        .append_activity_cancellation_requested(&execution_id, sim_clock.now())
        .await
        .unwrap();
    assert_eq!(CancelOutcome::Cancelled, outcome);

    let log = db_connection.get(&execution_id).await.unwrap();
    let locked = assert_matches!(
        &log.pending_state,
        PendingState::Cancelling(PendingStateCancelling::Locked(locked)) => locked
    );
    assert_eq!(lock_expires_at, locked.lock_expires_at);
    assert_matches!(
        log.last_event().event,
        ExecutionRequest::CancellationRequested
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancel_activity_on_paused_execution_requests_cancellation(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    create_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        sim_clock.now(),
    )
    .await;
    db_connection
        .append(
            execution_id.clone(),
            Version::new(1),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap();

    let outcome = db_connection
        .append_activity_cancellation_requested(&execution_id, sim_clock.now())
        .await
        .unwrap();
    assert_eq!(CancelOutcome::Cancelled, outcome);

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(
        log.pending_state,
        PendingState::Cancelling(PendingStateCancelling::PendingAt(_))
    );
    assert_matches!(
        log.last_event().event,
        ExecutionRequest::CancellationRequested
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancellation_requested_on_paused_workflow_should_keep_underlying_state(
    database: Database,
) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_workflow(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    let version = db_connection
        .append(
            execution_id.clone(),
            Version::new(1),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap();

    db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::CancellationRequested,
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    // Cancel supersedes pause: cancelling, not paused.
    assert_matches!(
        log.pending_state,
        PendingState::Cancelling(PendingStateCancelling::PendingAt(_))
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cannot_cancel_finished_execution(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    create_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        sim_clock.now(),
    )
    .await;
    let version = lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + Duration::from_secs(30),
        RunId::generate(),
    )
    .await;
    let version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Finished {
                    retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                    http_client_traces: None,
                },
            },
        )
        .await
        .unwrap();

    let err = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::CancellationRequested,
            },
        )
        .await
        .unwrap_err();
    assert_matches!(
        err,
        DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::AlreadyFinished)
    );

    drop(db_connection);
    db_close.close().await;
}

async fn create_cancellable_at(
    db_connection: &dyn DbConnection,
    sim_clock: &SimClock,
    execution_id: &ExecutionId,
    scheduled_at: DateTime<Utc>,
) {
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: CANCELLABLE_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at,
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancel_workflow_from_pending_at(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    create_cancellable_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        sim_clock.now(),
    )
    .await;

    let outcome = db_connection
        .cancel_workflow(&execution_id, sim_clock.now())
        .await
        .unwrap();
    assert_eq!(CancelOutcome::Cancelled, outcome);

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(
        log.pending_state,
        PendingState::Cancelling(PendingStateCancelling::PendingAt(_))
    );
    assert_matches!(
        log.last_event().event,
        ExecutionRequest::CancellationRequested
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancel_workflow_from_locked_should_keep_underlying_state(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    create_cancellable_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        sim_clock.now(),
    )
    .await;
    db_connection
        .lock_one(
            sim_clock.now(),
            ComponentId::dummy_activity(),
            DEPLOYMENT_ID_DUMMY,
            &execution_id,
            RunId::generate(),
            Version::new(1),
            ExecutorId::generate(),
            sim_clock.now() + Duration::from_secs(30),
            ComponentRetryConfig::ZERO,
        )
        .await
        .unwrap();

    let outcome = db_connection
        .cancel_workflow(&execution_id, sim_clock.now())
        .await
        .unwrap();
    assert_eq!(CancelOutcome::Cancelled, outcome);

    let log = db_connection.get(&execution_id).await.unwrap();
    // The lock is kept; the running workflow is fenced by the version bump.
    assert_matches!(
        log.pending_state,
        PendingState::Cancelling(PendingStateCancelling::Locked(_))
    );
    // ..., Locked, CancellationRequested
    assert_matches!(
        log.last_event().event,
        ExecutionRequest::CancellationRequested
    );
    assert_matches!(
        log.events[log.events.len() - 2].event,
        ExecutionRequest::Locked(_)
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancel_workflow_from_paused_should_keep_underlying_state(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    create_cancellable_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        sim_clock.now(),
    )
    .await;
    db_connection
        .pause_execution(&execution_id, sim_clock.now())
        .await
        .unwrap();

    let outcome = db_connection
        .cancel_workflow(&execution_id, sim_clock.now())
        .await
        .unwrap();
    assert_eq!(CancelOutcome::Cancelled, outcome);

    let log = db_connection.get(&execution_id).await.unwrap();
    // Cancel supersedes pause: cancelling, not paused.
    assert_matches!(
        log.pending_state,
        PendingState::Cancelling(PendingStateCancelling::PendingAt(_))
    );
    // ..., Paused, CancellationRequested (no Unpaused: paused executions are never running)
    assert_matches!(
        log.last_event().event,
        ExecutionRequest::CancellationRequested
    );
    assert_matches!(
        log.events[log.events.len() - 2].event,
        ExecutionRequest::Paused
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancel_workflow_rejects_non_cancellable(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    // SOME_FFQN has no `-cancellable` suffix.
    create_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        sim_clock.now(),
    )
    .await;

    let err = db_connection
        .cancel_workflow(&execution_id, sim_clock.now())
        .await
        .unwrap_err();
    let reason = assert_matches!(err, DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::IllegalState { reason, .. }) => reason);
    assert_eq!(
        "cannot cancel, execution is not a cancellable workflow",
        reason.as_ref()
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancel_workflow_second_call_is_already_cancelling(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    create_cancellable_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        sim_clock.now(),
    )
    .await;

    let first = db_connection
        .cancel_workflow(&execution_id, sim_clock.now())
        .await
        .unwrap();
    assert_eq!(CancelOutcome::Cancelled, first);
    let second = db_connection
        .cancel_workflow(&execution_id, sim_clock.now())
        .await
        .unwrap();
    assert_eq!(CancelOutcome::AlreadyCancelling, second);

    drop(db_connection);
    db_close.close().await;
}

/// `get_cancelling` (the driver's pick-up query) returns only `cancelling` rows,
/// excluding an active one.
#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn get_cancelling_returns_only_cancelling_rows(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let cancelling = ExecutionId::generate();
    create_cancellable_at(
        db_connection.as_ref(),
        &sim_clock,
        &cancelling,
        sim_clock.now(),
    )
    .await;
    db_connection
        .cancel_workflow(&cancelling, sim_clock.now())
        .await
        .unwrap();

    let active = ExecutionId::generate();
    create_at(db_connection.as_ref(), &sim_clock, &active, sim_clock.now()).await;

    assert_eq!(
        vec![cancelling],
        db_connection.get_cancelling(10).await.unwrap()
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cannot_lock_paused_execution(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    // Create with current scheduled time (ready to run)
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Pause the execution
    db_connection
        .append(
            execution_id.clone(),
            Version::new(1),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap();

    // Try to lock the paused execution - should fail
    let err = db_connection
        .lock_one(
            sim_clock.now(),
            ComponentId::dummy_activity(),
            DEPLOYMENT_ID_DUMMY,
            &execution_id,
            RunId::generate(),
            Version::new(2), // After pause event
            ExecutorId::generate(),
            sim_clock.now() + Duration::from_secs(30),
            ComponentRetryConfig::ZERO,
        )
        .await
        .unwrap_err();

    let reason = assert_matches!(err, DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::IllegalState { reason, .. }) => reason);
    assert_eq!("cannot lock, execution is paused", reason.as_ref());

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cannot_lock_cancelling_execution(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    // Create ready to run, then request cancellation from PendingAt.
    create_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        sim_clock.now(),
    )
    .await;
    db_connection
        .append(
            execution_id.clone(),
            Version::new(1),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::CancellationRequested,
            },
        )
        .await
        .unwrap();

    let err = db_connection
        .lock_one(
            sim_clock.now(),
            ComponentId::dummy_activity(),
            DEPLOYMENT_ID_DUMMY,
            &execution_id,
            RunId::generate(),
            Version::new(2), // After CancellationRequested event
            ExecutorId::generate(),
            sim_clock.now() + Duration::from_secs(30),
            ComponentRetryConfig::ZERO,
        )
        .await
        .unwrap_err();

    let reason = assert_matches!(err, DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::IllegalState { reason, .. }) => reason);
    assert_eq!("cannot lock, execution is cancelling", reason.as_ref());

    drop(db_connection);
    db_close.close().await;
}

/// A cancelling row must never be handed to a normal executor tick: the batch
/// pick-up queries carry `AND lifecycle = 'active'`. This exercises the SELECT
/// path, distinct from the `lock_one` rejection above.
#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn cancelling_execution_should_not_be_picked_up_by_pending_query(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    // Create ready to run, then request cancellation from PendingAt.
    create_at(
        db_connection.as_ref(),
        &sim_clock,
        &execution_id,
        sim_clock.now(),
    )
    .await;
    db_connection
        .append(
            execution_id.clone(),
            Version::new(1),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::CancellationRequested,
            },
        )
        .await
        .unwrap();

    let component_id = ComponentId::dummy_activity();
    let by_ffqns = db_connection
        .lock_pending_by_ffqns(
            1,
            sim_clock.now(),
            Arc::from([SOME_FFQN]),
            sim_clock.now(),
            component_id.clone(),
            DEPLOYMENT_ID_DUMMY,
            ExecutorId::generate(),
            sim_clock.now() + Duration::from_secs(30),
            RunId::generate(),
            ComponentRetryConfig::ZERO,
        )
        .await
        .unwrap();
    assert!(
        by_ffqns.is_empty(),
        "cancelling row must not be picked up by lock_pending_by_ffqns"
    );

    let by_digest = db_connection
        .lock_pending_by_component_digest(
            1,
            sim_clock.now(),
            &component_id,
            DEPLOYMENT_ID_DUMMY,
            sim_clock.now(),
            ExecutorId::generate(),
            sim_clock.now() + Duration::from_secs(30),
            RunId::generate(),
            ComponentRetryConfig::ZERO,
        )
        .await
        .unwrap();
    assert!(
        by_digest.is_empty(),
        "cancelling row must not be picked up by lock_pending_by_component_digest"
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn pause_then_join_next_then_unpause_should_restore_blocked_by_join_set(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();

    let execution_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    let lock_expiry = Duration::from_secs(30);
    let mut version = lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + lock_expiry,
        RunId::generate(),
    )
    .await;

    let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection.as_ref(),
    )
    .await;

    let child_execution_id = execution_id.next_level(&join_set_id);
    version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request: JoinSetRequest::ChildExecutionRequest {
                            child_execution_id,
                            target_ffqn: SOME_FFQN,
                            params: Params::empty(),
                            result: Ok(()),
                        },
                    },
                },
            },
        )
        .await
        .unwrap();

    let created_at = sim_clock.now();
    version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at,
                event: ExecutionRequest::Unlocked(Unlocked {
                    unlocked_at: created_at,
                    reason: "paused".into(),
                }),
            },
        )
        .await
        .unwrap();

    version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(log.pending_state, PendingState::Paused(..));

    version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinNext {
                        join_set_id: join_set_id.clone(),
                        run_expires_at: sim_clock.now() + lock_expiry,
                        closing: false,
                        requested_ffqn: Some(SOME_FFQN),
                    },
                },
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(
        log.pending_state,
        PendingState::Paused(PendingStatePaused::BlockedByJoinSet(
            PendingStateBlockedByJoinSet {
                join_set_id: actual_join_set_id,
                ..
            }
        )) if actual_join_set_id == join_set_id
    );

    db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Unpaused,
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(
        log.pending_state,
        PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet {
            join_set_id: actual_join_set_id,
            ..
        }) if actual_join_set_id == join_set_id
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn pause_with_pending_child_then_response_then_unpause_should_be_pending(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();

    let execution_id = ExecutionId::generate();
    // Create execution
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Lock the execution
    let lock_expiry = Duration::from_secs(30);
    let mut version = lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + lock_expiry,
        RunId::generate(),
    )
    .await;

    // Create join set
    let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection.as_ref(),
    )
    .await;

    // Submit child execution request
    let child_execution_id = execution_id.next_level(&join_set_id);
    db_connection
        .append(
            execution_id.clone(),
            version.clone(),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request: JoinSetRequest::ChildExecutionRequest {
                            child_execution_id: child_execution_id.clone(),
                            target_ffqn: SOME_FFQN,
                            params: Params::empty(),
                            result: Ok(()),
                        },
                    },
                },
            },
        )
        .await
        .unwrap();
    version = version.increment();

    // Append JoinNext - should make execution BlockedByJoinSet
    version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinNext {
                        join_set_id: join_set_id.clone(),
                        run_expires_at: sim_clock.now() + lock_expiry,
                        closing: false,
                        requested_ffqn: Some(SOME_FFQN),
                    },
                },
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(
        log.pending_state,
        PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet { .. })
    );

    // Pause the execution
    version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(log.pending_state, PendingState::Paused(..));

    // Append child execution response while paused
    db_connection
        .append_response(
            sim_clock.now(),
            execution_id.clone(),
            JoinSetResponseEvent {
                join_set_id: join_set_id.clone(),
                event: JoinSetResponse::ChildExecutionFinished {
                    child_execution_id,
                    finished_version: Version(1),
                    result: SupportedFunctionReturnValue::Ok(None),
                },
            },
        )
        .await
        .unwrap();

    // Should still be paused
    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(log.pending_state, PendingState::Paused(..));

    // Unpause - should be pending since response arrived
    db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Unpaused,
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(
        log.pending_state,
        PendingState::PendingAt(PendingStatePendingAt { .. })
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn pause_with_pending_delay_then_response_then_unpause_should_be_pending(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();

    let execution_id = ExecutionId::generate();
    // Create execution
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Lock the execution
    let lock_expiry = Duration::from_secs(30);
    let mut version = lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + lock_expiry,
        RunId::generate(),
    )
    .await;

    // Create join set
    let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection.as_ref(),
    )
    .await;

    // Submit delay request
    let delay_id = DelayId::new(&execution_id, &join_set_id);
    let delay_duration = Duration::from_mins(1);
    db_connection
        .append(
            execution_id.clone(),
            version.clone(),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request: JoinSetRequest::DelayRequest {
                            delay_id: delay_id.clone(),
                            expires_at: sim_clock.now() + delay_duration,
                            schedule_at: HistoryEventScheduleAt::In(delay_duration),
                            paused: false,
                        },
                    },
                },
            },
        )
        .await
        .unwrap();
    version = version.increment();

    // Append JoinNext - should make execution BlockedByJoinSet
    version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinNext {
                        join_set_id: join_set_id.clone(),
                        run_expires_at: sim_clock.now() + lock_expiry,
                        closing: false,
                        requested_ffqn: None,
                    },
                },
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(
        log.pending_state,
        PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet { .. })
    );

    // Pause the execution
    version = db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Paused,
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(log.pending_state, PendingState::Paused(..));

    // Append delay response while paused
    db_connection
        .append_response(
            sim_clock.now(),
            execution_id.clone(),
            JoinSetResponseEvent {
                join_set_id: join_set_id.clone(),
                event: JoinSetResponse::DelayFinished {
                    delay_id,
                    result: Ok(()),
                },
            },
        )
        .await
        .unwrap();

    // Should still be paused
    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(log.pending_state, PendingState::Paused(..));

    // Unpause - should be pending since response arrived
    db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Unpaused,
            },
        )
        .await
        .unwrap();

    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(
        log.pending_state,
        PendingState::PendingAt(PendingStatePendingAt { .. })
    );

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_backtrace(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let execution_id = ExecutionId::generate();

    let component_id = ComponentId::dummy_activity();

    // Create
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: component_id.clone(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();
    let lock_expiry = Duration::from_millis(100);
    let mut version = lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + lock_expiry, // lock expires at
        RunId::generate(),
    )
    .await;

    // Create joinset and a backtrace 1
    let join_set_id = JoinSetId::new(JoinSetKind::Named, "1".into()).unwrap();
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection.as_ref(),
    )
    .await;
    let backtrace_info_1_version = version.clone();
    let backtrace_info_1 = BacktraceInfo {
        execution_id: execution_id.clone(),
        component_id: component_id.clone(),
        version_min_including: backtrace_info_1_version.clone(),
        version_max_excluding: backtrace_info_1_version.increment(),
        wasm_backtrace: WasmBacktrace {
            frames: vec![FrameInfo {
                module: "m".to_string(),
                func_name: "f".to_string(),
                symbols: vec![FrameSymbol {
                    func_name: Some("f2".to_string()),
                    file: Some("file".to_string()),
                    line: Some(1),
                    col: Some(2),
                }],
            }],
        },
    };
    db_connection
        .append_backtrace(backtrace_info_1.clone())
        .await
        .unwrap();

    // Create joinset and a backtrace 2
    let join_set_id = JoinSetId::new(JoinSetKind::Named, "2".into()).unwrap();
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection.as_ref(),
    )
    .await;
    let backtrace_info_2_version = version.clone();
    let backtrace_info_2 = BacktraceInfo {
        execution_id: execution_id.clone(),
        component_id: component_id.clone(),
        version_min_including: backtrace_info_2_version.clone(),
        version_max_excluding: backtrace_info_2_version.increment(),
        wasm_backtrace: WasmBacktrace {
            frames: vec![FrameInfo {
                module: "m".to_string(),
                func_name: "f".to_string(),
                symbols: vec![],
            }],
        },
    };
    let inserted = db_connection
        .append_backtrace_batch(vec![backtrace_info_1.clone(), backtrace_info_2.clone()])
        .await
        .unwrap();
    assert_eq!(inserted, 1, "only the new backtrace should be inserted");
    let inserted = db_connection
        .append_backtrace_batch(vec![backtrace_info_1.clone(), backtrace_info_2.clone()])
        .await
        .unwrap();
    assert_eq!(inserted, 0, "repeated backtrace batches must be idempotent");

    let api_conn = db_pool.external_api_conn().await.unwrap();
    let found = api_conn
        .get_backtrace(&execution_id, BacktraceFilter::First)
        .await
        .unwrap();
    assert_eq!(backtrace_info_1, found);
    let found = api_conn
        .get_backtrace(
            &execution_id,
            BacktraceFilter::Specific(backtrace_info_1_version),
        )
        .await
        .unwrap();
    assert_eq!(backtrace_info_1, found);

    let found = api_conn
        .get_backtrace(&execution_id, BacktraceFilter::Last)
        .await
        .unwrap();
    assert_eq!(backtrace_info_2, found);
    let found = api_conn
        .get_backtrace(
            &execution_id,
            BacktraceFilter::Specific(backtrace_info_2_version),
        )
        .await
        .unwrap();
    assert_eq!(backtrace_info_2, found);

    drop(api_conn);
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn source_file_upsert_replaces_existing_mapping(database: Database) {
    set_up();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let api_conn = db_pool.external_api_conn().await.unwrap();
    let cas = db_pool.cas_conn().await.unwrap();
    let component_digest = ComponentId::dummy_activity().component_digest;

    // The source bytes live in the CAS; the DB only maps frame keys to digests.
    let upsert = async |frame_key: &str, is_suffix: bool, content: &str| {
        let digest = cas.write_blob(content.as_bytes()).await.unwrap();
        api_conn
            .upsert_source_mapping(&component_digest, frame_key, is_suffix, &digest)
            .await
            .unwrap();
    };
    upsert("exact.rs", false, "old exact").await;
    upsert("/src/suffix.rs", true, "old suffix").await;
    upsert("exact.rs", false, "new exact").await;
    upsert("/src/suffix.rs", true, "new suffix").await;

    let resolve = async |file: &str| {
        let digest = api_conn
            .resolve_source_digest(&component_digest, file)
            .await
            .unwrap()?;
        let bytes = cas.read_blob(&digest).await.unwrap().unwrap();
        Some(String::from_utf8(bytes).unwrap())
    };
    assert_eq!(Some("new exact".to_string()), resolve("exact.rs").await);
    assert_eq!(
        Some("new suffix".to_string()),
        resolve("/workspace/src/suffix.rs").await
    );

    drop(cas);
    drop(api_conn);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn wait_for_finished_result_should_fetch_before_racing_with_timeout(database: Database) {
    set_up();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();
    let sim_clock = SimClock::default();

    let execution_id = ExecutionId::generate();

    // Create
    let component_id = ComponentId::dummy_activity();
    let version = db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: component_id.clone(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Finish
    db_connection
        .append(
            execution_id.clone(),
            version,
            AppendRequest {
                event: ExecutionRequest::Finished {
                    retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                    http_client_traces: None,
                },
                created_at: sim_clock.now(),
            },
        )
        .await
        .unwrap();

    let res = db_connection
        .wait_for_finished_result(
            &execution_id,
            Some(Box::pin(std::future::ready(TimeoutOutcome::Cancel))),
        )
        .await
        .unwrap();
    assert_eq!(SUPPORTED_RETURN_VALUE_OK_EMPTY, res);

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn list_responses_empty_should_return_empty_list(database: Database) {
    // Regression test
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let api_conn = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();

    // Create execution
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    let list_result = api_conn
        .list_responses(
            &execution_id,
            Pagination::NewerThan {
                length: 10,
                cursor: 0,
                including_cursor: true,
            },
        )
        .await
        .unwrap();
    assert_eq!(0, list_result.responses.len());

    drop(api_conn);
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn list_responses_wrong_id_returns_empty_list(database: Database) {
    // Regression test
    set_up();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let api_conn = db_pool.external_api_conn().await.unwrap();
    let execution_id = ExecutionId::generate();

    let list_result = api_conn
        .list_responses(
            &execution_id,
            Pagination::NewerThan {
                length: 10,
                cursor: 0,
                including_cursor: true,
            },
        )
        .await
        .unwrap();
    assert_eq!(0, list_result.responses.len()); // According to contract NotFound is not returned.

    drop(api_conn);
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_list_responses(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let api_conn = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();

    // Create execution
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Lock and create join set
    let lock_expiry = Duration::from_millis(100);
    let mut version = lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + lock_expiry,
        RunId::generate(),
    )
    .await;
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection.as_ref(),
    )
    .await;

    // Add a delay request and response
    let delay_id = DelayId::new(&execution_id, &join_set_id);
    let delay_expires_at = sim_clock.now() + Duration::from_secs(10);
    db_connection
        .append(
            execution_id.clone(),
            version.clone(),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request: JoinSetRequest::DelayRequest {
                            delay_id: delay_id.clone(),
                            expires_at: delay_expires_at,
                            schedule_at: HistoryEventScheduleAt::Now,
                            paused: false,
                        },
                    },
                },
            },
        )
        .await
        .unwrap();
    db_connection
        .append_response(
            sim_clock.now(),
            execution_id.clone(),
            JoinSetResponseEvent {
                join_set_id: join_set_id.clone(),
                event: JoinSetResponse::DelayFinished {
                    delay_id: delay_id.clone(),
                    result: Ok(()),
                },
            },
        )
        .await
        .unwrap();

    // List responses - should have one response now
    let list_result = api_conn
        .list_responses(
            &execution_id,
            Pagination::NewerThan {
                length: 10,
                cursor: 0,
                including_cursor: true,
            },
        )
        .await
        .unwrap();
    assert_eq!(1, list_result.responses.len());
    assert_eq!(
        JoinSetResponseEvent {
            join_set_id: join_set_id.clone(),
            event: JoinSetResponse::DelayFinished {
                delay_id,
                result: Ok(()),
            },
        },
        list_result.responses[0].event.event
    );
    assert_eq!(ResponseCursor(1), list_result.responses[0].cursor);
    assert_eq!(ResponseCursor(1), list_result.max_cursor);

    drop(api_conn);
    drop(db_connection);
    db_close.close().await;
}

async fn append_delay_response(
    db_connection: &dyn DbConnectionTest,
    execution_id: &ExecutionId,
    join_set_id: &JoinSetId,
    version: &mut Version,
    index: u64,
    now: DateTime<Utc>,
) {
    let delay_id = DelayId::new_with_index(execution_id, join_set_id, index);
    *version = db_connection
        .append(
            execution_id.clone(),
            version.clone(),
            AppendRequest {
                created_at: now,
                event: ExecutionRequest::HistoryEvent {
                    event: HistoryEvent::JoinSetRequest {
                        join_set_id: join_set_id.clone(),
                        request: JoinSetRequest::DelayRequest {
                            delay_id: delay_id.clone(),
                            expires_at: now + Duration::from_secs(10),
                            schedule_at: HistoryEventScheduleAt::Now,
                            paused: false,
                        },
                    },
                },
            },
        )
        .await
        .unwrap();
    db_connection
        .append_response(
            now,
            execution_id.clone(),
            JoinSetResponseEvent {
                join_set_id: join_set_id.clone(),
                event: JoinSetResponse::DelayFinished {
                    delay_id,
                    result: Ok(()),
                },
            },
        )
        .await
        .unwrap();
}

/// Two executions with interleaved responses must each observe contiguous 1-based
/// cursors, independent of the global insertion order.
#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_list_responses_per_execution_cursor(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let api_conn = db_pool.external_api_conn().await.unwrap();

    // Create and lock two executions, each with its own join set. Create+lock
    // sequentially so `lock_pending_by_ffqns(1, ..)` targets the intended one.
    let lock_expiry = Duration::from_millis(100);
    let mut executions = Vec::new();
    for _ in 0..2 {
        let execution_id = ExecutionId::generate();
        let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
        db_connection
            .create(CreateRequest {
                created_at: sim_clock.now(),
                execution_id: execution_id.clone(),
                ffqn: SOME_FFQN,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: sim_clock.now(),
                component_id: ComponentId::dummy_activity(),
                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();
        let mut version = lock(
            db_connection.as_ref(),
            &execution_id,
            &sim_clock,
            ExecutorId::generate(),
            sim_clock.now() + lock_expiry,
            RunId::generate(),
        )
        .await;
        create_join_set(
            &execution_id,
            &mut version,
            join_set_id.clone(),
            sim_clock.now(),
            db_connection.as_ref(),
        )
        .await;
        executions.push((execution_id, join_set_id, version));
    }

    // Interleave three responses across the two executions: a, b, a, b, a, b.
    for index in 1..=3 {
        for (execution_id, join_set_id, version) in &mut executions {
            append_delay_response(
                db_connection.as_ref(),
                execution_id,
                join_set_id,
                version,
                index,
                sim_clock.now(),
            )
            .await;
        }
    }

    // Despite the interleaved global insertion order, each execution sees 1, 2, 3.
    for (execution_id, _, _) in &executions {
        let list_result = api_conn
            .list_responses(
                execution_id,
                Pagination::NewerThan {
                    length: 10,
                    cursor: 0,
                    including_cursor: true,
                },
            )
            .await
            .unwrap();
        let cursors: Vec<_> = list_result.responses.iter().map(|r| r.cursor).collect();
        assert_eq!(
            vec![ResponseCursor(1), ResponseCursor(2), ResponseCursor(3)],
            cursors
        );
        assert_eq!(ResponseCursor(3), list_result.max_cursor);
    }

    drop(api_conn);
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_list_responses_pagination_direction(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let api_conn = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();

    // Create execution
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Lock and create join set
    let lock_expiry = Duration::from_millis(100);
    let mut version = lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + lock_expiry,
        RunId::generate(),
    )
    .await;
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection.as_ref(),
    )
    .await;

    // Add 5 delay responses
    for i in 0u64..5 {
        let delay_id = DelayId::new_with_index(&execution_id, &join_set_id, i + 1);
        let delay_expires_at = sim_clock.now() + Duration::from_secs(10 + i);
        version = db_connection
            .append(
                execution_id.clone(),
                version.clone(),
                AppendRequest {
                    created_at: sim_clock.now(),
                    event: ExecutionRequest::HistoryEvent {
                        event: HistoryEvent::JoinSetRequest {
                            join_set_id: join_set_id.clone(),
                            request: JoinSetRequest::DelayRequest {
                                delay_id: delay_id.clone(),
                                expires_at: delay_expires_at,
                                schedule_at: HistoryEventScheduleAt::Now,
                                paused: false,
                            },
                        },
                    },
                },
            )
            .await
            .unwrap();
        db_connection
            .append_response(
                sim_clock.now(),
                execution_id.clone(),
                JoinSetResponseEvent {
                    join_set_id: join_set_id.clone(),
                    event: JoinSetResponse::DelayFinished {
                        delay_id,
                        result: Ok(()),
                    },
                },
            )
            .await
            .unwrap();
    }

    // Test NewerThan pagination (default, from oldest)
    let list_result = api_conn
        .list_responses(
            &execution_id,
            Pagination::NewerThan {
                length: 2,
                cursor: 0,
                including_cursor: true,
            },
        )
        .await
        .unwrap();
    assert_eq!(2, list_result.responses.len());
    assert_eq!(ResponseCursor(1), list_result.responses[0].cursor);
    assert_eq!(ResponseCursor(2), list_result.responses[1].cursor);
    assert_eq!(ResponseCursor(5), list_result.max_cursor);

    // Test NewerThan with cursor in the middle
    let list_result = api_conn
        .list_responses(
            &execution_id,
            Pagination::NewerThan {
                length: 2,
                cursor: 2,
                including_cursor: false,
            },
        )
        .await
        .unwrap();
    assert_eq!(2, list_result.responses.len());
    assert_eq!(ResponseCursor(3), list_result.responses[0].cursor);
    assert_eq!(ResponseCursor(4), list_result.responses[1].cursor);

    // Test OlderThan pagination (backwards, from newest)
    let list_result = api_conn
        .list_responses(
            &execution_id,
            Pagination::OlderThan {
                length: 2,
                cursor: u32::MAX,
                including_cursor: true,
            },
        )
        .await
        .unwrap();
    assert_eq!(2, list_result.responses.len());
    // Results are always in ascending order (oldest to newest)
    assert_eq!(ResponseCursor(4), list_result.responses[0].cursor);
    assert_eq!(ResponseCursor(5), list_result.responses[1].cursor);

    // Test OlderThan with cursor in the middle
    let list_result = api_conn
        .list_responses(
            &execution_id,
            Pagination::OlderThan {
                length: 2,
                cursor: 4,
                including_cursor: false,
            },
        )
        .await
        .unwrap();
    assert_eq!(2, list_result.responses.len());
    // Results are always in ascending order (oldest to newest)
    assert_eq!(ResponseCursor(2), list_result.responses[0].cursor);
    assert_eq!(ResponseCursor(3), list_result.responses[1].cursor);

    // Test OlderThan with including_cursor=true
    let list_result = api_conn
        .list_responses(
            &execution_id,
            Pagination::OlderThan {
                length: 2,
                cursor: 4,
                including_cursor: true,
            },
        )
        .await
        .unwrap();
    assert_eq!(2, list_result.responses.len());
    assert_eq!(ResponseCursor(3), list_result.responses[0].cursor);
    assert_eq!(ResponseCursor(4), list_result.responses[1].cursor);

    drop(api_conn);
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn test_list_execution_events_pagination_direction(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let api_conn = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();

    // Create execution - this creates version 0
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    // Lock and create join set - this adds more events
    let lock_expiry = Duration::from_millis(100);
    let mut version = lock(
        db_connection.as_ref(),
        &execution_id,
        &sim_clock,
        ExecutorId::generate(),
        sim_clock.now() + lock_expiry,
        RunId::generate(),
    )
    .await;
    create_join_set(
        &execution_id,
        &mut version,
        join_set_id.clone(),
        sim_clock.now(),
        db_connection.as_ref(),
    )
    .await;

    // Add 5 more events (delay requests)
    for i in 0u64..5 {
        let delay_id = DelayId::new_with_index(&execution_id, &join_set_id, i + 1);
        let delay_expires_at = sim_clock.now() + Duration::from_secs(10 + i);
        version = db_connection
            .append(
                execution_id.clone(),
                version.clone(),
                AppendRequest {
                    created_at: sim_clock.now(),
                    event: ExecutionRequest::HistoryEvent {
                        event: HistoryEvent::JoinSetRequest {
                            join_set_id: join_set_id.clone(),
                            request: JoinSetRequest::DelayRequest {
                                delay_id,
                                expires_at: delay_expires_at,
                                schedule_at: HistoryEventScheduleAt::Now,
                                paused: false,
                            },
                        },
                    },
                },
            )
            .await
            .unwrap();
    }

    // Get the max version for reference
    let all_events = api_conn
        .list_execution_events(
            &execution_id,
            Pagination::NewerThan {
                length: 100,
                cursor: 0,
                including_cursor: true,
            },
            false,
        )
        .await
        .unwrap();
    let max_version = all_events.max_version;
    let total_events = all_events.events.len();
    info!("Total events: {total_events}, max_version: {max_version:?}");

    // Test NewerThan pagination (default, from oldest)
    let list_result = api_conn
        .list_execution_events(
            &execution_id,
            Pagination::NewerThan {
                length: 3,
                cursor: 0,
                including_cursor: true,
            },
            false,
        )
        .await
        .unwrap();
    assert_eq!(3, list_result.events.len());
    assert_eq!(Version(0), list_result.events[0].version);
    assert_eq!(Version(1), list_result.events[1].version);
    assert_eq!(Version(2), list_result.events[2].version);

    // Test NewerThan with cursor in the middle (not including cursor)
    let list_result = api_conn
        .list_execution_events(
            &execution_id,
            Pagination::NewerThan {
                length: 2,
                cursor: 2,
                including_cursor: false,
            },
            false,
        )
        .await
        .unwrap();
    assert_eq!(2, list_result.events.len());
    assert_eq!(Version(3), list_result.events[0].version);
    assert_eq!(Version(4), list_result.events[1].version);

    // Test OlderThan pagination (backwards, from newest)
    let list_result = api_conn
        .list_execution_events(
            &execution_id,
            Pagination::OlderThan {
                length: 3,
                cursor: VersionType::MAX,
                including_cursor: true,
            },
            false,
        )
        .await
        .unwrap();
    assert_eq!(3, list_result.events.len());
    // Results are always in ascending order (oldest to newest)
    let last_version = max_version.0;
    assert_eq!(Version(last_version - 2), list_result.events[0].version);
    assert_eq!(Version(last_version - 1), list_result.events[1].version);
    assert_eq!(Version(last_version), list_result.events[2].version);

    // Test OlderThan with cursor in the middle
    let list_result = api_conn
        .list_execution_events(
            &execution_id,
            Pagination::OlderThan {
                length: 2,
                cursor: 4,
                including_cursor: false,
            },
            false,
        )
        .await
        .unwrap();
    assert_eq!(2, list_result.events.len());
    // Results are always in ascending order (oldest to newest)
    assert_eq!(Version(2), list_result.events[0].version);
    assert_eq!(Version(3), list_result.events[1].version);

    // Test OlderThan with including_cursor=true
    let list_result = api_conn
        .list_execution_events(
            &execution_id,
            Pagination::OlderThan {
                length: 2,
                cursor: 4,
                including_cursor: true,
            },
            false,
        )
        .await
        .unwrap();
    assert_eq!(2, list_result.events.len());
    assert_eq!(Version(3), list_result.events[0].version);
    assert_eq!(Version(4), list_result.events[1].version);

    drop(api_conn);
    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn deployment_insert_and_get(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let api_conn = db_pool.external_api_conn().await.unwrap();

    let deployment_id = concepts::prefixed_ulid::DeploymentId::generate();
    let now = sim_clock.now();
    let expected_digest = DeploymentRecord::compute_digest(r#"{"activities":[]}"#);
    let record = DeploymentRecord {
        deployment_id,
        description: Some("test deployment".to_string()),
        digest: expected_digest.clone(),
        created_at: now,
        last_active_at: None,
        status: DeploymentStatus::Inactive,
        deployment_toml: r#"{"activities":[]}"#.to_string(),
        obelisk_version: "0.0.0-test".to_string(),
        created_by: Some("test".to_string()),
        files: Vec::new(),
    };
    api_conn.insert_deployment(record).await.unwrap();

    let fetched = api_conn
        .get_deployment(deployment_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(deployment_id, fetched.deployment_id);
    assert_eq!(DeploymentStatus::Inactive, fetched.status);
    assert_eq!("0.0.0-test", fetched.obelisk_version);
    assert_eq!(Some("test".to_string()), fetched.created_by);
    assert_eq!(Some("test deployment".to_string()), fetched.description);
    assert_eq!(expected_digest, fetched.digest);
    assert!(fetched.last_active_at.is_none());

    drop(api_conn);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn deployment_files_roundtrip_and_missing_digests(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let api_conn = db_pool.external_api_conn().await.unwrap();
    let cas_conn = db_pool.cas_conn().await.unwrap();

    let deployment_id = concepts::prefixed_ulid::DeploymentId::generate();
    let now = sim_clock.now();
    let file_content = b"hello deployment file";
    let file_digest = DeploymentRecord::compute_digest("hello deployment file");
    // Before the blob is uploaded to the CAS, its size is not yet known (0).
    let file = DeploymentFileRecord {
        path: "scripts/a.js".to_string(),
        digest: file_digest.clone(),
        size: 0,
    };
    api_conn
        .insert_deployment(DeploymentRecord {
            deployment_id,
            description: None,
            digest: DeploymentRecord::compute_digest("{}"),
            created_at: now,
            last_active_at: None,
            status: DeploymentStatus::Inactive,
            deployment_toml: "{}".to_string(),
            obelisk_version: "0.0.0-test".to_string(),
            created_by: None,
            files: vec![file.clone()],
        })
        .await
        .unwrap();

    assert_eq!(
        vec![file.clone()],
        api_conn.list_deployment_files(deployment_id).await.unwrap()
    );
    assert_eq!(
        vec![file_digest.clone()],
        api_conn.missing_digests(deployment_id).await.unwrap()
    );
    assert!(!cas_conn.contains_blob(&file_digest).await.unwrap());
    assert!(cas_conn.read_blob(&file_digest).await.unwrap().is_none());

    // Content-addressed and idempotent: the digest is computed from the bytes,
    // and storing the same blob twice is a no-op.
    assert_eq!(
        file_digest,
        cas_conn.write_blob(file_content).await.unwrap()
    );
    assert_eq!(
        file_digest,
        cas_conn.write_blob(file_content).await.unwrap()
    );
    assert!(cas_conn.contains_blob(&file_digest).await.unwrap());
    assert_eq!(
        Some(file_content.to_vec()),
        cas_conn.read_blob(&file_digest).await.unwrap()
    );
    assert!(
        api_conn
            .missing_digests(deployment_id)
            .await
            .unwrap()
            .is_empty()
    );
    // After the blob is uploaded, the listed file reports its byte size.
    let file_with_size = DeploymentFileRecord {
        size: u64::try_from(file_content.len()).unwrap(),
        ..file
    };
    assert_eq!(
        vec![file_with_size],
        api_conn
            .get_deployment(deployment_id)
            .await
            .unwrap()
            .unwrap()
            .files
    );

    drop(cas_conn);
    drop(api_conn);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn deployment_activate(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let api_conn = db_pool.external_api_conn().await.unwrap();

    // No active deployment initially.
    assert!(api_conn.get_active_deployment().await.unwrap().is_none());

    let deployment_id = concepts::prefixed_ulid::DeploymentId::generate();
    let now = sim_clock.now();
    api_conn
        .insert_deployment(DeploymentRecord {
            deployment_id,
            description: None,
            digest: DeploymentRecord::compute_digest("{}"),
            created_at: now,
            last_active_at: None,
            status: DeploymentStatus::Inactive,
            deployment_toml: "{}".to_string(),
            obelisk_version: "0.0.0-test".to_string(),
            created_by: None,
            files: Vec::new(),
        })
        .await
        .unwrap();

    api_conn
        .activate_deployment(deployment_id, now)
        .await
        .unwrap();

    let active = api_conn.get_active_deployment().await.unwrap().unwrap();
    assert_eq!(deployment_id, active.deployment_id);
    assert_eq!(DeploymentStatus::Active, active.status);
    assert!(active.last_active_at.is_some());
    assert!(active.last_active_at.unwrap() >= now);

    drop(api_conn);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn deployment_only_one_active_allowed(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let api_conn = db_pool.external_api_conn().await.unwrap();

    let now = sim_clock.now();
    let id1 = concepts::prefixed_ulid::DeploymentId::generate();
    api_conn
        .insert_deployment(DeploymentRecord {
            deployment_id: id1,
            description: None,
            digest: DeploymentRecord::compute_digest("{}"),
            created_at: now,
            last_active_at: None,
            status: DeploymentStatus::Inactive,
            deployment_toml: "{}".to_string(),
            obelisk_version: "0.0.0-test".to_string(),
            created_by: None,
            files: Vec::new(),
        })
        .await
        .unwrap();
    api_conn.activate_deployment(id1, now).await.unwrap();

    // Activating a second deployment must deactivate the first.
    let id2 = concepts::prefixed_ulid::DeploymentId::generate();
    api_conn
        .insert_deployment(DeploymentRecord {
            deployment_id: id2,
            description: None,
            digest: DeploymentRecord::compute_digest("{}"),
            created_at: now,
            last_active_at: None,
            status: DeploymentStatus::Inactive,
            deployment_toml: "{}".to_string(),
            obelisk_version: "0.0.0-test".to_string(),
            created_by: None,
            files: Vec::new(),
        })
        .await
        .unwrap();
    api_conn.activate_deployment(id2, now).await.unwrap();

    // Only id2 is active now.
    let active = api_conn.get_active_deployment().await.unwrap().unwrap();
    assert_eq!(id2, active.deployment_id);

    // id1 must be inactive.
    let d1 = api_conn.get_deployment(id1).await.unwrap().unwrap();
    assert_eq!(DeploymentStatus::Inactive, d1.status);

    drop(api_conn);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn deployment_enqueue_active_clears_pending(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let api_conn = db_pool.external_api_conn().await.unwrap();

    let now = sim_clock.now();
    let active_id = concepts::prefixed_ulid::DeploymentId::generate();
    let pending_id = concepts::prefixed_ulid::DeploymentId::generate();
    for id in [active_id, pending_id] {
        api_conn
            .insert_deployment(DeploymentRecord {
                deployment_id: id,
                description: None,
                digest: DeploymentRecord::compute_digest("{}"),
                created_at: now,
                last_active_at: None,
                status: DeploymentStatus::Inactive,
                deployment_toml: "{}".to_string(),
                obelisk_version: "0.0.0-test".to_string(),
                created_by: None,
                files: Vec::new(),
            })
            .await
            .unwrap();
    }

    api_conn.activate_deployment(active_id, now).await.unwrap();
    assert_eq!(
        EnqueueOutcome::Enqueued,
        api_conn.enqueue_deployment(pending_id).await.unwrap()
    );
    assert_eq!(
        pending_id,
        api_conn
            .get_current_deployment()
            .await
            .unwrap()
            .unwrap()
            .deployment_id
    );

    assert_eq!(
        EnqueueOutcome::AlreadyActive,
        api_conn.enqueue_deployment(active_id).await.unwrap()
    );

    let active = api_conn.get_deployment(active_id).await.unwrap().unwrap();
    let pending = api_conn.get_deployment(pending_id).await.unwrap().unwrap();
    assert_eq!(DeploymentStatus::Active, active.status);
    assert_eq!(DeploymentStatus::Inactive, pending.status);
    assert_eq!(
        active_id,
        api_conn
            .get_current_deployment()
            .await
            .unwrap()
            .unwrap()
            .deployment_id
    );

    drop(api_conn);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn deployment_list(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let api_conn = db_pool.external_api_conn().await.unwrap();

    let now = sim_clock.now();
    let id1 = concepts::prefixed_ulid::DeploymentId::generate();
    let id2 = concepts::prefixed_ulid::DeploymentId::generate();
    for id in [id1, id2] {
        api_conn
            .insert_deployment(DeploymentRecord {
                deployment_id: id,
                description: None,
                digest: DeploymentRecord::compute_digest("{}"),
                created_at: now,
                last_active_at: None,
                status: DeploymentStatus::Inactive,
                deployment_toml: "{}".to_string(),
                obelisk_version: "0.0.0-test".to_string(),
                created_by: None,
                files: Vec::new(),
            })
            .await
            .unwrap();
    }

    let all = api_conn
        .list_deployments(Pagination::NewerThan {
            length: 10,
            cursor: None,
            including_cursor: false,
        })
        .await
        .unwrap();
    assert_eq!(2, all.len());

    drop(api_conn);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn list_logs_with_show_derived(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let api_conn = db_pool.external_api_conn().await.unwrap();

    let parent_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: parent_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    sim_clock.move_time_forward(Duration::from_secs(1));
    db_connection
        .append_log(LogInfoAppendRow {
            execution_id: parent_id.clone(),
            run_id: RunId::generate(),
            log_entry: LogEntry::Log {
                created_at: sim_clock.now(),
                level: concepts::storage::LogLevel::Info,
                message: "parent log".to_string(),
            },
        })
        .await
        .unwrap();

    sim_clock.move_time_forward(Duration::from_secs(1));
    let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    let child_id = parent_id.next_level(&join_set_id);
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: ExecutionId::Derived(child_id.clone()),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    db_connection
        .append_log(LogInfoAppendRow {
            execution_id: ExecutionId::Derived(child_id.clone()),
            run_id: RunId::generate(),
            log_entry: LogEntry::Log {
                created_at: sim_clock.now(),
                level: concepts::storage::LogLevel::Error,
                message: "child log".to_string(),
            },
        })
        .await
        .unwrap();

    sim_clock.move_time_forward(Duration::from_secs(1));
    let join_set_id_2 = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
    let grandchild_id = child_id.next_level(&join_set_id_2);
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: ExecutionId::Derived(grandchild_id.clone()),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    db_connection
        .append_log(LogInfoAppendRow {
            execution_id: ExecutionId::Derived(grandchild_id.clone()),
            run_id: RunId::generate(),
            log_entry: LogEntry::Log {
                created_at: sim_clock.now(),
                level: concepts::storage::LogLevel::Warn,
                message: "grandchild log".to_string(),
            },
        })
        .await
        .unwrap();

    sim_clock.move_time_forward(Duration::from_secs(1));

    let result = api_conn
        .list_logs(
            &parent_id,
            false,
            LogFilter::show_logs(Vec::new()),
            Pagination::OlderThan {
                length: 10,
                cursor: LogCursor(i64::MAX),
                including_cursor: false,
            },
        )
        .await
        .unwrap();

    assert_eq!(result.items.len(), 1);
    assert_eq!(result.items[0].execution_id, parent_id);
    assert_matches!(
        &result.items[0].log_entry,
        LogEntry::Log { message, .. } if message == "parent log"
    );

    let result = api_conn
        .list_logs(
            &parent_id,
            true,
            LogFilter::show_logs(Vec::new()),
            Pagination::OlderThan {
                length: 10,
                cursor: LogCursor(i64::MAX),
                including_cursor: false,
            },
        )
        .await
        .unwrap();

    assert_eq!(result.items.len(), 3);
    assert_eq!(
        result.items[0].execution_id,
        ExecutionId::Derived(grandchild_id.clone())
    );
    assert_matches!(
        &result.items[0].log_entry,
        LogEntry::Log { message, .. } if message == "grandchild log"
    );
    assert_eq!(
        result.items[1].execution_id,
        ExecutionId::Derived(child_id.clone())
    );
    assert_matches!(
        &result.items[1].log_entry,
        LogEntry::Log { message, .. } if message == "child log"
    );
    assert_eq!(result.items[2].execution_id, parent_id);
    assert_matches!(
        &result.items[2].log_entry,
        LogEntry::Log { message, .. } if message == "parent log"
    );

    let result = api_conn
        .list_logs(
            &parent_id,
            true,
            LogFilter::show_logs(Vec::new()),
            Pagination::OlderThan {
                length: 1,
                cursor: LogCursor(i64::MAX),
                including_cursor: false,
            },
        )
        .await
        .unwrap();

    assert_eq!(result.items.len(), 1);
    let first_cursor = result.items[0].cursor;

    let result = api_conn
        .list_logs(
            &parent_id,
            true,
            LogFilter::show_logs(Vec::new()),
            Pagination::OlderThan {
                length: 1,
                cursor: first_cursor,
                including_cursor: false,
            },
        )
        .await
        .unwrap();

    assert_eq!(result.items.len(), 1);
    assert_matches!(
        &result.items[0].log_entry,
        LogEntry::Log { message, .. } if message == "child log"
    );

    drop(db_connection);
    drop(api_conn);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn list_logs_paginates_equal_timestamps(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection_test().await.unwrap();
    let api_conn = db_pool.external_api_conn().await.unwrap();

    let execution_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    let created_at = sim_clock.now();
    for message in ["first", "second", "third"] {
        db_connection
            .append_log(LogInfoAppendRow {
                execution_id: execution_id.clone(),
                run_id: RunId::generate(),
                log_entry: LogEntry::Log {
                    created_at,
                    level: concepts::storage::LogLevel::Info,
                    message: message.to_string(),
                },
            })
            .await
            .unwrap();
    }

    let mut pagination = Pagination::NewerThan {
        length: 1,
        cursor: LogCursor(i64::MIN),
        including_cursor: false,
    };
    let mut messages = Vec::new();
    for _ in 0..3 {
        let result = api_conn
            .list_logs(
                &execution_id,
                false,
                LogFilter::show_logs(Vec::new()),
                pagination,
            )
            .await
            .unwrap();
        assert_eq!(result.items.len(), 1);
        let LogEntry::Log { message, .. } = &result.items[0].log_entry else {
            panic!("expected log entry");
        };
        messages.push(message.clone());
        pagination = result.next_page;
    }
    assert_eq!(messages, ["first", "second", "third"]);

    let mut pagination = Pagination::OlderThan {
        length: 1,
        cursor: LogCursor(i64::MAX),
        including_cursor: false,
    };
    let mut messages = Vec::new();
    for _ in 0..3 {
        let result = api_conn
            .list_logs(
                &execution_id,
                false,
                LogFilter::show_logs(Vec::new()),
                pagination,
            )
            .await
            .unwrap();
        assert_eq!(result.items.len(), 1);
        let LogEntry::Log { message, .. } = &result.items[0].log_entry else {
            panic!("expected log entry");
        };
        messages.push(message.clone());
        pagination = result.prev_page.unwrap();
    }
    assert_eq!(messages, ["third", "second", "first"]);

    drop(db_connection);
    drop(api_conn);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn create_paused_workflow(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let execution_id = ExecutionId::generate();
    let scheduled_at = sim_clock.now();
    // Create with paused: true
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: execution_id.clone(),
            ffqn: SOME_FFQN,
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at,
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: true,
        })
        .await
        .unwrap();

    // Verify state is paused immediately after creation
    let log = db_connection.get(&execution_id).await.unwrap();
    assert_matches!(log.pending_state, PendingState::Paused(..));

    // Paused execution cannot be locked
    let err = db_connection
        .lock_one(
            sim_clock.now(),
            ComponentId::dummy_activity(),
            DEPLOYMENT_ID_DUMMY,
            &execution_id,
            RunId::generate(),
            Version::new(2),
            ExecutorId::generate(),
            sim_clock.now() + Duration::from_secs(30),
            ComponentRetryConfig::ZERO,
        )
        .await
        .unwrap_err();
    let reason = assert_matches!(err, DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::IllegalState { reason, .. }) => reason);
    assert_eq!("cannot lock, execution is paused", reason.as_ref());

    // Unpause the execution
    db_connection
        .append(
            execution_id.clone(),
            Version::new(2),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::Unpaused,
            },
        )
        .await
        .unwrap();

    // Verify state is now pending at the original scheduled time
    let log = db_connection.get(&execution_id).await.unwrap();
    if let PendingState::PendingAt(PendingStatePendingAt {
        scheduled_at: actual,
        ..
    }) = log.pending_state
    {
        assert_eq!(actual, scheduled_at);
    } else {
        panic!("Expected PendingAt state, got {:?}", log.pending_state);
    }

    drop(db_connection);
    db_close.close().await;
}

#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn list_executions_filters_by_versioned_package_name(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let matching_execution_id = ExecutionId::generate();
    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: matching_execution_id.clone(),
            ffqn: FunctionFqn::new_static(
                "obelisk-client:api-http/executions@1.0.0-beta",
                "generate",
            ),
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    db_connection
        .create(CreateRequest {
            created_at: sim_clock.now(),
            execution_id: ExecutionId::generate(),
            ffqn: FunctionFqn::new_static(
                "obelisk-client:api-http/executions@2.0.0-beta",
                "generate",
            ),
            params: Params::empty(),
            parent: None,
            metadata: concepts::ExecutionMetadata::empty(),
            scheduled_at: sim_clock.now(),
            component_id: ComponentId::dummy_activity(),
            deployment_id: DEPLOYMENT_ID_DUMMY,
            scheduled_by: None,
            paused: false,
        })
        .await
        .unwrap();

    drop(db_connection);

    let api_conn = db_pool.external_api_conn().await.unwrap();
    let executions = api_conn
        .list_executions(
            ListExecutionsFilter {
                function_name_filter: Some(FunctionNameFilter::PackageName(
                    "obelisk-client:api-http@1.0.0-beta".to_string(),
                )),
                ..Default::default()
            },
            ExecutionListPagination::default(),
        )
        .await
        .unwrap();

    assert_eq!(1, executions.len());
    assert_eq!(matching_execution_id, executions[0].execution_id);
    assert_eq!(
        "obelisk-client:api-http/executions@1.0.0-beta.generate",
        executions[0].ffqn.to_string()
    );

    drop(api_conn);
    db_close.close().await;
}

/// Regression: every state filter renders SQL against `t_state`, which no longer
/// has the `is_paused` column (V12 replaced it with `lifecycle`). Exercise each
/// filter so a stale column reference fails loudly instead of only in production.
#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn list_executions_filters_by_state(database: Database) {
    set_up();
    let sim_clock = SimClock::default();
    let (_guard, db_pool, db_close) = database.set_up().await;
    let db_connection = db_pool.connection().await.unwrap();

    let create = |execution_id: ExecutionId| CreateRequest {
        created_at: sim_clock.now(),
        execution_id,
        ffqn: SOME_FFQN,
        params: Params::empty(),
        parent: None,
        metadata: concepts::ExecutionMetadata::empty(),
        scheduled_at: sim_clock.now(),
        component_id: ComponentId::dummy_activity(),
        deployment_id: DEPLOYMENT_ID_DUMMY,
        scheduled_by: None,
        paused: false,
    };

    let pending_id = ExecutionId::generate();
    db_connection
        .create(create(pending_id.clone()))
        .await
        .unwrap();
    let paused_id = ExecutionId::generate();
    db_connection
        .create(create(paused_id.clone()))
        .await
        .unwrap();
    let cancelling_id = ExecutionId::generate();
    db_connection
        .create(create(cancelling_id.clone()))
        .await
        .unwrap();
    db_connection
        .append(
            cancelling_id.clone(),
            Version::new(1),
            AppendRequest {
                created_at: sim_clock.now(),
                event: ExecutionRequest::CancellationRequested,
            },
        )
        .await
        .unwrap();
    drop(db_connection);

    let api_conn = db_pool.external_api_conn().await.unwrap();
    api_conn
        .pause_execution(&paused_id, sim_clock.now())
        .await
        .unwrap();

    let list = |state_filter: ExecutionStateFilter| {
        let api_conn = &api_conn;
        async move {
            api_conn
                .list_executions(
                    ListExecutionsFilter {
                        state_filters: vec![state_filter],
                        ..Default::default()
                    },
                    ExecutionListPagination::default(),
                )
                .await
                .unwrap()
                .into_iter()
                .map(|e| e.execution_id)
                .collect::<Vec<_>>()
        }
    };

    let now = sim_clock.now();
    // Each of these previously panicked on a missing `is_paused` column.
    assert_eq!(
        vec![pending_id.clone()],
        list(ExecutionStateFilter::Pending { now }).await
    );
    assert_eq!(
        vec![paused_id.clone()],
        list(ExecutionStateFilter::Paused).await
    );
    assert_eq!(
        vec![cancelling_id.clone()],
        list(ExecutionStateFilter::Cancelling).await
    );
    // Paused and cancelling rows are excluded from the active buckets (lifecycle = 'active').
    assert!(
        list(ExecutionStateFilter::Scheduled { now })
            .await
            .is_empty()
    );
    assert!(list(ExecutionStateFilter::Locked).await.is_empty());
    assert!(list(ExecutionStateFilter::Blocked).await.is_empty());
    assert!(list(ExecutionStateFilter::Finished).await.is_empty());

    drop(api_conn);
    db_close.close().await;
}
