feat(logging): add DEBUG/INFO/WARN logging across services (NCS-72) (#41)
Build & Test (NowChessSystems) TeamCity build finished

Reviewed-on: #41
Co-authored-by: Janis <janis.e.20@gmx.de>
Co-committed-by: Janis <janis.e.20@gmx.de>
This commit was merged in pull request #41.
This commit is contained in:
2026-05-02 17:33:27 +02:00
committed by Janis
parent 3c47d2b8c9
commit 804a4bf179
17 changed files with 321 additions and 107 deletions
@@ -9,6 +9,7 @@ import io.smallrye.jwt.build.Jwt
import jakarta.enterprise.context.ApplicationScoped import jakarta.enterprise.context.ApplicationScoped
import jakarta.inject.Inject import jakarta.inject.Inject
import jakarta.transaction.Transactional import jakarta.transaction.Transactional
import org.jboss.logging.Logger
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
import java.time.Instant import java.time.Instant
@@ -17,6 +18,8 @@ import java.util.UUID
@ApplicationScoped @ApplicationScoped
class AccountService: class AccountService:
private val log = Logger.getLogger(classOf[AccountService])
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
@Inject @Inject
var userAccountRepository: UserAccountRepository = uninitialized var userAccountRepository: UserAccountRepository = uninitialized
@@ -30,6 +33,7 @@ class AccountService:
@Transactional @Transactional
def register(req: RegisterRequest): Either[AccountError, UserAccount] = def register(req: RegisterRequest): Either[AccountError, UserAccount] =
log.infof("Registering user %s", req.username)
if userAccountRepository.findByUsername(req.username).isDefined then Left(AccountError.UsernameTaken(req.username)) if userAccountRepository.findByUsername(req.username).isDefined then Left(AccountError.UsernameTaken(req.username))
else if userAccountRepository.findByEmail(req.email).isDefined then else if userAccountRepository.findByEmail(req.email).isDefined then
Left(AccountError.EmailAlreadyRegistered(req.email)) Left(AccountError.EmailAlreadyRegistered(req.email))
@@ -40,15 +44,23 @@ class AccountService:
account.passwordHash = BcryptUtil.bcryptHash(req.password) account.passwordHash = BcryptUtil.bcryptHash(req.password)
account.createdAt = Instant.now() account.createdAt = Instant.now()
userAccountRepository.persist(account) userAccountRepository.persist(account)
log.infof("User %s registered successfully", req.username)
Right(account) Right(account)
def login(req: LoginRequest): Either[AccountError, String] = def login(req: LoginRequest): Either[AccountError, String] =
userAccountRepository.findByUsername(req.username) match userAccountRepository.findByUsername(req.username) match
case None => Left(AccountError.InvalidCredentials) case None =>
log.warnf("Login failed for unknown user %s", req.username)
Left(AccountError.InvalidCredentials)
case Some(account) => case Some(account) =>
if !BcryptUtil.matches(req.password, account.passwordHash) then Left(AccountError.InvalidCredentials) if !BcryptUtil.matches(req.password, account.passwordHash) then
else if account.banned then Left(AccountError.UserBanned) log.warnf("Login failed — invalid credentials for %s", req.username)
Left(AccountError.InvalidCredentials)
else if account.banned then
log.warnf("Login rejected — user %s is banned", req.username)
Left(AccountError.UserBanned)
else else
log.infof("User %s logged in successfully", req.username)
Right( Right(
Jwt Jwt
.issuer("nowchess") .issuer("nowchess")
@@ -65,6 +77,7 @@ class AccountService:
@Transactional @Transactional
def createBotAccount(ownerId: UUID, botName: String): Either[AccountError, BotAccount] = def createBotAccount(ownerId: UUID, botName: String): Either[AccountError, BotAccount] =
log.infof("Creating bot account %s for owner %s", botName, ownerId.toString)
userAccountRepository.findById(ownerId) match userAccountRepository.findById(ownerId) match
case None => Left(AccountError.UserNotFound) case None => Left(AccountError.UserNotFound)
case Some(owner) => case Some(owner) =>
@@ -77,6 +90,7 @@ class AccountService:
bot.token = generateBotToken(bot.id) bot.token = generateBotToken(bot.id)
bot.createdAt = Instant.now() bot.createdAt = Instant.now()
botAccountRepository.persist(bot) botAccountRepository.persist(bot)
log.infof("Bot account %s created for owner %s", botName, ownerId.toString)
Right(bot) Right(bot)
def getBotAccounts(ownerId: UUID): List[BotAccount] = def getBotAccounts(ownerId: UUID): List[BotAccount] =
@@ -93,6 +107,7 @@ class AccountService:
case None => Left(AccountError.BotNotFound) case None => Left(AccountError.BotNotFound)
case Some(_) => case Some(_) =>
botAccountRepository.delete(botId) botAccountRepository.delete(botId)
log.infof("Deleting bot account %s", botId.toString)
Right(()) Right(())
@Transactional @Transactional
@@ -146,6 +161,7 @@ class AccountService:
@Transactional @Transactional
def banUser(userId: UUID): Either[AccountError, UserAccount] = def banUser(userId: UUID): Either[AccountError, UserAccount] =
log.infof("Banning user %s", userId.toString)
userAccountRepository.findById(userId) match userAccountRepository.findById(userId) match
case None => Left(AccountError.UserNotFound) case None => Left(AccountError.UserNotFound)
case Some(user) => case Some(user) =>
@@ -156,6 +172,7 @@ class AccountService:
@Transactional @Transactional
def unbanUser(userId: UUID): Either[AccountError, UserAccount] = def unbanUser(userId: UUID): Either[AccountError, UserAccount] =
log.infof("Unbanning user %s", userId.toString)
userAccountRepository.findById(userId) match userAccountRepository.findById(userId) match
case None => Left(AccountError.UserNotFound) case None => Left(AccountError.UserNotFound)
case Some(user) => case Some(user) =>
@@ -6,6 +6,7 @@ import io.quarkus.redis.datasource.pubsub.PubSubCommands
import io.smallrye.mutiny.subscription.MultiEmitter import io.smallrye.mutiny.subscription.MultiEmitter
import jakarta.enterprise.context.ApplicationScoped import jakarta.enterprise.context.ApplicationScoped
import jakarta.inject.Inject import jakarta.inject.Inject
import org.jboss.logging.Logger
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
import java.util.function.Consumer import java.util.function.Consumer
@@ -13,6 +14,8 @@ import java.util.function.Consumer
@ApplicationScoped @ApplicationScoped
class BotRegistry: class BotRegistry:
private val log = Logger.getLogger(classOf[BotRegistry])
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
@Inject var redis: RedisDataSource = uninitialized @Inject var redis: RedisDataSource = uninitialized
@Inject var redisConfig: RedisConfig = uninitialized @Inject var redisConfig: RedisConfig = uninitialized
@@ -25,14 +28,17 @@ class BotRegistry:
val handler: Consumer[String] = msg => emitter.emit(msg) val handler: Consumer[String] = msg => emitter.emit(msg)
val subscriber = redis.pubsub(classOf[String]).subscribe(channel, handler) val subscriber = redis.pubsub(classOf[String]).subscribe(channel, handler)
connections.put(botId, (emitter, subscriber)) connections.put(botId, (emitter, subscriber))
log.infof("Bot %s registered", botId)
() ()
def unregister(botId: String): Unit = def unregister(botId: String): Unit =
Option(connections.remove(botId)).foreach { (_, subscriber) => Option(connections.remove(botId)).foreach { (_, subscriber) =>
subscriber.unsubscribe(s"${redisConfig.prefix}:bot:$botId:events") subscriber.unsubscribe(s"${redisConfig.prefix}:bot:$botId:events")
} }
log.infof("Bot %s unregistered", botId)
def dispatch(botId: String, event: String): Unit = def dispatch(botId: String, event: String): Unit =
log.debugf("Dispatching event to bot %s", botId)
redis.pubsub(classOf[String]).publish(s"${redisConfig.prefix}:bot:$botId:events", event) redis.pubsub(classOf[String]).publish(s"${redisConfig.prefix}:bot:$botId:events", event)
() ()
@@ -10,6 +10,7 @@ import jakarta.inject.Inject
import jakarta.ws.rs.* import jakarta.ws.rs.*
import jakarta.ws.rs.core.{MediaType, Response} import jakarta.ws.rs.core.{MediaType, Response}
import org.eclipse.microprofile.jwt.JsonWebToken import org.eclipse.microprofile.jwt.JsonWebToken
import org.jboss.logging.Logger
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
import java.util.function.Consumer import java.util.function.Consumer
@@ -18,6 +19,8 @@ import java.util.function.Consumer
@RolesAllowed(Array("**")) @RolesAllowed(Array("**"))
class BotEventResource: class BotEventResource:
private val log = Logger.getLogger(classOf[BotEventResource])
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
@Inject var registry: BotRegistry = uninitialized @Inject var registry: BotRegistry = uninitialized
@Inject var jwt: JsonWebToken = uninitialized @Inject var jwt: JsonWebToken = uninitialized
@@ -32,8 +35,10 @@ class BotEventResource:
val tokenType = Option(jwt.getClaim[AnyRef]("type")).map(_.toString).getOrElse("") val tokenType = Option(jwt.getClaim[AnyRef]("type")).map(_.toString).getOrElse("")
val subject = Option(jwt.getSubject).getOrElse("") val subject = Option(jwt.getSubject).getOrElse("")
if tokenType != "bot" || subject != botId then if tokenType != "bot" || subject != botId then
log.warnf("Unauthorized bot stream access — tokenType=%s subject=%s botId=%s", tokenType, subject, botId)
Multi.createFrom().failure(new ForbiddenException("Not authorized for this bot")) Multi.createFrom().failure(new ForbiddenException("Not authorized for this bot"))
else else
log.infof("Bot %s connected to event stream", botId)
Multi.createFrom().emitter[String] { emitter => Multi.createFrom().emitter[String] { emitter =>
registry.register(botId, emitter) registry.register(botId, emitter)
emitter.onTermination(() => registry.unregister(botId)) emitter.onTermination(() => registry.unregister(botId))
@@ -58,6 +63,7 @@ class BotEventResource:
@PathParam("uci") uci: String, @PathParam("uci") uci: String,
): Response = ): Response =
val playerId = Option(jwt.getSubject).getOrElse("") val playerId = Option(jwt.getSubject).getOrElse("")
val moveMsg = s"""{"type":"MOVE","uci":"$uci","playerId":"$playerId"}""" log.debugf("Bot move %s in game %s by player %s", uci, gameId, playerId)
val moveMsg = s"""{"type":"MOVE","uci":"$uci","playerId":"$playerId"}"""
redis.pubsub(classOf[String]).publish(s"${redisConfig.prefix}:game:$gameId:c2s", moveMsg) redis.pubsub(classOf[String]).publish(s"${redisConfig.prefix}:game:$gameId:c2s", moveMsg)
Response.ok().build() Response.ok().build()
@@ -9,7 +9,6 @@ quarkus:
server: server:
port: 9086 port: 9086
rest-client: rest-client:
connection-timeout: 5000
read-timeout: 10000 read-timeout: 10000
smallrye-openapi: smallrye-openapi:
info-title: NowChess Coordinator Service info-title: NowChess Coordinator Service
@@ -27,13 +27,24 @@ class CoordinatorGrpcServer extends CoordinatorServiceGrpc.CoordinatorServiceImp
override def heartbeatStream( override def heartbeatStream(
responseObserver: StreamObserver[CoordinatorCommand], responseObserver: StreamObserver[CoordinatorCommand],
): StreamObserver[HeartbeatFrame] = ): StreamObserver[HeartbeatFrame] =
log.info("New heartbeat stream connection established")
new StreamObserver[HeartbeatFrame]: new StreamObserver[HeartbeatFrame]:
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
private var lastInstanceId = "" private var lastInstanceId = ""
private var firstFrameSeen = false
// scalafix:on DisableSyntax.var // scalafix:on DisableSyntax.var
override def onNext(frame: HeartbeatFrame): Unit = override def onNext(frame: HeartbeatFrame): Unit =
lastInstanceId = frame.getInstanceId lastInstanceId = frame.getInstanceId
if !firstFrameSeen then
firstFrameSeen = true
log.infof(
"First heartbeat from instance %s (host=%s http=%d grpc=%d)",
frame.getInstanceId,
frame.getHostname,
frame.getHttpPort,
frame.getGrpcPort,
)
instanceRegistry instanceRegistry
.updateInstanceFromRedis(frame.getInstanceId) .updateInstanceFromRedis(frame.getInstanceId)
.subscribe() .subscribe()
@@ -16,10 +16,18 @@ class CoreGrpcClient:
private val channels = ConcurrentHashMap[String, ManagedChannel]() private val channels = ConcurrentHashMap[String, ManagedChannel]()
private def getChannel(host: String, port: Int): ManagedChannel = private def getChannel(host: String, port: Int): ManagedChannel =
channels.computeIfAbsent(s"$host:$port", _ => ManagedChannelBuilder.forAddress(host, port).usePlaintext().build()) channels.computeIfAbsent(
s"$host:$port",
_ =>
log.infof("Opening gRPC channel to %s:%d", host, port)
ManagedChannelBuilder.forAddress(host, port).usePlaintext().build(),
)
private def evictStaleChannel(host: String, port: Int): Unit = private def evictStaleChannel(host: String, port: Int): Unit =
Option(channels.remove(s"$host:$port")).foreach(_.shutdownNow()) Option(channels.remove(s"$host:$port")).foreach { ch =>
log.infof("Evicting stale gRPC channel to %s:%d", host, port)
ch.shutdownNow()
}
@PreDestroy @PreDestroy
def shutdown(): Unit = def shutdown(): Unit =
@@ -33,7 +41,9 @@ class CoreGrpcClient:
try try
val stub = CoordinatorServiceGrpc.newBlockingStub(getChannel(host, port)) val stub = CoordinatorServiceGrpc.newBlockingStub(getChannel(host, port))
val request = BatchResubscribeRequest.newBuilder().addAllGameIds(gameIds.asJava).build() val request = BatchResubscribeRequest.newBuilder().addAllGameIds(gameIds.asJava).build()
stub.batchResubscribeGames(request).getSubscribedCount val count = stub.batchResubscribeGames(request).getSubscribedCount
log.debugf("batchResubscribeGames %s:%d — subscribed %d games", host, port, count)
count
catch catch
case ex: Exception => case ex: Exception =>
log.warnf(ex, "batchResubscribeGames RPC failed for %s:%d", host, port) log.warnf(ex, "batchResubscribeGames RPC failed for %s:%d", host, port)
@@ -44,7 +54,9 @@ class CoreGrpcClient:
try try
val stub = CoordinatorServiceGrpc.newBlockingStub(getChannel(host, port)) val stub = CoordinatorServiceGrpc.newBlockingStub(getChannel(host, port))
val request = UnsubscribeGamesRequest.newBuilder().addAllGameIds(gameIds.asJava).build() val request = UnsubscribeGamesRequest.newBuilder().addAllGameIds(gameIds.asJava).build()
stub.unsubscribeGames(request).getUnsubscribedCount val count = stub.unsubscribeGames(request).getUnsubscribedCount
log.debugf("unsubscribeGames %s:%d — unsubscribed %d games", host, port, count)
count
catch catch
case ex: Exception => case ex: Exception =>
log.warnf(ex, "unsubscribeGames RPC failed for %s:%d", host, port) log.warnf(ex, "unsubscribeGames RPC failed for %s:%d", host, port)
@@ -55,7 +67,9 @@ class CoreGrpcClient:
try try
val stub = CoordinatorServiceGrpc.newBlockingStub(getChannel(host, port)) val stub = CoordinatorServiceGrpc.newBlockingStub(getChannel(host, port))
val request = EvictGamesRequest.newBuilder().addAllGameIds(gameIds.asJava).build() val request = EvictGamesRequest.newBuilder().addAllGameIds(gameIds.asJava).build()
stub.evictGames(request).getEvictedCount val count = stub.evictGames(request).getEvictedCount
log.debugf("evictGames %s:%d — evicted %d games", host, port, count)
count
catch catch
case ex: Exception => case ex: Exception =>
log.warnf(ex, "evictGames RPC failed for %s:%d", host, port) log.warnf(ex, "evictGames RPC failed for %s:%d", host, port)
@@ -9,6 +9,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
import de.nowchess.coordinator.dto.InstanceMetadata import de.nowchess.coordinator.dto.InstanceMetadata
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
import io.smallrye.mutiny.Uni import io.smallrye.mutiny.Uni
import org.jboss.logging.Logger
@ApplicationScoped @ApplicationScoped
class InstanceRegistry: class InstanceRegistry:
@@ -18,6 +19,7 @@ class InstanceRegistry:
private var redisPrefix = "nowchess" private var redisPrefix = "nowchess"
// scalafix:on DisableSyntax.var // scalafix:on DisableSyntax.var
private val log = Logger.getLogger(classOf[InstanceRegistry])
private val mapper = ObjectMapper() private val mapper = ObjectMapper()
private val instances = ConcurrentHashMap[String, InstanceMetadata]() private val instances = ConcurrentHashMap[String, InstanceMetadata]()
@@ -39,17 +41,30 @@ class InstanceRegistry:
.transformToUni { value => .transformToUni { value =>
try try
val metadata = mapper.readValue(value, classOf[InstanceMetadata]) val metadata = mapper.readValue(value, classOf[InstanceMetadata])
val isNew = !instances.containsKey(instanceId)
instances.put(instanceId, metadata) instances.put(instanceId, metadata)
if isNew then
log.infof("Instance %s joined registry (subscriptions=%d)", instanceId, metadata.subscriptionCount)
else
log.debugf(
"Instance %s updated (subscriptions=%d state=%s)",
instanceId,
metadata.subscriptionCount,
metadata.state,
)
Uni.createFrom().item(()) Uni.createFrom().item(())
catch case _: Exception => Uni.createFrom().item(()) catch
case ex: Exception =>
log.warnf(ex, "Failed to parse instance metadata for %s", instanceId)
Uni.createFrom().item(())
} }
.onFailure() .onFailure()
.recoverWithItem(()) .recoverWithItem(())
def markInstanceDead(instanceId: String): Unit = def markInstanceDead(instanceId: String): Unit =
instances.computeIfPresent(instanceId, (_, inst) => inst.copy(state = "DEAD")) instances.computeIfPresent(instanceId, (_, inst) => inst.copy(state = "DEAD"))
() log.infof("Instance %s marked dead", instanceId)
def removeInstance(instanceId: String): Unit = def removeInstance(instanceId: String): Unit =
instances.remove(instanceId) instances.remove(instanceId)
() log.infof("Instance %s removed from registry", instanceId)
@@ -7,11 +7,15 @@ import scala.compiletime.uninitialized
import de.nowchess.coordinator.proto.{CoordinatorServiceGrpc, *} import de.nowchess.coordinator.proto.{CoordinatorServiceGrpc, *}
import de.nowchess.chess.redis.GameRedisSubscriberManager import de.nowchess.chess.redis.GameRedisSubscriberManager
import io.grpc.stub.StreamObserver import io.grpc.stub.StreamObserver
import org.jboss.logging.Logger
import scala.jdk.CollectionConverters.* import scala.jdk.CollectionConverters.*
@GrpcService @GrpcService
@Singleton @Singleton
class CoordinatorServiceHandler extends CoordinatorServiceGrpc.CoordinatorServiceImplBase: class CoordinatorServiceHandler extends CoordinatorServiceGrpc.CoordinatorServiceImplBase:
private val log = Logger.getLogger(classOf[CoordinatorServiceHandler])
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
@Inject @Inject
private var gameSubscriberManager: GameRedisSubscriberManager = uninitialized private var gameSubscriberManager: GameRedisSubscriberManager = uninitialized
@@ -22,6 +26,7 @@ class CoordinatorServiceHandler extends CoordinatorServiceGrpc.CoordinatorServic
responseObserver: StreamObserver[BatchResubscribeResponse], responseObserver: StreamObserver[BatchResubscribeResponse],
): Unit = ): Unit =
val count = gameSubscriberManager.batchResubscribeGames(request.getGameIdsList) val count = gameSubscriberManager.batchResubscribeGames(request.getGameIdsList)
log.infof("Coordinator: batch resubscribe %d games → subscribed %d", request.getGameIdsList.size(), count)
val response = BatchResubscribeResponse val response = BatchResubscribeResponse
.newBuilder() .newBuilder()
.setSubscribedCount(count) .setSubscribedCount(count)
@@ -34,6 +39,7 @@ class CoordinatorServiceHandler extends CoordinatorServiceGrpc.CoordinatorServic
responseObserver: StreamObserver[UnsubscribeGamesResponse], responseObserver: StreamObserver[UnsubscribeGamesResponse],
): Unit = ): Unit =
val count = gameSubscriberManager.unsubscribeGames(request.getGameIdsList) val count = gameSubscriberManager.unsubscribeGames(request.getGameIdsList)
log.infof("Coordinator: unsubscribe %d games → unsubscribed %d", request.getGameIdsList.size(), count)
val response = UnsubscribeGamesResponse val response = UnsubscribeGamesResponse
.newBuilder() .newBuilder()
.setUnsubscribedCount(count) .setUnsubscribedCount(count)
@@ -46,6 +52,7 @@ class CoordinatorServiceHandler extends CoordinatorServiceGrpc.CoordinatorServic
responseObserver: StreamObserver[EvictGamesResponse], responseObserver: StreamObserver[EvictGamesResponse],
): Unit = ): Unit =
val count = gameSubscriberManager.evictGames(request.getGameIdsList) val count = gameSubscriberManager.evictGames(request.getGameIdsList)
log.infof("Coordinator: evict %d games → evicted %d", request.getGameIdsList.size(), count)
val response = EvictGamesResponse val response = EvictGamesResponse
.newBuilder() .newBuilder()
.setEvictedCount(count) .setEvictedCount(count)
@@ -58,6 +65,7 @@ class CoordinatorServiceHandler extends CoordinatorServiceGrpc.CoordinatorServic
responseObserver: StreamObserver[DrainInstanceResponse], responseObserver: StreamObserver[DrainInstanceResponse],
): Unit = ): Unit =
val migrated = gameSubscriberManager.drainInstance() val migrated = gameSubscriberManager.drainInstance()
log.infof("Coordinator: drain instance → migrated %d games", migrated)
val response = DrainInstanceResponse val response = DrainInstanceResponse
.newBuilder() .newBuilder()
.setGamesMigrated(migrated) .setGamesMigrated(migrated)
@@ -5,29 +5,63 @@ import de.nowchess.chess.client.CombinedExportResponse
import de.nowchess.core.proto.* import de.nowchess.core.proto.*
import io.quarkus.grpc.GrpcClient import io.quarkus.grpc.GrpcClient
import jakarta.enterprise.context.ApplicationScoped import jakarta.enterprise.context.ApplicationScoped
import org.jboss.logging.Logger
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
@ApplicationScoped @ApplicationScoped
class IoGrpcClientWrapper: class IoGrpcClientWrapper:
private val log = Logger.getLogger(classOf[IoGrpcClientWrapper])
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
@GrpcClient("io-grpc") @GrpcClient("io-grpc")
var stub: IoServiceGrpc.IoServiceBlockingStub = uninitialized var stub: IoServiceGrpc.IoServiceBlockingStub = uninitialized
// scalafix:on DisableSyntax.var // scalafix:on DisableSyntax.var
def exportCombined(ctx: GameContext): CombinedExportResponse = def exportCombined(ctx: GameContext): CombinedExportResponse =
val combined = stub.exportCombined(CoreProtoMapper.toProtoGameContext(ctx)) try
CombinedExportResponse(combined.getFen, combined.getPgn) val combined = stub.exportCombined(CoreProtoMapper.toProtoGameContext(ctx))
CombinedExportResponse(combined.getFen, combined.getPgn)
catch
case ex: Exception =>
log.warnf(ex, "IO gRPC exportCombined failed")
// scalafix:off DisableSyntax.throw
throw ex
// scalafix:on DisableSyntax.throw
def importFen(fen: String): GameContext = def importFen(fen: String): GameContext =
CoreProtoMapper.fromProtoGameContext(stub.importFen(ProtoImportFenRequest.newBuilder().setFen(fen).build())) try CoreProtoMapper.fromProtoGameContext(stub.importFen(ProtoImportFenRequest.newBuilder().setFen(fen).build()))
catch
case ex: Exception =>
log.warnf(ex, "IO gRPC importFen failed for fen %s", fen)
// scalafix:off DisableSyntax.throw
throw ex
// scalafix:on DisableSyntax.throw
def importPgn(pgn: String): GameContext = def importPgn(pgn: String): GameContext =
CoreProtoMapper.fromProtoGameContext(stub.importPgn(ProtoImportPgnRequest.newBuilder().setPgn(pgn).build())) try CoreProtoMapper.fromProtoGameContext(stub.importPgn(ProtoImportPgnRequest.newBuilder().setPgn(pgn).build()))
catch
case ex: Exception =>
log.warnf(ex, "IO gRPC importPgn failed")
// scalafix:off DisableSyntax.throw
throw ex
// scalafix:on DisableSyntax.throw
def exportFen(ctx: GameContext): String = def exportFen(ctx: GameContext): String =
stub.exportFen(CoreProtoMapper.toProtoGameContext(ctx)).getValue try stub.exportFen(CoreProtoMapper.toProtoGameContext(ctx)).getValue
catch
case ex: Exception =>
log.warnf(ex, "IO gRPC exportFen failed")
// scalafix:off DisableSyntax.throw
throw ex
// scalafix:on DisableSyntax.throw
def exportPgn(ctx: GameContext): String = def exportPgn(ctx: GameContext): String =
stub.exportPgn(CoreProtoMapper.toProtoGameContext(ctx)).getValue try stub.exportPgn(CoreProtoMapper.toProtoGameContext(ctx)).getValue
catch
case ex: Exception =>
log.warnf(ex, "IO gRPC exportPgn failed")
// scalafix:off DisableSyntax.throw
throw ex
// scalafix:on DisableSyntax.throw
@@ -7,6 +7,7 @@ import de.nowchess.api.rules.{PostMoveStatus, RuleSet}
import de.nowchess.core.proto.* import de.nowchess.core.proto.*
import io.quarkus.grpc.GrpcClient import io.quarkus.grpc.GrpcClient
import jakarta.enterprise.context.ApplicationScoped import jakarta.enterprise.context.ApplicationScoped
import org.jboss.logging.Logger
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
import scala.jdk.CollectionConverters.* import scala.jdk.CollectionConverters.*
@@ -14,28 +15,59 @@ import scala.jdk.CollectionConverters.*
@ApplicationScoped @ApplicationScoped
class RuleSetGrpcAdapter extends RuleSet: class RuleSetGrpcAdapter extends RuleSet:
private val log = Logger.getLogger(classOf[RuleSetGrpcAdapter])
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
@GrpcClient("rule-grpc") @GrpcClient("rule-grpc")
var stub: RuleServiceGrpc.RuleServiceBlockingStub = uninitialized var stub: RuleServiceGrpc.RuleServiceBlockingStub = uninitialized
// scalafix:on DisableSyntax.var // scalafix:on DisableSyntax.var
def candidateMoves(ctx: GameContext)(sq: Square): List[Move] = def candidateMoves(ctx: GameContext)(sq: Square): List[Move] =
val req = try
ProtoSquareRequest.newBuilder().setContext(CoreProtoMapper.toProtoGameContext(ctx)).setSquare(sq.toString).build() val req =
stub.candidateMoves(req).getMovesList.asScala.flatMap(CoreProtoMapper.fromProtoMove).toList ProtoSquareRequest
.newBuilder()
.setContext(CoreProtoMapper.toProtoGameContext(ctx))
.setSquare(sq.toString)
.build()
stub.candidateMoves(req).getMovesList.asScala.flatMap(CoreProtoMapper.fromProtoMove).toList
catch
case ex: Exception =>
log.warnf(ex, "Rule gRPC candidateMoves failed")
// scalafix:off DisableSyntax.throw
throw ex
// scalafix:on DisableSyntax.throw
def legalMoves(ctx: GameContext)(sq: Square): List[Move] = def legalMoves(ctx: GameContext)(sq: Square): List[Move] =
val req = try
ProtoSquareRequest.newBuilder().setContext(CoreProtoMapper.toProtoGameContext(ctx)).setSquare(sq.toString).build() val req =
stub.legalMoves(req).getMovesList.asScala.flatMap(CoreProtoMapper.fromProtoMove).toList ProtoSquareRequest
.newBuilder()
.setContext(CoreProtoMapper.toProtoGameContext(ctx))
.setSquare(sq.toString)
.build()
stub.legalMoves(req).getMovesList.asScala.flatMap(CoreProtoMapper.fromProtoMove).toList
catch
case ex: Exception =>
log.warnf(ex, "Rule gRPC legalMoves failed")
// scalafix:off DisableSyntax.throw
throw ex
// scalafix:on DisableSyntax.throw
def allLegalMoves(ctx: GameContext): List[Move] = def allLegalMoves(ctx: GameContext): List[Move] =
stub try
.allLegalMoves(CoreProtoMapper.toProtoGameContext(ctx)) stub
.getMovesList .allLegalMoves(CoreProtoMapper.toProtoGameContext(ctx))
.asScala .getMovesList
.flatMap(CoreProtoMapper.fromProtoMove) .asScala
.toList .flatMap(CoreProtoMapper.fromProtoMove)
.toList
catch
case ex: Exception =>
log.warnf(ex, "Rule gRPC allLegalMoves failed")
// scalafix:off DisableSyntax.throw
throw ex
// scalafix:on DisableSyntax.throw
def isCheck(ctx: GameContext): Boolean = def isCheck(ctx: GameContext): Boolean =
stub.isCheck(CoreProtoMapper.toProtoGameContext(ctx)).getValue stub.isCheck(CoreProtoMapper.toProtoGameContext(ctx)).getValue
@@ -56,19 +88,33 @@ class RuleSetGrpcAdapter extends RuleSet:
stub.isThreefoldRepetition(CoreProtoMapper.toProtoGameContext(ctx)).getValue stub.isThreefoldRepetition(CoreProtoMapper.toProtoGameContext(ctx)).getValue
def applyMove(ctx: GameContext)(move: Move): GameContext = def applyMove(ctx: GameContext)(move: Move): GameContext =
val req = ProtoMoveRequest try
.newBuilder() val req = ProtoMoveRequest
.setContext(CoreProtoMapper.toProtoGameContext(ctx)) .newBuilder()
.setMove(CoreProtoMapper.toProtoMove(move)) .setContext(CoreProtoMapper.toProtoGameContext(ctx))
.build() .setMove(CoreProtoMapper.toProtoMove(move))
CoreProtoMapper.fromProtoGameContext(stub.applyMove(req)) .build()
CoreProtoMapper.fromProtoGameContext(stub.applyMove(req))
catch
case ex: Exception =>
log.warnf(ex, "Rule gRPC applyMove failed")
// scalafix:off DisableSyntax.throw
throw ex
// scalafix:on DisableSyntax.throw
override def postMoveStatus(ctx: GameContext): PostMoveStatus = override def postMoveStatus(ctx: GameContext): PostMoveStatus =
val p = stub.postMoveStatus(CoreProtoMapper.toProtoGameContext(ctx)) try
PostMoveStatus( val p = stub.postMoveStatus(CoreProtoMapper.toProtoGameContext(ctx))
p.getIsCheckmate, PostMoveStatus(
p.getIsStalemate, p.getIsCheckmate,
p.getIsInsufficientMaterial, p.getIsStalemate,
p.getIsCheck, p.getIsInsufficientMaterial,
p.getIsThreefoldRepetition, p.getIsCheck,
) p.getIsThreefoldRepetition,
)
catch
case ex: Exception =>
log.warnf(ex, "Rule gRPC postMoveStatus failed")
// scalafix:off DisableSyntax.throw
throw ex
// scalafix:on DisableSyntax.throw
@@ -10,6 +10,10 @@ import de.nowchess.chess.observer.{GameEvent, Observer}
import de.nowchess.chess.registry.GameRegistry import de.nowchess.chess.registry.GameRegistry
import de.nowchess.chess.resource.GameDtoMapper import de.nowchess.chess.resource.GameDtoMapper
import io.quarkus.redis.datasource.RedisDataSource import io.quarkus.redis.datasource.RedisDataSource
import org.jboss.logging.Logger
object GameRedisPublisher:
private val log = Logger.getLogger(classOf[GameRedisPublisher])
class GameRedisPublisher( class GameRedisPublisher(
gameId: String, gameId: String,
@@ -23,57 +27,60 @@ class GameRedisPublisher(
) extends Observer: ) extends Observer:
def onGameEvent(event: GameEvent): Unit = def onGameEvent(event: GameEvent): Unit =
registry.get(gameId).foreach { entry => try
val dto = GameDtoMapper.toGameStateDto(entry, ioClient) GameRedisPublisher.log.debugf("Publishing game event for game %s", gameId)
val json = objectMapper.writeValueAsString(GameStateEventDto(dto)) registry.get(gameId).foreach { entry =>
redis.pubsub(classOf[String]).publish(s2cTopicName, json) val dto = GameDtoMapper.toGameStateDto(entry, ioClient)
val json = objectMapper.writeValueAsString(GameStateEventDto(dto))
redis.pubsub(classOf[String]).publish(s2cTopicName, json)
val clock = entry.engine.currentClockState val clock = entry.engine.currentClockState
val wb = GameWritebackEventDto( val wb = GameWritebackEventDto(
gameId = gameId, gameId = gameId,
fen = dto.fen, fen = dto.fen,
pgn = dto.pgn, pgn = dto.pgn,
moveCount = entry.engine.context.moves.size, moveCount = entry.engine.context.moves.size,
whiteId = entry.white.id.value, whiteId = entry.white.id.value,
whiteName = entry.white.displayName, whiteName = entry.white.displayName,
blackId = entry.black.id.value, blackId = entry.black.id.value,
blackName = entry.black.displayName, blackName = entry.black.displayName,
mode = entry.mode.toString, mode = entry.mode.toString,
resigned = entry.resigned, resigned = entry.resigned,
limitSeconds = entry.engine.timeControl match { limitSeconds = entry.engine.timeControl match {
case de.nowchess.api.game.TimeControl.Clock(l, _) => Some(l); case _ => None case de.nowchess.api.game.TimeControl.Clock(l, _) => Some(l); case _ => None
}, },
incrementSeconds = entry.engine.timeControl match { incrementSeconds = entry.engine.timeControl match {
case de.nowchess.api.game.TimeControl.Clock(_, i) => Some(i); case _ => None case de.nowchess.api.game.TimeControl.Clock(_, i) => Some(i); case _ => None
}, },
daysPerMove = entry.engine.timeControl match { daysPerMove = entry.engine.timeControl match {
case de.nowchess.api.game.TimeControl.Correspondence(d) => Some(d); case _ => None case de.nowchess.api.game.TimeControl.Correspondence(d) => Some(d); case _ => None
}, },
whiteRemainingMs = clock.collect { case c: LiveClockState => c.whiteRemainingMs }, whiteRemainingMs = clock.collect { case c: LiveClockState => c.whiteRemainingMs },
blackRemainingMs = clock.collect { case c: LiveClockState => c.blackRemainingMs }, blackRemainingMs = clock.collect { case c: LiveClockState => c.blackRemainingMs },
incrementMs = clock.collect { case c: LiveClockState => c.incrementMs }, incrementMs = clock.collect { case c: LiveClockState => c.incrementMs },
clockLastTickAt = clock.collect { case c: LiveClockState => c.lastTickAt.toEpochMilli }, clockLastTickAt = clock.collect { case c: LiveClockState => c.lastTickAt.toEpochMilli },
clockMoveDeadline = clock.collect { case c: CorrespondenceClockState => c.moveDeadline.toEpochMilli }, clockMoveDeadline = clock.collect { case c: CorrespondenceClockState => c.moveDeadline.toEpochMilli },
clockActiveColor = clock.map(_.activeColor.label.toLowerCase), clockActiveColor = clock.map(_.activeColor.label.toLowerCase),
pendingDrawOffer = entry.engine.pendingDrawOfferBy.map(_.label.toLowerCase), pendingDrawOffer = entry.engine.pendingDrawOfferBy.map(_.label.toLowerCase),
result = entry.engine.context.result.map { result = entry.engine.context.result.map {
case GameResult.Win(Color.White, _) => "white" case GameResult.Win(Color.White, _) => "white"
case GameResult.Win(Color.Black, _) => "black" case GameResult.Win(Color.Black, _) => "black"
case GameResult.Draw(_) => "draw" case GameResult.Draw(_) => "draw"
}, },
terminationReason = entry.engine.context.result.map { terminationReason = entry.engine.context.result.map {
case GameResult.Win(_, WinReason.Checkmate) => "checkmate" case GameResult.Win(_, WinReason.Checkmate) => "checkmate"
case GameResult.Win(_, WinReason.Resignation) => "resignation" case GameResult.Win(_, WinReason.Resignation) => "resignation"
case GameResult.Win(_, WinReason.TimeControl) => "timeout" case GameResult.Win(_, WinReason.TimeControl) => "timeout"
case GameResult.Draw(DrawReason.Stalemate) => "stalemate" case GameResult.Draw(DrawReason.Stalemate) => "stalemate"
case GameResult.Draw(DrawReason.InsufficientMaterial) => "insufficient_material" case GameResult.Draw(DrawReason.InsufficientMaterial) => "insufficient_material"
case GameResult.Draw(DrawReason.FiftyMoveRule) => "fifty_move" case GameResult.Draw(DrawReason.FiftyMoveRule) => "fifty_move"
case GameResult.Draw(DrawReason.ThreefoldRepetition) => "repetition" case GameResult.Draw(DrawReason.ThreefoldRepetition) => "repetition"
case GameResult.Draw(DrawReason.Agreement) => "agreement" case GameResult.Draw(DrawReason.Agreement) => "agreement"
}, },
redoStack = entry.engine.redoStackMoves.map(GameDtoMapper.moveToUci), redoStack = entry.engine.redoStackMoves.map(GameDtoMapper.moveToUci),
pendingTakebackRequest = entry.engine.pendingTakebackRequestBy.map(_.label.toLowerCase), pendingTakebackRequest = entry.engine.pendingTakebackRequestBy.map(_.label.toLowerCase),
) )
writebackEmit(objectMapper.writeValueAsString(wb)) writebackEmit(objectMapper.writeValueAsString(wb))
if entry.engine.context.result.isDefined then onGameOver(gameId) if entry.engine.context.result.isDefined then onGameOver(gameId)
} }
catch case ex: Exception => GameRedisPublisher.log.warnf(ex, "Failed to publish game event for game %s", gameId)
@@ -16,6 +16,7 @@ import jakarta.annotation.PreDestroy
import jakarta.enterprise.context.ApplicationScoped import jakarta.enterprise.context.ApplicationScoped
import jakarta.enterprise.inject.Instance import jakarta.enterprise.inject.Instance
import jakarta.inject.Inject import jakarta.inject.Inject
import org.jboss.logging.Logger
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
import scala.util.Try import scala.util.Try
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
@@ -24,6 +25,8 @@ import java.util.function.Consumer
@ApplicationScoped @ApplicationScoped
class GameRedisSubscriberManager: class GameRedisSubscriberManager:
private val log = Logger.getLogger(classOf[GameRedisSubscriberManager])
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
@Inject var redis: RedisDataSource = uninitialized @Inject var redis: RedisDataSource = uninitialized
@Inject var registry: GameRegistry = uninitialized @Inject var registry: GameRegistry = uninitialized
@@ -65,11 +68,12 @@ class GameRedisSubscriberManager:
) )
s2cObservers.put(gameId, obs) s2cObservers.put(gameId, obs)
registry.get(gameId).foreach(_.engine.subscribe(obs)) registry.get(gameId).foreach(_.engine.subscribe(obs))
log.debugf("Subscribed to game %s", gameId)
heartbeatServiceOpt.foreach(_.addGameSubscription(gameId)) heartbeatServiceOpt.foreach(_.addGameSubscription(gameId))
catch catch
case e: Exception => case e: Exception =>
System.err.println(s"Warning: Redis subscription failed for game $gameId: ${e.getMessage}") log.warnf(e, "Redis subscription failed for game %s", gameId)
() ()
def unsubscribeGame(gameId: String): Unit = def unsubscribeGame(gameId: String): Unit =
@@ -81,6 +85,7 @@ class GameRedisSubscriberManager:
} }
heartbeatServiceOpt.foreach(_.removeGameSubscription(gameId)) heartbeatServiceOpt.foreach(_.removeGameSubscription(gameId))
log.debugf("Unsubscribed from game %s", gameId)
private def handleC2sMessage(gameId: String, msg: String): Unit = private def handleC2sMessage(gameId: String, msg: String): Unit =
parseC2sMessage(msg) match parseC2sMessage(msg) match
@@ -97,6 +102,7 @@ class GameRedisSubscriberManager:
} }
private def handleMove(gameId: String, uci: String, playerId: Option[String]): Unit = private def handleMove(gameId: String, uci: String, playerId: Option[String]): Unit =
log.debugf("Processing move %s for game %s by player %s", uci, gameId, playerId.getOrElse("anonymous"))
registry.get(gameId).foreach { entry => registry.get(gameId).foreach { entry =>
entry.mode match entry.mode match
case GameMode.Open => entry.engine.processUserInput(uci) case GameMode.Open => entry.engine.processUserInput(uci)
@@ -127,6 +133,7 @@ class GameRedisSubscriberManager:
def batchResubscribeGames(gameIds: java.util.List[String]): Int = def batchResubscribeGames(gameIds: java.util.List[String]): Int =
gameIds.forEach(subscribeGame) gameIds.forEach(subscribeGame)
log.infof("Batch resubscribed %d games", gameIds.size())
gameIds.size() gameIds.size()
def unsubscribeGames(gameIds: java.util.List[String]): Int = def unsubscribeGames(gameIds: java.util.List[String]): Int =
@@ -135,12 +142,14 @@ class GameRedisSubscriberManager:
def evictGames(gameIds: java.util.List[String]): Int = def evictGames(gameIds: java.util.List[String]): Int =
gameIds.forEach(unsubscribeGame) gameIds.forEach(unsubscribeGame)
log.infof("Evicting %d games", gameIds.size())
gameIds.size() gameIds.size()
def drainInstance(): Int = def drainInstance(): Int =
val gameIds = new java.util.ArrayList(c2sListeners.keySet()) val gameIds = new java.util.ArrayList(c2sListeners.keySet())
val count = gameIds.size() val count = gameIds.size()
gameIds.forEach(unsubscribeGame) gameIds.forEach(unsubscribeGame)
log.infof("Draining instance, unsubscribing %d games", count)
count count
@PreDestroy @PreDestroy
@@ -17,6 +17,7 @@ import jakarta.enterprise.context.ApplicationScoped
import jakarta.inject.Inject import jakarta.inject.Inject
import org.eclipse.microprofile.rest.client.inject.RestClient import org.eclipse.microprofile.rest.client.inject.RestClient
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
import org.jboss.logging.Logger
import scala.util.Try import scala.util.Try
import java.nio.charset.StandardCharsets import java.nio.charset.StandardCharsets
import java.security.{MessageDigest, SecureRandom} import java.security.{MessageDigest, SecureRandom}
@@ -35,6 +36,7 @@ class RedisGameRegistry extends GameRegistry:
@Inject @RestClient var storeClient: StoreServiceClient = uninitialized @Inject @RestClient var storeClient: StoreServiceClient = uninitialized
// scalafix:on // scalafix:on
private val log = Logger.getLogger(classOf[RedisGameRegistry])
private val localEngines = ConcurrentHashMap[String, GameEntry]() private val localEngines = ConcurrentHashMap[String, GameEntry]()
private val rng = new SecureRandom() private val rng = new SecureRandom()
@@ -48,6 +50,12 @@ class RedisGameRegistry extends GameRegistry:
localEngines.put(entry.gameId, entry) localEngines.put(entry.gameId, entry)
val combined = ioClient.exportCombined(entry.engine.context) val combined = ioClient.exportCombined(entry.engine.context)
redis.value(classOf[String]).setex(cacheKey(entry.gameId), 1800L, toJson(entry, combined.fen, combined.pgn)) redis.value(classOf[String]).setex(cacheKey(entry.gameId), 1800L, toJson(entry, combined.fen, combined.pgn))
log.infof(
"Stored game %s in registry (white=%s black=%s)",
entry.gameId,
entry.white.displayName,
entry.black.displayName,
)
def get(gameId: String): Option[GameEntry] = def get(gameId: String): Option[GameEntry] =
Option(localEngines.get(gameId)) match Option(localEngines.get(gameId)) match
@@ -71,9 +79,15 @@ class RedisGameRegistry extends GameRegistry:
private def fromRedis(gameId: String): Option[GameEntry] = private def fromRedis(gameId: String): Option[GameEntry] =
readRedisDto(gameId) readRedisDto(gameId)
.flatMap(dto => Try(reconstruct(dto)).toOption) .flatMap { dto =>
Try(reconstruct(dto)).toOption.orElse {
log.warnf("Failed to reconstruct game %s from Redis", gameId)
None
}
}
.map { entry => .map { entry =>
localEngines.put(gameId, entry) localEngines.put(gameId, entry)
log.infof("Loaded game %s from Redis cache", gameId)
entry entry
} }
@@ -102,12 +116,15 @@ class RedisGameRegistry extends GameRegistry:
pendingDrawOffer = Option(record.pendingDrawOffer), pendingDrawOffer = Option(record.pendingDrawOffer),
) )
(dto, reconstruct(dto)) (dto, reconstruct(dto))
}.toOption } match
.map { case (dto, entry) => case scala.util.Success((dto, entry)) =>
log.infof("Loaded game %s from store service", gameId)
localEngines.put(gameId, entry) localEngines.put(gameId, entry)
redis.value(classOf[String]).setex(cacheKey(gameId), 1800L, objectMapper.writeValueAsString(dto)) redis.value(classOf[String]).setex(cacheKey(gameId), 1800L, objectMapper.writeValueAsString(dto))
entry Some(entry)
} case scala.util.Failure(ex) =>
log.warnf(ex, "Failed to load game %s from store service", gameId)
None
private def reconstruct(dto: GameCacheDto): GameEntry = private def reconstruct(dto: GameCacheDto): GameEntry =
val ctx = if dto.pgn.nonEmpty then ioClient.importPgn(dto.pgn) else GameContext.initial val ctx = if dto.pgn.nonEmpty then ioClient.importPgn(dto.pgn) else GameContext.initial
@@ -30,6 +30,7 @@ import jakarta.inject.Inject
import jakarta.ws.rs.* import jakarta.ws.rs.*
import jakarta.ws.rs.core.{MediaType, Response} import jakarta.ws.rs.core.{MediaType, Response}
import org.eclipse.microprofile.jwt.JsonWebToken import org.eclipse.microprofile.jwt.JsonWebToken
import org.jboss.logging.Logger
import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicReference
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
@@ -38,6 +39,8 @@ import scala.compiletime.uninitialized
@ApplicationScoped @ApplicationScoped
class GameResource: class GameResource:
private val log = Logger.getLogger(classOf[GameResource])
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
@Inject @Inject
var registry: GameRegistry = uninitialized var registry: GameRegistry = uninitialized
@@ -165,7 +168,13 @@ class GameResource:
val entry = newEntry(GameContext.initial, white, black, tc, mode) val entry = newEntry(GameContext.initial, white, black, tc, mode)
registry.store(entry) registry.store(entry)
subscriberManager.subscribeGame(entry.gameId) subscriberManager.subscribeGame(entry.gameId)
println(s"Created game ${entry.gameId}") log.infof(
"Game %s created — white=%s black=%s mode=%s",
entry.gameId,
white.displayName,
black.displayName,
mode.toString,
)
created(GameDtoMapper.toGameFullDto(entry, ioClient)) created(GameDtoMapper.toGameFullDto(entry, ioClient))
@GET @GET
@@ -182,6 +191,7 @@ class GameResource:
val entry = registry.get(gameId).getOrElse(throw GameNotFoundException(gameId)) val entry = registry.get(gameId).getOrElse(throw GameNotFoundException(gameId))
assertGameNotOver(entry) assertGameNotOver(entry)
val color = colorOf(entry) val color = colorOf(entry)
log.infof("Game %s — resign by %s", gameId, color.label)
entry.engine.resign(color) entry.engine.resign(color)
registry.update(entry.copy(resigned = true)) registry.update(entry.copy(resigned = true))
ok(OkResponseDto()) ok(OkResponseDto())
@@ -194,6 +204,7 @@ class GameResource:
val entry = registry.get(gameId).getOrElse(throw GameNotFoundException(gameId)) val entry = registry.get(gameId).getOrElse(throw GameNotFoundException(gameId))
assertGameNotOver(entry) assertGameNotOver(entry)
assertIsCurrentPlayer(entry) assertIsCurrentPlayer(entry)
log.debugf("Game %s — move %s by %s", gameId, uci, colorOf(entry).label)
if Parser.parseMove(uci).isEmpty then if Parser.parseMove(uci).isEmpty then
throw BadRequestException("INVALID_UCI", s"Invalid UCI notation: $uci", Some("uci")) throw BadRequestException("INVALID_UCI", s"Invalid UCI notation: $uci", Some("uci"))
applyMoveInput(entry.engine, uci).foreach(err => throw BadRequestException("INVALID_MOVE", err, Some("uci"))) applyMoveInput(entry.engine, uci).foreach(err => throw BadRequestException("INVALID_MOVE", err, Some("uci")))
@@ -284,6 +295,7 @@ class GameResource:
val entry = newEntry(ctx, white, black, tc) val entry = newEntry(ctx, white, black, tc)
registry.store(entry) registry.store(entry)
subscriberManager.subscribeGame(entry.gameId) subscriberManager.subscribeGame(entry.gameId)
log.infof("Imported FEN game %s", entry.gameId)
created(GameDtoMapper.toGameFullDto(entry, ioClient)) created(GameDtoMapper.toGameFullDto(entry, ioClient))
@POST @POST
@@ -295,6 +307,7 @@ class GameResource:
val entry = newEntry(ctx, DefaultWhite, DefaultBlack) val entry = newEntry(ctx, DefaultWhite, DefaultBlack)
registry.store(entry) registry.store(entry)
subscriberManager.subscribeGame(entry.gameId) subscriberManager.subscribeGame(entry.gameId)
log.infof("Imported PGN game %s", entry.gameId)
created(GameDtoMapper.toGameFullDto(entry, ioClient)) created(GameDtoMapper.toGameFullDto(entry, ioClient))
@GET @GET
@@ -6,6 +6,7 @@ import io.quarkus.redis.datasource.pubsub.PubSubCommands
import io.quarkus.websockets.next.* import io.quarkus.websockets.next.*
import io.smallrye.jwt.auth.principal.JWTParser import io.smallrye.jwt.auth.principal.JWTParser
import jakarta.inject.Inject import jakarta.inject.Inject
import org.jboss.logging.Logger
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
import scala.util.Try import scala.util.Try
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
@@ -14,6 +15,8 @@ import java.util.function.Consumer
@WebSocket(path = "/api/board/game/{gameId}/ws") @WebSocket(path = "/api/board/game/{gameId}/ws")
class GameWebSocketResource: class GameWebSocketResource:
private val log = Logger.getLogger(classOf[GameWebSocketResource])
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
@Inject @Inject
var redis: RedisDataSource = uninitialized var redis: RedisDataSource = uninitialized
@@ -40,6 +43,7 @@ class GameWebSocketResource:
.filter(_.nonEmpty) .filter(_.nonEmpty)
.flatMap(token => Try(jwtParser.parse(token)).toOption) .flatMap(token => Try(jwtParser.parse(token)).toOption)
.map(_.getSubject) .map(_.getSubject)
log.infof("Game WebSocket opened — gameId=%s playerId=%s", gameId, playerId.getOrElse("anonymous"))
val handler: Consumer[String] = msg => connection.sendText(msg).subscribe().`with`(_ => (), _ => ()) val handler: Consumer[String] = msg => connection.sendText(msg).subscribe().`with`(_ => (), _ => ())
val subscriber = redis.pubsub(classOf[String]).subscribe(s2cTopic(gameId), handler) val subscriber = redis.pubsub(classOf[String]).subscribe(s2cTopic(gameId), handler)
connections.put(connection.id(), ConnectionMeta(gameId, subscriber, playerId)) connections.put(connection.id(), ConnectionMeta(gameId, subscriber, playerId))
@@ -60,6 +64,7 @@ class GameWebSocketResource:
@OnClose @OnClose
def onClose(connection: WebSocketConnection): Unit = def onClose(connection: WebSocketConnection): Unit =
Option(connections.remove(connection.id())).foreach { meta => Option(connections.remove(connection.id())).foreach { meta =>
log.infof("Game WebSocket closed — gameId=%s", meta.gameId)
meta.subscriber.unsubscribe(s2cTopic(meta.gameId)) meta.subscriber.unsubscribe(s2cTopic(meta.gameId))
} }
@@ -6,6 +6,7 @@ import io.quarkus.redis.datasource.pubsub.PubSubCommands
import io.quarkus.websockets.next.* import io.quarkus.websockets.next.*
import io.smallrye.jwt.auth.principal.JWTParser import io.smallrye.jwt.auth.principal.JWTParser
import jakarta.inject.Inject import jakarta.inject.Inject
import org.jboss.logging.Logger
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
import scala.util.Try import scala.util.Try
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
@@ -14,6 +15,8 @@ import java.util.function.Consumer
@WebSocket(path = "/api/user/ws") @WebSocket(path = "/api/user/ws")
class UserWebSocketResource: class UserWebSocketResource:
private val log = Logger.getLogger(classOf[UserWebSocketResource])
// scalafix:off DisableSyntax.var // scalafix:off DisableSyntax.var
@Inject @Inject
var redis: RedisDataSource = uninitialized var redis: RedisDataSource = uninitialized
@@ -38,8 +41,11 @@ class UserWebSocketResource:
.map(_.getSubject) .map(_.getSubject)
userIdOpt match userIdOpt match
case None => connection.close().subscribe().`with`(_ => (), _ => ()) case None =>
log.warn("WebSocket opened with no valid JWT — closing connection")
connection.close().subscribe().`with`(_ => (), _ => ())
case Some(userId) => case Some(userId) =>
log.infof("User WebSocket opened — userId=%s", userId)
val handler: Consumer[String] = msg => connection.sendText(msg).subscribe().`with`(_ => (), _ => ()) val handler: Consumer[String] = msg => connection.sendText(msg).subscribe().`with`(_ => (), _ => ())
val subscriber = redis.pubsub(classOf[String]).subscribe(userTopic(userId), handler) val subscriber = redis.pubsub(classOf[String]).subscribe(userTopic(userId), handler)
connections.put(connection.id(), (userId, subscriber)) connections.put(connection.id(), (userId, subscriber))
@@ -48,6 +54,7 @@ class UserWebSocketResource:
@OnClose @OnClose
def onClose(connection: WebSocketConnection): Unit = def onClose(connection: WebSocketConnection): Unit =
log.infof("User WebSocket closed — connectionId=%s", connection.id())
Option(connections.remove(connection.id())).foreach { (userId, subscriber) => Option(connections.remove(connection.id())).foreach { (userId, subscriber) =>
subscriber.unsubscribe(userTopic(userId)) subscriber.unsubscribe(userTopic(userId))
} }
+1 -1
View File
@@ -1,7 +1,7 @@
#! /usr/bin/env bash #! /usr/bin/env bash
set -euo pipefail set -euo pipefail
./gradlew test ./gradlew test -Dquarkus.profile=test
if [ "$#" -eq 0 ]; then if [ "$#" -eq 0 ]; then
PYTHONUTF8=1 python3 jacoco-reporter/test_gaps.py PYTHONUTF8=1 python3 jacoco-reporter/test_gaps.py