use std::future::Future;

use crate::{
    error::TitoError,
    query::IndexQueryBuilder,
    types::{
        FieldValue, MigrateStats, ReverseIndex, TitoCursor, TitoEngine,
        TitoFindPayload, TitoKvPair,
        TitoModelOptions, TitoPaginated,
        TitoRelationshipConfig, TitoScanPayload, TitoTransaction,
    },
    utils::{next_string_lexicographically, previous_string_lexicographically},
};

enum MigrateResult {
    Done,
    Processed { next_cursor: String, modified: bool },
}
use base64::{decode, encode};
use chrono::Utc;
use serde::{de::DeserializeOwned, Serialize};
use serde_json::Value;

#[derive(Clone)]
pub struct TitoModel<E: TitoEngine, T> {
    pub model: T,
    pub engine: E,
    pub partition_count: u32,
}

impl<E: TitoEngine, T: crate::types::TitoModelConstraints> TitoModel<E, T> {
    pub fn new(engine: E, options: TitoModelOptions) -> Self {
        Self {
            model: T::default(),
            engine,
            partition_count: options.partition_count,
        }
    }

    pub fn relationships(&self) -> Vec<TitoRelationshipConfig> {
        self.model.relationships()
    }

    pub fn get_table(&self) -> String {
        format!("table:{}", self.model.table())
    }

    pub fn get_id_from_table(&self, key: String) -> String {
        let parts: Vec<&str> = key.split(':').collect();
        parts
            .last()
            .map(|last| last.to_string())
            .unwrap_or_else(|| key)
    }

    pub fn query_by_index(&self, index: impl Into<String>) -> IndexQueryBuilder<E, T> {
        IndexQueryBuilder::new(self.clone(), index.into())
    }

    fn decode_cursor(&self, cursor: String) -> Result<TitoCursor, TitoError> {
        let cursor = decode(cursor).map_err(|_err| {
            TitoError::DeserializationFailed("Failed to decode cursor".to_string())
        })?;
        if let Ok(value) = serde_json::from_slice::<TitoCursor>(&cursor) {
            return Ok(value);
        }
        return Err(TitoError::DeserializationFailed(
            "Failed to deserialize cursor".to_string(),
        ));
    }

    fn encode_cursors(&self, ids: Vec<Option<String>>) -> Result<String, TitoError> {
        let tikv_cursor = TitoCursor { ids };
        let json_bytes = serde_json::to_vec(&tikv_cursor).map_err(|_| {
            TitoError::SerializationFailed("Failed to serialize cursor".to_string())
        })?;
        Ok(encode(&json_bytes))
    }

    pub async fn tx<F, Fut, R, Err>(&self, f: F) -> Result<R, Err>
    where
        F: FnOnce(E::Transaction) -> Fut + Clone + Send,
        Fut: Future<Output = Result<R, Err>> + Send,
        Err: From<TitoError> + Send + Sync + std::fmt::Debug,
        R: Send,
    {
        self.engine.transaction(f).await
    }

    fn to_results(
        &self,
        items: impl IntoIterator<Item = TitoKvPair>,
    ) -> Result<Vec<(String, Value)>, TitoError> {
        let mut results = vec![];
        for (key_bytes, value_bytes) in items {
            let key = match String::from_utf8(key_bytes) {
                Ok(k) => k,
                Err(_) => {
                    continue;
                }
            };

            match serde_json::from_slice::<Value>(&value_bytes) {
                Ok(value) => results.push((key, value)),
                Err(_err) => continue,
            }
        }

        Ok(results)
    }
    async fn get(
        &self,
        key: &str,
        tx: &E::Transaction,
    ) -> Result<(String, Value), TitoError> {
        let key = key.to_string();

        match tx.get(key.clone()).await {
            Ok(Some(value)) => match serde_json::from_slice::<Value>(&value) {
                Ok(value) => Ok((key, value)),
                Err(e) => Err(TitoError::NotFound(format!(
                    "Failed to deserialize value for key '{}': {}",
                    key, e
                ))),
            },
            Ok(None) => Err(TitoError::NotFound(format!(
                "Key '{}' not found in database",
                key
            ))),
            Err(e) => Err(TitoError::NotFound(format!(
                "Failed to get key '{}': {}",
                key, e
            ))),
        }
    }
    pub async fn get_key(&self, key: &str, tx: &E::Transaction) -> Result<Value, TitoError> {
        let result = tx
            .get(key.to_string())
            .await
            .map_err(|e| TitoError::NotFound(e.to_string()))?;

        let result = result.ok_or(TitoError::NotFound("Not found".to_string()))?;

        serde_json::from_slice::<Value>(&result)
            .map_err(|_| TitoError::NotFound("Not found".to_string()))
    }

