New persistence system

.floo

@@ -0,0 +1,3 @@
+{
+    "url": "https://floobits.com/Frederikam/FredBoat"
+}

.flooignore

@@ -0,0 +1,11 @@
+extern
+node_modules
+tmp
+vendor
+.idea/workspace.xml
+.idea/misc.xml
+.idea
+
+fredboat.yml
+fredboat.yaml
+commons.yml

FredBoat/build.gradle

@@ -49,10 +49,13 @@ dependencies {
     }
 
     compile group: 'org.springframework.boot', name: 'spring-boot-starter-webflux', version: springBootVersion
+    compile group: 'org.springframework.boot', name: 'spring-boot-starter-data-mongodb-reactive', version: springBootVersion
+    compile group: 'org.springframework', name: 'spring-websocket', version: springWebSocketVersion
     compile group: 'it.unimi.dsi', name: 'fastutil', version: fastUtilVersion
     compile group: 'com.google.code.gson', name: 'gson', version: gsonVersion
     compile group: 'org.jetbrains.kotlin', name: 'kotlin-reflect', version: kotlinVersion
-    compile group: 'org.jetbrains.kotlinx', name: 'kotlinx-coroutines-reactor', version: '1.0.1'
+    compile group: 'org.jetbrains.kotlinx', name: 'kotlinx-coroutines-reactor', version: '1.1.1'
+    compile group: 'org.jetbrains.kotlinx', name: 'kotlinx-coroutines-jdk8', version: '1.1.1'
     compile group: 'org.springframework.amqp', name: 'spring-rabbit', version: amqpVersion
     compile group: 'com.fasterxml.jackson.module', name: 'jackson-module-kotlin', version: jacksonKotlinVersion
 
