review fixes

Author: mk-software-pl

designer/server/src/main/scala/pl/touk/nussknacker/ui/limits/LimitsService.scala

@@ -16,8 +16,6 @@ class LimitsService(
     scenarioStatusProvider: ScenarioStatusProvider
 ) {
 
-  // todo: check locking
-
   def checkScenarioLimitsBeforeDeployment(
       scenarioProcessingType: ProcessingType
   )(implicit user: LoggedUser): IO[Either[LimitError, Unit]] = {
@@ -49,7 +47,25 @@ class LimitsService(
     }
   }
 
-  private def checkGlobalLimits(): EitherT[IO, ActiveScenariosLimitExceededError, Unit] = ???
+  private def checkGlobalLimits()(implicit user: LoggedUser): EitherT[IO, ActiveScenariosLimitExceededError, Unit] = {
+    globalLimitsConfig.activeScenariosLimit match {
+      case Some(activeScenariosLimit) =>
+        EitherT { // todo: user nu user?
+          scenarioStatusProvider
+            .getActiveScenariosCountFor(activeScenariosLimitProvider.all.keys)
+            .map { activeScenariosCount =>
+              Either.cond(
+                test = activeScenariosCount < activeScenariosLimit.value,
+                right = (),
+                left = ActiveScenariosLimitExceededError(activeScenariosLimit.value)
+              )
+            }
+        }
+      case None =>
+        EitherT.right(IO.unit)
+    }
+  }
+
 }
 
 object LimitsService {

designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/scenariostatus/ScenarioStatusProvider.scala

@@ -79,20 +79,25 @@ class ScenarioStatusProvider(
     )
   }
 