    async fn put<P>(&self, key: String, payload: P, tx: &E::Transaction) -> Result<bool, TitoError>
    where
        P: Serialize + Unpin + std::marker::Send + Sync,
    {
        let mut value = serde_json::to_value(&payload)
            .map_err(|e| TitoError::SerializationFailed(e.to_string()))?;

        if let serde_json::Value::Object(ref mut map) = value {
            let now = Utc::now().timestamp();

            let existing = tx.get(&key).await;
            if existing.is_err() || existing.unwrap().is_none() {
                map.insert("created_at".to_string(), serde_json::json!(now));
            }
            map.insert("updated_at".to_string(), serde_json::json!(now));
        }

        let bytes = serde_json::to_vec(&value)
            .map_err(|e| TitoError::SerializationFailed(e.to_string()))?;

        tx.put(key, bytes)
            .await
            .map_err(|e| TitoError::CreateFailed(e.to_string()))?;

        Ok(true)
    }

    pub async fn delete(&self, key: String, tx: &E::Transaction) -> Result<bool, TitoError> {
        tx.delete(key)
            .await
            .map_err(|e| TitoError::DeleteFailed(e.to_string()))?;

        Ok(true)
    }

    pub fn to_paginated_items_with_cursor(
        &self,
        items: Vec<(String, Value)>,
        cursor: String,
    ) -> Result<TitoPaginated<T>, TitoError> {
        let mut results = vec![];

        for item in items.into_iter() {
            if let Ok(item) = serde_json::from_value::<T>(item.1) {
                results.push(item);
            }
        }

        let results = TitoPaginated::new(results, Some(cursor));

        Ok(results)
    }

    pub fn to_paginated_items(
        &self,
        items: Vec<(String, Value)>,
        has_more: bool,
    ) -> Result<TitoPaginated<T>, TitoError> {
        let mut results = vec![];
        let mut last_item: Option<String> = None;

        for item in items.into_iter() {
            last_item = Some(item.0.clone());
            if let Ok(item) = serde_json::from_value::<T>(item.1) {
                results.push(item);
            }
        }

        let cursor = if has_more && last_item.is_some() {
            Some(
                self.encode_cursors(vec![Some(last_item.unwrap())])
                    .expect("Failed to encode cursor"),
            )
        } else {
            None
        };

        let results = TitoPaginated::new(results, cursor);
        Ok(results)
    }

    async fn get_reverse_index(
        &self,
        key: &str,
        tx: &E::Transaction,
    ) -> Result<ReverseIndex, TitoError> {
        let result = tx.get(key.to_string()).await.map_err(|e| {
            TitoError::NotFound(format!(
                "Failed to get reverse index for key '{}': {}",
                key, e
            ))
        })?;

        let result = result.ok_or(TitoError::NotFound(format!(
            "Reverse index not found for key '{}'",
            key
        )))?;

        serde_json::from_slice::<ReverseIndex>(&result).map_err(|e| {
            TitoError::NotFound(format!(
                "Failed to deserialize reverse index for key '{}': {}",
                key, e
            ))
        })
    }
    pub fn get_nested_values(&self, json: &Value, field_path: &str) -> Option<Vec<FieldValue>> {
        let mut results = Vec::new();
        let mut to_process = vec![(json.clone(), 0)];
        let parts: Vec<&str> = field_path.split('.').collect();

        while let Some((current_value, depth)) = to_process.pop() {
            if depth == parts.len() {
                if let Some(obj) = current_value.as_object() {
                    for (key, value) in obj.iter() {
                        results.push(FieldValue::HashMapEntry {
                            key: key.clone(),
                            value: value.clone(),
                        });
                    }
                } else {
                    results.push(FieldValue::Simple(current_value));
                }
                continue;
            }

            match current_value.get(parts[depth]) {
                Some(nested) => {
                    if nested.is_array() {
                        if let Some(array) = nested.as_array() {
                            if array.is_empty() {
                                return None;
                            }
                            for item in array {
                                to_process.push((item.clone(), depth + 1));
                            }
                        }
                    } else {
                        to_process.push((nested.clone(), depth + 1));
                    }
                }
                None => return None,
            }
        }

        if results.is_empty() {
            None
        } else {
            Some(results)
        }
    }

