package at.procon.eventhub.importing;

import at.procon.eventhub.dto.EventHubPackageRequest;
import at.procon.eventhub.dto.EventSourceDto;
import at.procon.eventhub.dto.ImportRunStatus;
import at.procon.eventhub.importing.persistence.ImportCursorRepository;
import at.procon.eventhub.importing.persistence.ImportRunRepository;
import at.procon.eventhub.persistence.DataPackageRepository;
import at.procon.eventhub.persistence.EventSourceRepository;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Shared import-run orchestration for source providers. Provider services supply
 * the plan and the concrete batch executor; this class owns the common EventHub
 * run, package, and cursor lifecycle.
 */
public abstract class AbstractImportExecutionService<R extends ImportRunRequest, B extends ExtractionBatchResult> {

    private final Logger log = LoggerFactory.getLogger(getClass());

    private final EventSourceRepository eventSourceRepository;
    private final ImportRunRepository importRunRepository;
    private final DataPackageRepository dataPackageRepository;
    private final ImportCursorRepository importCursorRepository;

    protected AbstractImportExecutionService(
            EventSourceRepository eventSourceRepository,
            ImportRunRepository importRunRepository,
            DataPackageRepository dataPackageRepository,
            ImportCursorRepository importCursorRepository
    ) {
        this.eventSourceRepository = eventSourceRepository;
        this.importRunRepository = importRunRepository;
        this.dataPackageRepository = dataPackageRepository;
        this.importCursorRepository = importCursorRepository;
    }

    protected ImportRunResultDto createImportRun(R request, boolean executeImmediately) {
        ImportPlanDto plan = createPlan(request);
        int baseEventSourceId = eventSourceRepository.resolveSourceId(request.tenantKey(), request.eventSource());
        UUID importRunId = importRunRepository.createPlannedRun(
                baseEventSourceId,
                request,
                importRunMetadata(request, executeImmediately)
        );

        List<UUID> packageIds = new ArrayList<>();
        List<PlannedPackage> plannedPackages = new ArrayList<>();
        int batchNo = 1;
        try {
            for (ImportPlanItemDto item : plan.items()) {
                EventSourceDto itemEventSource = eventSourceForItem(request.eventSource(), item);
                int itemEventSourceId = eventSourceRepository.resolveSourceId(request.tenantKey(), itemEventSource);
                for (ImportTimeChunkDto chunk : plan.chunks()) {
                    EventHubPackageRequest packageInfo = packageRequestFor(request, itemEventSource, item, chunk);
                    String packageKey = packageKey(importRunId, packageInfo, item, chunk, batchNo);
                    UUID packageId = dataPackageRepository.createPlannedExtractionPackage(
                            importRunId,
                            itemEventSourceId,
                            packageKey,
                            packageInfo,
                            item.extractionCode(),
                            item.sourceKind(),
                            item.entityAxis(),
                            chunk == null ? null : chunk.occurredFrom(),
                            chunk == null ? null : chunk.occurredTo(),
                            batchNo,
                            packageMetadata(request, item, chunk, importRunId)
                    );
                    packageIds.add(packageId);
                    plannedPackages.add(new PlannedPackage(packageId, itemEventSourceId, item, chunk));
                    batchNo++;
                }
            }
            importRunRepository.markPlannedPackages(importRunId, packageIds.size());
            log.info("Created import run provider={} importRunId={} plannedPackages={} tenant={} mode={} strategy={} executeImmediately={}",
                    providerPackagePrefix(), importRunId, packageIds.size(), request.tenantKey(), request.mode(), request.acquisitionStrategy(), executeImmediately);

            if (executeImmediately) {
                executePlannedPackages(importRunId, request, plannedPackages);
                return new ImportRunResultDto(importRunId, ImportRunStatus.COMPLETED, packageIds.size(), plan, List.copyOf(packageIds));
            }
            return new ImportRunResultDto(importRunId, ImportRunStatus.PLANNED, packageIds.size(), plan, List.copyOf(packageIds));
        } catch (RuntimeException ex) {
            importRunRepository.markFailed(importRunId, ex.getMessage());
            throw ex;
        }
    }

    protected abstract ImportPlanDto createPlan(R request);

    protected abstract B executeBatch(
            UUID importRunId,
            UUID packageId,
            int eventSourceId,
            R request,
            ImportPlanItemDto planItem,
            ImportTimeChunkDto chunk
    );

    protected void beforeExecute(R request) {
    }

    protected EventSourceDto eventSourceForItem(EventSourceDto base, ImportPlanItemDto item) {
        return base;
    }

