use std::collections::HashSet;

use sqlx::SqlitePool;

use crate::model::errors::{AppError, ErrorSource};
use crate::model::thread::{
    AddMessageInput, MessageDto, MessageRecord, RunStatus, RunSummaryDto, ThreadRecord,
    ThreadSnapshotDto, ThreadStatus, ThreadSummaryDto,
};
use crate::persistence::repo::{
    message_repo, run_helper_repo, run_repo, thread_repo, tool_call_repo,
};

const DEFAULT_MESSAGE_PAGE_SIZE: i64 = 50;

pub struct ThreadManager {
    pool: SqlitePool,
}

impl ThreadManager {
    pub fn new(pool: SqlitePool) -> Self {
        Self { pool }
    }

    // -----------------------------------------------------------------------
    // Thread CRUD
    // -----------------------------------------------------------------------

    /// List threads for a workspace (sidebar query).
    pub async fn list(
        &self,
        workspace_id: &str,
        limit: Option<i64>,
        offset: Option<i64>,
    ) -> Result<Vec<ThreadSummaryDto>, AppError> {
        let records = thread_repo::list_by_workspace(
            &self.pool,
            workspace_id,
            limit.unwrap_or(100),
            offset.unwrap_or(0),
        )
        .await?;

        // Backfill the active-run start timestamp and elapsed running seconds
        // in one round-trip so the frontend workbench header can render
        // elapsed time without a follow-up IPC per thread.
        let thread_ids: Vec<String> = records.iter().map(|r| r.id.clone()).collect();
        let active_started =
            run_repo::list_active_run_started_at_ms_by_threads(&self.pool, &thread_ids).await?;
        let thread_elapsed =
            run_repo::list_thread_elapsed_running_seconds_by_threads(&self.pool, &thread_ids)
                .await?;

        Ok(records
            .into_iter()
            .map(|r| {
                let mut dto = ThreadSummaryDto::from(r);
                if let Some(ms) = active_started.get(&dto.id).copied() {
                    dto.active_run_started_at_ms = Some(ms);
                }
                if let Some(secs) = thread_elapsed.get(&dto.id).copied() {
                    dto.active_run_elapsed_seconds = Some(secs);
                }
                dto
            })
            .collect())
    }

    /// Create a new thread under a workspace.
    pub async fn create(
        &self,
        workspace_id: &str,
        title: Option<String>,
        profile_id: Option<String>,
    ) -> Result<ThreadSummaryDto, AppError> {
        let record = ThreadRecord {
            id: uuid::Uuid::now_v7().to_string(),
            workspace_id: workspace_id.to_string(),
            profile_id,
            title: title.unwrap_or_default(),
            status: ThreadStatus::Idle,
            summary: None,
            last_active_at: String::new(),
            created_at: String::new(),
            updated_at: String::new(),
        };

        thread_repo::insert(&self.pool, &record).await?;
        tracing::info!(thread_id = %record.id, workspace_id = %workspace_id, "thread created");

        // Re-fetch for server-set timestamps
        let saved = thread_repo::find_by_id(&self.pool, &record.id)
            .await?
            .ok_or_else(|| AppError::internal(ErrorSource::Thread, "failed to read back thread"))?;

        Ok(ThreadSummaryDto::from(saved))
    }

    /// Load a full thread snapshot (for UI recovery and run startup).
    pub async fn load(
        &self,
        id: &str,
        message_cursor: Option<String>,
        message_limit: Option<i64>,
    ) -> Result<ThreadSnapshotDto, AppError> {
        let thread = thread_repo::find_by_id(&self.pool, id)
            .await?
            .ok_or_else(|| AppError::not_found(ErrorSource::Thread, "thread"))?;

        let limit = message_limit.unwrap_or(DEFAULT_MESSAGE_PAGE_SIZE);

        let messages = message_repo::list_recent(
            &self.pool,
            id,
            message_cursor.as_deref(),
            limit + 1, // fetch one extra to detect has_more
        )
        .await?;

        let has_more = messages.len() as i64 > limit;
        let extra_messages = messages.len().saturating_sub(limit as usize);
        let messages: Vec<MessageDto> = messages
            .into_iter()
            .skip(extra_messages)
            .map(MessageDto::from)
            .collect();

        let active_run = run_repo::find_active_by_thread(&self.pool, id).await?;
        let latest_run = run_repo::find_latest_by_thread(&self.pool, id).await?;
        let mut run_ids: Vec<String> = messages
            .iter()
            .filter_map(|message| message.run_id.clone())
            .collect();
        if let Some(run) = active_run.as_ref() {
            if !run_ids.iter().any(|candidate| candidate == &run.id) {
                run_ids.push(run.id.clone());
            }
        }
        if let Some(run) = latest_run.as_ref() {
            if !run_ids.iter().any(|candidate| candidate == &run.id) {
                run_ids.push(run.id.clone());
            }
        }
        let tool_calls = tool_call_repo::list_by_run_ids(&self.pool, &run_ids).await?;
        let helpers = run_helper_repo::list_by_run_ids(&self.pool, &run_ids).await?;

        // Load task boards
        let task_boards =
            crate::core::task_board_manager::load_thread_task_boards(&self.pool, id).await?;
        let active_task_board_id = task_boards
            .iter()
            .find(|b| b.status == crate::model::task_board::TaskBoardStatus::Active)
            .map(|b| b.id.clone());

        Ok(ThreadSnapshotDto {
            thread: ThreadSummaryDto::from(thread),
            messages,
            has_more_messages: has_more,
            active_run,
            latest_run,
            tool_calls,
            helpers,
            task_boards,
            active_task_board_id,
        })
    }

