use rusqlite::{Connection, params};
use tracing::{debug, warn};

use crate::Config;
use crate::core::error::Result;
use crate::core::time::now_ms;
use crate::documents::SourceType;
use crate::gitlab::GitLabClient;
use crate::gitlab::transformers::{
    NormalizedDiscussion, NormalizedNote, NoteableRef, transform_discussion, transform_notes,
};
use crate::gitlab::types::GitLabDiscussion;
use crate::ingestion::dirty_tracker;
use crate::ingestion::storage::payloads::{StorePayloadOptions, store_payload};

use super::issues::IssueForDiscussionSync;

#[derive(Debug)]
pub struct NoteUpsertOutcome {
    pub local_note_id: i64,
    pub changed_semantics: bool,
}

#[derive(Debug, Default)]
pub struct IngestDiscussionsResult {
    pub discussions_fetched: usize,
    pub discussions_upserted: usize,
    pub notes_upserted: usize,
    pub stale_discussions_removed: usize,
}

// ═══════════════════════════════════════════════════════════════════════
//  Prefetch pattern — concurrent HTTP fetch, sequential DB write
// ═══════════════════════════════════════════════════════════════════════

#[derive(Debug)]
pub struct PrefetchedIssueDiscussions {
    pub issue: IssueForDiscussionSync,
    pub discussions: Vec<PrefetchedDiscussion>,
    pub fetch_error: Option<String>,
}

#[derive(Debug)]
pub struct PrefetchedDiscussion {
    pub raw: GitLabDiscussion,
    pub normalized: NormalizedDiscussion,
    pub notes: Vec<NormalizedNote>,
}

/// Prefetch all discussions for an issue (HTTP only, no DB writes).
/// This function is designed to be called concurrently via `join_all`.
pub async fn prefetch_issue_discussions(
    client: &GitLabClient,
    gitlab_project_id: i64,
    local_project_id: i64,
    issue: IssueForDiscussionSync,
) -> PrefetchedIssueDiscussions {
    debug!(issue_iid = issue.iid, "Prefetching discussions for issue");

    let raw_discussions = match client
        .fetch_all_issue_discussions(gitlab_project_id, issue.iid)
        .await
    {
        Ok(d) => d,
        Err(e) => {
            return PrefetchedIssueDiscussions {
                issue,
                discussions: Vec::new(),
                fetch_error: Some(e.to_string()),
            };
        }
    };

    let mut discussions = Vec::with_capacity(raw_discussions.len());

    for raw in raw_discussions {
        let normalized = transform_discussion(
            &raw,
            local_project_id,
            NoteableRef::Issue(issue.local_issue_id),
        );
        let notes = transform_notes(&raw, local_project_id);

        discussions.push(PrefetchedDiscussion {
            raw,
            normalized,
            notes,
        });
    }

    PrefetchedIssueDiscussions {
        issue,
        discussions,
        fetch_error: None,
    }
}

