use std::{
    cmp::Reverse,
    collections::{BTreeSet, BinaryHeap, HashMap},
};

use tokio::task::JoinSet;

use crate::{
    Client, Error, Result,
    domain::{DEFAULT_PAGE_SIZE, HistoryCommit, Page, RepositoryId},
    models,
};
use gitea_openapi::apis;

impl Client {
    pub async fn current_user_repositories_page(
        &self,
        page: i32,
        limit: i32,
    ) -> Result<Page<models::Repository>> {
        validate_page(page, limit)?;
        apis::user_api::user_current_list_repos(&self.configuration(), Some(page), Some(limit))
            .await
            .map(|items| Page::from_items(items, limit))
            .map_err(Error::generated)
    }

    pub async fn owned_repositories_page(
        &self,
        page: i32,
        limit: i32,
    ) -> Result<Page<models::Repository>> {
        let login = self
            .current_user()
            .await?
            .login
            .ok_or_else(|| Error::Generated("The server account has no username.".into()))?;
        let page = self.current_user_repositories_page(page, limit).await?;
        Ok(Page {
            has_more: page.has_more,
            items: page
                .items
                .into_iter()
                .filter(|repository| {
                    repository
                        .owner
                        .as_ref()
                        .and_then(|owner| owner.login.as_deref())
                        == Some(login.as_str())
                })
                .collect(),
        })
    }

    pub async fn branches(
        &self,
        repository: &RepositoryId,
        default_branch: &str,
    ) -> Result<Vec<String>> {
        let configuration = self.configuration();
        let mut branches = Vec::new();
        for page in 1.. {
            let batch = apis::repository_api::repo_list_branches(
                &configuration,
                &repository.owner,
                &repository.repository,
                Some(page),
                Some(100),
            )
            .await
            .map_err(Error::generated)?;
            let done = batch.len() < 100;
            branches.extend(batch.into_iter().filter_map(|branch| branch.name));
            if done {
                break;
            }
        }
        branches.sort_by_key(|branch| (branch != default_branch, branch.to_lowercase()));
        Ok(branches)
    }

    pub async fn repository_contents(
        &self,
        repository: &RepositoryId,
        path: &str,
    ) -> Result<Vec<models::ContentsResponse>> {
        let configuration = self.configuration();
        if path.is_empty() {
            apis::repository_api::repo_get_contents_list(
                &configuration,
                &repository.owner,
                &repository.repository,
                None,
            )
            .await
            .map_err(Error::generated)
        } else {
            apis::repository_api::repo_get_contents_ext(
                &configuration,
                &repository.owner,
                &repository.repository,
                path,
                None,
                None,
            )
            .await
            .map(|contents| contents.dir_contents.unwrap_or_default())
            .map_err(Error::generated)
        }
    }

    pub async fn repository_file(&self, repository: &RepositoryId, path: &str) -> Result<Vec<u8>> {
        apis::repository_api::repo_get_raw_file(
            &self.configuration(),
            &repository.owner,
            &repository.repository,
            path,
            None,
        )
        .await
        .map_err(Error::generated)?
        .bytes()
        .await
        .map(|bytes| bytes.to_vec())
        .map_err(Into::into)
    }

    pub async fn branch_history(
        &self,
        repository: &RepositoryId,
        branch: &str,
        path: Option<&str>,
        pages: u32,
    ) -> Result<Page<HistoryCommit>> {
        self.commits_for_ref(repository, Some(branch), path, pages)
            .await
            .map(|page| Page {
                has_more: page.has_more,
                items: page
                    .items
                    .into_iter()
                    .map(|commit| HistoryCommit {
                        commit,
                        top_lanes: Vec::new(),
                        bottom_lanes: Vec::new(),
                        node_lane: None,
                        top_connections: Vec::new(),
                        bottom_connections: Vec::new(),
                        refs: Vec::new(),
                        branch_starts: Vec::new(),
                    })
                    .collect(),
            })
    }

