feat(search-pipeline)!: project inside the batch, per root type (ADR 13)

packages/search-api-graphql/README.md

@@ -3,21 +3,21 @@
 The GraphQL surface for the [`@lde/search`](../search) core. **Both engine- and
 domain-agnostic:** it builds an executable
 [graphql-js](https://graphql.org/graphql-js/) `GraphQLSchema` from your whole
-[`SearchSchema`](../search/README.md#terminology) at runtime — one root query
+[`SearchSchema`](../search/README.md#terminology) at runtime – one root query
 field per `SearchType`, each searchable in its own way. All root fields are
 served by the same resolver implementation (no per-type code, no codegen);
 each root field gets its own instance of it, bound to that field’s
 `SearchType`, over any `SearchEngine`. It names neither your **domain** (each type’s GraphQL name
-is the `SearchType`’s own logical `name` — `Dataset`, `Person`, `CreativeWork`,
+is the `SearchType`’s own logical `name` – `Dataset`, `Person`, `CreativeWork`,
 …) nor your **engine** (the resolver calls the schema-bound `context.engine`, be it
 [`@lde/search-typesense`](../search-typesense) or another adapter).
 
 ## Runtime configuration, not codegen
 
 `buildGraphQLSchema(schema)` constructs the GraphQL schema once at startup from
-the field model — no SDL artifact, no generated resolver stubs. For you that
+the field model – no SDL artifact, no generated resolver stubs. For you that
 means: no codegen step in the build, no generated files to commit and review,
-and no stale artifact that can drift from the declaration — change the
+and no stale artifact that can drift from the declaration – change the
 `SearchType`, restart, and the API is current. (The flip side, no artifact
 showing contract changes as diffs, is restored by the
 [snapshot guard](#guarding-the-contract).) The field model
@@ -63,14 +63,17 @@ types such as a common `Agent`) are created once and reused across root types.
 
 `types` never filters: every `SearchType` in the schema you pass gets a root
 field (options for a type not in the schema are a build-time error). To expose
-only part of what you index, narrow the **schema argument**
-(`searchSchema(…)` is a cheap constructor):
+only part of what you index, narrow the **schema argument** you hand
+`buildGraphQLSchema` (`searchSchema(…)` is a cheap constructor, so build one per
+consumer):
 
 ```ts
-// Index all three types…
-projectGraph(quads, searchSchema(DATASET, PERSON, INTERNAL));
+// Index a superset: hand a three-type schema to the pipeline, which projects and
+// stores one collection per type (see @lde/search-pipeline). INTERNAL is indexed
+// (e.g. a label source references resolve against) but never served.
+const indexed = searchSchema(DATASET, PERSON, INTERNAL);
 
-// …but serve only two.
+// Serve a subset: the GraphQL API exposes only two of those types.
 const gqlSchema = buildGraphQLSchema(searchSchema(DATASET, PERSON));
 ```
 
@@ -98,7 +101,7 @@ const gqlSchema = buildGraphQLSchema(searchSchema(DATASET, PERSON));
 ## Guarding the contract
 
 Why the API, the index and a future REST surface cannot drift apart is the
-search family’s overall approach — one field model, one query IR — described
+search family’s overall approach – one field model, one query IR – described
 in [`@lde/search`](../search/README.md). Specific to this surface: the GraphQL
 contract is **frozen** (breaking to change), yet generated rather than
 handwritten, so nothing in the repo shows a contract change as a reviewable
@@ -114,8 +117,8 @@ it('keeps the public GraphQL contract stable', () => {
 ```
 
 The first run writes the emitted SDL to a committed snapshot file; every later
-run re-emits and diffs against it. Any contract change — your own schema edit,
+run re-emits and diffs against it. Any contract change – your own schema edit,
 or a new version of this library emitting different GraphQL for the same
-declaration — fails the test and shows the SDL diff, until you consciously
+declaration – fails the test and shows the SDL diff, until you consciously
 accept it (`vitest -u`) and the reviewer sees the contract change spelled out
 in the PR.

packages/search-pipeline/README.md

@@ -6,26 +6,41 @@ and a search engine’s document world, so indexing reuses the existing spine 
 Discovery → Selection → Iteration → change-gate (skip-unchanged) – instead of
 rebuilding it per consumer.
 
-The division of labour ([ADR 6](../../docs/decisions/0006-make-the-writer-transaction-aware.md)):
+The division of labour ([ADR 6](../../docs/decisions/0006-make-the-writer-transaction-aware.md),
+[ADR 13](../../docs/decisions/0013-project-inside-the-batch-per-root-type.md)):
 
 - [`@lde/search`](../search) owns the engine-agnostic projection (framed
   document + field model) and stays pipeline-free;
 - `@lde/search-<engine>` (e.g. [`@lde/search-typesense`](../search-typesense))
   is the engine adapter: a transactional, **single-collection** `Writer`
   consuming projected documents, owning the run lifecycle (Blue/green alias swap
   or In-place sweeps, cross-pod lock);
-- **this package** composes them: `searchIndexWriter` is the one type-changing
-  step (quad → document), shared across engines, and – since one schema declares
-  several root types – it fans each document out to the engine writer for **its**
-  type’s collection ([ADR 9](../../docs/decisions/0009-route-a-whole-schema-projection-to-per-type-collections.md)).
+- **this package** composes them:
+  - `searchStages` builds one projecting `Stage` per root type. Each stage
+    selects its own roots, extracts each root’s quads, and projects the
+    **root-complete batch** into documents paired with their `SearchType`
+    (`TypedSearchDocument`) – so projection happens inside the batch and memory
+    is bounded by `batchSize` roots, not the dataset
+    ([ADR 13](../../docs/decisions/0013-project-inside-the-batch-per-root-type.md));
+  - `searchIndexWriter` is the pipeline’s single terminal: an engine-agnostic
+    router that dispatches each document to the engine writer for **its**
+    type’s collection ([ADR 9](../../docs/decisions/0009-route-a-whole-schema-projection-to-per-type-collections.md)).
+    It owns no projection and buffers nothing.
+
+So a search pipeline is **one terminal, N stages**: `new Pipeline<TypedSearchDocument>({ datasetSelector, stages: searchStages(...), writers: searchIndexWriter(...) })`.
 
 ## Usage
 
 ```typescript
-import { Pipeline, Stage, SparqlConstructReader } from '@lde/pipeline';
+import { Pipeline, SparqlConstructReader } from '@lde/pipeline';
 import { searchSchema } from '@lde/search';
 import { BlueGreenRebuild } from '@lde/search-typesense';
-import { searchIndexWriter } from '@lde/search-pipeline';
+import {
+  searchStages,
+  selectByClass,
+  searchIndexWriter,
+  type TypedSearchDocument,
+} from '@lde/search-pipeline';
 
 // One schema, several root types: the `datasets` catalog plus the Organization
 // label collection its references resolve against.
@@ -60,14 +75,33 @@ const schema = searchSchema(
   },
 );
 
-const pipeline = new Pipeline({
+const dataset = schema.get('http://www.w3.org/ns/dcat#Dataset')!;
+const organization = schema.get('http://xmlns.com/foaf/0.1/Organization')!;
+
+const pipeline = new Pipeline<TypedSearchDocument>({
   datasetSelector,
-  stages: [
-    new Stage({
-      name: 'extract',
-      readers: new SparqlConstructReader({ query: '…' }),
-    }),
-  ],
+  // One stage per root type: each selects its own roots (here by class –
+  // `selectByClass` is a convenience for the object grain; a deployment writes
+  // its own selector where the entry point is a domain fact) and extracts each
+  // root’s quads. `rootVariable` couples the selector’s projected variable to
+  // what the projection reads; it must not be `dataset` (reserved by the reader).
+  stages: searchStages({
+    schema,
+    types: [
+      {
+        searchType: dataset,
+        rootVariable: 'root',
+        itemSelector: selectByClass(dataset),
+        readers: new SparqlConstructReader({ query: '…' }),
+      },
+      {
+        searchType: organization,
+        rootVariable: 'root',
+        itemSelector: selectByClass(organization),
+        readers: new SparqlConstructReader({ query: '…' }),
+      },
+    ],
+  }),
   // Each type gets its own collection – an independent blue/green rebuild –
   // named by the adapter from the type itself (`Dataset` → `datasets`,
   // `Organization` → `organizations`), so no naming map is passed here and the
@@ -101,14 +135,21 @@ Whatever the writers are named, the engine reading these collections must be
 given the same names (`collections`), or it reads the derived ones and finds an
 empty index.
 
-Each dataset’s extracted CONSTRUCT quads are buffered until the dataset
-completes, then projected once over the **whole** schema (`@lde/search`’s
-`projectGraph`) into one mixed, type-tagged stream, and each document is
-dispatched to the engine run for its type’s collection – before the engine acts
-on the dataset’s completion, so an In-place stale sweep never races its own
-documents. The run lifecycle (run context, per-dataset flush outcome,
-commit/abort) passes through unchanged: the engine writer’s update mode governs,
-and the pipeline never branches on it, nor on how many collections there are.
+Each stage projects its own root-complete batches (`@lde/search`’s
+`projectRoots`) into documents paired with its `SearchType`, and the terminal
+dispatches each to the engine run for its type’s collection as it arrives –
+before the engine acts on the dataset’s completion, so an In-place stale sweep
+never races its own documents. The run lifecycle (run context, per-dataset flush
+outcome, commit/abort) passes through unchanged: the engine writer’s update mode
+governs, and the pipeline never branches on it, nor on how many collections
+there are.
+
+**Root selection is the deployment’s contract.** A selector that does not bind
+the CONSTRUCT’s subject yields a batch that is not root-complete, and projection
+will then silently emit partial documents – a promise `@lde/pipeline` cannot
+check ([ADR 13](../../docs/decisions/0013-project-inside-the-batch-per-root-type.md)).
+`selectByClass(searchType)` is a convenience for the object grain (where the
+type’s `class` really is the source class), **not** a default.
 
 ## Per-collection isolation
 
@@ -130,7 +171,12 @@ See [ADR 9](../../docs/decisions/0009-route-a-whole-schema-projection-to-per-typ
 
 ## Memory
 
-Memory is bounded by one dataset’s extraction and released at each dataset
-flush – nothing accumulates across datasets. Streaming bounded entity batches
-_within_ one huge dataset needs the two-level iteration (dataset →
-entity-URI batches) and is not implemented yet.
+Memory is bounded by **`batchSize` roots**, not by the input
+([ADR 12](../../docs/decisions/0012-bound-memory-by-the-unit-of-work-not-the-input.md),
+[ADR 13](../../docs/decisions/0013-project-inside-the-batch-per-root-type.md)).
+Projection runs inside each root-complete batch and streams straight through the
+terminal to the engine run, which holds only its own batch – nothing accumulates
+across a dataset or across datasets, so the same pipeline indexes a 15-quad
+catalog entry and a dataset of millions of objects. The one irreducible atom is a
+single root’s quads: unbounded for a pathological root, and nothing here changes
+that.

packages/search-pipeline/package.json

@@ -26,11 +26,11 @@
   ],
   "dependencies": {
     "@lde/search": "^0.8.1",
-    "@rdfjs/types": "^2.0.1",
     "tslib": "^2.3.0"
   },
   "devDependencies": {
     "@lde/search-typesense": "^0.10.2",
+    "@rdfjs/types": "^2.0.1",
     "n3": "^2.1.1",
     "testcontainers": "^12.0.3",
     "typesense": "^3.0.6"

packages/search-pipeline/src/index.ts

@@ -1,2 +1,5 @@
+export { searchStages, selectByClass } from './search-stages.js';
+export type { SearchStagesOptions, SearchStageType } from './search-stages.js';
 export { searchIndexWriter } from './search-index-writer.js';
 export type { SearchIndexWriterOptions } from './search-index-writer.js';
+export type { TypedSearchDocument } from './typed-search-document.js';