// file: crates/ksp-job-backfill-lib/src/checkpoint.rs
// version: 3

/// Opaque caller-owned checkpoint for one controlled Backfill resumption.
///
/// The checkpoint carries no payload, endpoint, provider, URL or secret. Persistence of this
/// value is deliberately external to Store in v0.3.6; the Backfill library only validates and
/// consumes checkpoints supplied back by its caller.
#[derive(Clone, Eq, PartialEq)]
pub struct BackfillCheckpoint {
    job_id: ksp_job_api::JobId,
    scope_fingerprint: crate::BackfillScopeFingerprint,
    completed_prefix: usize,
    resume_before: std::option::Option<crate::BackfillSignature>,
}

impl crate::BackfillCheckpoint {
    /// Returns the logical Job identity that owns this checkpoint.
    #[must_use]
    pub const fn job_id(&self) -> &ksp_job_api::JobId {
        return &self.job_id;
    }

    /// Returns the semantic scope fingerprint bound to this checkpoint.
    #[must_use]
    pub const fn scope_fingerprint(&self) -> crate::BackfillScopeFingerprint {
        return self.scope_fingerprint;
    }

    /// Returns the number of candidates proven durable in one contiguous prefix.
    #[must_use]
    pub const fn completed_prefix(&self) -> usize {
        return self.completed_prefix;
    }

    /// Creates one internally proven checkpoint.
    pub(crate) fn new(
        job_id: ksp_job_api::JobId,
        scope_fingerprint: crate::BackfillScopeFingerprint,
        completed_prefix: usize,
        resume_before: std::option::Option<crate::BackfillSignature>,
    ) -> Self {
        return Self { job_id, scope_fingerprint, completed_prefix, resume_before };
    }

    /// Reissues this opaque frontier for a new caller-owned Job identity without changing scope or progress.
    pub(crate) fn reissue_for_job(mut self, job_id: ksp_job_api::JobId) -> Self {
        self.job_id = job_id;
        return self;
    }

    /// Returns the internal exclusive `before` cursor used only by controlled Before resumption.
    pub(crate) const fn resume_before(&self) -> std::option::Option<&crate::BackfillSignature> {
        return self.resume_before.as_ref();
    }
}

impl std::fmt::Debug for crate::BackfillCheckpoint {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        return formatter
            .debug_struct("BackfillCheckpoint")
            .field("job_id", &self.job_id)
            .field("scope_fingerprint", &self.scope_fingerprint)
            .field("completed_prefix", &self.completed_prefix)
            .field("has_resume_before", &self.resume_before.is_some())
            .finish();
    }
}

/// Private bounded bitmap tracking durable completions and their contiguous prefix.
pub(crate) struct CompletionFrontier {
    durable: std::vec::Vec<bool>,
    contiguous_completed: usize,
}

impl crate::CompletionFrontier {
    /// Creates one empty frontier for the exact bounded candidate count.
    pub(crate) fn new(candidate_count: usize) -> Self {
        return Self { durable: vec![false; candidate_count], contiguous_completed: 0 };
    }

    /// Seeds a previously proven replay prefix before processing the remaining candidates.
    pub(crate) fn seed_prefix(&mut self, completed_prefix: usize) -> ksp_core_lib::Result<()> {
        if completed_prefix > self.durable.len() {
            return std::result::Result::Err(checkpoint_error("checkpoint.completed_prefix"));
        }
        for index in 0..completed_prefix {
            self.durable[index] = true;
        }
        self.contiguous_completed = completed_prefix;
        return std::result::Result::Ok(());
    }

    /// Marks one candidate durable and advances only through the now-contiguous prefix.
    pub(crate) fn mark_durable(&mut self, index: usize) -> ksp_core_lib::Result<()> {
        let slot = match self.durable.get_mut(index) {
            std::option::Option::Some(slot) => slot,
            std::option::Option::None => return std::result::Result::Err(checkpoint_error("frontier.index")),
        };
        *slot = true;
        while self.contiguous_completed < self.durable.len() && self.durable[self.contiguous_completed] {
            self.contiguous_completed += 1;
        }
        return std::result::Result::Ok(());
    }

    /// Returns the number of durable candidates in the contiguous prefix.
    pub(crate) const fn contiguous_completed(&self) -> usize {
        return self.contiguous_completed;
    }
}

/// Validates one checkpoint against the Job, semantic scope and scope-specific cursor shape.
pub(crate) fn validate_request_checkpoint(
    request_job_id: &ksp_job_api::JobId,
    scope_fingerprint: crate::BackfillScopeFingerprint,
    scope_kind: crate::BackfillScopeKind,
    checkpoint: &crate::BackfillCheckpoint,
) -> ksp_core_lib::Result<()> {
    if checkpoint.job_id() != request_job_id {
        return std::result::Result::Err(checkpoint_error("checkpoint.job_id"));
    }
    if checkpoint.scope_fingerprint() != scope_fingerprint {
        return std::result::Result::Err(checkpoint_error("checkpoint.scope_fingerprint"));
    }
    if scope_kind != crate::BackfillScopeKind::BeforeAddress && checkpoint.resume_before().is_some() {
        return std::result::Result::Err(checkpoint_error("checkpoint.resume_before"));
    }
    return std::result::Result::Ok(());
}

