/***********************************************************************
 * Copyright (c) 2013-2025 Commonwealth Computer Research, Inc.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Apache License, Version 2.0
 * which accompanies this distribution and is available at
 * http://www.opensource.org/licenses/apache2.0.php.
 ***********************************************************************/

package org.locationtech.geomesa.features

import org.geotools.api.feature.simple.SimpleFeature
import org.junit.runner.RunWith
import org.locationtech.geomesa.features.SerializationOption.SerializationOptions
import org.locationtech.geomesa.features.avro.AvroFeatureSerializer
import org.locationtech.geomesa.features.kryo.{KryoFeatureSerializer, ProjectingKryoFeatureDeserializer}
import org.locationtech.geomesa.security
import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes
import org.locationtech.geomesa.utils.text.WKTUtils
import org.locationtech.jts.geom.Point
import org.specs2.matcher.{MatchResult, Matcher}
import org.specs2.mutable.Specification
import org.specs2.runner.JUnitRunner

import java.util.UUID

@RunWith(classOf[JUnitRunner])
class SimpleFeatureSerializersTest extends Specification {

  sequential

  val sftName = "SimpleFeatureSerializersTest"
  val sft = SimpleFeatureTypes.createType(sftName, "name:String,*geom:Point,dtg:Date")

  def getFeatures: Seq[SimpleFeature] = (0 until 6).map { i =>
    ScalaSimpleFeature.create(sft, i.toString, i.toString, "POINT(-110 30)", "2012-01-02T05:06:07.000Z")
  }

  def getFeaturesWithVisibility: Seq[SimpleFeature] = {
    import security._

    val features = getFeatures
    val visibilities = Seq("test&usa", "admin&user", "", null, "test", "user")

    features.zip(visibilities).map { case (sf, vis) =>
      sf.visibility = vis
      sf
    }

    features
  }

  "SimpleFeatureEncoder" should {

    "have a properly working apply() method" >> {
      val opts = SerializationOptions.withUserData

      // AVRO without options
      val avro1 = new AvroFeatureSerializer(sft)
      avro1 must beAnInstanceOf[AvroFeatureSerializer]
      avro1.options mustEqual SerializationOptions.none

      // AVRO with options
      val avro2 = new AvroFeatureSerializer(sft, opts)
      avro2 must beAnInstanceOf[AvroFeatureSerializer]
      avro2.options mustEqual opts

      // KRYO without options
      val kryo1 = KryoFeatureSerializer(sft)
      kryo1 must beAnInstanceOf[KryoFeatureSerializer]
      kryo1.options mustEqual SerializationOptions.none

      // KRYO with options
      val kryo2 = KryoFeatureSerializer(sft, opts)
      kryo2 must beAnInstanceOf[KryoFeatureSerializer]
      kryo2.options mustEqual opts
    }
  }

  "AvroFeatureSerializer" should {

    "be able to encode points" >> {
      val encoder = new AvroFeatureSerializer(sft)
      val features = getFeatures

      val encoded = features.map(encoder.serialize)
      encoded must not(beNull)
      encoded must have size features.size
    }

    "not include user data when not requested" >> {
      val encoder = new AvroFeatureSerializer(sft)
      val expected = getFeatures.map(encoder.serialize)

      val featuresWithVis = getFeaturesWithVisibility
      val actual = featuresWithVis.map(encoder.serialize)

      actual must haveSize(expected.size)

      forall(actual.zip(expected)) {
        case (a, e) => a mustEqual e
      }
    }

    "include user data when requested" >> {
      val noUserData = {
        val encoder = new AvroFeatureSerializer(sft, SerializationOptions.none)
        getFeatures.map(encoder.serialize)
      }
      val withUserData = {
        val encoder = new AvroFeatureSerializer(sft, SerializationOptions.withUserData)
        getFeaturesWithVisibility.map(encoder.serialize)
      }

      withUserData must haveSize(noUserData.size)

      forall(withUserData.zip(noUserData)) {
        case (y, n) => y.length must beGreaterThan(n.length)
      }
    }
  }

