//! Расшифровка звука встречи по кнопке «Расшифровать».
//!
//! ```text
//! дорожки ─► куски по паузам ─► распознавание (до 4 кусков сразу) ─► реплики по времени
//! ```
//!
//! Куски без речи не отправляются: это экономит деньги, и Whisper не сочиняет
//! фраз на тишине. Неудачный кусок повторяется один раз; если и повтор не
//! прошёл, на его месте в расшифровке остаётся пометка, а не молчаливая дыра.

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use meet_audio::chunker::{self, Chunk, Chunker};
use meet_audio::pcm::{self, samples_to_ms};
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;

use crate::openrouter::SttSegment;
use crate::transcript::{Segment, Speaker};

pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

/// Распознавание одного куска WAV. В приложении — OpenRouter, в тестах — подделка.
pub trait Stt: Send + Sync + 'static {
    fn transcribe(&self, wav: Vec<u8>) -> BoxFuture<'_, Result<Vec<SttSegment>, String>>;
}

pub const UNRECOGNIZED: &str = "*(этот кусок не распознался — нажмите «Расшифровать» ещё раз)*";

#[derive(Debug, Clone, Copy)]
pub struct Config {
    pub chunker: chunker::Config,
    /// Сколько кусков распознаётся одновременно.
    pub parallel: usize,
    pub retry_after: Duration,
}

impl Default for Config {
    fn default() -> Self {
        Self {
            chunker: chunker::Config::default(),
            parallel: 4,
            retry_after: Duration::from_secs(3),
        }
    }
}

#[derive(Debug, Clone, PartialEq)]
pub struct Outcome {
    pub segments: Vec<Segment>,
    /// Сколько кусков не распозналось.
    pub failed: usize,
    /// Последняя ошибка распознавания, чтобы показать человеку причину.
    pub last_error: Option<String>,
}

/// Режет дорожку на куски по паузам.
pub fn chunks_of(samples: &[i16], cfg: chunker::Config) -> Vec<Chunk> {
    let mut c = Chunker::new(cfg);
    let mut out = c.push(samples_to_ms(samples.len()), samples);
    out.extend(c.flush());
    out
}

/// Расшифровку отменил человек.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[error("Расшифровка отменена")]
pub struct Cancelled;

