package de.arxsilex.maptools.converter.worker

import de.arxsilex.maptools.converter.core.mtci.MtciBuildOptions
import de.arxsilex.maptools.converter.core.elevation.ElevationConversionRequest
import de.arxsilex.maptools.converter.core.elevation.ElevationConverter
import de.arxsilex.maptools.converter.core.elevation.ElevationExecutionPolicy
import de.arxsilex.maptools.converter.core.mapsforge.MapsforgeMapSource
import de.arxsilex.maptools.converter.core.mtci.MtciIntermediateBuilder
import de.arxsilex.maptools.converter.core.mtci.MtciSelectionMaterializer
import de.arxsilex.maptools.converter.core.mtci.MtciSpatialPhase
import de.arxsilex.maptools.converter.core.mtci.MtciSpatialSelector
import de.arxsilex.maptools.converter.core.pbf.MtciPbfWriter
import de.arxsilex.maptools.converter.core.pbf.PbfBlockScanner
import de.arxsilex.maptools.converter.core.pbf.PbfHeaderReader
import java.nio.file.AtomicMoveNotSupportedException
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.StandardCopyOption
import org.openstreetmap.osmosis.core.Osmosis
import java.util.concurrent.atomic.AtomicBoolean
import java.util.Comparator
import kotlin.concurrent.thread
import kotlin.system.exitProcess

fun main(arguments: Array<String>) {
    val exitCode = try {
        configureTemporaryDirectory(arguments)
        when (arguments.firstOrNull()) {
            "crop" -> runCrop(CropRequest.parse(arguments))
            "combine" -> runCombine(CombineRequest.parse(arguments))
            "mapsforge" -> runMapsforge(MapsforgeRequest.parse(arguments))
            "mapsforge-source" -> runMapsforgeSource(MapsforgeSourceRequest.parse(arguments))
            "poi-convert" -> runPoiConvert(PoiConvertRequest.parse(arguments))
            "poi-merge" -> runPoiMerge(PoiMergeRequest.parse(arguments))
            "elevation" -> runElevation(ElevationRequest.parse(arguments))
            else -> error(
                "Unbekannter Worker-Befehl. Erwartet: crop, combine, mapsforge, mapsforge-source, " +
                    "poi-convert, poi-merge oder elevation.",
            )
        }
        0
    } catch (exception: Throwable) {
        val rootCause = generateSequence(exception) { it.cause }.last()
        System.err.println(
            "MTC-WORKER status=failed message=${rootCause.message ?: rootCause.javaClass.simpleName}",
        )
        exception.printStackTrace(System.err)
        1
    }
    // Osmosis-Plugins können nicht-daemonisierte Bibliotheksthreads hinterlassen. Der Worker ist
    // absichtlich ein Einwegprozess und beendet sich daher nach einem definierten Ergebnis explizit.
    exitProcess(exitCode)
}

private fun runElevation(request: ElevationRequest) {
    Files.createDirectories(request.temporaryDirectory)
    val executionOptions = ElevationExecutionPolicy.detect()
    println(
        "MTC-WORKER phase=elevation status=started format=${request.outputFormat.name} " +
            "workers=${executionOptions.workerCount} bufferBytes=${executionOptions.bufferBytesPerWorker}",
    )
    var lastReportNanos = 0L
    val report = ElevationConverter.convert(
        request = ElevationConversionRequest(
            inputs = request.inputs,
            output = request.output,
            outputFormat = request.outputFormat,
            areas = request.areas,
            executionOptions = executionOptions,
        ),
        onProgress = { completed, total ->
            val now = System.nanoTime()
            if (completed == total || now - lastReportNanos >= PROGRESS_REPORT_INTERVAL_NANOS) {
                println("MTC-WORKER phase=elevation status=running current=$completed total=$total")
                lastReportNanos = now
            }
        },
    )
    val bytes = report.outputs.sumOf { Files.size(it) }
    println(
        "MTC-WORKER phase=elevation status=completed outputs=${report.outputs.size} " +
            "cells=${report.sampledCells} populated=${report.populatedCells} bytes=$bytes",
    )
}

