package de.arxsilex.maptools.converter.worker

import de.arxsilex.maptools.converter.core.PoiSchemaVersion
import java.nio.ByteBuffer
import java.nio.file.Path
import java.security.MessageDigest
import java.sql.Connection
import java.sql.DriverManager

internal data class PoiMergeReport(
    val sourceCount: Long,
    val outputCount: Long,
    val duplicateCount: Long,
)

/** Merges normalized Mapsforge POI databases while remapping local IDs and category trees. */
internal object MapsforgePoiMerger {
    fun merge(inputs: List<Path>, output: Path, schema: PoiSchemaVersion): PoiMergeReport {
        require(inputs.isNotEmpty()) { "Mindestens eine normalisierte POI-Datenbank ist erforderlich." }
        Class.forName("org.sqlite.JDBC")
        java.nio.file.Files.copy(inputs.first(), output)
        DriverManager.getConnection("jdbc:sqlite:${output.toAbsolutePath()}").use { target ->
            target.autoCommit = false
            try {
                if (schema == PoiSchemaVersion.V4) {
                    target.createStatement().use { it.execute("DROP TABLE IF EXISTS poi_data_fts") }
                }
                target.createStatement().use {
                    it.execute("CREATE TEMP TABLE merge_seen (signature BLOB PRIMARY KEY) WITHOUT ROWID")
                }
                val targetCategories = CategoryTree.read(target)
                val targetCategoryIds = targetCategories.paths.entries.associate { (id, path) -> path to id }.toMutableMap()
                var nextCategoryId = (targetCategories.paths.keys.maxOrNull() ?: 0L) + 1L
                var nextPoiId = target.singleLong("SELECT COALESCE(MAX(id), 0) FROM poi_data") + 1L
                var sourceCount = target.singleLong("SELECT COUNT(*) FROM poi_data")
                var duplicateCount = 0L

                PoiTargetWriter(target, schema).use { writer ->
                    val identityCategoryMap = targetCategories.paths.keys.associateWith { it }
                    target.forEachPoi(schema, identityCategoryMap) { poi -> writer.remember(poi.signature()) }

                    inputs.drop(1).forEach { input ->
                        val inputUri = input.toAbsolutePath().toUri()
                        DriverManager.getConnection("jdbc:sqlite:$inputUri?mode=ro").use { source ->
                            val categories = CategoryTree.read(source)
                            val categoryMap = mutableMapOf<Long, Long>()
                            categories.paths.entries.sortedBy { it.value.size }.forEach { (sourceId, path) ->
                                val targetId = targetCategoryIds[path] ?: run {
                                    val parentId = path.dropLast(1).takeIf(List<String>::isNotEmpty)?.let { parentPath ->
                                        targetCategoryIds[parentPath]
                                            ?: error("POI-Kategorie besitzt keinen zusammengeführten Elternknoten: $parentPath")
                                    }
                                    val createdId = nextCategoryId++
                                    target.prepareStatement(
                                        "INSERT INTO poi_categories (id, name, parent) VALUES (?, ?, ?)",
                                    ).use { insert ->
                                        insert.setLong(1, createdId)
                                        insert.setString(2, path.last())
                                        if (parentId == null) {
                                            insert.setNull(3, java.sql.Types.INTEGER)
                                        } else {
                                            insert.setLong(3, parentId)
                                        }
                                        insert.executeUpdate()
                                    }
                                    targetCategoryIds[path] = createdId
                                    createdId
                                }
                                categoryMap[sourceId] = targetId
                            }
                            sourceCount += source.singleLong("SELECT COUNT(*) FROM poi_data")
                            source.forEachPoi(schema, categoryMap) { sourcePoi ->
                                val signature = sourcePoi.signature()
                                if (!writer.remember(signature)) {
                                    duplicateCount++
                                } else {
                                    writer.insert(nextPoiId++, sourcePoi)
                                }
                            }
                        }
                    }
                }
                target.prepareStatement("UPDATE metadata SET value = ? WHERE name = 'writer'").use { update ->
                    update.setString(1, "MapTools Converter (merged Mapsforge POI v${schema.metadataVersion})")
                    update.executeUpdate()
                }
                if (schema == PoiSchemaVersion.V4) {
                    target.createStatement().use { statement ->
                        statement.execute(
                            "CREATE VIRTUAL TABLE poi_data_fts USING fts5(data, content='poi_data', content_rowid='id')",
                        )
                        statement.execute("INSERT INTO poi_data_fts(poi_data_fts) VALUES ('rebuild')")
                        statement.execute("CREATE INDEX IF NOT EXISTS poi_category_map_idx ON poi_category_map (category)")
                    }
                }
                val outputCount = target.singleLong("SELECT COUNT(*) FROM poi_data")
                require(outputCount == target.singleLong("SELECT COUNT(*) FROM poi_index")) {
                    "POI-Daten und räumlicher Index sind nach dem Zusammenführen unvollständig."
                }
                require(outputCount + duplicateCount == sourceCount) {
                    "Die POI-Zusammenführung hat Einträge verloren ($sourceCount/$outputCount/$duplicateCount)."
                }
                target.commit()
                target.autoCommit = true
                target.createStatement().use { it.execute("VACUUM") }
                return PoiMergeReport(sourceCount, outputCount, duplicateCount)
            } catch (exception: Exception) {
                target.rollback()
                throw exception
            } finally {
                if (!target.autoCommit) target.autoCommit = true
            }
        }
    }

