package com.daumkakao.s2graph.core

import HBaseElement._
import play.api.libs.json._
import org.apache.hadoop.hbase.client.HBaseAdmin
import org.apache.hadoop.hbase.HTableDescriptor
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.HColumnDescriptor
import org.apache.hadoop.hbase.client.Durability
import org.apache.hadoop.hbase.io.compress.Compression
import org.apache.hadoop.hbase.regionserver.BloomType
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding

/**
 * orchestrate kafka queue, db meta.
 */
object Management extends JSONParser {

  val hardLimit = 10000
  val defaultLimit = 100

  //  val labels = new ConcurrentHashMap[String, Label]
  /**
   * source, target, label, orderBy, extra properties
   * e	src	tgt	label	key:value,key:value...	key:value,key:value...
   * v src, key:value,key:value
   *
   * talk_friend 10 -> story:friend 10 ->
   */

  /**
   * label
   */
  /**
   * create {labelName, labelName_delete, labelName_update} labels
   */
  def createLabel(label: String,
    srcServiceName: String,
    srcColumnName: String,
    srcColumnType: String,
    tgtServiceName: String,
    tgtColumnName: String,
    tgtColumType: String,
    isDirected: Boolean = true,
    serviceName: String,
    indexProps: Seq[(String, JsValue)],
    props: Seq[(String, JsValue)],
    consistencyLevel: String,
    hTableName: Option[String],
    hTableTTL: Option[Int]): Label = {

    val idxProps = for ((k, v) <- indexProps; (innerVal, dataType) = toInnerVal(v)) yield (k, innerVal, dataType, true)
    val metaProps = for ((k, v) <- props; (innerVal, dataType) = toInnerVal(v)) yield (k, innerVal, dataType, false)

    val indexPropsWithType =
      for ((k, v) <- indexProps) yield {
        val (innerVal, dataType) = toInnerVal(v)
        (k, innerVal, dataType)
      }
    var cacheKey = s"serviceName=$srcServiceName"
    Service.expireCache(cacheKey)
    val srcService = tryOption(srcServiceName, Service.findByName)
    cacheKey = s"serviceName=$tgtServiceName"
    Service.expireCache(cacheKey)
    val tgtService = tryOption(tgtServiceName, Service.findByName)
    cacheKey = s"serviceName=$serviceName"
    Service.expireCache(cacheKey)
    val service = tryOption(serviceName, Service.findByName)

    cacheKey = s"label=$label"
    Label.expireCache(cacheKey)
    val labelOpt = Label.findByName(label, useCache = false)

    labelOpt match {
      case Some(l) =>
        throw new KGraphExceptions.LabelAlreadyExistException(s"Label name ${l.label} already exist.")
      case None =>
        Label.insertAll(label,
          srcService.id.get, srcColumnName, srcColumnType,
          tgtService.id.get, tgtColumnName, tgtColumType,
          isDirected, serviceName, service.id.get, idxProps ++ metaProps, consistencyLevel, hTableName, hTableTTL)
        Label.expireCache(cacheKey)
        Label.findByName(label).get
    }
  }

  def addIndex(labelStr: String, orderByKeys: Seq[(String, JsValue)]) = {
    val label = try {
      Label.findByName(labelStr).get
    } catch {
      case e: Throwable =>
        throw new KGraphExceptions.LabelNotExistException(labelStr)
    }
    val labelOrderTypes =
      for ((k, v) <- orderByKeys; (innerVal, dataType) = toInnerVal(v)) yield {

        val lblMeta = LabelMeta.findOrInsert(label.id.get, k, innerVal.toString, dataType, true)
        if (lblMeta.usedInIndex) lblMeta.seq
        else throw new KGraphExceptions.LabelMetaExistException(s"")
      }
    LabelIndex.findOrInsert(label.id.get, labelOrderTypes.toList, "")
  }

  def getServiceLable(label: String): Option[Label] = {
    Label.findByName(label)
  }

  /**
   *
   */

