use std::{collections::HashMap, ops::Bound, sync::Arc, time::Duration};

use anyhow::{Context, Result, anyhow, bail};
use async_trait::async_trait;
use log::{debug, info, warn};
use redb::{Database, ReadableTable, TableDefinition, TableError};
use serde::{Serialize, de::DeserializeOwned};
use tici_common::{cdc_file_name::CDCFileName, index::IndexKey};
use tokio::sync::oneshot;

use crate::{
    FragMetaWrapper, ShardMirrorMeta, ShardWriterMeta,
    utils::{
        batch_queue::{BatchExecutor, BatchQueue, Weighted, broadcast_error},
        metrics::{
            METADATA_SCAN_SKIPPED_REASON_DESERIALIZE, METADATA_SCAN_SKIPPED_REASON_ITERATOR,
            inc_metadata_scan_skipped_count,
        },
        schema::IndexMeta,
    },
};

const SHARD_WRITER_META_PREFIX: &str = "sw_";
const SHARD_MIRROR_META_PREFIX: &str = "sm_";
const INDEX_META_PREFIX: &str = "i_";
const FRAG_META_PREFIX: &str = "f_";
const SHARD_DEDUP_BITSET_PREFIX: &str = "db_";

fn shard_writer_meta_key(shard_id: u64) -> String {
    format!("{}{}", SHARD_WRITER_META_PREFIX, shard_id)
}

fn shard_mirror_meta_key(shard_id: u64) -> String {
    format!("{}{}", SHARD_MIRROR_META_PREFIX, shard_id)
}

pub fn frag_meta_key(shard_id: u64) -> String {
    format!("{}{}", FRAG_META_PREFIX, shard_id)
}

pub fn index_meta_key(index_key: &IndexKey) -> String {
    format!(
        "{}{}_{}_{}",
        INDEX_META_PREFIX, index_key.keyspace_id, index_key.table_id, index_key.index_id
    )
}

fn shard_dedup_bitset_key(shard_id: u64) -> String {
    format!("{}{}", SHARD_DEDUP_BITSET_PREFIX, shard_id)
}

fn metadata_prefix_upper_bound(prefix: &str) -> Option<String> {
    for (idx, ch) in prefix.char_indices().rev() {
        if let Some(next_ch) = next_valid_char(ch) {
            let mut upper_bound = String::with_capacity(idx + next_ch.len_utf8());
            upper_bound.push_str(&prefix[..idx]);
            upper_bound.push(next_ch);
            return Some(upper_bound);
        }
    }
    None
}

fn next_valid_char(ch: char) -> Option<char> {
    let mut next = u32::from(ch) + 1;
    while next <= u32::from(char::MAX) {
        if let Some(ch) = char::from_u32(next) {
            return Some(ch);
        }
        next += 1;
    }
    None
}

pub type MetadataMutator = Box<dyn FnOnce(&[u8]) -> Result<Vec<u8>> + Send>;
pub enum MetadataOp {
    // Overwrite the key with the given value.
    Put(Vec<u8>),
    // Write the value only if the key does not exist.
    PutIfAbsent(Vec<u8>),
    // Read the old value, apply a mutation function, and write the result back.
    // This is used for "Read-Modify-Write" atomicity within the batch.
    Modify(MetadataMutator),
    // Delete the key.
    Delete,
}

pub struct MetadataAction {
    key: String,
    op: MetadataOp,
}

pub struct MetadataBatchRequest(pub Vec<MetadataAction>);

impl Weighted for MetadataBatchRequest {
    fn weight(&self) -> usize {
        self.0.len().max(1)
    }
}

const METADATA_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("metadata_v2");

struct RedbExecutor {
    db: Arc<Database>,
}
#[async_trait]
impl BatchExecutor for RedbExecutor {
    type Request = MetadataBatchRequest;
    type Response = ();

