// file: crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs
// version: 10

/// Runtime-neutral boxed future resolving to one newer concrete RAW transaction ingest Worker snapshot.
pub type RawTransactionIngestSnapshotFuture<'a> =
    std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = crate::RawTransactionIngestSnapshot> + std::marker::Send + 'a>>;

/// Source-neutral lifecycle state of the productive RAW transaction ingest source.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub enum RawTransactionIngestSourceState {
    /// The productive source is active.
    Active,
    /// Transport is performing one bounded reconnect/replay attempt.
    Reconnecting,
    /// Cooperative source shutdown has started.
    Closing,
    /// The productive source closed cleanly.
    Closed,
    /// The productive source reached a terminal failure.
    Failed,
}

/// Stable source-neutral identifier of one run-local continuity gap.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct RawTransactionIngestGapId(u64);

impl crate::RawTransactionIngestGapId {
    /// Creates one internal gap identifier after the continuity ledger validated monotonicity.
    pub(crate) const fn new(value: u64) -> Self {
        return Self(value);
    }

    /// Returns the run-local numeric identifier.
    #[must_use]
    pub const fn value(self) -> u64 {
        return self.0;
    }
}

/// Source-neutral lifecycle state of one run-local continuity gap.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub enum RawTransactionIngestGapState {
    /// The gap is known and waiting for an admissible recovery proof.
    Pending,
    /// One bounded recovery mechanism is actively processing the gap.
    Repairing,
    /// The complete required interval has been proven covered.
    Repaired,
    /// The gap remains unresolved after the admissible bounded mechanisms are exhausted.
    Unresolved,
}

/// Source-neutral reason why one run-local continuity gap was opened.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub enum RawTransactionIngestGapReason {
    /// A slot was proven produced but its block material remained unavailable.
    HttpProducedBlockUnavailable,
    /// A previously observed transaction reference could not yet be materialized.
    KnownReferenceMissing,
    /// A logical live source became terminal before continuity was proven.
    SourceFailure,
    /// Transport reported bounded notification loss or overflow.
    TransportOverflow,
    /// A WebSocket reconnect opened a bounded continuity incident.
    WebSocketReconnect,
    /// Yellowstone replay retention could not cover the requested boundary.
    YellowstoneRetention,
}

/// Source-neutral mechanism last used while attempting to close one continuity gap.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub enum RawTransactionIngestRepairMethod {
    /// Transport-owned replay supplied material later proven sufficient.
    Replay,
    /// A distinct live source supplied explicit interval coverage.
    RedundantCoverage,
    /// A bounded HTTP slot/block discovery scan supplied coverage evidence.
    HttpScan,
    /// A produced slot required direct block material retrieval.
    BlockFetch,
    /// A known transaction reference required observed transaction hydration.
    TransactionHydration,
}

/// Safe source-neutral snapshot of one recent or still-open run-local continuity gap.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct RawTransactionIngestGapSnapshot {
    gap_id: crate::RawTransactionIngestGapId,
    start_slot: u64,
    end_slot: u64,
    state: crate::RawTransactionIngestGapState,
    reason: crate::RawTransactionIngestGapReason,
    last_method: std::option::Option<crate::RawTransactionIngestRepairMethod>,
}

impl crate::RawTransactionIngestGapSnapshot {
    /// Creates one safe gap projection from already validated continuity-ledger state.
    pub(crate) const fn new(
        gap_id: crate::RawTransactionIngestGapId,
        start_slot: u64,
        end_slot: u64,
        state: crate::RawTransactionIngestGapState,
        reason: crate::RawTransactionIngestGapReason,
        last_method: std::option::Option<crate::RawTransactionIngestRepairMethod>,
    ) -> Self {
        return Self { gap_id, start_slot, end_slot, state, reason, last_method };
    }

    /// Returns the run-local gap identifier.
    #[must_use]
    pub const fn gap_id(&self) -> crate::RawTransactionIngestGapId {
        return self.gap_id;
    }

    /// Returns the inclusive first slot of the gap.
    #[must_use]
    pub const fn start_slot(&self) -> u64 {
        return self.start_slot;
    }

    /// Returns the inclusive last slot of the gap.
    #[must_use]
    pub const fn end_slot(&self) -> u64 {
        return self.end_slot;
    }

    /// Returns the source-neutral gap lifecycle state.
    #[must_use]
    pub const fn state(&self) -> crate::RawTransactionIngestGapState {
        return self.state;
    }

    /// Returns the source-neutral reason that opened the gap.
    #[must_use]
    pub const fn reason(&self) -> crate::RawTransactionIngestGapReason {
        return self.reason;
    }

    /// Returns the last bounded recovery mechanism recorded for this gap, when any.
    #[must_use]
    pub const fn last_method(&self) -> std::option::Option<crate::RawTransactionIngestRepairMethod> {
        return self.last_method;
    }
}

/// Private bounded continuity-observability projection carried into the public latest-value Worker snapshot.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct RawTransactionIngestContinuitySnapshotProjection {
    gaps: std::vec::Vec<crate::RawTransactionIngestGapSnapshot>,
    open_gap_count: usize,
    repairing_gap_count: usize,
    repaired_gap_total: u64,
    unresolved_gap_total: u64,
    replay_repair_total: u64,
    redundant_coverage_repair_total: u64,
    http_scan_repair_total: u64,
    repair_block_fetch_total: u64,
    repair_transaction_hydration_total: u64,
    oldest_open_gap_start_slot: std::option::Option<u64>,
}

impl crate::RawTransactionIngestContinuitySnapshotProjection {
    /// Returns the empty continuity-observability projection before any gap exists.
    pub(crate) fn empty() -> Self {
        return Self {
            gaps: std::vec::Vec::new(),
            open_gap_count: 0,
            repairing_gap_count: 0,
            repaired_gap_total: 0,
            unresolved_gap_total: 0,
            replay_repair_total: 0,
            redundant_coverage_repair_total: 0,
            http_scan_repair_total: 0,
            repair_block_fetch_total: 0,
            repair_transaction_hydration_total: 0,
            oldest_open_gap_start_slot: std::option::Option::None,
        };
    }