    pub async fn build(&self, payload: T, tx: &E::Transaction) -> Result<T, TitoError>
    where
        T: serde::de::DeserializeOwned,
    {
        let value = serde_json::to_value(&payload).unwrap();

        let raw_id = payload.id();

        let id = format!("{}:{}", self.get_table(), raw_id);

        self.put(id.clone(), &value, tx).await?;

        let all_index_data = self.get_index_keys(id.clone(), &payload.clone(), &value)?;

        let mut all_index_keys = vec![];

        for data in all_index_data.clone() {
            all_index_keys.push(data.0.clone());
            self.put(data.0.clone(), &data.1, tx).await?;
        }

        let index_json_key = ReverseIndex {
            value: all_index_keys.clone(),
        };

        let reverse_key = format!("reverse-index:{}", id);

        self.put(reverse_key.clone(), index_json_key, tx).await?;

        let source_typed = format!("{}:{}", self.model.table(), payload.id());
        let targets = payload.references();
        self.sync_references(&source_typed, targets, tx).await?;

        Ok(payload)
    }

    async fn find_by_id_with_tx(
        &self,
        id: &str,
        rels: Vec<String>,
        tx: &E::Transaction,
    ) -> Result<T, TitoError>
    where
        T: serde::de::DeserializeOwned,
    {
        let id = format!("{}:{}", self.get_table(), id);

        let value = match self.get(&id, tx).await {
            Ok(value) => value,
            Err(e) => {
                return Err(TitoError::NotFound(format!(
                    "Failed to get record with id '{}': {}",
                    id, e
                )));
            }
        };

        let items = match self
            .fetch_and_stitch_relationships(vec![value], rels.clone(), tx)
            .await
        {
            Ok(value) => value,
            Err(e) => {
                return Err(TitoError::NotFound(format!(
                    "Failed to fetch relationships for id '{}' with rels {:?}: {}",
                    id, rels, e
                )));
            }
        };

        if let Some(value) = items.get(0) {
            serde_json::from_value(value.1.clone()).map_err(|err| {
                TitoError::NotFound(format!(
                    "Failed to deserialize record with id '{}': {}",
                    id, err
                ))
            })
        } else {
            Err(TitoError::NotFound(format!(
                "No record found with id '{}'",
                id
            )))
        }
    }

    pub async fn find_by_id(
        &self,
        id: &str,
        rels: Vec<String>,
        tx: Option<&E::Transaction>,
    ) -> Result<T, TitoError>
    where
        T: serde::de::DeserializeOwned,
    {
        match tx {
            Some(tx) => self.find_by_id_with_tx(id, rels, tx).await,
            None => {
                let id = id.to_string();
                self.tx(|tx| {
                    let id = id.clone();
                    let rels = rels.clone();
                    async move { self.find_by_id_with_tx(&id, rels, &tx).await }
                })
                .await
            }
        }
    }

    pub async fn scan(
        &self,
        payload: TitoScanPayload,
        tx: &E::Transaction,
    ) -> Result<(Vec<(String, Value)>, bool), TitoError>
    where
        T: DeserializeOwned,
    {
        let mut start_bound = format!("{}", payload.start);
        if let Some(cursor) = payload.cursor.clone() {
            let cursor = self.decode_cursor(cursor)?.first_id()?;
            let after_cursor = next_string_lexicographically(cursor);
            start_bound = after_cursor;
        }

        let end_bound = if let Some(end) = payload.end.clone() {
            end
        } else {
            next_string_lexicographically(payload.start.clone())
        };

        let limit = payload.limit.unwrap_or(u32::MAX);

        let limit_plus_one = if limit == u32::MAX {
            u32::MAX
        } else {
            limit + 1
        };

        let scan_stream = tx
            .scan(start_bound..end_bound, limit_plus_one)
            .await
            .map_err(|e| TitoError::NotFound(e.to_string()))?;

        let mut items = self.to_results(scan_stream)?;

        let has_more = if limit == u32::MAX {
            false
        } else {
            items.len() > limit as usize
        };

        if has_more {
            items.truncate(limit as usize);
        }

        Ok((items, has_more))
    }

