mad wicked sick tile grouping optimisations

modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/common/positions.kt

@@ -3,6 +3,8 @@ package dev.adamko.kafkatorio.schema.common
 import dev.adamko.kxstsgen.core.*
 import dev.adamko.kxstsgen.core.experiments.TupleSerializer
 import kotlin.math.floor
+import kotlin.math.pow
+import kotlin.math.roundToInt
 import kotlinx.serialization.Serializable
 
 /**
@@ -58,6 +60,33 @@ data class MapChunkPosition(
       )
     }
   }
+
+  private val bounds: MapChunkPositionBounds by lazy {
+    MapChunkPositionBounds(this, ChunkSize.CHUNK_032)
+  }
+
+  operator fun contains(tilePosition: MapTilePosition): Boolean {
+    return tilePosition in bounds
+  }
+}
+
+
+@Serializable
+data class MapChunkPositionBounds(
+  val leftTop: MapTilePosition,
+  val rightBottom: MapTilePosition,
+) : ClosedRange<MapTilePosition> {
+
+  constructor(chunkPosition: MapChunkPosition, chunkSize: ChunkSize) : this(
+    chunkPosition.leftTopTile(chunkSize),
+    chunkPosition.rightBottomTile(chunkSize),
+  )
+
+  override val start: MapTilePosition
+    get() = leftTop
+
+  override val endInclusive: MapTilePosition
+    get() = rightBottom
 }
 
 
@@ -71,7 +100,7 @@ data class MapChunkPosition(
 data class MapTilePosition(
   val x: Int,
   val y: Int,
-) {
+) : Comparable<MapTilePosition> {
   object Serializer : TupleSerializer<MapTilePosition>(
     "MapTilePosition",
     {
@@ -85,6 +114,12 @@ data class MapTilePosition(
       return MapTilePosition(x, y)
     }
   }
+
+  override operator fun compareTo(other: MapTilePosition): Int =
+    when {
+      other.x != x -> x.compareTo(other.x)
+      else         -> y.compareTo(other.y)
+    }
 }
 
 
@@ -100,21 +135,28 @@ operator fun MapTilePosition.div(divisor: Int) =
 operator fun MapTilePosition.plus(addend: Int) =
   MapTilePosition(x + addend, y + addend)
 
+operator fun MapTilePosition.plus(addend: MapTilePosition) =
+  MapTilePosition(x + addend.x, y + addend.y)
 
-fun MapTilePosition.toMapChunkPosition(chunkSize: Int = MAP_CHUNK_SIZE) =
+
+operator fun MapTilePosition.minus(subtrahend: MapTilePosition) =
+  MapTilePosition(x - subtrahend.x, y - subtrahend.y)
+
+
+fun MapTilePosition.toMapChunkPosition(chunkSize: ChunkSize) =
   MapChunkPosition(
-    floor(x.toDouble() / chunkSize.toDouble()).toInt(),
+    floor(x.toDouble() / chunkSize.lengthInTiles.toDouble()).toInt(),
-    floor(y.toDouble() / chunkSize.toDouble()).toInt(),
+    floor(y.toDouble() / chunkSize.lengthInTiles.toDouble()).toInt(),
   )
 
 
 fun MapEntityPosition.toMapTilePosition() =
   MapChunkPosition(floor(x).toInt(), floor(y).toInt())
 
-fun MapEntityPosition.toMapChunkPosition(chunkSize: Int = MAP_CHUNK_SIZE) =
+fun MapEntityPosition.toMapChunkPosition(chunkSize: ChunkSize) =
   MapChunkPosition(
-    floor(x / chunkSize.toDouble()).toInt(),
+    floor(x / chunkSize.lengthInTiles.toDouble()).toInt(),
-    floor(y / chunkSize.toDouble()).toInt(),
+    floor(y / chunkSize.lengthInTiles.toDouble()).toInt(),
   )
 
 
@@ -128,18 +170,17 @@ operator fun MapChunkPosition.minus(subtrahend: Int) =
   MapChunkPosition(x - subtrahend, y - subtrahend)
 
 
-fun MapChunkPosition.leftTopTile(chunkSize: Int = MAP_CHUNK_SIZE): MapTilePosition =
+fun MapChunkPosition.leftTopTile(chunkSize: ChunkSize): MapTilePosition =
   MapTilePosition(
-    x * chunkSize,
+    x * chunkSize.lengthInTiles,
-    y * chunkSize,
+    y * chunkSize.lengthInTiles,
   )
 
-
+fun MapChunkPosition.rightBottomTile(chunkSize: ChunkSize): MapTilePosition =
-fun MapChunkPosition.rightBottomTile(chunkSize: Int = MAP_CHUNK_SIZE): MapTilePosition =
+  leftTopTile(chunkSize) + (chunkSize.lengthInTiles - 1)
-  leftTopTile(chunkSize) + (chunkSize - 1)
 
 
-fun MapChunkPosition.iterateTiles(chunkSize: Int = MAP_CHUNK_SIZE): Iterator<MapTilePosition> {
+fun MapChunkPosition.iterateTiles(chunkSize: ChunkSize): Iterator<MapTilePosition> {
   val leftTop = leftTopTile(chunkSize)
   val rightBottom = rightBottomTile(chunkSize)
   return iterator {
@@ -158,3 +199,37 @@ fun MapChunkPosition.iterateTiles(chunkSize: Int = MAP_CHUNK_SIZE): Iterator<Map
 //
 //val MapChunkPosition.boundingBox: MapBoundingBox
 //  get() = MapBoundingBox(leftTopTile, rightBottomTile)
+
+
+@Serializable
+enum class ChunkSize(
+  val zoomLevel: Int,
+) : Comparable<ChunkSize> {
+  CHUNK_512(-1),
+  CHUNK_256(0),
+  CHUNK_128(1),
+  CHUNK_064(2),
+  CHUNK_032(3),
+  ;
+
+  /** 32, 64, 128, 256, or 512 */
+  val lengthInTiles: Int = 2f.pow(8 - zoomLevel).roundToInt()
+
+  init {
+    require(lengthInTiles > 0) { "tilesPerChunk $lengthInTiles must be positive" }
+    // 1000 & 0111 = 0000 =>  pow^2
+    // 1001 & 1000 = 1000 => !pow^2
+    require(lengthInTiles and (lengthInTiles - 1) == 0) {
+      "tilesPerChunk $lengthInTiles must be a power-of-two number"
+    }
+  }
+
+  companion object {
+    // cache values for better performance. KT-48872
+    val entries: Set<ChunkSize> = values().toSet()
+
+    val MAX: ChunkSize = entries.maxByOrNull { it.lengthInTiles }!!
+    val MIN: ChunkSize = entries.minByOrNull { it.lengthInTiles }!!
+    val STANDARD: ChunkSize = entries.first { it.zoomLevel == 0 }
+  }
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt

