use jacquard::IntoStatic;
use jacquard::deps::smol_str::SmolStr;
use jacquard::types::{
    aturi::AtUri, cid::Cid, did::Did, handle::Handle, nsid::Nsid, string::Datetime, string::Rkey,
};
use jacquard::{BosStr, Data, DefaultStr};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, IntoStatic)]
#[serde(bound(deserialize = "S: Deserialize<'de> + BosStr"))]
pub struct EnrichedVideo<S: BosStr = DefaultStr> {
    pub at_uri: AtUri<S>,
    pub cid: Option<Cid<S>>,
    pub title: S,
    pub duration_ns: u64,
    pub created_at: Datetime,
    pub creator: ProfileInfo<S>,
    pub thumbnail_url: Option<S>,
    pub livestream_uri: Option<AtUri<S>>,
    pub speakers: Vec<SpeakerInfo<S>>,
    pub talk: Option<vodplace_api::tv_ionosphere::talk::Talk<S>>,
}

#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, IntoStatic)]
#[serde(bound(deserialize = "S: Deserialize<'de> + BosStr"))]
pub struct ProfileInfo<S: BosStr = DefaultStr> {
    pub did: Did<S>,
    pub handle: Handle<S>,
    pub display_name: Option<S>,
    pub avatar_url: Option<S>,
}

#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, IntoStatic)]
#[serde(bound(deserialize = "S: Deserialize<'de> + BosStr"))]
pub struct VideoMetadata<S: BosStr = DefaultStr> {
    pub at_uri: AtUri<S>,
    pub title: S,
    pub duration_ns: u64,
    pub created_at: Datetime,
    pub creator: ProfileInfo<S>,
    pub thumbnail_url: Option<S>,
    pub livestream_uri: Option<AtUri<S>>,
    pub speakers: Vec<SpeakerInfo<S>>,
    pub talk: Option<vodplace_api::tv_ionosphere::talk::Talk<S>>,
    pub talk_uri: Option<AtUri<S>>,
}

#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, IntoStatic)]
#[serde(bound(deserialize = "S: Deserialize<'de> + BosStr"))]
pub struct SpeakerInfo<S: BosStr = DefaultStr> {
    pub name: S,
    pub did: Option<Did<S>>,
    pub handle: Option<Handle<S>>,
    pub avatar_url: Option<S>,
}

/// A single ionosphere comment, ready for client-side consumption.
///
/// Returned by `get_comments` as a flat list. The client builds the reply
/// tree and maps byte ranges to word indices using `build_comment_tree`.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
pub struct CommentData {
    pub uri: AtUri,
    pub text: SmolStr,
    pub author_did: Did,
    pub author_handle: Handle,
    pub created_at: Datetime,
    /// Byte range anchoring this comment to a substring of the transcript text.
    /// `None` means the comment is not anchored to any specific transcript position.
    pub anchor: Option<ByteRangeData>,
    /// The AT URI this comment is directed at (talk, transcript, or parent comment).
    pub subject_uri: AtUri,
    pub facets: Option<Vec<crate::chat::FacetData>>,
}

/// A validated byte range from a `tv.ionosphere.comment` anchor field.
///
/// Negative or inverted i64 values from the source lexicon are rejected during
/// conversion and the containing comment is treated as non-anchored.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
pub struct ByteRangeData {
    pub byte_start: usize,
    pub byte_end: usize,
}

const _: () = {
    fn _assert_server_fn_compatible()
    where
        Vec<EnrichedVideo>: serde::Serialize + serde::de::DeserializeOwned,
        VideoMetadata: serde::Serialize + serde::de::DeserializeOwned,
        Vec<CommentData>: serde::Serialize + serde::de::DeserializeOwned,
    {
    }
};

#[allow(dead_code)]
pub const CONSTELLATION_URL: &str = "https://constellation.microcosm.blue";

#[allow(dead_code)]
#[derive(Serialize, Deserialize, Debug, Clone, jacquard_derive::XrpcRequest)]
#[xrpc(
    nsid = "blue.microcosm.links.getBacklinks",
    method = Query,
    output = GetBacklinksResponse,
)]
#[serde(
    rename_all = "camelCase",
    bound(deserialize = "S: Deserialize<'de> + BosStr")
)]
pub struct GetBacklinksQuery<S: BosStr = DefaultStr> {
    pub subject: S,
    pub source: S,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub limit: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cursor: Option<S>,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub did: Vec<Did<S>>,
    /// Allow returning links in reverse order (default: false)
    #[serde(default)]
    pub reverse: bool,
}

#[allow(dead_code)]
#[derive(Serialize, Deserialize, Debug, Clone, IntoStatic)]
#[serde(bound(deserialize = "S: Deserialize<'de> + BosStr"))]
pub struct GetBacklinksResponse<S: BosStr = DefaultStr> {
    pub total: u32,
    pub records: Vec<BacklinkRecord<S>>,
    pub cursor: Option<S>,
}

#[allow(dead_code)]
#[derive(Serialize, Deserialize, Debug, Clone, IntoStatic)]
#[serde(bound(deserialize = "S: Deserialize<'de> + BosStr"))]
pub struct BacklinkRecord<S: BosStr = DefaultStr> {
    pub did: Did<S>,
    pub collection: Nsid<S>,
    pub rkey: Rkey<S>,
}

// --- Slingshot hydration proxy types ---

#[allow(dead_code)]
pub const SLINGSHOT_URL: &str = "https://slingshot.microcosm.blue";

/// User-Agent header value for outbound constellation/slingshot requests.
#[cfg(feature = "server")]
const USER_AGENT: &str = "vodplace/0.1.0";

/// User-Agent header components for xrpc requests.
#[cfg(feature = "server")]
const fn user_agent_header() -> (axum::http::HeaderName, axum::http::HeaderValue) {
    (
        axum::http::HeaderName::from_static("user-agent"),
        axum::http::HeaderValue::from_static(USER_AGENT),
    )
}

/// Request body for slingshot's hydrateQueryResponse endpoint.
#[derive(Serialize, Deserialize, Debug, Clone, jacquard_derive::XrpcRequest)]
#[xrpc(
    nsid = "com.bad-example.proxy.hydrateQuery",
    method = Procedure,
    output = HydrateQueryResponse,
)]
#[serde(
    rename_all = "camelCase",
    bound(deserialize = "S: Deserialize<'de> + BosStr")
)]
pub struct HydrateQueryRequest<S: BosStr = DefaultStr> {
    pub xrpc: S,
    pub atproto_proxy: S,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub authorization: Option<S>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub atproto_accept_labelers: Option<S>,
    pub params: Data<S>,
    pub hydration_sources: Vec<HydrationSource<S>>,
}

#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(bound(deserialize = "S: Deserialize<'de> + BosStr"))]
pub struct HydrationSource<S: BosStr = DefaultStr> {
    pub path: S,
    pub shape: Shape,
}

#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(rename_all = "kebab-case")]
pub enum Shape {
    StrongRef,
    AtUri,
    AtUriParts,
    Did,
    Handle,
    AtIdentifier,
}

/// Response from slingshot's hydrateQueryResponse endpoint.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(bound(deserialize = "S: Deserialize<'de> + BosStr"))]
pub struct HydrateQueryResponse<S: BosStr = DefaultStr> {
    pub output: Data<S>,
    #[serde(default)]
    pub records: BTreeMap<SmolStr, HydratedRecord<S>>,
    #[serde(default)]
    pub identifiers: BTreeMap<SmolStr, HydratedIdentifier<S>>,
}

#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(tag = "status", bound(deserialize = "S: Deserialize<'de> + BosStr"))]
pub enum HydratedRecord<S: BosStr = DefaultStr> {
    #[serde(rename = "found")]
    Found {
        uri: S,
        #[serde(skip_serializing_if = "Option::is_none")]
        cid: Option<Cid<S>>,
        value: Data<S>,
    },
    #[serde(rename = "pending")]
    Pending {
        reason: S,
        #[serde(rename = "followUp")]
        follow_up: S,
    },
    #[serde(rename = "error")]
    Error {
        reason: S,
        #[serde(rename = "shouldRetry")]
        should_retry: bool,
        #[serde(rename = "followUp")]
        follow_up: S,
    },
}

#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(tag = "status", bound(deserialize = "S: Deserialize<'de> + BosStr"))]
pub enum HydratedIdentifier<S: BosStr = DefaultStr> {
    #[serde(rename = "found")]
    Found {
        did: Did<S>,
        handle: Handle<S>,
        pds: S,
        signing_key: S,
    },
    #[serde(rename = "pending")]
    Pending {
        reason: S,
        #[serde(rename = "followUp")]
        follow_up: S,
    },
    #[serde(rename = "error")]
    Error {
        reason: S,
        #[serde(rename = "shouldRetry")]
        should_retry: bool,
        #[serde(rename = "followUp")]
        follow_up: S,
    },
}

#[cfg(any(test, feature = "server"))]
pub fn thumbnail_cdn_url(creator_did: &str, blob_cid: &str) -> jacquard::deps::smol_str::SmolStr {
    jacquard::deps::smol_str::format_smolstr!(
        "https://cdn.bsky.app/img/feed_thumbnail/plain/{}/{}@jpeg",
        creator_did,
        blob_cid
    )
}

// --- Server-only types and implementation ---

#[cfg(feature = "server")]
use {
    crate::lexicons::Video,
    crate::player::uri::playlist_url_for_at_uri,
    futures::future::join_all,
    jacquard::client::{AgentSessionExt, BasicClient},
    jacquard::deps::smol_str::{ToSmolStr, format_smolstr},
    jacquard::prelude::XrpcClient,
    jacquard::types::collection::Collection,
    jacquard::types::value::from_data,
    jacquard::xrpc::XrpcExt,
    jacquard_common::deps::fluent_uri::Uri,
    jacquard_common::types::ident::AtIdentifier,
    mini_moka_wasm::sync::Cache as MokaCache,
    std::sync::Arc,
    std::sync::OnceLock,
    std::time::Duration,
    tracing::warn,
    vodplace_api::app_bsky::actor::get_profile::GetProfile,
    vodplace_api::com_atproto::repo::list_records::ListRecords,
    vodplace_api::com_atproto::sync::get_repo::GetRepo,
    vodplace_api::com_atproto::sync::list_repos_by_collection::ListReposByCollection,
    vodplace_api::place_stream::livestream::Livestream,
};

