feat(database): batch lookups by key with findById and findAllBy

.changeset/batched-lookups.md

@@ -0,0 +1,5 @@
+---
+"@maple-dev/effect-orm": minor
+---
+
+`Db.findById(query, { key })` and `Db.findAllBy(query, { key })` in `/database` batch lookups by a key with Effect's `RequestResolver`: N concurrent calls send one statement, the query with `key IN (...)` added to its WHERE, and each row decodes through the query's row schema. A batch never mixes `Database`s, transactions or params, so a tenant filter in the query (a value or a `param.*`, given with `.with(params)`) keeps each statement to one tenant, and `requireTenantScope` checks it. On a dialect with a bound-parameter limit (D1: 100) a large batch is split into statements that each fit.

docs/database.md

@@ -93,6 +93,8 @@ Calling `withdraw(1, 30)` outside `transfer` does not compile: `requireTransacti
 | `retryContention(options?)` | Re-run on serialization failure or deadlock |
 | `Transaction` | The open transaction (`depth`, `isolationLevel`, `accessMode`) |
 | `isContention(error)` | Whether an error is a serialization failure or deadlock |
+| `findById(query, { key })` | A batched lookup of one row by a key: N concurrent calls send one statement ([Batching lookups](#batching-lookups)) |
+| `findAllBy(query, { key })` | A batched lookup of every row with a key, such as the children of a parent |
 
 `run`, `query`, `execute`, `transaction` and `retryContention` are also methods on
 `DatabaseApi`.
@@ -281,6 +283,82 @@ transaction it fails with `TransactionOptionsRejected`, because the database has
 aborted the whole transaction. `transaction(body, { retry: "contention" })` is shorthand for
 `retryContention(transaction(body))`, so each attempt starts with a fresh BEGIN.
 
+## Batching lookups
+
+Many fibers that each read one row by its id send one statement per fiber: the N+1 problem.
+`findById` and `findAllBy` collect the lookups that run at the same time into one batch, with
+Effect's `RequestResolver`, and send one statement for the batch: your query with
+`key IN (...)` added to its WHERE.
+
+```ts title="batched-lookups.ts"
+import { Effect } from "effect"
+import * as Db from "@maple-dev/effect-orm/database"
+import * as PG from "@maple-dev/effect-orm/postgres"
+
+const Users = PG.table("users", {
+	columns: { id: PG.int4, org_id: PG.text, name: PG.text },
+	primaryKey: ["id"],
+	tenantColumn: "org_id",
+})
+const Posts = PG.table("posts", {
+	columns: { id: PG.int4, org_id: PG.text, author_id: PG.int4, title: PG.text },
+	primaryKey: ["id"],
+	tenantColumn: "org_id",
+})
+
+// Built once, for example at module level. The tenant is a param.
+const userById = Db.findById(
+	PG.from(Users)
+		.select("id", "name")
+		.where(($) => [$.org_id.eq(PG.param.string("org"))]),
+	{ key: "id" },
+)
+const postsByAuthor = Db.findAllBy(
+	PG.from(Posts)
+		.select("id", "author_id", "title")
+		.where(($) => [$.org_id.eq(PG.param.string("org"))])
+		.orderBy(["id", "asc"]),
+	{ key: "author_id" },
+)
+
+// One SELECT for the users and one for their posts, whatever the number of ids.
+export const authors = (org: string, ids: ReadonlyArray<number>) =>
+	Effect.forEach(
+		ids,
+		(id) =>
+			Effect.all({ user: userById.with({ org })(id), posts: postsByAuthor.with({ org })(id) }, { concurrency: "unbounded" }),
+		{ concurrency: "unbounded" },
+	)
+```
+
+- **Rows** decode as `run` decodes them. The query must select `key` under its own name, so
+  each row can go back to the lookup that asked for it. That is a type error if it does not.
+- **Not found.** `findById` fails with `NoSuchElementError` for a key with no row; with more
+  than one row, the first one wins. `findAllBy` gives an empty array.
+- **Params.** Call the lookup directly when the query has no `param.*` markers. Otherwise
+  give them first with `.with(params)`, which checks them as `run` does.
+- **What shares a batch.** Lookups through the same `findById` that run at the same time, on
+  the same `Database`, in the same transaction (or both outside one), with equal params. A
+  lookup inside a transaction never runs on another connection, and one outside never runs
+  inside it. A key asked for twice is sent once.
+- **Tenant scope.** Put the tenant filter in the query, with a value or a param. Lookups with
+  different params never share a batch, so one statement never reads two tenants, and
+  [`requireTenantScope`](./tenant-scoping.md#enforcing-it) checks each batch statement as it
+  checks `run`.
+- **Bound-parameter limits.** On a dialect with a limit on the values one statement may
+  bind (D1: 100, Postgres: 65535), a large batch is split into statements that each fit,
+  with room for the query's own params. The statements run one after another on the
+  batch's connection. Each key is in exactly one of them, so its rows keep the query's
+  `ORDER BY`. ClickHouse inlines values and has no such limit.
+- **`limit` and `offset`** would apply to the whole batch, not to each key, so a query with
+  either fails with `QueryBuilderError` and sends nothing.
+
+The batch waits one `yieldNow`, the same as Effect's own resolvers, so the lookups must start
+together: through `Effect.forEach` or `Effect.all` with concurrency, or in fibers forked in the
+same step. Lookups in sequence send one statement each.
+
+_(Backed by `Request batching on PGlite`, in `src/database/resolver.test.ts`: one statement for N lookups, tenants and transactions kept apart; and `src/database/resolver.sqlite.test.ts`: a D1 batch over 100 bindings split into statements that fit.)_
+
 ## ClickHouse
 
 `clickhouseDialect` declares no transactions, so `transaction` fails with

docs/reference.md

@@ -514,6 +514,7 @@ placeholders, double-quoted identifiers.
 | `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` |
+| `findById`, `findAllBy`, `FindById`, `FindAllBy`, `LookupOptions`, `LookupError` — batched lookups by a key, one `IN` statement per batch | `/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/database.ts

