/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.gearpump.cluster.master

import akka.actor._
import akka.pattern.ask
import com.typesafe.config.{Config, ConfigFactory}
import org.apache.gearpump.Time.MilliSeconds
import org.apache.gearpump.cluster.AppMasterToMaster.{AppDataSaved, SaveAppDataFailed, _}
import org.apache.gearpump.cluster.AppMasterToWorker._
import org.apache.gearpump.cluster.{ApplicationStatus, ApplicationTerminalStatus}
import org.apache.gearpump.cluster.ClientToMaster._
import org.apache.gearpump.cluster.MasterToAppMaster.{AppMasterData, AppMasterDataRequest, AppMastersDataRequest, _}
import org.apache.gearpump.cluster.MasterToClient._
import org.apache.gearpump.cluster.WorkerToAppMaster.{ShutdownExecutorFailed, _}
import org.apache.gearpump.cluster.appmaster.{ApplicationMetaData, ApplicationRuntimeInfo}
import org.apache.gearpump.cluster.master.AppManager._
import org.apache.gearpump.cluster.master.InMemoryKVService.{GetKVResult, PutKVResult, PutKVSuccess, _}
import org.apache.gearpump.cluster.master.Master._
import org.apache.gearpump.util.Constants._
import org.apache.gearpump.util.{ActorUtil, TimeOutScheduler, Util, _}
import org.slf4j.Logger

import scala.concurrent.Future
import scala.util.{Failure, Success}

/**
 * AppManager is dedicated child of Master to manager all applications.
 */