  "AvroFeatureDeserializer" should {

    "be able to decode points" >> {
      val encoder = new AvroFeatureSerializer(sft)

      val features = getFeatures
      val encoded = features.map(encoder.serialize)

      val decoded = encoded.map(encoder.deserialize)
      decoded must equalFeatures(features, withoutUserData)
    }

    "be able to decode points with user data" >> {
      val encoder = new AvroFeatureSerializer(sft, SerializationOptions.withUserData)

      val features = getFeaturesWithVisibility
      val encoded = features.map(encoder.serialize)

      val decoded = encoded.map(encoder.deserialize)

      decoded must equalFeatures(features)
    }

    "work when user data were encoded but are not expected by decoder" >> {
      // in this case the encoded user data will be ignored
      val sf = getFeaturesWithVisibility.head
      val encoder = new AvroFeatureSerializer(sft, SerializationOptions.withUserData)
      val encoded = encoder.serialize(sf)

      val decoder = new AvroFeatureSerializer(sft, SerializationOptions.none)

      decoder.deserialize(encoded) must equalSF(sf, withoutUserData)
    }

    "fail when user data were not encoded but are expected by the decoder" >> {
      val encoder = new AvroFeatureSerializer(sft, SerializationOptions.none)
      val encoded = encoder.serialize(getFeaturesWithVisibility.head)

      val decoder = new AvroFeatureSerializer(sft, SerializationOptions.withUserData)

      decoder.deserialize(encoded) must throwA[Exception]
    }
  }

  "KryoFeatureEncoder" should {

    "be able to encode points" >> {
      val encoder = KryoFeatureSerializer(sft)
      val features = getFeatures

      val encoded = features.map(encoder.serialize)
      encoded must not(beNull)
      encoded must have size features.size
    }

    "not include visibilities when not requested" >> {
      val encoder = KryoFeatureSerializer(sft)
      val expected = getFeatures.map(encoder.serialize)

      val featuresWithVis = getFeaturesWithVisibility
      val actual = featuresWithVis.map(encoder.serialize)

      actual must haveSize(expected.size)

      forall(actual.zip(expected)) {
        case (a, e) => a mustEqual e
      }
    }

    "include user data when requested" >> {
      val noVis = {
        val encoder = KryoFeatureSerializer(sft, SerializationOptions.none)
        getFeatures.map(encoder.serialize)
      }
      val withVis = {
        val encoder = KryoFeatureSerializer(sft, SerializationOptions.withUserData)
        getFeaturesWithVisibility.map(encoder.serialize)
      }

      withVis must haveSize(noVis.size)

      forall(withVis.zip(noVis)) {
        case (y, n) => y.length must beGreaterThan(n.length)
      }
    }
  }

  "KryoFeatureDecoder" should {

    "be able to decode points" >> {
      val encoder = KryoFeatureSerializer(sft)
      val decoder = KryoFeatureSerializer(sft)

      val features = getFeatures
      val encoded = features.map(encoder.serialize)

      val decoded = encoded.map(decoder.deserialize)
      decoded must equalFeatures(features, withoutUserData)
    }

    "be able to decode points with user data" >> {
      val encoder = KryoFeatureSerializer(sft, SerializationOptions.withUserData)
      val decoder = KryoFeatureSerializer(sft, SerializationOptions.withUserData)

      val features = getFeaturesWithVisibility
      val encoded = features.map(encoder.serialize)

      val decoded = encoded.map(decoder.deserialize)

      forall(decoded.zip(features)) { case (d, sf) =>
        d.getID mustEqual sf.getID
        d.getAttributes mustEqual sf.getAttributes
        d.getUserData mustEqual sf.getUserData
      }
    }

    "work user data were encoded but are not expected by decoder" >> {
      // in this case the encoded user data will be ignored
      val sf = getFeaturesWithVisibility.head
      val encoder = KryoFeatureSerializer(sft, SerializationOptions.withUserData)
      val encoded = encoder.serialize(sf)

      val decoder = KryoFeatureSerializer(sft, SerializationOptions.none)

      decoder.deserialize(encoded) must equalSF(sf, withoutUserData)
    }

    "not fail when user data was not encoded but is expected by the decoder" >> {
      val encoder = KryoFeatureSerializer(sft, SerializationOptions.none)
      val encoded = encoder.serialize(getFeaturesWithVisibility.head)

      val decoder = KryoFeatureSerializer(sft, SerializationOptions.withUserData)

      val decoded = decoder.deserialize(encoded)
      decoded.getUserData.size mustEqual 0
    }
  }