-  def getActiveScenariosCountFor(processingType: ProcessingType)(implicit user: LoggedUser): IO[Int] = IO.fromFuture {
-    IO {
-      implicit val freshnessStatus: DataFreshnessPolicy = DataFreshnessPolicy.Fresh
-      deploymentStatusesProvider
-        .getBulkQueriedDeploymentStatusesForSupportedManagers(processingType :: Nil)
-        .map { statuses =>
-          statuses.getAllDeploymentStatuses
-            .count { status =>
-              status.status == SimpleStateStatus.DuringDeploy ||
-              status.status == SimpleStateStatus.Running ||
-              status.status == SimpleStateStatus.Restarting
-            }
-        }
+  def getActiveScenariosCountFor(processingTypes: Iterable[ProcessingType])(implicit user: LoggedUser): IO[Int] =
+    IO.fromFuture {
+      IO {
+        implicit val freshnessStatus: DataFreshnessPolicy = DataFreshnessPolicy.Fresh
+        deploymentStatusesProvider
+          .getBulkQueriedDeploymentStatusesForSupportedManagers(processingTypes)
+          .map { statuses =>
+            statuses.getAllDeploymentStatuses
+              .count { status =>
+                status.status == SimpleStateStatus.DuringDeploy ||
+                status.status == SimpleStateStatus.Running ||
+                status.status == SimpleStateStatus.Restarting
+              }
+          }
+      }
     }
+
+  def getActiveScenariosCountFor(processingType: ProcessingType)(implicit user: LoggedUser): IO[Int] = {
+    getActiveScenariosCountFor(processingType :: Nil)
   }
 
   private def getNonFragmentScenarioStatus[ScenarioShape, F[_]: Traverse](

designer/server/src/test/scala/pl/touk/nussknacker/test/base/it/NuResourcesTest.scala

@@ -157,8 +157,8 @@ trait NuResourcesTest
   protected val deploymentService: DeploymentService =
     new DeploymentService(
       dispatcher = dmDispatcher,
-      processValidator = processValidatorByProcessingType,
-      scenarioResolver = scenarioResolverByProcessingType,
+      processValidator = processValidatorByProcessingType(),
+      scenarioResolver = scenarioResolverByProcessingType(),
       actionService = actionService,
       additionalComponentConfigs = mapProcessingTypeDataProvider(),
       limitsService = new LimitsService(
@@ -259,9 +259,9 @@ trait NuResourcesTest
     new DBProcessService(
       processStateProvider,
       scenarioStatusPresenter,
-      newProcessPreparerByProcessingType,
+      newProcessPreparerByProcessingType(),
       processingTypeDataProvider.mapCombined(_.parametersService),
-      processResolverByProcessingType,
+      processResolverByProcessingType(),
       dbioRunner,
       futureFetchingScenarioRepository,
       actionRepository,
@@ -278,12 +278,12 @@ trait NuResourcesTest
   ): ScenarioTestService =
     new ScenarioTestService(
       testInfoProvider,
-      processResolver,
+      processResolver(),
       designerConfig.testDataSettings,
       new PreliminaryScenarioTestDataSerDe(designerConfig.testDataSettings),
-      new ProcessCounter(TestFactory.prepareSampleFragmentRepository),
+      new ProcessCounter(TestFactory.prepareSampleFragmentRepository()),
       new ScenarioTestExecutorServiceImpl(
-        new ScenarioResolver(sampleResolver, Streaming.stringify),
+        new ScenarioResolver(sampleResolver(), Streaming.stringify),
         deploymentManager
       )
     )
@@ -582,7 +582,7 @@ trait NuResourcesTest
       processName: ProcessName,
       isFragment: Boolean = false,
   ): ProcessId = {
-    val emptyProcess = newProcessPreparer.prepareEmptyProcess(processName, isFragment)
+    val emptyProcess = newProcessPreparer().prepareEmptyProcess(processName, isFragment)
     saveAndGetId(emptyProcess, Category1, isFragment).futureValue
   }
 

designer/server/src/test/scala/pl/touk/nussknacker/test/utils/domain/TestFactory.scala

@@ -16,7 +16,7 @@ import pl.touk.nussknacker.engine.management.FlinkStreamingPropertiesConfig
 import pl.touk.nussknacker.engine.util.Implicits.RichScalaMap
 import pl.touk.nussknacker.restmodel.scenariodetails.ScenarioParameters
 import pl.touk.nussknacker.security.Permission
-import pl.touk.nussknacker.test.config.WithSimplifiedDesignerConfig.{TestCategory, TestProcessingType}
+import pl.touk.nussknacker.test.config.WithSimplifiedDesignerConfig.TestCategory
 import pl.touk.nussknacker.test.config.WithSimplifiedDesignerConfig.TestProcessingType.Streaming
 import pl.touk.nussknacker.test.mock.{StubFragmentRepository, TestAdditionalUIConfigProvider}
 import pl.touk.nussknacker.ui.api.{RouteWithoutUser, RouteWithUser}
@@ -65,57 +65,73 @@ object TestFactory {
 
   val possibleValues: List[FixedExpressionValue] = List(FixedExpressionValue("a", "a"))
 
-  val processValidator: UIProcessValidator = ProcessTestData.testProcessValidator(fragmentResolver = sampleResolver)
+  def processValidator(processingTypes: List[String] = List(Streaming.stringify)): UIProcessValidator =
+    ProcessTestData.testProcessValidator(fragmentResolver = sampleResolver(processingTypes))
 
-  val flinkProcessValidator: UIProcessValidator = ProcessTestData.testProcessValidator(
-    fragmentResolver = sampleResolver,
-    scenarioProperties = FlinkStreamingPropertiesConfig.properties
-  )
+  def flinkProcessValidator(processingTypes: List[String] = List(Streaming.stringify)): UIProcessValidator =
+    ProcessTestData.testProcessValidator(
+      fragmentResolver = sampleResolver(processingTypes),
+      scenarioProperties = FlinkStreamingPropertiesConfig.properties
+    )
 
-  val processValidatorByProcessingType: ProcessingTypeDataProvider[UIProcessValidator, _] =
-    mapProcessingTypeDataProvider(Streaming.stringify -> flinkProcessValidator)
+  def processValidatorByProcessingType(
+      processingTypes: List[String] = List(Streaming.stringify)
+  ): ProcessingTypeDataProvider[UIProcessValidator, _] = {
+    val validator = flinkProcessValidator(processingTypes)
+    mapProcessingTypeDataProvider(processingTypes.map(pt => (pt, validator)): _*)
+  }
 
-  val processResolver = new UIProcessResolver(
-    processValidator,
+  def processResolver(processingTypes: List[String] = List(Streaming.stringify)) = new UIProcessResolver(
+    processValidator(processingTypes),
     ProcessDictSubstitutor(new SimpleDictRegistry(Map.empty))
   )
 
-  val processResolverByProcessingType: ProcessingTypeDataProvider[UIProcessResolver, _] =
-    mapProcessingTypeDataProvider(Streaming.stringify -> processResolver)
-
-  val scenarioParametersService: ScenarioParametersService = {
-    val combinations = Map(
-      TestProcessingType.Streaming.stringify ->
-        ScenarioParametersWithEngineSetupErrors(
-          ScenarioParameters(
-            ProcessingMode.UnboundedStream,
-            TestCategory.Category1.stringify,
-            EngineSetupName("Flink")
-          ),
-          List.empty
-        )
+  def processResolverByProcessingType(
+      processingTypes: List[String] = List(Streaming.stringify)
+  ): ProcessingTypeDataProvider[UIProcessResolver, _] = {
+    val resolver = processResolver(processingTypes)
+    mapProcessingTypeDataProvider(processingTypes.map(pt => (pt, resolver)): _*)
+  }
+
+  def scenarioParametersService(
+      processingTypes: List[String] = List(Streaming.stringify)
+  ): ScenarioParametersService = {
+    val scenarioParameters = ScenarioParametersWithEngineSetupErrors(
+      ScenarioParameters(
+        ProcessingMode.UnboundedStream,
+        TestCategory.Category1.stringify,
+        EngineSetupName("Flink")
+      ),
+      List.empty
     )
+    val combinations = processingTypes.map(pt => (pt, scenarioParameters)).toMap
     ScenarioParametersService.createUnsafe(combinations)
   }
 
-  val scenarioParametersServiceProvider: ProcessingTypeDataProvider[_, ScenarioParametersService] =
-    TestProcessingTypeDataProviderFactory.create(Map.empty, scenarioParametersService)
+  def scenarioParametersServiceProvider(
+      processingTypes: List[String] = List(Streaming.stringify)
+  ): ProcessingTypeDataProvider[_, ScenarioParametersService] =
+    TestProcessingTypeDataProviderFactory.create(Map.empty, scenarioParametersService(processingTypes))
 
   // It should be defined as method, because when it's defined as val then there is bug in IDEA at DefinitionPreparerSpec - it returns null
-  def prepareSampleFragmentRepository: StubFragmentRepository = new StubFragmentRepository(
-    Map(
-      Streaming.stringify -> List(ProcessTestData.sampleFragment)
+  def prepareSampleFragmentRepository(
+      processingTypes: List[String] = List(Streaming.stringify)
+  ): StubFragmentRepository =
+    new StubFragmentRepository(
+      processingTypes.map(pt => (pt, List(ProcessTestData.sampleFragment))).toMap
     )
-  )
 
-  def sampleResolver = new FragmentResolver(prepareSampleFragmentRepository)
+  def sampleResolver(processingTypes: List[String] = List(Streaming.stringify)) =
+    new FragmentResolver(prepareSampleFragmentRepository(processingTypes))
 
-  val scenarioResolver = new ScenarioResolver(sampleResolver, Streaming.stringify)
+  def scenarioResolver(processingType: String = Streaming.stringify) =
+    new ScenarioResolver(sampleResolver(processingType :: Nil), processingType)
 
-  def scenarioResolverByProcessingType: ProcessingTypeDataProvider[ScenarioResolver, _] =
-    mapProcessingTypeDataProvider(
-      Streaming.stringify -> scenarioResolver
-    )
+  def scenarioResolverByProcessingType(
+      processingTypes: List[String] = List(Streaming.stringify)
+  ): ProcessingTypeDataProvider[ScenarioResolver, _] = {
+    mapProcessingTypeDataProvider(processingTypes.map(pt => (pt, scenarioResolver(pt))): _*)
+  }
 
   def additionalComponentConfigsByProcessingType
       : ProcessingTypeDataProvider[Map[DesignerWideComponentId, ComponentAdditionalConfig], _] =
@@ -160,24 +176,32 @@ object TestFactory {
     new DBFetchingProcessRepository[DB](dbRef, newActionProcessRepository(dbRef), newScenarioLabelsRepository(dbRef))
       with DbioRepository
 
-  def newWriteProcessRepository(dbRef: DbRef, clock: Clock, modelVersions: Option[Int] = Some(1)) =
+  def newWriteProcessRepository(
+      dbRef: DbRef,
+      clock: Clock,
+      processingTypes: List[String] = List(Streaming.stringify),
+      modelVersions: Option[Int] = Some(1)
+  ) =
     new DBProcessRepository(
       dbRef,
       clock,
       newScenarioActivityRepository(dbRef, clock),
       newScenarioLabelsRepository(dbRef),
-      mapProcessingTypeDataProvider(modelVersions.map(Streaming.stringify -> _).toList: _*),
+      mapProcessingTypeDataProvider(modelVersions.map(mv => processingTypes.map(pt => (pt, mv))).toList.flatten: _*),
     )
 
   def newDummyWriteProcessRepository(): DBProcessRepository =
     newWriteProcessRepository(dummyDbRef, Clock.systemUTC())
 
-  def newScenarioGraphVersionService(dbRef: DbRef) = new ScenarioGraphVersionService(
-    newScenarioGraphVersionRepository(dbRef),
-    mapProcessingTypeDataProvider(Streaming.stringify -> processValidator),
-    scenarioResolverByProcessingType,
-    newDBIOActionRunner(dbRef)
-  )
+  def newScenarioGraphVersionService(dbRef: DbRef, processingTypes: List[String] = List(Streaming.stringify)) = {
+    val validator = processValidator(processingTypes)
+    new ScenarioGraphVersionService(
+      newScenarioGraphVersionRepository(dbRef),
+      mapProcessingTypeDataProvider(processingTypes.map(pt => (pt, validator)): _*),
+      scenarioResolverByProcessingType(processingTypes),
+      newDBIOActionRunner(dbRef)
+    )
+  }
 
   def newScenarioGraphVersionRepository(dbRef: DbRef) = new ScenarioGraphVersionRepository(dbRef)
 
@@ -197,17 +221,17 @@ object TestFactory {
   def asAdmin(route: RouteWithUser): Route =
     route.securedRouteWithErrorHandling(adminUser())
 
-  val newProcessPreparer: NewProcessPreparer =
+  def newProcessPreparer(processingType: String = Streaming.stringify): NewProcessPreparer =
     new NewProcessPreparer(
       ProcessTestData.streamingTypeSpecificInitialData,
       FlinkStreamingPropertiesConfig.properties,
-      new ScenarioPropertiesConfigFinalizer(TestAdditionalUIConfigProvider, Streaming.stringify)
+      new ScenarioPropertiesConfigFinalizer(TestAdditionalUIConfigProvider, processingType)
     )
 
-  val newProcessPreparerByProcessingType: ProcessingTypeDataProvider[NewProcessPreparer, _] =
-    mapProcessingTypeDataProvider(
-      Streaming.stringify -> newProcessPreparer
-    )
+  def newProcessPreparerByProcessingType(
+      processingTypes: List[String] = List(Streaming.stringify)
+  ): ProcessingTypeDataProvider[NewProcessPreparer, _] =
+    mapProcessingTypeDataProvider(processingTypes.map(pt => (pt, newProcessPreparer(pt))): _*)
 
   def withPermissions(route: RouteWithUser, permissions: Permission.Permission*): Route =
     route.securedRouteWithErrorHandling(user(permissions = permissions))

designer/server/src/test/scala/pl/touk/nussknacker/ui/api/ProcessesExportImportResourcesSpec.scala

@@ -47,7 +47,7 @@ class ProcessesExportImportResourcesSpec
     futureFetchingScenarioRepository,
     processService,
     scenarioActivityRepository,
-    processResolverByProcessingType,
+    processResolverByProcessingType(),
     dbioRunner,
   )
 

designer/server/src/test/scala/pl/touk/nussknacker/ui/definition/component/DefaultComponentServiceSpec.scala

@@ -919,9 +919,9 @@ class DefaultComponentServiceSpec
     new DBProcessService(
       scenarioStatusProvider = mock[ScenarioStatusProvider],
       scenarioStatusPresenter = mock[ScenarioStatusPresenter],
-      newProcessPreparers = TestFactory.newProcessPreparerByProcessingType,
+      newProcessPreparers = TestFactory.newProcessPreparerByProcessingType(),
       scenarioParametersServiceProvider = scenarioParametersServiceProvider,
-      processResolverByProcessingType = TestFactory.processResolverByProcessingType,
+      processResolverByProcessingType = TestFactory.processResolverByProcessingType(),
       dbioRunner = TestFactory.newDummyDBIOActionRunner(),
       fetchingProcessRepository = MockFetchingProcessRepository.withProcessesDetails(processes),
       scenarioActionRepository = TestFactory.newDummyActionRepository(),

designer/server/src/test/scala/pl/touk/nussknacker/ui/notifications/NotificationServiceTest.scala

@@ -314,8 +314,8 @@ class NotificationServiceTest
     )
     val deploymentService = new DeploymentService(
       managerDispatcher,
-      TestFactory.processValidatorByProcessingType,
-      TestFactory.scenarioResolverByProcessingType,
+      TestFactory.processValidatorByProcessingType(),
+      TestFactory.scenarioResolverByProcessingType(),
       actionService,
       TestFactory.additionalComponentConfigsByProcessingType,
       new LimitsService(

designer/server/src/test/scala/pl/touk/nussknacker/ui/process/DBProcessServiceSpec.scala

@@ -207,9 +207,9 @@ class DBProcessServiceSpec extends AnyFlatSpec with Matchers with PatientScalaFu
     new DBProcessService(
       scenarioStatusProvider = mock[ScenarioStatusProvider],
       scenarioStatusPresenter = mock[ScenarioStatusPresenter],
-      newProcessPreparers = TestFactory.newProcessPreparerByProcessingType,
-      scenarioParametersServiceProvider = TestFactory.scenarioParametersServiceProvider,
-      processResolverByProcessingType = TestFactory.processResolverByProcessingType,
+      newProcessPreparers = TestFactory.newProcessPreparerByProcessingType(),
+      scenarioParametersServiceProvider = TestFactory.scenarioParametersServiceProvider(),
+      processResolverByProcessingType = TestFactory.processResolverByProcessingType(),
       dbioRunner = TestFactory.newDummyDBIOActionRunner(),
       fetchingProcessRepository = MockFetchingProcessRepository.withProcessesDetails(processes),
       scenarioActionRepository = TestFactory.newDummyActionRepository(),

designer/server/src/test/scala/pl/touk/nussknacker/ui/process/deployment/TestDeploymentServiceFactory.scala

@@ -101,7 +101,7 @@ class TestDeploymentServiceFactory(dbRef: DbRef) {
                 deploymentManager,
                 EngineSetupName("mock"),
                 JobsRecoverySettings.noRecovery,
-                TestFactory.scenarioResolver
+                TestFactory.scenarioResolver()
               )
             )
           )
@@ -117,7 +117,7 @@ class TestDeploymentServiceFactory(dbRef: DbRef) {
       TestFactory.mapProcessingTypeDataProvider(
         Streaming.stringify -> ProcessTestData.testProcessValidator(validator = ProcessValidator.default(modelData))
       ),
-      TestFactory.scenarioResolverByProcessingType,
+      TestFactory.scenarioResolverByProcessingType(),
       actionService,
       additionalComponentConfigsByProcessingType,
       new LimitsService(

designer/server/src/test/scala/pl/touk/nussknacker/ui/process/migrate/HttpRemoteEnvironmentSpec.scala

@@ -136,7 +136,7 @@ object HttpRemoteEnvironmentSpec {
     mapProcessingTypeDataProvider(
       "streaming" -> new ProcessModelMigrator(new TestMigrations(1, 2))
     ),
-    mapProcessingTypeDataProvider("streaming" -> flinkProcessValidator)
+    mapProcessingTypeDataProvider("streaming" -> flinkProcessValidator("streaming" :: Nil))
   )
 
   private val impersonationSupport = new ImpersonationSupported {

designer/server/src/test/scala/pl/touk/nussknacker/ui/process/migrate/StandardRemoteEnvironmentSpec.scala

@@ -164,7 +164,7 @@ object StandardRemoteEnvironmentSpec {
       mapProcessingTypeDataProvider(
         "streaming" -> new ProcessModelMigrator(new TestMigrations(1, 2))
       ),
-      mapProcessingTypeDataProvider("streaming" -> flinkProcessValidator)
+      mapProcessingTypeDataProvider("streaming" -> flinkProcessValidator("streaming" :: Nil))
     )
 
     protected def fetchRemoteMigrationScenarioDescriptionVersion: FutureE[Int] = ???