Take superseding schema into account during validation (close #751)

modules/common-fs2/src/test/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/EventGenEtlPipelineSpec.scala

@@ -289,6 +289,12 @@ class EventGenEtlPipelineSpec extends Specification with CatsIO {
               .map(_.toJson(false))
               .map(_.foldWith(folder))
               .map(_.noSpaces)
+              .map(
+                _.replaceAll(
+                  "iglu:com.snowplowanalytics.snowplow/contexts/jsonschema/1-0-0",
+                  "iglu:com.snowplowanalytics.snowplow/contexts/jsonschema/1-0-1"
+                )
+              )
               .map(decode[IntermediateEvent])
               .rethrow
               .map(IntermediateEvent.pad)

modules/common-fs2/src/test/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/blackbox/BlackBoxTesting.scala

@@ -27,6 +27,7 @@ import cats.data.Validated.{Invalid, Valid}
 
 import io.circe.Json
 import io.circe.syntax._
+import io.circe.parser.{parse => jparse}
 
 import org.apache.thrift.TSerializer
 
@@ -116,10 +117,18 @@ object BlackBoxTesting extends Specification with CatsIO {
 
   private def checkEnriched(enriched: EnrichedEvent, expectedFields: Map[String, String]) = {
     val asMap = getMap(enriched)
-    val r = expectedFields.map { case (k, v) => asMap.get(k) must beSome(v) }
+    val r = expectedFields.map {
+      case (k, v) if k == "unstruct_event" || k == "contexts" || k == "derived_contexts" =>
+        compareJsons(asMap.getOrElse(k, ""), v) must beTrue
+      case (k, v) =>
+        asMap.get(k) must beSome(v)
+    }
     r.toList.reduce(_ and _)
   }
 
+  private def compareJsons(j1: String, j2: String): Boolean =
+    j1 == j2 || jparse(j1).toOption.get == jparse(j2).toOption.get
+
   private val enrichedFields = classOf[EnrichedEvent].getDeclaredFields()
   enrichedFields.foreach(_.setAccessible(true))
 

modules/common-fs2/src/test/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/blackbox/enrichments/ApiRequestEnrichmentSpec.scala

@@ -40,7 +40,7 @@ class ApiRequestEnrichmentSpec extends Specification with CatsIO {
       )
       val expected = Map(
         "unstruct_event" -> json"""{"schema":"iglu:com.snowplowanalytics.snowplow/unstruct_event/jsonschema/1-0-0","data":{"schema":"iglu:com.snowplowanalytics.snowplow-website/signup_form_submitted/jsonschema/1-0-0","data":{"name":"Bob®","email":"[email protected]","company":"SP","eventsPerMonth":"< 1 million","serviceType":"unsure"}}}""".noSpaces,
-        "contexts" -> json"""{"data":[{"data":{"osType":"OSX","appleIdfa":"some_appleIdfa","openIdfa":"some_Idfa","carrier":"some_carrier","deviceModel":"large","osVersion":"3.0.0","appleIdfa":"some_appleIdfa","androidIdfa":"some_androidIdfa","deviceManufacturer":"Amstrad"},"schema":"iglu:com.snowplowanalytics.snowplow/mobile_context/jsonschema/1-0-0"},{"data":{"longitude":10,"bearing":50,"speed":16,"altitude":20,"altitudeAccuracy":0.3,"latitudeLongitudeAccuracy":0.5,"latitude":7},"schema":"iglu:com.snowplowanalytics.snowplow/geolocation_context/jsonschema/1-0-0"}],"schema":"iglu:com.snowplowanalytics.snowplow/contexts/jsonschema/1-0-0"}""".noSpaces,
+        "contexts" -> json"""{"data":[{"data":{"osType":"OSX","appleIdfa":"some_appleIdfa","openIdfa":"some_Idfa","carrier":"some_carrier","deviceModel":"large","osVersion":"3.0.0","appleIdfa":"some_appleIdfa","androidIdfa":"some_androidIdfa","deviceManufacturer":"Amstrad"},"schema":"iglu:com.snowplowanalytics.snowplow/mobile_context/jsonschema/1-0-0"},{"data":{"longitude":10,"bearing":50,"speed":16,"altitude":20,"altitudeAccuracy":0.3,"latitudeLongitudeAccuracy":0.5,"latitude":7},"schema":"iglu:com.snowplowanalytics.snowplow/geolocation_context/jsonschema/1-0-0"}],"schema":"iglu:com.snowplowanalytics.snowplow/contexts/jsonschema/1-0-1"}""".noSpaces,
         "derived_contexts" -> json"""{"schema":"iglu:com.snowplowanalytics.snowplow/contexts/jsonschema/1-0-1","data":[{"schema":"iglu:com.statusgator/status_change/jsonschema/1-0-0","data":{"serviceName": "sp-api-request-enrichment"}}]}""".noSpaces
       )
       BlackBoxTesting.runTest(input, expected, Some(ApiRequestEnrichmentSpec.conf))

modules/common-fs2/src/test/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/blackbox/misc/TransactionSpec.scala

@@ -49,7 +49,7 @@ class TransactionSpec extends Specification with CatsIO {
         "txn_id" -> "28288",
         "tr_country" -> "UK",
         "tr_city" -> "London",
-        "contexts" -> json"""{"data":[{"schema":"iglu:com.snowplowanalytics.snowplow/uri_redirect/jsonschema/1-0-0","data":{"uri":"http://snowplowanalytics.com/"}}],"schema":"iglu:com.snowplowanalytics.snowplow/contexts/jsonschema/1-0-0"}""".noSpaces
+        "contexts" -> json"""{"data":[{"schema":"iglu:com.snowplowanalytics.snowplow/uri_redirect/jsonschema/1-0-0","data":{"uri":"http://snowplowanalytics.com/"}}],"schema":"iglu:com.snowplowanalytics.snowplow/contexts/jsonschema/1-0-1"}""".noSpaces
       )
       BlackBoxTesting.runTest(input, expected)
     }

modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/adapters/registry/snowplow/Tp2Adapter.scala

@@ -234,12 +234,13 @@ object Tp2Adapter extends Adapter {
             )
       _ <- client
              .check(sd)
-             .leftMap(e =>
-               NonEmptyList.one(
-                 FailureDetails.TrackerProtocolViolation
-                   .IgluError(sd.schema, e)
-               )
-             )
+             .leftMap({
+               case (e, _) =>
+                 NonEmptyList.one(
+                   FailureDetails.TrackerProtocolViolation
+                     .IgluError(sd.schema, e)
+                 )
+             })
              .subflatMap { _ =>
                schemaCriterion.matches(sd.schema) match {
                  case true => ().asRight

modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/EnrichmentManager.scala

@@ -80,10 +80,11 @@ object EnrichmentManager {
                                unstructEvent,
                                featureFlags.legacyEnrichmentOrder
                              )
-      _ <- EitherT.rightT[F, BadRow] {
-             if (enrichmentsContexts.nonEmpty)
-               enriched.derived_contexts = ME.formatDerivedContexts(enrichmentsContexts)
-           }
+      _ = {
+        ME.formatUnstructEvent(unstructEvent).foreach(e => enriched.unstruct_event = e)
+        ME.formatContexts(inputContexts).foreach(c => enriched.contexts = c)
+        ME.formatContexts(enrichmentsContexts).foreach(c => enriched.derived_contexts = c)
+      }
       _ <- IgluUtils
              .validateEnrichmentsContexts[F](client, enrichmentsContexts, raw, processor, enriched)
       _ <- EitherT.rightT[F, BadRow](

modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/EnrichmentRegistry.scala

@@ -65,7 +65,7 @@ object EnrichmentRegistry {
             )
       _ <- client
              .check(sd)
-             .leftMap(e => NonEmptyList.one(e.asJson.noSpaces))
+             .leftMap({ case (e, _) => NonEmptyList.one(e.asJson.noSpaces) })
              .subflatMap { _ =>
                EnrichmentConfigSchemaCriterion.matches(sd.schema) match {
                  case true => ().asRight
@@ -92,9 +92,10 @@ object EnrichmentRegistry {
                              )
                        _ <- client
                               .check(sd)
-                              .leftMap(e =>
-                                NonEmptyList.one(s"Enrichment with key '${sd.schema.toSchemaUri}` is invalid - ${e.asJson.noSpaces}")
-                              )
+                              .leftMap({
+                                case (e, _) =>
+                                  NonEmptyList.one(s"Enrichment with key '${sd.schema.toSchemaUri}` is invalid - ${e.asJson.noSpaces}")
+                              })
                        conf <- EitherT.fromEither[F](
                                  buildEnrichmentConfig(sd.schema, sd.data, localMode).toEither
                                )

modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/MiscEnrichments.scala

@@ -30,6 +30,9 @@ object MiscEnrichments {
   val ContextsSchema =
     SchemaKey("com.snowplowanalytics.snowplow", "contexts", "jsonschema", SchemaVer.Full(1, 0, 1))
 
+  val UnstructEventSchema =
+    SchemaKey("com.snowplowanalytics.snowplow", "unstruct_event", "jsonschema", SchemaVer.Full(1, 0, 0))
+
   /**
    * The version of this ETL. Appends this version to the supplied "host" ETL.
    * @param processor The version of the host ETL running this library
@@ -82,6 +85,11 @@ object MiscEnrichments {
     }
 
   /** Turn a list of custom contexts into a self-describing JSON property */
-  def formatDerivedContexts(derivedContexts: List[SelfDescribingData[Json]]): String =
-    SelfDescribingData(ContextsSchema, Json.arr(derivedContexts.map(_.normalize): _*)).asString
+  def formatContexts(contexts: List[SelfDescribingData[Json]]): Option[String] =
+    if (contexts.isEmpty) None
+    else Some(SelfDescribingData(ContextsSchema, Json.arr(contexts.map(_.normalize): _*)).asString)
+
+  /** Turn a unstruct event into a self-describing JSON property */
+  def formatUnstructEvent(unstructEvent: Option[SelfDescribingData[Json]]): Option[String] =
+    unstructEvent.map(e => SelfDescribingData(UnstructEventSchema, e.normalize).asString)
 }

modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/utils/IgluUtils.scala

@@ -17,14 +17,16 @@ import cats.data.{EitherT, NonEmptyList, Validated, ValidatedNel}
 import cats.effect.Clock
 import cats.implicits._
 
-import io.circe.Json
+import io.circe._
+import io.circe.syntax._
+import io.circe.generic.semiauto._
 
 import java.time.Instant
 
 import com.snowplowanalytics.iglu.client.{ClientError, IgluCirceClient}
 import com.snowplowanalytics.iglu.client.resolver.registries.RegistryLookup
 
-import com.snowplowanalytics.iglu.core.{SchemaCriterion, SchemaKey, SelfDescribingData}
+import com.snowplowanalytics.iglu.core.{SchemaCriterion, SchemaKey, SchemaVer, SelfDescribingData}
 import com.snowplowanalytics.iglu.core.circe.implicits._
 
 import com.snowplowanalytics.snowplow.badrows._
@@ -71,7 +73,11 @@ object IgluUtils {
                       .map(_.toValidatedNel)
       } yield (contexts, unstruct)
         .mapN { (c, ue) =>
-          (c, ue)
+          val supersedingInfoContexts = (c.flatMap(_.supersedingInfo) ::: ue.flatMap(_.supersedingInfo).toList).distinct
+            .map(_.toSdj)
+          val contexts = c.map(_.sdj) ::: supersedingInfoContexts
+          val unstruct = ue.map(_.sdj)
+          (contexts, unstruct)
         }
         .leftMap { schemaViolations =>
           buildSchemaViolationsBadRow(
@@ -98,27 +104,17 @@ object IgluUtils {
     client: IgluCirceClient[F],
     field: String = "ue_properties",
     criterion: SchemaCriterion = SchemaCriterion("com.snowplowanalytics.snowplow", "unstruct_event", "jsonschema", 1, 0)
-  ): F[Validated[FailureDetails.SchemaViolation, Option[SelfDescribingData[Json]]]] =
+  ): F[Validated[FailureDetails.SchemaViolation, Option[SdjExtractResult]]] =
     (Option(enriched.unstruct_event) match {
       case Some(rawUnstructEvent) =>
         for {
           // Validate input Json string and extract unstructured event
           unstruct <- extractInputData(rawUnstructEvent, field, criterion, client)
           // Parse Json unstructured event as SelfDescribingData[Json]
-          unstructSDJ <- SelfDescribingData
-                           .parse(unstruct)
-                           .leftMap(FailureDetails.SchemaViolation.NotIglu(unstruct, _))
-                           .toEitherT[F]
-          // Check SelfDescribingData[Json] of unstructured event
-          _ <- check(client, unstructSDJ)
-                 .leftMap {
-                   case (schemaKey, clientError) =>
-                     FailureDetails.SchemaViolation.IgluError(schemaKey, clientError)
-                 }
-                 .leftWiden[FailureDetails.SchemaViolation]
+          unstructSDJ <- parseAndValidateSDJ_sv(unstruct, client)
         } yield unstructSDJ.some
       case None =>
-        EitherT.rightT[F, FailureDetails.SchemaViolation](none[SelfDescribingData[Json]])
+        EitherT.rightT[F, FailureDetails.SchemaViolation](none[SdjExtractResult])
     }).toValidated
 
   /**
@@ -135,7 +131,7 @@ object IgluUtils {
     client: IgluCirceClient[F],
     field: String = "contexts",
     criterion: SchemaCriterion = SchemaCriterion("com.snowplowanalytics.snowplow", "contexts", "jsonschema", 1, 0)
-  ): F[ValidatedNel[FailureDetails.SchemaViolation, List[SelfDescribingData[Json]]]] =
+  ): F[ValidatedNel[FailureDetails.SchemaViolation, List[SdjExtractResult]]] =
     (Option(enriched.contexts) match {
       case Some(rawContexts) =>
         for {
@@ -153,7 +149,7 @@ object IgluUtils {
         } yield contextsSDJ
       case None =>
         EitherT.rightT[F, NonEmptyList[FailureDetails.SchemaViolation]](
-          List.empty[SelfDescribingData[Json]]
+          List.empty[SdjExtractResult]
         )
     }).toValidated
 
@@ -238,10 +234,13 @@ object IgluUtils {
   private def check[F[_]: Monad: RegistryLookup: Clock](
     client: IgluCirceClient[F],
     sdj: SelfDescribingData[Json]
-  ): EitherT[F, (SchemaKey, ClientError), Unit] =
+  ): EitherT[F, (SchemaKey, ClientError), Option[SchemaVer.Full]] =
     client
       .check(sdj)
-      .leftMap(clientErr => (sdj.schema, clientErr))
+      .leftMap({
+        case (e, None) => (sdj.schema, e)
+        case (e, Some(superseding)) => (sdj.schema.copy(version = superseding), e)
+      })
 
   /** Check a list of SDJs and merge the Iglu errors */
   private def checkList[F[_]: Monad: RegistryLookup: Clock](
@@ -259,20 +258,48 @@ object IgluUtils {
   private def parseAndValidateSDJ_sv[F[_]: Monad: RegistryLookup: Clock]( // _sv for SchemaViolation
     json: Json,
     client: IgluCirceClient[F]
-  ): EitherT[F, FailureDetails.SchemaViolation, SelfDescribingData[Json]] =
+  ): EitherT[F, FailureDetails.SchemaViolation, SdjExtractResult] =
     for {
       sdj <- SelfDescribingData
                .parse(json)
                .leftMap(FailureDetails.SchemaViolation.NotIglu(json, _))
                .toEitherT[F]
-      _ <- check(client, sdj)
-             .leftMap {
-               case (schemaKey, clientError) =>
-                 FailureDetails.SchemaViolation
-                   .IgluError(schemaKey, clientError): FailureDetails.SchemaViolation
+      supersedingSchema <- check(client, sdj)
+                             .leftMap {
+                               case (schemaKey, clientError) =>
+                                 FailureDetails.SchemaViolation
+                                   .IgluError(schemaKey, clientError): FailureDetails.SchemaViolation
 
-             }
-    } yield sdj
+                             }
+      supersedingInfo = supersedingSchema.map(s => SchemaSupersedingInfo(sdj.schema, s))
+      sdjUpdated = replaceSchemaVersion(sdj, supersedingInfo)
+    } yield SdjExtractResult(sdjUpdated, supersedingInfo)
+
+  private def replaceSchemaVersion(
+    sdj: SelfDescribingData[Json],
+    supersedingInfo: Option[SchemaSupersedingInfo]
+  ): SelfDescribingData[Json] =
+    supersedingInfo match {
+      case None => sdj
+      case Some(s) => sdj.copy(schema = sdj.schema.copy(version = s.supersededBy))
+    }
+
+  case class SchemaSupersedingInfo(originalSchema: SchemaKey, supersededBy: SchemaVer.Full) {
+    def toSdj: SelfDescribingData[Json] =
+      SelfDescribingData(SchemaSupersedingInfo.schemaKey, (this: SchemaSupersedingInfo).asJson)
+  }
+
+  object SchemaSupersedingInfo {
+    val schemaKey = SchemaKey("com.snowplowanalytics.iglu", "superseding_info", "jsonschema", SchemaVer.Full(1, 0, 0))
+
+    implicit val schemaVerFullEncoder: Encoder[SchemaVer.Full] =
+      Encoder.encodeString.contramap(v => v.asString)
+
+    implicit val schemaSupersedingInfoEncoder: Encoder[SchemaSupersedingInfo] =
+      deriveEncoder[SchemaSupersedingInfo]
+  }
+
+  case class SdjExtractResult(sdj: SelfDescribingData[Json], supersedingInfo: Option[SchemaSupersedingInfo])
 
   /** Build `BadRow.SchemaViolations` from a list of `FailureDetails.SchemaViolation`s */
   def buildSchemaViolationsBadRow(

modules/common/src/test/resources/iglu-schemas/schemas/com.acme/superseding_example/jsonschema/1-0-0

@@ -0,0 +1,25 @@
+{
+  "$schema": "http://iglucentral.com/schemas/com.snowplowanalytics.self-desc/schema/jsonschema/1-0-0#",
+  "description": "Superseding schema example",
+  "self": {
+    "vendor": "com.acme",
+    "name": "superseding_schema",
+    "format": "jsonschema",
+    "version": "1-0-0",
+    "supersededBy": "1-0-1"
+  },
+  "type": "object",
+  "properties": {
+    "field_a": {
+      "type": "string"
+    },
+    "field_b": {
+      "type": "string"
+    },
+    "field_c": {
+      "type": "string"
+    }
+  },
+  "required": ["field_a", "field_b"],
+  "additionalProperties": false
+}

modules/common/src/test/resources/iglu-schemas/schemas/com.acme/superseding_example/jsonschema/1-0-1

@@ -0,0 +1,27 @@
+{
+  "$schema": "http://iglucentral.com/schemas/com.snowplowanalytics.self-desc/schema/jsonschema/1-0-0#",
+  "description": "Superseding schema example",
+  "self": {
+    "vendor": "com.acme",
+    "name": "superseding_schema",
+    "format": "jsonschema",
+    "version": "1-0-1"
+  },
+  "type": "object",
+  "properties": {
+    "field_a": {
+      "type": "string"
+    },
+    "field_b": {
+      "type": "string"
+    },
+    "field_c": {
+      "type": "string"
+    },
+    "field_d": {
+      "type": "string"
+    }
+  },
+  "required": ["field_a", "field_b"],
+  "additionalProperties": false
+}

modules/common/src/test/resources/iglu-schemas/schemas/com.acme/superseding_example/jsonschema/2-0-0

@@ -0,0 +1,22 @@
+{
+  "$schema": "http://iglucentral.com/schemas/com.snowplowanalytics.self-desc/schema/jsonschema/1-0-0#",
+  "description": "Superseding schema example",
+  "self": {
+    "vendor": "com.acme",
+    "name": "superseding_schema",
+    "format": "jsonschema",
+    "version": "2-0-0",
+    "supersededBy": "2-0-1"
+  },
+  "type": "object",
+  "properties": {
+    "field_e": {
+      "type": "string"
+    },
+    "field_f": {
+      "type": "string"
+    }
+  },
+  "required": ["field_e", "field_f"],
+  "additionalProperties": false
+}

modules/common/src/test/resources/iglu-schemas/schemas/com.acme/superseding_example/jsonschema/2-0-1

@@ -0,0 +1,24 @@
+{
+  "$schema": "http://iglucentral.com/schemas/com.snowplowanalytics.self-desc/schema/jsonschema/1-0-0#",
+  "description": "Superseding schema example",
+  "self": {
+    "vendor": "com.acme",
+    "name": "superseding_schema",
+    "format": "jsonschema",
+    "version": "2-0-1"
+  },
+  "type": "object",
+  "properties": {
+    "field_e": {
+      "type": "string"
+    },
+    "field_f": {
+      "type": "string"
+    },
+    "field_g": {
+      "type": "string"
+    }
+  },
+  "required": ["field_e", "field_f"],
+  "additionalProperties": false
+}