    async fn execute_batch(&self, requests: Vec<Self::Request>) -> Vec<Result<Self::Response>> {
        let batch_size = requests.len();
        let db = self.db.clone();

        // Offload blocking I/O (database operations) to a dedicated blocking thread
        // to avoid blocking the Tokio async runtime.
        let result = tokio::task::spawn_blocking(move || {
            let write_txn = match db.begin_write().context("redb::begin_write failed") {
                Ok(txn) => txn,
                Err(e) => return broadcast_error(e, batch_size),
            };

            let mut results: Vec<Result<()>> = Vec::with_capacity(batch_size);

            {
                let mut table = match write_txn
                    .open_table(METADATA_TABLE)
                    .context("redb::open_table failed")
                {
                    Ok(t) => t,
                    Err(e) => return broadcast_error(e, batch_size),
                };

                for req in requests {
                    let op_result = (|| -> Result<()> {
                        let mut staged = HashMap::<String, Option<Vec<u8>>>::new();
                        for MetadataAction { key, op } in req.0 {
                            match op {
                                MetadataOp::Put(val) => {
                                    staged.insert(key, Some(val));
                                }
                                MetadataOp::PutIfAbsent(val) => {
                                    let exists = if let Some(staged_value) = staged.get(&key) {
                                        staged_value.is_some()
                                    } else {
                                        table
                                            .get(key.as_str())
                                            .with_context(|| {
                                                format!("redb::get failed, key: {}", key)
                                            })?
                                            .is_some()
                                    };
                                    if !exists {
                                        staged.insert(key, Some(val));
                                    }
                                }
                                MetadataOp::Modify(mutator) => {
                                    let old_val_owned = if let Some(staged_value) = staged.get(&key)
                                    {
                                        staged_value.clone()
                                    } else {
                                        table
                                            .get(key.as_str())
                                            .with_context(|| {
                                                format!("redb::get failed, key: {}", key)
                                            })?
                                            .map(|guard| guard.value().to_vec())
                                    };

                                    let old_value = old_val_owned
                                        .as_deref()
                                        .ok_or_else(|| anyhow!("Key not found, key: {}", key))?;

                                    let new_val = mutator(old_value).with_context(|| {
                                        format!("Mutator logic failed, key: {}", key)
                                    })?;
                                    staged.insert(key, Some(new_val));
                                }
                                MetadataOp::Delete => {
                                    staged.insert(key, None);
                                }
                            }
                        }

                        for (key, staged_value) in staged {
                            match staged_value {
                                Some(val) => {
                                    table.insert(key.as_str(), val.as_slice()).with_context(
                                        || format!("redb::insert failed, key: {}", key),
                                    )?;
                                }
                                None => {
                                    table.remove(key.as_str()).with_context(|| {
                                        format!("redb::remove failed, key: {}", key)
                                    })?;
                                }
                            }
                        }
                        Ok(())
                    })();

                    results.push(op_result);
                }
            }

            match write_txn.commit() {
                Ok(_) => results,
                Err(commit_err) => {
                    let err_msg = format!("redb::commit failed: {}", commit_err);
                    results
                        .into_iter()
                        .map(|res| match res {
                            Ok(_) => Err(anyhow!(err_msg.clone())),
                            Err(e) => Err(e.context(err_msg.clone())),
                        })
                        .collect()
                }
            }
        })
        .await;

        // Handle thread join errors (e.g., panic in blocking thread)
        match result {
            Ok(batch_results) => batch_results,
            Err(join_err) => (0..batch_size)
                .map(|_| Err(anyhow!("Task Join Error: {}", join_err)))
                .collect(),
        }
    }
}

const MAX_BATCH_WEIGHT: usize = 64;
const MAX_LATENCY_MS: Duration = Duration::from_millis(5);
const CONCURRENCY_LIMIT: usize = 1;
const BATCH_QUEUE_COMPONENT: &str = "metadata_store";
pub struct MetadataStore {
    queue: BatchQueue<RedbExecutor>,
    db: Arc<Database>,
}

impl MetadataStore {
    pub fn new(db_path: &str) -> Result<Self> {
        let db = Database::create(db_path)
            .with_context(|| format!("redb::create failed, db_path: {}", db_path))?;
        let db = Arc::new(db);
        let executor = RedbExecutor { db: db.clone() };
        // redb is a single writer model. Setting concurrency_limit to 1 allows the
        // batch logic to utilize "IO waiting time" to accumulate more data.
        let queue = BatchQueue::new(
            BATCH_QUEUE_COMPONENT,
            executor,
            MAX_BATCH_WEIGHT,
            MAX_LATENCY_MS,
            CONCURRENCY_LIMIT,
        );
        Ok(Self { queue, db })
    }

    pub async fn set_shard_writer_meta(
        &self,
        shard_id: u64,
        shard_meta: &ShardWriterMeta,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let key = shard_writer_meta_key(shard_id);
        self.submit_kv(key, shard_meta, MetadataOp::Put).await
    }

    pub async fn remove_shard_writer_meta(
        &self,
        shard_id: u64,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let key = shard_writer_meta_key(shard_id);
        self.submit_metadata_op(key, MetadataOp::Delete).await
    }

    pub async fn set_shard_mirror_meta(
        &self,
        shard_id: u64,
        shard_meta: &ShardMirrorMeta,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let key = shard_mirror_meta_key(shard_id);
        self.submit_kv(key, shard_meta, MetadataOp::Put).await
    }

    pub async fn remove_shard_mirror_meta(
        &self,
        shard_id: u64,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let key = shard_mirror_meta_key(shard_id);
        self.submit_metadata_op(key, MetadataOp::Delete).await
    }