private fun runPoiConvert(request: PoiConvertRequest) {
    Files.createDirectories(request.output.parent ?: error("POI-Ausgabe besitzt kein Verzeichnis."))
    Files.createDirectories(request.temporaryDirectory)
    require(!Files.exists(request.output)) { "Die POI-Ausgabedatei existiert bereits: ${request.output}" }
    val temporary = request.output.partial()
    Files.deleteIfExists(temporary)
    val progress = startHeartbeat("poi-convert")
    try {
        println(
            "MTC-WORKER phase=poi-convert status=started " +
                "source=${request.sourceSchema.metadataVersion} target=${request.targetSchema.metadataVersion}",
        )
        Files.copy(request.input, temporary)
        request.areas?.let { areas ->
            val report = MapsforgePoiSpatialFilter.filter(temporary, request.sourceSchema, areas)
            println(
                "MTC-WORKER phase=poi-filter status=completed " +
                    "selected=${report.selectedCount} source=${report.sourceCount}",
            )
        }
        MapsforgePoiSchemaConverter.convert(temporary, request.sourceSchema, request.targetSchema)
        moveAtomically(temporary, request.output)
        println("MTC-WORKER phase=poi-convert status=completed bytes=${Files.size(request.output)}")
    } finally {
        progress.set(false)
        Files.deleteIfExists(temporary)
    }
}

private fun runPoiMerge(request: PoiMergeRequest) {
    Files.createDirectories(request.output.parent ?: error("POI-Ausgabe besitzt kein Verzeichnis."))
    Files.createDirectories(request.temporaryDirectory)
    require(!Files.exists(request.output)) { "Die POI-Ausgabedatei existiert bereits: ${request.output}" }
    val outputTemporary = request.output.partial()
    val normalizedFiles = mutableListOf<Path>()
    val progress = startHeartbeat("poi-merge")
    Files.deleteIfExists(outputTemporary)
    try {
        println(
            "MTC-WORKER phase=poi-merge status=started sources=${request.sources.size} " +
                "target=${request.targetSchema.metadataVersion}",
        )
        request.sources.forEachIndexed { index, source ->
            val normalized = request.temporaryDirectory.resolve(
                "poi-merge-${ProcessHandle.current().pid()}-${index.toString().padStart(4, '0')}.poi",
            )
            Files.deleteIfExists(normalized)
            normalizedFiles.add(normalized)
            Files.copy(source.path, normalized)
            val report = MapsforgePoiSpatialFilter.filter(normalized, source.schema, request.areas)
            MapsforgePoiSchemaConverter.convert(normalized, source.schema, request.targetSchema)
            println(
                "MTC-WORKER phase=poi-filter status=completed source=${index + 1} " +
                    "selected=${report.selectedCount} total=${report.sourceCount}",
            )
        }
        val report = MapsforgePoiMerger.merge(normalizedFiles, outputTemporary, request.targetSchema)
        moveAtomically(outputTemporary, request.output)
        println(
            "MTC-WORKER phase=poi-merge status=completed sources=${request.sources.size} " +
                "input=${report.sourceCount} output=${report.outputCount} duplicates=${report.duplicateCount} " +
                "bytes=${Files.size(request.output)}",
        )
    } finally {
        progress.set(false)
        Files.deleteIfExists(outputTemporary)
        normalizedFiles.forEach(Files::deleteIfExists)
    }
}

private fun runMapsforgeSource(request: MapsforgeSourceRequest) {
    Files.createDirectories(request.output.parent ?: error("Ausgabeziel besitzt kein Verzeichnis."))
    require(!Files.exists(request.output)) { "Die Ausgabedatei existiert bereits: ${request.output}" }
    println("MTC-WORKER phase=mapsforge-source status=started source=${request.input.fileName}")
    var lastSourceReportNanos = 0L
    var lastWriteReportNanos = 0L
    var lastWritePhase = ""
    val report = MapsforgeMapSource.exportToPbf(
        source = request.input,
        output = request.output,
        areas = request.areas,
        namespace = request.namespace,
        temporaryDirectory = request.temporaryDirectory,
        onSourceProgress = { completed, total, selected ->
            val now = System.nanoTime()
            if (completed == total || now - lastSourceReportNanos >= PROGRESS_REPORT_INTERVAL_NANOS) {
                println(
                    "MTC-WORKER phase=mapsforge-source status=running " +
                        "current=$completed total=$total selected=$selected",
                )
                lastSourceReportNanos = now
            }
        },
        onProgress = { progress ->
            val now = System.nanoTime()
            val phase = progress.phase.name
            if (phase != lastWritePhase || now - lastWriteReportNanos >= PROGRESS_REPORT_INTERVAL_NANOS) {
                println(
                    "MTC-WORKER phase=mapsforge-source status=running " +
                        "writePhase=$phase current=${progress.writtenObjects} " +
                        "total=${progress.expectedObjects ?: 0L} blocks=${progress.writtenBlocks}",
                )
                lastWritePhase = phase
                lastWriteReportNanos = now
            }
        },
    )
    println(
        "MTC-WORKER phase=mapsforge-source status=completed " +
            "nodes=${report.nodeCount} ways=${report.wayCount} relations=${report.relationCount} " +
            "bytes=${report.outputBytes}",
    )
}