    pub async fn all_history(
        &self,
        repository: &RepositoryId,
        branches: &[String],
        path: Option<&str>,
        pages: u32,
    ) -> Result<Page<HistoryCommit>> {
        let mut tasks = JoinSet::new();
        for (lane, branch) in branches.iter().cloned().enumerate() {
            let (client, repository) = (self.clone(), repository.clone());
            let path = path.map(str::to_string);
            tasks.spawn(async move {
                let commits = client
                    .commits_for_ref(&repository, Some(&branch), path.as_deref(), pages)
                    .await?;
                Ok::<_, Error>((lane, branch, commits))
            });
        }

        let mut histories = Vec::new();
        while let Some(result) = tasks.join_next().await {
            histories.push(result.map_err(|error| Error::Generated(error.to_string()))??);
        }
        histories.sort_by_key(|(lane, _, _)| *lane);
        let has_more = histories.iter().any(|(_, _, page)| page.has_more);
        let histories = histories
            .into_iter()
            .map(|(lane, branch, page)| (lane, branch, page.items))
            .collect();
        let pull_refs = self.pull_refs(repository).await.unwrap_or_default();
        Ok(Page {
            items: build_graph(histories, pull_refs),
            has_more,
        })
    }

    pub async fn commit(&self, repository: &RepositoryId, sha: &str) -> Result<models::Commit> {
        apis::repository_api::repo_get_single_commit(
            &self.configuration(),
            &repository.owner,
            &repository.repository,
            sha,
            Some(false),
            Some(true),
            Some(true),
        )
        .await
        .map_err(Error::generated)
    }

    pub async fn commit_diff(&self, repository: &RepositoryId, sha: &str) -> Result<String> {
        apis::repository_api::repo_download_commit_diff_or_patch(
            &self.configuration(),
            &repository.owner,
            &repository.repository,
            sha,
            "diff",
        )
        .await
        .map_err(Error::generated)
    }

    async fn pull_refs(&self, repository: &RepositoryId) -> Result<PullMetadata> {
        let mut refs = PullMetadata::default();
        for page in 1.. {
            let batch = self.repository_pulls(repository, "all", page, 100).await?;
            for pull in batch.items {
                let Some(head) = pull.head else { continue };
                let Some(sha) = head.sha else { continue };
                let Some(label) = head.label.or(head.r#ref) else {
                    continue;
                };
                if !label.is_empty() {
                    refs.tips
                        .entry(sha.clone())
                        .or_default()
                        .push(label.clone());
                    if let Some(base) = pull.merge_base.filter(|base| !base.is_empty()) {
                        refs.starts.push(PullBranch {
                            head: sha,
                            base,
                            label,
                        });
                    }
                }
            }
            if !batch.has_more {
                break;
            }
        }
        Ok(refs)
    }

    async fn commits_for_ref(
        &self,
        repository: &RepositoryId,
        branch: Option<&str>,
        path: Option<&str>,
        pages: u32,
    ) -> Result<Page<models::Commit>> {
        let configuration = self.configuration();
        let mut commits = Vec::new();
        let mut has_more = false;
        for page in 1..=pages.max(1) {
            let batch = apis::repository_api::repo_get_all_commits(
                &configuration,
                &repository.owner,
                &repository.repository,
                branch,
                path,
                None,
                None,
                None,
                None,
                Some(false),
                Some(page as i32),
                Some(DEFAULT_PAGE_SIZE),
                None,
            )
            .await
            .map_err(Error::generated)?;
            has_more = batch.len() == DEFAULT_PAGE_SIZE as usize;
            commits.extend(batch);
            if !has_more {
                break;
            }
        }
        Ok(Page {
            items: commits,
            has_more,
        })
    }
}

#[derive(Default)]
struct PullMetadata {
    tips: HashMap<String, Vec<String>>,
    starts: Vec<PullBranch>,
}

struct PullBranch {
    head: String,
    base: String,
    label: String,
}

