//! SQLite-backed store for the voice-mode toggle.
//!
//! One row per `conversation_key`. Absent rows = OFF for every
//! reader, so callers never pre-seed.
//!
//! Owns its own SQLite file at `${state_dir}/voice.db` — the
//! plugin does not share a pool with any microapp. WAL mode lets
//! readers + writers overlap.
//!
//! Schema is richer than the SDK predecessor — adds per-row
//! provider override + speed + pitch + style + format + opaque
//! JSON bag so LLM `voice_set { speed: 0.9 }` calls persist
//! without a parallel side-table. Every override column is
//! nullable; `None` ⇒ fall back to the install-wide config.

use serde::{Deserialize, Serialize};
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use sqlx::SqlitePool;
use std::path::Path;
use std::str::FromStr;

use crate::error::Result;
use nexo_tool_meta::voice_addenda::DEFAULT_VOICE_ID;

/// DDL is idempotent — safe to invoke on every boot. Schema
/// migrations live in their own const for clarity; today we ship
/// v1 only.
const SCHEMA_V1: &str = r#"
CREATE TABLE IF NOT EXISTS voice_mode (
    conversation_key TEXT PRIMARY KEY,
    enabled          INTEGER NOT NULL,
    voice_id         TEXT NOT NULL,
    provider_id      TEXT NOT NULL,
    speed            REAL,
    pitch_pct        INTEGER,
    style            TEXT,
    format           TEXT,
    overrides_json   TEXT,
    updated_at_ms    INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_voice_mode_updated_at
    ON voice_mode(updated_at_ms);
"#;

/// Persisted state for a single conversation.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct VoiceModeRow {
    /// Application-defined conversation identifier. Stable string
    /// chosen by the microapp; the plugin treats it as opaque.
    pub conversation_key: String,
    /// `true` when outbound replies should be synthesised to audio.
    /// `false` (or absent row) ⇒ passthrough text reply.
    pub enabled: bool,
    /// Provider-agnostic voice id (canonical form from
    /// `voices.toml`, e.g. `"es-MX-DaliaNeural"`,
    /// `"el-Rachel-en-US"`, …). The provider knows how to
    /// translate to its own internal id when it differs.
    pub voice_id: String,
    /// Provider id (`"edge"`, `"elevenlabs"`, …). Persisted per
    /// row so a fallback synth that wins doesn't accidentally
    /// "stick" — only `voice_set` rewrites this.
    pub provider_id: String,
    /// Optional speed multiplier `0.5..=2.0`. `None` ⇒ provider
    /// default. Honored only if the provider supports tunable
    /// speed (Edge yes via SSML prosody rate, OpenAI no).
    pub speed: Option<f32>,
    /// Optional pitch percentage `-50..=50`. `None` ⇒ provider
    /// default.
    pub pitch_pct: Option<i32>,
    /// Optional provider-specific style id (ElevenLabs `style`,
    /// Cartesia voice variant, …). Opaque string.
    pub style: Option<String>,
    /// Output audio container/codec hint (`"ogg_opus"`, `"mp3"`).
    /// `None` ⇒ canonical `ogg_opus` from config.
    pub format: Option<String>,
    /// Opaque per-provider JSON overrides bag the LLM could
    /// emit via inline `[[tts:...]]` directives. Schema is
    /// provider-defined; the store just round-trips the text.
    pub overrides_json: Option<String>,
    /// Last write timestamp in epoch ms.
    pub updated_at_ms: i64,
}

/// Cloneable DAO. Clones share the underlying pool — open the
/// store once at boot, distribute clones into every handler.
#[derive(Debug, Clone)]
pub struct VoiceModeStore {
    pool: SqlitePool,
}