    /// Returns a copy carrying bounded gap details and current open-gap aggregates.
    pub(crate) fn with_gap_state(
        mut self,
        gaps: std::vec::Vec<crate::RawTransactionIngestGapSnapshot>,
        open_gap_count: usize,
        repairing_gap_count: usize,
        oldest_open_gap_start_slot: std::option::Option<u64>,
    ) -> Self {
        self.gaps = gaps;
        self.open_gap_count = open_gap_count;
        self.repairing_gap_count = repairing_gap_count;
        self.oldest_open_gap_start_slot = oldest_open_gap_start_slot;
        return self;
    }

    /// Returns a copy carrying cumulative source-neutral gap recovery totals.
    pub(crate) fn with_recovery_totals(
        mut self,
        repaired_gap_total: u64,
        unresolved_gap_total: u64,
        replay_repair_total: u64,
        redundant_coverage_repair_total: u64,
        http_scan_repair_total: u64,
    ) -> Self {
        self.repaired_gap_total = repaired_gap_total;
        self.unresolved_gap_total = unresolved_gap_total;
        self.replay_repair_total = replay_repair_total;
        self.redundant_coverage_repair_total = redundant_coverage_repair_total;
        self.http_scan_repair_total = http_scan_repair_total;
        return self;
    }

    /// Returns a copy carrying cumulative bounded material-recovery totals.
    pub(crate) fn with_material_totals(mut self, repair_block_fetch_total: u64, repair_transaction_hydration_total: u64) -> Self {
        self.repair_block_fetch_total = repair_block_fetch_total;
        self.repair_transaction_hydration_total = repair_transaction_hydration_total;
        return self;
    }

    /// Returns the bounded recent/open gap projections.
    pub(crate) fn gaps(&self) -> &[crate::RawTransactionIngestGapSnapshot] {
        return self.gaps.as_slice();
    }

    /// Returns the current number of open gaps.
    pub(crate) const fn open_gap_count(&self) -> usize {
        return self.open_gap_count;
    }

    /// Returns the current number of actively recovering gaps.
    pub(crate) const fn repairing_gap_count(&self) -> usize {
        return self.repairing_gap_count;
    }

    /// Returns the cumulative repaired-gap total.
    pub(crate) const fn repaired_gap_total(&self) -> u64 {
        return self.repaired_gap_total;
    }

    /// Returns the cumulative unresolved-gap total.
    pub(crate) const fn unresolved_gap_total(&self) -> u64 {
        return self.unresolved_gap_total;
    }

    /// Returns the cumulative replay recovery total.
    pub(crate) const fn replay_repair_total(&self) -> u64 {
        return self.replay_repair_total;
    }

    /// Returns the cumulative distinct-source coverage recovery total.
    pub(crate) const fn redundant_coverage_repair_total(&self) -> u64 {
        return self.redundant_coverage_repair_total;
    }

    /// Returns the cumulative bounded HTTP scan recovery total.
    pub(crate) const fn http_scan_repair_total(&self) -> u64 {
        return self.http_scan_repair_total;
    }

    /// Returns the cumulative direct block material recovery total.
    pub(crate) const fn repair_block_fetch_total(&self) -> u64 {
        return self.repair_block_fetch_total;
    }

    /// Returns the cumulative transaction hydration recovery total.
    pub(crate) const fn repair_transaction_hydration_total(&self) -> u64 {
        return self.repair_transaction_hydration_total;
    }

    /// Returns the oldest first slot among currently open gaps.
    pub(crate) const fn oldest_open_gap_start_slot(&self) -> std::option::Option<u64> {
        return self.oldest_open_gap_start_slot;
    }
}

/// Private latest-value source-processing projection emitted by the productive source task.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct RawTransactionIngestProcessingFrontierProjection {
    continuity_frontier_slot: std::option::Option<u64>,
    continuity_has_open_gaps: bool,
    continuity_policy_observed: bool,
    failed_source_losses_reconciled: bool,
    future_target_coverage: bool,
    continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection,
    hydration_pending: usize,
    processing_frontier_slot: std::option::Option<u64>,
    oldest_pending_slot: std::option::Option<u64>,
    source_state: std::option::Option<crate::RawTransactionIngestSourceState>,
    source_total: usize,
    source_active: usize,
    source_reconnecting: usize,
    source_failed: usize,
    source_reconnect_total: u64,
    source_replay_attempt_total: u64,
    source_continuity_gap_total: u64,
}

impl crate::RawTransactionIngestProcessingFrontierProjection {
    /// Returns the empty run-local processing projection used before the source observes work.
    pub(crate) fn empty() -> Self {
        return Self {
            continuity_frontier_slot: std::option::Option::None,
            continuity_has_open_gaps: false,
            continuity_policy_observed: false,
            failed_source_losses_reconciled: false,
            future_target_coverage: false,
            continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection::empty(),
            hydration_pending: 0,
            processing_frontier_slot: std::option::Option::None,
            oldest_pending_slot: std::option::Option::None,
            source_state: std::option::Option::None,
            source_total: 0,
            source_active: 0,
            source_reconnecting: 0,
            source_failed: 0,
            source_reconnect_total: 0,
            source_replay_attempt_total: 0,
            source_continuity_gap_total: 0,
        };
    }

    /// Creates one run-local processing projection from bounded source-owned state.
    pub(crate) fn new(hydration_pending: usize, processing_frontier_slot: std::option::Option<u64>, oldest_pending_slot: std::option::Option<u64>) -> Self {
        return Self {
            continuity_frontier_slot: std::option::Option::None,
            continuity_has_open_gaps: false,
            continuity_policy_observed: false,
            failed_source_losses_reconciled: false,
            future_target_coverage: false,
            continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection::empty(),
            hydration_pending,
            processing_frontier_slot,
            oldest_pending_slot,
            source_state: std::option::Option::None,
            source_total: 0,
            source_active: 0,
            source_reconnecting: 0,
            source_failed: 0,
            source_reconnect_total: 0,
            source_replay_attempt_total: 0,
            source_continuity_gap_total: 0,
        };
    }

    /// Returns the number of source signals still pending hydration/admission processing.
    pub(crate) const fn hydration_pending(&self) -> usize {
        return self.hydration_pending;
    }

    /// Returns the highest actually observed slot not blocked by older pending source work.
    pub(crate) const fn processing_frontier_slot(&self) -> std::option::Option<u64> {
        return self.processing_frontier_slot;
    }

