public void forEachEnabledFilterGroupedByStream(Consumer<GroupByStreamResult> consumer) {
        // Group all enabled filters by stream ID
        collection.aggregate(List.of(
                Aggregates.match(eq(FIELD_STATUS, StreamDestinationFilterRuleDTO.Status.ENABLED)),
                Aggregates.group("$" + FIELD_STREAM_ID, List.of(
                        Accumulators.push("filters", "$$ROOT")
                ))
        ), GroupByStreamResult.class).forEach(consumer);
    }
@Test
    @MongoDBFixtures("StreamDestinationFilterServiceTest-2024-07-01-1.json")
    void forEachEnabled() {
        final var resultBuilder = ImmutableMultimap.<String, String>builder();

        service.forEachEnabledFilterGroupedByStream(filter -> filter.filters().forEach(f -> resultBuilder.put(filter.streamId(), f.title())));

        final var result = resultBuilder.build();

        assertThat(result.keySet()).containsExactlyInAnyOrder("54e3deadbeefdeadbeef1000", "54e3deadbeefdeadbeef2000");
        assertThat(result.get("54e3deadbeefdeadbeef1000")).containsExactlyInAnyOrder("Test Filter 1", "Test Filter 3");
        assertThat(result.get("54e3deadbeefdeadbeef2000")).containsExactlyInAnyOrder("Test Filter 4");
    }