use std::collections::HashSet;

use parquet::file::metadata::ParquetMetaData;
use serde::{Deserialize, Serialize};

use crate::{FEATURE_REF_RECORD_LEN, PropertyProjection};

pub(crate) fn encode_feature_ref(feature: &FeatureRef) -> Vec<u8> {
    let mut out = Vec::with_capacity(FEATURE_REF_RECORD_LEN);
    out.extend_from_slice(&feature.row_number.to_le_bytes());
    out.extend_from_slice(&feature.row_group.unwrap_or(u32::MAX).to_le_bytes());
    out.extend_from_slice(&feature.row_in_group.unwrap_or(u32::MAX).to_le_bytes());
    out.extend_from_slice(&feature.part.unwrap_or(u16::MAX).to_le_bytes());
    out.extend_from_slice(&0u16.to_le_bytes());
    out.extend_from_slice(&0u32.to_le_bytes());
    out
}

pub(crate) fn encode_feature_wkb(feature: &FeatureRef, wkb: &[u8]) -> Vec<u8> {
    let mut out = encode_feature_ref(feature);
    out.extend_from_slice(wkb);
    out
}

/// Decode a fixed-width [`FeatureRef`] payload.
///
/// Returns `None` if the payload is shorter than [`FEATURE_REF_RECORD_LEN`].
pub fn decode_feature_ref_payload(payload: &[u8]) -> Option<FeatureRef> {
    if payload.len() < FEATURE_REF_RECORD_LEN {
        return None;
    }
    let row_number = u64::from_le_bytes(payload[0..8].try_into().ok()?);
    let row_group = decode_u32_option(payload[8..12].try_into().ok()?);
    let row_in_group = decode_u32_option(payload[12..16].try_into().ok()?);
    let part = decode_u16_option(payload[16..18].try_into().ok()?);
    Some(FeatureRef {
        row_number,
        row_group,
        row_in_group,
        part,
        feature_id: None,
    })
}

/// Decode a [`FeatureRef`] followed by WKB bytes.
///
/// This is the payload shape generated by [`PayloadPlan::RowWkb`]. Returns
/// `None` when the fixed feature-ref prefix is truncated.
pub fn decode_feature_wkb_payload(payload: &[u8]) -> Option<(FeatureRef, &[u8])> {
    let feature = decode_feature_ref_payload(payload)?;
    Some((feature, &payload[FEATURE_REF_RECORD_LEN..]))
}

fn decode_u32_option(bytes: [u8; 4]) -> Option<u32> {
    match u32::from_le_bytes(bytes) {
        u32::MAX => None,
        value => Some(value),
    }
}

fn decode_u16_option(bytes: [u8; 2]) -> Option<u16> {
    match u16::from_le_bytes(bytes) {
        u16::MAX => None,
        value => Some(value),
    }
}

pub(crate) fn unique_feature_count(features: &[FeatureRef]) -> usize {
    features
        .iter()
        .map(|feature| feature.row_number)
        .collect::<HashSet<_>>()
        .len()
}

pub(crate) fn entries_may_duplicate_rows(features: &[FeatureRef]) -> bool {
    let mut seen = HashSet::new();
    features
        .iter()
        .any(|feature| !seen.insert(feature.row_number))
}

pub(crate) fn source_fingerprint(meta: &ParquetMetaData) -> String {
    let mut hash = 0xcbf2_9ce4_8422_2325u64;
    hash = fnv(hash, &meta.file_metadata().num_rows().to_le_bytes());
    for col in meta.file_metadata().schema_descr().columns() {
        hash = fnv(hash, col.path().string().as_bytes());
        hash = fnv(hash, format!("{:?}", col.logical_type_ref()).as_bytes());
    }
    format!("fnv64:{hash:016x}")
}

fn fnv(mut hash: u64, bytes: &[u8]) -> u64 {
    for b in bytes {
        hash ^= u64::from(*b);
        hash = hash.wrapping_mul(0x100_0000_01b3);
    }
    hash
}

/// Stable reference back to a source feature.
///
/// # Example
///
/// ```rust
/// use packed_spatial_index_geo::FeatureRef;
///
/// let feature = FeatureRef {
///     row_number: 42,
///     row_group: None,
///     row_in_group: None,
///     part: Some(1),
///     feature_id: None,
/// };
/// assert_eq!(feature.row_number, 42);
/// assert_eq!(feature.part, Some(1));
/// ```
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FeatureRef {
    /// Absolute source row number.
    pub row_number: u64,
    /// Source row group when known.
    pub row_group: Option<u32>,
    /// Row offset within the row group when known.
    pub row_in_group: Option<u32>,
    /// Split part for duplicated index entries.
    pub part: Option<u16>,
    /// Optional feature identifier.
    pub feature_id: Option<String>,
}

impl FeatureRef {
    /// Create a feature ref from an absolute source row number.
    pub fn row_number(row_number: u64) -> Self {
        Self {
            row_number,
            row_group: None,
            row_in_group: None,
            part: None,
            feature_id: None,
        }
    }

    pub(crate) fn row_in_group(row_number: u64, row_group: u32, row_in_group: u32) -> Self {
        Self {
            row_number,
            row_group: Some(row_group),
            row_in_group: Some(row_in_group),
            part: None,
            feature_id: None,
        }
    }
}

/// Payload to attach to converted artifact entries or scan results.
///
/// # Example
///
/// ```no_run
/// use std::fs::File;
/// use packed_spatial_index_geo::{open, ConvertRequest, PayloadPlan, PropertyProjection};
///
/// let mut dataset = open(File::open("cities.parquet")?)?;
/// let bytes = dataset.convert(ConvertRequest {
///     payload: PayloadPlan::FeatureJson {
///         properties: PropertyProjection::AllNonGeometry,
///     },
///     ..ConvertRequest::default()
/// })?;
/// println!("{} bytes", bytes.len());
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// ```
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum PayloadPlan {
    /// Emit no payloads.
    None,
    /// Emit only fixed-width `FeatureRef` records.
    RowRef,
    /// Emit fixed-width `FeatureRef` records followed by WKB bytes.
    RowWkb,
    /// Emit GeoJSON Feature bytes with projected properties.
    FeatureJson {
        /// Property projection.
        properties: PropertyProjection,
    },
}
