// file: crates/ksp-job-api/tests/notifications.rs
// version: 1

//! External-consumer canaries for latest-value Job observation.

#[derive(Clone, Debug, Eq, PartialEq)]
struct TestSnapshot {
    completed: u64,
}

#[derive(Clone)]
struct TestSnapshotSource {
    current: std::sync::Arc<ksp_job_api::JobNotification<TestSnapshot>>,
}

impl TestSnapshotSource {
    fn new(current: ksp_job_api::JobNotification<TestSnapshot>) -> Self {
        return Self { current: std::sync::Arc::new(current) };
    }
}

impl ksp_job_api::JobSnapshotSource for TestSnapshotSource {
    type Snapshot = TestSnapshot;

    fn current(&self) -> ksp_job_api::JobNotification<Self::Snapshot> {
        return self.current.as_ref().clone();
    }

    fn wait_for_change(&self, observed: ksp_job_api::JobNotificationSequence) -> ksp_job_api::JobSnapshotFuture<'_, Self::Snapshot> {
        let current = ksp_job_api::JobSnapshotSource::current(self);
        assert!(current.sequence().is_after(observed));
        return std::boxed::Box::pin(std::future::ready(current));
    }
}

struct TestWake;

impl std::task::Wake for TestWake {
    fn wake(self: std::sync::Arc<Self>) {
        return;
    }
}

fn poll_ready<S>(mut future: ksp_job_api::JobSnapshotFuture<'_, S>) -> std::option::Option<ksp_job_api::JobNotification<S>> {
    let waker = std::task::Waker::from(std::sync::Arc::new(TestWake));
    let mut context = std::task::Context::from_waker(&waker);
    return match std::future::Future::poll(future.as_mut(), &mut context) {
        std::task::Poll::Ready(value) => std::option::Option::Some(value),
        std::task::Poll::Pending => std::option::Option::None,
    };
}

fn notification_sequence(value: u64) -> std::option::Option<ksp_job_api::JobNotificationSequence> {
    let mut sequence = ksp_job_api::JobNotificationSequence::initial();
    for _ in 0..value {
        sequence = match sequence.next() {
            std::result::Result::Ok(next) => next,
            std::result::Result::Err(_) => return std::option::Option::None,
        };
    }
    return std::option::Option::Some(sequence);
}

fn notification(sequence_value: u64, state: ksp_job_api::JobState, completed: u64) -> std::option::Option<ksp_job_api::JobNotification<TestSnapshot>> {
    let id = match ksp_job_api::JobId::new("external-job-001") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return std::option::Option::None,
    };
    let kind = match ksp_job_api::JobKindCode::new("backfill_raw") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return std::option::Option::None,
    };
    let sequence = match notification_sequence(sequence_value) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return std::option::Option::None,
    };
    return std::option::Option::Some(ksp_job_api::JobNotification::new(id, kind, sequence, state, TestSnapshot { completed }));
}

#[test]
fn pre_003_external_notification_contract_is_consumable_from_crate_root() {
    let notification = notification(3, ksp_job_api::JobState::Running, 2);
    assert!(notification.is_some());
    let notification = match notification {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    assert_eq!(notification.id().as_str(), "external-job-001");
    assert_eq!(notification.kind().as_str(), "backfill_raw");
    assert_eq!(notification.sequence().value(), 3);
    assert_eq!(notification.state(), ksp_job_api::JobState::Running);
    assert_eq!(notification.snapshot().completed, 2);
    return;
}

#[test]
fn pre_003_slow_and_independent_listeners_resynchronize_to_latest_value() {
    let latest = notification(8, ksp_job_api::JobState::Running, 7);
    assert!(latest.is_some());
    let source = match latest {
        std::option::Option::Some(value) => TestSnapshotSource::new(value),
        std::option::Option::None => return,
    };
    let listener_a = source.clone();
    let listener_b = source.clone();
    let observed_a = match notification_sequence(2) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let observed_b = match notification_sequence(6) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let changed_a = poll_ready(ksp_job_api::JobSnapshotSource::wait_for_change(&listener_a, observed_a));
    let changed_b = poll_ready(ksp_job_api::JobSnapshotSource::wait_for_change(&listener_b, observed_b));
    assert!(changed_a.is_some());
    assert!(changed_b.is_some());
    let changed_a = match changed_a {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let changed_b = match changed_b {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    assert_eq!(changed_a.sequence().value(), 8);
    assert_eq!(changed_b.sequence().value(), 8);
    assert_eq!(changed_a.snapshot().completed, 7);
    assert_eq!(changed_b.snapshot().completed, 7);
    return;
}

#[test]
fn pre_003_terminal_snapshot_remains_readable_from_shared_source() {
    let terminal = notification(9, ksp_job_api::JobState::Completed(ksp_job_api::JobCompletion::Partial), 8);
    assert!(terminal.is_some());
    let source = match terminal {
        std::option::Option::Some(value) => TestSnapshotSource::new(value),
        std::option::Option::None => return,
    };
    let cloned = source.clone();
    let current = ksp_job_api::JobSnapshotSource::current(&cloned);
    assert_eq!(current.sequence().value(), 9);
    assert_eq!(current.state(), ksp_job_api::JobState::Completed(ksp_job_api::JobCompletion::Partial));
    assert!(current.state().is_terminal());
    assert_eq!(current.snapshot().completed, 8);
    return;
}