use std::collections::BTreeMap;
use std::collections::HashMap;
#[cfg(feature = "server")]
#[derive(Debug, thiserror::Error)]
enum ThumbnailError {
    #[error("ffmpeg: {0}")]
    Ffmpeg(#[from] ffmpeg_next::Error),
    #[error("no video stream found")]
    NoVideoStream,
    #[error("no frames decoded")]
    NoFrames,
    #[error("image encoding failed: {0}")]
    Image(#[from] image::ImageError),
    #[error("pixel data invalid")]
    InvalidPixelData,
}

#[cfg(feature = "server")]
static CATALOG_CACHE: OnceLock<MokaCache<(), Vec<EnrichedVideo>>> = OnceLock::new();
#[cfg(feature = "server")]
static THUMBNAIL_CACHE: OnceLock<MokaCache<AtUri, bytes::Bytes>> = OnceLock::new();
#[cfg(feature = "server")]
static IONOSPHERE_CACHE: OnceLock<MokaCache<SmolStr, IonosphereData>> = OnceLock::new();
#[cfg(feature = "server")]
static CHAT_COLOUR_CACHE: OnceLock<MokaCache<SmolStr, Option<(u8, u8, u8)>>> = OnceLock::new();
#[cfg(feature = "server")]
static CLIENT: OnceLock<Arc<BasicClient>> = OnceLock::new();
#[cfg(feature = "server")]
static FFMPEG_INIT: std::sync::Once = std::sync::Once::new();
#[cfg(feature = "server")]
const CACHE_TTL: Duration = Duration::from_secs(3600);
#[cfg(feature = "server")]
const THUMBNAIL_CACHE_MAX: u64 = 200;
#[cfg(feature = "server")]
const IONOSPHERE_CACHE_MAX: u64 = 500;
#[cfg(feature = "server")]
const CHAT_COLOUR_CACHE_MAX: u64 = 1000;

#[cfg(feature = "server")]
fn shared_client() -> Arc<BasicClient> {
    CLIENT
        .get_or_init(|| Arc::new(BasicClient::unauthenticated()))
        .clone()
}

#[cfg(feature = "server")]
fn init_ffmpeg() {
    FFMPEG_INIT.call_once(|| {
        ffmpeg_next::init().expect("ffmpeg initialization failed");
    });
}

#[cfg(feature = "server")]
const RELAY_URL: &str = "https://lightrail.microcosm.blue";

/// Intermediate discovery result before enrichment.
#[cfg(feature = "server")]
struct DiscoveredVideo {
    at_uri: AtUri,
    cid: Option<Cid>,
    video: Video,
    creator_did: Did,
}

#[cfg(feature = "server")]
fn is_501(e: &jacquard_common::error::ClientError) -> bool {
    matches!(
        e.kind(),
        jacquard_common::error::ClientErrorKind::Http { status }
            if *status == axum::http::StatusCode::NOT_IMPLEMENTED
    )
}

/// Resolve the PDS endpoint URL for a DID.
#[cfg(feature = "server")]
async fn resolve_pds(client: &Arc<BasicClient>, did: &Did) -> Option<Uri<String>> {
    use jacquard::prelude::IdentityResolver;

    match client.pds_for_did(did).await {
        Ok(uri) => Some(uri),
        Err(e) => {
            warn!(did = %did.as_str(), "failed to resolve PDS: {e}");
            None
        }
    }
}

/// Discover video records from a DID via `sync.getRepo` CAR download.
///
/// Fallback for PDSes that don't implement `listRecords` (HTTP 501).
/// Downloads the full repo as a CAR file, parses the MST, and extracts
/// records under the `place.stream.video` collection.
#[cfg(feature = "server")]
async fn discover_from_repo(
    client: &Arc<BasicClient>,
    did: &Did,
    pds_uri: &Uri<String>,
) -> Vec<(AtUri, DiscoveredVideo)> {
    use jacquard_repo::{BlockStore, MemoryBlockStore, Repository, car::parse_car_bytes};

    let pds_ref = Uri::parse(pds_uri.as_str()).expect("already validated PDS URI");
    let request = GetRepo::new().did(did.clone()).build();
    let resp = match client.xrpc(pds_ref).send(&request).await {
        Ok(r) => r,
        Err(e) => {
            warn!(did = %did.as_str(), "sync.getRepo failed: {e}");
            return vec![];
        }
    };
    let output = match resp.into_output() {
        Ok(o) => o,
        Err(e) => {
            warn!(did = %did.as_str(), "failed to parse getRepo response: {e}");
            return vec![];
        }
    };

    let parsed = match parse_car_bytes(&output.body).await {
        Ok(p) => p,
        Err(e) => {
            warn!(did = %did.as_str(), "failed to parse CAR: {e}");
            return vec![];
        }
    };

    let storage = Arc::new(MemoryBlockStore::new());
    if let Err(e) = storage.put_many(parsed.blocks).await {
        warn!(did = %did.as_str(), "failed to store CAR blocks: {e}");
        return vec![];
    }

    let repo = match Repository::<SmolStr, _>::from_commit(storage, &parsed.root).await {
        Ok(r) => r,
        Err(e) => {
            warn!(did = %did.as_str(), "failed to load repo from CAR: {e}");
            return vec![];
        }
    };

    let nsid: Nsid = <Video>::nsid().into_static();
    let entries = match repo.list_collection_data(&nsid).await {
        Ok(e) => e,
        Err(e) => {
            warn!(did = %did.as_str(), "failed to list collection: {e}");
            return vec![];
        }
    };

    let mut results = vec![];
    for (rkey, _cid, data) in entries {
        let video: Video = match serde_ipld_dagcbor::from_slice(&data) {
            Ok(v) => v,
            Err(e) => {
                warn!(did = %did.as_str(), rkey = %rkey, "failed to parse video from CAR: {e}");
                continue;
            }
        };
        let at_uri = match AtUri::from_parts_owned(did, "place.stream.video", rkey.as_str()) {
            Ok(u) => u,
            Err(e) => {
                warn!(did = %did.as_str(), rkey = %rkey, "failed to construct AT URI: {e}");
                continue;
            }
        };
        results.push((
            at_uri.clone(),
            DiscoveredVideo {
                at_uri,
                cid: None,
                video,
                creator_did: did.clone(),
            },
        ));
    }

    results
}

#[cfg(feature = "server")]
async fn discover_from_pds(
    client: &Arc<BasicClient>,
    dids: &[Did],
) -> Vec<(AtUri, DiscoveredVideo)> {
    let nsid = <Video>::nsid().into_static();

    let futures = dids.iter().map(|did| {
        let client = client.clone();
        let nsid = nsid.clone();
        let did = did.clone();
        async move {
            let pds_uri = match resolve_pds(&client, &did).await {
                Some(u) => u,
                None => return vec![],
            };

            let mut results = vec![];
            let mut cursor: Option<SmolStr> = None;

            loop {
                let pds_ref = Uri::parse(pds_uri.as_str()).expect("already validated PDS URI");
                let request = ListRecords::new()
                    .repo(AtIdentifier::Did(did.clone()))
                    .collection(nsid.clone())
                    .limit(100i64)
                    .maybe_cursor(cursor)
                    .build();

                let resp = match client.xrpc(pds_ref).send(&request).await {
                    Ok(r) => r,
                    Err(e) => {
                        if is_501(&e) {
                            tracing::info!(did = %did.as_str(), "PDS returned 501 for listRecords, falling back to sync.getRepo");
                            return discover_from_repo(&client, &did, &pds_uri).await;
                        }
                        warn!(did = %did.as_str(), "PDS list_records failed: {e}");
                        break;
                    }
                };

                let output = match resp.into_output() {
                    Ok(o) => o,
                    Err(e) => {
                        warn!(did = %did.as_str(), "failed to parse PDS response: {e}");
                        break;
                    }
                };

                let next_cursor = output.cursor.clone();

                for record in output.records {
                    let video: Video = match from_data(&record.value) {
                        Ok(v) => v,
                        Err(e) => {
                            warn!(uri = %record.uri.as_str(), "failed to parse video record: {e}");
                            continue;
                        }
                    };
                    results.push((
                        record.uri.clone(),
                        DiscoveredVideo {
                            at_uri: record.uri,
                            cid: Some(record.cid),
                            video,
                            creator_did: did.clone(),
                        },
                    ));
                }

                match next_cursor {
                    Some(c) if !c.is_empty() => cursor = Some(c),
                    _ => break,
                }
            }

            results
        }
    });

    join_all(futures).await.into_iter().flatten().collect()
}

#[cfg(feature = "server")]
async fn discover_relay_dids(client: &Arc<BasicClient>) -> Vec<Did> {
    let nsid = <Video>::nsid().into_static();

    let mut all_dids = vec![];
    let mut cursor: Option<SmolStr> = None;

    loop {
        let relay_uri = match Uri::parse(RELAY_URL) {
            Ok(u) => u,
            Err(e) => {
                warn!("failed to parse relay URL: {e}");
                return vec![];
            }
        };

        let request = ListReposByCollection {
            collection: nsid.clone(),
            cursor,
            limit: Some(500),
        };

        match client.xrpc(relay_uri).send(&request).await {
            Ok(resp) => match resp.into_output() {
                Ok(output) => {
                    let next_cursor = output.cursor.clone();
                    all_dids.extend(output.repos.into_iter().map(|r| r.did));

                    match next_cursor {
                        Some(c) if !c.is_empty() => cursor = Some(c),
                        _ => break,
                    }
                }
                Err(e) => {
                    warn!("failed to parse relay response: {e}");
                    break;
                }
            },
            Err(e) => {
                warn!("relay discovery failed: {e}");
                break;
            }
        }
    }

    all_dids
}

#[cfg(feature = "server")]
async fn resolve_thumbnail(client: &Arc<BasicClient>, livestream_uri: &AtUri) -> Option<SmolStr> {
    let response = client
        .get_record::<Livestream, _>(livestream_uri)
        .await
        .ok()?;
    let output = response.into_output().ok()?;
    let thumb = output.value.thumb.as_ref()?;
    let cid = thumb.blob().cid();
    Some(thumbnail_cdn_url(
        livestream_uri.authority().as_ref(),
        cid.as_str(),
    ))
}

#[cfg(feature = "server")]
async fn resolve_profile(
    client: &Arc<BasicClient>,
    actor: &AtIdentifier,
) -> Result<ProfileInfo, ServerFnError> {
    let request = GetProfile::new().actor(actor.clone()).build();

    let resp = client.send(request).await.map_err(|e| {
        ServerFnError::new(format!(
            "profile resolution failed for {}: {e}",
            actor.as_str()
        ))
    })?;

    let profile = resp.into_output().map_err(|e| {
        ServerFnError::new(format!(
            "failed to parse profile for {}: {e}",
            actor.as_str()
        ))
    })?;

    Ok(ProfileInfo {
        did: profile.value.did,
        handle: profile.value.handle,
        display_name: profile.value.display_name,
        avatar_url: profile.value.avatar.map(|u| u.as_ref().to_smolstr()),
    })
}

#[cfg(feature = "server")]
pub(crate) fn catalog_cache() -> &'static MokaCache<(), Vec<EnrichedVideo>> {
    CATALOG_CACHE.get_or_init(|| {
        MokaCache::builder()
            .max_capacity(1)
            .time_to_live(CACHE_TTL)
            .build()
    })
}

#[cfg(feature = "server")]
fn thumbnail_cache() -> &'static MokaCache<AtUri, bytes::Bytes> {
    THUMBNAIL_CACHE.get_or_init(|| MokaCache::new(THUMBNAIL_CACHE_MAX))
}

