feat(database): keyset (cursor) pagination helper

.changeset/keyset-pagination.md

@@ -0,0 +1,5 @@
+---
+"@maple-dev/effect-orm": minor
+---
+
+`/database` adds `paginate(query, { orderBy, after, limit }, params?)`: keyset (cursor) pagination over a built query. It adds a seek on the sort key, the `ORDER BY` and `LIMIT limit + 1`, and returns `{ rows, nextCursor }`, with no `OFFSET`. With one direction the seek is a row comparison, `(a, b) < (?, ?)`; with mixed directions it is the expanded `OR` form, bounded on the first key. The key is one or more selected, non-null aliases; end it with a unique column. The cursor is base64url text, encoded through the key columns' schemas. A cursor that does not decode, or that was written for another sort key, fails with the new `InvalidCursor` error. Postgres, ClickHouse, SQLite and D1. A nullable key is refused at run time too, since dialects sort NULLs differently.

docs/database.md

@@ -84,6 +84,7 @@ Calling `withdraw(1, 30)` outside `transfer` does not compile: `requireTransacti
 | `layerSqlClient(options)` | A `Database` layer over the `SqlClient` in context |
 | `Database` | The service |
 | `run(query, params?)` | Compile a query for the database's dialect, run it, decode its rows |
+| `paginate(query, options, params?)` | One page of a query, by keyset, with a cursor for the next page ([Pagination](#pagination)) |
 | `sql\`...\`` | A statement with every `${value}` bound; `sql.identifier(name)` for a table or column name |
 | `query(statement, schema?)` | Run a statement and return its rows, decoded when a schema is given |
 | `execute(statement)` | Run a statement and discard its rows |
@@ -155,6 +156,55 @@ Build the `Database` wherever you build the client. If each request has its own
 `Database` per request from it: nothing request-scoped is stored in the `Database`, and the
 open transaction lives in fiber context.
 
