// file: crates/ksp-worker-raw-transaction-ingest-lib/src/persistence.rs
// version: 4

/// Canonical entity disposition produced by one Worker Store persistence attempt.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum RawTransactionIngestEntityPersistence {
    /// The canonical RAW transaction was inserted for the first time.
    Inserted,
    /// Identical canonical RAW transaction content was already durable.
    AlreadyPresent,
    /// Normal persistence respected an existing purge tombstone.
    SkippedPurged,
}

/// Observation disposition produced by one Worker Store persistence attempt.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum RawTransactionIngestObservationPersistence {
    /// The deterministic acquisition observation was inserted for the first time.
    Inserted,
    /// The same deterministic acquisition observation was already durable.
    AlreadyPresent,
    /// No observation was recorded because the canonical entity was intentionally skipped.
    NotRecorded,
}

/// Classified successful outcome of one atomic Worker Store persistence attempt.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) struct RawTransactionIngestPersistenceOutcome {
    entity: crate::RawTransactionIngestEntityPersistence,
    observation: crate::RawTransactionIngestObservationPersistence,
}

impl crate::RawTransactionIngestPersistenceOutcome {
    /// Returns the canonical entity persistence disposition.
    #[must_use]
    pub(crate) const fn entity(self) -> crate::RawTransactionIngestEntityPersistence {
        return self.entity;
    }

    /// Returns the acquisition-observation persistence disposition.
    #[must_use]
    pub(crate) const fn observation(self) -> crate::RawTransactionIngestObservationPersistence {
        return self.observation;
    }
}

/// Private backend-neutral Store persistence port used by the Worker and deterministic tests.
pub(crate) trait RawTransactionIngestPersistencePort: std::marker::Send + std::marker::Sync {
    /// Returns whether the persistence target belongs to the requested logical network.
    fn network_matches(&self, network: &ksp_store_lib::RawNetworkId) -> bool;

    /// Executes one atomic canonical RAW transaction plus observation Store write.
    fn persist_acquisition<'a>(
        &'a self,
        transaction: ksp_store_lib::RawTransaction,
        observation: ksp_store_lib::RawTransactionObservation,
        mode: ksp_store_lib::RawTransactionAcquisitionMode,
    ) -> ksp_store_lib::StoreApiFuture<'a, ksp_store_lib::Result<ksp_store_lib::RawAcquisitionWriteOutcome>>;

    /// Records one additional observation for an already durable canonical RAW transaction.
    fn record_observation<'a>(
        &'a self,
        observation: ksp_store_lib::RawTransactionObservation,
    ) -> ksp_store_lib::StoreApiFuture<'a, ksp_store_lib::Result<ksp_store_lib::RawObservationWriteOutcome>>;
}

impl crate::RawTransactionIngestPersistencePort for ksp_store_lib::Store {
    fn network_matches(&self, network: &ksp_store_lib::RawNetworkId) -> bool {
        return self.runtime_snapshot().network() == network;
    }

    fn persist_acquisition<'a>(
        &'a self,
        transaction: ksp_store_lib::RawTransaction,
        observation: ksp_store_lib::RawTransactionObservation,
        mode: ksp_store_lib::RawTransactionAcquisitionMode,
    ) -> ksp_store_lib::StoreApiFuture<'a, ksp_store_lib::Result<ksp_store_lib::RawAcquisitionWriteOutcome>> {
        return ksp_store_lib::RawTransactionWrite::persist_raw_transaction_acquisition(self, transaction, observation, mode);
    }

    fn record_observation<'a>(
        &'a self,
        observation: ksp_store_lib::RawTransactionObservation,
    ) -> ksp_store_lib::StoreApiFuture<'a, ksp_store_lib::Result<ksp_store_lib::RawObservationWriteOutcome>> {
        return ksp_store_lib::RawTransactionObservationWrite::record_raw_transaction_observation(self, observation);
    }
}

#[derive(Clone, Eq, Ord, PartialEq, PartialOrd)]
struct RawTransactionIngestPersistenceKey {
    network: ksp_store_lib::RawNetworkId,
    signature: ksp_store_lib::RawTransactionSignature,
}

#[derive(Clone, Eq, PartialEq)]
struct RawTransactionIngestCanonicalState {
    block_time: std::option::Option<ksp_store_lib::RawTimestamp>,
    content_hash: ksp_store_lib::RawContentHash,
    format_id: ksp_store_lib::RawFormatId,
    format_version: u32,
    slot: u64,
}

