@Override
    public PurgeExecutions.Output run(RunContext runContext) throws Exception {
        ExecutionService executionService = ((DefaultRunContext)runContext).getApplicationContext().getBean(ExecutionService.class);
        FlowService flowService = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowService.class);

        // validate that this namespace is authorized on the target namespace / all namespaces
        var flowInfo = runContext.flowInfo();
        if (namespace == null){
            flowService.checkAllowedAllNamespaces(flowInfo.tenantId(), flowInfo.tenantId(), flowInfo.namespace());
        } else if (!runContext.render(namespace).equals(flowInfo.namespace())) {
            flowService.checkAllowedNamespace(flowInfo.tenantId(), runContext.render(namespace), flowInfo.tenantId(), flowInfo.namespace());
        }

        ExecutionService.PurgeResult purgeResult = executionService.purge(
            purgeExecution,
            purgeLog,
            purgeMetric,
            purgeStorage,
            flowInfo.tenantId(),
            runContext.render(namespace),
            runContext.render(flowId),
            startDate != null ? ZonedDateTime.parse(runContext.render(startDate)) : null,
            ZonedDateTime.parse(runContext.render(endDate)),
            states
        );

        return Output.builder()
            .executionsCount(purgeResult.getExecutionsCount())
            .logsCount(purgeResult.getLogsCount())
            .storagesCount(purgeResult.getStoragesCount())
            .metricsCount(purgeResult.getMetricsCount())
            .build();
    }
@Test
    void deleted() throws Exception {
        String namespace = "deleted.namespace";
        String flowId = "deleted-flow-id";

        // create an execution to delete
        var execution = Execution.builder()
            .namespace(namespace)
            .flowId(flowId)
            .id(IdUtils.create())
            .state(new State().withState(State.Type.SUCCESS))
            .build();
        executionRepository.save(execution);
        executionRepository.delete(execution);

        var purge = PurgeExecutions.builder()
            .namespace(namespace)
            .flowId(flowId)
            .endDate(ZonedDateTime.now().plusMinutes(1).format(DateTimeFormatter.ISO_ZONED_DATE_TIME))
            .build();
        var runContext = runContextFactory.of(Map.of("flow", Map.of("namespace", namespace, "id", flowId)));
        var output = purge.run(runContext);

        assertThat(output.getExecutionsCount(), is(1));
    }