feat(analytics): feed Lichess PGN dumps into Spark batch jobs

Add GameSource: normalises game records into a shared schema and
selects backend via NOWCHESS_PGN_PATH. Unset = PostgreSQL game_records
(unchanged); set = a Lichess PGN dump (file or http(s) URL).

- Parse Lichess PGN with Spark SQL string functions only (no UDFs).
- URLs fetched once via SparkContext.addFile, distributed to executors.
- .pgn.zst decompressed in-process via zstd-jni, plain .pgn redistributed.
- All four batch jobs read through GameSource and skip JDBC write-back
  in PGN mode (Parquet/CSV output only).

Enables driving the analytics demo straight from
https://database.lichess.org standard dumps.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

modules/analytics/CHANGELOG.md

@@ -10,16 +10,3 @@
 ### Bug Fixes
 
 * **analytics:** upgrade Spark to 4.0.3 — 3.5.x has no official Docker image ([46af115](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/46af1154de34a8596cb6cb28c6fad7aba90f597c))
-##  (2026-06-19)
-
-### Features
-
-* **analytics:** add Dockerfile, CI workflow, and stable jar name for K8s deployment ([95215b6](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/95215b6a420fd526df1aa395f9b087556c8ad03b))
-* **analytics:** add PostgreSQL JDBC write-back to all four batch jobs ([0e0ea4c](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/0e0ea4c9893c6efed52e633e55d05ab3ed004502))
-* **analytics:** add Spark batch analytics module ([259b3bb](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/259b3bbb24c0f23326269b93f4b3c84012f727cd))
-* **analytics:** add Structured Streaming, MLlib clustering, GraphX jobs ([e1d80b9](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/e1d80b9331666feea191b1fd08aa762f3581c918))
-* **official-bots:** park expert bot on tournament server at startup ([#76](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/76)) ([751a58b](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/751a58b6061f7434115e229a7661894c76768bc2))
-
-### Bug Fixes
-
-* **analytics:** upgrade Spark to 4.0.3 — 3.5.x has no official Docker image ([46af115](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/46af1154de34a8596cb6cb28c6fad7aba90f597c))

modules/analytics/versions.env

@@ -1,3 +1,3 @@
 MAJOR=0
-MINOR=3
+MINOR=2
 PATCH=0

modules/official-bots/CHANGELOG.md

@@ -310,33 +310,3 @@
 ### Reverts
 
 * Revert "refactor: update metrics paths formatting in application.yml for clarity" ([3870566](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/38705663498d5f47c40dafe2f26198589ede8656))
-##  (2026-06-19)
-
-### Features
-
-* add initialization metrics for various services ([d438e97](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/d438e97f32bdde0bfc63c1b4a8cc810cdd093166))
-* add OpenTelemetry trace configuration with parentbased sampler ([3904d5a](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/3904d5ad8ad4930ddee65287a7bfab785a6148f5))
-* **analytics:** add Spark batch analytics module ([#70](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/70)) ([39f1657](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/39f1657e1db6e84889af338c43be8cb5c03c3ec3))
-* **config:** update application.yml for PostgreSQL and remove staging/production configurations ([2404e61](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/2404e6164c3b50ffccbea5238d636060d6abe4d6))
-* **config:** update application.yml for staging and production environments ([6113432](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/6113432a14c476a3a0dfc0d449e17d023697f2ba))
-* configure logging and add OpenTelemetry support ([#49](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/49)) ([d57c488](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/d57c4886612d1d92da0e1b79209fc83e6ef537a1))
-* **docker:** add .dockerignore and .gitignore files for build exclusions ([c987d8e](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/c987d8e258c0e6c4cfbdaa8381c64c410d7a2b83))
-* **docker:** add Dockerfiles for building Quarkus application in native and JVM modes ([3f2d2bb](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/3f2d2bb4c97fa8cddba66e1da4427c54236dfeed))
-* **docker:** add Dockerfiles for Quarkus application in JVM and native modes ([34b9933](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/34b993304670cf2aa62cd2f6460cee7b9864b08e))
-* **events:** migrate game-creation and bot flows to Redis Streams NCS-89 ([#62](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/62)) ([a24924c](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/a24924c23057db3d700a75dbc4333557789cd991))
-* NCS-78 Add Traceability to the Applications ([#46](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/46)) ([649566e](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/649566eb3fcf38f91c8896a739f74ea318af312d))
-* NCS-78 Add Traceability to the Applications ([#47](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/47)) ([87dfc6c](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/87dfc6c2bcce7f7d58fc641bd8d468a2e584c108))
-* NCS-82 add Swiss-system tournament module ([#55](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/55)) ([c5661de](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/c5661de4a0ebf4b33211f5a391840dcf744656b7))
-* **official-bots:** consume GameOver stream for bot cleanup ([#67](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/67)) ([db9d153](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/db9d1533912f4b41c4d1ca80ccffdde5d23d6ff6))
-* **official-bots:** park expert bot on tournament server at startup ([#75](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/75)) ([30295a4](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/30295a4bb95855ee8261c92278bb9ebc80ee12ee))
-* true-microservices ([#40](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/40)) ([5909242](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/590924254e8a2754de661a57a03e43f89ceb6299))
-
-### Bug Fixes
-
-* enable official bots to connect to external tournament server ([#71](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/71)) ([688d30e](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/688d30e2b10026923372be5fca3c63eaaee2de2a))
-* **official-bots:** configure JWT verification ([#72](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/72)) ([98c64fc](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/98c64fc0d56dc542beb31c75f4b9056d91de03cd))
-* **official-bots:** NCS-70-auto-register official bots with account service ([#59](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/59)) ([7117a93](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/7117a93376272094d0b1a6abf2121254ce396684))
-
-### Reverts
-
-* Revert "refactor: update metrics paths formatting in application.yml for clarity" ([3870566](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/38705663498d5f47c40dafe2f26198589ede8656))

modules/official-bots/versions.env

@@ -1,3 +1,3 @@
 MAJOR=0
-MINOR=20
+MINOR=19
 PATCH=0

modules/ws/CHANGELOG.md

@@ -259,32 +259,3 @@
 ### Reverts
 
 * Revert "refactor: update metrics paths formatting in application.yml for clarity" ([3870566](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/38705663498d5f47c40dafe2f26198589ede8656))
-##  (2026-06-19)
-
-### Features
-
-* add initialization metrics for various services ([d438e97](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/d438e97f32bdde0bfc63c1b4a8cc810cdd093166))
-* add OpenTelemetry trace configuration with parentbased sampler ([3904d5a](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/3904d5ad8ad4930ddee65287a7bfab785a6148f5))
-* **config:** update application.yml for PostgreSQL and remove staging/production configurations ([2404e61](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/2404e6164c3b50ffccbea5238d636060d6abe4d6))
-* **config:** update application.yml for staging and production environments ([6113432](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/6113432a14c476a3a0dfc0d449e17d023697f2ba))
-* configure logging and add OpenTelemetry support ([#49](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/49)) ([d57c488](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/d57c4886612d1d92da0e1b79209fc83e6ef537a1))
-* **docker:** add .dockerignore and .gitignore files for build exclusions ([c987d8e](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/c987d8e258c0e6c4cfbdaa8381c64c410d7a2b83))
-* **docker:** add Dockerfiles for building Quarkus application in native and JVM modes ([3f2d2bb](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/3f2d2bb4c97fa8cddba66e1da4427c54236dfeed))
-* **docker:** add Dockerfiles for Quarkus application in JVM and native modes ([34b9933](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/34b993304670cf2aa62cd2f6460cee7b9864b08e))
-* **docker:** add Dockerfiles for Quarkus application in JVM and native modes ([e5fe7d0](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/e5fe7d07a58e018151bb24f4ee37c06e72608ded))
-* **logging:** add DEBUG/INFO/WARN logging across services (NCS-72) ([#41](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/41)) ([804a4bf](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/804a4bf179e3dfb19e2be4390e7e543caf5237c6))
-* NCS-78 Add Traceability to the Applications ([#46](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/46)) ([649566e](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/649566eb3fcf38f91c8896a739f74ea318af312d))
-* NCS-78 Add Traceability to the Applications ([#47](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/47)) ([87dfc6c](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/87dfc6c2bcce7f7d58fc641bd8d468a2e584c108))
-* **reflection:** add native reflection configuration for tournament classes ([65bc6a7](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/65bc6a759937543df2d29905688bfa9e68d0c9d4))
-* true-microservices ([#40](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/40)) ([5909242](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/590924254e8a2754de661a57a03e43f89ceb6299))
-* update application.yml with new API root paths and add Micrometer and OpenTelemetry dependencies ([72ce262](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/72ce262bc491f94297700e6002fb5d0812e2cc2a))
-* **ws:** migrate challenge notifications to Redis Streams ([#66](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/66)) ([55f102c](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/55f102cbaa684be94a158b16aaa42a50b36afaf3))
-
-### Bug Fixes
-
-* NCS-122 authenticate WebSocket connections via first-message auth ([#73](https://git.janis-eccarius.de/NowChess/NowChessSystems/issues/73)) ([343e2bd](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/343e2bdd100649a96d96da8a6d98caad6de4ad14))
-* remove unused HTTP root-path configurations from application.yml ([3ed3e59](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/3ed3e59ee456d54cd3d65ece4f36623e256b9736))
-
-### Reverts
-
-* Revert "refactor: update metrics paths formatting in application.yml for clarity" ([3870566](https://git.janis-eccarius.de/NowChess/NowChessSystems/commit/38705663498d5f47c40dafe2f26198589ede8656))

modules/ws/src/main/scala/de/nowchess/ws/resource/GameWebSocketResource.scala

@@ -3,6 +3,7 @@ package de.nowchess.ws.resource
 import de.nowchess.ws.config.RedisConfig
 import io.micrometer.core.instrument.{Counter, Gauge, MeterRegistry}
 import io.quarkus.redis.datasource.RedisDataSource
+import io.quarkus.redis.datasource.pubsub.PubSubCommands
 import io.quarkus.websockets.next.*
 import io.smallrye.jwt.auth.principal.JWTParser
 import jakarta.annotation.PostConstruct
@@ -33,7 +34,6 @@ class GameWebSocketResource:
   // scalafix:on DisableSyntax.var
 
   private val connections = new ConcurrentHashMap[String, ConnectionMeta]()
-  private val pendingAuth = ConcurrentHashMap.newKeySet[String]()
 
   @PostConstruct
   def initializeMetrics(): Unit = {
@@ -64,60 +64,40 @@ class GameWebSocketResource:
     s"${redisConfig.prefix}:game:$gameId:c2s"
 
   @OnOpen
-  def onOpen(connection: WebSocketConnection): Unit =
+  def onOpen(connection: WebSocketConnection, handshake: HandshakeRequest): Unit =
     activeGauge
-    val gameId                    = connection.pathParam("gameId")
+    val gameId   = connection.pathParam("gameId")
+    val playerId = resolvePlayerId(handshake)
+    log.infof("Game WebSocket opened — gameId=%s playerId=%s", gameId, playerId.getOrElse("anonymous"))
     val handler: Consumer[String] = msg => connection.sendText(msg).subscribe().`with`(_ => (), _ => ())
     val subscriber                = redis.pubsub(classOf[String]).subscribe(s2cTopic(gameId), handler)
-    connections.put(connection.id(), ConnectionMeta(gameId, subscriber, None))
+    connections.put(connection.id(), ConnectionMeta(gameId, subscriber, playerId))
     connectionsOpened.increment()
-    pendingAuth.add(connection.id())
+    publishConnected(gameId, playerId)
-    log.infof("Game WebSocket opened — gameId=%s connId=%s awaiting auth", gameId, connection.id())
 
   @OnTextMessage
   def onTextMessage(connection: WebSocketConnection, message: String): Unit =
     messagesReceived.increment()
-    if pendingAuth.remove(connection.id()) then
+    Option(connections.get(connection.id())).foreach { meta =>
-      val playerIdOpt =
+      val enriched = meta.playerId match
-        parseAuthToken(message)
+        case Some(pid) => injectPlayerId(message, pid)
-          .flatMap(token => Try(jwtParser.parse(token)).toOption)
+        case None      => message
-          .map(_.getSubject)
+      redis.pubsub(classOf[String]).publish(c2sTopic(meta.gameId), enriched)
-      playerIdOpt match
+    }
-        case None =>
-          log.warnf("Game WebSocket auth failed — closing connId=%s", connection.id())
-          connection.close().subscribe().`with`(_ => (), _ => ())
-        case Some(playerId) =>
-          Option(connections.get(connection.id())).foreach { meta =>
-            connections.put(connection.id(), meta.copy(playerId = Some(playerId)))
-            publishConnected(meta.gameId, Some(playerId))
-          }
-    else
-      Option(connections.get(connection.id())).foreach { meta =>
-        val enriched = meta.playerId match
-          case Some(pid) => injectPlayerId(message, pid)
-          case None      => message
-        redis.pubsub(classOf[String]).publish(c2sTopic(meta.gameId), enriched)
-      }
 
   @OnClose
   def onClose(connection: WebSocketConnection): Unit =
-    pendingAuth.remove(connection.id())
     Option(connections.remove(connection.id())).foreach { meta =>
       log.infof("Game WebSocket closed — gameId=%s", meta.gameId)
       meta.subscriber.unsubscribe(s2cTopic(meta.gameId))
       connectionsClosed.increment()
     }
 
-  private def parseAuthToken(message: String): Option[String] =
+  private def resolvePlayerId(handshake: HandshakeRequest): Option[String] =
-    val trimmed = message.trim
+    Option(handshake.header("Authorization"))
-    if !trimmed.contains("\"type\":\"auth\"") then None
+      .filter(_.nonEmpty)
-    else
+      .flatMap(token => Try(jwtParser.parse(token)).toOption)
-      val start = trimmed.indexOf("\"token\":\"")
+      .map(_.getSubject)
-      if start < 0 then None
-      else
-        val valueStart = start + 9
-        val end        = trimmed.indexOf('"', valueStart)
-        if end < 0 then None else Some(trimmed.substring(valueStart, end)).filter(_.nonEmpty)
 
   private def publishConnected(gameId: String, playerId: Option[String]): Unit =
     val connectedMsg = playerId match

modules/ws/src/main/scala/de/nowchess/ws/resource/UserWebSocketResource.scala

@@ -32,7 +32,6 @@ class UserWebSocketResource:
   private val maxStreamLen = 1000L
 
   private val connections = new ConcurrentHashMap[String, (String, WebSocketConnection)]()
-  private val pendingAuth = ConcurrentHashMap.newKeySet[String]()
 
   private def userStreamKey(userId: String): String =
     s"${redisConfig.prefix}:user:$userId:events:stream"
@@ -40,34 +39,29 @@ class UserWebSocketResource:
   private def dlqKey: String = s"${redisConfig.prefix}:dlq"
 
   @OnOpen
-  def onOpen(connection: WebSocketConnection): Unit =
+  def onOpen(connection: WebSocketConnection, handshake: HandshakeRequest): Unit =
-    pendingAuth.add(connection.id())
+    val userIdOpt = Option(handshake.header("Authorization"))
+      .filter(_.nonEmpty)
+      .flatMap(token => Try(jwtParser.parse(token)).toOption)
+      .map(_.getSubject)
 
-  @OnTextMessage
+    userIdOpt match
-  def onTextMessage(connection: WebSocketConnection, message: String): Unit =
+      case None =>
-    if pendingAuth.remove(connection.id()) then
+        log.warn("WebSocket opened with no valid JWT — closing connection")
-      val userIdOpt =
+        connection.close().subscribe().`with`(_ => (), _ => ())
-        parseAuthToken(message)
+      case Some(userId) =>
-          .flatMap(token => Try(jwtParser.parse(token)).toOption)
+        log.infof("User WebSocket opened — userId=%s connId=%s", userId, connection.id())
-          .map(_.getSubject)
+        createGroupIfAbsent(userId, connection.id())
-      userIdOpt match
+        connections.put(connection.id(), (userId, connection))
-        case None =>
+        executor.submit(
-          log.warn("WebSocket opened with no valid JWT — closing connection")
+          new Runnable:
-          connection.close().subscribe().`with`(_ => (), _ => ())
+            def run(): Unit = pollLoop(connection.id(), userId, connection),
-        case Some(userId) =>
+        )
-          log.infof("User WebSocket opened — userId=%s connId=%s", userId, connection.id())
+        val connectedMsg = s"""{"type":"CONNECTED","userId":"$userId"}"""
-          createGroupIfAbsent(userId, connection.id())
+        connection.sendText(connectedMsg).subscribe().`with`(_ => (), _ => ())
-          connections.put(connection.id(), (userId, connection))
-          executor.submit(
-            new Runnable:
-              def run(): Unit = pollLoop(connection.id(), userId, connection),
-          )
-          val connectedMsg = s"""{"type":"CONNECTED","userId":"$userId"}"""
-          connection.sendText(connectedMsg).subscribe().`with`(_ => (), _ => ())
 
   @OnClose
   def onClose(connection: WebSocketConnection): Unit =
-    pendingAuth.remove(connection.id())
     log.infof("User WebSocket closed — connectionId=%s", connection.id())
     val userIdOpt = Option(connections.remove(connection.id())).map(_._1)
     userIdOpt.foreach { userId =>
@@ -134,14 +128,3 @@ class UserWebSocketResource:
     ) match
       case Failure(ex) => log.warnf(ex, "Failed to publish to stream %s", key)
       case Success(_)  => ()
-
-  private def parseAuthToken(message: String): Option[String] =
-    val trimmed = message.trim
-    if !trimmed.contains("\"type\":\"auth\"") then None
-    else
-      val start = trimmed.indexOf("\"token\":\"")
-      if start < 0 then None
-      else
-        val valueStart = start + 9
-        val end        = trimmed.indexOf('"', valueStart)
-        if end < 0 then None else Some(trimmed.substring(valueStart, end)).filter(_.nonEmpty)

modules/ws/versions.env

@@ -1,3 +1,3 @@
 MAJOR=0
-MINOR=17
+MINOR=16
 PATCH=0