    /// Update thread title.
    pub async fn update_title(&self, id: &str, title: &str) -> Result<(), AppError> {
        thread_repo::update_title(&self.pool, id, title).await
    }

    /// Update the profile bound to a thread.
    pub async fn update_profile(&self, id: &str, profile_id: Option<&str>) -> Result<(), AppError> {
        let updated = thread_repo::update_profile(&self.pool, id, profile_id).await?;
        if !updated {
            return Err(AppError::not_found(ErrorSource::Thread, "thread"));
        }
        Ok(())
    }

    /// Delete a thread and all its messages.
    pub async fn delete(&self, id: &str) -> Result<(), AppError> {
        let deleted = thread_repo::delete(&self.pool, id).await?;
        if !deleted {
            return Err(AppError::not_found(ErrorSource::Thread, "thread"));
        }
        tracing::info!(thread_id = %id, "thread deleted");
        Ok(())
    }

    // -----------------------------------------------------------------------
    // Messages
    // -----------------------------------------------------------------------

    /// Add a user message to a thread.
    pub async fn add_message(
        &self,
        thread_id: &str,
        input: AddMessageInput,
    ) -> Result<MessageDto, AppError> {
        // Verify thread exists
        thread_repo::find_by_id(&self.pool, thread_id)
            .await?
            .ok_or_else(|| AppError::not_found(ErrorSource::Thread, "thread"))?;

        let record = MessageRecord {
            id: uuid::Uuid::now_v7().to_string(),
            thread_id: thread_id.to_string(),
            run_id: None,
            role: input.role,
            content_markdown: input.content,
            parts_json: None,
            message_type: input
                .message_type
                .unwrap_or_else(|| "plain_message".to_string()),
            status: "completed".to_string(),
            metadata_json: input.metadata.map(|v| v.to_string()),
            attachments_json: None,
            created_at: String::new(),
        };

        message_repo::insert(&self.pool, &record).await?;

        // Touch thread activity timestamp
        thread_repo::touch_active(&self.pool, thread_id).await?;

        // Re-read to get server timestamp
        let messages = message_repo::list_recent(&self.pool, thread_id, None, 1).await?;
        let msg = messages.into_iter().last().ok_or_else(|| {
            AppError::internal(ErrorSource::Thread, "failed to read back message")
        })?;

        Ok(MessageDto::from(msg))
    }

    // -----------------------------------------------------------------------
    // Thread status derivation
    // -----------------------------------------------------------------------

    /// Derive and update thread status from the latest run.
    pub async fn sync_status(&self, thread_id: &str) -> Result<ThreadStatus, AppError> {
        let latest_run = run_repo::find_latest_by_thread(&self.pool, thread_id).await?;
        let status = derive_thread_status(latest_run.as_ref());
        thread_repo::update_status(&self.pool, thread_id, &status).await?;
        Ok(status)
    }

    // -----------------------------------------------------------------------
    // Crash recovery
    // -----------------------------------------------------------------------

    /// On app startup, mark any dangling active runs as interrupted,
    /// then sync affected thread statuses.
    pub async fn recover_interrupted_runs(&self) -> Result<(), AppError> {
        let mut thread_ids: HashSet<String> =
            run_repo::list_thread_ids_with_active_runs(&self.pool)
                .await?
                .into_iter()
                .collect();
        thread_ids.extend(thread_repo::list_ids_with_active_status(&self.pool).await?);

        let count = run_repo::interrupt_active_runs(&self.pool).await?;
        let tc_count = tool_call_repo::interrupt_active_tool_calls(&self.pool).await?;
        let helper_count = run_helper_repo::interrupt_active_helpers(&self.pool).await?;
        let reasoning_count = message_repo::discard_dangling_reasoning(&self.pool).await?;
        if count > 0 || tc_count > 0 || helper_count > 0 || reasoning_count > 0 {
            tracing::warn!(
                runs = count,
                tool_calls = tc_count,
                helpers = helper_count,
                reasoning = reasoning_count,
                "interrupted dangling runs/tool_calls/helpers/reasoning on startup"
            );
        }

        for thread_id in thread_ids {
            self.sync_status(&thread_id).await?;
        }

        Ok(())
    }
}

/// Derive ThreadStatus from the latest run's status.
fn derive_thread_status(latest_run: Option<&RunSummaryDto>) -> ThreadStatus {
    match latest_run {
        None => ThreadStatus::Idle,
        Some(run) => RunStatus::from_str(&run.status)
            .map(|s| s.to_thread_status())
            .unwrap_or(ThreadStatus::Idle),
    }
}