fn build_graph(
    histories: Vec<(usize, String, Vec<models::Commit>)>,
    extra_refs: PullMetadata,
) -> Vec<HistoryCommit> {
    let mut commits = HashMap::new();
    let mut ranks: HashMap<String, (usize, usize)> = HashMap::new();
    let mut refs: HashMap<String, Vec<String>> = HashMap::new();
    for (branch_index, branch, history) in histories {
        if let Some(sha) = history.iter().find_map(|commit| commit.sha.clone()) {
            refs.entry(sha).or_default().push(branch);
        }
        for (index, commit) in history.into_iter().enumerate() {
            let Some(sha) = commit.sha.clone() else {
                continue;
            };
            ranks
                .entry(sha.clone())
                .and_modify(|rank| *rank = (*rank).min((branch_index, index)))
                .or_insert((branch_index, index));
            commits.entry(sha).or_insert(commit);
        }
    }
    for (sha, labels) in extra_refs.tips {
        let row = refs.entry(sha).or_default();
        for label in labels {
            if !row.contains(&label) {
                row.push(label);
            }
        }
    }
    let mut branch_starts = branch_starts(&commits, extra_refs.starts);

    let mut children = HashMap::new();
    for sha in commits.keys() {
        children.insert(sha.clone(), 0_usize);
    }
    for commit in commits.values() {
        for parent in commit_parents(commit) {
            if let Some(count) = children.get_mut(parent) {
                *count += 1;
            }
        }
    }
    let mut ready = BinaryHeap::new();
    for (sha, count) in &children {
        if *count == 0 {
            ready.push(Reverse((ranks[sha], sha.clone())));
        }
    }
    let mut ordered = Vec::with_capacity(commits.len());
    while let Some(Reverse((_, sha))) = ready.pop() {
        let Some(commit) = commits.remove(&sha) else {
            continue;
        };
        for parent in commit_parents(&commit) {
            if let Some(count) = children.get_mut(parent) {
                *count -= 1;
                if *count == 0 {
                    ready.push(Reverse((ranks[parent], parent.to_string())));
                }
            }
        }
        ordered.push(HistoryCommit {
            commit,
            top_lanes: Vec::new(),
            bottom_lanes: Vec::new(),
            node_lane: None,
            top_connections: Vec::new(),
            bottom_connections: Vec::new(),
            refs: refs.remove(&sha).unwrap_or_default(),
            branch_starts: branch_starts.remove(&sha).unwrap_or_default(),
        });
    }
    layout_graph(&mut ordered);
    ordered
}

fn branch_starts(
    commits: &HashMap<String, models::Commit>,
    branches: Vec<PullBranch>,
) -> HashMap<String, Vec<String>> {
    let mut starts: HashMap<String, Vec<String>> = HashMap::new();
    for branch in branches {
        let Some(sha) = first_commit_after(commits, &branch.head, &branch.base) else {
            continue;
        };
        let labels = starts.entry(sha).or_default();
        if !labels.contains(&branch.label) {
            labels.push(branch.label);
        }
    }
    starts
}

fn first_commit_after(
    commits: &HashMap<String, models::Commit>,
    head: &str,
    base: &str,
) -> Option<String> {
    let mut sha = head;
    for _ in 0..commits.len() {
        let parent = commit_parents(commits.get(sha)?).next()?;
        if parent == base {
            return Some(sha.into());
        }
        sha = parent;
    }
    None
}

fn layout_graph(commits: &mut [HistoryCommit]) {
    let mut lanes: Vec<Option<String>> = Vec::new();
    for row in commits {
        let Some(sha) = row.commit.sha.as_deref() else {
            continue;
        };
        let matching_lanes: Vec<_> = lanes
            .iter()
            .enumerate()
            .filter_map(|(index, next)| (next.as_deref() == Some(sha)).then_some(index))
            .collect();
        let existing_node = matching_lanes.first().copied();
        let node = existing_node.unwrap_or_else(|| allocate_lane(&mut lanes, sha));
        row.top_lanes = lanes
            .iter()
            .enumerate()
            .filter_map(|(index, lane)| lane.as_ref().map(|_| index))
            .filter(|index| existing_node.is_some() || *index != node)
            .collect();
        let mut top_connections = BTreeSet::new();
        for duplicate in matching_lanes.into_iter().skip(1) {
            lanes[duplicate] = None;
            top_connections.insert(duplicate);
        }
        let mut bottom_connections = BTreeSet::new();
        let parents: Vec<_> = commit_parents(&row.commit).map(str::to_string).collect();
        lanes[node] = None;
        for (index, parent) in parents.into_iter().enumerate() {
            if index == 0 {
                lanes[node] = Some(parent);
            } else if let Some(existing) = lanes
                .iter()
                .position(|next| next.as_deref() == Some(&parent))
            {
                if node != existing {
                    bottom_connections.insert(existing);
                }
            } else {
                let branch = allocate_lane(&mut lanes, &parent);
                bottom_connections.insert(branch);
            }
        }
        row.node_lane = Some(node);
        row.top_connections = top_connections.into_iter().collect();
        row.bottom_connections = bottom_connections.into_iter().collect();
        row.bottom_lanes = lanes
            .iter()
            .enumerate()
            .filter_map(|(index, lane)| lane.as_ref().map(|_| index))
            .collect();
    }
}