@@ -72,6 +75,7 @@ dependencies {
     testCompile group: 'com.palantir.docker.compose', name: 'docker-compose-rule-junit4', version: dockerComposeRuleVersion
     testCompile group: 'org.springframework.boot', name: 'spring-boot-starter-test', version: springBootVersion
     testCompile group: 'org.postgresql', name: 'postgresql', version: postgresJdbcVersion
+    testCompile 'io.projectreactor:reactor-test:3.1.6.RELEASE'
     optional group: 'org.springframework.boot', name: 'spring-boot-configuration-processor', version: springBootVersion
 
     /* Explicitly set these, instead of getting outdated transitive ones */

FredBoat/src/main/java/fredboat/agent/GuildCacheInvalidationAgent.kt

@@ -3,18 +3,23 @@ package fredboat.agent
 import com.fredboat.sentinel.entities.GuildUnsubscribeRequest
 import fredboat.audio.lavalink.SentinelLavalink
 import fredboat.audio.player.PlayerRegistry
+import fredboat.db.mongo.PlayerRepository
+import fredboat.db.mongo.convertAndSave
 import fredboat.sentinel.GuildCache
 import fredboat.sentinel.InternalGuild
 import lavalink.client.io.Link
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Controller
+import reactor.core.publisher.Mono
+import java.time.Duration
 import java.util.concurrent.TimeUnit
 
 @Controller
 class GuildCacheInvalidationAgent(
         val guildCache: GuildCache,
         private val playerRegistry: PlayerRegistry,
+        private val playerRepository: PlayerRepository,
         private val lavalink: SentinelLavalink
 ) : FredBoatAgent("cache-invalidator", 5, TimeUnit.MINUTES) {
 
@@ -59,14 +64,19 @@ class GuildCacheInvalidationAgent(
     }
 
     fun invalidateGuild(guild: InternalGuild) {
-        try {
-            playerRegistry.destroyPlayer(guild)
-            lavalink.getExistingLink(guild)?.destroy()
-        } catch (e: Exception) {
-            log.error("Got exception when invaliding GuildPlayer and Link for {}", guild)
+        val mono = playerRegistry.getExisting(guild)?.let {
+            playerRepository.convertAndSave(it).timeout(Duration.ofSeconds(60))
+        } ?: Mono.empty()
+        mono.subscribe {
+            try {
+                playerRegistry.destroyPlayer(guild)
+                lavalink.getExistingLink(guild)?.destroy()
+            } catch (e: Exception) {
+                log.error("Got exception when invaliding GuildPlayer and Link for {}", guild)
+            }
+            guild.sentinel.sendAndForget(guild.routingKey, GuildUnsubscribeRequest(guild.id))
+            guildCache.cache.remove(guild.id)
         }
-        guild.sentinel.sendAndForget(guild.routingKey, GuildUnsubscribeRequest(guild.id))
-        guildCache.cache.remove(guild.id)
     }
 
 }

FredBoat/src/main/java/fredboat/agent/VoiceChannelCleanupAgent.kt

@@ -84,7 +84,7 @@ class VoiceChannelCleanupAgent(
                     } else if (vc != null && isBeingUsed(vc)) {
                         vcLastUsed[vcId] = System.currentTimeMillis()
                     } else {
-                        // Not being used! But there are users in te VC. Check if we've been here for a while.
+                        // Not being used! But there are users in the VC. Check if we've been here for a while.
 
                         vcId?.let {
                             if (!vcLastUsed.containsKey(vcId)) {

FredBoat/src/main/java/fredboat/api/MetricsController.java

@@ -69,11 +69,9 @@ private String buildAnswer(String[] includedParam) throws IOException {
         }
 
         Writer writer = new StringWriter();
-        try {
+        try (writer) {
             TextFormat.write004(writer, registry.filteredMetricFamilySamples(params));
             writer.flush();
-        } finally {
-            writer.close();
         }
 
         return writer.toString();

FredBoat/src/main/java/fredboat/audio/lavalink/SentinelLavalink.kt

@@ -1,39 +1,77 @@
 package fredboat.audio.lavalink
 
 import com.fredboat.sentinel.entities.VoiceServerUpdate
+import fredboat.audio.player.PlayerRegistry
 import fredboat.config.idString
 import fredboat.config.property.AppConfig
 import fredboat.config.property.LavalinkConfig
+import fredboat.db.mongo.ActivityMetricsController
 import fredboat.sentinel.Guild
+import fredboat.sentinel.GuildCache
 import fredboat.sentinel.Sentinel
+import fredboat.util.DiscordUtil
+import fredboat.ws.UserSessionHandler
 import lavalink.client.io.Lavalink
 import lavalink.client.io.metrics.LavalinkCollector
 import org.json.JSONObject
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Service
+import java.time.Duration
+import java.util.concurrent.TimeoutException
 
+@Suppress("ImplicitThis", "LeakingThis")
 @Service
 class SentinelLavalink(
         val sentinel: Sentinel,
+        val guildCache: GuildCache,
         val appConfig: AppConfig,
+        val activityMetrics: ActivityMetricsController,
+        val userSessionHandler: UserSessionHandler,
         lavalinkConfig: LavalinkConfig
 ) : Lavalink<SentinelLink>(
         sentinel.selfUser.idString,
         appConfig.shardCount
 ) {
 
+    lateinit var playerRegistry: PlayerRegistry
+    private val resumeKey = "fredboat-${sentinel.selfUser.idString}"
+
     companion object {
         lateinit var INSTANCE: SentinelLavalink
+        private const val DEFAULT_RESUME_TIMEOUT = 300 // 5 mins
+        private val log: Logger = LoggerFactory.getLogger(SentinelLavalink::class.java)
     }
 
     init {
-        @Suppress("LeakingThis")
         INSTANCE = this
-        lavalinkConfig.nodes.forEach { addNode(it.name, it.uri, it.password) }
-        @Suppress("LeakingThis")
+        setHoldEvents(true)
+        lavalinkConfig.nodes.forEach {
+            addNode(it.name, it.uri, it.password, resumeKey)
+                    .setResuming(resumeKey, DEFAULT_RESUME_TIMEOUT)
+        }
         LavalinkCollector(this).register<LavalinkCollector>()
     }
 
-    override fun buildNewLink(guildId: String) = SentinelLink(this, guildId)
+    override fun buildNewLink(guildId: String): SentinelLink {
+        val shardId = DiscordUtil.getShardId(guildId.toLong(), appConfig)
+        val shardMono = sentinel.tracker.awaitHello(shardId)
+        val guildMono = guildCache.getGuildMono(guildId.toLong()).doOnSuccess { guild ->
+            if (guild == null) {
+                log.warn("Built link for non-existing guild. This should not happen.")
+                return@doOnSuccess
+            }
+            playerRegistry.getOrCreate(guild).subscribe()
+        }
+
+        shardMono.timeout(Duration.ofSeconds(120))
+                .onErrorMap(TimeoutException::class.java) {
+                    RuntimeException("Attempted to resume Lavalink, but Sentinel hello timed out")
+                }.then(guildMono)
+                .subscribe()
+
+        return SentinelLink(this, guildId)
+    }
 
     fun getLink(guild: Guild) = getLink(guild.id.toString())
     fun getExistingLink(guild: Guild) = getExistingLink(guild.idString)

FredBoat/src/main/java/fredboat/audio/lavalink/SentinelLink.kt

@@ -4,6 +4,7 @@ import com.fredboat.sentinel.entities.AudioQueueRequest
 import com.fredboat.sentinel.entities.AudioQueueRequestEnum.*
 import fredboat.perms.InsufficientPermissionException
 import fredboat.perms.Permission.*
+import fredboat.sentinel.Guild
 import fredboat.sentinel.VoiceChannel
 import lavalink.client.io.Link
 import org.slf4j.Logger
@@ -59,6 +60,16 @@ class SentinelLink(val lavalink: SentinelLavalink, guildId: String) : Link(laval
         queueAudioConnect(channel.id)
     }
 
+    fun getChannel(guild: Guild): VoiceChannel? {
+        val id = channel ?: return null
+        val vc = guild.getVoiceChannel(id.toLong())
+
+        if (vc != null) return vc
+
+        log.warn("Lavalink appears to be connected to vc $id, which doesn't seem to exist in $guild")
+        return null
+    }
+
     override fun onVoiceWebSocketClosed(code: Int, reason: String, byRemote: Boolean) {
         val by = if (byRemote) "Discord" else "LLS"
         log.info("{}: Lavalink voice WS closed by {}, code {}: {}", guild, by, code, reason)
@@ -75,8 +86,8 @@ class SentinelLink(val lavalink: SentinelLavalink, guildId: String) : Link(laval
                 queueAudioConnect(vc.toLong())
             } else {
                 log.warn("{}: Got WS close code $code twice within $MIN_RETRY_INTERVAL ms, disconnecting " +
-                        " to prevent bouncing and getting stuck...", guild)
-                queueAudioDisconnect()
+                        "to prevent bouncing and getting stuck...", guild)
+                disconnect()
             }
         }
     }