private const val PROGRESS_REPORT_INTERVAL_NANOS = 250_000_000L

private fun runCombine(request: CombineRequest) {
    val outputParent = request.output.parent ?: error("Ausgabeziel besitzt kein Verzeichnis.")
    Files.createDirectories(outputParent)
    Files.createDirectories(request.temporaryDirectory)
    require(!Files.exists(request.output)) { "Die Ausgabedatei existiert bereits: ${request.output}" }
    val workDirectory = Files.createTempDirectory(request.temporaryDirectory, "combine-")
    val temporaryOutput = request.output.resolveSibling("${request.output.fileName}.partial")
    Files.deleteIfExists(temporaryOutput)
    try {
        val relevantInputs = request.inputs.filter { input ->
            PbfHeaderReader.read(input).bounds?.let { sourceBounds ->
                request.areas.any { sourceBounds.intersects(it.bounds) }
            } != false
        }
        require(relevantInputs.isNotEmpty()) { "Keine Quelle überschneidet den gewählten Exportbereich." }
        val croppedInputs = relevantInputs.mapIndexed { index, input ->
            if (input in request.preCroppedInputs) {
                println(
                    "MTC-WORKER phase=combine-crop status=skipped current=${index + 1} " +
                        "total=${relevantInputs.size} source=${input.fileName}",
                )
                return@mapIndexed input
            }
            println(
                "MTC-WORKER phase=combine-crop status=running current=${index + 1} total=${relevantInputs.size}",
            )
            val cropped = workDirectory.resolve("source-${index.toString().padStart(3, '0')}.osm.pbf")
            val cropRequest = CropRequest(
                input,
                cropped,
                request.bounds,
                request.cacheDirectory,
                request.temporaryDirectory,
                request.areas,
            )
            try {
                runFastCrop(cropRequest)
            } catch (exception: Throwable) {
                if (cropRequest.areas.any { it !is de.arxsilex.maptools.converter.core.GeoRectangleRegion }) throw exception
                Files.deleteIfExists(cropped)
                val fallback = cropped.resolveSibling("${cropped.fileName}.partial")
                Files.deleteIfExists(fallback)
                System.err.println(
                    "MTC-WORKER phase=combine-crop status=fallback source=${input.fileName} " +
                        "message=${exception.message ?: exception.javaClass.simpleName}",
                )
                runOsmosisCrop(cropRequest, fallback)
                moveAtomically(fallback, cropped)
            }
            cropped
        }
        if (croppedInputs.size == 1) {
            val single = croppedInputs.single()
            if (single in request.preCroppedInputs) {
                Files.copy(single, temporaryOutput)
                moveAtomically(temporaryOutput, request.output)
            } else {
                moveAtomically(single, request.output)
            }
        } else {
            println("MTC-WORKER phase=combine-merge status=started sources=${croppedInputs.size}")
            val heartbeat = startHeartbeat("combine-merge")
            try {
                Osmosis.run(request.osmosisMergeArguments(croppedInputs, temporaryOutput))
            } finally {
                heartbeat.set(false)
            }
            require(Files.isRegularFile(temporaryOutput) && Files.size(temporaryOutput) > 0L) {
                "Osmosis hat keine kombinierte PBF-Datei erzeugt."
            }
            moveAtomically(temporaryOutput, request.output)
        }
        PbfHeaderReader.read(request.output)
        println("MTC-WORKER phase=combine-merge status=completed bytes=${Files.size(request.output)}")
    } finally {
        Files.deleteIfExists(temporaryOutput)
        if (Files.exists(workDirectory)) deleteTree(workDirectory)
    }
}