    /// Returns the oldest actually observed slot that still owns pending source work.
    pub(crate) const fn oldest_pending_slot(&self) -> std::option::Option<u64> {
        return self.oldest_pending_slot;
    }

    /// Returns a copy carrying the latest source reconnect/replay projection.
    pub(crate) fn with_source_continuity(
        mut self,
        source_state: std::option::Option<crate::RawTransactionIngestSourceState>,
        source_reconnect_total: u64,
        source_replay_attempt_total: u64,
        source_continuity_gap_total: u64,
    ) -> Self {
        self.source_state = source_state;
        self.source_reconnect_total = source_reconnect_total;
        self.source_replay_attempt_total = source_replay_attempt_total;
        self.source_continuity_gap_total = source_continuity_gap_total;
        return self;
    }

    /// Returns a copy carrying source-neutral multi-source lifecycle counts.
    pub(crate) fn with_source_counts(mut self, source_total: usize, source_active: usize, source_reconnecting: usize, source_failed: usize) -> Self {
        self.source_total = source_total;
        self.source_active = source_active;
        self.source_reconnecting = source_reconnecting;
        self.source_failed = source_failed;
        return self;
    }

    /// Returns a copy carrying source-neutral run-local continuity evidence used only for Worker health classification.
    pub(crate) fn with_continuity_health(
        mut self,
        continuity_frontier_slot: std::option::Option<u64>,
        continuity_has_open_gaps: bool,
        future_target_coverage: bool,
    ) -> Self {
        self.continuity_frontier_slot = continuity_frontier_slot;
        self.continuity_has_open_gaps = continuity_has_open_gaps;
        self.continuity_policy_observed = true;
        self.future_target_coverage = future_target_coverage;
        return self;
    }

    /// Returns a copy carrying bounded source-neutral continuity observability.
    pub(crate) fn with_continuity_snapshot(mut self, continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection) -> Self {
        self.continuity_snapshot = continuity_snapshot;
        return self;
    }

    /// Returns the bounded source-neutral continuity observability carried by this projection.
    pub(crate) const fn continuity_snapshot(&self) -> &crate::RawTransactionIngestContinuitySnapshotProjection {
        return &self.continuity_snapshot;
    }

    /// Returns whether a run-local continuity policy projection has been observed.
    pub(crate) const fn continuity_policy_observed(&self) -> bool {
        return self.continuity_policy_observed;
    }

    /// Returns the gap-aware continuity frontier carried by this private projection.
    pub(crate) const fn continuity_frontier_slot(&self) -> std::option::Option<u64> {
        return self.continuity_frontier_slot;
    }

    /// Returns whether at least one run-local continuity gap remains unresolved.
    pub(crate) const fn continuity_has_open_gaps(&self) -> bool {
        return self.continuity_has_open_gaps;
    }

    /// Returns whether currently Active sources still cover the complete configured future `TargetCoverage`.
    pub(crate) const fn future_target_coverage(&self) -> bool {
        return self.future_target_coverage;
    }

    /// Returns a copy carrying whether every terminal Failed source has a reconciled source-loss gap.
    pub(crate) fn with_failed_source_losses_reconciled(mut self, failed_source_losses_reconciled: bool) -> Self {
        self.failed_source_losses_reconciled = failed_source_losses_reconciled;
        return self;
    }

    /// Returns whether every terminal Failed source has a reconciled source-loss gap.
    pub(crate) const fn failed_source_losses_reconciled(&self) -> bool {
        return self.failed_source_losses_reconciled;
    }

    /// Returns the configured logical source count carried by this private projection.
    pub(crate) const fn source_total(&self) -> usize {
        return self.source_total;
    }

    /// Returns the number of sources currently projected Active.
    pub(crate) const fn source_active(&self) -> usize {
        return self.source_active;
    }

    /// Returns the number of sources currently projected Reconnecting.
    pub(crate) const fn source_reconnecting(&self) -> usize {
        return self.source_reconnecting;
    }

    /// Returns the number of sources currently projected Failed.
    pub(crate) const fn source_failed(&self) -> usize {
        return self.source_failed;
    }

    /// Returns the latest source-neutral lifecycle state carried by this private projection.
    pub(crate) const fn source_state(&self) -> std::option::Option<crate::RawTransactionIngestSourceState> {
        return self.source_state;
    }

    /// Returns successful reconnects carried by this private projection.
    pub(crate) const fn source_reconnect_total(&self) -> u64 {
        return self.source_reconnect_total;
    }

    /// Returns replay-bearing reconnect attempts carried by this private projection.
    pub(crate) const fn source_replay_attempt_total(&self) -> u64 {
        return self.source_replay_attempt_total;
    }

    /// Returns proven replay-retention gaps carried by this private projection.
    pub(crate) const fn source_continuity_gap_total(&self) -> u64 {
        return self.source_continuity_gap_total;
    }
}

/// Complete safe latest-value snapshot of one continuous RAW transaction ingest Worker.
#[derive(Clone, Eq, PartialEq)]
pub struct RawTransactionIngestSnapshot {
    worker: ksp_worker_api::WorkerSnapshot,
    continuity_frontier_slot: std::option::Option<u64>,
    continuity_has_open_gaps: bool,
    continuity_policy_observed: bool,
    failed_source_losses_reconciled: bool,
    future_target_coverage: bool,
    gaps: std::vec::Vec<crate::RawTransactionIngestGapSnapshot>,
    open_gap_count: usize,
    repairing_gap_count: usize,
    repaired_gap_total: u64,
    unresolved_gap_total: u64,
    replay_repair_total: u64,
    redundant_coverage_repair_total: u64,
    http_scan_repair_total: u64,
    repair_block_fetch_total: u64,
    repair_transaction_hydration_total: u64,
    oldest_open_gap_start_slot: std::option::Option<u64>,
    admission_queue_capacity: usize,
    admission_queue_depth: usize,
    persistence_concurrency: usize,
    in_flight_persistence: usize,
    admitted_total: u64,
    canonicalized_total: u64,
    persisted_total: u64,
    entity_inserted_total: u64,
    entity_already_present_total: u64,
    entity_skipped_purged_total: u64,
    observation_inserted_total: u64,
    observation_already_present_total: u64,
    content_conflict_total: u64,
    store_failure_total: u64,
    source_failure_total: u64,
    backpressure_wait_total: u64,
    hydration_pending: usize,
    processing_frontier_slot: std::option::Option<u64>,
    oldest_pending_slot: std::option::Option<u64>,
    source_state: std::option::Option<crate::RawTransactionIngestSourceState>,
    source_total: usize,
    source_active: usize,
    source_reconnecting: usize,
    source_failed: usize,
    source_reconnect_total: u64,
    source_replay_attempt_total: u64,
    source_continuity_gap_total: u64,
}

