events-server: fix tile grouping, chunking, saving

- debounce chunk and tile updates
- make the tile output-dir configurable
- basic test for manual checking

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

@@ -134,9 +134,27 @@ fun MapChunkPosition.leftTopTile(chunkSize: Int = MAP_CHUNK_SIZE): MapTilePositi
     y * chunkSize,
   )
 
+
+fun MapChunkPosition.rightBottomTile(chunkSize: Int = MAP_CHUNK_SIZE): MapTilePosition =
+  leftTopTile(chunkSize) + (chunkSize - 1)
+
+
+fun MapChunkPosition.iterateTiles(chunkSize: Int = MAP_CHUNK_SIZE): Iterator<MapTilePosition> {
+  val leftTop = leftTopTile(chunkSize)
+  val rightBottom = rightBottomTile(chunkSize)
+  return iterator {
+    (leftTop.x..rightBottom.x).forEach { x ->
+      (leftTop.y..rightBottom.y).forEach { y ->
+        yield(MapTilePosition(x, y))
+      }
+    }
+  }
+}
+
+
 //val MapChunkPosition.rightBottomTile: MapTilePosition
 //  get() = leftTopTile + (MAP_CHUNK_SIZE - 1)
-//
+
 //
 //val MapChunkPosition.boundingBox: MapBoundingBox
 //  get() = MapBoundingBox(leftTopTile, rightBottomTile)

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