  "ProjectingKryoFeatureDecoder" should {

    "properly project features" >> {
      val encoder = KryoFeatureSerializer(sft)

      val projectedSft = SimpleFeatureTypes.createType("projectedTypeName", "*geom:Point")
      val projectingDecoder = new ProjectingKryoFeatureDeserializer(sft, projectedSft)

      val features = getFeatures
      val encoded = features.map(encoder.serialize)
      val decoded = encoded.map(projectingDecoder.deserialize)

      decoded.map(_.getID) mustEqual features.map(_.getID)
      decoded.map(_.getDefaultGeometry) mustEqual features.map(_.getDefaultGeometry)

      forall(decoded) { sf =>
        sf.getAttributeCount mustEqual 1
        sf.getAttribute(0) must beAnInstanceOf[Point]
        sf.getFeatureType mustEqual projectedSft
      }
    }

    "be able to decode points with user data" >> {
      val encoder = KryoFeatureSerializer(sft, SerializationOptions.withUserData)

      val projectedSft = SimpleFeatureTypes.createType("projectedTypeName", "*geom:Point")
      val decoder = new ProjectingKryoFeatureDeserializer(sft, projectedSft, SerializationOptions.withUserData)

      val features = getFeaturesWithVisibility
      val encoded = features.map(encoder.serialize)

      val decoded = encoded.map(decoder.deserialize)

      forall(features.zip(decoded)) { case (in, out) =>
        out.getUserData mustEqual in.getUserData
      }
    }
  }

  "SimpleFeatureSerializers" should {
    "serialize user data byte arrays, uuids, geometries, and lists" >> {

      val features = getFeatures
      var i = 0
      features.foreach { f =>
        f.getUserData.put("bytes", Array.fill[Byte](i)(i.toByte))
        f.getUserData.put("uuid", UUID.randomUUID())
        f.getUserData.put("point", WKTUtils.read(s"POINT (4$i 55)"))
        f.getUserData.put("geom", WKTUtils.read(s"LINESTRING (4$i 55, 4$i 56, 4$i 57)"))
        f.getUserData.put("list", java.util.Arrays.asList("foo", s"bar$i", 5, i))
        i += 1
      }

      def doTest(typ: SerializationType.SerializationType): MatchResult[Any] = {
        val serializer = typ match {
          case SerializationType.KRYO => KryoFeatureSerializer(sft, SerializationOptions.withUserData)
          case SerializationType.AVRO => new AvroFeatureSerializer(sft, SerializationOptions.withUserData)
        }
        foreach(features) { feature =>
          val reserialized = serializer.deserialize(serializer.serialize(feature))
          // note: can't compare whole map at once as byte arrays aren't considered equal in that comparison
          foreach(Seq("bytes", "uuid", "point", "geom", "list")) { key =>
            reserialized.getUserData.get(key) mustEqual feature.getUserData.get(key)
          }
        }
      }

      doTest(SerializationType.KRYO)
      doTest(SerializationType.AVRO).pendingUntilFixed("Avro user data serialization")
    }
  }

  type MatcherFactory[T] = (T) => Matcher[T]
  type UserDataMap = java.util.Map[AnyRef, AnyRef]

  val withoutUserData: MatcherFactory[UserDataMap] = {
    expected: UserDataMap => actual: UserDataMap => actual.isEmpty must beTrue
  }

  val withUserData: MatcherFactory[UserDataMap] = {
    expected: UserDataMap => actual: UserDataMap => actual mustEqual expected
  }

  def equalFeatures(expected: Seq[SimpleFeature], withUserDataMatcher: MatcherFactory[UserDataMap] = withUserData): Matcher[Seq[SimpleFeature]] = {
    actual: Seq[SimpleFeature] => {
      actual must not(beNull)
      actual must haveSize(expected.size)

      forall(actual zip expected) {
        case (act, exp) => act must equalSF(exp, withUserDataMatcher)
      }
    }
  }

  def equalSF(expected: SimpleFeature, matchUserData: MatcherFactory[UserDataMap] = withUserData): Matcher[SimpleFeature] = {
    sf: SimpleFeature => {
      sf.getID mustEqual expected.getID
      sf.getDefaultGeometry mustEqual expected.getDefaultGeometry
      sf.getAttributes mustEqual expected.getAttributes
      sf.getUserData must matchUserData(expected.getUserData)
    }
  }
}
