/**
 * Executes expired session cleanup in Redis.
 *
 * Spring Session Redis uses multiple keys per session:
 * 1. Hash key (namespace:sessions:{id}) - Stores session data with 35 min TTL
 * 2. ZSet entry - Tracks session expiration times for batch cleanup
 *
 * Process flow:
 * 1. Query ZSet for sessions that should be expired
 * 2. Access their hash keys to trigger Redis expiration events
 * - Redis only guarantees expiration events when keys are accessed
 * - Spring Session listens for these events to cleanup related keys
 * 3. Remove processed entries from the ZSet
 *
 * Implementation details:
 * - Uses Lua script for efficient multi-key EXISTS checks
 * - Handles quoted session IDs in ZSet vs unquoted in hash keys
 * - Relies on Redis keyspace events for actual session deletion
 * - Bounded elastic scheduling for background processing
 *
 * @param context Operation boundaries and limits for ZSet query
 * @return Mono<Void> completing when cleanup process finishes
 * @throws RedisConnectionException if Redis operations fail
 *
 * @see "https://docs.spring.io/spring-session/reference/configuration/reactive-redis-indexed.html#how-spring-session-cleans-up-expired-sessions"
 */
 private fun cleanupExpiredSessions(context: CleanupContext): Mono<Void> =
 redisOperations.opsForZSet()
 .reverseRangeByScore(redisKeyExpirations, context.range, context.limit)
 .collectList()
 .flatMap { sessionIds ->
 if (sessionIds.isEmpty()) {
 logger.debug("No expired sessions found")
 Mono.empty()
 } else {
 logger.debug("Found {} expired sessions to remove", sessionIds.size)
 // Prepare session keys (remove quotes around session id in zset)
 val sessionKeys = sessionIds.map { sessionId ->
 "${sessionProperties.redis?.sessionNamespace}:sessions:${sessionId.trim('"')}"
 }.also { keys ->
 logger.debug("Checking Redis keys: {}", keys)
 }
 // First read the hash keys to trigger expiration event
 val script = """
 local keys = ARGV
 local count = 0
 for i, key in ipairs(keys) do
 count = count + redis.call('EXISTS', key)
 end
 return count
 """
 redisOperations.execute(
 RedisScript.of(script, Long::class.java),
 listOf(), // KEYS list is empty
 sessionKeys
 )
 .doOnNext { count ->
 logger.debug("Accessed {} session hash keys", count)
 }
 .flatMap { _ ->
 redisOperations.opsForZSet()
 .remove(redisKeyExpirations, *sessionIds.toTypedArray())
 .doOnSuccess { logger.debug("Successfully removed {} expired sessions", sessionIds.size) }
 .doOnError { e -> logger.error("Error during removal: ${e.message}") }
 }
 .then()
 }
 }
 .subscribeOn(Schedulers.boundedElastic())