diff --git a/.sqlx/query-098c1e752ec1e822d32a76fdfbae5d8bfae01c8e4942518a15bc16f7fa5e025b.json b/.sqlx/query-098c1e752ec1e822d32a76fdfbae5d8bfae01c8e4942518a15bc16f7fa5e025b.json new file mode 100644 index 0000000..5accd30 --- /dev/null +++ b/.sqlx/query-098c1e752ec1e822d32a76fdfbae5d8bfae01c8e4942518a15bc16f7fa5e025b.json @@ -0,0 +1,139 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n p.uri, p.did, p.rkey, p.cid, p.isrc, p.duration, p.track_name, p.played_time,\n p.processed_time, p.release_mbid, p.release_name, p.recording_mbid,\n p.submission_client_agent, p.music_service_base_domain, p.origin_url,\n profile.did AS \"profile_did?\", profile.handle AS profile_handle,\n profile.display_name AS profile_display_name, profile.avatar AS profile_avatar,\n COALESCE(\n json_agg(\n json_build_object(\n 'artistMbId', ae.mbid,\n 'artistName', ptae.artist_name\n )\n ) FILTER (WHERE ptae.artist_name IS NOT NULL),\n '[]'\n ) AS artists\n FROM plays p\n LEFT JOIN profiles profile ON p.did = profile.did\n LEFT JOIN play_to_artists_extended as ptae ON p.uri = ptae.play_uri\n LEFT JOIN artists_extended as ae ON ptae.artist_id = ae.id\n WHERE p.did = $1\n AND ($2::timestamptz IS NULL OR (p.processed_time, p.uri) < ($2, $3))\n GROUP BY p.uri, p.did, p.rkey, p.cid, p.isrc, p.duration, p.track_name, p.played_time,\n p.processed_time, p.release_mbid, p.release_name, p.recording_mbid,\n p.submission_client_agent, p.music_service_base_domain, p.origin_url,\n profile.did, profile.handle, profile.display_name, profile.avatar\n ORDER BY p.processed_time DESC, p.uri DESC\n LIMIT $4\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "uri", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "did", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "rkey", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "cid", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "isrc", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "duration", + "type_info": "Int4" + }, + { + "ordinal": 6, + "name": "track_name", + "type_info": "Text" + }, + { + "ordinal": 7, + "name": "played_time", + "type_info": "Timestamptz" + }, + { + "ordinal": 8, + "name": "processed_time", + "type_info": "Timestamptz" + }, + { + "ordinal": 9, + "name": "release_mbid", + "type_info": "Uuid" + }, + { + "ordinal": 10, + "name": "release_name", + "type_info": "Text" + }, + { + "ordinal": 11, + "name": "recording_mbid", + "type_info": "Uuid" + }, + { + "ordinal": 12, + "name": "submission_client_agent", + "type_info": "Text" + }, + { + "ordinal": 13, + "name": "music_service_base_domain", + "type_info": "Text" + }, + { + "ordinal": 14, + "name": "origin_url", + "type_info": "Text" + }, + { + "ordinal": 15, + "name": "profile_did?", + "type_info": "Text" + }, + { + "ordinal": 16, + "name": "profile_handle", + "type_info": "Text" + }, + { + "ordinal": 17, + "name": "profile_display_name", + "type_info": "Text" + }, + { + "ordinal": 18, + "name": "profile_avatar", + "type_info": "Text" + }, + { + "ordinal": 19, + "name": "artists", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "Text", + "Timestamptz", + "Text", + "Int8" + ] + }, + "nullable": [ + false, + false, + false, + false, + true, + true, + false, + true, + true, + true, + true, + true, + true, + true, + true, + false, + true, + true, + true, + null + ] + }, + "hash": "098c1e752ec1e822d32a76fdfbae5d8bfae01c8e4942518a15bc16f7fa5e025b" +} diff --git a/apps/amethyst/app/(tabs)/profile/[handle].tsx b/apps/amethyst/app/(tabs)/profile/[handle].tsx index 2141ea9..dfc4c84 100644 --- a/apps/amethyst/app/(tabs)/profile/[handle].tsx +++ b/apps/amethyst/app/(tabs)/profile/[handle].tsx @@ -1,5 +1,11 @@ -import { useEffect, useState } from "react"; -import { ActivityIndicator, Image, View } from "react-native"; +import { useCallback, useEffect, useRef, useState } from "react"; +import { + ActivityIndicator, + Image, + View, + type NativeScrollEvent, + type NativeSyntheticEvent, +} from "react-native"; import { Link, Stack, useLocalSearchParams } from "expo-router"; import PlayFeedCard from "@/components/teal/PlayFeedCard"; import RightRail from "@/components/teal/RightRail"; @@ -40,8 +46,11 @@ export default function ProfileScreen() { const [did, setDid] = useState(null); const [profile, setProfile] = useState(null); const [plays, setPlays] = useState([]); + const [cursor, setCursor] = useState(); + const [loadingMore, setLoadingMore] = useState(false); const [isBlueskyFallback, setIsBlueskyFallback] = useState(false); const [error, setError] = useState(null); + const loadingMoreRef = useRef(false); const pdsAgent = useStore((state) => state.pdsAgent); useEffect(() => { @@ -49,12 +58,17 @@ export default function ProfileScreen() { async function load() { if (!actor) return; try { + setError(null); + setDid(null); + setProfile(null); + setPlays([]); + setCursor(undefined); const resolved = actor.startsWith("did:") ? actor : await resolveHandle(actor); if (!mounted) return; setDid(resolved); - const feedRes = await getActorFeed(resolved, 50); + const feedRes = await getActorFeed(resolved, 30); let nextProfile: DisplayProfile | null = null; let nextIsBlueskyFallback = false; @@ -78,6 +92,7 @@ export default function ProfileScreen() { setProfile(nextProfile); setIsBlueskyFallback(nextIsBlueskyFallback); setPlays(feedRes.plays); + setCursor(feedRes.cursor); } catch (e) { if (mounted) setError(e instanceof Error ? e.message : String(e)); } @@ -88,10 +103,47 @@ export default function ProfileScreen() { }; }, [actor]); + const loadMore = useCallback(() => { + if (!did || !cursor || loadingMoreRef.current) return; + loadingMoreRef.current = true; + setLoadingMore(true); + getActorFeed(did, 30, cursor) + .then((feedRes) => { + setPlays((current) => { + const knownUris = new Set(current.map((play) => play.uri)); + return [ + ...current, + ...feedRes.plays.filter( + (play) => !play.uri || !knownUris.has(play.uri), + ), + ]; + }); + setCursor(feedRes.cursor); + }) + .catch((e) => { + setError(e instanceof Error ? e.message : String(e)); + }) + .finally(() => { + loadingMoreRef.current = false; + setLoadingMore(false); + }); + }, [cursor, did]); + + const handleScroll = useCallback( + (event: NativeSyntheticEvent) => { + const { contentOffset, contentSize, layoutMeasurement } = + event.nativeEvent; + const remaining = + contentSize.height - (contentOffset.y + layoutMeasurement.height); + if (remaining < 800) loadMore(); + }, + [loadMore], + ); + const isSelf = did === pdsAgent?.did; return ( - }> + } onScroll={handleScroll}> @@ -178,6 +230,16 @@ export default function ProfileScreen() { /> )) )} + {loadingMore && ( + + + + )} + {plays.length > 0 && !cursor && ( + + You reached the beginning of this listener's indexed plays. + + )} )} diff --git a/apps/amethyst/components/teal/PlayFeedCard.tsx b/apps/amethyst/components/teal/PlayFeedCard.tsx index 56d2153..6ea839d 100644 --- a/apps/amethyst/components/teal/PlayFeedCard.tsx +++ b/apps/amethyst/components/teal/PlayFeedCard.tsx @@ -54,7 +54,7 @@ export function musicHref(play: PlayView) { export default function PlayFeedCard({ play, compact }: PlayFeedCardProps) { const [blueskyAuthor, setBlueskyAuthor] = useState(); const [artFailed, setArtFailed] = useState(false); - const indexedAuthor = play.author; + const indexedAuthor = play.author as FeedAuthor | undefined; const authorProfile = indexedAuthor || blueskyAuthor; const authorDid = authorProfile?.did || play.authorDid; diff --git a/apps/amethyst/lib/teal/api.ts b/apps/amethyst/lib/teal/api.ts index dbaf1cf..cf3fc05 100644 --- a/apps/amethyst/lib/teal/api.ts +++ b/apps/amethyst/lib/teal/api.ts @@ -64,11 +64,15 @@ export function getLatestPlays(limit = 50, cursor?: string) { ); } -export function getActorFeed(authorDID: string, limit = 50) { - return getXrpc<{ plays: PlayView[] }>("fm.teal.alpha.feed.getActorFeed", { - authorDID, - limit, - }); +export function getActorFeed(authorDID: string, limit = 30, cursor?: string) { + return getXrpc<{ plays: PlayView[]; cursor?: string }>( + "fm.teal.alpha.feed.getActorFeed", + { + authorDID, + limit, + cursor, + }, + ); } export function getPlayByUri(uri: string) { diff --git a/apps/aqua/src/repos/feed_play.rs b/apps/aqua/src/repos/feed_play.rs index b76578b..bd117c7 100644 --- a/apps/aqua/src/repos/feed_play.rs +++ b/apps/aqua/src/repos/feed_play.rs @@ -3,11 +3,27 @@ use jacquard_common::from_json_value; use jacquard_common::types::string::{AtUri, Did}; use types::fm_teal::alpha::feed::{Artist, PlayView}; -use super::{mbid_uri, mini_profile, pg::PgDataSource, utc_to_atrium_datetime}; +use super::{ + mbid_uri, mini_profile, + pg::PgDataSource, + stats::{LatestPlaysCursor, decode_latest_cursor, encode_latest_cursor}, + utc_to_atrium_datetime, +}; + +pub struct ActorFeedPage { + pub plays: Vec, + pub cursor: Option, +} #[async_trait] pub trait FeedPlayRepo: Send + Sync { async fn get_feed_play(&self, identity: &str) -> anyhow::Result>; + async fn get_actor_feed( + &self, + author_did: &str, + limit: Option, + cursor: Option<&str>, + ) -> anyhow::Result; async fn get_feed_plays_for_profile( &self, identities: &[String], @@ -84,6 +100,119 @@ impl FeedPlayRepo for PgDataSource { })) } + async fn get_actor_feed( + &self, + author_did: &str, + limit: Option, + cursor: Option<&str>, + ) -> anyhow::Result { + let limit = limit.unwrap_or(20).clamp(1, 50) as i64; + let query_limit = limit + 1; + let cursor = decode_latest_cursor(cursor)?; + let cursor_time = cursor + .as_ref() + .map(|cursor| { + time::OffsetDateTime::parse( + &cursor.processed_time, + &time::format_description::well_known::Rfc3339, + ) + }) + .transpose()?; + + let rows = sqlx::query!( + r#" + SELECT + p.uri, p.did, p.rkey, p.cid, p.isrc, p.duration, p.track_name, p.played_time, + p.processed_time, p.release_mbid, p.release_name, p.recording_mbid, + p.submission_client_agent, p.music_service_base_domain, p.origin_url, + profile.did AS "profile_did?", profile.handle AS profile_handle, + profile.display_name AS profile_display_name, profile.avatar AS profile_avatar, + COALESCE( + json_agg( + json_build_object( + 'artistMbId', ae.mbid, + 'artistName', ptae.artist_name + ) + ) FILTER (WHERE ptae.artist_name IS NOT NULL), + '[]' + ) AS artists + FROM plays p + LEFT JOIN profiles profile ON p.did = profile.did + LEFT JOIN play_to_artists_extended as ptae ON p.uri = ptae.play_uri + LEFT JOIN artists_extended as ae ON ptae.artist_id = ae.id + WHERE p.did = $1 + AND ($2::timestamptz IS NULL OR (p.processed_time, p.uri) < ($2, $3)) + GROUP BY p.uri, p.did, p.rkey, p.cid, p.isrc, p.duration, p.track_name, p.played_time, + p.processed_time, p.release_mbid, p.release_name, p.recording_mbid, + p.submission_client_agent, p.music_service_base_domain, p.origin_url, + profile.did, profile.handle, profile.display_name, profile.avatar + ORDER BY p.processed_time DESC, p.uri DESC + LIMIT $4 + "#, + author_did, + cursor_time, + cursor.as_ref().map(|cursor| cursor.uri.as_str()), + query_limit + ) + .fetch_all(&self.db) + .await?; + + let has_more = rows.len() > limit as usize; + let mut plays = Vec::with_capacity(rows.len().min(limit as usize)); + let mut next_cursor = None; + for row in rows.into_iter().take(limit as usize) { + next_cursor = Some(LatestPlaysCursor { + processed_time: row + .processed_time + .unwrap_or_else(time::OffsetDateTime::now_utc) + .format(&time::format_description::well_known::Rfc3339)?, + uri: row.uri.clone(), + }); + + let artists: Vec = match row.artists { + Some(value) => from_json_value::>(value).unwrap_or_default(), + None => vec![], + }; + + plays.push(PlayView { + track_name: row.track_name.clone().into(), + author: mini_profile( + row.profile_did, + row.profile_handle, + row.profile_display_name, + row.profile_avatar, + ), + uri: AtUri::try_from(row.uri.clone()).ok(), + cid: Some(row.cid.clone().into()), + author_did: Did::new_owned(&row.did).ok(), + rkey: Some(row.rkey.clone().into()), + track_mb_id: row.recording_mbid.map(mbid_uri), + recording_mb_id: row.recording_mbid.map(mbid_uri), + duration: row.duration.map(|d| d as i64), + artists, + release_name: row.release_name.clone().map(|s| s.into()), + release_mb_id: row.release_mbid.map(mbid_uri), + isrc: row.isrc.map(|s| s.into()), + origin_url: row.origin_url.map(|s| s.into()), + music_service_base_domain: row.music_service_base_domain.map(|s| s.into()), + submission_client_agent: row.submission_client_agent.map(|s| s.into()), + played_time: row + .played_time + .map(|dt| utc_to_atrium_datetime(crate::repos::time_to_chrono_utc(dt))), + extra_data: Default::default(), + }); + } + + Ok(ActorFeedPage { + plays, + cursor: if has_more { + next_cursor.as_ref().map(encode_latest_cursor).transpose()? + } else { + None + }, + }) + } + async fn get_feed_plays_for_profile( &self, identities: &[String], diff --git a/apps/aqua/src/xrpc/feed.rs b/apps/aqua/src/xrpc/feed.rs index aaaf759..2244621 100644 --- a/apps/aqua/src/xrpc/feed.rs +++ b/apps/aqua/src/xrpc/feed.rs @@ -63,6 +63,7 @@ pub struct GetActorFeedQuery { #[derive(Serialize)] pub struct GetActorFeedResponse { plays: Vec, + cursor: Option, } pub async fn get_actor_feed( @@ -75,12 +76,13 @@ pub async fn get_actor_feed( return Err((StatusCode::BAD_REQUEST, "authorDID is required".to_string())); } - // Cursor and limit are accepted for lexicon compatibility; repository pagination is deferred. - let _ = (query.limit, query.cursor); - - match repo.get_feed_plays_for_profile(&[query.author_did]).await { - Ok(plays) => Ok(axum::Json(GetActorFeedResponse { - plays: plays.into_static(), + match repo + .get_actor_feed(&query.author_did, query.limit, query.cursor.as_deref()) + .await + { + Ok(page) => Ok(axum::Json(GetActorFeedResponse { + cursor: page.cursor, + plays: page.plays.into_static(), })), Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e.to_string())), } diff --git a/lexicons/fm.teal.alpha/feed/getActorFeed.json b/lexicons/fm.teal.alpha/feed/getActorFeed.json index 51b6c9d..645953a 100644 --- a/lexicons/fm.teal.alpha/feed/getActorFeed.json +++ b/lexicons/fm.teal.alpha/feed/getActorFeed.json @@ -36,6 +36,10 @@ "type": "ref", "ref": "fm.teal.alpha.feed.defs#playView" } + }, + "cursor": { + "type": "string", + "description": "Opaque cursor for the next page of plays" } } } diff --git a/todo.md b/todo.md index b360f2d..936e18c 100644 --- a/todo.md +++ b/todo.md @@ -40,7 +40,7 @@ This file is the working handoff for the Teal-native Teal clone. Keep it updated ## Next: Aqua And Lexicons -- [ ] Add pagination support for `fm.teal.alpha.feed.getActorFeed` cursor and limit parameters. +- [x] Add pagination support for `fm.teal.alpha.feed.getActorFeed` cursor and limit parameters. - [x] Add keyset pagination and infinite scrolling for the global latest-play feed. - [x] Run SQLx prepare against the development Postgres instance and commit refreshed query cache data. - [ ] Resolve the existing Satellite SQLx offline-cache gap so `pnpm turbo run test:rust` passes without a live Docker hostname.