package at.procon.eventhub.yellowfox.service;

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.EventType;
import at.procon.eventhub.dto.ImportCursorStateDto;
import at.procon.eventhub.dto.ImportScopeDto;
import at.procon.eventhub.dto.SourceGroupType;
import at.procon.eventhub.importing.ImportPlanItemDto;
import at.procon.eventhub.importing.ImportTimeChunkDto;
import at.procon.eventhub.importing.persistence.ImportCursorRepository;
import at.procon.eventhub.service.EventHubIngestionService;
import at.procon.eventhub.yellowfox.dto.YellowFoxD8BookingDto;
import at.procon.eventhub.yellowfox.dto.YellowFoxD8ExtractionBatchResultDto;
import at.procon.eventhub.yellowfox.dto.YellowFoxD8ImportRequest;
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.UUID;
import org.apache.camel.ProducerTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StreamUtils;

@Service
@ConditionalOnExpression("T(org.springframework.util.StringUtils).hasText('${eventhub.yellow-fox.datasource.jdbc-url:}')")
public class JdbcYellowFoxD8BookingExtractionBatchExecutor implements YellowFoxD8ExtractionBatchExecutor {

    private static final Logger log = LoggerFactory.getLogger(JdbcYellowFoxD8BookingExtractionBatchExecutor.class);
    private static final String EXTRACTION_CODE = "YELLOWFOX_D8_BOOKING";
    private static final int PROGRESS_LOG_INTERVAL = 5000;

    private final NamedParameterJdbcTemplate jdbcTemplate;
    private final EventHubIngestionService ingestionService;
    private final ProducerTemplate producerTemplate;
    private final ResourceLoader resourceLoader;
    private final ImportCursorRepository importCursorRepository;
    private final EventHubProperties properties;
    private final YellowFoxD8BookingRowMapper rowMapper;
    private final YellowFoxD8BookingEventMapper eventMapper;
    private final YellowFoxD8IgnitionTransitionDetector ignitionTransitionDetector;

    public JdbcYellowFoxD8BookingExtractionBatchExecutor(
            @Qualifier("yellowFoxNamedParameterJdbcTemplate") NamedParameterJdbcTemplate jdbcTemplate,
            EventHubIngestionService ingestionService,
            ProducerTemplate producerTemplate,
            ResourceLoader resourceLoader,
            ImportCursorRepository importCursorRepository,
            EventHubProperties properties,
            YellowFoxD8BookingRowMapper rowMapper,
            YellowFoxD8BookingEventMapper eventMapper,
            YellowFoxD8IgnitionTransitionDetector ignitionTransitionDetector
    ) {
        this.jdbcTemplate = jdbcTemplate;
        this.ingestionService = ingestionService;
        this.producerTemplate = producerTemplate;
        this.resourceLoader = resourceLoader;
        this.importCursorRepository = importCursorRepository;
        this.properties = properties;
        this.rowMapper = rowMapper;
        this.eventMapper = eventMapper;
        this.ignitionTransitionDetector = ignitionTransitionDetector;
    }