@@ -40,5 +40,13 @@ export {
 	type TransactionError,
 } from "./database/errors"
 export { QuerySummary, withQuerySummary } from "./database/tracing"
+export {
+	findAllBy,
+	findById,
+	type FindAllBy,
+	type FindById,
+	type LookupError,
+	type LookupOptions,
+} from "./database/resolver"
 export { sql, type SqlIdentifier, type SqlRawText, type SqlTemplate } from "./database/sql"
 export type { DialectTransactions, IsolationLevel, TransactionSettings } from "./ch/dialect"

src/database/database.ts

@@ -170,7 +170,8 @@ interface OpenState {
 	open: boolean
 	readonly client: SqlClient.SqlClient
 }
-class TransactionOpen extends Context.Service<TransactionOpen, OpenState>()("@maple-dev/effect-orm/TransactionOpen") {}
+/** @internal — also read by `resolver.ts` to keep batches inside one transaction. */
+export class TransactionOpen extends Context.Service<TransactionOpen, OpenState>()("@maple-dev/effect-orm/TransactionOpen") {}
 
 // A defect from the body, carried through `withTransaction` as a failure so the
 // transaction still rolls back and any defect that comes out of it is known to

src/database/resolver.sqlite.test.ts

@@ -0,0 +1,67 @@
+// Batched lookups on SQLite and D1. D1 binds at most 100 values per statement,
+// so a large batch goes in chunks that each fit, with room for the query's own
+// params; SQLite's limit is far above any batch here, so it sends one.
+
+import * as D1Client from "@effect/sql-d1/D1Client"
+import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient"
+import { Effect, Layer } 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 Users = SQLite.table("users", { id: SQLite.integer, org: SQLite.text, name: SQLite.text }, { tenantColumn: "org" })
+const ids = Array.from({ length: 250 }, (_, i) => i + 1)
+
+const lookups = (dialect: Dialect, client: Layer.Layer<SqlClient.SqlClient, unknown>) =>
+	Effect.runPromise(
+		Effect.gen(function* () {
+			const statements: Array<string> = []
+			const db = Db.fromSqlClient(yield* SqlClient.SqlClient, {
+				dialect,
+				requireTenantScope: true,
+				observe: (statement) => Effect.sync(() => void statements.push(statement.sql)),
+			})
+			yield* db.execute({ sql: "CREATE TABLE users (id INTEGER PRIMARY KEY, org TEXT NOT NULL, name TEXT NOT NULL)" })
+			for (const id of ids) yield* db.execute({ sql: "INSERT INTO users VALUES (?, ?, ?)", parameters: [id, id % 2 === 0 ? "a" : "b", `u${id}`] })
+			statements.length = 0
+			const userById = Db.findById(
+				SQLite.from(Users)
+					.select("id", "name")
+					.where(($) => [$.org.eq(SQLite.param.string("org"))]),
+				{ key: "id" },
+			)
+			const evens = ids.filter((id) => id % 2 === 0)
+			const found = yield* Effect.forEach(evens, (id) => userById.with({ org: "a" })(id), { concurrency: "unbounded" }).pipe(
+				Effect.provideService(Db.Database, db),
+			)
+			return { found, statements }
+		}).pipe(Effect.provide(client)),
+	)
+
+it("splits a D1 batch over the 100-binding limit into statements that each fit", async () => {
+	const mf = new Miniflare({
+		modules: true,
+		script: "export default { fetch() { return new Response('resolver test') } }",
+		compatibilityDate: "2026-07-01",
+		d1Databases: ["DB"],
+	})
+	try {
+		const binding = await mf.getD1Database("DB")
+		const { found, statements } = await lookups(SQLite.d1Dialect, D1Client.layer({ db: binding }))
+		expect(found).toEqual(ids.filter((id) => id % 2 === 0).map((id) => ({ id, name: `u${id}` })))
+		// 125 keys plus the tenant param: 99 keys per statement.
+		expect(statements).toHaveLength(2)
+		for (const sql of statements) expect(sql.split("?").length - 1).toBeLessThanOrEqual(100)
+	} finally {
+		await mf.dispose()
+	}
+}, 30_000)
+
+it("sends one statement for the same batch on SQLite", async () => {
+	const { found, statements } = await lookups(SQLite.sqliteDialect, SqliteClient.layer({ filename: ":memory:" }))
+	expect(found).toHaveLength(125)
+	expect(statements).toHaveLength(1)
+})