private fun runCrop(request: CropRequest) {
    val outputParent = request.output.parent ?: error("Ausgabeziel besitzt kein Verzeichnis.")
    Files.createDirectories(outputParent)
    require(!Files.exists(request.output)) { "Die Ausgabedatei existiert bereits: ${request.output}" }
    val temporary = request.output.resolveSibling("${request.output.fileName}.partial")
    Files.deleteIfExists(temporary)
    var fallbackProgress: AtomicBoolean? = null
    try {
        println("MTC-WORKER phase=crop status=started")
        try {
            runFastCrop(request)
        } catch (exception: Throwable) {
            if (request.areas.any { it !is de.arxsilex.maptools.converter.core.GeoRectangleRegion }) throw exception
            Files.deleteIfExists(request.output.resolveSibling("${request.output.fileName}.partial"))
            Files.deleteIfExists(request.output)
            System.err.println(
                "MTC-WORKER phase=crop status=fallback message=${exception.message ?: exception.javaClass.simpleName}",
            )
            fallbackProgress = startHeartbeat("crop")
            runOsmosisCrop(request, temporary)
            require(Files.isRegularFile(temporary) && Files.size(temporary) > 0L) {
                "Osmosis hat keine gültige Ausgabedatei erzeugt."
            }
            moveAtomically(temporary, request.output)
        }
        println("MTC-WORKER phase=crop status=completed bytes=${Files.size(request.output)}")
    } finally {
        fallbackProgress?.set(false)
        Files.deleteIfExists(temporary)
    }
}

private fun runFastCrop(request: CropRequest) {
    Files.createDirectories(request.cacheDirectory)
    cleanupStaleIntermediateDirectories(request.cacheDirectory)
    println("MTC-WORKER phase=mtci-index status=started")
    val index = PbfBlockScanner.scan(request.input, onProgress = { scan ->
        if (scan.blocksRead % 100 == 0 || scan.bytesRead == scan.totalBytes) {
            println(
                "MTC-WORKER phase=mtci-index status=running current=${scan.bytesRead} total=${scan.totalBytes}",
            )
        }
    })
    val cacheKey = "${index.sourceSha256}-${index.sourceSize}"
    val fullMtci = request.cacheDirectory.resolve(cacheKey)
    val manifest = fullMtci.resolve("manifest.json")
    val cacheValid = Files.isRegularFile(manifest) &&
        Files.readString(manifest).contains("\"sourceSha256\": \"${index.sourceSha256}\"")
    if (!cacheValid) {
        if (Files.exists(fullMtci)) deleteTree(fullMtci)
        println("MTC-WORKER phase=mtci-build status=started")
        MtciIntermediateBuilder.build(
            sourcePath = request.input,
            index = index,
            outputDirectory = fullMtci,
            sourceBounds = PbfHeaderReader.read(request.input).bounds,
            options = MtciBuildOptions(),
            onProgress = { build ->
                if (build.completedBlocks % 50 == 0 || build.completedBlocks == build.totalBlocks) {
                    println(
                        "MTC-WORKER phase=mtci-build status=running " +
                            "current=${build.completedBlocks} total=${build.totalBlocks}",
                    )
                }
            },
        )
        println("MTC-WORKER phase=mtci-build status=completed")
    } else {
        println("MTC-WORKER phase=mtci-build status=cached")
    }

    val selectionTotals = readSelectionTotals(manifest)
    val totalSelectionWork = selectionTotals.nodes + selectionTotals.ways * 2L + selectionTotals.relations
    println("MTC-WORKER phase=mtci-select status=started")
    val selection = MtciSpatialSelector.analyze(fullMtci, request.areas, onProgress = { spatial ->
        val completedWork = when (spatial.phase) {
            MtciSpatialPhase.NODES -> spatial.visitedObjects
            MtciSpatialPhase.WAYS -> selectionTotals.nodes + spatial.visitedObjects
            MtciSpatialPhase.RELATIONS ->
                selectionTotals.nodes + selectionTotals.ways + spatial.visitedObjects
            MtciSpatialPhase.RELATION_DEPENDENCIES ->
                selectionTotals.nodes + selectionTotals.ways + selectionTotals.relations
            MtciSpatialPhase.WAY_DEPENDENCIES ->
                selectionTotals.nodes + selectionTotals.ways + selectionTotals.relations + spatial.visitedObjects
        }.coerceAtMost(totalSelectionWork)
        println(
            "MTC-WORKER phase=mtci-select status=running current=$completedWork " +
                "total=$totalSelectionWork",
        )
    })
    println(
        "MTC-WORKER phase=mtci-select status=completed " +
            "current=$totalSelectionWork total=$totalSelectionWork",
    )
    val selectedMtci = request.cacheDirectory.resolve("selection-${java.util.UUID.randomUUID()}")
    try {
        println("MTC-WORKER phase=mtci-materialize status=started")
        val materialized = MtciSelectionMaterializer.materialize(
            sourceDirectory = fullMtci,
            selection = selection,
            outputDirectory = selectedMtci,
            onProgress = { materialization ->
                println(
                    "MTC-WORKER phase=mtci-materialize-${materialization.phase.name.lowercase()} status=running " +
                        "current=${materialization.writtenObjects} total=${materialization.expectedObjects}",
                )
            },
        )
        println("MTC-WORKER phase=mtci-pbf status=started")
        MtciPbfWriter.write(
            sourceDirectory = selectedMtci,
            outputPath = request.output,
            bounds = request.bounds,
            onProgress = { pbf ->
                if (pbf.writtenBlocks % 10 == 0) {
                    val total = when (pbf.phase) {
                        de.arxsilex.maptools.converter.core.pbf.MtciPbfWritePhase.NODES -> materialized.nodeCount
                        de.arxsilex.maptools.converter.core.pbf.MtciPbfWritePhase.WAYS -> materialized.wayCount
                        de.arxsilex.maptools.converter.core.pbf.MtciPbfWritePhase.RELATIONS -> materialized.relationCount
                    }
                    println(
                        "MTC-WORKER phase=mtci-pbf-${pbf.phase.name.lowercase()} status=running " +
                            "current=${pbf.writtenObjects} total=$total blocks=${pbf.writtenBlocks}",
                    )
                }
            },
        )
        PbfHeaderReader.read(request.output)
        println("MTC-WORKER phase=mtci-pbf status=completed")
    } finally {
        if (Files.exists(selectedMtci)) deleteTree(selectedMtci)
    }
}

