//! Запись одной встречи: дорожки звука → куски по паузам → распознавание →
//! реплики в расшифровке, и раз в несколько минут — живой конспект.
//!
//! ```text
//! поток «я»          ─┐                         ┌─ store.append_segments
//! (микрофон)          ├─ куски ─► распознавание ─┤
//! поток «собеседники» ┘  (канал)   (одна задача) └─ живой конспект
//! (звук Zoom)
//! ```
//!
//! Каждая дорожка читается своим потоком: захват блокирующий, а медленная
//! сеть не должна терять звук. Распознавание — одна асинхронная задача, куски
//! идут по очереди: так реплики одной дорожки не обгоняют друг друга.
//! Звук каждой дорожки пишется на диск сразу — после сбоя его можно
//! распознать заново.
//!
//! Модуль не знает про Tauri и OpenRouter: распознавание — через
//! [`Transcriber`], события — через колбэк. Так его проверяют тесты.

use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use meet_audio::chunker::{self, Chunk, Chunker};
use meet_audio::pcm::{self, WavWriter};
use meet_audio::source::{Capture, CaptureError};
use tokio::sync::mpsc;

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

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

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

#[derive(Debug, Clone, PartialEq, serde::Serialize)]
#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
pub enum Event {
    /// Новые реплики уже в расшифровке.
    Segments(Vec<Segment>),
    /// Уровень дорожки, dBFS, несколько раз в секунду.
    Level {
        speaker: Speaker,
        dbfs: f32,
    },
    LiveSummary(String),
    /// Дорожка перестала писаться; остальные продолжают.
    TrackFailed {
        speaker: Speaker,
        error: String,
    },
    /// Кусок не распознался и после повтора; звук остался на диске.
    SttFailed(String),
}

pub struct Track {
    pub speaker: Speaker,
    pub capture: Box<dyn Capture>,
}

#[derive(Debug, Clone, Copy)]
pub struct Config {
    pub chunker: chunker::Config,
    /// Как часто обновлять живой конспект.
    pub live_every: Duration,
    /// Пауза перед повтором неудачного распознавания.
    pub retry_after: Duration,
}

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

type Emit = Arc<dyn Fn(Event) + Send + Sync>;

pub struct Session {
    pub meeting_id: String,
    stop: Arc<AtomicBool>,
    paused: Arc<AtomicBool>,
    threads: Vec<JoinHandle<u64>>,
    worker: tokio::task::JoinHandle<()>,
}

impl Session {
    pub fn start(
        meeting_id: String,
        store: Arc<Store>,
        tracks: Vec<Track>,
        ai: Arc<dyn Transcriber>,
        emit: Emit,
        cfg: Config,
        rt: &tokio::runtime::Handle,
    ) -> Self {
        let stop = Arc::new(AtomicBool::new(false));
        let paused = Arc::new(AtomicBool::new(false));
        let started = Instant::now();
        let (tx, rx) = mpsc::unbounded_channel();
        let threads = tracks
            .into_iter()
            .map(|track| {
                let wav_path = store.audio_path(&meeting_id, &track.speaker).ok();
                let ctx = TrackCtx {
                    stop: stop.clone(),
                    paused: paused.clone(),
                    started,
                    tx: tx.clone(),
                    emit: emit.clone(),
                    chunker: cfg.chunker,
                };
                std::thread::Builder::new()
                    .name(format!("meet-track-{}", track.speaker.label()))
                    .spawn(move || run_track(track, wav_path, ctx))
                    .expect("поток дорожки")
            })
            .collect();
        drop(tx);
        let worker = rt.spawn(run_worker(meeting_id.clone(), store, rx, ai, emit, cfg));
        Self {
            meeting_id,
            stop,
            paused,
            threads,
            worker,
        }
    }

    pub fn set_paused(&self, paused: bool) {
        self.paused.store(paused, Ordering::Relaxed);
    }

    pub fn is_paused(&self) -> bool {
        self.paused.load(Ordering::Relaxed)
    }

    /// Останавливает дорожки, дожидается распознавания всех кусков и
    /// возвращает длительность записи, мс.
    pub async fn stop(self) -> u64 {
        self.stop.store(true, Ordering::Relaxed);
        let threads = self.threads;
        let duration = tokio::task::spawn_blocking(move || {
            threads
                .into_iter()
                .filter_map(|t| t.join().ok())
                .max()
                .unwrap_or(0)
        })
        .await
        .unwrap_or(0);
        let _ = self.worker.await;
        duration
    }
}

struct TrackCtx {
    stop: Arc<AtomicBool>,
    paused: Arc<AtomicBool>,
    started: Instant,
    tx: mpsc::UnboundedSender<(Speaker, Chunk)>,
    emit: Emit,
    chunker: chunker::Config,
}

