// file: ks-pipeline/src/backfill.rs
// version: 11

//! Reusable HTTP transaction backfill orchestration.

use futures_util::FutureExt; // rust-rules: trait-import
use futures_util::StreamExt; // rust-rules: trait-import
use sha2::Digest; // rust-rules: trait-import

const AFTER_HISTORY_PAGE_SIZE: u16 = 1000;
const CANCELLATION_POLL_INTERVAL_MS: u64 = 25;

/// Address category used to identify one targeted signature history campaign.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum BackfillAddressKind {
    /// Executable Solana program address.
    Program,
    /// Token mint address.
    Token,
    /// Pool account address.
    Pool,
}

impl BackfillAddressKind {
    /// Returns the stable filter code prefix.
    pub fn code(&self) -> &'static str {
        return match self {
            crate::BackfillAddressKind::Program => "program",
            crate::BackfillAddressKind::Token => "token",
            crate::BackfillAddressKind::Pool => "pool",
        };
    }
}

/// Chronological direction relative to one anchor signature.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum BackfillDirection {
    /// Fetch signatures older than the anchor.
    Before,
    /// Fetch the nearest signatures newer than the anchor.
    After,
}

impl BackfillDirection {
    /// Returns the stable direction code.
    pub fn code(&self) -> &'static str {
        return match self {
            crate::BackfillDirection::Before => "before",
            crate::BackfillDirection::After => "after",
        };
    }
}

/// Candidate source used by one backfill campaign.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum BackfillSource {
    /// Explicit transaction signatures supplied by the operator.
    ExplicitSignatures(std::vec::Vec<std::string::String>),
    /// Signatures discovered from one address history.
    AddressHistory {
        /// Semantic category of the address.
        kind: crate::BackfillAddressKind,
        /// Program, token mint or pool address.
        address: std::string::String,
        /// Optional anchor signature. `Before` without an anchor starts from the latest history page.
        anchor_signature: std::option::Option<std::string::String>,
        /// Chronological direction relative to the anchor.
        direction: crate::BackfillDirection,
        /// Maximum number of signatures to hydrate.
        limit: usize,
    },
}

/// Complete bounded backfill request.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BackfillRequest {
    /// Endpoint role used by the HTTP pool.
    pub role: std::string::String,
    /// Requested commitment for history and transaction calls.
    pub commitment: std::string::String,
    /// Candidate source.
    pub source: crate::BackfillSource,
    /// Maximum RPC page size between 1 and 1000.
    pub page_size: u16,
    /// Maximum number of history pages inspected.
    pub max_pages: u32,
    /// Operator concurrency cap before endpoint role limits are applied.
    pub max_concurrent_requests: u32,
    /// Number of retries after the initial HTTP attempt.
    pub max_retries: u32,
}

impl BackfillRequest {
    /// Validates request bounds and source identifiers.
    pub fn validate(&self) -> ks_core::Result<()> {
        if self.role.trim().is_empty() {
            return std::result::Result::Err(ks_core::Error::config(
                "backfill endpoint role must not be empty",
            ));
        }
        let transaction_config_result =
            ks_onchain_transport::GetTransactionConfig::new(self.commitment.clone(), 0);
        if let std::result::Result::Err(error) = transaction_config_result {
            return std::result::Result::Err(error);
        }
        if self.page_size == 0 || self.page_size > 1000 {
            return std::result::Result::Err(ks_core::Error::config(
                "backfill page size must be between 1 and 1000",
            ));
        }
        if self.max_pages == 0 {
            return std::result::Result::Err(ks_core::Error::config(
                "backfill max pages must be greater than zero",
            ));
        }
        if self.max_concurrent_requests == 0 {
            return std::result::Result::Err(ks_core::Error::config(
                "backfill concurrency must be greater than zero",
            ));
        }
        match &self.source {
            crate::BackfillSource::ExplicitSignatures(signatures) => {
                if signatures.is_empty() {
                    return std::result::Result::Err(ks_core::Error::config(
                        "explicit signature backfill requires at least one signature",
                    ));
                }
                for signature in signatures {
                    let validation_result =
                        ks_onchain_transport::validate_transaction_signature_text(
                            signature.as_str(),
                            "explicit backfill signature",
                        );
                    if let std::result::Result::Err(error) = validation_result {
                        return std::result::Result::Err(error);
                    }
                }
            },
            crate::BackfillSource::AddressHistory {
                address,
                anchor_signature,
                direction,
                limit,
                ..
            } => {
                let address_result = ks_onchain_transport::validate_solana_pubkey_text(
                    address.as_str(),
                    "backfill history address",
                );
                if let std::result::Result::Err(error) = address_result {
                    return std::result::Result::Err(error);
                }
                if let std::option::Option::Some(signature) = anchor_signature {
                    let signature_result =
                        ks_onchain_transport::validate_transaction_signature_text(
                            signature.as_str(),
                            "backfill anchor signature",
                        );
                    if let std::result::Result::Err(error) = signature_result {
                        return std::result::Result::Err(error);
                    }
                } else if *direction == crate::BackfillDirection::After {
                    return std::result::Result::Err(ks_core::Error::config(
                        "newer address history backfill requires an anchor signature",
                    ));
                }
                if *limit == 0 {
                    return std::result::Result::Err(ks_core::Error::config(
                        "address history backfill limit must be greater than zero",
                    ));
                }
            },
        }
        return std::result::Result::Ok(());
    }

    /// Returns a stable filter code for transaction observations.
    pub fn filter_code(&self) -> std::string::String {
        return match &self.source {
            crate::BackfillSource::ExplicitSignatures(_) => "explicit_signatures".to_string(),
            crate::BackfillSource::AddressHistory { kind, anchor_signature, direction, .. } => {
                if *direction == crate::BackfillDirection::Before && anchor_signature.is_none() {
                    return format!("{}_latest", kind.code());
                }
                format!("{}_{}", kind.code(), direction.code())
            },
        };
    }
}

/// Backfill progress severity.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum BackfillProgressLevel {
    /// Diagnostic detail.
    Debug,
    /// Normal campaign information.
    Info,
    /// Recoverable issue or retry.
    Warning,
    /// Non-recoverable item or campaign failure.
    Error,
}

impl BackfillProgressLevel {
    /// Returns the stable lowercase level code.
    pub fn code(&self) -> &'static str {
        return match self {
            crate::BackfillProgressLevel::Debug => "debug",
            crate::BackfillProgressLevel::Info => "info",
            crate::BackfillProgressLevel::Warning => "warn",
            crate::BackfillProgressLevel::Error => "error",
        };
    }
}

/// One operator-visible progress event.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BackfillProgressEvent {
    /// UTC timestamp rendered in RFC 3339.
    pub timestamp: std::string::String,
    /// Severity.
    pub level: crate::BackfillProgressLevel,
    /// Human-readable message.
    pub message: std::string::String,
    /// Number of candidates completed when known.
    pub completed: std::option::Option<u64>,
    /// Total candidates when known.
    pub total: std::option::Option<u64>,
}

impl BackfillProgressEvent {
    fn new(
        level: crate::BackfillProgressLevel,
        message: impl std::convert::Into<std::string::String>,
        completed: std::option::Option<u64>,
        total: std::option::Option<u64>,
    ) -> Self {
        return Self {
            timestamp: chrono::Utc::now().to_rfc3339(),
            level,
            message: message.into(),
            completed,
            total,
        };
    }
}

/// Progress and cancellation contract implemented by applications.
pub trait BackfillObserver: Sync {
    /// Receives one progress event.
    fn on_progress(&self, event: &crate::BackfillProgressEvent);
    /// Returns true when the operator requested cancellation.
    fn is_cancelled(&self) -> bool;
}