#[cfg(feature = "server")]
fn extract_thumbnail_sync(hls_url: &str) -> Result<Vec<u8>, ThumbnailError> {
    init_ffmpeg();

    let mut ictx = ffmpeg_next::format::input(&hls_url)?;
    let stream = ictx
        .streams()
        .best(ffmpeg_next::media::Type::Video)
        .ok_or(ThumbnailError::NoVideoStream)?;
    let stream_index = stream.index();

    let context = ffmpeg_next::codec::context::Context::from_parameters(stream.parameters())?;
    let mut decoder = context.decoder().video()?;

    let mut scaler = ffmpeg_next::software::scaling::Context::get(
        decoder.format(),
        decoder.width(),
        decoder.height(),
        ffmpeg_next::format::Pixel::RGB24,
        640,
        360,
        ffmpeg_next::software::scaling::Flags::BILINEAR,
    )?;

    let mut frame = ffmpeg_next::frame::Video::empty();
    let mut rgb_frame = ffmpeg_next::frame::Video::empty();
    let mut frames_checked = 0u32;
    const MAX_FRAMES: u32 = 30;

    for (stream, packet) in ictx.packets() {
        if stream.index() != stream_index {
            continue;
        }
        decoder.send_packet(&packet)?;

        while decoder.receive_frame(&mut frame).is_ok() {
            frames_checked += 1;
            scaler.run(&frame, &mut rgb_frame)?;

            if !is_mostly_black(&rgb_frame) {
                return encode_rgb_frame_to_jpeg(&rgb_frame);
            }

            if frames_checked >= MAX_FRAMES {
                // Give up looking for non-black, use last frame
                return encode_rgb_frame_to_jpeg(&rgb_frame);
            }
        }
    }

    Err(ThumbnailError::NoFrames)
}

#[cfg(feature = "server")]
fn is_mostly_black(frame: &ffmpeg_next::frame::Video) -> bool {
    let data = frame.data(0);
    let width = frame.width() as usize;
    let height = frame.height() as usize;
    let stride = frame.stride(0);

    let mut pixel_sum: u64 = 0;
    for y in 0..height {
        let row_start = y * stride;
        let row_end = row_start + (width * 3);
        if row_end <= data.len() {
            for &byte in &data[row_start..row_end] {
                pixel_sum += byte as u64;
            }
        }
    }

    let total_bytes = (width * height * 3) as u64;
    // Threshold: average brightness below ~6% (15/255)
    pixel_sum < total_bytes * 15
}

#[cfg(feature = "server")]
fn encode_rgb_frame_to_jpeg(frame: &ffmpeg_next::frame::Video) -> Result<Vec<u8>, ThumbnailError> {
    let width = frame.width();
    let height = frame.height();
    let stride = frame.stride(0);
    let data = frame.data(0);

    // Copy pixel data row-by-row (stride may be wider than width*3)
    let mut pixels = Vec::with_capacity((width as usize) * (height as usize) * 3);
    for y in 0..height as usize {
        let row_start = y * stride;
        let row_end = row_start + (width as usize * 3);
        pixels.extend_from_slice(&data[row_start..row_end]);
    }

    let img =
        image::RgbImage::from_raw(width, height, pixels).ok_or(ThumbnailError::InvalidPixelData)?;

    let mut buf = std::io::Cursor::new(Vec::new());
    img.write_to(&mut buf, image::ImageFormat::Jpeg)?;
    Ok(buf.into_inner())
}

#[cfg(feature = "server")]
pub async fn generate_thumbnail(at_uri: &AtUri) -> Option<bytes::Bytes> {
    let cache = thumbnail_cache();

    // Check cache first
    if let Some(bytes) = cache.get(at_uri) {
        return Some(bytes);
    }

    let hls_url = playlist_url_for_at_uri(at_uri.as_str());
    let at_uri_owned = at_uri.clone();

    let result = tokio::task::spawn_blocking(move || extract_thumbnail_sync(&hls_url))
        .await
        .ok()?
        .ok()?;

    let result = bytes::Bytes::from(result);
    cache.insert(at_uri_owned, result.clone());
    Some(result)
}

#[cfg(feature = "server")]
pub fn get_cached_thumbnail(at_uri: &AtUri) -> Option<bytes::Bytes> {
    thumbnail_cache().get(at_uri)
}

#[cfg(feature = "server")]
async fn enrich(
    client: &Arc<BasicClient>,
    discovered: HashMap<AtUri, DiscoveredVideo>,
) -> Vec<EnrichedVideo> {
    // Collect unique DIDs for profile resolution
    let unique_dids: Vec<Did> = discovered
        .values()
        .map(|d| d.creator_did.clone())
        .collect::<std::collections::HashSet<_>>()
        .into_iter()
        .collect();

    // Resolve profiles in parallel
    let actor_ids: Vec<AtIdentifier> = unique_dids
        .iter()
        .map(|did| AtIdentifier::Did(did.clone()))
        .collect();
    let profile_futures = actor_ids.iter().map(|actor| resolve_profile(client, actor));
    let profiles: HashMap<Did, ProfileInfo> = join_all(profile_futures)
        .await
        .into_iter()
        .filter_map(|r| match r {
            Ok(p) => Some((p.did.clone(), p)),
            Err(e) => {
                warn!("profile resolution failed during enrichment: {e}");
                None
            }
        })
        .collect();

    // Resolve thumbnails in parallel: try livestream CDN first, then ffmpeg fallback
    let discovered_vec: Vec<_> = discovered.into_values().collect();
    let thumb_futures: Vec<_> = discovered_vec
        .iter()
        .map(|d| {
            let client = client.clone();
            let livestream_uri = d.video.livestream.as_ref().map(|sr| sr.uri.clone());
            let at_uri = d.at_uri.clone();
            async move {
                // Try livestream CDN thumbnail first
                if let Some(ref uri) = livestream_uri {
                    if let Some(url) = resolve_thumbnail(&client, uri).await {
                        return Some(url);
                    }
                }
                // Fallback: generate from HLS via ffmpeg
                if generate_thumbnail(&at_uri).await.is_some() {
                    let encoded = percent_encoding::utf8_percent_encode(
                        at_uri.as_str(),
                        percent_encoding::NON_ALPHANUMERIC,
                    );
                    Some(format_smolstr!("/api/thumbnail?uri={encoded}"))
                } else {
                    None
                }
            }
        })
        .collect();
    // Resolve ionosphere data in parallel
    let iono_futures: Vec<_> = discovered_vec
        .iter()
        .map(|d| {
            let client = client.clone();
            let at_uri = d.at_uri.clone();
            async move { resolve_ionosphere(&client, at_uri.as_str()).await }
        })
        .collect();

    let (thumbnails, ionosphere): (Vec<_>, Vec<_>) =
        futures::future::join(join_all(thumb_futures), join_all(iono_futures)).await;

    // Assemble EnrichedVideos
    discovered_vec
        .into_iter()
        .zip(thumbnails)
        .zip(ionosphere)
        .map(|((d, thumb), iono)| {
            let creator = profiles
                .get(&d.creator_did)
                .cloned()
                .unwrap_or_else(|| ProfileInfo {
                    did: d.creator_did,
                    handle: "handle.invalid".parse().expect("valid handle"),
                    display_name: None,
                    avatar_url: None,
                });

            let livestream_uri = d.video.livestream.as_ref().map(|sr| sr.uri.clone());

            EnrichedVideo {
                at_uri: d.at_uri,
                cid: d.cid,
                title: d.video.title,
                duration_ns: d.video.duration,
                created_at: d.video.created_at,
                creator,
                thumbnail_url: thumb,
                livestream_uri,
                speakers: iono.speakers,
                talk: iono.talk,
            }
        })
        .collect()
}

#[cfg(feature = "server")]
async fn query_backlinks(
    client: &Arc<BasicClient>,
    subject: &str,
    source_field: &str,
) -> Option<GetBacklinksResponse> {
    let query = GetBacklinksQuery {
        subject: SmolStr::new(subject),
        source: SmolStr::new(source_field),
        limit: Some(1),
        cursor: None,
        did: vec![],
        reverse: false,
    };
    let constellation_uri = Uri::parse(CONSTELLATION_URL).ok()?;
    let (ua_name, ua_value) = user_agent_header();
    let resp = client
        .xrpc(constellation_uri)
        .header(ua_name, ua_value)
        .send(&query)
        .await
        .ok()?;
    resp.into_output().ok()
}

#[cfg(feature = "server")]
#[derive(Default, Clone)]
struct IonosphereData {
    speakers: Vec<SpeakerInfo>,
    talk: Option<vodplace_api::tv_ionosphere::talk::Talk>,
    talk_uri: Option<AtUri>,
}

#[cfg(feature = "server")]
fn ionosphere_cache() -> &'static MokaCache<SmolStr, IonosphereData> {
    IONOSPHERE_CACHE.get_or_init(|| {
        MokaCache::builder()
            .max_capacity(IONOSPHERE_CACHE_MAX)
            .time_to_live(CACHE_TTL)
            .build()
    })
}

#[cfg(feature = "server")]
fn chat_colour_cache() -> &'static MokaCache<SmolStr, Option<(u8, u8, u8)>> {
    CHAT_COLOUR_CACHE.get_or_init(|| {
        MokaCache::builder()
            .max_capacity(CHAT_COLOUR_CACHE_MAX)
            .time_to_live(CACHE_TTL)
            .build()
    })
}

#[cfg(feature = "server")]
async fn resolve_ionosphere(client: &Arc<BasicClient>, subject: &str) -> IonosphereData {
    let cache = ionosphere_cache();
    let key = subject.to_smolstr();
    if let Some(cached) = cache.get(&key) {
        return cached;
    }

    let data = resolve_ionosphere_uncached(client, subject).await;
    cache.insert(key, data.clone());
    data
}

