Fix the problem of incorrect old data passed to watcher during updates (

halo-dev#4959)

#### What type of PR is this?

/kind bug
/area core
/milestone 2.11.0

#### What this PR does / why we need it:

This PR resolves the problem of incorrect old data passed to watcher during updates. As shown in the following line, the old value should be `old` instead of `extension` from outside.

https://github.com/halo-dev/halo/blob/7a84f553005b2d8047ccdb0acf473693384a7b51/application/src/main/java/run/halo/app/extension/ReactiveExtensionClientImpl.java#L172

#### Does this PR introduce a user-facing change?

```release-note
None
```

api/src/main/java/run/halo/app/extension/JsonExtension.java

@@ -17,6 +17,7 @@
 import java.io.IOException;
 import java.time.Instant;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 
 /**
@@ -117,6 +118,23 @@ public MetadataOperator getMetadataOrCreate() {
         return new ObjectNodeMetadata(metadataNode);
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        JsonExtension that = (JsonExtension) o;
+        return Objects.equals(objectNode, that.objectNode);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(objectNode);
+    }
+
     class ObjectNodeMetadata implements MetadataOperator {
 
         private final ObjectNode objectNode;

application/src/main/java/run/halo/app/extension/ReactiveExtensionClientImpl.java

@@ -4,12 +4,13 @@
 import static org.springframework.util.StringUtils.hasText;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Comparator;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Predicate;
 import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.data.util.Predicates;
@@ -137,42 +138,31 @@ public <E extends Extension> Mono<E> create(E extension) {
     @SuppressWarnings("unchecked")
     public <E extends Extension> Mono<E> update(E extension) {
         // Refactor the atomic reference if we have a better solution.
-        final var statusChangeOnly = new AtomicBoolean(false);
-        return getLatest(extension)
-            .map(old -> new JsonExtension(objectMapper, old))
-            .flatMap(oldJsonExt -> {
-                var newJsonExt = new JsonExtension(objectMapper, extension);
-                // reset some mandatory fields
-                var oldMetadata = oldJsonExt.getMetadata();
-                var newMetadata = newJsonExt.getMetadata();
-                newMetadata.setCreationTimestamp(oldMetadata.getCreationTimestamp());
-                newMetadata.setGenerateName(oldMetadata.getGenerateName());
-
-                var oldObjectNode = oldJsonExt.getInternal().deepCopy();
-                var newObjectNode = newJsonExt.getInternal().deepCopy();
-                if (Objects.equals(oldObjectNode, newObjectNode)) {
-                    // if no data were changed, just skip updating.
-                    return Mono.empty();
-                }
-                // check status is changed
-                oldObjectNode.remove("status");
-                newObjectNode.remove("status");
-                if (Objects.equals(oldObjectNode, newObjectNode)) {
-                    statusChangeOnly.set(true);
-                }
-                return Mono.just(newJsonExt);
-            })
-            .map(converter::convertTo)
-            .flatMap(extensionStore -> client.update(extensionStore.getName(),
-                extensionStore.getVersion(),
-                extensionStore.getData()))
-            .map(updated -> converter.convertFrom((Class<E>) extension.getClass(), updated))
-            .doOnNext(updated -> {
-                if (!statusChangeOnly.get()) {
-                    watchers.onUpdate(extension, updated);
-                }
-            })
-            .switchIfEmpty(Mono.defer(() -> Mono.just(extension)));
+        return getLatest(extension).flatMap(old -> {
+            var oldJsonExt = new JsonExtension(objectMapper, old);
+            var newJsonExt = new JsonExtension(objectMapper, extension);
+            // reset some mandatory fields
+            var oldMetadata = oldJsonExt.getMetadata();
+            var newMetadata = newJsonExt.getMetadata();
+            newMetadata.setCreationTimestamp(oldMetadata.getCreationTimestamp());
+            newMetadata.setGenerateName(oldMetadata.getGenerateName());
+
+            if (Objects.equals(oldJsonExt, newJsonExt)) {
+                // skip updating if not data changed.
+                return Mono.just(extension);
+            }
+
+            var onlyStatusChanged =
+                isOnlyStatusChanged(oldJsonExt.getInternal(), newJsonExt.getInternal());
+
+            var store = this.converter.convertTo(newJsonExt);
+            var updated = client.update(store.getName(), store.getVersion(), store.getData())
+                .map(ext -> converter.convertFrom((Class<E>) extension.getClass(), ext));
+            if (!onlyStatusChanged) {
+                updated = updated.doOnNext(ext -> watchers.onUpdate(old, ext));
+            }
+            return updated;
+        });
     }
 
     private Mono<? extends Extension> getLatest(Extension extension) {
@@ -199,4 +189,26 @@ public void watch(Watcher watcher) {
         this.watchers.addWatcher(watcher);
     }
 
+    private static boolean isOnlyStatusChanged(ObjectNode oldNode, ObjectNode newNode) {
+        if (Objects.equals(oldNode, newNode)) {
+            return false;
+        }
+        // WARNING!!!
+        // Do not edit the ObjectNode
+        var oldFields = new HashSet<String>();
+        var newFields = new HashSet<String>();
+        oldNode.fieldNames().forEachRemaining(oldFields::add);
+        newNode.fieldNames().forEachRemaining(newFields::add);
+        oldFields.remove("status");
+        newFields.remove("status");
+        if (!Objects.equals(oldFields, newFields)) {
+            return false;
+        }
+        for (var field : oldFields) {
+            if (!Objects.equals(oldNode.get(field), newNode.get(field))) {
+                return false;
+            }
+        }
+        return true;
+    }
 }

application/src/test/java/run/halo/app/extension/FakeExtension.java

@@ -1,12 +1,21 @@
 package run.halo.app.extension;
 
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
 @GVK(group = "fake.halo.run",
     version = "v1alpha1",
     kind = "Fake",
     plural = "fakes",
     singular = "fake")
+@Data
+@ToString(callSuper = true)
+@EqualsAndHashCode(callSuper = true)
 public class FakeExtension extends AbstractExtension {
 
+    private FakeStatus status = new FakeStatus();
+
     public static FakeExtension createFake(String name) {
         var metadata = new Metadata();
         metadata.setName(name);
@@ -15,4 +24,8 @@ public static FakeExtension createFake(String name) {
         return fake;
     }
 
+    @Data
+    public static class FakeStatus {
+        private String state;
+    }
 }

application/src/test/java/run/halo/app/extension/ReactiveExtensionClientTest.java

@@ -455,6 +455,37 @@ void shouldNotUpdateIfExtensionNotChange() {
         verify(storeClient, never()).update(any(), any(), any());
     }
 
+    @Test
+    void shouldUpdateIfExtensionStatusChangedOnly() {
+        var fake = createFakeExtension("fake", 2L);
+        fake.getStatus().setState("new-state");
+        var storeName = "/registry/fake.halo.run/fakes/fake";
+        when(converter.convertTo(any())).thenReturn(
+            createExtensionStore(storeName, 2L));
+        when(storeClient.update(any(), any(), any())).thenReturn(
+            Mono.just(createExtensionStore(storeName, 2L)));
+        when(storeClient.fetchByName(storeName)).thenReturn(
+            Mono.just(createExtensionStore(storeName, 1L)));
+
+        var oldFake = createFakeExtension("fake", 2L);
+        oldFake.getStatus().setState("old-state");
+
+        var updatedFake = createFakeExtension("fake", 3L);
+        when(converter.convertFrom(same(FakeExtension.class), any()))
+            .thenReturn(oldFake)
+            .thenReturn(updatedFake);
+
+        StepVerifier.create(client.update(fake))
+            .expectNext(updatedFake)
+            .verifyComplete();
+
+        verify(storeClient).fetchByName(storeName);
+        verify(converter).convertTo(isA(JsonExtension.class));
+        verify(converter, times(2)).convertFrom(same(FakeExtension.class), any());
+        verify(storeClient)
+            .update(eq("/registry/fake.halo.run/fakes/fake"), eq(2L), any());
+    }
+
     @Test
     void shouldUpdateUnstructuredSuccessfully() throws JsonProcessingException {
         var fake = createUnstructured();
@@ -539,6 +570,13 @@ void shouldNotWatchOnUpdateIfExtensionNotChange() {
             verify(watcher, never()).onUpdate(any(), any());
         }
 
+        @Test
+        void shouldNotWatchOnUpdateIfExtensionStatusChangeOnly() {
+            shouldUpdateIfExtensionStatusChangedOnly();
+
+            verify(watcher, never()).onUpdate(any(), any());
+        }
+
         @Test
         void shouldWatchOnDeleteSuccessfully() {
             doNothing().when(watcher).onDelete(any());