#[cfg(test)]
mod tests {
    use std::future::Future;
    use std::sync::Arc;
    use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

    use pd_proto_tso::{KeyspaceGroup, RequestHeader, TsoRequest};
    use pd_service_tso::{
        AllocatorConfig, GrpcCode, InMemoryEtcdStore, KeyspaceGroupManager, TsoRuntime,
        TsoRuntimeConfig, TsoService,
    };

    const TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN: &str = "tso count should be positive";

    fn default_group() -> KeyspaceGroup {
        KeyspaceGroup {
            id: 0,
            user_kind: "basic".to_string(),
            split_state: None,
            merge_state: None,
            members: Vec::new(),
            keyspaces: vec![0],
        }
    }

    fn build_service_with_primary(cluster_id: u64, now_ms: i64) -> TsoService {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager =
            KeyspaceGroupManager::new(cluster_id, store, AllocatorConfig::default(), "node");
        manager
            .register_group(default_group())
            .expect("register default group");
        manager
            .campaign_group_primary(0, now_ms)
            .expect("campaign primary");
        TsoService::new(cluster_id, manager)
    }

    fn tso_request(cluster_id: u64, count: u32) -> TsoRequest {
        TsoRequest {
            header: RequestHeader {
                cluster_id,
                keyspace_id: 0,
                keyspace_group_id: 0,
                sender_id: 0,
            },
            count,
            dc_location: String::new(),
        }
    }

    async fn wait_until<F, Fut>(
        label: &str,
        timeout: Duration,
        poll_interval: Duration,
        mut predicate: F,
    ) where
        F: FnMut() -> Fut,
        Fut: Future<Output = bool>,
    {
        let deadline = Instant::now() + timeout;
        loop {
            if predicate().await {
                return;
            }
            if Instant::now() >= deadline {
                panic!("timeout while waiting for {label}");
            }
            tokio::time::sleep(poll_interval).await;
        }
    }

    #[test]
    fn chaos_forced_primary_stepdown_blocks_tso_until_recampaign_and_preserves_monotonicity() {
        let service = build_service_with_primary(1, now_ms());

        let before = service
            .tso(now_ms(), tso_request(1, 1))
            .expect("initial tso");

        service
            .manager_mut()
            .step_down_group_primary(0)
            .expect("force primary step-down");
        let after_step_down = service
            .tso(now_ms(), tso_request(1, 1))
            .expect_err("step-down should reject tso requests");
        assert_eq!(after_step_down.code, GrpcCode::Unknown);
        assert!(
            after_step_down.message.contains("is not served")
                || after_step_down.message.contains("not leader")
                || after_step_down.message.contains("not primary")
                || after_step_down.message.contains("no leader"),
            "unexpected step-down rejection message: {}",
            after_step_down.message
        );

        service
            .manager_mut()
            .campaign_group_primary(0, now_ms())
            .expect("re-campaign primary");
        let recovered = service
            .tso(now_ms(), tso_request(1, 1))
            .expect("tso after recovery");

        assert!(
            recovered.timestamp.compose() > before.timestamp.compose(),
            "timestamp sequence must remain monotonic across forced primary flap"
        );
    }

    #[test]
    fn chaos_lease_expiration_blocks_tso_until_keepalive_and_preserves_monotonicity() {
        let start_ms = 10_000;
        let service = build_service_with_primary(1, start_ms);

        let before = service
            .tso(start_ms, tso_request(1, 1))
            .expect("initial tso");

        let lease_ms = AllocatorConfig::default().lease_ms;
        let expired_now = start_ms + lease_ms + 1;
        let expired = service
            .tso(expired_now, tso_request(1, 1))
            .expect_err("expired lease should reject tso request");
        assert_eq!(expired.code, GrpcCode::Unknown);
        assert!(
            expired.message.contains("is not served")
                || expired.message.contains("not leader")
                || expired.message.contains("not primary")
                || expired.message.contains("no leader"),
            "unexpected lease-expiry rejection message: {}",
            expired.message
        );

        service
            .manager_mut()
            .keep_group_primary_alive(0, expired_now)
            .expect("keepalive after lease expiry");
        let recovered = service
            .tso(expired_now + 1, tso_request(1, 1))
            .expect("tso after keepalive");
        assert!(
            recovered.timestamp.compose() > before.timestamp.compose(),
            "timestamp sequence must remain monotonic after lease-expiry recovery"
        );
    }

