@ExecuteOn(TaskExecutors.IO)
    @Post(uri = "validate", consumes = MediaType.APPLICATION_YAML)
    @Operation(tags = {"Flows"}, summary = "Validate a list of flows")
    public List<ValidateConstraintViolation> validateFlows(
        @Parameter(description = "A list of flows") @Body String flows
    ) {
        AtomicInteger index = new AtomicInteger(0);
        return Stream
            .of(flows.split("\\n+---\\n*?"))
            .map(flow -> {
                ValidateConstraintViolation.ValidateConstraintViolationBuilder<?, ?> validateConstraintViolationBuilder = ValidateConstraintViolation.builder();
                validateConstraintViolationBuilder.index(index.getAndIncrement());

                try {
                    Flow flowParse = yamlFlowParser.parse(flow, Flow.class);
                    Integer sentRevision = flowParse.getRevision();
                    if (sentRevision != null) {
                        Integer lastRevision = Optional.ofNullable(flowRepository.lastRevision(tenantService.resolveTenant(), flowParse.getNamespace(), flowParse.getId()))
                            .orElse(0);
                        validateConstraintViolationBuilder.outdated(!sentRevision.equals(lastRevision + 1));
                    }

                    validateConstraintViolationBuilder.deprecationPaths(flowService.deprecationPaths(flowParse));
                    List<String> warnings = new ArrayList<>();
                    warnings.addAll(flowService.warnings(flowParse));
                    warnings.addAll(flowService.relocations(flow).stream().map(relocation -> relocation.from() + " is replaced by " + relocation.to()).toList());
                    validateConstraintViolationBuilder.warnings(warnings);
                    validateConstraintViolationBuilder.flow(flowParse.getId());
                    validateConstraintViolationBuilder.namespace(flowParse.getNamespace());

                    modelValidator.validate(pluginDefaultService.injectDefaults(flowParse));
                } catch (ConstraintViolationException e) {
                    validateConstraintViolationBuilder.constraints(e.getMessage());
                } catch (RuntimeException re) {
                    // In case of any error, we add a validation violation so the error is displayed in the UI.
                    // We may change that by throwing an internal error and handle it in the UI, but this should not occur except for rare cases
                    // in dev like incompatible plugin versions.
                    log.error("Unable to validate the flow", re);
                    validateConstraintViolationBuilder.constraints("Unable to validate the flow: " + re.getMessage());
                }

                return validateConstraintViolationBuilder.build();
            })
            .collect(Collectors.toList());
    }
@Test
    void validateFlows() throws IOException {
        URL resource = TestsUtils.class.getClassLoader().getResource("flows/validateMultipleValidFlows.yaml");
        String flow = Files.readString(Path.of(Objects.requireNonNull(resource).getPath()), Charset.defaultCharset());

        String firstFlowSource = flow.split("(?m)^---")[0];
        Flow firstFlow = parseFlow(firstFlowSource);
        jdbcFlowRepository.create(firstFlow, firstFlowSource, firstFlow);

        HttpResponse<List<ValidateConstraintViolation>> response = client.toBlocking().exchange(POST("/api/v1/flows/validate", flow).contentType(MediaType.APPLICATION_YAML), Argument.listOf(ValidateConstraintViolation.class));

        List<ValidateConstraintViolation> body = response.body();
        assertThat(body.size(), is(2));
        // We don't send any revision while the flow already exists so it's outdated
        assertThat(body.getFirst().isOutdated(), is(true));
        assertThat(body.getFirst().getDeprecationPaths(), hasSize(3));
        assertThat(body.getFirst().getDeprecationPaths(), containsInAnyOrder("tasks[1]", "tasks[1].additionalProperty", "listeners"));
        assertThat(body.getFirst().getWarnings().size(), is(1));
        assertThat(body.getFirst().getWarnings().getFirst(), containsString("The system namespace is reserved for background workflows"));
        assertThat(body.get(1).isOutdated(), is(false));
        assertThat(body.get(1).getDeprecationPaths(), containsInAnyOrder("tasks[0]", "tasks[1]"));
        assertThat(body, everyItem(
            Matchers.hasProperty("constraints", is(nullValue()))
        ));

        resource = TestsUtils.class.getClassLoader().getResource("flows/validateMultipleInvalidFlows.yaml");
        flow = Files.readString(Path.of(Objects.requireNonNull(resource).getPath()), Charset.defaultCharset());

        response = client.toBlocking().exchange(POST("/api/v1/flows/validate", flow).contentType(MediaType.APPLICATION_YAML), Argument.listOf(ValidateConstraintViolation.class));

        body = response.body();
        assertThat(body.size(), is(2));
        assertThat(body.getFirst().getConstraints(), containsString("Unrecognized field \"unknownProp\""));
        assertThat(body.get(1).getConstraints(), containsString("Invalid type: io.kestra.plugin.core.debug.UnknownTask"));
    }