LNG-88 Canonicalize streams in service calls and if statements (#565)

model/inline/src/main/scala/aqua/model/inline/RawValueInliner.scala

@@ -2,7 +2,12 @@ package aqua.model.inline
 
 import aqua.model.inline.state.{Arrows, Counter, Exports, Mangler}
 import aqua.model.*
-import aqua.model.inline.raw.{ApplyFunctorRawInliner, ApplyPropertiesRawInliner, CallArrowRawInliner, CollectionRawInliner}
+import aqua.model.inline.raw.{
+  ApplyFunctorRawInliner,
+  ApplyPropertiesRawInliner,
+  CallArrowRawInliner,
+  CollectionRawInliner
+}
 import aqua.raw.ops.*
 import aqua.raw.value.*
 import aqua.types.{ArrayType, OptionType, StreamType}
@@ -107,10 +112,20 @@ object RawValueInliner extends Logging {
   ): State[S, List[(ValueModel, Option[OpModel.Tree])]] =
     values.traverse(valueToModel(_))
 
+  /**
+   * Unfold all arguments and make CallModel
+   * @param flatStreamArguments canonicalize and flatten all stream arguments if true
+   */
   def callToModel[S: Mangler: Exports: Arrows](
-    call: Call
+    call: Call,
+    flatStreamArguments: Boolean
   ): State[S, (CallModel, Option[OpModel.Tree])] =
-    valueListToModel(call.args).map { list =>
+    valueListToModel(call.args).flatMap { args =>
+      if (flatStreamArguments)
+        args.map(arg => TagInliner.flat(arg._1, arg._2, true)).sequence
+      else
+        State.pure(args)
+    }.map { list =>
       (
         CallModel(
           list.map(_._1),

model/inline/src/main/scala/aqua/model/inline/TagInliner.scala

@@ -49,10 +49,8 @@ object TagInliner extends Logging {
     }
   }
 
-  private def flat[S: Mangler](vm: ValueModel, op: Option[OpModel.Tree], flatStream: Boolean) = {
+  def flat[S: Mangler](vm: ValueModel, op: Option[OpModel.Tree], flatStream: Boolean): State[S, (ValueModel, Option[OpModel.Tree])] = {
     vm match {
-      // flatten stream, because in via we are using `fold`
-      // and `fold` will hang on stream
       case v @ VarModel(n, StreamType(t), l) if flatStream =>
         val canonName = v.name + "_canon"
         for {
@@ -134,8 +132,7 @@ object TagInliner extends Logging {
       case ForTag(item, iterable) =>
         for {
           vp <- valueToModel(iterable)
-          (vN, pN) = vp
-          flattened <- flat(vN, pN, true)
+          flattened <- flat(vp._1, vp._2, true)
           (v, p) = flattened
           n <- Mangler[S].findAndForbidName(item)
           elementType = iterable.`type` match {

model/inline/src/main/scala/aqua/model/inline/raw/CallArrowRawInliner.scala

@@ -23,7 +23,7 @@ object CallArrowRawInliner extends RawInliner[CallArrowRaw] with Logging {
       case Some(serviceId) =>
         logger.trace(Console.BLUE + s"call service id $serviceId" + Console.RESET)
         for {
-          cd <- callToModel(call)
+          cd <- callToModel(call, true)
           sd <- valueToModel(serviceId)
         } yield cd._1.exportTo.map(_.asVar.resolveWith(exports)) -> Inline(
           Map.empty,
@@ -44,7 +44,7 @@ object CallArrowRawInliner extends RawInliner[CallArrowRaw] with Logging {
           arrows.get(funcName) match {
             case Some(fn) =>
               logger.trace(Console.YELLOW + s"Call arrow $funcName" + Console.RESET)
-              callToModel(call).flatMap { case (cm, p) =>
+              callToModel(call, false).flatMap { case (cm, p) =>
                 ArrowInliner
                   .callArrowRet(fn, cm)
                   .map { case (body, vars) =>

model/inline/src/test/scala/aqua/model/inline/ArrowInlinerSpec.scala

@@ -51,6 +51,8 @@ class ArrowInlinerSpec extends AnyFlatSpec with Matchers {
     val streamType = StreamType(ScalarType.string)
     val streamVar = VarRaw("records", streamType)
     val streamModel = VarModel("records", StreamType(ScalarType.string))
+    val canonName = streamVar.name + "_canon"
+    val canonModel = VarModel(canonName, CanonStreamType(ScalarType.string))
     val cbType = ArrowType(ProductType(ArrayType(ScalarType.string) :: Nil), ProductType(Nil))
     val cbVal = VarModel("cb-pass", cbType)
 
@@ -112,11 +114,14 @@ class ArrowInlinerSpec extends AnyFlatSpec with Matchers {
 
     model.equalsOrShowDiff(
       RestrictionModel(streamVar.name, true).wrap(
-        CallServiceModel(
-          LiteralModel("\"test-service\"", LiteralType.string),
-          "some-call",
-          CallModel(streamModel :: Nil, Nil)
-        ).leaf
+        SeqModel.wrap(
+          CanonicalizeModel(streamModel, CallModel.Export(canonModel.name, canonModel.`type`)).leaf,
+          CallServiceModel(
+            LiteralModel("\"test-service\"", LiteralType.string),
+            "some-call",
+            CallModel(canonModel :: Nil, Nil)
+          ).leaf
+        )
       )
     ) should be(true)
 
@@ -232,8 +237,10 @@ class ArrowInlinerSpec extends AnyFlatSpec with Matchers {
 
     val returnType = ArrayType(ArrayType(ScalarType.string))
     val streamType = StreamType(ArrayType(ScalarType.string))
+    val canonType = CanonStreamType(ArrayType(ScalarType.string))
     val recordsVar = VarRaw("records", streamType)
     val recordsModel = VarModel(recordsVar.name, recordsVar.baseType)
+    val canonModel = VarModel(recordsVar.name + "_canon", canonType)
     val innerRecordsVar = VarRaw("inner-records", StreamType(ArrayType(ScalarType.string)))
     val innerName = "inner"
 
@@ -290,11 +297,14 @@ class ArrowInlinerSpec extends AnyFlatSpec with Matchers {
           "get_records",
           CallModel(Nil, CallModel.Export(recordsModel.name, recordsModel.`type`) :: Nil)
         ).leaf,
-        CallServiceModel(
-          LiteralModel("\"callbackSrv\"", LiteralType.string),
-          "response",
-          CallModel(recordsModel :: Nil, Nil)
-        ).leaf
+        SeqModel.wrap(
+          CanonicalizeModel(recordsModel, CallModel.Export(canonModel.name, canonType)).leaf,
+          CallServiceModel(
+            LiteralModel("\"callbackSrv\"", LiteralType.string),
+            "response",
+            CallModel(canonModel :: Nil, Nil)
+          ).leaf
+        )
       )
     ) should be(true)