/// Private bounded run-local cache serializing repeated canonical identities before Store writes.
pub(crate) struct RawTransactionIngestPersistenceConvergence {
    max_entries: usize,
    entries: std::sync::Mutex<
        std::collections::BTreeMap<
            RawTransactionIngestPersistenceKey,
            std::sync::Arc<tokio::sync::Mutex<std::option::Option<RawTransactionIngestCanonicalState>>>,
        >,
    >,
}

impl crate::RawTransactionIngestPersistenceConvergence {
    /// Creates one bounded run-local convergence cache using the caller-provided effective runtime bound.
    #[must_use]
    pub(crate) fn new(max_entries: usize) -> Self {
        return Self { max_entries, entries: std::sync::Mutex::new(std::collections::BTreeMap::new()) };
    }

    fn entry(
        &self,
        key: RawTransactionIngestPersistenceKey,
    ) -> std::option::Option<std::sync::Arc<tokio::sync::Mutex<std::option::Option<RawTransactionIngestCanonicalState>>>> {
        let mut entries = match self.entries.lock() {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(poisoned) => poisoned.into_inner(),
        };
        if let std::option::Option::Some(entry) = entries.get(&key) {
            return std::option::Option::Some(std::sync::Arc::clone(entry));
        }
        if entries.len() >= self.max_entries {
            let removable = entries.iter().find_map(|(candidate, entry)| {
                if std::sync::Arc::strong_count(entry) == 1 {
                    return std::option::Option::Some(candidate.clone());
                }
                return std::option::Option::None;
            });
            if let std::option::Option::Some(removable) = removable {
                entries.remove(&removable);
            }
        }
        if entries.len() >= self.max_entries {
            return std::option::Option::None;
        }
        let entry = std::sync::Arc::new(tokio::sync::Mutex::new(std::option::Option::None));
        entries.insert(key, std::sync::Arc::clone(&entry));
        return std::option::Option::Some(entry);
    }
}

/// Persists one already-canonical Worker acquisition through the private Store port in `Normal` mode.
pub(crate) async fn persist_raw_transaction_ingest_acquisition<P>(
    port: &P,
    acquisition: ksp_raw_transaction_lib::RawTransactionAcquisition,
) -> ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>
where
    P: crate::RawTransactionIngestPersistencePort + ?Sized,
{
    let reference = acquisition.transaction().reference().clone();
    if !port.network_matches(reference.network()) {
        return std::result::Result::Err(crate::runtime_error("persistence.store_network_mismatch"));
    }
    if acquisition.observation().transaction() != &reference {
        return std::result::Result::Err(crate::runtime_error("persistence.acquisition_reference_mismatch"));
    }
    let (transaction, observation) = acquisition.into_parts();
    let result = port.persist_acquisition(transaction, observation, ksp_store_lib::RawTransactionAcquisitionMode::Normal).await;
    return match result {
        std::result::Result::Ok(outcome) => map_store_outcome(outcome),
        std::result::Result::Err(error) => map_store_error(error),
    };
}