/// Resolves the exclusive Before cursor from a validated checkpoint or the original scope anchor.
pub(crate) fn resume_before_cursor(request: &crate::BackfillRequest) -> std::option::Option<std::string::String> {
    if request.scope().kind() != crate::BackfillScopeKind::BeforeAddress {
        return std::option::Option::None;
    }
    if let std::option::Option::Some(checkpoint) = request.checkpoint()
        && let std::option::Option::Some(cursor) = checkpoint.resume_before()
    {
        return std::option::Option::Some(cursor.as_str().to_owned());
    }
    return request.scope().anchor().map(|anchor| return anchor.as_str().to_owned());
}

/// Resolves the replay prefix skipped only by After and Explicit execution.
pub(crate) fn execution_resume_offset(request: &crate::BackfillRequest, discovery: &crate::BackfillDiscovery) -> ksp_core_lib::Result<usize> {
    let checkpoint = match request.checkpoint() {
        std::option::Option::Some(checkpoint) => checkpoint,
        std::option::Option::None => return std::result::Result::Ok(0),
    };
    return match request.scope().kind() {
        crate::BackfillScopeKind::AfterAddress | crate::BackfillScopeKind::ExplicitSignatures => {
            if checkpoint.completed_prefix() > discovery.candidates().len() {
                return std::result::Result::Err(checkpoint_error("checkpoint.completed_prefix"));
            }
            std::result::Result::Ok(checkpoint.completed_prefix())
        },
        crate::BackfillScopeKind::LatestAddress | crate::BackfillScopeKind::BeforeAddress => std::result::Result::Ok(0),
    };
}

/// Builds the safe next checkpoint from one drained contiguous completion frontier.
pub(crate) fn checkpoint_from_frontier(
    request: &crate::BackfillRequest,
    discovery: &crate::BackfillDiscovery,
    frontier: &crate::CompletionFrontier,
) -> ksp_core_lib::Result<crate::BackfillCheckpoint> {
    let validation = crate::validate_discovery_identity(request, discovery);
    if let std::result::Result::Err(error) = validation {
        return std::result::Result::Err(error);
    }
    let local_prefix = frontier.contiguous_completed();
    if request.scope().kind() == crate::BackfillScopeKind::AfterAddress && discovery.boundary() == crate::BackfillDiscoveryBoundary::AfterAnchorNotReached {
        if let std::option::Option::Some(checkpoint) = request.checkpoint() {
            return std::result::Result::Ok(checkpoint.clone());
        }
        return std::result::Result::Ok(crate::BackfillCheckpoint::new(request.job_id().clone(), request.scope_fingerprint(), 0, std::option::Option::None));
    }
    let (completed_prefix, resume_before) = match request.scope().kind() {
        crate::BackfillScopeKind::LatestAddress => (local_prefix, std::option::Option::None),
        crate::BackfillScopeKind::BeforeAddress => {
            let previous = request.checkpoint().map_or(0, crate::BackfillCheckpoint::completed_prefix);
            let completed_prefix = match previous.checked_add(local_prefix) {
                std::option::Option::Some(value) => value,
                std::option::Option::None => return std::result::Result::Err(checkpoint_error("checkpoint.completed_prefix")),
            };
            let resume_before = if local_prefix == 0 {
                match request.checkpoint() {
                    std::option::Option::Some(checkpoint) => checkpoint.resume_before().cloned(),
                    std::option::Option::None => request.scope().anchor().cloned(),
                }
            } else {
                let index = local_prefix - 1;
                let candidate = match discovery.candidates().get(index) {
                    std::option::Option::Some(candidate) => candidate,
                    std::option::Option::None => return std::result::Result::Err(checkpoint_error("frontier.index")),
                };
                std::option::Option::Some(candidate.identity().signature().clone())
            };
            (completed_prefix, resume_before)
        },
        crate::BackfillScopeKind::AfterAddress | crate::BackfillScopeKind::ExplicitSignatures => (local_prefix, std::option::Option::None),
    };
    return std::result::Result::Ok(crate::BackfillCheckpoint::new(request.job_id().clone(), request.scope_fingerprint(), completed_prefix, resume_before));
}

/// Validates that one discovery result belongs to the request network and semantic scope.
pub(crate) fn validate_discovery_identity(request: &crate::BackfillRequest, discovery: &crate::BackfillDiscovery) -> ksp_core_lib::Result<()> {
    if discovery.network() != request.network() {
        return std::result::Result::Err(checkpoint_error("discovery.network"));
    }
    if discovery.scope_fingerprint() != request.scope_fingerprint() {
        return std::result::Result::Err(checkpoint_error("discovery.scope_fingerprint"));
    }
    return std::result::Result::Ok(());
}

fn checkpoint_error(field: &'static str) -> ksp_core_lib::Error {
    return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_CHECKPOINT_INVALID, "invalid Backfill checkpoint/frontier state").with_context("field", field);
}

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