project re-organisation, and rename

modules/events-processor/build.gradle.kts

@@ -0,0 +1,44 @@
+plugins {
+  id("dev.adamko.factoriowebmap.archetype.kotlin-jvm")
+  kotlin("plugin.serialization")
+  application
+}
+
+description = """
+    Receive raw Factorio Events from Kafka and process them into targeted topics or data formats. 
+    
+    Factorio events -> Mod -> Kafka -> ***Factorio Events Processor*** -> processed events
+  """.trimIndent()
+
+val projectId: String by project.extra
+
+@Suppress("UnstableApiUsage")
+dependencies {
+
+  implementation("org.apache.kafka:kafka-streams:3.0.0")
+
+  implementation(projects.modules.eventsSchema)
+
+//  implementation("com.ionspin.kotlin:bignum:0.3.3")
+
+  implementation(enforcedPlatform(libs.kotlinx.serialization.bom))
+  implementation(libs.kotlinx.serialization.core)
+  implementation(libs.kotlinx.serialization.json)
+
+  implementation(platform(libs.http4k.bom))
+//  implementation(libs.http4k.core)
+  implementation(libs.http4k.cloudnative)
+//  implementation(libs.http4k.format.json)
+  implementation(libs.http4k.format.yaml) {
+    because("parsing yaml properties files")
+  }
+
+  implementation(libs.http4k.format.kotlinx)
+
+  implementation(libs.bundles.logging)
+
+}
+
+application {
+  mainClass.set("dev.adamko.factorioevents.processor.mainKt")
+}

modules/events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/FactorioEventLogConsumer.kt

@@ -0,0 +1,7 @@
+package dev.adamko.factorioevents.processor
+
+/**
+ * Process the raw messages from a server.
+ */
+class FactorioEventLogConsumer {
+}

modules/events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/FactorioEventsTopology.kt

@@ -0,0 +1,20 @@
+package dev.adamko.factorioevents.processor
+
+import java.util.Properties
+import org.apache.kafka.streams.KafkaStreams
+import org.apache.kafka.streams.StreamsBuilder
+
+class FactorioEventsTopology {
+
+  fun build() {
+
+    val topology = StreamsBuilder().build()
+
+
+    val properties = Properties()
+
+    val kafkaStreams = KafkaStreams(topology, properties)
+
+  }
+
+}

modules/events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/StreamsExceptionHandler.kt

@@ -0,0 +1,11 @@
+package dev.adamko.factorioevents.processor
+
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse
+
+class StreamsExceptionHandler : StreamsUncaughtExceptionHandler {
+  override fun handle(exception: Throwable): StreamThreadExceptionResponse {
+    println("${exception.message} - ${exception.stackTrace}")
+    return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION
+  }
+}

modules/events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/config/ApplicationProperties.kt

@@ -0,0 +1,67 @@
+package dev.adamko.factorioevents.processor.config
+
+import com.fasterxml.jackson.databind.JsonNode
+import dev.adamko.factorioevents.processor.config.PortVal.Companion.portVal
+import java.io.StringReader
+import java.time.Clock
+import java.util.Properties
+import org.http4k.cloudnative.env.Environment
+import org.http4k.cloudnative.env.EnvironmentKey
+import org.http4k.core.Credentials
+import org.http4k.core.Uri
+import org.http4k.format.Json
+import org.http4k.lens.boolean
+import org.http4k.lens.map
+import org.http4k.lens.nonEmptyString
+import org.http4k.lens.uri
+
+/**
+ * Load properties, in descending priority order, from
+ *
+ * 1. Environment variables
+ * 2. This app's env file (as defined by the environment variable [Lenses.APPLICATION_PROPERTIES_FILE_PATH])
+ * # 3. A resources-dir yaml file
+ *
+ */
+class ApplicationProperties(
+  vararg environments: Environment = arrayOf(
+    Environment.env(),
+    Environment.fromEnvVarPath(),
+    Environment.fromResourcesFile(".secret.config.yml")
+  )
+) {
+  private val appProps: Environment =
+    environments.reduce { acc, environment -> acc.overrides(environment) }
+
+  fun clock(): Clock = Clock.systemUTC()
+
+  //<editor-fold desc="Server properties">
+
+  /**
+   * The port of the server
+   *
+   * Set by GCP ([docs](https://cloud.google.com/run/docs/reference/container-contract#env-vars)).
+   */
+  val serverPort: PortVal
+    get() = EnvironmentKey.portVal().defaulted("PORT", PortVal.DEFAULT)[appProps]
+
+  //</editor-fold>
+
+
+  //<editor-fold desc="Kafka properties">
+
+  val kafkaConfig: Properties
+    get() = EnvironmentKey
+      .nonEmptyString()
+      .map {
+        it.reader().use { sr ->
+          Properties().also { p ->
+            p.load(sr)
+          }
+        }
+      }
+      .required("kafka.config")[appProps]
+
+  //</editor-fold>
+
+}