    pub async fn find_by_ids_raw(
        &self,
        ids: Vec<String>,
        rels: Vec<String>,
        tx: &E::Transaction,
    ) -> Result<Vec<(String, Value)>, TitoError>
    where
        T: DeserializeOwned,
    {
        let ids = ids
            .into_iter()
            .map(|id| format!("{}:{}", self.get_table(), id))
            .collect();

        let items = self.batch_get(ids, tx).await?;
        let items = self.fetch_and_stitch_relationships(items, rels, tx).await?;

        Ok(items)
    }

    async fn find_by_ids_with_tx(
        &self,
        ids: Vec<String>,
        rels: Vec<String>,
        tx: &E::Transaction,
    ) -> Result<Vec<T>, TitoError>
    where
        T: DeserializeOwned,
    {
        let items = self.find_by_ids_raw(ids, rels, tx).await?;

        let mut result = vec![];

        for value in items.into_iter() {
            if let Ok(item) = serde_json::from_value::<T>(value.1) {
                result.push(item);
            }
        }

        Ok(result)
    }

    pub async fn find_by_ids(
        &self,
        ids: Vec<String>,
        rels: Vec<String>,
        tx: Option<&E::Transaction>,
    ) -> Result<Vec<T>, TitoError>
    where
        T: DeserializeOwned,
    {
        match tx {
            Some(tx) => self.find_by_ids_with_tx(ids, rels, tx).await,
            None => {
                self.tx(|tx| {
                    let ids = ids.clone();
                    let rels = rels.clone();
                    async move { self.find_by_ids_with_tx(ids, rels, &tx).await }
                })
                .await
            }
        }
    }

    pub async fn scan_reverse(
        &self,
        payload: TitoScanPayload,
        tx: &E::Transaction,
    ) -> Result<(Vec<(String, Value)>, bool), TitoError>
    where
        T: DeserializeOwned,
    {
        let start_bound = format!("{}", payload.start.clone());

        let mut end_bound = if let Some(end) = payload.end {
            end
        } else {
            next_string_lexicographically(payload.start.clone())
        };

        if let Some(cursor) = payload.cursor {
            let cursor = self.decode_cursor(cursor)?.first_id()?;
            let after_cursor = previous_string_lexicographically(cursor.clone());
            end_bound = after_cursor;
        }

        let limit = payload.limit.unwrap_or(u32::MAX);

        let limit_plus_one = if limit == u32::MAX {
            u32::MAX
        } else {
            limit + 1
        };

        let scan_stream = tx
            .scan_reverse(start_bound..end_bound, limit_plus_one)
            .await
            .map_err(|e| TitoError::NotFound(e.to_string()))?;

        let mut items = self.to_results(scan_stream)?;

        let has_more = if limit == u32::MAX {
            false
        } else {
            items.len() > limit as usize
        };

        if has_more {
            items.truncate(limit as usize);
        }

        Ok((items, has_more))
    }

    pub async fn update(&self, payload: T, tx: &E::Transaction) -> Result<bool, TitoError>
    where
        T: serde::de::DeserializeOwned,
    {
        let raw_id = payload.id();

        let _ = self.delete_by_id(&raw_id, tx).await;

        self.build(payload, tx).await?;

        Ok(true)
    }

    pub fn get_last_id(&self, key: String) -> Option<String> {
        let parts: Vec<&str> = key.split(':').collect();
        parts.last().map(|last| last.to_string())
    }

    pub async fn batch_get(
        &self,
        keys: Vec<String>,
        tx: &E::Transaction,
    ) -> Result<Vec<(String, Value)>, TitoError> {
        match tx.batch_get(keys.clone()).await {
            Ok(res) => self.to_results(res.into_iter()),
            Err(e) => Err(TitoError::NotFound(format!(
                "Failed to batch get keys {:?}: {}",
                keys, e
            ))),
        }
    }

    pub async fn delete_by_id(&self, raw_id: &str, tx: &E::Transaction) -> Result<bool, TitoError> {
        let id = format!("{}:{}", self.get_table(), raw_id);
        let reverse_index_key = format!("reverse-index:{}", id);

        let source_typed = format!("{}:{}", self.model.table(), raw_id);
        self.clear_references(&source_typed, tx).await?;

        let reverse_index = match self.get_reverse_index(&reverse_index_key, tx).await {
            Ok(idx) => idx,
            Err(_) => return Err(TitoError::NotFound(format!("Entity not found: {}", id))),
        };

        let mut keys = reverse_index.value;

        keys.push(id.clone());
        keys.push(reverse_index_key);

        for key in keys.into_iter() {
            self.delete(key, tx).await?;
        }

        Ok(true)
    }