/// Расшифровывает дорожки. `progress(готово, всего)` зовётся после каждого куска.
/// По `cancel` неотправленные куски не отправляются, идущие обрываются.
pub async fn run(
    tracks: Vec<(Speaker, Vec<i16>)>,
    stt: Arc<dyn Stt>,
    cfg: Config,
    cancel: &CancellationToken,
    progress: impl Fn(usize, usize),
) -> Result<Outcome, Cancelled> {
    let jobs: Vec<(Speaker, Chunk)> = tracks
        .into_iter()
        .flat_map(|(speaker, samples)| {
            chunks_of(&samples, cfg.chunker)
                .into_iter()
                .map(move |c| (speaker.clone(), c))
        })
        .collect();
    let total = jobs.len();
    progress(0, total);

    let mut set = JoinSet::new();
    let mut pending = jobs.into_iter();
    let mut segments = Vec::new();
    let mut failed = 0;
    let mut last_error = None;
    let mut done = 0;
    loop {
        while set.len() < cfg.parallel.max(1) {
            let Some((speaker, chunk)) = pending.next() else {
                break;
            };
            let stt = stt.clone();
            let retry_after = cfg.retry_after;
            set.spawn(async move {
                let wav = pcm::encode_wav(&chunk.samples);
                let mut result = stt.transcribe(wav.clone()).await;
                if result.is_err() {
                    tokio::time::sleep(retry_after).await;
                    result = stt.transcribe(wav).await;
                }
                (speaker, chunk.start_ms, result)
            });
        }
        let joined = tokio::select! {
            joined = set.join_next() => joined,
            _ = cancel.cancelled() => {
                set.abort_all();
                return Err(Cancelled);
            }
        };
        let Some(joined) = joined else {
            break;
        };
        let (speaker, start_ms, result) = joined.expect("задача распознавания упала");
        match result {
            Ok(list) => segments.extend(list.into_iter().map(|s| Segment {
                start_ms: start_ms + s.start_ms,
                speaker: speaker.clone(),
                text: s.text,
            })),
            Err(e) => {
                failed += 1;
                last_error = Some(e);
                segments.push(Segment {
                    start_ms,
                    speaker,
                    text: UNRECOGNIZED.into(),
                });
            }
        }
        done += 1;
        progress(done, total);
    }
    segments.sort_by_key(|s| s.start_ms);
    Ok(Outcome {
        segments,
        failed,
        last_error,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use meet_audio::pcm::{ms_to_samples, SAMPLE_RATE};
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Mutex;

    fn speech(ms: u64) -> Vec<i16> {
        (0..ms_to_samples(ms))
            .map(|i| ((i as f32 / SAMPLE_RATE as f32 * std::f32::consts::TAU * 250.0).sin() * 9000.0) as i16)
            .collect()
    }

    fn silence(ms: u64) -> Vec<i16> {
        vec![0; ms_to_samples(ms)]
    }

    /// Подделка: реплика «N мс» на кусок; `fail` — сколько первых вызовов упадут.
    struct Fake {
        calls: AtomicUsize,
        fail: usize,
        in_flight: AtomicUsize,
        max_in_flight: AtomicUsize,
    }

    impl Fake {
        fn new(fail: usize) -> Arc<Self> {
            Arc::new(Self {
                calls: AtomicUsize::new(0),
                fail,
                in_flight: AtomicUsize::new(0),
                max_in_flight: AtomicUsize::new(0),
            })
        }
    }

    impl Stt for Fake {
        fn transcribe(&self, wav: Vec<u8>) -> BoxFuture<'_, Result<Vec<SttSegment>, String>> {
            Box::pin(async move {
                let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
                self.max_in_flight.fetch_max(now, Ordering::SeqCst);
                tokio::time::sleep(Duration::from_millis(20)).await;
                self.in_flight.fetch_sub(1, Ordering::SeqCst);
                let n = self.calls.fetch_add(1, Ordering::SeqCst);
                if n < self.fail {
                    return Err("сеть".into());
                }
                let ms = samples_to_ms(pcm::decode_wav(&wav).unwrap().len());
                Ok(vec![SttSegment {
                    start_ms: 200,
                    text: format!("{ms} мс"),
                }])
            })
        }
    }

    fn cfg() -> Config {
        Config {
            retry_after: Duration::from_millis(5),
            ..Config::default()
        }
    }

    #[tokio::test]
    async fn two_tracks_become_one_timeline() {
        let mut me = speech(7_000);
        me.extend(silence(1_500));
        me.extend(speech(3_000));
        let mut others = silence(2_000);
        others.extend(speech(8_000));
        let log = Mutex::new(Vec::new());
        let out = run(
            vec![(Speaker::Me, me), (Speaker::Others, others)],
            Fake::new(0),
            cfg(),
            &CancellationToken::new(),
            |d, t| log.lock().unwrap().push((d, t)),
        )
        .await
        .unwrap();
        assert_eq!(out.failed, 0);
        assert_eq!(out.segments.len(), 3, "{:#?}", out.segments);
        let starts: Vec<u64> = out.segments.iter().map(|s| s.start_ms).collect();
        assert!(starts.windows(2).all(|w| w[0] <= w[1]), "по времени: {starts:?}");
        assert!(out
            .segments
            .iter()
            .any(|s| s.speaker == Speaker::Others && s.start_ms >= 2_000));
        assert_eq!(log.lock().unwrap().first(), Some(&(0, 3)));
        assert_eq!(log.lock().unwrap().last(), Some(&(3, 3)));
    }

    #[tokio::test]
    async fn chunks_run_in_parallel_but_not_more_than_allowed() {
        let long = speech(28_000 * 10);
        let fake = Fake::new(0);
        let out = run(
            vec![(Speaker::Unknown, long)],
            fake.clone(),
            Config { parallel: 3, ..cfg() },
            &CancellationToken::new(),
            |_, _| {},
        )
        .await
        .unwrap();
        assert_eq!(out.segments.len(), 10);
        let max = fake.max_in_flight.load(Ordering::SeqCst);
        assert!((2..=3).contains(&max), "одновременно было {max}");
    }

    #[tokio::test]
    async fn failed_chunk_is_retried_then_marked() {
        let one = speech(3_000);
        let out = run(
            vec![(Speaker::Unknown, one.clone())],
            Fake::new(1),
            cfg(),
            &CancellationToken::new(),
            |_, _| {},
        )
        .await
        .unwrap();
        assert_eq!(out.failed, 0, "одна ошибка — повтор прошёл");
        let out = run(
            vec![(Speaker::Unknown, one)],
            Fake::new(2),
            cfg(),
            &CancellationToken::new(),
            |_, _| {},
        )
        .await
        .unwrap();
        assert_eq!(out.failed, 1);
        assert_eq!(out.segments[0].text, UNRECOGNIZED);
        assert_eq!(out.last_error.as_deref(), Some("сеть"));
    }

    #[tokio::test]
    async fn cancel_stops_the_work() {
        let fake = Fake::new(0);
        let cancel = CancellationToken::new();
        let stop = cancel.clone();
        let out = run(
            vec![(Speaker::Unknown, speech(28_000 * 20))],
            fake.clone(),
            Config { parallel: 2, ..cfg() },
            &cancel,
            move |done, _| {
                if done == 2 {
                    stop.cancel();
                }
            },
        )
        .await;
        assert_eq!(out, Err(Cancelled));
        assert!(
            fake.calls.load(Ordering::SeqCst) < 20,
            "после отмены куски не отправляются"
        );
    }

    #[tokio::test]
    async fn silence_costs_nothing() {
        let fake = Fake::new(0);
        let out = run(
            vec![(Speaker::Others, silence(60_000))],
            fake.clone(),
            cfg(),
            &CancellationToken::new(),
            |_, _| {},
        )
        .await
        .unwrap();
        assert!(out.segments.is_empty());
        assert_eq!(fake.calls.load(Ordering::SeqCst), 0);
    }
}