tiles from kafka -> image tiles... maybe???

modules/events-server/build.gradle.kts

@@ -31,10 +31,11 @@ dependencies {
 
   implementation(platform(libs.http4k.bom))
   implementation(libs.bundles.http4k)
-
   implementation(libs.http4k.format.kotlinx)
 
   implementation(libs.bundles.logging)
+
+  implementation("com.sksamuel.scrimage:scrimage-core:4.0.24")
 }
 
 application {

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/GlobalKTableDefinition.kt

@@ -0,0 +1,31 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams
+
+import org.apache.kafka.streams.state.QueryableStoreTypes.timestampedKeyValueStore
+
+/**
+ * * update topic [topicName]
+ * * plus table-source topic [tableName]
+ * * plus state store [globalStoreName]
+ *
+ * equals a GlobalTableDefinition
+ */
+abstract class GlobalKTableDefinition<K, V>(
+    val tableName: String,
+    override val serdes: KeyValueSerdes<K, V>,
+    val globalStoreName: String = "$tableName-store",
+    /**
+     * The backing topic, from which this table will be filled.
+     *
+     * Updates to items can be sent or received with this topic.
+     */
+    final override val topicName: String = "$tableName-topic",
+    val storeType: TimestampedQueryStoreType<K, V> = timestampedKeyValueStore(),
+) : TopicDefinition<K, V> {
+  val topicNames: Set<String>
+
+  init {
+    require(tableName != topicName)
+    require(tableName != globalStoreName)
+    topicNames = setOf(tableName, topicName)
+  }
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamExtensions.kt

@@ -0,0 +1,111 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams
+
+import dev.adamko.kafkatorio.extensions.kafkastreams.KStreamExtensions.flatMap
+import dev.adamko.kafkatorio.extensions.kafkastreams.KStreamExtensions.groupBy
+import dev.adamko.kafkatorio.extensions.kafkastreams.KStreamExtensions.map
+import dev.adamko.kafkatorio.extensions.kafkastreams.KStreamExtensions.mapValues
+import dev.adamko.kafkatorio.extensions.kafkastreams.KafkaStreamsExtensions.withName
+import dev.adamko.kafkatorio.extensions.kafkastreams.KeyPairExtensions.kvPair
+import dev.adamko.kafkatorio.extensions.kafkastreams.KeyPairExtensions.toKeyValue
+import org.apache.kafka.streams.KeyValue
+import org.apache.kafka.streams.kstream.Grouped
+import org.apache.kafka.streams.kstream.KGroupedStream
+import org.apache.kafka.streams.kstream.KStream
+import org.apache.kafka.streams.kstream.Transformer
+import org.apache.kafka.streams.processor.ProcessorContext
+
+object KStreamExtensions {
+
+//  /** @see KStream.map */
+//  inline fun <inK, inV, reified outK, reified outV : TopicData<outK>> KStream<inK, inV>.map(
+//    name: String,
+//    crossinline mapper: (key: inK, value: inV) -> outV
+//  ): KStream<outK, outV> =
+//    map({ k, v -> mapper(k, v).asKeyValue() }, withName(name))
+
+  /** @see KStream.map */
+  inline fun <inK, inV, reified outK, reified outV> KStream<inK, inV>.map(
+    name: String,
+    crossinline mapper: (key: inK, value: inV) -> Pair<outK, outV>
+  ): KStream<outK, outV> =
+    map({ k, v -> mapper(k, v).toKeyValue() }, withName(name))
+
+  /** Wrapper for [KStream.mapValues] */
+  fun <K, inV, outV> KStream<K, inV>.mapValues(
+    name: String,
+    mapper: (key: K, value: inV) -> outV
+  ): KStream<K, outV> =
+    mapValues({ k, v -> mapper(k, v) }, withName(name))
+
+//  /** @see KStream.flatMap */
+//  inline fun <inK, reified outK, inV, reified outV : TopicData<outK>> KStream<inK, inV>.flatMap(
+//    name: String,
+//    crossinline mapper: (key: inK, value: inV) -> Iterable<outV>
+//  ): KStream<outK, outV> = flatMap(
+//    { k, v -> mapper(k, v).map { a: outV -> a.asKeyValue() } },
+//    withName(name)
+//  )
+
+  /** @see KStream.flatMap */
+  inline fun <reified inK, inV, reified outK, reified outV> KStream<inK, inV>.flatMap(
+    name: String,
+    crossinline mapper: (key: inK, value: inV) -> Iterable<Pair<outK, outV>>
+  ): KStream<outK, outV> = flatMap(
+    { k, v -> mapper(k, v).map { it.toKeyValue() } },
+    withName(name)
+  )
+
+  /** @see KStream.groupBy */
+  inline fun <reified inK, outK, reified V> KStream<inK, V>.groupBy(
+    grouped: Grouped<outK, V>,
+    crossinline keySelector: KeyValue<inK, V>.() -> outK
+  ): KGroupedStream<outK, V> =
+    groupBy(
+      { k, v -> (k kvPair v).keySelector() },
+      grouped
+    )
+
+
+
+//  @JvmName("flatMapKeyValuePairs")
+//  inline fun <inK, reified outK, inV, reified outV> KStream<inK, inV>.flatMap(
+//      name: String,
+//      crossinline mapper: (key: inK, value: inV) -> Iterable<Pair<outK, outV>>
+//  ): KStream<outK, outV> =
+//      flatMap(
+//          { k, v -> mapper(k, v).map { it.asKeyPair() } },
+//          withName(name)
+//      )
+
+//  inline fun <reified K, inV, reified outV> KStream<K, inV>.mapValues(
+//      name: String,
+//      crossinline mapper: (readOnlyKey: K, value: inV) -> outV
+//  ): KStream<K, outV> =
+//      map({ k, v -> (k to mapper(v)).asKeyPair() }, KafkaStreamsExtensions.withName(name))
+
+}
+
+
+//open class TransformerKt<inK, inV, outK, outV>(
+//) : Transformer<inK, inV, KeyValue<outK, outV>> {
+//
+//  val onClose: TransformerKt<inK, inV, outK, outV>.() -> Unit = {}
+//  val transformer: TransformerKt<inK, inV, outK, outV>.(key: inK, value: inV) -> KeyValue<outK, outV>? =
+//    {}
+//
+//  private lateinit var _context: ProcessorContext
+//  val context: ProcessorContext?
+//    get() = if (this::_context.isInitialized) _context else null
+//
+//  override fun init(context: ProcessorContext) {
+//    this._context = context
+//  }
+//
+//  override fun close() {
+//    onClose()
+//  }
+//
+//  override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
+//    return transformer(key, value)
+//  }
+//}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamTransform.kt

@@ -0,0 +1,90 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams
+
+import dev.adamko.kafkatorio.extensions.kafkastreams.KeyPairExtensions.toKeyValue
+import kotlin.experimental.ExperimentalTypeInference
+import org.apache.kafka.streams.KeyValue
+import org.apache.kafka.streams.kstream.KStream
+import org.apache.kafka.streams.kstream.Transformer
+import org.apache.kafka.streams.kstream.TransformerSupplier
+import org.apache.kafka.streams.processor.ProcessorContext
+
+/** @see KStream.transform */
+inline fun <inK, inV, reified outK, reified outV> KStream<inK, inV>.transformKt(
+  crossinline block: BuildTransformer<inK, inV, outK, outV>
+): KStream<outK, outV> {
+
+  val transformer = object : Transformer<inK, inV, KeyValue<outK, outV>> {
+
+    private lateinit var scope: TransformerScope<inK, inV, outK, outV>
+
+    override fun init(procContext: ProcessorContext) {
+      scope = TransformerScope(procContext)
+      scope.block()
+    }
+
+    override fun close() {
+      scope.close()
+    }
+
+    override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
+      return scope.mapper(key, value)?.toKeyValue()
+    }
+  }
+
+  return transform({ transformer })
+}
+
+/** @see KStream.transform */
+@OptIn(ExperimentalTypeInference::class)
+inline fun <inK, inV, reified outK, reified outV, outKV : KeyValue<outK, outV>> KStream<inK, inV>.transformKt2(
+  @BuilderInference block: Transformer<inK, inV, outKV>.() -> Unit
+): KStream<outK, outV> {
+
+  val transformer = object : Transformer<inK, inV, KeyValue<outK, outV>> {
+
+    private lateinit var scope: TransformerScope<inK, inV, outK, outV>
+
+    override fun init(procContext: ProcessorContext) {
+      scope = TransformerScope(procContext)
+    }
+
+    override fun close() {
+      scope.close()
+
+      sequence<Long> {
+
+      }
+    }
+
+    override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
+      return scope.mapper(key, value)?.toKeyValue()
+    }
+  }
+
+  return transform(TransformerSupplier { transformer })
+}
+
+typealias BuildTransformer<inK, inV, outK, outV> = TransformerScope<inK, inV, outK, outV>.() -> Unit
+
+typealias TransformerMapper<inK, inV, outK, outV> = (key: inK, value: inV) -> Pair<outK, outV>?
+typealias TransformerCloseHandler = (ProcessorContext) -> Unit
+
+open class TransformerScope<inK, inV, outK, outV>(
+  val processorContext: ProcessorContext,
+) : AutoCloseable {
+
+  private var closeHandler: TransformerCloseHandler = {}
+  lateinit var mapper: TransformerMapper<inK, inV, outK, outV>
+
+  fun handleClose(closeHandler: TransformerCloseHandler) {
+    this.closeHandler = closeHandler
+  }
+
+  fun mapEach(mapper: TransformerMapper<inK, inV, outK, outV>) {
+    this.mapper = mapper
+  }
+
+  override fun close() {
+    closeHandler(processorContext)
+  }
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KTableExtensions.kt

@@ -0,0 +1,19 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams
+
+import dev.adamko.kafkatorio.extensions.kafkastreams.KTableExtensions.mapValues
+import dev.adamko.kafkatorio.extensions.kafkastreams.KafkaStreamsExtensions.withName
+import org.apache.kafka.streams.kstream.KTable
+
+object KTableExtensions {
+
+  /** @see [KTable.mapValues] */
+  fun <K, inV, outV> KTable<K, inV>.mapValues(
+    name: String,
+    mapper: (readOnlyKey: K, value: inV) -> outV
+  ): KTable<K, outV> =
+    mapValues(
+      { k, v -> mapper(k, v) },
+      withName(name)
+    )
+
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KafkaStreamsExtensions.kt

@@ -0,0 +1,93 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams
+
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.streams.kstream.Grouped
+import org.apache.kafka.streams.kstream.Joined
+import org.apache.kafka.streams.kstream.Materialized
+import org.apache.kafka.streams.kstream.Named
+import org.apache.kafka.streams.kstream.Produced
+import org.apache.kafka.streams.kstream.Repartitioned
+import org.apache.kafka.streams.processor.StateStore
+
+/** Helpful extensions & shortcuts for Kafka Streams */
+object KafkaStreamsExtensions {
+
+  /** @see [Named] */
+  fun withName(name: String): Named = Named.`as`(name)
+
+//  @Suppress("FunctionName")
+//  inline fun <reified K, reified V, S : StateStore> MaterializedAs(
+//      tableDefinition: GlobalTableDefinition<K, V>
+//  ) = MaterializedAs<K, V, S>(
+//      tableDefinition.globalStoreName,
+//      tableDefinition.serdes.keySerde,
+//      tableDefinition.serdes.valueSerde
+//  )
+
+  /** @see Materialized */
+  inline fun <reified K, reified V, S : StateStore> materializedAs(
+    name: String,
+    keySerde: Serde<K>,//= kxsJsonSerde(),
+    valueSerde: Serde<V>,//= kxsJsonSerde(),
+  ): Materialized<K, V, S> =
+    Materialized.`as`<K, V, S>(name)
+      .withKeySerde(keySerde)
+      .withValueSerde(valueSerde)
+
+  /** @see Repartitioned */
+  inline fun <reified Key, reified Val> repartitionedAs(
+    name: String,
+    keySerde: Serde<Key>,
+    valueSerde: Serde<Val>,
+  ): Repartitioned<Key, Val> =
+    Repartitioned.`as`<Key, Val>(name)
+      .withKeySerde(keySerde)
+      .withValueSerde(valueSerde)
+
+  /** @see Produced */
+  fun <K, V> producedAs(name: String): Produced<K, V> =
+    Produced.`as`(name)
+
+  /** @see Produced */
+  inline fun <reified Key, reified Val> producedAs(
+    name: String,
+    keySerde: Serde<Key>,
+    valueSerde: Serde<Val>,
+  ): Produced<Key, Val> =
+    producedAs<Key, Val>(name)
+      .withKeySerde(keySerde)
+      .withValueSerde(valueSerde)
+
+//  @Suppress("FunctionName")
+//  fun <Key, Val, V_other> JoinedAs(name: String): Joined<Key, Val, V_other> =
+//      Joined.`as`(name)
+
+  /** @see Joined */
+  inline fun <reified Key, reified Val, reified V_other> joinedAs(
+    name: String,
+    keySerde: Serde<Key>,// = kxsJsonSerde(),
+    valueSerde: Serde<Val>,//= kxsJsonSerde(),
+    otherValueSerde: Serde<V_other>?,//= kxsJsonSerde(),
+  ): Joined<Key, Val, V_other> =
+    Joined.`as`<Key, Val, V_other>(name)
+      .withKeySerde(keySerde)
+      .withValueSerde(valueSerde)
+      .withOtherValueSerde(otherValueSerde)
+
+  /** @see [Grouped.`as`] */
+  inline fun <reified Key, reified Val> groupedAs(
+    name: String,
+    keySerde: Serde<Key>?, //= kxsJsonSerde(),
+    valueSerde: Serde<Val>?,// = kxsJsonSerde(),
+  ): Grouped<Key, Val> =
+    Grouped.`as`<Key, Val>(name)
+      .withKeySerde(keySerde)
+      .withValueSerde(valueSerde)
+
+//  fun <K, V> InteractiveQueryService.getQueryableStore(
+//      tableDef: GlobalTableDefinition<K, V>
+//  ): Optional<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>> {
+//    return getQueryableStore(tableDef.globalStoreName, tableDef.storeType)
+//  }
+
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyPairExtensions.kt

@@ -0,0 +1,35 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams
+
+import org.apache.kafka.streams.KeyValue
+import org.apache.kafka.streams.state.ValueAndTimestamp
+
+object KeyPairExtensions {
+
+  inline fun <reified K, reified V> Pair<K, V>.toKeyValue(): KeyValue<K, V> =
+    KeyValue.pair(first, second)
+
+  infix fun <K, V> K.kvPair(value: V): KeyValue<K, V> = KeyValue(this, value)
+
+  infix fun <V> V.timestamped(timestamp: Long): ValueAndTimestamp<V> =
+    ValueAndTimestamp.make(this, timestamp)
+
+
+  @JvmName("keyValueGetKey")
+  operator fun <K, V> KeyValue<K, V>.component1(): K = key
+  @JvmName("keyValueGetValue")
+  operator fun <K, V> KeyValue<K, V>.component2(): V = value
+
+
+  @JvmName("keyValueTimestampGetKey")
+  operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component1(): K = key
+  @JvmName("keyValueTimestampGetValue")
+  operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component2(): V = value.value()
+  @JvmName("keyValueTimestampGetTimestamp")
+  operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component3(): Long = value.timestamp()
+
+
+  @JvmName("valueTimestampGetValue")
+  operator fun <V> ValueAndTimestamp<V>.component1(): V = value()
+  @JvmName("valueTimestampGetTimestamp")
+  operator fun <V> ValueAndTimestamp<V>.component2(): Long = timestamp()
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyValueSerdes.kt

@@ -0,0 +1,40 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams
+
+import dev.adamko.kafkatorio.extensions.kafkastreams.KotlinxSerializationExtensions.kxsJsonSerde
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.streams.kstream.Consumed
+import org.apache.kafka.streams.kstream.Produced
+import org.apache.kafka.streams.kstream.Repartitioned
+
+/**
+ * A wrapper for two [Serde]s - [one for the Key][keySerde], [another for the Value][valueSerde].
+ */
+open class KeyValueSerdes<K, V>(
+  val keySerde: Serde<K>,
+  val valueSerde: Serde<V>,
+) {
+
+  fun consumer(name: String): Consumed<K, V> =
+    Consumed.with(keySerde, valueSerde).withName(name)
+
+  fun producer(name: String): Produced<K, V> =
+    Produced.with(keySerde, valueSerde).withName(name)
+
+  fun repartitioned(name: String): Repartitioned<K, V> =
+    Repartitioned.with(keySerde, valueSerde).withName(name)
+
+  companion object {
+    inline fun <reified K, reified V> kxsJson(): KeyValueSerdes<K, V> =
+      KeyValueSerdes(
+        kxsJsonSerde(),
+        kxsJsonSerde(),
+      )
+
+    fun string(): KeyValueSerdes<String, String> =
+      KeyValueSerdes<String, String>(
+        Serdes.String(),
+        Serdes.String(),
+      )
+  }
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KotlinxSerializationExtensions.kt

@@ -0,0 +1,55 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams
+
+import kotlinx.serialization.decodeFromString
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
+import org.apache.kafka.common.serialization.Deserializer
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.common.serialization.Serializer
+
+object KotlinxSerializationExtensions {
+
+  inline fun <reified T> Json.kafkaSerializer() = Serializer { topic: String, data: T ->
+    runCatching {
+      encodeToString(data).encodeToByteArray()
+//      jsonMapper.encodeToString(data as T).encodeToByteArray()
+    }.getOrElse { e ->
+      println(
+        """
+            Exception on encodeToString 
+            Topic: $topic
+            topicData: $data
+            topicData as T: ${data as T}
+          """.trimIndent()
+      )
+      e.printStackTrace()
+      throw e
+    }
+  }
+
+  inline fun <reified T> Json.kafkaDeserializer() = Deserializer { topic: String, data: ByteArray ->
+    runCatching {
+      decodeFromString<T>(data.decodeToString())
+    }.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> Json.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/extensions/kafkastreams/TopicData.kt

@@ -0,0 +1,17 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams
+
+import org.apache.kafka.streams.KeyValue
+
+interface TopicData<K> {
+
+  val topicKey: K
+
+  companion object {
+    fun <K, T : TopicData<K>> T.toPair(): Pair<K, T> = topicKey to this
+    fun <K, T : TopicData<K>> T.toKeyValue(): KeyValue<K, T> = KeyValue.pair(topicKey, this)
+
+    operator fun <K, T : TopicData<K>> T.component1(): K = topicKey
+    operator fun <K, T : TopicData<K>> T.component2(): T = this
+
+  }
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicDefinition.kt

@@ -0,0 +1,42 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams;
+
+/**
+ * A definition for all topics that are created in Kafka.
+ *
+ * These can be used to create KTopics, KTables, or GlobalKTables.
+ */
+interface TopicDefinition<K, V> {
+
+  /** unique kafka processor id */
+  val pid: String
+    get() = this::class.simpleName!!
+
+  /**
+   * Unique name for each [TopicDefinition]
+   *
+   * https://devshawn.com/blog/apache-kafka-topic-naming-conventions/
+   *
+   * https://www.xeotek.com/topic-naming-conventions-how-do-i-name-my-topics-5-recommendations-with-examples/
+   *
+   * Domain: warehouse
+   * Operation: stock-management
+   * Data type:
+   *   - cdc=all instances of a specific object/thing/item,
+   *   - cmd=command (mutation),
+   *   - fct=fact (information (complete or partial)
+   * about an event at specific point in time)
+   * - sys=internal system topic - not meant for external use
+   * - tbl=a global table store (must be paired with another topic)
+   * Description: the data type
+   *
+   * Version: stored in message header
+   *
+   * ```
+   * warehouse.stock-management.sts/cdc/cmd.warehouse-item
+   * ```
+   */
+  val topicName: String
+
+  val serdes: KeyValueSerdes<K, V>
+
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/aliases.kt

@@ -0,0 +1,7 @@
+package dev.adamko.kafkatorio.extensions.kafkastreams
+
+import org.apache.kafka.streams.state.QueryableStoreType
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore
+import org.apache.kafka.streams.state.ValueAndTimestamp
+
+typealias TimestampedQueryStoreType<K, V> = QueryableStoreType<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>>

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

@@ -1,17 +1,44 @@
 package dev.adamko.kafkatorio.processor
 
+import com.sksamuel.scrimage.ImmutableImage
+import com.sksamuel.scrimage.nio.PngWriter
+import com.sksamuel.scrimage.pixels.Pixel
+import com.sksamuel.scrimage.pixels.PixelTools
 import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate
 import dev.adamko.kafkatorio.events.schema.FactorioEvent
+import dev.adamko.kafkatorio.events.schema.FactorioObjectData
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
+import dev.adamko.kafkatorio.events.schema.MapChunk
+import dev.adamko.kafkatorio.events.schema.MapChunkPosition
+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.MapTiles
+import dev.adamko.kafkatorio.extensions.kafkastreams.KafkaStreamsExtensions.materializedAs
+import dev.adamko.kafkatorio.extensions.kafkastreams.KeyPairExtensions.toKeyValue
+import dev.adamko.kafkatorio.extensions.kafkastreams.KotlinxSerializationExtensions.serde
+import dev.adamko.kafkatorio.extensions.kafkastreams.TopicData
+import dev.adamko.kafkatorio.extensions.kafkastreams.TopicData.Companion.toKeyValue
+import java.awt.image.BufferedImage
+import java.io.File
 import java.time.Duration
+import kotlin.math.roundToInt
+import kotlinx.serialization.Serializable
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
 import org.apache.kafka.common.serialization.Serdes
 import org.apache.kafka.streams.KafkaStreams
+import org.apache.kafka.streams.StoreQueryParameters
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.TopologyDescription
 import org.apache.kafka.streams.kstream.Consumed
+import org.apache.kafka.streams.kstream.KStream
+import org.apache.kafka.streams.kstream.KTable
 import org.apache.kafka.streams.kstream.Produced
+import org.apache.kafka.streams.state.QueryableStoreTypes
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore
+import org.apache.kafka.streams.state.ValueAndTimestamp
+
 
 class KafkatorioTopology(
   private val websocketServer: WebsocketServer,
@@ -20,36 +47,25 @@ class KafkatorioTopology(
 
   private val sourceTopic = "factorio-server-log"
 
+  private lateinit var tilePrototypesStore: ReadOnlyKeyValueStore<String, ValueAndTimestamp<MapTilePrototype>>
   fun build() {
 
-    builder.stream(
-      sourceTopic,
-      Consumed.with(Serdes.String(), Serdes.String())
-    )
-      .mapValues { readOnlyKey, value ->
-        println("Mapping $readOnlyKey:$value")
-        jsonMapper.decodeFromString<KafkatorioPacket>(value)
-      }
-      .peek { _, value ->
-        websocketServer.sendMessage(jsonMapper.encodeToString(value))
-      }
-      .to(
-        { _, value, _ ->
-//        println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
-          when (value) {
-            is FactorioEvent               ->
-              "kafkatorio.${value.packetType.name}.${value.data.objectName.name}"
-            is FactorioConfigurationUpdate ->
-              "kafkatorio.${value.packetType.name}"
-          }
-        },
-        Produced.with(Serdes.String(), KafkatorioPacketSerde)
-      )
+    parseAndSplitServerLog()
+    sendPlayerUpdatesToWebSocket()
+//    createMapTileImages()
+    val tilePrototypesTable: KTable<String, MapTilePrototype> = tilePrototypesTable()
 
     val topology = builder.build()
 
     val streams = KafkaStreams(topology, appProps.kafkaConfig)
 
+    tilePrototypesStore = streams.store(
+      StoreQueryParameters.fromNameAndType(
+        tilePrototypesTable.queryableStoreName(),
+        QueryableStoreTypes.timestampedKeyValueStore<String, MapTilePrototype>()
+      )
+    )
+
     streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
 
     Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(1)) })
@@ -62,4 +78,272 @@ class KafkatorioTopology(
     streams.start()
   }
 
+  private fun parseAndSplitServerLog() {
+
+    builder.stream(
+      sourceTopic,
+      Consumed.with(Serdes.String(), Serdes.String())
+    )
+      .mapValues { readOnlyKey, value ->
+        println("Mapping $readOnlyKey:$value")
+        jsonMapper.decodeFromString<KafkatorioPacket>(value)
+      }
+      .to(
+        { _, value, _ ->
+//        println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
+          when (value) {
+            is FactorioEvent               ->
+              "kafkatorio.${value.packetType.name}.${value.data.objectName.name}"
+            is FactorioConfigurationUpdate ->
+              "kafkatorio.${value.packetType.name}.FactorioConfigurationUpdate"
+          }
+        },
+        Produced.with(Serdes.String(), KafkatorioPacketSerde)
+      )
+  }
+
+  private fun sendPlayerUpdatesToWebSocket() {
+    builder.stream(
+      "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}",
+      Consumed.with(Serdes.String(), KafkatorioPacketSerde)
+    )
+      .foreach { _, value ->
+        websocketServer.sendMessage(jsonMapper.encodeToString(value))
+      }
+  }
+
+  private fun tilePrototypesTable(): KTable<String, MapTilePrototype> {
+    return builder.stream(
+      "kafkatorio.${KafkatorioPacket.PacketType.CONFIG}.FactorioConfigurationUpdate",
+      Consumed.with(Serdes.String(), jsonMapper.serde<FactorioConfigurationUpdate>())
+    )
+      .flatMap { _, update: FactorioConfigurationUpdate ->
+        update
+          .prototypes
+          .filterIsInstance<MapTilePrototype>()
+          .map { proto -> proto.name to proto }
+          .map { it.toKeyValue() }
+      }
+      .toTable(
+        materializedAs(
+          "kafkatorio.prototypes.map-tiles",
+          jsonMapper.serde(),
+          jsonMapper.serde()
+        )
+      )
+  }
+  
+  private fun mapChunksStream() {
+
+    val luaTilesUpdatesStream: KStream<String, MapTiles> = builder.stream(
+      "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}",
+      Consumed.with(Serdes.String(), jsonMapper.serde<FactorioEvent>())
+    )
+      .filter { _, event: FactorioEvent -> event.data is MapTiles }
+      .mapValues { _, event: FactorioEvent -> (event.data as? MapTiles)!! }
+
+    val chunkTilesUpdateStream: KStream<String, MapTiles> = builder.stream(
+      "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
+      Consumed.with(Serdes.String(), jsonMapper.serde<FactorioEvent>())
+    )
+      .filter { _, event: FactorioEvent -> event.data is MapChunk }
+      .mapValues { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles }
+
+    @Serializable
+    data class TileUpdateRecordKey(
+      val surfaceIndex: Int,
+      val tilePosition: MapTilePosition,
+    )
+
+    @Serializable
+    data class TileUpdateRecord(
+      val surfaceIndex: Int,
+      val tilePosition: MapTilePosition,
+      val tile: MapTile
+    ) : TopicData<TileUpdateRecordKey> {
+      override val topicKey: TileUpdateRecordKey =
+        TileUpdateRecordKey(surfaceIndex, tilePosition)
+    }
+
+    val allTileUpdatesStream: KStream<TileUpdateRecordKey, TileUpdateRecord> =
+      luaTilesUpdatesStream
+        .merge(chunkTilesUpdateStream)
+        .flatMap { _, mapTiles: MapTiles ->
+          mapTiles.tiles.map { tile ->
+            TileUpdateRecord(
+              mapTiles.surfaceIndex,
+              tile.position,
+              tile
+            ).toKeyValue()
+          }
+        }
+
+    @Serializable
+    data class MapChunkDataPosition(
+      val position: MapChunkPosition,
+      val surfaceIndex: Int,
+    )
+
+    @Serializable
+    data class MapChunkData(
+      val position: MapChunkPosition,
+      val surfaceIndex: Int,
+      val tiles: Set<MapTile>,
+    ) : TopicData<MapChunkDataPosition> {
+      override val topicKey = MapChunkDataPosition(position, surfaceIndex)
+
+      constructor(key: MapChunkDataPosition, tiles: Set<MapTile>) :
+          this(key.position, key.surfaceIndex, tiles)
+    }
+
+    val chunksTable =
+      allTileUpdatesStream
+        .groupBy { _: TileUpdateRecordKey, value: TileUpdateRecord ->
+          MapChunkDataPosition(
+            value.tilePosition.toMapChunkPosition(),
+            value.surfaceIndex,
+          )
+        }
+        .aggregate(
+          { setOf() },
+        ) { _: MapChunkDataPosition, v: TileUpdateRecord, tiles: Set<MapTile> ->
+          tiles + v.tile
+        }
+        .mapValues { chunkPosition: MapChunkDataPosition, tiles: Set<MapTile> ->
+          MapChunkData(chunkPosition, tiles)
+        }
+
+    chunksTable.toStream()
+      .foreach { key: MapChunkDataPosition, value: MapChunkData ->
+
+
+        value.tiles
+          .map { tile ->
+
+            val colour = tilePrototypesStore.get(tile.prototypeName)
+              .value()
+              .mapColour
+              .asDecimal()
+
+
+            val pixelColour = PixelTools.argb(
+              colour.alpha.roundToInt(),
+              colour.red.roundToInt(),
+              colour.green.roundToInt(),
+              colour.blue.roundToInt(),
+            )
+
+            Pixel(
+              tile.position.x,
+              tile.position.y,
+              pixelColour
+            )
+          }
+
+        val chunkImg = ImmutableImage.create(32, 32, BufferedImage.TYPE_INT_ARGB)
+
+        val zoom = 1u
+        chunkImg.output(
+          PngWriter.MaxCompression,
+          File("${key.surfaceIndex}/$zoom/${key.position.x}/${key.position.y}.png")
+        )
+      }
+  }
+
+//
+//  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/serdes.kt

@@ -12,7 +12,6 @@ import org.http4k.format.ConfigurableKotlinxSerialization
 import org.http4k.format.asConfigurable
 import org.http4k.format.withStandardMappings
 
-
 object KafkatorioPacketSerde : Serde<KafkatorioPacket> {
   override fun serializer() = Serializer<KafkatorioPacket> { _, message ->
     jsonMapper.encodeToString<KafkatorioPacket>(message).encodeToByteArray()
@@ -23,7 +22,7 @@ object KafkatorioPacketSerde : Serde<KafkatorioPacket> {
   }
 }
 
-val jsonMapper = Json {
+val jsonMapper: Json = Json {
   prettyPrint = true
   prettyPrintIndent = "  "
   serializersModule = jsonMapperKafkatorio.serializersModule

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

@@ -0,0 +1,68 @@
+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 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
+
+class FactorioGameState (
+private val builder: StreamsBuilder = StreamsBuilder(),
+) {
+  private val sourceTopic = "factorio-server-log"
+
+
+  private fun gatherGameState() {
+
+    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()) }
+//      )
+  }
+
+
+}