+### Pagination
+
+`paginate` reads a query one page at a time by keyset. It adds a seek on the sort key, the
+`ORDER BY` and `LIMIT limit + 1`, then returns `limit` rows and a cursor for the next page. The
+extra row only tells if there is a next page. There is no `OFFSET`: the database goes to the
+cursor through an index on the key, and does not read and drop all earlier rows. A row added
+before the cursor does not move later pages.
+
+```ts
+const page = yield* Db.paginate(
+	PG.from(Posts)
+		.select("id", "createdAt", "title")
+		.where(($) => [$.orgId.eq(PG.param.string("orgId"))]),
+	{ orderBy: [["createdAt", "desc"], ["id", "desc"]], after: request.cursor, limit: 50 },
+	{ orgId },
+)  // { rows: ReadonlyArray<{ id; createdAt; title }>, nextCursor: Option<string> }
+```
+
+| Option | Meaning |
+| --- | --- |
+| `orderBy` | One or more selected aliases, each `"asc"` or `"desc"`. A nullable alias is a type error |
+| `after` | The `nextCursor` of the previous page; leave it out for the first page |
+| `limit` | Rows per page, a positive integer. Anything else fails with `QueryBuilderError` |
+
+The key must be unique, so end it with a unique column such as the primary key. `paginate`
+cannot check this: rows that tie on the whole key can be skipped at a page boundary.
+
+With one direction the seek is a row comparison, `("createdAt", "id") < ($2, $3)`, which
+Postgres and SQLite (D1 too) serve from an index on `(createdAt, id)` and ClickHouse compares
+as a tuple. With mixed directions it is `a >= x AND (a > x OR (a = x AND b < y))`.
+
+A NULL key has no place in the seek, and dialects disagree on where NULLs sort: ascending,
+SQLite puts them first, Postgres and ClickHouse last. So a nullable key is a type error,
+and a defect at run time when the types were bypassed.
+
+The cursor is base64url text. It holds the sort key and the last row's key values, encoded
+through the key columns' schemas, so a `DateTime.Utc` key comes back as the same instant. It
+is not encrypted or signed: do not put a value in the key that the reader must not see. A
+cursor that does not decode, or that was written for another `orderBy`, fails with
+`InvalidCursor`. It usually comes from a request, so answer it as a bad request.
+
+The query keeps its own `where`, so the tenant filter and params still apply. `paginate`
+replaces the query's `orderBy` and `limit`. A query with `offset`, `groupBy`, `having` or
+`distinctOn`, or a key that is not a selected alias, is a defect.
+
+`paginate` uses the `Database` in context; it is not a method on `DatabaseApi`. Its types are
+`PaginateOptions`, `Page`, `KeysetOrder`, `KeysetColumn` (an alias that can be a key) and
+`KeysetDirection`.
+
 ## Transactions
 
 ```ts

docs/recipes.md

@@ -143,7 +143,8 @@ console.log(compiled.sql) // ... LIMIT 25 OFFSET 25
 Validate page sizes and indices as bounded non-negative integers before constructing a query.
 An offset is not a cursor: concurrent inserts can shift pages, and large offsets still require
 work. For a raw event feed, use a fixed time range and a cursor over a unique, stable sort key
-(such as timestamp plus event ID). The builder does not create or validate cursors for you.
+(such as timestamp plus event ID). `Db.paginate` does this for a query that is not grouped; see
+[Pagination](./database.md#pagination).
 
 ## Preserve large integer IDs
 

docs/reference.md

@@ -512,8 +512,8 @@ placeholders, double-quoted identifiers.
 | `ClickHouseStatementFromString`, `splitTerminalClauses`, `maskLiteralsAndComments`               | `/sql`            |
 | Schema tooling: `isSchemaObject`, `makeSnapshot`, `renderSchema`, `diffSchemas`, `diffPgSchemas`, `fromDrizzleSnapshot`, snapshot and entity schemas (`PgEnumEntity`, `PgUniqueEntity`, …), `Hint`; for SQLite, `sqliteEntitiesOf`, `renderSqliteSchema`, `renderSqliteCreateTable`, `renderSqliteCreateIndex`, `diffSqliteSchemas`, `fromDrizzleSqliteSnapshot`, `SqliteMigrationOp`, `renderSqliteOp`, `renderSqliteMigrationSql`, `labelOfSqlite`, `rebuildName`, the `SqliteSnapshot` schema and the `SqliteTableEntity`, `SqliteColumnEntity`, `SqliteIndexEntity`, `SqliteUniqueEntity` schemas | `/schema`         |
 | `run`, `status`, `verify`, `baseline`, `MigrationDriver` and the migrate errors                  | `/migrate`        |
-| `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `afterCommit`, `retryContention`, `Transaction` | `/database` |
-| `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` |
+| `Database`, `run`, `paginate`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `afterCommit`, `retryContention`, `Transaction` | `/database` |
+| `DatabaseError`, `InvalidCursor`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` |
 | `withQuerySummary`, `QuerySummary` — the `db.query.summary` of the statements an effect runs, for spans | `/database` |
 | `ClickhouseHttpClient`, `layer`, `make`, `ClickhouseHttpClientConfig` — a ClickHouse `SqlClient` over `fetch`, for Workers | `/clickhouse-http` |
 | `defineConfig`, `generate`, `check`, `pull`, `loadSchema`, `readMigrations`, `analyze`, `KitError` | `/kit`            |

src/ch/compile.ts

@@ -1289,15 +1289,15 @@ export function selectedAliasesOf(query: CHQuery<any, any, any>): ReadonlyArray<
 }
 
 /** Evaluate a query's SELECT callback without compiling it. */
-function selectExprsOf(query: CHQuery<any, any, any>): Record<string, unknown> | undefined {
+export function selectExprsOf(query: CHQuery<any, any, any>): Record<string, unknown> | undefined {
 	return selectRecordOf(query._state, makeAccessor(query._state))
 }
 
 /**
  * A query's SELECT list against accessor `$`: its callback's record, or for a
  * bare `select()` over a subquery or union, every alias that source selects.
  */
-function selectRecordOf(state: CHQueryState, $: any): Record<string, any> | undefined {
+export function selectRecordOf(state: CHQueryState, $: any): Record<string, any> | undefined {
 	if (state.selectAll) {
 		const source = state.fromUnion ? unionExprsOf(state.fromUnion._state.queries) : state.fromQuery ? selectExprsOf(state.fromQuery) : undefined
 		return source === undefined ? undefined : Object.fromEntries(Object.keys(source).map((alias) => [alias, $[alias]]))

src/ch/expr.ts

@@ -584,9 +584,18 @@ export function makeExpr<T>(
 			arith<number | Extract<T | R, null>>(fragment, "-", n, schema),
 		mod: <R extends number | null>(n: R | Expr<R>) => arith<Quotient<T, R>>(fragment, "%", n, schema),
 	}
+	operands.set(self, operand)
 	return self
 }
 
+// How each expression writes a plain value on the right of a comparison, for
+// SQL that compares outside the comparison methods: a row value `(a, b) > (x, y)`.
+const operands = new WeakMap<Expr<any, any>, (value: unknown) => SqlFragment>()
+
+/** A plain value as `expr`'s comparisons write it: through the column's codec when `expr` is a column. */
+export const operandOf = (expr: Expr<any, any>, value: unknown): SqlFragment =>
+	operands.get(expr)?.(value) ?? toFragment(value)
+
 /**
  * An expression with no declared result type — {@link untypedExpr}'s
  * counterpart for a caller assembling its own fragment.

src/database.ts

@@ -31,6 +31,7 @@ export {
 } from "./database/database"
 export {
 	DatabaseError,
+	InvalidCursor,
 	TenantScopeViolation,
 	TransactionClosed,
 	TransactionCommitFailed,
@@ -39,6 +40,14 @@ export {
 	TransactionUnsupported,
 	type TransactionError,
 } from "./database/errors"
+export {
+	paginate,
+	type KeysetColumn,
+	type KeysetDirection,
+	type KeysetOrder,
+	type Page,
+	type PaginateOptions,
+} from "./database/paginate"
 export { QuerySummary, withQuerySummary } from "./database/tracing"
 export { sql, type SqlIdentifier, type SqlRawText, type SqlTemplate } from "./database/sql"
 export type { DialectTransactions, IsolationLevel, TransactionSettings } from "./ch/dialect"

src/database/errors.ts

@@ -82,3 +82,13 @@ export type TransactionError =
 	| TransactionCommitFailed
 	| TransactionRollbackFailed
 	| TransactionClosed
+
+/**
+ * `paginate` was given a cursor it cannot read: not one it wrote, one written
+ * for another sort order, or one changed on the way. The cursor usually comes
+ * from a request, so this is an error to answer (a 400), not a defect.
+ */
+export class InvalidCursor extends Schema.TaggedError<InvalidCursor>()("@maple-dev/effect-orm/InvalidCursor", {
+	message: Schema.String,
+	cause: Schema.Defect(),
+}) {}

src/database/paginate.sqlite.test.ts

@@ -0,0 +1,82 @@
+// `paginate` on SQLite and a local D1: the row comparison and the mixed-direction
+// seek both walk every row once, timestamps (stored as text) round-trip through
+// the cursor, and a nullable key is refused even when the type check is bypassed.
+
+import * as D1Client from "@effect/sql-d1/D1Client"
+import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient"
+import { Cause, DateTime, Effect, Exit, Layer, Option } from "effect"
+import * as SqlClient from "effect/sql/SqlClient"
+import { Miniflare } from "miniflare"
+import { expect, it } from "vitest"
+import * as Db from "../database"
+import * as SQLite from "../sqlite"
+import type { Dialect } from "../ch/dialect"
+
+const Posts = SQLite.table(
+	"posts",
+	{ id: SQLite.integer, org: SQLite.text, at: SQLite.timestamp, note: SQLite.nullable(SQLite.text) },
+	{ tenantColumn: "org" },
+)
+const byOrg = SQLite.from(Posts)
+	.select("id", "at", "note")
+	.where(($) => [$.org.eq(SQLite.param.string("org"))])
+type Row = { readonly id: number; readonly at: DateTime.Utc; readonly note: string | null }
+
+const walk = (orderBy: Db.PaginateOptions<Row>["orderBy"]) =>
+	Effect.gen(function* () {
+		const pages: Array<ReadonlyArray<number>> = []
+		let after: string | undefined
+		for (;;) {
+			const page = yield* Db.paginate(byOrg, { orderBy, limit: 4, after }, { org: "a" })
+			pages.push(page.rows.map((row) => row.id))
+			if (Option.isNone(page.nextCursor)) return pages
+			after = page.nextCursor.value
+		}
+	})
+
+const pagesOn = (dialect: Dialect, client: Layer.Layer<SqlClient.SqlClient, unknown>) =>
+	Effect.runPromise(
+		Effect.gen(function* () {
+			const db = Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect, requireTenantScope: true })
+			yield* db.execute({ sql: "CREATE TABLE posts (id INTEGER PRIMARY KEY, org TEXT NOT NULL, at TEXT NOT NULL, note TEXT)" })
+			// Ten posts in org "a" whose timestamps tie in pairs, with NULL notes on
+			// the odd ids, and one post in org "b".
+			const rows = Array.from({ length: 10 }, (_, i) => ({
+				id: i + 1,
+				org: "a",
+				at: DateTime.makeUnsafe(Date.UTC(2026, 0, 1, 0, Math.ceil((i + 1) / 2))),
+				note: (i + 1) % 2 === 0 ? `n${i + 1}` : null,
+			}))
+			yield* db.run(SQLite.insertInto(Posts).values([...rows, { id: 11, org: "b", at: rows[0]!.at, note: null }]).crossTenant())
+			const run = <A, E>(effect: Effect.Effect<A, E, Db.Database>) => Effect.provideService(effect, Db.Database, db)
+			const desc = yield* run(walk([["at", "desc"], ["id", "desc"]]))
+			const mixed = yield* run(walk([["at", "asc"], ["id", "desc"]]))
+			const nullable = yield* run(Effect.exit(Db.paginate(byOrg, { orderBy: [["note", "asc"]] as any, limit: 4 }, { org: "a" })))
+			return { desc, mixed, nullable }
+		}).pipe(Effect.provide(client)),
+	)
+
+const check = ({ desc, mixed, nullable }: Awaited<ReturnType<typeof pagesOn>>) => {
+	expect(desc).toEqual([[10, 9, 8, 7], [6, 5, 4, 3], [2, 1]])
+	expect(mixed).toEqual([[2, 1, 4, 3], [6, 5, 8, 7], [10, 9]])
+	expect(Exit.isFailure(nullable) && Cause.hasDies(nullable.cause)).toBe(true)
+	expect(String(Exit.isFailure(nullable) ? Cause.squash(nullable.cause) : "")).toMatch(/can be NULL/)
+}
+
+it("walks keyset pages on SQLite", async () => {
+	check(await pagesOn(SQLite.sqliteDialect, SqliteClient.layer({ filename: ":memory:" })))
+})
+
+it("walks keyset pages on D1", async () => {
+	const mf = new Miniflare({
+		modules: true,
+		script: "export default { fetch() { return new Response('paginate test') } }",
+		compatibilityDate: "2026-07-01",
+		d1Databases: ["DB"],
+	})
+	try {
+		check(await pagesOn(SQLite.d1Dialect, D1Client.layer({ db: await mf.getD1Database("DB") })))
+	} finally {
+		await mf.dispose()
+	}
+}, 30_000)

src/database/paginate.test-d.ts

@@ -0,0 +1,35 @@
+// Type-level tests for `paginate`: the page's rows, and which keys it accepts.
+
+import { Effect } from "effect"
+import { expectTypeOf } from "expect-type"
+import * as PG from "../postgres"
+import * as Db from "../database"
+
+const Items = PG.table("items", {
+	columns: { id: PG.int4, org: PG.text, note: PG.nullable(PG.text) },
+	primaryKey: ["id"],
+})
+const items = PG.from(Items)
+	.select("id", "note")
+	.where(($) => [$.org.eq(PG.param.string("org"))])
+type Row = { readonly id: number; readonly note: string | null }
+
+// The page holds the query's rows and an optional next cursor.
+expectTypeOf(Db.paginate(items, { orderBy: [["id", "asc"]], limit: 10 }, { org: "a" })).toEqualTypeOf<
+	Effect.Effect<
+		Db.Page<Row>,
+		Db.InvalidCursor | Db.DatabaseError | PG.QueryBuilderError | PG.CompiledQueryDecodeError,
+		Db.Database
+	>
+>()
+
+// Only a selected, non-null alias can be a key.
+expectTypeOf<Db.KeysetColumn<Row>>().toEqualTypeOf<"id">()
+// @ts-expect-error -- a nullable column cannot order a keyset
+Db.paginate(items, { orderBy: [["note", "asc"]], limit: 10 }, { org: "a" })
+// @ts-expect-error -- "org" is not selected
+Db.paginate(items, { orderBy: [["org", "asc"]], limit: 10 }, { org: "a" })
+// @ts-expect-error -- the key needs at least one column
+Db.paginate(items, { orderBy: [], limit: 10 }, { org: "a" })
+// @ts-expect-error -- a missing param is a type error, as for `run`
+Db.paginate(items, { orderBy: [["id", "asc"]], limit: 10 })