impl crate::RawTransactionIngestSnapshot {
    /// Returns the common Worker projection carried by this concrete snapshot.
    #[must_use]
    pub const fn worker_snapshot(&self) -> &ksp_worker_api::WorkerSnapshot {
        return &self.worker;
    }

    /// Reports whether continuity policy has emitted a run-local health projection.
    #[must_use]
    pub const fn continuity_policy_observed(&self) -> bool {
        return self.continuity_policy_observed;
    }

    /// Returns the latest run-local continuity frontier when continuity policy has observed one.
    #[must_use]
    pub const fn continuity_frontier_slot(&self) -> std::option::Option<u64> {
        return self.continuity_frontier_slot;
    }

    /// Reports whether the continuity policy currently retains one or more open gaps.
    #[must_use]
    pub const fn continuity_has_open_gaps(&self) -> bool {
        return self.continuity_has_open_gaps;
    }

    /// Reports whether all failed-source continuity obligations have been reconciled for this run.
    #[must_use]
    pub const fn failed_source_losses_reconciled(&self) -> bool {
        return self.failed_source_losses_reconciled;
    }

    /// Reports whether current source coverage proves the active target can continue into future slots.
    #[must_use]
    pub const fn future_target_coverage(&self) -> bool {
        return self.future_target_coverage;
    }

    /// Returns the configured bounded admission queue capacity.
    #[must_use]
    pub const fn admission_queue_capacity(&self) -> usize {
        return self.admission_queue_capacity;
    }

    /// Returns the latest observed number of queued ingress entries awaiting supervisor admission.
    #[must_use]
    pub const fn admission_queue_depth(&self) -> usize {
        return self.admission_queue_depth;
    }

    /// Returns the configured maximum number of concurrent Store persistence operations.
    #[must_use]
    pub const fn persistence_concurrency(&self) -> usize {
        return self.persistence_concurrency;
    }

    /// Returns the latest observed number of in-flight Store persistence operations.
    #[must_use]
    pub const fn in_flight_persistence(&self) -> usize {
        return self.in_flight_persistence;
    }

    /// Returns the number of ingress entries removed from the bounded admission queue.
    #[must_use]
    pub const fn admitted_total(&self) -> u64 {
        return self.admitted_total;
    }

    /// Returns the number of admitted ingress entries successfully canonicalized through the Common RAW contract.
    #[must_use]
    pub const fn canonicalized_total(&self) -> u64 {
        return self.canonicalized_total;
    }

    /// Returns the number of successful atomic Store persistence outcomes.
    #[must_use]
    pub const fn persisted_total(&self) -> u64 {
        return self.persisted_total;
    }

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

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

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

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

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

    /// Returns the number of durable Store content conflicts observed by this run.
    #[must_use]
    pub const fn content_conflict_total(&self) -> u64 {
        return self.content_conflict_total;
    }

    /// Returns the number of non-conflict Store persistence failures observed by this run.
    #[must_use]
    pub const fn store_failure_total(&self) -> u64 {
        return self.store_failure_total;
    }

    /// Returns the number of source-task failures observed by this run.
    #[must_use]
    pub const fn source_failure_total(&self) -> u64 {
        return self.source_failure_total;
    }

    /// Returns the number of supervisor dequeues that observed the bounded admission queue at full capacity.
    #[must_use]
    pub const fn backpressure_wait_total(&self) -> u64 {
        return self.backpressure_wait_total;
    }

    /// Returns the latest number of source signals pending hydration/admission processing.
    #[must_use]
    pub const fn hydration_pending(&self) -> usize {
        return self.hydration_pending;
    }

    /// Returns the run-local processing frontier over actually observed source work.
    #[must_use]
    pub const fn processing_frontier_slot(&self) -> std::option::Option<u64> {
        return self.processing_frontier_slot;
    }

    /// Returns the oldest actually observed slot that still owns pending source work.
    #[must_use]
    pub const fn oldest_pending_slot(&self) -> std::option::Option<u64> {
        return self.oldest_pending_slot;
    }

    /// Returns the bounded source-neutral gap projections retained in this latest-value snapshot.
    ///
    /// Every open gap is retained while capacity remains bounded; recent repaired entries may fill unused projection capacity.
    #[must_use]
    pub fn gaps(&self) -> &[crate::RawTransactionIngestGapSnapshot] {
        return self.gaps.as_slice();
    }

    /// Returns the number of currently open run-local continuity gaps.
    #[must_use]
    pub const fn open_gap_count(&self) -> usize {
        return self.open_gap_count;
    }

    /// Returns the number of currently active gap recoveries.
    #[must_use]
    pub const fn repairing_gap_count(&self) -> usize {
        return self.repairing_gap_count;
    }

    /// Returns the number of gaps cumulatively proven repaired during this run.
    #[must_use]
    pub const fn repaired_gap_total(&self) -> u64 {
        return self.repaired_gap_total;
    }

    /// Returns the number of gaps cumulatively classified unresolved during this run.
    #[must_use]
    pub const fn unresolved_gap_total(&self) -> u64 {
        return self.unresolved_gap_total;
    }

    /// Returns the number of gaps whose latest successful recovery proof used replay.
    #[must_use]
    pub const fn replay_repair_total(&self) -> u64 {
        return self.replay_repair_total;
    }

    /// Returns the number of gaps whose latest successful recovery proof used distinct-source coverage.
    #[must_use]
    pub const fn redundant_coverage_repair_total(&self) -> u64 {
        return self.redundant_coverage_repair_total;
    }

    /// Returns the number of gaps whose latest successful recovery proof used a bounded HTTP scan.
    #[must_use]
    pub const fn http_scan_repair_total(&self) -> u64 {
        return self.http_scan_repair_total;
    }

