Implement akka-http-session RefreshTokenStorage backed by an actor (fix

#372)

server/src/main/scala/com.olegych.scastie.web/ServerMain.scala

@@ -44,7 +44,7 @@ object ServerMain {
     implicit val materializer: ActorMaterializer = ActorMaterializer()
 
     val github = new Github
-    val session = new GithubUserSession
+    val session = new GithubUserSession(system)
     val userDirectives = new UserDirectives(session)
 
     val progressActor =

server/src/main/scala/com.olegych.scastie.web/oauth2/GithubUserSession.scala

@@ -8,19 +8,19 @@ import com.softwaremill.session._
 import com.typesafe.config.ConfigFactory
 
 import scala.collection.parallel.mutable.ParTrieMap
-import scala.concurrent.ExecutionContext
 import com.typesafe.scalalogging.Logger
 
 import scala.util.Try
 import java.util.UUID
 import java.nio.file._
 
+import akka.actor.ActorSystem
+
 import scala.collection.JavaConverters._
 
 import System.{lineSeparator => nl}
 
-class GithubUserSession()(implicit val executionContext: ExecutionContext) {
-
+class GithubUserSession(system: ActorSystem) {
   val logger = Logger("GithubUserSession")
 
   private val configuration =
@@ -49,11 +49,7 @@ class GithubUserSession()(implicit val executionContext: ExecutionContext) {
       (id: String) => Try { UUID.fromString(id) }
     )
   implicit val sessionManager = new SessionManager[UUID](sessionConfig)
-  implicit val refreshTokenStorage = new InMemoryRefreshTokenStorage[UUID] {
-    def log(msg: String): Unit =
-      if (msg.startsWith("Looking up token for selector")) () // boring
-      else logger.info(msg)
-  }
+  implicit val refreshTokenStorage = new ActorRefreshTokenStorage(system)
 
   private def readSessionsFile(): Array[(UUID, User)] = {
     if (Files.exists(usersSessions)) {

server/src/main/scala/com.olegych.scastie.web/oauth2/InMemoryRefreshTokenStorage.scala

@@ -0,0 +1,69 @@
+package com.olegych.scastie.web.oauth2
+
+import com.softwaremill.session.{
+  RefreshTokenData,
+  RefreshTokenStorage,
+  RefreshTokenLookupResult
+}
+import akka.actor.{Actor, ActorSystem, Props}
+import akka.pattern.ask
+import akka.util.Timeout
+
+import scala.concurrent.Future
+import scala.concurrent.duration._
+import scala.collection.mutable
+
+import java.util.UUID
+
+private[oauth2] case class SessionStorage(session: UUID,
+                                          tokenHash: String,
+                                          expires: Long)
+
+class ActorRefreshTokenStorage(system: ActorSystem)
+    extends RefreshTokenStorage[UUID] {
+  import system.dispatcher
+  implicit private val timeout = Timeout(10.seconds)
+  private val impl = system.actorOf(Props(new ActorRefreshTokenStorageImpl()))
+
+  def lookup(selector: String): Future[Option[RefreshTokenLookupResult[UUID]]] =
+    (impl ? Lookup(selector)).mapTo[Option[RefreshTokenLookupResult[UUID]]]
+
+  def store(data: RefreshTokenData[UUID]): Future[Unit] = {
+    impl ! Store(data)
+    Future.successful(())
+  }
+  def remove(selector: String): Future[Unit] = {
+    impl ! Remove(selector)
+    Future.successful(())
+  }
+  def schedule[S](after: Duration)(op: => Future[S]): Unit = {
+    after match {
+      case finite: FiniteDuration => system.scheduler.scheduleOnce(finite)(op)
+      case _: Duration.Infinite   => ()
+    }
+  }
+}
+
+private[oauth2] case class Lookup(selector: String)
+private[oauth2] case class Store(data: RefreshTokenData[UUID])
+private[oauth2] case class Remove(selector: String)
+
+class ActorRefreshTokenStorageImpl() extends Actor {
+  private val storage = mutable.Map[String, SessionStorage]()
+  override def receive: Receive = {
+    case Lookup(selector) =>
+      val lookupResult =
+        storage
+          .get(selector)
+          .map(
+            s =>
+              RefreshTokenLookupResult(s.tokenHash, s.expires, () => s.session)
+          )
+      sender ! lookupResult
+    case Store(data) =>
+      storage.put(data.selector,
+                  SessionStorage(data.forSession, data.tokenHash, data.expires))
+    case Remove(selector) =>
+      storage.remove(selector)
+  }
+}

server/src/main/scala/com.olegych.scastie.web/oauth2/UserDirectives.scala

@@ -9,7 +9,11 @@ import com.softwaremill.session._
 import SessionDirectives._
 import SessionOptions._
 
-class UserDirectives(session: GithubUserSession) {
+import scala.concurrent.ExecutionContext
+
+class UserDirectives(
+    session: GithubUserSession
+)(implicit val executionContext: ExecutionContext) {
   import session._
 
   def optionalLogin: Directive1[Option[User]] =

server/src/main/scala/com.olegych.scastie.web/routes/OAuth2Routes.scala

@@ -3,18 +3,24 @@ package web
 package routes
 
 import oauth2._
+
 import com.softwaremill.session.SessionDirectives._
 import com.softwaremill.session.SessionOptions._
 import com.softwaremill.session.CsrfDirectives._
 import com.softwaremill.session.CsrfOptions._
+
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.model.Uri.Query
 import akka.http.scaladsl.model.StatusCodes.TemporaryRedirect
 import akka.http.scaladsl.model.headers.Referer
 import akka.http.scaladsl.server.Directives._
 import akka.http.scaladsl.server.Route
 
-class OAuth2Routes(github: Github, session: GithubUserSession) {
+import scala.concurrent.ExecutionContext
+
+class OAuth2Routes(github: Github, session: GithubUserSession)(
+    implicit val executionContext: ExecutionContext
+) {
   import session._
 
   val routes: Route =