    pub async fn has_inbound_references(
        _engine: &E,
        target_id: &str,
        tx: &E::Transaction,
    ) -> Result<bool, TitoError> {
        let prefix = format!("reference:i:target:{}:", target_id);
        let end = next_string_lexicographically(prefix.clone());
        let results = tx
            .scan(prefix.clone()..end, 1)
            .await
            .map_err(|e| TitoError::QueryFailed(e.to_string()))?;
        Ok(!results.is_empty())
    }

    pub async fn inbound_references(
        _engine: &E,
        target_id: &str,
        limit: u32,
        tx: &E::Transaction,
    ) -> Result<Vec<String>, TitoError> {
        let prefix = format!("reference:i:target:{}:", target_id);
        let end = next_string_lexicographically(prefix.clone());
        let items = tx
            .scan(prefix.clone()..end, limit)
            .await
            .map_err(|e| TitoError::QueryFailed(e.to_string()))?;
        let mut out = Vec::new();
        for (k, _) in items {
            if let Ok(s) = String::from_utf8(k) {
                if let Some(src) = s.split("source:").nth(1) {
                    out.push(src.to_string());
                }
            }
        }
        Ok(out)
    }

    pub async fn outbound_references(
        _engine: &E,
        source_id: &str,
        limit: u32,
        tx: &E::Transaction,
    ) -> Result<Vec<String>, TitoError> {
        let prefix = format!("reference:o:source:{}:", source_id);
        let end = next_string_lexicographically(prefix.clone());
        let items = tx
            .scan(prefix.clone()..end, limit)
            .await
            .map_err(|e| TitoError::QueryFailed(e.to_string()))?;
        let mut out = Vec::new();
        for (k, _) in items {
            if let Ok(s) = String::from_utf8(k) {
                if let Some(dst) = s.split("target:").nth(1) {
                    out.push(dst.to_string());
                }
            }
        }
        Ok(out)
    }

    async fn clear_references(
        &self,
        source_id: &str,
        tx: &E::Transaction,
    ) -> Result<(), TitoError> {
        let prefix = format!("reference:o:source:{}:", source_id);
        let end = next_string_lexicographically(prefix.clone());
        let items = tx
            .scan(prefix.clone()..end, 10_000)
            .await
            .map_err(|e| TitoError::QueryFailed(e.to_string()))?;
        for (k, _) in items {
            if let Ok(s) = String::from_utf8(k.clone()) {
                if let Some(dst) = s.split("target:").nth(1) {
                    tx.delete(k.clone())
                        .await
                        .map_err(|e| TitoError::DeleteFailed(e.to_string()))?;
                    let rev = format!("reference:i:target:{}:source:{}", dst, source_id);
                    let _ = tx.delete(rev).await;
                }
            }
        }
        Ok(())
    }

    async fn sync_references(
        &self,
        source_id: &str,
        targets: Vec<String>,
        tx: &E::Transaction,
    ) -> Result<(), TitoError> {
        self.clear_references(source_id, tx).await?;
        use std::collections::HashSet;
        let unique: HashSet<String> = targets.into_iter().filter(|t| !t.is_empty()).collect();
        for dst in unique.into_iter() {
            let fwd = format!("reference:o:source:{}:target:{}", source_id, dst);
            let marker = serde_json::to_vec(&1)
                .map_err(|e| TitoError::SerializationFailed(e.to_string()))?;
            tx.put(&fwd, marker.clone())
                .await
                .map_err(|e| TitoError::CreateFailed(e.to_string()))?;
            let rev = format!("reference:i:target:{}:source:{}", dst, source_id);
            tx.put(&rev, marker.clone())
                .await
                .map_err(|e| TitoError::CreateFailed(e.to_string()))?;
        }
        Ok(())
    }

    pub async fn find(&self, payload: TitoFindPayload) -> Result<TitoPaginated<T>, TitoError>
    where
        T: DeserializeOwned,
    {
        let start_bound = format!("{}:{}", self.get_table(), payload.start);

        self.tx(|tx| {
            let start_bound = start_bound.clone();
            let payload = payload.clone();
            async move {
                let (scan_stream, has_more) = self
                    .scan(
                        TitoScanPayload {
                            start: start_bound,
                            end: None,
                            limit: payload.limit,
                            cursor: payload.cursor.clone(),
                        },
                        &tx,
                    )
                    .await?;

                let items = self
                    .fetch_and_stitch_relationships(scan_stream, payload.rels, &tx)
                    .await?;

                self.to_paginated_items(items, has_more)
            }
        })
        .await
    }

