Something went wrong. Try again.
Streamplace AtmosphereConf 2026 VOD JAM
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578use 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); }}