modules/events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/config/Lenses.kt

@@ -0,0 +1,38 @@
+package dev.adamko.factorioevents.processor.config
+
+import java.io.File
+import org.http4k.cloudnative.env.Environment
+import org.http4k.cloudnative.env.EnvironmentKey
+import org.http4k.core.Credentials
+import org.http4k.lens.BiDiLens
+import org.http4k.lens.BiDiMapping
+
+
+internal object Lenses {
+
+  /** The location of this app's env file. */
+  val APPLICATION_PROPERTIES_FILE_PATH: BiDiLens<Environment, File?> =
+    EnvironmentKey
+      .map(
+        { path: String -> File(path) },
+        { file: File -> file.canonicalPath },
+      )
+      .optional("APPLICATION_PROPERTIES_FILE_PATH")
+
+  /** Bidirectional [String] and [Credentials] mapper. */
+  val credentialsLens: BiDiMapping<String, Credentials> = run {
+    val separator = ":"
+    BiDiMapping(
+      // from String to Credentials
+      { property ->
+        property.split(separator, limit = 2)
+          .takeIf { it.size == 2 }
+          ?.let { (user, pass) -> Credentials(user, pass) }
+          ?: throw IllegalArgumentException("Credentials are invalid")
+      },
+      // from Credentials to String
+      { it.user + separator + it.password }
+    )
+  }
+
+}

modules/events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/config/PortVal.kt

@@ -0,0 +1,30 @@
+package dev.adamko.factorioevents.processor.config
+
+import org.http4k.cloudnative.env.Port
+import org.http4k.lens.BiDiLensSpec
+import org.http4k.lens.StringBiDiMappings.int
+import org.http4k.lens.map
+
+/**
+ * Improved (slightly) version of http4k's [Port].
+ */
+@JvmInline
+value class PortVal(val value: Int) {
+
+  init {
+    require(value in VALID_RANGE) {
+      "Out of range. Port: '$value', range: $VALID_RANGE"
+    }
+  }
+
+  operator fun invoke(): Int = value
+
+  companion object {
+    val DEFAULT = PortVal(8080)
+    val VALID_RANGE: IntRange
+      get() = (1..65535)
+
+    fun <IN : Any> BiDiLensSpec<IN, String>.portVal(): BiDiLensSpec<IN, PortVal> =
+      map(int().map(::PortVal, PortVal::value))
+  }
+}

modules/events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/config/environmentExt.kt

