package at.procon.dip.ingestion.adapter;

import at.procon.dip.domain.access.DocumentAccessContext;
import at.procon.dip.domain.document.CanonicalDocumentMetadata;
import at.procon.dip.domain.document.SourceType;
import at.procon.dip.domain.document.service.DocumentService;
import at.procon.dip.ingestion.config.DipIngestionProperties;
import at.procon.dip.ingestion.dto.ImportedDocumentResult;
import at.procon.dip.ingestion.service.GenericDocumentImportService;
import at.procon.dip.ingestion.service.TedPackageChildImportProcessor;
import at.procon.dip.ingestion.service.TedPackageExpansionService;
import at.procon.dip.ingestion.spi.DocumentIngestionAdapter;
import at.procon.dip.ingestion.spi.IngestionResult;
import at.procon.dip.ingestion.spi.OriginalContentStoragePolicy;
import at.procon.dip.ingestion.spi.SourceDescriptor;
import at.procon.dip.runtime.condition.ConditionalOnRuntimeMode;
import at.procon.dip.runtime.config.RuntimeMode;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

@Component
@ConditionalOnRuntimeMode(RuntimeMode.NEW)
@RequiredArgsConstructor
@Slf4j
public class TedPackageDocumentIngestionAdapter implements DocumentIngestionAdapter {

    private final DipIngestionProperties properties;
    private final GenericDocumentImportService importService;
    private final TedPackageExpansionService expansionService;
    private final TedPackageChildImportProcessor childImportProcessor;
    private final DocumentService documentService;

    @Override
    public boolean supports(SourceDescriptor sourceDescriptor) {
        return sourceDescriptor.sourceType() == SourceType.TED_PACKAGE
                && properties.isEnabled()
                && properties.isTedPackageAdapterEnabled();
    }

    @Override
    @Transactional(propagation = Propagation.NOT_SUPPORTED)
    public IngestionResult ingest(SourceDescriptor sourceDescriptor) {
        SourceDescriptor packageRootSource = buildPackageRootSource(sourceDescriptor);

        Map<String, String> rootAttributes = new LinkedHashMap<>(sourceDescriptor.attributes() == null ? Map.of() : sourceDescriptor.attributes());
        rootAttributes.putIfAbsent("packageId", sourceDescriptor.sourceIdentifier());
        rootAttributes.putIfAbsent("title", sourceDescriptor.fileName() != null ? sourceDescriptor.fileName() : sourceDescriptor.sourceIdentifier());
        rootAttributes.put("wrapperDocument", Boolean.TRUE.toString());
        rootAttributes.put("importBatchId", properties.getTedPackageImportBatchId());

        ImportedDocumentResult packageDocument = importService.importDocument(new SourceDescriptor(
                sourceDescriptor.accessContext() == null ? DocumentAccessContext.publicDocument() : sourceDescriptor.accessContext(),
                SourceType.TED_PACKAGE,
                sourceDescriptor.sourceIdentifier(),
                packageRootSource.sourceUri(),
                sourceDescriptor.fileName(),
                sourceDescriptor.mediaType() == null ? "application/gzip" : sourceDescriptor.mediaType(),
                packageRootSource.binaryContent(),
                null,
                sourceDescriptor.receivedAt() == null ? OffsetDateTime.now() : sourceDescriptor.receivedAt(),
                OriginalContentStoragePolicy.SKIP,
                rootAttributes
        ));

        List<String> warnings = new ArrayList<>(packageDocument.warnings());
        List<CanonicalDocumentMetadata> documents = new ArrayList<>();
        documents.add(documentService.getMetadata(packageDocument.document().getId()));

        List<OrderedChildImportResult> childResults = properties.getTedPackageChildParallelism() > 1
                ? processChildrenInParallel(packageDocument, sourceDescriptor)
                : processChildrenSequentially(packageDocument, sourceDescriptor);

        childResults.stream()
                .sorted(Comparator.comparingInt(OrderedChildImportResult::sortOrder))
                .forEach(orderedResult -> {
                    TedPackageChildImportProcessor.ChildImportResult result = orderedResult.result();
                    if (result.childDocumentMetadata() != null) {
                        documents.add(result.childDocumentMetadata());
                    }
                    if (StringUtils.hasText(result.warning())) {
                        warnings.add(result.warning());
                    }
                });

        return new IngestionResult(documents, warnings);
    }

    private List<OrderedChildImportResult> processChildrenSequentially(ImportedDocumentResult packageDocument,
                                                                       SourceDescriptor sourceDescriptor) {
        AtomicInteger sortOrder = new AtomicInteger();
        List<OrderedChildImportResult> results = new ArrayList<>();
        streamPackageEntries(sourceDescriptor, entry -> {
            int order = sortOrder.incrementAndGet();
            TedPackageChildImportProcessor.ChildImportResult result = childImportProcessor.processChild(
                    packageDocument.document().getId(),
                    sourceDescriptor.sourceIdentifier(),
                    sourceDescriptor.receivedAt(),
                    sourceDescriptor.accessContext(),
                    entry,
                    order
            );
            results.add(new OrderedChildImportResult(order, result));
        });
        return results;
    }

