//! SQLite implementation of the EmailRepository.
//!
//! Manages email messages with support for:
//! - IMAP synchronization tracking (message_id, imap_uid)
//! - Threading via in_reply_to and thread_id
//! - Read/archived status
//! - Project associations
//! - Snoozing and waiting-for-response tracking

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use sqlx::SqlitePool;
use std::collections::HashSet;
use goingson_core::{
    CoreError, Email, EmailAccountId, EmailId, EmailRepository, EmailThread, NewEmail,
    NewEmailWithTracking, ProjectId, Result, UserId,
};
use std::collections::HashMap;

use crate::utils::{format_datetime, format_datetime_now, format_datetime_opt, parse_datetime, parse_uuid, parse_uuid_opt};

/// Column list for SELECT queries - avoids duplication across methods.
const EMAIL_SELECT_COLUMNS: &str = r#"e.id, e.project_id, p.name as project_name, e.from_address, e.to_address,
    e.subject, e.body, e.html_body, e.is_read, e.is_archived, e.received_at, e.message_id,
    e.in_reply_to, e.thread_id, e.email_account_id, e.is_outgoing, e.imap_uid, e.source_folder,
    e.attachment_meta, e.labels, e.is_draft, e.cc_address, e.bcc_address, e.draft_account_id,
    e.snoozed_until, e.waiting_for_response, e.waiting_since, e.expected_response_date"#;

#[derive(Debug, Clone, sqlx::FromRow)]
struct EmailRow {
    pub id: String,
    pub project_id: Option<String>,
    pub project_name: Option<String>,
    pub from_address: String,
    pub to_address: String,
    pub subject: String,
    pub body: String,
    pub html_body: Option<String>,
    pub is_read: i32,
    pub is_archived: i32,
    pub received_at: String,
    pub message_id: Option<String>,
    pub in_reply_to: Option<String>,
    pub thread_id: Option<String>,
    pub email_account_id: Option<String>,
    pub is_outgoing: i32,
    pub imap_uid: Option<i64>,
    pub source_folder: Option<String>,
    pub attachment_meta: Option<String>,
    pub labels: String,
    pub is_draft: i32,
    pub cc_address: Option<String>,
    pub bcc_address: Option<String>,
    pub draft_account_id: Option<String>,
    pub snoozed_until: Option<String>,
    pub waiting_for_response: i32,
    pub waiting_since: Option<String>,
    pub expected_response_date: Option<String>,
}

impl TryFrom<EmailRow> for Email {
    type Error = CoreError;

    fn try_from(row: EmailRow) -> std::result::Result<Self, Self::Error> {
        Ok(Email {
            id: parse_uuid(&row.id)?.into(),
            project_id: parse_uuid_opt(row.project_id.as_deref())?.map(Into::into),
            project_name: row.project_name,
            from: row.from_address,
            to: row.to_address,
            subject: row.subject,
            body: row.body,
            html_body: row.html_body,
            is_read: row.is_read != 0,
            is_archived: row.is_archived != 0,
            received_at: parse_datetime(&row.received_at)?,
            message_id: row.message_id,
            in_reply_to: row.in_reply_to,
            thread_id: row.thread_id,
            email_account_id: parse_uuid_opt(row.email_account_id.as_deref())?.map(Into::into),
            is_outgoing: row.is_outgoing != 0,
            imap_uid: row.imap_uid,
            source_folder: row.source_folder,
            attachment_meta: row.attachment_meta,
            labels: serde_json::from_str(&row.labels).unwrap_or_default(),
            is_draft: row.is_draft != 0,
            cc_address: row.cc_address,
            bcc_address: row.bcc_address,
            draft_account_id: parse_uuid_opt(row.draft_account_id.as_deref())?.map(Into::into),
            snoozed_until: row.snoozed_until.as_ref().map(|s| parse_datetime(s)).transpose()?,
            waiting_for_response: row.waiting_for_response != 0,
            waiting_since: row.waiting_since.as_ref().map(|s| parse_datetime(s)).transpose()?,
            expected_response_date: row.expected_response_date.as_ref().map(|s| parse_datetime(s)).transpose()?,
        })
    }
}

