add scheduler service

fupelaqu · fupelaqu · commit 5d86db11c742 · 2023-02-03T14:14:10.000+01:00
diff --git a/api/src/main/scala/app/softnetwork/scheduler/api/SchedulerApi.scala b/api/src/main/scala/app/softnetwork/scheduler/api/SchedulerApi.scala
@@ -1,25 +1,29 @@
 package app.softnetwork.scheduler.api
 
 import akka.actor.typed.ActorSystem
-import app.softnetwork.persistence.jdbc.query.JdbcSchema.SchemaType
+import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
 import app.softnetwork.persistence.jdbc.query.{JdbcJournalProvider, JdbcSchema, JdbcSchemaProvider}
 import app.softnetwork.scheduler.handlers.SchedulerHandler
 import app.softnetwork.scheduler.launch.SchedulerApplication
 import app.softnetwork.scheduler.persistence.query.Entity2SchedulerProcessorStream
 
-trait SchedulerApi extends SchedulerApplication with JdbcSchemaProvider {
+import scala.concurrent.Future
 
-  def jdbcSchemaType: SchemaType = this.schemaType
+trait SchedulerApi extends SchedulerApplication with JdbcSchemaProvider {
 
   override def entity2SchedulerProcessorStream: ActorSystem[_] => Entity2SchedulerProcessorStream =
     sys =>
       new Entity2SchedulerProcessorStream()
         with SchedulerHandler
         with JdbcJournalProvider
         with JdbcSchemaProvider {
-        override lazy val schemaType: JdbcSchema.SchemaType = jdbcSchemaType
+        override lazy val schemaType: JdbcSchema.SchemaType = SchedulerApi.this.schemaType
 
         override implicit def system: ActorSystem[_] = sys
       }
 
+  override def grpcServices
+    : ActorSystem[_] => Seq[PartialFunction[HttpRequest, Future[HttpResponse]]] = system =>
+    Seq(SchedulerServiceApiHandler.partial(schedulerServer(system))(system))
+
 }
diff --git a/build.sbt b/build.sbt
@@ -31,7 +31,7 @@ ThisBuild / organization := "app.softnetwork"
 
 name := "scheduler"
 
-ThisBuild / version := "0.2.4"
+ThisBuild / version := "0.2.5"
 
 ThisBuild / scalaVersion := "2.12.15"
 
diff --git a/common/build.sbt b/common/build.sbt
@@ -8,6 +8,8 @@ name := "scheduler-common"
 
 libraryDependencies ++= Seq(
   "com.markatta" %% "akron" % "1.2" excludeAll(ExclusionRule(organization = "com.typesafe.akka"), ExclusionRule(organization = "org.scala-lang.modules")),
+  // session
+  "app.softnetwork.session" %% "session-core" % Versions.session,
   "app.softnetwork.api" %% "generic-server-api" % Versions.server,
   "app.softnetwork.protobuf" %% "scalapb-extensions" % "0.1.5"
 )
diff --git a/common/src/main/resources/softnetwork-scheduler.conf b/common/src/main/resources/softnetwork-scheduler.conf
@@ -1,5 +1,6 @@
 softnetwork{
   scheduler {
+    path = "scheduler"
     # id = "*"
     akka-node-role = scheduler
     reset-scheduler {
diff --git a/common/src/main/scala/app/softnetwork/scheduler/config/SchedulerSettings.scala b/common/src/main/scala/app/softnetwork/scheduler/config/SchedulerSettings.scala
@@ -19,6 +19,8 @@ object SchedulerSettings extends StrictLogging {
       case Right(schedulerConfig) => schedulerConfig
     }
 
+  val SchedulerPath: String = config.getString("softnetwork.scheduler.path")
+
   def tag(persistenceId: String): String = s"scheduler-to-$persistenceId"
 }
 
diff --git a/core/src/main/scala/app/softnetwork/scheduler/launch/SchedulerApplication.scala b/core/src/main/scala/app/softnetwork/scheduler/launch/SchedulerApplication.scala
@@ -1,18 +1,8 @@
 package app.softnetwork.scheduler.launch
 
-import akka.actor.typed.ActorSystem
-import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
-import app.softnetwork.api.server.launch.HealthCheckApplication
+import app.softnetwork.api.server.launch.Application
 import app.softnetwork.persistence.query.SchemaProvider
-import app.softnetwork.scheduler.api.SchedulerServiceApiHandler
 
-import scala.concurrent.Future
-
-trait SchedulerApplication extends HealthCheckApplication with SchedulerGuardian {
+trait SchedulerApplication extends Application with SchedulerRoutes {
   _: SchemaProvider =>
-
-  override def grpcServices
-    : ActorSystem[_] => Seq[PartialFunction[HttpRequest, Future[HttpResponse]]] = system =>
-    Seq(SchedulerServiceApiHandler.partial(schedulerServer(system))(system))
-
 }
diff --git a/core/src/main/scala/app/softnetwork/scheduler/launch/SchedulerGuardian.scala b/core/src/main/scala/app/softnetwork/scheduler/launch/SchedulerGuardian.scala
@@ -11,11 +11,12 @@ import app.softnetwork.scheduler.persistence.query.{
   Scheduler2EntityProcessorStream
 }
 import app.softnetwork.scheduler.persistence.typed.SchedulerBehavior
+import app.softnetwork.session.launch.SessionGuardian
 import com.typesafe.scalalogging.StrictLogging
 
 import scala.util.{Failure, Success, Try}
 
-trait SchedulerGuardian extends PersistenceGuardian with StrictLogging {
+trait SchedulerGuardian extends SessionGuardian with StrictLogging {
   _: SchemaProvider =>
 
   def schedulerEntities: ActorSystem[_] => Seq[PersistentEntity[_, _, _, _]] = _ =>
@@ -25,7 +26,8 @@ trait SchedulerGuardian extends PersistenceGuardian with StrictLogging {
 
   /** initialize all entities
     */
-  override def entities: ActorSystem[_] => Seq[PersistentEntity[_, _, _, _]] = schedulerEntities
+  override def entities: ActorSystem[_] => Seq[PersistentEntity[_, _, _, _]] = system =>
+    sessionEntities(system) ++ schedulerEntities(system)
 
   def entity2SchedulerProcessorStream: ActorSystem[_] => Entity2SchedulerProcessorStream
 
diff --git a/core/src/main/scala/app/softnetwork/scheduler/launch/SchedulerRoutes.scala b/core/src/main/scala/app/softnetwork/scheduler/launch/SchedulerRoutes.scala
@@ -0,0 +1,15 @@
+package app.softnetwork.scheduler.launch
+
+import akka.actor.typed.ActorSystem
+import akka.http.scaladsl.server.Route
+import app.softnetwork.api.server.ApiRoutes
+import app.softnetwork.persistence.query.SchemaProvider
+import app.softnetwork.scheduler.service.SchedulerService
+
+trait SchedulerRoutes extends ApiRoutes with SchedulerGuardian { _: SchemaProvider =>
+
+  def schedulerService: ActorSystem[_] => SchedulerService = system => SchedulerService(system)
+
+  override def apiRoutes(system: ActorSystem[_]): Route = schedulerService(system).route
+
+}
diff --git a/core/src/main/scala/app/softnetwork/scheduler/service/SchedulerService.scala b/core/src/main/scala/app/softnetwork/scheduler/service/SchedulerService.scala
@@ -0,0 +1,143 @@
+package app.softnetwork.scheduler.service
+
+import akka.actor.typed.ActorSystem
+import akka.http.scaladsl.model.{HttpResponse, StatusCodes}
+import akka.http.scaladsl.server.{Directives, Route}
+import app.softnetwork.api.server._
+import app.softnetwork.persistence.service.Service
+import app.softnetwork.scheduler.config.SchedulerSettings
+import app.softnetwork.scheduler.handlers.{SchedulerDao, SchedulerHandler}
+import app.softnetwork.scheduler.message.{
+  AddCronTab,
+  AddSchedule,
+  CronTabAdded,
+  CronTabRemoved,
+  RemoveCronTab,
+  RemoveSchedule,
+  ScheduleAdded,
+  ScheduleRemoved,
+  SchedulerCommand,
+  SchedulerCommandResult,
+  SchedulerErrorMessage
+}
+import app.softnetwork.scheduler.model._
+import app.softnetwork.serialization._
+import app.softnetwork.session.service.SessionService
+import com.softwaremill.session.CsrfDirectives.randomTokenCsrfProtection
+import com.softwaremill.session.CsrfOptions.checkHeader
+import com.typesafe.scalalogging.StrictLogging
+import de.heikoseeberger.akkahttpjson4s.Json4sSupport
+import org.json4s.jackson.Serialization
+import org.json4s.{jackson, Formats}
+import org.softnetwork.session.model.Session
+
+trait SchedulerService
+    extends SessionService
+    with Directives
+    with DefaultComplete
+    with Json4sSupport
+    with StrictLogging
+    with Service[SchedulerCommand, SchedulerCommandResult]
+    with SchedulerDao
+    with SchedulerHandler {
+
+  implicit def serialization: Serialization.type = jackson.Serialization
+
+  override implicit def formats: Formats = commonFormats
+
+  import Session._
+
+  val route: Route = {
+    pathPrefix(SchedulerSettings.SchedulerPath) {
+      // check anti CSRF token
+      randomTokenCsrfProtection(checkHeader) {
+        // check if a session exists
+        _requiredSession(ec) { session =>
+          // only administrators should be allowed to access this resource
+          if (session.admin) {
+            schedules ~ crons ~ scheduler
+          } else {
+            complete(HttpResponse(StatusCodes.Unauthorized))
+          }
+        }
+      }
+    }
+  }
+
+  lazy val schedules: Route = pathPrefix("schedules") {
+    post {
+      entity(as[Schedule]) { schedule =>
+        run(
+          SchedulerSettings.SchedulerConfig.id.getOrElse("*"),
+          AddSchedule(schedule)
+        ) completeWith {
+          case r: ScheduleAdded => complete(HttpResponse(StatusCodes.OK, entity = r))
+          case r: SchedulerErrorMessage =>
+            complete(HttpResponse(StatusCodes.BadRequest, entity = r))
+          case _ => complete(HttpResponse(StatusCodes.BadRequest))
+        }
+      }
+    } ~ delete {
+      entity(as[RemoveSchedule]) { cmd =>
+        run(SchedulerSettings.SchedulerConfig.id.getOrElse("*"), cmd) completeWith {
+          case r: ScheduleRemoved => complete(HttpResponse(StatusCodes.OK, entity = r))
+          case r: SchedulerErrorMessage =>
+            complete(HttpResponse(StatusCodes.BadRequest, entity = r))
+          case _ => complete(HttpResponse(StatusCodes.BadRequest))
+        }
+      }
+    }
+  }
+
+  lazy val crons: Route = pathPrefix("crons") {
+    post {
+      entity(as[CronTab]) { cronTab =>
+        run(SchedulerSettings.SchedulerConfig.id.getOrElse("*"), AddCronTab(cronTab)) completeWith {
+          case r: CronTabAdded => complete(HttpResponse(StatusCodes.OK, entity = r))
+          case r: SchedulerErrorMessage =>
+            complete(HttpResponse(StatusCodes.BadRequest, entity = r))
+          case _ => complete(HttpResponse(StatusCodes.BadRequest))
+        }
+      }
+    } ~ delete {
+      entity(as[RemoveCronTab]) { cmd =>
+        run(SchedulerSettings.SchedulerConfig.id.getOrElse("*"), cmd) completeWith {
+          case r: CronTabRemoved => complete(HttpResponse(StatusCodes.OK, entity = r))
+          case r: SchedulerErrorMessage =>
+            complete(HttpResponse(StatusCodes.BadRequest, entity = r))
+          case _ => complete(HttpResponse(StatusCodes.BadRequest))
+        }
+      }
+    }
+  }
+
+  lazy val scheduler: Route =
+    get {
+      pathSuffix(Segment) { value =>
+        getScheduler(Some(value))
+      } ~ pathEnd {
+        getScheduler(None)
+      }
+    }
+
+  private[this] def getScheduler(scheduler: Option[String]): Route = {
+    loadScheduler(scheduler) completeWith {
+      case Some(s) =>
+        complete(
+          HttpResponse(
+            StatusCodes.OK,
+            entity = s
+          )
+        )
+      case _ => complete(HttpResponse(StatusCodes.NotFound))
+    }
+  }
+}
+
+object SchedulerService {
+  def apply(sys: ActorSystem[_]): SchedulerService = {
+    new SchedulerService {
+      override implicit def system: ActorSystem[_] = sys
+    }
+  }
+}
diff --git a/project/src/main/scala/app/softnetwork/sbt/build/Versions.scala b/project/src/main/scala/app/softnetwork/sbt/build/Versions.scala
@@ -8,5 +8,7 @@ object Versions {
 
   val server = "0.2.8"
 
+  val session = "0.1.3"
+
   val scalatest = "3.1.1"
 }
diff --git a/testkit/build.sbt b/testkit/build.sbt
@@ -8,5 +8,6 @@ name := "scheduler-testkit"
 
 libraryDependencies ++= Seq(
   "app.softnetwork.api" %% "generic-server-api-testkit" % Versions.server,
+  "app.softnetwork.session" %% "session-testkit" % Versions.session,
   "app.softnetwork.persistence" %% "persistence-core-testkit" % Versions.genericPersistence
 )
diff --git a/testkit/src/main/scala/app/softnetwork/scheduler/scalatest/SchedulerRouteTestKit.scala b/testkit/src/main/scala/app/softnetwork/scheduler/scalatest/SchedulerRouteTestKit.scala
@@ -0,0 +1,68 @@
+package app.softnetwork.scheduler.scalatest
+
+import akka.actor.typed.ActorSystem
+import akka.http.scaladsl.model.StatusCodes
+import akka.http.scaladsl.server.Route
+import app.softnetwork.api.server.config.ServerSettings.RootPath
+import app.softnetwork.scheduler.api.SchedulerGrpcServices
+import app.softnetwork.scheduler.config.SchedulerSettings.SchedulerPath
+import app.softnetwork.scheduler.launch.SchedulerRoutes
+import app.softnetwork.scheduler.message.{RemoveCronTab, RemoveSchedule}
+import app.softnetwork.scheduler.model.{CronTab, Schedule, Scheduler}
+import app.softnetwork.serialization._
+import app.softnetwork.session.scalatest.{SessionServiceRoute, SessionTestKit}
+import org.scalatest.{Assertion, Suite}
+
+trait SchedulerRouteTestKit
+    extends SessionTestKit
+    with SchedulerTestKit
+    with SchedulerRoutes
+    with SchedulerGrpcServices {
+  _: Suite =>
+
+  lazy val path = s"/$RootPath/$SchedulerPath"
+
+  override def apiRoutes(system: ActorSystem[_]): Route =
+    SessionServiceRoute(system).route ~ schedulerService(system).route
+
+  def addSchedule(schedule: Schedule): Assertion = {
+    withCookies(Post(s"$path/schedules", schedule)) ~> routes ~> check {
+      status shouldEqual StatusCodes.OK
+    }
+  }
+
+  def removeSchedule(persistenceId: String, entityId: String, key: String): Assertion = {
+    withCookies(
+      Delete(
+        s"$path/schedules",
+        RemoveSchedule(persistenceId, entityId, key)
+      )
+    ) ~> routes ~> check {
+      status shouldEqual StatusCodes.OK
+    }
+  }
+
+  def addCronTab(cronTab: CronTab): Assertion = {
+    withCookies(Post(s"$path/crons", cronTab)) ~> routes ~> check {
+      status shouldEqual StatusCodes.OK
+    }
+  }
+
+  def removeCronTab(persistenceId: String, entityId: String, key: String): Assertion = {
+    withCookies(
+      Delete(
+        s"$path/crons",
+        RemoveCronTab(persistenceId, entityId, key)
+      )
+    ) ~> routes ~> check {
+      status shouldEqual StatusCodes.OK
+    }
+  }
+
+  def loadScheduler(): Scheduler = {
+    withCookies(Get(s"$path")) ~> routes ~> check {
+      status shouldEqual StatusCodes.OK
+      responseAs[Scheduler]
+    }
+  }
+}
diff --git a/testkit/src/test/scala/app/softnetwork/scheduler/service/SchedulerServiceSpec.scala b/testkit/src/test/scala/app/softnetwork/scheduler/service/SchedulerServiceSpec.scala
@@ -0,0 +1,33 @@
+package app.softnetwork.scheduler.service
+
+import app.softnetwork.scheduler.model.{CronTab, Schedule}
+import app.softnetwork.scheduler.scalatest.SchedulerRouteTestKit
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class SchedulerServiceSpec extends AnyWordSpecLike with SchedulerRouteTestKit {
+
+  val schedule: Schedule = Schedule("s", "0", "add", 1, Some(true), None, None)
+
+  val cronTab: CronTab = CronTab("c", "*", "cron", "* * * * *")
+
+  "scheduler service" must {
+    "add schedule" in {
+      createSession("admin", admin = Some(true))
+      addSchedule(schedule)
+    }
+    "add cron tab" in {
+      addCronTab(cronTab)
+    }
+    "load scheduler" in {
+      val scheduler = loadScheduler()
+      assert(scheduler.schedules.exists(_.uuid == schedule.uuid))
+      assert(scheduler.cronTabs.exists(_.uuid == cronTab.uuid))
+    }
+    "remove schedule" in {
+      removeSchedule(schedule.persistenceId, schedule.entityId, schedule.key)
+    }
+    "remove cron tab" in {
+      removeCronTab(cronTab.persistenceId, cronTab.entityId, cronTab.key)
+    }
+  }
+}

Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,8 @@ name := "scheduler-common"`
`8`	`8`
`9`	`9`	`libraryDependencies ++= Seq(`
`10`	`10`	`"com.markatta" %% "akron" % "1.2" excludeAll(ExclusionRule(organization = "com.typesafe.akka"), ExclusionRule(organization = "org.scala-lang.modules")),`
	`11`	`+ // session`
	`12`	`+ "app.softnetwork.session" %% "session-core" % Versions.session,`
`11`	`13`	`"app.softnetwork.api" %% "generic-server-api" % Versions.server,`
`12`	`14`	`"app.softnetwork.protobuf" %% "scalapb-extensions" % "0.1.5"`
`13`	`15`	`)`
Original file line number	Diff line number	Diff line change
`@@ -19,6 +19,8 @@ object SchedulerSettings extends StrictLogging {`
`19`	`19`	`case Right(schedulerConfig) => schedulerConfig`
`20`	`20`	`}`
`21`	`21`
	`22`	`+ val SchedulerPath: String = config.getString("softnetwork.scheduler.path")`
	`23`	`+`
`22`	`24`	`def tag(persistenceId: String): String = s"scheduler-to-$persistenceId"`
`23`	`25`	`}`
`24`	`26`
Original file line number	Diff line number	Diff line change
`@@ -8,5 +8,7 @@ object Versions {`
`8`	`8`
`9`	`9`	`val server = "0.2.8"`
`10`	`10`
	`11`	`+ val session = "0.1.3"`
	`12`	`+`
`11`	`13`	`val scalatest = "3.1.1"`
`12`	`14`	`}`