Fix the problem that deletable extensions created by plugins cannot be recycled (#4526)

#### What type of PR is this?

/kind bug
/area core
/milestone 2.9.x

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

As I mentioned in <https://github.com/halo-dev/halo/issues/4519>, some extensions which are deletable cannot be recycled by GC. This PR provides an ability to watch scheme changes and recycles deletable extensions.

#### Which issue(s) this PR fixes:

Fixes https://github.com/halo-dev/halo/issues/4519

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

```release-note
修复因重启后部分可被回收的资源一直处于删除中的状态
```

application/src/main/java/run/halo/app/extension/gc/GcReconciler.java

@@ -10,6 +10,7 @@ import run.halo.app.extension.Extension;
 import run.halo.app.extension.ExtensionClient;
 import run.halo.app.extension.ExtensionConverter;
 import run.halo.app.extension.SchemeManager;
+import run.halo.app.extension.SchemeWatcherManager;
 import run.halo.app.extension.controller.Controller;
 import run.halo.app.extension.controller.ControllerBuilder;
 import run.halo.app.extension.controller.DefaultController;
@@ -29,12 +30,18 @@ class GcReconciler implements Reconciler<GcRequest> {
 
     private final SchemeManager schemeManager;
 
-    GcReconciler(ExtensionClient client, ExtensionStoreClient storeClient,
+    private final SchemeWatcherManager schemeWatcherManager;
-        ExtensionConverter converter, SchemeManager schemeManager) {
+
+    GcReconciler(ExtensionClient client,
+        ExtensionStoreClient storeClient,
+        ExtensionConverter converter,
+        SchemeManager schemeManager,
+        SchemeWatcherManager schemeWatcherManager) {
         this.client = client;
         this.storeClient = storeClient;
         this.converter = converter;
         this.schemeManager = schemeManager;
+        this.schemeWatcherManager = schemeWatcherManager;
     }
 
 
@@ -56,7 +63,7 @@ class GcReconciler implements Reconciler<GcRequest> {
     @Override
     public Controller setupWith(ControllerBuilder builder) {
         var queue = new DefaultQueue<GcRequest>(Instant::now, Duration.ofMillis(500));
-        var synchronizer = new GcSynchronizer(client, queue, schemeManager);
+        var synchronizer = new GcSynchronizer(client, queue, schemeManager, schemeWatcherManager);
         return new DefaultController<>(
             "garbage-collector-controller",
             this,

application/src/main/java/run/halo/app/extension/gc/GcSynchronizer.java

@@ -1,10 +1,14 @@
 package run.halo.app.extension.gc;
 
+import static run.halo.app.extension.Comparators.compareCreationTimestamp;
+
 import java.util.function.Predicate;
 import run.halo.app.extension.Extension;
 import run.halo.app.extension.ExtensionClient;
 import run.halo.app.extension.Scheme;
 import run.halo.app.extension.SchemeManager;
+import run.halo.app.extension.SchemeWatcherManager;
+import run.halo.app.extension.SchemeWatcherManager.SchemeRegistered;
 import run.halo.app.extension.Watcher;
 import run.halo.app.extension.controller.RequestQueue;
 import run.halo.app.extension.controller.Synchronizer;
@@ -13,10 +17,10 @@ class GcSynchronizer implements Synchronizer<GcRequest> {
 
     private final ExtensionClient client;
 
-    private final RequestQueue<GcRequest> queue;
-
     private final SchemeManager schemeManager;
 
+    private final SchemeWatcherManager schemeWatcherManager;
+
     private boolean disposed = false;
 
     private boolean started = false;
@@ -24,11 +28,11 @@ class GcSynchronizer implements Synchronizer<GcRequest> {
     private final Watcher watcher;
 
     GcSynchronizer(ExtensionClient client, RequestQueue<GcRequest> queue,
-        SchemeManager schemeManager) {
+        SchemeManager schemeManager, SchemeWatcherManager schemeWatcherManager) {
         this.client = client;
-        this.queue = queue;
         this.schemeManager = schemeManager;
         this.watcher = new GcWatcher(queue);
+        this.schemeWatcherManager = schemeWatcherManager;
     }
 
     @Override
@@ -51,10 +55,17 @@ class GcSynchronizer implements Synchronizer<GcRequest> {
             return;
         }
         this.started = true;
+        this.schemeWatcherManager.register(event -> {
+            if (event instanceof SchemeRegistered registeredEvent) {
+                var newScheme = registeredEvent.getNewScheme();
+                client.list(newScheme.type(), deleted(), compareCreationTimestamp(true))
+                    .forEach(watcher::onDelete);
+            }
+        });
         client.watch(watcher);
         schemeManager.schemes().stream()
             .map(Scheme::type)
-            .forEach(type -> client.list(type, deleted(), null)
+            .forEach(type -> client.list(type, deleted(), compareCreationTimestamp(true))
                 .forEach(watcher::onDelete));
     }
 

application/src/test/java/run/halo/app/extension/gc/GcSynchronizerTest.java

@@ -0,0 +1,52 @@
+package run.halo.app.extension.gc;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.verify;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import run.halo.app.extension.ExtensionClient;
+import run.halo.app.extension.SchemeManager;
+import run.halo.app.extension.SchemeWatcherManager;
+import run.halo.app.extension.SchemeWatcherManager.SchemeWatcher;
+
+@ExtendWith(MockitoExtension.class)
+class GcSynchronizerTest {
+
+    @Mock
+    ExtensionClient client;
+
+    @Mock
+    SchemeManager schemeManager;
+
+    @Mock
+    SchemeWatcherManager schemeWatcherManager;
+
+    @InjectMocks
+    GcSynchronizer synchronizer;
+
+    @Test
+    void shouldStartNormally() {
+        synchronizer.start();
+
+        assertFalse(synchronizer.isDisposed());
+        verify(schemeWatcherManager).register(any(SchemeWatcher.class));
+        verify(client).watch(isA(GcWatcher.class));
+        verify(schemeManager).schemes();
+    }
+
+    @Test
+    void shouldDisposeSuccessfully() {
+        assertFalse(synchronizer.isDisposed());
+
+        synchronizer.dispose();
+
+        assertTrue(synchronizer.isDisposed());
+    }
+}