/// Final counters and pagination information for one campaign.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BackfillSummary {
    /// Unique capture session identifier.
    pub capture_session_id: std::string::String,
    /// Stable filter code.
    pub filter_code: std::string::String,
    /// Endpoint role.
    pub role: std::string::String,
    /// Provider used for transaction hydration.
    pub provider: std::string::String,
    /// Endpoint code used for transaction hydration.
    pub endpoint_code: std::string::String,
    /// Number of history pages fetched.
    pub pages_fetched: u64,
    /// Number of unique candidates selected.
    pub candidates_selected: u64,
    /// Number of candidates admitted to the bounded execution queue.
    pub candidates_started: u64,
    /// Number of candidates that reached a terminal outcome.
    pub candidates_completed: u64,
    /// Number of admitted candidates interrupted before a terminal outcome.
    pub candidates_cancelled: u64,
    /// Number of selected candidates never admitted after cancellation.
    pub candidates_not_started: u64,
    /// Number of source transactions received and normalized.
    pub transactions_received: u64,
    /// Number of canonical transaction rows inserted.
    pub canonical_inserted: u64,
    /// Number of canonical inserts skipped by idempotence.
    pub canonical_skipped: u64,
    /// Number of signatures skipped because they already existed before hydration.
    pub existing_skipped: u64,
    /// Number of missing transaction results after retries.
    pub missing: u64,
    /// Number of failed candidates after retries or persistence errors.
    pub failed: u64,
    /// Number of observation rows inserted.
    pub observations_inserted: u64,
    /// Number of source attempts performed.
    pub attempts: u64,
    /// Whether the campaign stopped after a cancellation request.
    pub cancelled: bool,
    /// Optional cursor suitable for continuing an older-history scan.
    pub resume_before_signature: std::option::Option<std::string::String>,
    /// Campaign start time.
    pub started_at: std::string::String,
    /// Campaign completion time.
    pub finished_at: std::string::String,
}

#[derive(Clone, Debug)]
struct BackfillCandidate {
    signature: std::string::String,
    slot: std::option::Option<u64>,
    detected_at: chrono::DateTime<chrono::Utc>,
}

#[derive(Clone, Copy, Debug)]
struct HttpRoleLimits {
    requests_per_second: u32,
    max_concurrent_requests: u32,
    pause_after_rate_limit_ms: u64,
}

#[derive(Debug)]
struct RequestPacer {
    interval: std::time::Duration,
    next_start: tokio::sync::Mutex<tokio::time::Instant>,
}

impl RequestPacer {
    fn new(requests_per_second: u32) -> Self {
        let micros = 1_000_000_u64.div_ceil(u64::from(requests_per_second));
        return Self {
            interval: std::time::Duration::from_micros(micros),
            next_start: tokio::sync::Mutex::new(tokio::time::Instant::now()),
        };
    }

    async fn wait(&self) {
        let mut next_start = self.next_start.lock().await;
        let now = tokio::time::Instant::now();
        if *next_start > now {
            tokio::time::sleep(*next_start - now).await;
        }
        *next_start = tokio::time::Instant::now() + self.interval;
    }
}

async fn wait_until_cancelled<O>(observer: &O)
where
    O: crate::BackfillObserver + Sync,
{
    loop {
        if observer.is_cancelled() {
            return;
        }
        tokio::time::sleep(std::time::Duration::from_millis(CANCELLATION_POLL_INTERVAL_MS)).await;
    }
}

async fn await_or_cancelled<O, F, T>(observer: &O, future: F) -> std::option::Option<T>
where
    O: crate::BackfillObserver + Sync,
    F: std::future::Future<Output = T>,
{
    if observer.is_cancelled() {
        return std::option::Option::None;
    }
    tokio::pin!(future);
    let outcome = tokio::select! {
        _ = wait_until_cancelled(observer) => std::option::Option::None,
        value = &mut future => std::option::Option::Some(value),
    };
    return outcome;
}

async fn wait_with_cancellation<O>(observer: &O, duration: std::time::Duration) -> bool
where
    O: crate::BackfillObserver + Sync,
{
    let outcome = await_or_cancelled(observer, tokio::time::sleep(duration)).await;
    return outcome.is_some();
}

#[derive(Clone, Debug)]
struct CandidateOutcome {
    transactions_received: u64,
    canonical_inserted: u64,
    canonical_skipped: u64,
    existing_skipped: u64,
    missing: u64,
    failed: u64,
    observations_inserted: u64,
    attempts: u64,
    cancelled: bool,
}

impl CandidateOutcome {
    fn empty() -> Self {
        return Self {
            transactions_received: 0,
            canonical_inserted: 0,
            canonical_skipped: 0,
            existing_skipped: 0,
            missing: 0,
            failed: 0,
            observations_inserted: 0,
            attempts: 0,
            cancelled: false,
        };
    }
}

#[derive(Debug)]
struct CandidateTaskResult {
    index: usize,
    result: ks_core::Result<CandidateOutcome>,
}

#[derive(Debug)]
struct CandidateCompletionFrontier {
    completed: std::vec::Vec<bool>,
    contiguous_completed: usize,
}

impl CandidateCompletionFrontier {
    fn new(candidate_count: usize) -> Self {
        return Self {
            completed: vec![false; candidate_count],
            contiguous_completed: 0,
        };
    }

    fn mark_completed(&mut self, index: usize) {
        if let std::option::Option::Some(completed) = self.completed.get_mut(index) {
            *completed = true;
        }
        while self.completed.get(self.contiguous_completed).copied().unwrap_or(false) {
            self.contiguous_completed += 1;
        }
    }

    fn contiguous_completed(&self) -> usize {
        return self.contiguous_completed;
    }
}

