//! CRC file reading functionality.

use tracing::instrument;

use super::Crc;
use crate::metrics::reporter::CRC_READ_COMPLETED_SPAN;
use crate::path::{AsUrl as _, ParsedLogPath};
use crate::{DeltaResult, Engine, Error};

/// Attempt to read and parse a CRC file.
///
/// Reads raw bytes via the storage handler and deserializes with serde_json.
///
/// Returns `Ok(Crc)` on success, `Err` on any failure (file not readable, corrupt JSON,
/// missing required fields). The caller should handle errors gracefully by falling back to log
/// replay.
///
/// Reports metrics: `CrcReadCompleted`.
#[instrument(name = CRC_READ_COMPLETED_SPAN, skip_all, fields(report, bytes_read))]
pub(crate) fn try_read_crc_file(engine: &dyn Engine, crc_path: &ParsedLogPath) -> DeltaResult<Crc> {
    let storage = engine.storage_handler();
    let url = crc_path.location.as_url().clone();
    let data = storage
        .read_files(vec![(url, None)])?
        .next()
        .ok_or_else(|| Error::generic("CRC file read returned no data"))??;
    tracing::Span::current().record("bytes_read", data.len() as u64);
    let crc: Crc = serde_json::from_slice(&data)?;
    Ok(crc)
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::path::PathBuf;
    use std::sync::Arc;

    use test_utils::assert_result_error_with_message;

    use super::*;
    use crate::actions::{Format, Metadata, Protocol};
    use crate::engine::sync::SyncEngine;
    use crate::metrics::MetricEvent;
    use crate::path::ParsedLogPath;
    use crate::table_features::TableFeature;
    use crate::utils::test_utils::{install_thread_local_metrics_reporter, CapturingReporter};

    fn test_table_root(dir: &str) -> url::Url {
        let path = std::fs::canonicalize(PathBuf::from(dir)).unwrap();
        url::Url::from_directory_path(path).unwrap()
    }

    #[test]
    fn test_read_crc_file() {
        let reporter = Arc::new(CapturingReporter::default());
        let _guard = install_thread_local_metrics_reporter(reporter.clone());

        let engine = SyncEngine::new();
        let table_root = test_table_root("./tests/data/crc-full/");
        let crc_path = ParsedLogPath::create_parsed_crc(&table_root, 0);

        // Read and parse the CRC file
        let crc = try_read_crc_file(&engine, &crc_path).unwrap();

        // Verify basic fields
        let stats = crc.file_stats().unwrap();
        assert_eq!(stats.table_size_bytes(), 5259);
        assert_eq!(stats.num_files(), 10);
        assert_eq!(crc.in_commit_timestamp_opt, Some(1694758257000));

        // Verify protocol
        let expected_protocol = Protocol::new_unchecked(
            3,
            7,
            Some(vec![TableFeature::DeletionVectors]),
            Some(vec![
                TableFeature::DomainMetadata,
                TableFeature::ClusteredTable,
                TableFeature::DeletionVectors,
                TableFeature::RowTracking,
            ]),
        );
        assert_eq!(crc.protocol, expected_protocol);

        // Verify metadata
        let expected_metadata = Metadata::new_unchecked(
            "6ca3020b-3cd9-4048-82e3-1417a0abb98f",
            None,
            None,
            Format::default(),
            r#"{"type":"struct","fields":[{"name":"id","type":"long","nullable":true,"metadata":{}}]}"#,
            vec![],
            Some(1694758256009),
            HashMap::from([
                (
                    "delta.enableDeletionVectors".to_string(),
                    "true".to_string(),
                ),
                (
                    "delta.checkpoint.writeStatsAsStruct".to_string(),
                    "true".to_string(),
                ),
                ("delta.enableRowTracking".to_string(), "true".to_string()),
                (
                    "delta.checkpoint.writeStatsAsJson".to_string(),
                    "false".to_string(),
                ),
                (
                    "delta.rowTracking.materializedRowCommitVersionColumnName".to_string(),
                    "_row-commit-version-col-2f60dcc1-9e36-4424-95e7-799b707e4ddb".to_string(),
                ),
                (
                    "delta.rowTracking.materializedRowIdColumnName".to_string(),
                    "_row-id-col-4cbc7924-f662-4db1-aa59-22c23f59eb5d".to_string(),
                ),
            ]),
        );
        assert_eq!(crc.metadata, expected_metadata);

        // Verify domain metadatas
        let dms = crc.domain_metadata_state.expect_complete();
        assert_eq!(dms.len(), 3);

        assert!(dms["delta.clustering"]
            .configuration()
            .contains("clusteringColumns"));
        assert!(dms["delta.rowTracking"]
            .configuration()
            .contains("rowIdHighWaterMark"));
        assert!(dms["myApp.metadata"].configuration().contains("key"));

        // Verify set transactions
        let txns = crc.set_transaction_state.expect_complete();
        assert_eq!(txns.len(), 2);
        assert_eq!(txns["spark-app-1"].version, 42);
        assert_eq!(txns["spark-app-1"].last_updated, Some(1694758250000));
        assert_eq!(txns["streaming-job-abc"].version, 100);
        assert_eq!(txns["streaming-job-abc"].last_updated, Some(1694758255000));

        // Verify file size histogram was deserialized (all 10 files in bin 0, < 8KB)
        let hist = stats.file_size_histogram().unwrap();
        assert_eq!(hist.sorted_bin_boundaries.len(), 95);
        assert_eq!(hist.file_counts[0], 10);
        assert_eq!(hist.total_bytes[0], 5259);
        // All other bins should be zero
        assert!(hist.file_counts[1..].iter().all(|&c| c == 0));
        assert!(hist.total_bytes[1..].iter().all(|&b| b == 0));

        // These fields are on the Crc struct but not yet round-tripped through CrcRaw.
        assert!(crc.txn_id.is_none());
        assert!(crc.all_files.is_none());
        assert!(crc.num_deleted_records_opt.is_none());
        assert!(crc.num_deletion_vectors_opt.is_none());
        assert!(crc.deleted_record_counts_histogram_opt.is_none());

        // Verify CrcReadCompleted metric was emitted
        let crc_events: Vec<_> = reporter
            .events()
            .into_iter()
            .filter(|e| matches!(e, MetricEvent::CrcReadCompleted { .. }))
            .collect();
        assert_eq!(crc_events.len(), 1);
        assert!(
            matches!(&crc_events[0], MetricEvent::CrcReadCompleted { bytes_read, .. } if *bytes_read > 0)
        );
    }

    #[test]
    fn test_read_malformed_crc_file_emits_metric_then_fails() {
        let reporter = Arc::new(CapturingReporter::default());
        let _guard = install_thread_local_metrics_reporter(reporter.clone());

        let engine = SyncEngine::new();
        let table_root = test_table_root("./tests/data/crc-malformed/");
        let crc_path = ParsedLogPath::create_parsed_crc(&table_root, 0);

        assert_result_error_with_message(try_read_crc_file(&engine, &crc_path), "expected value");

        // CrcReadCompleted is emitted after storage read, even when JSON parse fails
        let crc_events: Vec<_> = reporter
            .events()
            .into_iter()
            .filter(|e| matches!(e, MetricEvent::CrcReadCompleted { .. }))
            .collect();
        assert_eq!(crc_events.len(), 1);
        assert!(
            matches!(&crc_events[0], MetricEvent::CrcReadCompleted { bytes_read, .. } if *bytes_read > 0)
        );
    }
}