  def toLabelWithDirectionAndOp(label: Label, direction: String): Option[LabelWithDirection] = {
    for {
      labelId <- label.id
      dir = GraphUtil.toDirection(direction)
    } yield LabelWithDirection(labelId, dir)
  }

  def tryOption[A, R](key: A, f: A => Option[R]) = {
    f(key) match {
      case None => throw new KGraphExceptions.InternalException(s"$key is not found in DB. create $key first.")
      case Some(r) => r
    }
  }

  def toEdge(ts: Long, operation: String, srcId: String, tgtId: String,
    labelStr: String, direction: String = "", props: String): Edge = {

    val label = tryOption(labelStr, getServiceLable)

    val src = toInnerVal(srcId, label.srcColumnType)
    val tgt = toInnerVal(tgtId, label.tgtColumnType)

    val srcVertex = Vertex(CompositeId(label.srcColumn.id.get, src, true, true), ts)
    val tgtVertex = Vertex(CompositeId(label.tgtColumn.id.get, tgt, true, true), ts)
    val dir = if (direction == "") GraphUtil.toDirection(label.direction) else GraphUtil.toDirection(direction)
    val labelWithDir = LabelWithDirection(label.id.get, dir)
    val op = tryOption(operation, GraphUtil.toOp)

    val jsObject = Json.parse(props).asOpt[JsObject].getOrElse(Json.obj())
    val parsedProps = toProps(label, jsObject).toMap
    val propsWithTs = parsedProps.map(kv => (kv._1 -> InnerValWithTs(kv._2, ts))) ++ Map(LabelMeta.timeStampSeq -> InnerValWithTs(InnerVal.withLong(ts), ts))
    Edge(srcVertex, tgtVertex, labelWithDir, op, ts, version = ts, propsWithTs)

  }

  def toVertex(ts: Long, operation: String, id: String, serviceName: String, columnName: String, props: String): Vertex = {
    Service.findByName(serviceName) match {
      case None => throw new RuntimeException(s"$serviceName does not exist. create service first.")
      case Some(service) =>
        ServiceColumn.find(service.id.get, columnName) match {
          case None => throw new RuntimeException(s"$columnName is not exist. create service column first.")
          case Some(col) =>
            val idVal = toInnerVal(id, col.columnType)
            val op = tryOption(operation, GraphUtil.toOp)
            val jsObject = Json.parse(props).asOpt[JsObject].getOrElse(Json.obj())
            val parsedProps = toProps(col, jsObject).toMap
            Vertex(CompositeId(col.id.get, idVal, isEdge = false, useHash = true), ts, parsedProps, op = op)
        }
    }
  }

  def toProps(column: ServiceColumn, js: JsObject): Seq[(Byte, InnerVal)] = {

    val props = for {
      (k, v) <- js.fields
    } yield {
      val colMeta = ColumnMeta.findOrInsert(column.id.get, k)
      val (innerVal, dataType) = toInnerVal(v)
      (colMeta.seq, innerVal)
    }
    //    Logger.debug(s"vertex.ToProps: $column, $js => $props")
    props

  }

  def toProps(label: Label, js: JsObject): Seq[(Byte, InnerVal)] = {

    val props = for {
      (k, v) <- js.fields
      meta <- label.metaPropsInvMap.get(k)
      //        meta <- LabelMeta.findByName(label.id.get, k)
      //      meta = tryOption((label.id.get, k), LabelMeta.findByName)
      innerVal <- jsValueToInnerVal(v, meta.dataType)
    } yield {
      (meta.seq, innerVal)
    }
    //    Logger.error(s"toProps: $js => $props")
    props

  }

  val idTableName = "id"
  val cf = "a"
  val idColName = "id"
  val regionCnt = 10

