- MapTile/LuaTile no longer a FactorioObjectData (it doesn't get sent as a packet)

- trying to optimise the topology so it's faster... almost there but the records are now too large
- map tile pngs, multiple zoom levels
- use coroutines for faster map tiles

modules/events-mod/src/main/typescript/generated/kafkatorio-schema/object-data.d.ts

@@ -1,6 +1,6 @@
 // Generated by TypeScriptGenerator - do not edit this file manually
 
-type ObjectName = "LuaPlayer" | "LuaEntity" | "LuaSurface" | "MapChunk" | "LuaTile" | "LuaTiles" | "ConsoleChatMessage";
+type ObjectName = "LuaPlayer" | "LuaEntity" | "LuaSurface" | "MapChunk" | "LuaTiles" | "ConsoleChatMessage";
 
 interface FactorioObjectData {
   objectName: ObjectName;
@@ -37,12 +37,6 @@ interface SurfaceData extends FactorioObjectData {
   objectName: ObjectName;
 }
 
-interface MapTile extends FactorioObjectData {
-  objectName: ObjectName;
-  position: MapTilePosition;
-  prototypeName: string;
-}
-
 interface MapTiles extends FactorioObjectData {
   objectName: ObjectName;
   surfaceIndex: int;

modules/events-mod/src/main/typescript/generated/kafkatorio-schema/schema.d.ts

@@ -41,3 +41,9 @@ interface MapChunkPosition {
   x: int;
   y: int;
 }
+
+interface MapTile {
+  objectName: string;
+  position: MapTilePosition;
+  prototypeName: string;
+}

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

@@ -38,7 +38,7 @@ sealed class FactorioObjectData {
     LuaEntity,
     LuaSurface,
     MapChunk,
-    LuaTile,
+    //    LuaTile,
     LuaTiles,
     ConsoleChatMessage,
   }
@@ -66,7 +66,7 @@ object FactorioObjectDataJsonSerializer : JsonContentPolymorphicSerializer<Facto
       FactorioObjectData.ObjectName.LuaEntity          -> EntityData.serializer()
       FactorioObjectData.ObjectName.LuaSurface         -> SurfaceData.serializer()
       FactorioObjectData.ObjectName.MapChunk           -> MapChunk.serializer()
-      FactorioObjectData.ObjectName.LuaTile            -> MapTile.serializer()
+//      FactorioObjectData.ObjectName.LuaTile            -> MapTile.serializer()
       FactorioObjectData.ObjectName.LuaTiles           -> MapTiles.serializer()
       FactorioObjectData.ObjectName.ConsoleChatMessage -> ConsoleChatMessage.serializer()
     }
@@ -141,9 +141,11 @@ data class MapTiles(
 data class MapTile(
   val prototypeName: String,
   val position: MapTilePosition,
-) : FactorioObjectData() {
-  @EncodeDefault
-  override val objectName = ObjectName.LuaTile
+)
+//  : FactorioObjectData()
+{
+  //  @EncodeDefault
+  val objectName: String = ""
 }
 
 @Serializable

modules/events-server/build.gradle.kts

@@ -17,7 +17,7 @@ val projectId: String by project.extra
 @Suppress("UnstableApiUsage") // platform + version-catalog is incubating
 dependencies {
 
-  implementation("org.apache.kafka:kafka-streams:3.0.0")
+  implementation("org.apache.kafka:kafka-streams:3.1.0")
 
   implementation(projects.modules.eventsSchema)
 
@@ -38,8 +38,8 @@ dependencies {
 
   implementation("com.sksamuel.scrimage:scrimage-core:4.0.24")
 
-//  val kotkaVersion = "0.0.11"
-  val kotkaVersion = "store-iterators-SNAPSHOT"
+  val kotkaVersion = "v0.0.12"
+//  val kotkaVersion = "store-iterators-SNAPSHOT"
   implementation("com.github.adamko-dev.kotka-streams:kotka-streams:$kotkaVersion") {
     isChanging = true
   }

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

@@ -1,22 +1,14 @@
 package dev.adamko.kafkatorio.processor
 
-import dev.adamko.kafkatorio.events.schema.Colour
-import dev.adamko.kafkatorio.events.schema.MapTile
-import dev.adamko.kafkatorio.processor.topology.PrototypeName
-import dev.adamko.kafkatorio.processor.topology.TileUpdateRecordKey
-import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPixels
-import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPosition
-import dev.adamko.kafkatorio.processor.topology.aggregateWebMapTiles
-import dev.adamko.kafkatorio.processor.topology.allMapTilesTable
+import dev.adamko.kafkatorio.processor.topology.buildFactorioServerMap
+import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream
 import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer
-import dev.adamko.kafkatorio.processor.topology.saveTileImages
-import dev.adamko.kafkatorio.processor.topology.splitFactorioServerLog
-import dev.adamko.kafkatorio.processor.topology.tilePrototypeColourTable
+import dev.adamko.kafkatorio.processor.topology.saveMapTiles
+import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
 import java.time.Duration
 import org.apache.kafka.streams.KafkaStreams
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.TopologyDescription
-import org.apache.kafka.streams.kstream.KTable
 
 
 class KafkatorioTopology(
@@ -30,17 +22,25 @@ class KafkatorioTopology(
 
   fun build() {
 
-    splitFactorioServerLog(builder)
+    val packets = factorioServerPacketStream(builder)
+
+    splitFactorioServerPacketStream(packets)
 
     playerUpdatesToWsServer(websocketServer, builder)
 
-    val allMapTilesTable: KTable<TileUpdateRecordKey, MapTile> = allMapTilesTable(builder)
-    val tilePrototypeColourTable: KTable<PrototypeName, Colour> = tilePrototypeColourTable(builder)
+    val serverMapDataTable = buildFactorioServerMap(packets)
 
-    val webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels> =
-      aggregateWebMapTiles(allMapTilesTable, tilePrototypeColourTable)
+    saveMapTiles(serverMapDataTable)
 
-    saveTileImages(webMapTiles)
+
+    // the old way...
+//    val allMapTilesTable: KTable<TileUpdateRecordKey, MapTile> = allMapTilesTable(builder)
+//    val tilePrototypeColourTable: KTable<PrototypeName, Colour> = tilePrototypeColourTable(builder)
+//
+//    val webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels> =
+//      aggregateWebMapTiles(allMapTilesTable, tilePrototypeColourTable)
+//
+//    saveTileImages(webMapTiles)
 
     val topology = builder.build()
 

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

@@ -0,0 +1,127 @@
+package dev.adamko.kafkatorio.processor.topology
+
+import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate
+import dev.adamko.kafkatorio.events.schema.FactorioEvent
+import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
+import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
+import dev.adamko.kafkatorio.events.schema.MapChunk
+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.processor.serdes.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kotka.extensions.groupedAs
+import dev.adamko.kotka.extensions.materializedAs
+import dev.adamko.kotka.extensions.materializedWith
+import dev.adamko.kotka.extensions.streams.aggregate
+import dev.adamko.kotka.extensions.tables.mapValues
+import dev.adamko.kotka.kxs.serde
+import kotlinx.serialization.Serializable
+import org.apache.kafka.streams.kstream.KStream
+import org.apache.kafka.streams.kstream.KTable
+
+
+@Serializable
+data class FactorioServerMap(
+  val serverId: FactorioServerId,
+  val tilePrototypes: Map<PrototypeName, MapTilePrototype>,
+  val surfaces: Map<SurfaceIndex, FactorioMapSurface>,
+)
+
+
+@Serializable
+data class FactorioMapSurface(
+  val index: SurfaceIndex,
+  val tiles: Map<MapTilePosition, PrototypeName>,
+)
+
+/** Aggregate [KafkatorioPacket]s into a single [FactorioServerMap] (per [serverId][FactorioServerId]). */
+fun buildFactorioServerMap(
+  factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket>
+): KTable<FactorioServerId, FactorioServerMap> {
+
+  return factorioServerPacketStream
+    .groupByKey(
+      groupedAs("factorio-server-map-data-grouping", jsonMapper.serde(), jsonMapper.serde())
+    )
+    .aggregate<FactorioServerId, KafkatorioPacket, FactorioServerMapAggregator>(
+      "factorio-server-map-data-aggregate",
+      materializedAs("factorio-server-map-aggregate-store", kxsBinary.serde(), kxsBinary.serde()),
+      { FactorioServerMapAggregator() },
+    ) { _: FactorioServerId, packet: KafkatorioPacket, aggregate: FactorioServerMapAggregator ->
+      aggregate + packet
+    }
+    .mapValues(
+      "finalise-factorio-server-map-aggregate",
+      materializedWith(kxsBinary.serde(), kxsBinary.serde())
+    ) { readOnlyKey, aggregate ->
+
+      val surfaces = aggregate.surfaces.mapValues { (_, surface) ->
+        FactorioMapSurface(surface.index, surface.tiles)
+      }
+
+      FactorioServerMap(readOnlyKey, aggregate.tilePrototypes, surfaces)
+    }
+
+}
+
+
+@Serializable
+private data class FactorioServerMapAggregator(
+  val tilePrototypes: MutableMap<PrototypeName, MapTilePrototype> = mutableMapOf(),
+  val surfaces: MutableMap<SurfaceIndex, FactorioMapSurfaceAggregator> = mutableMapOf(),
+) {
+
+  operator fun plus(packet: KafkatorioPacket): FactorioServerMapAggregator {
+    return when (packet) {
+      is FactorioEvent               -> {
+        when (val data = packet.data) {
+          is MapChunk -> plus(data)
+          is MapTiles -> plus(data)
+//        is MapTile  -> {} // hmmm... no surface index
+          else        -> this
+        }
+      }
+      is FactorioPrototypes          -> plus(packet)
+      is FactorioConfigurationUpdate -> this
+    }
+  }
+
+  operator fun plus(mapTiles: MapTiles): FactorioServerMapAggregator {
+//    println("FactorioServerMapAggregator: adding ${mapTiles.tiles.size} MapTiles from surface ${mapTiles.surfaceIndex}")
+
+    val surfaceIndex = SurfaceIndex(mapTiles.surfaceIndex)
+
+    val surface = surfaces.getOrPut(surfaceIndex) { FactorioMapSurfaceAggregator(surfaceIndex) }
+
+    val surfaceTiles = mapTiles.tiles.associate { it.position to PrototypeName(it.prototypeName) }
+    surface.tiles.putAll(surfaceTiles)
+
+    return this
+  }
+
+  operator fun plus(mapChunk: MapChunk): FactorioServerMapAggregator {
+//    println("FactorioServerMapAggregator: adding mapChunk ${mapChunk.position}")
+    return plus(mapChunk.tiles)
+  }
+
+  operator fun plus(packet: FactorioPrototypes): FactorioServerMapAggregator {
+//    println("FactorioServerMapAggregator: adding FactorioPrototypes")
+
+    val packetPrototypes = packet.prototypes
+      .filterIsInstance<MapTilePrototype>()
+      .associateBy { PrototypeName(it.name) }
+
+    tilePrototypes.putAll(packetPrototypes)
+
+    return this
+  }
+
+}
+
+
+@Serializable
+private data class FactorioMapSurfaceAggregator(
+  val index: SurfaceIndex,
+  val tiles: MutableMap<MapTilePosition, PrototypeName> = mutableMapOf(),
+)

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

@@ -35,24 +35,24 @@ fun allMapTilesTable(
   builder: StreamsBuilder
 ): KTable<TileUpdateRecordKey, MapTile> {
 
-  val luaTilesUpdatesStream: KStream<FactorioPacketKey, MapTiles> =
-    builder.stream<FactorioPacketKey, FactorioEvent>(
+  val luaTilesUpdatesStream: KStream<FactorioServerId, MapTiles> =
+    builder.stream<FactorioServerId, FactorioEvent>(
       "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
       consumedAs("consume-map-tiles-packets", jsonMapper.serde(), jsonMapper.serde())
     )
-      .filter("events.filter.map-tiles") { _: FactorioPacketKey, event: FactorioEvent ->
+      .filter("events.filter.map-tiles") { _: FactorioServerId, event: FactorioEvent ->
         event.data is MapTiles
       }
       .mapValues("events.extract-map-tiles") { _, event: FactorioEvent ->
         (event.data as? MapTiles)!!
       }
 
-  val chunkTilesUpdateStream: KStream<FactorioPacketKey, MapTiles> =
-    builder.stream<FactorioPacketKey, FactorioEvent>(
+  val chunkTilesUpdateStream: KStream<FactorioServerId, MapTiles> =
+    builder.stream<FactorioServerId, FactorioEvent>(
       "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}",
       consumedAs("consume-map-chunk-packets", jsonMapper.serde(), jsonMapper.serde())
     )
-      .filter("events.filter.map-chunks") { _: FactorioPacketKey, event: FactorioEvent ->
+      .filter("events.filter.map-chunks") { _: FactorioServerId, event: FactorioEvent ->
         event.data is MapChunk
       }
       .mapValues("events.extract-map-chunks") { _, event: FactorioEvent ->

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

@@ -1,17 +1,9 @@
 package dev.adamko.kafkatorio.processor.topology
 
-import dev.adamko.kafkatorio.events.schema.ConsoleChatMessage
-import dev.adamko.kafkatorio.events.schema.EntityData
 import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate
 import dev.adamko.kafkatorio.events.schema.FactorioEvent
 import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
-import dev.adamko.kafkatorio.events.schema.MapChunk
-import dev.adamko.kafkatorio.events.schema.MapTile
-import dev.adamko.kafkatorio.events.schema.MapTiles
-import dev.adamko.kafkatorio.events.schema.PlayerData
-import dev.adamko.kafkatorio.events.schema.SurfaceData
-import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
 import dev.adamko.kafkatorio.processor.KafkatorioTopology
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kotka.extensions.consumedAs
@@ -19,25 +11,18 @@ import dev.adamko.kotka.extensions.producedAs
 import dev.adamko.kotka.extensions.streams.map
 import dev.adamko.kotka.extensions.streams.to
 import dev.adamko.kotka.kxs.serde
-import kotlinx.serialization.Serializable
 import kotlinx.serialization.decodeFromString
 import org.apache.kafka.common.serialization.Serdes
 import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.kstream.KStream
 import org.apache.kafka.streams.processor.RecordContext
 
 
-@Serializable
-data class FactorioPacketKey(
-  val serverId: FactorioServerId,
-  /** arbitrary distinctive key */
-  val key: String,
-)
-
-fun splitFactorioServerLog(
+fun factorioServerPacketStream(
   builder: StreamsBuilder = StreamsBuilder(),
-) {
+): KStream<FactorioServerId, KafkatorioPacket> {
 
-  builder.stream(
+  return builder.stream(
     KafkatorioTopology.sourceTopic,
     consumedAs(
       "read-raw-packets-from-server",
@@ -46,18 +31,22 @@ fun splitFactorioServerLog(
     )
   )
     .map("decode-packets") { serverId: String, value: String ->
-//        println("Mapping $readOnlyKey:$value")
       val packet = jsonMapper.decodeFromString<KafkatorioPacket>(value)
-      val key: String = createPacketKey(packet)
-      FactorioPacketKey(FactorioServerId(serverId), key) to packet
+      FactorioServerId(serverId) to packet
     }
+}
+
+fun splitFactorioServerPacketStream(
+  factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket>
+) {
+  factorioServerPacketStream
     .to(
       producedAs(
         "split-server-log",
         jsonMapper.serde(),
         jsonMapper.serde(),
       )
-    ) { _: FactorioPacketKey, value: KafkatorioPacket, _: RecordContext ->
+    ) { _: FactorioServerId, value: KafkatorioPacket, _: RecordContext ->
 //        println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
       when (value) {
         is FactorioEvent               ->
@@ -68,28 +57,34 @@ fun splitFactorioServerLog(
           "kafkatorio.${value.packetType.name}.all"
       }
     }
-
 }
 
-private fun createPacketKey(packet: KafkatorioPacket): String {
-
-  return when (packet) {
-    is FactorioEvent      -> factorioEventKey(packet)
-    is FactorioConfigurationUpdate,
-    is FactorioPrototypes -> packet.hashCode().toString()
-  }
-}
-
-private fun factorioEventKey(event: FactorioEvent): String {
-  val id: String = when (val data = event.data) {
-    is ConsoleChatMessage -> data.authorPlayerIndex?.toString()
-    is EntityData         -> data.type
-    is MapChunk           -> data.position.toString()
-    is MapTile            -> data.position.toString()
-    is MapTiles           -> data.tiles.firstOrNull()?.position?.toMapChunkPosition()?.toString()
-    is PlayerData         -> (data.characterUnitNumber?.toString() ?: data.name)
-    is SurfaceData        -> data.index.toString()
-  } ?: event.tick.toString()
-
-  return id
-}
+//@Serializable
+//data class FactorioPacketKey(
+//  val serverId: FactorioServerId,
+//  /** arbitrary distinctive key */
+//  val key: String,
+//)
+//
+//private fun createPacketKey(packet: KafkatorioPacket): String {
+//
+//  return when (packet) {
+//    is FactorioEvent      -> factorioEventKey(packet)
+//    is FactorioConfigurationUpdate,
+//    is FactorioPrototypes -> packet.hashCode().toString()
+//  }
+//}
+//
+//private fun factorioEventKey(event: FactorioEvent): String {
+//  val id: String = when (val data = event.data) {
+//    is ConsoleChatMessage -> data.authorPlayerIndex?.toString()
+//    is EntityData         -> data.type
+//    is MapChunk           -> data.position.toString()
+////    is MapTile            -> data.position.toString()
+//    is MapTiles           -> data.tiles.firstOrNull()?.position?.toMapChunkPosition()?.toString()
+//    is PlayerData         -> (data.characterUnitNumber?.toString() ?: data.name)
+//    is SurfaceData        -> data.index.toString()
+//  } ?: event.tick.toString()
+//
+//  return id
+//}

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

@@ -0,0 +1,198 @@
+package dev.adamko.kafkatorio.processor.topology
+
+import com.sksamuel.scrimage.ImmutableImage
+import com.sksamuel.scrimage.ScaleMethod
+import com.sksamuel.scrimage.nio.PngWriter
+import dev.adamko.kafkatorio.events.schema.Colour
+import dev.adamko.kafkatorio.events.schema.MapChunkPosition
+import dev.adamko.kafkatorio.events.schema.MapTilePosition
+import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
+import java.awt.image.BufferedImage
+import java.io.File
+import kotlin.math.abs
+import kotlin.time.Duration.Companion.seconds
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.conflate
+import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.flatMapMerge
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.mapLatest
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import org.apache.kafka.streams.kstream.KTable
+
+
+private const val WEB_MAP_TILE_SIZE_PX = 256
+
+private val cScope = CoroutineScope(
+  Dispatchers.Default + SupervisorJob() + CoroutineName("f-server-map-updates")
+)
+
+/** A bridge between Kafka and Kotlin */
+private val serverMapDataFlow: MutableStateFlow<FactorioServerMap?> = MutableStateFlow(null)
+
+
+fun saveMapTiles(
+  serverMapTable: KTable<FactorioServerId, FactorioServerMap>
+) {
+
+  serverMapTable
+    .toStream()
+    .foreach { _, value ->
+      runBlocking(Dispatchers.Default) {
+        serverMapDataFlow.emit(value)
+      }
+    }
+
+  cScope.launch {
+    println("launching serverMapDataFlow merging")
+
+
+    serverMapDataFlow
+      .filterNotNull()
+
+    serverMapDataFlow
+      .filterNotNull()
+      .conflate()
+//      .transform { serverMapData ->
+      .flatMapMerge(10) { serverMapData ->
+        println("flatMapMerge serverMapData ${serverMapData.serverId}, surfaces ${serverMapData.surfaces.keys}")
+
+        val prototypes = serverMapData.tilePrototypes
+
+        serverMapData
+          .surfaces
+          .values
+//          .forEach { surfaceData ->
+          .asFlow()
+          .flatMapMerge(10) { surfaceData ->
+            println("flatMapMerge surfaceData ${surfaceData.index}")
+
+            // get all tile colours for the surface
+            val surfaceTileColours = surfaceData.tiles
+              .entries
+              .mapNotNull { (tilePosition, prototypeName) ->
+                prototypes[prototypeName]?.let { prototype ->
+                  tilePosition to prototype.mapColour
+                }
+              }
+            println("surfaceTileColours.size ${surfaceTileColours.size}")
+
+            // split a surface into chunks, per zoom-level
+            ZoomLevel.values
+//              .forEach { zoom ->
+              .asFlow()
+              .flatMapMerge(10) { zoom ->
+                println("flatMapMerge zoom $zoom")
+
+                surfaceTileColours
+                  .groupBy { (tilePosition, _) ->
+                    tilePosition.toMapChunkPosition(zoom.chunkSize)
+                  }
+                  .entries
+//                    .forEach { (chunkPosition, tileColours) ->
+                  .asFlow()
+                  .mapLatest { (chunkPosition, tileColours) ->
+                    println("mapLatest chunkPosition $chunkPosition")
+
+                    // mapLatest + delay = an alternative to debounce-per-key?
+                    delay(30.seconds)
+
+                    println("emitting ChunkData $chunkPosition")
+
+                    ChunkData(
+                      surfaceData.index,
+                      zoom,
+                      chunkPosition,
+                      tileColours.toMap()
+                    )
+                  }
+              }
+          }
+      }
+      .onEach { chunkData ->
+
+        println("starting to save chunkData ${chunkData.chunkPosition}...")
+
+        val chunkImageFile = File(chunkData.filename.value)
+        if (chunkImageFile.parentFile.mkdirs()) {
+          println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
+        }
+
+        val image =
+          createMapTileImage(chunkData.chunkPosition, chunkData.tileColours, chunkData.zoomLevel)
+
+        println("saving map tile $chunkImageFile")
+        val savedTile = image.output(PngWriter.NoCompression, chunkImageFile)
+        println("savedTile: $savedTile")
+
+      }
+      .launchIn(this)
+  }
+}
+
+
+@JvmInline
+private value class TilePngFilename(
+  val value: String,
+) {
+
+  constructor(
+    surfaceIndex: SurfaceIndex,
+    zoom: Int,
+    chunkX: Int,
+    chunkY: Int,
+  ) : this("src/main/resources/kafkatorio-web-map/s${surfaceIndex.index}/z$zoom/x${chunkX}/y${chunkY}.png")
+
+}
+
+
+private data class ChunkData(
+  val surfaceIndex: SurfaceIndex,
+  val zoomLevel: ZoomLevel,
+  val chunkPosition: MapChunkPosition,
+  val tileColours: Map<MapTilePosition, Colour>,
+) {
+  val filename = TilePngFilename(surfaceIndex, zoomLevel.level, chunkPosition.x, chunkPosition.y)
+}
+
+private fun createMapTileImage(
+  chunkPosition: MapChunkPosition,
+  chunkColours: Map<MapTilePosition, Colour>,
+  z: ZoomLevel
+): ImmutableImage {
+
+
+  val chunkImage = ImmutableImage.filled(
+    z.chunkSize,
+    z.chunkSize,
+    COLOUR_TRANSPARENT.awt(),
+    BufferedImage.TYPE_INT_ARGB
+  )
+
+  val chunkOriginX: Int = chunkPosition.x * z.chunkSize
+  val chunkOriginY: Int = chunkPosition.y * z.chunkSize
+
+  chunkColours.forEach { (tilePosition, colour) ->
+
+    val rgbColour = colour.toRgbColor()
+
+    val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX))
+    val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY))
+
+    chunkImage.setColor(
+      pixelX,
+      pixelY,
+      rgbColour
+    )
+  }
+
+  return chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.Progressive)
+}

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

@@ -2,9 +2,11 @@ package dev.adamko.kafkatorio.processor.topology
 
 import com.sksamuel.scrimage.color.RGBColor
 import dev.adamko.kafkatorio.events.schema.Colour
+import kotlin.math.pow
 import kotlin.math.roundToInt
 import kotlinx.serialization.Serializable
 
+
 @Serializable
 @JvmInline
 value class SurfaceIndex(
@@ -13,12 +15,14 @@ value class SurfaceIndex(
   override fun toString() = "$index"
 }
 
+
 @JvmInline
 @Serializable
 value class FactorioServerId(val id: String) {
   override fun toString() = id
 }
 
+
 fun Colour.toRgbColor(): RGBColor {
   return RGBColor(
     red.roundToInt(),
@@ -27,3 +31,26 @@ fun Colour.toRgbColor(): RGBColor {
     alpha.roundToInt(),
   )
 }
+
+
+enum class ZoomLevel(
+  val level: Int
+) {
+  ZOOM_0(0),
+  ZOOM_1(1),
+  ZOOM_2(2),
+  ZOOM_3(3),
+  ZOOM_4(4),
+  ZOOM_5(5),
+  ;
+
+  val chunkSize: Int = 2f.pow(10 - level).roundToInt()
+
+  init {
+    require(chunkSize > 0) { "chunkSize must be positive" }
+  }
+
+  companion object {
+    val values: List<ZoomLevel> = values().asList() // better performance, KT-48872
+  }
+}

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

@@ -14,7 +14,7 @@ fun playerUpdatesToWsServer(
   websocketServer: WebsocketServer,
   builder: StreamsBuilder = StreamsBuilder(),
 ) {
-  builder.stream<FactorioPacketKey, KafkatorioPacket>(
+  builder.stream<FactorioServerId, KafkatorioPacket>(
     "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}",
     consumedAs("stream-player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde())
   )

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

@@ -49,7 +49,7 @@ value class PrototypeName(val name: String) {
 
 /** Get the [Colour] of each [MapTilePrototype] */
 fun tilePrototypeColourTable(builder: StreamsBuilder): KTable<PrototypeName, Colour> {
-  return builder.stream<FactorioPacketKey, FactorioPrototypes>(
+  return builder.stream<FactorioServerId, FactorioPrototypes>(
     "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
     consumedAs(
       "consume-all-prototypes-packets",

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

@@ -108,7 +108,7 @@ fun aggregateWebMapTiles(
           kxsBinary.serde(),
           kxsBinary.serde(),
         )
-      ) { pos: TileUpdateRecordKey, px: WebMapTilePixel ->
+      ) { pos: TileUpdateRecordKey, _: WebMapTilePixel ->
         val (chunkX, chunkY) = pos.tilePosition.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE)
         val chunkPos = WebMapTileChunkPosition(
           WEB_MAP_IMAGE_TILE_SIZE,

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

@@ -14,7 +14,7 @@ import kotlin.math.roundToInt
 import org.apache.kafka.streams.kstream.KTable
 
 
-private val COLOUR_TRANSPARENT = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f))
+val COLOUR_TRANSPARENT = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f))
 
 fun saveTileImages(
   webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels>
@@ -54,12 +54,7 @@ private fun saveMapTilesPng(
 
   pixels.forEach { (tilePosition, colour) ->
 
-    val rgbColour = RGBColor(
-      colour.red.roundToInt(),
-      colour.green.roundToInt(),
-      colour.blue.roundToInt(),
-      colour.alpha.roundToInt(),
-    )
+    val rgbColour = colour.toRgbColor()
 
     val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX))
     val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY))

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

@@ -36,7 +36,7 @@ fun createFactorioMap() = Maps {
       tms = true
 //      minZoom = -2
 //      maxZoom = 1
-      maxNativeZoom = 1
+      maxNativeZoom = 5
       minNativeZoom = 1
       noWrap = true
       updateWhenIdle = true

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

@@ -46,7 +46,7 @@ class WebsocketService(
             is ConsoleChatMessage,
             is EntityData,
             is MapChunk,
-            is MapTile,
+//            is MapTile,
             is SurfaceData,
             is MapTiles   -> {
             }