use std::net::SocketAddr;
use std::time::Duration;

use anyhow::Context;
use axum::Router;
use ci::CiService;
use config::scoped::{ScopedConfig, WebhookServiceConfig};
use tokio::net::TcpListener;
use tokio_util::sync::CancellationToken;
use tracing::{info, warn};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    export::telemetry::init_tracing();

    let cfg = WebhookServiceConfig::load().context("loading configuration")?;
    if !cfg.webhook.enabled() {
        info!("webhook is disabled by config; exiting");
        return Ok(());
    }
    info!("connecting to database / redis / storage");
    let state = export::AppState::with_config(cfg.clone().into_app_config())
        .await
        .context("building AppState")?;
    let service = CiService::new(state);
    info!("dependencies ready");

    let cancel = CancellationToken::new();

    let consumer = if cfg.ci.enabled {
        Some(tokio::spawn(consume_events(
            service.clone(),
            cancel.clone(),
        )))
    } else {
        info!("ci is disabled by config; scheduler not started");
        None
    };
    let scheduler = if cfg.ci.enabled {
        Some(tokio::spawn(run_scheduler(
            service.clone(),
            cfg.ci.scheduler_interval_secs,
            cancel.clone(),
        )))
    } else {
        None
    };
    let outbound = tokio::spawn(run_outbound(
        service.clone(),
        cfg.ci.scheduler_interval_secs,
        cancel.clone(),
    ));

    let addr = SocketAddr::new(
        cfg.webhook
            .bind_addr()
            .parse()
            .context("webhook bind addr")?,
        cfg.webhook.port(),
    );
    let listener = TcpListener::bind(addr)
        .await
        .with_context(|| format!("binding webhook {addr}"))?;
    let runner_routes = ci::routes::runner_router().with_state(service.clone());
    let router: Router = Router::new()
        .nest("/api/v1", runner_routes)
        .merge(ci::actions_runtime::router().with_state(service.clone()))
        .merge(export::health::router_with(
            Some(service.state.database.clone()),
            service.state.redis.clone(),
        ));
    info!(%addr, "webhook scheduler listening");

    let server_cancel = cancel.clone();
    axum::serve(listener, router.into_make_service())
        .with_graceful_shutdown(async move { server_cancel.cancelled().await })
        .await
        .context("webhook server error")?;

    cancel.cancel();
    if let Some(consumer) = consumer {
        let _ = consumer.await;
    }
    if let Some(scheduler) = scheduler {
        let _ = scheduler.await;
    }
    let _ = outbound.await;
    info!("webhook shut down");
    Ok(())
}

async fn consume_events(service: CiService, cancel: CancellationToken) {
    use ci::events::{decode_event, publish_event_to_dlq};
    use git::sync::queue::SyncQueue;

    let topic = service.config().ci.event_queue_topic.clone();
    let dlq = service.config().ci.event_dlq_topic.clone();
    // BRPOP blocks longer than the Redis request timeout, so the consumer
    // needs its own dedicated connection (same as the sync worker).
    let queue = match SyncQueue::with_consumer_conns(service.state.redis.clone(), 1).await {
        Ok(queue) => queue,
        Err(error) => {
            warn!(%error, "failed to open ci event consumer connection");
            return;
        }
    };
    let mut attempts: std::collections::HashMap<uuid::Uuid, u32> = std::collections::HashMap::new();
    loop {
        let payload = tokio::select! {
            _ = cancel.cancelled() => break,
            res = queue.consume(&topic, 5, 0) => match res {
                Ok(Some(payload)) => payload,
                Ok(None) => continue,
                Err(error) => {
                    warn!(%error, "event consume failed");
                    tokio::time::sleep(Duration::from_secs(2)).await;
                    continue;
                }
            },
        };
        let event = match decode_event(&payload) {
            Ok(event) => event,
            Err(error) => {
                warn!(%error, "dropping malformed ci event");
                let _ = publish_event_to_dlq(&service.state.redis, &dlq, &payload).await;
                continue;
            }
        };
        let event_id = match &event {
            ci_protocol::events::CiEvent::Push(push) => push.id,
            ci_protocol::events::CiEvent::PullRequest(pr) => pr.id,
            ci_protocol::events::CiEvent::Trigger(event) => event.id,
        };
        match service.handle_event(&event).await {
            Ok(created) => {
                attempts.remove(&event_id);
                if !created.is_empty() {
                    info!(event = %event_id, runs = created.len(), "ci event produced runs");
                }
            }
            Err(error) => {
                let count = attempts.entry(event_id).or_insert(0);
                *count += 1;
                if *count > 5 {
                    attempts.remove(&event_id);
                    warn!(%error, event = %event_id, "ci event failed permanently; sending to DLQ");
                    let _ = publish_event_to_dlq(&service.state.redis, &dlq, &payload).await;
                } else {
                    warn!(%error, event = %event_id, attempt = *count, "ci event failed; retrying");
                    tokio::time::sleep(Duration::from_secs(2 * u64::from(*count))).await;
                    let _ = queue.publish_raw(&topic, None, &payload).await;
                }
            }
        }
    }
    info!("event consumer stopped");
}

async fn run_outbound(service: CiService, interval_secs: u64, cancel: CancellationToken) {
    let interval = Duration::from_secs(interval_secs.clamp(1, 30));
    let mut ticker = tokio::time::interval(interval);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        tokio::select! {
            _ = cancel.cancelled() => break,
            _ = ticker.tick() => {}
        }
        if let Err(error) = service.deliver_pending_repo_webhooks().await {
            warn!(%error, "outbound webhook delivery failed");
        }
    }
    info!("outbound webhook loop stopped");
}

async fn run_scheduler(service: CiService, interval_secs: u64, cancel: CancellationToken) {
    if let Err(error) = service.discover_actions_schedules().await {
        warn!(%error,"initial schedule discovery failed");
    }
    let interval = Duration::from_secs(interval_secs.max(1));
    let mut ticker = tokio::time::interval(interval);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    let mut ticks: u64 = 0;
    loop {
        tokio::select! {
            _ = cancel.cancelled() => break,
            _ = ticker.tick() => {}
        }
        if let Err(error) = service.advance_jobs().await {
            warn!(%error, "advance jobs failed");
        }
        if let Err(error) = service.enforce_concurrency().await {
            warn!(%error, "concurrency enforcement failed");
        }
        if let Err(error) = service.expire_leases().await {
            warn!(%error, "lease expiry failed");
        }
        if let Err(error) = service.apply_cancellations().await {
            warn!(%error, "cancellation sweep failed");
        }
        if let Err(error) = service.finalize_runs().await {
            warn!(%error, "run finalization failed");
        }
        ticks = ticks.wrapping_add(1);
        if let Err(error) = service.run_due_actions_schedules().await {
            warn!(%error,"scheduled workflow execution failed");
        }
        if ticks.is_multiple_of(3600)
            && let Err(error) = service.prune_expired().await
        {
            warn!(%error, "retention prune failed");
        }
        if ticks.is_multiple_of(30)
            && let Err(error) = service.reap_stale_runners().await
        {
            warn!(%error, "stale runner sweep failed");
        }
    }
    info!("scheduler stopped");
}