/// Поток одной дорожки. Возвращает позицию конца звука, мс.
fn run_track(mut track: Track, wav_path: Option<std::path::PathBuf>, ctx: TrackCtx) -> u64 {
    let mut chunker = Chunker::new(ctx.chunker);
    let mut wav = wav_path.and_then(|p| WavWriter::create(&p).ok());
    let mut level_buf: Vec<i16> = Vec::new();
    let mut last_level = Instant::now();
    let mut last_flush = Instant::now();
    let speaker = track.speaker.clone();
    let send = |chunks: Vec<Chunk>| {
        for c in chunks {
            let _ = ctx.tx.send((speaker.clone(), c));
        }
    };
    while !ctx.stop.load(Ordering::Relaxed) {
        let samples = match track.capture.read() {
            Ok(s) => s,
            Err(CaptureError::Ended) => break,
            Err(e) => {
                (ctx.emit)(Event::TrackFailed {
                    speaker: speaker.clone(),
                    error: e.to_string(),
                });
                break;
            }
        };
        if samples.is_empty() || ctx.paused.load(Ordering::Relaxed) {
            continue;
        }
        let now_ms = ctx.started.elapsed().as_millis() as u64;
        if let Some(w) = wav.as_mut() {
            let _ = w.write(&samples);
            if last_flush.elapsed() >= Duration::from_secs(5) {
                let _ = w.flush();
                last_flush = Instant::now();
            }
        }
        level_buf.extend_from_slice(&samples);
        if last_level.elapsed() >= Duration::from_millis(200) {
            (ctx.emit)(Event::Level {
                speaker: speaker.clone(),
                dbfs: pcm::level_dbfs(&level_buf),
            });
            level_buf.clear();
            last_level = Instant::now();
        }
        send(chunker.push(now_ms, &samples));
    }
    send(chunker.flush());
    (ctx.emit)(Event::Level {
        speaker: speaker.clone(),
        dbfs: -90.0,
    });
    chunker.position_ms()
}