    /// Returns the number of repaired gaps whose last recorded mechanism was direct block material retrieval.
    #[must_use]
    pub const fn repair_block_fetch_total(&self) -> u64 {
        return self.repair_block_fetch_total;
    }

    /// Returns the number of repaired gaps whose last recorded mechanism was transaction hydration.
    #[must_use]
    pub const fn repair_transaction_hydration_total(&self) -> u64 {
        return self.repair_transaction_hydration_total;
    }

    /// Returns the oldest first slot among currently open continuity gaps.
    #[must_use]
    pub const fn oldest_open_gap_start_slot(&self) -> std::option::Option<u64> {
        return self.oldest_open_gap_start_slot;
    }

    /// Returns the configured number of logical live sources for this Worker run.
    #[must_use]
    pub const fn source_total(&self) -> usize {
        return self.source_total;
    }

    /// Returns the latest number of sources projected Active.
    #[must_use]
    pub const fn source_active(&self) -> usize {
        return self.source_active;
    }

    /// Returns the latest number of sources projected Reconnecting.
    #[must_use]
    pub const fn source_reconnecting(&self) -> usize {
        return self.source_reconnecting;
    }

    /// Returns the latest number of sources projected Failed.
    #[must_use]
    pub const fn source_failed(&self) -> usize {
        return self.source_failed;
    }

    /// Returns the latest source-neutral lifecycle state when the productive source has started.
    #[must_use]
    pub const fn source_state(&self) -> std::option::Option<crate::RawTransactionIngestSourceState> {
        return self.source_state;
    }

    /// Returns successful automatic Yellowstone reconnects observed by this Worker run.
    #[must_use]
    pub const fn source_reconnect_total(&self) -> u64 {
        return self.source_reconnect_total;
    }

    /// Returns Yellowstone replay-bearing reconnect attempts observed by this Worker run.
    #[must_use]
    pub const fn source_replay_attempt_total(&self) -> u64 {
        return self.source_replay_attempt_total;
    }

    /// Returns replay retention gaps conservatively proven by Transport during this Worker run.
    #[must_use]
    pub const fn source_continuity_gap_total(&self) -> u64 {
        return self.source_continuity_gap_total;
    }
}

impl std::fmt::Debug for crate::RawTransactionIngestSnapshot {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        return formatter
            .debug_struct("RawTransactionIngestSnapshot")
            .field("worker", &self.worker)
            .field("gaps", &self.gaps)
            .field("open_gap_count", &self.open_gap_count)
            .field("repairing_gap_count", &self.repairing_gap_count)
            .field("repaired_gap_total", &self.repaired_gap_total)
            .field("unresolved_gap_total", &self.unresolved_gap_total)
            .field("replay_repair_total", &self.replay_repair_total)
            .field("redundant_coverage_repair_total", &self.redundant_coverage_repair_total)
            .field("http_scan_repair_total", &self.http_scan_repair_total)
            .field("repair_block_fetch_total", &self.repair_block_fetch_total)
            .field("repair_transaction_hydration_total", &self.repair_transaction_hydration_total)
            .field("oldest_open_gap_start_slot", &self.oldest_open_gap_start_slot)
            .field("admission_queue_capacity", &self.admission_queue_capacity)
            .field("admission_queue_depth", &self.admission_queue_depth)
            .field("persistence_concurrency", &self.persistence_concurrency)
            .field("in_flight_persistence", &self.in_flight_persistence)
            .field("admitted_total", &self.admitted_total)
            .field("canonicalized_total", &self.canonicalized_total)
            .field("persisted_total", &self.persisted_total)
            .field("entity_inserted_total", &self.entity_inserted_total)
            .field("entity_already_present_total", &self.entity_already_present_total)
            .field("entity_skipped_purged_total", &self.entity_skipped_purged_total)
            .field("observation_inserted_total", &self.observation_inserted_total)
            .field("observation_already_present_total", &self.observation_already_present_total)
            .field("content_conflict_total", &self.content_conflict_total)
            .field("store_failure_total", &self.store_failure_total)
            .field("source_failure_total", &self.source_failure_total)
            .field("backpressure_wait_total", &self.backpressure_wait_total)
            .field("hydration_pending", &self.hydration_pending)
            .field("processing_frontier_slot", &self.processing_frontier_slot)
            .field("oldest_pending_slot", &self.oldest_pending_slot)
            .field("source_state", &self.source_state)
            .field("source_total", &self.source_total)
            .field("source_active", &self.source_active)
            .field("source_reconnecting", &self.source_reconnecting)
            .field("source_failed", &self.source_failed)
            .field("source_reconnect_total", &self.source_reconnect_total)
            .field("source_replay_attempt_total", &self.source_replay_attempt_total)
            .field("source_continuity_gap_total", &self.source_continuity_gap_total)
            .finish();
    }
}

/// Cloneable latest-value source exposing concrete and common Worker snapshots from one shared watch state.
#[derive(Clone)]
pub struct RawTransactionIngestSnapshotSource {
    receiver: tokio::sync::watch::Receiver<crate::RawTransactionIngestSnapshot>,
}

impl crate::RawTransactionIngestSnapshotSource {
    /// Returns the complete current concrete snapshot without replaying intermediate updates.
    #[must_use]
    pub fn current(&self) -> crate::RawTransactionIngestSnapshot {
        return self.receiver.borrow().clone();
    }

    /// Reports whether the private publisher has closed after the runtime task returned.
    #[must_use]
    pub(crate) fn is_closed(&self) -> bool {
        return self.receiver.has_changed().is_err();
    }

    /// Waits for one concrete snapshot newer than `observed`, coalescing intermediate updates to the latest value.
    #[must_use]
    pub fn wait_for_change(&self, observed: ksp_worker_api::WorkerSnapshotSequence) -> crate::RawTransactionIngestSnapshotFuture<'_> {
        let mut receiver = self.receiver.clone();
        return std::boxed::Box::pin(async move {
            loop {
                let current = receiver.borrow_and_update().clone();
                if current.worker_snapshot().sequence().is_after(observed) {
                    return current;
                }
                let changed = receiver.changed().await;
                if changed.is_err() {
                    return receiver.borrow_and_update().clone();
                }
            }
        });
    }
}

