@VisibleForTesting
    ImmutableList<EventWithContext> eventsFromAggregationResult(EventFactory eventFactory, AggregationEventProcessorParameters parameters, AggregationResult result)
            throws EventProcessorException {
        final ImmutableList.Builder<EventWithContext> eventsWithContext = ImmutableList.builder();
        final Set<String> sourceStreams = eventStreamService.buildEventSourceStreams(getStreams(parameters),
                result.sourceStreams());

        for (final AggregationKeyResult keyResult : result.keyResults()) {
            if (!satisfiesConditions(keyResult)) {
                LOG.debug("Skipping result <{}> because the conditions <{}> don't match", keyResult, config.conditions());
                continue;
            }

            final String keyString = String.join("|", keyResult.key());
            final String eventMessage = createEventMessageString(keyString, keyResult);

            // Extract event time and range from the key result or use query time range as fallback.
            // These can be different, e.g. during catch up processing.
            final DateTime eventTime = keyResult.timestamp().orElse(result.effectiveTimerange().to());
            final Event event = eventFactory.createEvent(eventDefinition, eventTime, eventMessage);
            // The keyResult timestamp is set to the end of the range
            event.setTimerangeStart(keyResult.timestamp().map(t -> t.minus(config.searchWithinMs())).orElse(parameters.timerange().getFrom()));
            event.setTimerangeEnd(keyResult.timestamp().orElse(parameters.timerange().getTo()));

            event.setReplayInfo(EventReplayInfo.builder()
                    .timerangeStart(event.getTimerangeStart())
                    .timerangeEnd(event.getTimerangeEnd())
                    .query(config.query())
                    .streams(sourceStreams)
                    .filters(config.filters())
                    .build());
            sourceStreams.forEach(event::addSourceStream);

            final Map<String, Object> fields = new HashMap<>();

            // Each group value will be a separate field in the message to make it usable as event fields.
            //
            // Example result:
            //   groupBy=["application_name", "username"]
            //   result-key=["sshd", "jane"]
            //
            // Message fields:
            //   application_name=sshd
            //   username=jane
            for (int i = 0; i < config.groupBy().size(); i++) {
                try {
                    fields.put(config.groupBy().get(i), keyResult.key().get(i));
                } catch (IndexOutOfBoundsException e) {
                    throw new EventProcessorException(
                            "Couldn't create events for: " + eventDefinition.title() + " (possibly due to non-existing grouping fields)",
                            false, eventDefinition.id(), eventDefinition, e);
                }
            }

            // Group By fields need to be saved on the event so they are available to the subsequent notification events
            event.setGroupByFields(fields.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().toString())));

            // The field name for the series value is composed of the series function and field. We don't take the
            // series ID into account because it would be very hard to use for the user. That means a series with
            // the same function and field but different ID would overwrite a previous one.
            // This shouldn't be a problem though, because the same function and field will always compute the same
            // value.
            //
            // Examples:
            //   aggregation_value_count_source=42
            //   aggregation_value_card_anonid=23
            for (AggregationSeriesValue seriesValue : keyResult.seriesValues()) {
                final String function = seriesValue.series().type().toLowerCase(Locale.ROOT);
                final Optional<String> field = fieldFromSeries(seriesValue.series());

                final String fieldName = field.map(f -> String.format(Locale.ROOT, "aggregation_value_%s_%s", function, f))
                        .orElseGet(() -> String.format(Locale.ROOT, "aggregation_value_%s", function));

                fields.put(fieldName, seriesValue.value());
            }

            // This is the concatenated key value
            fields.put("aggregation_key", keyString);

            // TODO: Can we find a useful source value?
            final Message message = messageFactory.createMessage(eventMessage, "", result.effectiveTimerange().to());
            message.addFields(fields);

            // Ask any event query modifier for its state and collect it into the event modifier state
            final Map<String, Object> eventModifierState = eventQueryModifiers.stream()
                    .flatMap(modifier -> modifier.eventModifierData(result.additionalResults()).entrySet().stream())
                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

            LOG.debug("Creating event {}/{} - {} {} ({})", eventDefinition.title(), eventDefinition.id(), keyResult.key(), seriesString(keyResult), fields);

            eventsWithContext.add(EventWithContext.builder()
                    .event(event)
                    .messageContext(message)
                    .eventModifierState(eventModifierState)
                    .build());
        }

        return eventsWithContext.build();
    }
@Test
    public void testEventsFromAggregationResultWithEmptyResultAndNoConfiguredStreamsUsesAllStreamsAsSourceStreams() throws EventProcessorException {
        final DateTime now = DateTime.now(DateTimeZone.UTC);
        final AbsoluteRange timerange = AbsoluteRange.create(now.minusHours(1), now.minusHours(1).plusMillis(SEARCH_WINDOW_MS));

        // We expect to get the end of the aggregation timerange as event time
        final TestEvent event1 = new TestEvent(timerange.to());
        final TestEvent event2 = new TestEvent(timerange.to());
        when(eventFactory.createEvent(any(EventDefinition.class), any(DateTime.class), anyString()))
                .thenReturn(event1)  // first invocation return value
                .thenReturn(event2); // second invocation return value

        final EventDefinitionDto eventDefinitionDto = buildEventDefinitionDto(ImmutableSet.of(), ImmutableList.of(), null, emptyList());
        final AggregationEventProcessorParameters parameters = AggregationEventProcessorParameters.builder()
                .timerange(timerange)
                .build();

        final AggregationEventProcessor eventProcessor = new AggregationEventProcessor(eventDefinitionDto, searchFactory, eventProcessorDependencyCheck, stateService, moreSearch,
                eventStreamService, messages, notificationService, permittedStreams, Set.of(), messageFactory);
        final AggregationResult result = buildAggregationResult(timerange, timerange.to(), ImmutableList.of("one", "two"));
        final ImmutableList<EventWithContext> eventsWithContext = eventProcessor.eventsFromAggregationResult(eventFactory, parameters, result);

        assertThat(eventsWithContext).hasSize(1);

        assertThat(eventsWithContext.get(0)).satisfies(eventWithContext -> {
            final Event event = eventWithContext.event();

            assertThat(event.getId()).isEqualTo(event1.getId());
            assertThat(event.getMessage()).isEqualTo(event1.getMessage());
            assertThat(event.getEventTimestamp()).isEqualTo(timerange.to());
            assertThat(event.getTimerangeStart()).isEqualTo(timerange.from());
            assertThat(event.getTimerangeEnd()).isEqualTo(timerange.to());
            // Must contain all existing streams but the default event streams!
            assertThat(event.getSourceStreams()).containsOnly(
                    "stream-1",
                    "stream-2",
                    "stream-3",
                    StreamImpl.DEFAULT_STREAM_ID
            );

            final Message message = eventWithContext.messageContext().orElse(null);

            assertThat(message).isNotNull();
            assertThat(message.getField("group_field_one")).isEqualTo("one");
            assertThat(message.getField("group_field_two")).isEqualTo("two");
            assertThat(message.getField("aggregation_key")).isEqualTo("one|two");
            assertThat(message.getField("aggregation_value_count")).isEqualTo(0.0d);
        });
    }