@SuppressWarnings("unchecked")
    @Override
    public Output run(RunContext runContext) throws Exception {
        Logger logger = runContext.logger();
        String renderedNamespace = runContext.render(this.namespace);

        final Namespace namespace = runContext.storage().namespace(renderedNamespace);

        List<String> renderedFiles;
        if (files instanceof String filesString) {
            renderedFiles = List.of(runContext.render(filesString));
        } else if (files instanceof List<?> filesList) {
            renderedFiles = runContext.render((List<String>) filesList);
        } else {
            throw new IllegalArgumentException("Files must be a String or a list of String");
        }

        List<NamespaceFile> matched = namespace.findAllFilesMatching(PathMatcherPredicate.matches(renderedFiles));
        long count = matched
            .stream()
            .map(Rethrow.throwFunction(file -> {
                if (namespace.delete(file)) {
                    logger.debug(String.format("Deleted %s", (file.path())));
                    return true;
                }
                return false;
            }))
            .filter(Boolean::booleanValue)
            .count();

        runContext.metric(Counter.of("deleted", count));
        return Output.builder().build();
    }
@Test
    void shouldDeleteNamespaceFilesForMatchingExpression() throws Exception {
        // Given
        String namespaceId = "io.kestra." + IdUtils.create();

        DeleteFiles deleteFiles = DeleteFiles.builder()
            .id(DeleteFiles.class.getSimpleName())
            .type(DeleteFiles.class.getName())
            .files(List.of("**test1*"))
            .namespace("{{ inputs.namespace }}")
            .build();

        final RunContext runContext = TestsUtils.mockRunContext(this.runContextFactory, deleteFiles, Map.of("namespace", namespaceId));
        final Namespace namespace = runContext.storage().namespace(namespaceId);

        namespace.putFile(Path.of("/a/b/test1.txt"), new ByteArrayInputStream("1".getBytes(StandardCharsets.UTF_8)));
        namespace.putFile(Path.of("/a/b/test2.txt"), new ByteArrayInputStream("2".getBytes(StandardCharsets.UTF_8)));

        assertThat(namespace.all("/a/b/", false).size(), is(2));

        // When
        assertThat(deleteFiles.run(runContext), notNullValue());

        // Then
        assertThat(namespace.all("/a/b/", false).size(), is(1));
    }