feat(ws): migrate challenge notifications to Redis Streams

Replace pub/sub publish in EventPublisher with XADD to user event stream.
UserWebSocketResource subscribes via XREADGROUP consumer group (per-connection
group, '$' offset). DLQ after maxRetries=3 on delivery failure. Poll loop
uses connection identity to prevent thread leak on reconnect.

Closes NCS-104
https://knockoutwhist.youtrack.cloud/issue/NCS-104

modules/account/src/main/scala/de/nowchess/account/service/EventPublisher.scala

@@ -43,19 +43,26 @@ class EventPublisher:
     val payload = objectMapper.createObjectNode()
     payload.put("challengeId", challengeId)
     payload.put("challengerName", challengerName)
-    publish(s"${redisConfig.prefix}:user:$destUserId:events", EventType.ChallengeCreated, payload)
+    publishToUserStream(destUserId, EventType.ChallengeCreated, payload)
 
   def publishChallengeAccepted(challengerId: String, challengeId: String, gameId: String): Unit =
     val payload = objectMapper.createObjectNode()
     payload.put("challengeId", challengeId)
     payload.put("gameId", gameId)
-    publish(s"${redisConfig.prefix}:user:$challengerId:events", EventType.ChallengeAccepted, payload)
+    publishToUserStream(challengerId, EventType.ChallengeAccepted, payload)
 
-  private def publish(
-      channel: String,
+  private def publishToUserStream(
+      userId: String,
       eventType: EventType,
       payload: com.fasterxml.jackson.databind.node.ObjectNode,
   ): Unit =
     val envelope = EventEnvelope.of(eventType, payload)
-    redis.pubsub(classOf[String]).publish(channel, objectMapper.writeValueAsString(envelope))
+    val json     = objectMapper.writeValueAsString(envelope)
+    redis
+      .stream(classOf[String])
+      .xadd(
+        s"${redisConfig.prefix}:user:$userId:events:stream",
+        new XAddArgs().maxlen(maxStreamLen).nearlyExactTrimming(),
+        Map("data" -> json).asJava,
+      )
     ()

modules/account/src/test/scala/de/nowchess/account/service/EventPublisherTest.scala

@@ -0,0 +1,56 @@
+package de.nowchess.account.service
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
+import de.nowchess.account.config.RedisConfig
+import io.quarkus.redis.datasource.RedisDataSource
+import io.quarkus.redis.datasource.stream.{StreamCommands, XAddArgs}
+import org.junit.jupiter.api.{BeforeEach, Test}
+import org.mockito.ArgumentMatchers.*
+import org.mockito.Mockito.*
+import scala.compiletime.uninitialized
+
+class EventPublisherTest:
+
+  // scalafix:off DisableSyntax.var
+  private var redis: RedisDataSource                              = uninitialized
+  private var streamCmds: StreamCommands[String, String, Nothing] = uninitialized
+  private var redisConfig: RedisConfig                            = uninitialized
+  // scalafix:on DisableSyntax.var
+
+  private val objectMapper = new ObjectMapper().registerModule(new JavaTimeModule())
+
+  @BeforeEach
+  def setup(): Unit =
+    redis = mock(classOf[RedisDataSource])
+    streamCmds = mock(classOf[StreamCommands[String, String, Nothing]])
+    redisConfig = mock(classOf[RedisConfig])
+    when(redis.stream(classOf[String])).thenReturn(streamCmds)
+    when(redisConfig.prefix).thenReturn("nowchess")
+
+  private def publisher: EventPublisher =
+    val p = new EventPublisher
+    p.redis = redis
+    p.redisConfig = redisConfig
+    p.objectMapper = objectMapper
+    p
+
+  @Test
+  def publishChallengeCreatedWritesToUserStream(): Unit =
+    publisher.publishChallengeCreated("user1", "ch1", "Alice")
+    verify(streamCmds).xadd(
+      org.mockito.ArgumentMatchers.eq("nowchess:user:user1:events:stream"),
+      any(classOf[XAddArgs]),
+      any(),
+    )
+    verify(redis, never()).pubsub(any(classOf[Class[?]]))
+
+  @Test
+  def publishChallengeAcceptedWritesToUserStream(): Unit =
+    publisher.publishChallengeAccepted("user2", "ch1", "game42")
+    verify(streamCmds).xadd(
+      org.mockito.ArgumentMatchers.eq("nowchess:user:user2:events:stream"),
+      any(classOf[XAddArgs]),
+      any(),
+    )
+    verify(redis, never()).pubsub(any(classOf[Class[?]]))