#[cfg(feature = "server")]
async fn resolve_ionosphere_uncached(client: &Arc<BasicClient>, subject: &str) -> IonosphereData {
    use vodplace_api::tv_ionosphere::speaker::Speaker;
    use vodplace_api::tv_ionosphere::talk::Talk;

    let backlinks = match query_backlinks(client, subject, "tv.ionosphere.talk:videoUri").await {
        Some(bl) if !bl.records.is_empty() => bl,
        _ => return IonosphereData::default(),
    };

    let rec = &backlinks.records[0];
    let talk_uri: AtUri = match AtUri::from_parts_owned(
        &rec.did,
        rec.collection.as_ref(),
        &rec.rkey,
    ) {
        Ok(u) => u,
        Err(e) => {
            warn!(did = %rec.did, rkey = %rec.rkey.as_str(), "failed to construct talk AT URI: {e}");
            return IonosphereData::default();
        }
    };

    let talk: Talk = match client.get_record::<Talk, _>(&talk_uri).await {
        Ok(resp) => match resp.into_output() {
            Ok(out) => out.value,
            Err(e) => {
                warn!(uri = %talk_uri.as_str(), "failed to parse talk record: {e}");
                return IonosphereData::default();
            }
        },
        Err(e) => {
            warn!(uri = %talk_uri.as_str(), "failed to fetch talk record: {e}");
            return IonosphereData::default();
        }
    };

    let speaker_uris = talk.speaker_uris.clone().unwrap_or_default();

    let speaker_futures = speaker_uris.iter().map(|uri| {
        let client = client.clone();
        let uri = uri.clone();
        async move {
            let resp = client.get_record::<Speaker, _>(&uri).await.ok()?;
            let out = resp.into_output().ok()?;
            let speaker = out.value;

            // Resolve profile for avatar -- by DID if available, otherwise by handle
            let actor: Option<AtIdentifier> = if let Some(ref did) = speaker.did {
                Some(AtIdentifier::Did(did.clone()))
            } else if let Some(ref h) = speaker.handle {
                h.parse().ok().map(AtIdentifier::Handle)
            } else {
                None
            };

            let (handle, avatar_url) = match actor {
                Some(actor) => match resolve_profile(&client, &actor).await {
                    Ok(profile) => (Some(profile.handle), profile.avatar_url),
                    Err(_) => (speaker.handle.and_then(|h| h.parse().ok()), None),
                },
                None => (speaker.handle.and_then(|h| h.parse().ok()), None),
            };

            Some(SpeakerInfo {
                name: speaker.name,
                did: speaker.did,
                handle,
                avatar_url,
            })
        }
    });

    let speakers: Vec<SpeakerInfo> = futures::future::join_all(speaker_futures)
        .await
        .into_iter()
        .flatten()
        .collect();

    IonosphereData {
        speakers,
        talk: Some(talk),
        talk_uri: Some(talk_uri),
    }
}

#[cfg(feature = "server")]
pub(crate) async fn fetch_video_metadata(at_uri: &AtUri) -> Result<VideoMetadata, ServerFnError> {
    let client = shared_client();
    let creator_ident: AtIdentifier = at_uri.authority().into_static();

    // Fetch video record
    let video_resp = client
        .get_record::<crate::lexicons::Video, _>(at_uri)
        .await
        .map_err(|e| ServerFnError::new(format!("failed to fetch video record: {e}")))?;
    let video_out = video_resp
        .into_output()
        .map_err(|e| ServerFnError::new(format!("failed to parse video record: {e}")))?;
    let video = video_out.value;

    // Resolve creator profile + thumbnail + ionosphere in parallel
    let profile_fut = resolve_profile(&client, &creator_ident);
    let thumb_fut = async {
        if let Some(ref ls) = video.livestream {
            resolve_thumbnail(&client, &ls.uri).await
        } else {
            None
        }
    };
    let iono_fut = resolve_ionosphere(&client, at_uri.as_str());

    let (creator_result, thumbnail_url, iono) =
        futures::future::join3(profile_fut, thumb_fut, iono_fut).await;

    let creator = creator_result?;

    let livestream_uri = video.livestream.as_ref().map(|ls| ls.uri.clone());

    Ok(VideoMetadata {
        at_uri: at_uri.clone(),
        title: video.title,
        duration_ns: video.duration,
        created_at: video.created_at,
        creator,
        thumbnail_url,
        livestream_uri,
        speakers: iono.speakers,
        talk: iono.talk,
        talk_uri: iono.talk_uri,
    })
}

#[cfg(feature = "server")]
pub(crate) async fn build_catalog() -> Result<Vec<EnrichedVideo>, ServerFnError> {
    use tracing::info;

    let client = shared_client();

    // Step 1: Discover DIDs from relay
    let relay_dids = discover_relay_dids(&client).await;
    info!(count = relay_dids.len(), "relay DID discovery complete");
    if relay_dids.is_empty() {
        warn!("relay returned no DIDs for place.stream.video");
        return Ok(vec![]);
    }

    // Step 2: Fan out to each DID's PDS and fetch all video records (paginated)
    let pds_results = discover_from_pds(&client, &relay_dids).await;
    info!(count = pds_results.len(), "PDS record discovery complete");
    let catalog: HashMap<AtUri, DiscoveredVideo> = pds_results.into_iter().collect();

    // Step 3: Enrich -- resolve thumbnails and profiles
    let videos = enrich(&client, catalog).await;
    info!(count = videos.len(), "catalog enrichment complete");
    Ok(videos)
}

use dioxus::prelude::*;

#[server]
pub async fn get_video_catalog() -> Result<Vec<EnrichedVideo>, ServerFnError> {
    use crate::chat_cache::shared_cache;

    let moka = catalog_cache();

    // Layer 1: in-memory moka cache (hot path)
    if let Some(videos) = moka.get(&()) {
        return Ok(videos);
    }

    // Layer 2: sqlite persistent cache
    let db = shared_cache();
    let now_secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs() as i64;

    if let Ok(Some((videos, updated_at))) = db.load_video_catalog::<EnrichedVideo>().await {
        let age_secs = now_secs - updated_at;

        // Populate moka from sqlite
        moka.insert((), videos.clone());

        if age_secs < CACHE_TTL.as_secs() as i64 {
            return Ok(videos);
        }

        // Stale — return cached data now and refresh in the background
        tokio::spawn(async move {
            if let Ok(fresh) = build_catalog().await {
                let db = shared_cache();
                let moka = catalog_cache();
                if let Err(e) = db.store_video_catalog(&fresh).await {
                    warn!("failed to persist refreshed catalog to sqlite: {e}");
                }
                moka.insert((), fresh);
            }
        });

        return Ok(videos);
    }

    // Layer 3: cold start — no cached data at all, fetch synchronously
    let videos = match build_catalog().await {
        Ok(v) => v,
        Err(_first_err) => build_catalog().await?,
    };

    if let Err(e) = db.store_video_catalog(&videos).await {
        warn!("failed to persist catalog to sqlite: {e}");
    }
    moka.insert((), videos.clone());
    Ok(videos)
}

#[server]
pub async fn get_video_metadata(at_uri: AtUri) -> Result<VideoMetadata, ServerFnError> {
    fetch_video_metadata(&at_uri).await
}

/// Fetch the ionosphere transcript for a talk, if one exists.
///
/// Queries constellation for `tv.ionosphere.transcript:talkUri` backlinks,
/// then fetches the transcript record. Returns `None` if no transcript
/// exists or the fetch fails -- transcript is non-critical.
#[server]
pub async fn get_transcript(
    talk_uri: AtUri,
) -> Result<Option<vodplace_subtitle::IonosphereInput>, ServerFnError> {
    use vodplace_api::tv_ionosphere::transcript::{Transcript, TranscriptRecord};

    let client = shared_client();
    let backlinks = match query_backlinks(
        &client,
        talk_uri.as_str(),
        "tv.ionosphere.transcript:talkUri",
    )
    .await
    {
        Some(bl) if !bl.records.is_empty() => bl,
        _ => return Ok(None),
    };

    let rec = &backlinks.records[0];
    let transcript_uri: AtUri =
        AtUri::from_parts_owned(&rec.did, rec.collection.as_ref(), &rec.rkey)
            .map_err(|e| ServerFnError::new(format!("failed to construct transcript URI: {e}")))?;

    let transcript: Transcript = client
        .get_record::<TranscriptRecord, _>(&transcript_uri)
        .await
        .map_err(|e| ServerFnError::new(format!("failed to fetch transcript: {e}")))?
        .into_output()
        .map_err(|e| ServerFnError::new(format!("failed to parse transcript: {e}")))?
        .value;

    Ok(Some(vodplace_subtitle::IonosphereInput {
        text: transcript.text.to_string(),
        start_ms: transcript.start_ms,
        timings: transcript.timings,
    }))
}