  def getAdmin(zkAddr: String) = {
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", zkAddr)

    val adm = new HBaseAdmin(conf)
    adm
  }
  def enableTable(zkAddr: String, tableName: String) = {
    getAdmin(zkAddr).enableTable(tableName)
  }
  def disableTable(zkAddr: String, tableName: String) = {
    getAdmin(zkAddr).disableTable(tableName)
  }
  def dropTable(zkAddr: String, tableName: String) = {
    getAdmin(zkAddr).disableTable(tableName)
    getAdmin(zkAddr).deleteTable(tableName)
  }
//  def deleteEdgesByLabelIds(zkAddr: String,
//    tableName: String,
//    labelIds: String = "",
//    minTs: Long = 0L,
//    maxTs: Long = Long.MaxValue,
//    include: Boolean = true) = {
//    val conf = HBaseConfiguration.create()
//    val longTimeout = "1200000"
//    conf.set("hbase.rpc.timeout", longTimeout)
//    conf.set("hbase.client.operation.timeout", longTimeout)
//    conf.set("hbase.client.scanner.timeout.period", longTimeout)
//    conf.set("hbase.zookeeper.quorum", zkAddr)
//    val conn = HConnectionManager.createConnection(conf)
//    val table = conn.getTable(tableName.getBytes)
//    var builder = DeleteLabelsArgument.newBuilder()
//    val scanner = Scan.newBuilder()
//
//    scanner.setTimeRange(TimeRange.newBuilder().setFrom(minTs).setTo(maxTs))
//    /**
//     *  when we clean up all data does not match current database ids
//     *  we will delete row completely
//     */
//    if (!include) scanner.setFilter(ProtobufUtil.toFilter(new FirstKeyOnlyFilter))
//
//    builder.setScan(scanner)
//    for (id <- labelIds.split(",")) {
//      builder.addId(id.toInt)
//    }
//
//    val argument = builder.build()
//
//    val regionStats = table.coprocessorService(classOf[GraphStatService], null, null,
//      new Batch.Call[GraphStatService, Long]() {
//        override def call(counter: GraphStatService): Long = {
//          val controller: ServerRpcController = new ServerRpcController()
//          val rpcCallback: BlockingRpcCallback[CountResponse] = new BlockingRpcCallback[CountResponse]()
//
//          if (include) {
//            counter.cleanUpDeleteLabelsRows(controller, argument, rpcCallback)
//          } else {
//            counter.cleanUpDeleteLabelsRowsExclude(controller, argument, rpcCallback)
//          }
//
//          val response: CountResponse = rpcCallback.get()
//          if (controller.failedOnException()) throw controller.getFailedOn()
//          if (response != null && response.hasCount()) {
//            response.getCount()
//          } else {
//            0L
//          }
//        }
//      })
//
//    //    regionStats.map(kv => Bytes.toString(kv._1) -> kv._2) ++ Map("total" -> regionStats.values().sum)
//  }
  def createTable(zkAddr: String, tableName: String, cfs: List[String], regionCnt: Int, ttl: Option[Int]) = {
    try {
      val admin = getAdmin(zkAddr)
      println(admin)
      if (!admin.tableExists(tableName)) {
        println("createTable")
        val desc = new HTableDescriptor(TableName.valueOf(tableName))
        desc.setDurability(Durability.ASYNC_WAL)
        for (cf <- cfs) {
          val columnDesc = new HColumnDescriptor(cf)
            .setCompressionType(Compression.Algorithm.GZ)
            .setBloomFilterType(BloomType.ROW)
            .setDataBlockEncoding(DataBlockEncoding.FAST_DIFF)
            .setMaxVersions(1)
            .setTimeToLive(2147483647)
            .setMinVersions(0)
            .setBlocksize(32768)
            .setBlockCacheEnabled(true)
          if (ttl.isDefined) columnDesc.setTimeToLive(ttl.get)
          desc.addFamily(columnDesc)
        }

        if (regionCnt <= 1) admin.createTable(desc)
        else admin.createTable(desc, getStartKey(regionCnt), getEndKey(regionCnt), regionCnt)
      } else {
        // already exist
      }
    } catch {
      case e: Throwable => println(e)
    }
  }
  // we only use murmur hash to distribute row key.
  private def getStartKey(regionCount: Int) = {
    Bytes.toBytes((Int.MaxValue / regionCount))
  }

  private def getEndKey(regionCount: Int) = {
    Bytes.toBytes((Int.MaxValue / regionCount * (regionCount - 1)))
  }


}