//! Candidate provenance and ordering; contains no market-data or broker I/O.
use std::collections::{BTreeMap, BTreeSet};

use chrono::NaiveDate;
use serde::{Deserialize, Serialize};

pub const CANDIDATE_SOURCES_SCHEMA: u32 = 1;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CandidateSourceMode {
    Manual,
    FilteredManual,
    Automatic,
    Mixed,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CandidateSourcePriority {
    #[default]
    ManualFirst,
    AutomaticFirst,
    ListOrder,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CandidateSourcePolicy {
    pub schema_version: u32,
    pub mode: CandidateSourceMode,
    #[serde(default)]
    pub priority: CandidateSourcePriority,
    #[serde(default)]
    pub merged_order: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CandidateMember {
    pub symbol: String,
    pub manual: bool,
    pub automatic: bool,
}

fn symbols(values: &[String], label: &str) -> Result<Vec<String>, String> {
    let mut seen = BTreeSet::new();
    values.iter().map(|value| {
        let symbol = value.trim().to_ascii_uppercase();
        if !symbol.rsplit_once('.').is_some_and(|(code, exchange)| {
            code.len() == 6 && code.bytes().all(|byte| byte.is_ascii_digit())
                && matches!(exchange, "SH" | "SZ" | "BJ")
        }) {
            return Err(format!("{label}: invalid qualified security code {value}"));
        }
        if !seen.insert(symbol.clone()) {
            return Err(format!("{label}: duplicate security {symbol}"));
        }
        Ok(symbol)
    }).collect()
}

impl CandidateSourcePolicy {
    pub fn validate(&self) -> Result<(), String> {
        if self.schema_version != CANDIDATE_SOURCES_SCHEMA {
            return Err("candidate_sources schema_version must be 1".into());
        }
        symbols(&self.merged_order, "candidate_sources.merged_order")?;
        if self.mode != CandidateSourceMode::Mixed && self.priority != CandidateSourcePriority::ManualFirst {
            return Err("candidate source priority only applies to mixed sources".into());
        }
        if self.priority != CandidateSourcePriority::ListOrder && !self.merged_order.is_empty() {
            return Err("merged_order requires list_order priority".into());
        }
        Ok(())
    }

    pub fn uses_screen(&self) -> bool {
        self.mode != CandidateSourceMode::Manual
    }

    pub fn validate_screen_binding(&self, manual: &[String], has_screen: bool) -> Result<(), String> {
        self.validate()?;
        symbols(manual, "manual candidates")?;
        if self.uses_screen() != has_screen {
            return Err("candidate source mode and screen contract must agree".into());
        }
        if self.mode == CandidateSourceMode::FilteredManual && manual.is_empty() {
            return Err("filtered_manual requires manual members; an empty scope must not become all-market".into());
        }
        Ok(())
    }
}

/// Overlap between two valid sources denotes one member with both provenance
/// flags. Duplicates *within* a source are invalid evidence, not fixed by dedup.
pub fn resolve_candidates(
    policy: &CandidateSourcePolicy,
    manual: &[String],
    automatic: Option<&[String]>,
) -> Result<Vec<CandidateMember>, String> {
    policy.validate_screen_binding(manual, automatic.is_some())?;
    let manual = symbols(manual, "manual candidates")?;
    let automatic = automatic.map(|values| symbols(values, "automatic candidates")).transpose()?.unwrap_or_default();
    let manual_set = manual.iter().cloned().collect::<BTreeSet<_>>();
    let auto_set = automatic.iter().cloned().collect::<BTreeSet<_>>();
    if policy.mode == CandidateSourceMode::FilteredManual && !auto_set.is_subset(&manual_set) {
        return Err("filtered_manual snapshot contains a security outside the manual scope".into());
    }
    let mut ordered = match policy.mode {
        CandidateSourceMode::Manual => manual.clone(),
        CandidateSourceMode::FilteredManual | CandidateSourceMode::Automatic => automatic.clone(),
        CandidateSourceMode::Mixed => {
            let (first, second) = if policy.priority == CandidateSourcePriority::AutomaticFirst {
                (&automatic, &manual)
            } else { (&manual, &automatic) };
            let mut union = first.clone();
            let mut seen = first.iter().cloned().collect::<BTreeSet<_>>();
            union.extend(second.iter().filter(|symbol| seen.insert((*symbol).clone())).cloned());
            union
        }
    };
    if policy.priority == CandidateSourcePriority::ListOrder {
        let present = ordered.iter().cloned().collect::<BTreeSet<_>>();
        let prefix = symbols(&policy.merged_order, "candidate_sources.merged_order")?
            .into_iter().filter(|symbol| present.contains(symbol)).collect::<Vec<_>>();
        let selected = prefix.iter().cloned().collect::<BTreeSet<_>>();
        let tail = ordered.into_iter().filter(|symbol| !selected.contains(symbol));
        ordered = prefix.into_iter().chain(tail).collect();
    }
    Ok(ordered.into_iter().map(|symbol| CandidateMember {
        manual: manual_set.contains(&symbol), automatic: auto_set.contains(&symbol), symbol,
    }).collect())
}

/// Raw daily automatic candidates remain unchanged. Every resolved list is
/// derived by the shared kernel; absent dates never inherit yesterday's list.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CandidateSourceBook {
    pub schema_version: u32,
    pub policy: CandidateSourcePolicy,
    pub manual_symbols: Vec<String>,
    pub automatic_symbols_by_date: BTreeMap<NaiveDate, Vec<String>>,
    pub source_snapshot_sha256: String,
    pub source_coverage_sha256: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub execution_symbols: Option<Vec<String>>,
}

impl CandidateSourceBook {
    pub fn resolve(&self) -> Result<BTreeMap<NaiveDate, Vec<CandidateMember>>, String> {
        if self.schema_version != CANDIDATE_SOURCES_SCHEMA || !self.policy.uses_screen() {
            return Err("candidate source book requires schema 1 and a screened source".into());
        }
        for value in [&self.source_snapshot_sha256, &self.source_coverage_sha256] {
            if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
                return Err("candidate source book requires snapshot and coverage SHA256".into());
            }
        }
        if self.automatic_symbols_by_date.is_empty() {
            return Err("candidate source book requires explicit covered trading dates".into());
        }
        let scope = self.execution_symbols.as_ref().map(|values| symbols(values, "candidate execution scope")
            .map(|values| values.into_iter().collect::<BTreeSet<_>>())).transpose()?;
        self.automatic_symbols_by_date.iter().map(|(day, values)| {
            resolve_candidates(&self.policy, &self.manual_symbols, Some(values))
                .map(|members| (*day, members.into_iter().filter(|member| scope.as_ref().is_none_or(|scope| scope.contains(&member.symbol))).collect()))
        }).collect()
    }

    pub fn resolved_symbols(&self) -> Result<BTreeMap<NaiveDate, Vec<String>>, String> {
        Ok(self.resolve()?.into_iter().map(|(date, values)|
            (date, values.into_iter().map(|member| member.symbol).collect())).collect())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    fn list(values: &[&str]) -> Vec<String> { values.iter().map(|value| value.to_string()).collect() }
    fn policy(mode: CandidateSourceMode, priority: CandidateSourcePriority) -> CandidateSourcePolicy {
        CandidateSourcePolicy { schema_version: 1, mode, priority, merged_order: vec![] }
    }
    #[test]
    fn mixed_sources_preserve_priority_and_both_provenances() {
        let manual = list(&["600000.SH", "000001.SZ"]);
        let automatic = list(&["000002.SZ", "000001.SZ"]);
        for (priority, expected) in [
            (CandidateSourcePriority::ManualFirst, list(&["600000.SH", "000001.SZ", "000002.SZ"])),
            (CandidateSourcePriority::AutomaticFirst, list(&["000002.SZ", "000001.SZ", "600000.SH"])),
        ] {
            let result = resolve_candidates(&policy(CandidateSourceMode::Mixed, priority), &manual, Some(&automatic)).unwrap();
            assert_eq!(result.iter().map(|value| value.symbol.clone()).collect::<Vec<_>>(), expected);
            let overlap = result.iter().find(|value| value.symbol == "000001.SZ").unwrap();
            assert!(overlap.manual && overlap.automatic);
        }
    }
    #[test]
    fn list_order_reuses_explicit_prefix_and_appends_new_candidates() {
        let mut p = policy(CandidateSourceMode::Mixed, CandidateSourcePriority::ListOrder);
        p.merged_order = list(&["000002.SZ", "600036.SH", "600000.SH"]);
        let result = resolve_candidates(&p, &list(&["600000.SH", "000001.SZ"]), Some(&list(&["000002.SZ", "000003.SZ"]))).unwrap();
        assert_eq!(result.into_iter().map(|row| row.symbol).collect::<Vec<_>>(), list(&["000002.SZ", "600000.SH", "000001.SZ", "000003.SZ"]));
    }
    #[test]
    fn missing_snapshot_duplicate_input_and_empty_filtered_scope_fail() {
        let p = policy(CandidateSourceMode::Mixed, CandidateSourcePriority::ManualFirst);
        assert!(resolve_candidates(&p, &[], None).is_err());
        assert!(resolve_candidates(&p, &[], Some(&list(&["000001.SZ", "000001.sz"]))).is_err());
        let p = policy(CandidateSourceMode::FilteredManual, CandidateSourcePriority::ManualFirst);
        assert!(resolve_candidates(&p, &[], Some(&[])).unwrap_err().contains("all-market"));
        assert!(resolve_candidates(&p, &list(&["000001.SZ"]), Some(&list(&["600000.SH"]))).is_err());
    }
    #[test]
    fn zero_automatic_day_keeps_manual_members_without_inheriting_old_auto_targets() {
        let day1 = NaiveDate::from_ymd_opt(2026, 9, 9).unwrap();
        let day2 = NaiveDate::from_ymd_opt(2026, 9, 10).unwrap();
        let book = CandidateSourceBook { schema_version: 1,
            policy: policy(CandidateSourceMode::Mixed, CandidateSourcePriority::AutomaticFirst),
            manual_symbols: list(&["510300.SH"]),
            automatic_symbols_by_date: BTreeMap::from([(day1, list(&["000001.SZ"])), (day2, vec![])]),
            source_snapshot_sha256: "a".repeat(64), source_coverage_sha256: "b".repeat(64), execution_symbols:None };
        let result = book.resolved_symbols().unwrap();
        assert_eq!(result[&day1], list(&["000001.SZ", "510300.SH"]));
        assert_eq!(result[&day2], list(&["510300.SH"]));
        let mut auto = book; auto.policy = policy(CandidateSourceMode::Automatic, CandidateSourcePriority::ManualFirst);
        assert!(auto.resolved_symbols().unwrap()[&day2].is_empty());
    }
}