/// Fetch all `tv.ionosphere.comment` records that reference a talk (and optionally
/// its transcript), returning a flat list of `CommentData` for client-side tree building.
///
/// Uses slingshot hydration for efficient batch fetching, with a fallback to direct
/// Constellation backlink queries when slingshot is unavailable.
///
/// Both `talk_uri` and `transcript_uri` may be subjects of comments. The results are
/// deduplicated so a comment referencing both appears only once.
#[server]
pub async fn get_comments(
    talk_uri: AtUri,
    transcript_uri: Option<AtUri>,
) -> Result<Vec<CommentData>, ServerFnError> {
    use crate::chat::convert_bsky_facets;
    use vodplace_api::tv_ionosphere::comment::{Comment, CommentRecord};

    let client = shared_client();
    let slingshot_uri = Uri::parse(SLINGSHOT_URL)
        .map_err(|e| ServerFnError::new(format!("bad slingshot URL: {e}")))?;
    let constellation_uri = Uri::parse(CONSTELLATION_URL)
        .map_err(|e| ServerFnError::new(format!("bad constellation URL: {e}")))?;

    // Discover transcript URI server-side if not provided, so we also find
    // comments whose subject is the transcript (not just the talk).
    let transcript_uri = match transcript_uri {
        Some(uri) => Some(uri),
        None => {
            query_backlinks(&client, talk_uri.as_str(), "tv.ionosphere.transcript:talkUri")
                .await
                .and_then(|bl| {
                    bl.records.first().and_then(|rec| {
                        AtUri::from_parts_owned(&rec.did, rec.collection.as_ref(), &rec.rkey).ok()
                    })
                })
        }
    };

    // Collect all unique comment AT URIs from backlink queries.
    // We query for comments on the talk and, if provided, the transcript.
    let mut comment_uri_strs: Vec<SmolStr> = Vec::new();
    let mut seen_uris = std::collections::HashSet::new();

    let subjects_to_query: Vec<SmolStr> = {
        let mut v = vec![SmolStr::new(talk_uri.as_str())];
        if let Some(ref t) = transcript_uri {
            v.push(SmolStr::new(t.as_str()));
        }
        v
    };

    for subject_str in &subjects_to_query {
        // Try slingshot first. An empty Ok result is valid (no comments exist);
        // only fall back to constellation on actual failure.
        let uris_to_add = match fetch_comment_backlinks_slingshot(&client, &slingshot_uri, subject_str).await {
            Ok(uris) => uris,
            Err(e) => {
                warn!("slingshot comment backlinks failed for {subject_str}, trying constellation: {e}");
                match fetch_comment_backlinks_direct(&client, &constellation_uri, subject_str).await {
                    Ok(uris) => uris,
                    Err(e2) => {
                        warn!("constellation comment backlinks also failed for {subject_str}: {e2}");
                        vec![]
                    }
                }
            }
        };

        for uri_str in uris_to_add {
            if seen_uris.insert(uri_str.clone()) {
                comment_uri_strs.push(uri_str);
            }
        }
    }

    if comment_uri_strs.is_empty() {
        return Ok(vec![]);
    }

    // Fetch each comment record from its author's PDS in parallel.
    let fetch_futures = comment_uri_strs.iter().filter_map(|uri_str| {
        let at_uri: AtUri = match uri_str.parse() {
            Ok(u) => u,
            Err(e) => {
                warn!("failed to parse comment URI {uri_str}: {e}");
                return None;
            }
        };
        let client = client.clone();
        Some(async move {
            let resp = match client.get_record::<CommentRecord, _>(&at_uri).await {
                Ok(r) => r,
                Err(e) => {
                    warn!(uri = %at_uri.as_str(), "failed to fetch comment record: {e}");
                    return None;
                }
            };
            let output = match resp.into_output() {
                Ok(o) => o,
                Err(e) => {
                    warn!(uri = %at_uri.as_str(), "failed to parse comment record: {e}");
                    return None;
                }
            };
            Some((at_uri, output.value))
        })
    });

    let fetched: Vec<(AtUri, Comment)> = join_all(fetch_futures)
        .await
        .into_iter()
        .flatten()
        .collect();

    // Resolve unique author handles via lightweight DID document lookup.
    // This is much cheaper than resolve_profile (no GetProfile API call).
    let unique_dids: Vec<Did> = {
        let mut seen = std::collections::HashSet::new();
        fetched
            .iter()
            .filter_map(|(uri, _)| {
                let did_str = uri.authority().as_ref().to_string();
                if seen.insert(did_str.clone()) {
                    did_str.parse::<Did>().ok()
                } else {
                    None
                }
            })
            .collect()
    };

    let handle_futures = unique_dids.iter().map(|did| {
        let client = client.clone();
        let did = did.clone();
        async move {
            match crate::identity::resolve_handle_for_did(&*client, &did).await {
                Ok(handle) => Some((did, handle)),
                Err(e) => {
                    warn!(did = %did.as_str(), "failed to resolve comment author handle: {e}");
                    None
                }
            }
        }
    });

    let handle_map: HashMap<Did, Handle> = join_all(handle_futures)
        .await
        .into_iter()
        .flatten()
        .collect();

    let fallback_handle: Handle = "handle.invalid".parse().expect("valid handle literal");

    // Convert fetched records to CommentData.
    let comments: Vec<CommentData> = fetched
        .into_iter()
        .filter_map(|(uri, comment)| {
            let author_did: Did = match uri.authority().as_ref().parse() {
                Ok(d) => d,
                Err(e) => {
                    warn!(uri = %uri.as_str(), "failed to parse author DID from comment URI: {e}");
                    return None;
                }
            };

            let author_handle = handle_map
                .get(&author_did)
                .cloned()
                .unwrap_or_else(|| fallback_handle.clone());

            let anchor = comment.anchor.and_then(|a| {
                // Reject negative values — treat comment as non-anchored.
                let byte_start = usize::try_from(a.byte_start).ok()?;
                let byte_end = usize::try_from(a.byte_end).ok()?;
                // Reject inverted ranges — treat comment as non-anchored.
                if byte_start > byte_end {
                    return None;
                }
                Some(ByteRangeData {
                    byte_start,
                    byte_end,
                })
            });

            let facets = comment.facets.as_deref().and_then(|fs| {
                let v = convert_bsky_facets(fs);
                if v.is_empty() { None } else { Some(v) }
            });

            Some(CommentData {
                uri,
                text: SmolStr::new(comment.text.as_ref() as &str),
                author_did,
                author_handle,
                created_at: comment.created_at,
                anchor,
                subject_uri: comment.subject,
                facets,
            })
        })
        .collect();

    Ok(comments)
}

/// Query Constellation backlinks for `tv.ionosphere.comment:subject` via slingshot hydration.
///
/// Returns a list of AT URI strings for matching comments. Does NOT hydrate the record
/// bodies — we fetch those separately since comments may be from many different PDSes.
#[cfg(feature = "server")]
async fn fetch_comment_backlinks_slingshot(
    client: &Arc<BasicClient>,
    slingshot_uri: &Uri<&str>,
    subject: &str,
) -> Result<Vec<SmolStr>, String> {
    use jacquard::types::value::{from_data, to_data};

    let query = GetBacklinksQuery {
        subject: SmolStr::new(subject),
        source: SmolStr::new("tv.ionosphere.comment:subject"),
        limit: Some(100),
        cursor: None,
        did: vec![],
        reverse: false,
    };

    let params_data =
        to_data(&query).map_err(|e| format!("failed to convert params to Data: {e}"))?;

    let hydrate_req = HydrateQueryRequest {
        xrpc: SmolStr::new("blue.microcosm.links.getBacklinks"),
        atproto_proxy: SmolStr::new("did:web:constellation.microcosm.blue#constellation"),
        authorization: None,
        atproto_accept_labelers: None,
        params: params_data,
        hydration_sources: vec![],
    };

    let (ua_name, ua_value) = user_agent_header();
    let resp = client
        .xrpc(slingshot_uri.clone())
        .header(ua_name, ua_value)
        .send(&hydrate_req)
        .await
        .map_err(|e| format!("slingshot request failed: {e}"))?;

    let hydrate_output: HydrateQueryResponse = resp
        .into_output()
        .map_err(|e| format!("slingshot response parse failed: {e}"))?;

    let backlinks: GetBacklinksResponse =
        from_data(&hydrate_output.output).map_err(|e| format!("backlinks parse failed: {e}"))?;

    let uris: Vec<SmolStr> = backlinks
        .records
        .iter()
        .filter_map(|rec| {
            let uri: AtUri =
                AtUri::from_parts_owned(&rec.did, rec.collection.as_ref(), &rec.rkey).ok()?;
            Some(SmolStr::new(uri.as_str()))
        })
        .collect();

    Ok(uris)
}

/// Query Constellation backlinks for `tv.ionosphere.comment:subject` directly.
///
/// Used as fallback when slingshot is unavailable.
#[cfg(feature = "server")]
async fn fetch_comment_backlinks_direct(
    client: &Arc<BasicClient>,
    constellation_uri: &Uri<&str>,
    subject: &str,
) -> Result<Vec<SmolStr>, String> {
    let query = GetBacklinksQuery {
        subject: SmolStr::new(subject),
        source: SmolStr::new("tv.ionosphere.comment:subject"),
        limit: Some(100),
        cursor: None,
        did: vec![],
        reverse: false,
    };

    let (ua_name, ua_value) = user_agent_header();
    let resp = client
        .xrpc(constellation_uri.clone())
        .header(ua_name, ua_value)
        .send(&query)
        .await
        .map_err(|e| format!("constellation request failed: {e}"))?;

    let backlinks: GetBacklinksResponse = resp
        .into_output()
        .map_err(|e| format!("constellation response parse failed: {e}"))?;

    let uris: Vec<SmolStr> = backlinks
        .records
        .iter()
        .filter_map(|rec| {
            let uri: AtUri =
                AtUri::from_parts_owned(&rec.did, rec.collection.as_ref(), &rec.rkey).ok()?;
            Some(SmolStr::new(uri.as_str()))
        })
        .collect();

    Ok(uris)
}

/// Resolve the chat replay context for a VOD: the streamer DID, anchor time,
/// and wall-clock window of messages to fetch.
#[server]
pub async fn resolve_chat_context(
    livestream_uri: AtUri,
    video_created_at: Datetime,
    video_duration_ns: u64,
    talk_starts_at: Option<Datetime>,
    talk_ends_at: Option<Datetime>,
) -> Result<crate::chat::ChatContext, ServerFnError> {
    use crate::chat::{self, ChatContext};
    use jacquard::prelude::IdentityResolver;

    let client = shared_client();
    let streamer_did: Did = match livestream_uri.authority().into_static() {
        AtIdentifier::Did(did) => did,
        AtIdentifier::Handle(handle) => client
            .resolve_handle(&handle)
            .await
            .map_err(|e| ServerFnError::new(format!("failed to resolve handle: {e}")))?,
    };

    // Fetch livestream for its timing data
    let livestream_resp = client
        .get_record::<Livestream, _>(&livestream_uri)
        .await
        .map_err(|e| ServerFnError::new(format!("failed to fetch livestream: {e}")))?;
    let livestream = livestream_resp
        .into_output()
        .map_err(|e| ServerFnError::new(format!("failed to parse livestream: {e}")))?
        .value;

    let anchor = chat::resolve_anchor_time_from_fields(
        talk_starts_at.as_ref(),
        Some(&livestream.created_at),
        &video_created_at,
    );
    let end_ms = chat::resolve_end_time_from_fields(
        talk_ends_at.as_ref(),
        livestream.ended_at.as_ref(),
        &anchor,
        video_duration_ns,
    );

    const FUZZ_MINUTES: u32 = 5;
    let (window_start_ms, window_end_ms) = chat::compute_time_window(&anchor, end_ms, FUZZ_MINUTES);

    Ok(ChatContext {
        streamer_did,
        anchor,
        window_start_ms,
        window_end_ms,
    })
}