    private data class PoiRecord(
        val latitude: Double,
        val longitude: Double,
        val data: String?,
        val categories: List<Long>,
    ) {
        fun signature(): ByteArray {
            val digest = MessageDigest.getInstance("SHA-256")
            digest.update(ByteBuffer.allocate(16).putLong(latitude.toBits()).putLong(longitude.toBits()).array())
            val encodedData = data?.toByteArray(Charsets.UTF_8) ?: byteArrayOf()
            digest.update(ByteBuffer.allocate(4).putInt(if (data == null) -1 else encodedData.size).array())
            digest.update(encodedData)
            categories.sorted().forEach { digest.update(ByteBuffer.allocate(8).putLong(it).array()) }
            return digest.digest()
        }
    }

    private data class CategoryTree(val paths: Map<Long, List<String>>) {
        companion object {
            fun read(connection: Connection): CategoryTree {
                data class Category(val name: String, val parent: Long?)
                val categories = linkedMapOf<Long, Category>()
                connection.createStatement().use { statement ->
                    statement.executeQuery("SELECT id, name, parent FROM poi_categories").use { result ->
                        while (result.next()) {
                            val parentValue = result.getLong("parent")
                            val parentIsNull = result.wasNull()
                            categories[result.getLong("id")] = Category(
                                result.getString("name") ?: "",
                                parentValue.takeUnless { parentIsNull },
                            )
                        }
                    }
                }
                val resolved = mutableMapOf<Long, List<String>>()
                fun resolve(id: Long, active: MutableSet<Long>): List<String> {
                    resolved[id]?.let { return it }
                    require(active.add(id)) { "Zyklische POI-Kategoriehierarchie bei ID $id." }
                    val category = categories[id] ?: error("Fehlende POI-Kategorie $id.")
                    val path = category.parent?.let { resolve(it, active) }.orEmpty() + category.name
                    active.remove(id)
                    resolved[id] = path
                    return path
                }
                categories.keys.forEach { resolve(it, mutableSetOf()) }
                return CategoryTree(resolved)
            }
        }
    }

    private class PoiTargetWriter(
        connection: Connection,
        private val schema: PoiSchemaVersion,
    ) : AutoCloseable {
        private val seen = connection.prepareStatement("INSERT OR IGNORE INTO merge_seen (signature) VALUES (?)")
        private val data = connection.prepareStatement("INSERT INTO poi_data (id, data) VALUES (?, ?)")
        private val index = connection.prepareStatement(
            when (schema) {
                PoiSchemaVersion.V3 -> "INSERT INTO poi_index (id, lat, lon) VALUES (?, ?, ?)"
                PoiSchemaVersion.V4 ->
                    "INSERT INTO poi_index (id, minLat, maxLat, minLon, maxLon) VALUES (?, ?, ?, ?, ?)"
            },
        )
        private val category = connection.prepareStatement(
            "INSERT INTO poi_category_map (id, category) VALUES (?, ?)",
        )

        fun remember(signature: ByteArray): Boolean {
            seen.setBytes(1, signature)
            return seen.executeUpdate() == 1
        }

        fun insert(id: Long, poi: PoiRecord) {
            data.setLong(1, id)
            data.setString(2, poi.data)
            data.executeUpdate()

            index.setLong(1, id)
            index.setDouble(2, poi.latitude)
            if (schema == PoiSchemaVersion.V3) {
                index.setDouble(3, poi.longitude)
            } else {
                index.setDouble(3, poi.latitude)
                index.setDouble(4, poi.longitude)
                index.setDouble(5, poi.longitude)
            }
            index.executeUpdate()

            poi.categories.distinct().forEach { categoryId ->
                category.setLong(1, id)
                category.setLong(2, categoryId)
                category.addBatch()
            }
            category.executeBatch()
        }

        override fun close() {
            category.close()
            index.close()
            data.close()
            seen.close()
        }
    }

    private fun Connection.forEachPoi(
        schema: PoiSchemaVersion,
        categoryMap: Map<Long, Long>,
        consume: (PoiRecord) -> Unit,
    ) {
        val coordinates = when (schema) {
            PoiSchemaVersion.V3 -> "i.lat, i.lon"
            PoiSchemaVersion.V4 -> "(i.minLat + i.maxLat) / 2.0, (i.minLon + i.maxLon) / 2.0"
        }
        val sql = "SELECT d.id, d.data, $coordinates, m.category " +
            "FROM poi_data d JOIN poi_index i ON i.id = d.id " +
            "LEFT JOIN poi_category_map m ON m.id = d.id ORDER BY d.id, m.category"
        createStatement().use { statement ->
            statement.fetchSize = 4_096
            statement.executeQuery(sql).use { result ->
                var currentId: Long? = null
                var latitude = 0.0
                var longitude = 0.0
                var data: String? = null
                val categories = mutableListOf<Long>()
                fun emit() {
                    if (currentId != null) consume(PoiRecord(latitude, longitude, data, categories.toList()))
                }
                while (result.next()) {
                    val id = result.getLong(1)
                    if (currentId != id) {
                        emit()
                        currentId = id
                        data = result.getString(2)
                        latitude = result.getDouble(3)
                        longitude = result.getDouble(4)
                        categories.clear()
                    }
                    val sourceCategory = result.getLong(5)
                    if (!result.wasNull()) {
                        categories += categoryMap[sourceCategory]
                            ?: error("POI verweist auf eine unbekannte Kategorie $sourceCategory.")
                    }
                }
                emit()
            }
        }
    }

    private fun Connection.singleLong(sql: String): Long = createStatement().use { statement ->
        statement.executeQuery(sql).use { result ->
            check(result.next()) { "Die POI-Datenbankabfrage lieferte kein Ergebnis." }
            result.getLong(1)
        }
    }
}
