Repository: kestra-io/kestra
Issue: Creating a flow from JSON didn't use the custom label deserializer
Labels: bug

### Expected Behavior

Creating a flow from JSON must use the custom label deserializer otherwise labe as map is not possibe.

### Actual Behaviour

Using the following cURL example:

```
curl --location 'http://localhost:8080/api/v1/flows' \
--header 'Content-Type: application/json' \
--header 'Accept: text/json' \
--data '{
    "id": "hello-world-5",
    "namespace": "dev",
    "revision": 1,
    "labels":{
        "key": "value"
    },
    "tasks": [
        {
            "id": "hello",
            "type": "io.kestra.core.tasks.log.Log",
            "message": "Kestra team wishes you a great day! 👋"
        }
    ]
}'
```

When the flow is created it has one label `value=null`  as it reads the map as a list as the `ListOrMapOfLabelDeserializer` is not used.

### Steps To Reproduce

_No response_

### Environment Information

- Kestra Version:
- Operating System (OS / Docker / Kubernetes):
- Java Version (If not docker):


### Example flow

_No response_

Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/models/flows/Flow.java b/core/src/main/java/io/kestra/core/models/flows/Flow.java
index e29e7d1df6..a12b51a664 100644
--- a/core/src/main/java/io/kestra/core/models/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/models/flows/Flow.java
@@ -10,7 +10,6 @@
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.DeletedInterface;
 import io.kestra.core.models.Label;
-import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.listeners.Listener;
 import io.kestra.core.models.tasks.FlowableTask;
@@ -29,17 +28,16 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.*;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import javax.validation.ConstraintViolation;
 import javax.validation.ConstraintViolationException;
 import javax.validation.Valid;
 import javax.validation.constraints.*;
-import java.util.*;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 @SuperBuilder(toBuilder = true)
 @Getter
-@AllArgsConstructor
 @NoArgsConstructor
 @Introspected
 @ToString