    pub async fn set_index_meta(
        &self,
        index_key: &IndexKey,
        index_meta: &IndexMeta,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let key = index_meta_key(index_key);
        // IndexMeta is immutable, so we use PutIfAbsent to ensure it's only set once.
        self.submit_kv(key, index_meta, MetadataOp::PutIfAbsent)
            .await
    }

    pub async fn update_last_cdc_file(
        &self,
        shard_id: u64,
        last_file: CDCFileName,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let key = shard_writer_meta_key(shard_id);
        self.atomic_update(key, move |mut meta: ShardWriterMeta| {
            meta.cdc_checkpoint.last_file = last_file;
            Ok(meta)
        })
        .await
    }

    pub async fn set_frag_meta(
        &self,
        shard_id: u64,
        frag_meta: &FragMetaWrapper,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let key = frag_meta_key(shard_id);
        self.submit_kv(key, frag_meta, MetadataOp::Put).await
    }

    pub async fn remove_frag_meta(&self, shard_id: u64) -> Result<oneshot::Receiver<Result<()>>> {
        let key = frag_meta_key(shard_id);
        self.submit_metadata_op(key, MetadataOp::Delete).await
    }

    pub async fn set_shard_dedup_bitset(
        &self,
        shard_id: u64,
        dedup_bitsets: &[u8],
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let key = shard_dedup_bitset_key(shard_id);
        self.submit_metadata_op(key, MetadataOp::Put(dedup_bitsets.to_vec()))
            .await
    }

    pub async fn remove_shard_dedup_bitset(
        &self,
        shard_id: u64,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let key = shard_dedup_bitset_key(shard_id);
        self.submit_metadata_op(key, MetadataOp::Delete).await
    }

    pub async fn persist_reader_local_shard_metadata(
        &self,
        shard_mirror_meta: &ShardMirrorMeta,
        index_meta: &IndexMeta,
        frag_meta: &FragMetaWrapper,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let shard_id = shard_mirror_meta.shard_header.shard_id;
        let index_meta_json = serde_json::to_vec(index_meta).context("Serialization error")?;
        let frag_meta_json = serde_json::to_vec(frag_meta).context("Serialization error")?;
        let shard_mirror_meta_json =
            serde_json::to_vec(shard_mirror_meta).context("Serialization error")?;
        self.submit_metadata_ops(vec![
            MetadataAction {
                key: index_meta_key(&shard_mirror_meta.index_key),
                op: MetadataOp::PutIfAbsent(index_meta_json),
            },
            MetadataAction {
                key: frag_meta_key(shard_id),
                op: MetadataOp::Put(frag_meta_json),
            },
            MetadataAction {
                key: shard_mirror_meta_key(shard_id),
                op: MetadataOp::Put(shard_mirror_meta_json),
            },
        ])
        .await
    }

    pub async fn persist_worker_local_shard_metadata(
        &self,
        shard_writer_meta: &ShardWriterMeta,
        index_meta: &IndexMeta,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let shard_id = shard_writer_meta.shard_header.shard_id;
        let index_meta_json = serde_json::to_vec(index_meta).context("Serialization error")?;
        let shard_writer_meta_json =
            serde_json::to_vec(shard_writer_meta).context("Serialization error")?;
        self.submit_metadata_ops(vec![
            MetadataAction {
                key: index_meta_key(&shard_writer_meta.index_key),
                op: MetadataOp::PutIfAbsent(index_meta_json),
            },
            MetadataAction {
                key: shard_writer_meta_key(shard_id),
                op: MetadataOp::Put(shard_writer_meta_json),
            },
        ])
        .await
    }

    pub async fn remove_reader_local_shard_metadata(
        &self,
        shard_id: u64,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        self.submit_metadata_ops(vec![
            MetadataAction {
                key: frag_meta_key(shard_id),
                op: MetadataOp::Delete,
            },
            MetadataAction {
                key: shard_mirror_meta_key(shard_id),
                op: MetadataOp::Delete,
            },
            MetadataAction {
                key: shard_dedup_bitset_key(shard_id),
                op: MetadataOp::Delete,
            },
        ])
        .await
    }

