package at.procon.eventhub.importing.extraction;

import at.procon.eventhub.config.EventHubProperties;
import at.procon.eventhub.dto.AcquisitionStrategy;
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.EventHubPackageResult;
import at.procon.eventhub.dto.EventSourceDto;
import at.procon.eventhub.dto.ImportCursorStateDto;
import at.procon.eventhub.dto.ImportScopeDto;
import at.procon.eventhub.importing.ExtractionBatchResult;
import at.procon.eventhub.importing.ImportPlanItemDto;
import at.procon.eventhub.importing.ImportRunRequest;
import at.procon.eventhub.importing.ImportTimeChunkDto;
import at.procon.eventhub.importing.persistence.ImportCursorRepository;
import at.procon.eventhub.service.EventHubIngestionService;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import org.apache.camel.ProducerTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.util.StreamUtils;

public abstract class AbstractJdbcExtractionBatchExecutor<R extends ImportRunRequest, B extends ExtractionBatchResult>
        implements ExtractionBatchExecutor<R, B> {

    private static final Logger log = LoggerFactory.getLogger(AbstractJdbcExtractionBatchExecutor.class);
    private static final int EVENT_EXTRACTION_PROGRESS_LOG_INTERVAL = 5000;

    private final NamedParameterJdbcTemplate jdbcTemplate;
    private final EventHubIngestionService ingestionService;
    private final ProducerTemplate producerTemplate;
    private final EventHubProperties eventHubProperties;
    private final ResourceLoader resourceLoader;
    private final ImportCursorRepository importCursorRepository;

    protected AbstractJdbcExtractionBatchExecutor(
            NamedParameterJdbcTemplate jdbcTemplate,
            EventHubIngestionService ingestionService,
            ProducerTemplate producerTemplate,
            EventHubProperties eventHubProperties,
            ResourceLoader resourceLoader,
            ImportCursorRepository importCursorRepository
    ) {
        this.jdbcTemplate = jdbcTemplate;
        this.ingestionService = ingestionService;
        this.producerTemplate = producerTemplate;
        this.eventHubProperties = eventHubProperties;
        this.resourceLoader = resourceLoader;
        this.importCursorRepository = importCursorRepository;
    }

    @Override
    public B execute(
            UUID importRunId,
            UUID packageId,
            int eventSourceId,
            R request,
            ImportPlanItemDto planItem,
            ImportTimeChunkDto chunk
    ) {
        ExtractionDefinition<R> definition = findDefinition(planItem.extractionCode())
                .orElseThrow(() -> new IllegalArgumentException("No extraction definition for " + planItem.extractionCode()));

        ImportScopeDto chunkScope = chunkScope(request.importScope(), chunk);
        EventHubPackageRequest packageInfo = packageInfo(importRunId, request, planItem, chunk, chunkScope);
        ExtractionContext<R> context = new ExtractionContext<>(
                importRunId,
                packageId,
                eventSourceId,
                request,
                planItem,
                chunk,
                packageInfo.eventSource(),
                packageInfo
        );

        ImportCursorStateDto cursor = findCursor(eventSourceId, request, planItem);

        Map<String, Object> params = parameters(request, chunkScope, cursor);
        String sql = loadSql(definition.sqlResource());
        ExtractedEventStats stats = new ExtractedEventStats();
        List<EventHubEventDto> pendingEvents = new ArrayList<>(jdbcPersistBatchSize());
        log.info("Reading EventHub events provider={} tenant={} importRunId={} packageId={} extractionCode={} sourceKind={} chunk={} occurredFrom={} occurredTo={} ingestMode={}",
                providerPackagePrefix(), request.tenantKey(), importRunId, packageId, planItem.extractionCode(), planItem.sourceKind(),
                chunk.sequence(), chunk.occurredFrom(), chunk.occurredTo(), jdbcExtractionIngestMode());
        jdbcTemplate.query(sql, params, rs -> {
            EventHubEventDto event = definition.rowMapper().map(rs, stats.eventsMapped(), context);
            pendingEvents.add(event);
            stats.accept(event);
            if (pendingEvents.size() >= jdbcPersistBatchSize()) {
                flushPersistBatch(request, importRunId, packageId, planItem, chunk, packageInfo, pendingEvents, stats);
            }
            if (stats.eventsMapped() % EVENT_EXTRACTION_PROGRESS_LOG_INTERVAL == 0) {
                logEventExtractionProgress(request, importRunId, packageId, planItem, chunk, stats);
            }
        });
        flushPersistBatch(request, importRunId, packageId, planItem, chunk, packageInfo, pendingEvents, stats);
        logEventExtractionFinished(request, importRunId, packageId, planItem, chunk, stats);

        return resultFor(packageId, planItem, chunk, cursor, stats);
    }

    protected abstract Optional<ExtractionDefinition<R>> findDefinition(String code);

    protected abstract EventSourceDto eventSourceFor(R request, ImportPlanItemDto planItem);

    protected abstract B resultFor(
            UUID packageId,
            ImportPlanItemDto planItem,
            ImportTimeChunkDto chunk,
            ImportCursorStateDto cursor,
            ExtractedEventStats stats
    );

    protected String providerPackagePrefix() {
        return "SOURCE";
    }

    protected int jdbcPersistBatchSize() {
        return Math.max(1, eventHubProperties.getBatch().getCompletionSize());
    }

    protected EventHubProperties.JdbcExtractionIngestMode jdbcExtractionIngestMode() {
        return eventHubProperties.getTachograph().getJdbcExtractionIngestMode();
    }

    protected int eventsInsertedOrSubmitted(ExtractedEventStats stats) {
        return stats.eventsInserted();
    }

    protected String camelBatchPersistUri() {
        return "direct:eventhub-batch-persist-input";
    }

    private void flushPersistBatch(
            R request,
            UUID importRunId,
            UUID extractionPackageId,
            ImportPlanItemDto planItem,
            ImportTimeChunkDto chunk,
            EventHubPackageRequest packageInfo,
            List<EventHubEventDto> pendingEvents,
            ExtractedEventStats stats
    ) {
        if (pendingEvents.isEmpty()) {
            return;
        }
        int batchNo = stats.nextPersistBatchNo();
        List<EventHubEventDto> eventsToPersist = List.copyOf(pendingEvents);
        pendingEvents.clear();

        EventHubEventBatchDto batch = new EventHubEventBatchDto(
                packageInfo.externalPackageId() + ":JDBC-" + batchNo,
                packageInfo,
                DataPackageType.DB_EXTRACT,
                occurredFrom(eventsToPersist, chunk),
                occurredTo(eventsToPersist, chunk),
                eventsToPersist,
                persistBatchMetadata(request, importRunId, extractionPackageId, planItem, chunk, batchNo, eventsToPersist)
        );
        EventHubPackageResult result = persistBatch(batch);
        stats.acceptPersistResult(result);
        log.info("Persisted EventHub extraction batch provider={} tenant={} importRunId={} extractionPackageId={} extractionCode={} sourceKind={} chunk={} batchNo={} received={} inserted={}",
                providerPackagePrefix(), request.tenantKey(), importRunId, extractionPackageId, planItem.extractionCode(), planItem.sourceKind(),
                chunk.sequence(), batchNo, result.receivedCount(), result.insertedCount());
    }

    private EventHubPackageResult persistBatch(EventHubEventBatchDto batch) {
        if (jdbcExtractionIngestMode() == EventHubProperties.JdbcExtractionIngestMode.CAMEL_ROUTE) {
            return producerTemplate.requestBody(camelBatchPersistUri(), batch, EventHubPackageResult.class);
        }
        return ingestionService.ingest(batch);
    }

    private Map<String, Object> persistBatchMetadata(
            R request,
            UUID importRunId,
            UUID extractionPackageId,
            ImportPlanItemDto planItem,
            ImportTimeChunkDto chunk,
            int batchNo,
            List<EventHubEventDto> events
    ) {
        Map<String, Object> metadata = new LinkedHashMap<>();
        metadata.put("ingestMode", jdbcExtractionIngestMode().name());
        metadata.put("importRunId", importRunId.toString());
        metadata.put("extractionPackageId", extractionPackageId.toString());
        metadata.put("tenantKey", request.tenantKey());
        metadata.put("mode", request.mode().name());
        metadata.put("acquisitionStrategy", request.acquisitionStrategy().name());
        metadata.put("eventFamily", planItem.eventFamily().name());
        metadata.put("sourceKind", planItem.sourceKind());
        metadata.put("extractionCode", planItem.extractionCode());
        metadata.put("entityAxis", planItem.entityAxis());
        metadata.put("chunkSequence", chunk.sequence());
        metadata.put("chunkOccurredFrom", chunk.occurredFrom() == null ? null : chunk.occurredFrom().toString());
        metadata.put("chunkOccurredTo", chunk.occurredTo() == null ? null : chunk.occurredTo().toString());
        metadata.put("batchNo", batchNo);
        metadata.put("receivedEventCount", events.size());
        metadata.put("sourcePackageRefPolicy", "Original source package is preserved per acquired event.");
        return metadata;
    }

    private OffsetDateTime occurredFrom(List<EventHubEventDto> events, ImportTimeChunkDto chunk) {
        if (chunk.occurredFrom() != null) {
            return chunk.occurredFrom();
        }
        return events.stream()
                .map(EventHubEventDto::occurredAt)
                .filter(value -> value != null)
                .min(OffsetDateTime::compareTo)
                .orElse(null);
    }

    private OffsetDateTime occurredTo(List<EventHubEventDto> events, ImportTimeChunkDto chunk) {
        if (chunk.occurredTo() != null) {
            return chunk.occurredTo();
        }
        return events.stream()
                .map(EventHubEventDto::occurredAt)
                .filter(value -> value != null)
                .max(OffsetDateTime::compareTo)
                .orElse(null);
    }

    private void logEventExtractionProgress(
            R request,
            UUID importRunId,
            UUID packageId,
            ImportPlanItemDto planItem,
            ImportTimeChunkDto chunk,
            ExtractedEventStats stats
    ) {
        log.info("EventHub event extraction progress provider={} tenant={} importRunId={} packageId={} extractionCode={} sourceKind={} chunk={} mapped={} byType={}",
                providerPackagePrefix(), request.tenantKey(), importRunId, packageId, planItem.extractionCode(), planItem.sourceKind(),
                chunk.sequence(), stats.eventsMapped(), stats.eventTypeCounts());
    }

    private void logEventExtractionFinished(
            R request,
            UUID importRunId,
            UUID packageId,
            ImportPlanItemDto planItem,
            ImportTimeChunkDto chunk,
            ExtractedEventStats stats
    ) {
        log.info("Finished EventHub event extraction provider={} tenant={} importRunId={} packageId={} extractionCode={} sourceKind={} chunk={} mapped={} byType={}",
                providerPackagePrefix(), request.tenantKey(), importRunId, packageId, planItem.extractionCode(), planItem.sourceKind(),
                chunk.sequence(), stats.eventsMapped(), stats.eventTypeCounts());
    }

    protected Map<String, Object> parameters(R request, ImportScopeDto scope, ImportCursorStateDto cursor) {
        Map<String, Object> params = new HashMap<>();
        String organisationId = scope == null || scope.rootSourceOrganisation() == null
                ? null
                : scope.rootSourceOrganisation().sourceEntityId();
        params.put("tenantKey", request.tenantKey());
        params.put("occurredFrom", scope == null ? null : scope.occurredFrom());
        params.put("occurredTo", scope == null ? null : scope.occurredTo());
        params.put("organisationId", organisationId);
        params.put("includeChildren", scope != null && scope.includeChildren());
        params.put("sourcePackageWatermarkEnabled", request.acquisitionStrategy() == AcquisitionStrategy.SOURCE_PACKAGE_WATERMARK ? 1 : 0);
        params.put("lastSourcePackageImportedAt", cursor == null ? null : cursor.lastSourcePackageImportedAt());
        params.put("lastSourcePackageId", cursor == null ? null : cursor.lastSourcePackageId());
        params.put("lastSourcePackageIdNumeric", parseLong(cursor == null ? null : cursor.lastSourcePackageId()));
        params.put("lastSourceRowUpdatedAt", cursor == null ? null : cursor.lastSourceRowUpdatedAt());
        params.put("lastOccurredTo", cursor == null ? null : cursor.lastOccurredTo());
        return params;
    }

    protected ImportCursorStateDto findCursor(int eventSourceId, R request, ImportPlanItemDto planItem) {
        String scopeHash = request.importScope() == null ? "NO_SCOPE" : request.importScope().stableKey();
        ImportCursorStateDto cursor = importCursorRepository.findCursor(
                request.tenantKey(),
                eventSourceId,
                scopeHash,
                planItem.eventFamily(),
                planItem.sourceKind(),
                request.acquisitionStrategy()
        );
        if (cursor != null || !shouldBootstrapWatermarkCursor(request)) {
            return cursor;
        }
        return importCursorRepository.findLatestCursor(
                request.tenantKey(),
                eventSourceId,
                scopeHash,
                planItem.eventFamily(),
                planItem.sourceKind()
        );
    }

    private boolean shouldBootstrapWatermarkCursor(R request) {
        return request.mode() == at.procon.eventhub.dto.ImportMode.INCREMENTAL_UPDATE
                && (request.acquisitionStrategy() == AcquisitionStrategy.SOURCE_ROW_WATERMARK
                || request.acquisitionStrategy() == AcquisitionStrategy.SOURCE_PACKAGE_WATERMARK);
    }

    protected OffsetDateTime lastSourcePackageImportedAt(ExtractedEventStats stats, ImportCursorStateDto cursor) {
        return stats.lastSourcePackageImportedAt() == null
                ? cursor == null ? null : cursor.lastSourcePackageImportedAt()
                : stats.lastSourcePackageImportedAt();
    }

    protected String lastSourcePackageId(ExtractedEventStats stats, ImportCursorStateDto cursor) {
        if (stats.lastSourcePackageIdByImportedAt() != null) {
            return stats.lastSourcePackageIdByImportedAt();
        }
        if (stats.maxSourcePackageId() != null) {
            return stats.maxSourcePackageId();
        }
        return cursor == null ? null : cursor.lastSourcePackageId();
    }

    private EventHubPackageRequest packageInfo(
            UUID importRunId,
            R request,
            ImportPlanItemDto planItem,
            ImportTimeChunkDto chunk,
            ImportScopeDto chunkScope
    ) {
        return new EventHubPackageRequest(
                request.tenantKey(),
                eventSourceFor(request, planItem),
                request.sourceGroup(),
                chunkScope,
                planItem.eventFamily().name(),
                chunk.occurredFrom() == null ? null : chunk.occurredFrom().toLocalDate(),
                providerPackagePrefix() + ":" + planItem.sourceKind() + ":" + planItem.extractionCode() + ":RUN-" + importRunId + ":CHUNK-" + chunk.sequence()
        );
    }

    private ImportScopeDto chunkScope(ImportScopeDto scope, ImportTimeChunkDto chunk) {
        if (scope == null) {
            return ImportScopeDto.tenantAll(chunk.occurredFrom(), chunk.occurredTo());
        }
        return new ImportScopeDto(
                scope.type(),
                scope.rootSourceOrganisation(),
                scope.includeChildren(),
                chunk.occurredFrom(),
                chunk.occurredTo()
        );
    }

    private String loadSql(String location) {
        Resource resource = resourceLoader.getResource(location);
        try (var inputStream = resource.getInputStream()) {
            return StreamUtils.copyToString(inputStream, StandardCharsets.UTF_8);
        } catch (IOException e) {
            throw new IllegalStateException("Cannot load extraction SQL resource " + location, e);
        }
    }

    private int compareSourcePackageId(String left, String right) {
        Integer leftInt = parseInteger(left);
        Integer rightInt = parseInteger(right);
        if (leftInt != null && rightInt != null) {
            return leftInt.compareTo(rightInt);
        }
        return left.compareTo(right);
    }

    private Integer parseInteger(String value) {
        if (value == null || value.isBlank()) {
            return null;
        }
        try {
            return Integer.parseInt(value.trim());
        } catch (NumberFormatException ignored) {
            return null;
        }
    }

    private Long parseLong(String value) {
        if (value == null || value.isBlank()) {
            return null;
        }
        try {
            return Long.parseLong(value.trim());
        } catch (NumberFormatException ignored) {
            return null;
        }
    }

    protected final class ExtractedEventStats {
        private int eventsMapped;
        private OffsetDateTime lastSourcePackageImportedAt;
        private String lastSourcePackageIdByImportedAt;
        private String maxSourcePackageId;
        private final Map<String, Integer> eventTypeCounts = new LinkedHashMap<>();

        public int eventsMapped() {
            return eventsMapped;
        }

        public OffsetDateTime lastSourcePackageImportedAt() {
            return lastSourcePackageImportedAt;
        }

        public String lastSourcePackageIdByImportedAt() {
            return lastSourcePackageIdByImportedAt;
        }

        public String maxSourcePackageId() {
            return maxSourcePackageId;
        }

        private int eventsInserted;
        private int persistBatchNo;

        public int eventsInserted() {
            return eventsInserted;
        }

        public Map<String, Integer> eventTypeCounts() {
            return eventTypeCounts;
        }

        private int nextPersistBatchNo() {
            persistBatchNo++;
            return persistBatchNo;
        }

        private void acceptPersistResult(EventHubPackageResult result) {
            if (result != null) {
                eventsInserted += result.insertedCount();
            }
        }

        private void accept(EventHubEventDto event) {
            eventsMapped++;
            eventTypeCounts.merge(eventTypeKey(event), 1, Integer::sum);
            if (event.sourcePackageRef() == null) {
                return;
            }

            OffsetDateTime importedAt = event.sourcePackageRef().importedIntoSourceAt();
            String sourcePackageId = event.sourcePackageRef().sourcePackageId();
            if (importedAt != null) {
                if (lastSourcePackageImportedAt == null || importedAt.compareTo(lastSourcePackageImportedAt) > 0) {
                    lastSourcePackageImportedAt = importedAt;
                    lastSourcePackageIdByImportedAt = sourcePackageId;
                } else if (importedAt.compareTo(lastSourcePackageImportedAt) == 0
                        && sourcePackageId != null
                        && !sourcePackageId.isBlank()
                        && (lastSourcePackageIdByImportedAt == null
                                || compareSourcePackageId(sourcePackageId, lastSourcePackageIdByImportedAt) > 0)) {
                    lastSourcePackageIdByImportedAt = sourcePackageId;
                }
            }
            if (sourcePackageId != null
                    && !sourcePackageId.isBlank()
                    && (maxSourcePackageId == null || compareSourcePackageId(sourcePackageId, maxSourcePackageId) > 0)) {
                maxSourcePackageId = sourcePackageId;
            }
        }

        private String eventTypeKey(EventHubEventDto event) {
            String domain = event.eventDomain() == null ? "UNKNOWN_DOMAIN" : event.eventDomain().name();
            String type = event.eventType() == null ? "UNKNOWN_EVENT" : event.eventType().name();
            return domain + "/" + type;
        }
    }
}