/// Fetch chat messages for a VOD time window, using slingshot hydration with
/// Constellation backlink discovery and SQLite caching.
#[server]
pub async fn get_chat_messages(
    streamer_did: Did,
    window_start_ms: i64,
    window_end_ms: i64,
    anchor_ms: i64,
) -> Result<crate::chat::ChatMessageBatch, ServerFnError> {
    use crate::chat::{
        ChatMessageBatch, ChatMessageView, FacetData, ReplySnippet, convert_facets,
        to_vod_relative_ms, truncate_preview,
    };
    use crate::chat_cache::{CachedChatMessage, FetchDirection, shared_cache};

    let client = shared_client();
    let cache = shared_cache();
    let streamer_did_str = streamer_did.as_str();

    // Step 1: Check fetch cursors
    let forward_cursor = cache
        .get_fetch_cursor(streamer_did_str, FetchDirection::Forward)
        .await
        .map_err(|e| ServerFnError::new(format!("cache fetch_cursor forward failed: {e}")))?;
    let reverse_cursor = cache
        .get_fetch_cursor(streamer_did_str, FetchDirection::Reverse)
        .await
        .map_err(|e| ServerFnError::new(format!("cache fetch_cursor reverse failed: {e}")))?;

    let fwd_boundary = forward_cursor.as_ref().map_or(i64::MIN, |c| c.boundary_ms);
    let rev_boundary = reverse_cursor.as_ref().map_or(i64::MAX, |c| c.boundary_ms);

    let forward_covered = fwd_boundary >= window_end_ms;
    let reverse_covered = rev_boundary <= window_start_ms;

    // Step 2: Determine has_more from fetching
    let mut has_more = false;

    // Step 3: Fetch via slingshot if not fully cached
    if !forward_covered || !reverse_covered {
        let slingshot_uri = Uri::parse(SLINGSHOT_URL)
            .map_err(|e| ServerFnError::new(format!("bad slingshot URL: {e}")))?;

        // Fetch forward direction if needed (older → newer)
        if !forward_covered {
            let cursor_val = forward_cursor.as_ref().and_then(|c| c.cursor.as_deref());
            if fetch_and_cache_messages(
                &client,
                cache,
                &slingshot_uri,
                streamer_did_str,
                FetchDirection::Forward,
                cursor_val,
                fwd_boundary,
                window_end_ms,
            )
            .await
            .map_err(ServerFnError::new)?
            {
                has_more = true;
            }
        }

        // Fetch reverse direction if needed (newer → older)
        if !reverse_covered {
            let cursor_val = reverse_cursor.as_ref().and_then(|c| c.cursor.as_deref());
            if fetch_and_cache_messages(
                &client,
                cache,
                &slingshot_uri,
                streamer_did_str,
                FetchDirection::Reverse,
                cursor_val,
                rev_boundary,
                window_start_ms,
            )
            .await
            .map_err(ServerFnError::new)?
            {
                has_more = true;
            }
        }
    }

    // Step 4: Query cache for the requested time window
    let cached = cache
        .get_messages(streamer_did_str, window_start_ms, window_end_ms)
        .await
        .map_err(|e| ServerFnError::new(format!("cache query failed: {e}")))?;

    // Step 6: Resolve author handles (deduplicated within this request)
    let unique_dids: Vec<Did> = {
        let mut seen = std::collections::HashSet::new();
        cached
            .iter()
            .filter_map(|m| {
                if seen.insert(m.author_did.clone()) {
                    Some(m.author_did.parse::<Did>().ok())
                } else {
                    None
                }
            })
            .flatten()
            .collect()
    };

    let mut handle_map: HashMap<Did, Handle> = HashMap::new();
    {
        let handle_futures = unique_dids.iter().map(|did| {
            let client = client.clone();
            let did = did.clone();
            async move {
                match crate::identity::resolve_handle_for_did(&*client, &did).await {
                    Ok(handle) => Some((did, handle)),
                    Err(e) => {
                        warn!(did = %did.as_str(), "handle resolution failed: {e}");
                        None
                    }
                }
            }
        });
        for (did, handle) in join_all(handle_futures).await.into_iter().flatten() {
            handle_map.insert(did, handle);
        }
    }

    let fallback_handle: Handle = "handle.invalid".parse().expect("valid handle literal");

    // Build a lookup map from rkey to cached message for reply snippet resolution
    let msg_by_rkey: HashMap<&str, &CachedChatMessage> =
        cached.iter().map(|m| (m.rkey.as_str(), m)).collect();

    // Step 7: Build ChatMessageViews with reply snippets
    let messages: Vec<ChatMessageView> = cached
        .iter()
        .map(|msg| {
            let author_did: Did = msg
                .author_did
                .parse()
                .unwrap_or_else(|_| "did:plc:invalid".parse().unwrap());
            let author_handle = handle_map
                .get(&author_did)
                .cloned()
                .unwrap_or_else(|| fallback_handle.clone());

            let vod_relative = to_vod_relative_ms(msg.created_at_ms, anchor_ms);

            let facets: Vec<FacetData> = msg
                .facets_json
                .as_deref()
                .and_then(|json| {
                    serde_json::from_str::<Vec<vodplace_api::place_stream::richtext::facet::Facet>>(
                        json,
                    )
                    .ok()
                })
                .map(|f| convert_facets(&f))
                .unwrap_or_default();

            // Build reply snippet if this message is a reply
            let reply_snippet = msg.reply_parent_uri.as_deref().and_then(|parent_uri_str| {
                let parent_rkey = parent_uri_str.rsplit('/').next()?;
                let parent_msg = msg_by_rkey.get(parent_rkey)?;
                let parent_did: Did = parent_msg.author_did.parse().ok()?;
                let parent_handle = handle_map
                    .get(&parent_did)
                    .cloned()
                    .unwrap_or_else(|| fallback_handle.clone());
                Some(ReplySnippet {
                    author_did: parent_did,
                    author_handle: parent_handle,
                    text_preview: truncate_preview(&parent_msg.text, 80),
                    parent_tid: SmolStr::new(parent_rkey),
                })
            });

            ChatMessageView {
                tid: SmolStr::new(&msg.rkey),
                text: SmolStr::new(&msg.text),
                author_did: author_did.clone(),
                author_handle,
                vod_relative_ms: vod_relative,
                reply_snippet,
                facets,
            }
        })
        .collect();

    Ok(ChatMessageBatch { messages, has_more })
}

/// Fetch chat profile colours for a set of author DIDs.
///
/// Each author stores a `place.stream.chat.profile` record in their own repo at
/// `at://author_did/place.stream.chat.profile/self`. Results are cached with
/// a 1-hour TTL. Missing or erroring profiles return `None` for that DID —
/// callers should fall back to the CSS default colour.
#[server]
pub async fn get_chat_colours(
    author_dids: Vec<Did>,
) -> Result<HashMap<String, Option<(u8, u8, u8)>>, ServerFnError> {
    use vodplace_api::place_stream::chat::profile::ProfileRecord;

    let client = shared_client();
    let cache = chat_colour_cache();

    let fetch_futures = author_dids.into_iter().map(|did| {
        let client = client.clone();
        let cache = cache;
        async move {
            let did_str = did.as_str().to_string();
            let cache_key = SmolStr::new(&did_str);
            // Check cache first
            if let Some(cached) = cache.get(&cache_key) {
                return (did_str, cached);
            }

            // Construct the profile URI: at://did/place.stream.chat.profile/self
            let profile_uri_str = format!("at://{}/place.stream.chat.profile/self", did.as_str());
            let profile_uri: AtUri = match profile_uri_str.parse() {
                Ok(u) => u,
                Err(e) => {
                    warn!(did = %did.as_str(), "failed to parse profile URI: {e}");
                    cache.insert(cache_key, None);
                    return (did_str, None);
                }
            };

            let colour = match client.get_record::<ProfileRecord, _>(&profile_uri).await {
                Ok(resp) => match resp.into_output() {
                    Ok(output) => output.value.color.map(|c| {
                        (
                            c.red.clamp(0, 255) as u8,
                            c.green.clamp(0, 255) as u8,
                            c.blue.clamp(0, 255) as u8,
                        )
                    }),
                    Err(e) => {
                        warn!(did = %did.as_str(), "failed to parse chat profile: {e}");
                        None
                    }
                },
                Err(_) => {
                    // Profile not found or PDS unreachable — silently return None.
                    None
                }
            };

            cache.insert(cache_key, colour);
            (did_str, colour)
        }
    });

    let results: HashMap<String, Option<(u8, u8, u8)>> =
        join_all(fetch_futures).await.into_iter().collect();

    Ok(results)
}

/// Fetch one page of messages for a given direction, store them in cache, and update the cursor.
///
/// Tries slingshot first; falls back to direct Constellation fetch if slingshot fails.
///
/// The `old_boundary` is the cursor boundary before this fetch (used as fallback when no
/// messages are returned). The `window_limit_ms` is the far edge of the requested time
/// window for this direction: `window_end_ms` for forward, `window_start_ms` for reverse.
///
/// Returns `true` if the server has more results that have not yet been fetched into the
/// window (i.e. the caller should set `has_more`).
#[cfg(feature = "server")]
async fn fetch_and_cache_messages(
    client: &Arc<BasicClient>,
    cache: &crate::chat_cache::ChatCache,
    slingshot_uri: &Uri<&str>,
    streamer_did: &str,
    direction: crate::chat_cache::FetchDirection,
    cursor: Option<&str>,
    old_boundary: i64,
    window_limit_ms: i64,
) -> Result<bool, String> {
    use crate::chat_cache::FetchDirection;

    let reverse = direction == FetchDirection::Reverse;
    let dir_label = if reverse { "reverse" } else { "forward" };

    let (messages, response_cursor) =
        match fetch_via_slingshot(client, slingshot_uri, streamer_did, reverse, cursor).await {
            Ok(result) => result,
            Err(e) => {
                warn!("slingshot {dir_label} fetch failed, trying fallback: {e}");
                match fetch_fallback(client, streamer_did, reverse, cursor).await {
                    Ok(result) => result,
                    Err(e2) => {
                        warn!("fallback {dir_label} fetch also failed: {e2}");
                        return Ok(false);
                    }
                }
            }
        };

    // For forward direction the new boundary is the latest message timestamp;
    // for reverse it is the earliest. Computed once and reused for cursor update + has_more.
    let new_boundary = if reverse {
        messages
            .iter()
            .map(|m| m.created_at_ms)
            .min()
            .unwrap_or(old_boundary)
    } else {
        messages
            .iter()
            .map(|m| m.created_at_ms)
            .max()
            .unwrap_or(old_boundary)
    };

    // Store successfully parsed messages (only if we have any to store)
    if !messages.is_empty() {
        cache
            .store_messages(&messages)
            .await
            .map_err(|e| format!("cache store failed: {e}"))?;
    }
    // Always advance the cursor when constellation gave us one — prevents stall
    // loops when all records in a page are Pending/Error/unparseable.
    cache
        .set_fetch_cursor(
            streamer_did,
            direction,
            response_cursor.as_deref(),
            new_boundary,
        )
        .await
        .map_err(|e| format!("cache set_cursor failed: {e}"))?;

    // has_more: the server has more pages and the boundary hasn't yet reached the window edge.
    let has_more = response_cursor.is_some()
        && if reverse {
            new_boundary > window_limit_ms
        } else {
            new_boundary < window_limit_ms
        };

    Ok(has_more)
}

