Merge pull request #38 from pmorris-dev/pagination-phase1

feat: add keyset pagination to sqlx-sqlite-toolkit

Author: jjhafer

crates/sqlx-sqlite-toolkit/README.md

@@ -127,6 +127,61 @@ tx.commit().await?;
 // Or: tx.rollback().await?;
 ```
 
+### Pagination
+
+When working with large result sets, loading all rows at once can cause
+performance degradation and excessive memory usage. The toolkit provides
+built-in pagination via `fetch_page` to fetch data in fixed-size pages,
+keeping memory bounded and queries fast regardless of total row count.
+
+#### Why Keyset Pagination
+
+The toolkit uses keyset (cursor-based) pagination rather than traditional
+OFFSET-based pagination. With OFFSET, the database must scan and discard
+all skipped rows on every page request, making deeper pages progressively
+slower. Keyset pagination uses indexed column values from the last row of
+the current page to seek directly to the next page, keeping query time
+constant no matter how far you paginate.
+
+```rust
+use sqlx_sqlite_toolkit::pagination::KeysetColumn;
+
+let keyset = vec![
+   KeysetColumn::asc("category"),
+   KeysetColumn::desc("score"),
+   KeysetColumn::asc("id"),
+];
+
+// First page
+let page = db.fetch_page(
+   "SELECT id, title, category, score FROM posts".into(),
+   vec![],
+   keyset.clone(),
+   25,
+).await?;
+
+// Next page (forward) — pass the cursor from the previous page
+if let Some(cursor) = page.next_cursor {
+   let next = db.fetch_page(
+      "SELECT id, title, category, score FROM posts".into(),
+      vec![],
+      keyset.clone(),
+      25,
+   ).after(cursor.clone()).await?;
+
+   // Previous page (backward) — rows are returned in original sort order
+   let prev = db.fetch_page(
+      "SELECT id, title, category, score FROM posts".into(),
+      vec![],
+      keyset,
+      25,
+   ).before(cursor).await?;
+}
+```
+
+The base query must not contain `ORDER BY` or `LIMIT` clauses — the builder
+appends these automatically based on the keyset definition.
+
 ### Cross-Database Queries
 
 Attach other databases using the builder pattern:
@@ -183,6 +238,7 @@ cleanup_all_transactions(&interruptible, &regular).await;
 | `begin_interruptible_transaction()` | Begin interruptible transaction (builder) |
 | `fetch_all(query, values)` | Fetch all rows as JSON maps |
 | `fetch_one(query, values)` | Fetch single row or `None` |
+| `fetch_page(query, values, keyset, page_size)` | Keyset pagination (builder, supports `.after()`, `.before()`, `.attach()`) |
 | `acquire_writer()` | Acquire exclusive `WriterGuard` |
 | `run_migrations(migrator)` | Run pending migrations |
 | `close()` | Close connection |
@@ -214,6 +270,13 @@ All errors provide an `error_code()` method returning a machine-readable string:
 | `NO_ACTIVE_TRANSACTION` | Remove from empty state |
 | `INVALID_TRANSACTION_TOKEN` | Wrong transaction ID |
 | `IO_ERROR` | File system error |
+| `EMPTY_KEYSET_COLUMNS` | Keyset pagination requires at least one column |
+| `INVALID_PAGE_SIZE` | Page size must be greater than zero |
+| `CURSOR_LENGTH_MISMATCH` | Cursor value count does not match keyset column count |
+| `INVALID_PAGINATION_QUERY` | Base query contains top-level ORDER BY or LIMIT |
+| `CURSOR_COLUMN_NOT_FOUND` | Keyset column not found in query results |
+| `INVALID_COLUMN_NAME` | Keyset column name contains invalid characters |
+| `CONFLICTING_CURSORS` | Both `after` and `before` cursors provided |
 
 ## Development
 

crates/sqlx-sqlite-toolkit/src/builders.rs

@@ -9,6 +9,7 @@ use serde_json::Value as JsonValue;
 use sqlx_sqlite_conn_mgr::AttachedSpec;
 
 use crate::Error;
+use crate::pagination::{KeysetColumn, KeysetPage, build_paginated_query};
 use crate::wrapper::{DatabaseWrapper, WriteQueryResult, bind_value};
 
 /// Builder for SELECT queries returning multiple rows
@@ -153,6 +154,188 @@ impl IntoFuture for FetchOneBuilder {
    }
 }
 
+/// Internal cursor position for forward vs backward pagination.
+enum CursorPosition {
+   Forward(Vec<JsonValue>),
+   Backward(Vec<JsonValue>),
+}
+
+/// Builder for paginated SELECT queries using keyset (cursor-based) pagination
+pub struct FetchPageBuilder {
+   db: Arc<sqlx_sqlite_conn_mgr::SqliteDatabase>,
+   query: String,
+   values: Vec<JsonValue>,
+   keyset: Vec<KeysetColumn>,
+   page_size: usize,
+   cursor: Option<CursorPosition>,
+   attached: Vec<AttachedSpec>,
+}
+
+impl FetchPageBuilder {
+   pub(crate) fn new(
+      db: Arc<sqlx_sqlite_conn_mgr::SqliteDatabase>,
+      query: String,
+      values: Vec<JsonValue>,
+      keyset: Vec<KeysetColumn>,
+      page_size: usize,
+   ) -> Self {
+      Self {
+         db,
+         query,
+         values,
+         keyset,
+         page_size,
+         cursor: None,
+         attached: Vec::new(),
+      }
+   }
+
+   /// Set the cursor for fetching the next page (forward pagination).
+   ///
+   /// Pass the `next_cursor` from a previous `KeysetPage` to fetch the page
+   /// that follows it in the original sort order.
+   pub fn after(mut self, cursor: Vec<JsonValue>) -> Self {
+      self.cursor = Some(CursorPosition::Forward(cursor));
+      self
+   }
+
+   /// Set the cursor for fetching the previous page (backward pagination).
+   ///
+   /// Pass a cursor to fetch the page that precedes it in the original sort
+   /// order. Rows are returned in the original sort order (not reversed).
+   pub fn before(mut self, cursor: Vec<JsonValue>) -> Self {
+      self.cursor = Some(CursorPosition::Backward(cursor));
+      self
+   }
+
+   /// Attach additional databases for this query
+   pub fn attach(mut self, attached: Vec<AttachedSpec>) -> Self {
+      self.attached = attached;
+      self
+   }
+
+   /// Execute the paginated query and return a page of results
+   pub async fn execute(self) -> Result<KeysetPage, Error> {
+      // Validate inputs
+      if self.keyset.is_empty() {
+         return Err(Error::EmptyKeysetColumns);
+      }
+      if self.page_size == 0 {
+         return Err(Error::InvalidPageSize);
+      }
+
+      // Extract cursor values and direction
+      let (cursor_values, backward) = match self.cursor {
+         Some(CursorPosition::Forward(vals)) => (Some(vals), false),
+         Some(CursorPosition::Backward(vals)) => (Some(vals), true),
+         None => (None, false),
+      };
+
+      if let Some(ref vals) = cursor_values
+         && vals.len() != self.keyset.len()
+      {
+         return Err(Error::CursorLengthMismatch {
+            cursor_len: vals.len(),
+            keyset_len: self.keyset.len(),
+         });
+      }
+
+      // Build paginated SQL — pass the user's bind count so cursor
+      // placeholders are numbered $N+1, $N+2, … and never collide with
+      // the user's $1, $2, … (or positional ?) parameters.
+      let (sql, cursor_bind_values) = build_paginated_query(
+         &self.query,
+         &self.keyset,
+         cursor_values.as_deref(),
+         self.page_size,
+         backward,
+         self.values.len(),
+      )?;
+
+      // Combine user values + cursor bind values
+      let mut all_values = self.values;
+      all_values.extend(cursor_bind_values);
+
+      // Execute query
+      let rows = if self.attached.is_empty() {
+         let pool = self.db.read_pool()?;
+         let mut q = sqlx::query(&sql);
+         for value in all_values {
+            q = bind_value(q, value);
+         }
+         q.fetch_all(pool).await?
+      } else {
+         let mut conn =
+            sqlx_sqlite_conn_mgr::acquire_reader_with_attached(&self.db, self.attached).await?;
+
+         let mut q = sqlx::query(&sql);
+         for value in all_values {
+            q = bind_value(q, value);
+         }
+         let rows = sqlx::Executor::fetch_all(&mut *conn, q).await?;
+
+         // Explicit cleanup
+         conn.detach_all().await?;
+         rows
+      };
+
+      // Decode rows
+      let mut decoded = decode_rows(rows)?;
+
+      // Determine has_more by checking if we got more rows than page_size
+      let has_more = decoded.len() > self.page_size;
+      if has_more {
+         decoded.truncate(self.page_size);
+      }
+
+      // Reverse rows when paginating backward to restore original sort order
+      if backward {
+         decoded.reverse();
+      }
+
+      // Extract continuation cursor: first row if backward, last row if forward
+      let cursor_row = if backward {
+         decoded.first()
+      } else {
+         decoded.last()
+      };
+
+      let next_cursor = if has_more {
+         if let Some(row) = cursor_row {
+            let mut cursor_vals = Vec::with_capacity(self.keyset.len());
+            for col in &self.keyset {
+               let value = row
+                  .get(&col.name)
+                  .ok_or_else(|| Error::CursorColumnNotFound {
+                     column: col.name.clone(),
+                  })?;
+               cursor_vals.push(value.clone());
+            }
+            Some(cursor_vals)
+         } else {
+            None
+         }
+      } else {
+         None
+      };
+
+      Ok(KeysetPage {
+         rows: decoded,
+         next_cursor,
+         has_more,
+      })
+   }
+}
+
+impl IntoFuture for FetchPageBuilder {
+   type Output = Result<KeysetPage, Error>;
+   type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send>>;
+
+   fn into_future(self) -> Self::IntoFuture {
+      Box::pin(self.execute())
+   }
+}
+
 /// Builder for write queries (INSERT/UPDATE/DELETE)
 pub struct ExecuteBuilder {
    db: DatabaseWrapper,