async fn run_worker(
    meeting_id: String,
    store: Arc<Store>,
    mut rx: mpsc::UnboundedReceiver<(Speaker, Chunk)>,
    ai: Arc<dyn Transcriber>,
    emit: Emit,
    cfg: Config,
) {
    let mut last_summary = Instant::now();
    let mut new_since_summary = 0usize;
    while let Some((speaker, chunk)) = rx.recv().await {
        let wav = pcm::encode_wav(&chunk.samples);
        let mut result = ai.transcribe(wav.clone()).await;
        if result.is_err() {
            tokio::time::sleep(cfg.retry_after).await;
            result = ai.transcribe(wav).await;
        }
        let segments: Vec<Segment> = match result {
            Ok(list) => list
                .into_iter()
                .map(|s| Segment {
                    start_ms: chunk.start_ms + s.start_ms,
                    speaker: speaker.clone(),
                    text: s.text,
                })
                .collect(),
            Err(e) => {
                emit(Event::SttFailed(e));
                continue;
            }
        };
        if segments.is_empty() {
            continue;
        }
        if store.append_segments(&meeting_id, &segments).is_ok() {
            new_since_summary += segments.len();
            emit(Event::Segments(segments));
        }
        if new_since_summary > 0 && last_summary.elapsed() >= cfg.live_every {
            if let Ok(detail) = store.get(&meeting_id) {
                if let Ok(text) = ai.live_summary(detail.transcript).await {
                    let _ = store.save_live_summary(&meeting_id, &text);
                    emit(Event::LiveSummary(text));
                }
            }
            last_summary = Instant::now();
            new_since_summary = 0;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Local;
    use meet_audio::pcm::{ms_to_samples, SAMPLE_RATE};
    use meet_audio::source::Replay;
    use std::sync::Mutex;

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

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

    /// Подделка: на каждый кусок — одна реплика «кусок N (длина мс)».
    struct Fake {
        calls: Mutex<usize>,
        fail_first: bool,
    }

    impl Transcriber for Fake {
        fn transcribe(&self, wav: Vec<u8>) -> BoxFuture<'_, Result<Vec<SttSegment>, String>> {
            Box::pin(async move {
                let mut n = self.calls.lock().unwrap();
                *n += 1;
                if self.fail_first && *n == 1 {
                    return Err("сеть".into());
                }
                let ms = pcm::samples_to_ms(pcm::decode_wav(&wav).unwrap().len());
                Ok(vec![SttSegment {
                    start_ms: 100,
                    text: format!("кусок {n} ({ms} мс)"),
                }])
            })
        }
        fn live_summary(&self, t: Transcript) -> BoxFuture<'_, Result<String, String>> {
            Box::pin(async move { Ok(format!("реплик: {}", t.segments.len())) })
        }
    }

    fn fast() -> Config {
        Config {
            live_every: Duration::ZERO,
            retry_after: Duration::from_millis(10),
            ..Config::default()
        }
    }

    async fn record(tracks: Vec<Track>, fail_first: bool) -> (Arc<Store>, String, Vec<Event>, u64) {
        let dir = tempfile::tempdir().unwrap().keep();
        let store = Arc::new(Store::new(dir));
        let meeting = store.create_recording("Тест", Local::now()).unwrap();
        let events = Arc::new(Mutex::new(Vec::new()));
        let sink = events.clone();
        let session = Session::start(
            meeting.id.clone(),
            store.clone(),
            tracks,
            Arc::new(Fake {
                calls: Mutex::new(0),
                fail_first,
            }),
            Arc::new(move |e| sink.lock().unwrap().push(e)),
            fast(),
            &tokio::runtime::Handle::current(),
        );
        // Ждём, пока дорожки дочитают звук (Replay заканчивается сам).
        tokio::time::sleep(Duration::from_millis(1500)).await;
        let duration = session.stop().await;
        let events = events.lock().unwrap().clone();
        (store, meeting.id, events, duration)
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn two_tracks_become_one_ordered_transcript() {
        // «Я»: 7 с речи, пауза, 3 с речи. «Собеседники»: 2 с тишины, 8 с речи.
        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 tracks = vec![
            Track {
                speaker: Speaker::Me,
                capture: Box::new(Replay::new(me, 50.0)),
            },
            Track {
                speaker: Speaker::Others,
                capture: Box::new(Replay::new(others, 50.0)),
            },
        ];
        let (store, id, events, duration) = record(tracks, false).await;

        let t = store.get(&id).unwrap().transcript;
        assert_eq!(t.segments.len(), 3, "{:#?}", t.segments);
        let starts: Vec<u64> = t.segments.iter().map(|s| s.start_ms).collect();
        let mut sorted = starts.clone();
        sorted.sort();
        assert_eq!(starts, sorted, "расшифровка упорядочена по времени");
        assert!(t.segments.iter().any(|s| s.speaker == Speaker::Me));
        assert!(t.segments.iter().any(|s| s.speaker == Speaker::Others));
        assert!(duration >= 10_000, "длительность по звуку: {duration}");

        assert!(events.iter().any(|e| matches!(e, Event::Segments(_))));
        assert!(events.iter().any(|e| matches!(e, Event::Level { .. })));
        assert!(events.iter().any(|e| matches!(e, Event::LiveSummary(_))));
        assert!(store.get(&id).unwrap().live_summary.is_some());
        // Дорожки лежат на диске целиком.
        let me_wav = std::fs::read(store.audio_path(&id, &Speaker::Me).unwrap()).unwrap();
        assert_eq!(pcm::decode_wav(&me_wav).unwrap().len(), ms_to_samples(11_500));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn failed_request_is_retried_once() {
        let tracks = vec![Track {
            speaker: Speaker::Me,
            capture: Box::new(Replay::new(speech(3_000), 50.0)),
        }];
        let (store, id, events, _) = record(tracks, true).await;
        assert_eq!(store.get(&id).unwrap().transcript.segments.len(), 1);
        assert!(!events.iter().any(|e| matches!(e, Event::SttFailed(_))));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn silent_recording_sends_nothing() {
        let tracks = vec![Track {
            speaker: Speaker::Others,
            capture: Box::new(Replay::new(silence(20_000), 100.0)),
        }];
        let (store, id, events, _) = record(tracks, false).await;
        assert!(store.get(&id).unwrap().transcript.segments.is_empty());
        assert!(!events.iter().any(|e| matches!(e, Event::Segments(_))));
    }

    struct Broken;
    impl Capture for Broken {
        fn read(&mut self) -> Result<Vec<i16>, CaptureError> {
            Err(CaptureError::Loopback("устройство пропало".into()))
        }
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn broken_track_is_reported_and_the_other_keeps_going() {
        let tracks = vec![
            Track {
                speaker: Speaker::Others,
                capture: Box::new(Broken),
            },
            Track {
                speaker: Speaker::Me,
                capture: Box::new(Replay::new(speech(3_000), 50.0)),
            },
        ];
        let (store, id, events, _) = record(tracks, false).await;
        assert!(events.iter().any(|e| matches!(
            e,
            Event::TrackFailed {
                speaker: Speaker::Others,
                ..
            }
        )));
        assert_eq!(store.get(&id).unwrap().transcript.segments.len(), 1);
    }
}