/// Fetch chat messages via slingshot hydration proxy.
///
/// Returns `(cached_messages, response_cursor)`. The response cursor is `Some` if
/// Constellation has more results.
#[cfg(feature = "server")]
async fn fetch_via_slingshot(
    client: &Arc<BasicClient>,
    slingshot_uri: &Uri<&str>,
    streamer_did: &str,
    reverse: bool,
    cursor: Option<&str>,
) -> Result<(Vec<crate::chat_cache::CachedChatMessage>, Option<String>), String> {
    use crate::chat_cache::CachedChatMessage;
    use jacquard::types::value::{from_data, to_data};
    use vodplace_api::place_stream::chat::message::Message as ChatMessage;

    let query = GetBacklinksQuery {
        subject: SmolStr::new(streamer_did),
        source: SmolStr::new("place.stream.chat.message:streamer"),
        limit: Some(100),
        cursor: cursor.map(SmolStr::new),
        did: vec![],
        reverse,
    };
    let params_data: Data =
        to_data(&query).map_err(|e| format!("failed to convert params to Data: {e}"))?;

    let hydrate_req = HydrateQueryRequest {
        xrpc: SmolStr::new("blue.microcosm.links.getBacklinks"),
        atproto_proxy: SmolStr::new("did:web:constellation.microcosm.blue#constellation"),
        authorization: None,
        atproto_accept_labelers: None,
        params: params_data,
        hydration_sources: vec![HydrationSource {
            path: SmolStr::new("records[]"),
            shape: Shape::AtUriParts,
        }],
    };

    let (ua_name, ua_value) = user_agent_header();
    let resp = client
        .xrpc(slingshot_uri.clone())
        .header(ua_name, ua_value)
        .send(&hydrate_req)
        .await
        .map_err(|e| format!("slingshot request failed: {e}"))?;
    let hydrate_output: HydrateQueryResponse = resp
        .into_output()
        .map_err(|e| format!("slingshot response parse failed: {e}"))?;

    // Extract backlinks cursor from output
    let backlinks: GetBacklinksResponse =
        from_data(&hydrate_output.output).map_err(|e| format!("backlinks parse failed: {e}"))?;
    let response_cursor = backlinks.cursor.map(|s| s.to_string());

    // Extract hydrated records, collecting pending URIs for follow-up fetches
    let mut messages = Vec::new();
    let mut pending_uri_strs: Vec<&SmolStr> = Vec::new();
    for (uri_str, record) in &hydrate_output.records {
        match record {
            HydratedRecord::Found { value, .. } => {
                let chat_msg: ChatMessage = match from_data(value) {
                    Ok(m) => m,
                    Err(e) => {
                        warn!("failed to parse hydrated chat message {uri_str}: {e}");
                        continue;
                    }
                };

                // Extract author DID and rkey from the AT URI
                let (author_did, rkey) = match parse_at_uri_parts(uri_str) {
                    Some(parts) => parts,
                    None => {
                        warn!("failed to parse AT URI parts from hydrated record key: {uri_str}");
                        continue;
                    }
                };

                messages.push(message_to_cached(
                    &chat_msg,
                    streamer_did,
                    &author_did,
                    &rkey,
                ));
            }
            HydratedRecord::Pending { reason, .. } => {
                // Slingshot hasn't cached this record yet but it may be available on the PDS.
                // Collect for individual fetch below.
                warn!("slingshot record pending for {uri_str}: {reason}");
                pending_uri_strs.push(uri_str);
            }
            HydratedRecord::Error { reason, .. } => {
                warn!("slingshot record error for {uri_str}: {reason}");
            }
        }
    }

    // Follow up on pending records — fetch them individually from the PDS.
    // Slingshot marks records Pending when its eager firehose cache hasn't ingested
    // them yet; the PDS itself usually has them.
    if !pending_uri_strs.is_empty() {
        use vodplace_api::place_stream::chat::message::MessageRecord;

        warn!(
            "{} pending record(s) from slingshot, fetching individually",
            pending_uri_strs.len()
        );

        let fetch_futures = pending_uri_strs.iter().filter_map(|uri_str| {
            let at_uri: AtUri = match uri_str.parse() {
                Ok(u) => u,
                Err(e) => {
                    warn!("failed to parse pending URI {uri_str}: {e}");
                    return None;
                }
            };
            let client = client.clone();
            let streamer_did = streamer_did.to_string();
            Some(async move {
                let resp = match client.get_record::<MessageRecord, _>(&at_uri).await {
                    Ok(r) => r,
                    Err(e) => {
                        warn!("failed to fetch pending record {}: {e}", at_uri.as_str());
                        return None;
                    }
                };
                let output = match resp.into_output() {
                    Ok(o) => o,
                    Err(e) => {
                        warn!("failed to parse pending record {}: {e}", at_uri.as_str());
                        return None;
                    }
                };
                let chat_msg = output.value;
                let (author_did, rkey) = match parse_at_uri_parts(at_uri.as_str()) {
                    Some(parts) => parts,
                    None => {
                        warn!(
                            "failed to parse AT URI parts from pending record: {}",
                            at_uri.as_str()
                        );
                        return None;
                    }
                };
                Some(message_to_cached(
                    &chat_msg,
                    &streamer_did,
                    &author_did,
                    &rkey,
                ))
            })
        });

        let pending_results: Vec<CachedChatMessage> = join_all(fetch_futures)
            .await
            .into_iter()
            .flatten()
            .collect();
        messages.extend(pending_results);
    }

    Ok((messages, response_cursor))
}

/// Parse author DID and rkey from an AT URI string like
/// `at://did:plc:xxx/place.stream.chat.message/rkey123`.
#[cfg(feature = "server")]
fn parse_at_uri_parts(uri_str: &str) -> Option<(String, String)> {
    let at_uri: AtUri = uri_str.parse().ok()?;
    let authority = at_uri.authority().as_ref().to_string();
    let rkey = at_uri.rkey()?.as_ref().to_string();
    Some((authority, rkey))
}

/// Convert a parsed chat message record into a `CachedChatMessage` for storage.
///
/// Extracted to avoid duplicating the field mapping between `fetch_via_slingshot`
/// and `fetch_fallback`.
#[cfg(feature = "server")]
fn message_to_cached(
    msg: &vodplace_api::place_stream::chat::message::Message,
    streamer_did: &str,
    author_did: &str,
    rkey: &str,
) -> crate::chat_cache::CachedChatMessage {
    use crate::chat::datetime_to_ms;
    crate::chat_cache::CachedChatMessage {
        streamer_did: streamer_did.to_string(),
        author_did: author_did.to_string(),
        rkey: rkey.to_string(),
        created_at_ms: datetime_to_ms(&msg.created_at),
        text: msg.text.to_string(),
        reply_parent_uri: msg
            .reply
            .as_ref()
            .map(|r| r.parent.uri.as_ref().to_string()),
        reply_root_uri: msg.reply.as_ref().map(|r| r.root.uri.as_ref().to_string()),
        facets_json: msg
            .facets
            .as_ref()
            .and_then(|f| serde_json::to_string(f).ok()),
    }
}

/// Fallback: fetch chat messages directly via Constellation + individual get_record calls.
///
/// Used when slingshot is unavailable. Slower but functional.
#[cfg(feature = "server")]
async fn fetch_fallback(
    client: &Arc<BasicClient>,
    streamer_did: &str,
    reverse: bool,
    cursor: Option<&str>,
) -> Result<(Vec<crate::chat_cache::CachedChatMessage>, Option<String>), String> {
    use crate::chat_cache::CachedChatMessage;
    use vodplace_api::place_stream::chat::message::MessageRecord;

    let constellation_uri =
        Uri::parse(CONSTELLATION_URL).map_err(|e| format!("bad constellation URL: {e}"))?;

    let query = GetBacklinksQuery {
        subject: SmolStr::new(streamer_did),
        source: SmolStr::new("place.stream.chat.message:streamer"),
        limit: Some(100),
        cursor: cursor.map(SmolStr::new),
        did: vec![],
        reverse,
    };

    let (ua_name, ua_value) = user_agent_header();
    let resp = client
        .xrpc(constellation_uri)
        .header(ua_name, ua_value)
        .send(&query)
        .await
        .map_err(|e| format!("constellation request failed: {e}"))?;
    let backlinks: GetBacklinksResponse = resp
        .into_output()
        .map_err(|e| format!("constellation response parse failed: {e}"))?;
    let response_cursor = backlinks.cursor.map(|s| s.to_string());

    // Fetch each record from its author's PDS in parallel
    let record_futures = backlinks.records.iter().map(|rec| {
        let client = client.clone();
        let did = rec.did.clone();
        let collection = rec.collection.clone();
        let rkey = rec.rkey.clone();
        let streamer_did = streamer_did.to_string();
        async move {
            let uri: AtUri = match AtUri::from_parts_owned(&did, collection.as_ref(), &rkey) {
                Ok(u) => u,
                Err(e) => {
                    warn!(
                        did = %did.as_str(),
                        rkey = %rkey.as_str(),
                        "failed to construct chat message URI: {e}"
                    );
                    return None;
                }
            };

            let resp = match client.get_record::<MessageRecord, _>(&uri).await {
                Ok(r) => r,
                Err(e) => {
                    warn!(uri = %uri.as_str(), "failed to fetch chat message: {e}");
                    return None;
                }
            };

            let output = match resp.into_output() {
                Ok(o) => o,
                Err(e) => {
                    warn!(uri = %uri.as_str(), "failed to parse chat message: {e}");
                    return None;
                }
            };

            let chat_msg = output.value;

            Some(message_to_cached(
                &chat_msg,
                &streamer_did,
                did.as_str(),
                rkey.as_str(),
            ))
        }
    });

    let messages: Vec<CachedChatMessage> = join_all(record_futures)
        .await
        .into_iter()
        .flatten()
        .collect();

    Ok((messages, response_cursor))
}

