icydb-core 0.98.1

//! Module: executor::planning::continuation::scalar
//! Responsibility: scalar continuation planning/runtime bindings for access resume behavior.
//! Does not own: cursor token encoding policy or planner semantic ownership.
//! Boundary: consumes validated continuation contracts and computes scalar resume inputs.

use crate::{
    db::{
        access::LoweredKey,
        cursor::{
            ContinuationSignature, CursorBoundary, PlannedCursor, RangeToken,
            decode_pk_cursor_boundary_storage_key_for_name,
            effective_keep_count_for_limit as continuation_keep_count_for_limit,
            effective_page_offset_for_window as continuation_page_offset_for_window,
            range_token_anchor_key, range_token_from_validated_cursor_anchor,
        },
        direction::Direction,
        executor::{
            AccessScanContinuationInput, ContinuationMode, RouteContinuationPlan,
            planning::route::LoadOrderRouteContract,
        },
        query::plan::{AccessPlannedQuery, ContinuationPolicy},
    },
    error::InternalError,
};

///
/// ScalarContinuationContext
///
/// Normalized scalar continuation runtime state.
/// Carries the validated cursor plus pre-derived boundary and index-range anchor
/// bindings so load/route code does not decode cursor internals directly.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db::executor) struct ScalarContinuationContext {
    cursor_boundary: Option<CursorBoundary>,
    index_range_token: Option<RangeToken>,
    continuation_signature: Option<ContinuationSignature>,
}

impl ScalarContinuationContext {
    /// Construct one empty scalar continuation runtime for initial executions.
    #[must_use]
    pub(in crate::db::executor) const fn initial() -> Self {
        Self {
            cursor_boundary: None,
            index_range_token: None,
            continuation_signature: None,
        }
    }

    /// Build one scalar runtime cursor binding bundle from one planned cursor.
    #[must_use]
    pub(in crate::db::executor) fn new(cursor: PlannedCursor) -> Self {
        let cursor_boundary = cursor.boundary().cloned();
        let index_range_token = cursor
            .index_range_anchor()
            .map(range_token_from_validated_cursor_anchor);

        Self {
            cursor_boundary,
            index_range_token,
            continuation_signature: None,
        }
    }

    /// Build one runtime-ready scalar continuation context from validated
    /// cursor state plus the immutable continuation signature.
    #[must_use]
    pub(in crate::db::executor) fn for_runtime(
        cursor: PlannedCursor,
        continuation_signature: ContinuationSignature,
    ) -> Self {
        let mut continuation = Self::new(cursor);
        continuation.continuation_signature = Some(continuation_signature);

        continuation
    }

    /// Borrow optional scalar cursor boundary.
    #[must_use]
    pub(in crate::db::executor) const fn cursor_boundary(&self) -> Option<&CursorBoundary> {
        self.cursor_boundary.as_ref()
    }

    /// Return whether this scalar continuation context has one cursor boundary.
    #[must_use]
    pub(in crate::db::executor) const fn has_cursor_boundary(&self) -> bool {
        self.cursor_boundary.is_some()
    }

    /// Validate scalar cursor-boundary decode for PK fast-path eligibility gates.
    ///
    /// This preserves PK fast-path cursor error classification while keeping
    /// boundary decode authority in continuation runtime.
    pub(in crate::db::executor) fn validate_pk_fast_path_boundary(
        &self,
        primary_key_name: &str,
    ) -> Result<(), InternalError> {
        let _ = decode_pk_cursor_boundary_storage_key_for_name(
            self.cursor_boundary(),
            primary_key_name,
        )
        .map_err(crate::db::cursor::CursorPlanError::into_internal_error)?;

        Ok(())
    }

    /// Borrow optional index-range continuation anchor token.
    #[must_use]
    pub(in crate::db::executor) const fn index_range_token(&self) -> Option<&RangeToken> {
        self.index_range_token.as_ref()
    }

    /// Return whether this scalar continuation context has one index-range anchor.
    #[must_use]
    pub(in crate::db::executor) const fn has_index_range_anchor(&self) -> bool {
        self.index_range_token.is_some()
    }

