Merge remote-tracking branch 'origin/main' into overlay2-rework

config/vkcup/2021/settings.json

@@ -2,5 +2,6 @@
   "type": "cf",
   "apiKey": "$creds.codeforces_key",
   "apiSecret": "$creds.codeforces_secret",
-  "contestId": 1563
+  "contestId": 1563,
+  "asManager": false
 }

src/backend/src/main/kotlin/org/icpclive/data/DataBus.kt

@@ -8,6 +8,7 @@ import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.StateFlow
 import org.icpclive.api.*
 import org.icpclive.api.tunning.AdvancedProperties
+import org.icpclive.scoreboard.ScoreboardAndContestInfo
 import org.icpclive.util.completeOrThrow
 import org.icpclive.service.AnalyticsAction
 import org.icpclive.service.FeaturedRunAction
@@ -23,7 +24,8 @@ object DataBus {
     // flow of run ids that need to be braking news
     val queueFeaturedRunsFlow = CompletableDeferred<FlowCollector<FeaturedRunAction>>()
     val tickerFlow = CompletableDeferred<Flow<TickerEvent>>()
-    private val scoreboardFlow = Array(OptimismLevel.values().size) { CompletableDeferred<Flow<Scoreboard>>() }
+    private val legacyScoreboardFlow = Array(OptimismLevel.entries.size) { CompletableDeferred<Flow<LegacyScoreboard>>() }
+    private val scoreboardFlow = Array(OptimismLevel.entries.size) { CompletableDeferred<Flow<ScoreboardAndContestInfo>>() }
     val statisticFlow = CompletableDeferred<Flow<SolutionsStatistic>>()
     val advancedPropertiesFlow = CompletableDeferred<Flow<AdvancedProperties>>()
     val analyticsActionsFlow = CompletableDeferred<Flow<AnalyticsAction>>()
@@ -43,9 +45,6 @@ object DataBus {
     val teamInterestingFlow = CompletableDeferred<Flow<List<CurrentTeamState>>>()
     val socialEvents = CompletableDeferred<Flow<SocialEvent>>()
 
-    fun setScoreboardEvents(level: OptimismLevel, flow: Flow<Scoreboard>) {
-        scoreboardFlow[level.ordinal].completeOrThrow(flow)
-    }
-
+    fun setScoreboardEvents(level: OptimismLevel, flow: Flow<ScoreboardAndContestInfo>) { scoreboardFlow[level.ordinal].completeOrThrow(flow) }
     suspend fun getScoreboardEvents(level: OptimismLevel) = scoreboardFlow[level.ordinal].await()
 }

src/backend/src/main/kotlin/org/icpclive/overlay/Routing.kt

@@ -2,21 +2,35 @@ package org.icpclive.overlay
 
 import io.ktor.server.routing.*
 import io.ktor.server.websocket.*
+import kotlinx.coroutines.flow.*
 import org.icpclive.Config
 import org.icpclive.api.OptimismLevel
 import org.icpclive.data.DataBus
+import org.icpclive.scoreboard.ScoreboardAndContestInfo
+import org.icpclive.scoreboard.toLegacyScoreboard
 import org.icpclive.util.sendJsonFlow
 
+private inline fun <reified T> Route.setUpScoreboard(crossinline process: (Flow<ScoreboardAndContestInfo>) -> Flow<T>) {
+    webSocket("/normal") { sendJsonFlow(process(DataBus.getScoreboardEvents(OptimismLevel.NORMAL))) }
+    webSocket("/optimistic") { sendJsonFlow(process(DataBus.getScoreboardEvents(OptimismLevel.OPTIMISTIC))) }
+    webSocket("/pessimistic") { sendJsonFlow(process(DataBus.getScoreboardEvents(OptimismLevel.PESSIMISTIC))) }
+}
+
 fun Route.configureOverlayRouting() {
     webSocket("/mainScreen") { sendJsonFlow(DataBus.mainScreenFlow.await()) }
     webSocket("/contestInfo") { sendJsonFlow(DataBus.contestInfoFlow.await()) }
     webSocket("/queue") { sendJsonFlow(DataBus.queueFlow.await()) }
     webSocket("/statistics") { sendJsonFlow(DataBus.statisticFlow.await()) }
     webSocket("/ticker") { sendJsonFlow(DataBus.tickerFlow.await()) }
     route("/scoreboard") {
-        webSocket("/normal") { sendJsonFlow(DataBus.getScoreboardEvents(OptimismLevel.NORMAL)) }
-        webSocket("/optimistic") { sendJsonFlow(DataBus.getScoreboardEvents(OptimismLevel.OPTIMISTIC)) }
-        webSocket("/pessimistic") { sendJsonFlow(DataBus.getScoreboardEvents(OptimismLevel.PESSIMISTIC)) }
+        setUpScoreboard { flow -> flow.map { it.scoreboardSnapshot.toLegacyScoreboard(it.info) } }
+        route("v2") {
+            setUpScoreboard { flow ->
+                flow.withIndex().map {
+                    if (it.index == 0) it.value.scoreboardSnapshot else it.value.scoreboardDiff
+                }
+            }
+        }
     }
     route("/svgAchievement"){
         configureSvgAtchievementRouting(Config.mediaDirectory)

src/backend/src/main/kotlin/org/icpclive/service/ICPCStatisticsService.kt

@@ -5,37 +5,32 @@ import kotlinx.coroutines.flow.*
 import org.icpclive.api.*
 import org.icpclive.util.completeOrThrow
 import org.icpclive.data.DataBus
+import org.icpclive.scoreboard.ScoreboardAndContestInfo
 
 class ICPCStatisticsService {
-    suspend fun run(scoreboardFlow: Flow<Scoreboard>, contestInfoFlow: Flow<ContestInfo>) {
+    suspend fun run(scoreboardFlow: Flow<ScoreboardAndContestInfo>) {
         coroutineScope {
-            combine(
-                scoreboardFlow,
-                contestInfoFlow.map { it.problems.size }.distinctUntilChanged(),
-                ::Pair,
-            ).map { (scoreboard, problemNumber) ->
-                if (scoreboard.rows.isEmpty()) {
-                    SolutionsStatistic(List(problemNumber) { ICPCProblemSolutionsStatistic(0, 0, 0) })
-                } else {
-                    SolutionsStatistic(
-                        List(scoreboard.rows[0].problemResults.size) { problemId ->
-                            var success = 0
-                            var wrong = 0
-                            var pending = 0
+            scoreboardFlow.conflate().map {
+                val problems = it.info.problems.size
+                SolutionsStatistic(
+                    List(problems) { problemId ->
+                        var success = 0
+                        var wrong = 0
+                        var pending = 0
 
-                            for(row in scoreboard.rows) {
-                                val p = row.problemResults[problemId]
-                                require(p is ICPCProblemResult)
-                                success += if (p.isSolved) 1 else 0
-                                wrong += if (!p.isSolved && p.wrongAttempts > 0 && p.pendingAttempts == 0) 1 else 0
-                                pending += if (!p.isSolved && p.pendingAttempts > 0) 1 else 0
-                            }
-
-                            return@List ICPCProblemSolutionsStatistic(success, wrong, pending)
+                        for (row in it.scoreboardSnapshot.rows.values) {
+                            val p = row.problemResults[problemId]
+                            require(p is ICPCProblemResult)
+                            success += if (p.isSolved) 1 else 0
+                            wrong += if (!p.isSolved && p.wrongAttempts > 0 && p.pendingAttempts == 0) 1 else 0
+                            pending += if (!p.isSolved && p.pendingAttempts > 0) 1 else 0
                         }
-                    )
-                }
-            }.stateIn(this)
+
+                        return@List ICPCProblemSolutionsStatistic(success, wrong, pending)
+                    }
+                )
+            }
+                .stateIn(this)
                 .also { DataBus.statisticFlow.completeOrThrow(it) }
         }
     }

src/backend/src/main/kotlin/org/icpclive/service/IOIStatisticsService.kt

@@ -1,58 +1,57 @@
 package org.icpclive.service
 
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.*
 import org.icpclive.api.*
 import org.icpclive.util.completeOrThrow
 import org.icpclive.data.DataBus
+import org.icpclive.scoreboard.ScoreboardAndContestInfo
+import kotlin.time.Duration.Companion.milliseconds
 
 class IOIStatisticsService {
-    suspend fun run(scoreboardFlow: Flow<Scoreboard>, contestInfoFlow: Flow<ContestInfo>) {
+    suspend fun run(scoreboardFlow: Flow<ScoreboardAndContestInfo>) {
         coroutineScope {
-            combine(
-                scoreboardFlow,
-                contestInfoFlow.map { it.problems.size }.distinctUntilChanged(),
-                ::Pair,
-            ).map { (scoreboard, problemNumber) ->
-                if (scoreboard.rows.isEmpty()) {
-                    SolutionsStatistic(List(problemNumber) { IOIProblemSolutionsStatistic(emptyList(), 0) })
-                } else {
-                    SolutionsStatistic(
-                        List(scoreboard.rows[0].problemResults.size) { problemId ->
-                            val listScore = mutableListOf<Double>();
-                            var pending = 0
-                            for(row in scoreboard.rows) {
-                                val p = row.problemResults[problemId]
-                                require(p is IOIProblemResult)
-                                if(p.score != null) {
-                                    listScore.add(p.score!!)
-                                } else {
-                                    ++pending
-                                }
+            scoreboardFlow.conflate().transform {
+                emit(it)
+                delay(100.milliseconds)
+            }.map {
+                val problems = it.info.problems.size
+                SolutionsStatistic(
+                    List(problems) { problemId ->
+                        val listScore = mutableListOf<Double>()
+                        var pending = 0
+                        for (row in it.scoreboardSnapshot.rows.values) {
+                            val p = row.problemResults[problemId]
+                            require(p is IOIProblemResult)
+                            if (p.score != null) {
+                                listScore.add(p.score!!)
+                            } else {
+                                ++pending
                             }
+                        }
 
-                            listScore.sortDescending()
+                        listScore.sortDescending()
 
-                            val entityList = mutableListOf<IOIProblemEntity>()
-                            var currentCount = 0
-                            var currentScore = 0.0
-                            listScore.forEach {
-                                if(it != currentScore) {
-                                    entityList.add(IOIProblemEntity(currentCount, currentScore))
+                        val entityList = mutableListOf<IOIProblemEntity>()
+                        var currentCount = 0
+                        var currentScore = 0.0
+                        listScore.forEach {
+                            if (it != currentScore) {
+                                entityList.add(IOIProblemEntity(currentCount, currentScore))
 
-                                    currentCount = 1
-                                    currentScore = it
-                                } else {
-                                    ++currentCount
-                                }
+                                currentCount = 1
+                                currentScore = it
+                            } else {
+                                ++currentCount
                             }
+                        }
 
-                            entityList.add(IOIProblemEntity(currentCount, currentScore))
+                        entityList.add(IOIProblemEntity(currentCount, currentScore))
 
-                            return@List IOIProblemSolutionsStatistic(entityList, pending)
-                        }
-                    )
-                }
+                        return@List IOIProblemSolutionsStatistic(entityList, pending)
+                    }
+                )
             }.stateIn(this)
                 .also { DataBus.statisticFlow.completeOrThrow(it) }
         }

src/backend/src/main/kotlin/org/icpclive/service/ScoreboardService.kt

@@ -4,24 +4,19 @@ import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.*
 import org.icpclive.api.*
 import org.icpclive.cds.ContestUpdate
-import org.icpclive.cds.adapters.stateGroupedByTeam
 import org.icpclive.data.DataBus
-import org.icpclive.scoreboard.calculateScoreboard
+import org.icpclive.scoreboard.*
 import org.icpclive.util.getLogger
-import kotlin.time.Duration
-
 
 class ScoreboardService(val optimismLevel: OptimismLevel) {
-
     suspend fun run(
         updates: Flow<ContestUpdate>,
     ) {
         logger.info("Scoreboard service for optimismLevel=${optimismLevel} started")
         coroutineScope {
-            updates.stateGroupedByTeam()
-                .calculateScoreboard(optimismLevel)
-                .stateIn(this, SharingStarted.Eagerly, Scoreboard(Duration.ZERO, emptyList()))
-                .let { DataBus.setScoreboardEvents(optimismLevel, it) }
+            val scoreboardFlow = updates.calculateScoreboard(optimismLevel)
+                .shareIn(this, SharingStarted.Eagerly, replay = 1)
+            DataBus.setScoreboardEvents(optimismLevel, scoreboardFlow)
         }
     }
 

src/backend/src/main/kotlin/org/icpclive/service/ServiceRoot.kt

@@ -25,30 +25,15 @@ fun CoroutineScope.launchServices(loader: Flow<ContestUpdate>) {
                 launch { ScoreboardService(OptimismLevel.OPTIMISTIC).run(loaded) }
                 launch { ScoreboardService(OptimismLevel.PESSIMISTIC).run(loaded) }
                 launch { ScoreboardService(OptimismLevel.NORMAL).run(loaded) }
-                launch {
-                    ICPCStatisticsService().run(
-                        DataBus.getScoreboardEvents(OptimismLevel.NORMAL),
-                        DataBus.contestInfoFlow.await()
-                    )
-                }
+                launch { ICPCStatisticsService().run(DataBus.getScoreboardEvents(OptimismLevel.NORMAL)) }
             }
 
             ContestResultType.IOI -> {
                 launch { ScoreboardService(OptimismLevel.NORMAL).run(loaded) }
                 DataBus.setScoreboardEvents(OptimismLevel.OPTIMISTIC, DataBus.getScoreboardEvents(OptimismLevel.NORMAL))
-                DataBus.setScoreboardEvents(
-                    OptimismLevel.PESSIMISTIC,
-                    DataBus.getScoreboardEvents(OptimismLevel.NORMAL)
-                )
-
-                //DataBus.statisticFlow.completeOrThrow(emptyFlow())
-                launch {
-                    IOIStatisticsService().run(
-                        DataBus.getScoreboardEvents(OptimismLevel.NORMAL),
-                        DataBus.contestInfoFlow.await()
-                    )
-                }
+                DataBus.setScoreboardEvents(OptimismLevel.PESSIMISTIC, DataBus.getScoreboardEvents(OptimismLevel.NORMAL))
 
+                launch { IOIStatisticsService().run(DataBus.getScoreboardEvents(OptimismLevel.NORMAL)) }
             }
         }
         val generatedAnalyticsMessages = Config.analyticsTemplatesFile?.let {

src/backend/src/main/kotlin/org/icpclive/service/TeamSpotlightService.kt

@@ -1,12 +1,14 @@
 package org.icpclive.service
 
+import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.serialization.Serializable
 import org.icpclive.api.*
 import org.icpclive.data.DataBus
+import org.icpclive.scoreboard.ScoreboardAndContestInfo
 import org.icpclive.util.getLogger
 import org.icpclive.util.intervalFlow
 import kotlin.time.Duration.Companion.seconds
@@ -119,47 +121,50 @@ class TeamSpotlightService(
     suspend fun run(
         info: StateFlow<ContestInfo>,
         runs: Flow<RunInfo>,
-        scoreboard: Flow<Scoreboard>,
+        scoreboard: Flow<ScoreboardAndContestInfo>,
         // analyticsMessage: Flow<AnalyticsMessage>
         addScoreRequests: Flow<AddTeamScoreRequest>? = null,
     ) {
         val runIds = mutableSetOf<Int>()
-        merge(
-            intervalFlow(settings.scoreboardPushInterval).map { ScoreboardPushTrigger },
-            runs.filter { !it.isHidden },
-            addScoreRequests ?: emptyFlow(),
-            DataBus.socialEvents.await(),
-        ).collect { update ->
-            when (update) {
-                is RunInfo -> {
-                    if (update.time + 60.seconds > info.value.currentContestTime) {
-                        if (update.isJudged || update.id !in runIds) {
-                            runIds += update.id
-                            mutex.withLock {
-                                getTeamInQueue(update.teamId).addAccent(TeamRunAccent(update))
+        coroutineScope {
+            val scoreboardState = scoreboard.map { it.scoreboardSnapshot }.stateIn(this)
+            merge(
+                intervalFlow(settings.scoreboardPushInterval).map { ScoreboardPushTrigger },
+                runs.filter { !it.isHidden },
+                addScoreRequests ?: emptyFlow(),
+                DataBus.socialEvents.await(),
+            ).collect { update ->
+                when (update) {
+                    is RunInfo -> {
+                        if (update.time + 60.seconds > info.value.currentContestTime) {
+                            if (update.isJudged || update.id !in runIds) {
+                                runIds += update.id
+                                mutex.withLock {
+                                    getTeamInQueue(update.teamId).addAccent(TeamRunAccent(update))
+                                }
                             }
                         }
                     }
-                }
 
-                is ScoreboardPushTrigger -> {
-                    scoreboard.first().rows.filter { it.rank <= settings.scoreboardLowestRank }.forEach {
-                        mutex.withLock {
-                            getTeamInQueue(it.teamId).addAccent(TeamScoreboardPlace(it.rank))
+                    is ScoreboardPushTrigger -> {
+                        scoreboardState.value.order.zip(scoreboardState.value.ranks).takeWhile { it.second <= settings.scoreboardLowestRank }.forEach {
+                            mutex.withLock {
+                                getTeamInQueue(it.first).addAccent(TeamScoreboardPlace(it.second))
+                            }
                         }
                     }
-                }
 
-                is AddTeamScoreRequest -> {
-                    mutex.withLock {
-                        getTeamInQueue(update.teamId).addAccent(ExternalScoreAddAccent(update.score))
+                    is AddTeamScoreRequest -> {
+                        mutex.withLock {
+                            getTeamInQueue(update.teamId).addAccent(ExternalScoreAddAccent(update.score))
+                        }
                     }
-                }
 
-                is SocialEvent -> {
-                    update.teamIds.forEach { teamId ->
-                        mutex.withLock {
-                            getTeamInQueue(teamId).addAccent(SocialEventAccent)
+                    is SocialEvent -> {
+                        update.teamIds.forEach { teamId ->
+                            mutex.withLock {
+                                getTeamInQueue(teamId).addAccent(SocialEventAccent)
+                            }
                         }
                     }
                 }