    pub fn get_shard_dedup_bitset(&self, shard_id: u64) -> Result<Option<Vec<u8>>> {
        let key = shard_dedup_bitset_key(shard_id);
        let read_txn = self.db.begin_read().context("redb::begin_read failed")?;
        let table = match read_txn.open_table(METADATA_TABLE) {
            Ok(table) => table,
            Err(TableError::TableDoesNotExist(_)) => return Ok(None),
            Err(err) => {
                bail!("Failed to open table {}: {:?}", METADATA_TABLE, err);
            }
        };
        Ok(table
            .get(key.as_str())
            .with_context(|| format!("redb::get failed, key: {}", key))?
            .map(|value| value.value().to_vec()))
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn insert_raw_for_test(&self, key: &str, value: &str) -> Result<()> {
        let write_txn = self.db.begin_write().context("redb::begin_write failed")?;
        {
            let mut table = write_txn
                .open_table(METADATA_TABLE)
                .context("redb::open_table failed")?;
            table
                .insert(key, value.as_bytes())
                .with_context(|| format!("redb::insert failed, key: {}", key))?;
        }
        write_txn.commit().context("redb::commit failed")?;
        Ok(())
    }

    pub async fn atomic_update<T, F>(
        &self,
        key: String,
        mutator: F,
    ) -> Result<oneshot::Receiver<Result<()>>>
    where
        T: Serialize + DeserializeOwned + Send + 'static,
        F: FnOnce(T) -> Result<T> + Send + 'static,
    {
        let wrapper_fn = Box::new(move |old_json: &[u8]| -> Result<Vec<u8>> {
            let old_obj: T = serde_json::from_slice(old_json).with_context(|| {
                format!(
                    "Failed to deserialize old json in atomic_update, value len: {}",
                    old_json.len()
                )
            })?;

            let new_obj = mutator(old_obj)?;

            let new_json = serde_json::to_vec(&new_obj)
                .with_context(|| "Failed to serialize new value in RMW")?;

            Ok(new_json)
        });

        self.submit_metadata_op(key, MetadataOp::Modify(wrapper_fn))
            .await
    }

    async fn submit_kv<T, F>(
        &self,
        key: String,
        value: &T,
        op_ctor: F,
    ) -> Result<oneshot::Receiver<Result<()>>>
    where
        T: Serialize,
        F: FnOnce(Vec<u8>) -> MetadataOp,
    {
        let value_json = serde_json::to_vec(value).context("Serialization error")?;

        self.submit_metadata_op(key, op_ctor(value_json)).await
    }

    async fn submit_metadata_op(
        &self,
        key: String,
        op: MetadataOp,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        self.submit_metadata_ops(vec![MetadataAction { key, op }])
            .await
    }

    async fn submit_metadata_ops(
        &self,
        ops: Vec<MetadataAction>,
    ) -> Result<oneshot::Receiver<Result<()>>> {
        let req = MetadataBatchRequest(ops);
        self.queue.submit(req).await
    }

    /// Scans all shard writer metadata entries under the writer prefix.
    ///
    /// Entries that fail to read or deserialize are skipped and one warning
    /// summary is logged instead of failing the whole scan.
    pub fn scan_shard_writer_metas_skip_invalid(&self) -> Result<Vec<(String, ShardWriterMeta)>> {
        self.scan_by_prefix_skip_invalid(SHARD_WRITER_META_PREFIX)
    }

    /// Scans all shard mirror metadata entries.
    ///
    /// If a redb range item cannot be read or a JSON value cannot be
    /// deserialized as [`ShardMirrorMeta`], that entry is skipped and a
    /// warning summary is logged. Use this for best-effort restore when
    /// local metadata may contain stray or corrupted rows.
    pub fn scan_shard_mirror_metas_skip_invalid(&self) -> Result<Vec<(String, ShardMirrorMeta)>> {
        self.scan_by_prefix_skip_invalid(SHARD_MIRROR_META_PREFIX)
    }

    /// Scans all index metadata entries under the index prefix.
    ///
    /// Entries that fail to read or deserialize are skipped and one warning
    /// summary is logged instead of failing the whole scan.
    pub fn scan_index_metas_skip_invalid(&self) -> Result<Vec<(String, IndexMeta)>> {
        self.scan_by_prefix_skip_invalid(INDEX_META_PREFIX)
    }

    /// Scans all fragment metadata entries under the fragment prefix.
    ///
    /// Entries that fail to read from redb or deserialize as
    /// [`FragMetaWrapper`] are skipped and one warning summary is logged
    /// (there is no strict variant exposed for fragment metadata).
    pub fn scan_frag_metas_skip_invalid(&self) -> Result<Vec<(String, FragMetaWrapper)>> {
        self.scan_by_prefix_skip_invalid(FRAG_META_PREFIX)
    }

    /// Omits rows that cannot be read from redb or deserialized from JSON,
    /// logging one warning summary for skipped rows.
    fn scan_by_prefix_skip_invalid<T>(&self, prefix: &str) -> Result<Vec<(String, T)>>
    where
        T: DeserializeOwned + Send + 'static,
    {
        self.scan_by_prefix_impl(prefix, true)
    }

    /// Scans JSON metadata entries for keys starting with `prefix`.
    ///
    /// When `skip_invalid` is `false`, the scan fails on the first redb access
    /// error or JSON parse error. When `true`, those rows are skipped with
    /// per-row debug logs and one warning summary.
    fn scan_by_prefix_impl<T>(&self, prefix: &str, skip_invalid: bool) -> Result<Vec<(String, T)>>
    where
        T: DeserializeOwned + Send + 'static,
    {
        let db = self.db.clone();
        let prefix = prefix.to_string();
        let read_txn = db.begin_read().context("redb::begin_read failed")?;
        let table = match read_txn.open_table(METADATA_TABLE) {
            Ok(table) => table,
            Err(TableError::TableDoesNotExist(_)) => {
                info!("Table {} does not exist", METADATA_TABLE);
                return Ok(Vec::new());
            }
            Err(err) => {
                bail!("Failed to open table {}: {:?}", METADATA_TABLE, err);
            }
        };

        let mut results = Vec::new();
        let mut scanned = 0;
        let mut skipped = 0;
        let mut iterator_skipped = 0;
        let mut deserialize_skipped = 0;
        let range_end = metadata_prefix_upper_bound(&prefix);
        let range: (Bound<&str>, Bound<&str>) = (
            Bound::Included(prefix.as_str()),
            range_end
                .as_deref()
                .map_or(Bound::Unbounded, Bound::Excluded),
        );
        for item in table.range::<&str>(range).context("redb::range failed")? {
            let (key_access, value_access) = match item {
                Ok(item) => item,
                Err(err) if skip_invalid => {
                    scanned += 1;
                    skipped += 1;
                    iterator_skipped += 1;
                    debug!(
                        "Failed to access metadata item while scanning prefix {}, skip it: {:?}",
                        prefix, err
                    );
                    continue;
                }
                Err(err) => return Err(err).context("redb item access failed"),
            };
            let key_str = key_access.value();
            if !key_str.starts_with(&prefix) {
                break;
            }
            scanned += 1;
            let value = value_access.value();
            let meta: T = match serde_json::from_slice(value) {
                Ok(meta) => meta,
                Err(err) if skip_invalid => {
                    skipped += 1;
                    deserialize_skipped += 1;
                    debug!(
                        "Failed to deserialize metadata for key {}, skip it: {:?}",
                        key_str, err
                    );
                    continue;
                }
                Err(err) => {
                    return Err(err).with_context(|| {
                        format!(
                            "Failed to deserialize metadata for key: {}, value len: {}",
                            key_str,
                            value.len()
                        )
                    });
                }
            };
            results.push((key_str.to_string(), meta));
        }
        if skipped > 0 {
            inc_metadata_scan_skipped_count(
                &prefix,
                METADATA_SCAN_SKIPPED_REASON_ITERATOR,
                iterator_skipped,
            );
            inc_metadata_scan_skipped_count(
                &prefix,
                METADATA_SCAN_SKIPPED_REASON_DESERIALIZE,
                deserialize_skipped,
            );
            warn!(
                "scan {} scanned {} rows, skipped {} invalid rows (iterator: {}, deserialize: {})",
                prefix, scanned, skipped, iterator_skipped, deserialize_skipped
            );
        }
        Ok(results)
    }
}

#[cfg(test)]
mod tests {
    use serde::{Deserialize, Serialize};
    use tempfile::NamedTempFile;