    pub async fn migrate<F, Fut>(
        &self,
        f: F,
        checkpoint: &str,
    ) -> Result<MigrateStats, TitoError>
    where
        F: Fn(T) -> Fut + Send + Sync + Clone,
        Fut: Future<Output = Result<Option<T>, TitoError>> + Send,
        T: DeserializeOwned,
    {
        let table_prefix = format!("{}:", self.get_table());
        let end_key = next_string_lexicographically(table_prefix.clone());
        let mut stats = MigrateStats::default();

        let mut cursor = self
            .load_migrate_checkpoint(checkpoint)
            .await?
            .unwrap_or(table_prefix);

        loop {
            let f = f.clone();
            let checkpoint_key = checkpoint.to_string();

            let result = self
                .engine
                .transaction(|tx| {
                    let cursor = cursor.clone();
                    let end_key = end_key.clone();
                    let checkpoint_key = checkpoint_key.clone();

                    async move {
                        let kvs = tx
                            .scan(cursor.as_bytes()..end_key.as_bytes(), 1)
                            .await
                            .map_err(|e| TitoError::QueryFailed(e.to_string()))?;

                        if kvs.is_empty() {
                            return Ok(MigrateResult::Done);
                        }

                        let key = String::from_utf8(kvs[0].0.clone())
                            .map_err(|_| TitoError::DeserializationFailed("Invalid key".into()))?;

                        let record: T = serde_json::from_slice(&kvs[0].1)
                            .map_err(|e| TitoError::DeserializationFailed(e.to_string()))?;

                        let modified = match f(record).await? {
                            Some(updated) => {
                                self.update(updated, &tx).await?;
                                true
                            }
                            None => false,
                        };

                        let next_cursor = next_string_lexicographically(key.clone());
                        tx.put(checkpoint_key.as_bytes(), next_cursor.as_bytes())
                            .await
                            .map_err(|e| TitoError::UpdateFailed(e.to_string()))?;

                        Ok(MigrateResult::Processed {
                            next_cursor,
                            modified,
                        })
                    }
                })
                .await?;

            match result {
                MigrateResult::Done => {
                    self.delete_migrate_checkpoint(checkpoint).await?;
                    break;
                }
                MigrateResult::Processed {
                    next_cursor,
                    modified,
                } => {
                    stats.processed += 1;
                    if modified {
                        stats.modified += 1;
                    } else {
                        stats.skipped += 1;
                    }
                    cursor = next_cursor;
                }
            }
        }

        Ok(stats)
    }

    async fn load_migrate_checkpoint(&self, key: &str) -> Result<Option<String>, TitoError> {
        self.engine
            .transaction(|tx| {
                let key = key.to_string();
                async move {
                    match tx.get(key.as_bytes()).await {
                        Ok(Some(v)) => Ok(Some(String::from_utf8(v).map_err(|_| {
                            TitoError::DeserializationFailed("Invalid checkpoint".into())
                        })?)),
                        Ok(None) => Ok(None),
                        Err(e) => Err(TitoError::QueryFailed(e.to_string())),
                    }
                }
            })
            .await
    }

    async fn delete_migrate_checkpoint(&self, key: &str) -> Result<(), TitoError> {
        self.engine
            .transaction(|tx| {
                let key = key.to_string();
                async move {
                    tx.delete(key.as_bytes())
                        .await
                        .map_err(|e| TitoError::DeleteFailed(e.to_string()))
                }
            })
            .await
    }

    pub async fn find_all(&self) -> Result<TitoPaginated<T>, TitoError> {
        let table_name = self.get_table();
        let start_key = format!("{}:", table_name);
        let end_key = next_string_lexicographically(start_key.clone());

        self.tx(|tx| {
            let start_key = start_key.clone();
            let end_key = end_key.clone();
            async move {
                let (items, _has_more) = self
                    .scan(
                        TitoScanPayload {
                            start: start_key,
                            end: Some(end_key),
                            limit: None,
                            cursor: None,
                        },
                        &tx,
                    )
                    .await?;

                let results: Vec<T> = items
                    .iter()
                    .map(|(_, value)| {
                        serde_json::from_value::<T>(value.clone()).map_err(|_| {
                            TitoError::DeserializationFailed("Failed to deserialize".to_string())
                        })
                    })
                    .collect::<Result<_, _>>()?;

                Ok(TitoPaginated::new(results, None))
            }
        })
        .await
    }
}
