Fix DataTableRowWatcher to work under CsvDataTableStore

DataTableRowWatcher diffed snapshot row counts via DataTableStore.getRows(),
which returns Stream.empty() under CsvDataTableStore. The watcher therefore
saw zero rows in production (`mod`, SaaS) regardless of what the inner recipe
actually emitted, so all six recipes that depend on it (LibraryUpgrade,
ScalaVersionUpgrade, KotlinVersionUpgrade, GroovyVersionUpgrade,
ParentPomUpgrade, DependencyVulnerabilityCheck) silently produced no
UpgradesAndMigrations rows.

Replace the read-back approach with a transparent decorator that wraps the
existing store for the duration of the inner-recipe call, intercepts
insertRow on the matching DataTable bucket, and forwards every call to the
delegate. Works under any DataTableStore implementation.

Author: bmuschko

build.gradle.kts

@@ -37,6 +37,7 @@ dependencies {
     testImplementation("org.openrewrite:rewrite-java-21:${rewriteVersion}")
     testImplementation(gradleApi())
     testImplementation("org.openrewrite.gradle.tooling:model:${rewriteVersion}")
+    testImplementation("de.siegmar:fastcsv:3.+")
     testRuntimeOnly("junit:junit:4.+")
 }
 

src/main/java/io/moderne/devcenter/internal/DataTableRowWatcher.java

@@ -15,39 +15,77 @@
  */
 package io.moderne.devcenter.internal;
 
+import org.jspecify.annotations.Nullable;
 import org.openrewrite.DataTable;
 import org.openrewrite.DataTableExecutionContextView;
 import org.openrewrite.DataTableStore;
 import org.openrewrite.ExecutionContext;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
