Scala Common Enrich: ensured that all timestamp fields are nonnegative (

closes #1938)

3-enrich/scala-common-enrich/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/EventEnrichments.scala

@@ -77,6 +77,19 @@ object EventEnrichments {
     }
   }
 
+  /**
+   * Validate that the collector timestamp is set and valid
+   *
+   * @param collectorTstamp
+   * @return Validated collector timestamp
+   */
+  def validateCollectorTstamp(collectorTstamp: Option[DateTime]): Validation[String, String] = {
+    collectorTstamp match {
+      case None => "No collector_tstamp set".fail
+      case Some(t) => extractTimestamp("collector_tstamp", t.getMillis.toString)
+    }
+  }
+
   /**
    * Extracts the timestamp from the
    * format as laid out in the Tracker
@@ -95,7 +108,12 @@ object EventEnrichments {
   val extractTimestamp: (String, String) => ValidatedString = (field, tstamp) =>
     try {
       val dt = new DateTime(tstamp.toLong)
-      toTimestamp(dt).success
+      val timestampString = toTimestamp(dt)
+      if (timestampString.startsWith("-")) {
+        s"Field [$field]: [$tstamp] is formatted as [$timestampString] which isn't Redshift-compatible".fail
+      } else {
+        timestampString.success
+      }
     } catch {
       case nfe: NumberFormatException =>
         "Field [%s]: [%s] is not in the expected format (ms since epoch)".format(field, tstamp).fail