/// Executes one bounded HTTP backfill and persists canonical transactions plus observations.
pub async fn execute_http_backfill<S, O>(
    http_pool: &ks_onchain_transport::HttpEndpointPool,
    store: &S,
    request: &crate::BackfillRequest,
    observer: &O,
) -> ks_core::Result<crate::BackfillSummary>
where
    S: ks_store::RawTransactionStore + Sync,
    O: crate::BackfillObserver + Sync,
{
    let validation_result = request.validate();
    if let std::result::Result::Err(error) = validation_result {
        tracing::error!(target: crate::TRACING_TARGET, action = "validate_http_backfill", role = %request.role, error = %error, "HTTP backfill request validation failed");
        return std::result::Result::Err(error);
    }
    let started_at = chrono::Utc::now();
    let capture_session_id = uuid::Uuid::new_v4().to_string();
    let filter_code = request.filter_code();
    tracing::info!(target: crate::TRACING_TARGET, action = "execute_http_backfill", capture_session_id = %capture_session_id, filter_code = %filter_code, role = %request.role, commitment = %request.commitment, page_size = request.page_size, max_pages = request.max_pages, requested_concurrency = request.max_concurrent_requests, max_retries = request.max_retries, "HTTP backfill campaign started");
    emit(
        observer,
        crate::BackfillProgressLevel::Info,
        format!("backfill session {capture_session_id} started with filter {filter_code}"),
        std::option::Option::None,
        std::option::Option::None,
    );
    let transaction_client_result =
        http_pool.select_client_for_role_and_method(request.role.as_str(), "getTransaction");
    let transaction_client = match transaction_client_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => {
            tracing::error!(target: crate::TRACING_TARGET, action = "select_backfill_transaction_client", capture_session_id = %capture_session_id, filter_code = %filter_code, role = %request.role, error = %error, "backfill transaction client selection failed");
            return std::result::Result::Err(error);
        },
    };
    let transaction_limits_result =
        role_limits(&transaction_client, request.role.as_str(), "get_transaction");
    let transaction_limits = match transaction_limits_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => {
            tracing::error!(target: crate::TRACING_TARGET, action = "resolve_backfill_role_limits", capture_session_id = %capture_session_id, filter_code = %filter_code, role = %request.role, endpoint_name = %transaction_client.endpoint_name(), error = %error, "backfill role limit resolution failed");
            return std::result::Result::Err(error);
        },
    };
    let discovery_result = collect_candidates(http_pool, request, observer).await;
    let discovery = match discovery_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => {
            tracing::error!(target: crate::TRACING_TARGET, action = "collect_backfill_candidates", capture_session_id = %capture_session_id, filter_code = %filter_code, role = %request.role, error = %error, "backfill candidate discovery failed");
            return std::result::Result::Err(error);
        },
    };
    let total = match u64::try_from(discovery.candidates.len()) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => u64::MAX,
    };
    let ordered_signatures = discovery
        .candidates
        .iter()
        .map(|candidate| return candidate.signature.clone())
        .collect::<std::vec::Vec<std::string::String>>();
    let discovery_resume_before_signature = discovery.resume_before_signature.clone();
    let mut summary = crate::BackfillSummary {
        capture_session_id: capture_session_id.clone(),
        filter_code: filter_code.clone(),
        role: request.role.clone(),
        provider: transaction_client.provider().to_string(),
        endpoint_code: transaction_client.endpoint_name().to_string(),
        pages_fetched: discovery.pages_fetched,
        candidates_selected: total,
        candidates_started: 0,
        candidates_completed: 0,
        candidates_cancelled: 0,
        candidates_not_started: 0,
        transactions_received: 0,
        canonical_inserted: 0,
        canonical_skipped: 0,
        existing_skipped: 0,
        missing: 0,
        failed: 0,
        observations_inserted: 0,
        attempts: 0,
        cancelled: false,
        resume_before_signature: discovery_resume_before_signature.clone(),
        started_at: started_at.to_rfc3339(),
        finished_at: started_at.to_rfc3339(),
    };
    if observer.is_cancelled() {
        tracing::warn!(target: crate::TRACING_TARGET, action = "execute_http_backfill", capture_session_id = %capture_session_id, filter_code = %filter_code, candidates_selected = total, decision = "cancelled_before_admission", "HTTP backfill cancelled before candidate admission");
        summary.cancelled = true;
        summary.candidates_not_started = total;
        summary.resume_before_signature = resolved_resume_before_signature(
            request,
            ordered_signatures.as_slice(),
            0,
            discovery_resume_before_signature,
            true,
        );
        summary.finished_at = chrono::Utc::now().to_rfc3339();
        return std::result::Result::Ok(summary);
    }
    let effective_concurrency = std::cmp::max(
        1_usize,
        std::cmp::min(
            request.max_concurrent_requests as usize,
            transaction_limits.max_concurrent_requests as usize,
        ),
    );
    let pacer = std::sync::Arc::new(RequestPacer::new(transaction_limits.requests_per_second));
    let transaction_config_result =
        ks_onchain_transport::GetTransactionConfig::new(request.commitment.clone(), 0);
    let transaction_config = match transaction_config_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let mut pending = discovery.candidates.into_iter().enumerate();
    let mut in_flight = futures_util::stream::FuturesUnordered::new();
    let mut completion_frontier = CandidateCompletionFrontier::new(ordered_signatures.len());
    let create_task = |index: usize, candidate: BackfillCandidate| {
        let pacer_value = std::sync::Arc::clone(&pacer);
        let capture_session_id_value = capture_session_id.clone();
        let filter_code_value = filter_code.clone();
        let transaction_client_value = transaction_client.clone();
        let transaction_config_value = transaction_config.clone();
        return async move {
            let result = hydrate_candidate(
                store,
                observer,
                &transaction_client_value,
                &transaction_config_value,
                request,
                &capture_session_id_value,
                &filter_code_value,
                candidate,
                pacer_value.as_ref(),
                transaction_limits.pause_after_rate_limit_ms,
            )
            .await;
            return CandidateTaskResult { index, result };
        }
        .boxed();
    };
    while in_flight.len() < effective_concurrency && !observer.is_cancelled() {
        let next_candidate = pending.next();
        let (index, candidate) = match next_candidate {
            std::option::Option::Some(value) => value,
            std::option::Option::None => break,
        };
        summary.candidates_started += 1;
        in_flight.push(create_task(index, candidate));
    }
    while let std::option::Option::Some(task) = in_flight.next().await {
        let index = task.index;
        match task.result {
            std::result::Result::Ok(outcome) => {
                if outcome.cancelled {
                    summary.candidates_cancelled += 1;
                    summary.cancelled = true;
                    emit(
                        observer,
                        crate::BackfillProgressLevel::Debug,
                        format!("candidate {} cancelled before terminal processing", index + 1),
                        std::option::Option::Some(summary.candidates_completed),
                        std::option::Option::Some(total),
                    );
                } else {
                    summary.candidates_completed += 1;
                    summary.transactions_received += outcome.transactions_received;
                    summary.canonical_inserted += outcome.canonical_inserted;
                    summary.canonical_skipped += outcome.canonical_skipped;
                    summary.existing_skipped += outcome.existing_skipped;
                    summary.missing += outcome.missing;
                    summary.failed += outcome.failed;
                    summary.observations_inserted += outcome.observations_inserted;
                    summary.attempts += outcome.attempts;
                    completion_frontier.mark_completed(index);
                    emit(
                        observer,
                        crate::BackfillProgressLevel::Info,
                        format!("processed {}/{} candidates", summary.candidates_completed, total),
                        std::option::Option::Some(summary.candidates_completed),
                        std::option::Option::Some(total),
                    );
                }
            },
            std::result::Result::Err(error) => {
                tracing::error!(target: crate::TRACING_TARGET, action = "hydrate_backfill_candidate", capture_session_id = %capture_session_id, filter_code = %filter_code, candidate_index = index, error = %error, "backfill candidate task failed without a normal outcome");
                summary.candidates_completed += 1;
                summary.failed += 1;
                completion_frontier.mark_completed(index);
                emit(
                    observer,
                    crate::BackfillProgressLevel::Error,
                    format!("candidate {} failed without a normal outcome: {error}", index + 1),
                    std::option::Option::Some(summary.candidates_completed),
                    std::option::Option::Some(total),
                );
            },
        }
        if observer.is_cancelled() {
            summary.cancelled = true;
        } else {
            let next_candidate = pending.next();
            if let std::option::Option::Some((next_index, next_candidate_value)) = next_candidate {
                summary.candidates_started += 1;
                in_flight.push(create_task(next_index, next_candidate_value));
            }
        }
    }
    if observer.is_cancelled() {
        summary.cancelled = true;
    }
    summary.candidates_not_started = total.saturating_sub(summary.candidates_started);
    if summary.candidates_cancelled > 0 || summary.candidates_not_started > 0 {
        summary.cancelled = true;
    }
    summary.resume_before_signature = resolved_resume_before_signature(
        request,
        ordered_signatures.as_slice(),
        completion_frontier.contiguous_completed(),
        discovery_resume_before_signature,
        summary.cancelled,
    );
    summary.finished_at = chrono::Utc::now().to_rfc3339();
    let completion_level = if summary.cancelled {
        crate::BackfillProgressLevel::Warning
    } else {
        crate::BackfillProgressLevel::Info
    };
    if summary.failed > 0 {
        tracing::error!(target: crate::TRACING_TARGET, action = "execute_http_backfill", capture_session_id = %summary.capture_session_id, filter_code = %summary.filter_code, endpoint_name = %summary.endpoint_code, candidates_selected = summary.candidates_selected, candidates_completed = summary.candidates_completed, failed = summary.failed, missing = summary.missing, cancelled = summary.cancelled, "HTTP backfill completed with failed candidates");
    } else if summary.cancelled {
        tracing::warn!(target: crate::TRACING_TARGET, action = "execute_http_backfill", capture_session_id = %summary.capture_session_id, filter_code = %summary.filter_code, endpoint_name = %summary.endpoint_code, candidates_selected = summary.candidates_selected, candidates_completed = summary.candidates_completed, candidates_cancelled = summary.candidates_cancelled, candidates_not_started = summary.candidates_not_started, "HTTP backfill campaign cancelled");
    } else {
        tracing::info!(target: crate::TRACING_TARGET, action = "execute_http_backfill", capture_session_id = %summary.capture_session_id, filter_code = %summary.filter_code, endpoint_name = %summary.endpoint_code, candidates_selected = summary.candidates_selected, candidates_completed = summary.candidates_completed, canonical_inserted = summary.canonical_inserted, canonical_skipped = summary.canonical_skipped, existing_skipped = summary.existing_skipped, missing = summary.missing, observations_inserted = summary.observations_inserted, "HTTP backfill campaign completed");
    }
    emit(
        observer,
        completion_level,
        format!(
            "backfill completed: completed={}, candidate_cancelled={}, not_started={}, inserted={}, skipped={}, existing={}, missing={}, failed={}, observations={}",
            summary.candidates_completed,
            summary.candidates_cancelled,
            summary.candidates_not_started,
            summary.canonical_inserted,
            summary.canonical_skipped,
            summary.existing_skipped,
            summary.missing,
            summary.failed,
            summary.observations_inserted
        ),
        std::option::Option::Some(summary.candidates_completed),
        std::option::Option::Some(total),
    );
    return std::result::Result::Ok(summary);
}

