refactor: wip

Author: dieppa

core/flamingock-core-commons/src/main/java/io/flamingock/internal/common/core/response/data/ChangeResult.java

@@ -113,6 +113,10 @@ public boolean isAlreadyApplied() {
         return status == ChangeStatus.ALREADY_APPLIED;
     }
 
+    public boolean isNotReached() {
+        return status == ChangeStatus.NOT_REACHED;
+    }
+
     public static Builder builder() {
         return new Builder();
     }

core/flamingock-core-commons/src/main/java/io/flamingock/internal/common/core/response/data/ChangeStatus.java

@@ -20,27 +20,33 @@
  */
 public enum ChangeStatus {
     /**
-     * Change was successfully applied.
+     * Change was successfully applied during this run by the operation.
      */
     APPLIED,
 
     /**
-     * Change was already applied in a previous execution.
+     * Change was found already applied — either by the executor (SKIP action during this run)
+     * or by the planner (audit-confirmed without invoking the executor).
      */
     ALREADY_APPLIED,
 
     /**
-     * Change failed during execution.
+     * Change failed during execution this run.
      */
     FAILED,
 
     /**
-     * Change failed but was successfully rolled back.
+     * Change failed during execution this run and was successfully rolled back (transactional
+     * auto-rollback).
      */
     ROLLED_BACK,
 
     /**
-     * Change was not reached due to a prior failure.
+     * No positive information about this change in this run. Either the operation didn't process
+     * it (executor stopped on a prior failure, the stage was unreached, etc.) or the planner
+     * found no audit entry confirming it was applied. Default initial status assigned to every
+     * loaded change at {@code PipelineRun} construction; writers transition records forward as
+     * they learn facts.
      */
     NOT_REACHED
 }

core/flamingock-core-commons/src/main/java/io/flamingock/internal/common/core/response/data/ExecuteResponseData.java

