- factorio version bump

- mod: add a couple of commands for emitting chunks
- mod: decrease on_tick event rate
- mod: version bump typed-factorio, tstl
- kafka admin: decrease replication
- kafka admin: filter out existing-topics when creating
- events server: add more serdes, hopefully more performant than json
- events server: lots of work trying to improve performance...
- events server: increase tile size to 256
- events server: fix web-map-tile-image x/y location
- kafka: version bump
- kafka: increase default partition size
- kafka ui: version bump
-

modules/events-server/build.gradle.kts

@@ -26,6 +26,7 @@ dependencies {
   implementation(platform(libs.kotlinx.serialization.bom))
   implementation(libs.kotlinx.serialization.core)
   implementation(libs.kotlinx.serialization.json)
+  implementation("org.jetbrains.kotlinx:kotlinx-serialization-protobuf")
 
   implementation(libs.kotlinx.coroutines)
 

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt

@@ -2,11 +2,13 @@ package dev.adamko.kafkatorio.admin
 
 import dev.adamko.kafkatorio.events.schema.FactorioObjectData
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
+import dev.adamko.kafkatorio.processor.KafkatorioTopology
 import java.util.concurrent.TimeUnit
 import mu.KotlinLogging
 import org.apache.kafka.clients.admin.Admin
 import org.apache.kafka.clients.admin.CreateTopicsResult
 import org.apache.kafka.clients.admin.NewTopic
+import org.apache.kafka.clients.admin.TopicListing
 
 private val logger = KotlinLogging.logger {}
 
@@ -29,7 +31,7 @@ object CreateTopics {
     KafkatorioPacket.PacketType.values().forEach { packetType ->
 
       val kafkatorioTopics = buildSet {
-//        add(KafkatorioTopology.sourceTopic)
+        add(KafkatorioTopology.sourceTopic)
         when (packetType) {
           KafkatorioPacket.PacketType.EVENT      ->
             FactorioObjectData.ObjectName.values().forEach { objectName ->
@@ -41,7 +43,11 @@ object CreateTopics {
             add("kafkatorio.${packetType.name}.all")
         }
       }
+
+      val currentTopics = currentTopics().map { it.name() }
+
       val result = kafkatorioTopics
+        .filterNot { it in currentTopics }
         .run {
           logger.info("Creating $size topics")
           kafkaAdmin.createTopics { this }
@@ -68,10 +74,18 @@ object CreateTopics {
 
   private fun Admin.createTopics(
     numPartitions: Int = 10,
-    replicationFactor: Short = 5,
+    replicationFactor: Short = 1,
     topicNames: () -> Collection<String>,
   ): CreateTopicsResult =
-    createTopics(topicNames().distinct().map {
-      NewTopic(it, numPartitions, replicationFactor)
-    })
+    createTopics(topicNames()
+      .distinct()
+      .map {
+        NewTopic(it, numPartitions, replicationFactor)
+      })
+
+  private fun currentTopics(): MutableCollection<TopicListing> {
+    return kafkaAdmin.listTopics()
+      .listings()
+      .get()
+  }
 }

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

@@ -1,9 +1,9 @@
 package dev.adamko.kafkatorio.processor
 
 import dev.adamko.kafkatorio.events.schema.MapTile
-import dev.adamko.kafkatorio.events.schema.MapTilePosition
 import dev.adamko.kafkatorio.events.schema.MapTilePrototype
 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
@@ -33,7 +33,7 @@ class KafkatorioTopology(
     splitFactorioServerLog(builder)
     playerUpdatesToWsServer(websocketServer, builder)
 
-    val allMapTilesTable: KTable<MapTilePosition, MapTile> = allMapTilesTable(builder)
+    val allMapTilesTable: KTable<TileUpdateRecordKey, MapTile> = allMapTilesTable(builder)
     val tilePrototypesTable: KTable<PrototypeName, MapTilePrototype> = prototypesTable(builder)
 
     val webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels> =

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

@@ -0,0 +1,105 @@
+package dev.adamko.kafkatorio.processor.serdes
+
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+import java.io.DataInput
+import java.io.DataInputStream
+import java.io.DataOutput
+import java.io.DataOutputStream
+import java.util.zip.DeflaterOutputStream
+import kotlinx.serialization.BinaryFormat
+import kotlinx.serialization.DeserializationStrategy
+import kotlinx.serialization.SerializationStrategy
+import kotlinx.serialization.descriptors.SerialDescriptor
+import kotlinx.serialization.encoding.AbstractDecoder
+import kotlinx.serialization.encoding.AbstractEncoder
+import kotlinx.serialization.encoding.CompositeDecoder
+import kotlinx.serialization.encoding.CompositeEncoder
+import kotlinx.serialization.modules.EmptySerializersModule
+import kotlinx.serialization.modules.SerializersModule
+
+
+val kxsBinaryMapper: KxsBinary = KxsBinary()
+
+/** [`https://github.com/Kotlin/kotlinx.serialization/blob/master/docs/formats.md#efficient-binary-format`](https://github.com/Kotlin/kotlinx.serialization/blob/master/docs/formats.md#efficient-binary-format) */
+class KxsBinary(
+  override val serializersModule: SerializersModule = EmptySerializersModule
+) : BinaryFormat {
+
+  override fun <T> encodeToByteArray(serializer: SerializationStrategy<T>, value: T): ByteArray {
+    val output = ByteArrayOutputStream()
+    val encoder = KxsDataOutputEncoder(DataOutputStream(output), serializersModule)
+    encoder.encodeSerializableValue(serializer, value)
+    return output.toByteArray()
+  }
+
+  override fun <T> decodeFromByteArray(
+    deserializer: DeserializationStrategy<T>,
+    bytes: ByteArray
+  ): T {
+    val input = ByteArrayInputStream(bytes)
+    val decoder = KxsDataInputDecoder(DataInputStream(input), serializersModule)
+    return decoder.decodeSerializableValue(deserializer)
+  }
+}
+
+class KxsDataOutputEncoder(
+  private val output: DataOutput,
+  override val serializersModule: SerializersModule = EmptySerializersModule,
+) : AbstractEncoder() {
+
+  override fun encodeBoolean(value: Boolean) = output.writeByte(if (value) 1 else 0)
+  override fun encodeByte(value: Byte) = output.writeByte(value.toInt())
+  override fun encodeShort(value: Short) = output.writeShort(value.toInt())
+  override fun encodeInt(value: Int) = output.writeInt(value)
+  override fun encodeLong(value: Long) = output.writeLong(value)
+  override fun encodeFloat(value: Float) = output.writeFloat(value)
+  override fun encodeDouble(value: Double) = output.writeDouble(value)
+  override fun encodeChar(value: Char) = output.writeChar(value.code)
+  override fun encodeString(value: String) = output.writeUTF(value)
+  override fun encodeEnum(enumDescriptor: SerialDescriptor, index: Int) = output.writeInt(index)
+
+  override fun beginCollection(
+    descriptor: SerialDescriptor,
+    collectionSize: Int
+  ): CompositeEncoder {
+    encodeInt(collectionSize)
+    return this
+  }
+
+  override fun encodeNull() = encodeBoolean(false)
+  override fun encodeNotNullMark() = encodeBoolean(true)
+}
+
+class KxsDataInputDecoder(
+  private val input: DataInput,
+  override val serializersModule: SerializersModule = EmptySerializersModule,
+  var elementsCount: Int = 0,
+) : AbstractDecoder() {
+  private var elementIndex = 0
+  override fun decodeBoolean(): Boolean = input.readByte().toInt() != 0
+  override fun decodeByte(): Byte = input.readByte()
+  override fun decodeShort(): Short = input.readShort()
+  override fun decodeInt(): Int = input.readInt()
+  override fun decodeLong(): Long = input.readLong()
+  override fun decodeFloat(): Float = input.readFloat()
+  override fun decodeDouble(): Double = input.readDouble()
+  override fun decodeChar(): Char = input.readChar()
+  override fun decodeString(): String = input.readUTF()
+  override fun decodeEnum(enumDescriptor: SerialDescriptor): Int = input.readInt()
+
+  override fun decodeElementIndex(descriptor: SerialDescriptor): Int {
+    if (elementIndex == elementsCount) return CompositeDecoder.DECODE_DONE
+    return elementIndex++
+  }
+
+  override fun beginStructure(descriptor: SerialDescriptor): CompositeDecoder =
+    KxsDataInputDecoder(input, serializersModule, descriptor.elementsCount)
+
+  override fun decodeSequentially(): Boolean = true
+
+  override fun decodeCollectionSize(descriptor: SerialDescriptor): Int =
+    decodeInt().also { elementsCount = it }
+
+  override fun decodeNotNullMark(): Boolean = decodeBoolean()
+}

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

@@ -1,4 +1,4 @@
-package dev.adamko.kafkatorio.processor
+package dev.adamko.kafkatorio.processor.serdes
 
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
 import dev.adamko.kafkatorio.events.schema.jsonMapperKafkatorio

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

@@ -0,0 +1,59 @@
+package dev.adamko.kafkatorio.processor.serdes
+
+import dev.adamko.kafkatorio.events.schema.jsonMapperKafkatorio
+import kotlinx.serialization.BinaryFormat
+import kotlinx.serialization.decodeFromByteArray
+import kotlinx.serialization.encodeToByteArray
+import kotlinx.serialization.protobuf.ProtoBuf
+import org.apache.kafka.common.serialization.Deserializer
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.common.serialization.Serializer
+
+val protoMapper: ProtoBuf = ProtoBuf {
+  serializersModule = jsonMapperKafkatorio.serializersModule
+}
+
+
+inline fun <reified T> BinaryFormat.kafkaSerializer() =
+  Serializer { topic: String, data: T ->
+    runCatching {
+      encodeToByteArray(data)
+    }.getOrElse { e ->
+      println(
+        """
+            Exception on encodeToString 
+            Topic: $topic
+            topicData: $data
+            topicData as T: $data
+          """.trimIndent()
+      )
+      e.printStackTrace()
+      throw e
+    }
+  }
+
+inline fun <reified T> BinaryFormat.kafkaDeserializer() =
+  Deserializer { topic: String, data: ByteArray ->
+    runCatching {
+      decodeFromByteArray<T>(data)
+    }.getOrElse { e ->
+      println(
+        """
+            Exception on decodeFromString,    
+            Topic: $topic
+            topicData: $data
+            topicData as T: ${data as? T}
+          """.trimIndent()
+      )
+      e.printStackTrace()
+      throw e
+    }
+  }
+
+inline fun <reified T> ProtoBuf.serde() = object : Serde<T> {
+
+  override fun serializer(): Serializer<T> = kafkaSerializer()
+
+  override fun deserializer(): Deserializer<T> = kafkaDeserializer()
+
+}

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

@@ -1,16 +1,16 @@
 package dev.adamko.kafkatorio.processor.topology
 
-import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
-import dev.adamko.kafkatorio.events.schema.MapChunkPosition
-import dev.adamko.kafkatorio.events.schema.MapTilePosition
-import dev.adamko.kafkatorio.events.schema.MapTilePrototype
-import dev.adamko.kafkatorio.processor.jsonMapper
+//import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
+//import dev.adamko.kafkatorio.events.schema.MapChunkPosition
+//import dev.adamko.kafkatorio.events.schema.MapTilePosition
+//import dev.adamko.kafkatorio.events.schema.MapTilePrototype
+//import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 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.Consumed
-import org.apache.kafka.streams.kstream.KGroupedStream
+//import kotlinx.serialization.decodeFromString
+//import org.apache.kafka.common.serialization.Serdes
+//import org.apache.kafka.streams.StreamsBuilder
+//import org.apache.kafka.streams.kstream.Consumed
+//import org.apache.kafka.streams.kstream.KGroupedStream
 
 private const val sourceTopic = "factorio-server-log"
 
@@ -20,55 +20,55 @@ value class FactorioServerId(val id: String) {
   override fun toString() = id
 }
 
-@Serializable
-data class FactorioServerState(
-  val serverId: FactorioServerId,
-  val mapTilePrototypes: Set<MapTilePrototype>,
-  val surfaceIndexes: Set<SurfaceIndex>,
-)
-
-
-private fun gatherGameState(builder: StreamsBuilder = StreamsBuilder()) {
-
-  val serverPackets: KGroupedStream<String, KafkatorioPacket> =
-    builder.stream(
-      sourceTopic,
-      Consumed.with(Serdes.String(), Serdes.String())
-    )
-      .mapValues { readOnlyKey, value ->
-        println("Mapping $readOnlyKey:$value")
-        jsonMapper.decodeFromString<KafkatorioPacket>(value)
-      }
-      .groupByKey()
-
-  @Serializable
-  data class TileSurfaceData(
-    val surfaceIndex: Int,
-    val position: MapTilePosition,
-    val prototype: MapTilePrototype,
-  )
-
-  @Serializable
-  data class MapChunkData(
-    val position: MapChunkPosition,
-    val tiles: List<TileSurfaceData>,
-  )
-
-  @Serializable
-  data class FactorioSurfaceData(
-    val surfaceIndex: Int,
-    val chunks: List<MapChunkData>,
-  )
-
-  @Serializable
-  data class FactorioServerState(
-    val surfaces: Map<Int, FactorioSurfaceData>,
-    val tilePrototypes: List<MapTilePrototype>,
-  )
-
-
-//    serverPackets
-//      .aggregate<FactorioServerState>(
-//        Initializer { FactorioServerState("", emptyMap(), emptyList()) }
-//      )
-}
+//@Serializable
+//data class FactorioServerState(
+//  val serverId: FactorioServerId,
+//  val mapTilePrototypes: Set<MapTilePrototype>,
+//  val surfaceIndexes: Set<SurfaceIndex>,
+//)
+//
+//
+//private fun gatherGameState(builder: StreamsBuilder = StreamsBuilder()) {
+//
+//  val serverPackets: KGroupedStream<String, KafkatorioPacket> =
+//    builder.stream(
+//      sourceTopic,
+//      Consumed.with(Serdes.String(), Serdes.String())
+//    )
+//      .mapValues { readOnlyKey, value ->
+//        println("Mapping $readOnlyKey:$value")
+//        jsonMapper.decodeFromString<KafkatorioPacket>(value)
+//      }
+//      .groupByKey()
+//
+//  @Serializable
+//  data class TileSurfaceData(
+//    val surfaceIndex: Int,
+//    val position: MapTilePosition,
+//    val prototype: MapTilePrototype,
+//  )
+//
+//  @Serializable
+//  data class MapChunkData(
+//    val position: MapChunkPosition,
+//    val tiles: List<TileSurfaceData>,
+//  )
+//
+//  @Serializable
+//  data class FactorioSurfaceData(
+//    val surfaceIndex: Int,
+//    val chunks: List<MapChunkData>,
+//  )
+//
+//  @Serializable
+//  data class FactorioServerState(
+//    val surfaces: Map<Int, FactorioSurfaceData>,
+//    val tilePrototypes: List<MapTilePrototype>,
+//  )
+//
+//
+////    serverPackets
+////      .aggregate<FactorioServerState>(
+////        Initializer { FactorioServerState("", emptyMap(), emptyList()) }
+////      )
+//}

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

@@ -7,7 +7,7 @@ import dev.adamko.kafkatorio.events.schema.MapChunk
 import dev.adamko.kafkatorio.events.schema.MapTile
 import dev.adamko.kafkatorio.events.schema.MapTilePosition
 import dev.adamko.kafkatorio.events.schema.MapTiles
-import dev.adamko.kafkatorio.processor.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.extensions.materializedWith
 import dev.adamko.kotka.extensions.streams.filter
@@ -17,12 +17,12 @@ import dev.adamko.kotka.extensions.streams.toTable
 import dev.adamko.kotka.kxs.serde
 import java.time.Duration
 import kotlinx.serialization.Serializable
-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.kstream.KTable
 import org.apache.kafka.streams.kstream.Suppressed
 
+
 @Serializable
 data class TileUpdateRecordKey(
   val surfaceIndex: SurfaceIndex,
@@ -32,26 +32,42 @@ data class TileUpdateRecordKey(
 
 fun allMapTilesTable(
   builder: StreamsBuilder
-): KTable<MapTilePosition, MapTile> {
+): KTable<TileUpdateRecordKey, MapTile> {
 
-  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 luaTilesUpdatesStream: KStream<FactorioPacketKey, MapTiles> =
+    builder.stream<FactorioPacketKey, 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 ->
+        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 }
+  val chunkTilesUpdateStream: KStream<FactorioPacketKey, MapTiles> =
+    builder.stream<FactorioPacketKey, 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 ->
+        event.data is MapChunk
+      }
+      .mapValues("events.extract-map-chunks") { _, event: FactorioEvent ->
+        (event.data as? MapChunk)!!.tiles
+      }
 
   return luaTilesUpdatesStream
     .merge(chunkTilesUpdateStream)
     .flatMap("flat-map-all-tiles") { _, value ->
-      value.tiles.map { it.position to it }
+      value.tiles.map { tile ->
+        val key = TileUpdateRecordKey(
+          SurfaceIndex(value.surfaceIndex),
+          tile.position
+        )
+        key to tile
+      }
     }
     .toTable(
       "all-tiles-keyed-on-position",
@@ -66,21 +82,3 @@ fun allMapTilesTable(
       )
     )
 }
-
-//fun buildTilesTable(
-//  builder: StreamsBuilder
-//): KTable<TileUpdateRecordKey, TileUpdateRecord> {
-//
-//  return luaTilesUpdatesStream
-//    .merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream)
-//    .flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles ->
-//      mapTiles.tiles.map { tile ->
-//        TileUpdateRecord(
-//          SurfaceIndex(mapTiles.surfaceIndex),
-//          tile.position,
-//          tile,
-//        )
-//      }
-//    }
-//    .toTable(materializedWith(jsonMapper.serde(), jsonMapper.serde()))
-//}

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

@@ -1,22 +1,39 @@
 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.processor.KafkatorioPacketSerde
+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.serdes.KafkatorioPacketSerde
 import dev.adamko.kafkatorio.processor.KafkatorioTopology
-import dev.adamko.kafkatorio.processor.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.extensions.producedAs
-import dev.adamko.kotka.extensions.streams.mapValues
+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.processor.RecordContext
 
 
+@Serializable
+data class FactorioPacketKey(
+  val serverId: FactorioServerId,
+  /** arbitrary distinctive key */
+  val key: String,
+)
+
 fun splitFactorioServerLog(
   builder: StreamsBuilder = StreamsBuilder(),
 ) {
@@ -29,17 +46,19 @@ fun splitFactorioServerLog(
       Serdes.String(),
     )
   )
-    .mapValues("decode-packets") { _: String, value: String ->
+    .map("decode-packets") { serverId: String, value: String ->
 //        println("Mapping $readOnlyKey:$value")
-      jsonMapper.decodeFromString<KafkatorioPacket>(value)
+      val packet = jsonMapper.decodeFromString<KafkatorioPacket>(value)
+      val key: String = createPacketKey(packet)
+      FactorioPacketKey(FactorioServerId(serverId), key) to packet
     }
     .to(
       producedAs(
         "split-server-log",
-        Serdes.String(),
+        jsonMapper.serde(),
         KafkatorioPacketSerde
       )
-    ) { _: String, value: KafkatorioPacket, _: RecordContext ->
+    ) { _: FactorioPacketKey, value: KafkatorioPacket, _: RecordContext ->
 //        println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
       when (value) {
         is FactorioEvent               ->
@@ -52,3 +71,26 @@ fun splitFactorioServerLog(
     }
 
 }
+
+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/model.kt

@@ -3,8 +3,9 @@ package dev.adamko.kafkatorio.processor.topology
 import kotlinx.serialization.Serializable
 
 @Serializable
-data class SurfaceIndex(
-  val surfaceIndex: Int
+@JvmInline
+value class SurfaceIndex(
+  val index: Int
 ) {
-  override fun toString() = "$surfaceIndex"
+  override fun toString() = "$index"
 }

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

@@ -2,9 +2,9 @@ package dev.adamko.kafkatorio.processor.topology
 
 import dev.adamko.kafkatorio.events.schema.FactorioObjectData
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
-import dev.adamko.kafkatorio.processor.KafkatorioPacketSerde
+import dev.adamko.kafkatorio.processor.serdes.KafkatorioPacketSerde
 import dev.adamko.kafkatorio.processor.WebsocketServer
-import dev.adamko.kafkatorio.processor.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import kotlinx.serialization.encodeToString
 import org.apache.kafka.common.serialization.Serdes
 import org.apache.kafka.streams.StreamsBuilder

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

@@ -3,12 +3,10 @@ package dev.adamko.kafkatorio.processor.topology
 import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
 import dev.adamko.kafkatorio.events.schema.MapTilePrototype
-import dev.adamko.kafkatorio.processor.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kotka.extensions.consumedAs
-//import dev.adamko.kotka.extensions.groupedAs
 import dev.adamko.kotka.extensions.materializedWith
 import dev.adamko.kotka.extensions.streams.flatMap
-//import dev.adamko.kotka.extensions.streams.*
 import dev.adamko.kotka.kxs.serde
 import kotlinx.serialization.Serializable
 import org.apache.kafka.common.serialization.Serdes

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

@@ -5,7 +5,7 @@ import dev.adamko.kafkatorio.events.schema.MapTile
 import dev.adamko.kafkatorio.events.schema.MapTilePosition
 import dev.adamko.kafkatorio.events.schema.MapTilePrototype
 import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
-import dev.adamko.kafkatorio.processor.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kotka.extensions.groupedAs
 import dev.adamko.kotka.extensions.materializedAs
 import dev.adamko.kotka.extensions.namedAs
@@ -19,26 +19,16 @@ import kotlinx.serialization.Serializable
 import org.apache.kafka.streams.kstream.KTable
 import org.apache.kafka.streams.kstream.Suppressed
 
-/** The tiles of the webmap will be this many pixels high/wide */
-const val WEB_MAP_IMAGE_TILE_SIZE = 32
 
-//@Serializable
-//data class MapChunkDataPosition(
-//  val position: MapChunkPosition,
-//  val surfaceIndex: SurfaceIndex,
-//)
-//
-//@Serializable
-//data class MapChunkData(
-//  val chunkPosition: MapChunkDataPosition,
-//  val tiles: Set<MapTile>,
-//)
+/** The tiles of the webmap will be this many pixels high/wide */
+const val WEB_MAP_IMAGE_TILE_SIZE = 256
 
 @Serializable
 data class WebMapTileChunkPosition(
   val chunkSize: Int,
   val x: Int,
   val y: Int,
+  val surfaceIndex: SurfaceIndex,
 )
 
 @Serializable
@@ -59,11 +49,11 @@ private data class WebMapTileChunkPixelsAcc(
 )
 
 fun aggregateWebMapTiles(
-  allMapTilesTable: KTable<MapTilePosition, MapTile>,
+  allMapTilesTable: KTable<TileUpdateRecordKey, MapTile>,
   tilePrototypesTable: KTable<PrototypeName, MapTilePrototype>,
 ): KTable<WebMapTileChunkPosition, WebMapTileChunkPixels> {
 
-  val webmapTileColours: KTable<MapTilePosition, WebMapTilePixel> =
+  val webmapTileColours: KTable<TileUpdateRecordKey, WebMapTilePixel> =
     allMapTilesTable
       .join(
         tilePrototypesTable,
@@ -87,9 +77,14 @@ fun aggregateWebMapTiles(
           jsonMapper.serde(),
           jsonMapper.serde(),
         )
-      ) { pos: MapTilePosition, px: WebMapTilePixel ->
-        val (chunkX, chunkY) = pos.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE)
-        val chunkPos = WebMapTileChunkPosition(WEB_MAP_IMAGE_TILE_SIZE, chunkX, chunkY)
+      ) { pos: TileUpdateRecordKey, px: WebMapTilePixel ->
+        val (chunkX, chunkY) = pos.tilePosition.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE)
+        val chunkPos = WebMapTileChunkPosition(
+          WEB_MAP_IMAGE_TILE_SIZE,
+          chunkX,
+          chunkY,
+          pos.surfaceIndex
+        )
         (chunkPos to px).toKeyValue()
       }
       .aggregate(
@@ -103,7 +98,11 @@ fun aggregateWebMapTiles(
           acc.apply { pixels.minus(oldValue) }
         },
         namedAs("web-map-tile-colours-into-$WEB_MAP_IMAGE_TILE_SIZE-chunks"),
-        materializedAs("web-map-tile-colour-chunks-aggregate", jsonMapper.serde(), jsonMapper.serde())
+        materializedAs(
+          "web-map-tile-colour-chunks-aggregate",
+          jsonMapper.serde(),
+          jsonMapper.serde()
+        )
       )
       .suppress(
         Suppressed.untilTimeLimit<WebMapTileChunkPosition>(

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

@@ -3,7 +3,6 @@ package dev.adamko.kafkatorio.processor.topology
 import com.sksamuel.scrimage.ImmutableImage
 import com.sksamuel.scrimage.color.RGBColor
 import com.sksamuel.scrimage.nio.PngWriter
-import dev.adamko.kafkatorio.events.schema.MAP_CHUNK_SIZE
 import dev.adamko.kotka.extensions.tables.toStream
 import java.awt.Color
 import java.awt.image.BufferedImage
@@ -43,15 +42,14 @@ private fun saveMapTilesPng(
   chunkPosition: WebMapTileChunkPosition,
   pixels: Set<WebMapTilePixel>
 ) {
-  val surfaceIndex = 1 // TODO get surface index
-
-  val chunkOriginX = chunkPosition.x * chunkPosition.chunkSize
-  val chunkOriginY = chunkPosition.y * chunkPosition.chunkSize
+  val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize
+  val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize
+  val surfaceIndex: Int = chunkPosition.surfaceIndex.index
 
   val chunkImage =
     ImmutableImage.filled(
-      MAP_CHUNK_SIZE,
-      MAP_CHUNK_SIZE,
+      chunkPosition.chunkSize,
+      chunkPosition.chunkSize,
       Color.BLACK,
       BufferedImage.TYPE_INT_ARGB
     )
@@ -79,7 +77,7 @@ private fun saveMapTilesPng(
   val zoom = 1u
   val file =
     File(
-      "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkOriginX}/y${chunkOriginY}.png"
+      "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkPosition.x}/y${chunkPosition.y}.png"
     )
 
   if (file.parentFile.mkdirs()) {