private fun runOsmosisCrop(request: CropRequest, output: Path) {
    if (request.areas.size == 1) {
        Osmosis.run(request.osmosisArguments(output))
        return
    }
    val workDirectory = Files.createTempDirectory(request.temporaryDirectory, "osmosis-multi-")
    try {
        val pieces = request.areas.mapIndexed { index, area ->
            val piece = workDirectory.resolve("area-${index.toString().padStart(3, '0')}.osm.pbf")
            val single = request.copy(bounds = area.bounds, areas = listOf(area))
            Osmosis.run(single.osmosisArguments(piece))
            piece
        }
        if (pieces.size == 1) {
            moveAtomically(pieces.single(), output)
        } else {
            val merge = CombineRequest(
                inputs = pieces,
                output = output,
                bounds = request.bounds,
                cacheDirectory = request.cacheDirectory,
                temporaryDirectory = request.temporaryDirectory,
                areas = request.areas,
            )
            Osmosis.run(merge.osmosisMergeArguments(pieces, output))
        }
    } finally {
        deleteTree(workDirectory)
    }
}

private data class SelectionTotals(val nodes: Long, val ways: Long, val relations: Long)

private fun readSelectionTotals(manifest: Path): SelectionTotals {
    val json = Files.readString(manifest)
    fun count(name: String): Long = Regex("\\\"$name\\\"\\s*:\\s*([0-9]+)")
        .find(json)
        ?.groupValues
        ?.get(1)
        ?.toLongOrNull()
        ?: error("Im MTCI-Manifest fehlt $name.")
    return SelectionTotals(
        nodes = count("nodeCount"),
        ways = count("wayCount"),
        relations = count("relationCount"),
    )
}