    @Override
    public YellowFoxD8ExtractionBatchResultDto execute(
            UUID importRunId,
            UUID packageId,
            int eventSourceId,
            YellowFoxD8ImportRequest request,
            ImportPlanItemDto planItem,
            ImportTimeChunkDto chunk
    ) {
        if (!EXTRACTION_CODE.equals(planItem.extractionCode())) {
            throw new IllegalArgumentException("Unsupported YellowFox extraction code " + planItem.extractionCode());
        }

        ImportScopeDto chunkScope = chunkScope(request.importScope(), chunk);
        ImportCursorStateDto cursor = findCursor(eventSourceId, request, planItem);
        QuerySpec query = buildQuerySpec(request, chunkScope, cursor);
        EventHubPackageRequest packageInfo = packageInfo(importRunId, request, planItem, chunk, chunkScope);
        Stats stats = new Stats();
        List<EventHubEventDto> pendingEvents = new ArrayList<>(jdbcPersistBatchSize());
        YellowFoxD8IgnitionTransitionDetector.Session ignitionSession = ignitionTransitionDetector
                .newSession(properties.getYellowFox().isEmitInitialIgnitionSnapshot());

        log.info("Reading YellowFox D8 bookings tenant={} importRunId={} packageId={} chunk={} occurredFrom={} occurredTo={} fleetId={} strategy={} ingestMode={}",
                request.tenantKey(), importRunId, packageId, chunk.sequence(), chunk.occurredFrom(), chunk.occurredTo(), query.fleetId(),
                request.acquisitionStrategy(), jdbcExtractionIngestMode());

        jdbcTemplate.query(query.sql(), query.params(), rs -> {
            stats.sourceRowsRead++;
            YellowFoxD8BookingDto booking = rowMapper.map(
                    rs,
                    request.tenantKey(),
                    request.eventSource().sourceInstanceKey(),
                    request.eventSource().tenantProviderSettingKey()
            );
            stats.acceptSourceRow(booking);
            if (!hasEventReference(booking)) {
                stats.skippedRows++;
                return;
            }
            EventHubEventDto primaryEvent = eventMapper.map(booking);
            pendingEvents.add(primaryEvent);
            stats.acceptEvent(primaryEvent);
            EventHubEventDto ignitionEvent = ignitionSession.detect(booking);
            if (ignitionEvent != null) {
                pendingEvents.add(ignitionEvent);
                stats.acceptEvent(ignitionEvent);
            }
            if (pendingEvents.size() >= jdbcPersistBatchSize()) {
                flushPersistBatch(request, importRunId, packageId, planItem, chunk, packageInfo, pendingEvents, stats);
            }
            if (stats.sourceRowsRead % PROGRESS_LOG_INTERVAL == 0) {
                log.info("YellowFox D8 extraction progress tenant={} importRunId={} packageId={} rows={} events={} inserted={} byType={}",
                        request.tenantKey(), importRunId, packageId, stats.sourceRowsRead, stats.eventsMapped, stats.eventsInserted, stats.eventTypeCounts);
            }
        });
        flushPersistBatch(request, importRunId, packageId, planItem, chunk, packageInfo, pendingEvents, stats);

        log.info("Finished YellowFox D8 extraction tenant={} importRunId={} packageId={} rows={} events={} inserted={} skippedRows={} byType={}",
                request.tenantKey(), importRunId, packageId, stats.sourceRowsRead, stats.eventsMapped, stats.eventsInserted, stats.skippedRows, stats.eventTypeCounts);

        return new YellowFoxD8ExtractionBatchResultDto(
                packageId,
                planItem.extractionCode(),
                planItem.sourceKind(),
                stats.sourceRowsRead,
                stats.eventsMapped,
                stats.eventsInserted,
                stats.skippedRows,
                true,
                null,
                stats.lastEventId,
                stats.lastOccurredAt,
                stats.lastOccurredAt,
                stats.eventTypeCounts
        );
    }

    EventHubProperties.JdbcExtractionIngestMode jdbcExtractionIngestMode() {
        return properties.getYellowFox().getJdbcExtractionIngestMode();
    }

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

    QuerySpec buildQuerySpec(YellowFoxD8ImportRequest request, ImportScopeDto scope, ImportCursorStateDto cursor) {
        Map<String, Object> params = new HashMap<>();
        StringBuilder filters = new StringBuilder("where 1 = 1");

        if (scope != null && scope.occurredFrom() != null) {
            params.put("occurredFrom", scope.occurredFrom());
            filters.append("\n  and b.utc >= :occurredFrom");
        }
        if (scope != null && scope.occurredTo() != null) {
            params.put("occurredTo", scope.occurredTo());
            filters.append("\n  and b.utc < :occurredTo");
        }

        Integer fleetId = fleetId(request);
        if (fleetId != null) {
            params.put("fleetId", fleetId);
            filters.append("\n  and f.id = :fleetId");
        }

        appendCursorFilter(filters, params, request, cursor);

        return new QuerySpec(applyFilters(loadSqlTemplate(), filters.toString()), params, fleetId);
    }

    ImportCursorStateDto findCursor(int eventSourceId, YellowFoxD8ImportRequest 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 Integer fleetId(YellowFoxD8ImportRequest request) {
        if (request.sourceGroup() == null || request.sourceGroup().type() != SourceGroupType.FLEET) {
            return null;
        }
        String value = request.sourceGroup().sourceEntityId() == null
                ? request.sourceGroup().code()
                : request.sourceGroup().sourceEntityId();
        if (value == null || value.isBlank()) {
            return null;
        }
        try {
            return Integer.parseInt(value.trim());
        } catch (NumberFormatException ex) {
            throw new IllegalArgumentException("YellowFox D8 fleet sourceEntityId/code must be numeric for JDBC import: " + value, ex);
        }
    }

    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 boolean hasEventReference(YellowFoxD8BookingDto booking) {
        return (booking.driverRef() != null && booking.driverRef().hasAnyReference())
                || (booking.vehicleRef() != null && booking.vehicleRef().hasAnyReference());
    }

    private void flushPersistBatch(
            YellowFoxD8ImportRequest request,
            UUID importRunId,
            UUID extractionPackageId,
            ImportPlanItemDto planItem,
            ImportTimeChunkDto chunk,
            EventHubPackageRequest packageInfo,
            List<EventHubEventDto> pendingEvents,
            Stats 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 YellowFox extraction batch tenant={} importRunId={} extractionPackageId={} extractionCode={} sourceKind={} chunk={} batchNo={} received={} inserted={}",
                request.tenantKey(), importRunId, extractionPackageId, planItem.extractionCode(), planItem.sourceKind(),
                chunk.sequence(), batchNo, result.receivedCount(), result.insertedCount());
    }