    use super::*;
    use crate::utils::metrics::{
        METADATA_SCAN_SKIPPED_REASON_DESERIALIZE, METADATA_SCAN_SKIPPED_REASON_ITERATOR,
        get_metadata_scan_skipped_count,
    };

    #[derive(Debug, Serialize, Deserialize, PartialEq, Clone)]
    struct TestMeta {
        name: String,
        version: u64,
    }

    #[test]
    fn test_metadata_prefix_upper_bound() {
        assert_eq!(metadata_prefix_upper_bound(""), None);
        assert_eq!(metadata_prefix_upper_bound("t_").as_deref(), Some("t`"));
        assert_eq!(
            metadata_prefix_upper_bound("t_\u{10ffff}").as_deref(),
            Some("t`")
        );
        assert_eq!(metadata_prefix_upper_bound("\u{10ffff}"), None);
        assert_eq!(
            metadata_prefix_upper_bound("a\u{d7ff}").as_deref(),
            Some("a\u{e000}")
        );
    }

    impl MetadataStore {
        #[cfg(test)]
        pub fn get_test_value<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>> {
            let db = &self.db;
            let r_txn = db.begin_read()?;
            let table = r_txn.open_table(METADATA_TABLE)?;

            match table.get(key)? {
                Some(access_guard) => {
                    let val = access_guard.value();
                    let t: T =
                        serde_json::from_slice(val).context("Failed to deserialize value")?;
                    Ok(Some(t))
                }
                None => Ok(None),
            }
        }
    }