impl VoiceModeStore {
    /// Open (or create) the plugin's SQLite file. WAL mode +
    /// busy-timeout 5s tolerates a parallel reader. Idempotent
    /// DDL — safe to call on every boot.
    pub async fn open(db_path: &Path) -> Result<Self> {
        if let Some(parent) = db_path.parent() {
            // Parent directory missing on a fresh state dir is the
            // common case — the daemon hands us a path before any
            // file at it exists. `create_dir_all` is no-op when the
            // dir already exists, so this is safe to call always.
            tokio::fs::create_dir_all(parent).await.ok();
        }
        let opts = SqliteConnectOptions::from_str(&format!(
            "sqlite://{}?mode=rwc",
            db_path.to_string_lossy()
        ))
        .map_err(|e| crate::error::VoiceError::Other(anyhow::anyhow!(e)))?
        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
        .busy_timeout(std::time::Duration::from_secs(5));
        let pool = SqlitePoolOptions::new()
            .max_connections(8)
            .connect_with(opts)
            .await?;
        Self::open_with_pool(pool).await
    }

    /// Open on top of a caller-supplied pool. Useful for tests
    /// (in-memory DB) + the embedded build target.
    pub async fn open_with_pool(pool: SqlitePool) -> Result<Self> {
        for stmt in SCHEMA_V1.split(';').filter(|s| !s.trim().is_empty()) {
            sqlx::query(stmt).execute(&pool).await?;
        }
        Ok(Self { pool })
    }

    /// Read the current state. Returns `None` when no row exists
    /// — every reader treats absent rows as "voice mode OFF".
    pub async fn get(&self, conversation_key: &str) -> Result<Option<VoiceModeRow>> {
        let row: Option<(i64, String, String, Option<f32>, Option<i32>, Option<String>, Option<String>, Option<String>, i64)> = sqlx::query_as(
            "SELECT enabled, voice_id, provider_id, speed, pitch_pct, style, format,
                    overrides_json, updated_at_ms
             FROM voice_mode WHERE conversation_key = ?1",
        )
        .bind(conversation_key)
        .fetch_optional(&self.pool)
        .await?;
        Ok(row.map(|(enabled, voice_id, provider_id, speed, pitch_pct, style, format, overrides_json, updated_at_ms)| VoiceModeRow {
            conversation_key: conversation_key.to_string(),
            enabled: enabled != 0,
            voice_id,
            provider_id,
            speed,
            pitch_pct,
            style,
            format,
            overrides_json,
            updated_at_ms,
        }))
    }

    /// Hot-path convenience: returns the row only when one exists
    /// AND `enabled` is `true`. Used by `voice_reply_transform` to
    /// short-circuit the synth pipeline on every outbound text.
    pub async fn get_active(&self, conversation_key: &str) -> Result<Option<VoiceModeRow>> {
        Ok(self.get(conversation_key).await?.filter(|r| r.enabled))
    }

    /// Toggle voice mode for `conversation_key`. Fills omitted
    /// fields from the existing row (or the install defaults for
    /// fresh rows). Returns the row as persisted.
    #[allow(clippy::too_many_arguments)]
    pub async fn upsert(
        &self,
        conversation_key: &str,
        enabled: bool,
        voice_id: Option<&str>,
        provider_id: Option<&str>,
        speed: Option<f32>,
        pitch_pct: Option<i32>,
        style: Option<&str>,
        format: Option<&str>,
        overrides_json: Option<&str>,
    ) -> Result<VoiceModeRow> {
        let existing = self.get(conversation_key).await?;
        let resolved_voice = voice_id
            .map(str::to_string)
            .or_else(|| existing.as_ref().map(|r| r.voice_id.clone()))
            .unwrap_or_else(|| DEFAULT_VOICE_ID.to_string());
        let resolved_provider = provider_id
            .map(str::to_string)
            .or_else(|| existing.as_ref().map(|r| r.provider_id.clone()))
            .unwrap_or_else(|| "edge".to_string());
        let resolved_speed = speed.or_else(|| existing.as_ref().and_then(|r| r.speed));
        let resolved_pitch = pitch_pct.or_else(|| existing.as_ref().and_then(|r| r.pitch_pct));
        let resolved_style = style
            .map(str::to_string)
            .or_else(|| existing.as_ref().and_then(|r| r.style.clone()));
        let resolved_format = format
            .map(str::to_string)
            .or_else(|| existing.as_ref().and_then(|r| r.format.clone()));
        let resolved_overrides = overrides_json
            .map(str::to_string)
            .or_else(|| existing.as_ref().and_then(|r| r.overrides_json.clone()));
        let now_ms = chrono::Utc::now().timestamp_millis();

        sqlx::query(
            "INSERT INTO voice_mode (
                conversation_key, enabled, voice_id, provider_id,
                speed, pitch_pct, style, format, overrides_json, updated_at_ms
             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
             ON CONFLICT(conversation_key) DO UPDATE SET
                enabled        = excluded.enabled,
                voice_id       = excluded.voice_id,
                provider_id    = excluded.provider_id,
                speed          = excluded.speed,
                pitch_pct      = excluded.pitch_pct,
                style          = excluded.style,
                format         = excluded.format,
                overrides_json = excluded.overrides_json,
                updated_at_ms  = excluded.updated_at_ms",
        )
        .bind(conversation_key)
        .bind(if enabled { 1_i64 } else { 0_i64 })
        .bind(&resolved_voice)
        .bind(&resolved_provider)
        .bind(resolved_speed)
        .bind(resolved_pitch)
        .bind(&resolved_style)
        .bind(&resolved_format)
        .bind(&resolved_overrides)
        .bind(now_ms)
        .execute(&self.pool)
        .await?;

        Ok(VoiceModeRow {
            conversation_key: conversation_key.to_string(),
            enabled,
            voice_id: resolved_voice,
            provider_id: resolved_provider,
            speed: resolved_speed,
            pitch_pct: resolved_pitch,
            style: resolved_style,
            format: resolved_format,
            overrides_json: resolved_overrides,
            updated_at_ms: now_ms,
        })
    }

