diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c7bc0414..d14c31d67 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,37 @@ Semver applies from 1.0.0. A breaking change to a documented API needs a major ### Added +- **`findById` batches itself — one microtask of point lookups is one `where "id" in (…)`.** A page that resolves an author per row sent one `select … where "id" = $1` per row. Inside a request, `postgresRepo()` now collects the lookups issued in the same microtask and sends one statement for all of them: + + ```ts + // One statement, not one per post. findById's signature and its meaning are unchanged. + const authors = await Promise.all(posts.map((post) => users.findById(post.authorId, { orgId }))); + ``` + + No `dataloader()`, no `batch()`, nothing to opt into: the capability lives inside the method that already exists, which is the only place it can reach code already written. The batch is keyed by context identity — a `WeakMap`, so it dies with the request, the shape `@ultimat3/query`'s request memo has one tier up — and by a scope key covering every input to the statement except the id, so another tenant, another soft-delete visibility, another projection, another entity or another client is a different statement and never joins one. What goes out is the statement each single lookup would have been served by, `in` instead of `=`: the tenant predicate and the `deleted_at is null` clause are inside it, so an id whose row is missing, soft-deleted or another tenant's still reads as `null`. Past 500 ids it becomes several whole statements rather than one Postgres refuses for its bind count, and it declines outright — sending exactly what it always sent — with no request in scope (a job, a script), on a composite primary key, or on a scope it cannot compare. The window is one microtask and closes before the statement is sent, so a sequential `for … of` loop shares nothing through this path — what batches one is the sibling preload below. Proved against a real Postgres in `pg-driver.live.test.ts`: five lookups, one statement, the same five answers. +- **A `for … of` loop over a page is two statements — the sibling-aware preload.** A microtask window cannot see a sequential loop: its `await` closes the window before the next lookup exists. `findMany` now leaves its page's foreign key **values** behind for the request, so the first `findById` for any one of them resolves that key for **every** row of the page in one statement and the rest of the loop is served from memory. On by default; `postgresDriver({ jitPreload: false })` is the one switch that turns it off, where the driver is already constructed — not an `app.config.ts` key, because nothing reads config at the seam that builds a repository and a switch the framework cannot read is a switch that does nothing: + + ```ts + const page = await posts.findMany({ orgId }); + for (const post of page.rows) { + // Two statements for the whole loop: the page, then one `select … where "id" in (…)` over + // every author on it. Every lookup after the first is memory. + const author = await users.findById(post.authorId, { orgId }); + } + ``` + + Nothing new to write — the relation is the `references()` already declared, and `findById` keeps its signature, so the fix reaches loops that are already written. The scope guard is a security boundary, not a tuning knob: a preloaded row is served only to a lookup with the same scope key the batch above uses (tenant predicate, soft-delete visibility, projection, entity), the same client, and no write to that entity since — anything else reads the statement it always read. The preload statement is that same statement widened to the page's ids, so its tenant predicate and `deleted_at is null` clause are in the SQL Postgres runs; a page's ids can therefore never resolve rows outside the reader's own scope. The index holds values and not rows, keyed by context identity, so it pins nothing and dies with the request; a page of a hundred rows by one author is still one bind; a nullable key that resolved to nothing is not a row to go looking for; past 500 ids it becomes whole statements; and with no request in scope (a job, a script) a page leaves nothing behind at all. +- **`preload(name)` — the join a `for … of` loop would have earned, without writing the loop.** The two paths above catch a lookup pattern already written; `preload()` is the same join, named on the chain instead of triggered by a loop: + + ```ts + export const db = database({ orgs, posts, members }); + + // Two statements: the page, then one `select … where "id" in (…)` over its authors. + const page = await db.posts.where({ orgId }).preload('author').page(); + page.rows[0].author; // the member row, or null — always present + ``` + + One vocabulary, `preload('')` — no `include`, `join` or `with`, and nothing new to declare: the name is the `references()` already written, resolved by `relationNamed()` when `preload()` is called, so an unknown one is `X_PRELOAD_UNKNOWN_RELATION` on the chain and not a page later. A `belongsTo` attaches the row or `null`, a `hasMany` an array — always present, so "no author" and "nobody preloaded the author" never read the same. Several `preload()` calls resolve concurrently, never one after the other, and naming one relation twice is one statement; keys are chunked at `MAX_IDS_PER_STATEMENT` (500) exactly as a coalesced point read already is, so past 500 keys a relation costs several statements rather than one Postgres refuses for its bind count, and a relation with more rows than one page holds costs another keyset page rather than a silent truncation. The page's own tenant predicate carries onto the related read only when the other entity's tenant column has the same name — a value that scopes one entity is a guess on another, and a guess here is a cross-tenant read — so a differently-named tenant column carries nothing, and carrying nothing is not silent: the related read builds its own plan and `assertScoped` refuses it there as `X_TENANCY_UNSCOPED`. The related read is `findMany`, so `deleted_at is null` applies exactly as it does anywhere else. Attached to a copy of each row — `{ ...row }`, because the in-memory driver hands back the row it stores — after `select()`, which is widened internally with the relation's local key so a projection can drop neither the key a preload reads nor the relation it attaches; `plan().select` reports the widened list. `page()`, `all()` and `one()` preload; `count()` and `plan()` do not, since a count reads no rows. A table reaches only the entities its own `database()` call named, through the driver that call was given, so a preload against memory means what a preload against Postgres means — a relation whose other end is outside the set is `X_INVARIANT_VIOLATED` naming the `database({ … })` call that fixes it, and `tableFor(entity, repo)` built by hand reaches no other table. `preload('author')` returns `ReadBuilder` — `unknown`, not a generated type, because the name is a string resolved at runtime and the row on the other side is parsed by its own entity. - **`relationMap()` — the foreign keys an entity already declared, readable at query time.** `ColumnMeta.references` was resolved in exactly one place, `describe.ts`, to spell a DDL constraint; at query time nothing could answer "what is a post's author". `@ultimat3/entity` now derives a named map from the same thunks — `belongsTo` from an entity's own foreign keys, `hasMany` from the inbound ones: ```ts diff --git a/docs/architecture/06-data-layer.md b/docs/architecture/06-data-layer.md index 9fd89cec9..60f8bab0c 100644 --- a/docs/architecture/06-data-layer.md +++ b/docs/architecture/06-data-layer.md @@ -49,8 +49,9 @@ Generated per entity. The **only** place SQL lives. | Method | Notes | |---|---| -| `findById(id)` | tenant filter applied automatically; returns `null`, never throws on miss | -| `findMany(args)` | cursor-paginated only (below) | +| `findById(id)` | tenant filter applied automatically; returns `null`, never throws on miss. Inside a request, the calls issued in one microtask are **one** `where "id" in (…)` ([`coalesce.ts`](../../packages/entity/src/coalesce.ts)) — same scope, same soft-delete filter, one round trip. An id that is a foreign key on a page this request read resolves that key for the whole page instead ([`jit-preload.ts`](../../packages/entity/src/jit-preload.ts)), which is what batches a sequential `for … of` loop | +| `findMany(args)` | cursor-paginated only (below); leaves its page's foreign key values behind for the preload above | +| `preload(name)` | not on `Repo` — a `ReadBuilder` chain method, `db..preload('')`, resolved by `page()`/`all()`/`one()`. One more `where in (…)` per named relation, over the page `findMany` already read ([`preload.ts`](../../packages/entity/src/preload.ts)); several relations resolve concurrently, and naming one twice is one statement | | `insert(values)` / `update(id, patch)` | invariants run before the statement | | `delete(id)` | soft-deletes when the entity declares `deletedAt`, hard-deletes when it does not | | `deleteWhere(filter)` | delete by equality filter; resolves with the **number of rows removed** | @@ -59,6 +60,16 @@ Generated per entity. The **only** place SQL lives. | `Transactor.run(fn)` | joins the ambient transaction if one exists, opens one otherwise | | custom | added in the feature's `repo.ts`, returning schema-parsed rows | +One family, three members: + +| Path | Trigger | Cost | +|---|---|---| +| Microtask coalescing ([`coalesce.ts`](../../packages/entity/src/coalesce.ts)) | several `findById` calls land in one microtask | one `where id in (…)` — nothing to opt into | +| Sibling JIT preload ([`jit-preload.ts`](../../packages/entity/src/jit-preload.ts)) | the first `findById` whose id is a foreign key on a page this request read | one `where in (…)` for the whole page; every later lookup in the loop is memory — nothing to opt into, and `postgresDriver({ jitPreload: false })` is the one switch that turns it off | +| Eager preload ([`preload.ts`](../../packages/entity/src/preload.ts)) | `.preload('')` named on the chain | one `where in (…)` per relation, resolved before the caller sees a row | + +Reach for `preload()` when the relation is part of what the page *is* — a list rendered with its authors, rows handed to something that will not call back into the repo, or a read a reviewer should see stated rather than inferred from a loop. The other two ask for nothing: a same-microtask fan-out and a sequential `for … of` loop already get them for free. All three read their keys through one file ([`batch-read.ts`](../../packages/entity/src/batch-read.ts)) — one spelling of a key, one 500-id bind cap — so a bound and a key's identity cannot disagree between them. What each may widen is its own: the two implicit paths share a scope key and refuse to widen across a tenant, a soft-delete visibility, a projection or an entity, because they answer a statement the caller already sent; `preload()` sends its own, so it carries the chain's tenant predicate onto it and refuses when it cannot. + The two filtered writes are not conveniences. `delete(id)` and `update(id, patch)` both need a single-column primary key — `singleKeyOf` throws `X_INVARIANT_VIOLATED` on a composite one — so on a join table (`likes`, `blocks`, `participants`) they are the only write paths there are. Without them a composite-key row is create-only: `likes` could be liked and never unliked, and `participants.lastReadAt` could never be marked read. Four properties make them safe to be the only filtered writes: | Property | Mechanism | diff --git a/packages/entity/CLAUDE.md b/packages/entity/CLAUDE.md index d43ee1c5e..6cc89fa93 100644 --- a/packages/entity/CLAUDE.md +++ b/packages/entity/CLAUDE.md @@ -28,6 +28,75 @@ Columns + invariants; the row type is derived from the columns. Tier 2. that is how a `unique()` column shipped a migration failing on `42P07` and money's currency shipped as `char(1)`. A new operator, column kind or write path is not done until it round-trips there. +- **A point lookup batches itself, and the batch is never wider than the statement it replaces.** + `findById` called several times in one microtask of one request is one `select … where "id" in + (…)` — `coalesce.ts`, keyed by ctx identity (a `WeakMap`, so the batch dies with the request, the + shape `@ultimat3/query`'s request memo has one tier up) and by a scope key covering **every** + input to the statement except the id. Two tenants, two soft-delete visibilities, two projections, + two entities or two clients therefore never share one: a coalesced statement has to be one each + of the singles would have been served by, or a caller is answered with rows their own statement + could never have returned. It declines rather than guesses — no request in scope, a composite + key, a predicate value it cannot render — and declining is just the statement `findById` always + sent, which is why `findById` keeps its signature and there is no `batch()` to opt into. The + window closes before the statement goes out, so a lookup arriving mid-flight opens the next batch + instead of joining ids already on the wire, and past `MAX_IDS_PER_STATEMENT` a batch becomes + several whole statements rather than one Postgres refuses for its bind count. A sequential + `for … of` loop shares no microtask — its `await` ends the window — which is what the sibling + preload below is for. +- **A page batches the loop it causes, and a preloaded row is only ever served to the statement + that read it.** `findMany` leaves its page's foreign key *values* behind (`jit-preload.ts`, + `tagSiblings`), so the first `findById` for any one of them resolves that key for every row of + the page in one `in` statement and the rest of a `for … of` loop is memory. Five rules, none + optional. **The scope guard is a security boundary**: a preloaded row is served only under the + *same* `scopeKey` the coalescer uses — same tenant predicate, same soft-delete visibility, same + projection, same entity — and the preload statement is that scope widened to the page's ids, so + a page read under one tenant can never resolve another tenant's rows, whichever tenant asks. + **Same client, or nothing**: a bucket filled through the ambient pool is not read through a + pinned one, which is also what stops a row read inside a transaction being served after it — + `db()` hands back a different client once the transaction is over, rolled back or not. + **A write drops it**: `postgresRepo`'s `writing()` is the one place every write goes out, and it + calls `forgetPreloaded(entity.$name)` *before* the statement, so a row a request changed is + re-read and never served from a page read before it. **Values, not rows**: the index is keyed by + id and holds ids, so a page early in a long request pins its keys and not its rows, and it dies + with the request like every other per-ctx store here. **Declining is the old behaviour**: no + request in scope, an id no page indexed, a key that resolved to nothing — the caller reads the + statement it always read. `MAX_IDS_PER_STATEMENT` bounds the preload exactly as it bounds a + batch. What both share — the scope key, `keyOf`, the one `in` statement — lives in + `batch-read.ts` so the two can never disagree about when a shared statement is legal. + **One switch, where the driver is built**: `postgresDriver({ jitPreload: false })` / + `postgresRepo(entity, { jitPreload: false })` turns the tagging off. Never an `app.config.ts` + key — nothing reads config at the seam that builds a repository, so a `database.jitPreload` + field would be a switch the framework cannot read, which is a switch that does nothing. +- **`preload(name)` shares `batch-read.ts` with the coalescer and the JIT preload above, but + keeps no request-scoped cache of its own.** `keyOf`, `MAX_IDS_PER_STATEMENT` and + `statementChunks` come from the same file, so a bind-count bound and a key's identity can + never disagree across the three — but `preload()` reads its scope straight off the chain's + own `where` and issues its statement every call; nothing here declines to an old statement + the way the coalescer or the JIT preload can, because there is no old statement to decline + to — a chain that calls `preload('author')` always gets the extra statement. **Tenancy is + carried, never inferred, and that is a security boundary, not a convenience**: + `tenantScope()` carries the page's own tenant predicate onto the related read only when + **both** entities are scoped by a column of that same name — a value that scopes one entity + is a guess on another, and serving a guessed scope is a cross-tenant read. Both ends are + checked, never the target's alone: a source scoped by `workspaceId` may still carry an + ordinary `orgId` predicate of its own, and matching on the target's column name would lift + that filter into the target's tenant scope and attach rows from a tenant nobody proved this + reader owns. A differently-named column carries nothing, on purpose, so the related read + builds an unscoped plan of its own and `assertScoped` refuses it as `X_TENANCY_UNSCOPED` + rather than let it pass. **Reach is the same `database()` set the two bullets above already answer + to**: `RelatedTables` is the resolver `database()` hands every table it builds, an + entity-name → `{ entity, repo }` map closed over the same call, so `preload('author')` + resolves `author` only when that call named the entity the relation points at — outside it + is `X_INVARIANT_VIOLATED`, never a reach around the handle. `tableFor(entity, repo)` built + by hand takes no `related` resolver, so the identical call fails the identical way with + `related` itself `undefined`. **A projection cannot drop what a preload needs**: + `select()` widens its own field list with each preloaded relation's local key, so + `plan().select` — the projection that actually runs — always carries it, though the row + type the caller sees still names only what they picked. **Attachment copies, never + mutates**: a preloaded relation is written onto `{ ...row }`, because the in-memory driver + hands back the row it stores and attaching directly would leak the relation into the table + itself. **Preloading terminals only**: `page()`, `all()` and `one()` resolve every named + relation; `count()` and `plan()` do not, since neither reads a row to attach one to. - **Cursor pagination only.** OFFSET is wrong under concurrent writes: an insert before the offset shifts every later page, so a client silently skips and repeats rows. No `offset` on `FindManyArgs` or the builder; the primary key is always the last sort key, so the order is @@ -155,6 +224,10 @@ Columns + invariants; the row type is derived from the columns. Tier 2. | `repo.ts` / `tenancy.ts` | `Repo` + `memoryDriver`'s repo, tx rollback; `QueryPlan` + `assertScoped()` | | `plan.ts` / `cursor.ts` | the plan both drivers execute; the one keyset cursor codec | | `pg-driver.ts` | `postgresDriver()`, `postgresRepo()`, `postgresTransactor()` | +| `coalesce.ts` | one microtask of `findById` calls → one `where id in (…)`, per request | +| `batch-read.ts` | what a shared point read is made of — the scope key, `keyOf`, the one `in` statement | +| `jit-preload.ts` | a page's foreign key values → one `in` statement for the whole `for … of` loop | +| `preload.ts` | the relation `preload()` names → one related-rows statement → attached to the page | | `pg-sql.ts` / `pg-row.ts` | plan → parameterised SQL; physical row ⇄ entity row (money is two columns) | | `registry.ts` | duplicate detection, `describeEntities()` for the manifest, `references()` per entry | | `relations.ts` | `relationMap()`/`relationsFor()`/`relationNamed()` — the FKs as a named `belongsTo`/`hasMany` map | diff --git a/packages/entity/README.md b/packages/entity/README.md index ad858d884..394891bdc 100644 --- a/packages/entity/README.md +++ b/packages/entity/README.md @@ -182,8 +182,31 @@ named subset — a `belongsTo` to an entity outside it is still reported, a `has sides. An entity holding its own keys reads `entity.$references()`: same closure, so a foreign key is read once and not twice. -Nothing consumes the map yet — a preload is the next slice. It is exported because the derivation -is a fact about the schema, not an implementation detail of whoever traverses it first. +`preload()`, below, is what consumes it — exported because the derivation is a fact about the +schema, not an implementation detail of whoever traverses it first. + +## Preloading a relation + +`As of 2026-08`. A single `findById` batches itself, and a `for … of` loop over a page batches the +loop it causes — reach for `preload()` to carry the relation from the start, without waiting on +either pattern to trigger it: + +```ts +export const db = database({ orgs, posts, members }); + +// Two statements: the page, then one `select … where "id" in (…)` over its authors. +const page = await db.posts.where({ orgId }).preload('author').page(); +page.rows[0].author; // the member row, or null — always present +``` + +| | | +|---|---| +| Vocabulary | one method, `preload('')` — no `include`, `join` or `with` | +| Unknown name | `X_PRELOAD_UNKNOWN_RELATION` at `preload()` itself, not a page later | +| Shape | `belongsTo` attaches the row or `null`; `hasMany` an array — always present | +| Statements | one extra per relation, resolved concurrently; naming one twice is one statement | +| Tenancy | carried onto the related read only when the other entity's tenant column shares the name; otherwise `X_TENANCY_UNSCOPED` refuses the related read rather than guess | +| Terminals | `page()`, `all()`, `one()` preload; `count()` and `plan()` don't — neither reads a row | ## Two drivers, one meaning @@ -208,6 +231,54 @@ that enqueued it. Every value is bound to `$n` and every identifier is resolved through the entity, so a column name can only be one the entity declared and a row value can never become SQL. +## Point lookups batch themselves + +`As of 2026-08`: + +```ts +const repo = postgresRepo(users); +// One statement, not one per post: the lookups issued in this microtask are one `in`. +const authors = await Promise.all(posts.map((post) => repo.findById(post.authorId, { orgId }))); +``` + +`findById` keeps its signature and gets faster. There is no `dataloader()`, no `batch()` and +nothing to opt into: inside a request the point lookups issued in one microtask become one +`select … where "id" in ($1, $2, …)` carrying the scope each of them carried, and outside one — a +job, a script — it is the single statement it always was. + +| | | +|---|---| +| Window | one microtask, closed before the statement is sent | +| Lifetime | the request; the batch is keyed by context identity and dies with it | +| Never shared with | another tenant, another soft-delete visibility, another projection, another entity, another client | +| Wider than 500 ids | several whole statements, never one Postgres refuses | +| An id with no row | `null`, exactly as the single statement answered | + +## A page batches the loop it causes + +`As of 2026-08`. A sequential loop shares no microtask — its `await` ends the window before the +next lookup exists. So a page leaves its foreign key values behind, and the first lookup for any +one of them resolves that key for every row of the page: + +```ts +const page = await postgresRepo(posts).findMany({ orgId }); +for (const post of page.rows) { + // Two statements for the whole loop: the page, then one `in` over every author on it. + const author = await postgresRepo(users).findById(post.authorId, { orgId }); +} +``` + +Nothing new to write: the relation is the `references()` the column already declares. + +| | | +|---|---| +| Served to | a lookup with the same scope key, the same client, and no write to that entity since — the preload statement *is* the statement it was widened from | +| Scope | the tenant predicate and `deleted_at is null` are in the preload statement, so a page's ids can never resolve rows outside the reader's own scope | +| A write | drops what was preloaded for that entity, before the statement goes out, so a changed row is re-read and never served from before it | +| Held | the ids, never the rows; keyed by context identity, so it dies with the request | +| Declines to the old statement | no request in scope, an id no page indexed, a key that resolved to nothing | +| Switched off | `postgresDriver({ jitPreload: false })`, where the driver is constructed — the one switch. Not an `app.config.ts` key: nothing reads config at the seam that builds a repository | + ## Tenancy is a guard `tenant: 'orgId'` on the entity names the column outright. Omit it and it is inferred — a @@ -227,7 +298,7 @@ the invariants, which makes a seed a test of the schema as well. ## Errors `X_ENTITY_DUPLICATE` · `X_INVARIANT_VIOLATED` · `X_TENANCY_UNSCOPED` · `X_DB_DRIFT` · -`X_NOT_FOUND` · `X_WRITE_UNFILTERED` · `X_PATCH_EMPTY` +`X_NOT_FOUND` · `X_WRITE_UNFILTERED` · `X_PATCH_EMPTY` · `X_PRELOAD_UNKNOWN_RELATION` ## Boundaries diff --git a/packages/entity/src/batch-read.ts b/packages/entity/src/batch-read.ts new file mode 100644 index 000000000..227db2e18 --- /dev/null +++ b/packages/entity/src/batch-read.ts @@ -0,0 +1,134 @@ +// Single responsibility: what a batched point read is made of — the scope two lookups must share +// before one statement is allowed to answer both, and the one `select … where in (…)` that +// answers them. The microtask coalescer and the sibling-aware preload both read ids through here, +// so the two can never disagree about when a shared statement is legal. + +import type { DbClient } from '@ultimat3/db'; +import type { EntityCore } from './entity'; +import { decodeRow, type PhysicalRow } from './pg-row'; +import { type ReadShape, selectStatement } from './pg-sql'; +import type { QueryPlan } from './tenancy'; + +/** + * Postgres binds at most 65535 parameters, so a batch wider than this becomes several statements + * rather than one the driver refuses. Splitting can only cost round trips; not splitting would + * fail reads that succeeded one at a time, and a batcher that breaks a working program is worse + * than no batcher. + */ +export const MAX_IDS_PER_STATEMENT = 500; + +/** The primary key, in the three spellings a batch needs: the plan's, the table's, and the type. */ +export interface KeyColumn { + readonly property: string; + readonly column: string; + readonly kind: string; +} + +/** + * One row, or the failure decoding it. A column the table no longer matches is that row's + * problem: a caller whose own row decoded must still be handed it, exactly as it would have been + * by the statement it did not share. + */ +export type Answer = { readonly row: unknown } | { readonly error: unknown }; + +/** + * The string both sides of a batch are filed under — the requested id, and the key of a row that + * came back. Postgres compares a `uuid` as a value and prints it lower-cased, so an id handed in + * upper case matches the row *there* and would miss it here, which would make an answer depend on + * whether some other lookup shared the microtask. Every other kind compares by its bytes, and + * lower-casing a text key would merge two rows Postgres keeps apart. + */ +export const keyOf = (kind: string, value: unknown): string => + kind === 'uuid' ? String(value).toLowerCase() : String(value); + +/** A scope value has to be comparable as a string, or two scopes cannot be told apart. */ +const scopeValue = (value: unknown): string | undefined => { + if (value === null || value === undefined) return 'null'; + if (value instanceof Date) return `date:${value.getTime()}`; + const kind = typeof value; + return kind === 'string' || kind === 'number' || kind === 'bigint' || kind === 'boolean' + ? `${kind}:${String(value)}` + : undefined; +}; + +/** + * Everything about the read except the id. Two lookups may share one statement only when this + * matches: another tenant's predicate, a different projection or a different soft-delete + * visibility is a different query, and merging them would answer a caller with rows the statement + * they asked for could never have returned. + * + * `undefined` when a predicate value cannot be rendered — an object scope is one this cannot prove + * two lookups share, and a batch is only ever an optimisation, so it declines instead of guessing. + */ +export const scopeKey = ( + entity: EntityCore, + scoped: QueryPlan, + shape: ReadShape, +): string | undefined => { + const predicates: string[][] = []; + for (const predicate of scoped.where) { + const value = scopeValue(predicate.value); + if (value === undefined) return undefined; + predicates.push([predicate.column, predicate.op, value]); + } + // JSON rather than a joined string: a value carrying the separator cannot forge a boundary. + return JSON.stringify([ + entity.$name, + entity.$table, + shape.includeDeleted, + scoped.select ?? null, + predicates, + ]); +}; + +/** + * One read, minus the ids. This is the whole of what a shared statement is allowed to vary in — + * a batch that widens anything else here is a batch answering a caller with rows their own + * statement could never have returned. + */ +export interface PointRead { + readonly entity: EntityCore; + /** A pinned client and the ambient pool are two places to read from, never one batch. */ + readonly client: DbClient; + /** The plan with the id predicate removed — the scope every id in the batch shares. */ + readonly scoped: QueryPlan; + readonly shape: ReadShape; + readonly key: KeyColumn; +} + +/** The one statement, filed by key — the same builder, the same scope, `in` instead of `=`. */ +export const readByIds = async ( + read: PointRead, + ids: readonly unknown[], +): Promise> => { + const { entity, scoped, key } = read; + const found = await read.client.query( + selectStatement( + entity, + { ...scoped, where: [{ column: key.property, op: 'in', value: ids }, ...scoped.where] }, + read.shape, + ids.length, + ), + ); + const answers = new Map(); + for (const physical of found) { + // Filed under the value the statement matched on, which is readable whether or not the rest + // of the row decodes — so a drifted column fails one caller instead of everyone in the batch. + const filedAt = keyOf(key.kind, physical[key.column]); + try { + answers.set(filedAt, { row: decodeRow(entity, physical) }); + } catch (error) { + answers.set(filedAt, { error }); + } + } + return answers; +}; + +/** One statement's worth at a time — a batch too wide for the bind count is several, never one. */ +export const statementChunks = (values: readonly T[]): readonly (readonly T[])[] => { + const chunks: T[][] = []; + for (let from = 0; from < values.length; from += MAX_IDS_PER_STATEMENT) { + chunks.push(values.slice(from, from + MAX_IDS_PER_STATEMENT)); + } + return chunks; +}; diff --git a/packages/entity/src/coalesce.test.ts b/packages/entity/src/coalesce.test.ts new file mode 100644 index 000000000..f3fabdd09 --- /dev/null +++ b/packages/entity/src/coalesce.test.ts @@ -0,0 +1,280 @@ +// What one microtask of point lookups buys a request: one statement, an answer per id, and the +// scope that statement carried. The scope guard is the claim — a coalesced row is served only to a +// lookup the statement WAS, so a batch can never answer with rows its caller's own could not. + +import { afterAll, beforeEach, describe, expect, test } from 'bun:test'; +import { createContext, runWithContext } from '@ultimat3/core'; +import { + createRecordingClient, + type DbClient, + type RecordingClient, + setDbClient, +} from '@ultimat3/db'; +import { MAX_IDS_PER_STATEMENT } from './batch-read'; +import { money, text, timestamp, uuid } from './columns'; +import { entity } from './entity'; +import { postgresRepo } from './pg-driver'; +import { clearRegistry } from './registry'; + +const invoices = entity('coalesce_test_invoices', { + columns: { + id: uuid().primaryKey(), + orgId: uuid().tenant(), + reference: text({ max: 40 }), + total: money(), + deletedAt: timestamp().nullable(), + }, +}); + +const notes = entity('coalesce_test_notes', { + columns: { id: uuid().primaryKey(), body: text() }, +}); + +const likes = entity('coalesce_test_likes', { + columns: { postId: uuid().primaryKey(), userId: uuid().primaryKey() }, +}); + +const idAt = (index: number): string => + `00000000-0000-7000-8000-${String(index).padStart(12, '0')}`; + +const ORG = idAt(1); +const OTHER_ORG = idAt(2); + +/** What Bun.SQL hands back: snake_case names, int8 as a string. */ +const physical = (id: string, over: Record = {}): Record => ({ + id, + org_id: ORG, + reference: `INV-${id.slice(-2)}`, + total_minor: '100', + total_currency: 'EUR', + deleted_at: null, + ...over, +}); + +let client: RecordingClient; + +beforeEach(() => { + client = createRecordingClient(); + setDbClient(client); +}); + +afterAll(() => { + setDbClient(undefined); + clearRegistry(); +}); + +const repo = () => postgresRepo(invoices); +const inRequest = (work: () => Promise): Promise => runWithContext(createContext(), work); + +describe('findById coalescing', () => { + test('point lookups issued in one microtask become one statement', async () => { + // Reversed on purpose: rows come back in whatever order Postgres returns them, so a caller + // served by position instead of by id would get another caller's row. + client.on('select', { rows: [physical(idAt(11)), physical(idAt(10))] }); + const rows = await inRequest(() => + Promise.all([ + repo().findById(idAt(10), { orgId: ORG }), + repo().findById(idAt(11), { orgId: ORG }), + repo().findById(idAt(12), { orgId: ORG }), + ]), + ); + + expect(client.statements).toHaveLength(1); + expect(client.texts[0]).toContain('"id" in ($1, $2, $3)'); + expect(client.statements[0]?.values).toEqual([idAt(10), idAt(11), idAt(12), ORG, 3]); + expect(rows[0]?.id).toBe(idAt(10)); + expect(rows[1]?.id).toBe(idAt(11)); + // An id the statement did not answer for is `findById`'s null, not a rejection. + expect(rows[2]).toBeNull(); + }); + + test('the coalesced statement carries the scope the single one carried', async () => { + await inRequest(() => + Promise.all([ + repo().findById(idAt(10), { orgId: ORG }), + repo().findById(idAt(11), { orgId: ORG }), + ]), + ); + expect(client.texts[0]).toContain('"org_id" = $3'); + expect(client.texts[0]).toContain('"deleted_at" is null'); + expect(client.texts[0]).toContain('order by "id" asc'); + }); + + test('a row is decoded by the columns that declared it', async () => { + client.on('select', { rows: [physical(idAt(10))] }); + const [row] = await inRequest(() => + Promise.all([ + repo().findById(idAt(10), { orgId: ORG }), + repo().findById(idAt(11), { orgId: ORG }), + ]), + ); + expect(row?.total).toEqual({ minor: 100n, currency: 'EUR' }); + expect(row?.deletedAt).toBeNull(); + }); + + test('the same id twice is one bind and one row', async () => { + client.on('select', { rows: [physical(idAt(10))] }); + const [first, second] = await inRequest(() => + Promise.all([ + repo().findById(idAt(10), { orgId: ORG }), + repo().findById(idAt(10), { orgId: ORG }), + ]), + ); + expect(client.statements).toHaveLength(1); + expect(client.statements[0]?.values).toEqual([idAt(10), ORG, 1]); + // The in-memory driver hands every caller the same row object; so does this one. + expect(first).toBe(second); + }); + + test('a uuid asked for in upper case is the row Postgres would have matched', async () => { + const mixed = '0000000a-000b-7000-8000-00000000000c'; + client.on('select', { rows: [physical(mixed)] }); + const [upper, lower] = await inRequest(() => + Promise.all([ + repo().findById(mixed.toUpperCase(), { orgId: ORG }), + repo().findById(mixed, { orgId: ORG }), + ]), + ); + // One bind: Postgres compares a uuid as a value and would have matched either spelling, so + // the two spellings must not become two lookups — or the answer would depend on the batch. + expect(client.statements[0]?.values).toEqual([mixed.toUpperCase(), ORG, 1]); + expect(upper?.id).toBe(mixed); + expect(lower).toBe(upper); + }); + + test('a row that no longer decodes fails its own caller, not the batch', async () => { + // `reference` is not-null, so a null from the database means the table no longer matches the + // entity. That is this row's problem: the caller of the other id still gets their row. + client.on('select', { + rows: [physical(idAt(10), { reference: null }), physical(idAt(11))], + }); + await inRequest(async () => { + // Both issued before either is asserted on: an assertion between them would be an await, + // and the batch would be two batches. + const broken = repo().findById(idAt(10), { orgId: ORG }); + const fine = repo().findById(idAt(11), { orgId: ORG }); + await Promise.all([ + expect(broken).rejects.toBeUltimateError('X_INVARIANT_VIOLATED'), + expect(fine).resolves.toHaveProperty('reference', 'INV-11'), + ]); + }); + expect(client.statements).toHaveLength(1); + }); + + test('a batch wider than the cap becomes whole statements, never one Postgres refuses', async () => { + const ids = Array.from({ length: MAX_IDS_PER_STATEMENT + 1 }, (_, index) => idAt(100 + index)); + const rows = await inRequest(() => + Promise.all(ids.map((id) => repo().findById(id, { orgId: ORG }))), + ); + expect(client.statements).toHaveLength(2); + // ids + the org predicate + the limit. + expect(client.statements[0]?.values).toHaveLength(MAX_IDS_PER_STATEMENT + 2); + expect(client.statements[1]?.values).toHaveLength(3); + expect(rows).toHaveLength(MAX_IDS_PER_STATEMENT + 1); + expect(rows.every((row) => row === null)).toBe(true); + }); + + test('a failed statement fails every caller in it', async () => { + let sent = 0; + const fail = (): Promise => { + sent += 1; + return Promise.reject(new RangeError('boom')); + }; + const broken: DbClient = { query: fail, one: fail, execute: fail }; + const pinned = postgresRepo(invoices, { client: broken }); + await inRequest(async () => { + const first = pinned.findById(idAt(10), { orgId: ORG }); + const second = pinned.findById(idAt(11), { orgId: ORG }); + // Held before either is asserted on: the batch rejects every caller in one tick, and an + // assertion is an await, so the second rejection would arrive with nobody holding it. + const held = Promise.allSettled([first, second]); + await expect(first).rejects.toThrow('boom'); + await expect(second).rejects.toThrow('boom'); + await held; + }); + expect(sent).toBe(1); + }); +}); + +describe('what never shares a statement', () => { + test('with no request in scope every lookup is the statement it always was', async () => { + await Promise.all([ + repo().findById(idAt(10), { orgId: ORG }), + repo().findById(idAt(11), { orgId: ORG }), + repo().findById(idAt(12), { orgId: ORG }), + ]); + expect(client.statements).toHaveLength(3); + expect(client.texts.every((statement) => statement.includes('"id" = $1'))).toBe(true); + }); + + test('two tenants never coalesce into one statement', async () => { + await inRequest(() => + Promise.all([ + repo().findById(idAt(10), { orgId: ORG }), + repo().findById(idAt(11), { orgId: OTHER_ORG }), + ]), + ); + expect(client.statements).toHaveLength(2); + expect(client.statements[0]?.values).toEqual([idAt(10), ORG, 1]); + expect(client.statements[1]?.values).toEqual([idAt(11), OTHER_ORG, 1]); + }); + + test('a lookup that reveals soft-deleted rows never joins one that hides them', async () => { + // `RepoOptions` is what `findById` takes, and `shapeOf` reads `includeDeleted` off it — the + // live suite passes it, so the two visibilities have to be two statements. + const revealed = { orgId: ORG, includeDeleted: true }; + await inRequest(() => + Promise.all([repo().findById(idAt(10), { orgId: ORG }), repo().findById(idAt(11), revealed)]), + ); + expect(client.statements).toHaveLength(2); + expect(client.texts.filter((statement) => statement.includes('"deleted_at" is null'))).toEqual([ + client.texts[0] ?? '', + ]); + }); + + test('two entities are two statements', async () => { + await inRequest(() => + Promise.all([ + repo().findById(idAt(10), { orgId: ORG }), + postgresRepo(notes).findById(idAt(11)), + ]), + ); + expect(client.statements).toHaveLength(2); + expect(client.texts[0]).toContain('from "coalesce_test_invoices"'); + expect(client.texts[1]).toContain('from "coalesce_test_notes"'); + }); + + test('a batch belongs to one request, so two requests are two statements', async () => { + await Promise.all([ + inRequest(() => repo().findById(idAt(10), { orgId: ORG })), + inRequest(() => repo().findById(idAt(11), { orgId: ORG })), + ]); + expect(client.statements).toHaveLength(2); + }); + + test('the window is one microtask: a lookup after an await opens the next statement', async () => { + await inRequest(async () => { + await repo().findById(idAt(10), { orgId: ORG }); + await repo().findById(idAt(11), { orgId: ORG }); + }); + expect(client.statements).toHaveLength(2); + }); +}); + +describe('the guards a point lookup already had', () => { + test('a tenant-scoped lookup without an org predicate never reaches a statement', async () => { + await inRequest(async () => { + await expect(repo().findById(idAt(10))).rejects.toBeUltimateError('X_TENANCY_UNSCOPED'); + }); + expect(client.statements).toHaveLength(0); + }); + + test('a composite primary key is still refused, batching or not', async () => { + await inRequest(async () => { + await expect(postgresRepo(likes).findById(idAt(10))).rejects.toBeUltimateError( + 'X_INVARIANT_VIOLATED', + ); + }); + expect(client.statements).toHaveLength(0); + }); +}); diff --git a/packages/entity/src/coalesce.ts b/packages/entity/src/coalesce.ts new file mode 100644 index 000000000..6bd252678 --- /dev/null +++ b/packages/entity/src/coalesce.ts @@ -0,0 +1,175 @@ +// Single responsibility: collapse the point lookups one request issues in the same microtask into +// a single `where id in (…)`, and hand a lookup a page already answered straight to the preload. +// A list that resolves an author per row is the N+1 this removes, and it removes it without adding +// a second way to read — `findById` keeps its signature and its meaning, and pays for one round +// trip instead of one per row. + +import { type Ctx, tryUseContext } from '@ultimat3/core'; +import type { DbClient } from '@ultimat3/db'; +import { + type Answer, + type KeyColumn, + keyOf, + type PointRead, + readByIds, + scopeKey, + statementChunks, +} from './batch-read'; +import type { EntityCore } from './entity'; +import { preloadedFindById } from './jit-preload'; +import { physicalName } from './pg-row'; +import type { ReadShape } from './pg-sql'; +import type { QueryPlan } from './tenancy'; + +/** One caller's lookup: the row it asked for, and the two ends of the promise it is holding. */ +interface Pending { + readonly id: unknown; + /** What the answer will be filed under — `keyOf(id)`, not the id itself. */ + readonly key: string; + readonly row: Promise; + readonly settle: (row: unknown) => void; + readonly fail: (error: unknown) => void; +} + +/** One statement in the making: the ids collected so far, and the read that will send them. */ +interface Batch { + /** Two lookups share a statement only when they read from the same place. */ + readonly client: DbClient; + /** Keyed by `String(id)`, so the same id asked for twice is one bind and one row. */ + readonly pending: Map; + readonly load: (ids: readonly unknown[]) => Promise>; +} + +/** + * Per request, keyed by ctx identity, so a batch dies with the request that opened it — the shape + * `@ultimat3/query`'s request memo has one tier up. `entity` cannot import that one (tier 2 to + * tier 3 is upward), so it owns this one. + */ +const requests = new WeakMap>(); + +const batchesFor = (ctx: Ctx): Map => { + const key: object = ctx; + const existing = requests.get(key); + if (existing !== undefined) return existing; + const created = new Map(); + requests.set(key, created); + return created; +}; + +const pendingFor = (id: unknown, key: string): Pending => { + // The executor runs synchronously, so both are assigned before this returns. TypeScript cannot + // see through the callback, which is all the definite assignments claim. + let settle!: (row: unknown) => void; + let fail!: (error: unknown) => void; + const row = new Promise((resolve, reject) => { + settle = resolve; + fail = reject; + }); + return { id, key, row, settle, fail }; +}; + +const openBatch = ( + batches: Map, + key: string, + client: DbClient, + load: Batch['load'], +): Batch => { + const batch: Batch = { client, pending: new Map(), load }; + batches.set(key, batch); + // The window is one microtask: every lookup issued before the current synchronous run ends + // shares this statement. It closes here, before the statement is sent, so a lookup arriving + // mid-flight opens the next batch instead of joining ids already on the wire. + queueMicrotask(() => { + if (batches.get(key) === batch) batches.delete(key); + void flush(batch); + }); + return batch; +}; + +const flush = async (batch: Batch): Promise => { + const waiting = [...batch.pending.values()]; + batch.pending.clear(); + // One statement at a time: a batch wide enough to split must not take the pool with it. + for (const chunk of statementChunks(waiting)) { + try { + const answers = await batch.load(chunk.map((entry) => entry.id)); + for (const entry of chunk) { + const answer = answers.get(entry.key); + // An id the statement did not answer for is a row that is not there — `findById`'s `null`, + // never a rejection, and never another caller's row. + if (answer === undefined) entry.settle(null); + else if ('error' in answer) entry.fail(answer.error); + else entry.settle(answer.row); + } + } catch (error) { + // The statement failed, so everyone in it gets the failure the single statement would have + // handed them. Every one of these promises was returned to a caller, so none goes unhandled. + for (const entry of chunk) entry.fail(error); + } + } +}; + +/** + * The shared read this lookup joins, or `undefined` when there is none to join: no request in + * scope, a composite key, a scope this cannot compare, or a client the open batch does not read + * from. Declining is always correct — the caller sends the one statement it always sent. + * + * Two shapes, one seam. A page already read answers first (one statement for a whole sequential + * loop), and what no page can answer joins the microtask batch. + */ +export const coalesceFindById = ( + entity: EntityCore, + client: DbClient, + plan: QueryPlan, + shape: ReadShape, + id: unknown, +): Promise | undefined => { + const ctx = tryUseContext(); + const [keyColumn] = entity.$primaryKey; + const declared = keyColumn === undefined ? undefined : entity.$columns[keyColumn]; + if ( + ctx === undefined || + keyColumn === undefined || + declared === undefined || + entity.$primaryKey.length !== 1 + ) { + return undefined; + } + // A seek positions a page, never a point lookup. If one ever reaches here the statement is not + // the one this batches. + if (shape.seek !== undefined) return undefined; + const at = plan.where.findIndex( + (predicate) => + predicate.column === keyColumn && predicate.op === 'eq' && predicate.value === id, + ); + if (at === -1) return undefined; + const scoped: QueryPlan = { ...plan, where: plan.where.filter((_, index) => index !== at) }; + const key = scopeKey(entity, scoped, shape); + if (key === undefined) return undefined; + + const keyColumnRef: KeyColumn = { + property: keyColumn, + column: physicalName(entity, keyColumn), + kind: declared.$meta.kind, + }; + const read: PointRead = { entity, client, scoped, shape, key: keyColumnRef }; + // A page whose foreign keys this id is one of answers for every row of that page at once, which + // is the only thing that batches a `for … of` loop: its `await` already ended the microtask. + const preloaded = preloadedFindById(ctx, read, key, id); + if (preloaded !== undefined) return preloaded; + + const batches = batchesFor(ctx); + const open = batches.get(key); + // A pinned client and the ambient pool are two places to read from, and a batch is one + // statement: a lookup that does not share the open batch's client sends its own. + if (open !== undefined && open.client !== client) return undefined; + const batch = open ?? openBatch(batches, key, client, (ids) => readByIds(read, ids)); + + const filedAt = keyOf(keyColumnRef.kind, id); + const already = batch.pending.get(filedAt); + const pending = already ?? pendingFor(id, filedAt); + if (already === undefined) batch.pending.set(filedAt, pending); + // One batch is one entity — the key fixed that before the batch existed — so this re-attaches + // the row type the store erased rather than asserting anything new about it. + return pending.row as Promise; +}; diff --git a/packages/entity/src/database.ts b/packages/entity/src/database.ts index d9eea348b..8ab725f05 100644 --- a/packages/entity/src/database.ts +++ b/packages/entity/src/database.ts @@ -2,6 +2,7 @@ // writes a repository class per entity, and nobody can reach a table that is not in the set. import type { EntityCore } from './entity'; +import type { RelatedTables } from './preload'; import type { Table } from './query'; import { tableFor } from './query'; import type { Repo } from './repo'; @@ -53,9 +54,18 @@ export const database = ( options: DatabaseOptions = {}, ): Database => { const driver = options.driver ?? defaultDriver(); + // Keyed by entity name, which is what a relation names — the object key is the caller's spelling + // of it. This handle is the whole of what a preload can reach: a table reads the entities its own + // `database()` call named, through the driver that call was given, so a preload against memory + // means what a preload against Postgres means. + const declared = new Map(Object.values(entities).map((entity) => [entity.$name, entity])); + const related: RelatedTables = (entityName) => { + const entity = declared.get(entityName); + return entity === undefined ? undefined : { entity, repo: driver.repo(entity) }; + }; const tables: Record = {}; for (const [key, entity] of Object.entries(entities)) { - tables[key] = tableFor(entity, driver.repo(entity)); + tables[key] = tableFor(entity, driver.repo(entity), related); } // Built key by key from `entities`, so each table is the one `Database` names. return tables as Database; diff --git a/packages/entity/src/index.ts b/packages/entity/src/index.ts index 86defa65b..d246144c9 100644 --- a/packages/entity/src/index.ts +++ b/packages/entity/src/index.ts @@ -46,7 +46,8 @@ export { } from './invariants'; export type { PostgresDriverOptions } from './pg-driver'; export { postgresDriver, postgresRepo, postgresTransactor } from './pg-driver'; -export type { ReadBuilder, Table } from './query'; +export type { RelatedTable, RelatedTables } from './preload'; +export type { Preloaded, ReadBuilder, Table } from './query'; export { tableFor } from './query'; export type { ColumnDescription, diff --git a/packages/entity/src/jit-preload.test.ts b/packages/entity/src/jit-preload.test.ts new file mode 100644 index 000000000..0ff266d45 --- /dev/null +++ b/packages/entity/src/jit-preload.test.ts @@ -0,0 +1,356 @@ +// What a page of rows buys the lookups that follow it. The headline is the sequential loop — +// `for … of` awaits between iterations, so the microtask coalescer cannot see two of its lookups +// at once and only the page they came from can batch them. The rest is the scope guard: a +// preloaded row is served to a lookup the preload statement WAS, and to no other. + +import { afterAll, beforeEach, describe, expect, test } from 'bun:test'; +import { createContext, runWithContext } from '@ultimat3/core'; +import { + createRecordingClient, + type DbClient, + type RecordingClient, + setDbClient, +} from '@ultimat3/db'; +import { MAX_IDS_PER_STATEMENT } from './batch-read'; +import { text, timestamp, uuid } from './columns'; +import { entity } from './entity'; +import { postgresRepo } from './pg-driver'; +import { clearRegistry } from './registry'; + +const users = entity('jit_test_users', { + columns: { + id: uuid().primaryKey(), + orgId: uuid().tenant(), + name: text({ max: 40 }), + deletedAt: timestamp().nullable(), + }, +}); + +const posts = entity('jit_test_posts', { + columns: { + id: uuid().primaryKey(), + orgId: uuid().tenant(), + authorId: uuid().references(() => users.id), + /** Nullable on purpose: a post nobody reviewed is data, not a key to go looking for. */ + reviewerId: uuid() + .references(() => users.id) + .nullable(), + title: text({ max: 80 }), + }, +}); + +const idAt = (index: number): string => + `00000000-0000-7000-8000-${String(index).padStart(12, '0')}`; + +const ORG = idAt(1); +const OTHER_ORG = idAt(2); + +/** What Bun.SQL hands back: snake_case names. */ +const postRow = (id: string, authorId: string, over: Record = {}): unknown => ({ + id, + org_id: ORG, + author_id: authorId, + reviewer_id: null, + title: `P-${id.slice(-2)}`, + ...over, +}); + +const userRow = (id: string, over: Record = {}): unknown => ({ + id, + org_id: ORG, + name: `U-${id.slice(-2)}`, + deleted_at: null, + ...over, +}); + +let client: RecordingClient; + +beforeEach(() => { + client = createRecordingClient(); + setDbClient(client); +}); + +afterAll(() => { + setDbClient(undefined); + clearRegistry(); +}); + +const postRepo = () => postgresRepo(posts); +const userRepo = () => postgresRepo(users); +const inRequest = (work: () => Promise): Promise => runWithContext(createContext(), work); + +/** A page of three posts by three authors, and the three authors it will be asked for. */ +const aPageOfThree = (): void => { + client.on('from "jit_test_posts"', { + rows: [postRow(idAt(10), idAt(20)), postRow(idAt(11), idAt(21)), postRow(idAt(12), idAt(22))], + }); + client.on('from "jit_test_users"', { + rows: [userRow(idAt(20)), userRow(idAt(21)), userRow(idAt(22))], + }); +}; + +const authorsOfThePage = async (): Promise => { + const page = await postRepo().findMany({ orgId: ORG }); + const names: (string | null)[] = []; + // A `for … of` awaits between iterations, so no two of these lookups share a microtask. + for (const post of page.rows) { + const author = await userRepo().findById(post.authorId, { orgId: ORG }); + names.push(author?.name ?? null); + } + return names; +}; + +describe('a page batches the lookups it causes', () => { + test('a sequential loop over a page is two statements, not one per row', async () => { + aPageOfThree(); + const names = await inRequest(authorsOfThePage); + + expect(client.statements).toHaveLength(2); + expect(client.texts[1]).toContain('"id" in ($1, $2, $3)'); + expect(names).toEqual(['U-20', 'U-21', 'U-22']); + }); + + test('the concurrent form is the same two statements', async () => { + aPageOfThree(); + const names = await inRequest(async () => { + const page = await postRepo().findMany({ orgId: ORG }); + const authors = await Promise.all( + page.rows.map((post) => userRepo().findById(post.authorId, { orgId: ORG })), + ); + return authors.map((author) => author?.name ?? null); + }); + + expect(client.statements).toHaveLength(2); + expect(names).toEqual(['U-20', 'U-21', 'U-22']); + }); + + test('forty rows by one author is one bind, not forty statements', async () => { + client.on('from "jit_test_posts"', { + rows: Array.from({ length: 40 }, (_, index) => postRow(idAt(100 + index), idAt(20))), + }); + client.on('from "jit_test_users"', { rows: [userRow(idAt(20))] }); + const names = await inRequest(authorsOfThePage); + + expect(client.statements).toHaveLength(2); + expect(client.statements[1]?.values).toEqual([idAt(20), ORG, 1]); + expect(names.every((name) => name === 'U-20')).toBe(true); + }); + + test('rows are served by id, never by the order they came back in', async () => { + client.on('from "jit_test_posts"', { + rows: [postRow(idAt(10), idAt(20)), postRow(idAt(11), idAt(21))], + }); + // Reversed on purpose: a caller served by position would get the other row. + client.on('from "jit_test_users"', { rows: [userRow(idAt(21)), userRow(idAt(20))] }); + + expect(await inRequest(authorsOfThePage)).toEqual(['U-20', 'U-21']); + }); + + test('an id the preload did not find is null, exactly as the single statement answered', async () => { + aPageOfThree(); + client.on('from "jit_test_users"', { rows: [userRow(idAt(20))] }); + + expect(await inRequest(authorsOfThePage)).toEqual(['U-20', null, null]); + expect(client.statements).toHaveLength(2); + }); + + test('two lookups of one preloaded id are one row object', async () => { + aPageOfThree(); + const [first, second] = await inRequest(async () => { + await postRepo().findMany({ orgId: ORG }); + return [ + await userRepo().findById(idAt(20), { orgId: ORG }), + await userRepo().findById(idAt(20), { orgId: ORG }), + ]; + }); + + expect(client.statements).toHaveLength(2); + expect(first).toBe(second); + }); + + test('a key that resolved to nothing is not a row to go looking for', async () => { + client.on('from "jit_test_posts"', { + rows: [ + postRow(idAt(10), idAt(20), { reviewer_id: idAt(30) }), + postRow(idAt(11), idAt(21), { reviewer_id: null }), + postRow(idAt(12), idAt(22), { reviewer_id: idAt(31) }), + ], + }); + await inRequest(async () => { + await postRepo().findMany({ orgId: ORG }); + await userRepo().findById(idAt(30), { orgId: ORG }); + }); + + // The two reviewers that exist, and neither the null nor any author: a second key to the same + // entity is its own group, and a key nobody set contributes nothing to it. + expect(client.statements[1]?.values).toEqual([idAt(30), idAt(31), ORG, 2]); + }); + + test('a page wider than the cap preloads whole statements, never one Postgres refuses', async () => { + const rows = Array.from({ length: MAX_IDS_PER_STATEMENT + 1 }, (_, index) => + postRow(idAt(1000 + index), idAt(2000 + index)), + ); + client.on('from "jit_test_posts"', { rows }); + await inRequest(async () => { + const page = await postRepo().findMany({ orgId: ORG, limit: rows.length }); + await userRepo().findById(page.rows[0]?.authorId ?? '', { orgId: ORG }); + }); + + expect(client.statements).toHaveLength(3); + // ids + the org predicate + the limit. + expect(client.statements[1]?.values).toHaveLength(MAX_IDS_PER_STATEMENT + 2); + expect(client.statements[2]?.values).toHaveLength(3); + }); +}); + +describe('the scope a preloaded row may be served to', () => { + test('the preload statement carries the scope the single lookups carried', async () => { + aPageOfThree(); + await inRequest(authorsOfThePage); + + expect(client.texts[1]).toContain('"org_id" = $4'); + expect(client.texts[1]).toContain('"deleted_at" is null'); + expect(client.statements[1]?.values).toEqual([idAt(20), idAt(21), idAt(22), ORG, 3]); + }); + + test('another tenant is never served from this one, and never shares its statement', async () => { + aPageOfThree(); + await inRequest(async () => { + await postRepo().findMany({ orgId: ORG }); + await userRepo().findById(idAt(20), { orgId: ORG }); + await userRepo().findById(idAt(20), { orgId: OTHER_ORG }); + }); + + expect(client.statements).toHaveLength(3); + // Its own statement, carrying its own predicate: the other tenant's rows are unreachable from + // it, so the answer is the one this caller's single statement would have produced. + expect(client.statements[2]?.values).toEqual([idAt(20), idAt(21), idAt(22), OTHER_ORG, 3]); + }); + + test('a lookup that reveals soft-deleted rows is never served the ones that hide them', async () => { + aPageOfThree(); + // `RepoOptions` is what `findById` takes and `shapeOf` reads `includeDeleted` off it — the + // live suite passes it, so the two visibilities have to be two statements. + const revealed = { orgId: ORG, includeDeleted: true }; + await inRequest(async () => { + await postRepo().findMany({ orgId: ORG }); + await userRepo().findById(idAt(20), { orgId: ORG }); + await userRepo().findById(idAt(20), revealed); + }); + + expect(client.statements).toHaveLength(3); + expect(client.texts[2]).not.toContain('"deleted_at" is null'); + }); + + test('another client is another place to read from', async () => { + aPageOfThree(); + const pinned = createRecordingClient(); + pinned.on('from "jit_test_users"', { rows: [userRow(idAt(20))] }); + await inRequest(async () => { + await postRepo().findMany({ orgId: ORG }); + await userRepo().findById(idAt(20), { orgId: ORG }); + await postgresRepo(users, { client: pinned }).findById(idAt(20), { orgId: ORG }); + }); + + expect(client.statements).toHaveLength(2); + expect(pinned.statements).toHaveLength(1); + }); + + test('a write to the entity is read back, never served from before it', async () => { + aPageOfThree(); + client.on('update "jit_test_users"', { rows: [userRow(idAt(20), { name: 'renamed' })] }); + const after = await inRequest(async () => { + await postRepo().findMany({ orgId: ORG }); + await userRepo().findById(idAt(20), { orgId: ORG }); + await userRepo().update(idAt(20), { name: 'renamed' }, { orgId: ORG }); + return userRepo().findById(idAt(20), { orgId: ORG }); + }); + + // page, preload, update, and the read the update forced. + expect(client.statements).toHaveLength(4); + expect(client.texts[3]).toContain('"id" in'); + // `U-20`, not `renamed`: the forced re-read matches the `from "jit_test_users"` stub, which + // never learned about the update. The statement count above is what proves the invalidation. + expect(after?.name).toBe('U-20'); + }); + + test('a write to another entity leaves this one preloaded', async () => { + aPageOfThree(); + client.on('insert into "jit_test_posts"', { rows: [postRow(idAt(13), idAt(20))] }); + await inRequest(async () => { + await postRepo().findMany({ orgId: ORG }); + await userRepo().findById(idAt(20), { orgId: ORG }); + await postRepo().insert({ id: idAt(13), orgId: ORG, authorId: idAt(20), title: 'P-13' }); + await userRepo().findById(idAt(21), { orgId: ORG }); + }); + + expect(client.statements).toHaveLength(3); + }); +}); + +describe('what a page never batches', () => { + test('an id that is no page’s foreign key is the statement it always was', async () => { + aPageOfThree(); + await inRequest(async () => { + await userRepo().findById(idAt(90), { orgId: ORG }); + await userRepo().findById(idAt(91), { orgId: ORG }); + }); + + expect(client.statements).toHaveLength(2); + // One id each, and neither carries the page's authors: an id no page indexed reaches the + // microtask batch, which a sequential pair of lookups never shares either. + expect(client.statements[0]?.values).toEqual([idAt(90), ORG, 1]); + expect(client.statements[1]?.values).toEqual([idAt(91), ORG, 1]); + }); + + test('with no request in scope a page leaves nothing behind', async () => { + aPageOfThree(); + await authorsOfThePage(); + + expect(client.statements).toHaveLength(4); + expect(client.texts.slice(1).every((statement) => statement.includes('"id" = $1'))).toBe(true); + }); + + test('a page belongs to one request, so the next request reads for itself', async () => { + aPageOfThree(); + await inRequest(() => postRepo().findMany({ orgId: ORG })); + await inRequest(() => userRepo().findById(idAt(20), { orgId: ORG })); + + expect(client.statements).toHaveLength(2); + // The id it was asked for, alone: the page's siblings died with the request that read them. + expect(client.statements[1]?.values).toEqual([idAt(20), ORG, 1]); + }); + + test('a row that no longer decodes fails its own caller, not the page', async () => { + aPageOfThree(); + // `name` is not-null, so a null from the database means the table no longer matches the + // entity. That is this row's problem: the caller of the other id still gets their row. + client.on('from "jit_test_users"', { + rows: [userRow(idAt(20), { name: null }), userRow(idAt(21))], + }); + await inRequest(async () => { + await postRepo().findMany({ orgId: ORG }); + await expect(userRepo().findById(idAt(20), { orgId: ORG })).rejects.toBeUltimateError( + 'X_INVARIANT_VIOLATED', + ); + await expect(userRepo().findById(idAt(21), { orgId: ORG })).resolves.toHaveProperty( + 'name', + 'U-21', + ); + }); + + expect(client.statements).toHaveLength(2); + }); + + test('a preload that fails fails the caller it was widened for', async () => { + aPageOfThree(); + const fail = (): Promise => Promise.reject(new RangeError('boom')); + const broken: DbClient = { query: fail, one: fail, execute: fail }; + await inRequest(async () => { + await postRepo().findMany({ orgId: ORG }); + await expect( + postgresRepo(users, { client: broken }).findById(idAt(20), { orgId: ORG }), + ).rejects.toThrow('boom'); + }); + }); +}); diff --git a/packages/entity/src/jit-preload.ts b/packages/entity/src/jit-preload.ts new file mode 100644 index 000000000..57d31e332 --- /dev/null +++ b/packages/entity/src/jit-preload.ts @@ -0,0 +1,216 @@ +// Single responsibility: the sibling-aware preload. A point lookup whose id is a foreign key on a +// page already read resolves that key for EVERY row of the page in one statement, and the rest of +// the loop is served from memory. A sequential `for … of` loop awaits between iterations, so its +// lookups never share a microtask and the coalescer cannot see them; this is what batches them. +// +// The trigger carries an id, not a row, so what a page leaves behind is an index of its foreign +// key VALUES rather than a map keyed by row identity: an id is a thing that can be looked up in +// it, and it holds values, so it pins no rows for the request's lifetime. +// +// The scope guard is a security boundary, not a tuning knob. A preloaded row is served only to a +// lookup with the same scope key, the same client and no write since — anything else reads the +// statement it always read. + +import type { Ctx } from '@ultimat3/core'; +import { tryUseContext } from '@ultimat3/core'; +import type { DbClient } from '@ultimat3/db'; +import { type Answer, keyOf, type PointRead, readByIds, statementChunks } from './batch-read'; +import type { EntityCore } from './entity'; + +/** The rows one page's worth of foreign keys resolved to, under one scope. */ +interface Bucket { + /** The entity the rows belong to — what a write invalidates. */ + readonly entity: string; + /** Where the rows were read from. A pinned client and the ambient pool are two places. */ + readonly client: DbClient; + /** Writes to `entity` when the bucket was opened. A later write makes every row of it stale. */ + readonly generation: number; + /** Never rejects: a failure is an `Answer`, so an id nobody asks for cannot go unhandled. */ + readonly rows: Map>; +} + +interface Store { + /** `[targetEntity, targetProperty]` -> id key -> every id the page carried for that key. */ + readonly siblings: Map>; + /** Scope key -> the rows preloaded under it. */ + readonly preloaded: Map; + /** Entity name -> writes this request has issued against it. */ + readonly writes: Map; +} + +/** + * Per request, keyed by ctx identity, so a page's siblings and everything preloaded from them die + * with the request that read them — the shape the microtask coalescer's store has, one file over. + */ +const requests = new WeakMap(); + +const storeFor = (ctx: Ctx): Store => { + const key: object = ctx; + const existing = requests.get(key); + if (existing !== undefined) return existing; + const created: Store = { siblings: new Map(), preloaded: new Map(), writes: new Map() }; + requests.set(key, created); + return created; +}; + +/** Both ends of the edge: a key pointing at another column of the same entity is another edge. */ +const siblingKey = (targetEntity: string, targetProperty: string): string => + JSON.stringify([targetEntity, targetProperty]); + +const writesTo = (store: Store, entity: string): number => store.writes.get(entity) ?? 0; + +/** + * What a page of rows leaves behind: for each foreign key it declares, the distinct values that + * page carried, filed under every one of them. A later `findById` for any of those values is a + * lookup this page can answer for all of them. + * + * Values only. Rows are not held, so a page read early in a long request costs its keys and not + * its rows, and a group of one still counts — a hundred posts by one author is one statement for + * the whole loop rather than a hundred. + */ +export const tagSiblings = (entity: EntityCore, rows: readonly Row[]): void => { + if (rows.length === 0) return; + const ctx = tryUseContext(); + // No request, no store: a job or a script reads the statement it always read. Asked first — + // resolving the foreign keys is a pass over the columns, and nothing would read the answer. + if (ctx === undefined) return; + const references = entity.$references(); + if (references.length === 0) return; + const store = storeFor(ctx); + for (const reference of references) { + // The declaring column's own kind: a foreign key mirrors the key it points at, and a value is + // filed here exactly as `findById` will spell it when it comes looking. + const kind = entity.$columns[reference.property]?.$meta.kind; + if (kind === undefined) continue; + const ids: unknown[] = []; + const keys = new Set(); + for (const row of rows) { + const value = (row as Record)[reference.property]; + // A nullable key that resolved to nothing is data, not a row to go looking for. + if (value === null || value === undefined) continue; + const key = keyOf(kind, value); + if (keys.has(key)) continue; + keys.add(key); + ids.push(value); + } + if (ids.length === 0) continue; + const at = siblingKey(reference.targetEntity, reference.targetProperty); + const index = store.siblings.get(at) ?? new Map(); + for (const key of keys) index.set(key, ids); + store.siblings.set(at, index); + } +}; + +/** + * A write makes every preloaded row of that entity stale, so the bucket holding them is not read + * again. Called before the statement goes out: a row read back afterwards is the row the write + * left, and one read concurrently with it was concurrent either way. + */ +export const forgetPreloaded = (entityName: string): void => { + const ctx = tryUseContext(); + if (ctx === undefined) return; + const store = requests.get(ctx); + if (store === undefined) return; + store.writes.set(entityName, writesTo(store, entityName) + 1); +}; + +/** The bucket this lookup may be served from, or `undefined` — a stale one is dropped, not read. */ +const usableBucket = ( + store: Store, + read: PointRead, + scope: string, +): Bucket | undefined => { + const open = store.preloaded.get(scope); + if (open === undefined) return undefined; + if (open.client === read.client && open.generation === writesTo(store, open.entity)) return open; + store.preloaded.delete(scope); + return undefined; +}; + +type Settlers = ReadonlyMap void>; + +/** One statement at a time: a page wide enough to split must not take the pool with it. */ +const fill = async ( + read: PointRead, + ids: readonly unknown[], + settlers: Settlers, +): Promise => { + for (const chunk of statementChunks(ids)) { + let answers: ReadonlyMap; + try { + answers = await readByIds(read, chunk); + } catch (error) { + // The statement failed, so it fails for everyone it was widened to cover — which is what + // the single statement each of them would have sent would have done. + for (const id of chunk) settlers.get(keyOf(read.key.kind, id))?.({ error }); + continue; + } + for (const id of chunk) { + const at = keyOf(read.key.kind, id); + // An id the statement did not answer for is a row that is not there — `findById`'s null. + settlers.get(at)?.(answers.get(at) ?? { row: null }); + } + } +}; + +/** + * The statement, and one promise per id it will answer for. Only the ids the bucket does not + * already hold are read: a second lookup into a page it already resolved is memory, not a wire. + */ +const preload = (read: PointRead, bucket: Bucket, ids: readonly unknown[]): void => { + const settlers = new Map void>(); + const wanted: unknown[] = []; + for (const id of ids) { + const at = keyOf(read.key.kind, id); + if (bucket.rows.has(at)) continue; + // The executor runs synchronously, so `settle` is assigned before the promise is stored. + let settle!: (answer: Answer) => void; + bucket.rows.set( + at, + new Promise((resolve) => { + settle = resolve; + }), + ); + settlers.set(at, settle); + wanted.push(id); + } + // Nothing awaits this: an id nobody asks for still settles, and it settles with an `Answer` + // rather than a rejection, so a failed statement cannot go unhandled. + void fill(read, wanted, settlers); +}; + +const answered = (answer: Promise): Promise => + answer.then((settled) => + 'error' in settled ? Promise.reject(settled.error) : (settled.row as Row | null), + ); + +/** + * The row this lookup is served from a page already read, or `undefined` when no page can answer + * it: nothing read in this request, an id that is no page's foreign key, a write since, or + * another client. Declining is always correct — the caller reads the statement it always read. + */ +export const preloadedFindById = ( + ctx: Ctx, + read: PointRead, + scope: string, + id: unknown, +): Promise | undefined => { + const store = requests.get(ctx); + if (store === undefined) return undefined; + const filedAt = keyOf(read.key.kind, id); + const bucket = usableBucket(store, read, scope); + const already = bucket?.rows.get(filedAt); + if (already !== undefined) return answered(already); + const ids = store.siblings.get(siblingKey(read.entity.$name, read.key.property))?.get(filedAt); + if (ids === undefined) return undefined; + const target = bucket ?? { + entity: read.entity.$name, + client: read.client, + generation: writesTo(store, read.entity.$name), + rows: new Map>(), + }; + store.preloaded.set(scope, target); + preload(read, target, ids); + const answer = target.rows.get(filedAt); + return answer === undefined ? undefined : answered(answer); +}; diff --git a/packages/entity/src/pg-driver.live.test.ts b/packages/entity/src/pg-driver.live.test.ts index 4fb6c59b4..c4f3257f5 100644 --- a/packages/entity/src/pg-driver.live.test.ts +++ b/packages/entity/src/pg-driver.live.test.ts @@ -10,13 +10,14 @@ // `db-integration.test.ts`; CI's `postgres` service container sets it. import { afterAll, beforeAll, describe, expect, test } from 'bun:test'; -import { isUltimateError } from '@ultimat3/core'; +import { createContext, isUltimateError, runWithContext } from '@ultimat3/core'; import { createPostgresClient, generateMigration, type PostgresClient, raw, setDbClient, + setStatementObserver, } from '@ultimat3/db'; import { boolean, money, text, timestamp, uuid } from './columns'; import { database } from './database'; @@ -243,6 +244,95 @@ describe.skipIf(!hasPostgres)('live · postgres · postgresDriver', () => { expect(revealed?.deletedAt).toBeInstanceOf(Date); }); + test('point lookups in one request are one statement, and it answers like the singles did', async () => { + const first = await write(acme, 'INV-batch-1'); + const second = await write(acme, 'INV-batch-2'); + const hidden = await write(acme, 'INV-batch-3'); + await repo().delete(hidden.id, { orgId: acme }); + const foreign = await write(other, 'INV-batch-4'); + + // The observer is how "one statement" is asserted against a real server: there is no + // recording client here, and the count is the whole claim. + const reads: string[] = []; + setStatementObserver({ + onStatement: (event) => { + if (event.text.includes('from "pg_live_invoices"')) reads.push(event.text); + }, + }); + try { + const rows = await runWithContext(createContext(), () => + Promise.all([ + repo().findById(first.id, { orgId: acme }), + repo().findById(second.id, { orgId: acme }), + repo().findById(hidden.id, { orgId: acme }), + repo().findById(foreign.id, { orgId: acme }), + repo().findById(first.id, { orgId: acme }), + ]), + ); + + expect(reads).toHaveLength(1); + // Five lookups, four binds: the repeated id is one of them. + expect(reads[0]).toContain('"id" in ($1, $2, $3, $4)'); + // Every answer is the one the single statement gave — Postgres applied the tenant predicate + // and the `deleted_at is null` clause inside the coalesced statement, not the process after. + expect(rows.map((row) => row?.reference ?? null)).toEqual([ + 'INV-batch-1', + 'INV-batch-2', + null, + null, + 'INV-batch-1', + ]); + } finally { + setStatementObserver(undefined); + } + }); + + test('a sequential loop over a page of tags is two statements against the invoices it points at', async () => { + // The JIT-preload half of the same claim, against a real server: `findMany` on `invoiceTags` + // leaves its page's `invoiceId` values behind, so a `for … of` loop that follows — one + // `await` per iteration, no two lookups sharing a microtask — costs one widened `in` query + // for the whole page, not one `select` per row. + const org = await newOrg('jit-live'); + const tags = () => postgresRepo(invoiceTags); + const written = await Promise.all( + ['JIT-1', 'JIT-2', 'JIT-3'].map((reference) => write(org, reference)), + ); + for (const invoice of written) { + await tags().insert({ invoiceId: invoice.id, orgId: org, label: 'primary', note: null }); + } + + // Filtered to the two tables under test, as the sibling test above is: any incidental + // statement in this window — a connection-checkout `set`, a pool keepalive, a `begin` — would + // otherwise read as "expected 2, got 3" with nothing naming the intruder. + const statements: string[] = []; + setStatementObserver({ + onStatement: (event) => { + if (/from "pg_live_(?:invoice_tags|invoices)"/.test(event.text)) + statements.push(event.text); + }, + }); + try { + const references = await runWithContext(createContext(), async () => { + const page = await tags().findMany({ orgId: org }); + const seen: string[] = []; + for (const tagRow of page.rows) { + const invoice = await repo().findById(tagRow.invoiceId, { orgId: org }); + seen.push(invoice?.reference ?? 'missing'); + } + return seen; + }); + + // The tags page, and one `in (…)` statement for the three invoices it named — never three. + expect(statements).toHaveLength(2); + const invoiceReads = statements.filter((text) => text.includes('from "pg_live_invoices"')); + expect(invoiceReads).toHaveLength(1); + expect(invoiceReads[0]).toContain('"id" in ($1, $2, $3)'); + expect(references.slice().sort()).toEqual(['JIT-1', 'JIT-2', 'JIT-3']); + } finally { + setStatementObserver(undefined); + } + }); + test('delete on an entity without a soft-delete column removes the row', async () => { const target = await database({ orgs }, { driver: postgresDriver() }).orgs.insert({ slug: 'hard-delete', diff --git a/packages/entity/src/pg-driver.test.ts b/packages/entity/src/pg-driver.test.ts index 8f2f29df8..e6f3460d2 100644 --- a/packages/entity/src/pg-driver.test.ts +++ b/packages/entity/src/pg-driver.test.ts @@ -1,4 +1,5 @@ import { afterAll, beforeEach, describe, expect, test } from 'bun:test'; +import { createContext, runWithContext } from '@ultimat3/core'; import { createRecordingClient, type RecordingClient, setDbClient } from '@ultimat3/db'; import { boolean, money, text, timestamp, uuid } from './columns'; import { database } from './database'; @@ -29,8 +30,9 @@ const invoices = entity('pg_test_invoices', { type Invoice = typeof invoices.$row; const ORG = '00000000-0000-7000-8000-0000000000a1'; -const _OTHER_ORG = '00000000-0000-7000-8000-0000000000a2'; +const OTHER_ORG = '00000000-0000-7000-8000-0000000000a2'; const ID = '00000000-0000-7000-8000-000000000101'; +const OTHER_ID = '00000000-0000-7000-8000-000000000102'; /** What Bun.SQL hands back: snake_case names, int8 as a string, timestamptz as an ISO string. */ const physical = (over: Record = {}): Record => ({ @@ -306,3 +308,39 @@ describe('composition', () => { expect(client.texts).not.toContain('COMMIT'); }); }); + +describe('postgresRepo() jitPreload config', () => { + /** + * A page of invoices carries `orgId`, a foreign key to `orgs`, so an enabled page leaves both + * org ids behind and the sequential loop that follows costs one widened statement for the two of + * them. Disabled, it is the two statements a loop always sent — which is the only difference the + * switch makes, and the reason this counts statements rather than asserting the page came back. + */ + const pageThenLoop = async (jitPreload: boolean): Promise => { + client.on('from "pg_test_invoices"', { + rows: [physical({ id: ID }), physical({ id: OTHER_ID, org_id: OTHER_ORG })], + }); + client.on('from "pg_test_orgs"', { + rows: [ + { id: ORG, slug: 'ours' }, + { id: OTHER_ORG, slug: 'theirs' }, + ], + }); + await runWithContext(createContext(), async () => { + const page = await postgresRepo(invoices, { jitPreload }).findMany({ orgId: ORG, limit: 2 }); + // A `for … of` awaits between iterations, so no two of these share a microtask: only the + // page they came from can batch them. + for (const invoice of page.rows) await postgresRepo(orgs).findById(invoice.orgId); + }); + return client.statements.length; + }; + + test('defaults to enabled, so the page batches the loop that follows it', async () => { + expect(await pageThenLoop(true)).toBe(2); + expect(lastText()).toContain('"id" in ($1, $2)'); + }); + + test('disabled, the loop after the page is the statements it always sent', async () => { + expect(await pageThenLoop(false)).toBe(3); + }); +}); diff --git a/packages/entity/src/pg-driver.ts b/packages/entity/src/pg-driver.ts index 8420d99e9..2ea0f866b 100644 --- a/packages/entity/src/pg-driver.ts +++ b/packages/entity/src/pg-driver.ts @@ -14,11 +14,13 @@ import { type TransactionOptions, withTransaction, } from '@ultimat3/db'; +import { coalesceFindById } from './coalesce'; import { snake } from './column'; import { cursorFor, seekFrom, valueAt } from './cursor'; import type { Driver } from './database'; import { type EntityCore, SOFT_DELETE_COLUMN } from './entity'; import { notFound } from './errors'; +import { forgetPreloaded, tagSiblings } from './jit-preload'; import { bindValues, decodeRow, type PhysicalRow } from './pg-row'; import { countStatement, @@ -39,6 +41,12 @@ export interface PostgresDriverOptions { * globally with `setDbClient()`. */ readonly client?: DbClient | undefined; + /** + * Preload foreign keys resolved by a page into a request-scoped cache, so the first + * `findById` for any one of them resolves that key for the whole page in one statement. + * Default: true. + */ + readonly jitPreload?: boolean | undefined; } const shapeOf = (args: FindManyArgs, seek?: readonly unknown[]): ReadShape => ({ @@ -76,10 +84,26 @@ export const postgresRepo = ( ) : deleteStatement(entity, plan); + /** + * Every write goes out through here, which makes it the ONE place the request's preloaded rows + * are dropped: a row this statement changes must not be served afterwards from a page read + * before it. Before the statement, not after — a row read back afterwards is the row this write + * left, and one read concurrently with it was concurrent either way. + */ + const writing = (send: () => Promise): Promise => { + forgetPreloaded(entity.$name); + return send(); + }; + return { async findById(id, options) { const plan = idPlan(entity, id, options, 'findById'); - return one(plan, options ?? {}); + const args = options ?? {}; + // Every point lookup a request issues in one microtask is one statement: a page that + // resolves an author per row pays for one round trip, not one per row. The statement is the + // one this call would have sent — same scope, same soft-delete filter, `in` instead of `=` — + // and with no request in scope (a job, a script) it is exactly that statement, alone. + return coalesceFindById(entity, client(), plan, shapeOf(args), id) ?? one(plan, args); }, async findMany(args = {}) { @@ -90,6 +114,10 @@ export const postgresRepo = ( selectStatement(entity, plan, shapeOf(args, seekFrom(entity, plan)), plan.limit + 1), ); const rows = found.slice(0, plan.limit).map((row) => decodeRow(entity, row)); + // What the page leaves behind: its foreign key values, so the first `findById` for any one + // of them resolves the key for the whole page. A `for … of` loop over these rows costs one + // statement, not one per row — its `await` already ended the coalescing window. + if (config.jitPreload !== false) tagSiblings(entity, rows); const last = rows.at(-1); return { rows, @@ -102,8 +130,8 @@ export const postgresRepo = ( async insert(values) { entity.$assert(values); - const written = await client().one( - insertStatement(entity, bindValues(entity, values)), + const written = await writing(() => + client().one(insertStatement(entity, bindValues(entity, values))), ); // `returning *` is the row Postgres actually stored, defaults included. return written === null ? values : decodeRow(entity, written); @@ -117,8 +145,8 @@ export const postgresRepo = ( if (current === null) throw notFound(entity.$name, id); return current; } - const written = await client().one( - updateStatement(entity, plan, values, shapeOf(options ?? {})), + const written = await writing(() => + client().one(updateStatement(entity, plan, values, shapeOf(options ?? {}))), ); if (written === null) throw notFound(entity.$name, id); const after = decodeRow(entity, written); @@ -131,13 +159,17 @@ export const postgresRepo = ( async delete(id, options) { const plan = idPlan(entity, id, options, 'delete'); - if ((await client().execute(removal(plan))) === 0) throw notFound(entity.$name, id); + if ((await writing(() => client().execute(removal(plan)))) === 0) { + throw notFound(entity.$name, id); + } }, // No `X_NOT_FOUND` here: a filter that matches nothing is a fact the caller asked for, not a // failed address. The count is the answer, which is why it is a `number` and not `void`. async deleteWhere(filter, options) { - return client().execute(removal(deletePlan(entity, filter, options, 'deleteWhere'))); + // The plan is built before the write is announced: a refused filter never happened. + const plan = deletePlan(entity, filter, options, 'deleteWhere'); + return writing(() => client().execute(removal(plan))); }, async updateWhere(filter, patch, options) { @@ -149,7 +181,7 @@ export const postgresRepo = ( // re-asserts every row it writes, and a JS-only invariant (`kind: 'assert'`, `sql: null`) // has no CHECK for Postgres to have enforced. Inside `withTransaction` the throw takes the // whole statement with it. - const written = await client().query(statement); + const written = await writing(() => client().query(statement)); for (const row of written) entity.$assert(decodeRow(entity, row)); return written.length; }, diff --git a/packages/entity/src/preload-statements.test.ts b/packages/entity/src/preload-statements.test.ts new file mode 100644 index 000000000..06bcfc353 --- /dev/null +++ b/packages/entity/src/preload-statements.test.ts @@ -0,0 +1,160 @@ +// What `preload()` costs on the wire: one extra statement per relation, over the distinct keys the +// page carried, carrying the scope the page was read under. Counted against the recording client, +// because "one statement" is a claim about SQL and only SQL can answer it. + +import { afterAll, beforeEach, describe, expect, test } from 'bun:test'; +import { createRecordingClient, type RecordingClient, setDbClient } from '@ultimat3/db'; +import { text, timestamp, uuid } from './columns'; +import { database } from './database'; +import { entity } from './entity'; +import { postgresDriver } from './pg-driver'; +import { clearRegistry } from './registry'; + +const orgs = entity('preload_stmt_orgs', { + columns: { id: uuid().primaryKey(), slug: text({ max: 40 }) }, +}); + +const members = entity('preload_stmt_members', { + columns: { + id: uuid().primaryKey(), + orgId: uuid() + .references(() => orgs.id) + .tenant(), + email: text({ max: 120 }), + deletedAt: timestamp().nullable(), + }, +}); + +const posts = entity('preload_stmt_posts', { + columns: { + id: uuid().primaryKey(), + orgId: uuid() + .references(() => orgs.id) + .tenant(), + authorId: uuid().references(() => members.id), + reviewerId: uuid() + .references(() => members.id) + .nullable(), + title: text({ max: 120 }), + }, +}); + +const idAt = (index: number): string => + `00000000-0000-7000-8000-${String(index).padStart(12, '0')}`; + +const ORG = idAt(1); +const ANA = idAt(10); +const BEN = idAt(11); + +/** What Bun.SQL hands back: snake_case names, nothing decoded. */ +const postRow = (index: number, authorId: string, reviewerId: string | null = null) => ({ + id: idAt(index), + org_id: ORG, + author_id: authorId, + reviewer_id: reviewerId, + title: `Post ${index}`, +}); + +const memberRow = (id: string) => ({ + id, + org_id: ORG, + email: `${id}@example.com`, + deleted_at: null, +}); + +let client: RecordingClient; + +beforeEach(() => { + client = createRecordingClient(); + setDbClient(client); + // Three posts, two authors: what the preload binds is the distinct set, not the page. + client.on('preload_stmt_posts', { + rows: [postRow(20, ANA), postRow(21, ANA, BEN), postRow(22, BEN)], + }); + client.on('preload_stmt_members', { rows: [memberRow(ANA), memberRow(BEN)] }); +}); + +afterAll(() => { + setDbClient(undefined); + clearRegistry(); +}); + +const db = () => database({ orgs, members, posts }, { driver: postgresDriver() }); + +const preloadStatement = (): string => client.texts[1] ?? ''; + +describe('what a preload costs', () => { + test('a page and its relation are two statements, whatever the page holds', async () => { + const rows = await db().posts.where({ orgId: ORG }).preload('author').all(); + + expect(rows).toHaveLength(3); + expect(client.texts).toHaveLength(2); + expect(preloadStatement()).toContain('from "preload_stmt_members"'); + }); + + test('the keys are the distinct ones the page carried — three rows, two binds', async () => { + await db().posts.where({ orgId: ORG }).preload('author').all(); + + expect(preloadStatement()).toContain('"id" in ($1, $2)'); + expect(client.statements[1]?.values.slice(0, 2)).toEqual([ANA, BEN]); + }); + + test('two relations are one statement each', async () => { + await db().posts.where({ orgId: ORG }).preload('author').preload('reviewer').all(); + + expect(client.texts).toHaveLength(3); + }); + + test('naming one relation twice is one statement — a chain is not a queue', async () => { + await db().posts.where({ orgId: ORG }).preload('author').preload('author').all(); + + expect(client.texts).toHaveLength(2); + }); + + test('a page with no key to resolve sends no second statement', async () => { + client.on('preload_stmt_posts', { rows: [] }); + + expect(await db().posts.where({ orgId: ORG }).preload('author').all()).toEqual([]); + expect(client.texts).toHaveLength(1); + }); +}); + +describe('what the preload statement carries', () => { + test('the tenant predicate the page was read under is inside the SQL Postgres runs', async () => { + await db().posts.where({ orgId: ORG }).preload('author').all(); + + expect(preloadStatement()).toContain('"org_id" ='); + expect(client.statements[1]?.values).toContain(ORG); + }); + + test('the soft-delete clause is the one any other read of that entity carries', async () => { + await db().posts.where({ orgId: ORG }).preload('author').all(); + + expect(preloadStatement()).toContain('"deleted_at" is null'); + }); + + test('a projection cannot drop the key the preload reads', async () => { + await db() + .posts.where({ orgId: ORG }) + .select({ id: true, title: true }) + .preload('author') + .all(); + + // The caller named two columns; the statement asks for the third because the framework does. + expect(client.texts[0]).toContain('"author_id"'); + }); + + test('a hasMany reads the other side by its own foreign key', async () => { + await db().members.where({ orgId: ORG }).preload('preload_stmt_postsByAuthor').all(); + + expect(preloadStatement()).toContain('from "preload_stmt_posts"'); + expect(preloadStatement()).toContain('"author_id" in'); + }); + + test('count() reads no row, so it attaches nothing and sends nothing extra', async () => { + client.on('count(', { rows: [{ count: '3' }] }); + + expect(await db().posts.where({ orgId: ORG }).preload('author').count()).toBe(3); + expect(client.texts).toHaveLength(1); + }); +}); diff --git a/packages/entity/src/preload.test.ts b/packages/entity/src/preload.test.ts new file mode 100644 index 000000000..befa12fd8 --- /dev/null +++ b/packages/entity/src/preload.test.ts @@ -0,0 +1,264 @@ +// What `preload('')` means, in the driver a test can seed: which rows attach, what a +// missing one reads as, and the four refusals. The statement count it costs is pinned against the +// recording client in `preload-statements.test.ts` — the same feature, from the wire's side. + +import { afterAll, beforeEach, describe, expect, test } from 'bun:test'; +import { text, timestamp, uuid } from './columns'; +import { database, memoryDriver } from './database'; +import { entity } from './entity'; +import { tableFor } from './query'; +import { clearRegistry } from './registry'; +import { memoryRepo } from './repo'; + +/** The tenant root: scoped by nobody, which is what makes it the target of a `belongsTo`. */ +const orgs = entity('preload_test_orgs', { + columns: { id: uuid().primaryKey(), slug: text({ max: 40 }) }, +}); + +const members = entity('preload_test_members', { + columns: { + id: uuid().primaryKey(), + orgId: uuid() + .references(() => orgs.id) + .tenant(), + email: text({ max: 120 }), + }, +}); + +const posts = entity('preload_test_posts', { + columns: { + id: uuid().primaryKey(), + orgId: uuid() + .references(() => orgs.id) + .tenant(), + authorId: uuid().references(() => members.id), + /** Nullable on purpose: a post nobody reviewed is data, not a broken key. */ + reviewerId: uuid() + .references(() => members.id) + .nullable(), + title: text({ max: 120 }), + deletedAt: timestamp().nullable(), + }, +}); + +/** Unscoped itself, pointing at a scoped entity — the read a tenant predicate cannot be guessed for. */ +const audits = entity('preload_test_audits', { + columns: { + id: uuid().primaryKey(), + memberId: uuid().references(() => members.id), + action: text({ max: 40 }), + }, +}); + +/** + * Scoped by `workspaceId`, and carrying an ordinary `orgId` column that is a filter and not its + * tenancy — the one shape where matching the target's column name alone would carry a predicate + * nobody proved this reader owns. + */ +const tickets = entity('preload_test_tickets', { + columns: { + id: uuid().primaryKey(), + workspaceId: uuid().tenant(), + orgId: uuid(), + memberId: uuid().references(() => members.id), + }, +}); + +const idAt = (index: number): string => + `00000000-0000-7000-8000-${String(index).padStart(12, '0')}`; + +/** + * A `hasMany` is named for the entity the rows come from, and both inbound keys want that name — + * so every member of the group takes its long form. Spelled out here because the entity names in + * a test file carry a prefix and the derived name carries it too. + */ +const BY_AUTHOR = 'preload_test_postsByAuthor'; +const BY_REVIEWER = 'preload_test_postsByReviewer'; + +const ORG = idAt(1); +const OTHER_ORG = idAt(2); +const ANA = idAt(10); +const BEN = idAt(11); +const THEIRS = idAt(12); + +type Db = ReturnType; + +const handle = () => + database({ orgs, members, posts, audits, tickets }, { driver: memoryDriver() }); + +let db: Db; + +beforeEach(async () => { + db = handle(); + await db.orgs.insert({ id: ORG, slug: 'ours' }); + await db.orgs.insert({ id: OTHER_ORG, slug: 'theirs' }); + await db.members.insert({ id: ANA, orgId: ORG, email: 'ana@example.com' }); + await db.members.insert({ id: BEN, orgId: ORG, email: 'ben@example.com' }); + await db.members.insert({ id: THEIRS, orgId: OTHER_ORG, email: 'them@example.com' }); + await db.posts.insert({ id: idAt(20), orgId: ORG, authorId: ANA, title: 'First' }); + await db.posts.insert({ + id: idAt(21), + orgId: ORG, + authorId: ANA, + reviewerId: BEN, + title: 'Second', + }); + await db.posts.insert({ id: idAt(22), orgId: ORG, authorId: BEN, title: 'Third' }); +}); + +afterAll(() => { + clearRegistry(); +}); + +const field = (row: unknown, key: string): unknown => (row as Record)[key]; + +describe('what attaches', () => { + test('a belongsTo attaches the row itself, under the relation the foreign key names', async () => { + const rows = await db.posts.where({ orgId: ORG }).orderBy('id').preload('author').all(); + + expect(rows).toHaveLength(3); + expect(field(rows[0], 'author')).toMatchObject({ id: ANA, email: 'ana@example.com' }); + expect(field(rows[2], 'author')).toMatchObject({ id: BEN }); + }); + + test('a key that resolved to nothing attaches null — never a missing property', async () => { + const rows = await db.posts.where({ orgId: ORG }).orderBy('id').preload('reviewer').all(); + + expect(field(rows[0], 'reviewer')).toBeNull(); + expect(field(rows[1], 'reviewer')).toMatchObject({ id: BEN }); + // Present on every row: "nobody reviewed it" and "nobody preloaded the reviewer" must differ. + expect(rows.every((row) => 'reviewer' in (row as object))).toBe(true); + }); + + test('a hasMany attaches an array, empty rather than absent when the page has none', async () => { + const rows = await db.members.where({ orgId: ORG }).orderBy('id').preload(BY_AUTHOR).all(); + + expect( + (field(rows[0], BY_AUTHOR) as readonly unknown[]).map((row) => field(row, 'id')), + ).toEqual([idAt(20), idAt(21)]); + expect(field(rows[1], BY_AUTHOR)).toHaveLength(1); + // One relation named is one relation attached: the other inbound key is nobody's business. + expect(field(rows[0], BY_REVIEWER)).toBeUndefined(); + + const reviewed = await db.members + .where({ orgId: ORG }) + .orderBy('id') + .preload(BY_REVIEWER) + .all(); + expect(field(reviewed[0], BY_REVIEWER)).toEqual([]); + }); + + test('several relations attach together, each under its own name', async () => { + const [row] = await db.posts + .where({ orgId: ORG, id: idAt(21) }) + .preload('author') + .preload('reviewer') + .all(); + + expect(field(row, 'author')).toMatchObject({ id: ANA }); + expect(field(row, 'reviewer')).toMatchObject({ id: BEN }); + }); + + test('every terminal that returns rows preloads; count() has no row to attach to', async () => { + const chain = () => db.posts.where({ orgId: ORG }).orderBy('id').preload('author'); + + expect(field((await chain().page()).rows[0], 'author')).toMatchObject({ id: ANA }); + expect(field(await chain().one(), 'author')).toMatchObject({ id: ANA }); + expect(await chain().count()).toBe(3); + }); + + test('a projection narrows the columns, never the relations', async () => { + const chain = db.posts + .where({ orgId: ORG }) + .orderBy('id') + .select({ id: true, title: true }) + .preload('author'); + const [row] = await chain.all(); + + expect(field(row, 'author')).toMatchObject({ id: ANA }); + // The key the preload reads is asked for even though the caller did not name it. + expect(chain.plan().select).toEqual(['id', 'title', 'authorId']); + }); + + test('the attachment is a copy — the driver keeps handing back an unpreloaded row', async () => { + await db.posts.where({ orgId: ORG }).preload('author').all(); + const [plain] = await db.posts.where({ orgId: ORG, id: idAt(20) }).all(); + + expect('author' in (plain as object)).toBe(false); + }); +}); + +describe('the refusals', () => { + test('an unknown relation fails on the chain, not one page later', () => { + let error: unknown; + try { + db.posts.where({ orgId: ORG }).preload('authorr'); + } catch (thrown) { + error = thrown; + } + expect(error).toBeUltimateError('X_PRELOAD_UNKNOWN_RELATION'); + // The names that do exist are in the fix, because a relation is derived and listed nowhere. + expect(String((error as { fix: string }).fix)).toContain('author'); + }); + + test('a relation whose other end the database() call never named is refused', async () => { + const partial = database({ posts }, { driver: memoryDriver() }); + await partial.posts.insert({ id: idAt(20), orgId: ORG, authorId: ANA, title: 'Alone' }); + + await expect( + partial.posts.where({ orgId: ORG }).preload('author').all(), + ).rejects.toBeUltimateError('X_INVARIANT_VIOLATED'); + }); + + test('a table built by hand reaches no other table, and says so the same way', async () => { + const row = posts.$parse({ id: idAt(20), orgId: ORG, authorId: ANA, title: 'Alone' }); + const bare = tableFor(posts, memoryRepo(posts, [row])); + + await expect(bare.where({ orgId: ORG }).preload('author').all()).rejects.toBeUltimateError( + 'X_INVARIANT_VIOLATED', + ); + }); + + test('a tenant scope is carried, never guessed — an unscoped source refuses a scoped target', async () => { + await db.audits.insert({ id: idAt(30), memberId: ANA, action: 'login' }); + + await expect(db.audits.preload('member').all()).rejects.toBeUltimateError('X_TENANCY_UNSCOPED'); + }); + + test('a predicate that only shares the target’s column name is not this source’s tenancy', async () => { + // `tickets` is scoped by `workspaceId`; its `orgId` is an ordinary filter. Carrying it would + // scope the members read to a tenant the reader never proved they own. + await db.tickets.insert({ id: idAt(31), workspaceId: idAt(3), orgId: ORG, memberId: ANA }); + + await expect( + db.tickets + .where({ workspaceId: idAt(3), orgId: ORG }) + .preload('member') + .all(), + ).rejects.toBeUltimateError('X_TENANCY_UNSCOPED'); + }); +}); + +describe('the scope the page was read under', () => { + test('another tenant’s row is not attached, whichever tenant asks', async () => { + // A post pointing across the tenant boundary: the FK is there, the row is not readable. + await db.posts.insert({ id: idAt(23), orgId: ORG, authorId: THEIRS, title: 'Crossed' }); + + const [crossed] = await db.posts + .where({ orgId: ORG, id: idAt(23) }) + .preload('author') + .all(); + + expect(field(crossed, 'author')).toBeNull(); + }); + + test('a soft-deleted row is as invisible to a preload as it is to any other read', async () => { + await db.posts.delete(idAt(20), { orgId: ORG }); + + const [ana] = await db.members.where({ orgId: ORG, id: ANA }).preload(BY_AUTHOR).all(); + + expect((field(ana, BY_AUTHOR) as readonly unknown[]).map((row) => field(row, 'id'))).toEqual([ + idAt(21), + ]); + }); +}); diff --git a/packages/entity/src/preload.ts b/packages/entity/src/preload.ts new file mode 100644 index 000000000..5acce8e6d --- /dev/null +++ b/packages/entity/src/preload.ts @@ -0,0 +1,184 @@ +// Single responsibility: the eager preload. `preload('author')` is one extra +// `select … where in (…)` over the values the page already carries, and its rows attached to +// that page under the relation's name — the declarative form of what a point lookup batches for +// itself, and the exact line an N+1 warning tells the reader to write. +// +// Nothing new is declared: the relation is the `references()` already written (`relations.ts`). +// The other table is read through the same driver this one came from, so a preload against the +// in-memory driver means what a preload against Postgres means. + +import { keyOf, MAX_IDS_PER_STATEMENT, statementChunks } from './batch-read'; +import { valueAt } from './cursor'; +import type { EntityCore } from './entity'; +import { EntityError } from './errors'; +import type { Relation } from './relations'; +import type { Repo } from './repo'; +import type { Predicate } from './tenancy'; + +/** The other side of a relation: the entity, and where its rows live. */ +export interface RelatedTable { + readonly entity: EntityCore; + readonly repo: Repo; +} + +/** + * How one table reaches another. `database()` supplies it over the set it was declared with and + * through the driver that set was given, so a preload reads rows from where this table's own rows + * come from — and a table can no more preload an entity the set never named than it can be indexed + * off `db` by one. + */ +export type RelatedTables = (entityName: string) => RelatedTable | undefined; + +/** + * Not `invariantViolated`: its fix opens `x entity explain`, which describes invariants nobody + * wrote here. What repairs this is one edit to the `database()` call — a relation whose other end + * is outside the set is a set that is missing an entity, and the preload declines rather than + * reaching around the handle for it. + */ +const unreachable = (entityName: string, relation: Relation): EntityError => + new EntityError({ + code: 'X_INVARIANT_VIOLATED', + cause: `${entityName}.preload('${relation.name}') reads ${relation.to}, which this table cannot reach — a table reads the entities its own database() call named`, + fix: `x entities list --json # then widen the database() call that built db.${entityName} to database({ ${entityName}, ${relation.to} }, options)`, + }); + +/** + * The tenant predicate the page was read under, carried onto the related read when BOTH entities + * are scoped by a column of that same name. Never across two differently-named tenant columns: a + * value that scopes one entity is a guess on another, and a guess here is a cross-tenant read. + * + * Both ends are checked, not just the target's. A source scoped by `workspaceId` may still carry + * an ordinary `orgId` predicate of its own — a filter, not its tenancy — and matching on the + * target's column name alone would lift that filter into the target's tenant scope and hand the + * preload rows from a tenant nobody proved this reader owns. + * + * Carrying nothing is not a failure of this function — the related read builds its own plan, so + * `assertScoped` refuses it there, in the words a caller can act on. + */ +const tenantScope = ( + source: EntityCore, + target: EntityCore, + where: readonly Predicate[], +): readonly Predicate[] => { + const column = target.$tenantColumn; + if (column === null || source.$tenantColumn !== column) return []; + const carried = where.find((predicate) => predicate.column === column && predicate.op === 'eq'); + return carried === undefined ? [] : [carried]; +}; + +/** + * Every row on the other side, in as few statements as the bind count allows: one per 500 keys — + * the bound a batched point read already lives under — and one more only when a page comes back + * genuinely full. A `belongsTo` over a page of 50 is exactly one statement. + * + * The page loop is what keeps a `hasMany` honest: a relation with more rows than one page holds + * costs another statement rather than silently returning the first page of them. + */ +const relatedRows = async ( + target: RelatedTable, + relation: Relation, + values: readonly unknown[], + scope: readonly Predicate[], +): Promise => { + const rows: unknown[] = []; + for (const chunk of statementChunks(values)) { + let cursor: string | null = null; + do { + const page = await target.repo.findMany({ + where: [{ column: relation.remoteKey, op: 'in', value: chunk }, ...scope], + limit: MAX_IDS_PER_STATEMENT, + cursor, + }); + rows.push(...page.rows); + cursor = page.nextCursor; + } while (cursor !== null); + } + return rows; +}; + +/** Related rows filed under the key they attach to — in the order the read returned them. */ +const indexed = async ( + target: RelatedTable, + relation: Relation, + kind: string, + values: readonly unknown[], + scope: readonly Predicate[], +): Promise> => { + const index = new Map(); + if (values.length === 0) return index; + const found = await relatedRows(target, relation, values, scope); + for (const row of found) { + const at = keyOf(kind, valueAt(row, relation.remoteKey)); + const bucket = index.get(at); + if (bucket === undefined) index.set(at, [row]); + else bucket.push(row); + } + return index; +}; + +/** The distinct keys a page carried, spelled as the batch spells them. A null key is not a key. */ +const distinctKeys = (kind: string, keys: readonly unknown[]): readonly unknown[] => { + const seen = new Set(); + const values: unknown[] = []; + for (const key of keys) { + if (key === null || key === undefined) continue; + const at = keyOf(kind, key); + if (seen.has(at)) continue; + seen.add(at); + values.push(key); + } + return values; +}; + +export interface PreloadRead { + readonly entity: EntityCore; + /** `undefined` for a table built by hand — `tableFor(entity, repo)` reaches no other table. */ + readonly related: RelatedTables | undefined; + readonly relations: readonly Relation[]; + /** The page's own predicates: what the related read inherits its tenant scope from. */ + readonly where: readonly Predicate[]; +} + +/** + * The page, with every named relation attached to it. `source` carries the key values (a + * projection may have dropped them from the rows the caller sees) and `rows` are the rows the + * caller gets, in the same order — the attachment is by position, so the two are always one page. + * + * A relation resolves to a row or `null` when it is a `belongsTo` and to an array when it is a + * `hasMany`, always present: "this post has no author" and "nobody preloaded the author" must not + * read the same at the call site. Relations resolve concurrently — two `preload()` calls are two + * statements in flight, never one after the other — and attach in the order they were named. + */ +export const preloaded = async ( + read: PreloadRead, + source: readonly Source[], + rows: readonly Row[], +): Promise => { + if (read.relations.length === 0 || rows.length === 0) return rows; + const resolved = await Promise.all( + read.relations.map(async (relation) => { + const target = read.related?.(relation.to); + if (target === undefined) throw unreachable(read.entity.$name, relation); + // The declaring column's own kind: a foreign key mirrors the key it points at, so both ends + // of the edge are spelled the way a batched point read already spells them. + const kind = read.entity.$columns[relation.localKey]?.$meta.kind ?? ''; + const keys = source.map((row) => valueAt(row, relation.localKey)); + const scope = tenantScope(read.entity, target.entity, read.where); + const index = await indexed(target, relation, kind, distinctKeys(kind, keys), scope); + return { relation, kind, keys, index }; + }), + ); + // A copy per row: the in-memory driver hands back the row it stores, and attaching to that one + // would write a relation into the table itself. + const attached = rows.map((row) => ({ ...row }) as Record); + for (const { relation, kind, keys, index } of resolved) { + for (const [position, row] of attached.entries()) { + const key = keys[position]; + const found = key === null || key === undefined ? undefined : index.get(keyOf(kind, key)); + row[relation.name] = relation.kind === 'hasMany' ? (found ?? []) : (found?.[0] ?? null); + } + } + // Built from the caller's own rows, one property added per relation named — which is exactly + // what `Row & { [name]: unknown }` says at the call site. + return attached as readonly Row[]; +}; diff --git a/packages/entity/src/query.ts b/packages/entity/src/query.ts index d7b90a191..e1c251f5e 100644 --- a/packages/entity/src/query.ts +++ b/packages/entity/src/query.ts @@ -5,10 +5,21 @@ import type { EntityCore } from './entity'; import { namedColumns } from './plan'; +import type { RelatedTables } from './preload'; +import { preloaded } from './preload'; +import type { Relation } from './relations'; +import { relationNamed } from './relations'; import type { Page, Repo, RepoOptions } from './repo'; import type { Operator, Predicate, QueryPlan, SortDirection, SortKey } from './tenancy'; import type { ColumnMap, IdOf, Insertable } from './types'; +/** + * What a preloaded relation adds to a row. `unknown` because the name is a string resolved at + * runtime against the relation map: the row on the other side is parsed by its own entity, never + * asserted into shape here. + */ +export type Preloaded = { readonly [K in Name]: unknown }; + export interface ReadBuilder { /** Equality on the columns given. `where({ orgId })` is what satisfies the tenancy guard. */ where(filter: Partial): ReadBuilder; @@ -20,6 +31,18 @@ export interface ReadBuilder { select( fields: { readonly [P in K]: true }, ): ReadBuilder>; + /** + * One relation, read for the whole page in one extra `where in (…)` and attached to every + * row under its own name — the eager form of the batching a point lookup does for itself, and + * the line an N+1 warning names. The relation is the `references()` already declared, so there + * is nothing to declare here; a name no foreign key produces is `X_PRELOAD_UNKNOWN_RELATION`, + * listing the ones that exist. + * + * A `belongsTo` attaches the row or `null`, a `hasMany` an array — always present, so "no + * author" never reads like "nobody preloaded the author". Attached after the projection: a + * `select()` narrows the columns, never the relations. + */ + preload(relation: Name): ReadBuilder>; /** The terminal: one bounded page and the cursor that continues it. */ page(): Promise>; all(): Promise; @@ -55,9 +78,18 @@ interface State { readonly limit: number; readonly cursor: string | null; readonly select: readonly string[] | undefined; + /** Resolved when `preload()` was called, so an unknown name fails at the chain and not a page later. */ + readonly preload: readonly Relation[]; } -const EMPTY: State = { where: [], orderBy: [], limit: 50, cursor: null, select: undefined }; +const EMPTY: State = { + where: [], + orderBy: [], + limit: 50, + cursor: null, + select: undefined, + preload: [], +}; const asRecord = (value: unknown): Readonly> => typeof value === 'object' && value !== null ? (value as Record) : {}; @@ -67,18 +99,37 @@ const builder = ( repo: Repo, state: State, pick: (row: Source) => Row, + related?: RelatedTables, ): ReadBuilder => { const next = (patch: Partial): ReadBuilder => - builder(entity, repo, { ...state, ...patch }, pick); + builder(entity, repo, { ...state, ...patch }, pick, related); + + /** + * A projection that drops a key the page is about to be preloaded on would send the statement + * looking for a column it did not select. The framework asks for what it needs; `pick` still + * hands the caller only the columns they named. + */ + const selected: readonly string[] | undefined = + state.select === undefined + ? undefined + : [...new Set([...state.select, ...state.preload.map((relation) => relation.localKey)])]; const args = () => ({ where: state.where, orderBy: state.orderBy, limit: state.limit, cursor: state.cursor, - ...(state.select === undefined ? {} : { select: state.select }), + ...(selected === undefined ? {} : { select: selected }), }); + /** The page the caller gets: projected, then every named relation attached to it by position. */ + const attach = (rows: readonly Source[]): Promise => + preloaded( + { entity, related, relations: state.preload, where: state.where }, + rows, + rows.map(pick), + ); + return { where: (filter) => next({ @@ -102,25 +153,47 @@ const builder = ( select(fields: { readonly [P in K]: true }) { // The predicate is what carries the literal key type through `Object.keys`. const keys = Object.keys(fields).filter((key): key is K => Object.hasOwn(fields, key)); - return builder>(entity, repo, { ...state, select: keys }, (row) => { - const source = pick(row); - const picked = {} as Pick; - for (const key of keys) picked[key] = source[key]; - return picked; - }); + return builder>( + entity, + repo, + { ...state, select: keys }, + (row) => { + const source = pick(row); + const picked = {} as Pick; + for (const key of keys) picked[key] = source[key]; + return picked; + }, + related, + ); + }, + + preload(relation: Name) { + // Resolved here, so a name no foreign key produces fails on the chain rather than one page + // later — and naming one relation twice is one statement, not two identical ones. + const resolved = relationNamed(entity.$name, relation); + const already = state.preload.some((held) => held.name === resolved.name); + return builder>( + entity, + repo, + { ...state, preload: already ? state.preload : [...state.preload, resolved] }, + // The relation is attached after the projection, so `pick` is unchanged and the row type + // is the only thing that grows — one cast, where a runtime name becomes a static one. + pick as (row: Source) => Row & Preloaded, + related, + ); }, page: async () => { const result = await repo.findMany(args()); - return { rows: result.rows.map(pick), nextCursor: result.nextCursor }; + return { rows: await attach(result.rows), nextCursor: result.nextCursor }; }, - all: async () => (await repo.findMany(args())).rows.map(pick), + all: async () => attach((await repo.findMany(args())).rows), one: async () => { const { rows } = await repo.findMany({ ...args(), limit: 1 }); const row = rows[0]; - return row === undefined ? null : pick(row); + return row === undefined ? null : ((await attach([row]))[0] ?? null); }, count: () => repo.count(args()), @@ -131,7 +204,9 @@ const builder = ( orderBy: state.orderBy, limit: state.limit, ...(state.cursor === null ? {} : { cursor: state.cursor }), - ...(state.select === undefined ? {} : { select: state.select }), + // The projection actually sent, preload keys included: a plan that is safe to log is only + // useful if it is the plan that ran. + ...(selected === undefined ? {} : { select: selected }), }), }; }; @@ -161,8 +236,10 @@ const touch = (entity: EntityCore, patch: Partial): Partial export const tableFor = ( entity: EntityCore, repo: Repo, + /** How this table reaches another — `database()` passes it; a table built by hand has none. */ + related?: RelatedTables, ): Table => ({ - ...builder(entity, repo, EMPTY, (row) => row), + ...builder(entity, repo, EMPTY, (row) => row, related), insert: async (values, options) => repo.insert(entity.$parse(values), options), update: async (id, patch, options) => repo.update(id, touch(entity, patch), options), delete: async (id, options) => repo.delete(id, options), diff --git a/wiki/Entities-And-Migrations.md b/wiki/Entities-And-Migrations.md index 5f15bc7f1..cac31caa5 100644 --- a/wiki/Entities-And-Migrations.md +++ b/wiki/Entities-And-Migrations.md @@ -129,6 +129,88 @@ Removing the key is a real option, not a hedge: inference (steps 2–4) still ap | Live queries | the tenant predicate is part of the matcher, not a post-filter | | Vector search | tenant + policy filters applied **in SQL**, so similarity search cannot leak across tenants | +## Point lookups batch themselves + +`findById` called several times in one microtask of one request is **one** statement, `As of 2026-08`: + +```ts +// One `select … where "id" in ($1, $2, $3, …)`, not one statement per post. +const authors = await Promise.all(posts.map((post) => users.findById(post.authorId, { orgId }))); +``` + +Nothing to opt into — no `dataloader()`, no `batch()`. `findById` keeps its signature and its meaning, which is the only way the fix reaches code that is already written. + +| Property | Behavior | +|---|---| +| Window | one microtask, closed before the statement goes out. A sequential `for … of` loop shares no microtask — its `await` ends the window — so what batches one is the page it is looping over (below) | +| Lifetime | one request. The batch is keyed by context identity, so it dies with the request and never crosses one | +| Never shared with | another tenant, another soft-delete visibility, another projection, another entity, another client. The coalesced statement has to be one each single lookup would have been served by | +| Scope | the tenant predicate and `deleted_at is null` are **inside** the statement, so an id that is missing, soft-deleted or another tenant's still reads as `null` — never another caller's row | +| Wide batches | past 500 ids, several whole statements rather than one Postgres refuses for its bind count | +| Outside a request | a job, a script, a migration: the single statement it always was | + +## A page batches the loop it causes + +A `for … of` loop awaits between iterations, so no two of its lookups share a microtask. The page +they came from batches them instead, `As of 2026-08`: + +```ts +const page = await db.posts.findMany({ orgId }); +for (const post of page.rows) { + // Two statements for the whole loop: the page, then one `where "id" in (…)` over every author + // on it. Every lookup after the first is memory. + const author = await db.users.findById(post.authorId, { orgId }); +} +``` + +Nothing new to write, and nothing to opt into: the relation is the `references()` the column +already declares, so the fix reaches loops that are already written. + +| Property | Behavior | +|---|---| +| Trigger | the first `findById` whose id is a foreign key value on a page this request read. It resolves that key for **every** row of the page, in one statement | +| Never shared with | a different scope key (another tenant, another soft-delete visibility, another projection, another entity), another client, or the same entity after a write to it | +| Scope | the tenant predicate and `deleted_at is null` are inside the preload statement — it is the statement each single lookup would have sent, widened to the page's ids — so a page's ids can never resolve rows outside the reader's own scope | +| A write | drops what was preloaded for that entity before the statement goes out, so a row this request changed is re-read, never served from a page read before it | +| Lifetime | one request. What is held is the ids, not the rows, keyed by context identity | +| Wide pages | past 500 ids, whole statements — the same bound the microtask batch has | +| Outside a request | a job, a script: a page leaves nothing behind, and every lookup is the statement it always was | + +## Preload states a relation the loop would infer + +The eager, declarative third of the family: the relation is named on the chain and resolved +before the caller sees a row, `As of 2026-08`: + +```ts +export const db = database({ orgs, posts, members }); + +// Two statements: the page, then one `select … where "id" in (…)` over its authors. +const page = await db.posts.where({ orgId }).preload('author').page(); +page.rows[0].author; // the member row, or null — always present +``` + +Nothing new to declare: `'author'` is the relation the `authorId` foreign key on `posts` already +produces — `preload()` names it, it does not define it. A name no foreign key produces is +`X_PRELOAD_UNKNOWN_RELATION`, resolved and thrown when `preload()` is called, on the chain and not +a page later. + +| Property | Behavior | +|---|---| +| Shape | a `belongsTo` attaches the row or `null`; a `hasMany` attaches an array, empty when there are none. Always present, so "no author" and "nobody preloaded the author" cannot read the same | +| Statements | one extra statement per named relation, resolved **concurrently** — two `preload()` calls are two statements in flight, never one after the other. Naming one relation twice is one statement | +| Terminals | `page()`, `all()`, `one()` resolve every named relation. `count()` and `plan()` do not — a count reads no rows | +| Projection | attached after `select()`. `select()` is widened internally with the relation's own key, so a projection can drop neither the key the preload reads nor the relation it attaches — `plan().select` reports the widened list, the one that actually ran | +| Tenancy | the page's own tenant predicate carries onto the related read only when **both** entities are scoped by a column of that same name — a value that scopes one entity is a guess on another, and a source scoped by `workspaceId` may carry an ordinary `orgId` predicate that is a filter and not its tenancy. Carrying nothing is not silence: the related read builds its own plan, so it is refused as `X_TENANCY_UNSCOPED` rather than answered with a guess | +| Soft delete | the related read is `findMany`, so `deleted_at is null` applies exactly as it does to any other read | +| Wide relations | past 500 ids, whole statements — the same bound the two batches above share. A `hasMany` wider than one page costs another keyset page rather than a silent truncation | +| Reach | a table reads only the entities its own `database()` call named. A relation to one outside the set is `X_INVARIANT_VIOLATED`; the fix names the missing entity in that same `database({ … })` call | + +Reach for it when the relation is part of what the page *is* — a list rendered with its authors, +rows handed to something that will not call back into the repo, or a read a reviewer should see +stated rather than inferred from a loop. The other two members of the family ask for nothing: +`findById` batches a same-microtask fan-out for itself, and a `for … of` loop over a page is +already two statements, above. + ## Invariants | Rule | Behavior | @@ -223,5 +305,6 @@ The same clone mechanism powers test parallelism: `bun test --workers 8` gives e | `X_ENTITY_DUPLICATE` | two entities on the same table | rename one, or merge them | | `X_INVARIANT_VIOLATED` | a write broke a named invariant | fix the caller, or change the invariant and generate a migration | | `X_TENANCY_UNSCOPED` | a query without a tenant predicate | go through the repo | +| `X_PRELOAD_UNKNOWN_RELATION` | `preload('')` named a relation no `references()` produces | pick one of the relation names the error lists, or add the `.references()` call that creates it | Full list with `--json` shapes: [Error codes](Error-Codes). Source: [`docs/idea/02-primitives.md`](https://github.com/developerz-ai/ultimate/blob/main/docs/idea/02-primitives.md), [`docs/idea/10-testing.md`](https://github.com/developerz-ai/ultimate/blob/main/docs/idea/10-testing.md). diff --git a/wiki/Queries-And-Live-Queries.md b/wiki/Queries-And-Live-Queries.md index 2615e8e1d..4a484d26d 100644 --- a/wiki/Queries-And-Live-Queries.md +++ b/wiki/Queries-And-Live-Queries.md @@ -24,7 +24,7 @@ export const liveFeed = query({ | `policy` | yes | `can('')`, optionally with a predicate over `{ input, actor }`. Evaluated at HTTP call, client hook, subscribe, **and per delivered row** | | `live` | no — default `false` | registers the query with the incremental matcher. Requires a deterministic, bounded `sql` | | `persist` | no — default `false` | tier 3. Swaps the client result store from memory to IndexedDB and makes the mutator queue durable. Implies `live: true`. v2 | -| `sql` | yes | `(input) => SqlSource`. Built with `from()` from `@ultimat3/query` or an `@ultimat3/entity` repo plan — no ORM in the graph. SQL-transparent: `toSQL()` prints the statement verbatim so an agent can read it and self-correct | +| `sql` | yes | `(input) => SqlSource`. `from()` (`@ultimat3/query`) wraps an already-resolved `@ultimat3/entity` repo call and restates `where`/`orderBy`/`limit` for the matcher to read back; `select`/`preload` happen inside that repo call, before `from()` ever sees a row. No ORM in the graph. SQL-transparent: `toSQL()` prints the statement verbatim so an agent can read it and self-correct | | `mcp` | no — default not exposed | `{ expose: true, description }` makes the read an MCP tool. Opt-in, unlike an action: a read hands rows to an agent, so silence exposes nothing | | `mcp.visibleTo` | no | roles that may see the projected tool; a caller whose role is not named gets ToolNotFound, never Forbidden — the policy still decides every call | | cache tags | derived | acquired automatically from the tables `sql` touches. Never hand-declared on a query | @@ -142,6 +142,8 @@ Three components on one page calling `liveFeed({ orgId })` resolve **one** query Streamed `` holes ([Routes and render modes](Routes-And-Render-Modes)) are the common case: independent holes, one round trip to Postgres. The same memo is what keeps an uncached lookup called once per row of a list to one round trip — `As of 2026-08`, per row is what it used to cost. +The memo collapses the *same* read asked twice. Fifty *different* row lookups collapse one layer down, in the repo: `findById` issued across one microtask of a request is one `where "id" in (…)` ([Entities and migrations → Point lookups batch themselves](Entities-And-Migrations#point-lookups-batch-themselves)) — or, named on the chain instead of inferred from a loop, one `preload()` per relation ([Entities and migrations → Preload states a relation the loop would infer](Entities-And-Migrations#preload-states-a-relation-the-loop-would-infer)). + ## Every cached query carries a tag The contract. **Not yet a gate** — `As of 2026-08` `X_CACHE_UNTAGGED_QUERY` is reserved: no code path raises it, and `x errors explain X_CACHE_UNTAGGED_QUERY` refuses it ([Error codes → Reserved codes](Error-Codes#reserved-codes)).