/// Write prefetched discussions to the database (sequential DB writes).
pub fn write_prefetched_issue_discussions(
    conn: &Connection,
    config: &Config,
    local_project_id: i64,
    prefetched: PrefetchedIssueDiscussions,
) -> Result<IngestDiscussionsResult> {
    let mut result = IngestDiscussionsResult::default();
    let issue = &prefetched.issue;

    if let Some(error) = &prefetched.fetch_error {
        warn!(issue_iid = issue.iid, error = %error, "Prefetch failed for issue");
        return Ok(result);
    }

    let run_seen_at = now_ms();
    let mut seen_discussion_ids: Vec<String> = Vec::with_capacity(prefetched.discussions.len());

    for disc in &prefetched.discussions {
        result.discussions_fetched += 1;
        let notes_count = disc.notes.len();

        let tx = conn.unchecked_transaction()?;

        let payload_bytes = serde_json::to_vec(&disc.raw)?;
        let payload_id = store_payload(
            &tx,
            StorePayloadOptions {
                project_id: Some(local_project_id),
                resource_type: "discussion",
                gitlab_id: &disc.raw.id,
                json_bytes: &payload_bytes,
                compress: config.storage.compress_raw_payloads,
            },
        )?;

        upsert_discussion(&tx, &disc.normalized, payload_id)?;

        let local_discussion_id: i64 = tx.query_row(
            "SELECT id FROM discussions WHERE project_id = ? AND gitlab_discussion_id = ?",
            (local_project_id, &disc.normalized.gitlab_discussion_id),
            |row| row.get(0),
        )?;

        dirty_tracker::mark_dirty_tx(&tx, SourceType::Discussion, local_discussion_id)?;

        // Mark child note documents dirty (they inherit parent metadata)
        tx.execute(
            "INSERT INTO dirty_sources (source_type, source_id, queued_at)
             SELECT 'note', n.id, ?1
             FROM notes n
             WHERE n.discussion_id = ?2 AND n.is_system = 0
             ON CONFLICT(source_type, source_id) DO UPDATE SET queued_at = excluded.queued_at, attempt_count = 0",
            params![now_ms(), local_discussion_id],
        )?;

        for note in &disc.notes {
            let outcome = upsert_note_for_issue(&tx, local_discussion_id, note, run_seen_at, None)?;
            if !note.is_system && outcome.changed_semantics {
                dirty_tracker::mark_dirty_tx(&tx, SourceType::Note, outcome.local_note_id)?;
            }
        }

        sweep_stale_issue_notes(&tx, local_discussion_id, run_seen_at)?;

        tx.commit()?;

        result.discussions_upserted += 1;
        result.notes_upserted += notes_count;
        seen_discussion_ids.push(disc.normalized.gitlab_discussion_id.clone());
    }

    // Only do stale removal if fetch succeeded
    let removed = remove_stale_discussions(conn, issue.local_issue_id, &seen_discussion_ids)?;
    result.stale_discussions_removed = removed;

    update_issue_sync_timestamp(conn, issue.local_issue_id, issue.updated_at)?;

    Ok(result)
}

// ═══════════════════════════════════════════════════════════════════════
//  Database helpers
// ═══════════════════════════════════════════════════════════════════════

fn upsert_discussion(
    conn: &Connection,
    discussion: &crate::gitlab::transformers::NormalizedDiscussion,
    payload_id: i64,
) -> Result<()> {
    conn.execute(
        "INSERT INTO discussions (
            gitlab_discussion_id, project_id, issue_id, merge_request_id, noteable_type,
            individual_note, first_note_at, last_note_at, last_seen_at,
            resolvable, resolved, raw_payload_id
        ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)
        ON CONFLICT(project_id, gitlab_discussion_id) DO UPDATE SET
            first_note_at = excluded.first_note_at,
            last_note_at = excluded.last_note_at,
            last_seen_at = excluded.last_seen_at,
            resolvable = excluded.resolvable,
            resolved = excluded.resolved,
            raw_payload_id = excluded.raw_payload_id",
        (
            &discussion.gitlab_discussion_id,
            discussion.project_id,
            discussion.issue_id,
            discussion.merge_request_id,
            &discussion.noteable_type,
            discussion.individual_note,
            discussion.first_note_at,
            discussion.last_note_at,
            discussion.last_seen_at,
            discussion.resolvable,
            discussion.resolved,
            payload_id,
        ),
    )?;
    Ok(())
}

