massive tile processing speed improvement

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

@@ -16,7 +16,6 @@ import dev.adamko.kafkatorio.events.schema.MapTilePosition
 import dev.adamko.kafkatorio.events.schema.MapTilePrototype
 import dev.adamko.kafkatorio.events.schema.MapTiles
 import dev.adamko.kafkatorio.events.schema.converters.leftTopTile
-import dev.adamko.kafkatorio.events.schema.converters.rightBottomTile
 import dev.adamko.kafkatorio.events.schema.converters.toHexadecimal
 import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
 import dev.adamko.kotka.extensions.consumedAs
@@ -24,14 +23,16 @@ import dev.adamko.kotka.extensions.groupedAs
 import dev.adamko.kotka.extensions.materializedAs
 import dev.adamko.kotka.extensions.materializedWith
 import dev.adamko.kotka.extensions.producedAs
+import dev.adamko.kotka.extensions.streams.aggregate
 import dev.adamko.kotka.extensions.streams.filter
-import dev.adamko.kotka.extensions.streams.flatMap
+import dev.adamko.kotka.extensions.streams.groupBy
 import dev.adamko.kotka.extensions.streams.map
 import dev.adamko.kotka.extensions.streams.mapValues
 import dev.adamko.kotka.extensions.streams.merge
+import dev.adamko.kotka.extensions.streams.reduce
 import dev.adamko.kotka.extensions.streams.to
-import dev.adamko.kotka.extensions.streams.toTable
 import dev.adamko.kotka.extensions.tables.join
+import dev.adamko.kotka.extensions.tables.mapValues
 import dev.adamko.kotka.kxs.serde
 import dev.adamko.kotka.topicdata.TopicRecord
 import dev.adamko.kotka.topicdata.flatMapTopicRecords
@@ -42,6 +43,7 @@ import java.time.Duration
 import kotlin.math.abs
 import kotlin.math.roundToInt
 import kotlinx.serialization.Serializable
+import kotlinx.serialization.Transient
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
 import org.apache.kafka.common.serialization.Serdes
@@ -68,37 +70,41 @@ class KafkatorioTopology(
     const val sourceTopic = "factorio-server-log"
   }
 
-  private lateinit var tilePrototypesStore: ReadOnlyKeyValueStore<String, ValueAndTimestamp<MapTilePrototype>>
+//  private lateinit var tilePrototypesStore: ReadOnlyKeyValueStore<String, ValueAndTimestamp<MapTilePrototype>>
 
   fun build() {
 
     parseAndSplitServerLog()
     sendPlayerUpdatesToWebSocket()
 //    createMapTileImages()
-    val tilePrototypesTable: KTable<String, MapTilePrototype> = tilePrototypesTable()
+    val tilePrototypesTable: KTable<SurfaceIndex, SurfacePrototypesData> = tilePrototypesTable()
-    mapChunksStream(tilePrototypesTable)
+
+    val tilesTable: KTable<TileUpdateRecordKey, TileUpdateRecord> = buildTilesTable()
+
+    val tilesGroupedByChunk: KTable<MapChunkDataPosition, MapChunkData> =
+      buildChunkedTilesTable(tilesTable)
+
+    createChunkTileImage(tilePrototypesTable, tilesGroupedByChunk)
 
     val topology = builder.build()
 
     val streams = KafkaStreams(topology, appProps.kafkaConfig)
+
 //    streams.cleanUp()
-
     streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
-
+    Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(1)) })
-    Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(10)) })
-
 
     val description: TopologyDescription = topology.describe()
     println(description)
 
     streams.start()
 
-    tilePrototypesStore = streams.store(
+//    tilePrototypesStore = streams.store(
-      StoreQueryParameters.fromNameAndType(
+//      StoreQueryParameters.fromNameAndType(
-        tilePrototypesTable.queryableStoreName(),
+//        tilePrototypesTable.queryableStoreName(),
-        QueryableStoreTypes.timestampedKeyValueStore<String, MapTilePrototype>()
+//        QueryableStoreTypes.timestampedKeyValueStore<String, MapTilePrototype>()
-      )
+//      )
-    )
+//    )
   }
 
   private fun parseAndSplitServerLog() {
@@ -124,11 +130,11 @@ class KafkatorioTopology(
       ) { _: String, value: KafkatorioPacket, _: RecordContext ->
 //        println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
         when (value) {
-          is FactorioEvent               ->
+          is FactorioEvent ->
             "kafkatorio.${value.packetType.name}.${value.data.objectName.name}"
           is FactorioConfigurationUpdate ->
             "kafkatorio.${value.packetType.name}.FactorioConfigurationUpdate"
-          is FactorioPrototypes          ->
+          is FactorioPrototypes ->
             "kafkatorio.${value.packetType.name}.all"
         }
       }
@@ -146,30 +152,67 @@ class KafkatorioTopology(
       }
   }
 
