// file: crates/ksp-job-backfill-lib/src/execution.rs
// version: 4

use futures_util::StreamExt; // rust-rules: trait-import

/// Bounded result of one concurrent Backfill candidate execution pass.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BackfillExecutionBatch {
    candidate_count: usize,
    admitted_count: usize,
    finished_count: usize,
    durable_count: usize,
    inserted_count: usize,
    already_present_count: usize,
    purged_count: usize,
    missing_count: usize,
    conflict_count: usize,
    observation_inserted_count: usize,
    observation_already_present_count: usize,
    cancelled_count: usize,
    hole_count: usize,
    maximum_in_flight: usize,
    local_contiguous_completed: usize,
    discovery_partial: bool,
    cancelled: bool,
    checkpoint: crate::BackfillCheckpoint,
    failure_code: std::option::Option<ksp_core_lib::ErrorCode>,
}

impl crate::BackfillExecutionBatch {
    /// Returns the number of candidates present in the bounded discovery result.
    #[must_use]
    pub const fn candidate_count(&self) -> usize {
        return self.candidate_count;
    }

    /// Returns the number of candidates admitted into hydration during this execution pass.
    #[must_use]
    pub const fn admitted_count(&self) -> usize {
        return self.admitted_count;
    }

    /// Returns the number of admitted candidates whose hydration/persistence future reached a known outcome.
    #[must_use]
    pub const fn finished_count(&self) -> usize {
        return self.finished_count;
    }

    /// Returns the number of candidates whose Store outcome is durable for checkpoint advancement.
    #[must_use]
    pub const fn durable_count(&self) -> usize {
        return self.durable_count;
    }

    /// Returns the number of newly inserted canonical RAW entities.
    #[must_use]
    pub const fn inserted_count(&self) -> usize {
        return self.inserted_count;
    }

    /// Returns the number of canonical RAW entities already durable.
    #[must_use]
    pub const fn already_present_count(&self) -> usize {
        return self.already_present_count;
    }

    /// Returns the number of purge tombstones respected by normal persistence.
    #[must_use]
    pub const fn purged_count(&self) -> usize {
        return self.purged_count;
    }

    /// Returns the number of candidates whose `getTransaction` result was missing.
    #[must_use]
    pub const fn missing_count(&self) -> usize {
        return self.missing_count;
    }

    /// Returns the number of Store content conflicts.
    #[must_use]
    pub const fn conflict_count(&self) -> usize {
        return self.conflict_count;
    }

    /// Returns the number of newly inserted acquisition observations.
    #[must_use]
    pub const fn observation_inserted_count(&self) -> usize {
        return self.observation_inserted_count;
    }

    /// Returns the number of acquisition observations already durable.
    #[must_use]
    pub const fn observation_already_present_count(&self) -> usize {
        return self.observation_already_present_count;
    }

    /// Returns the number of admitted candidates cancelled before Store submission.
    #[must_use]
    pub const fn cancelled_count(&self) -> usize {
        return self.cancelled_count;
    }

    /// Returns the number of known candidate outcomes that intentionally block the contiguous frontier.
    #[must_use]
    pub const fn hole_count(&self) -> usize {
        return self.hole_count;
    }

    /// Returns the greatest number of candidate futures simultaneously in flight.
    #[must_use]
    pub const fn maximum_in_flight(&self) -> usize {
        return self.maximum_in_flight;
    }

    /// Returns the durable contiguous prefix completed inside the current discovery result.
    #[must_use]
    pub const fn local_contiguous_completed(&self) -> usize {
        return self.local_contiguous_completed;
    }

    /// Returns the safe caller-owned checkpoint after draining all work already admitted.
    #[must_use]
    pub const fn checkpoint(&self) -> &crate::BackfillCheckpoint {
        return &self.checkpoint;
    }

    /// Returns the first stable fatal error code that stopped new admissions, when one occurred.
    #[must_use]
    pub const fn failure_code(&self) -> std::option::Option<ksp_core_lib::ErrorCode> {
        return self.failure_code;
    }

    /// Returns whether cooperative cancellation stopped admissions or cancelled an admitted pre-Store operation.
    #[must_use]
    pub const fn was_cancelled(&self) -> bool {
        return self.cancelled;
    }

    /// Returns whether discovery or candidate outcomes left the pass incomplete.
    #[must_use]
    pub const fn is_partial(&self) -> bool {
        return self.discovery_partial || self.hole_count != 0 || self.failure_code.is_some() || self.cancelled;
    }
}

