use std::time::{Duration, SystemTime, UNIX_EPOCH};

use axum::{Json, extract::State};
use bobbin_types::sh_tangled::actor::get_trending::GetTrendingOutput;
use futures::stream::{self, StreamExt};
use jacquard_common::types::did::Did;
use tokio::time::Instant;

use crate::{
    AppState, FETCH_CONCURRENCY, XrpcError,
    view::{build_profile_detailed, build_repo_view_basic},
};

const OUTPUT_LIMIT: usize = 4;
// +1 are fallbacks in case a returned person is the viewer or a repo isnt resolvable
const PEOPLE_QUERY_LIMIT: usize = OUTPUT_LIMIT + 1;
const REPO_QUERY_LIMIT: usize = OUTPUT_LIMIT + 1;
const TRENDING_WINDOW: Duration = Duration::from_secs(7 * 24 * 60 * 60);
const CACHE_TTL: Duration = Duration::from_secs(60 * 60);

#[derive(Clone, Default)]
pub(crate) struct TrendingCandidates {
    people: Vec<Did>,
    repos: Vec<Did>,
}

#[axum::debug_handler]
pub(crate) async fn get_trending(
    State(state): State<AppState>,
) -> Result<Json<GetTrendingOutput>, XrpcError> {
    let permit = state.heavy_permit()?;

    let candidates = {
        let mut cached = state.trending_cache.lock().await;
        let now = Instant::now();
        if cached
            .as_ref()
            .is_none_or(|(refreshed_at, _)| now - *refreshed_at >= CACHE_TTL)
        {
            let candidates = TrendingCandidates {
                people: state.edges.most_followed(PEOPLE_QUERY_LIMIT),
                repos: state
                    .edges
                    .top_starred_since(trending_cutoff_micros(), REPO_QUERY_LIMIT),
            };
            if state.coverage.snapshot().is_ready() {
                *cached = Some((now, candidates.clone()));
            }
            candidates
        } else {
            cached
                .as_ref()
                .expect("fresh trending cache entry")
                .1
                .clone()
        }
    };

    let people = stream::iter(candidates.people)
        .map({
            let state = &state;
            move |did| async move { build_profile_detailed(state, None, &did).await }
        })
        .buffered(FETCH_CONCURRENCY)
        .collect()
        .await;

    let hydrated = stream::iter(candidates.repos)
        .map({
            let state = &state;
            move |did| async move {
                let result = build_repo_view_basic(state, None, &did).await;
                (did, result)
            }
        })
        .buffered(FETCH_CONCURRENCY)
        .collect::<Vec<_>>()
        .await;
    let repos = hydrated
        .into_iter()
        .filter_map(|(did, result)| match result {
            Ok(repo) => Some(repo),
            Err(error) => {
                tracing::warn!(repo_did = %did, %error, "skipping trending repo, hydration failed");
                None
            }
        })
        .take(OUTPUT_LIMIT)
        .collect();

    drop(permit);
    Ok(Json(GetTrendingOutput {
        people,
        repos,
        extra_data: None,
    }))
}

fn trending_cutoff_micros() -> u64 {
    let elapsed = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .saturating_sub(TRENDING_WINDOW);
    elapsed
        .as_secs()
        .saturating_mul(1_000_000)
        .saturating_add(elapsed.subsec_micros().into())
}

#[cfg(test)]
mod tests {
    use super::*;
    use bobbin_edge_index::EdgeStore;
    use bobbin_runtime::RuntimeHasher;
    use bobbin_types::{
        edges::Edge,
        ids::{SubjectRef, nsid_static},
    };
    use jacquard_common::types::string::AtUri;

    fn did(value: &str) -> Did {
        Did::new_owned(value).unwrap()
    }

    #[test]
    fn trending_candidates_are_ranked_from_the_edge_index() {
        let edges = EdgeStore::new(RuntimeHasher::default());
        let alpha = did("did:plc:alpha");
        let beta = did("did:plc:beta");
        let compiler = did("did:plc:compiler");
        let editor = did("did:plc:editor");
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_micros() as u64;

        [
            ("did:plc:a", "a1", alpha.clone()),
            ("did:plc:b", "a2", alpha.clone()),
            ("did:plc:c", "b1", beta.clone()),
        ]
        .into_iter()
        .for_each(|(author, rkey, subject)| {
            edges.add(Edge {
                kind: nsid_static("org.tangled.graph.follow"),
                subject: SubjectRef::Did(subject),
                source: AtUri::new_owned(format!("at://{author}/org.tangled.graph.follow/{rkey}"))
                    .unwrap(),
                sort_micros: now,
            });
        });
        [
            ("did:plc:a", "c1", compiler.clone(), now),
            ("did:plc:b", "c2", compiler.clone(), now),
            ("did:plc:c", "e1", editor.clone(), now),
        ]
        .into_iter()
        .for_each(|(author, rkey, subject, sort_micros)| {
            edges.add(Edge {
                kind: nsid_static("org.tangled.feed.star"),
                subject: SubjectRef::Did(subject),
                source: AtUri::new_owned(format!("at://{author}/org.tangled.feed.star/{rkey}"))
                    .unwrap(),
                sort_micros,
            });
        });

        let candidates = TrendingCandidates {
            people: edges.most_followed(PEOPLE_QUERY_LIMIT),
            repos: edges.top_starred_since(trending_cutoff_micros(), REPO_QUERY_LIMIT),
        };

        assert_eq!(candidates.people, vec![alpha, beta]);
        assert_eq!(candidates.repos, vec![compiler, editor]);
    }
}