From 7bb7eb6c2c14d70b4de248c3143004b64ca1b685 Mon Sep 17 00:00:00 2001 From: Trezy Date: Tue, 12 May 2026 15:17:54 -0500 Subject: [PATCH] fix: use cursors for pagination in spaces endpoints Signed-off-by: Trezy Signed-off-by: Trezy --- .../experimental/spaces/managing-spaces.md | 2 +- .../docs/experimental/spaces/records.md | 2 +- src/db.rs | 14 ++++ src/lua/atproto_api.rs | 4 +- src/lua/db_api.rs | 15 +--- src/spaces/db.rs | 79 +++++++++++-------- src/spaces/routes.rs | 12 +-- 7 files changed, 67 insertions(+), 61 deletions(-) diff --git a/packages/docs/content/docs/experimental/spaces/managing-spaces.md b/packages/docs/content/docs/experimental/spaces/managing-spaces.md index d588d7a..581ce49 100644 --- a/packages/docs/content/docs/experimental/spaces/managing-spaces.md +++ b/packages/docs/content/docs/experimental/spaces/managing-spaces.md @@ -95,7 +95,7 @@ curl 'https://happyview.example.com/xrpc/dev.happyview.space.listSpaces?limit=20 "isOwner": true } ], - "cursor": "2026-05-09T12:00:00Z" + "cursor": "MjAyNi0wNS0wOVQxMjowMDowMFp8YXRzOi8vZGlkOnBsYzphYmMxMjMvY29tLmV4YW1wbGUuZm9ydW0vbWFpbg" } ``` diff --git a/packages/docs/content/docs/experimental/spaces/records.md b/packages/docs/content/docs/experimental/spaces/records.md index 25acf2e..a848ff7 100644 --- a/packages/docs/content/docs/experimental/spaces/records.md +++ b/packages/docs/content/docs/experimental/spaces/records.md @@ -162,7 +162,7 @@ curl 'https://happyview.example.com/xrpc/dev.happyview.space.listRecords?space=a "cid": "bafyrei..." } ], - "cursor": "2026-05-09T12:00:00Z" + "cursor": "MjAyNi0wNS0wOVQxMjowMDowMFp8YXRzOi8vZGlkOnBsYzphYmMxMjMvY29tLmV4YW1wbGUuZm9ydW0vbWFpbg" } ``` diff --git a/src/db.rs b/src/db.rs index f8fc963..19f571a 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1,3 +1,4 @@ +use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; use chrono::{DateTime, Utc}; use regex::Regex; use serde::Deserialize; @@ -185,6 +186,19 @@ pub fn now_rfc3339() -> String { Utc::now().to_rfc3339() } +/// Encode a pagination cursor from a timestamp and URI. +pub fn encode_cursor(timestamp: &str, uri: &str) -> String { + BASE64.encode(format!("{timestamp}|{uri}")) +} + +/// Decode a pagination cursor into (timestamp, uri). Returns None if invalid. +pub fn decode_cursor(cursor: &str) -> Option<(String, String)> { + let decoded = BASE64.decode(cursor).ok()?; + let s = String::from_utf8(decoded).ok()?; + let (ts, uri) = s.split_once('|')?; + Some((ts.to_string(), uri.to_string())) +} + /// Connect to the configured database and run migrations. pub async fn connect(url: &str, backend: DatabaseBackend) -> AnyPool { sqlx::any::install_default_drivers(); diff --git a/src/lua/atproto_api.rs b/src/lua/atproto_api.rs index e1669a6..e70e488 100644 --- a/src/lua/atproto_api.rs +++ b/src/lua/atproto_api.rs @@ -429,7 +429,7 @@ pub fn register_atproto_api( } }; - let records = crate::spaces::db::list_space_records( + let (records, next_cursor) = crate::spaces::db::list_space_records( &state.db, state.db_backend, &space.id, @@ -442,8 +442,6 @@ pub fn register_atproto_api( .await .map_err(|e| mlua::Error::runtime(format!("record query failed: {e}")))?; - let next_cursor = records.last().map(|r| r.indexed_at.clone()); - let result = lua.create_table()?; let records_table = lua.create_table()?; for (i, record) in records.iter().enumerate() { diff --git a/src/lua/db_api.rs b/src/lua/db_api.rs index e2fb031..1937267 100644 --- a/src/lua/db_api.rs +++ b/src/lua/db_api.rs @@ -5,20 +5,7 @@ use sqlx::{Column, Row}; use std::sync::Arc; use crate::AppState; -use crate::db::{DatabaseBackend, adapt_sql}; - -/// Encode a cursor from created_at timestamp and uri. -fn encode_cursor(created_at: &str, uri: &str) -> String { - BASE64.encode(format!("{created_at}|{uri}")) -} - -/// Decode a cursor into (created_at, uri). Returns None if invalid. -fn decode_cursor(cursor: &str) -> Option<(String, String)> { - let decoded = BASE64.decode(cursor).ok()?; - let s = String::from_utf8(decoded).ok()?; - let (ts, uri) = s.split_once('|')?; - Some((ts.to_string(), uri.to_string())) -} +use crate::db::{DatabaseBackend, adapt_sql, decode_cursor, encode_cursor}; /// Register the `db` table with database query functions. pub fn register_db_api(lua: &Lua, state: Arc) -> LuaResult<()> { diff --git a/src/spaces/db.rs b/src/spaces/db.rs index f6eae7b..63196fc 100644 --- a/src/spaces/db.rs +++ b/src/spaces/db.rs @@ -1,4 +1,4 @@ -use crate::db::{DatabaseBackend, adapt_sql, now_rfc3339}; +use crate::db::{DatabaseBackend, adapt_sql, decode_cursor, encode_cursor, now_rfc3339}; use crate::error::AppError; use crate::spaces::types::*; @@ -123,28 +123,24 @@ pub async fn list_spaces_for_user( did: &str, limit: i64, cursor: Option<&str>, -) -> Result, AppError> { - let (sql, has_cursor) = if cursor.is_some() { - ( - adapt_sql( - "SELECT s.did, s.owner_did, s.type_nsid, s.skey, sm.created_at FROM space_members sm JOIN spaces s ON s.id = sm.space_id WHERE sm.member_did = ? AND sm.created_at > ? ORDER BY sm.created_at ASC LIMIT ?", - backend, - ), - true, +) -> Result<(Vec, Option), AppError> { + let decoded_cursor = cursor.and_then(decode_cursor); + + let sql = if decoded_cursor.is_some() { + adapt_sql( + "SELECT s.did, s.owner_did, s.type_nsid, s.skey, sm.created_at FROM space_members sm JOIN spaces s ON s.id = sm.space_id WHERE sm.member_did = ? AND (sm.created_at > ? OR (sm.created_at = ? AND ('ats://' || s.did || '/' || s.type_nsid || '/' || s.skey) > ?)) ORDER BY sm.created_at ASC, ('ats://' || s.did || '/' || s.type_nsid || '/' || s.skey) ASC LIMIT ?", + backend, ) } else { - ( - adapt_sql( - "SELECT s.did, s.owner_did, s.type_nsid, s.skey, sm.created_at FROM space_members sm JOIN spaces s ON s.id = sm.space_id WHERE sm.member_did = ? ORDER BY sm.created_at ASC LIMIT ?", - backend, - ), - false, + adapt_sql( + "SELECT s.did, s.owner_did, s.type_nsid, s.skey, sm.created_at FROM space_members sm JOIN spaces s ON s.id = sm.space_id WHERE sm.member_did = ? ORDER BY sm.created_at ASC, ('ats://' || s.did || '/' || s.type_nsid || '/' || s.skey) ASC LIMIT ?", + backend, ) }; let mut query = sqlx::query_as::<_, (String, String, String, String, String)>(&sql).bind(did); - if has_cursor { - query = query.bind(cursor.unwrap()); + if let Some((ref ts, ref uri)) = decoded_cursor { + query = query.bind(ts.as_str()).bind(ts.as_str()).bind(uri.as_str()); } query = query.bind(limit); @@ -153,7 +149,7 @@ pub async fn list_spaces_for_user( .await .map_err(|e| AppError::Internal(format!("failed to list spaces for user: {e}")))?; - Ok(rows + let views: Vec = rows .into_iter() .map( |(space_did, owner_did, type_nsid, skey, created_at)| SpaceView { @@ -162,7 +158,15 @@ pub async fn list_spaces_for_user( created_at, }, ) - .collect()) + .collect(); + + let next_cursor = if views.len() as i64 == limit { + views.last().map(|v| encode_cursor(&v.created_at, &v.uri)) + } else { + None + }; + + Ok((views, next_cursor)) } pub async fn update_space( @@ -491,7 +495,9 @@ pub async fn list_space_records( limit: i64, cursor: Option<&str>, reverse: bool, -) -> Result, AppError> { +) -> Result<(Vec, Option), AppError> { + let decoded_cursor = cursor.and_then(decode_cursor); + let mut conditions = vec!["space_id = ?".to_string()]; if repo.is_some() { conditions.push("author_did = ?".to_string()); @@ -499,19 +505,17 @@ pub async fn list_space_records( if collection.is_some() { conditions.push("collection = ?".to_string()); } - let (cursor_op, order) = if reverse { - ("indexed_at < ?", "DESC") - } else { - ("indexed_at > ?", "ASC") - }; - if cursor.is_some() { - conditions.push(cursor_op.to_string()); + let (cursor_cmp, order) = if reverse { ("<", "DESC") } else { (">", "ASC") }; + if decoded_cursor.is_some() { + conditions.push(format!( + "(indexed_at {cursor_cmp} ? OR (indexed_at = ? AND uri {cursor_cmp} ?))" + )); } let where_clause = conditions.join(" AND "); let raw = format!( - "SELECT uri, space_id, author_did, collection, rkey, record, cid, indexed_at FROM space_records WHERE {} ORDER BY indexed_at {} LIMIT ?", - where_clause, order + "SELECT uri, space_id, author_did, collection, rkey, record, cid, indexed_at FROM space_records WHERE {} ORDER BY indexed_at {}, uri {} LIMIT ?", + where_clause, order, order ); let sql = adapt_sql(&raw, backend); @@ -522,8 +526,8 @@ pub async fn list_space_records( if let Some(c) = collection { query = query.bind(c); } - if let Some(cur) = cursor { - query = query.bind(cur); + if let Some((ref ts, ref uri)) = decoded_cursor { + query = query.bind(ts.as_str()).bind(ts.as_str()).bind(uri.as_str()); } query = query.bind(limit); @@ -532,7 +536,18 @@ pub async fn list_space_records( .await .map_err(|e| AppError::Internal(format!("failed to list space records: {e}")))?; - rows.into_iter().map(parse_record_row).collect() + let records: Vec = rows + .into_iter() + .map(parse_record_row) + .collect::>()?; + + let next_cursor = if records.len() as i64 == limit { + records.last().map(|r| encode_cursor(&r.indexed_at, &r.uri)) + } else { + None + }; + + Ok((records, next_cursor)) } pub async fn insert_space_record( diff --git a/src/spaces/routes.rs b/src/spaces/routes.rs index 02b4dcf..7022c8f 100644 --- a/src/spaces/routes.rs +++ b/src/spaces/routes.rs @@ -498,7 +498,7 @@ async fn list_spaces( let did = query.did.unwrap_or_else(|| claims.did().to_string()); let limit = query.limit.unwrap_or(50).min(100); - let views = db::list_spaces_for_user( + let (views, cursor) = db::list_spaces_for_user( &state.db, state.db_backend, &did, @@ -507,12 +507,6 @@ async fn list_spaces( ) .await?; - let cursor = if views.len() as i64 == limit { - views.last().map(|v| v.created_at.clone()) - } else { - None - }; - let spaces_json: Vec = views .into_iter() .map(|v| { @@ -915,7 +909,7 @@ async fn list_records( let limit = query.limit.unwrap_or(50).min(100); let reverse = query.reverse.unwrap_or(false); - let records = db::list_space_records( + let (records, cursor) = db::list_space_records( &state.db, state.db_backend, &space.id, @@ -927,8 +921,6 @@ async fn list_records( ) .await?; - let cursor = records.last().map(|r| r.indexed_at.clone()); - let records_json: Vec = records .into_iter() .map(|r| { -- 2.51.2