fn allocate_lane(lanes: &mut Vec<Option<String>>, sha: &str) -> usize {
    if let Some(index) = lanes.iter().position(Option::is_none) {
        lanes[index] = Some(sha.into());
        index
    } else {
        lanes.push(Some(sha.into()));
        lanes.len() - 1
    }
}

fn commit_parents(commit: &models::Commit) -> impl Iterator<Item = &str> {
    commit
        .parents
        .as_deref()
        .unwrap_or_default()
        .iter()
        .filter_map(|parent| parent.sha.as_deref())
}

fn validate_page(page: i32, limit: i32) -> Result<()> {
    if page < 1 || limit < 1 {
        return Err(Error::InvalidInput(
            "page and limit must be positive".into(),
        ));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    fn commit(sha: &str, parents: &[&str]) -> models::Commit {
        models::Commit {
            sha: Some(sha.into()),
            parents: Some(
                parents
                    .iter()
                    .map(|sha| models::CommitMeta {
                        sha: Some((*sha).into()),
                        ..Default::default()
                    })
                    .collect(),
            ),
            ..Default::default()
        }
    }

    #[test]
    fn lays_out_shared_commit_graph() {
        let rows = build_graph(
            vec![
                (
                    0,
                    "main".into(),
                    vec![
                        commit("merge", &["left", "right"]),
                        commit("left", &["base"]),
                        commit("base", &[]),
                    ],
                ),
                (
                    1,
                    "feature".into(),
                    vec![commit("right", &["base"]), commit("base", &[])],
                ),
            ],
            PullMetadata::default(),
        );
        assert_eq!(rows.len(), 4);
        assert!(rows.iter().any(|row| row.refs == ["main"]));
        assert!(rows.iter().any(|row| row.refs == ["feature"]));
        let merge = rows
            .iter()
            .find(|row| row.commit.sha.as_deref() == Some("merge"))
            .unwrap();
        assert_eq!(merge.node_lane, Some(0));
        assert_eq!(merge.bottom_connections, [1]);
        let right = rows
            .iter()
            .find(|row| row.commit.sha.as_deref() == Some("right"))
            .unwrap();
        assert_eq!(right.node_lane, Some(1));
        let base = rows
            .iter()
            .find(|row| row.commit.sha.as_deref() == Some("base"))
            .unwrap();
        assert_eq!(base.node_lane, Some(0));
        assert_eq!(base.top_connections, [1]);
    }

    #[test]
    fn labels_first_commit_of_pull_branch() {
        let rows = build_graph(
            vec![(
                0,
                "main".into(),
                vec![
                    commit("merge", &["main", "head"]),
                    commit("head", &["first"]),
                    commit("first", &["base"]),
                    commit("main", &["base"]),
                    commit("base", &[]),
                ],
            )],
            PullMetadata {
                tips: HashMap::from([("head".into(), vec!["feature".into()])]),
                starts: vec![PullBranch {
                    head: "head".into(),
                    base: "base".into(),
                    label: "feature".into(),
                }],
            },
        );
        let first = rows
            .iter()
            .find(|row| row.commit.sha.as_deref() == Some("first"))
            .unwrap();
        assert_eq!(first.branch_starts, ["feature"]);
        assert!(
            rows.iter()
                .find(|row| row.commit.sha.as_deref() == Some("head"))
                .unwrap()
                .refs
                .contains(&"feature".into())
        );
    }
}