fn resolved_resume_before_signature(
    request: &crate::BackfillRequest,
    ordered_signatures: &[std::string::String],
    contiguous_completed: usize,
    discovery_resume_before_signature: std::option::Option<std::string::String>,
    cancelled: bool,
) -> std::option::Option<std::string::String> {
    if !cancelled {
        return discovery_resume_before_signature;
    }
    return match &request.source {
        crate::BackfillSource::AddressHistory {
            anchor_signature,
            direction: crate::BackfillDirection::Before,
            ..
        } => {
            if contiguous_completed == 0 {
                return anchor_signature.clone();
            }
            if let std::option::Option::Some(signature) =
                ordered_signatures.get(contiguous_completed - 1)
            {
                return std::option::Option::Some(signature.clone());
            }
            anchor_signature.clone()
        },
        crate::BackfillSource::ExplicitSignatures(_)
        | crate::BackfillSource::AddressHistory {
            direction: crate::BackfillDirection::After,
            ..
        } => std::option::Option::None,
    };
}

struct CandidateDiscovery {
    candidates: std::vec::Vec<BackfillCandidate>,
    pages_fetched: u64,
    resume_before_signature: std::option::Option<std::string::String>,
}

async fn collect_candidates<O>(
    http_pool: &ks_onchain_transport::HttpEndpointPool,
    request: &crate::BackfillRequest,
    observer: &O,
) -> ks_core::Result<CandidateDiscovery>
where
    O: crate::BackfillObserver + Sync,
{
    return match &request.source {
        crate::BackfillSource::ExplicitSignatures(signatures) => {
            std::result::Result::Ok(explicit_candidates(signatures))
        },
        crate::BackfillSource::AddressHistory {
            address,
            anchor_signature,
            direction,
            limit,
            ..
        } => {
            let client_result = http_pool.select_client_for_role_and_method(
                request.role.as_str(),
                "getSignaturesForAddress",
            );
            let client = match client_result {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
            let limits_result =
                role_limits(&client, request.role.as_str(), "get_signatures_for_address");
            let limits = match limits_result {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
            let pacer = RequestPacer::new(limits.requests_per_second);
            match direction {
                crate::BackfillDirection::Before => {
                    collect_before_candidates(
                        &client,
                        request,
                        observer,
                        address,
                        anchor_signature.as_deref(),
                        *limit,
                        &pacer,
                        limits.pause_after_rate_limit_ms,
                    )
                    .await
                },
                crate::BackfillDirection::After => {
                    collect_after_candidates(
                        &client,
                        request,
                        observer,
                        address,
                        match anchor_signature.as_deref() {
                            std::option::Option::Some(value) => value,
                            std::option::Option::None => {
                                return std::result::Result::Err(ks_core::Error::config(
                                    "newer address history backfill requires an anchor signature",
                                ));
                            },
                        },
                        *limit,
                        &pacer,
                        limits.pause_after_rate_limit_ms,
                    )
                    .await
                },
            }
        },
    };
}

fn explicit_candidates(signatures: &[std::string::String]) -> CandidateDiscovery {
    let mut seen = std::collections::HashSet::<std::string::String>::new();
    let mut candidates = std::vec::Vec::<BackfillCandidate>::new();
    let detected_at = chrono::Utc::now();
    for signature in signatures {
        let trimmed = signature.trim();
        if trimmed.is_empty() || !seen.insert(trimmed.to_string()) {
            continue;
        }
        candidates.push(BackfillCandidate {
            signature: trimmed.to_string(),
            slot: std::option::Option::None,
            detected_at,
        });
    }
    return CandidateDiscovery {
        candidates,
        pages_fetched: 0,
        resume_before_signature: std::option::Option::None,
    };
}

async fn collect_before_candidates<O>(
    client: &ks_onchain_transport::HttpClient,
    request: &crate::BackfillRequest,
    observer: &O,
    address: &str,
    anchor_signature: std::option::Option<&str>,
    limit: usize,
    pacer: &RequestPacer,
    pause_after_rate_limit_ms: u64,
) -> ks_core::Result<CandidateDiscovery>
where
    O: crate::BackfillObserver + Sync,
{
    let mut candidates = std::vec::Vec::<BackfillCandidate>::new();
    let mut seen = std::collections::HashSet::<std::string::String>::new();
    let mut before = anchor_signature.map(str::to_string);
    let mut pages_fetched = 0_u64;
    while candidates.len() < limit && pages_fetched < u64::from(request.max_pages) {
        if observer.is_cancelled() {
            break;
        }
        let remaining = limit - candidates.len();
        let page_limit = std::cmp::min(usize::from(request.page_size), remaining);
        let page_limit_u16 = match u16::try_from(page_limit) {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => {
                return std::result::Result::Err(ks_core::Error::config(format!(
                    "backfill page limit conversion failed: {error}"
                )));
            },
        };
        let config_result = ks_onchain_transport::GetSignaturesForAddressConfig::new(
            request.commitment.clone(),
            page_limit_u16,
            before.clone(),
            std::option::Option::None,
            std::option::Option::None,
        );
        let config = match config_result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        let page_result = fetch_signature_page(
            client,
            address,
            &config,
            request.max_retries,
            pacer,
            pause_after_rate_limit_ms,
            observer,
        )
        .await;
        let page = match page_result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        if observer.is_cancelled() {
            break;
        }
        pages_fetched += 1;
        append_signature_candidates(&mut candidates, &mut seen, &page);
        before = page.last().map(|item| return item.signature.clone());
        emit(
            observer,
            crate::BackfillProgressLevel::Debug,
            format!("history page {pages_fetched} returned {} signatures", page.len()),
            std::option::Option::None,
            std::option::Option::None,
        );
        if page.len() < page_limit || page.is_empty() {
            break;
        }
    }
    let resume_before_signature = before;
    return std::result::Result::Ok(CandidateDiscovery {
        candidates,
        pages_fetched,
        resume_before_signature,
    });
}

async fn collect_after_candidates<O>(
    client: &ks_onchain_transport::HttpClient,
    request: &crate::BackfillRequest,
    observer: &O,
    address: &str,
    anchor_signature: &str,
    limit: usize,
    pacer: &RequestPacer,
    pause_after_rate_limit_ms: u64,
) -> ks_core::Result<CandidateDiscovery>
where
    O: crate::BackfillObserver + Sync,
{
    let page_size = AFTER_HISTORY_PAGE_SIZE;
    let mut nearest_newer = std::collections::VecDeque::<BackfillCandidate>::with_capacity(limit);
    let mut seen = std::collections::HashSet::<std::string::String>::new();
    let mut before = std::option::Option::None;
    let until = std::option::Option::Some(anchor_signature.to_string());
    let mut pages_fetched = 0_u64;
    let mut scanned = 0_u64;
    let mut boundary_reached = false;
    while pages_fetched < u64::from(request.max_pages) {
        if observer.is_cancelled() {
            break;
        }
        let config_result = ks_onchain_transport::GetSignaturesForAddressConfig::new(
            request.commitment.clone(),
            page_size,
            before.clone(),
            until.clone(),
            std::option::Option::None,
        );
        let config = match config_result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        let page_result = fetch_signature_page(
            client,
            address,
            &config,
            request.max_retries,
            pacer,
            pause_after_rate_limit_ms,
            observer,
        )
        .await;
        let page = match page_result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        if observer.is_cancelled() {
            break;
        }
        pages_fetched += 1;
        let page_count = match u64::try_from(page.len()) {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => {
                return std::result::Result::Err(ks_core::Error::config(format!(
                    "after-history page length conversion failed: {error}"
                )));
            },
        };
        scanned = scanned.saturating_add(page_count);
        append_nearest_newer_candidates(&mut nearest_newer, &mut seen, &page, limit);
        before = page.last().map(|item| return item.signature.clone());
        let oldest_scanned_slot = page.last().map(|item| return item.slot);
        emit(
            observer,
            crate::BackfillProgressLevel::Debug,
            format!(
                "newer-history page={pages_fetched} page_size={} scanned={scanned} oldest_scanned_slot={oldest_scanned_slot:?}",
                page.len()
            ),
            std::option::Option::None,
            std::option::Option::None,
        );
        if page.len() < usize::from(page_size) {
            boundary_reached = true;
            break;
        }
    }
    if !boundary_reached && !observer.is_cancelled() {
        return std::result::Result::Err(ks_core::Error::config(format!(
            "anchor boundary was not reached within {} pages after scanning {scanned} signatures; increase max_pages",
            request.max_pages
        )));
    }
    return std::result::Result::Ok(CandidateDiscovery {
        candidates: nearest_newer.into_iter().collect(),
        pages_fetched,
        resume_before_signature: std::option::Option::None,
    });
}

fn append_nearest_newer_candidates(
    candidates: &mut std::collections::VecDeque<BackfillCandidate>,
    seen: &mut std::collections::HashSet<std::string::String>,
    page: &[ks_onchain_transport::AddressSignatureInfo],
    limit: usize,
) {
    let detected_at = chrono::Utc::now();
    for item in page {
        if !seen.insert(item.signature.clone()) {
            continue;
        }
        candidates.push_back(BackfillCandidate {
            signature: item.signature.clone(),
            slot: std::option::Option::Some(item.slot),
            detected_at,
        });
        if candidates.len() > limit {
            candidates.pop_front();
        }
    }
}

fn append_signature_candidates(
    candidates: &mut std::vec::Vec<BackfillCandidate>,
    seen: &mut std::collections::HashSet<std::string::String>,
    page: &[ks_onchain_transport::AddressSignatureInfo],
) {
    let detected_at = chrono::Utc::now();
    for item in page {
        if !seen.insert(item.signature.clone()) {
            continue;
        }
        candidates.push(BackfillCandidate {
            signature: item.signature.clone(),
            slot: std::option::Option::Some(item.slot),
            detected_at,
        });
    }
}

async fn fetch_signature_page<O>(
    client: &ks_onchain_transport::HttpClient,
    address: &str,
    config: &ks_onchain_transport::GetSignaturesForAddressConfig,
    max_retries: u32,
    pacer: &RequestPacer,
    pause_after_rate_limit_ms: u64,
    observer: &O,
) -> ks_core::Result<std::vec::Vec<ks_onchain_transport::AddressSignatureInfo>>
where
    O: crate::BackfillObserver + Sync,
{
    let mut attempt = 0_u32;
    loop {
        if observer.is_cancelled() {
            return std::result::Result::Ok(std::vec::Vec::new());
        }
        let pacing_result = await_or_cancelled(observer, pacer.wait()).await;
        if pacing_result.is_none() {
            return std::result::Result::Ok(std::vec::Vec::new());
        }
        let request_result =
            await_or_cancelled(observer, client.get_signatures_for_address(address, config)).await;
        let result = match request_result {
            std::option::Option::Some(value) => value,
            std::option::Option::None => return std::result::Result::Ok(std::vec::Vec::new()),
        };
        match result {
            std::result::Result::Ok(page) => return std::result::Result::Ok(page),
            std::result::Result::Err(error) => {
                if attempt >= max_retries {
                    return std::result::Result::Err(error);
                }
                let error_text = error.to_string();
                let delay_ms =
                    retry_delay_ms(error_text.as_str(), attempt, pause_after_rate_limit_ms);
                emit(
                    observer,
                    crate::BackfillProgressLevel::Warning,
                    format!(
                        "getSignaturesForAddress retry {} after {} ms: {}",
                        attempt + 1,
                        delay_ms,
                        error_text
                    ),
                    std::option::Option::None,
                    std::option::Option::None,
                );
                let wait_completed =
                    wait_with_cancellation(observer, std::time::Duration::from_millis(delay_ms))
                        .await;
                if !wait_completed {
                    return std::result::Result::Ok(std::vec::Vec::new());
                }
                attempt += 1;
            },
        }
    }
}

async fn hydrate_candidate<S, O>(
    store: &S,
    observer: &O,
    client: &ks_onchain_transport::HttpClient,
    config: &ks_onchain_transport::GetTransactionConfig,
    request: &crate::BackfillRequest,
    capture_session_id: &str,
    filter_code: &str,
    candidate: BackfillCandidate,
    pacer: &RequestPacer,
    pause_after_rate_limit_ms: u64,
) -> ks_core::Result<CandidateOutcome>
where
    S: ks_store::RawTransactionStore + Sync,
    O: crate::BackfillObserver + Sync,
{
    let mut outcome = CandidateOutcome::empty();
    if observer.is_cancelled() {
        outcome.cancelled = true;
        return std::result::Result::Ok(outcome);
    }
    let signature_model = ks_lib::MdSignature(candidate.signature.clone());
    let exists_result =
        ks_store::RawTransactionStore::has_raw_transaction_signature(store, &signature_model).await;
    let exists = match exists_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    if exists {
        outcome.existing_skipped = 1;
        return std::result::Result::Ok(outcome);
    }
    let mut attempt = 0_u32;
    loop {
        if observer.is_cancelled() {
            outcome.cancelled = true;
            return std::result::Result::Ok(outcome);
        }
        let pacing_result = await_or_cancelled(observer, pacer.wait()).await;
        if pacing_result.is_none() {
            outcome.cancelled = true;
            return std::result::Result::Ok(outcome);
        }
        outcome.attempts += 1;
        let request_result = await_or_cancelled(
            observer,
            client.get_transaction_acquisition(candidate.signature.as_str(), config),
        )
        .await;
        let acquisition_result = match request_result {
            std::option::Option::Some(value) => value,
            std::option::Option::None => {
                outcome.cancelled = true;
                return std::result::Result::Ok(outcome);
            },
        };
        let received_at = chrono::Utc::now();
        match acquisition_result {
            std::result::Result::Ok(acquisition) => {
                let metadata_result = source_payload_metadata(&acquisition.source_json);
                let (payload_size_bytes, source_payload_hash) = match metadata_result {
                    std::result::Result::Ok(value) => value,
                    std::result::Result::Err(error) => {
                        return std::result::Result::Err(error);
                    },
                };
                match acquisition.canonical_transaction {
                    std::option::Option::Some(transaction) => {
                        outcome.transactions_received = 1;
                        let normalized_at = chrono::Utc::now();
                        let raw_insert_result =
                            ks_store::RawTransactionInsert::from_canonical(&transaction);
                        let raw_insert = match raw_insert_result {
                            std::result::Result::Ok(value) => value,
                            std::result::Result::Err(error) => {
                                let observation_result = persist_failed_observation(
                                    store,
                                    client,
                                    request,
                                    capture_session_id,
                                    filter_code,
                                    &candidate,
                                    attempt,
                                    received_at,
                                    std::option::Option::Some(normalized_at),
                                    payload_size_bytes,
                                    source_payload_hash,
                                    "canonical_normalization_failed",
                                    error.to_string(),
                                )
                                .await;
                                if let std::result::Result::Ok(count) = observation_result {
                                    outcome.observations_inserted += count;
                                }
                                tracing::error!(target: crate::TRACING_TARGET, action = "normalize_backfill_transaction", capture_session_id, filter_code, signature = %candidate.signature, attempt, error = %error, "canonical transaction normalization failed");
                                outcome.failed = 1;
                                return std::result::Result::Ok(outcome);
                            },
                        };
                        let persist_result = ks_store::RawTransactionStore::insert_raw_transaction(
                            store,
                            &raw_insert,
                        )
                        .await;
                        let persist_outcome = match persist_result {
                            std::result::Result::Ok(value) => value,
                            std::result::Result::Err(error) => {
                                let observation_result = persist_failed_observation(
                                    store,
                                    client,
                                    request,
                                    capture_session_id,
                                    filter_code,
                                    &candidate,
                                    attempt,
                                    received_at,
                                    std::option::Option::Some(normalized_at),
                                    payload_size_bytes,
                                    source_payload_hash,
                                    "canonical_persist_failed",
                                    error.to_string(),
                                )
                                .await;
                                if let std::result::Result::Ok(count) = observation_result {
                                    outcome.observations_inserted += count;
                                }
                                tracing::error!(target: crate::TRACING_TARGET, action = "persist_backfill_transaction", capture_session_id, filter_code, signature = %candidate.signature, attempt, error = %error, "canonical transaction persistence failed");
                                outcome.failed = 1;
                                return std::result::Result::Ok(outcome);
                            },
                        };
                        outcome.canonical_inserted += persist_outcome.inserted_count;
                        outcome.canonical_skipped += persist_outcome.skipped_count;
                        let observation_result = persist_success_observation(
                            store,
                            client,
                            request,
                            capture_session_id,
                            filter_code,
                            &candidate,
                            transaction.slot,
                            attempt,
                            received_at,
                            normalized_at,
                            payload_size_bytes,
                            source_payload_hash,
                        )
                        .await;
                        let observation_count = match observation_result {
                            std::result::Result::Ok(value) => value,
                            std::result::Result::Err(error) => {
                                tracing::error!(target: crate::TRACING_TARGET, action = "persist_backfill_observation", capture_session_id, filter_code, signature = %candidate.signature, attempt, error = %error, "successful backfill observation persistence failed");
                                outcome.failed = 1;
                                emit(
                                    observer,
                                    crate::BackfillProgressLevel::Error,
                                    format!(
                                        "observation persistence failed for {}: {}",
                                        candidate.signature, error
                                    ),
                                    std::option::Option::None,
                                    std::option::Option::None,
                                );
                                return std::result::Result::Ok(outcome);
                            },
                        };
                        outcome.observations_inserted += observation_count;
                        return std::result::Result::Ok(outcome);
                    },
                    std::option::Option::None => {
                        let observation_result = persist_missing_observation(
                            store,
                            client,
                            request,
                            capture_session_id,
                            filter_code,
                            &candidate,
                            attempt,
                            received_at,
                            payload_size_bytes,
                            source_payload_hash,
                        )
                        .await;
                        let observation_count = match observation_result {
                            std::result::Result::Ok(value) => value,
                            std::result::Result::Err(error) => {
                                return std::result::Result::Err(error);
                            },
                        };
                        outcome.observations_inserted += observation_count;
                        if attempt >= request.max_retries {
                            outcome.missing = 1;
                            return std::result::Result::Ok(outcome);
                        }
                    },
                }
            },
            std::result::Result::Err(error) => {
                let error_text = error.to_string();
                let observation_result = persist_failed_observation(
                    store,
                    client,
                    request,
                    capture_session_id,
                    filter_code,
                    &candidate,
                    attempt,
                    received_at,
                    std::option::Option::None,
                    std::option::Option::None,
                    std::option::Option::None,
                    error_code(error_text.as_str()),
                    error_text.clone(),
                )
                .await;
                let observation_count = match observation_result {
                    std::result::Result::Ok(value) => value,
                    std::result::Result::Err(observation_error) => {
                        return std::result::Result::Err(observation_error);
                    },
                };
                outcome.observations_inserted += observation_count;
                if attempt >= request.max_retries {
                    tracing::error!(target: crate::TRACING_TARGET, action = "hydrate_backfill_candidate", capture_session_id, filter_code, endpoint_name = %client.endpoint_name(), provider = %client.provider(), signature = %candidate.signature, attempt, max_retries = request.max_retries, error_code = error_code(error_text.as_str()), error_message = %error_text, "getTransaction failed after all retries");
                    outcome.failed = 1;
                    return std::result::Result::Ok(outcome);
                }
                let delay_ms =
                    retry_delay_ms(error_text.as_str(), attempt, pause_after_rate_limit_ms);
                emit(
                    observer,
                    crate::BackfillProgressLevel::Warning,
                    format!(
                        "getTransaction retry {} for {} after {} ms: {}",
                        attempt + 1,
                        candidate.signature,
                        delay_ms,
                        error_text
                    ),
                    std::option::Option::None,
                    std::option::Option::None,
                );
                let wait_completed =
                    wait_with_cancellation(observer, std::time::Duration::from_millis(delay_ms))
                        .await;
                if !wait_completed {
                    outcome.cancelled = true;
                    return std::result::Result::Ok(outcome);
                }
            },
        }
        attempt += 1;
    }
}

async fn persist_success_observation<S>(
    store: &S,
    client: &ks_onchain_transport::HttpClient,
    request: &crate::BackfillRequest,
    capture_session_id: &str,
    filter_code: &str,
    candidate: &BackfillCandidate,
    slot: u64,
    attempt: u32,
    received_at: chrono::DateTime<chrono::Utc>,
    normalized_at: chrono::DateTime<chrono::Utc>,
    payload_size_bytes: std::option::Option<u64>,
    source_payload_hash: std::option::Option<std::string::String>,
) -> ks_core::Result<u64>
where
    S: ks_store::RawTransactionStore + Sync,
{
    let observation_result = base_observation(
        client,
        request,
        capture_session_id,
        filter_code,
        candidate,
        attempt,
        received_at,
    );
    let observation_base = match observation_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let identity_result = observation_base
        .with_transaction_identity(candidate.signature.clone(), std::option::Option::Some(slot));
    let observation_with_identity = match identity_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let observation_with_timings = observation_with_identity.with_timings(
        std::option::Option::Some(candidate.detected_at),
        std::option::Option::Some(normalized_at),
    );
    let payload_result =
        observation_with_timings.with_payload_metadata(payload_size_bytes, source_payload_hash);
    let observation = match payload_result {
        std::result::Result::Ok(value) => {
            value.with_status(ks_store::TransactionObservationStatus::Persisted)
        },
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    return insert_observation(store, &observation).await;
}

async fn persist_missing_observation<S>(
    store: &S,
    client: &ks_onchain_transport::HttpClient,
    request: &crate::BackfillRequest,
    capture_session_id: &str,
    filter_code: &str,
    candidate: &BackfillCandidate,
    attempt: u32,
    received_at: chrono::DateTime<chrono::Utc>,
    payload_size_bytes: std::option::Option<u64>,
    source_payload_hash: std::option::Option<std::string::String>,
) -> ks_core::Result<u64>
where
    S: ks_store::RawTransactionStore + Sync,
{
    let observation_result = base_observation(
        client,
        request,
        capture_session_id,
        filter_code,
        candidate,
        attempt,
        received_at,
    );
    let observation_base = match observation_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let identity_result =
        observation_base.with_transaction_identity(candidate.signature.clone(), candidate.slot);
    let observation_with_identity = match identity_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let observation_with_timings = observation_with_identity
        .with_timings(std::option::Option::Some(candidate.detected_at), std::option::Option::None);
    let payload_result =
        observation_with_timings.with_payload_metadata(payload_size_bytes, source_payload_hash);
    let observation = match payload_result {
        std::result::Result::Ok(value) => {
            value.with_status(ks_store::TransactionObservationStatus::Missing)
        },
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    return insert_observation(store, &observation).await;
}

async fn persist_failed_observation<S>(
    store: &S,
    client: &ks_onchain_transport::HttpClient,
    request: &crate::BackfillRequest,
    capture_session_id: &str,
    filter_code: &str,
    candidate: &BackfillCandidate,
    attempt: u32,
    received_at: chrono::DateTime<chrono::Utc>,
    normalized_at: std::option::Option<chrono::DateTime<chrono::Utc>>,
    payload_size_bytes: std::option::Option<u64>,
    source_payload_hash: std::option::Option<std::string::String>,
    error_code: &str,
    error_message: std::string::String,
) -> ks_core::Result<u64>
where
    S: ks_store::RawTransactionStore + Sync,
{
    let observation_result = base_observation(
        client,
        request,
        capture_session_id,
        filter_code,
        candidate,
        attempt,
        received_at,
    );
    let observation_base = match observation_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let identity_result =
        observation_base.with_transaction_identity(candidate.signature.clone(), candidate.slot);
    let observation_with_identity = match identity_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let observation_with_timings = observation_with_identity
        .with_timings(std::option::Option::Some(candidate.detected_at), normalized_at);
    let payload_result =
        observation_with_timings.with_payload_metadata(payload_size_bytes, source_payload_hash);
    let observation_with_payload = match payload_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let error_result = observation_with_payload
        .with_error(error_code.to_string(), std::option::Option::Some(error_message));
    let observation = match error_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    return insert_observation(store, &observation).await;
}

fn base_observation(
    client: &ks_onchain_transport::HttpClient,
    request: &crate::BackfillRequest,
    capture_session_id: &str,
    filter_code: &str,
    candidate: &BackfillCandidate,
    attempt: u32,
    received_at: chrono::DateTime<chrono::Utc>,
) -> ks_core::Result<ks_store::TransactionObservationInsert> {
    let observation_key = format!(
        "backfill:{}:{}:{}:{}",
        capture_session_id,
        client.endpoint_name(),
        candidate.signature,
        attempt
    );
    let observation_result = ks_store::TransactionObservationInsert::new(
        observation_key,
        client.provider().to_string(),
        "solana_http_json_rpc".to_string(),
        "getTransaction".to_string(),
        ks_store::TransactionObservationOrigin::Backfill,
        received_at,
    );
    let observation = match observation_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let endpoint_result = observation.with_endpoint_code(client.endpoint_name().to_string());
    let endpoint_observation = match endpoint_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let commitment_result = endpoint_observation.with_commitment(request.commitment.clone());
    let commitment_observation = match commitment_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    return commitment_observation.with_capture_context(
        std::option::Option::Some(capture_session_id.to_string()),
        std::option::Option::Some(filter_code.to_string()),
    );
}

async fn insert_observation<S>(
    store: &S,
    observation: &ks_store::TransactionObservationInsert,
) -> ks_core::Result<u64>
where
    S: ks_store::RawTransactionStore + Sync,
{
    let insert_result =
        ks_store::RawTransactionStore::insert_transaction_observation(store, observation).await;
    let outcome = match insert_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    return std::result::Result::Ok(outcome.inserted_count);
}

fn source_payload_metadata(
    source: &serde_json::Value,
) -> ks_core::Result<(std::option::Option<u64>, std::option::Option<std::string::String>)> {
    let bytes_result = serde_json::to_vec(source);
    let bytes = match bytes_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => {
            return std::result::Result::Err(ks_core::Error::json(format!(
                "cannot serialize source getTransaction payload for metrics: {error}"
            )));
        },
    };
    let size = match u64::try_from(bytes.len()) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => {
            return std::result::Result::Err(ks_core::Error::json(format!(
                "source payload size conversion failed: {error}"
            )));
        },
    };
    let digest = sha2::Sha256::digest(bytes.as_slice());
    return std::result::Result::Ok((
        std::option::Option::Some(size),
        std::option::Option::Some(lower_hex(digest.as_ref())),
    ));
}

fn lower_hex(bytes: &[u8]) -> std::string::String {
    const HEX_DIGITS: &[u8; 16] = b"0123456789abcdef";
    let mut output = std::string::String::with_capacity(bytes.len().saturating_mul(2));
    for byte in bytes {
        output.push(char::from(HEX_DIGITS[usize::from(byte >> 4)]));
        output.push(char::from(HEX_DIGITS[usize::from(byte & 0x0f)]));
    }
    return output;
}

fn role_limits(
    client: &ks_onchain_transport::HttpClient,
    required_role: &str,
    request_kind: &str,
) -> ks_core::Result<HttpRoleLimits> {
    for role in &client.endpoint_config().roles {
        if !role.enabled || role.role != required_role {
            continue;
        }
        let supports =
            role.request_kinds.iter().any(|kind| return kind == request_kind || kind == "*");
        if !supports {
            continue;
        }
        return std::result::Result::Ok(HttpRoleLimits {
            requests_per_second: role.requests_per_second,
            max_concurrent_requests: role.max_concurrent_requests,
            pause_after_rate_limit_ms: role.pause_after_rate_limit_ms,
        });
    }
    return std::result::Result::Err(ks_core::Error::config(format!(
        "endpoint '{}' does not expose role '{}' for '{}'",
        client.endpoint_name(),
        required_role,
        request_kind
    )));
}

fn retry_delay_ms(error_text: &str, attempt: u32, pause_after_rate_limit_ms: u64) -> u64 {
    if error_text.contains("429") || error_text.to_ascii_lowercase().contains("rate limit") {
        return pause_after_rate_limit_ms;
    }
    let exponent = std::cmp::min(attempt, 4);
    return 250_u64.saturating_mul(1_u64 << exponent);
}

fn error_code(error_text: &str) -> &'static str {
    if error_text.contains("429") || error_text.to_ascii_lowercase().contains("rate limit") {
        return "http_rate_limited";
    }
    if error_text.to_ascii_lowercase().contains("timeout") {
        return "http_timeout";
    }
    return "http_rpc_failed";
}

fn emit<O>(
    observer: &O,
    level: crate::BackfillProgressLevel,
    message: impl std::convert::Into<std::string::String>,
    completed: std::option::Option<u64>,
    total: std::option::Option<u64>,
) where
    O: crate::BackfillObserver + Sync,
{
    let event = crate::BackfillProgressEvent::new(level, message, completed, total);
    observer.on_progress(&event);
}

#[cfg(test)]
mod tests {
    fn signature(byte: u8) -> std::string::String {
        return bs58::encode([byte; 64]).into_string();
    }

    fn pubkey(byte: u8) -> std::string::String {
        return bs58::encode([byte; 32]).into_string();
    }

    #[test]
    fn explicit_candidates_are_deduplicated_in_input_order() {
        let first = signature(1);
        let second = signature(2);
        let discovery = super::explicit_candidates(&[first.clone(), second.clone(), first.clone()]);
        assert_eq!(discovery.candidates.len(), 2);
        assert_eq!(discovery.candidates[0].signature, first);
        assert_eq!(discovery.candidates[1].signature, second);
    }

    #[test]
    fn address_request_requires_non_zero_limit() {
        let request = crate::BackfillRequest {
            role: "history_backfill".to_string(),
            commitment: "confirmed".to_string(),
            source: crate::BackfillSource::AddressHistory {
                kind: crate::BackfillAddressKind::Program,
                address: pubkey(3),
                anchor_signature: std::option::Option::Some(signature(4)),
                direction: crate::BackfillDirection::Before,
                limit: 0,
            },
            page_size: 100,
            max_pages: 10,
            max_concurrent_requests: 2,
            max_retries: 2,
        };
        assert!(request.validate().is_err());
    }

    #[test]
    fn before_address_request_accepts_missing_anchor() {
        let request = crate::BackfillRequest {
            role: "history_backfill".to_string(),
            commitment: "confirmed".to_string(),
            source: crate::BackfillSource::AddressHistory {
                kind: crate::BackfillAddressKind::Program,
                address: pubkey(40),
                anchor_signature: std::option::Option::None,
                direction: crate::BackfillDirection::Before,
                limit: 10,
            },
            page_size: 100,
            max_pages: 10,
            max_concurrent_requests: 2,
            max_retries: 2,
        };
        assert!(request.validate().is_ok());
        assert_eq!(request.filter_code(), "program_latest");
    }

    #[test]
    fn after_address_request_rejects_missing_anchor() {
        let request = crate::BackfillRequest {
            role: "history_backfill".to_string(),
            commitment: "confirmed".to_string(),
            source: crate::BackfillSource::AddressHistory {
                kind: crate::BackfillAddressKind::Program,
                address: pubkey(41),
                anchor_signature: std::option::Option::None,
                direction: crate::BackfillDirection::After,
                limit: 10,
            },
            page_size: 100,
            max_pages: 10,
            max_concurrent_requests: 2,
            max_retries: 2,
        };
        assert!(request.validate().is_err());
    }

    #[test]
    fn filter_code_distinguishes_target_and_direction() {
        let request = crate::BackfillRequest {
            role: "history_backfill".to_string(),
            commitment: "confirmed".to_string(),
            source: crate::BackfillSource::AddressHistory {
                kind: crate::BackfillAddressKind::Pool,
                address: pubkey(5),
                anchor_signature: std::option::Option::Some(signature(6)),
                direction: crate::BackfillDirection::After,
                limit: 10,
            },
            page_size: 100,
            max_pages: 10,
            max_concurrent_requests: 2,
            max_retries: 2,
        };
        assert_eq!(request.filter_code(), "pool_after");
    }

    #[test]
    fn nearest_newer_candidates_keep_only_entries_closest_to_anchor() {
        let page = (1_u8..=6_u8)
            .map(|byte| {
                return ks_onchain_transport::AddressSignatureInfo {
                    signature: signature(byte),
                    slot: u64::from(byte),
                    err: std::option::Option::None,
                    memo: std::option::Option::None,
                    block_time: std::option::Option::None,
                    confirmation_status: std::option::Option::Some("confirmed".to_string()),
                };
            })
            .collect::<std::vec::Vec<ks_onchain_transport::AddressSignatureInfo>>();
        let mut candidates = std::collections::VecDeque::new();
        let mut seen = std::collections::HashSet::new();
        super::append_nearest_newer_candidates(&mut candidates, &mut seen, &page, 3);
        let signatures = candidates
            .into_iter()
            .map(|candidate| return candidate.signature)
            .collect::<std::vec::Vec<std::string::String>>();
        assert_eq!(signatures, vec![signature(4), signature(5), signature(6)]);
    }

    #[test]
    fn nearest_newer_candidates_deduplicate_page_entries() {
        let duplicate = signature(7);
        let page = vec![
            ks_onchain_transport::AddressSignatureInfo {
                signature: duplicate.clone(),
                slot: 7,
                err: std::option::Option::None,
                memo: std::option::Option::None,
                block_time: std::option::Option::None,
                confirmation_status: std::option::Option::Some("confirmed".to_string()),
            },
            ks_onchain_transport::AddressSignatureInfo {
                signature: duplicate,
                slot: 7,
                err: std::option::Option::None,
                memo: std::option::Option::None,
                block_time: std::option::Option::None,
                confirmation_status: std::option::Option::Some("confirmed".to_string()),
            },
        ];
        let mut candidates = std::collections::VecDeque::new();
        let mut seen = std::collections::HashSet::new();
        super::append_nearest_newer_candidates(&mut candidates, &mut seen, &page, 5);
        assert_eq!(candidates.len(), 1);
    }

    #[test]
    fn lower_hex_encodes_digest_bytes_without_lower_hex_trait() {
        assert_eq!(super::lower_hex(&[0x00, 0xab, 0xcd, 0xef]), "00abcdef");
    }

    #[test]
    fn completion_frontier_advances_only_across_contiguous_results() {
        let mut frontier = super::CandidateCompletionFrontier::new(4);
        frontier.mark_completed(1);
        assert_eq!(frontier.contiguous_completed(), 0);
        frontier.mark_completed(0);
        assert_eq!(frontier.contiguous_completed(), 2);
        frontier.mark_completed(3);
        assert_eq!(frontier.contiguous_completed(), 2);
        frontier.mark_completed(2);
        assert_eq!(frontier.contiguous_completed(), 4);
    }

    #[test]
    fn cancelled_before_resume_uses_last_contiguous_candidate() {
        let anchor = signature(20);
        let candidates = vec![signature(21), signature(22), signature(23)];
        let request = crate::BackfillRequest {
            role: "history_backfill".to_string(),
            commitment: "confirmed".to_string(),
            source: crate::BackfillSource::AddressHistory {
                kind: crate::BackfillAddressKind::Program,
                address: pubkey(24),
                anchor_signature: std::option::Option::Some(anchor),
                direction: crate::BackfillDirection::Before,
                limit: 3,
            },
            page_size: 100,
            max_pages: 10,
            max_concurrent_requests: 2,
            max_retries: 2,
        };
        let resume = super::resolved_resume_before_signature(
            &request,
            candidates.as_slice(),
            2,
            std::option::Option::Some(signature(25)),
            true,
        );
        assert_eq!(resume, std::option::Option::Some(candidates[1].clone()));
    }

    #[test]
    fn cancelled_before_resume_keeps_anchor_when_nothing_completed() {
        let anchor = signature(30);
        let request = crate::BackfillRequest {
            role: "history_backfill".to_string(),
            commitment: "confirmed".to_string(),
            source: crate::BackfillSource::AddressHistory {
                kind: crate::BackfillAddressKind::Token,
                address: pubkey(31),
                anchor_signature: std::option::Option::Some(anchor.clone()),
                direction: crate::BackfillDirection::Before,
                limit: 2,
            },
            page_size: 100,
            max_pages: 10,
            max_concurrent_requests: 2,
            max_retries: 2,
        };
        let resume = super::resolved_resume_before_signature(
            &request,
            &[signature(32), signature(33)],
            0,
            std::option::Option::Some(signature(34)),
            true,
        );
        assert_eq!(resume, std::option::Option::Some(anchor));
    }

    #[test]
    fn cancelled_latest_scan_without_completed_candidate_restarts_from_latest() {
        let request = crate::BackfillRequest {
            role: "history_backfill".to_string(),
            commitment: "confirmed".to_string(),
            source: crate::BackfillSource::AddressHistory {
                kind: crate::BackfillAddressKind::Program,
                address: pubkey(42),
                anchor_signature: std::option::Option::None,
                direction: crate::BackfillDirection::Before,
                limit: 2,
            },
            page_size: 100,
            max_pages: 10,
            max_concurrent_requests: 2,
            max_retries: 2,
        };
        let resume = super::resolved_resume_before_signature(
            &request,
            &[signature(43), signature(44)],
            0,
            std::option::Option::Some(signature(45)),
            true,
        );
        assert_eq!(resume, std::option::Option::None);
    }

    #[test]
    fn retry_delay_uses_configured_pause_for_rate_limit() {
        assert_eq!(super::retry_delay_ms("HTTP 429", 2, 1500), 1500);
        assert_eq!(super::retry_delay_ms("timeout", 2, 1500), 1000);
    }

    struct TestObserver {
        cancelled: std::sync::atomic::AtomicBool,
    }

    impl TestObserver {
        fn new(cancelled: bool) -> Self {
            return Self {
                cancelled: std::sync::atomic::AtomicBool::new(cancelled),
            };
        }

        fn cancel(&self) {
            self.cancelled.store(true, std::sync::atomic::Ordering::SeqCst);
        }
    }

    impl crate::BackfillObserver for TestObserver {
        fn on_progress(&self, _event: &crate::BackfillProgressEvent) {}

        fn is_cancelled(&self) -> bool {
            return self.cancelled.load(std::sync::atomic::Ordering::SeqCst);
        }
    }

    #[tokio::test]
    async fn cancellable_retry_wait_returns_immediately_when_already_cancelled() {
        let observer = TestObserver::new(true);
        let started = std::time::Instant::now();
        let completed =
            super::wait_with_cancellation(&observer, std::time::Duration::from_secs(60)).await;
        assert!(!completed);
        assert!(started.elapsed() < std::time::Duration::from_secs(1));
    }

    #[tokio::test]
    async fn long_running_rpc_future_is_cancelled_cooperatively() {
        let observer = TestObserver::new(false);
        let operation = super::await_or_cancelled(
            &observer,
            tokio::time::sleep(std::time::Duration::from_secs(60)),
        );
        let cancellation = async {
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            observer.cancel();
        };
        let (result, ()) = tokio::join!(operation, cancellation);
        assert!(result.is_none());
    }
}