    #[tokio::test]
    async fn test_submit_kv_sync() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        let key = "shard_100";
        let data = TestMeta {
            name: "shard_main".into(),
            version: 1,
        };

        let rx = service
            .submit_kv(key.to_string(), &data, MetadataOp::Put)
            .await?;

        let res = rx.await?;
        assert!(res.is_ok(), "Database write failed");

        let saved: TestMeta = service
            .get_test_value(key)?
            .ok_or_else(|| anyhow::anyhow!("Key not found"))?;
        assert_eq!(saved, data);

        Ok(())
    }

    #[tokio::test]
    async fn test_shard_dedup_bitset_round_trip() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        assert_eq!(service.get_shard_dedup_bitset(7)?, None);

        let bitset_bytes = vec![1_u8, 2, 3, 4];
        service
            .set_shard_dedup_bitset(7, &bitset_bytes)
            .await?
            .await??;
        assert_eq!(service.get_shard_dedup_bitset(7)?, Some(bitset_bytes));

        service.remove_shard_dedup_bitset(7).await?.await??;
        assert_eq!(service.get_shard_dedup_bitset(7)?, None);
        Ok(())
    }

    #[tokio::test]
    async fn test_mixed_metadata_and_bytes_requests() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        let meta_key = "mixed_meta";
        let meta = TestMeta {
            name: "mixed".into(),
            version: 42,
        };
        let bitset_bytes = vec![7_u8, 8, 9];

        let meta_rx = service
            .submit_kv(meta_key.to_string(), &meta, MetadataOp::Put)
            .await?;
        let bitset_rx = service.set_shard_dedup_bitset(9, &bitset_bytes).await?;

        meta_rx.await??;
        bitset_rx.await??;

        let saved: TestMeta = service
            .get_test_value(meta_key)?
            .ok_or_else(|| anyhow::anyhow!("Key not found"))?;
        assert_eq!(saved, meta);
        assert_eq!(service.get_shard_dedup_bitset(9)?, Some(bitset_bytes));

        Ok(())
    }

    #[tokio::test]
    async fn test_atomic_update() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        let key = "config_v1".to_string();
        let initial = TestMeta {
            name: "init".into(),
            version: 10,
        };

        let rx_init = service
            .submit_kv(key.clone(), &initial, MetadataOp::Put)
            .await?;
        rx_init.await??;

        let rx_update = service
            .atomic_update(key.clone(), |mut t: TestMeta| {
                t.version += 1;
                t.name = "updated".into();
                Ok(t)
            })
            .await?;

        rx_update.await??;

        let updated: TestMeta = service
            .get_test_value(&key)?
            .ok_or_else(|| anyhow::anyhow!("Key not found"))?;
        assert_eq!(updated.version, 11);
        assert_eq!(updated.name, "updated");

        Ok(())
    }

    #[tokio::test]
    async fn test_put_if_absent() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        let key = "lock_file";
        let val1 = TestMeta {
            name: "holder_A".into(),
            version: 1,
        };
        let val2 = TestMeta {
            name: "holder_B".into(),
            version: 2,
        };

        let rx1 = service
            .submit_kv(key.to_string(), &val1, MetadataOp::PutIfAbsent)
            .await?;
        rx1.await??;

        let rx2 = service
            .submit_kv(key.to_string(), &val2, MetadataOp::PutIfAbsent)
            .await?;
        rx2.await??;

        let current: TestMeta = service
            .get_test_value(key)?
            .ok_or_else(|| anyhow::anyhow!("Key not found"))?;
        assert_eq!(current, val1);

        Ok(())
    }

    #[tokio::test]
    async fn test_atomic_update_fail_if_missing() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        let rx = service
            .atomic_update("missing_key".to_string(), |t: TestMeta| Ok(t))
            .await?;

        let batch_result = rx.await?;

        assert!(batch_result.is_err());
        assert!(
            batch_result
                .unwrap_err()
                .to_string()
                .contains("Key not found")
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_batch_partial_failure() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        let key_success_1 = "valid_key_1".to_string();
        let key_fail = "missing_key".to_string();
        let key_success_2 = "valid_key_2".to_string();

        let data_1 = TestMeta {
            name: "success_1".into(),
            version: 1,
        };
        let data_2 = TestMeta {
            name: "success_2".into(),
            version: 2,
        };

        let rx1 = service
            .submit_kv(key_success_1.clone(), &data_1, MetadataOp::Put)
            .await?;
        let rx2 = service
            .atomic_update(key_fail.clone(), |d: TestMeta| Ok(d))
            .await?;
        let rx3 = service
            .submit_kv(key_success_2.clone(), &data_2, MetadataOp::Put)
            .await?;

        let res1 = rx1.await?;
        let res2 = rx2.await?;
        let res3 = rx3.await?;

        assert!(res1.is_ok(), "Request 1 should succeed");
        assert!(res2.is_err(), "Request 2 should fail");
        let err_msg = res2.unwrap_err().to_string();
        assert!(
            err_msg.contains("Key not found"),
            "Error message should indicate missing key, got: {}",
            err_msg
        );
        assert!(
            res3.is_ok(),
            "Request 3 should succeed despite Request 2 failing"
        );

        let saved_1: TestMeta = service
            .get_test_value(&key_success_1)?
            .ok_or_else(|| anyhow::anyhow!("Key not found"))?;
        assert_eq!(saved_1, data_1, "Data 1 should be persisted");

        let saved_2: TestMeta = service
            .get_test_value(&key_success_2)?
            .ok_or_else(|| anyhow::anyhow!("Key not found"))?;
        assert_eq!(saved_2, data_2, "Data 2 should be persisted");

        Ok(())
    }

    #[tokio::test]
    async fn test_multi_op_request_is_atomic() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        let key = "atomic_key".to_string();
        let initial = TestMeta {
            name: "initial".into(),
            version: 1,
        };
        service
            .submit_kv(key.clone(), &initial, MetadataOp::Put)
            .await?
            .await??;

        let rx = service
            .submit_metadata_ops(vec![
                MetadataAction {
                    key: key.clone(),
                    op: MetadataOp::Delete,
                },
                MetadataAction {
                    key: "missing_key".to_string(),
                    op: MetadataOp::Modify(Box::new(|_| Ok(Vec::new()))),
                },
            ])
            .await?;
        let res = rx.await?;
        assert!(res.is_err(), "batched request should fail");

        let saved: TestMeta = service
            .get_test_value(&key)?
            .ok_or_else(|| anyhow::anyhow!("Key not found"))?;
        assert_eq!(
            saved, initial,
            "failed batched request should not partially apply"
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_scan_prefix_isolation() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        // Group A: "a_"
        let a1 = TestMeta {
            name: "A1".into(),
            version: 1,
        };
        let a2 = TestMeta {
            name: "A2".into(),
            version: 2,
        };

        // Group B: "b_"
        let b1 = TestMeta {
            name: "B1".into(),
            version: 10,
        };
        let b2 = TestMeta {
            name: "B2".into(),
            version: 20,
        };
        let b3 = TestMeta {
            name: "B3".into(),
            version: 30,
        };

        // Group C: "c_"
        let c1 = TestMeta {
            name: "C1".into(),
            version: 100,
        };

        service
            .submit_kv("a_1".to_string(), &a1, MetadataOp::Put)
            .await?
            .await??;
        service
            .submit_kv("a_2".to_string(), &a2, MetadataOp::Put)
            .await?
            .await??;

        service
            .submit_kv("b_1".to_string(), &b1, MetadataOp::Put)
            .await?
            .await??;
        service
            .submit_kv("b_2".to_string(), &b2, MetadataOp::Put)
            .await?
            .await??;
        service
            .submit_kv("b_3".to_string(), &b3, MetadataOp::Put)
            .await?
            .await??;

        service
            .submit_kv("c_1".to_string(), &c1, MetadataOp::Put)
            .await?
            .await??;

        let results: Vec<(String, TestMeta)> = service.scan_by_prefix_impl("b_", false)?;
        assert_eq!(
            results.len(),
            3,
            "Should find exactly 3 items starting with 'b_'"
        );
        assert_eq!(results[0], ("b_1".to_string(), b1));
        assert_eq!(results[1], ("b_2".to_string(), b2));
        assert_eq!(results[2], ("b_3".to_string(), b3));

        let empty_results: Vec<(String, TestMeta)> = service.scan_by_prefix_impl("x_", false)?;
        assert!(
            empty_results.is_empty(),
            "Should return empty list for unused prefix"
        );
        let a_results: Vec<(String, TestMeta)> = service.scan_by_prefix_impl("a_", false)?;
        assert_eq!(a_results.len(), 2);
        assert_eq!(a_results[0], ("a_1".to_string(), a1));
        assert_eq!(a_results[1], ("a_2".to_string(), a2));

        Ok(())
    }

    #[tokio::test]
    async fn test_scan_mixed_types() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        #[derive(Serialize, Deserialize, Debug, PartialEq)]
        struct OtherMeta {
            desc: String,
        }

        let val_test = TestMeta {
            name: "target".into(),
            version: 1,
        };
        let val_other = OtherMeta {
            desc: "noise".into(),
        };

        service
            .submit_kv("t_1".to_string(), &val_test, MetadataOp::Put)
            .await?
            .await??;
        service
            .submit_kv("o_1".to_string(), &val_other, MetadataOp::Put)
            .await?
            .await??;

        let res: Vec<(String, TestMeta)> = service.scan_by_prefix_impl("t_", false)?;
        assert_eq!(res.len(), 1);
        assert_eq!(res[0], ("t_1".to_string(), val_test));

        let res_err: Result<Vec<(String, TestMeta)>> = service.scan_by_prefix_impl("o_", false);
        assert!(
            res_err.is_err(),
            "Should fail to deserialize OtherMeta as TestMeta"
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_scan_prefix_skip_invalid_skips_bad_rows() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;
        let prefix = format!("t_{}_", uuid::Uuid::new_v4().simple());

        let valid = TestMeta {
            name: "target".into(),
            version: 1,
        };
        let valid_key = format!("{prefix}1");
        service
            .submit_kv(valid_key.clone(), &valid, MetadataOp::Put)
            .await?
            .await??;

        {
            let write_txn = service.db.begin_write()?;
            {
                let mut table = write_txn.open_table(METADATA_TABLE)?;
                table.insert(format!("{prefix}bad_json").as_str(), "{bad json".as_bytes())?;
                table.insert(format!("{prefix}bad_type").as_str(), b"[]".as_slice())?;
            }
            write_txn.commit()?;
        }

        let deserialize_skipped_before =
            get_metadata_scan_skipped_count(&prefix, METADATA_SCAN_SKIPPED_REASON_DESERIALIZE);
        let iterator_skipped_before =
            get_metadata_scan_skipped_count(&prefix, METADATA_SCAN_SKIPPED_REASON_ITERATOR);

        let strict_res: Result<Vec<(String, TestMeta)>> =
            service.scan_by_prefix_impl(&prefix, false);
        assert!(strict_res.is_err());
        assert_eq!(
            get_metadata_scan_skipped_count(&prefix, METADATA_SCAN_SKIPPED_REASON_DESERIALIZE),
            deserialize_skipped_before
        );
        assert_eq!(
            get_metadata_scan_skipped_count(&prefix, METADATA_SCAN_SKIPPED_REASON_ITERATOR),
            iterator_skipped_before
        );

        let skip_invalid_res: Vec<(String, TestMeta)> =
            service.scan_by_prefix_skip_invalid(&prefix)?;
        assert_eq!(skip_invalid_res, vec![(valid_key, valid)]);
        assert_eq!(
            get_metadata_scan_skipped_count(&prefix, METADATA_SCAN_SKIPPED_REASON_DESERIALIZE)
                - deserialize_skipped_before,
            2
        );
        assert_eq!(
            get_metadata_scan_skipped_count(&prefix, METADATA_SCAN_SKIPPED_REASON_ITERATOR)
                - iterator_skipped_before,
            0
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_scan_shard_writer_metas_skip_invalid_skips_bad_rows() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        let valid = crate::test_utils::test_get_shard_writer_meta();
        let valid_key = shard_writer_meta_key(valid.shard_header.shard_id);
        service
            .set_shard_writer_meta(valid.shard_header.shard_id, &valid)
            .await?
            .await??;

        service.insert_raw_for_test(&shard_writer_meta_key(1001), "{bad json")?;
        service.insert_raw_for_test(&shard_writer_meta_key(1002), "[]")?;

        let skip_invalid_res = service.scan_shard_writer_metas_skip_invalid()?;
        assert_eq!(skip_invalid_res.len(), 1);
        assert_eq!(skip_invalid_res[0].0, valid_key);
        assert_eq!(
            skip_invalid_res[0].1.shard_header.shard_id,
            valid.shard_header.shard_id
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_remove() -> Result<()> {
        let file = NamedTempFile::new()?;
        let path = file
            .path()
            .to_str()
            .context("Failed to convert path to string")?;
        let service = MetadataStore::new(path)?;

        let shard_id = 12345;
        let key = super::shard_writer_meta_key(shard_id);
        let data = TestMeta {
            name: "to_be_deleted".into(),
            version: 1,
        };

        service
            .submit_kv(key.clone(), &data, MetadataOp::Put)
            .await?
            .await??;

        let saved: Option<TestMeta> = service.get_test_value(&key)?;
        assert_eq!(saved, Some(data.clone()));

        service.remove_shard_writer_meta(shard_id).await?.await??;

        let deleted: Option<TestMeta> = service.get_test_value(&key)?;
        assert!(deleted.is_none(), "Key should be removed after deletion");

        let res = service.remove_shard_writer_meta(shard_id).await?.await?;
        assert!(
            res.is_ok(),
            "Second removal should be idempotent/successful"
        );

        Ok(())
    }
}