impl ksp_worker_api::WorkerSnapshotSource for crate::RawTransactionIngestSnapshotSource {
    fn current(&self) -> ksp_worker_api::WorkerSnapshot {
        return self.receiver.borrow().worker_snapshot().clone();
    }

    fn wait_for_change(&self, observed: ksp_worker_api::WorkerSnapshotSequence) -> ksp_worker_api::WorkerSnapshotFuture<'_> {
        let mut receiver = self.receiver.clone();
        return std::boxed::Box::pin(async move {
            loop {
                let current = receiver.borrow_and_update().worker_snapshot().clone();
                if current.sequence().is_after(observed) {
                    return current;
                }
                let changed = receiver.changed().await;
                if changed.is_err() {
                    return receiver.borrow_and_update().worker_snapshot().clone();
                }
            }
        });
    }
}

impl std::fmt::Debug for crate::RawTransactionIngestSnapshotSource {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let current = self.receiver.borrow();
        return formatter
            .debug_struct("RawTransactionIngestSnapshotSource")
            .field("sequence", &current.worker_snapshot().sequence())
            .field("state", &current.worker_snapshot().state())
            .finish();
    }
}

/// Private latest-value publisher and checked counter owner shared by the Worker supervisor.
pub(crate) struct RawTransactionIngestSnapshotPublisher {
    sender: tokio::sync::watch::Sender<crate::RawTransactionIngestSnapshot>,
    snapshot: crate::RawTransactionIngestSnapshot,
}

impl crate::RawTransactionIngestSnapshotPublisher {
    /// Creates the initial `Starting` snapshot stream for one validated Worker run.
    pub(crate) fn new(
        settings: &crate::RawTransactionIngestSettings,
        lifecycle: &ksp_worker_api::WorkerLifecycle,
        source_total: usize,
    ) -> (crate::RawTransactionIngestSnapshotPublisher, crate::RawTransactionIngestSnapshotSource) {
        let worker = ksp_worker_api::WorkerSnapshot::new(
            lifecycle.id().clone(),
            lifecycle.kind().clone(),
            ksp_worker_api::WorkerSnapshotSequence::initial(),
            lifecycle.state(),
            ksp_worker_api::WorkerHealth::Unknown,
            ksp_worker_api::WorkerActivity::Unknown,
        );
        let snapshot = crate::RawTransactionIngestSnapshot {
            worker,
            continuity_frontier_slot: std::option::Option::None,
            continuity_has_open_gaps: false,
            continuity_policy_observed: false,
            failed_source_losses_reconciled: false,
            future_target_coverage: false,
            gaps: std::vec::Vec::new(),
            open_gap_count: 0,
            repairing_gap_count: 0,
            repaired_gap_total: 0,
            unresolved_gap_total: 0,
            replay_repair_total: 0,
            redundant_coverage_repair_total: 0,
            http_scan_repair_total: 0,
            repair_block_fetch_total: 0,
            repair_transaction_hydration_total: 0,
            oldest_open_gap_start_slot: std::option::Option::None,
            admission_queue_capacity: settings.admission_queue_capacity(),
            admission_queue_depth: 0,
            persistence_concurrency: settings.persistence_concurrency(),
            in_flight_persistence: 0,
            admitted_total: 0,
            canonicalized_total: 0,
            persisted_total: 0,
            entity_inserted_total: 0,
            entity_already_present_total: 0,
            entity_skipped_purged_total: 0,
            observation_inserted_total: 0,
            observation_already_present_total: 0,
            content_conflict_total: 0,
            store_failure_total: 0,
            source_failure_total: 0,
            backpressure_wait_total: 0,
            hydration_pending: 0,
            processing_frontier_slot: std::option::Option::None,
            oldest_pending_slot: std::option::Option::None,
            source_state: std::option::Option::None,
            source_total,
            source_active: 0,
            source_reconnecting: 0,
            source_failed: 0,
            source_reconnect_total: 0,
            source_replay_attempt_total: 0,
            source_continuity_gap_total: 0,
        };
        let (sender, receiver) = tokio::sync::watch::channel(snapshot.clone());
        return (Self { sender, snapshot }, crate::RawTransactionIngestSnapshotSource { receiver });
    }

    /// Forces one terminal latest value without advancing the sequence, used only when sequence publication itself is exhausted or invalid.
    pub(crate) fn force_terminal(&mut self, state: ksp_worker_api::WorkerState) {
        let worker = ksp_worker_api::WorkerSnapshot::new(
            self.snapshot.worker.id().clone(),
            self.snapshot.worker.kind().clone(),
            self.snapshot.worker.sequence(),
            state,
            health_for_state(state, self.snapshot.worker.health(), &self.snapshot),
            ksp_worker_api::WorkerActivity::Idle,
        );
        self.snapshot.worker = worker;
        self.snapshot.admission_queue_depth = 0;
        self.snapshot.in_flight_persistence = 0;
        self.sender.send_replace(self.snapshot.clone());
        return;
    }

    /// Publishes one lifecycle/depth refresh with a strictly newer common sequence.
    pub(crate) fn publish_state(
        &mut self,
        state: ksp_worker_api::WorkerState,
        admission_queue_depth: usize,
        in_flight_persistence: usize,
    ) -> ksp_core_lib::Result<()> {
        return self.publish(state, admission_queue_depth, in_flight_persistence);
    }

