use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use uuid::Uuid;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Cursor {
#[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime,
pub id: Uuid,
}
#[derive(Debug, thiserror::Error)]
pub enum CursorError {
#[error("cursor is not valid base64")]
Base64,
#[error("cursor payload is malformed")]
Payload,
}
impl Cursor {
pub fn encode(&self) -> String {
let json = serde_json::to_vec(self).expect("cursor is serializable");
URL_SAFE_NO_PAD.encode(json)
}
pub fn decode(token: &str) -> Result<Self, CursorError> {
let bytes = URL_SAFE_NO_PAD
.decode(token.as_bytes())
.map_err(|_| CursorError::Base64)?;
serde_json::from_slice(&bytes).map_err(|_| CursorError::Payload)
}
}
use crate::cursor::Cursor;
use serde::Serialize;
use sqlx::PgPool;
use time::OffsetDateTime;
use uuid::Uuid;
#[derive(Debug, Serialize, sqlx::FromRow)]
pub struct Event {
pub id: Uuid,
pub kind: String,
pub payload: serde_json::Value,
#[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime,
}
pub struct Page {
pub items: Vec<Event>,
pub next_cursor: Option<Cursor>,
}
pub async fn fetch_page(
pool: &PgPool,
after: Option<Cursor>,
limit: i64,
) -> Result<Page, sqlx::Error> {
let (bound_ts, bound_id) = match &after {
Some(c) => (c.created_at, c.id),
None => (OffsetDateTime::now_utc(), Uuid::nil()),
};
let mut rows = sqlx::query_as::<_, Event>(
"SELECT id, kind, payload, created_at
FROM events
WHERE ($1::bool OR (created_at, id) < ($2, $3))
ORDER BY created_at DESC, id DESC
LIMIT $4",
)
.bind(after.is_none())
.bind(bound_ts)
.bind(bound_id)
.bind(limit + 1)
.fetch_all(pool)
.await?;
let has_more = rows.len() as i64 > limit;
if has_more {
rows.truncate(limit as usize);
}
let next_cursor = if has_more {
rows.last().map(|e| Cursor { created_at: e.created_at, id: e.id })
} else {
None
};
Ok(Page { items: rows, next_cursor })
}
use crate::cursor::Cursor;
use crate::repository::{self, Event};
use async_stream::try_stream;
use axum::{
body::Body,
extract::{Query, State},
http::{header, StatusCode},
response::{IntoResponse, Response},
};
use serde::Deserialize;
use sqlx::PgPool;
#[derive(Deserialize)]
pub struct PageParams {
cursor: Option<String>,
#[serde(default = "default_limit")]
limit: i64,
}
fn default_limit() -> i64 {
50
}
pub async fn stream_events(
State(pool): State<PgPool>,
Query(params): Query<PageParams>,
) -> Result<Response, StatusCode> {
let after = match params.cursor.as_deref() {
Some(token) => Some(Cursor::decode(token).map_err(|_| StatusCode::BAD_REQUEST)?),
None => None,
};
let limit = params.limit.clamp(1, 500);
let page = repository::fetch_page(&pool, after, limit)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let next = page.next_cursor.map(|c| c.encode());
let items = page.items;
let body = Body::from_stream(try_stream! {
for event in items {
let line = serde_json::to_string(&event)
.map_err(std::io::Error::other)?;
yield format!("{line}\n");
}
let meta = serde_json::json!({ "next_cursor": next });
yield format!("{meta}\n");
});
Ok(([(header::CONTENT_TYPE, "application/x-ndjson")], body).into_response())
}
This snippet shows how a REST endpoint returns cursor-based (keyset) pagination instead of the more common LIMIT/OFFSET approach. Offset pagination degrades on large tables because the database must scan and discard every skipped row, and it also drifts when rows are inserted or deleted between requests. Keyset pagination avoids both problems by remembering the last row seen — the (created_at, id) pair — and asking for rows strictly after it, which lets Postgres seek directly into an index.
In cursor.rs, the Cursor type encodes that boundary. It carries a timestamp and an id, and is serialized into an opaque, URL-safe base64 token via encode so clients treat it as a blob rather than reverse-engineering the ordering columns. decode parses the token back, returning a CursorError on malformed input so the handler can respond with a clean 400 rather than panicking. Encoding the compound key is what makes the pagination stable: ties on created_at are broken deterministically by id.
In repository.rs, fetch_page runs the keyset query. The WHERE (created_at, id) < ($1, $2) row-value comparison expresses "everything ordered before this cursor" in a single index-friendly predicate, and it requests limit + 1 rows. That extra row is a cheap lookahead: if it comes back, there is another page, so next_cursor is derived from the last item the client will actually keep. The first request passes no cursor and starts from the newest rows.
In handler.rs, stream_events ties it together. Rather than buffering the whole page into a Vec and one large JSON body, it writes newline-delimited JSON (application/x-ndjson) through an async_stream so each record is flushed as it is serialized — useful when pages are large or the client wants to begin processing immediately. The final line is a metadata record carrying next_cursor, letting the caller pass it straight back on the next request. The Query extractor validates the incoming token up front, and an invalid cursor becomes a StatusCode::BAD_REQUEST. The main trade-off of keyset pagination is that it only supports sequential next/prev navigation, not jumping to an arbitrary page number, which is exactly the right constraint for streaming feeds and infinite scroll.
Share this code
Here's the card — post it anywhere.