fn upsert_note_for_issue(
    conn: &Connection,
    discussion_id: i64,
    note: &NormalizedNote,
    last_seen_at: i64,
    payload_id: Option<i64>,
) -> Result<NoteUpsertOutcome> {
    // Pre-read for semantic change detection
    let existing = conn
        .query_row(
            "SELECT id, body, note_type, resolved, resolved_by,
                    position_old_path, position_new_path, position_old_line, position_new_line,
                    position_type, position_line_range_start, position_line_range_end,
                    position_base_sha, position_start_sha, position_head_sha
             FROM notes WHERE gitlab_id = ?",
            params![note.gitlab_id],
            |row| {
                Ok((
                    row.get::<_, i64>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, Option<String>>(2)?,
                    row.get::<_, bool>(3)?,
                    row.get::<_, Option<String>>(4)?,
                    row.get::<_, Option<String>>(5)?,
                    row.get::<_, Option<String>>(6)?,
                    row.get::<_, Option<i32>>(7)?,
                    row.get::<_, Option<i32>>(8)?,
                    row.get::<_, Option<String>>(9)?,
                    row.get::<_, Option<i32>>(10)?,
                    row.get::<_, Option<i32>>(11)?,
                    row.get::<_, Option<String>>(12)?,
                    row.get::<_, Option<String>>(13)?,
                    row.get::<_, Option<String>>(14)?,
                ))
            },
        )
        .ok();

    let changed_semantics = match &existing {
        Some((
            _id,
            body,
            note_type,
            resolved,
            resolved_by,
            pos_old_path,
            pos_new_path,
            pos_old_line,
            pos_new_line,
            pos_type,
            pos_range_start,
            pos_range_end,
            pos_base_sha,
            pos_start_sha,
            pos_head_sha,
        )) => {
            *body != note.body
                || *note_type != note.note_type
                || *resolved != note.resolved
                || *resolved_by != note.resolved_by
                || *pos_old_path != note.position_old_path
                || *pos_new_path != note.position_new_path
                || *pos_old_line != note.position_old_line
                || *pos_new_line != note.position_new_line
                || *pos_type != note.position_type
                || *pos_range_start != note.position_line_range_start
                || *pos_range_end != note.position_line_range_end
                || *pos_base_sha != note.position_base_sha
                || *pos_start_sha != note.position_start_sha
                || *pos_head_sha != note.position_head_sha
        }
        None => true,
    };

    conn.execute(
        "INSERT INTO notes (
            gitlab_id, discussion_id, project_id, note_type, is_system,
            author_id, author_username, body, created_at, updated_at, last_seen_at,
            position, resolvable, resolved, resolved_by, resolved_at,
            position_old_path, position_new_path, position_old_line, position_new_line,
            position_type, position_line_range_start, position_line_range_end,
            position_base_sha, position_start_sha, position_head_sha,
            raw_payload_id
        ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26, ?27)
        ON CONFLICT(gitlab_id) DO UPDATE SET
            body = excluded.body,
            note_type = excluded.note_type,
            author_id = excluded.author_id,
            updated_at = excluded.updated_at,
            last_seen_at = excluded.last_seen_at,
            resolvable = excluded.resolvable,
            resolved = excluded.resolved,
            resolved_by = excluded.resolved_by,
            resolved_at = excluded.resolved_at,
            position_old_path = excluded.position_old_path,
            position_new_path = excluded.position_new_path,
            position_old_line = excluded.position_old_line,
            position_new_line = excluded.position_new_line,
            position_type = excluded.position_type,
            position_line_range_start = excluded.position_line_range_start,
            position_line_range_end = excluded.position_line_range_end,
            position_base_sha = excluded.position_base_sha,
            position_start_sha = excluded.position_start_sha,
            position_head_sha = excluded.position_head_sha,
            raw_payload_id = COALESCE(excluded.raw_payload_id, raw_payload_id)",
        params![
            note.gitlab_id,
            discussion_id,
            note.project_id,
            &note.note_type,
            note.is_system,
            note.author_id,
            &note.author_username,
            &note.body,
            note.created_at,
            note.updated_at,
            last_seen_at,
            note.position,
            note.resolvable,
            note.resolved,
            &note.resolved_by,
            note.resolved_at,
            &note.position_old_path,
            &note.position_new_path,
            note.position_old_line,
            note.position_new_line,
            &note.position_type,
            note.position_line_range_start,
            note.position_line_range_end,
            &note.position_base_sha,
            &note.position_start_sha,
            &note.position_head_sha,
            payload_id,
        ],
    )?;

    let local_note_id: i64 = conn.query_row(
        "SELECT id FROM notes WHERE gitlab_id = ?",
        params![note.gitlab_id],
        |row| row.get(0),
    )?;

    Ok(NoteUpsertOutcome {
        local_note_id,
        changed_semantics,
    })
}

