holochain_state 0.8.0-dev.5

//! `DhtStore` methods that back the Kitsune2 `OpStore` (gossip/sync).
//!
//! These are the cross-table op reads K2 issues during gossip plus the
//! slice-hash cache access. The reads are generic over any handle that
//! borrows a read pool, so both `DhtStoreRead` and the writable
//! `DhtStore<DbWrite<Dht>>` expose them; the slice-hash write needs
//! `DbWrite`. All `sqlx::Error`s map into `StateQueryError`.

use holochain_conductor_api::{OpTimingDump, OpTimingsCursor, OpTimingsDump};
use holochain_data::kind::Dht;
use holochain_data::models::dht::OpTimingRow;
use holochain_data::DbWrite;
use holochain_types::prelude::{AgentPubKey, DhtOpHash, OpValidity, Timestamp};

use super::DhtStore;
use crate::mutations::{StateMutationError, StateMutationResult};
use crate::query::{StateQueryError, StateQueryResult};

impl<Db> DhtStore<Db>
where
    Db: AsRef<holochain_data::DbRead<Dht>>,
{
    /// `(op_hash, basis, size)` triples for every integrated, locally-
    /// validated op whose authored timestamp falls in `[start, end)`.
    /// Used by K2 `retrieve_op_hashes_in_time_slice`.
    pub async fn op_hashes_in_time_slice(
        &self,
        arc_start: u32,
        arc_end: u32,
        start: Timestamp,
        end: Timestamp,
    ) -> crate::query::StateQueryResult<Vec<holochain_data::models::dht::K2OpHashRow>> {
        self.db
            .as_ref()
            .op_hashes_in_time_slice(arc_start, arc_end, start.as_micros(), end.as_micros())
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Up to `limit` `(hash, basis, when_integrated, size)` tuples with
    /// `when_integrated >= t_min`. Used by K2 `retrieve_op_ids_bounded`'s
    /// inner paging loop.
    pub async fn op_ids_since_time_batch(
        &self,
        arc_start: u32,
        arc_end: u32,
        t_min: Timestamp,
        limit: u32,
    ) -> crate::query::StateQueryResult<Vec<holochain_data::models::dht::K2OpIdSinceRow>> {
        self.db
            .as_ref()
            .op_ids_since_time_batch(arc_start, arc_end, t_min.as_micros(), limit)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Subset of `op_hashes` we hold in limbo or as locally-validated
    /// integrated ops (cache-only copies excluded), with their basis hashes.
    /// Used by K2 `filter_out_existing_ops`.
    pub async fn check_op_hashes_present(
        &self,
        op_hashes: &[Vec<u8>],
    ) -> crate::query::StateQueryResult<Vec<holochain_data::models::dht::K2OpPresentRow>> {
        self.db
            .as_ref()
            .check_op_hashes_present(op_hashes)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Joined chain-op rows (ChainOp joined with Action, left-joined with
    /// Entry) for wire reconstruction. Filtered to `locally_validated = 1`.
    pub async fn get_chain_ops_for_wire(
        &self,
        op_hashes: &[Vec<u8>],
    ) -> crate::query::StateQueryResult<Vec<holochain_data::models::dht::K2ChainOpForWireRow>> {
        self.db
            .as_ref()
            .get_chain_ops_for_wire(op_hashes)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Warrant rows for wire reconstruction.
    pub async fn get_warrants_for_wire(
        &self,
        op_hashes: &[Vec<u8>],
    ) -> crate::query::StateQueryResult<Vec<holochain_data::models::dht::K2WarrantForWireRow>> {
        self.db
            .as_ref()
            .get_warrants_for_wire(op_hashes)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Earliest authored timestamp in the arc, across both `ChainOp`
    /// (joined with `Action`) and `Warrant`. `None` if no rows match.
    pub async fn earliest_authored_timestamp_in_arc(
        &self,
        arc_start: u32,
        arc_end: u32,
    ) -> crate::query::StateQueryResult<Option<Timestamp>> {
        let micros = self
            .db
            .as_ref()
            .earliest_authored_timestamp_in_arc(arc_start, arc_end)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)?;
        Ok(micros.map(Timestamp::from_micros))
    }

    /// `(validation_limbo, integration_limbo, integrated)` counts for the
    /// integration-state report, each as `usize`. `integrated` counts
    /// locally-validated `ChainOp` rows (GET-cached copies excluded) plus all
    /// `WarrantOp` rows; `integration_limbo` is the limbo subset ready for
    /// integration and `validation_limbo` the remainder still in validation.
    pub async fn limbo_state_counts(
        &self,
    ) -> crate::query::StateQueryResult<(usize, usize, usize)> {
        let (validation_limbo, integration_limbo, integrated) = self
            .db
            .as_ref()
            .limbo_state_counts()
            .await
            .map_err(crate::query::StateQueryError::Sqlx)?;
        Ok((
            validation_limbo.max(0) as usize,
            integration_limbo.max(0) as usize,
            integrated.max(0) as usize,
        ))
    }

    /// Return integrated chain-op rows after `after`, bounded by `limit`.
    ///
    /// `None` for `after` starts at the beginning, while `None` for `limit`
    /// returns the full remaining set.
    ///
    /// # Errors
    ///
    /// Returns an error if `limit` is zero or the database query fails.
    pub async fn integrated_chain_ops_for_dump(
        &self,
        after: Option<(i64, &[u8])>,
        limit: Option<u32>,
    ) -> crate::query::StateQueryResult<Vec<holochain_data::models::dht::DumpChainOpRow>> {
        if limit == Some(0) {
            return Err(crate::query::StateQueryError::InvalidInput(
                "dump limit must be greater than zero".to_string(),
            ));
        }
        self.db
            .as_ref()
            .integrated_chain_ops_for_dump(after, limit)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Return one global DHT-op page ordered by `(when_received, op_hash)`.
    ///
    /// `None` for `after` starts at the beginning, while `None` for `limit`
    /// returns the full remaining set.
    ///
    /// # Errors
    ///
    /// Returns an error if `limit` is zero or the database query fails.
    pub async fn dht_ops_page_for_dump(
        &self,
        after: Option<(i64, &DhtOpHash)>,
        limit: Option<u32>,
    ) -> crate::query::StateQueryResult<holochain_data::models::dht::DumpOpPage> {
        if limit == Some(0) {
            return Err(crate::query::StateQueryError::InvalidInput(
                "dump limit must be greater than zero".to_string(),
            ));
        }
        self.db
            .as_ref()
            .dht_ops_page_for_dump(after, limit)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// One page of DHT-op lifecycle timings ordered by `(when_received, hash)`.
    ///
    /// `after` is exclusive, so passing the previous page's cursor resumes
    /// strictly after it. `None` for `limit` returns the full remaining set.
    ///
    /// # Errors
    ///
    /// Returns an error if `limit` is zero, the database query fails, or a
    /// stored validation status is not a value this version can decode.
    pub async fn op_timings_page_for_dump(
        &self,
        after: Option<&OpTimingsCursor>,
        limit: Option<u32>,
    ) -> StateQueryResult<OpTimingsDump> {
        if limit == Some(0) {
            return Err(StateQueryError::InvalidInput(
                "dump limit must be greater than zero".to_string(),
            ));
        }
        let page = self
            .db
            .as_ref()
            .op_timings_page(
                after.map(|cursor| (cursor.when_received, &cursor.hash)),
                limit,
            )
            .await
            .map_err(StateQueryError::Sqlx)?;

        Ok(OpTimingsDump {
            timings: page
                .rows
                .into_iter()
                .map(op_timing_dump)
                .collect::<StateQueryResult<Vec<_>>>()?,
            cursor: page.cursor.map(|cursor| OpTimingsCursor {
                when_received: cursor.when_received,
                hash: cursor.hash,
            }),
        })
    }

    /// Limbo chain-op rows for the integration dump. `ready` selects the
    /// integration-limbo subset; `!ready` selects the validation-limbo subset.
    pub async fn limbo_chain_ops_for_dump(
        &self,
        ready: bool,
    ) -> crate::query::StateQueryResult<Vec<holochain_data::models::dht::K2ChainOpForWireRow>> {
        self.db
            .as_ref()
            .limbo_chain_ops_for_dump(ready)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Every integrated warrant row for the integration dump.
    pub async fn integrated_warrants_for_dump(
        &self,
    ) -> crate::query::StateQueryResult<Vec<holochain_data::models::dht::K2WarrantForWireRow>> {
        self.db
            .as_ref()
            .integrated_warrants_for_dump()
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Chain-op rows authored and shared by `author`, joined for wire
    /// reconstruction. Excludes private `CreateEntry` ops so private entries
    /// never leak into the published set. Used by the consistency-check
    /// harness to gather a cell's published ops.
    pub async fn ops_to_publish_for_wire(
        &self,
        author: &AgentPubKey,
    ) -> crate::query::StateQueryResult<Vec<holochain_data::models::dht::K2ChainOpForWireRow>> {
        self.db
            .as_ref()
            .ops_to_publish_for_wire(author)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Total integrated op + warrant count (no `locally_validated` filter
    /// — preserves the old DHT+cache combined count).
    pub async fn total_integrated_op_count(&self) -> crate::query::StateQueryResult<u64> {
        let n = self
            .db
            .as_ref()
            .count_integrated_ops()
            .await
            .map_err(crate::query::StateQueryError::Sqlx)?;
        Ok(n)
    }

    /// Number of stored slices for the arc, or 0 if none.
    ///
    /// K2 assigns slice indices consecutively from 0, so the count is the
    /// highest stored index + 1 (matching the kitsune2 reference op-store);
    /// the query layer adds the one. Returned to K2 as `slice_hash_count`.
    pub async fn slice_hash_count(
        &self,
        arc_start: u32,
        arc_end: u32,
    ) -> crate::query::StateQueryResult<u64> {
        self.db
            .as_ref()
            .slice_hash_count(arc_start, arc_end)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Single stored slice hash, if any.
    pub async fn get_slice_hash(
        &self,
        arc_start: u32,
        arc_end: u32,
        slice_index: u64,
    ) -> crate::query::StateQueryResult<Option<Vec<u8>>> {
        self.db
            .as_ref()
            .get_slice_hash(arc_start, arc_end, slice_index)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }

    /// Every `(slice_index, hash)` pair stored for the arc.
    pub async fn get_slice_hashes(
        &self,
        arc_start: u32,
        arc_end: u32,
    ) -> crate::query::StateQueryResult<Vec<holochain_data::models::dht::SliceHashIndexedRow>> {
        self.db
            .as_ref()
            .get_slice_hashes(arc_start, arc_end)
            .await
            .map_err(crate::query::StateQueryError::Sqlx)
    }
}

/// Decode one raw op-timings row into its typed dump form.
///
/// # Errors
///
/// Returns an error if the row's `validation_status` is not a value this
/// version can decode.
fn op_timing_dump(row: OpTimingRow) -> StateQueryResult<OpTimingDump> {
    Ok(OpTimingDump {
        op_hash: DhtOpHash::from_raw_36(row.hash),
        when_received: Timestamp::from_micros(row.when_received),
        when_integrated: row.when_integrated.map(Timestamp::from_micros),
        abandoned_at: row.abandoned_at.map(Timestamp::from_micros),
        validation_status: row
            .validation_status
            .map(OpValidity::try_from)
            .transpose()
            .map_err(|status| {
                StateQueryError::Other(format!("invalid stored validation status {status}"))
            })?,
        locally_validated: row.locally_validated.map(|flag| flag != 0),
    })
}

impl DhtStore<DbWrite<Dht>> {
    /// Insert or replace the slice hash for `(arc, slice_index)`. K2
    /// `store_slice_hash`.
    pub async fn store_slice_hash(
        &self,
        arc_start: u32,
        arc_end: u32,
        slice_index: u64,
        hash: &[u8],
    ) -> StateMutationResult<()> {
        self.db
            .insert_slice_hash(arc_start, arc_end, slice_index, hash)
            .await
            .map_err(StateMutationError::from)?;
        Ok(())
    }
}