-
-import static java.util.stream.Collectors.toList;
+import java.util.Objects;
+import java.util.stream.Stream;
 
 public class DataTableRowWatcher<Row> {
     private final DataTable<Row> dataTable;
     private final ExecutionContext ctx;
-    private int snapshotSize;
+    private final List<Row> captured = new ArrayList<>();
+    private @Nullable DataTableStore originalStore;
 
     public DataTableRowWatcher(DataTable<Row> dataTable, ExecutionContext ctx) {
         this.dataTable = dataTable;
         this.ctx = ctx;
     }
 
     public void start() {
-        DataTableStore store = DataTableExecutionContextView.view(ctx).getDataTableStore();
-        snapshotSize = (int) store.getRows(dataTable.getName(), dataTable.getGroup()).count();
+        DataTableExecutionContextView view = DataTableExecutionContextView.view(ctx);
+        originalStore = view.getDataTableStore();
+        view.setDataTableStore(new RecordingStore(originalStore));
     }
 
-    @SuppressWarnings("unchecked")
     public List<Row> stop() {
-        DataTableStore store = DataTableExecutionContextView.view(ctx).getDataTableStore();
-        List<Row> allRows = (List<Row>) store.getRows(dataTable.getName(), dataTable.getGroup())
-                .collect(toList());
-        if (allRows.size() > snapshotSize) {
-            return new ArrayList<>(allRows.subList(snapshotSize, allRows.size()));
+        DataTableExecutionContextView.view(ctx).setDataTableStore(
+                Objects.requireNonNull(originalStore, "stop() called before start()"));
+        return new ArrayList<>(captured);
+    }
+
+    private boolean matchesTarget(DataTable<?> table) {
+        if (!table.getName().equals(dataTable.getName())) {
+            return false;
+        }
+        return Objects.equals(bucketKey(table), bucketKey(dataTable));
+    }
+
+    private static @Nullable String bucketKey(DataTable<?> table) {
+        return table.getGroup() != null ? table.getGroup() : table.getInstanceName();
+    }
+
+    private final class RecordingStore implements DataTableStore {
+        private final DataTableStore delegate;
+
+        RecordingStore(DataTableStore delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public <R> void insertRow(DataTable<R> table, ExecutionContext ctx, R row) {
+            delegate.insertRow(table, ctx, row);
+            if (matchesTarget(table)) {
+                @SuppressWarnings("unchecked")
+                Row casted = (Row) row;
+                captured.add(casted);
+            }
+        }
+
+        @Override
+        public Stream<?> getRows(String dataTableName, @Nullable String group) {
+            return delegate.getRows(dataTableName, group);
+        }
+
+        @Override
+        public Collection<DataTable<?>> getDataTables() {
+            return delegate.getDataTables();
         }
-        return new ArrayList<>();
     }
 }

src/test/java/io/moderne/devcenter/DevCenterTest.java

@@ -15,22 +15,32 @@
  */
 package io.moderne.devcenter;
 
+import de.siegmar.fastcsv.reader.CommentStrategy;
+import de.siegmar.fastcsv.reader.CsvReader;
+import de.siegmar.fastcsv.reader.NamedCsvRecord;
 import io.moderne.devcenter.table.UpgradesAndMigrations;
 import org.intellij.lang.annotations.Language;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.openrewrite.CsvDataTableStore;
+import org.openrewrite.DataTableExecutionContextView;
 import org.openrewrite.DocumentExample;
+import org.openrewrite.InMemoryExecutionContext;
 import org.openrewrite.Recipe;
 import org.openrewrite.config.Environment;
 import org.openrewrite.config.YamlResourceLoader;
 import org.openrewrite.test.RewriteTest;
 
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.List;
 import java.util.Properties;
 import java.util.stream.Stream;
@@ -248,6 +258,99 @@ public List<Recipe> getRecipeList() {
         );
     }
 
+    @Test
+    void libraryUpgradeUnderCsvDataTableStore(@TempDir Path tempDir) throws Exception {
+        Path csvDir = tempDir.resolve("datatables");
+        Files.createDirectories(csvDir);
+        CsvDataTableStore store = new CsvDataTableStore(csvDir);
+        InMemoryExecutionContext ctx = new InMemoryExecutionContext();
+        DataTableExecutionContextView.view(ctx).setDataTableStore(store);
+
+        @Language("yaml") var recipe = """
+          type: specs.openrewrite.org/v1beta/recipe
+          name: io.moderne.devcenter.SpringBoot4Card
+          displayName: Move to Spring Boot 4.0
+          description: Spring Boot 4.0 upgrade card.
+          recipeList:
+            - io.moderne.devcenter.LibraryUpgrade:
+                cardName: Move to Spring Boot 4.0
+                groupIdPattern: org.springframework.boot
+                artifactIdPattern: '*'
+                version: 4.0.0
+                upgradeRecipe: io.moderne.java.spring.boot4.UpgradeSpringBoot_4_0
+          """;
+        rewriteRun(
+          spec ->
+            spec.recipeFromYaml(recipe, "io.moderne.devcenter.SpringBoot4Card")
+              .executionContext(ctx)
+              .beforeRecipe(withToolingApi()),
+          //language=Groovy
+          buildGradle(
+            """
+              plugins {
+                  id "java"
+              }
+              repositories {
+                  mavenCentral()
+              }
+              dependencies {
+                  implementation "org.springframework.boot:spring-boot-starter:3.1.2"
+              }
+              """,
+            """
+              plugins {
+                  id "java"
+              }
+              repositories {
+                  mavenCentral()
+              }
+              dependencies {
+                  /*~~(org.springframework.boot:spring-boot-autoconfigure:3.1.2,org.springframework.boot:spring-boot-starter-logging:3.1.2,org.springframework.boot:spring-boot-starter:3.1.2,org.springframework.boot:spring-boot:3.1.2)~~>*/implementation "org.springframework.boot:spring-boot-starter:3.1.2"
+              }
+              """
+          )
+        );
+        store.close();
+
+        Path depsCsv = csvDir.resolve("org.openrewrite.maven.table.DependenciesInUse.csv");
+        assertThat(depsCsv).exists();
+        assertThat(readCsvRows(depsCsv))
+          .as("Dependency is resolved and written to DependenciesInUse.csv")
+          .anySatisfy(row -> {
+              assertThat(row.getField("groupId")).isEqualTo("org.springframework.boot");
+              assertThat(row.getField("artifactId")).isEqualTo("spring-boot-starter");
+              assertThat(row.getField("version")).isEqualTo("3.1.2");
+          });
+
+        Path upgradesCsv = findCsv(csvDir, "io.moderne.devcenter.table.UpgradesAndMigrations");
+        assertThat(readCsvRows(upgradesCsv))
+          .singleElement()
+          .satisfies(row -> {
+              assertThat(row.getField("card")).isEqualTo("Move to Spring Boot 4.0");
+              assertThat(row.getField("value")).isEqualTo("Major");
+              assertThat(row.getField("currentMinimumVersion")).isEqualTo("3.1.2");
+          });
+    }
+
+    private static List<NamedCsvRecord> readCsvRows(Path csv) throws IOException {
+        try (CsvReader<NamedCsvRecord> reader = CsvReader.builder()
+                .commentCharacter('#')
+                .commentStrategy(CommentStrategy.SKIP)
+                .ofNamedCsvRecord(csv)) {
+            return reader.stream().toList();
+        }
+    }
+
+    private static Path findCsv(Path dir, String namePrefix) throws IOException {
+        try (Stream<Path> files = Files.list(dir)) {
+            return files
+              .filter(p -> p.getFileName().toString().startsWith(namePrefix))
+              .findFirst()
+              .orElseThrow(() -> new AssertionError(
+                "No CSV starting with '" + namePrefix + "' in " + dir));
+        }
+    }
+
     @DocumentExample
     @Test
     void devcenterWithMultipleLibraryUpgradeRecipesHasCorrectData() {