// file: crates/ksp-job-backfill-lib/unit_tests/execution.rs
// version: 2

#[derive(Clone, Copy)]
enum FakeDisposition {
    Durable,
    Missing,
    Conflict,
    Failure,
}

#[derive(Clone, Copy)]
struct FakePlan {
    pending_polls: usize,
    disposition: FakeDisposition,
}

struct FakeProcessor {
    plans: std::vec::Vec<FakePlan>,
    active: std::sync::atomic::AtomicUsize,
    maximum: std::sync::atomic::AtomicUsize,
    calls: std::sync::atomic::AtomicUsize,
}

impl FakeProcessor {
    fn new(plans: std::vec::Vec<FakePlan>) -> Self {
        return Self {
            plans,
            active: std::sync::atomic::AtomicUsize::new(0),
            maximum: std::sync::atomic::AtomicUsize::new(0),
            calls: std::sync::atomic::AtomicUsize::new(0),
        };
    }

    fn maximum(&self) -> usize {
        return self.maximum.load(std::sync::atomic::Ordering::SeqCst);
    }

    fn calls(&self) -> usize {
        return self.calls.load(std::sync::atomic::Ordering::SeqCst);
    }
}

impl super::CandidateProcessor for FakeProcessor {
    fn process<'a>(&'a self, candidate: &'a crate::BackfillCandidate) -> super::CandidateProcessFuture<'a> {
        self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        let first_byte = candidate.identity().signature().as_str().as_bytes()[0];
        let plan_index = usize::from(first_byte.saturating_sub(b'1'));
        let plan = match self.plans.get(plan_index) {
            std::option::Option::Some(plan) => *plan,
            std::option::Option::None => FakePlan { pending_polls: 0, disposition: FakeDisposition::Failure },
        };
        return std::boxed::Box::pin(async move {
            let mut pending_polls = plan.pending_polls;
            let mut started = false;
            std::future::poll_fn(|context| {
                if !started {
                    started = true;
                    let active = self.active.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
                    self.maximum.fetch_max(active, std::sync::atomic::Ordering::SeqCst);
                }
                if pending_polls != 0 {
                    pending_polls -= 1;
                    context.waker().wake_by_ref();
                    return std::task::Poll::Pending;
                }
                self.active.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
                return std::task::Poll::Ready(());
            })
            .await;
            return planned_outcome(candidate, plan.disposition);
        });
    }
}

fn planned_outcome(candidate: &crate::BackfillCandidate, disposition: FakeDisposition) -> ksp_core_lib::Result<crate::BackfillPersistenceOutcome> {
    if matches!(disposition, FakeDisposition::Failure) {
        return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_RAW_CONVERSION_INVALID, "planned candidate failure"));
    }
    let fill = candidate.identity().signature().as_str().as_bytes()[0];
    let reference =
        ksp_store_lib::RawTransactionReference::new(candidate.identity().network().clone(), ksp_store_lib::RawTransactionSignature::new([fill; 64]));
    let (entity, observation) = match disposition {
        FakeDisposition::Durable => (crate::BackfillEntityPersistence::AlreadyPresent, crate::BackfillObservationPersistence::AlreadyPresent),
        FakeDisposition::Missing => (crate::BackfillEntityPersistence::Missing, crate::BackfillObservationPersistence::NotApplicable),
        FakeDisposition::Conflict => (crate::BackfillEntityPersistence::Conflict, crate::BackfillObservationPersistence::NotRecorded),
        FakeDisposition::Failure => return std::result::Result::Err(super::execution_error("test.disposition")),
    };
    return std::result::Result::Ok(crate::BackfillPersistenceOutcome::new(reference, entity, observation));
}

fn signature(character: char) -> std::option::Option<crate::BackfillSignature> {
    return crate::BackfillSignature::new(character.to_string().repeat(crate::MIN_BACKFILL_SIGNATURE_TEXT_BYTES)).ok();
}

