use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use cache::CacheClient;
use config::git::SyncSettings;
use export::{AppDatabase, AppRedis};
use models::repo::repos;
use sea_orm::EntityTrait;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::{Instrument as _, debug, error, info, warn};

use crate::sync::error::{SyncError, SyncResult};
use crate::sync::event::PostReceiveTask;
use crate::sync::queue::SyncQueue;
use crate::sync::sync::RepoSync;
use crate::sync::webhook::WebhookTx;

const ATTEMPT_KEY_PREFIX: &str = "git:sync:attempts:";
const ATTEMPT_TTL_SECS: u64 = 3600;
const REQUEUE_BACKOFF_SECS: u64 = 5;

#[derive(Clone)]
pub struct SyncWorker {
    db: Arc<AppDatabase>,
    redis: AppRedis,
    cache: Arc<CacheClient>,
    repo_root: PathBuf,
    queue: SyncQueue,
    webhook: Arc<dyn WebhookTx>,
    settings: SyncSettings,
    cancel: CancellationToken,
}

pub struct RunningWorker {
    handles: Vec<JoinHandle<()>>,
}

impl RunningWorker {
    pub async fn shutdown(self) -> SyncResult<()> {
        for h in self.handles {
            if !h.is_finished() {
                h.abort();
                let _ = h.await;
            }
        }
        Ok(())
    }
}

impl SyncWorker {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        db: Arc<AppDatabase>,
        redis: AppRedis,
        cache: Arc<CacheClient>,
        repo_root: PathBuf,
        queue: SyncQueue,
        webhook: Arc<dyn WebhookTx>,
        settings: SyncSettings,
        cancel: CancellationToken,
    ) -> Self {
        let concurrency = settings.concurrency.max(1);
        info!(
            concurrency,
            topic = %settings.queue_topic,
            "starting sync worker"
        );
        Self {
            db,
            redis,
            cache,
            repo_root,
            queue,
            webhook,
            settings,
            cancel,
        }
    }

    pub async fn run(self) -> SyncResult<RunningWorker> {
        let concurrency = self.settings.concurrency.max(1);
        let topic = self.settings.queue_topic.clone();
        let brpop = self.settings.brpop_timeout_secs;

        let mut queue = self.queue;
        if queue.consumer_conns().is_empty() {
            queue = SyncQueue::with_consumer_conns(queue.redis().clone(), concurrency).await?;
        }

        let mut handles = Vec::with_capacity(concurrency);
        for i in 0..concurrency {
            let db = self.db.clone();
            let redis = self.redis.clone();
            let cache = self.cache.clone();
            let repo_root = self.repo_root.clone();
            let queue = queue.clone();
            let webhook = self.webhook.clone();
            let settings = self.settings.clone();
            let cancel = self.cancel.clone();
            let topic = topic.clone();

            handles.push(tokio::spawn(async move {
                run_consumer(
                    i, &db, &redis, &cache, &repo_root, &queue, webhook, &settings, &cancel,
                    &topic, brpop,
                )
                .await;
            }));
        }
        Ok(RunningWorker { handles })
    }
}

