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

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

use meet_audio::chunker::{self, Chunker};
use meet_audio::pcm::{self, ms_to_samples, samples_to_ms, WavFile};
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 trait Samples: Send + Sync + 'static {
    fn len(&self) -> usize;
    fn read(&self, start: usize, n: usize) -> io::Result<Vec<i16>>;
    fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

impl Samples for WavFile {
    fn len(&self) -> usize {
        WavFile::len(self)
    }
    fn read(&self, start: usize, n: usize) -> io::Result<Vec<i16>> {
        WavFile::read(self, start, n)
    }
}

impl Samples for Vec<i16> {
    fn len(&self) -> usize {
        self.as_slice().len()
    }
    fn read(&self, start: usize, n: usize) -> io::Result<Vec<i16>> {
        let start = start.min(self.as_slice().len());
        let end = (start + n).min(self.as_slice().len());
        Ok(self[start..end].to_vec())
    }
}

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>,
}

/// Ход расшифровки: сколько кусков готово и какие реплики добавились.
#[derive(Debug)]
pub struct Update<'a> {
    pub done: usize,
    pub total: usize,
    /// Новые реплики по порядку времени; пусто, пока не готов кусок раньше.
    pub fresh: &'a [Segment],
}

/// Кусок для распознавания: где он в дорожке. Самого звука здесь нет.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Span {
    pub start_ms: u64,
    pub samples: usize,
}

/// Блок, которым дорожка проходит через нарезку.
const BLOCK_MS: u64 = 10_000;