    #[test]
    fn chaos_process_pause_simulation_forces_lease_timeout_then_recovers() {
        let start_ms = 50_000;
        let service = build_service_with_primary(1, start_ms);

        let before = service
            .tso(start_ms, tso_request(1, 1))
            .expect("initial tso");

        // Simulate a long process pause (`SIGSTOP`-style) by advancing time beyond lease TTL.
        let lease_ms = AllocatorConfig::default().lease_ms;
        let paused_until_ms = start_ms + lease_ms.saturating_mul(3);
        let timeout = service
            .tso(paused_until_ms, tso_request(1, 1))
            .expect_err("long pause should expire lease and reject tso");
        assert_eq!(timeout.code, GrpcCode::Unknown);

        service
            .manager_mut()
            .keep_group_primary_alive(0, paused_until_ms)
            .expect("refresh lease after pause");
        let recovered = service
            .tso(paused_until_ms + 1, tso_request(1, 1))
            .expect("tso after simulated pause recovery");
        assert!(
            recovered.timestamp.compose() > before.timestamp.compose(),
            "timestamp sequence must remain monotonic after pause-induced lease timeout"
        );
    }

    #[test]
    fn chaos_clock_jump_backward_still_preserves_monotonicity() {
        let now = 20_000;
        let service = build_service_with_primary(1, now);

        let first = service
            .tso(now + 1, tso_request(1, 1))
            .expect("first tso at higher physical time");
        let second = service
            .tso(now - 500, tso_request(1, 1))
            .expect("second tso at lower physical time");
        let third = service
            .tso(now + 2, tso_request(1, 8))
            .expect("third tso after clock jump recovery");

        assert!(
            second.timestamp.compose() > first.timestamp.compose(),
            "clock rollback must not break monotonic sequence"
        );
        assert!(
            third.timestamp.compose() > second.timestamp.compose(),
            "monotonicity must continue after rollback handling"
        );
    }

    #[test]
    fn chaos_membership_resync_gate_rejects_ingress_then_recovers() {
        let service = build_service_with_primary(1, 10_000);
        service.set_membership_resync_blocking(true);

        let blocked = service
            .tso(10_100, tso_request(1, 1))
            .expect_err("membership resync gate must reject ingress");
        assert_eq!(blocked.code, GrpcCode::Unavailable);
        assert_eq!(blocked.retry_after_ms, Some(100));
        assert!(blocked.message.contains("membership resync"));

        service.set_membership_resync_blocking(false);
        let recovered = service
            .tso(10_200, tso_request(1, 1))
            .expect("tso should recover after membership resync gate clears");
        assert_eq!(recovered.count, 1);
    }

    #[test]
    fn chaos_malformed_zero_count_request_is_rejected_with_parity_token() {
        let service = build_service_with_primary(1, 10_000);

        let err = service
            .tso(10_100, tso_request(1, 0))
            .expect_err("zero-count request must fail");
        assert_eq!(err.code, GrpcCode::Unknown);
        assert_eq!(err.message, TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN);
    }

    #[test]
    fn chaos_expected_primary_mismatch_steps_down_until_recampaign() {
        let now = 20_000;
        let service = build_service_with_primary(1, now);
        let before = service.tso(now, tso_request(1, 1)).expect("initial tso");

        service
            .manager_mut()
            .apply_group_expected_primary(0, Some("remote:0".to_string()), now)
            .expect("apply remote expected primary");

        let stepped_down = service
            .tso(now + 1, tso_request(1, 1))
            .expect_err("expected-primary mismatch should step down serving");
        assert_eq!(stepped_down.code, GrpcCode::Unknown);
        assert!(
            stepped_down.message.contains("is not served")
                || stepped_down.message.contains("not leader")
                || stepped_down.message.contains("not primary")
                || stepped_down.message.contains("no leader"),
            "unexpected expected-primary step-down rejection: {}",
            stepped_down.message
        );

        service
            .manager_mut()
            .apply_group_expected_primary(0, Some("node:0".to_string()), now + 1)
            .expect("restore local expected primary");
        service
            .manager_mut()
            .campaign_group_primary(0, now + 2)
            .expect("re-campaign after expected-primary restoration");

        let recovered = service
            .tso(now + 3, tso_request(1, 1))
            .expect("tso after recampaign");
        assert!(
            recovered.timestamp.compose() > before.timestamp.compose(),
            "timestamp sequence must remain monotonic after expected-primary stepdown/recovery"
        );
    }