#[allow(clippy::too_many_arguments)]
async fn run_consumer(
    id: usize,
    db: &Arc<AppDatabase>,
    redis: &AppRedis,
    cache: &Arc<CacheClient>,
    repo_root: &Path,
    queue: &SyncQueue,
    webhook: Arc<dyn WebhookTx>,
    settings: &SyncSettings,
    cancel: &CancellationToken,
    topic: &str,
    brpop_timeout: u64,
) {
    debug!(consumer = id, "consumer started");
    loop {
        if cancel.is_cancelled() {
            break;
        }
        let payload = tokio::select! {
            biased;
            _ = cancel.cancelled() => break,
            res = queue.consume(topic, brpop_timeout, id) => match res {
                Ok(Some(p)) => p,
                Ok(None) => continue,
                Err(e) => {
                    warn!(consumer = id, error = %e, "queue consume error; backing off");
                    tokio::time::sleep(Duration::from_secs(REQUEUE_BACKOFF_SECS)).await;
                    continue;
                }
            }
        };

        let task: PostReceiveTask = match serde_json::from_slice(&payload) {
            Ok(t) => t,
            Err(e) => {
                error!(
                    consumer = id,
                    error = %e,
                    payload_len = payload.len(),
                    "invalid sync task payload; dropping"
                );
                continue;
            }
        };

        let started = std::time::Instant::now();
        debug!(
            consumer = id,
            task_id = %task.id,
            repo_id = %task.repo_id,
            refs = task.refs.len(),
            "sync task received"
        );
        let span = tracing::info_span!(
            "git.sync",
            task_id = %task.id,
            repo_id = %task.repo_id,
            refs = task.refs.len(),
            request_id = tracing::field::Empty
        );
        if let Some(request_id) = task.request_id.as_deref() {
            span.record("request_id", request_id);
        }
        export::telemetry::set_parent_from_traceparent(&span, task.trace.as_deref());

        let repo = match repos::Entity::find_by_id(task.repo_id).one(db.read()).await {
            Ok(Some(r)) => r,
            Ok(None) => {
                warn!(
                    consumer = id,
                    task_id = %task.id,
                    repo_id = %task.repo_id,
                    "repo not found; skipping"
                );
                continue;
            }
            Err(e) => {
                error!(
                    consumer = id,
                    task_id = %task.id,
                    repo_id = %task.repo_id,
                    error = %e,
                    "repo lookup failed; skipping"
                );
                continue;
            }
        };

        let rsync = RepoSync::new(
            db.clone(),
            redis.clone(),
            cache.clone(),
            repo,
            repo_root.to_path_buf(),
            settings.clone(),
        );

        match export::middleware::with_request_id(
            task.request_id.as_deref(),
            rsync.sync().instrument(span.clone()),
        )
        .await
        {
            Ok(()) => {
                export::metrics::sync_job("ok");
                info!(
                    consumer = id,
                    task_id = %task.id,
                    repo_id = %task.repo_id,
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    "sync task completed"
                );
                if let Err(e) =
                    crate::sync::pulls::refresh_pulls_after_sync(db, rsync.repo_model(), &task)
                        .await
                {
                    warn!(
                        consumer = id,
                        task_id = %task.id,
                        repo_id = %task.repo_id,
                        error = %e,
                        "pull request refresh failed"
                    );
                }
                if let Err(e) = export::middleware::with_request_id(
                    task.request_id.as_deref(),
                    webhook.dispatch(&task).instrument(span.clone()),
                )
                .await
                {
                    warn!(
                        consumer = id,
                        task_id = %task.id,
                        repo_id = %task.repo_id,
                        error = %e,
                        "webhook dispatch failed"
                    );
                }
            }
            Err(e) if e.is_locked() => {
                export::metrics::sync_job("locked");
                warn!(
                    consumer = id,
                    task_id = %task.id,
                    repo_id = %task.repo_id,
                    "sync locked; requeueing"
                );
                if let Err(e2) = requeue_or_dlq(queue, settings, &task, &payload).await {
                    error!(
                        consumer = id,
                        task_id = %task.id,
                        repo_id = %task.repo_id,
                        error = %e2,
                        "requeue failed"
                    );
                }
            }
            Err(e) => {
                export::metrics::sync_job("error");
                error!(
                    consumer = id,
                    task_id = %task.id,
                    repo_id = %task.repo_id,
                    error = %e,
                    "sync task failed"
                );
                if let Err(e2) = requeue_or_dlq(queue, settings, &task, &payload).await {
                    error!(
                        consumer = id,
                        task_id = %task.id,
                        repo_id = %task.repo_id,
                        error = %e2,
                        "requeue failed"
                    );
                }
            }
        }
    }
    debug!(consumer = id, "consumer stopped");
}

async fn requeue_or_dlq(
    queue: &SyncQueue,
    settings: &SyncSettings,
    task: &PostReceiveTask,
    payload: &[u8],
) -> SyncResult<()> {
    let key = format!("{ATTEMPT_KEY_PREFIX}{}", task.id);
    let attempts = queue.incr_with_ttl(&key, ATTEMPT_TTL_SECS).await?;

    if attempts as u64 > settings.max_lock_retries {
        let _ = queue.delete(&key).await;
        queue
            .publish_raw(&settings.dlq_topic, Some(task.repo_id.as_bytes()), payload)
            .await?;
        warn!(
            task_id = %task.id,
            repo_id = %task.repo_id,
            attempts,
            "task exceeded retry limit; sent to DLQ"
        );
        return Ok(());
    }

    tokio::time::sleep(Duration::from_secs(REQUEUE_BACKOFF_SECS)).await;
    queue
        .publish_raw(
            &settings.queue_topic,
            Some(task.repo_id.as_bytes()),
            payload,
        )
        .await?;
    debug!(
        task_id = %task.id,
        repo_id = %task.repo_id,
        attempts,
        "re-queued retry task"
    );
    Ok(())
}

pub async fn run_once(
    worker: &SyncWorker,
    task: PostReceiveTask,
    payload: Vec<u8>,
) -> SyncResult<()> {
    let repo = repos::Entity::find_by_id(task.repo_id)
        .one(worker.db.read())
        .await
        .map_err(SyncError::Db)?
        .ok_or_else(|| SyncError::NotFound(format!("repo {}", task.repo_id)))?;

    let rsync = RepoSync::new(
        worker.db.clone(),
        worker.redis.clone(),
        worker.cache.clone(),
        repo,
        worker.repo_root.clone(),
        worker.settings.clone(),
    );

    let result = rsync.sync().await;
    if result.is_ok() {
        let _ = worker.webhook.dispatch(&task).await;
    } else if let Err(e) = requeue_or_dlq(&worker.queue, &worker.settings, &task, &payload).await {
        warn!(error = %e, "requeue failed in run_once");
    }
    result
}