use axum::extract::State;
use axum::response::Response;
use futures::future::OptionFuture;
use futures::stream::{self, StreamExt, TryStreamExt};
use lexicons::org_tangled::feed::list_comments::ListComments;
use serde::Deserialize;

use bobbin_edge_index::{
    EdgeItem, EdgePage, EdgeStore, PageCursor, PageLimit, PageOffset, PageToken, SortDir,
};
use bobbin_types::edges::editor_in;
use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi};
use bobbin_types::org_tangled::actor::{ProfileViewBasic, ProfileViewDetailed, ViewerState};
use bobbin_types::org_tangled::embed::commit::View as CommitEmbedView;
use bobbin_types::org_tangled::feed::CommentView;
use bobbin_types::org_tangled::feed::comment::{Comment, CommentRecord};
use bobbin_types::org_tangled::feed::get_timeline::{
    FollowEvent, RepoEvent, StarEvent, TimelineItem, TimelineItemEvent,
};
use bobbin_types::org_tangled::feed::star::{Star, StarRecord, StarSubject};
use bobbin_types::org_tangled::graph::follow::{Follow, FollowRecord};
use bobbin_types::org_tangled::repo::manifest::{Manifest, ManifestRecord, ManifestSource};
use jacquard_axum::service_auth::ExtractOptionalServiceAuth;
use jacquard_common::types::string::{AtUri, Datetime, Did, Handle};
use jacquard_common::xrpc::XrpcResp;
use jacquard_common::{DefaultStr, IntoStatic as _};

use crate::view::{avatar_cid, build_repo_view_basic, build_repo_view_basic_from_record};
use crate::{
    AppState, HasSubject, HitProvenance, RecordView, SubjectQuery, XrpcError, XrpcQuery, fetch,
    hydrate_record_stream, json_stream, paged_tail, parse_cursor, parse_limit, parse_start,
    parse_subject,
};

const REPO_NSID: &str = "org.tangled.repo.manifest";
const STAR_NSID: &str = "org.tangled.feed.star";
const FOLLOW_NSID: &str = "org.tangled.graph.follow";
const STAR_BY_NSID: &str = "org.tangled.feed.star.by";
const FOLLOW_BY_NSID: &str = "org.tangled.graph.follow.by";

const HYDRATE_CONCURRENCY: usize = 8;

pub(crate) async fn list_comments(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<ListComments>,
) -> Result<Response, XrpcError> {
    let subject = parse_subject(&SubjectQuery::Uri(q.subject), CommentRecord::SHAPE)?;
    let offset = q
        .offset
        .map(|offset| {
            u64::try_from(offset)
                .map(PageOffset::new)
                .map_err(|_| XrpcError::InvalidParams("offset must not be negative".into()))
        })
        .transpose()?;
    let start = parse_start(q.cursor.as_deref(), offset)?;
    let limit = parse_limit(
        q.limit
            .map(|limit| {
                u32::try_from(limit).map_err(|_| {
                    XrpcError::InvalidParams("limit must be between 1 and 1000".into())
                })
            })
            .transpose()?,
    )?;
    let dir = match q.order.as_deref() {
        Some("asc") => SortDir::Asc,
        _ => SortDir::Desc,
    };

    let nsid = nsid_static(CommentRecord::NSID);
    let page = state
        .edges
        .list(&EdgeKey::new(nsid.clone(), subject), start, limit, dir);
    let permit = state.heavy_permit()?;
    let owned = state.clone();
    let items = hydrate_record_stream::<Comment<DefaultStr>>(
        &state,
        nsid,
        page.items,
        HitProvenance::Indexed,
    )
    .map_ok(move |view| {
        let state = owned.clone();
        async move { Ok(comment_view(&state, view).await) }
    })
    .try_buffered(HYDRATE_CONCURRENCY)
    .try_filter_map(|view| async move { Ok(view) });

    Ok(json_stream::<CommentView<DefaultStr>, _>(
        "items",
        items,
        paged_tail(page.next.map(PageToken::encode_token), page.total),
        permit,
    ))
}