fn sweep_stale_issue_notes(
    conn: &Connection,
    discussion_id: i64,
    last_seen_at: i64,
) -> Result<usize> {
    // Step 1: Delete note documents for stale notes
    conn.execute(
        "DELETE FROM documents WHERE source_type = 'note' AND source_id IN
         (SELECT id FROM notes WHERE discussion_id = ?1 AND last_seen_at < ?2 AND is_system = 0)",
        params![discussion_id, last_seen_at],
    )?;

    // Step 2: Delete dirty_sources entries for stale notes
    conn.execute(
        "DELETE FROM dirty_sources WHERE source_type = 'note' AND source_id IN
         (SELECT id FROM notes WHERE discussion_id = ?1 AND last_seen_at < ?2 AND is_system = 0)",
        params![discussion_id, last_seen_at],
    )?;

    // Step 3: Delete the stale notes themselves
    let deleted = conn.execute(
        "DELETE FROM notes WHERE discussion_id = ?1 AND last_seen_at < ?2",
        params![discussion_id, last_seen_at],
    )?;
    if deleted > 0 {
        debug!(discussion_id, deleted, "Swept stale issue notes");
    }
    Ok(deleted)
}

fn remove_stale_discussions(
    conn: &Connection,
    issue_id: i64,
    seen_ids: &[String],
) -> Result<usize> {
    if seen_ids.is_empty() {
        let deleted = conn.execute("DELETE FROM discussions WHERE issue_id = ?", [issue_id])?;
        return Ok(deleted);
    }

    const CHUNK_SIZE: usize = 500;

    let total_deleted = if seen_ids.len() > CHUNK_SIZE {
        conn.execute(
            "CREATE TEMP TABLE IF NOT EXISTS _temp_seen_discussions (id TEXT PRIMARY KEY)",
            [],
        )?;

        conn.execute("DELETE FROM _temp_seen_discussions", [])?;

        for chunk in seen_ids.chunks(CHUNK_SIZE) {
            let placeholders: Vec<&str> = chunk.iter().map(|_| "(?)").collect();
            let sql = format!(
                "INSERT OR IGNORE INTO _temp_seen_discussions (id) VALUES {}",
                placeholders.join(", ")
            );

            let params: Vec<&dyn rusqlite::ToSql> =
                chunk.iter().map(|s| s as &dyn rusqlite::ToSql).collect();
            conn.execute(&sql, params.as_slice())?;
        }

        let deleted = conn.execute(
            "DELETE FROM discussions
             WHERE issue_id = ?1
             AND gitlab_discussion_id NOT IN (SELECT id FROM _temp_seen_discussions)",
            [issue_id],
        )?;

        conn.execute("DROP TABLE IF EXISTS _temp_seen_discussions", [])?;
        deleted
    } else {
        let placeholders: Vec<&str> = seen_ids.iter().map(|_| "?").collect();
        let sql = format!(
            "DELETE FROM discussions WHERE issue_id = ?1 AND gitlab_discussion_id NOT IN ({})",
            placeholders.join(", ")
        );

        let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(issue_id)];
        for id in seen_ids {
            params.push(Box::new(id.clone()));
        }

        let param_refs: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect();
        conn.execute(&sql, param_refs.as_slice())?
    };

    Ok(total_deleted)
}

fn update_issue_sync_timestamp(conn: &Connection, issue_id: i64, updated_at: i64) -> Result<()> {
    conn.execute(
        "UPDATE issues SET discussions_synced_for_updated_at = ? WHERE id = ?",
        (updated_at, issue_id),
    )?;
    Ok(())
}

#[cfg(test)]
#[path = "discussions_tests.rs"]
mod tests;
