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

/// 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>>;

/// Private latest-value processing-frontier projection emitted by the productive source task.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) struct RawTransactionIngestProcessingFrontierProjection {
    hydration_pending: usize,
    processing_frontier_slot: std::option::Option<u64>,
    oldest_pending_slot: std::option::Option<u64>,
}

impl crate::RawTransactionIngestProcessingFrontierProjection {
    /// Returns the empty run-local processing projection used before the source observes work.
    pub(crate) const fn empty() -> Self {
        return Self {
            hydration_pending: 0,
            processing_frontier_slot: std::option::Option::None,
            oldest_pending_slot: std::option::Option::None,
        };
    }

    /// Creates one run-local processing projection from bounded source-owned state.
    pub(crate) const fn new(
        hydration_pending: usize,
        processing_frontier_slot: std::option::Option<u64>,
        oldest_pending_slot: std::option::Option<u64>,
    ) -> Self {
        return Self { hydration_pending, processing_frontier_slot, oldest_pending_slot };
    }

    /// 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;
    }
}

/// Complete safe latest-value snapshot of one continuous RAW transaction ingest Worker.
#[derive(Clone, Eq, PartialEq)]
pub struct RawTransactionIngestSnapshot {
    worker: ksp_worker_api::WorkerSnapshot,
    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>,
}

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;
    }

    /// 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;
    }
}

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("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)
            .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,
    ) -> (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,
            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,
        };
        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()),
            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.hydration_pending = projection.hydration_pending();
        self.snapshot.processing_frontier_slot = projection.processing_frontier_slot();
        self.snapshot.oldest_pending_slot = projection.oldest_pending_slot();
        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()),
            activity_for_state(state, admission_queue_depth, in_flight_persistence, self.snapshot.hydration_pending),
        );
        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,
) -> ksp_worker_api::WorkerActivity {
    if admission_queue_depth > 0 || in_flight_persistence > 0 || hydration_pending > 0 {
        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) -> ksp_worker_api::WorkerHealth {
    return match state {
        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;