    /// Derive route continuation mode from scalar continuation context shape.
    #[must_use]
    pub(in crate::db::executor) const fn route_continuation_mode(&self) -> ContinuationMode {
        match (self.has_cursor_boundary(), self.has_index_range_anchor()) {
            (_, true) => ContinuationMode::IndexRangeAnchor,
            (true, false) => ContinuationMode::CursorBoundary,
            (false, false) => ContinuationMode::Initial,
        }
    }

    /// Derive one route continuation plan from scalar runtime state and planner policy.
    ///
    /// This keeps continuation/window derivation in continuation authority so
    /// route planning consumes one pre-derived continuation contract.
    #[must_use]
    pub(in crate::db::executor) fn route_continuation_plan(
        &self,
        plan: &AccessPlannedQuery,
        continuation_policy: ContinuationPolicy,
    ) -> RouteContinuationPlan {
        RouteContinuationPlan::from_scalar_access_window_plan(
            self.route_continuation_mode(),
            continuation_policy,
            plan.scalar_access_window_plan(self.has_cursor_boundary()),
        )
    }

    /// Build access-stream continuation input for routed stream resolution.
    #[must_use]
    pub(in crate::db::executor) fn access_scan_input(
        &self,
        direction: Direction,
    ) -> AccessScanContinuationInput<'_> {
        AccessScanContinuationInput::new(self.previous_index_range_anchor(), direction)
    }

    /// Assert scalar route-continuation invariants against this runtime context.
    ///
    /// Keeps scalar continuation protocol sanity checks centralized in
    /// continuation runtime so load entrypoints consume one invariant boundary.
    pub(in crate::db::executor) fn debug_assert_route_continuation_invariants(
        &self,
        plan: &AccessPlannedQuery,
        route_continuation: RouteContinuationPlan,
    ) {
        debug_assert!(
            route_continuation.strict_advance_required_when_applied(),
            "route invariant: continuation executions must enforce strict advancement policy",
        );
        debug_assert_eq!(
            route_continuation.effective_offset(),
            continuation_page_offset_for_window(plan, self.cursor_boundary().is_some()),
            "route window effective offset must match logical plan offset semantics",
        );
    }

    /// Borrow optional scalar cursor boundary for post-access cursor semantics.
    #[must_use]
    pub(in crate::db::executor) const fn post_access_cursor_boundary(
        &self,
    ) -> Option<&CursorBoundary> {
        self.cursor_boundary()
    }

    /// Return whether this continuation context represents a resumed page.
    #[must_use]
    pub(in crate::db::executor) const fn continuation_applied(&self) -> bool {
        self.cursor_boundary.is_some()
    }

    /// Derive effective keep count (`offset + limit`) under this continuation context.
    #[must_use]
    pub(in crate::db::executor) fn keep_count_for_limit_window(
        &self,
        plan: &AccessPlannedQuery,
        limit: u32,
    ) -> usize {
        continuation_keep_count_for_limit(plan, self.continuation_applied(), limit)
    }

    /// Borrow optional previous index-range anchor.
    #[must_use]
    pub(in crate::db::executor) fn previous_index_range_anchor(&self) -> Option<&LoweredKey> {
        self.index_range_token().map(range_token_anchor_key)
    }

    /// Borrow continuation signature for this runtime continuation context.
    #[must_use]
    pub(in crate::db::executor) const fn continuation_signature(&self) -> ContinuationSignature {
        self.continuation_signature
            .expect("runtime scalar continuation context requires signature")
    }

    /// Validate load scan-budget hint preconditions under this continuation context.
    ///
    /// Bounded load scan hints are only valid for non-continuation executions on
    /// streaming-safe access shapes where access order is already final.
    pub(in crate::db::executor) fn validate_load_scan_budget_hint(
        &self,
        scan_budget_hint: Option<usize>,
        load_order_route_contract: LoadOrderRouteContract,
    ) -> Result<(), InternalError> {
        if scan_budget_hint.is_some() {
            if self.continuation_applied() {
                return Err(InternalError::query_executor_invariant(
                    "load page scan budget hint requires non-continuation execution",
                ));
            }
            if !load_order_route_contract.allows_streaming_load() {
                return Err(InternalError::query_executor_invariant(
                    "load page scan budget hint requires streaming-safe access shape",
                ));
            }
        }

        Ok(())
    }
}