//! Threaded list assembly: the paginated thread view and the per-thread fetch.
//!
//! `list_threaded` is four queries in sequence (count, ranked thread summary,
//! the page's full emails, then the assembly pass), which is why it has a
//! module to itself.

use std::collections::HashMap;

use goingson_core::{CoreError, Email, EmailThread, Result, UserId};
use rusqlite::{Connection, params, params_from_iter};

use crate::utils::{bind_placeholders, query_all};

use super::row::{EMAIL_SELECT_COLUMNS, EmailRow};

/// One page of threads, newest first, with the total thread count.
pub(super) fn list_threaded(
    conn: &Connection,
    user_id: UserId,
    include_archived: bool,
    offset: Option<i64>,
    limit: Option<i64>,
    folder: Option<&str>,
    label: Option<&str>,
) -> Result<(Vec<EmailThread>, i64)> {
    let uid = user_id.to_string();
    let archived_filter = if include_archived {
        ""
    } else {
        "AND e.is_archived = 0"
    };
    let folder_filter = folder.map_or("", |_| "AND e.source_folder = ?");
    let label_filter = label.map_or(
        "",
        |_| "AND EXISTS (SELECT 1 FROM json_each(e.labels) j WHERE j.value = ?)",
    );
    // Defense-in-depth: clamp before binding. A negative LIMIT means
    // unbounded in SQLite (would load the whole mailbox); a negative OFFSET
    // is ignored. Matches task_repo/search_repo.
    const MAX_PAGE_LIMIT: i64 = 1000;
    let offset_val = offset.unwrap_or(0).max(0);
    let limit_val = limit.unwrap_or(50).clamp(0, MAX_PAGE_LIMIT);

    // Query 1: Get total thread count
    let count_sql = format!(
        "SELECT COUNT(DISTINCT COALESCE(e.thread_id, e.id)) FROM emails e WHERE e.user_id = ? AND e.is_draft = 0 {archived_filter} {folder_filter} {label_filter}"
    );
    // The optional folder/label filters add a placeholder each, so the binds are
    // assembled in the same order the filters were spliced into the SQL.
    let mut filter_binds: Vec<String> = vec![uid.clone()];
    if let Some(f) = folder {
        filter_binds.push(f.to_string());
    }
    if let Some(l) = label {
        filter_binds.push(l.to_string());
    }

    let total: i64 = conn
        .query_row(&count_sql, params_from_iter(&filter_binds), |row| {
            row.get(0)
        })
        .map_err(CoreError::database)?;

    if total == 0 {
        return Ok((vec![], 0));
    }

    // Query 2: Thread summary, group by thread, get latest received_at, count, unread status
    #[allow(dead_code)]
    struct ThreadSummary {
        thread_key: String,
        latest_received_at: String, // needed for SQL ORDER BY
        thread_count: i64,
        unread_count: i64,
        latest_email_id: String,
    }

    // Rank emails within each thread by recency in a single pass with window
    // functions, then keep the latest row per thread. This replaces a
    // correlated subquery that rescanned `emails` once per thread group
    // (O(threads x emails)); the partition's MAX(received_at) is the rn = 1
    // row, and the thread-wide counts come from window aggregates.
    let summary_sql = format!(
        r"WITH ranked AS (
                SELECT
                    e.id AS email_id,
                    COALESCE(e.thread_id, e.id) AS thread_key,
                    e.received_at AS received_at,
                    ROW_NUMBER() OVER (
                        PARTITION BY COALESCE(e.thread_id, e.id)
                        ORDER BY e.received_at DESC, e.id DESC
                    ) AS rn,
                    COUNT(*) OVER (PARTITION BY COALESCE(e.thread_id, e.id)) AS thread_count,
                    SUM(CASE WHEN e.is_read = 0 THEN 1 ELSE 0 END)
                        OVER (PARTITION BY COALESCE(e.thread_id, e.id)) AS unread_count
                FROM emails e
                WHERE e.user_id = ? AND e.is_draft = 0 {archived_filter} {folder_filter} {label_filter}
            )
            SELECT
                thread_key,
                received_at AS latest_received_at,
                thread_count,
                unread_count,
                email_id AS latest_email_id
            FROM ranked
            WHERE rn = 1
            ORDER BY latest_received_at DESC
            LIMIT ? OFFSET ?",
    );

    let mut summary_binds: Vec<rusqlite::types::Value> = filter_binds
        .iter()
        .map(|s| rusqlite::types::Value::from(s.clone()))
        .collect();
    summary_binds.push(limit_val.into());
    summary_binds.push(offset_val.into());

    let summaries = query_all(conn, &summary_sql, params_from_iter(summary_binds), |row| {
        Ok(ThreadSummary {
            thread_key: row.get("thread_key")?,
            latest_received_at: row.get("latest_received_at")?,
            thread_count: row.get("thread_count")?,
            unread_count: row.get("unread_count")?,
            latest_email_id: row.get("latest_email_id")?,
        })
    })?;

    if summaries.is_empty() {
        return Ok((vec![], total));
    }

    // Query 3: Fetch full emails for the page's most-recent-email IDs
    let email_ids: Vec<String> = summaries
        .iter()
        .map(|s| s.latest_email_id.clone())
        .collect();
    let placeholders = bind_placeholders(email_ids.len());
    let emails_sql = format!(
        "SELECT {EMAIL_SELECT_COLUMNS} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.id IN ({placeholders}) AND e.user_id = ?"
    );

    let mut email_binds: Vec<String> = Vec::with_capacity(email_ids.len() + 2);
    email_binds.push(uid.clone());
    email_binds.extend(email_ids.iter().cloned());
    email_binds.push(uid.clone());

    let rows = query_all(
        conn,
        &emails_sql,
        params_from_iter(email_binds),
        EmailRow::from_row,
    )?;

    let email_map: HashMap<String, Email> = rows
        .into_iter()
        .filter_map(|row| {
            let id_str = row.id.clone();
            Email::try_from(row).ok().map(|e| (id_str, e))
        })
        .collect();

    // Assemble threads in summary order
    let threads: Vec<EmailThread> = summaries
        .into_iter()
        .filter_map(|s| {
            let email = email_map.get(&s.latest_email_id)?.clone();
            Some(EmailThread {
                thread_id: s.thread_key,
                most_recent_email: email,
                thread_count: s.thread_count as usize,
                has_unread: s.unread_count > 0,
            })
        })
        .collect();

    Ok((threads, total))
}

/// Every email in one thread, oldest first.
pub(super) fn list_by_thread(
    conn: &Connection,
    user_id: UserId,
    thread_id: &str,
) -> Result<Vec<Email>> {
    let query = format!(
        "SELECT {EMAIL_SELECT_COLUMNS} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.thread_id = ? ORDER BY e.received_at ASC"
    );
    let rows = query_all(
        conn,
        &query,
        params![user_id.to_string(), user_id.to_string(), thread_id],
        EmailRow::from_row,
    )?;
    rows.into_iter().map(Email::try_from).collect()
}