-  private fun tilePrototypesTable(): KTable<String, MapTilePrototype> {
+  @Serializable
+  data class SurfaceIndex(
+    val surfaceIndex: Int
+  ) {
+    override fun toString() = "$surfaceIndex"
+  }
+
+  @Serializable
+  data class SurfacePrototypesData(
+    val surfaceIndex: SurfaceIndex,
+    val mapTilePrototypes: Set<MapTilePrototype>,
+  )
+
+  private fun tilePrototypesTable(): KTable<SurfaceIndex, SurfacePrototypesData> {
 
 //    val tilePrototypesStreamName = "kafkatorio.stream.prototypes-tiles"
 //    val tilePrototypesStoreName = "kafkatorio.store.prototypes-tiles"
 
     return builder.stream(
       "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
-      Consumed.with(Serdes.String(), jsonMapper.serde<FactorioPrototypes>())
+      consumedAs(
+        "consume-all-prototypes-packets",
+        Serdes.String(),
+        jsonMapper.serde<FactorioPrototypes>()
+      )
     )
-      .flatMap("flatten-map-tile-prototypes") { _, prototypes: FactorioPrototypes ->
+      .map("flatten-map-tile-prototypes") { _, prototypes: FactorioPrototypes ->
-        prototypes.prototypes
+        val tilePrototypes = prototypes.prototypes
           .filterIsInstance<MapTilePrototype>()
-          .map { proto -> proto.name to proto }
+          .toSet()
+        val key = SurfaceIndex(1)   // TODO make surfaceIndex come from Factorio, not hard coded
+        val data = SurfacePrototypesData(key, tilePrototypes)
+        key to data
       }
       .peek { _, value ->
-        println("tile prototype: ${value.name}")
+        println("tile prototypes: ${value.mapTilePrototypes.joinToString { it.name }}")
       }
-      .toTable(
+      .groupByKey(
-        materializedAs(
+        groupedAs(
-          "kafkatorio.prototypes.map-tiles",
+          "grouping-tile-prototypes-by-surface-index",
           jsonMapper.serde(),
           jsonMapper.serde()
         )
       )
+      .reduce(
+        "group-surface-prototypes",
+        materializedWith(
+//          "kafkatorio.surface-prototypes",
+          jsonMapper.serde(),
+          jsonMapper.serde()
+        )
+      ) { value1: SurfacePrototypesData, value2: SurfacePrototypesData ->
+        value1.copy(mapTilePrototypes = value1.mapTilePrototypes + value2.mapTilePrototypes)
+      }
+//      .toTable(
+//        materializedAs(
+//          "kafkatorio.prototypes.map-tiles",
+//          jsonMapper.serde(),
+//          jsonMapper.serde()
+//        )
+//      )
 //      .to(
 //        tilePrototypesStreamName,
 //      )
@@ -190,137 +233,166 @@ class KafkatorioTopology(
   }
 
   @Serializable
-  data class MapTileWithPrototype(
+  data class TileUpdateRecordKey(
-    val position: MapTilePosition,
+    val surfaceIndex: SurfaceIndex,
-    val prototype: MapTilePrototype,
+    val tilePosition: MapTilePosition,
   )
 
+  @Serializable
+  data class TileUpdateRecord(
+    val surfaceIndex: SurfaceIndex,
+    val tilePosition: MapTilePosition,
+    val tile: MapTile,
+  ) : TopicRecord<TileUpdateRecordKey> {
+    @Transient
+    override val topicKey: TileUpdateRecordKey = TileUpdateRecordKey(surfaceIndex, tilePosition)
+  }
+
+  private fun buildTilesTable(): KTable<TileUpdateRecordKey, TileUpdateRecord> {
+
+    val luaTilesUpdatesStream: KStream<String, MapTiles> = builder.stream(
+      "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
+      consumedAs("consume-map-tiles-packets", Serdes.String(), jsonMapper.serde<FactorioEvent>())
+    )
+      .filter("events.filter.map-tiles") { _: String, event: FactorioEvent -> event.data is MapTiles }
+      .mapValues("events.extract-map-tiles") { _, event: FactorioEvent -> (event.data as? MapTiles)!! }
+
+    val chunkTilesUpdateStream: KStream<String, MapTiles> = builder.stream(
+      "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}",
+      consumedAs("consume-map-chunk-packets", Serdes.String(), jsonMapper.serde<FactorioEvent>())
+    )
+      .filter("events.filter.map-chunks") { _: String, event: FactorioEvent -> event.data is MapChunk }
+      .mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles }
+
+    return luaTilesUpdatesStream
+      .merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream)
+      .flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles ->
+//          println("all map tiles update ${mapTiles.tiles.size}")
+        mapTiles.tiles.map { tile ->
+          TileUpdateRecord(
+            SurfaceIndex(mapTiles.surfaceIndex),
+            tile.position,
+            tile,
+          )
+        }
+      }
+      .toTable(materializedWith(jsonMapper.serde(), jsonMapper.serde()))
+//      .suppress(
+//        Suppressed.untilTimeLimit(
+//          Duration.ofSeconds(30),
+//          Suppressed.BufferConfig
+//            .maxRecords(100)
+//            .emitEarlyWhenFull()
+//        )
+//      )
+  }
+
   @Serializable
   data class MapChunkDataPosition(
     val position: MapChunkPosition,
-    val surfaceIndex: Int,
+    val surfaceIndex: SurfaceIndex,
   )
 
   @Serializable
   data class MapChunkData(
     val chunkPosition: MapChunkDataPosition,
     val tiles: Set<MapTile>,
-    val tilePrototypes: Set<MapTilePrototype>,
   )
 
-  private fun mapChunksStream(tilePrototypesTable: KTable<String, MapTilePrototype>) {
+  private fun buildChunkedTilesTable(
-
+    tilesTable: KTable<TileUpdateRecordKey, TileUpdateRecord>
-    val luaTilesUpdatesStream: KStream<String, MapTiles> = builder.stream(
+  ): KTable<MapChunkDataPosition, MapChunkData> {
-      "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
-      Consumed.with(Serdes.String(), jsonMapper.serde<FactorioEvent>())
-    )
-//      .peek { k, v ->
-//        println("trying to handle MapChunk update... $k: ${v.eventType}")
-//      }
-      .filter("events.filter.map-tiles") { _: String, event: FactorioEvent -> event.data is MapTiles }
-      .mapValues("events.extract-map-tiles") { _, event: FactorioEvent -> (event.data as? MapTiles)!! }
-//      .peek { key, value ->
-//        println("MapChunk update $key, tiles count: ${value.tiles.size}")
-//      }
-
-
-    val chunkTilesUpdateStream: KStream<String, MapTiles> = builder.stream(
-      "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}",
-      Consumed.with(Serdes.String(), jsonMapper.serde<FactorioEvent>())
-    )
-      .peek { k, v ->
-        println("trying to handle LuaTiles update... $k: ${v.eventType} / ${v.data is MapChunk}")
-      }
-      .filter("events.filter.map-chunks") { _: String, event: FactorioEvent -> event.data is MapChunk }
-      .mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles }
-      .peek { key, value ->
-        println("LuaTiles update $key, tiles count: ${value.tiles.size}")
-      }
 
     @Serializable
-    data class TileUpdateRecordKey(
+    data class MapChunkDataAggregator(
-      val surfaceIndex: Int,
+      val tiles: MutableSet<MapTile> = mutableSetOf(),
-      val tilePosition: MapTilePosition,
     )
 
-    @Serializable
+    return tilesTable
-    data class TileUpdateRecord(
-      val surfaceIndex: Int,
-      val tilePosition: MapTilePosition,
-      val tile: MapTile,
-      val tilePrototype: MapTilePrototype?,
-    ) : TopicRecord<TileUpdateRecordKey> {
-      override val topicKey: TileUpdateRecordKey = TileUpdateRecordKey(surfaceIndex, tilePosition)
-    }
-
-    val allTileUpdatesStream: KStream<TileUpdateRecordKey, TileUpdateRecord> =
-      luaTilesUpdatesStream
-        .merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream)
-        .flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles ->
-//          println("all map tiles update ${mapTiles.tiles.size}")
-          mapTiles.tiles.map { tile ->
-            TileUpdateRecord(
-              mapTiles.surfaceIndex,
-              tile.position,
-              tile,
-              null
-            )
-          }
-        }
-        .toTable(
-          "all-tile-updates-table",
-          materializedWith(jsonMapper.serde(), jsonMapper.serde())
-        )
-        .join(
-          tilePrototypesTable,
-          "enrich-tiles-with-prototypes",
-          materializedWith(jsonMapper.serde(), jsonMapper.serde()),
-          { it.tile.prototypeName },
-        ) { v1: TileUpdateRecord, v2: MapTilePrototype ->
-          v1.copy(tilePrototype = v2)
-        }
-        .toStream()
-        .filter { _, value ->
-          value.tilePrototype != null
-        }
-
-    val chunksTable: KTable<MapChunkDataPosition, MapChunkData> =
-      allTileUpdatesStream
-        .map("map-tile-updates-to-MapChunkData") { _: TileUpdateRecordKey, value: TileUpdateRecord ->
-          val key = MapChunkDataPosition(
-            value.tilePosition.toMapChunkPosition(),
-            value.surfaceIndex,
-          )
-          key to MapChunkData(
-            key,
-            setOf(value.tile),
-            setOfNotNull(value.tilePrototype)
-          )
-        }
-        // group all tiles by chunk position
-        .groupByKey(
-          groupedAs("group-tile-updates-per-chunk-position", jsonMapper.serde(), jsonMapper.serde())
-        )
-        .reduce { tileA, tileB ->
-          tileA.copy(
-            tiles = tileA.tiles + tileB.tiles,
-            tilePrototypes = tileA.tilePrototypes + tileB.tilePrototypes
-          )
-        }
-        .suppress(
-          Suppressed.untilTimeLimit(
-            Duration.ofSeconds(30),
-            Suppressed.BufferConfig.maxRecords(32*32*2)
-          )
-        )
-
-    chunksTable
       .toStream()
-      .foreach { key: MapChunkDataPosition, value: MapChunkData ->
+//      .repartition(
+//        repartitionedAs(
+//          "repartition-all-tiles-before-chunk-grouping",
+//          jsonMapper.serde(),
+//          jsonMapper.serde()
+//        )
+//      )
+      .groupBy<TileUpdateRecordKey, TileUpdateRecord, MapChunkDataPosition>(
+        groupedAs("grouping-tiles-by-chunk-pos", jsonMapper.serde(), jsonMapper.serde()),
+      ) { key: TileUpdateRecordKey, value: TileUpdateRecord ->
+        MapChunkDataPosition(
+          value.tilePosition.toMapChunkPosition(),
+          key.surfaceIndex
+        )
+      }
+      .aggregate<MapChunkDataPosition, TileUpdateRecord, MapChunkDataAggregator>(
+        "reducing-tiles-grouped-by-chunk-position",
+        materializedWith(jsonMapper.serde(), jsonMapper.serde()),
+        { MapChunkDataAggregator() },
+      ) { _: MapChunkDataPosition, value: TileUpdateRecord, aggregate: MapChunkDataAggregator ->
+        aggregate.tiles.add(value.tile)
+        aggregate
+      }
+      // TODO maybe set serdes here?
+      .mapValues(
+        "tiles-grouped-by-chunk-position-finalise-aggregation",
+        materializedWith(jsonMapper.serde(), jsonMapper.serde()),
+      ) { k, v ->
+        MapChunkData(k, v.tiles)
+      }
+
+  }
+
+  private fun createChunkTileImage(
+    tilePrototypesTable: KTable<SurfaceIndex, SurfacePrototypesData>,
+    tilesGroupedByChunk: KTable<MapChunkDataPosition, MapChunkData>,
+  ) {
+
+    @Serializable
+    data class ChunkTilesAndProtos(
+      val chunkTiles: MapChunkData,
+      val protos: SurfacePrototypesData
+    )
+
+//    tilesGroupedByChunk
+//      .join<ChunkTilesAndProtos, SurfaceIndex, SurfacePrototypesData>(
+//        tilePrototypesTable,
+//        { cd: MapChunkData -> cd.chunkPosition.surfaceIndex },
+//        { chunkTiles: MapChunkData, protos: SurfacePrototypesData ->
+//          ChunkTilesAndProtos(chunkTiles, protos)
+//        },
+//        namedAs("joining-chunks-tiles-prototypes"),
+//        materializedAs(
+//          "joined-chunked-tiles-with-prototypes",
+//          jsonMapper.serde(),
+//          jsonMapper.serde()
+//        ),
+//      )
+    tilesGroupedByChunk
+      .join<MapChunkDataPosition, MapChunkData, SurfaceIndex, SurfacePrototypesData, ChunkTilesAndProtos>(
+        tilePrototypesTable,
+        "enrich-tiles-with-prototype-colour",
+        materializedAs(
+          "joined-chunked-tiles-with-prototypes",
+          jsonMapper.serde(),
+          jsonMapper.serde()
+        ),
+        { it.chunkPosition.surfaceIndex },
+      ) { chunkTiles: MapChunkData, protos: SurfacePrototypesData ->
+        ChunkTilesAndProtos(chunkTiles, protos)
+      }
+      .suppress(
+        Suppressed.untilTimeLimit(
+          Duration.ofSeconds(30),
+          Suppressed.BufferConfig.maxRecords(30)
+        )
+      )
+      .toStream()
+      .foreach { _, (chunkTiles: MapChunkData, protos: SurfacePrototypesData) ->
         runCatching {
-          saveMapTilesPng(key, value)
+          saveMapTilesPng(chunkTiles, protos)
         }.onFailure { e ->
-          println("error saving map tile png chunk:${value.chunkPosition}")
+          println("error saving map tile png chunk:${chunkTiles.chunkPosition}")
           e.printStackTrace()
           throw e
         }
@@ -328,14 +400,16 @@ class KafkatorioTopology(
   }
 
 
-  private fun saveMapTilesPng(key: MapChunkDataPosition, chunk: MapChunkData) {
+  private fun saveMapTilesPng(chunkTiles: MapChunkData, protos: SurfacePrototypesData) {
+
+    val chunkPos = chunkTiles.chunkPosition
 
     val tilePrototypes: Map<String, MapTilePrototype?> =
-      chunk.tilePrototypes.associateBy { it.name }.withDefault { null }
+      protos.mapTilePrototypes.associateBy { it.name }.withDefault { null }
 
-    val chunkOriginX = chunk.chunkPosition.position.leftTopTile.x
+    val chunkOriginX = chunkTiles.chunkPosition.position.leftTopTile.x
 //    val chunkOriginY = chunk.chunkPosition.position.rightBottomTile.y
-    val chunkOriginY = chunk.chunkPosition.position.leftTopTile.y
+    val chunkOriginY = chunkTiles.chunkPosition.position.leftTopTile.y
 
     val chunkImage =
       ImmutableImage.filled(
@@ -345,7 +419,7 @@ class KafkatorioTopology(
         BufferedImage.TYPE_INT_ARGB
       )
 
-    chunk.tiles.forEach { tile ->
+    chunkTiles.tiles.forEach { tile ->
 
       val prototypeColour = tilePrototypes
         .getValue(tile.prototypeName)
@@ -378,7 +452,9 @@ class KafkatorioTopology(
 
     val zoom = 1u
     val file =
-      File("src/main/resources/kafkatorio-web-map/s${key.surfaceIndex}/z$zoom/x${key.position.x}/y${key.position.y}.png")
+      File(
+        "src/main/resources/kafkatorio-web-map/s${chunkPos.surfaceIndex}/z$zoom/x${chunkOriginX}/y${chunkOriginY}.png"
+      )
 
     if (file.parentFile.mkdirs()) {
       println("created new map tile parentFile directory ${file.absolutePath}")
@@ -389,100 +465,4 @@ class KafkatorioTopology(
     chunkImage.output(PngWriter.MaxCompression, file)
   }
 
-//
-//  private fun createMapTileImages() {
-//
-//    val mapTilesByPrototypeName: KStream<String, MapTile> =
-//      builder.stream(
-//        "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
-//        Consumed.with(Serdes.String(), KafkatorioPacketSerde)
-//      )
-//        .filter { _, packet -> packet is FactorioEvent && packet.data is MapTiles }
-//        .mapValues { packet -> ((packet as? FactorioEvent)?.data as? MapTiles)!! }
-//        .flatMap { _, mapTiles ->
-//          mapTiles
-//            .tiles
-//            .map { tile -> tile.prototypeName to tile }
-//            .map { it.toKeyValue() }
-//        }
-////        .groupByKey()
-//
-//    @Serializable
-//    data class TileSurfaceCoord(
-//      val surfaceIndex: Int,
-//      val position: MapTilePosition,
-//    )
-//
-//    @Serializable
-//    data class TileSurfaceState(
-//      val surfaceIndex: Int,
-//      val position: MapTilePosition,
-//      val prototypeName: String,
-//    )
-//
-//    val mapTilesTable: KTable<TileSurfaceCoord, TileSurfaceState> =
-//      builder.stream(
-//        "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
-//        Consumed.with(Serdes.String(), KafkatorioPacketSerde)
-//      )
-//        .filter { _, packet -> packet is FactorioEvent && packet.data is MapTiles }
-//        .mapValues { packet -> ((packet as? FactorioEvent)?.data as? MapTiles)!! }
-//        .flatMap({ _, mapTiles: MapTiles ->
-//          mapTiles.tiles.map { tile: MapTile ->
-//            val key = TileSurfaceCoord(mapTiles.surfaceIndex, tile.position)
-//            val value = TileSurfaceState(mapTiles.surfaceIndex, tile.position, tile.prototypeName)
-//            (key to value).toKeyValue()
-//          }
-//        }
-//        )
-//        .toTable(
-//          materializedAs(
-//            "kafkatorio.table.map-tile-status",
-//            jsonMapper.serde(),
-//            jsonMapper.serde(),
-//          )
-//        )
-//
-//
-//    val tilePrototypesTable: KTable<String, MapTilePrototype> =
-//      builder.stream(
-//        "kafkatorio.${KafkatorioPacket.PacketType.CONFIG}.FactorioConfigurationUpdate",
-//        Consumed.with(Serdes.String(), KafkatorioPacketSerde)
-//      )
-//        .flatMap(KeyValueMapper<String, KafkatorioPacket, List<KeyValue<String, MapTilePrototype>>> { _, packet ->
-//          when (packet) {
-//            is FactorioConfigurationUpdate ->
-//              packet.prototypes
-//                .filterIsInstance<MapTilePrototype>()
-//                .map { proto -> proto.name to proto }
-//                .map { it.toKeyValue() }
-//            is FactorioEvent               ->
-//              emptyList()
-//          }
-//        })
-//        .toTable(
-//          materializedAs("kafkatorio.table.tile-prototypes", Serdes.String(), jsonMapper.serde())
-//        )
-//
-//
-//    @Serializable
-//    data class TileSurfaceData(
-//      val surfaceIndex: Int,
-//      val position: MapTilePosition,
-//      val prototype: MapTilePrototype,
-//    )
-//
-//    val enrichedTiles: KTable<TileSurfaceCoord, TileSurfaceData> =
-//      mapTilesTable.join<TileSurfaceData, String, MapTilePrototype>(
-//        tilePrototypesTable,
-//        { tile: TileSurfaceState ->
-//          tile.prototypeName
-//        },
-//        { tile: TileSurfaceState, proto: MapTilePrototype ->
-//          TileSurfaceData(tile.surfaceIndex, tile.position, proto)
-//        },
-//      )
-//
-//  }
-
 }

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

@@ -3,6 +3,7 @@ package dev.adamko.kafkatorio.processor
 
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
 import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.http4k.server.PolyHandler
 import org.http4k.server.Undertow
@@ -10,7 +11,7 @@ import org.http4k.server.asServer
 
 val appProps = ApplicationProperties()
 
-fun main() = runBlocking {
+fun main() = runBlocking<Unit> {
 
   val wsServer = WebsocketServer()
   val tileServer = WebMapTileServer()
@@ -18,11 +19,16 @@ fun main() = runBlocking {
   val topology = KafkatorioTopology(wsServer)
   topology.build()
 
-  PolyHandler(
+  val webServer = PolyHandler(
     tileServer.build(),
     wsServer.build(),
   ).asServer(Undertow(9073))
-    .start()
+
-    .block()
+//  Runtime.getRuntime().addShutdownHook(Thread {
+//    webServer.stop()
+//  })
+//  launch {
+    webServer.start().block()
+//  }
 
 }

modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt

@@ -34,11 +34,16 @@ fun createFactorioMap() = Maps {
     ) {
       attribution = "kafkatorio"
       tms = true
-      minZoom = 1
+      minZoom = -2
-//      maxZoom = 1
+      maxZoom = 1
       maxNativeZoom = 1
+      minNativeZoom = 1
     }.addTo(this)
 
+    L.scale {
+
+    }
+
     with(options) {
       crs = CRS.Simple
     }