/// Persists one canonical acquisition through the bounded cross-source convergence cache.
pub(crate) async fn persist_raw_transaction_ingest_converged_acquisition<P>(
    port: &P,
    acquisition: ksp_raw_transaction_lib::RawTransactionAcquisition,
    convergence: &crate::RawTransactionIngestPersistenceConvergence,
) -> ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>
where
    P: crate::RawTransactionIngestPersistencePort + ?Sized,
{
    let reference = acquisition.transaction().reference().clone();
    if !port.network_matches(reference.network()) {
        return std::result::Result::Err(crate::runtime_error("persistence.store_network_mismatch"));
    }
    if acquisition.observation().transaction() != &reference {
        return std::result::Result::Err(crate::runtime_error("persistence.acquisition_reference_mismatch"));
    }
    let canonical_state = canonical_state(acquisition.transaction());
    let key = RawTransactionIngestPersistenceKey { network: reference.network().clone(), signature: reference.signature() };
    let entry = convergence.entry(key);
    let entry = match entry {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return crate::persist_raw_transaction_ingest_acquisition(port, acquisition).await,
    };
    let mut known_state = entry.lock().await;
    if let std::option::Option::Some(known_state_value) = known_state.as_ref() {
        if known_state_value != &canonical_state {
            return std::result::Result::Err(crate::content_conflict_error());
        }
        let (_transaction, observation) = acquisition.into_parts();
        let result = port.record_observation(observation).await;
        return match result {
            std::result::Result::Ok(outcome) => map_additional_observation_outcome(outcome),
            std::result::Result::Err(error) => map_store_error(error),
        };
    }
    let outcome = crate::persist_raw_transaction_ingest_acquisition(port, acquisition).await;
    let outcome = match outcome {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    if outcome.entity() != crate::RawTransactionIngestEntityPersistence::SkippedPurged {
        *known_state = std::option::Option::Some(canonical_state);
    }
    return std::result::Result::Ok(outcome);
}

fn canonical_state(transaction: &ksp_store_lib::RawTransaction) -> RawTransactionIngestCanonicalState {
    return RawTransactionIngestCanonicalState {
        block_time: transaction.block_time(),
        content_hash: transaction.payload().content_hash(),
        format_id: transaction.payload().format_id().clone(),
        format_version: transaction.payload().format_version(),
        slot: transaction.slot(),
    };
}

fn map_store_error(error: ksp_core_lib::Error) -> ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome> {
    if error.code() == ksp_store_lib::ERROR_CODE_RAW_CONFLICT {
        return std::result::Result::Err(crate::content_conflict_error());
    }
    return std::result::Result::Err(crate::store_error(error.code()));
}

fn map_store_outcome(outcome: ksp_store_lib::RawAcquisitionWriteOutcome) -> ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome> {
    let entity = outcome.entity();
    let observation = outcome.observation();
    if entity == ksp_store_lib::RawEntityWriteOutcome::Inserted && observation == ksp_store_lib::RawObservationWriteOutcome::Inserted {
        return std::result::Result::Ok(crate::RawTransactionIngestPersistenceOutcome {
            entity: crate::RawTransactionIngestEntityPersistence::Inserted,
            observation: crate::RawTransactionIngestObservationPersistence::Inserted,
        });
    }
    if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::Inserted {
        return std::result::Result::Ok(crate::RawTransactionIngestPersistenceOutcome {
            entity: crate::RawTransactionIngestEntityPersistence::AlreadyPresent,
            observation: crate::RawTransactionIngestObservationPersistence::Inserted,
        });
    }
    if entity == ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent && observation == ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent {
        return std::result::Result::Ok(crate::RawTransactionIngestPersistenceOutcome {
            entity: crate::RawTransactionIngestEntityPersistence::AlreadyPresent,
            observation: crate::RawTransactionIngestObservationPersistence::AlreadyPresent,
        });
    }
    if entity == ksp_store_lib::RawEntityWriteOutcome::SkippedPurged && observation == ksp_store_lib::RawObservationWriteOutcome::NotRecorded {
        return std::result::Result::Ok(crate::RawTransactionIngestPersistenceOutcome {
            entity: crate::RawTransactionIngestEntityPersistence::SkippedPurged,
            observation: crate::RawTransactionIngestObservationPersistence::NotRecorded,
        });
    }
    return std::result::Result::Err(crate::runtime_error("persistence.store_outcome_invalid"));
}

fn map_additional_observation_outcome(
    outcome: ksp_store_lib::RawObservationWriteOutcome,
) -> ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome> {
    return match outcome {
        ksp_store_lib::RawObservationWriteOutcome::Inserted => std::result::Result::Ok(crate::RawTransactionIngestPersistenceOutcome {
            entity: crate::RawTransactionIngestEntityPersistence::AlreadyPresent,
            observation: crate::RawTransactionIngestObservationPersistence::Inserted,
        }),
        ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent => std::result::Result::Ok(crate::RawTransactionIngestPersistenceOutcome {
            entity: crate::RawTransactionIngestEntityPersistence::AlreadyPresent,
            observation: crate::RawTransactionIngestObservationPersistence::AlreadyPresent,
        }),
        ksp_store_lib::RawObservationWriteOutcome::NotRecorded => {
            std::result::Result::Err(crate::runtime_error("persistence.additional_observation_not_recorded"))
        },
        _ => std::result::Result::Err(crate::runtime_error("persistence.additional_observation_outcome_invalid")),
    };
}

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