@@ -64,7 +62,6 @@ public boolean hasIgnoreMarker(final AnnotatedMember m) {
     @Pattern(regexp = "[a-z0-9._-]+")
     String namespace;
 
-    @With
     @Min(value = 1)
     Integer revision;
 
@@ -75,7 +72,6 @@ public boolean hasIgnoreMarker(final AnnotatedMember m) {
     @Schema(implementation = Object.class, anyOf = {List.class, Map.class})
     List<Label> labels;
 
-
     @Valid
     List<Input<?>> inputs;
 
@@ -322,21 +318,9 @@ public String generateSource() {
     }
 
     public Flow toDeleted() {
-        return new Flow(
-            this.id,
-            this.namespace,
-            this.revision + 1,
-            this.description,
-            this.labels,
-            this.inputs,
-            this.variables,
-            this.tasks,
-            this.errors,
-            this.listeners,
-            this.triggers,
-            this.taskDefaults,
-            this.disabled,
-            true
-        );
+        return this.toBuilder()
+            .revision(this.revision + 1)
+            .deleted(true)
+            .build();
     }
 }
diff --git a/core/src/main/java/io/kestra/core/models/flows/FlowWithException.java b/core/src/main/java/io/kestra/core/models/flows/FlowWithException.java
index abf1c5a5d2..4e09794ba3 100644
--- a/core/src/main/java/io/kestra/core/models/flows/FlowWithException.java
+++ b/core/src/main/java/io/kestra/core/models/flows/FlowWithException.java
@@ -1,12 +1,14 @@
 package io.kestra.core.models.flows;
 
 import io.micronaut.core.annotation.Introspected;
-import lombok.*;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
 @SuperBuilder(toBuilder = true)
 @Getter
-@AllArgsConstructor
 @NoArgsConstructor
 @Introspected
 @ToString
diff --git a/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java b/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java
index 6c950f01c4..be9049b12a 100644
--- a/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java
+++ b/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java
@@ -2,20 +2,16 @@
 
 import io.kestra.core.services.FlowService;
 import io.micronaut.core.annotation.Introspected;
-import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.ToString;
 import lombok.experimental.SuperBuilder;
-import lombok.extern.slf4j.Slf4j;
 
 @SuperBuilder(toBuilder = true)
 @Getter
-@AllArgsConstructor
 @NoArgsConstructor
 @Introspected
 @ToString
-@Slf4j
 public class FlowWithSource extends Flow {
     String source;
 
diff --git a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
index 8f10e1f2ff..13defaef92 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
@@ -348,6 +348,10 @@ public FlowWithSource update(Flow flow, Flow previous, String flowSource, Flow f
 
     @SneakyThrows
     private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowSource) throws ConstraintViolationException {
+        if (flow instanceof FlowWithSource) {
+            flow = ((FlowWithSource) flow).toFlow();
+        }
+
         // flow exists, return it
         Optional<FlowWithSource> exists = this.findByIdWithSource(flow.getNamespace(), flow.getId());
         if (exists.isPresent() && exists.get().isUpdatable(flow, flowSource)) {
@@ -357,9 +361,9 @@ private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowS
         List<FlowWithSource> revisions = this.findRevisions(flow.getNamespace(), flow.getId());
 
         if (revisions.size() > 0) {
-            flow = flow.withRevision(revisions.get(revisions.size() - 1).getRevision() + 1);
+            flow = flow.toBuilder().revision(revisions.get(revisions.size() - 1).getRevision() + 1).build();
         } else {
-            flow = flow.withRevision(1);
+            flow = flow.toBuilder().revision(1).build();
         }
 
         Map<Field<Object>, Object> fields = this.jdbcRepository.persistFields(flow);
@@ -376,6 +380,10 @@ private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowS
     @SneakyThrows
     @Override
     public Flow delete(Flow flow) {
+        if (flow instanceof FlowWithSource) {
+            flow = ((FlowWithSource) flow).toFlow();
+        }
+
         Optional<Flow> revision = this.findById(flow.getNamespace(), flow.getId(), Optional.of(flow.getRevision()));
         if (revision.isEmpty()) {
             throw new IllegalStateException("Flow " + flow.getId() + " doesn't exists");
diff --git a/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java b/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java
index 31abf66595..9e3e49e8dc 100644
--- a/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java
+++ b/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java
@@ -195,6 +195,10 @@ public FlowWithSource update(Flow flow, Flow previous, String flowSource, Flow f
     }
 
     private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowSource) throws ConstraintViolationException {
+        if (flow instanceof FlowWithSource) {
+            flow = ((FlowWithSource) flow).toFlow();
+        }
+
         // flow exists, return it
         Optional<Flow> exists = this.findById(flow.getNamespace(), flow.getId());
         Optional<String> existsSource = this.findSourceById(flow.getNamespace(), flow.getId());
@@ -205,9 +209,9 @@ private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowS
         List<FlowWithSource> revisions = this.findRevisions(flow.getNamespace(), flow.getId());
 
         if (revisions.size() > 0) {
-            flow = flow.withRevision(revisions.get(revisions.size() - 1).getRevision() + 1);
+            flow = flow.toBuilder().revision(revisions.get(revisions.size() - 1).getRevision() + 1).build();
         } else {
-            flow = flow.withRevision(1);
+            flow = flow.toBuilder().revision(1).build();
         }
 
         this.flows.put(flowId(flow), flow);
@@ -222,6 +226,10 @@ private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowS
 
     @Override
     public Flow delete(Flow flow) {
+        if (flow instanceof FlowWithSource) {
+            flow = ((FlowWithSource) flow).toFlow();
+        }
+
         if (this.findById(flow.getNamespace(), flow.getId(), Optional.of(flow.getRevision())).isEmpty()) {
             throw new IllegalStateException("Flow " + flow.getId() + " doesn't exists");
         }
@@ -232,8 +240,9 @@ public Flow delete(Flow flow) {
         this.flows.remove(flowId(deleted));
         this.revisions.put(deleted.uid(), deleted);
 
+        Flow finalFlow = flow;
         ListUtils.emptyOnNull(flow.getTriggers())
-            .forEach(abstractTrigger -> triggerQueue.delete(Trigger.of(flow, abstractTrigger)));
+            .forEach(abstractTrigger -> triggerQueue.delete(Trigger.of(finalFlow, abstractTrigger)));
 
         eventPublisher.publishEvent(new CrudEvent<>(flow, CrudEventType.DELETE));
 


Test patch:
diff --git a/core/src/test/java/io/kestra/core/services/FlowTopologyServiceTest.java b/core/src/test/java/io/kestra/core/services/FlowTopologyServiceTest.java
index 70e7042aec..553c383dab 100644
--- a/core/src/test/java/io/kestra/core/services/FlowTopologyServiceTest.java
+++ b/core/src/test/java/io/kestra/core/services/FlowTopologyServiceTest.java
@@ -173,14 +173,14 @@ public void multipleCondition() {
 
     @Test
     public void self1() {
-        Flow flow = parse("flows/valids/trigger-multiplecondition-listener.yaml").withRevision(1);
+        Flow flow = parse("flows/valids/trigger-multiplecondition-listener.yaml").toBuilder().revision(1).build();
 
         assertThat(flowTopologyService.isChild(flow, flow), nullValue());
     }
 
     @Test
     public void self() {
-        Flow flow = parse("flows/valids/trigger-flow-listener.yaml").withRevision(1);
+        Flow flow = parse("flows/valids/trigger-flow-listener.yaml").toBuilder().revision(1).build();
 
         assertThat(flowTopologyService.isChild(flow, flow), nullValue());
     }
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java
index 40b9b336f7..e2f8c18722 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java
@@ -10,6 +10,7 @@
 import io.kestra.core.models.hierarchies.FlowGraph;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.validations.ValidateConstraintViolation;
+import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.serializers.YamlFlowParser;
 import io.kestra.core.tasks.debugs.Return;
 import io.kestra.core.tasks.flows.Sequential;
@@ -251,7 +252,18 @@ void createFlow() {
         Flow get = parseFlow(client.toBlocking().retrieve(HttpRequest.GET("/api/v1/flows/" + flow.getNamespace() + "/" + flow.getId()), String.class));
         assertThat(get.getId(), is(flow.getId()));
         assertThat(get.getInputs().get(0).getName(), is("a"));
+    }
+
+    @Test
+    void createFlowWithJsonLabels() {
+        Map<String, Object> flow = JacksonMapper.toMap(generateFlow("io.kestra.unittest", "a"));
+        flow.put("labels", Map.of("a", "b"));
+
+        Flow result = parseFlow(client.toBlocking().retrieve(POST("/api/v1/flows", flow), String.class));
 
+        assertThat(result.getId(), is(flow.get("id")));
+        assertThat(result.getLabels().get(0).key(), is("a"));
+        assertThat(result.getLabels().get(0).value(), is("b"));
     }
 
     @Test