package at.procon.eventhub.service;

import at.procon.eventhub.dto.EventHubEventBatchDto;
import at.procon.eventhub.dto.EventHubEventDto;
import at.procon.eventhub.dto.EventHubPackageRequest;
import at.procon.eventhub.dto.EventHubPackageResult;
import at.procon.eventhub.dto.EventSourceDto;
import at.procon.eventhub.persistence.DataPackageRepository;
import at.procon.eventhub.persistence.EventRepository;
import at.procon.eventhub.persistence.EventSourceRepository;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;

@Service
public class EventHubIngestionService {

    private static final Logger log = LoggerFactory.getLogger(EventHubIngestionService.class);

    private final EventSourceRepository eventSourceRepository;
    private final DataPackageRepository dataPackageRepository;
    private final EventRepository eventRepository;
    private final EventHubEventValidator validator;
    private final EventHubEventSorter sorter;
    private final TransactionTemplate transactionTemplate;

    public EventHubIngestionService(
            EventSourceRepository eventSourceRepository,
            DataPackageRepository dataPackageRepository,
            EventRepository eventRepository,
            EventHubEventValidator validator,
            EventHubEventSorter sorter,
            PlatformTransactionManager transactionManager
    ) {
        this.eventSourceRepository = eventSourceRepository;
        this.dataPackageRepository = dataPackageRepository;
        this.eventRepository = eventRepository;
        this.validator = validator;
        this.sorter = sorter;
        this.transactionTemplate = new TransactionTemplate(transactionManager);
    }

    public EventHubPackageResult ingest(EventHubEventBatchDto batch) {
        if (batch == null || batch.events().isEmpty()) {
            return new EventHubPackageResult(null, batch == null ? null : batch.packageKey(), 0, 0);
        }
        long startedAtNanos = System.nanoTime();

        EventHubPackageRequest packageInfo = batch.packageInfo();
        if (packageInfo == null) {
            packageInfo = batch.events().getFirst().packageInfo();
        }
        if (packageInfo == null || packageInfo.eventSource() == null) {
            throw new IllegalArgumentException("packageInfo.eventSource must be set before ingestion");
        }

        EventSourceDto eventSource = packageInfo.eventSource();
        String tenantKey = packageInfo.tenantKey();
        int eventSourceId = eventSourceRepository.resolveSourceId(packageInfo.tenantKey(), eventSource);
        List<EventHubEventDto> sortedEvents = sorter.sort(batch.events());
        sortedEvents.forEach(validator::validate);

        UUID packageId = dataPackageRepository.createPackage(
                eventSourceId,
                batch.packageKey(),
                packageInfo,
                batch.packageType(),
                batch.occurredFrom(),
                batch.occurredTo(),
                batch.metadata()
        );

        try {
            Integer insertedCount = transactionTemplate.execute(status ->
                    eventRepository.batchInsert(packageId, tenantKey, eventSourceId, sortedEvents)
            );
            if (insertedCount == null) {
                throw new IllegalStateException("EventHub batch insert returned no result for package " + batch.packageKey());
            }
            dataPackageRepository.markImported(packageId, insertedCount);
            long elapsedMs = Math.max(1L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
            long receivedPerSecond = Math.max(1L, Math.round(sortedEvents.size() * 1000.0 / elapsedMs));
            log.info("Imported EventHub acquisition package packageId={} packageKey={} source={} receivedCount={} insertedCount={} elapsedMs={} receivedPerSecond={} byType={}",
                    packageId, batch.packageKey(), eventSource.stableKey(), sortedEvents.size(), insertedCount, elapsedMs, receivedPerSecond, eventTypeCounts(sortedEvents));
            return new EventHubPackageResult(packageId, batch.packageKey(), sortedEvents.size(), insertedCount);
        } catch (RuntimeException ex) {
            try {
                dataPackageRepository.markFailed(packageId, ex.getMessage());
            } catch (RuntimeException markFailedEx) {
                ex.addSuppressed(markFailedEx);
                log.warn("Failed to mark EventHub acquisition package as failed packageId={} packageKey={}",
                        packageId, batch.packageKey(), markFailedEx);
            }
            throw ex;
        }
    }

    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;
    }
}
