Content indexer — orchestration, diff, upsert/delete

CLAUDE.md

@@ -49,7 +49,11 @@ extraction, Meilisearch indexing with scheduled refresh, and four MCP tools
 │   ├── ContentSourceStrategy.java           # per-source-type discover/fetch seam
 │   ├── WebsiteSourceStrategy.java           # website strategy: crawler + fetcher + extractor
 │   ├── FetchOutcome.java                    # INDEX/UNCHANGED/DELETE/SKIP + optional document
-│   └── SourceStrategyRegistry.java          # selects a strategy by ContentSource.type()
+│   ├── SourceStrategyRegistry.java          # selects a strategy by ContentSource.type()
+│   ├── ContentIndexer.java                  # orchestrates discover→diff→fetch→upsert/delete
+│   ├── IndexReport.java                     # per-pass counters
+│   ├── ContentIndexStore.java               # index read/write seam (StoredDocument)
+│   └── MeilisearchContentIndexStore.java    # ContentIndexStore backed by MeilisearchClient
 ├── src/main/resources/application.yaml      # datasource, JPA, OAuth2, MCP, Meilisearch, content config
 ├── src/test/java/com/openelements/content/  # behavior tests (context, MCP enabled/disabled, search-down, jsoup)
 └── docs/
@@ -95,7 +99,13 @@ categories/preview image/locale) from OpenGraph/Article `<meta>`, JSON-LD, and `
 wires crawler → fetcher → extractor and maps the fetch result to a `FetchOutcome`
 (`INDEX`/`UNCHANGED`/`DELETE`/`SKIP`); `SourceStrategyRegistry` selects the strategy by
 `ContentSource.type()`, so a future `git` strategy (spec 016) plugs in as a bean with no indexer
-changes.
+changes. `ContentIndexer` is the orchestration engine: discover → diff discovered `lastmod` against
+the state read from the index (`ContentIndexStore`; the index *is* the state) → fetch only new/changed
+items via the strategy → batch-upsert, and delete documents that 404 or vanished from discovery.
+`MeilisearchContentIndexStore` implements the store over `MeilisearchClient` (paged `multiSearch` to
+read state, `addDocuments`+`waitForTask` to upsert, `deleteDocument`), since the library's
+`BatchWriter` is package-private. The indexer is the reusable engine for the bootstrap step (009) and
+refresh scheduler (010).
 
 > **Key gotcha:** the library ships no Spring Boot auto-configuration and couples MCP to a JPA
 > datasource, so `@Import({ McpConfiguration, SearchConfig })` alone does **not** boot. See

docs/specs/008-content-indexer/steps.md

@@ -0,0 +1,61 @@
+# Implementation Steps: Content Indexer
+
+## Step 1: `IndexReport`
+
+- [x] Record `IndexReport(discovered, upserted, unchanged, deleted, skipped)`
+
+**Related behaviors:** all (report counters)
+
+---
+
+## Step 2: `ContentIndexStore` seam
+
+- [x] Interface `loadState(source) → id→StoredDocument`, `upsert(docs) → count`, `delete(id)`; nested `StoredDocument(url, lastmod)`
+- [x] `MeilisearchContentIndexStore` production impl: paged filtered `multiSearch` (testable `parseHits`), batched `addDocuments` + `waitForTask` (BatchWriter is package-private), `deleteDocument`; gated on `meilisearch.enabled`
+
+**Related behaviors:** underpins all (diff/upsert/delete)
+
+---
+
+## Step 3: `ContentIndexer`
+
+- [x] `@Component` (gated on `meilisearch.enabled`) taking `SourceStrategyRegistry` + `ContentIndexStore`
+- [x] `indexSource`: discover → load stored state → per item: unchanged (lastmod equal) skip; else strategy.fetch → INDEX(collect)/UNCHANGED/DELETE/SKIP; batch upsert; delete stored-but-not-discovered
+- [x] `streamAllDocuments`: lazy discover→fetch→INDEX map stream for bootstrap
+- [x] Idempotent (upsert by stable id); fault-tolerant (SKIP one, continue)
+
+**Related behaviors:** new/unchanged/changed/304/vanished/404/idempotent/skip/dedupe/stream
+
+---
+
+## Step 4: Tests
+
+- [x] `ContentIndexerTest` (10) — in-memory store + stub strategy: every behavior
+- [x] `MeilisearchContentIndexStoreTest` (3) — `parseHits` parsing (hits, empty, missing id)
+
+**Acceptance criteria:**
+- [x] All tests pass (`mvn test`); build green
+
+---
+
+## Behavior Coverage
+
+| Scenario | Layer | Covered in Step |
+|----------|-------|-----------------|
+| New pages are upserted | Backend | Step 4 (`newPagesAreUpserted`) |
+| Unchanged pages are skipped | Backend | Step 4 (`unchangedPagesAreSkipped`) |
+| Changed page is re-fetched and upserted | Backend | Step 4 (`changedPageIsReFetched`) |
+| Conditional GET short-circuits a lastmod false-positive | Backend | Step 4 (`conditionalGetShortCircuits`) |
+| Vanished URL is deleted | Backend | Step 4 (`vanishedUrlIsDeleted`) |
+| 404 on fetch deletes the document | Backend | Step 4 (`notFoundDeletesDocument`) |
+| Re-running is idempotent | Backend | Step 4 (`reRunIsIdempotent`) |
+| One failing page does not abort the batch | Backend | Step 4 (`oneFailingPageIsSkipped`) |
+| Stable id prevents duplicates | Backend | Step 4 (`stableIdPreventsDuplicates`) |
+| streamAllDocuments yields every in-scope document | Backend | Step 4 (`streamAllDocumentsYieldsAll`) |
+
+All scenarios are backend; there is no frontend in this spec.
+
+## Notes / deviations
+
+- The design suggested `BatchWriter.write(...)`, but `BatchWriter` is package-private in `spring-services` and not reusable from this package. Upsert uses `MeilisearchClient.addDocuments` in batches with `waitForTask`.
+- A `ContentIndexStore` seam was introduced (not in the original design) to encapsulate the Meilisearch read/write and make the orchestration unit-testable without a live instance. The production impl's `parseHits` is unit-tested; the `multiSearch` paging / `addDocuments` batching is thin delegation exercised end-to-end by specs 009/010 against real Meilisearch.

docs/specs/INDEX.md

@@ -13,7 +13,7 @@ roadmap step, to be implemented sequentially (each builds on the previous).
 | 005 | 005-page-fetcher | Page fetcher | backend, crawler | `PageFetcher` — robust HTTP fetch (ETag/IMS, rate limit, retry) | #9 | done |
 | 006 | 006-content-extractor | Content extractor | backend, crawler | `ContentExtractor` — jsoup content + metadata extraction | #11 | done |
 | 007 | 007-source-strategy | Source strategy | backend, architecture | `ContentSourceStrategy` interface + `WebsiteSourceStrategy` | #13 | done |
-| 008 | 008-content-indexer | Content indexer | backend, search, crawler | `ContentIndexer` — orchestration, diff, upsert/delete | — | open |
+| 008 | 008-content-indexer | Content indexer | backend, search, crawler | `ContentIndexer` — orchestration, diff, upsert/delete | #15 | done |
 | 009 | 009-content-bootstrap-step | Content bootstrap step | backend, search | `ContentBootstrapStep` — initial reindex via `SearchIndexBootstrapStep` | — | open |
 | 010 | 010-content-refresh-scheduler | Content refresh scheduler | backend, scheduling | `ContentRefreshScheduler` — `@Scheduled` incremental re-crawl | — | open |
 | 011 | 011-content-search-service | Content search service | backend, search | `ContentSearchService` — `multiSearch` + highlighting facade | — | open |

src/main/java/com/openelements/content/ContentIndexStore.java

@@ -0,0 +1,47 @@
+package com.openelements.content;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The persistence seam the {@link ContentIndexer} uses to read and write index state.
+ *
+ * <p>The Meilisearch content index <em>is</em> the state: {@link #loadState(String)} returns the
+ * stored change-marker per document so the indexer can diff without a separate datastore. Isolating
+ * these three operations keeps the indexer's orchestration logic testable with an in-memory
+ * implementation, independent of a running Meilisearch.
+ */
+public interface ContentIndexStore {
+
+    /**
+     * Loads the currently-indexed documents for a source.
+     *
+     * @param source the source id
+     * @return a map of document id → {@link StoredDocument} (its URL and change marker)
+     */
+    Map<String, StoredDocument> loadState(String source);
+
+    /**
+     * Upserts documents (keyed by their {@code id}); existing documents with the same id are replaced.
+     *
+     * @param documents document attribute maps (from {@link ContentDocument#toMap()})
+     * @return the number of documents written
+     */
+    int upsert(List<Map<String, Object>> documents);
+
+    /**
+     * Removes a document from the index.
+     *
+     * @param id the stable document id
+     */
+    void delete(String id);
+
+    /**
+     * The stored projection of an indexed document needed for diffing.
+     *
+     * @param url     the document URL
+     * @param lastmod the stored change marker, or {@code null}
+     */
+    record StoredDocument(String url, String lastmod) {
+    }
+}

src/main/java/com/openelements/content/ContentIndexer.java

@@ -0,0 +1,121 @@
+package com.openelements.content;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Stream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+/**
+ * Orchestrates the ingestion pipeline for one source: Discover → Diff → Fetch → Extract → Index.
+ *
+ * <p>Discovery yields URLs with change markers; the indexer diffs these against the state stored in
+ * the index (via {@link ContentIndexStore}) and only fetches new or changed items. Fetched documents
+ * are upserted; items reported gone (404) or no longer present in discovery are deleted. Upserts are
+ * keyed by the stable document id, so re-running is idempotent, and a single failing page is skipped
+ * without aborting the pass.
+ *
+ * <p>This is the reusable engine invoked by the bootstrap step (spec 009) and the refresh scheduler
+ * (spec 010). It is only created when the Meilisearch stack is enabled.
+ */
+@Component
+@ConditionalOnProperty(prefix = "openelements.meilisearch", name = "enabled", havingValue = "true")
+public class ContentIndexer {
+
+    private static final Logger log = LoggerFactory.getLogger(ContentIndexer.class);
+
+    private final SourceStrategyRegistry strategyRegistry;
+    private final ContentIndexStore indexStore;
+
+    public ContentIndexer(SourceStrategyRegistry strategyRegistry, ContentIndexStore indexStore) {
+        this.strategyRegistry = strategyRegistry;
+        this.indexStore = indexStore;
+    }
+
+    /**
+     * Runs a full incremental pass over one source.
+     *
+     * @param source the source to index
+     * @return a report of what happened
+     */
+    public IndexReport indexSource(ContentSource source) {
+        ContentSourceStrategy strategy = strategyRegistry.forSource(source);
+        List<DiscoveredItem> discovered = strategy.discover(source);
+        Map<String, ContentIndexStore.StoredDocument> stored = indexStore.loadState(source.id());
+
+        List<Map<String, Object>> toUpsert = new ArrayList<>();
+        Set<String> discoveredIds = new HashSet<>();
+        int upserted = 0;
+        int unchanged = 0;
+        int deleted = 0;
+        int skipped = 0;
+
+        for (DiscoveredItem item : discovered) {
+            String id = ContentDocument.id(source.id(), item.url());
+            discoveredIds.add(id);
+            ContentIndexStore.StoredDocument existing = stored.get(id);
+
+            if (existing != null && isUnchanged(existing.lastmod(), item.lastmod())) {
+                unchanged++;
+                continue;
+            }
+
+            FetchOutcome outcome = strategy.fetch(source, item);
+            switch (outcome.result()) {
+                case INDEX -> {
+                    toUpsert.add(outcome.document().toMap());
+                    upserted++;
+                }
+                case UNCHANGED -> unchanged++;
+                case DELETE -> {
+                    if (existing != null) {
+                        indexStore.delete(id);
+                        deleted++;
+                    }
+                }
+                case SKIP -> skipped++;
+            }
+        }
+
+        if (!toUpsert.isEmpty()) {
+            indexStore.upsert(toUpsert);
+        }
+
+        // Documents still in the index but no longer discovered have been removed at the source.
+        for (String storedId : stored.keySet()) {
+            if (!discoveredIds.contains(storedId)) {
+                indexStore.delete(storedId);
+                deleted++;
+            }
+        }
+
+        IndexReport report = new IndexReport(discovered.size(), upserted, unchanged, deleted, skipped);
+        log.info("Indexed source {}: {}", source.id(), report);
+        return report;
+    }
+
+    /**
+     * Lazily fetches and yields every in-scope document of a source, without diffing against the
+     * index — used for a full (re)build by the bootstrap step (spec 009).
+     *
+     * @param source the source
+     * @return a lazy stream of document attribute maps ready for the index
+     */
+    public Stream<Map<String, Object>> streamAllDocuments(ContentSource source) {
+        ContentSourceStrategy strategy = strategyRegistry.forSource(source);
+        return strategy.discover(source).stream()
+            .map(item -> strategy.fetch(source, item))
+            .filter(outcome -> outcome.result() == FetchOutcome.Result.INDEX)
+            .map(outcome -> outcome.document().toMap());
+    }
+
+    /** A discovered item is unchanged only when it carries a change marker equal to the stored one. */
+    private static boolean isUnchanged(String storedLastmod, String discoveredLastmod) {
+        return discoveredLastmod != null && discoveredLastmod.equals(storedLastmod);
+    }
+}

src/main/java/com/openelements/content/IndexReport.java

@@ -0,0 +1,13 @@
+package com.openelements.content;
+
+/**
+ * Summary of one indexing pass over a source.
+ *
+ * @param discovered number of items discovery returned
+ * @param upserted   documents fetched/extracted and written to the index
+ * @param unchanged  items skipped because they were unchanged (by {@code lastmod} diff or a 304)
+ * @param deleted    documents removed (vanished from discovery, or 404 on fetch)
+ * @param skipped    items that failed to fetch or extract and were skipped
+ */
+public record IndexReport(int discovered, int upserted, int unchanged, int deleted, int skipped) {
+}