[NU-1960] API for messages preview in source

designer/server/src/main/scala/pl/touk/nussknacker/ui/api/NodesApiHttpService.scala

@@ -8,6 +8,7 @@ import pl.touk.nussknacker.engine.ModelData
 import pl.touk.nussknacker.engine.api.graph.{ProcessProperties, ScenarioGraph}
 import pl.touk.nussknacker.engine.api.process.{ProcessName, ProcessingType}
 import pl.touk.nussknacker.engine.api.typed.typing.TypingResult
+import pl.touk.nussknacker.engine.graph.node.SourceNodeData
 import pl.touk.nussknacker.engine.spel.ExpressionSuggestion
 import pl.touk.nussknacker.restmodel.definition.UIValueParameter
 import pl.touk.nussknacker.restmodel.scenariodetails.ScenarioWithDetails
@@ -16,6 +17,7 @@ import pl.touk.nussknacker.ui.api.BaseHttpService.CustomAuthorizationError
 import pl.touk.nussknacker.ui.api.description.NodesApiEndpoints
 import pl.touk.nussknacker.ui.api.description.NodesApiEndpoints.Dtos
 import pl.touk.nussknacker.ui.api.description.NodesApiEndpoints.Dtos.NodesError.{
+  FetchLatestRecordsError,
   MalformedTypingResult,
   NoPermission,
   NoProcessingType,
@@ -34,11 +36,12 @@ import pl.touk.nussknacker.ui.api.description.NodesApiEndpoints.Dtos.{
   decodeVariableTypes,
   prepareTypingResultDecoder
 }
-import pl.touk.nussknacker.ui.api.utils.ScenarioHttpServiceExtensions
 import pl.touk.nussknacker.ui.api.utils.ScenarioDetailsOps._
+import pl.touk.nussknacker.ui.api.utils.ScenarioHttpServiceExtensions
 import pl.touk.nussknacker.ui.process.ProcessService
 import pl.touk.nussknacker.ui.process.processingtype.provider.ProcessingTypeDataProvider
 import pl.touk.nussknacker.ui.process.repository.ProcessDBQueryRepository.ProcessNotFoundError
+import pl.touk.nussknacker.ui.process.test.ScenarioTestService
 import pl.touk.nussknacker.ui.security.api.{AuthManager, LoggedUser}
 import pl.touk.nussknacker.ui.suggester.ExpressionSuggester
 import pl.touk.nussknacker.ui.validation.{NodeValidator, ParametersValidator, UIProcessValidator}
@@ -52,6 +55,7 @@ class NodesApiHttpService(
     processingTypeToNodeValidator: ProcessingTypeDataProvider[NodeValidator, _],
     processingTypeToExpressionSuggester: ProcessingTypeDataProvider[ExpressionSuggester, _],
     processingTypeToParametersValidator: ProcessingTypeDataProvider[ParametersValidator, _],
+    processingTypeToScenarioTestServices: ProcessingTypeDataProvider[ScenarioTestService, _],
     protected override val scenarioService: ProcessService
 )(override protected implicit val executionContext: ExecutionContext)
     extends BaseHttpService(authManager)
@@ -144,6 +148,39 @@ class NodesApiHttpService(
       }
   }
 