private[cluster] class AppManager(kvService: ActorRef, launcher: AppMasterLauncherFactory)
  extends Actor with Stash with TimeOutScheduler {

  private val LOG: Logger = LogUtil.getLogger(getClass)
  private val systemConfig: Config = context.system.settings.config

  private val appTotalRetries: Int = systemConfig.getInt(Constants.APPLICATION_TOTAL_RETRIES)

  implicit val timeout = FUTURE_TIMEOUT
  implicit val executionContext = context.dispatcher

  // Next available appId
  private var nextAppId: Int = 1

  private var applicationRegistry = Map.empty[Int, ApplicationRuntimeInfo]
  private var appResultListeners = Map.empty[Int, List[ActorRef]]

  private var appMasterRestartPolicies = Map.empty[Int, RestartPolicy]

  def receive: Receive = null

  kvService ! GetKV(MASTER_GROUP, MASTER_STATE)
  context.become(waitForMasterState)

  def waitForMasterState: Receive = {
    case GetKVSuccess(_, result) =>
      val masterState = result.asInstanceOf[MasterState]
      if (masterState != null) {
        this.nextAppId = masterState.maxId + 1
        this.applicationRegistry = masterState.applicationRegistry
      }
      context.become(receiveHandler)
      unstashAll()
    case GetKVFailed(ex) =>
      LOG.error("Failed to get master state, shutting down master to avoid data corruption...")
      context.parent ! PoisonPill
    case msg =>
      LOG.info(s"Get message ${msg.getClass.getSimpleName}")
      stash()
  }

  def receiveHandler: Receive = {
    val msg = "Application Manager started. Ready for application submission..."
    LOG.info(msg)
    clientMsgHandler orElse appMasterMessage orElse selfMsgHandler orElse workerMessage orElse
      appDataStoreService orElse terminationWatch
  }

  def clientMsgHandler: Receive = {
    case SubmitApplication(app, jar, username) =>
      LOG.info(s"Submit Application ${app.name}($nextAppId) by $username...")
      val client = sender()
      if (applicationNameExist(app.name)) {
        client ! SubmitApplicationResult(Failure(
          new Exception(s"Application name ${app.name} already existed")))
      } else {
        context.actorOf(launcher.props(nextAppId, APPMASTER_DEFAULT_EXECUTOR_ID, app, jar, username,
          context.parent, Some(client)), s"launcher${nextAppId}_${Util.randInt()}")
        appMasterRestartPolicies += nextAppId -> new RestartPolicy(appTotalRetries)

        val appRuntimeInfo = ApplicationRuntimeInfo(nextAppId, app.name,
          user = username,
          submissionTime = System.currentTimeMillis(),
          config = app.clusterConfig,
          status = ApplicationStatus.PENDING)
        applicationRegistry += nextAppId -> appRuntimeInfo
        val appMetaData = ApplicationMetaData(nextAppId, 0, app, jar, username)
        kvService ! PutKV(nextAppId.toString, APP_METADATA, appMetaData)

        nextAppId += 1
        kvService ! PutKV(MASTER_GROUP, MASTER_STATE, MasterState(nextAppId, applicationRegistry))
      }

    case RestartApplication(appId) =>
      val client = sender()
      (kvService ? GetKV(appId.toString, APP_METADATA)).asInstanceOf[Future[GetKVResult]].map {
        case GetKVSuccess(_, result) =>
          val metaData = result.asInstanceOf[ApplicationMetaData]
          if (metaData != null) {
            LOG.info(s"Shutting down the application (restart), $appId")
            self ! ShutdownApplication(appId)
            self.tell(SubmitApplication(metaData.appDesc, metaData.jar, metaData.username), client)
          } else {
            client ! SubmitApplicationResult(Failure(
              new Exception(s"Failed to restart, because the application $appId does not exist.")
            ))
          }
        case GetKVFailed(ex) =>
          client ! SubmitApplicationResult(Failure(
            new Exception(s"Unable to obtain the Master State. " +
              s"Application $appId will not be restarted.")
          ))
      }

    case ShutdownApplication(appId) =>
      LOG.info(s"App Manager Shutting down application $appId")
      val appInfo = applicationRegistry.get(appId).
        filter(!_.status.isInstanceOf[ApplicationTerminalStatus])
      appInfo match {
        case Some(info) =>
          killAppMaster(appId, info.worker)
          sender ! ShutdownApplicationResult(Success(appId))
          // Here we use the function to make sure the status is consistent because
          // sending another message to self will involve timing problem
          this.onApplicationStatusChanged(appId, ApplicationStatus.TERMINATED,
            System.currentTimeMillis(), null)
        case None =>
          val errorMsg = s"Failed to find registration information for appId: $appId"
          LOG.error(errorMsg)
          sender ! ShutdownApplicationResult(Failure(new Exception(errorMsg)))
      }

    case ResolveAppId(appId) =>
      val appMaster = applicationRegistry.get(appId).map(_.appMaster)
      appMaster match {
        case Some(appMasterActor) =>
          sender ! ResolveAppIdResult(Success(appMasterActor))
        case None =>
          sender ! ResolveAppIdResult(Failure(new Exception(s"Can not find Application: $appId")))
      }

    case AppMastersDataRequest =>
      var appMastersData = collection.mutable.ListBuffer[AppMasterData]()
      applicationRegistry.foreach(pair => {
        val (id, info: ApplicationRuntimeInfo) = pair
        val appMasterPath = ActorUtil.getFullPath(context.system, info.appMaster)
        val workerPath = Option(info.worker).map(worker =>
          ActorUtil.getFullPath(context.system, worker))
        appMastersData += AppMasterData(
          info.status, id, info.appName, appMasterPath, workerPath.orNull,
          info.submissionTime, info.startTime, info.finishTime, info.user)
      })
      sender ! AppMastersData(appMastersData.toList)

    case QueryAppMasterConfig(appId) =>
      val config = applicationRegistry.get(appId).map(_.config).getOrElse(ConfigFactory.empty())
      sender ! AppMasterConfig(config)

    case appMasterDataRequest: AppMasterDataRequest =>
      val appId = appMasterDataRequest.appId
      val appRuntimeInfo = applicationRegistry.get(appId)
      appRuntimeInfo match {
        case Some(info) =>
          val appMasterPath = ActorUtil.getFullPath(context.system, info.appMaster.path)
          val workerPath = Option(info.worker).map(
            worker => ActorUtil.getFullPath(context.system, worker.path)).orNull
          sender ! AppMasterData(
            info.status, appId, info.appName, appMasterPath, workerPath,
            info.submissionTime, info.startTime, info.finishTime, info.user)
        case None =>
          sender ! AppMasterData(ApplicationStatus.NONEXIST)
      }

    case RegisterAppResultListener(appId) =>
      val listenerList = appResultListeners.getOrElse(appId, List.empty[ActorRef])
      appResultListeners += appId -> (listenerList :+ sender())
  }

  def workerMessage: Receive = {
    case ShutdownExecutorSucceed(appId, executorId) =>
      LOG.info(s"Shut down executor $executorId for application $appId successfully")
    case failed: ShutdownExecutorFailed =>
      LOG.error(failed.reason)
  }

  def appMasterMessage: Receive = {
    case RegisterAppMaster(appId, appMaster, workerInfo) =>
      val appInfo = applicationRegistry.get(appId)
      appInfo match {
        case Some(info) =>
          LOG.info(s"Register AppMaster for app: $appId")
          val updatedInfo = info.onAppMasterRegistered(appMaster, workerInfo.ref)
          context.watch(appMaster)
          applicationRegistry += appId -> updatedInfo
          kvService ! PutKV(MASTER_GROUP, MASTER_STATE, MasterState(nextAppId, applicationRegistry))
          sender ! AppMasterRegistered(appId)
        case None =>
          LOG.error(s"Can not find submitted application $appId")
      }

    case ApplicationStatusChanged(appId, newStatus, timeStamp, error) =>
      onApplicationStatusChanged(appId, newStatus, timeStamp, error)
  }

  private def onApplicationStatusChanged(appId: Int, newStatus: ApplicationStatus,
      timeStamp: MilliSeconds, error: Throwable): Unit = {
    applicationRegistry.get(appId) match {
      case Some(appRuntimeInfo) =>
        if (appRuntimeInfo.status.canTransitTo(newStatus)) {
          var updatedStatus: ApplicationRuntimeInfo = null
          LOG.info(s"Application $appId change to ${newStatus.toString} at $timeStamp")
          newStatus match {
            case ApplicationStatus.ACTIVE =>
              updatedStatus = appRuntimeInfo.onAppMasterActivated(timeStamp)
              sender ! AppMasterActivated(appId)
            case succeeded@ApplicationStatus.SUCCEEDED =>
              killAppMaster(appId, appRuntimeInfo.worker)
              updatedStatus = appRuntimeInfo.onFinalStatus(timeStamp, succeeded)
              sendAppResultToListeners(appId, ApplicationSucceeded(appId))
            case failed@ApplicationStatus.FAILED =>
              killAppMaster(appId, appRuntimeInfo.worker)
              updatedStatus = appRuntimeInfo.onFinalStatus(timeStamp, failed)
              sendAppResultToListeners(appId, ApplicationFailed(appId, error))
            case terminated@ApplicationStatus.TERMINATED =>
              updatedStatus = appRuntimeInfo.onFinalStatus(timeStamp, terminated)
              sendAppResultToListeners(appId, ApplicationTerminated(appId))
            case status =>
              LOG.error(s"App $appId should not change it's status to $status")
          }

          if (newStatus.isInstanceOf[ApplicationTerminalStatus]) {
            kvService ! DeleteKVGroup(appId.toString)
          }
          applicationRegistry += appId -> updatedStatus
          kvService ! PutKV(MASTER_GROUP, MASTER_STATE, MasterState(nextAppId, applicationRegistry))
        } else {
          LOG.error(s"Application $appId tries to switch status ${appRuntimeInfo.status} " +
            s"to $newStatus")
        }
      case None =>
        LOG.error(s"Can not find application runtime info for appId $appId when it's " +
          s"status changed to ${newStatus.toString}")
    }
  }

  private def sendAppResultToListeners(appId: Int, result: ApplicationResult): Unit = {
    appResultListeners.get(appId).foreach {
      _.foreach { client =>
        client ! result
      }
    }
  }

  def appDataStoreService: Receive = {
    case SaveAppData(appId, key, value) =>
      val client = sender()
      (kvService ? PutKV(appId.toString, key, value)).asInstanceOf[Future[PutKVResult]].map {
        case PutKVSuccess =>
          client ! AppDataSaved
        case PutKVFailed(k, ex) =>
          client ! SaveAppDataFailed
      }
    case GetAppData(appId, key) =>
      val client = sender()
      (kvService ? GetKV(appId.toString, key)).asInstanceOf[Future[GetKVResult]].map {
        case GetKVSuccess(_, value) =>
          client ! GetAppDataResult(key, value)
        case GetKVFailed(ex) =>
          client ! GetAppDataResult(key, null)
      }
  }

  def terminationWatch: Receive = {
    case terminate: Terminated =>
      LOG.info(s"AppMaster(${terminate.actor.path}) is terminated, " +
        s"network down: ${terminate.getAddressTerminated}")

      // Now we assume that the only normal way to stop the application is submitting a
      // ShutdownApplication request
      val application = applicationRegistry.find { appInfo =>
        val (_, runtimeInfo) = appInfo
        terminate.actor.equals(runtimeInfo.appMaster) &&
          !runtimeInfo.status.isInstanceOf[ApplicationTerminalStatus]
      }
      if (application.nonEmpty) {
        val appId = application.get._1
        (kvService ? GetKV(appId.toString, APP_METADATA)).asInstanceOf[Future[GetKVResult]].map {
          case GetKVSuccess(_, result) =>
            val appMetadata = result.asInstanceOf[ApplicationMetaData]
            if (appMetadata != null) {
              LOG.info(s"Recovering application, $appId")
              val updatedInfo = application.get._2.copy(status = ApplicationStatus.PENDING)
              applicationRegistry += appId -> updatedInfo
              self ! RecoverApplication(appMetadata)
            } else {
              LOG.error(s"Cannot find application meta data for $appId")
            }
          case GetKVFailed(ex) =>
            LOG.error(s"Cannot find master state to recover")
        }
      }
  }

  def selfMsgHandler: Receive = {
    case RecoverApplication(state) =>
      val appId = state.appId
      if (appMasterRestartPolicies.get(appId).get.allowRestart) {
        LOG.info(s"AppManager Recovering Application $appId...")
        kvService ! PutKV(MASTER_GROUP, MASTER_STATE,
          MasterState(this.nextAppId, applicationRegistry))
        context.actorOf(launcher.props(appId, APPMASTER_DEFAULT_EXECUTOR_ID, state.appDesc,
          state.jar, state.username, context.parent, None), s"launcher${appId}_${Util.randInt()}")
      } else {
        LOG.error(s"Application $appId failed too many times")
      }
  }

  private def killAppMaster(appId: Int, worker: ActorRef): Unit = {
    val workerPath = Option(worker).map(_.path).orNull
    LOG.info(s"Shutdown AppMaster at $workerPath, appId: $appId, executorId:" +
      s" $APPMASTER_DEFAULT_EXECUTOR_ID")
    val shutdown = ShutdownExecutor(appId, APPMASTER_DEFAULT_EXECUTOR_ID,
      s"AppMaster $appId shutdown requested by master...")
    sendMsgWithTimeOutCallBack(worker, shutdown, 30000, shutDownExecutorTimeOut())
  }

  private def applicationNameExist(appName: String): Boolean = {
    applicationRegistry.values.exists { info =>
      info.appName == appName && !info.status.isInstanceOf[ApplicationTerminalStatus]
    }
  }

  private def shutDownExecutorTimeOut(): Unit = {
    LOG.error(s"Shut down executor time out")
  }
}

object AppManager {
  final val APP_METADATA = "app_metadata"
  // The id is used in KVStore
  final val MASTER_STATE = "master_state"

  case class RecoverApplication(appMetaData: ApplicationMetaData)

  case class MasterState(maxId: Int, applicationRegistry: Map[Int, ApplicationRuntimeInfo])
}