    protected Map<String, Object> importRunMetadata(R request, boolean executeImmediately) {
        return Map.of(
                "note", executeImmediately
                        ? "Created import run and executing planned extraction packages."
                        : "Created import run and planned extraction packages.",
                "packageModel", "EventHub data packages are extraction batches; original source packages are SourcePackageRefDto.",
                "executeImmediately", executeImmediately
        );
    }

    protected Map<String, Object> packageMetadata(R request, ImportPlanItemDto item, ImportTimeChunkDto chunk, UUID importRunId) {
        Map<String, Object> metadata = new LinkedHashMap<>();
        metadata.put("importRunId", importRunId.toString());
        metadata.put("mode", request.mode().name());
        metadata.put("acquisitionStrategy", request.acquisitionStrategy().name());
        metadata.put("refreshMasterDataFirst", request.refreshMasterDataFirst());
        metadata.put("eventFamily", item.eventFamily().name());
        metadata.put("sourceKind", item.sourceKind());
        metadata.put("extractionCode", item.extractionCode());
        metadata.put("sourceTables", item.sourceTables());
        metadata.put("entityAxis", item.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("sourcePackageRefPolicy", "Original source package is preserved per acquired event when extraction returns it.");
        return metadata;
    }

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

    protected String externalPackageId(R request, ImportPlanItemDto item, ImportTimeChunkDto chunk) {
        String scope = request.importScope() == null ? "NO_SCOPE" : request.importScope().stableKey();
        return providerPackagePrefix() + ":" + item.sourceKind() + ":" + item.extractionCode() + ":" + item.eventFamily()
                + ":" + scope + ":CHUNK-" + chunk.sequence();
    }

    protected String packageKey(UUID importRunId, EventHubPackageRequest packageInfo, ImportPlanItemDto item, ImportTimeChunkDto chunk, int batchNo) {
        return packageInfo.tenantKey()
                + ":" + packageInfo.eventSource().stableKey()
                + ":" + item.eventFamily()
                + ":" + item.extractionCode()
                + ":RUN-" + importRunId
                + ":CHUNK-" + chunk.sequence()
                + ":BATCH-" + batchNo;
    }

    private void executePlannedPackages(UUID importRunId, R request, List<PlannedPackage> plannedPackages) {
        importRunRepository.markRunning(importRunId);
        beforeExecute(request);
        List<B> results = new ArrayList<>();
        for (PlannedPackage plannedPackage : plannedPackages) {
            dataPackageRepository.markImporting(plannedPackage.packageId());
            try {
                B result = executeBatch(
                        importRunId,
                        plannedPackage.packageId(),
                        plannedPackage.eventSourceId(),
                        request,
                        plannedPackage.planItem(),
                        plannedPackage.chunk()
                );
                dataPackageRepository.updateEventCount(plannedPackage.packageId(), result.eventsInserted());
                if (!result.eventTypeCounts().isEmpty()) {
                    dataPackageRepository.mergeMetadata(plannedPackage.packageId(), Map.of(
                            "extractedEventTypeCounts", result.eventTypeCounts()
                    ));
                }
                results.add(result);
                dataPackageRepository.markImported(plannedPackage.packageId(), result.eventsInserted());
                if (result.executed()) {
                    importCursorRepository.advanceCursor(
                            request.tenantKey(),
                            plannedPackage.eventSourceId(),
                            request.importScope() == null ? "NO_SCOPE" : request.importScope().stableKey(),
                            plannedPackage.planItem().eventFamily(),
                            plannedPackage.planItem().sourceKind(),
                            request.acquisitionStrategy(),
                            result
                    );
                }
            } catch (RuntimeException ex) {
                dataPackageRepository.markFailed(plannedPackage.packageId(), ex.getMessage());
                throw ex;
            }
        }
        importRunRepository.markCompleted(importRunId);
        log.info("Completed import run provider={} importRunId={} packages={} insertedEvents={} executedBatches={}",
                providerPackagePrefix(),
                importRunId,
                plannedPackages.size(),
                results.stream().mapToInt(ExtractionBatchResult::eventsInserted).sum(),
                results.stream().filter(ExtractionBatchResult::executed).count());
    }

    private EventHubPackageRequest packageRequestFor(
            R request,
            EventSourceDto itemEventSource,
            ImportPlanItemDto item,
            ImportTimeChunkDto chunk
    ) {
        return new EventHubPackageRequest(
                request.tenantKey(),
                itemEventSource,
                request.sourceGroup(),
                chunkScope(request.importScope(), chunk),
                item.eventFamily().name(),
                chunk.occurredFrom() == null ? null : chunk.occurredFrom().toLocalDate(),
                externalPackageId(request, item, chunk)
        );
    }

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

    private record PlannedPackage(UUID packageId, int eventSourceId, ImportPlanItemDto planItem, ImportTimeChunkDto chunk) {
    }
}