@@ -1,6 +1,7 @@
 package dev.adamko.kafkatorio.processor
 
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.processor.topology.colourMapChunks
 import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream
 import dev.adamko.kafkatorio.processor.topology.groupMapChunks
 import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer
@@ -54,7 +55,7 @@ internal class KafkatorioTopology(
 
   private fun groupTilesMapChunks() {
     val builder = StreamsBuilder()
-    val topology = groupMapChunks(builder)
+    val topology = colourMapChunks(builder)
     launchTopology("groupTilesMapChunks", topology)
   }
 

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt

@@ -0,0 +1,313 @@
+package dev.adamko.kafkatorio.processor.topology
+
+import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
+import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kafkatorio.schema.common.ChunkSize
+import dev.adamko.kafkatorio.schema.common.ColourHex
+import dev.adamko.kafkatorio.schema.common.leftTopTile
+import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
+import dev.adamko.kafkatorio.schema.packets.MapChunkUpdate
+import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate
+import dev.adamko.kotka.extensions.consumedAs
+import dev.adamko.kotka.extensions.groupedAs
+import dev.adamko.kotka.extensions.materializedAs
+import dev.adamko.kotka.extensions.producedAs
+import dev.adamko.kotka.extensions.repartitionedAs
+import dev.adamko.kotka.extensions.streams.filter
+import dev.adamko.kotka.extensions.streams.map
+import dev.adamko.kotka.extensions.streams.mapValues
+import dev.adamko.kotka.extensions.streams.peek
+import dev.adamko.kotka.extensions.streams.reduce
+import dev.adamko.kotka.extensions.tableJoined
+import dev.adamko.kotka.extensions.tables.join
+import dev.adamko.kotka.extensions.tables.toStream
+import dev.adamko.kotka.kxs.serde
+import kotlin.collections.component1
+import kotlin.collections.component2
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.Topology
+import org.apache.kafka.streams.kstream.KStream
+import org.apache.kafka.streams.kstream.KTable
+
+
+private const val pid: String = "colourMapChunks"
+
+
+/** Add colour to each [MapChunkUpdate] */
+fun colourMapChunks(builder: StreamsBuilder): Topology {
+
+  val mapChunksStream: KStream<FactorioServerId, MapChunkUpdate> = builder.streamPacketData()
+  val protosStream: KStream<FactorioServerId, PrototypesUpdate> = builder.streamPacketData()
+
+  val protosTable: KTable<FactorioServerId, TileColourDict> =
+    protosStream.createTilePrototypeTable()
+
+  val chunkTilesColouredTable032: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
+    mapChunksStream
+      .convertToServerMapChunkTiles()
+      .reduceServerMapChunkTilesToTable()
+      .enrichWithColourData(protosTable)
+
+  chunkTilesColouredTable032.streamTo(ChunkSize.CHUNK_032, TOPIC_GROUPED_MAP_CHUNKS_STATE)
+
+  val groupedMapChunks = builder.stream(
+    TOPIC_GROUPED_MAP_CHUNKS_STATE,
+    consumedAs(
+      "$pid.consume.grouped-map-chunks",
+      kxsBinary.serde<ServerMapChunkId>(),
+      kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
+    )
+  )
+
+  mapOf(
+    ChunkSize.CHUNK_032 to ChunkSize.CHUNK_064,
+    ChunkSize.CHUNK_064 to ChunkSize.CHUNK_128,
+    ChunkSize.CHUNK_128 to ChunkSize.CHUNK_256,
+    ChunkSize.CHUNK_256 to ChunkSize.CHUNK_512,
+  ).forEach { (from, to) ->
+    groupedMapChunks
+      .filter("$pid.combine-chunks.filter-$from-to-$to") { key, _ ->
+      key?.chunkSize == from
+    }.groupByChunkPosition(to)
+      .streamTo(to, TOPIC_GROUPED_MAP_CHUNKS_STATE)
+  }
+
+//  val chunkTilesColouredTable064 =
+//    chunkTilesColouredTable032.groupByChunkPosition(ChunkSize.CHUNK_064)
+//  chunkTilesColouredTable064.streamTo(ChunkSize.CHUNK_064, TOPIC_GROUPED_MAP_CHUNKS_STATE)
+//
+//  val chunkTilesColouredTable128 =
+//    chunkTilesColouredTable064.groupByChunkPosition(ChunkSize.CHUNK_128)
+//  chunkTilesColouredTable128.streamTo(ChunkSize.CHUNK_128, TOPIC_GROUPED_MAP_CHUNKS_STATE)
+//
+//  val chunkTilesColouredTable256 =
+//    chunkTilesColouredTable128.groupByChunkPosition(ChunkSize.CHUNK_256)
+//  chunkTilesColouredTable256.streamTo(ChunkSize.CHUNK_256, TOPIC_GROUPED_MAP_CHUNKS_STATE)
+//
+//  val chunkTilesColouredTable512 =
+//    chunkTilesColouredTable256.groupByChunkPosition(ChunkSize.CHUNK_512)
+//  chunkTilesColouredTable512.streamTo(ChunkSize.CHUNK_512, TOPIC_GROUPED_MAP_CHUNKS_STATE)
+
+  return builder.build()
+}
+
+
+private fun KStream<FactorioServerId, MapChunkUpdate>.convertToServerMapChunkTiles()
+    : KStream<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> {
+
+  val pid = "$pid.convertToMapTiles"
+
+  return map("$pid.map") { serverId: FactorioServerId, update: MapChunkUpdate ->
+
+    val mapTileList = update.tileDictionary?.toMapTileList() ?: emptyList()
+
+    val (validMapTilesList, invalidMapTilesList) = mapTileList.partition {
+      it.position in update.key.chunkPosition
+    }
+
+    if (invalidMapTilesList.isNotEmpty()) {
+      println("WARNING [reduceMapTilesToTable] MapChunkUpdate contained out-of-bounds tiles $invalidMapTilesList")
+    }
+
+    val validMapTilesToProtoHashCode =
+      validMapTilesList.associate { it.position to TileProtoHashCode(it) }
+
+    val chunkId = ServerMapChunkId(
+      serverId = serverId,
+      chunkPosition = update.key.chunkPosition,
+      surfaceIndex = update.key.surfaceIndex,
+      chunkSize = ChunkSize.CHUNK_032,
+    )
+
+    val chunkTiles = ServerMapChunkTiles(chunkId, validMapTilesToProtoHashCode)
+
+    chunkId to chunkTiles
+  }
+}
+
+
+private fun KStream<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>.reduceServerMapChunkTilesToTable()
+    : KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> {
+
+  val pid = "$pid.reduceServerMapChunkTilesToTable"
+
+  return repartition(
+    repartitionedAs(
+      "$pid.pre-table-repartition",
+      kxsBinary.serde<ServerMapChunkId>(),
+      kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
+      // force, otherwise KTable-KTable FK join doesn't work
+      numberOfPartitions = 1,
+    )
+  ).groupByKey(
+    groupedAs(
+      "$pid.group-by-key",
+      kxsBinary.serde<ServerMapChunkId>(),
+      kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
+    )
+  ).reduce(
+    "$pid.reduce",
+    materializedAs(
+      "$pid.reduce.store",
+      kxsBinary.serde<ServerMapChunkId>(),
+      kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
+    )
+  ) { chunkTiles, otherChunkTiles ->
+    chunkTiles + otherChunkTiles
+  }
+}
+
+
+private fun KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>.enrichWithColourData(
+  protosTable: KTable<FactorioServerId, TileColourDict>,
+): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
+
+  val pid = "$pid.enrichWithColourData"
+
+  return this.join(
+    other = protosTable,
+    tableJoined = tableJoined("$pid.join-tiles-with-prototypes"),
+    materialized = materializedAs(
+      "$pid.join-tiles-with-prototypes.store",
+      kxsBinary.serde(),
+      kxsBinary.serde(),
+    ),
+    foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<TileProtoHashCode> ->
+      chunkTiles.chunkId.serverId
+    }
+  ) { tiles: ServerMapChunkTiles<TileProtoHashCode>, colourDict: TileColourDict ->
+//        println("joining tiles:${tiles.map.size} with colourDict:${colourDict.map.size}")
+
+    val missingProtos = mutableSetOf<TileProtoHashCode>()
+
+    val tileColours = tiles.map.mapValues { (_, code) ->
+      colourDict.map.getOrElse(code) {
+        missingProtos.add(code)
+        ColourHex.TRANSPARENT
+      }
+    }
+
+    if (missingProtos.isNotEmpty()) {
+      println("[$pid] missing ${missingProtos.size} tile prototypes: ${missingProtos.joinToString { "${it.code}" }}")
+    }
+
+    println("[$pid] Set tile colours for chunk ${tiles.chunkId}")
+
+    ServerMapChunkTiles(tiles.chunkId, tileColours)
+  }
+}
+
+
+private fun KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.streamTo(
+  chunkSize: ChunkSize,
+  outputTopic: String = TOPIC_GROUPED_MAP_CHUNKS_STATE,
+) {
+  val pid = "$pid.output-chunk.${chunkSize.name}"
+
+  toStream("$pid.stream")
+    .filter("$pid.filter-tiles-not-empty") { _, v ->
+      !v?.map.isNullOrEmpty()
+    }
+    .mapValues("$pid.map-not-null") { _, v ->
+      requireNotNull(v)
+    }
+    .peek("$pid.print-group-result") { _, v ->
+      println("Grouping map tiles result: ${v.chunkId} / ${v.map.size}")
+    }
+    .to(
+      outputTopic,
+      producedAs(
+        "$pid.grouped-map-chunks",
+        kxsBinary.serde(),
+        kxsBinary.serde(),
+      )
+    )
+}
+
+
+///**
+// * Group [MapChunkUpdate]s by the Chunk position.
+// *
+// * They should already be grouped, but do it again to make sure, and to filter out empty updates.
+// */
+//private fun KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.groupByChunkPosition(
+//  chunkSize: ChunkSize = ChunkSize.CHUNK_032
+//): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
+//
+//  val pid = "$pid.groupByChunkPosition.${chunkSize.name}"
+//
+//  return toStream("$pid.intake-stream")
+//    .map("$pid.change-chunk-size") { chunkId: ServerMapChunkId, chunkTiles: ServerMapChunkTiles<ColourHex>? ->
+//
+//      val newChunkPosition = chunkId.chunkPosition
+//        .leftTopTile(chunkId.chunkSize)
+//        .toMapChunkPosition(chunkSize)
+//
+//      val newId = chunkId.copy(
+//        chunkSize = chunkSize,
+//        chunkPosition = newChunkPosition,
+//      )
+//
+//      val newChunkTiles = chunkTiles?.map ?: emptyMap()
+//
+//      newId to ServerMapChunkTiles(chunkId = newId, map = newChunkTiles)
+//    }.groupByKey(
+//      groupedAs(
+//        "$pid.group-by-key",
+//        kxsBinary.serde<ServerMapChunkId>(),
+//        kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
+//      )
+//    ).reduce(
+//      "$pid.reduce",
+//      materializedAs(
+//        "$pid.reduce.store",
+//        kxsBinary.serde<ServerMapChunkId>(),
+//        kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
+//      )
+//    ) { chunkTiles, otherChunkTiles ->
+//      chunkTiles + otherChunkTiles
+//    }
+//}
+
+/**
+ * Group [MapChunkUpdate]s by the Chunk position.
+ *
+ * They should already be grouped, but do it again to make sure, and to filter out empty updates.
+ */
+private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.groupByChunkPosition(
+  chunkSize: ChunkSize = ChunkSize.CHUNK_032
+): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
+
+  val pid = "$pid.groupByChunkPosition.${chunkSize.name}"
+
+  return map("$pid.change-chunk-size") { chunkId: ServerMapChunkId, chunkTiles: ServerMapChunkTiles<ColourHex>? ->
+
+    val newChunkPosition = chunkId.chunkPosition
+      .leftTopTile(chunkId.chunkSize)
+      .toMapChunkPosition(chunkSize)
+
+    val newId = chunkId.copy(
+      chunkSize = chunkSize,
+      chunkPosition = newChunkPosition,
+    )
+
+    val newChunkTiles = chunkTiles?.map ?: emptyMap()
+
+    newId to ServerMapChunkTiles(chunkId = newId, map = newChunkTiles)
+  }.groupByKey(
+    groupedAs(
+      "$pid.group-by-key",
+      kxsBinary.serde<ServerMapChunkId>(),
+      kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
+    )
+  ).reduce(
+    "$pid.reduce",
+    materializedAs(
+      "$pid.reduce.store",
+      kxsBinary.serde<ServerMapChunkId>(),
+      kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
+    )
+  ) { chunkTiles, otherChunkTiles ->
+    chunkTiles + otherChunkTiles
+  }
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt

@@ -4,6 +4,7 @@ import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
 import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED
 import dev.adamko.kafkatorio.processor.misc.DebounceProcessor.Companion.addDebounceProcessor
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kafkatorio.schema.common.ChunkSize
 import dev.adamko.kafkatorio.schema.common.ColourHex
 import dev.adamko.kafkatorio.schema.common.MapChunkPosition
 import dev.adamko.kafkatorio.schema.common.MapTile
@@ -61,7 +62,7 @@ fun groupMapChunks(builder: StreamsBuilder): Topology {
   val protosStream: KStream<FactorioServerId, PrototypesUpdate> = builder.streamPacketData()
 
   val tileProtoColourDict: KTable<FactorioServerId, TileColourDict> =
-    tileProtoColourDictionary(protosStream)
+    protosStream.createTilePrototypeTable()
 
   val groupedMapChunkTiles: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
     groupTilesIntoChunksWithColours(
@@ -128,7 +129,7 @@ private fun groupTilesIntoChunksWithColours(
           mapTiles
             .tiles
             .groupBy { tile ->
-              tile.position.toMapChunkPosition(standardChunkSize.lengthInTiles)
+              tile.position.toMapChunkPosition(standardChunkSize)
             }
             .map { (chunkPos, tiles) ->
 

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt

@@ -13,38 +13,3 @@ import kotlinx.serialization.Serializable
 value class FactorioServerId(private val id: String) {
   override fun toString() = id
 }
-
-
-@Serializable
-enum class ChunkSize(
-  val zoomLevel: Int,
-) : Comparable<ChunkSize> {
-  CHUNK_512(-1),
-  CHUNK_256(0),
-  CHUNK_128(1),
-  CHUNK_064(2),
-  CHUNK_032(3),
-  ;
-
-  /** 32, 64, 128, 256, or 512 */
-  val lengthInTiles: Int = 2f.pow(8 - zoomLevel).roundToInt()
-
-  init {
-    require(lengthInTiles > 0) { "tilesPerChunk $lengthInTiles must be positive" }
-    // 1000 & 0111 = 0000 =>  pow^2
-    // 1001 & 1000 = 1000 => !pow^2
-    require(lengthInTiles and (lengthInTiles - 1) == 0) {
-      "tilesPerChunk $lengthInTiles must be a power-of-two number"
-    }
-  }
-
-  companion object {
-    // cache values for better performance. KT-48872
-    val entries: Set<ChunkSize> = values().toSet()
-
-    val MAX: ChunkSize = entries.maxByOrNull { it.lengthInTiles }!!
-    val MIN: ChunkSize = entries.minByOrNull { it.lengthInTiles }!!
-    val STANDARD: ChunkSize = entries.first { it.zoomLevel == 0 }
-
-  }
-}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt

@@ -2,15 +2,22 @@ package dev.adamko.kafkatorio.processor.topology
 
 import dev.adamko.kafkatorio.processor.WebsocketServer
 import dev.adamko.kafkatorio.processor.admin.topicName
-import dev.adamko.kafkatorio.processor.serdes.jsonMapper
+import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
 import dev.adamko.kafkatorio.schema.packets.PlayerUpdate
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.kxs.serde
 import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
 import org.apache.kafka.streams.StreamsBuilder
 
 
+private val jsonMapper: Json = Json {
+  prettyPrint = false
+  serializersModule = jsonMapperKafkatorio.serializersModule
+}
+
+
 fun playerUpdatesToWsServer(
   websocketServer: WebsocketServer,
   builder: StreamsBuilder = StreamsBuilder(),

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt

@@ -32,12 +32,12 @@ value class TileColourDict(val map: Map<TileProtoHashCode, ColourHex>)
 
 
 /** Get the latest map-tile colours per server. */
-fun tileProtoColourDictionary(
+fun KStream<FactorioServerId, PrototypesUpdate>.createTilePrototypeTable()
-  factorioServerPacketStream: KStream<FactorioServerId, PrototypesUpdate>
+    : KTable<FactorioServerId, TileColourDict> {
-): KTable<FactorioServerId, TileColourDict> {
-  val pid = "tileProtoColourDictionary"
 
-  return factorioServerPacketStream.mapValues(
+  val pid = "createTilePrototypeTable"
+
+  return mapValues(
     "$pid.map-values"
   ) { _, protoPacket: PrototypesUpdate ->
     val map = protoPacket

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/saveMapTiles.kt

@@ -4,11 +4,13 @@ import com.sksamuel.scrimage.ImmutableImage
 import com.sksamuel.scrimage.ScaleMethod
 import com.sksamuel.scrimage.color.RGBColor
 import com.sksamuel.scrimage.nio.PngWriter
+import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
 import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED
 import dev.adamko.kafkatorio.processor.admin.TOPIC_SUBDIVIDED_MAP_TILES
 import dev.adamko.kafkatorio.processor.admin.TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED
 import dev.adamko.kafkatorio.processor.misc.DebounceProcessor.Companion.addDebounceProcessor
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kafkatorio.schema.common.ChunkSize
 import dev.adamko.kafkatorio.schema.common.ColourHex
 import dev.adamko.kafkatorio.schema.common.MapTilePosition
 import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
@@ -42,28 +44,28 @@ fun saveMapTiles(
   tileDirectory: Path,
 ): Topology {
 
-  val groupedMapChunkTiles: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
+//  val groupedMapChunkTiles: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
-    builder.stream(
+//    builder.stream(
-      TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED,
+//      TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED,
-      consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()),
+//      consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()),
-    )
+//    )
-
+//
-  groupedMapChunkTiles
+//  groupedMapChunkTiles
-    .flatMapSubdividedChunk()
+//    .flatMapSubdividedChunk()
-    .to(
+//    .to(
-      TOPIC_SUBDIVIDED_MAP_TILES,
+//      TOPIC_SUBDIVIDED_MAP_TILES,
-      producedAs(
+//      producedAs(
-        "$pid.produce.grouped-map-chunks",
+//        "$pid.produce.grouped-map-chunks",
-        kxsBinary.serde(),
+//        kxsBinary.serde(),
-        kxsBinary.serde(),
+//        kxsBinary.serde(),
-      )
+//      )
-    )
+//    )
 
   val subdividedMapChunkTilesDebounced: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
     builder.stream(
-      TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED,
+      TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED,
       consumedAs(
-        "$pid.consume.debounced-subdivided-chunks",
+        "$pid.consume.grouped-map-chunks",
         kxsBinary.serde(),
         kxsBinary.serde(),
       ),
@@ -74,8 +76,8 @@ fun saveMapTiles(
   return builder.build()
     .addDebounceProcessor<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>(
       namePrefix = pid,
-      sourceTopic = TOPIC_SUBDIVIDED_MAP_TILES,
+      sourceTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE,
-      sinkTopic = TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED,
+      sinkTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED,
       inactivityDuration = 15.seconds,
       keySerde = kxsBinary.serde(),
       valueSerde = kxsBinary.serde(),
@@ -83,35 +85,35 @@ fun saveMapTiles(
 }
 
 
-private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.flatMapSubdividedChunk() =
+//private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.flatMapSubdividedChunk() =
-  flatMap("$pid.subdivide-chunks") { initialKey, initialChunk ->
+//  flatMap("$pid.subdivide-chunks") { initialKey, initialChunk ->
-
+//
-    val (chunks, time) = measureTimedValue {
+//    val (chunks, time) = measureTimedValue {
-      ChunkSize.entries.flatMap { chunkSize ->
+//      ChunkSize.entries.flatMap { chunkSize ->
-        initialChunk.map
+//        initialChunk.map
-          .entries
+//          .entries
-          .groupingBy { (tilePosition, _) ->
+//          .groupingBy { (tilePosition, _) ->
-            val chunkPosition = tilePosition.toMapChunkPosition(chunkSize.lengthInTiles)
+//            val chunkPosition = tilePosition.toMapChunkPosition(chunkSize)
-            initialKey.copy(
+//            initialKey.copy(
-              chunkSize = chunkSize,
+//              chunkSize = chunkSize,
-              chunkPosition = chunkPosition,
+//              chunkPosition = chunkPosition,
-            )
+//            )
-          }.fold(
+//          }.fold(
-            initialValueSelector = { _, _ -> mutableMapOf<MapTilePosition, ColourHex>() }
+//            initialValueSelector = { _, _ -> mutableMapOf<MapTilePosition, ColourHex>() }
-          ) { _, accumulator, (tilePosition, colour) ->
+//          ) { _, accumulator, (tilePosition, colour) ->
-            accumulator[tilePosition] = colour
+//            accumulator[tilePosition] = colour
-            accumulator
+//            accumulator
-          }
+//          }
-          .map { (chunkId, tiles) ->
+//          .map { (chunkId, tiles) ->
-            chunkId to ServerMapChunkTiles(chunkId, tiles)
+//            chunkId to ServerMapChunkTiles(chunkId, tiles)
-          }
+//          }
-      }
+//      }
-    }
+//    }
-
+//
-    println("subdivided ${initialKey.chunkPosition} to ${chunks.size} chunks in $time")
+//    println("subdivided ${initialKey.chunkPosition} to ${chunks.size} chunks in $time")
-
+//
-    chunks
+//    chunks
-  }
+//  }
 
 
 private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.forEachChunkSaveImage(

modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt

@@ -1,6 +1,7 @@
 package dev.adamko.kafkatorio.processor.topology
 
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kafkatorio.schema.common.ChunkSize
 import dev.adamko.kafkatorio.schema.common.ColourHex
 import dev.adamko.kafkatorio.schema.common.MapChunkPosition
 import dev.adamko.kafkatorio.schema.common.MapTilePosition
@@ -142,11 +143,11 @@ class SaveMapChunksTest : FunSpec({
       println("chunk size: ${chunkId.chunkSize} ${chunkId.chunkSize.lengthInTiles}")
 
       val map = chunkId.chunkPosition
-        .iterateTiles(chunkId.chunkSize.lengthInTiles)
+        .iterateTiles(chunkId.chunkSize)
         .asSequence()
         .map { tilePos ->
           val colour =
-            tileColours.getOrPut(tilePos.toMapChunkPosition(ChunkSize.MIN.lengthInTiles)) {
+            tileColours.getOrPut(tilePos.toMapChunkPosition(ChunkSize.MIN)) {
               colourHexArb().sample(RandomSource.default()).value
             }
           tilePos to colour