    EventHubPackageResult persistBatch(EventHubEventBatchDto batch) {
        if (jdbcExtractionIngestMode() == EventHubProperties.JdbcExtractionIngestMode.CAMEL_ROUTE) {
            return producerTemplate.requestBody("direct:eventhub-batch-persist-input", batch, EventHubPackageResult.class);
        }
        return ingestionService.ingest(batch);
    }

    private Map<String, Object> persistBatchMetadata(
            YellowFoxD8ImportRequest 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("slotPolicy", "eventtype 0/2 = DRIVER, eventtype 1/3 = CO_DRIVER");
        metadata.put("ignitionPolicy", "Store ignition state on every D8 detail; emit separate ignition event only on state change.");
        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 EventHubPackageRequest packageInfo(
            UUID importRunId,
            YellowFoxD8ImportRequest 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(),
                "YELLOWFOX_D8:" + planItem.sourceKind() + ":" + planItem.extractionCode() + ":RUN-" + importRunId + ":CHUNK-" + chunk.sequence()
        );
    }

    private EventSourceDto eventSourceFor(YellowFoxD8ImportRequest request, ImportPlanItemDto planItem) {
        return new EventSourceDto(
                "YELLOWFOX",
                planItem.sourceKind(),
                "YELLOWFOX_D8",
                request.eventSource().sourceInstanceKey(),
                request.eventSource().tenantProviderSettingKey(),
                request.eventSource().externalFleetKey()
        );
    }

    private void appendCursorFilter(
            StringBuilder filters,
            Map<String, Object> params,
            YellowFoxD8ImportRequest request,
            ImportCursorStateDto cursor
    ) {
        if (request.acquisitionStrategy() != AcquisitionStrategy.SOURCE_ROW_WATERMARK || cursor == null) {
            return;
        }

        OffsetDateTime lastOccurredTo = cursor.lastOccurredTo();
        String lastSourceRowId = cursor.lastSourcePackageId();
        if (lastOccurredTo == null) {
            return;
        }
        if (properties.getYellowFox().getOccurredAtOverlap() != null
                && !properties.getYellowFox().getOccurredAtOverlap().isZero()) {
            lastOccurredTo = lastOccurredTo.minus(properties.getYellowFox().getOccurredAtOverlap());
            lastSourceRowId = null;
        }

        params.put("lastOccurredTo", lastOccurredTo);
        if (lastSourceRowId == null || lastSourceRowId.isBlank()) {
            filters.append("\n  and b.utc >= :lastOccurredTo");
            return;
        }

        params.put("lastSourceRowId", lastSourceRowId);
        filters.append("""

  and (
        b.utc > :lastOccurredTo
        or (
            b.utc = :lastOccurredTo
            and b.eventid > :lastSourceRowId
        )
      )""");
    }

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

    private String applyFilters(String sqlTemplate, String filters) {
        if (!sqlTemplate.contains("/*__FILTERS__*/")) {
            throw new IllegalStateException("YellowFox D8 extraction SQL template is missing filter marker");
        }
        return sqlTemplate.replace("/*__FILTERS__*/", filters);
    }

    private String loadSqlTemplate() {
        Resource resource = resourceLoader.getResource("classpath:sql/yellowfox/d8-bookings.sql");
        try (var in = resource.getInputStream()) {
            return StreamUtils.copyToString(in, StandardCharsets.UTF_8);
        } catch (IOException ex) {
            throw new IllegalStateException("Failed to load YellowFox D8 extraction SQL", ex);
        }
    }

    static record QuerySpec(String sql, Map<String, Object> params, Integer fleetId) {
    }

    private static class Stats {
        private int sourceRowsRead;
        private int eventsMapped;
        private int eventsInserted;
        private int skippedRows;
        private OffsetDateTime lastOccurredAt;
        private String lastEventId;
        private final Map<String, Integer> eventTypeCounts = new LinkedHashMap<>();
        private int persistBatchNo;

        private void acceptSourceRow(YellowFoxD8BookingDto booking) {
            if (booking.occurredAt() != null) {
                lastOccurredAt = booking.occurredAt();
                lastEventId = booking.eventId();
            }
        }

        private void acceptEvent(EventHubEventDto event) {
            eventsMapped++;
            EventType type = event.eventType();
            String key = event.eventDomain().name() + "/" + (type == null ? "UNKNOWN" : type.name());
            eventTypeCounts.merge(key, 1, Integer::sum);
        }

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

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