@@ -0,0 +1,73 @@
+package dev.adamko.factorioevents.processor.config
+
+import java.io.File
+import java.io.FileNotFoundException
+import org.http4k.cloudnative.env.Environment
+import org.http4k.cloudnative.env.MapEnvironment
+import org.http4k.cloudnative.env.fromYaml
+import org.http4k.lens.Lens
+
+
+/**
+ * Load properties from either a `.yaml` or `.env` file.
+ *
+ * (An improved version of [Environment.fromResource])
+ */
+internal fun Environment.Companion.fromResourcesFile(
+  resource: String,
+  default: Environment = EMPTY
+) =
+  try {
+    val file =
+      this::class.java
+        .getResource("/${resource.removePrefix("/")}")
+        ?.file
+        ?.let { File(it) }
+    fromFile(file, default)
+  } catch (e: FileNotFoundException) {
+    default
+  }
+
+/**
+ * Try fetching a properties file the location provided by the
+ * [`JVM`][jvmProperties] or [`System`][env] environment variable
+ * [Lenses.APPLICATION_PROPERTIES_FILE_PATH], if it exists.
+ *
+ * If it does not exist, return [default].
+ */
+internal fun Environment.Companion.fromEnvVarPath(
+  location: Lens<Environment, File?> = Lenses.APPLICATION_PROPERTIES_FILE_PATH,
+  default: Environment = EMPTY,
+): Environment {
+  val file = (jvmProperties() overrides env())[location]
+
+  return fromFile(file, default)
+}
+
+/**
+ * Load properties from either a `.yaml` or `.env` file.
+ *
+ * (An improved version of [Environment.from])
+ */
+internal fun Environment.Companion.fromFile(
+  file: File?,
+  default: Environment = EMPTY,
+): Environment {
+  return file
+    ?.takeIf { it.isFile }
+    .let {
+      when (it?.extension) {
+        "yml", "yaml" -> fromYaml(it)
+        "env"         -> from(it)
+        else          -> default
+      }
+    }
+}
+
+/** The same as [Environment.ENV], but refreshes on each invocation (which is useful for testing). */
+internal fun Environment.Companion.env(): Environment =
+  MapEnvironment.from(System.getenv().toProperties())
+
+/** The same as [Environment.JVM_PROPERTIES], but refreshes on each invocation (which is useful for testing). */
+internal fun Environment.Companion.jvmProperties(): Environment =
+  MapEnvironment.from(System.getProperties())

modules/events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/main.kt

@@ -0,0 +1,50 @@
+package dev.adamko.factorioevents.processor
+
+import dev.adamko.factorioevents.model.FactorioEvent
+import dev.adamko.factorioevents.model.jsonMapper
+import dev.adamko.factorioevents.processor.config.ApplicationProperties
+import java.time.Duration
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.streams.KafkaStreams
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.TopologyDescription
+import org.apache.kafka.streams.kstream.Consumed
+import org.apache.kafka.streams.processor.TopicNameExtractor
+import org.http4k.format.KotlinxSerialization
+
+fun main() {
+
+  val appProps = ApplicationProperties()
+
+  val builder = StreamsBuilder()
+
+  builder.stream(
+    "factorio-server-log",
+    Consumed.with(Serdes.String(), Serdes.String())
+  )
+    .mapValues { readOnlyKey, value ->
+      println("Mapping $readOnlyKey:$value")
+//      jsonMapper.decodeFromString<FactorioEvent<*>>(value)
+      KotlinxSerialization.asA<FactorioEvent<*>>(value)
+    }
+    .to(TopicNameExtractor { key, value, recordContext ->
+      value.data.objectName
+    })
+
+  val topology = builder.build()
+  val properties = appProps.kafkaConfig
+
+  val streams = KafkaStreams(topology, properties)
+
+  streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
+
+  Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(1)) })
+
+//  streams.cleanUp()
+
+  val description: TopologyDescription = topology.describe()
+  println(description)
+
+  streams.start()
+
+}

modules/events-processor/src/main/resources/logback.xml

@@ -0,0 +1,21 @@
+<configuration>
+
+  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+    <withJansi>true</withJansi>
+    <!-- encoders are assigned the type
+         ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+    <encoder>
+      <pattern>%cyan(%d{HH:mm:ss.SSS}) %gray([%thread]) %highlight(%-5level) %magenta(%logger{36}) - %msg%n
+      </pattern>
+    </encoder>
+  </appender>
+
+  <logger name="org.apache.kafka" level="info" additivity="false">
+    <appender-ref ref="CONSOLE"/>
+  </logger>
+
+  <root level="info">
+    <appender-ref ref="STDOUT"/>
+  </root>
+
+</configuration>