/// Режет дорожку на куски по паузам, читая её блоками.
pub fn spans_of(source: &dyn Samples, cfg: chunker::Config) -> io::Result<Vec<Span>> {
    let mut c = Chunker::new(cfg);
    let block = ms_to_samples(BLOCK_MS);
    let mut spans = Vec::new();
    let mut pos = 0;
    let span = |ch: chunker::Chunk| Span {
        start_ms: ch.start_ms,
        samples: ch.samples.len(),
    };
    while pos < source.len() {
        let samples = source.read(pos, block)?;
        if samples.is_empty() {
            break;
        }
        pos += samples.len();
        spans.extend(c.push(samples_to_ms(pos), &samples).into_iter().map(span));
    }
    spans.extend(c.flush().into_iter().map(span));
    Ok(spans)
}

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

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error(transparent)]
    Cancelled(#[from] Cancelled),
    #[error("Не читается звук: {0}")]
    Read(#[from] io::Error),
}

/// Расшифровывает дорожки. `progress` зовётся в начале и после каждого куска.
/// По `cancel` неотправленные куски не отправляются, идущие обрываются.
pub async fn run(
    tracks: Vec<(Speaker, Arc<dyn Samples>)>,
    stt: Arc<dyn Stt>,
    cfg: Config,
    cancel: &CancellationToken,
    progress: impl Fn(Update<'_>),
) -> Result<Outcome, Error> {
    // Границы кусков — в отдельном потоке: это чтение всего звука с диска.
    let planned = tokio::task::spawn_blocking(move || {
        tracks
            .into_iter()
            .map(|(speaker, source)| {
                let spans = spans_of(source.as_ref(), cfg.chunker)?;
                Ok::<_, io::Error>((speaker, source, spans))
            })
            .collect::<Result<Vec<_>, _>>()
    })
    .await
    .expect("нарезка упала")?;
    let mut jobs: Vec<(Speaker, Arc<dyn Samples>, Span)> = planned
        .into_iter()
        .flat_map(|(speaker, source, spans)| {
            spans
                .into_iter()
                .map(move |span| (speaker.clone(), source.clone(), span))
        })
        .collect();
    // По времени, чтобы первыми распознавались и показывались ранние куски.
    jobs.sort_by_key(|(_, _, span)| span.start_ms);
    let total = jobs.len();
    progress(Update {
        done: 0,
        total,
        fresh: &[],
    });

    let mut set = JoinSet::new();
    let mut pending = jobs.into_iter().enumerate();
    let mut segments = Vec::new();
    // Готовые куски, которые ждут, пока распознаются куски раньше них.
    let mut waiting: BTreeMap<usize, Vec<Segment>> = BTreeMap::new();
    let mut next_out = 0;
    let mut failed = 0;
    let mut last_error = None;
    let mut done = 0;
    loop {
        while set.len() < cfg.parallel.max(1) {
            let Some((index, (speaker, source, span))) = pending.next() else {
                break;
            };
            let stt = stt.clone();
            let retry_after = cfg.retry_after;
            set.spawn(async move {
                let start = ms_to_samples(span.start_ms);
                let samples = tokio::task::spawn_blocking(move || source.read(start, span.samples))
                    .await
                    .expect("чтение упало");
                let result = match samples {
                    Ok(samples) => {
                        let wav = pcm::encode_wav(&samples);
                        drop(samples);
                        let mut result = stt.transcribe(wav.clone()).await;
                        if result.is_err() {
                            tokio::time::sleep(retry_after).await;
                            result = stt.transcribe(wav).await;
                        }
                        result
                    }
                    Err(e) => Err(format!("не читается звук: {e}")),
                };
                (index, speaker, span.start_ms, result)
            });
        }
        let joined = tokio::select! {
            joined = set.join_next() => joined,
            _ = cancel.cancelled() => {
                set.abort_all();
                return Err(Cancelled.into());
            }
        };
        let Some(joined) = joined else {
            break;
        };
        let (index, speaker, start_ms, result) = joined.expect("задача распознавания упала");
        let mut got: Vec<Segment> = match result {
            Ok(list) => list
                .into_iter()
                .map(|s| Segment {
                    start_ms: start_ms + s.start_ms,
                    speaker: speaker.clone(),
                    text: s.text,
                })
                .collect(),
            Err(e) => {
                failed += 1;
                last_error = Some(e);
                vec![Segment {
                    start_ms,
                    speaker,
                    text: UNRECOGNIZED.into(),
                }]
            }
        };
        got.sort_by_key(|s| s.start_ms);
        waiting.insert(index, got);
        let mut fresh = Vec::new();
        while let Some(ready) = waiting.remove(&next_out) {
            fresh.extend(ready);
            next_out += 1;
        }
        done += 1;
        progress(Update {
            done,
            total,
            fresh: &fresh,
        });
        segments.extend(fresh);
    }
    segments.sort_by_key(|s| s.start_ms);
    Ok(Outcome {
        segments,
        failed,
        last_error,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use meet_audio::pcm::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)]
    }

    fn track(speaker: Speaker, samples: Vec<i16>) -> (Speaker, Arc<dyn Samples>) {
        (speaker, Arc::new(samples))
    }

    /// Подделка: реплика «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![track(Speaker::Me, me), track(Speaker::Others, others)],
            Fake::new(0),
            cfg(),
            &CancellationToken::new(),
            |u| log.lock().unwrap().push((u.done, u.total)),
        )
        .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![track(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![track(Speaker::Unknown, one.clone())],
            Fake::new(1),
            cfg(),
            &CancellationToken::new(),
            |_| {},
        )
        .await
        .unwrap();
        assert_eq!(out.failed, 0, "одна ошибка — повтор прошёл");
        let out = run(
            vec![track(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![track(Speaker::Unknown, speech(28_000 * 20))],
            fake.clone(),
            Config { parallel: 2, ..cfg() },
            &cancel,
            move |u| {
                if u.done == 2 {
                    stop.cancel();
                }
            },
        )
        .await;
        assert!(matches!(out, Err(Error::Cancelled(_))), "{out:?}");
        assert!(
            fake.calls.load(Ordering::SeqCst) < 20,
            "после отмены куски не отправляются"
        );
    }

    #[tokio::test]
    async fn silence_costs_nothing() {
        let fake = Fake::new(0);
        let out = run(
            vec![track(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);
    }

    #[test]
    fn spans_by_blocks_match_cutting_the_whole_track() {
        // Нарезка блоками с диска должна дать те же куски, что нарезка целиком.
        let mut all = Vec::new();
        for i in 0..40 {
            all.extend(speech(3_000 + (i % 7) * 900));
            all.extend(silence(800 + (i % 3) * 400));
        }
        let mut whole = Chunker::new(chunker::Config::default());
        let mut expect = whole.push(samples_to_ms(all.len()), &all);
        expect.extend(whole.flush());
        let spans = spans_of(&all, chunker::Config::default()).unwrap();
        assert_eq!(spans.len(), expect.len());
        for (span, chunk) in spans.iter().zip(&expect) {
            assert_eq!(span.start_ms, chunk.start_ms);
            assert_eq!(span.samples, chunk.samples.len());
            let start = ms_to_samples(span.start_ms);
            assert_eq!(
                &all[start..start + span.samples],
                chunk.samples.as_slice(),
                "кусок читается с того же места"
            );
        }
    }

    #[tokio::test]
    async fn fresh_segments_come_in_time_order() {
        let mut me = Vec::new();
        let mut others = Vec::new();
        for _ in 0..12 {
            me.extend(speech(9_000));
            me.extend(silence(1_000));
            others.extend(silence(4_000));
            others.extend(speech(5_000));
            others.extend(silence(1_000));
        }
        let seen = Mutex::new(Vec::new());
        let out = run(
            vec![track(Speaker::Me, me), track(Speaker::Others, others)],
            Fake::new(0),
            Config { parallel: 4, ..cfg() },
            &CancellationToken::new(),
            |u| seen.lock().unwrap().extend(u.fresh.iter().cloned()),
        )
        .await
        .unwrap();
        let seen = seen.into_inner().unwrap();
        assert_eq!(seen.len(), out.segments.len(), "все реплики показаны по ходу");
        // Кусок уходит наружу, только когда готовы все куски раньше: начала кусков
        // (время реплики минус 200 мс подделки) идут без возвратов назад.
        let starts: Vec<u64> = seen.iter().map(|s| s.start_ms - 200).collect();
        assert!(starts.windows(2).all(|w| w[0] <= w[1]), "{starts:?}");
    }

    /// Длинная встреча: две дорожки по `MEET_LONG_MINUTES` минут (по умолчанию
    /// 120) на диске, распознавание — подделка. Проверяет, что расшифровка
    /// доходит до конца, время реплик не уезжает, а звук не читается в память
    /// целиком. Тяжёлый (пишет ~460 МБ), поэтому по запросу:
    /// `cargo test -p skillok-meet --release long_recording -- --ignored --nocapture`,
    /// пиковую память меряет `/usr/bin/time -l`.
    #[tokio::test(flavor = "multi_thread")]
    #[ignore]
    async fn long_recording() {
        use meet_audio::pcm::{WavFile, WavWriter};
        let minutes: u64 = std::env::var("MEET_LONG_MINUTES")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(120);
        let dir = std::env::temp_dir().join(format!("meet-long-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        // Реплика 8 с, пауза 1 с; у собеседников — со сдвигом на 4,5 с.
        let write = |name: &str, offset_ms: u64| {
            let path = dir.join(name);
            let mut w = WavWriter::create(&path).unwrap();
            w.write(&silence(offset_ms)).unwrap();
            let mut left = minutes * 60_000 - offset_ms;
            while left > 0 {
                let talk = left.min(8_000);
                w.write(&speech(talk)).unwrap();
                left -= talk;
                let pause = left.min(1_000);
                w.write(&silence(pause)).unwrap();
                left -= pause;
            }
            w.flush().unwrap();
            path
        };
        let me = write("me.wav", 0);
        let others = write("others.wav", 4_500);
        let started = std::time::Instant::now();
        let fake = Fake::new(0);
        let out = run(
            vec![
                (
                    Speaker::Me,
                    Arc::new(WavFile::open(&me).unwrap()) as Arc<dyn Samples>,
                ),
                (
                    Speaker::Others,
                    Arc::new(WavFile::open(&others).unwrap()) as Arc<dyn Samples>,
                ),
            ],
            fake.clone(),
            Config { parallel: 8, ..cfg() },
            &CancellationToken::new(),
            |_| {},
        )
        .await
        .unwrap();
        let calls = fake.calls.load(Ordering::SeqCst);
        let last = out.segments.last().unwrap().start_ms;
        println!(
            "{minutes} мин × 2 дорожки: {calls} кусков, {} реплик, последняя на {}, {:.1} с",
            out.segments.len(),
            crate::transcript::format_timestamp(last),
            started.elapsed().as_secs_f32()
        );
        assert_eq!(out.failed, 0);
        assert_eq!(out.segments.len(), calls, "по реплике на кусок");
        let expected = 2 * minutes * 60 / 9;
        assert!(
            calls as u64 >= expected - 4 && calls as u64 <= expected + 4,
            "{calls} кусков, ждали ~{expected}"
        );
        assert!(
            last > (minutes * 60 - 20) * 1000,
            "время не уехало: последняя реплика {last} мс"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }
}