/*
 * Copyright 2015-2016 IBM Corporation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package system.packages

import system.utils.KafkaUtils

import org.apache.kafka.clients.producer.ProducerRecord;

import scala.concurrent.duration.DurationInt
import scala.language.postfixOps

import org.junit.runner.RunWith
import org.scalatest.BeforeAndAfterAll
import org.scalatest.FlatSpec
import org.scalatest.Matchers
import org.scalatest.junit.JUnitRunner

import spray.json.DefaultJsonProtocol._
import spray.json.JsObject
import spray.json.JsString
import spray.json.pimpAny

import common.JsHelpers
import common.TestHelpers
import common.Wsk
import common.WskActorSystem
import common.WskProps
import common.WskTestHelpers
import ActionHelper._

import java.util.Base64
import java.nio.charset.StandardCharsets

@RunWith(classOf[JUnitRunner])
class MessageHubFeedTests
  extends FlatSpec
  with Matchers
  with WskActorSystem
  with BeforeAndAfterAll
  with TestHelpers
  with WskTestHelpers
  with JsHelpers {

  val topic = "test"
  val sessionTimeout = 10 seconds

  val messagingPackage = "/whisk.system/messaging"
  val messageHubFeed = "messageHubFeed"
  val messageHubProduce = "messageHubProduce"

  val consumerInitTime = 10000 // ms

  val kafkaUtils = new KafkaUtils

  implicit val wskprops = WskProps()
  val wsk = new Wsk()
  val actionName = s"${messagingPackage}/${messageHubFeed}"

  behavior of "Message Hub feed action"

  it should "reject invocation when topic argument is missing" in {
    val expectedOutput = JsObject(
      "error" -> JsString("You must supply a 'topic' parameter.")
    )

    runActionWithExpectedResult(actionName, "dat/missingTopic.json", expectedOutput, false)
  }

  it should "reject invocation when kafka_brokers_sasl argument is missing" in {
    val expectedOutput = JsObject(
      "error" -> JsString("You must supply a 'kafka_brokers_sasl' parameter.")
    )

    runActionWithExpectedResult(actionName, "dat/missingBrokers.json", expectedOutput, false)
  }

  it should "reject invocation when kafka_admin_url argument is missing" in {
    val expectedOutput = JsObject("error" -> JsString(
      "You must supply a 'kafka_admin_url' parameter.")
    )

    runActionWithExpectedResult(actionName, "dat/missingAdminURL.json", expectedOutput, false)
  }

  it should "reject invocation when user argument is missing" in {
    val expectedOutput = JsObject(
      "error" -> JsString("You must supply a 'user' parameter to authenticate with Message Hub.")
    )

    runActionWithExpectedResult(actionName, "dat/missingUser.json", expectedOutput, false)
  }

  it should "reject invocation when password argument is missing" in {
    val expectedOutput = JsObject(
      "error" -> JsString("You must supply a 'password' parameter to authenticate with Message Hub.")
    )

    runActionWithExpectedResult(actionName, "dat/missingPassword.json", expectedOutput, false)
  }

  it should "reject invocation when isJSONData and isBinaryValue are both enable" in {
    val expectedOutput = JsObject(
      "error" -> JsString("isJSONData and isBinaryValue cannot both be enabled."))

    runActionWithExpectedResult(actionName, "dat/multipleValueTypes.json", expectedOutput, false)
  }

  it should "fire a trigger when a binary message is posted to message hub" in withAssetCleaner(wskprops) {
    val currentTime = s"${System.currentTimeMillis}"

    (wp, assetHelper) =>
      val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
      println(s"Creating trigger ${triggerName}")

      createTrigger(assetHelper, triggerName, parameters = Map(
        "user" -> kafkaUtils.getAsJson("user"),
        "password" -> kafkaUtils.getAsJson("password"),
        "api_key" -> kafkaUtils.getAsJson("api_key"),
        "kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
        "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
        "topic" -> topic.toJson,
        "isBinaryKey" -> true.toJson,
        "isBinaryValue" -> true.toJson))

      // It takes a moment for the consumer to fully initialize.
      println("Giving the consumer a moment to get ready")
      Thread.sleep(consumerInitTime)

      // key to use for the produced message
      val key = "TheKey"
      val encodedCurrentTime = Base64.getEncoder.encodeToString(currentTime.getBytes(StandardCharsets.UTF_8))
      val encodedKey = Base64.getEncoder.encodeToString(key.getBytes(StandardCharsets.UTF_8))

      withActivation(wsk.activation, wsk.action.invoke(s"$messagingPackage/$messageHubProduce", Map(
        "user" -> kafkaUtils.getAsJson("user"),
        "password" -> kafkaUtils.getAsJson("password"),
        "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
        "topic" -> topic.toJson,
        "key" -> key.toJson,
        "value" -> currentTime.toJson))) {
        _.response.success shouldBe true
      }

      println("Polling for activations")
      val activations = wsk.activation.pollFor(N = 1, Some(triggerName), retries = 60)
      assert(activations.length > 0)

      val matchingActivations = for {
        id <- activations
        activation = wsk.activation.waitForActivation(id)
        if (activation.isRight && activation.right.get.fields.get("response").toString.contains(encodedCurrentTime))
      } yield activation.right.get

      assert(matchingActivations.length == 1)

      val activation = matchingActivations.head
      activation.getFieldPath("response", "success") shouldBe Some(true.toJson)

      // assert that there exists a message in the activation which has the expected keys and values
      val messages = KafkaUtils.messagesInActivation(activation, field = "value", value = encodedCurrentTime)
      assert(messages.length == 1)

      val message = messages.head
      message.getFieldPath("topic") shouldBe Some(topic.toJson)
      message.getFieldPath("key") shouldBe Some(encodedKey.toJson)
  }

  it should "not fire a single trigger with an oversized payload" in withAssetCleaner(wskprops) {
    // payload size should be under the payload limit, but greater than 50% of the limit
    val testPayloadSize = 600000

    // create trigger
    val currentTime = s"${System.currentTimeMillis}"

    (wp, assetHelper) =>
      val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
      println(s"Creating trigger ${triggerName}")

      createTrigger(assetHelper, triggerName, parameters = Map(
        "user" -> kafkaUtils.getAsJson("user"),
        "password" -> kafkaUtils.getAsJson("password"),
        "api_key" -> kafkaUtils.getAsJson("api_key"),
        "kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
        "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
        "topic" -> topic.toJson,
        "isBinaryKey" -> false.toJson,
        "isBinaryValue" -> false.toJson))

      // It takes a moment for the consumer to fully initialize.
      println("Giving the consumer a moment to get ready")
      Thread.sleep(consumerInitTime)

      // Rapidly produce two messages whose size are each greater than half the allowed payload limit.
      // This should ensure that the feed fires these as two separate triggers.
      println("Rapidly producing two large messages")
      val producer = kafkaUtils.createProducer()
      val firstMessage = new ProducerRecord(topic, "key", generateMessage(s"first${currentTime}", testPayloadSize))
      val secondMessage = new ProducerRecord(topic, "key", generateMessage(s"second${currentTime}", testPayloadSize))
      producer.send(firstMessage)
      producer.send(secondMessage)
      producer.close()

      // verify there are two trigger activations required to handle these messages
      println("Polling for activations")
      val activations = wsk.activation.pollFor(N = 2, Some(triggerName), retries = 60)

      println("Verifying activation content")
      val matchingActivations = for {
        id <- activations
        activation = wsk.activation.waitForActivation(id)
        if (activation.isRight && (activation.right.get.fields.get("response").toString.contains(s"first${currentTime}") ||
          activation.right.get.fields.get("response").toString.contains(s"second${currentTime}")))
      } yield activation.right.get

      assert(matchingActivations.length == 2)
  }

  it should "not fire a trigger for a single oversized message" in withAssetCleaner(wskprops) {
    // payload size should be over the payload limit
    val testPayloadSize = 2000000

    // create trigger
    val currentTime = s"${System.currentTimeMillis}"

    (wp, assetHelper) =>
      val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
      println(s"Creating trigger ${triggerName}")

      createTrigger(assetHelper, triggerName, parameters = Map(
        "user" -> kafkaUtils.getAsJson("user"),
        "password" -> kafkaUtils.getAsJson("password"),
        "api_key" -> kafkaUtils.getAsJson("api_key"),
        "kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
        "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
        "topic" -> topic.toJson,
        "isBinaryKey" -> false.toJson,
        "isBinaryValue" -> false.toJson))

      // It takes a moment for the consumer to fully initialize.
      println("Giving the consumer a moment to get ready")
      Thread.sleep(consumerInitTime)

      println("Producing an oversized message")
      val producer = kafkaUtils.createProducer()
      val bigMessage = new ProducerRecord(topic, "key", generateMessage(s"${currentTime}", testPayloadSize))
      producer.send(bigMessage)
      producer.close()

      // verify there are no activations that match
      println("Polling for activations")
      val activations = wsk.activation.pollFor(N = 1, Some(triggerName), retries = 60)

      println("Verifying activation content")
      val matchingActivations = for {
        id <- activations
        activation = wsk.activation.waitForActivation(id)
        if (activation.isRight && (activation.right.get.fields.get("response").toString.contains(s"first${currentTime}")))
      } yield activation.right.get

      assert(matchingActivations.length == 0)
  }

  def createTrigger(assetHelper: AssetCleaner, name: String, parameters: Map[String, spray.json.JsValue]) = {
    val feedCreationResult = assetHelper.withCleaner(wsk.trigger, name) {
      (trigger, _) =>
        trigger.create(name, feed = Some(s"$messagingPackage/$messageHubFeed"), parameters = parameters)
    }

    withActivation(wsk.activation, feedCreationResult, initialWait = 5 seconds, totalWait = 60 seconds) {
      activation =>
        // should be successful
        activation.response.success shouldBe true
    }
  }

  def generateMessage(prefix: String, size: Int): String = {
    val longString = Array.fill[String](size)("0").mkString("")
    s"${prefix}${longString}"
  }
}