    private List<OrderedChildImportResult> processChildrenInParallel(ImportedDocumentResult packageDocument,
                                                                     SourceDescriptor sourceDescriptor) {
        int parallelism = Math.max(1, properties.getTedPackageChildParallelism());
        int maxInFlight = Math.max(parallelism, properties.getTedPackageChildMaxInFlight());
        log.info("Processing TED package {} with parallel child import (parallelism={}, maxInFlight={})",
                sourceDescriptor.sourceIdentifier(), parallelism, maxInFlight);

        ExecutorService executor = Executors.newFixedThreadPool(parallelism, new TedPackageChildThreadFactory());
        ExecutorCompletionService<OrderedChildImportResult> completionService = new ExecutorCompletionService<>(executor);
        List<OrderedChildImportResult> results = new ArrayList<>();
        AtomicInteger sortOrder = new AtomicInteger();
        AtomicInteger submitted = new AtomicInteger();
        AtomicInteger completed = new AtomicInteger();

        try {
            streamPackageEntries(sourceDescriptor, entry -> {
                int order = sortOrder.incrementAndGet();
                completionService.submit(() -> new OrderedChildImportResult(order, childImportProcessor.processChild(
                        packageDocument.document().getId(),
                        sourceDescriptor.sourceIdentifier(),
                        sourceDescriptor.receivedAt(),
                        sourceDescriptor.accessContext(),
                        entry,
                        order
                )));
                int currentSubmitted = submitted.incrementAndGet();
                while (currentSubmitted - completed.get() >= maxInFlight) {
                    results.add(awaitNextChildResult(completionService, executor));
                    completed.incrementAndGet();
                }
            });

            while (completed.get() < submitted.get()) {
                results.add(awaitNextChildResult(completionService, executor));
                completed.incrementAndGet();
            }
            return results;
        } finally {
            executor.shutdownNow();
        }
    }

    private OrderedChildImportResult awaitNextChildResult(ExecutorCompletionService<OrderedChildImportResult> completionService,
                                                          ExecutorService executor) {
        try {
            Future<OrderedChildImportResult> completedFuture = completionService.take();
            return completedFuture.get();
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
            throw new IllegalStateException("Interrupted while waiting for TED package child import completion", e);
        } catch (ExecutionException e) {
            executor.shutdownNow();
            Throwable cause = e.getCause();
            if (cause instanceof RuntimeException runtimeException) {
                throw runtimeException;
            }
            throw new IllegalStateException("TED package child import failed", cause);
        }
    }

    private SourceDescriptor buildPackageRootSource(SourceDescriptor sourceDescriptor) {
        if (StringUtils.hasText(sourceDescriptor.sourceUri()) && isReadablePath(sourceDescriptor.sourceUri())) {
            return new SourceDescriptor(
                    sourceDescriptor.accessContext(),
                    sourceDescriptor.sourceType(),
                    sourceDescriptor.sourceIdentifier(),
                    sourceDescriptor.sourceUri(),
                    sourceDescriptor.fileName(),
                    sourceDescriptor.mediaType(),
                    null,
                    sourceDescriptor.textContent(),
                    sourceDescriptor.receivedAt(),
                    sourceDescriptor.originalContentStoragePolicy(),
                    sourceDescriptor.attributes()
            );
        }
        return sourceDescriptor;
    }

    private void streamPackageEntries(SourceDescriptor sourceDescriptor,
                                      TedPackageExpansionService.TedPackageEntryConsumer consumer) {
        if (StringUtils.hasText(sourceDescriptor.sourceUri()) && isReadablePath(sourceDescriptor.sourceUri())) {
            expansionService.streamEntries(Path.of(sourceDescriptor.sourceUri()), consumer);
            return;
        }
        byte[] packageBytes = sourceDescriptor.binaryContent();
        if (packageBytes == null || packageBytes.length == 0) {
            throw new IllegalArgumentException("TED package adapter requires tar.gz bytes or a readable local sourceUri");
        }
        expansionService.streamEntries(packageBytes, consumer);
    }

    private boolean isReadablePath(String sourceUri) {
        try {
            Path path = Path.of(sourceUri);
            return Files.exists(path) && Files.isReadable(path) && Files.isRegularFile(path);
        } catch (Exception e) {
            return false;
        }
    }

    private record OrderedChildImportResult(int sortOrder, TedPackageChildImportProcessor.ChildImportResult result) {
    }

    private static final class TedPackageChildThreadFactory implements ThreadFactory {
        private final AtomicInteger counter = new AtomicInteger();

        @Override
        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(runnable, "dip-ted-child-" + counter.incrementAndGet());
            thread.setDaemon(true);
            return thread;
        }
    }
}
