Make NVP codec isomorphic

NVP (Name-Value-Pair) type casts all the source values into plain `String`.
This causes problems when a value is encoded differently and a different
type is required when recovering a failed event.

We need a smarter way of dealing with that. Ideally Name-Value-Pair should
be removed, but until then we need an alternative codec that will somehow
restore the type.

core/src/main/scala/com/snowplowanalytics/snowplow/event/recovery/inspect/transform.scala

@@ -24,6 +24,7 @@ import io.circe.parser.{parse => parseJson}
 import domain._
 import com.snowplowanalytics.snowplow.badrows.NVP
 import scala.util.matching.Regex
+import json.nvpsDecoder
 
 /** A blueprint for transformation operations on JSON objects
   */
@@ -198,7 +199,7 @@ private[inspect] object transform {
       .flatMap(encode)
 
   private[this] val indexF = (name: String) =>
-    (json: ACursor) => (Option(index(name)) <*> json.as[List[NVP]].toOption).filter(_ >= 0)
+    (json: ACursor) => (Option(index(name)) <*> json.as[List[NVP]](nvpsDecoder).toOption).filter(_ >= 0)
 
   private[this] val index = (name: String) => (nvps: List[NVP]) => nvps.indexWhere(_.name == name)
 

core/src/main/scala/com/snowplowanalytics/snowplow/event/recovery/json.scala

@@ -15,15 +15,16 @@
 package com.snowplowanalytics.snowplow
 package event.recovery
 
-import cats.syntax.functor._
-import io.circe.{Decoder, Encoder, Printer}
+import cats.implicits._
+import io.circe.{Decoder, DecodingFailure, Encoder, HCursor, Json, JsonObject, Printer}
 import io.circe.syntax._
 import io.circe.generic.semiauto.{deriveDecoder, deriveEncoder}
 import io.circe.generic.extras.semiauto.{deriveEnumerationDecoder, deriveEnumerationEncoder}
 import com.snowplowanalytics.snowplow.badrows._
 import com.snowplowanalytics.iglu.core.circe.implicits._
 import config._
 import config.conditions._
+import badrows.NVP
 
 object json {
   def path(str: String): Seq[String] = {
@@ -156,4 +157,51 @@ object json {
     private[this] def field[A: Decoder](fieldName: String)(b: BadRow): Option[A] =
       b.selfDescribingData.data.hcursor.get[A](fieldName).toOption
   }
+
+  implicit val nvpsDecoder: Decoder[List[NVP]] = new Decoder[List[NVP]] {
+    final def apply(cur: HCursor): Decoder.Result[List[NVP]] =
+      cur.focus match {
+        case Some(json) =>
+          json.fold(
+            DecodingFailure("query string (payload.raw.parameters) can not be null", cur.history).asLeft,
+            b =>
+              DecodingFailure(
+                s"query string (payload.raw.parameters) can not be boolean, [$b] provided",
+                cur.history
+              ).asLeft,
+            n =>
+              DecodingFailure(
+                s"query string (payload.raw.parameters) cannot be number, [$n] provided",
+                cur.history
+              ).asLeft,
+            s =>
+              DecodingFailure(
+                s"query string (payload.raw.parameters) cannot be string, [$s] provided",
+                cur.history
+              ).asLeft,
+            a => arrayToNVPs(a),
+            o => objectToNVPs(o)
+          )
+        case None =>
+          DecodingFailure("query string (payload.raw.parameters) is missing", cur.history).asLeft
+      }
+  }
+
+  private def arrayToNVPs(arr: Vector[Json]): Decoder.Result[List[NVP]] =
+    arr.toList.traverse(_.as[NVP])
+
+  private def objectToNVPs(obj: JsonObject): Decoder.Result[List[NVP]] =
+    obj.toList.traverse { case (key, value) =>
+      value
+        .fold(
+          None.asRight,
+          b => Some(b.toString).asRight,
+          n => Some(n.toString).asRight,
+          s => Some(s).asRight,
+          a => DecodingFailure(s"NVP parameter cannot be an array, [$a] provided", List.empty).asLeft,
+          o => DecodingFailure(s"NVP parameter cannot be an object, [$o] provided", List.empty).asLeft
+        )
+        .map(v => NVP(key, v))
+    }
+
 }

core/src/main/scala/com/snowplowanalytics/snowplow/event/recovery/utils/payload.scala

@@ -17,7 +17,6 @@ package event.recovery
 package util
 
 import scala.collection.JavaConverters._
-import cats.syntax.either._
 import io.circe.Json
 import io.circe.parser._
 import domain._
@@ -26,6 +25,8 @@ import CollectorPayload.thrift.model1.CollectorPayload
 import org.joda.time.DateTime
 import java.util.UUID
 import com.snowplowanalytics.iglu.core.SchemaKey
+import io.circe._
+import cats.implicits._
 
 object payload {
 
@@ -72,7 +73,7 @@ object payload {
       extractData(e)
         .flatMap { case (schema, data) =>
           data
-            .as[List[NVP]]
+            .as[List[NVP]](json.nvpsDecoder)
             .map(nvps => querystring.fromNVP(nvps :+ NVP("schema", Some(schema))))
             .leftMap(err => UncoerciblePayload(e, err.toString))
         }

core/src/test/scala/com/snowplowanalytics/snowplow/event/recovery/inspect/AddSpec.scala

@@ -53,7 +53,7 @@ class AddSpec extends AnyWordSpec with ScalaCheckPropertyChecks with EitherValue
           add(Seq("payload", "querystring"), List(suffix).asJson)(br.asJson).leftMap(_.message)
         )
         f <- EitherT.fromOption[Id](a.hcursor.downField("payload").downField("querystring").focus, "empty cursor")
-        o <- EitherT.fromEither[Id](f.as[List[NVP]].leftMap(_.message))
+        o <- EitherT.fromEither[Id](f.as[List[NVP]](json.nvpsDecoder).leftMap(_.message))
       } yield o).value.value should contain theSameElementsAs (value :+ suffix)
 
     }