    /// Records one dequeued ingress that failed canonicalization and publishes the resulting latest value.
    pub(crate) fn record_admission_failure(
        &mut self,
        state: ksp_worker_api::WorkerState,
        admission_queue_depth: usize,
        in_flight_persistence: usize,
        backpressure_wait_observed: bool,
    ) -> ksp_core_lib::Result<()> {
        let admitted_total = match checked_counter(self.snapshot.admitted_total, "admitted_total") {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        let backpressure_wait_total =
            match checked_optional_counter(self.snapshot.backpressure_wait_total, backpressure_wait_observed, "backpressure_wait_total") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        self.snapshot.admitted_total = admitted_total;
        self.snapshot.backpressure_wait_total = backpressure_wait_total;
        return self.publish(state, admission_queue_depth, in_flight_persistence);
    }

    /// Records one successfully canonicalized admitted ingress and publishes the resulting latest value.
    pub(crate) fn record_admission_success(
        &mut self,
        state: ksp_worker_api::WorkerState,
        admission_queue_depth: usize,
        in_flight_persistence: usize,
        backpressure_wait_observed: bool,
    ) -> ksp_core_lib::Result<()> {
        let admitted_total = match checked_counter(self.snapshot.admitted_total, "admitted_total") {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        let backpressure_wait_total =
            match checked_optional_counter(self.snapshot.backpressure_wait_total, backpressure_wait_observed, "backpressure_wait_total") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
        let canonicalized_total = match checked_counter(self.snapshot.canonicalized_total, "canonicalized_total") {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        self.snapshot.admitted_total = admitted_total;
        self.snapshot.backpressure_wait_total = backpressure_wait_total;
        self.snapshot.canonicalized_total = canonicalized_total;
        return self.publish(state, admission_queue_depth, in_flight_persistence);
    }

    /// Publishes one latest run-local processing-frontier projection emitted by the source task.
    pub(crate) fn record_processing_frontier(
        &mut self,
        state: ksp_worker_api::WorkerState,
        admission_queue_depth: usize,
        in_flight_persistence: usize,
        projection: crate::RawTransactionIngestProcessingFrontierProjection,
    ) -> ksp_core_lib::Result<()> {
        self.snapshot.continuity_frontier_slot = projection.continuity_frontier_slot();
        self.snapshot.continuity_has_open_gaps = projection.continuity_has_open_gaps();
        self.snapshot.continuity_policy_observed = projection.continuity_policy_observed();
        self.snapshot.failed_source_losses_reconciled = projection.failed_source_losses_reconciled();
        self.snapshot.future_target_coverage = projection.future_target_coverage();
        let continuity_snapshot = projection.continuity_snapshot();
        self.snapshot.gaps = continuity_snapshot.gaps().to_vec();
        self.snapshot.open_gap_count = continuity_snapshot.open_gap_count();
        self.snapshot.repairing_gap_count = continuity_snapshot.repairing_gap_count();
        self.snapshot.repaired_gap_total = continuity_snapshot.repaired_gap_total();
        self.snapshot.unresolved_gap_total = continuity_snapshot.unresolved_gap_total();
        self.snapshot.replay_repair_total = continuity_snapshot.replay_repair_total();
        self.snapshot.redundant_coverage_repair_total = continuity_snapshot.redundant_coverage_repair_total();
        self.snapshot.http_scan_repair_total = continuity_snapshot.http_scan_repair_total();
        self.snapshot.repair_block_fetch_total = continuity_snapshot.repair_block_fetch_total();
        self.snapshot.repair_transaction_hydration_total = continuity_snapshot.repair_transaction_hydration_total();
        self.snapshot.oldest_open_gap_start_slot = continuity_snapshot.oldest_open_gap_start_slot();
        self.snapshot.hydration_pending = projection.hydration_pending();
        self.snapshot.processing_frontier_slot = projection.processing_frontier_slot();
        self.snapshot.oldest_pending_slot = projection.oldest_pending_slot();
        self.snapshot.source_state = projection.source_state();
        if projection.source_total() > 0 {
            self.snapshot.source_total = projection.source_total();
            self.snapshot.source_active = projection.source_active();
            self.snapshot.source_reconnecting = projection.source_reconnecting();
            self.snapshot.source_failed = projection.source_failed();
        }
        self.snapshot.source_reconnect_total = projection.source_reconnect_total();
        self.snapshot.source_replay_attempt_total = projection.source_replay_attempt_total();
        self.snapshot.source_continuity_gap_total = projection.source_continuity_gap_total();
        return self.publish(state, admission_queue_depth, in_flight_persistence);
    }

    /// Records one failed private source task without retaining provider-specific error material.
    pub(crate) fn record_source_failure(
        &mut self,
        state: ksp_worker_api::WorkerState,
        admission_queue_depth: usize,
        in_flight_persistence: usize,
    ) -> ksp_core_lib::Result<()> {
        let source_failure_total = match checked_counter(self.snapshot.source_failure_total, "source_failure_total") {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        self.snapshot.source_failure_total = source_failure_total;
        return self.publish(state, admission_queue_depth, in_flight_persistence);
    }

    /// Records one successful Store persistence outcome and publishes the resulting latest value.
    pub(crate) fn record_persistence_success(
        &mut self,
        state: ksp_worker_api::WorkerState,
        admission_queue_depth: usize,
        in_flight_persistence: usize,
        outcome: crate::RawTransactionIngestPersistenceOutcome,
    ) -> ksp_core_lib::Result<()> {
        let persisted_total = match checked_counter(self.snapshot.persisted_total, "persisted_total") {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        let mut entity_inserted_total = self.snapshot.entity_inserted_total;
        let mut entity_already_present_total = self.snapshot.entity_already_present_total;
        let mut entity_skipped_purged_total = self.snapshot.entity_skipped_purged_total;
        let mut observation_inserted_total = self.snapshot.observation_inserted_total;
        let mut observation_already_present_total = self.snapshot.observation_already_present_total;
        match outcome.entity() {
            crate::RawTransactionIngestEntityPersistence::Inserted => {
                entity_inserted_total = match checked_counter(entity_inserted_total, "entity_inserted_total") {
                    std::result::Result::Ok(value) => value,
                    std::result::Result::Err(error) => return std::result::Result::Err(error),
                };
            },
            crate::RawTransactionIngestEntityPersistence::AlreadyPresent => {
                entity_already_present_total = match checked_counter(entity_already_present_total, "entity_already_present_total") {
                    std::result::Result::Ok(value) => value,
                    std::result::Result::Err(error) => return std::result::Result::Err(error),
                };
            },
            crate::RawTransactionIngestEntityPersistence::SkippedPurged => {
                entity_skipped_purged_total = match checked_counter(entity_skipped_purged_total, "entity_skipped_purged_total") {
                    std::result::Result::Ok(value) => value,
                    std::result::Result::Err(error) => return std::result::Result::Err(error),
                };
            },
        }
        match outcome.observation() {
            crate::RawTransactionIngestObservationPersistence::Inserted => {
                observation_inserted_total = match checked_counter(observation_inserted_total, "observation_inserted_total") {
                    std::result::Result::Ok(value) => value,
                    std::result::Result::Err(error) => return std::result::Result::Err(error),
                };
            },
            crate::RawTransactionIngestObservationPersistence::AlreadyPresent => {
                observation_already_present_total = match checked_counter(observation_already_present_total, "observation_already_present_total") {
                    std::result::Result::Ok(value) => value,
                    std::result::Result::Err(error) => return std::result::Result::Err(error),
                };
            },
            crate::RawTransactionIngestObservationPersistence::NotRecorded => {},
        }
        self.snapshot.persisted_total = persisted_total;
        self.snapshot.entity_inserted_total = entity_inserted_total;
        self.snapshot.entity_already_present_total = entity_already_present_total;
        self.snapshot.entity_skipped_purged_total = entity_skipped_purged_total;
        self.snapshot.observation_inserted_total = observation_inserted_total;
        self.snapshot.observation_already_present_total = observation_already_present_total;
        return self.publish(state, admission_queue_depth, in_flight_persistence);
    }

    /// Records one classified persistence fault counter and publishes the resulting latest value.
    pub(crate) fn record_persistence_fault(
        &mut self,
        state: ksp_worker_api::WorkerState,
        admission_queue_depth: usize,
        in_flight_persistence: usize,
        code: ksp_core_lib::ErrorCode,
    ) -> ksp_core_lib::Result<()> {
        if code == crate::ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT {
            let next = match checked_counter(self.snapshot.content_conflict_total, "content_conflict_total") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
            self.snapshot.content_conflict_total = next;
        } else if code == crate::ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED {
            let next = match checked_counter(self.snapshot.store_failure_total, "store_failure_total") {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => return std::result::Result::Err(error),
            };
            self.snapshot.store_failure_total = next;
        }
        return self.publish(state, admission_queue_depth, in_flight_persistence);
    }

    fn publish(&mut self, state: ksp_worker_api::WorkerState, admission_queue_depth: usize, in_flight_persistence: usize) -> ksp_core_lib::Result<()> {
        let sequence = match self.snapshot.worker.sequence().next() {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => return std::result::Result::Err(crate::counter_exhausted_error("snapshot_sequence")),
        };
        let worker = ksp_worker_api::WorkerSnapshot::new(
            self.snapshot.worker.id().clone(),
            self.snapshot.worker.kind().clone(),
            sequence,
            state,
            health_for_state(state, self.snapshot.worker.health(), &self.snapshot),
            activity_for_state(state, admission_queue_depth, in_flight_persistence, self.snapshot.hydration_pending, self.snapshot.source_state),
        );
        self.snapshot.worker = worker;
        self.snapshot.admission_queue_depth = admission_queue_depth;
        self.snapshot.in_flight_persistence = in_flight_persistence;
        self.sender.send_replace(self.snapshot.clone());
        return std::result::Result::Ok(());
    }
}

fn activity_for_state(
    state: ksp_worker_api::WorkerState,
    admission_queue_depth: usize,
    in_flight_persistence: usize,
    hydration_pending: usize,
    source_state: std::option::Option<crate::RawTransactionIngestSourceState>,
) -> ksp_worker_api::WorkerActivity {
    if admission_queue_depth > 0
        || in_flight_persistence > 0
        || hydration_pending > 0
        || matches!(
            source_state,
            std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting)
                | std::option::Option::Some(crate::RawTransactionIngestSourceState::Closing)
        )
    {
        return ksp_worker_api::WorkerActivity::Active;
    }
    return match state {
        ksp_worker_api::WorkerState::Running
        | ksp_worker_api::WorkerState::Stopping
        | ksp_worker_api::WorkerState::Stopped
        | ksp_worker_api::WorkerState::Faulted(_) => ksp_worker_api::WorkerActivity::Idle,
        _ => ksp_worker_api::WorkerActivity::Unknown,
    };
}

fn checked_counter(current: u64, field: &'static str) -> ksp_core_lib::Result<u64> {
    return match current.checked_add(1) {
        std::option::Option::Some(value) => std::result::Result::Ok(value),
        std::option::Option::None => std::result::Result::Err(crate::counter_exhausted_error(field)),
    };
}

fn checked_optional_counter(current: u64, increment: bool, field: &'static str) -> ksp_core_lib::Result<u64> {
    if !increment {
        return std::result::Result::Ok(current);
    }
    return checked_counter(current, field);
}

fn health_for_state(
    state: ksp_worker_api::WorkerState,
    previous: ksp_worker_api::WorkerHealth,
    snapshot: &crate::RawTransactionIngestSnapshot,
) -> ksp_worker_api::WorkerHealth {
    if state == ksp_worker_api::WorkerState::Running && snapshot.continuity_policy_observed {
        let source_total = snapshot.source_total;
        let source_active = snapshot.source_active;
        let source_failed = snapshot.source_failed;
        if snapshot.source_reconnecting > 0
            || snapshot.continuity_has_open_gaps
            || snapshot.continuity_frontier_slot != snapshot.processing_frontier_slot
            || !snapshot.future_target_coverage
        {
            return ksp_worker_api::WorkerHealth::Unhealthy;
        }
        if source_total > 0 && snapshot.source_active == snapshot.source_total {
            return ksp_worker_api::WorkerHealth::Healthy;
        }
        if source_total > 0
            && source_active < source_total
            && source_failed > 0
            && snapshot.failed_source_losses_reconciled
            && source_failed == source_total - source_active
        {
            return ksp_worker_api::WorkerHealth::Degraded;
        }
        return ksp_worker_api::WorkerHealth::Unhealthy;
    }
    return match state {
        ksp_worker_api::WorkerState::Running if snapshot.source_failed > 0 => ksp_worker_api::WorkerHealth::Unhealthy,
        ksp_worker_api::WorkerState::Running if snapshot.source_reconnecting > 0 => ksp_worker_api::WorkerHealth::Degraded,
        ksp_worker_api::WorkerState::Running if snapshot.source_total > 0 && snapshot.source_active < snapshot.source_total => {
            ksp_worker_api::WorkerHealth::Degraded
        },
        ksp_worker_api::WorkerState::Running => ksp_worker_api::WorkerHealth::Healthy,
        ksp_worker_api::WorkerState::Stopping | ksp_worker_api::WorkerState::Stopped => previous,
        ksp_worker_api::WorkerState::Faulted(_) => ksp_worker_api::WorkerHealth::Unhealthy,
        _ => ksp_worker_api::WorkerHealth::Unknown,
    };
}

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