+  expose {
+    nodesApiEndpoints.fetchLatestRecordsForNodeEndpoint
+      .serverSecurityLogic(authorizeKnownUser[NodesError])
+      .serverLogicEitherT { implicit loggedUser =>
+        { case (scenarioName, numberOfRecords, fetchLatestRecordsDto) =>
+          for {
+            scenarioWithDetails <- getScenarioWithDetailsByName(scenarioName)
+            scenarioTestService = processingTypeToScenarioTestServices.forProcessingTypeUnsafe(
+              scenarioWithDetails.processingType
+            )
+            sourceNodeData <- EitherT.fromEither[Future](fetchLatestRecordsDto.nodeData match {
+              case source: SourceNodeData => Right(source)
+              case other =>
+                Left(FetchLatestRecordsError(s"Expected SourceNodeData but got: ${other.getClass.getSimpleName}"))
+            })
+            parametersDefinition <- EitherT[Future, NodesError, String](
+              scenarioTestService.getDataFromSource(
+                fetchLatestRecordsDto.processProperties.toMetaData(scenarioName),
+                sourceNodeData,
+                numberOfRecords
+              ) match {
+                case Left(error) =>
+                  logger.error(s"Error during fetching latest records for node=[${sourceNodeData.id}]: $error")
+                  Future(Left(FetchLatestRecordsError(error)))
+                case Right(rawScenarioTestData) =>
+                  Future(Right(rawScenarioTestData.content))
+              }
+            )
+          } yield parametersDefinition
+        }
+      }
+  }
+
   expose {
     nodesApiEndpoints.parametersValidationEndpoint
       .serverSecurityLogic(authorizeKnownUser[NodesError])

designer/server/src/main/scala/pl/touk/nussknacker/ui/api/description/NodesApiEndpoints.scala

@@ -10,7 +10,7 @@ import io.circe.{Decoder, Encoder, Json, KeyDecoder, KeyEncoder}
 import org.springframework.util.ClassUtils
 import pl.touk.nussknacker.engine.additionalInfo.{AdditionalInfo, MarkdownAdditionalInfo}
 import pl.touk.nussknacker.engine.api.CirceUtil._
-import pl.touk.nussknacker.engine.api.{LayoutData, ProcessAdditionalFields}
+import pl.touk.nussknacker.engine.api.{LayoutData, ProcessAdditionalFields, StreamMetaData}
 import pl.touk.nussknacker.engine.api.definition.{
   FixedExpressionValue,
   FixedExpressionValueWithIcon,
@@ -54,6 +54,7 @@ import pl.touk.nussknacker.security.AuthCredentials
 import pl.touk.nussknacker.ui.api.TapirCodecs.ScenarioGraphCodec._
 import pl.touk.nussknacker.ui.api.TapirCodecs.ScenarioNameCodec._
 import pl.touk.nussknacker.ui.api.description.NodesApiEndpoints.Dtos.NodesError.{
+  FetchLatestRecordsError,
   MalformedTypingResult,
   NoProcessingType,
   NoScenario
@@ -349,6 +350,63 @@ class NodesApiEndpoints(auth: EndpointInput[AuthCredentials]) extends BaseEndpoi
       .withSecurity(auth)
   }
 
+  lazy val fetchLatestRecordsForNodeEndpoint: SecuredEndpoint[
+    (ProcessName, Int, FetchLatestRecordsRequestDto),
+    NodesError,
+    String,
+    Any
+  ] = {
+    baseNuApiEndpoint
+      .summary("Fetch latest records for specific node")
+      .tag("Nodes")
+      .post
+      .in(
+        "nodes" / path[ProcessName]("scenarioName") / "fetchLatestRecordsForNode" / path[Int](
+          "numberOfRecords"
+        )
+      )
+      .in(
+        jsonBody[FetchLatestRecordsRequestDto]
+          .example(
+            Example.of(
+              summary = Some("Basic fetch request"),
+              value = FetchLatestRecordsRequestDto(
+                ProcessProperties(StreamMetaData()),
+                Source("sourceId", SourceRef("source", List.empty), None)
+              )
+            )
+          )
+      )
+      .out(
+        statusCode(Ok).and(
+          stringBody
+            .examples(
+              List(
+                Example.of(
+                  summary = Some("Simple scenario test data in json stringify form"),
+                  value = "{name: John}"
+                )
+              )
+            )
+        )
+      )
+      .errorOut(
+        oneOf[NodesError](
+          fetchLatestRecordsErrorExample,
+          noScenarioExample
+        )
+      )
+      .withSecurity(auth)
+  }
+
+  private val simpleGraphExample: Example[ScenarioGraph] = Example.of(
+    ScenarioGraph(
+      ProcessProperties(StreamMetaData()),
+      List(),
+      List(),
+    )
+  )
+
   lazy val parametersValidationEndpoint: SecuredEndpoint[
     (ProcessingType, ParametersValidationRequestDto),
     NodesError,
@@ -561,6 +619,18 @@ object NodesApiEndpoints {
           )
       )
 
+    val fetchLatestRecordsErrorExample: EndpointOutput.OneOfVariant[FetchLatestRecordsError] =
+      oneOfVariantFromMatchType(
+        NotFound,
+        plainBody[FetchLatestRecordsError]
+          .example(
+            Example.of(
+              summary = Some("Fetching error"),
+              value = FetchLatestRecordsError("Fetching error")
+            )
+          )
+      )
+
     val malformedTypingResultExample: EndpointOutput.OneOfVariant[MalformedTypingResult] =
       oneOfVariantFromMatchType(
         BadRequest,
@@ -1472,6 +1542,7 @@ object NodesApiEndpoints {
     sealed trait NodesError
 
     object NodesError {
+      final case class FetchLatestRecordsError(msg: String)             extends NodesError
       final case class NoScenario(scenarioName: ProcessName)            extends NodesError
       final case class NoProcessingType(processingType: ProcessingType) extends NodesError
       final case object NoPermission                                    extends NodesError with CustomAuthorizationError
@@ -1495,8 +1566,20 @@ object NodesApiEndpoints {
         )
       }
 
+      implicit val fetchLatestRecordsErrorCodec: Codec[String, FetchLatestRecordsError, CodecFormat.TextPlain] = {
+        BaseEndpointDefinitions.toTextPlainCodecSerializationOnly[FetchLatestRecordsError](e =>
+          s"Error during fetching latest records: \n${e.msg}"
+        )
+      }
+
     }
 
+    @derive(schema, encoder, decoder)
+    final case class FetchLatestRecordsRequestDto(
+        processProperties: ProcessProperties,
+        nodeData: NodeData
+    )
+
   }
 
 }

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

@@ -2,14 +2,15 @@ package pl.touk.nussknacker.ui.process.test
 
 import com.carrotsearch.sizeof.RamUsageEstimator
 import com.typesafe.scalalogging.LazyLogging
-import pl.touk.nussknacker.engine.api.ProcessVersion
 import pl.touk.nussknacker.engine.api.definition.{DualParameterEditor, Parameter, StringParameterEditor}
 import pl.touk.nussknacker.engine.api.editor.DualEditorMode
 import pl.touk.nussknacker.engine.api.graph.ScenarioGraph
 import pl.touk.nussknacker.engine.api.test.ScenarioTestData
 import pl.touk.nussknacker.engine.api.typed.typing.Typed
+import pl.touk.nussknacker.engine.api.{MetaData, ProcessVersion}
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
 import pl.touk.nussknacker.engine.definition.test.{TestInfoProvider, TestingCapabilities}
+import pl.touk.nussknacker.engine.graph.node.SourceNodeData
 import pl.touk.nussknacker.engine.testmode.TestProcess.TestResults
 import pl.touk.nussknacker.restmodel.definition.UISourceParameters
 import pl.touk.nussknacker.ui.api.TestDataSettings
@@ -85,6 +86,16 @@ class ScenarioTestService(
     } yield rawTestData
   }
 
+  def getDataFromSource(
+      metaData: MetaData,
+      sourceNodeData: SourceNodeData,
+      size: Int
+  ): Either[String, RawScenarioTestData] = {
+    testInfoProvider.generateTestDataForSource(metaData, sourceNodeData, size).flatMap { generatedData =>
+      preliminaryScenarioTestDataSerDe.serialize(generatedData)
+    }
+  }
+
   def performTest(
       scenarioGraph: ScenarioGraph,
       processVersion: ProcessVersion,

designer/server/src/main/scala/pl/touk/nussknacker/ui/server/AkkaHttpBasedRouteProvider.scala

@@ -404,6 +404,7 @@ class AkkaHttpBasedRouteProvider(
         processingTypeToParametersValidator = processingTypeDataProvider.mapValues(v =>
           new ParametersValidator(v.designerModelData.modelData, v.deploymentData.scenarioPropertiesConfig.keys)
         ),
+        processingTypeToScenarioTestServices = scenarioTestService,
         scenarioService = processService,
       )