/// SQLite-backed implementation of [`EmailRepository`].
///
/// Manages email messages with threading support, snoozing, and
/// waiting-for-response tracking. Integrates with IMAP sync via message_id.
pub struct SqliteEmailRepository { pool: SqlitePool }

impl SqliteEmailRepository {
    /// Creates a new repository instance with the given connection pool.
    #[tracing::instrument(skip_all)]
    pub fn new(pool: SqlitePool) -> Self { Self { pool } }
}

#[async_trait]
impl EmailRepository for SqliteEmailRepository {
    #[tracing::instrument(skip_all)]
    async fn list_all(&self, user_id: UserId, include_archived: bool) -> Result<Vec<Email>> {
        let archived_filter = if include_archived { "" } else { "AND e.is_archived = 0" };
        let query = format!(
            "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.is_draft = 0 {} ORDER BY e.received_at DESC",
            EMAIL_SELECT_COLUMNS, archived_filter
        );
        let rows = sqlx::query_as::<_, EmailRow>(&query).bind(user_id.to_string()).bind(user_id.to_string()).fetch_all(&self.pool).await.map_err(CoreError::database)?;
        rows.into_iter().map(Email::try_from).collect()
    }

    #[tracing::instrument(skip_all)]
    async fn list_threaded(&self, 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(|_| "AND e.source_folder = ?").unwrap_or("");
        let label_filter = label.map(|_| "AND EXISTS (SELECT 1 FROM json_each(e.labels) j WHERE j.value = ?)").unwrap_or("");
        let offset_val = offset.unwrap_or(0);
        let limit_val = limit.unwrap_or(50);

        // 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
        );
        let mut count_q = sqlx::query_as::<_, (i64,)>(&count_sql).bind(&uid);
        if let Some(f) = folder { count_q = count_q.bind(f); }
        if let Some(l) = label { count_q = count_q.bind(l); }
        let (total,) = count_q.fetch_one(&self.pool).await.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
        #[derive(sqlx::FromRow)]
        #[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,
        }

        let summary_sql = format!(
            r#"SELECT
                COALESCE(e.thread_id, e.id) AS thread_key,
                MAX(e.received_at) AS latest_received_at,
                COUNT(*) AS thread_count,
                SUM(CASE WHEN e.is_read = 0 THEN 1 ELSE 0 END) AS unread_count,
                (SELECT e2.id FROM emails e2
                 WHERE COALESCE(e2.thread_id, e2.id) = COALESCE(e.thread_id, e.id)
                   AND e2.user_id = ? AND e2.is_draft = 0 {} {} {}
                 ORDER BY e2.received_at DESC LIMIT 1) AS latest_email_id
            FROM emails e
            WHERE e.user_id = ? AND e.is_draft = 0 {} {} {}
            GROUP BY COALESCE(e.thread_id, e.id)
            ORDER BY latest_received_at DESC
            LIMIT ? OFFSET ?"#,
            archived_filter, folder_filter, label_filter,
            archived_filter, folder_filter, label_filter,
        );

        let mut summary_q = sqlx::query_as::<_, ThreadSummary>(&summary_sql).bind(&uid);
        if let Some(f) = folder { summary_q = summary_q.bind(f); }
        if let Some(l) = label { summary_q = summary_q.bind(l); }
        summary_q = summary_q.bind(&uid);
        if let Some(f) = folder { summary_q = summary_q.bind(f); }
        if let Some(l) = label { summary_q = summary_q.bind(l); }
        let summaries = summary_q
            .bind(limit_val)
            .bind(offset_val)
            .fetch_all(&self.pool)
            .await
            .map_err(CoreError::database)?;

        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 = email_ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
        let emails_sql = format!(
            "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.id IN ({}) AND e.user_id = ?",
            EMAIL_SELECT_COLUMNS, placeholders
        );

        let mut q = sqlx::query_as::<_, EmailRow>(&emails_sql).bind(&uid);
        for id in &email_ids {
            q = q.bind(id);
        }
        q = q.bind(&uid);

        let rows = q.fetch_all(&self.pool).await.map_err(CoreError::database)?;

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

    #[tracing::instrument(skip_all)]
    async fn list_by_project(&self, user_id: UserId, project_id: ProjectId) -> Result<Vec<Email>> {
        let query = format!(
            "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.project_id = ? ORDER BY e.received_at DESC",
            EMAIL_SELECT_COLUMNS
        );
        let rows = sqlx::query_as::<_, EmailRow>(&query).bind(user_id.to_string()).bind(user_id.to_string()).bind(project_id.to_string()).fetch_all(&self.pool).await.map_err(CoreError::database)?;
        rows.into_iter().map(Email::try_from).collect()
    }

    #[tracing::instrument(skip_all)]
    async fn list_by_addresses(&self, user_id: UserId, addresses: &[&str]) -> Result<Vec<Email>> {
        if addresses.is_empty() {
            return Ok(Vec::new());
        }
        let placeholders = addresses.iter().map(|_| "?").collect::<Vec<_>>().join(",");
        let query = format!(
            "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND (LOWER(e.from_address) IN ({placeholders}) OR LOWER(e.to_address) IN ({placeholders})) ORDER BY e.received_at DESC LIMIT 200",
            EMAIL_SELECT_COLUMNS
        );
        let mut q = sqlx::query_as::<_, EmailRow>(&query)
            .bind(user_id.to_string())
            .bind(user_id.to_string());
        // Bind addresses twice (once for from_address IN, once for to_address IN)
        for _ in 0..2 {
            for addr in addresses {
                q = q.bind(addr.to_lowercase());
            }
        }
        let rows = q.fetch_all(&self.pool).await.map_err(CoreError::database)?;
        rows.into_iter().map(Email::try_from).collect()
    }

    #[tracing::instrument(skip_all)]
    async fn list_unlinked(&self, user_id: UserId) -> Result<Vec<Email>> {
        let query = format!(
            "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.project_id IS NULL AND e.is_archived = 0 ORDER BY e.received_at DESC",
            EMAIL_SELECT_COLUMNS
        );
        let rows = sqlx::query_as::<_, EmailRow>(&query).bind(user_id.to_string()).bind(user_id.to_string()).fetch_all(&self.pool).await.map_err(CoreError::database)?;
        rows.into_iter().map(Email::try_from).collect()
    }

    #[tracing::instrument(skip_all)]
    async fn get_by_id(&self, id: EmailId, user_id: UserId) -> Result<Option<Email>> {
        let query = format!(
            "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.id = ? AND e.user_id = ?",
            EMAIL_SELECT_COLUMNS
        );
        let row = sqlx::query_as::<_, EmailRow>(&query).bind(user_id.to_string()).bind(id.to_string()).bind(user_id.to_string()).fetch_optional(&self.pool).await.map_err(CoreError::database)?;
        row.map(Email::try_from).transpose()
    }

    #[tracing::instrument(skip_all)]
    async fn create(&self, user_id: UserId, email: NewEmail) -> Result<Email> {
        let id = EmailId::new();
        let received_at = format_datetime(&email.received_at.unwrap_or_else(Utc::now));
        sqlx::query("INSERT INTO emails (id, user_id, project_id, from_address, to_address, subject, body, is_read, received_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)")
            .bind(id.to_string()).bind(user_id.to_string()).bind(email.project_id.map(|p| p.to_string()))
            .bind(&email.from_address).bind(&email.to_address).bind(&email.subject).bind(&email.body)
            .bind(if email.is_read { 1 } else { 0 }).bind(&received_at)
            .execute(&self.pool).await.map_err(CoreError::database)?;
        self.get_by_id(id, user_id).await?.ok_or_else(|| CoreError::internal("Failed to retrieve created email"))
    }

    #[tracing::instrument(skip_all)]
    async fn create_with_tracking(&self, user_id: UserId, email: NewEmailWithTracking) -> Result<Email> {
        let id = goingson_core::deterministic_email_id(email.message_id.as_deref());
        let received_at = format_datetime(&email.received_at.unwrap_or_else(Utc::now));
        sqlx::query("INSERT INTO emails (id, user_id, project_id, from_address, to_address, subject, body, html_body, is_read, is_archived, received_at, message_id, in_reply_to, thread_id, email_account_id, is_outgoing, imap_uid, source_folder, attachment_meta) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
            .bind(id.to_string()).bind(user_id.to_string()).bind(email.project_id.map(|p| p.to_string()))
            .bind(&email.from_address).bind(&email.to_address).bind(&email.subject).bind(&email.body).bind(&email.html_body)
            .bind(if email.is_read { 1 } else { 0 }).bind(if email.is_archived { 1 } else { 0 }).bind(&received_at)
            .bind(&email.message_id).bind(&email.in_reply_to).bind(&email.thread_id)
            .bind(email.email_account_id.map(|a| a.to_string()))
            .bind(if email.is_outgoing { 1 } else { 0 }).bind(email.imap_uid).bind(&email.source_folder)
            .bind(&email.attachment_meta)
            .execute(&self.pool).await.map_err(CoreError::database)?;
        self.get_by_id(id, user_id).await?.ok_or_else(|| CoreError::internal("Failed to retrieve created email"))
    }

    #[tracing::instrument(skip_all)]
    async fn create_with_tracking_batch(&self, user_id: UserId, emails: Vec<NewEmailWithTracking>) -> Result<usize> {
        if emails.is_empty() {
            return Ok(0);
        }

        let mut count = 0usize;
        let uid = user_id.to_string();

        let mut tx = self.pool.begin().await.map_err(CoreError::database)?;

        for email in emails {
            let id = goingson_core::deterministic_email_id(email.message_id.as_deref());
            let received_at = format_datetime(&email.received_at.unwrap_or_else(Utc::now));
            let result = sqlx::query("INSERT OR IGNORE INTO emails (id, user_id, project_id, from_address, to_address, subject, body, html_body, is_read, is_archived, received_at, message_id, in_reply_to, thread_id, email_account_id, is_outgoing, imap_uid, source_folder, attachment_meta) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
                .bind(id.to_string()).bind(&uid).bind(email.project_id.map(|p| p.to_string()))
                .bind(&email.from_address).bind(&email.to_address).bind(&email.subject).bind(&email.body).bind(&email.html_body)
                .bind(if email.is_read { 1 } else { 0 }).bind(if email.is_archived { 1 } else { 0 }).bind(&received_at)
                .bind(&email.message_id).bind(&email.in_reply_to).bind(&email.thread_id)
                .bind(email.email_account_id.map(|a| a.to_string()))
                .bind(if email.is_outgoing { 1 } else { 0 }).bind(email.imap_uid).bind(&email.source_folder)
                .bind(&email.attachment_meta)
                .execute(&mut *tx).await.map_err(CoreError::database)?;
            if result.rows_affected() > 0 {
                count += 1;
            }
        }

        tx.commit().await.map_err(CoreError::database)?;
        Ok(count)
    }

    #[tracing::instrument(skip_all)]
    async fn delete(&self, id: EmailId, user_id: UserId) -> Result<bool> {
        let result = sqlx::query("DELETE FROM emails WHERE id = ? AND user_id = ?").bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
        Ok(result.rows_affected() > 0)
    }

    #[tracing::instrument(skip_all)]
    async fn mark_read(&self, id: EmailId, user_id: UserId) -> Result<bool> {
        let result = sqlx::query("UPDATE emails SET is_read = 1 WHERE id = ? AND user_id = ?").bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
        Ok(result.rows_affected() > 0)
    }

    #[tracing::instrument(skip_all)]
    async fn mark_unread(&self, id: EmailId, user_id: UserId) -> Result<bool> {
        let result = sqlx::query("UPDATE emails SET is_read = 0 WHERE id = ? AND user_id = ?").bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
        Ok(result.rows_affected() > 0)
    }

    #[tracing::instrument(skip_all)]
    async fn archive(&self, id: EmailId, user_id: UserId) -> Result<bool> {
        let result = sqlx::query("UPDATE emails SET is_archived = 1 WHERE id = ? AND user_id = ?").bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
        Ok(result.rows_affected() > 0)
    }

    #[tracing::instrument(skip_all)]
    async fn unarchive(&self, id: EmailId, user_id: UserId) -> Result<bool> {
        let result = sqlx::query("UPDATE emails SET is_archived = 0 WHERE id = ? AND user_id = ?").bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
        Ok(result.rows_affected() > 0)
    }

    #[tracing::instrument(skip_all)]
    async fn update_source_folder(&self, id: EmailId, user_id: UserId, new_folder: &str) -> Result<bool> {
        let result = sqlx::query("UPDATE emails SET source_folder = ? WHERE id = ? AND user_id = ?").bind(new_folder).bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
        Ok(result.rows_affected() > 0)
    }

    #[tracing::instrument(skip_all)]
    async fn mark_all_read(&self, user_id: UserId) -> Result<u64> {
        let result = sqlx::query("UPDATE emails SET is_read = 1 WHERE user_id = ? AND is_read = 0").bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
        Ok(result.rows_affected())
    }

    #[tracing::instrument(skip_all)]
    async fn link_to_project(&self, id: EmailId, user_id: UserId, project_id: Option<ProjectId>) -> Result<bool> {
        let result = sqlx::query("UPDATE emails SET project_id = ? WHERE id = ? AND user_id = ?").bind(project_id.map(|p| p.to_string())).bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
        Ok(result.rows_affected() > 0)
    }

    #[tracing::instrument(skip_all)]
    async fn count_unread(&self, user_id: UserId) -> Result<i64> {
        let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM emails WHERE user_id = ? AND is_read = 0").bind(user_id.to_string()).fetch_one(&self.pool).await.map_err(CoreError::database)?;
        Ok(row.0)
    }

    #[tracing::instrument(skip_all)]
    async fn exists_by_message_id(&self, user_id: UserId, message_id: &str) -> Result<bool> {
        let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM emails WHERE user_id = ? AND message_id = ?").bind(user_id.to_string()).bind(message_id).fetch_one(&self.pool).await.map_err(CoreError::database)?;
        Ok(row.0 > 0)
    }

    #[tracing::instrument(skip_all)]
    async fn exists_by_message_ids(&self, user_id: UserId, message_ids: &[&str]) -> Result<HashSet<String>> {
        if message_ids.is_empty() {
            return Ok(HashSet::new());
        }

        let placeholders = message_ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
        let query = format!(
            "SELECT message_id FROM emails WHERE user_id = ? AND message_id IN ({})",
            placeholders
        );

        let mut q = sqlx::query_as::<_, (String,)>(&query).bind(user_id.to_string());
        for msg_id in message_ids {
            q = q.bind(*msg_id);
        }

        let rows = q.fetch_all(&self.pool).await
            .map_err(CoreError::database)?;

        Ok(rows.into_iter().map(|(id,)| id).collect())
    }

    #[tracing::instrument(skip_all)]
    async fn exists_as_senders(&self, user_id: UserId, addresses: &[&str]) -> Result<HashSet<String>> {
        if addresses.is_empty() {
            return Ok(HashSet::new());
        }

        let placeholders = addresses.iter().map(|_| "?").collect::<Vec<_>>().join(",");
        let query = format!(
            "SELECT DISTINCT LOWER(from_address) FROM emails WHERE user_id = ? AND LOWER(from_address) IN ({})",
            placeholders
        );

        let mut q = sqlx::query_as::<_, (String,)>(&query).bind(user_id.to_string());
        for addr in addresses {
            q = q.bind(addr.to_lowercase());
        }

        let rows = q.fetch_all(&self.pool).await
            .map_err(CoreError::database)?;

        Ok(rows.into_iter().map(|(a,)| a).collect())
    }

    #[tracing::instrument(skip_all)]
    async fn snooze(&self, id: EmailId, user_id: UserId, until: DateTime<Utc>) -> Result<Option<Email>> {
        let until_str = format_datetime(&until);
        let result = sqlx::query("UPDATE emails SET snoozed_until = ? WHERE id = ? AND user_id = ?")
            .bind(&until_str)
            .bind(id.to_string())
            .bind(user_id.to_string())
            .execute(&self.pool)
            .await
            .map_err(CoreError::database)?;
        if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
    }

    #[tracing::instrument(skip_all)]
    async fn unsnooze(&self, id: EmailId, user_id: UserId) -> Result<Option<Email>> {
        let result = sqlx::query("UPDATE emails SET snoozed_until = NULL WHERE id = ? AND user_id = ?")
            .bind(id.to_string())
            .bind(user_id.to_string())
            .execute(&self.pool)
            .await
            .map_err(CoreError::database)?;
        if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
    }

    #[tracing::instrument(skip_all)]
    async fn list_snoozed(&self, user_id: UserId) -> Result<Vec<Email>> {
        let query = format!(
            "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.snoozed_until IS NOT NULL AND datetime(e.snoozed_until) > datetime('now') ORDER BY e.snoozed_until ASC",
            EMAIL_SELECT_COLUMNS
        );
        let rows = sqlx::query_as::<_, EmailRow>(&query)
            .bind(user_id.to_string())
            .bind(user_id.to_string())
            .fetch_all(&self.pool)
            .await
            .map_err(CoreError::database)?;
        rows.into_iter().map(Email::try_from).collect()
    }

    #[tracing::instrument(skip_all)]
    async fn mark_waiting(&self, id: EmailId, user_id: UserId, expected_response: Option<DateTime<Utc>>) -> Result<Option<Email>> {
        let now = format_datetime_now();
        let expected = format_datetime_opt(expected_response);

        let result = sqlx::query(
            "UPDATE emails SET waiting_for_response = 1, waiting_since = ?, expected_response_date = ? WHERE id = ? AND user_id = ?"
        )
            .bind(&now)
            .bind(&expected)
            .bind(id.to_string())
            .bind(user_id.to_string())
            .execute(&self.pool)
            .await
            .map_err(CoreError::database)?;

        if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
    }

    #[tracing::instrument(skip_all)]
    async fn clear_waiting(&self, id: EmailId, user_id: UserId) -> Result<Option<Email>> {
        let result = sqlx::query(
            "UPDATE emails SET waiting_for_response = 0, waiting_since = NULL, expected_response_date = NULL WHERE id = ? AND user_id = ?"
        )
            .bind(id.to_string())
            .bind(user_id.to_string())
            .execute(&self.pool)
            .await
            .map_err(CoreError::database)?;

        if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
    }

    #[tracing::instrument(skip_all)]
    async fn list_waiting(&self, user_id: UserId) -> Result<Vec<Email>> {
        let query = format!(
            "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.waiting_for_response = 1 ORDER BY e.expected_response_date ASC",
            EMAIL_SELECT_COLUMNS
        );
        let rows = sqlx::query_as::<_, EmailRow>(&query)
            .bind(user_id.to_string())
            .bind(user_id.to_string())
            .fetch_all(&self.pool)
            .await
            .map_err(CoreError::database)?;
        rows.into_iter().map(Email::try_from).collect()
    }

    #[tracing::instrument(skip_all)]
    async fn list_by_thread(&self, user_id: UserId, thread_id: &str) -> Result<Vec<Email>> {
        let query = format!(
            "SELECT {} 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",
            EMAIL_SELECT_COLUMNS
        );
        let rows = sqlx::query_as::<_, EmailRow>(&query)
            .bind(user_id.to_string())
            .bind(user_id.to_string())
            .bind(thread_id)
            .fetch_all(&self.pool)
            .await
            .map_err(CoreError::database)?;
        rows.into_iter().map(Email::try_from).collect()
    }

    #[tracing::instrument(skip_all)]
    async fn list_drafts(&self, user_id: UserId) -> Result<Vec<Email>> {
        let query = format!(
            "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.is_draft = 1 ORDER BY e.received_at DESC",
            EMAIL_SELECT_COLUMNS
        );
        let rows = sqlx::query_as::<_, EmailRow>(&query)
            .bind(user_id.to_string())
            .bind(user_id.to_string())
            .fetch_all(&self.pool)
            .await
            .map_err(CoreError::database)?;
        rows.into_iter().map(Email::try_from).collect()
    }

    #[tracing::instrument(skip_all)]
    async fn save_draft(&self, id: EmailId, user_id: UserId, from: &str, to: &str, cc: Option<&str>, bcc: Option<&str>, subject: &str, body: &str, account_id: Option<EmailAccountId>, in_reply_to: Option<&str>, _references: Option<&str>, thread_id: Option<&str>) -> Result<Email> {
        let now = format_datetime_now();
        let account_id_str = account_id.map(|a: EmailAccountId| a.to_string());

        // Upsert: update if exists, insert if not
        sqlx::query(r#"
            INSERT INTO emails (id, user_id, from_address, to_address, cc_address, bcc_address, subject, body,
                is_read, is_archived, is_draft, is_outgoing, received_at, draft_account_id, in_reply_to, thread_id)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, 0, 1, 1, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                from_address = excluded.from_address,
                to_address = excluded.to_address,
                cc_address = excluded.cc_address,
                bcc_address = excluded.bcc_address,
                subject = excluded.subject,
                body = excluded.body,
                received_at = excluded.received_at,
                draft_account_id = excluded.draft_account_id,
                in_reply_to = excluded.in_reply_to,
                thread_id = excluded.thread_id
        "#)
            .bind(id.to_string())
            .bind(user_id.to_string())
            .bind(from)
            .bind(to)
            .bind(cc)
            .bind(bcc)
            .bind(subject)
            .bind(body)
            .bind(&now)
            .bind(&account_id_str)
            .bind(in_reply_to)
            .bind(thread_id)
            .execute(&self.pool)
            .await
            .map_err(CoreError::database)?;

        self.get_by_id(id, user_id).await?.ok_or_else(|| CoreError::internal("Failed to retrieve saved draft"))
    }

    #[tracing::instrument(skip_all)]
    async fn get_by_message_id(&self, user_id: UserId, message_id: &str) -> Result<Option<Email>> {
        let query = format!(
            "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.message_id = ?",
            EMAIL_SELECT_COLUMNS
        );
        let row = sqlx::query_as::<_, EmailRow>(&query)
            .bind(user_id.to_string())
            .bind(user_id.to_string())
            .bind(message_id)
            .fetch_optional(&self.pool)
            .await
            .map_err(CoreError::database)?;
        row.map(Email::try_from).transpose()
    }

    #[tracing::instrument(skip_all)]
    async fn update_labels(&self, id: EmailId, user_id: UserId, labels: &[String]) -> Result<Option<Email>> {
        let labels_json = serde_json::to_string(labels).unwrap_or_else(|_| "[]".to_string());
        let result = sqlx::query("UPDATE emails SET labels = ? WHERE id = ? AND user_id = ?")
            .bind(&labels_json)
            .bind(id.to_string())
            .bind(user_id.to_string())
            .execute(&self.pool)
            .await
            .map_err(CoreError::database)?;
        if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
    }

    #[tracing::instrument(skip_all)]
    async fn list_folders(&self, user_id: UserId) -> Result<Vec<String>> {
        let rows: Vec<(String,)> = sqlx::query_as(
            "SELECT DISTINCT source_folder FROM emails WHERE user_id = ? AND source_folder IS NOT NULL AND is_draft = 0 ORDER BY source_folder ASC"
        )
            .bind(user_id.to_string())
            .fetch_all(&self.pool)
            .await
            .map_err(CoreError::database)?;
        Ok(rows.into_iter().map(|r| r.0).collect())
    }

    #[tracing::instrument(skip_all)]
    async fn list_labels(&self, user_id: UserId) -> Result<Vec<String>> {
        // Extract all unique labels across all emails via JSON parsing
        let rows: Vec<(String,)> = sqlx::query_as(
            "SELECT DISTINCT j.value FROM emails e, json_each(e.labels) j WHERE e.user_id = ? AND e.is_draft = 0 ORDER BY j.value ASC"
        )
            .bind(user_id.to_string())
            .fetch_all(&self.pool)
            .await
            .map_err(CoreError::database)?;
        Ok(rows.into_iter().map(|r| r.0).collect())
    }
}