package at.procon.eventhub.camel;

import at.procon.eventhub.dto.DataPackageType;
import at.procon.eventhub.dto.EventHubEventBatchDto;
import at.procon.eventhub.dto.EventHubEventDto;
import at.procon.eventhub.dto.EventHubPackageRequest;
import at.procon.eventhub.dto.ImportScopeDto;
import at.procon.eventhub.service.EventHubEventSorter;
import java.time.OffsetDateTime;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.springframework.stereotype.Component;

@Component
public class EventHubBatchBuildProcessor implements Processor {

    private final EventHubEventSorter sorter;

    public EventHubBatchBuildProcessor(EventHubEventSorter sorter) {
        this.sorter = sorter;
    }

    @Override
    public void process(Exchange exchange) {
        @SuppressWarnings("unchecked")
        List<EventHubEventDto> events = exchange.getMessage().getBody(List.class);
        List<EventHubEventDto> sortedEvents = sorter.sort(events);

        String aggregatePackageKey = exchange.getMessage().getHeader(EventHubHeaders.PACKAGE_KEY, String.class);
        String packageKey = aggregatePackageKey + ":CAMEL-" + exchange.getExchangeId();
        EventHubPackageRequest packageInfo = exchange.getMessage().getHeader(EventHubHeaders.PACKAGE_INFO, EventHubPackageRequest.class);
        if (packageInfo == null && !sortedEvents.isEmpty()) {
            packageInfo = sortedEvents.getFirst().packageInfo();
        }

        ImportScopeDto importScope = packageInfo == null ? null : packageInfo.importScope();
        OffsetDateTime occurredFrom = importScope != null && importScope.occurredFrom() != null
                ? importScope.occurredFrom()
                : sortedEvents.getFirst().occurredAt();
        OffsetDateTime occurredTo = importScope != null && importScope.occurredTo() != null
                ? importScope.occurredTo()
                : sortedEvents.getLast().occurredAt();

        Map<String, Object> metadata = new HashMap<>();
        metadata.put("camelRouteId", exchange.getFromRouteId());
        metadata.put("packageKey", packageKey);
        metadata.put("aggregatePackageKey", aggregatePackageKey);
        metadata.put("camelExchangeId", exchange.getExchangeId());
        metadata.put("eventCount", sortedEvents.size());
        metadata.put("receivedEventTypeCounts", eventTypeCounts(sortedEvents));
        if (packageInfo != null) {
            metadata.put("tenantKey", packageInfo.tenantKey());
            metadata.put("eventSource", packageInfo.eventSource().stableKey());
            metadata.put("sourceGroup", packageInfo.sourceGroup() == null ? null : packageInfo.sourceGroup().stableKey());
            metadata.put("importScope", packageInfo.importScope() == null ? null : packageInfo.importScope().stableKey());
            metadata.put("eventFamily", packageInfo.eventFamily());
            metadata.put("businessDate", packageInfo.businessDate());
            metadata.put("externalPackageId", packageInfo.externalPackageId());
        }

        exchange.getMessage().setBody(new EventHubEventBatchDto(
                packageKey,
                packageInfo,
                DataPackageType.CAMEL_BATCH,
                occurredFrom,
                occurredTo,
                sortedEvents,
                metadata
        ));
    }

    private Map<String, Integer> eventTypeCounts(List<EventHubEventDto> events) {
        Map<String, Integer> counts = new LinkedHashMap<>();
        for (EventHubEventDto event : events) {
            String domain = event.eventDomain() == null ? "UNKNOWN_DOMAIN" : event.eventDomain().name();
            String type = event.eventType() == null ? "UNKNOWN_EVENT" : event.eventType().name();
            counts.merge(domain + "/" + type, 1, Integer::sum);
        }
        return counts;
    }
}