fn explicit_request(characters: &[char], concurrency: usize) -> std::option::Option<crate::BackfillRequest> {
    let mut signatures = std::vec::Vec::with_capacity(characters.len());
    for character in characters {
        let value = match signature(*character) {
            std::option::Option::Some(value) => value,
            std::option::Option::None => return std::option::Option::None,
        };
        signatures.push(value);
    }
    let scope = match crate::BackfillScope::explicit_signatures(signatures) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return std::option::Option::None,
    };
    let job_id = match ksp_job_api::JobId::new("backfill:execution-test") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return std::option::Option::None,
    };
    let network = match ksp_store_lib::RawNetworkId::new("devnet") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return std::option::Option::None,
    };
    return crate::BackfillRequest::new(
        job_id,
        network,
        ksp_onchain_transport_lib::HttpRoleName::new("history"),
        crate::BackfillCommitment::Confirmed,
        scope,
        100,
        10,
        characters.len(),
        concurrency,
        std::option::Option::None,
    )
    .ok();
}

fn discovery(request: &crate::BackfillRequest, characters: &[char]) -> std::option::Option<crate::BackfillDiscovery> {
    let mut candidates = std::vec::Vec::with_capacity(characters.len());
    for character in characters {
        let signature = match signature(*character) {
            std::option::Option::Some(value) => value,
            std::option::Option::None => return std::option::Option::None,
        };
        let identity = crate::BackfillCandidateIdentity::new(request.network().clone(), signature);
        candidates.push(crate::BackfillCandidate::new(identity, std::option::Option::None));
    }
    return std::option::Option::Some(crate::BackfillDiscovery::new(
        request.network().clone(),
        request.scope_fingerprint(),
        candidates,
        0,
        crate::BackfillDiscoveryBoundary::ExplicitInput,
    ));
}

#[tokio::test]
async fn pre_008_execution_is_bounded_and_reconciles_out_of_order_durable_completions() {
    let request = match explicit_request(&['1', '2', '3', '4'], 2) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let discovery = match discovery(&request, &['1', '2', '3', '4']) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let processor = FakeProcessor::new(std::vec![
        FakePlan { pending_polls: 6, disposition: FakeDisposition::Durable },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
        FakePlan { pending_polls: 2, disposition: FakeDisposition::Durable },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
    ]);
    let result = super::execute_with_processor(&processor, &request, &discovery, std::option::Option::None, std::option::Option::None).await;
    let batch = match result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return,
    };
    assert_eq!(batch.admitted_count(), 4);
    assert_eq!(batch.finished_count(), 4);
    assert_eq!(batch.durable_count(), 4);
    assert_eq!(batch.hole_count(), 0);
    assert_eq!(batch.maximum_in_flight(), 2);
    assert_eq!(processor.maximum(), 2);
    assert_eq!(batch.local_contiguous_completed(), 4);
    assert_eq!(batch.checkpoint().completed_prefix(), 4);
    assert!(!batch.is_partial());
    return;
}

#[tokio::test]
async fn pre_008_missing_is_non_fatal_but_blocks_frontier_while_later_candidates_continue() {
    let request = match explicit_request(&['1', '2', '3'], 2) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let discovery = match discovery(&request, &['1', '2', '3']) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let processor = FakeProcessor::new(std::vec![
        FakePlan { pending_polls: 1, disposition: FakeDisposition::Durable },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Missing },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
    ]);
    let batch = match super::execute_with_processor(&processor, &request, &discovery, std::option::Option::None, std::option::Option::None).await {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return,
    };
    assert_eq!(batch.admitted_count(), 3);
    assert_eq!(batch.finished_count(), 3);
    assert_eq!(batch.durable_count(), 2);
    assert_eq!(batch.hole_count(), 1);
    assert_eq!(batch.local_contiguous_completed(), 1);
    assert_eq!(batch.checkpoint().completed_prefix(), 1);
    assert_eq!(batch.failure_code(), std::option::Option::None);
    assert!(batch.is_partial());
    return;
}