/// Internal latest-value execution facts published after each known candidate completion.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct BackfillExecutionProgress {
    admitted_count: usize,
    finished_count: usize,
    inserted_count: usize,
    already_present_count: usize,
    purged_count: usize,
    missing_count: usize,
    conflict_count: usize,
    observation_inserted_count: usize,
    observation_already_present_count: usize,
    cancelled_count: usize,
    hole_count: usize,
    maximum_in_flight: usize,
    contiguous_completed: usize,
    checkpoint: crate::BackfillCheckpoint,
}

impl crate::BackfillExecutionProgress {
    /// Returns the cumulative admission count.
    pub(crate) const fn admitted_count(&self) -> usize {
        return self.admitted_count;
    }

    /// Returns the cumulative finished count.
    pub(crate) const fn finished_count(&self) -> usize {
        return self.finished_count;
    }

    /// Returns the cumulative inserted entity count.
    pub(crate) const fn inserted_count(&self) -> usize {
        return self.inserted_count;
    }

    /// Returns the cumulative already-present entity count.
    pub(crate) const fn already_present_count(&self) -> usize {
        return self.already_present_count;
    }

    /// Returns the cumulative purged entity count.
    pub(crate) const fn purged_count(&self) -> usize {
        return self.purged_count;
    }

    /// Returns the cumulative missing count.
    pub(crate) const fn missing_count(&self) -> usize {
        return self.missing_count;
    }

    /// Returns the cumulative conflict count.
    pub(crate) const fn conflict_count(&self) -> usize {
        return self.conflict_count;
    }

    /// Returns the cumulative inserted observation count.
    pub(crate) const fn observation_inserted_count(&self) -> usize {
        return self.observation_inserted_count;
    }

    /// Returns the cumulative already-present observation count.
    pub(crate) const fn observation_already_present_count(&self) -> usize {
        return self.observation_already_present_count;
    }

    /// Returns the cumulative cancelled candidate count.
    pub(crate) const fn cancelled_count(&self) -> usize {
        return self.cancelled_count;
    }

    /// Returns the cumulative hole count.
    pub(crate) const fn hole_count(&self) -> usize {
        return self.hole_count;
    }

    /// Returns the maximum observed in-flight count.
    pub(crate) const fn maximum_in_flight(&self) -> usize {
        return self.maximum_in_flight;
    }

    /// Returns the cumulative safe contiguous prefix represented by the progress checkpoint.
    pub(crate) const fn contiguous_completed(&self) -> usize {
        return self.contiguous_completed;
    }

    /// Returns the safe checkpoint proven at this progress position.
    pub(crate) const fn checkpoint(&self) -> &crate::BackfillCheckpoint {
        return &self.checkpoint;
    }
}

/// Executes one bounded discovered candidate set with request-owned hydration concurrency.
///
/// This compatibility entry point has no external cancellation handle. Transport owns retries and
/// endpoint selection; Store submissions are always awaited to completion.
pub async fn execute_backfill_discovery(
    transport: &ksp_onchain_transport_lib::HttpTransportPool,
    store: &ksp_store_lib::Store,
    request: &crate::BackfillRequest,
    discovery: &crate::BackfillDiscovery,
) -> ksp_core_lib::Result<crate::BackfillExecutionBatch> {
    let validation = crate::validate_discovery_identity(request, discovery);
    if let std::result::Result::Err(error) = validation {
        return std::result::Result::Err(error);
    }
    let processor = RuntimeCandidateProcessor { transport, store, request, cancellation: std::option::Option::None };
    return execute_with_processor(&processor, request, discovery, std::option::Option::None, std::option::Option::None).await;
}

/// Executes one bounded discovery with cooperative cancellation and concrete progress publication.
pub(crate) async fn execute_backfill_discovery_cancellable(
    transport: &ksp_onchain_transport_lib::HttpTransportPool,
    store: &ksp_store_lib::Store,
    request: &crate::BackfillRequest,
    discovery: &crate::BackfillDiscovery,
    cancellation: &crate::BackfillCancellationSignal,
    publisher: &crate::BackfillRuntimePublisher,
) -> ksp_core_lib::Result<crate::BackfillExecutionBatch> {
    let processor = RuntimeCandidateProcessor { transport, store, request, cancellation: std::option::Option::Some(cancellation) };
    return execute_with_processor(&processor, request, discovery, std::option::Option::Some(cancellation), std::option::Option::Some(publisher)).await;
}

