//! Functional speech publication uses real recognition and real Google embeddings.
//! Retrieval queries derive from recognized evidence; this is not a relevance benchmark.
use anyhow::{Context, Result};
use searchi_core::SearchIndex;
use searchi_service::{
    FfmpegTools, HybridSearch, MediaLimits, SearchOptions, SpeechProvider, SpeechTranscription,
};
use searchi_speech::{Device, SpeechWorker};
use std::{
    fs,
    path::PathBuf,
    sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
};

struct FailOnceNative {
    worker: SpeechWorker,
    fingerprint: String,
    fail_once: bool,
    calls: Arc<AtomicUsize>,
}
impl SpeechProvider for FailOnceNative {
    fn fingerprint(&self) -> &str {
        &self.fingerprint
    }
    fn transcribe(&mut self, pcm: &[f32], rate: u32) -> Result<SpeechTranscription> {
        self.calls.fetch_add(1, Ordering::SeqCst);
        if self.fail_once {
            self.fail_once = false;
            anyhow::bail!("Injected optional recognition failure");
        }
        self.worker.transcribe(pcm, rate)
    }
}
fn required(name: &str) -> Result<PathBuf> {
    std::env::var_os(name)
        .map(PathBuf::from)
        .with_context(|| format!("Set {name}"))
}
fn native() -> Result<SpeechWorker> {
    SpeechWorker::load(
        &required("SEARCHI_SPEECH_WORKER")?,
        &required("SEARCHI_SPEECH_BASE")?,
        &required("SEARCHI_SPEECH_VAD")?,
        Device::Cpu,
    )
}
fn raw_vectors(db: &std::path::Path) -> Result<Vec<Vec<u8>>> {
    let connection = rusqlite::Connection::open(db)?;
    let mut statement=connection.prepare("SELECT vector FROM semantic_chunks WHERE generation=(SELECT active_generation FROM semantic_meta WHERE singleton=1) AND source_type='audio' ORDER BY ordinal")?;
    let rows = statement
        .query_map([], |row| row.get(0))?
        .collect::<rusqlite::Result<Vec<_>>>()?;
    Ok(rows)
}

#[test]
#[ignore = "Requires verified real speech assets/runtime, Google assets/runtime and short real speech WAV; CPU only"]
fn genuine_speech_policy_upgrade_retries_only_speech_and_preserves_warmed_query_reader(
) -> Result<()> {
    assert_eq!(std::env::var("SEARCHI_FORCE_CPU")?, "1");
    let model = required("SEARCHI_GEMMA_MODEL_DIR")?;
    let gemma = required("SEARCHI_GEMMA_WORKER")?;
    let source = required("SEARCHI_TEST_SPEECH_AUDIO")?;
    anyhow::ensure!(
        fs::metadata(&source)?.len() <= 64 * 1024 * 1024,
        "Test source exceeds byte budget"
    );
    let tools = FfmpegTools {
        ffmpeg: required("SEARCHI_TEST_FFMPEG")?,
        ffprobe: required("SEARCHI_TEST_FFPROBE")?,
    };
    let limits = MediaLimits::default();
    let metadata = searchi_media::probe(&tools, &source, &limits)?;
    anyhow::ensure!(
        metadata.has_audio && metadata.duration_ms <= 30000,
        "Functional fixture must contain real speech in at most thirty seconds"
    );
    let dir = tempfile::tempdir()?;
    let root = dir.path().join("media");
    fs::create_dir(&root)?;
    let path = root.join("meeting.wav");
    fs::copy(&source, &path)?;
    let db = dir.path().join("index.db");
    let mut lexical = SearchIndex::open(&db)?;
    lexical.index_folder(&root)?;
    let mut indexer = HybridSearch::open(&db, &model, &gemma, 1)?;
    let raw = indexer.index_embeddings_with_media(&tools, &limits)?;
    assert!(raw.failures.is_empty());
    assert_eq!(raw.audio_inference_calls, 1);
    let original_vectors = raw_vectors(&db)?;
    assert_eq!(original_vectors.len(), 1);
    indexer = indexer.with_speech_provider(Box::new(native()?))?;
    let speech = indexer.index_embeddings_with_media(&tools, &limits)?;
    assert!(speech.speech_failures.is_empty());
    assert_eq!(speech.audio_inference_calls, 0);
    assert_eq!(speech.speech_windows, 1);
    assert!(speech.speech_chunks > 0);
    assert_eq!(raw_vectors(&db)?, original_vectors);
    let connection = rusqlite::Connection::open(&db)?;
    let recognized:String=connection.query_row("SELECT snippet FROM semantic_chunks WHERE generation=(SELECT active_generation FROM semantic_meta WHERE singleton=1) AND source_type='speech' ORDER BY ordinal LIMIT 1",[],|row|row.get(0))?;
    let mut reader = HybridSearch::open_query(&db, &model, &gemma, 1)?;
    let query = format!("someone says {recognized}");
    let hits = reader.search(&query, 1, &SearchOptions::default())?;
    assert_eq!(hits.len(), 1);
    assert_eq!(hits[0].match_kind, "semantic_speech");
    let location = hits[0]
        .source
        .as_ref()
        .context("recognized source timestamp")?;
    assert_eq!(location.kind, "timestamp_ms");
    let excerpt = reader
        .speech_excerpt_matching(&hits[0].path, location.number, &hits[0].snippet, 4096)?
        .context("verified recognized evidence")?;
    assert_eq!(excerpt.provenance, "local_speech_recognition");
    assert_eq!(excerpt.text, hits[0].snippet);
    let calls = Arc::new(AtomicUsize::new(0));
    let worker = native()?;
    let fingerprint = format!("{};functional-fault-injection-v1", worker.fingerprint());
    indexer = indexer.with_speech_provider(Box::new(FailOnceNative {
        worker,
        fingerprint,
        fail_once: true,
        calls: calls.clone(),
    }))?;
    let partial = indexer.index_embeddings_with_media(&tools, &limits)?;
    assert_eq!(partial.speech_failures.len(), 1);
    assert_eq!(partial.audio_inference_calls, 0);
    assert_eq!(raw_vectors(&db)?, original_vectors);
    let status = HybridSearch::embedding_status(&db)?;
    assert_eq!(status.speech_failures.len(), 1);
    let retry = indexer.index_embeddings_with_media(&tools, &limits)?;
    assert!(retry.speech_failures.is_empty());
    assert_eq!(retry.audio_inference_calls, 0);
    assert_eq!(calls.load(Ordering::SeqCst), 2);
    assert!(retry.speech_chunks > 0);
    assert_eq!(raw_vectors(&db)?, original_vectors);
    let current = HybridSearch::embedding_status(&db)?;
    assert!(current.speech_failures.is_empty());
    assert_eq!(
        current.corpus_revision, status.corpus_revision,
        "Partial publication must not reset automatic retry identity"
    );
    assert_eq!(current.speech_policy, status.speech_policy);
    // The retained foreground Google query session adopts the compatible publication.
    let hits = reader.search(&query, 1, &SearchOptions::default())?;
    assert_eq!(hits[0].match_kind, "semantic_speech");
    Ok(())
}