use stibble_server::{
    api, bmc, irail,
    provider::{FeedFailure, FeedStage},
    quota, state,
    storage::Store,
    villo,
};

use std::sync::Arc;
use std::time::Duration;

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "info".into()),
        )
        .init();
    let database =
        std::env::var("STIBBLE_DB").unwrap_or_else(|_| "/data/stibble.sqlite".into());
    let extension =
        std::env::var("SPATIALITE_EXTENSION").expect("SPATIALITE_EXTENSION is required");
    let store = Store::open(database, extension)
        .await
        .expect("open spatial store");
    let key = std::env::var("BMC_PARTNER_KEY").expect("BMC_PARTNER_KEY is required");
    let base = std::env::var("BMC_BASE")
        .unwrap_or_else(|_| "https://api-management-opendata-production.azure-api.net".into());
    let quota = Arc::new(quota::Quota::new(500, 12000));
    let client = Arc::new(bmc::Client::new(base, key, quota));
    let (train_refresh_tx, train_refresh_rx) = tokio::sync::mpsc::channel(64);
    let app_state = state::AppState::new(store.clone(), train_refresh_tx);

    let maintenance_store = store.clone();
    tokio::spawn(async move {
        loop {
            if let Err(error) = maintenance_store.maintain(chrono::Utc::now()).await {
                tracing::warn!("storage maintenance failed: {error:#}");
            }
            tokio::time::sleep(Duration::from_secs(3600)).await;
        }
    });

    let liveboard_store = store.clone();
    let liveboard_http = app_state.irail_http.clone();
    let liveboard_quota = app_state.irail_quota.clone();
    tokio::spawn(async move {
        if let Err(error) = irail::run_liveboard_worker(
            liveboard_store,
            liveboard_http,
            liveboard_quota,
            train_refresh_rx,
        )
        .await
        {
            tracing::warn!("irail liveboard worker stopped: {error:#}");
        }
    });

    let s = store.clone();
    let c = client.clone();
    tokio::spawn(async move {
        loop {
            let result = match c.fetch_static().await {
                Ok(batch) => s.apply_static(batch).await.map_err(|source| FeedFailure {
                    stage: FeedStage::Commit,
                    source,
                }),
                Err(failure) => Err(failure),
            };
            match result {
                Ok(()) => {
                    tracing::info!("static data refreshed");
                    tokio::time::sleep(Duration::from_secs(24 * 3600)).await;
                }
                Err(failure) => {
                    let message = format!("{:#}", failure.source);
                    if let Err(error) = s
                        .record_feed_failure(
                            "bmc",
                            "static",
                            failure.stage,
                            &message,
                            chrono::Utc::now(),
                        )
                        .await
                    {
                        tracing::warn!("static failure recording failed: {error:#}");
                    }
                    tracing::warn!("static refresh failed: {failure}");
                    tokio::time::sleep(Duration::from_secs(300)).await;
                }
            }
        }
    });

    let s = store.clone();
    let c = client.clone();
    tokio::spawn(async move {
        loop {
            let waiting = match c.fetch_waiting().await {
                Ok(batch) => s.apply_waiting(batch).await.map_err(|source| FeedFailure {
                    stage: FeedStage::Commit,
                    source,
                }),
                Err(failure) => Err(failure),
            };
            match waiting {
                Ok(()) => tracing::info!("waiting data refreshed"),
                Err(failure) => {
                    let message = format!("{:#}", failure.source);
                    if let Err(error) = s
                        .record_feed_failure(
                            "bmc",
                            "waiting",
                            failure.stage,
                            &message,
                            chrono::Utc::now(),
                        )
                        .await
                    {
                        tracing::warn!("waiting failure recording failed: {error:#}");
                    }
                    tracing::warn!("waiting refresh failed: {failure}");
                }
            }
            let positions = match c.fetch_positions().await {
                Ok(batch) => s.apply_positions(batch).await.map_err(|source| FeedFailure {
                    stage: FeedStage::Commit,
                    source,
                }),
                Err(failure) => Err(failure),
            };
            match positions {
                Ok(()) => tracing::info!("positions data refreshed"),
                Err(failure) => {
                    let message = format!("{:#}", failure.source);
                    if let Err(error) = s
                        .record_feed_failure(
                            "bmc",
                            "positions",
                            failure.stage,
                            &message,
                            chrono::Utc::now(),
                        )
                        .await
                    {
                        tracing::warn!("positions failure recording failed: {error:#}");
                    }
                    tracing::warn!("positions refresh failed: {failure}");
                }
            }
            tokio::time::sleep(Duration::from_secs(30)).await;
        }
    });

    let s = store.clone();
    let c = client.clone();
    tokio::spawn(async move {
        loop {
            let result = match c.fetch_disruptions().await {
                Ok(batch) => s.apply_disruptions(batch).await.map_err(|source| FeedFailure {
                    stage: FeedStage::Commit,
                    source,
                }),
                Err(failure) => Err(failure),
            };
            match result {
                Ok(()) => tracing::info!("disruptions data refreshed"),
                Err(failure) => {
                    let message = format!("{:#}", failure.source);
                    if let Err(error) = s
                        .record_feed_failure(
                            "bmc",
                            "disruptions",
                            failure.stage,
                            &message,
                            chrono::Utc::now(),
                        )
                        .await
                    {
                        tracing::warn!("disruptions failure recording failed: {error:#}");
                    }
                    tracing::warn!("disruptions refresh failed: {failure}");
                }
            }
            tokio::time::sleep(Duration::from_secs(300)).await;
        }
    });

    let irail_store = store.clone();
    let irail_http = app_state.irail_http.clone();
    let irail_quota = app_state.irail_quota.clone();
    tokio::spawn(async move {
        loop {
            let result = match irail::refresh_stations(&irail_http, &irail_quota).await {
                Ok(batch) => irail_store
                    .apply_train_stations(batch)
                    .await
                    .map_err(|source| FeedFailure {
                        stage: FeedStage::Commit,
                        source,
                    }),
                Err(failure) => Err(failure),
            };
            match result {
                Ok(()) => {
                    tracing::info!("irail stations refreshed");
                    tokio::time::sleep(Duration::from_secs(24 * 3600)).await;
                }
                Err(failure) => {
                    let message = format!("{:#}", failure.source);
                    if let Err(error) = irail_store
                        .record_feed_failure(
                            "irail",
                            "stations",
                            failure.stage,
                            &message,
                            chrono::Utc::now(),
                        )
                        .await
                    {
                        tracing::warn!("irail stations failure recording failed: {error:#}");
                    }
                    tracing::warn!("irail stations refresh failed: {failure}");
                    tokio::time::sleep(Duration::from_secs(300)).await;
                }
            }
        }
    });

    let villo_state = app_state.clone();
    let villo_store = store.clone();
    tokio::spawn(async move {
        loop {
            let result = match villo::fetch_discovery(
                &villo_state.irail_http,
                &villo_state.villo_quota,
            )
            .await
            {
                Ok(batch) => villo_store
                    .apply_villo_discovery(batch)
                    .await
                    .map_err(|source| FeedFailure {
                        stage: FeedStage::Commit,
                        source,
                    }),
                Err(failure) => Err(failure),
            };
            match result {
                Ok(()) => {
                    tracing::info!("villo discovery refreshed");
                    tokio::time::sleep(Duration::from_secs(24 * 3600)).await;
                }
                Err(failure) => {
                    let message = format!("{:#}", failure.source);
                    if let Err(error) = villo_store
                        .record_feed_failure(
                            "villo",
                            "discovery",
                            failure.stage,
                            &message,
                            chrono::Utc::now(),
                        )
                        .await
                    {
                        tracing::warn!("villo discovery failure recording failed: {error:#}");
                    }
                    tracing::warn!("villo discovery refresh failed: {failure}");
                    tokio::time::sleep(Duration::from_secs(300)).await;
                }
            }
        }
    });

    let villo_state = app_state.clone();
    let villo_store = store.clone();
    tokio::spawn(async move {
        loop {
            let status_url = match villo_store.villo_status_url().await {
                Ok(Some(status_url)) => status_url,
                Ok(None) => {
                    tokio::time::sleep(Duration::from_secs(60)).await;
                    continue;
                }
                Err(source) => {
                    tracing::warn!("villo status URL read failed: {source:#}");
                    tokio::time::sleep(Duration::from_secs(60)).await;
                    continue;
                }
            };
            let result = match villo::fetch_status(
                &villo_state.irail_http,
                &villo_state.villo_quota,
                &status_url,
            )
            .await
            {
                Ok(batch) => villo_store
                    .apply_villo_status(batch)
                    .await
                    .map_err(|source| FeedFailure {
                        stage: FeedStage::Commit,
                        source,
                    }),
                Err(failure) => Err(failure),
            };
            match result {
                Ok(()) => tracing::info!("villo status refreshed"),
                Err(failure) => {
                    let message = format!("{:#}", failure.source);
                    if let Err(error) = villo_store
                        .record_feed_failure(
                            "villo",
                            "status",
                            failure.stage,
                            &message,
                            chrono::Utc::now(),
                        )
                        .await
                    {
                        tracing::warn!("villo status failure recording failed: {error:#}");
                    }
                    tracing::warn!("villo status refresh failed: {failure}");
                }
            }
            tokio::time::sleep(Duration::from_secs(60)).await;
        }
    });

    let addr = std::env::var("BIND_ADDR").unwrap_or_else(|_| "0.0.0.0:8080".into());
    let listener = tokio::net::TcpListener::bind(&addr).await.expect("bind");
    tracing::info!("listening on {addr}");
    axum::serve(listener, api::router(app_state.clone()))
        .await
        .expect("serve");
}