src/database/resolver.test-d.ts

@@ -0,0 +1,39 @@
+// Type-level tests: what `findById` and `findAllBy` take and give.
+
+import type { Cause, Effect } from "effect"
+import { expectTypeOf } from "expect-type"
+import * as CH from "../ch/index"
+import * as PG from "../postgres"
+import * as Db from "../database"
+
+const Users = PG.table("users", { columns: { id: PG.int4, org: PG.text, name: PG.text }, primaryKey: ["id"] })
+type Row = { readonly id: number; readonly name: string }
+
+const userById = Db.findById(CH.from(Users).select("id", "name"), { key: "id" })
+expectTypeOf(userById(1)).toEqualTypeOf<Effect.Effect<Row, Cause.NoSuchElementError | Db.LookupError, Db.Database>>()
+// @ts-expect-error -- the key is the column's type
+userById("1")
+
+const usersByName = Db.findAllBy(CH.from(Users).select("id", "name"), { key: "name" })
+expectTypeOf(usersByName("Ann")).toEqualTypeOf<Effect.Effect<ReadonlyArray<Row>, Db.LookupError, Db.Database>>()
+
+// The key must be a selected column of the FROM table.
+// @ts-expect-error -- `org` is not selected
+Db.findById(CH.from(Users).select("id", "name"), { key: "org" })
+// @ts-expect-error -- not a column
+Db.findById(CH.from(Users).select("id", "name"), { key: "nope" })
+
+// A query with params takes them through `with`, checked as `run` checks them.
+const inOrg = Db.findById(
+	CH.from(Users)
+		.select("id", "name")
+		.where(($) => [$.org.eq(CH.param.string("org"))]),
+	{ key: "id" },
+)
+expectTypeOf(inOrg.with({ org: "a" })(1)).toEqualTypeOf<
+	Effect.Effect<Row, Cause.NoSuchElementError | Db.LookupError, Db.Database>
+>()
+// @ts-expect-error -- the params are required
+inOrg(1)
+// @ts-expect-error -- a param of the wrong type
+inOrg.with({ org: 1 })