    #[test]
    fn chaos_actor_panic_during_manager_lock_does_not_break_future_tso_serving() {
        let service = Arc::new(build_service_with_primary(1, now_ms()));
        let poisoned = Arc::clone(&service);
        let panic_handle = std::thread::spawn(move || {
            let _guard = poisoned.manager_mut();
            panic!("simulated actor panic while holding manager lock");
        });
        assert!(panic_handle.join().is_err(), "panic should be observed");

        let response = service
            .tso(now_ms(), tso_request(1, 1))
            .expect("service should recover from poisoned manager lock");
        assert_eq!(response.count, 1);
    }

    #[tokio::test]
    async fn chaos_unreachable_etcd_runtime_forces_watch_resync_block_and_stepdown() {
        let cluster_id = 1;
        let start_ms = 10_000;
        let lease_ttl_secs = 1_i64;
        let unplanned_failover_p99_budget =
            Duration::from_secs((lease_ttl_secs.saturating_mul(2)).max(5) as u64);
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager =
            KeyspaceGroupManager::new(cluster_id, store, AllocatorConfig::default(), "node");
        manager
            .register_group(default_group())
            .expect("register default group");
        manager
            .campaign_group_primary(0, start_ms)
            .expect("campaign primary");

        let service = Arc::new(tokio::sync::RwLock::new(TsoService::new(
            cluster_id, manager,
        )));
        {
            let guard = service.read().await;
            guard
                .tso(start_ms, tso_request(cluster_id, 1))
                .expect("precondition: service is initially serving");
        }

        let _runtime = TsoRuntime::start(
            Arc::clone(&service),
            TsoRuntimeConfig {
                cluster_id,
                etcd_endpoints: vec!["http://127.0.0.1:1".to_string()],
                campaign_interval: Duration::from_millis(50),
                keepalive_interval: Duration::from_millis(50),
                watch_retry_backoff: Duration::from_millis(50),
                lease_ttl_secs,
                ..TsoRuntimeConfig::default()
            },
        );

        let fail_closed_started = Instant::now();
        wait_until(
            "membership-resync fail-closed gate after etcd connectivity loss",
            Duration::from_secs(10),
            Duration::from_millis(50),
            || {
                let service = Arc::clone(&service);
                async move {
                    let guard = service.read().await;
                    let manager = guard.manager();
                    guard.is_membership_resync_blocking()
                        && manager
                            .group_runtime_snapshot(0, now_ms())
                            .map(|snapshot| !snapshot.is_primary)
                            .unwrap_or(false)
                }
            },
        )
        .await;
        let fail_closed_elapsed = fail_closed_started.elapsed();
        assert!(
            fail_closed_elapsed <= unplanned_failover_p99_budget,
            "fail-closed stepdown exceeded unplanned-failover p99 budget {:?}: {:?}",
            unplanned_failover_p99_budget,
            fail_closed_elapsed
        );

        let guard = service.read().await;
        let blocked = guard
            .tso(start_ms + 1, tso_request(cluster_id, 1))
            .expect_err("watch-resync fail-closed gate must reject ingress");
        assert_eq!(blocked.code, GrpcCode::Unavailable);
        assert_eq!(blocked.retry_after_ms, Some(100));
        assert!(blocked.message.contains("membership resync"));
    }

    fn now_ms() -> i64 {
        let duration = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default();
        (duration.as_millis().min(i64::MAX as u128)) as i64
    }
}