use std::time::Duration;

use backon::{ExponentialBuilder, Retryable};
use tokio::time::Instant;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};

use crate::heads::RepoHeads;
use crate::index::LanguageIndex;
use crate::live::run_live;
use crate::source::{LanguageSource, Page, SourceFuture};

const PAGE_LIMIT: usize = 500;
const RETRY_MIN: Duration = Duration::from_secs(1);
const RETRY_MAX: Duration = Duration::from_secs(60);

pub async fn run(
    source: &dyn LanguageSource,
    index: &LanguageIndex,
    heads: &RepoHeads,
    cancel: &CancellationToken,
) {
    if backfill(source, index, cancel).await.is_none() {
        return;
    }
    info!(repos = index.len(), "cutting over to live language updates");
    run_live(source, index, heads, cancel).await;
}

async fn backfill(
    source: &dyn LanguageSource,
    index: &LanguageIndex,
    cancel: &CancellationToken,
) -> Option<()> {
    let started = Instant::now();
    let (mut cursor, mut pages, mut repos) = (None, 0, 0);
    loop {
        let page = next_page(source, cursor.as_deref(), cancel).await?;
        pages += 1;
        repos += page.repos.len();
        for repo in &page.repos {
            let sizes = repo
                .languages
                .iter()
                .map(|(name, size)| (name.as_str(), *size));
            index.apply(&repo.repo, repo.commit, sizes);
        }
        debug!(
            page = pages,
            repos = page.repos.len(),
            "loaded a page of indexed languages"
        );
        cursor = page.cursor;
        if cursor.is_none() {
            break;
        }
    }
    info!(repos, pages, elapsed = ?started.elapsed(), "backfilled repo languages from the mirror");
    Some(())
}

async fn next_page(
    source: &dyn LanguageSource,
    cursor: Option<&str>,
    cancel: &CancellationToken,
) -> Option<Page> {
    let backoff = ExponentialBuilder::default()
        .with_min_delay(RETRY_MIN)
        .with_max_delay(RETRY_MAX)
        .without_max_times();
    let page: SourceFuture<'_, _> =
        Box::pin((|| source.page(cursor, PAGE_LIMIT)).retry(backoff).notify(
            |error, retry_in| warn!(%error, ?retry_in, "loading indexed languages failed"),
        ));
    cancel.run_until_cancelled(page).await?.ok()
}

#[cfg(test)]
mod tests {
    use tokio::time::sleep;

    use super::*;
    use crate::fake::{Scripted, did, oid};
    use crate::source::{Fetched, IndexedRepo};

    fn page(repos: &[(&str, char, &str)], cursor: Option<&str>) -> Result<Page, String> {
        Ok(Page {
            repos: repos
                .iter()
                .map(|(repo, commit, language)| IndexedRepo {
                    repo: did(repo),
                    commit: oid(*commit),
                    languages: vec![((*language).into(), 1)],
                })
                .collect(),
            cursor: cursor.map(Into::into),
        })
    }

    #[tokio::test(start_paused = true)]
    async fn backfill_follows_the_cursor_and_retries_a_failed_page() {
        let source = Scripted::paging(
            vec![
                page(&[("a", 'a', "Rust")], Some("did:plc:a")),
                Err("mirror returned 502".into()),
                page(&[("b", 'b', "Go")], None),
            ],
            vec![],
        );
        let index = LanguageIndex::new();
        assert!(
            backfill(&source, &index, &CancellationToken::new())
                .await
                .is_some()
        );
        assert_eq!(
            source.cursors(),
            [None, Some("did:plc:a".into()), Some("did:plc:a".into())],
        );
        assert_eq!(index.primary(&did("a")).unwrap().name, "Rust");
        assert_eq!(index.primary(&did("b")).unwrap().name, "Go");
    }

    #[tokio::test(start_paused = true)]
    async fn heads_seen_during_backfill_are_only_fetched_when_they_moved() {
        let source = Scripted::paging(
            vec![page(&[("a", 'a', "Rust"), ("b", 'b', "Go")], None)],
            vec![Fetched::Languages(vec![("Zig".into(), 3)])],
        );
        let (index, heads, cancel) = (
            LanguageIndex::new(),
            RepoHeads::new(),
            CancellationToken::new(),
        );
        heads.observe(did("a"), Some(oid('a')));
        heads.observe(did("b"), Some(oid('f')));

        let stop = cancel.clone();
        tokio::spawn(async move {
            sleep(Duration::from_secs(1)).await;
            stop.cancel();
        });
        run(&source, &index, &heads, &cancel).await;

        assert_eq!(source.asked(), [(did("b"), oid('f'))]);
        assert_eq!(index.commit(&did("a")), Some(oid('a')));
        assert_eq!(index.primary(&did("b")).unwrap().name, "Zig");
    }

    #[tokio::test(start_paused = true)]
    async fn cancelling_during_backfill_stops_without_going_live() {
        let source = Scripted::paging(vec![], vec![]);
        let cancel = CancellationToken::new();
        cancel.cancel();
        assert!(
            backfill(&source, &LanguageIndex::new(), &cancel)
                .await
                .is_none()
        );
    }
}