src/database/resolver.test.ts

@@ -0,0 +1,152 @@
+// Request batching: `findById` and `findAllBy` turn N concurrent lookups into
+// one `WHERE key IN (...)` statement. Counted on PGlite through `observe`.
+
+import { PgliteClient } from "@effect/sql-pglite"
+import { assert, expect, layer } from "@effect/vitest"
+import { Cause, Effect, Exit, Layer, Result } from "effect"
+import * as SqlClient from "effect/sql/SqlClient"
+import * as CH from "../ch/index"
+import * as PG from "../postgres"
+import { postgresDialect } from "../pg/dialect"
+import * as Db from "../database"
+
+const statements: Array<string> = []
+
+const DatabaseLive = Layer.effect(
+	Db.Database,
+	Effect.gen(function* () {
+		const sql = yield* SqlClient.SqlClient
+		return Db.fromSqlClient(sql, {
+			dialect: postgresDialect,
+			observe: (statement) => Effect.sync(() => void statements.push(statement.sql)),
+		})
+	}),
+).pipe(Layer.provideMerge(PgliteClient.layer({ postgresqlconf: "timezone = 'UTC'" })))
+
+const Users = PG.table("users", {
+	columns: { id: PG.int4, org: PG.text, name: PG.text },
+	primaryKey: ["id"],
+	tenantColumn: "org",
+})
+
+const Live = Layer.effectDiscard(
+	Effect.gen(function* () {
+		yield* Db.execute(Db.sql`CREATE TABLE users (id int4 PRIMARY KEY, org text NOT NULL, name text NOT NULL)`)
+		yield* Db.execute(Db.sql`INSERT INTO users VALUES (1, 'a', 'Ann'), (2, 'a', 'Bob'), (3, 'b', 'Cy')`)
+	}),
+).pipe(Layer.provideMerge(DatabaseLive))
+
+/** Run `effect` and return the statements it sent. */
+const counted = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
+	Effect.gen(function* () {
+		statements.length = 0
+		const value = yield* effect
+		return { value, sent: [...statements] }
+	})
+
+layer(Live, { excludeTestServices: true })("Request batching on PGlite", (it) => {
+	it.effect("findById sends one statement for N concurrent lookups and decodes each row", () =>
+		Effect.gen(function* () {
+			const userById = Db.findById(CH.from(Users).select("id", "name").crossTenant(), { key: "id" })
+			const { value, sent } = yield* counted(Effect.forEach([1, 2, 3], userById, { concurrency: "unbounded" }))
+			expect(value).toEqual([
+				{ id: 1, name: "Ann" },
+				{ id: 2, name: "Bob" },
+				{ id: 3, name: "Cy" },
+			])
+			assert.strictEqual(sent.length, 1)
+			expect(sent[0]).toContain(" IN (")
+		}),
+	)
+
+	it.effect("findById fails a missing key with NoSuchElementError and sends a repeated key once", () =>
+		Effect.gen(function* () {
+			const userById = Db.findById(CH.from(Users).select("id", "name").crossTenant(), { key: "id" })
+			const { value, sent } = yield* counted(
+				Effect.forEach([1, 1, 9], (id) => Effect.result(userById(id)), { concurrency: "unbounded" }),
+			)
+			assert(Result.isSuccess(value[0]) && Result.isSuccess(value[1]))
+			expect(value[1].success).toEqual({ id: 1, name: "Ann" })
+			assert(Result.isFailure(value[2]))
+			expect(value[2].failure).toBeInstanceOf(Cause.NoSuchElementError)
+			assert.strictEqual(sent.length, 1)
+			expect(sent[0]).toMatch(/IN \(\$1, \$2\)/)
+		}),
+	)
+
+	it.effect("one batch never mixes tenants: each params value gets its own single-tenant statement", () =>
+		Effect.gen(function* () {
+			const strict = Db.fromSqlClient(yield* SqlClient.SqlClient, {
+				dialect: postgresDialect,
+				requireTenantScope: true,
+				observe: (statement) => Effect.sync(() => void statements.push(statement.sql)),
+			})
+			const inOrg = CH.from(Users)
+				.select("id", "name")
+				.where(($) => [$.org.eq(CH.param.string("org"))])
+			const userById = Db.findById(inOrg, { key: "id" })
+			const lookups = [
+				userById.with({ org: "a" })(1),
+				userById.with({ org: "b" })(3),
+				userById.with({ org: "a" })(2),
+				// Another tenant's row is not found, even with its id.
+				Effect.flip(userById.with({ org: "b" })(1)),
+			] as const
+			const { value, sent } = yield* counted(
+				Effect.all(lookups, { concurrency: "unbounded" }).pipe(Effect.provideService(Db.Database, strict)),
+			)
+			expect(value.slice(0, 3)).toEqual([
+				{ id: 1, name: "Ann" },
+				{ id: 3, name: "Cy" },
+				{ id: 2, name: "Bob" },
+			])
+			expect(value[3]).toBeInstanceOf(Cause.NoSuchElementError)
+			assert.strictEqual(sent.length, 2)
+
+			// Without a tenant filter, requireTenantScope refuses the batch statement.
+			const unscoped = Db.findById(CH.from(Users).select("id", "name"), { key: "id" })
+			const exit = yield* Effect.exit(unscoped(1).pipe(Effect.provideService(Db.Database, strict)))
+			assert(Exit.isFailure(exit))
+			expect(Cause.squash(exit.cause)).toBeInstanceOf(Db.TenantScopeViolation)
+		}),
+	)
+
+	it.effect("a lookup inside a transaction never shares a batch with one outside it", () =>
+		Effect.gen(function* () {
+			const userById = Db.findById(CH.from(Users).select("id", "name").crossTenant(), { key: "id" })
+			const { value, sent } = yield* counted(
+				Effect.all([Db.transaction(userById(1)), userById(2)], { concurrency: "unbounded" }),
+			)
+			expect(value).toEqual([
+				{ id: 1, name: "Ann" },
+				{ id: 2, name: "Bob" },
+			])
+			assert.strictEqual(sent.length, 2)
+		}),
+	)
+
+	it.effect("findAllBy groups every row under its key, and a key with no rows gets an empty array", () =>
+		Effect.gen(function* () {
+			const usersInOrg = Db.findAllBy(CH.from(Users).select("id", "org").orderBy(["id", "asc"]).crossTenant(), { key: "org" })
+			const { value, sent } = yield* counted(Effect.forEach(["a", "b", "z"], usersInOrg, { concurrency: "unbounded" }))
+			expect(value).toEqual([
+				[
+					{ id: 1, org: "a" },
+					{ id: 2, org: "a" },
+				],
+				[{ id: 3, org: "b" }],
+				[],
+			])
+			assert.strictEqual(sent.length, 1)
+		}),
+	)
+
+	it.effect("refuses a query with limit or offset, which would apply to the whole batch", () =>
+		Effect.gen(function* () {
+			const firstUser = Db.findById(CH.from(Users).select("id", "name").limit(1).crossTenant(), { key: "id" })
+			const { value, sent } = yield* counted(Effect.flip(firstUser(1)))
+			expect(value).toBeInstanceOf(CH.QueryBuilderError)
+			assert.strictEqual(sent.length, 0)
+		}),
+	)
+})