private fun cleanupStaleIntermediateDirectories(cacheDirectory: Path) {
    Files.list(cacheDirectory).use { paths ->
        paths.filter { path ->
            val name = path.fileName.toString()
            name.startsWith(".selection-") || name.startsWith("selection-") || name.contains(".partial-")
        }.forEach { path -> if (Files.isDirectory(path)) deleteTree(path) }
    }
}

private fun deleteTree(path: Path) {
    val normalized = path.toAbsolutePath().normalize()
    if (!Files.exists(normalized)) return
    Files.walk(normalized).use { paths ->
        paths.sorted(Comparator.reverseOrder()).forEach(Files::deleteIfExists)
    }
}

private fun moveAtomically(source: Path, target: Path) {
    try {
        Files.move(source, target, StandardCopyOption.ATOMIC_MOVE)
    } catch (_: AtomicMoveNotSupportedException) {
        Files.move(source, target)
    }
}

private fun runMapsforge(request: MapsforgeRequest) {
    Files.createDirectories(request.temporaryDirectory)
    request.mapOutput?.let { output ->
        val tagMapping = Files.createTempFile(request.temporaryDirectory, "tag-mapping-", ".xml")
        try {
            MapsforgeTagMapping.write(request.content, tagMapping)
            runMapsforgeOutput("map", output, request.mapArguments(output.partial(), tagMapping))
        } finally {
            Files.deleteIfExists(tagMapping)
        }
    }
    request.poiOutput?.let { output ->
        runMapsforgeOutput("poi", output, request.poiArguments(output.partial())) { temporary ->
            if (request.poiSchemaVersion == de.arxsilex.maptools.converter.core.PoiSchemaVersion.V3) {
                println("MTC-WORKER phase=mapsforge-poi-schema-v3 status=started")
                MapsforgePoiSchemaConverter.convertV4ToV3(temporary)
                println("MTC-WORKER phase=mapsforge-poi-schema-v3 status=completed")
            }
        }
    }
}

private fun runMapsforgeOutput(
    kind: String,
    output: java.nio.file.Path,
    arguments: Array<String>,
    afterWrite: (java.nio.file.Path) -> Unit = {},
) {
    Files.createDirectories(output.toAbsolutePath().parent)
    require(!Files.exists(output)) { "Die Ausgabedatei existiert bereits: $output" }
    val temporary = output.partial()
    Files.deleteIfExists(temporary)
    val progress = startHeartbeat("mapsforge-$kind")
    try {
        val processor = arguments.firstOrNull { it.startsWith("type=") }?.substringAfter('=')
        val processorDetail = processor?.let { " processor=$it" }.orEmpty()
        println("MTC-WORKER phase=mapsforge-$kind status=started$processorDetail")
        Osmosis.run(arguments)
        require(Files.isRegularFile(temporary) && Files.size(temporary) > 0L) {
            "Der Mapsforge-$kind-Writer hat keine Datei erzeugt."
        }
        afterWrite(temporary)
        try {
            Files.move(temporary, output, StandardCopyOption.ATOMIC_MOVE)
        } catch (_: AtomicMoveNotSupportedException) {
            Files.move(temporary, output)
        }
        println("MTC-WORKER phase=mapsforge-$kind status=completed bytes=${Files.size(output)}")
    } finally {
        progress.set(false)
        Files.deleteIfExists(temporary)
    }
}

private fun startHeartbeat(phase: String): AtomicBoolean {
    val active = AtomicBoolean(true)
    val startedAt = System.nanoTime()
    thread(name = "maptools-worker-progress-$phase", isDaemon = true) {
        while (active.get()) {
            Thread.sleep(5_000)
            if (active.get()) {
                val seconds = (System.nanoTime() - startedAt) / 1_000_000_000L
                println("MTC-WORKER phase=$phase status=running elapsedSeconds=$seconds")
            }
        }
    }
    return active
}

private fun configureTemporaryDirectory(arguments: Array<String>) {
    val index = arguments.indexOf("--temp-dir")
    val directory = if (index >= 0 && index + 1 < arguments.size) {
        Path.of(arguments[index + 1])
    } else {
        Path.of(".maptools-temp")
    }.toAbsolutePath().normalize()
    Files.createDirectories(directory)
    System.setProperty("java.io.tmpdir", directory.toString())
}

private fun java.nio.file.Path.partial(): java.nio.file.Path = resolveSibling("${fileName}.partial")
