ldelements · ddeboer · Jul 23, 2026
diff --git a/AGENTS.md b/AGENTS.md
@@ -82,7 +82,7 @@ Each package uses conditional exports with a `development` condition for local d
 - Test files use `.test.ts` suffix in `test/` directory
 - Fixtures in `test/fixtures/`
 - HTTP mocking with Nock
-- Tests that start a local SPARQL endpoint (`@lde/local-sparql-endpoint`) must use unique ports across packages to avoid conflicts when Nx runs tests in parallel. Current port allocations: `dataset-registry-client` (3002), `pipeline` sparqlQuery (3001), `pipeline` executor (3003), `pipeline` provenance store (3004), `pipeline-void` namespace-normalization (3005–3006), `search-pipeline` extraction round-trip (3007)
+- Tests that start a local SPARQL endpoint (`@lde/local-sparql-endpoint`) must use unique ports across packages to avoid conflicts when Nx runs tests in parallel. Current port allocations: `dataset-registry-client` (3002), `pipeline` sparqlQuery (3001), `pipeline` executor (3003), `pipeline` provenance store (3004), `pipeline-void` namespace-normalization (3005–3006), `search-pipeline` extraction round-trip (3007), `search-pipeline` searchIndexerPipeline end-to-end (3008)
 
 ### Key Dependencies
 