    /// Remove the row entirely. Equivalent to "voice mode never
    /// touched for this conversation" — every reader treats this
    /// the same as `enabled = false` since absent rows are OFF.
    /// Used by `voice_unset` when the caller wants to forget
    /// overrides too (vs `upsert(enabled=false)` which keeps the
    /// voice for next time).
    pub async fn clear(&self, conversation_key: &str) -> Result<bool> {
        let res = sqlx::query("DELETE FROM voice_mode WHERE conversation_key = ?1")
            .bind(conversation_key)
            .execute(&self.pool)
            .await?;
        Ok(res.rows_affected() > 0)
    }

    /// Enumerate enabled conversations. Drives admin RPC `stats`
    /// + audit migrations + drain timing. Order: most-recently
    /// updated first.
    pub async fn list_enabled(&self, limit: u32) -> Result<Vec<VoiceModeRow>> {
        let rows: Vec<(String, String, String, Option<f32>, Option<i32>, Option<String>, Option<String>, Option<String>, i64)> = sqlx::query_as(
            "SELECT conversation_key, voice_id, provider_id, speed, pitch_pct, style,
                    format, overrides_json, updated_at_ms
             FROM voice_mode
             WHERE enabled = 1
             ORDER BY updated_at_ms DESC
             LIMIT ?1",
        )
        .bind(limit as i64)
        .fetch_all(&self.pool)
        .await?;
        Ok(rows
            .into_iter()
            .map(|(conversation_key, voice_id, provider_id, speed, pitch_pct, style, format, overrides_json, updated_at_ms)| VoiceModeRow {
                conversation_key,
                enabled: true,
                voice_id,
                provider_id,
                speed,
                pitch_pct,
                style,
                format,
                overrides_json,
                updated_at_ms,
            })
            .collect())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use sqlx::sqlite::SqliteConnectOptions;

    async fn fresh() -> VoiceModeStore {
        let opts = SqliteConnectOptions::from_str("sqlite::memory:").unwrap();
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(opts)
            .await
            .unwrap();
        VoiceModeStore::open_with_pool(pool).await.unwrap()
    }

    #[tokio::test]
    async fn get_returns_none_for_unknown_conversation() {
        let s = fresh().await;
        assert!(s.get("ana:session:zzz").await.unwrap().is_none());
        assert!(s.get_active("ana:session:zzz").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn upsert_then_get_round_trips_full_row() {
        let s = fresh().await;
        let row = s
            .upsert(
                "ana:session:1",
                true,
                Some("es-CO-SalomeNeural"),
                Some("elevenlabs"),
                Some(1.1),
                Some(5),
                Some("conversational"),
                Some("ogg_opus"),
                Some(r#"{"stability":0.5}"#),
            )
            .await
            .unwrap();
        assert!(row.enabled);
        assert_eq!(row.voice_id, "es-CO-SalomeNeural");
        assert_eq!(row.provider_id, "elevenlabs");
        assert_eq!(row.speed, Some(1.1));
        assert_eq!(row.pitch_pct, Some(5));
        assert_eq!(row.style.as_deref(), Some("conversational"));
        assert_eq!(row.format.as_deref(), Some("ogg_opus"));
        assert_eq!(row.overrides_json.as_deref(), Some(r#"{"stability":0.5}"#));

        let back = s.get("ana:session:1").await.unwrap().unwrap();
        assert_eq!(row, back);
    }

    #[tokio::test]
    async fn upsert_omitted_fields_carry_over() {
        let s = fresh().await;
        s.upsert(
            "k",
            true,
            Some("es-CO-SalomeNeural"),
            Some("edge"),
            Some(0.9),
            None,
            None,
            None,
            None,
        )
        .await
        .unwrap();
        // Second upsert toggles off but supplies NONE everywhere
        // else. Existing voice/provider/speed must persist.
        let row = s
            .upsert("k", false, None, None, None, None, None, None, None)
            .await
            .unwrap();
        assert!(!row.enabled);
        assert_eq!(row.voice_id, "es-CO-SalomeNeural");
        assert_eq!(row.provider_id, "edge");
        assert_eq!(row.speed, Some(0.9));
    }

    #[tokio::test]
    async fn upsert_fresh_row_uses_install_defaults() {
        let s = fresh().await;
        let row = s
            .upsert("k", true, None, None, None, None, None, None, None)
            .await
            .unwrap();
        assert_eq!(row.voice_id, DEFAULT_VOICE_ID);
        assert_eq!(row.provider_id, "edge");
        assert_eq!(row.speed, None);
    }

    #[tokio::test]
    async fn get_active_short_circuits_when_disabled() {
        let s = fresh().await;
        s.upsert(
            "k",
            false,
            Some("v"),
            Some("edge"),
            None,
            None,
            None,
            None,
            None,
        )
        .await
        .unwrap();
        assert!(s.get_active("k").await.unwrap().is_none());

        s.upsert("k", true, None, None, None, None, None, None, None)
            .await
            .unwrap();
        assert!(s.get_active("k").await.unwrap().is_some());
    }

    #[tokio::test]
    async fn clear_removes_row() {
        let s = fresh().await;
        s.upsert(
            "k",
            true,
            Some("v"),
            Some("edge"),
            None,
            None,
            None,
            None,
            None,
        )
        .await
        .unwrap();
        assert!(s.clear("k").await.unwrap());
        assert!(s.get("k").await.unwrap().is_none());
        // Idempotent: second clear returns false.
        assert!(!s.clear("k").await.unwrap());
    }

    #[tokio::test]
    async fn list_enabled_orders_by_recency_and_filters_disabled() {
        let s = fresh().await;
        // Two enabled, one disabled. Insert in a way that the
        // disabled row has the freshest updated_at to confirm
        // it's filtered, not just last.
        s.upsert(
            "a",
            true,
            Some("va"),
            Some("edge"),
            None,
            None,
            None,
            None,
            None,
        )
        .await
        .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(2)).await;
        s.upsert(
            "b",
            true,
            Some("vb"),
            Some("edge"),
            None,
            None,
            None,
            None,
            None,
        )
        .await
        .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(2)).await;
        s.upsert(
            "c",
            false,
            Some("vc"),
            Some("edge"),
            None,
            None,
            None,
            None,
            None,
        )
        .await
        .unwrap();

        let enabled = s.list_enabled(10).await.unwrap();
        assert_eq!(enabled.len(), 2);
        // Most-recently updated of the enabled rows first.
        assert_eq!(enabled[0].conversation_key, "b");
        assert_eq!(enabled[1].conversation_key, "a");
    }
}
