holochain_state 0.7.0-dev.28

//! `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_data::kind::Dht;
use holochain_data::DbWrite;
use holochain_types::prelude::Timestamp;

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

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

    /// 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.max(0) as u64)
    }

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

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(())
    }
}