@@ -61,7 +61,7 @@ internal class KafkatorioTopology(
   private fun saveTiles() {
     val builder = StreamsBuilder()
 //    val topology = saveMapTiles(builder)
-    val topology = saveMapTiles2(builder)
+    val topology = saveMapTiles2(builder, appProps.webmapTileDir)
     launchTopology("saveTiles", topology)
   }
 

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

@@ -17,11 +17,18 @@ private const val DOMAIN = "kafkatorio"
 const val TOPIC_SRC_SERVER_LOG = "$DOMAIN.src.server-log"
 
 const val TOPIC_GROUPED_MAP_CHUNKS_STATE = "$DOMAIN.state.map-chunks.grouped"
+const val TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED = "$TOPIC_GROUPED_MAP_CHUNKS_STATE.debounced"
+
+const val TOPIC_SUBDIVIDED_MAP_TILES = "$DOMAIN.map-tiles.subdivided"
+const val TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED = "$TOPIC_SUBDIVIDED_MAP_TILES.debounced"
 
 
 fun allTopics(): Set<String> = buildSet {
   add(TOPIC_SRC_SERVER_LOG)
   add(TOPIC_GROUPED_MAP_CHUNKS_STATE)
+  add(TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED)
+  add(TOPIC_SUBDIVIDED_MAP_TILES)
+  add(TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED)
 
   addAll(packetTopicNames.values)
 }

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

@@ -4,6 +4,7 @@ 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.KafkatorioTopology.Companion.rootJob
 import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
 import dev.adamko.kafkatorio.schema.common.ColourHex
@@ -82,8 +83,7 @@ value class TilePngFilename(
   val value: String,
 ) {
   constructor(id: ServerMapChunkId) : this(buildString {
-    append("src/main/resources/kafkatorio-web-map")
+    append("s${id.surfaceIndex}")
-    append("/s${id.surfaceIndex}")
     append("/z${id.chunkSize.zoomLevel}")
     append("/x${id.chunkPosition.x}")
     append("/y${id.chunkPosition.y}")
@@ -198,7 +198,7 @@ private class ServerMapChunkHandler : CoroutineScope {
                 .tiles
                 .entries
                 .groupBy(
-                  { (tile, _) -> tile.toMapChunkPosition(chunkSize.tilesPerChunk) }
+                  { (tile, _) -> tile.toMapChunkPosition(chunkSize.lengthInTiles) }
                 ) { (tilePosition, colour) ->
                   tilePosition to colour
                 }
@@ -245,14 +245,14 @@ private class ServerMapChunkHandler : CoroutineScope {
   ): ImmutableImage {
 
     val chunkImage = ImmutableImage.filled(
-      chunkSize.tilesPerChunk,
+      chunkSize.lengthInTiles,
-      chunkSize.tilesPerChunk,
+      chunkSize.lengthInTiles,
       TRANSPARENT_AWT,
       BufferedImage.TYPE_INT_ARGB
     )
 
-    val chunkOriginX: Int = chunkPosition.x * chunkSize.tilesPerChunk
+    val chunkOriginX: Int = chunkPosition.x * chunkSize.lengthInTiles
-    val chunkOriginY: Int = chunkPosition.y * chunkSize.tilesPerChunk
+    val chunkOriginY: Int = chunkPosition.y * chunkSize.lengthInTiles
 
     chunkColours.forEach { (tilePosition, colour) ->
 

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

@@ -3,18 +3,25 @@ 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.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.ColourHex
+import dev.adamko.kafkatorio.schema.common.MapTilePosition
 import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
 import dev.adamko.kotka.extensions.consumedAs
+import dev.adamko.kotka.extensions.producedAs
 import dev.adamko.kotka.extensions.streams.flatMap
 import dev.adamko.kotka.extensions.streams.forEach
 import dev.adamko.kotka.kxs.serde
 import java.awt.image.BufferedImage
-import java.io.File
+import java.nio.file.Path
 import kotlin.io.path.fileSize
 import kotlin.math.abs
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.measureTimedValue
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.Topology
 import org.apache.kafka.streams.kstream.KStream
@@ -22,78 +29,130 @@ import org.apache.kafka.streams.kstream.KStream
 
 private const val WEB_MAP_TILE_SIZE_PX = 256
 
-
+/**
+ * @param[tileDirectory] [dev.adamko.kafkatorio.processor.config.ApplicationProperties.webmapTileDir]
+ */
 fun saveMapTiles2(
   builder: StreamsBuilder,
+  tileDirectory: Path,
 ): Topology {
   val pid = "saveMapTiles"
 
   val groupedMapChunkTiles: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
     builder.stream(
-      TOPIC_GROUPED_MAP_CHUNKS_STATE,
+      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.flatMap("$pid.subdivide-chunks") { initialKey, initialChunk ->
-    ChunkSize.entries.flatMap { chunkSize ->
-      initialChunk.map.entries
-        .groupBy(
-          { (tile, _) -> tile.toMapChunkPosition(chunkSize.tilesPerChunk) }
-        ) { (tilePosition, colour) ->
-          tilePosition to colour
-        }
-        .map { (chunkPos, tiles) ->
-          val key = initialKey.copy(
-            chunkSize = chunkSize,
-            chunkPosition = chunkPos
-          )
 
-          val value = ServerMapChunkTiles(key, tiles.toMap())
+    val (chunks, time) = measureTimedValue {
+      ChunkSize.entries.flatMap { chunkSize ->
+        initialChunk.map
+          .entries
+          .groupingBy { (tilePosition, _) ->
+            val chunkPosition = tilePosition.toMapChunkPosition(chunkSize.lengthInTiles)
+//            println("$tilePosition groupingBy $chunkPosition")
+            initialKey.copy(
+              chunkSize = chunkSize,
+              chunkPosition = chunkPosition,
+            )
+          }.fold(
+            initialValueSelector = { _, _ -> mutableMapOf<MapTilePosition, ColourHex>() }
+          ) { _, accumulator, (tilePosition, colour) ->
+//            println("fold $tilePosition $colour")
+            accumulator[tilePosition] = colour
+            accumulator
+          }
+          .map { (chunkId, tiles) ->
+//            println("map $chunkId ${tiles.size}")
+            chunkId to ServerMapChunkTiles(chunkId, tiles)
+          }
+      }
 
-          key to value
-        }
     }
-  }.forEach("$pid.save-tiles") { key, value ->
+    println("subdivided ${initialKey.chunkPosition} to ${chunks.size} chunks in $time")
-    val filename = TilePngFilename(key)
-    val chunkColours = value.map
 
-    val chunkImage = ImmutableImage.filled(
+    chunks
-      key.chunkSize.tilesPerChunk,
+  }.to(
-      key.chunkSize.tilesPerChunk,
+    TOPIC_SUBDIVIDED_MAP_TILES,
-      TRANSPARENT_AWT,
+    producedAs(
-      BufferedImage.TYPE_INT_ARGB
+      "$pid.produce.grouped-map-chunks",
+      kxsBinary.serde(),
+      kxsBinary.serde(),
+    )
+  )
+
+  val subdividedMapChunkTilesDebounced: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
+    builder.stream(
+      TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED,
+      consumedAs(
+        "$pid.consume.debounced-subdivided-chunks",
+        kxsBinary.serde(),
+        kxsBinary.serde(),
+      ),
     )
 
-    val chunkOriginX: Int = key.chunkPosition.x * key.chunkSize.tilesPerChunk
+  subdividedMapChunkTilesDebounced
-    val chunkOriginY: Int = key.chunkPosition.y * key.chunkSize.tilesPerChunk
+    .forEach("$pid.save-tiles") { chunkId: ServerMapChunkId, tileColours: ServerMapChunkTiles<ColourHex> ->
 
-    chunkColours.forEach { (tilePosition, colour) ->
+      val chunkImage = ImmutableImage.filled(
-      val rgbColour = colour.toRgbColor()
+        chunkId.chunkSize.lengthInTiles,
+        chunkId.chunkSize.lengthInTiles,
+        TRANSPARENT_AWT,
+        BufferedImage.TYPE_INT_ARGB
+      )
 
-      val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX))
+      val chunkOriginX: Int = chunkId.chunkPosition.x * chunkId.chunkSize.lengthInTiles
-      val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY))
+      val chunkOriginY: Int = chunkId.chunkPosition.y * chunkId.chunkSize.lengthInTiles
 
-      chunkImage.setColor(pixelX, pixelY, rgbColour)
+      tileColours.map.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)
+      }
+
+      val scaledImage =
+        chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale)
+
+      val filename = TilePngFilename(chunkId)
+      saveTile(filename, tileDirectory, scaledImage)
     }
 
-    val img = chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale)
-
-    val chunkImageFile = File(filename.value)
-    if (chunkImageFile.parentFile.mkdirs()) {
-      println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
-    }
-
-    val sizeBefore = chunkImageFile.takeIf { it.exists() }?.toPath()?.fileSize()
-
-    val savedTile = img.output(PngWriter.NoCompression, chunkImageFile)
-
-    val sizeAfter = savedTile.takeIf { it.exists() }?.toPath()?.fileSize()
-    if (sizeBefore != sizeAfter) {
-//          println("savedTile $sizeBefore/$sizeAfter: $savedTile")
-    } else {
-      println("savedTile NO-SIZE-CHANGE $sizeBefore/$sizeAfter: $savedTile")
-    }
-  }
-
   return builder.build()
+    .addDebounceProcessor<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>(
+      namePrefix = pid,
+      sourceTopic = TOPIC_SUBDIVIDED_MAP_TILES,
+      sinkTopic = TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED,
+      inactivityDuration = 15.seconds,
+      keySerde = kxsBinary.serde(),
+      valueSerde = kxsBinary.serde(),
+    )
+}
+
+@Synchronized
+fun saveTile(
+  filename: TilePngFilename,
+  tileDirectory: Path,
+  img: ImmutableImage,
+) {
+  val chunkImageFile = tileDirectory.resolve(filename.value).toFile()
+  if (chunkImageFile.parentFile.mkdirs()) {
+    println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
+  }
+  val sizeBefore = chunkImageFile.takeIf { it.exists() }?.toPath()?.fileSize()
+
+  val savedTile = img.output(PngWriter.NoCompression, chunkImageFile)
+
+  val sizeAfter = savedTile.takeIf { it.exists() }?.toPath()?.fileSize()
+
+  val sizeDescription = if (sizeBefore != sizeAfter && sizeBefore != null) {
+    "$sizeBefore/$sizeAfter"
+  } else {
+    "$sizeAfter"
+  }
+  println("savedTile $sizeDescription: $savedTile")
 }

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

@@ -1,6 +1,8 @@
 package dev.adamko.kafkatorio.processor.topology
 
 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.ColourHex
 import dev.adamko.kafkatorio.schema.common.MapChunkPosition
@@ -13,7 +15,7 @@ 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.groupedAs
-import dev.adamko.kotka.extensions.materializedWith
+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
@@ -27,6 +29,7 @@ import dev.adamko.kotka.extensions.tables.toStream
 import dev.adamko.kotka.kxs.serde
 import kotlin.collections.component1
 import kotlin.collections.component2
+import kotlin.time.Duration.Companion.seconds
 import kotlinx.serialization.Serializable
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.Topology
@@ -82,12 +85,20 @@ fun groupMapChunks(builder: StreamsBuilder): Topology {
       TOPIC_GROUPED_MAP_CHUNKS_STATE,
       producedAs(
         "$pid.grouped-map-chunks",
-        kxsBinary.serde<ServerMapChunkId>(),
+        kxsBinary.serde(),
-        kxsBinary.serde<ServerMapChunkTiles<ColourHex>>()
+        kxsBinary.serde(),
       )
     )
 
   return builder.build()
+    .addDebounceProcessor<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>(
+      namePrefix = pid,
+      sourceTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE,
+      sinkTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED,
+      inactivityDuration = 15.seconds,
+      keySerde = kxsBinary.serde(),
+      valueSerde = kxsBinary.serde(),
+    )
 }
 
 
@@ -122,7 +133,7 @@ private fun groupTilesIntoChunksWithColours(
           mapTiles
             .tiles
             .groupBy { tile ->
-              tile.position.toMapChunkPosition(standardChunkSize.tilesPerChunk)
+              tile.position.toMapChunkPosition(standardChunkSize.lengthInTiles)
             }
             .map { (chunkPos, tiles) ->
 
@@ -165,9 +176,8 @@ private fun groupTilesIntoChunksWithColours(
       )
       .reduce(
         "server-map-data.tiles.reduce",
-        materializedWith(
+        materializedAs(
-//        materializedAs(
+          "server-map-data.tiles.reduce.store",
-//          "server-map-data.tiles.reduce.store",
           kxsBinary.serde<ServerMapChunkId>(),
           kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
         )
@@ -185,10 +195,10 @@ private fun groupTilesIntoChunksWithColours(
       .join(
         other = tileProtoColourDict,
         tableJoined = tableJoined("server-map-data.join-tiles-with-prototypes"),
-        materialized = materializedWith(
+        materialized = materializedAs(
-//          "server-map-data.join-tiles-with-prototypes.store",
+          "server-map-data.join-tiles-with-prototypes.store",
-          kxsBinary.serde<ServerMapChunkId>(),
+          kxsBinary.serde(),
-          kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
+          kxsBinary.serde(),
         ),
         foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<TileProtoHashCode> ->
           chunkTiles.chunkId.serverId

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

@@ -23,14 +23,15 @@ enum class ChunkSize(
   CHUNK_032(3),
   ;
 
-  val tilesPerChunk: Int = 2f.pow(8 - zoomLevel).roundToInt()
+  /** 32, 64, 128, 256, or 512 */
+  val lengthInTiles: Int = 2f.pow(8 - zoomLevel).roundToInt()
 
   init {
-    require(tilesPerChunk > 0) { "tilesPerChunk $tilesPerChunk must be positive" }
+    require(lengthInTiles > 0) { "tilesPerChunk $lengthInTiles must be positive" }
     // 1000 & 0111 = 0000 =>  pow^2
     // 1001 & 1000 = 1000 => !pow^2
-    require(tilesPerChunk and (tilesPerChunk - 1) == 0) {
+    require(lengthInTiles and (lengthInTiles - 1) == 0) {
-      "tilesPerChunk $tilesPerChunk must be a power-of-two number"
+      "tilesPerChunk $lengthInTiles must be a power-of-two number"
     }
   }
 
@@ -38,8 +39,8 @@ enum class ChunkSize(
     // cache values for better performance. KT-48872
     val entries: Set<ChunkSize> = values().toSet()
 
-    val MAX: ChunkSize = entries.maxByOrNull { it.tilesPerChunk }!!
+    val MAX: ChunkSize = entries.maxByOrNull { it.lengthInTiles }!!
-    val MIN: ChunkSize = entries.minByOrNull { it.tilesPerChunk }!!
+    val MIN: ChunkSize = entries.minByOrNull { it.lengthInTiles }!!
     val STANDARD: ChunkSize = entries.first { it.zoomLevel == 0 }
 
   }

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

@@ -0,0 +1,195 @@
+package dev.adamko.kafkatorio.processor.topology
+
+import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kafkatorio.schema.common.ColourHex
+import dev.adamko.kafkatorio.schema.common.MapChunkPosition
+import dev.adamko.kafkatorio.schema.common.MapTilePosition
+import dev.adamko.kafkatorio.schema.common.SurfaceIndex
+import dev.adamko.kafkatorio.schema.common.iterateTiles
+import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
+import dev.adamko.kotka.kxs.serde
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.property.Arb
+import io.kotest.property.PropTestConfig
+import io.kotest.property.RandomSource
+import io.kotest.property.arbitrary.arbitrary
+import io.kotest.property.arbitrary.int
+import io.kotest.property.arbitrary.uByte
+import io.kotest.property.checkAll
+import java.nio.file.Path
+import java.time.LocalTime
+import java.time.format.DateTimeFormatter
+import kotlin.io.path.Path
+import kotlin.io.path.absolutePathString
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.minutes
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.toJavaDuration
+import kotlinx.coroutines.delay
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.TestInputTopic
+import org.apache.kafka.streams.TopologyTestDriver
+
+
+class SaveMapChunksTest : FunSpec({
+
+  test("save tile") {
+
+
+    Scenario().use { scenario ->
+
+      val chunkTilesArb = serverMapChunkTilesArb(scenario.serverId)
+
+      checkAll(
+        config = PropTestConfig(iterations = 1),
+        chunkTilesArb,
+      ) { chunkTiles ->
+
+        scenario.groupedMapChunksInputTopic.pipeInput(
+          chunkTiles.chunkId,
+          chunkTiles,
+        )
+
+        scenario += 1.minutes
+        delay(5.seconds)
+        scenario += 1.minutes
+//        delay(5.seconds)
+        scenario += 1.minutes
+//        delay(5.seconds)
+        scenario += 1.minutes
+//        delay(5.seconds)
+
+        println("file:///${scenario.outputDir.absolutePathString()}")
+      }
+    }
+  }
+
+}) {
+
+  class Scenario(
+    streamsBuilder: StreamsBuilder = StreamsBuilder(),
+
+    groupedMapChunksInputTopicName: String = "kafkatorio.state.map-chunks.grouped.debounced",
+
+    val serverId: FactorioServerId = FactorioServerId("test-server-id"),
+  ) : AutoCloseable {
+
+    val outputDir: Path = run {
+      val now = LocalTime.now().format(DateTimeFormatter.ISO_TIME).filter { it.isLetterOrDigit() }
+      Path("build/test/save-map-chunks-${now}")
+    }
+    private val topology = saveMapTiles2(streamsBuilder, outputDir)
+    val testDriver: TopologyTestDriver = TopologyTestDriver(topology)
+
+    val groupedMapChunksInputTopic: TestInputTopic<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
+      testDriver.createInputTopic(
+        groupedMapChunksInputTopicName,
+        kxsBinary.serde<ServerMapChunkId>().serializer(),
+        kxsBinary.serde<ServerMapChunkTiles<ColourHex>>().serializer(),
+      )
+
+    operator fun plusAssign(wallClockTime: Duration) {
+      testDriver.advanceWallClockTime(wallClockTime.toJavaDuration())
+    }
+
+    override fun close() {
+      testDriver.close()
+    }
+  }
+
+  companion object {
+
+    fun mapTilePositionArb(
+      xArb: Arb<Int> = Arb.int(),
+      yArb: Arb<Int> = Arb.int(),
+    ): Arb<MapTilePosition> = arbitrary {
+      MapTilePosition(xArb.bind(), yArb.bind())
+    }
+
+    fun mapChunkPositionArb(
+      xArb: Arb<Int> = Arb.int(-10..10),
+      yArb: Arb<Int> = Arb.int(-10..10),
+    ): Arb<MapChunkPosition> = arbitrary {
+      MapChunkPosition(xArb.bind(), yArb.bind())
+    }
+
+    fun colourHexArb(
+      redArb: Arb<UByte> = Arb.uByte(),
+      greenArb: Arb<UByte> = Arb.uByte(),
+      blueArb: Arb<UByte> = Arb.uByte(),
+      alphaArb: Arb<UByte> = Arb.uByte(),
+    ): Arb<ColourHex> = arbitrary {
+      ColourHex(
+        redArb.bind(),
+        greenArb.bind(),
+        blueArb.bind(),
+//        alphaArb.bind(),
+        255u,
+      )
+//      ColourHex(
+//       255u,
+//        0u,
+//        0u,
+//        255u,
+//      )
+    }
+
+    fun serverMapChunkIdArb(
+      serverId: FactorioServerId,
+      chunkPosition: Arb<MapChunkPosition> = mapChunkPositionArb(),
+      surfaceIndex: SurfaceIndex = SurfaceIndex(1u),
+      chunkSize: ChunkSize = ChunkSize.MAX,
+    ): Arb<ServerMapChunkId> = arbitrary {
+      ServerMapChunkId(
+        serverId = serverId,
+        chunkPosition = chunkPosition.bind(),
+        surfaceIndex = surfaceIndex,
+        chunkSize = chunkSize,
+      )
+    }
+
+    fun tilesInChunkArb(
+      chunkId: ServerMapChunkId
+    ): Arb<Map<MapTilePosition, ColourHex>> = arbitrary {
+
+      val tileColours = mutableMapOf<MapChunkPosition, ColourHex>()
+
+      println("chunk size: ${chunkId.chunkSize} ${chunkId.chunkSize.lengthInTiles}")
+
+      val map = chunkId.chunkPosition
+        .iterateTiles(chunkId.chunkSize.lengthInTiles)
+        .asSequence()
+        .map { tilePos ->
+          val colour =
+            tileColours.getOrPut(tilePos.toMapChunkPosition(ChunkSize.MIN.lengthInTiles)) {
+              colourHexArb().sample(RandomSource.default()).value
+            }
+          tilePos to colour
+        }
+//        .take(2500)
+        .toMap()
+
+      println(map.size)
+
+      map
+    }
+
+    fun serverMapChunkTilesArb(
+      serverId: FactorioServerId,
+    ): Arb<ServerMapChunkTiles<ColourHex>> {
+
+      val serverMapChunkId = serverMapChunkIdArb(serverId)
+
+      return arbitrary {
+        val chunkId = serverMapChunkId.bind()
+
+        val tileColoursArb = tilesInChunkArb(serverMapChunkId.bind())
+
+        ServerMapChunkTiles(
+          chunkId = chunkId,
+          map = tileColoursArb.bind()
+        )
+      }
+    }
+  }
+}