diff --git a/packages/search-pipeline/README.md b/packages/search-pipeline/README.md
@@ -34,6 +34,62 @@ The division of labour ([ADR 6](../../docs/decisions/0006-make-the-writer-transa
 
 So a search pipeline is **one terminal, N stages**: `new Pipeline<TypedSearchDocument>({ datasetSelector, stages: searchStages(...), writers: searchIndexWriter(...) })`.
 
+## Quick start: `searchIndexerPipeline`
+
+For the common **object grain** – index every IRI-identified instance of each
+root type’s source class – `searchIndexerPipeline` wires that whole composition
+and returns the ready-to-run `Pipeline`. It generates one stage per root type
+in the schema, each selecting the type’s non-blank roots by class
+(`selectByClass`; a blank node has no stable document key, so it can never
+become a document), extracting with the schema-generated CONSTRUCT, and routing
+every document to the engine writer for its type’s collection. The consumer
+supplies only the domain (the schema, which datasets) and the deployment shell
+(the engine writer, the SPARQL/import adapter):
+
+```typescript
+import { searchIndexerPipeline } from '@lde/search-pipeline';
+import {
+  FileProvenanceStore,
+  ImportResolver,
+  SparqlDistributionResolver,
+} from '@lde/pipeline';
+import { ConsoleReporter } from '@lde/pipeline-console-reporter';
+import { createQlever } from '@lde/sparql-qlever';
+import { InPlaceRebuild } from '@lde/search-typesense';
+
+const pipeline = searchIndexerPipeline({
+  schema,
+  datasets,
+  // Import each dataset’s data dump into a local QLever and query that.
+  distributionResolver: new ImportResolver(new SparqlDistributionResolver(), {
+    ...createQlever({ mode: 'docker', image: 'adfreiburg/qlever:latest' }),
+    strategy: 'import',
+  }),
+  writerFor: (searchType) => new InPlaceRebuild(typesenseClient, searchType),
+  // Optional: skip datasets whose source and pipeline version are unchanged.
+  provenanceStore: new FileProvenanceStore({ path: 'data/provenance.json' }),
+  pipelineVersion: '1',
+  reporter: new ConsoleReporter(),
+});
+await pipeline.run();
+```
+
+Each role in that wiring maps to a package:
+
+| Role                                   | Package                                                                                                                                                                                                                |
+| -------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| Dataset selection                      | [`@lde/pipeline`](../pipeline) – `Dataset[]` directly, or a `DatasetSelector` such as `RegistrySelector`                                                                                                               |
+| Distribution resolution and import     | [`@lde/pipeline`](../pipeline) `ImportResolver` + [`@lde/sparql-qlever`](../sparql-qlever) `createQlever` (the concrete [`@lde/sparql-server`](../sparql-server)/[`@lde/sparql-importer`](../sparql-importer) adapter) |
+| Root selection, extraction, projection | this package (`searchStages` over [`@lde/search`](../search))                                                                                                                                                          |
+| Engine writer, one collection per type | [`@lde/search-typesense`](../search-typesense) – `InPlaceRebuild` or `BlueGreenRebuild`                                                                                                                                |
+| Provenance (skip-unchanged)            | [`@lde/pipeline`](../pipeline) – `FileProvenanceStore`, or a triplestore-backed store                                                                                                                                  |
+| Reporting                              | [`@lde/pipeline-console-reporter`](../pipeline-console-reporter) `ConsoleReporter`                                                                                                                                     |
+
+A deployment that needs more – a bespoke root selector (where the entry point
+is a domain fact, not a class), per-stage tuning, non-SPARQL readers, or
+quad-level plugins – composes the parts directly, as below; the convenience
+owns no capability of its own.
+
 ## Usage
 
 ```typescript

diff --git a/packages/search-pipeline/src/index.ts b/packages/search-pipeline/src/index.ts
@@ -1,3 +1,5 @@
+export { searchIndexerPipeline } from './search-indexer-pipeline.js';
+export type { SearchIndexerPipelineOptions } from './search-indexer-pipeline.js';
 export { searchStages, selectByClass } from './search-stages.js';
 export type { SearchStagesOptions, SearchStageType } from './search-stages.js';
 export { extractionQuery, extractionQueryString } from './extraction.js';

diff --git a/packages/search-pipeline/src/search-indexer-pipeline.ts b/packages/search-pipeline/src/search-indexer-pipeline.ts
@@ -0,0 +1,119 @@
+import type { Dataset } from '@lde/dataset';
+import {
+  ManualDatasetSelection,
+  Pipeline,
+  type DatasetSelector,
+  type DistributionResolver,
+  type ProgressReporter,
+  type ProvenanceStore,
+  type Writer,
+} from '@lde/pipeline';
+import type { RootType, SearchDocument, SearchSchema } from '@lde/search';
+import { searchIndexWriter } from './search-index-writer.js';
+import { searchStages, selectByClass } from './search-stages.js';
+import type { TypedSearchDocument } from './typed-search-document.js';
+
+/** Options for {@link searchIndexerPipeline}. */
+export interface SearchIndexerPipelineOptions {
+  /**
+   * The declarative schema driving the indexer: one stage and one engine
+   * collection per {@link RootType} in it. Reference Types are absent from
+   * `schema.values()`, so none ever earns a stage or a collection.
+   */
+  schema: SearchSchema;
+  /**
+   * Which datasets to index – the deployment’s domain fact. Pass the
+   * {@link Dataset}s directly, or a {@link DatasetSelector} (e.g. a
+   * `RegistrySelector`) when selection is dynamic.
+   */
+  datasets: Dataset[] | DatasetSelector;
+  /**
+   * How each dataset’s data becomes queryable – the deployment’s engine
+   * choice. Typically an `ImportResolver` wrapping a
+   * `SparqlDistributionResolver` with a `@lde/sparql-qlever` `createQlever`
+   * import path, so data-dump distributions are imported into a local
+   * endpoint. Defaults to the {@link Pipeline} default (a bare
+   * `SparqlDistributionResolver`), which serves only datasets that publish a
+   * live SPARQL endpoint.
+   */
+  distributionResolver?: DistributionResolver;
+  /**
+   * The engine writer that owns a given root type’s collection – e.g. a
+   * `@lde/search-typesense` `InPlaceRebuild` or `BlueGreenRebuild` bound to
+   * that type. Called once per {@link RootType} in the schema.
+   */
+  writerFor: (searchType: RootType) => Writer<SearchDocument>;
+  /**
+   * Optional per-dataset processing memory: skip a dataset whose source
+   * fingerprint and {@link pipelineVersion} both match the stored record.
+   * Requires {@link pipelineVersion}.
+   */
+  provenanceStore?: ProvenanceStore;
+  /**
+   * Opaque, consumer-declared version of the indexer’s output-affecting logic
+   * (schema, projection). Required when {@link provenanceStore} is set – a
+   * skip-enabled pipeline with no version would silently freeze.
+   */
+  pipelineVersion?: string;
+  /**
+   * Observer(s) of pipeline lifecycle events, e.g. a
+   * `@lde/pipeline-console-reporter` `ConsoleReporter`.
+   */
+  reporter?: ProgressReporter | readonly ProgressReporter[];
+}
+
+/**
+ * Wire the common **object-grain** search indexer and return the ready-to-run
+ * {@link Pipeline}: one projecting stage per {@link RootType} in the schema,
+ * each selecting the type’s roots by its source class with blank-node subjects
+ * excluded ({@link selectByClass} – a blank node has no stable document key),
+ * extracting with the schema-generated CONSTRUCT, and projecting inside the
+ * batch; plus the single {@link searchIndexWriter} terminal routing each
+ * document to the engine writer for its type’s collection.
+ *
+ * The consumer supplies only the domain and the deployment shell: the schema,
+ * which datasets, the engine writer per type, and (optionally) the
+ * SPARQL/import adapter, provenance and reporting.
+ *
+ * ```ts
+ * const pipeline = searchIndexerPipeline({
+ *   schema,
+ *   datasets,
+ *   distributionResolver: new ImportResolver(new SparqlDistributionResolver(), {
+ *     ...createQlever({ mode: 'docker', image: 'adfreiburg/qlever:latest' }),
+ *     strategy: 'import',
+ *   }),
+ *   writerFor: (searchType) => new InPlaceRebuild(typesenseClient, searchType),
+ * });
+ * await pipeline.run();
+ * ```
+ *
+ * A deployment that needs more – a bespoke root selector (the entry point is a
+ * domain fact, not a class), per-stage tuning (`batchSize`, `maxConcurrency`),
+ * non-SPARQL readers, or quad-level plugins – composes {@link searchStages},
+ * {@link searchIndexWriter} and {@link Pipeline} directly; this convenience
+ * owns no capability of its own.
+ */
+export function searchIndexerPipeline(
+  options: SearchIndexerPipelineOptions,
+): Pipeline<TypedSearchDocument> {
+  const { schema, datasets } = options;
+  return new Pipeline<TypedSearchDocument>({
+    datasetSelector: Array.isArray(datasets)
+      ? new ManualDatasetSelection(datasets)
+      : datasets,
+    distributionResolver: options.distributionResolver,
+    stages: searchStages({
+      schema,
+      types: [...schema.values()].map((searchType) => ({
+        searchType,
+        rootVariable: 'root',
+        itemSelector: selectByClass(searchType),
+      })),
+    }),
+    writers: searchIndexWriter({ schema, writerFor: options.writerFor }),
+    provenanceStore: options.provenanceStore,
+    pipelineVersion: options.pipelineVersion,
+    reporter: options.reporter,
+  });
+}
diff --git a/packages/search-pipeline/test/search-indexer-pipeline.integration.test.ts b/packages/search-pipeline/test/search-indexer-pipeline.integration.test.ts
@@ -0,0 +1,137 @@
+import { fileURLToPath } from 'node:url';
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import { Dataset, Distribution } from '@lde/dataset';
+import {
+  ProbedDistributions,
+  ResolvedDistribution,
+  type DistributionResolver,
+  type ProgressReporter,
+  type Writer,
+} from '@lde/pipeline';
+import {
+  defineSearchType,
+  searchSchema,
+  type RootType,
+  type SearchDocument,
+} from '@lde/search';
+import {
+  startSparqlEndpoint,
+  teardownSparqlEndpoint,
+} from '@lde/local-sparql-endpoint';
+import { searchIndexerPipeline } from '../src/search-indexer-pipeline.js';
+
+const SCHEMA = 'https://schema.org/';
+
+// The Drapo-shaped schema of the extraction round-trip test, both types root:
+// the convenience derives one stage and one collection per root type.
+const person = defineSearchType({
+  name: 'Person',
+  class: `${SCHEMA}Person`,
+  fields: [
+    {
+      name: 'label',
+      kind: 'text',
+      path: `<${SCHEMA}name>`,
+      locales: ['und'],
+      output: true,
+      searchable: { weight: 3 },
+    },
+  ],
+});
+
+const creativeWork = defineSearchType({
+  name: 'CreativeWork',
+  class: `${SCHEMA}CreativeWork`,
+  fields: [
+    {
+      name: 'name',
+      kind: 'text',
+      path: `<${SCHEMA}name>`,
+      locales: ['nl', 'en'],
+      output: true,
+      searchable: { weight: 5 },
+    },
+  ],
+});
+
+const schema = searchSchema(creativeWork, person);
+
+describe('searchIndexerPipeline end to end', () => {
+  const port = 3008;
+  const distribution = Distribution.sparql(
+    new URL(`http://localhost:${port}/sparql`),
+  );
+  const dataset = new Dataset({
+    iri: new URL('http://example.org/dataset/drapo'),
+    distributions: [distribution],
+  });
+
+  // Resolve straight to the local endpoint: probing and importing are the
+  // resolver’s own concern, tested with @lde/pipeline – here it is a seam.
+  const resolver: DistributionResolver = {
+    probe: async () => new ProbedDistributions(dataset, [], null),
+    resolve: async () => new ResolvedDistribution(distribution, []),
+  };
+
+  beforeAll(async () => {
+    const fixture = fileURLToPath(
+      new URL('./fixtures/drapo-sample.ttl', import.meta.url),
+    );
+    await startSparqlEndpoint(port, fixture);
+  }, 60_000);
+
+  afterAll(async () => {
+    await teardownSparqlEndpoint();
+  });
+
+  it('indexes every root type’s IRI roots into its own committed collection', async () => {
+    const received = new Map<string, SearchDocument[]>();
+    const committed: string[] = [];
+    const writerFor = (searchType: RootType): Writer<SearchDocument> => ({
+      openRun: async () => ({
+        write: async (_dataset, documents) => {
+          const collection = received.get(searchType.name) ?? [];
+          received.set(searchType.name, collection);
+          for await (const document of documents) {
+            collection.push(document);
+          }
+        },
+        commit: async () => {
+          committed.push(searchType.name);
+        },
+        abort: async () => undefined,
+      }),
+    });
+
+    // The pipeline isolates a stage failure per dataset instead of throwing
+    // (e.g. framing crashing on a blank-node root), so capture it explicitly.
+    const stageFailures: Error[] = [];
+    const reporter: ProgressReporter = {
+      stageFailed: (_stage, error) => {
+        stageFailures.push(error);
+      },
+    };
+
+    await searchIndexerPipeline({
+      schema,
+      datasets: [dataset],
+      distributionResolver: resolver,
+      writerFor,
+      reporter,
+    }).run();
+
+    expect(stageFailures).toEqual([]);
+    // Both roots per type – and not the fixture’s blank-node CreativeWork,
+    // which selectByClass must exclude (it has no stable document key).
+    expect(
+      received
+        .get('CreativeWork')
+        ?.map((document) => document.id)
+        .sort(),
+    ).toEqual(['https://ex/cw/1', 'https://ex/cw/2']);
+    expect(received.get('Person')?.map((document) => document.id)).toEqual([
+      'https://ex/p/1',
+    ]);
+    expect(committed.sort()).toEqual(['CreativeWork', 'Person']);
+  });
+});