@@ -42,6 +42,7 @@ public class ExecuteResponseData {
     private int appliedChanges;
     private int skippedChanges;
     private int failedChanges;
+    private int notReachedChanges;
 
     // Per-stage breakdown
     private List<StageResult> stages;
@@ -64,6 +65,7 @@ private ExecuteResponseData(Builder builder) {
         this.appliedChanges = builder.appliedChanges;
         this.skippedChanges = builder.skippedChanges;
         this.failedChanges = builder.failedChanges;
+        this.notReachedChanges = builder.notReachedChanges;
         this.stages = builder.stages != null ? builder.stages : new ArrayList<>();
     }
 
@@ -171,6 +173,14 @@ public void setFailedChanges(int failedChanges) {
         this.failedChanges = failedChanges;
     }
 
+    public int getNotReachedChanges() {
+        return notReachedChanges;
+    }
+
+    public void setNotReachedChanges(int notReachedChanges) {
+        this.notReachedChanges = notReachedChanges;
+    }
+
     public List<StageResult> getStages() {
         return stages;
     }
@@ -205,6 +215,7 @@ public static class Builder {
         private int appliedChanges;
         private int skippedChanges;
         private int failedChanges;
+        private int notReachedChanges;
         private List<StageResult> stages = new ArrayList<>();
 
         public Builder status(ExecutionStatus status) {
@@ -272,6 +283,11 @@ public Builder failedChanges(int failedChanges) {
             return this;
         }
 
+        public Builder notReachedChanges(int notReachedChanges) {
+            this.notReachedChanges = notReachedChanges;
+            return this;
+        }
+
         public Builder stages(List<StageResult> stages) {
             this.stages = stages;
             return this;

core/flamingock-core-commons/src/main/java/io/flamingock/internal/common/core/response/data/ExecutionReportFormatter.java

@@ -73,10 +73,11 @@ public static String summary(ExecuteResponseData result) {
         }
 
         String counts = String.format(
-                "; changes applied=%d, failed=%d, skipped=%d; duration=%dms",
+                "; changes newly_applied=%d, already_applied=%d, failed=%d, not_reached=%d; duration=%dms",
                 result.getAppliedChanges(),
-                result.getFailedChanges(),
                 result.getSkippedChanges(),
+                result.getFailedChanges(),
+                result.getNotReachedChanges(),
                 result.getTotalDurationMs());
 
         StringBuilder sb = new StringBuilder(headline).append(counts);
@@ -110,9 +111,10 @@ public static String report(ExecuteResponseData result) {
           .append(NEWLINE);
         sb.append(" Changes:   ")
           .append(result.getTotalChanges()).append(" total — ")
-          .append(result.getAppliedChanges()).append(" applied, ")
-          .append(result.getSkippedChanges()).append(" already at target state, ")
-          .append(result.getFailedChanges()).append(" failed")
+          .append(result.getAppliedChanges()).append(" newly applied, ")
+          .append(result.getSkippedChanges()).append(" already applied, ")
+          .append(result.getFailedChanges()).append(" failed, ")
+          .append(result.getNotReachedChanges()).append(" not reached")
           .append(NEWLINE);
 
         // Per-stage breakdown: include any stage the operation touched (state != NOT_STARTED) OR
@@ -181,7 +183,7 @@ private static void appendStageBlock(StringBuilder sb, StageResult stage) {
                     : 0;
             int reportableCount = alreadyAppliedCount > 0 ? alreadyAppliedCount : stage.getTotalChanges();
             sb.append("   ").append(label).append(' ').append(name)
-              .append("  (").append(reportableCount).append(" changes already at target state)")
+              .append("  (").append(reportableCount).append(" changes already applied)")
               .append(NEWLINE);
             return;
         }
@@ -195,10 +197,12 @@ private static void appendStageBlock(StringBuilder sb, StageResult stage) {
         // ROLLED_BACK is widened into the failed bucket — the change did not succeed even though
         // the system was left clean. See PipelineRun.toResponse() for the matching aggregate logic.
         int failed = (int) changes.stream().filter(c -> c != null && (c.isFailed() || c.isRolledBack())).count();
+        int notReached = (int) changes.stream().filter(c -> c != null && c.isNotReached()).count();
         sb.append("               changes: ")
-          .append(applied).append(" applied, ")
-          .append(skipped).append(" already at target state, ")
-          .append(failed).append(" failed")
+          .append(applied).append(" newly applied, ")
+          .append(skipped).append(" already applied, ")
+          .append(failed).append(" failed, ")
+          .append(notReached).append(" not reached")
           .append(NEWLINE);
 
         StageState state = stage.getState();

core/flamingock-core-commons/src/test/java/io/flamingock/internal/common/core/response/data/ExecutionReportFormatterTest.java

@@ -122,10 +122,10 @@ void reportSuccessHeadlineAndPerStageBlock() {
         String out = ExecutionReportFormatter.report(result);
         assertTrue(out.contains("Flamingock execution report — SUCCESS"), out);
         assertTrue(out.contains("Stages:    1 total — 1 completed, 0 failed, 0 up to date, 0 not reached"), out);
-        assertTrue(out.contains("Changes:   1 total — 1 applied, 0 already at target state, 0 failed"), out);
+        assertTrue(out.contains("Changes:   1 total — 1 newly applied, 0 already applied, 0 failed, 0 not reached"), out);
         assertTrue(out.contains("[COMPLETED]"), out);
         assertTrue(out.contains("only-stage"), out);
-        assertTrue(out.contains("1 applied, 0 already at target state, 0 failed"), out);
+        assertTrue(out.contains("1 newly applied, 0 already applied, 0 failed, 0 not reached"), out);
         assertFalse(out.contains("Not reached"), "no Not-reached section when all stages reached: " + out);
     }
 
@@ -243,7 +243,7 @@ void reportCountsRolledBackChangeAsFailedAndListsItsId() {
                 .build();
 
         String out = ExecutionReportFormatter.report(result);
-        assertTrue(out.contains("0 applied, 5 already at target state, 1 failed"),
+        assertTrue(out.contains("0 newly applied, 5 already applied, 1 failed"),
                 "per-stage block must count ROLLED_BACK as failed: " + out);
         assertTrue(out.contains("failed change(s): InsertEmployeeSeedData"),
                 "failed change(s) line must include the rolled-back change ID: " + out);
@@ -292,12 +292,12 @@ void reportAllUpToDateShowsNoChangesHeadlineAndUpToDateRowsInBreakdown() {
         assertTrue(out.contains("Flamingock execution report — NO CHANGES"),
                 "headline must render NO_CHANGES with a space: " + out);
         assertTrue(out.contains("Stages:    1 total — 0 completed, 0 failed, 1 up to date, 0 not reached"), out);
-        assertTrue(out.contains("Changes:   6 total — 0 applied, 6 already at target state, 0 failed"), out);
+        assertTrue(out.contains("Changes:   6 total — 0 newly applied, 6 already applied, 0 failed, 0 not reached"), out);
         assertTrue(out.contains("Per-stage breakdown:"),
                 "per-stage section should include [UP TO DATE] rows: " + out);
         assertTrue(out.contains("[UP TO DATE]"), out);
         assertTrue(out.contains("database-init"), out);
-        assertTrue(out.contains("(6 changes already at target state)"), out);
+        assertTrue(out.contains("(6 changes already applied)"), out);
         assertFalse(out.contains("Not reached"),
                 "Not-reached section must be omitted when nothing reached and nothing not-evaluated: " + out);
     }
@@ -321,7 +321,7 @@ void reportAllUpToDateForCloudCaseRendersCountOnlyWhenChangesEmpty() {
 
         String out = ExecutionReportFormatter.report(result);
         assertTrue(out.contains("[UP TO DATE]"), out);
-        assertTrue(out.contains("(6 changes already at target state)"),
+        assertTrue(out.contains("(6 changes already applied)"),
                 "cloud-style up-to-date row should use totalChanges when per-change list is empty: " + out);
     }
 
@@ -359,14 +359,14 @@ void reportPartialCoverageMixesReachedUpToDateAndNotReached() {
         ExecuteResponseData result = ExecuteResponseData.builder()
                 .status(ExecutionStatus.FAILED)
                 .totalStages(4).completedStages(1).upToDateStages(1).notReachedStages(1).failedStages(1)
-                .totalChanges(10).appliedChanges(3).skippedChanges(3).failedChanges(1)
+                .totalChanges(10).appliedChanges(3).skippedChanges(3).failedChanges(1).notReachedChanges(3)
                 .totalDurationMs(170)
                 .stages(Arrays.asList(block1, block2, block3, block4))
                 .build();
 
         String out = ExecutionReportFormatter.report(result);
         assertTrue(out.contains("Stages:    4 total — 1 completed, 1 failed, 1 up to date, 1 not reached"), out);
-        assertTrue(out.contains("Changes:   10 total — 3 applied, 3 already at target state, 1 failed"), out);
+        assertTrue(out.contains("Changes:   10 total — 3 newly applied, 3 already applied, 1 failed, 3 not reached"), out);
         assertTrue(out.contains("Per-stage breakdown:"), out);
         assertTrue(out.contains("[COMPLETED] block-1"), out);
         assertTrue(out.contains("[UP TO DATE]") && out.contains("block-2"), out);

core/flamingock-core/src/main/java/io/flamingock/internal/core/pipeline/run/PipelineRun.java

@@ -34,11 +34,10 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashSet;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.stream.Collectors;
 
 public class PipelineRun {
@@ -165,13 +164,16 @@ public void markStageStarted(String stageName) {
     }
 
     public void markStageCompleted(String stageName, StageResult result) {
-        // Incoming result from the executor already carries state=COMPLETED. Preserve the
-        // totalChanges / plannerVerdict fields set on the in-memory StageResult before
-        // execution started — the executor doesn't know about them and would otherwise reset
-        // them to defaults.
+        // Incoming result from the executor already carries state=COMPLETED. Merge per-change
+        // records by ID: the executor's records win for the IDs it processed; records for
+        // changes the executor didn't process (i.e. NOT_REACHED records from PipelineRun
+        // construction) survive. Also preserve totalChanges / plannerVerdict — the executor
+        // doesn't know about them.
         StageRun run = lookupOrThrow(stageName);
         StageResult current = run.getResult();
+        List<ChangeResult> merged = mergeChangeRecords(current.getChanges(), result.getChanges());
         run.setResult(StageResult.builder(result)
+                .changes(merged)
                 .totalChanges(current.getTotalChanges())
                 .plannerVerdict(current.getPlannerVerdict())
                 .build());
@@ -186,13 +188,45 @@ public void markStageFailed(String stageName, StageExecutionException exception)
         );
         StageRun run = lookupOrThrow(stageName);
         StageResult current = run.getResult();
+        // Same merge semantics as markStageCompleted: executor's records win for processed
+        // IDs; NOT_REACHED records for changes the executor never reached (stage stopped on
+        // a failure) stay intact.
+        List<ChangeResult> merged = mergeChangeRecords(current.getChanges(), resultFromExecutor.getChanges());
         run.setResult(StageResult.builder(resultFromExecutor)
                 .state(StageState.failed(errorInfo))
+                .changes(merged)
                 .totalChanges(current.getTotalChanges())
                 .plannerVerdict(current.getPlannerVerdict())
                 .build());
     }
 
+    /**
+     * Merges two per-change record lists keyed by change ID. Incoming records (typically from
+     * the executor) take precedence for the IDs they carry; existing records (e.g. initial
+     * NOT_REACHED entries from {@code StageRun} construction) for any other ID survive. Order
+     * follows the first occurrence — existing entries keep their position; new IDs from
+     * {@code incoming} are appended.
+     */
+    private static List<ChangeResult> mergeChangeRecords(List<ChangeResult> existing,
+                                                          List<ChangeResult> incoming) {
+        LinkedHashMap<String, ChangeResult> byId = new LinkedHashMap<>();
+        if (existing != null) {
+            for (ChangeResult r : existing) {
+                if (r != null && r.getChangeId() != null) {
+                    byId.put(r.getChangeId(), r);
+                }
+            }
+        }
+        if (incoming != null) {
+            for (ChangeResult r : incoming) {
+                if (r != null && r.getChangeId() != null) {
+                    byId.put(r.getChangeId(), r);
+                }
+            }
+        }
+        return new ArrayList<>(byId.values());
+    }
+
     /**
      * Stage-scope failure for non-{@link StageExecutionException} Throwables (no enriched
      * {@link StageResult} from the executor; we rebuild the existing one with a Failed state).
@@ -256,11 +290,15 @@ public void markStageVerdict(String stageName, PlannerVerdict verdict) {
     }
 
     /**
-     * Adds {@link ChangeStatus#ALREADY_APPLIED} {@link ChangeResult}s for the given change IDs.
-     * Defensive merge: for any change ID the operation has already recorded a result for, the
-     * planner's would-be record is dropped — the operation's observation is more authoritative.
-     * This keeps the iterative planner→operation→planner cycle safe when audit re-reads see
-     * APPLIED entries the operation just wrote.
+     * Upgrades {@link ChangeStatus#NOT_REACHED} records to {@link ChangeStatus#ALREADY_APPLIED}
+     * for the given change IDs. Defensive merge: only upgrades records currently at NOT_REACHED;
+     * any record the operation (or a prior writer) has already populated with positive info
+     * (APPLIED / FAILED / ROLLED_BACK / ALREADY_APPLIED) is left untouched.
+     *
+     * <p>This codifies "operation wins, planner fills gaps" against the fully-populated record
+     * set that {@code StageRun} initializes with NOT_REACHED entries. As a fallback, IDs without
+     * an existing record (shouldn't happen — every loaded change starts with one — but defensive)
+     * get appended as fresh ALREADY_APPLIED.
      */
     public void markStageAlreadyAppliedFromAudit(String stageName, List<String> changeIds) {
         if (changeIds == null || changeIds.isEmpty()) {
@@ -271,21 +309,34 @@ public void markStageAlreadyAppliedFromAudit(String stageName, List<String> chan
         List<ChangeResult> merged = current.getChanges() != null
                 ? new ArrayList<>(current.getChanges())
                 : new ArrayList<>();
-        Set<String> existing = new HashSet<>();
-        for (ChangeResult c : merged) {
+        Map<String, Integer> indexById = new HashMap<>();
+        for (int i = 0; i < merged.size(); i++) {
+            ChangeResult c = merged.get(i);
             if (c != null && c.getChangeId() != null) {
-                existing.add(c.getChangeId());
+                indexById.put(c.getChangeId(), i);
             }
         }
         for (String id : changeIds) {
-            if (id == null || existing.contains(id)) {
+            if (id == null) {
+                continue;
+            }
+            Integer idx = indexById.get(id);
+            if (idx == null) {
+                // No existing record (defensive): append.
+                merged.add(ChangeResult.builder()
+                        .changeId(id)
+                        .status(ChangeStatus.ALREADY_APPLIED)
+                        .build());
+                indexById.put(id, merged.size() - 1);
                 continue;
             }
-            merged.add(ChangeResult.builder()
-                    .changeId(id)
-                    .status(ChangeStatus.ALREADY_APPLIED)
-                    .build());
-            existing.add(id);
+            if (merged.get(idx).isNotReached()) {
+                merged.set(idx, ChangeResult.builder()
+                        .changeId(id)
+                        .status(ChangeStatus.ALREADY_APPLIED)
+                        .build());
+            }
+            // else: operation (or prior planner write) has positive info — respect it.
         }
         run.setResult(StageResult.builder(current)
                 .changes(merged)
@@ -327,6 +378,7 @@ public ExecuteResponseData toResponse() {
         int appliedChanges = 0;
         int skippedChanges = 0;
         int failedChanges = 0;
+        int notReachedChanges = 0;
         boolean anyFailed = false;
 
         for (StageRun stageRun : stageRuns) {
@@ -369,6 +421,8 @@ public ExecuteResponseData toResponse() {
                         // ROLLED_BACK counts as a failed attempt for the user-facing aggregate.
                         // The per-change record retains the precise status for downstream consumers.
                         failedChanges++;
+                    } else if (change.getStatus() == ChangeStatus.NOT_REACHED) {
+                        notReachedChanges++;
                     }
                 }
             }
@@ -403,6 +457,7 @@ public ExecuteResponseData toResponse() {
                 .appliedChanges(appliedChanges)
                 .skippedChanges(skippedChanges)
                 .failedChanges(failedChanges)
+                .notReachedChanges(notReachedChanges)
                 .stages(stages)
                 .build();
     }