#[cfg(feature = "server")]
pub async fn thumbnail_handler(
    axum::extract::Query(params): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> axum::response::Response {
    use axum::http::{StatusCode, header};
    use axum::response::IntoResponse;

    let Some(uri_str) = params.get("uri") else {
        return (StatusCode::BAD_REQUEST, "missing uri parameter").into_response();
    };

    let at_uri: AtUri = match uri_str.parse() {
        Ok(u) => u,
        Err(_) => return (StatusCode::BAD_REQUEST, "invalid AT URI").into_response(),
    };

    // Try cache first, then generate on miss
    let bytes = if let Some(cached) = get_cached_thumbnail(&at_uri) {
        cached
    } else if let Some(generated) = generate_thumbnail(&at_uri).await {
        generated
    } else {
        return (StatusCode::NOT_FOUND, "thumbnail generation failed").into_response();
    };

    (
        StatusCode::OK,
        [
            (header::CONTENT_TYPE, "image/jpeg"),
            (header::CACHE_CONTROL, "public, max-age=86400"),
        ],
        bytes,
    )
        .into_response()
}

pub fn filter_and_sort(videos: &[EnrichedVideo], query: &str) -> Vec<EnrichedVideo> {
    let query = query.to_lowercase();
    let contains = |s: &str| s.to_lowercase().contains(&query);

    let mut filtered: Vec<_> = if query.is_empty() {
        videos.to_vec()
    } else {
        videos
            .iter()
            .filter(|v| {
                contains(v.title.as_ref())
                    || contains(v.creator.handle.as_ref())
                    || v.creator.display_name.as_deref().is_some_and(&contains)
                    || v.speakers.iter().any(|s| contains(s.name.as_ref()))
                    || v.talk.as_ref().is_some_and(|t| {
                        t.description.as_deref().is_some_and(&contains)
                            || t.category.as_deref().is_some_and(&contains)
                            || t.talk_type.as_deref().is_some_and(&contains)
                            || t.room.as_deref().is_some_and(&contains)
                    })
            })
            .cloned()
            .collect()
    };

    let sort_date = |v: &EnrichedVideo| {
        v.talk
            .as_ref()
            .and_then(|t| t.starts_at.clone())
            .unwrap_or_else(|| v.created_at.clone())
    };
    filtered.sort_by(|a, b| sort_date(b).cmp(&sort_date(a)));
    filtered
}

#[cfg(test)]
mod tests {
    use super::*;
    use jacquard::deps::smol_str::SmolStr;

    fn test_video(uri: &str, title: &str) -> EnrichedVideo {
        EnrichedVideo {
            at_uri: uri.parse().unwrap(),
            cid: None,
            title: SmolStr::new(title),
            duration_ns: 60_000_000_000,
            created_at: Datetime::now(),
            creator: ProfileInfo {
                did: "did:plc:test123".parse().unwrap(),
                handle: "test.bsky.social".parse().unwrap(),
                display_name: None,
                avatar_url: None,
            },
            thumbnail_url: None,
            livestream_uri: None,
            speakers: vec![],
            talk: None,
        }
    }

    #[test]
    fn enriched_video_roundtrip() {
        let mut video = test_video("at://did:plc:test123/place.stream.video/abc", "Test Video");
        video.creator.display_name = Some(SmolStr::new("Test User"));
        video.thumbnail_url = Some(SmolStr::new("https://example.com/thumb.jpg"));

        let json = serde_json::to_string(&video).unwrap();
        let roundtripped: EnrichedVideo = serde_json::from_str(&json).unwrap();
        assert_eq!(video, roundtripped);
    }

    #[test]
    fn thumbnail_cdn_url_format() {
        let url = thumbnail_cdn_url(
            "did:plc:abc123",
            "bafyreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
        );
        assert_eq!(
            url.as_str(),
            "https://cdn.bsky.app/img/feed_thumbnail/plain/did:plc:abc123/bafyreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa@jpeg"
        );
    }

    #[test]
    fn deduplication_by_at_uri() {
        let video1 = test_video("at://did:plc:aaa/place.stream.video/111", "Video A");
        let video2 = test_video(
            "at://did:plc:aaa/place.stream.video/111",
            "Video A duplicate",
        );
        let video3 = test_video("at://did:plc:bbb/place.stream.video/222", "Video B");

        // Mirror the actual catalog deduplication pattern: HashMap<AtUri, _>
        let mut map = std::collections::HashMap::new();
        // First source inserts
        map.insert(video1.at_uri.clone(), video1);
        // Duplicate from second source -- or_insert preserves first
        map.entry(video2.at_uri.clone()).or_insert(video2);
        // Different URI inserts fine
        map.insert(video3.at_uri.clone(), video3);

        assert_eq!(map.len(), 2);
        let uri_a: AtUri = "at://did:plc:aaa/place.stream.video/111".parse().unwrap();
        let uri_b: AtUri = "at://did:plc:bbb/place.stream.video/222".parse().unwrap();
        assert!(map.contains_key(&uri_a));
        assert!(map.contains_key(&uri_b));
        // Verify first-wins semantics
        assert_eq!(map[&uri_a].title.as_str(), "Video A");
    }

    #[test]
    fn pds_results_override_ufos() {
        let ufos_video = test_video("at://did:plc:aaa/place.stream.video/111", "From ufos");
        let mut pds_video = test_video("at://did:plc:aaa/place.stream.video/111", "From PDS");
        pds_video.cid = Some(
            "bafyreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                .parse()
                .unwrap(),
        );

        let mut map = std::collections::HashMap::new();
        // ufos inserts first
        map.insert(ufos_video.at_uri.clone(), ufos_video);
        // PDS overwrites (insert, not entry/or_insert) -- PDS has CIDs
        map.insert(pds_video.at_uri.clone(), pds_video);

        let uri: AtUri = "at://did:plc:aaa/place.stream.video/111".parse().unwrap();
        assert_eq!(map[&uri].title.as_str(), "From PDS");
        assert!(map[&uri].cid.is_some());
    }

    fn test_video_with_creator(
        uri: &str,
        title: &str,
        handle: &str,
        display_name: Option<&str>,
    ) -> EnrichedVideo {
        EnrichedVideo {
            at_uri: uri.parse().unwrap(),
            cid: None,
            title: SmolStr::new(title),
            duration_ns: 60_000_000_000,
            created_at: Datetime::now(),
            creator: ProfileInfo {
                did: "did:plc:test123".parse().unwrap(),
                handle: handle.parse().unwrap(),
                display_name: display_name.map(SmolStr::new),
                avatar_url: None,
            },
            thumbnail_url: None,
            livestream_uri: None,
            speakers: vec![],
            talk: None,
        }
    }

    #[test]
    fn filter_empty_query_returns_all() {
        let videos = vec![
            test_video("at://did:plc:aaa/place.stream.video/1", "Alpha"),
            test_video("at://did:plc:bbb/place.stream.video/2", "Beta"),
        ];
        let result = filter_and_sort(&videos, "");
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn filter_by_title() {
        let videos = vec![
            test_video("at://did:plc:aaa/place.stream.video/1", "Rust Talk"),
            test_video("at://did:plc:bbb/place.stream.video/2", "Go Workshop"),
        ];
        let result = filter_and_sort(&videos, "rust");
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].title.as_str(), "Rust Talk");
    }

    #[test]
    fn filter_by_handle() {
        let videos = vec![
            test_video_with_creator(
                "at://did:plc:aaa/place.stream.video/1",
                "Some Video",
                "alice.bsky.social",
                None,
            ),
            test_video_with_creator(
                "at://did:plc:bbb/place.stream.video/2",
                "Other Video",
                "bob.bsky.social",
                None,
            ),
        ];
        let result = filter_and_sort(&videos, "alice");
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].creator.handle.as_str(), "alice.bsky.social");
    }

    #[test]
    fn filter_by_display_name() {
        let videos = vec![
            test_video_with_creator(
                "at://did:plc:aaa/place.stream.video/1",
                "Some Video",
                "alice.bsky.social",
                Some("Alice Wonderland"),
            ),
            test_video_with_creator(
                "at://did:plc:bbb/place.stream.video/2",
                "Other Video",
                "bob.bsky.social",
                Some("Bob Builder"),
            ),
        ];
        let result = filter_and_sort(&videos, "wonderland");
        assert_eq!(result.len(), 1);
        assert_eq!(
            result[0].creator.display_name.as_deref(),
            Some("Alice Wonderland")
        );
    }

    #[test]
    fn filter_is_case_insensitive() {
        let videos = vec![test_video(
            "at://did:plc:aaa/place.stream.video/1",
            "LOUD TITLE",
        )];
        let result = filter_and_sort(&videos, "loud");
        assert_eq!(result.len(), 1);
    }

    #[test]
    fn filter_no_match_returns_empty() {
        let videos = vec![test_video(
            "at://did:plc:aaa/place.stream.video/1",
            "Something",
        )];
        let result = filter_and_sort(&videos, "nonexistent");
        assert!(result.is_empty());
    }

    #[test]
    fn sort_newest_first() {
        use chrono::{FixedOffset, TimeZone};
        let utc = FixedOffset::east_opt(0).unwrap();

        let mut older = test_video("at://did:plc:aaa/place.stream.video/1", "Older");
        older.created_at = Datetime::new(utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).unwrap());

        let mut newer = test_video("at://did:plc:bbb/place.stream.video/2", "Newer");
        newer.created_at = Datetime::new(utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap());

        // Pass them in wrong order
        let videos = vec![older, newer];
        let result = filter_and_sort(&videos, "");
        assert_eq!(result[0].title.as_str(), "Newer");
        assert_eq!(result[1].title.as_str(), "Older");
    }

    #[test]
    fn thumbnail_url_none_without_livestream() {
        let video = test_video("at://did:plc:aaa/place.stream.video/1", "No Thumbnail");
        assert!(video.livestream_uri.is_none());
        assert!(video.thumbnail_url.is_none());
    }

    #[test]
    fn empty_speakers_without_talk() {
        let video = test_video("at://did:plc:aaa/place.stream.video/1", "No Talk Data");
        assert!(video.talk.is_none());
        assert!(video.speakers.is_empty());
    }

    #[test]
    fn comment_data_serialization_roundtrip() {
        use crate::chat::{FacetData, FacetFeatureData};

        let original = CommentData {
            uri: "at://did:plc:test000000000000000000000001/tv.ionosphere.comment/abc"
                .parse()
                .unwrap(),
            text: SmolStr::new("test comment"),
            author_did: "did:plc:test000000000000000000001".parse().unwrap(),
            author_handle: "test.bsky.social".parse().unwrap(),
            created_at: "2026-04-01T12:00:00.000Z".parse().unwrap(),
            anchor: Some(ByteRangeData {
                byte_start: 10,
                byte_end: 20,
            }),
            subject_uri: "at://did:plc:test000000000000000000000001/tv.ionosphere.talk/xyz"
                .parse()
                .unwrap(),
            facets: Some(vec![FacetData {
                byte_start: 0,
                byte_end: 4,
                features: vec![FacetFeatureData::Link {
                    uri: SmolStr::new("https://example.com"),
                }],
            }]),
        };

        let json = serde_json::to_string(&original).expect("serialize");
        let roundtripped: CommentData = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(original, roundtripped);
    }
}