tidy up shared config, start setting up kafka processor

modules/factorio-events-processor/build.gradle.kts

@@ -1,5 +1,7 @@
+
 plugins {
   id("dev.adamko.factoriowebmap.archetype.kotlin-jvm")
+  application
 }
 
 description =
@@ -12,3 +14,21 @@ description =
 val projectId: String by project.extra
 val buildDir: Directory = layout.buildDirectory.dir(projectId).get()
 
+@Suppress("UnstableApiUsage")
+dependencies {
+
+  implementation("org.apache.kafka:kafka-streams:3.0.0")
+
+  implementation(platform(libs.http4k.bom))
+  implementation(libs.http4k.core)
+  implementation(libs.http4k.cloudnative)
+  implementation(libs.http4k.format.json)
+  implementation(libs.http4k.format.yaml)
+
+  implementation(libs.bundles.logging)
+
+}
+
+application {
+  mainClass.set("dev.adamko.factorioevents.processor.mainKt")
+}

modules/factorio-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/factorio-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/factorio-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.REPLACE_THREAD
+  }
+}

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

@@ -0,0 +1,68 @@
+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.Jackson
+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/factorio-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/factorio-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/factorio-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/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/main.kt

@@ -0,0 +1,19 @@
+package dev.adamko.factorioevents.processor
+
+import dev.adamko.factorioevents.processor.config.ApplicationProperties
+import org.apache.kafka.streams.KafkaStreams
+import org.apache.kafka.streams.StreamsBuilder
+
+fun main() {
+
+  val appProps = ApplicationProperties()
+
+  val topology = StreamsBuilder().build()
+
+  val properties = appProps.kafkaConfig
+
+  val streams = KafkaStreams(topology, properties)
+
+  streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
+
+}