type CandidateProcessFuture<'a> =
    std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = ksp_core_lib::Result<crate::BackfillPersistenceOutcome>> + std::marker::Send + 'a>>;

trait CandidateProcessor {
    fn process<'a>(&'a self, candidate: &'a crate::BackfillCandidate) -> CandidateProcessFuture<'a>;
}

struct RuntimeCandidateProcessor<'a> {
    transport: &'a ksp_onchain_transport_lib::HttpTransportPool,
    store: &'a ksp_store_lib::Store,
    request: &'a crate::BackfillRequest,
    cancellation: std::option::Option<&'a crate::BackfillCancellationSignal>,
}

impl CandidateProcessor for RuntimeCandidateProcessor<'_> {
    fn process<'a>(&'a self, candidate: &'a crate::BackfillCandidate) -> CandidateProcessFuture<'a> {
        return std::boxed::Box::pin(async move {
            let received_at = match current_raw_timestamp() {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
            let hydration_future = crate::hydrate_backfill_candidate(self.transport, self.request, candidate, received_at);
            let hydration = match self.cancellation {
                std::option::Option::Some(signal) => signal.run_cancellable(hydration_future).await,
                std::option::Option::None => hydration_future.await,
            };
            let hydration = match hydration {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
            if self.cancellation.is_some_and(crate::BackfillCancellationSignal::is_requested) {
                return std::result::Result::Err(cancelled_error());
            }
            // Deliberately not cancellable once Store submission begins: a submitted durable write is drained.
            return crate::persist_backfill_hydration(self.store, hydration).await;
        });
    }
}

#[allow(clippy::too_many_lines)]
async fn execute_with_processor<P>(
    processor: &P,
    request: &crate::BackfillRequest,
    discovery: &crate::BackfillDiscovery,
    cancellation: std::option::Option<&crate::BackfillCancellationSignal>,
    publisher: std::option::Option<&crate::BackfillRuntimePublisher>,
) -> ksp_core_lib::Result<crate::BackfillExecutionBatch>
where
    P: CandidateProcessor,
{
    let validation = crate::validate_discovery_identity(request, discovery);
    if let std::result::Result::Err(error) = validation {
        return std::result::Result::Err(error);
    }
    let resume_offset = match crate::execution_resume_offset(request, discovery) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let mut frontier = crate::CompletionFrontier::new(discovery.candidates().len());
    if matches!(request.scope().kind(), crate::BackfillScopeKind::AfterAddress | crate::BackfillScopeKind::ExplicitSignatures) {
        let seeded = frontier.seed_prefix(resume_offset);
        if let std::result::Result::Err(error) = seeded {
            return std::result::Result::Err(error);
        }
    }
    let mut next_index = resume_offset;
    let mut admitted_count = 0_usize;
    let mut finished_count = 0_usize;
    let mut durable_count = 0_usize;
    let mut inserted_count = 0_usize;
    let mut already_present_count = 0_usize;
    let mut purged_count = 0_usize;
    let mut missing_count = 0_usize;
    let mut conflict_count = 0_usize;
    let mut observation_inserted_count = 0_usize;
    let mut observation_already_present_count = 0_usize;
    let mut cancelled_count = 0_usize;
    let mut hole_count = 0_usize;
    let mut maximum_in_flight = 0_usize;
    let mut failure_code = std::option::Option::<ksp_core_lib::ErrorCode>::None;
    let mut cancellation_requested = false;
    let mut in_flight = futures_util::stream::FuturesUnordered::new();
    loop {
        if !cancellation_requested && cancellation.is_some_and(crate::BackfillCancellationSignal::is_requested) {
            cancellation_requested = true;
            let progress = progress_from_state(
                request,
                discovery,
                &frontier,
                admitted_count,
                finished_count,
                inserted_count,
                already_present_count,
                purged_count,
                missing_count,
                conflict_count,
                observation_inserted_count,
                observation_already_present_count,
                cancelled_count,
                hole_count,
                maximum_in_flight,
            );
            let progress = match progress {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
            let published = publish_progress(publisher, &progress, true, true);
            if let std::result::Result::Err(error) = published {
                return std::result::Result::Err(error);
            }
        }
        while failure_code.is_none()
            && !cancellation_requested
            && next_index < discovery.candidates().len()
            && in_flight.len() < request.hydration_concurrency()
        {
            let index = next_index;
            let candidate = &discovery.candidates()[index];
            let future = processor.process(candidate);
            in_flight.push(async move {
                return (index, future.await);
            });
            next_index += 1;
            admitted_count = match checked_increment(admitted_count, "admitted_count") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
            maximum_in_flight = std::cmp::max(maximum_in_flight, in_flight.len());
        }
        let completed = in_flight.next().await;
        let (index, result) = match completed {
            std::option::Option::Some(completed) => completed,
            std::option::Option::None => break,
        };
        finished_count = match checked_increment(finished_count, "finished_count") {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        match result {
            std::result::Result::Ok(outcome) => {
                let classified = classify_persistence(
                    &outcome,
                    &mut inserted_count,
                    &mut already_present_count,
                    &mut purged_count,
                    &mut missing_count,
                    &mut conflict_count,
                    &mut observation_inserted_count,
                    &mut observation_already_present_count,
                );
                if let std::result::Result::Err(error) = classified {
                    return std::result::Result::Err(error);
                }
                if persistence_advances_frontier(&outcome) {
                    let marked = frontier.mark_durable(index);
                    if let std::result::Result::Err(error) = marked {
                        return std::result::Result::Err(error);
                    }
                    durable_count = match checked_increment(durable_count, "durable_count") {
                        std::result::Result::Ok(value) => value,
                        std::result::Result::Err(error) => return std::result::Result::Err(error),
                    };
                } else {
                    hole_count = match checked_increment(hole_count, "hole_count") {
                        std::result::Result::Ok(value) => value,
                        std::result::Result::Err(error) => return std::result::Result::Err(error),
                    };
                    if outcome.entity() == crate::BackfillEntityPersistence::Conflict && failure_code.is_none() {
                        failure_code = std::option::Option::Some(ksp_store_lib::ERROR_CODE_RAW_CONFLICT);
                    }
                }
            },
            std::result::Result::Err(error) => {
                hole_count = match checked_increment(hole_count, "hole_count") {
                    std::result::Result::Ok(value) => value,
                    std::result::Result::Err(increment_error) => return std::result::Result::Err(increment_error),
                };
                if error.code() == crate::ERROR_CODE_BACKFILL_CANCELLED {
                    cancelled_count = match checked_increment(cancelled_count, "cancelled_count") {
                        std::result::Result::Ok(value) => value,
                        std::result::Result::Err(increment_error) => return std::result::Result::Err(increment_error),
                    };
                    cancellation_requested = true;
                } else if failure_code.is_none() {
                    failure_code = std::option::Option::Some(error.code());
                }
            },
        }
        let progress = progress_from_state(
            request,
            discovery,
            &frontier,
            admitted_count,
            finished_count,
            inserted_count,
            already_present_count,
            purged_count,
            missing_count,
            conflict_count,
            observation_inserted_count,
            observation_already_present_count,
            cancelled_count,
            hole_count,
            maximum_in_flight,
        );
        let progress = match progress {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        let draining = cancellation_requested || failure_code.is_some();
        let published = publish_progress(publisher, &progress, cancellation_requested, draining);
        if let std::result::Result::Err(error) = published {
            return std::result::Result::Err(error);
        }
    }
    let checkpoint = match crate::checkpoint_from_frontier(request, discovery, &frontier) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    return std::result::Result::Ok(crate::BackfillExecutionBatch {
        candidate_count: discovery.candidates().len(),
        admitted_count,
        finished_count,
        durable_count,
        inserted_count,
        already_present_count,
        purged_count,
        missing_count,
        conflict_count,
        observation_inserted_count,
        observation_already_present_count,
        cancelled_count,
        hole_count,
        maximum_in_flight,
        local_contiguous_completed: frontier.contiguous_completed(),
        discovery_partial: discovery.is_partial(),
        cancelled: cancellation_requested,
        checkpoint,
        failure_code,
    });
}

#[allow(clippy::too_many_arguments)]
fn progress_from_state(
    request: &crate::BackfillRequest,
    discovery: &crate::BackfillDiscovery,
    frontier: &crate::CompletionFrontier,
    admitted_count: usize,
    finished_count: usize,
    inserted_count: usize,
    already_present_count: usize,
    purged_count: usize,
    missing_count: usize,
    conflict_count: usize,
    observation_inserted_count: usize,
    observation_already_present_count: usize,
    cancelled_count: usize,
    hole_count: usize,
    maximum_in_flight: usize,
) -> ksp_core_lib::Result<crate::BackfillExecutionProgress> {
    let checkpoint = match crate::checkpoint_from_frontier(request, discovery, frontier) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    return std::result::Result::Ok(crate::BackfillExecutionProgress {
        admitted_count,
        finished_count,
        inserted_count,
        already_present_count,
        purged_count,
        missing_count,
        conflict_count,
        observation_inserted_count,
        observation_already_present_count,
        cancelled_count,
        hole_count,
        maximum_in_flight,
        contiguous_completed: checkpoint.completed_prefix(),
        checkpoint,
    });
}

fn publish_progress(
    publisher: std::option::Option<&crate::BackfillRuntimePublisher>,
    progress: &crate::BackfillExecutionProgress,
    cancelling: bool,
    draining: bool,
) -> ksp_core_lib::Result<()> {
    if let std::option::Option::Some(publisher) = publisher {
        let published = publisher.publish_execution_progress(progress, cancelling, draining);
        if let std::result::Result::Err(error) = published {
            return std::result::Result::Err(error);
        }
    }
    return std::result::Result::Ok(());
}

#[allow(clippy::too_many_arguments)]
fn classify_persistence(
    outcome: &crate::BackfillPersistenceOutcome,
    inserted_count: &mut usize,
    already_present_count: &mut usize,
    purged_count: &mut usize,
    missing_count: &mut usize,
    conflict_count: &mut usize,
    observation_inserted_count: &mut usize,
    observation_already_present_count: &mut usize,
) -> ksp_core_lib::Result<()> {
    match outcome.entity() {
        crate::BackfillEntityPersistence::Inserted => {
            *inserted_count = match checked_increment(*inserted_count, "inserted_count") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        },
        crate::BackfillEntityPersistence::AlreadyPresent => {
            *already_present_count = match checked_increment(*already_present_count, "already_present_count") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        },
        crate::BackfillEntityPersistence::SkippedPurged => {
            *purged_count = match checked_increment(*purged_count, "purged_count") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        },
        crate::BackfillEntityPersistence::Missing => {
            *missing_count = match checked_increment(*missing_count, "missing_count") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        },
        crate::BackfillEntityPersistence::Conflict => {
            *conflict_count = match checked_increment(*conflict_count, "conflict_count") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        },
    }
    match outcome.observation() {
        crate::BackfillObservationPersistence::Inserted => {
            *observation_inserted_count = match checked_increment(*observation_inserted_count, "observation_inserted_count") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        },
        crate::BackfillObservationPersistence::AlreadyPresent => {
            *observation_already_present_count = match checked_increment(*observation_already_present_count, "observation_already_present_count") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        },
        crate::BackfillObservationPersistence::NotRecorded | crate::BackfillObservationPersistence::NotApplicable => {},
    }
    return std::result::Result::Ok(());
}

fn persistence_advances_frontier(outcome: &crate::BackfillPersistenceOutcome) -> bool {
    return matches!(
        outcome.entity(),
        crate::BackfillEntityPersistence::Inserted | crate::BackfillEntityPersistence::AlreadyPresent | crate::BackfillEntityPersistence::SkippedPurged
    );
}

fn checked_increment(value: usize, field: &'static str) -> ksp_core_lib::Result<usize> {
    return value.checked_add(1).ok_or_else(|| return execution_error(field));
}

fn current_raw_timestamp() -> ksp_core_lib::Result<ksp_store_lib::RawTimestamp> {
    let duration = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
    let duration = match duration {
        std::result::Result::Ok(duration) => duration,
        std::result::Result::Err(_) => return std::result::Result::Err(execution_error("clock.before_epoch")),
    };
    let millis = match u64::try_from(duration.as_millis()) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return std::result::Result::Err(execution_error("clock.millis")),
    };
    return ksp_store_lib::RawTimestamp::from_unix_millis(millis);
}

fn cancelled_error() -> ksp_core_lib::Error {
    return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_CANCELLED, "Backfill candidate cancelled before durable Store submission");
}

fn execution_error(field: &'static str) -> ksp_core_lib::Error {
    return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_EXECUTION_INVALID, "invalid bounded Backfill execution state").with_context("field", field);
}

#[cfg(test)]
#[path = "../unit_tests/execution.rs"]
mod tests;