async fn comment_view(
    state: &AppState,
    view: RecordView<Comment<DefaultStr>>,
) -> Option<CommentView<DefaultStr>> {
    let RecordView { uri, cid, value } = view;
    let embed = OptionFuture::from(value.embed.map(|commit| async {
        let change = state
            .change_ids
            .get(
                state.mirror.as_deref(),
                &commit.repo,
                commit.commit.oid.as_ref(),
            )
            .await;
        CommitEmbedView {
            change,
            commit: commit.commit.oid,
            repo: commit.repo,
            extra_data: None,
        }
    }))
    .await;
    Some(CommentView {
        body: value.body,
        cid: cid?,
        created_at: value.created_at,
        editor: editor_in(value.extra_data.as_ref()),
        embed,
        reply_to: value.reply_to,
        subject: value.subject,
        uri,
        extra_data: None,
    })
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct GetTimelineQuery {
    #[serde(default)]
    following_only: bool,
    limit: Option<u32>,
    cursor: Option<String>,
}

#[axum::debug_handler]
pub(crate) async fn get_timeline(
    State(state): State<AppState>,
    ExtractOptionalServiceAuth(auth): ExtractOptionalServiceAuth,
    XrpcQuery(q): XrpcQuery<GetTimelineQuery>,
) -> Result<Response, XrpcError> {
    let limit = parse_limit(q.limit)?;
    let cursor = parse_cursor(q.cursor.as_deref())?;
    let permit = state.heavy_permit()?;
    let viewer = auth.as_ref().map(|auth| auth.did().into_static());

    // Prepare timeline skeleton
    let page = if q.following_only {
        // Following feed: the viewer's followed set, then a time-ordered k-way merge.
        let viewer = viewer.as_ref().ok_or_else(|| {
            XrpcError::AuthRequired("followingOnly requires an authenticated viewer".into())
        })?;
        let followed = followed_dids(&state, &viewer).await;
        following_skeleton(&state, &followed, cursor, limit)
    } else {
        global_skeleton(&state, cursor, limit)
    };

    // Hydrate skeleton
    let hydrated = stream::iter(
        page.items
            .iter()
            .map(|it| hydrate_item(&state, viewer.as_ref(), it))
            .collect::<Vec<_>>(),
    )
    .buffered(HYDRATE_CONCURRENCY)
    .collect::<Vec<_>>()
    .await;

    let mut feed: Vec<TimelineItem<DefaultStr>> = Vec::with_capacity(hydrated.len());
    for (item, r) in page.items.iter().zip(hydrated) {
        match r {
            Ok(item) => feed.push(item),
            Err(e) => {
                tracing::warn!(uri = %item.uri, error = %e, "skipping timeline item, hydration failed")
            }
        }
    }

    let items = stream::iter(feed.into_iter().map(Ok::<_, XrpcError>));
    Ok(json_stream::<TimelineItem<DefaultStr>, _>(
        "feed",
        items,
        paged_tail(page.next.map(PageToken::encode_token), page.total),
        permit,
    ))
}

async fn followed_dids(state: &AppState, viewer: &Did<DefaultStr>) -> Vec<Did<DefaultStr>> {
    let key = EdgeKey::new(nsid_static(FOLLOW_BY_NSID), SubjectRef::Did(viewer.clone()));
    state
        .edges
        .sources_for(&key)
        .iter()
        .filter_map(follow_subject)
        .collect()
}

// follow records are keyed by the followed DID
fn follow_subject(uri: &AtUri<DefaultStr>) -> Option<Did<DefaultStr>> {
    Did::new_owned(uri.rkey()?.as_ref()).ok()
}

fn following_skeleton(
    state: &AppState,
    followed: &[Did<DefaultStr>],
    cursor: PageCursor,
    limit: PageLimit,
) -> EdgePage {
    let keys: Vec<EdgeKey> = followed
        .iter()
        .flat_map(|u| {
            [REPO_NSID, STAR_BY_NSID, FOLLOW_BY_NSID]
                .map(|kind| EdgeKey::new(nsid_static(kind), SubjectRef::Did(u.clone())))
        })
        .collect();
    state.edges.list_multi(&keys, cursor, limit, SortDir::Desc)
}

fn global_skeleton(state: &AppState, cursor: PageCursor, limit: PageLimit) -> EdgePage {
    let keys = [REPO_NSID, STAR_NSID, FOLLOW_NSID]
        .map(|kind| EdgeKey::new(nsid_static(kind), SubjectRef::Global));
    state.edges.list_multi(&keys, cursor, limit, SortDir::Desc)
}

async fn hydrate_item(
    state: &AppState,
    viewer: Option<&Did<DefaultStr>>,
    item: &EdgeItem,
) -> Result<TimelineItem<DefaultStr>, XrpcError> {
    let collection = item
        .uri
        .collection()
        .ok_or_else(|| XrpcError::InvalidParams("timeline uri missing collection".into()))?;
    let (event, event_at) = match collection.as_ref() {
        REPO_NSID => hydrate_repo_event(state, viewer, &item.uri).await?,
        STAR_NSID => hydrate_star_event(state, viewer, &item.uri).await?,
        FOLLOW_NSID => hydrate_follow_event(state, viewer, &item.uri).await?,
        other => {
            return Err(XrpcError::InvalidRecord(format!(
                "unexpected timeline collection: {other}"
            )));
        }
    };
    Ok(TimelineItem::new().event(event).event_at(event_at).build())
}

async fn hydrate_repo_event(
    state: &AppState,
    viewer: Option<&Did<DefaultStr>>,
    uri: &AtUri<DefaultStr>,
) -> Result<(TimelineItemEvent<DefaultStr>, Datetime), XrpcError> {
    let (body, repo) = fetch::<ManifestRecord, Manifest<DefaultStr>>(state, uri).await?;
    let repo_did = owner_did_from_aturi(uri)
        .ok_or_else(|| XrpcError::InvalidParams("repo uri authority must be a did".into()))?;
    let view = build_repo_view_basic_from_record(state, viewer, &repo, repo_did).await?;
    let source = match &repo.source {
        Some(ManifestSource::SourceTangled(upstream)) => {
            build_repo_view_basic(state, viewer, &upstream.did)
                .await
                .ok()
        }
        _ => None,
    };
    let event_at = repo.created_at.clone();
    let ev = RepoEvent::new()
        .uri(body.uri.clone())
        .cid(body.cid.clone())
        .repo(view)
        .source(source)
        .build();
    Ok((TimelineItemEvent::RepoEvent(Box::new(ev)), event_at))
}

async fn hydrate_star_event(
    state: &AppState,
    viewer: Option<&Did<DefaultStr>>,
    uri: &AtUri<DefaultStr>,
) -> Result<(TimelineItemEvent<DefaultStr>, Datetime), XrpcError> {
    let (body, star) = fetch::<StarRecord, Star<DefaultStr>>(state, uri).await?;
    let starrer = owner_did_from_aturi(uri)
        .ok_or_else(|| XrpcError::InvalidParams("star uri authority must be a did".into()))?;
    let actor = build_profile_basic(state, viewer, &starrer).await;
    let repo_did = match &star.subject {
        StarSubject::Repo(r) => r.identity.clone(),
        StarSubject::Record(_) => {
            return Err(XrpcError::InvalidRecord(
                "timeline star subject is not a repo".into(),
            ));
        }
    };
    let repo = build_repo_view_basic(state, viewer, &repo_did).await?;
    let event_at = star.created_at.clone();
    let ev = StarEvent::new()
        .uri(body.uri.clone())
        .cid(body.cid.clone())
        .actor(actor)
        .repo(repo)
        .build();
    Ok((TimelineItemEvent::StarEvent(Box::new(ev)), event_at))
}

async fn hydrate_follow_event(
    state: &AppState,
    viewer: Option<&Did<DefaultStr>>,
    uri: &AtUri<DefaultStr>,
) -> Result<(TimelineItemEvent<DefaultStr>, Datetime), XrpcError> {
    let (body, follow) = fetch::<FollowRecord, Follow<DefaultStr>>(state, uri).await?;
    let follower = owner_did_from_aturi(uri)
        .ok_or_else(|| XrpcError::InvalidParams("follow uri authority must be a did".into()))?;
    let actor = build_profile_basic(state, viewer, &follower).await;
    let subject = follow_subject(uri)
        .ok_or_else(|| XrpcError::InvalidParams("follow rkey must be a did".into()))?;
    let subject = build_profile_detailed(state, viewer, &subject).await;
    let event_at = follow.created_at.clone();
    let ev = FollowEvent::new()
        .uri(body.uri.clone())
        .cid(body.cid.clone())
        .actor(actor)
        .subject(subject)
        .build();
    Ok((TimelineItemEvent::FollowEvent(Box::new(ev)), event_at))
}

async fn build_profile_basic(
    state: &AppState,
    viewer: Option<&Did<DefaultStr>>,
    did: &Did<DefaultStr>,
) -> ProfileViewBasic<DefaultStr> {
    let handle = resolve_handle(state, did);
    let avatar = None; // state.avatar.as_ref().and_then(|s| s.url(did));
    let avatar_cid = avatar_cid(state, did);
    let viewer_state = build_actor_viewer_state(&state.edges, viewer, did);
    ProfileViewBasic::new()
        .did(did.clone())
        .handle(handle)
        .maybe_avatar(avatar)
        .maybe_avatar_cid(avatar_cid)
        .maybe_viewer(viewer_state)
        .build()
}

async fn build_profile_detailed(
    state: &AppState,
    viewer: Option<&Did<DefaultStr>>,
    did: &Did<DefaultStr>,
) -> ProfileViewDetailed<DefaultStr> {
    let handle = resolve_handle(state, did);
    let avatar = None; // state.avatar.as_ref().and_then(|s| s.url(did));
    let avatar_cid = avatar_cid(state, did);
    let viewer_state = build_actor_viewer_state(&state.edges, viewer, did);
    let followers = state.edges.count(&EdgeKey::new(
        nsid_static(FOLLOW_NSID),
        SubjectRef::Did(did.clone()),
    )) as i64;
    let follows = state.edges.count(&EdgeKey::new(
        nsid_static(FOLLOW_BY_NSID),
        SubjectRef::Did(did.clone()),
    )) as i64;
    ProfileViewDetailed::new()
        .did(did.clone())
        .handle(handle)
        .followers_count(followers)
        .follows_count(follows)
        .maybe_avatar(avatar)
        .maybe_avatar_cid(avatar_cid)
        .maybe_viewer(viewer_state)
        .build()
}

fn build_actor_viewer_state(
    edges: &EdgeStore,
    viewer: Option<&Did<DefaultStr>>,
    subject: &Did<DefaultStr>,
) -> Option<ViewerState<DefaultStr>> {
    let viewer = viewer?;
    if viewer == subject {
        return None;
    }
    let following = edges.viewer_source(
        &EdgeKey::new(nsid_static(FOLLOW_NSID), SubjectRef::Did(subject.clone())),
        viewer.as_str(),
    );
    let followed_by = edges.viewer_source(
        &EdgeKey::new(nsid_static(FOLLOW_NSID), SubjectRef::Did(viewer.clone())),
        subject.as_str(),
    );
    (following.is_some() || followed_by.is_some()).then(|| ViewerState {
        following,
        followed_by,
        ..Default::default()
    })
}

fn resolve_handle(state: &AppState, did: &Did<DefaultStr>) -> Handle<DefaultStr> {
    state
        .identity
        .get_by_did(did)
        .ok()
        .map(|doc| doc.handle)
        .unwrap_or_else(|| {
            Handle::new_static("handle.invalid").expect("handle.invalid is a valid handle")
        })
}