#[tokio::test]
async fn pre_008_conflict_stops_new_admissions_and_drains_already_in_flight_work() {
    let request = match explicit_request(&['1', '2', '3', '4'], 2) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let discovery = match discovery(&request, &['1', '2', '3', '4']) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let processor = FakeProcessor::new(std::vec![
        FakePlan { pending_polls: 6, disposition: FakeDisposition::Durable },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Conflict },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
    ]);
    let batch = match super::execute_with_processor(&processor, &request, &discovery, std::option::Option::None, std::option::Option::None).await {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return,
    };
    assert_eq!(processor.calls(), 2);
    assert_eq!(batch.admitted_count(), 2);
    assert_eq!(batch.finished_count(), 2);
    assert_eq!(batch.durable_count(), 1);
    assert_eq!(batch.hole_count(), 1);
    assert_eq!(batch.local_contiguous_completed(), 1);
    assert_eq!(batch.failure_code(), std::option::Option::Some(ksp_store_lib::ERROR_CODE_RAW_CONFLICT));
    assert!(batch.is_partial());
    return;
}

#[tokio::test]
async fn pre_008_explicit_resume_skips_only_the_checkpointed_contiguous_prefix() {
    let request = match explicit_request(&['1', '2', '3', '4'], 2) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let checkpoint = crate::BackfillCheckpoint::new(request.job_id().clone(), request.scope_fingerprint(), 2, std::option::Option::None);
    let request = match request.with_checkpoint(checkpoint) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return,
    };
    let discovery = match discovery(&request, &['1', '2', '3', '4']) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let processor = FakeProcessor::new(std::vec![
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Failure },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Failure },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
    ]);
    let batch = match super::execute_with_processor(&processor, &request, &discovery, std::option::Option::None, std::option::Option::None).await {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return,
    };
    assert_eq!(processor.calls(), 2);
    assert_eq!(batch.admitted_count(), 2);
    assert_eq!(batch.local_contiguous_completed(), 4);
    assert_eq!(batch.checkpoint().completed_prefix(), 4);
    assert_eq!(batch.failure_code(), std::option::Option::None);
    return;
}

#[tokio::test]
async fn pre_009_cancellation_stops_admission_and_drains_already_admitted_candidate_work() {
    let request = match explicit_request(&['1', '2', '3', '4'], 2) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let discovery = match discovery(&request, &['1', '2', '3', '4']) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return,
    };
    let processor = FakeProcessor::new(std::vec![
        FakePlan { pending_polls: 12, disposition: FakeDisposition::Durable },
        FakePlan { pending_polls: 12, disposition: FakeDisposition::Durable },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
        FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable },
    ]);
    let (cancel_sender, cancel_receiver) = tokio::sync::watch::channel(false);
    let control = crate::runtime::BackfillRuntimeControl::new(cancel_sender);
    let signal = crate::runtime::BackfillCancellationSignal::new(control.token(), cancel_receiver);
    let execution = super::execute_with_processor(&processor, &request, &discovery, std::option::Option::Some(&signal), std::option::Option::None);
    let cancellation = async {
        loop {
            if processor.calls() >= 2 {
                break;
            }
            tokio::task::yield_now().await;
        }
        assert!(control.request_cancellation());
    };
    let (result, ()) = tokio::join!(execution, cancellation);
    let batch = match result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return,
    };
    assert_eq!(processor.calls(), 2);
    assert_eq!(batch.admitted_count(), 2);
    assert_eq!(batch.finished_count(), 2);
    assert_eq!(batch.durable_count(), 2);
    assert_eq!(batch.cancelled_count(), 0);
    assert!(batch.was_cancelled());
    assert_eq!(batch.checkpoint().completed_prefix(), 2);
    return;
}