package com.daumkakao.s2graph.core
import HBaseElement._
import org.apache.hadoop.hbase.Cell
import org.apache.hadoop.hbase.client.{ Delete, Mutation, Put, Result }
import org.hbase.async.{HBaseRpc, DeleteRequest, PutRequest}
import org.slf4j.LoggerFactory
import scala.collection.JavaConversions._
import scala.collection.mutable.ListBuffer
import play.api.libs.json.Json

case class EdgeWithIndexInverted(srcVertex: Vertex, tgtVertex: Vertex, labelWithDir: LabelWithDirection, op: Byte, version: Long, props: Map[Byte, InnerValWithTs]) {
  import GraphConstant._
  import Edge._

  //  Logger.error(s"EdgeWithIndexInverted${this.toString}")
  lazy val lastModifiedAt = props.map(_._2.ts).max

  lazy val rowKey = EdgeRowKey(srcVertex.id, labelWithDir, LabelIndex.defaultSeq, isInverted = true)

  lazy val qualifier = EdgeQualifierInverted(tgtVertex.id)

  // only for toString.
  lazy val label = Label.findById(labelWithDir.labelId)
  lazy val propsWithoutTs = props.map(kv => (kv._1 -> kv._2.innerVal))

  lazy val value = EdgeValueInverted(op, props.toList)

  def buildPut() = {
    val put = new Put(rowKey.bytes)
    put.add(edgeCf, qualifier.bytes, version, value.bytes)
  }
  def buildPutAsync() = {
    new PutRequest(label.hbaseTableName.getBytes, rowKey.bytes, edgeCf, qualifier.bytes, value.bytes, version)
  }

  def isSame(other: Any): Boolean = {
    val ret = this.toString == other.toString
    logger.debug(s"EdgeWithIndexInverted\n$this\n$other\n$ret")
    ret
  }

  override def toString(): String = {
    val ls = ListBuffer(lastModifiedAt, GraphUtil.fromOp(op), "e",
      srcVertex.innerId, tgtVertex.innerId, label.label)
    //    if (!propsWithName.isEmpty) ls += Json.toJson(propsWithName)
    if (!props.isEmpty) ls += props
    ls.mkString("\t")
  }
}
case class EdgeWithIndex(srcVertex: Vertex, tgtVertex: Vertex, labelWithDir: LabelWithDirection, op: Byte, version: Long, labelIndexSeq: Byte, props: Map[Byte, InnerVal]) {

  assert(props.get(LabelMeta.timeStampSeq).isDefined)

  lazy val ts = props(LabelMeta.timeStampSeq).longV.get

  import GraphConstant._
  import Edge._

  lazy val rowKey = EdgeRowKey(srcVertex.id, labelWithDir, labelIndexSeq, isInverted = false)
  lazy val labelIndex = LabelIndex.findByLabelIdAndSeq(labelWithDir.labelId, labelIndexSeq).get
  lazy val defaultIndexMetas = labelIndex.sortKeyTypes.map(meta => meta.seq -> meta.defaultInnerVal).toMap
  lazy val labelIndexMetaSeqs = labelIndex.metaSeqs

  lazy val orders = for (k <- labelIndexMetaSeqs) yield {
    props.get(k) match {
      case None =>
        /**
         *  TODO: agly hack
         * 	now we double store target vertex.innerId/srcVertex.innerId for easy development. later fix this to only store id once
         */
        val v = k match {
          case LabelMeta.timeStampSeq => InnerVal.withLong(ts)
          case LabelMeta.toSeq => tgtVertex.innerId
          case LabelMeta.fromSeq => //srcVertex.innerId
            // for now, it does not make sense to build index on srcVertex.innerId since all edges have same data.
            throw new RuntimeException("_from on indexProps is not supported")
          case _ => defaultIndexMetas(k)
        }
        //        val v = if (k == LabelMeta.timeStampSeq) InnerVal.withLong(ts) else defaultIndexMetas(k)
        (k -> v)
      case Some(v) => (k -> v)
    }
  }
  lazy val metas = for ((k, v) <- props if !defaultIndexMetas.containsKey(k)) yield (k -> v)

  lazy val qualifier = EdgeQualifier(orders, tgtVertex.id, op)
  lazy val value = EdgeValue(metas.toList)

  lazy val hasAllPropsForIndex = orders.length == labelIndexMetaSeqs.length

  // only for toString.
  lazy val label = Label.findById(labelWithDir.labelId)

  def buildPuts(): List[Put] = {
    if (!hasAllPropsForIndex) {
      logger.error(s"$this dont have all props for index")
      List.empty[Put]
    } else {
      val put = new Put(rowKey.bytes)
      //    Logger.debug(s"$this")
      //      Logger.debug(s"EdgeWithIndex.buildPut: $rowKey, $qualifier, $value")
      put.add(edgeCf, qualifier.bytes, version, value.bytes)
      List(put)
    }
  }
  def buildPutsAsync(): List[PutRequest] = {
    if (!hasAllPropsForIndex) {
      logger.error(s"$this dont have all props for index")
      List.empty[PutRequest]
    } else {
      List(new PutRequest(label.hbaseTableName.getBytes, rowKey.bytes, edgeCf, qualifier.bytes, value.bytes, version))
    }
  }
  def buildDeletes(): List[Delete] = {
    if (!hasAllPropsForIndex) List.empty[Delete]
    else {
      val delete = new Delete(rowKey.bytes)
      delete.deleteColumns(edgeCf, qualifier.bytes, version)
      List(delete)
    }
  }
  def buildDeletesAsync(): List[DeleteRequest] = {
    if (!hasAllPropsForIndex) List.empty[DeleteRequest]
    else {
      List(new DeleteRequest(label.hbaseTableName.getBytes, rowKey.bytes, edgeCf, qualifier.bytes, version))
    }
  }
  override def toString(): String = {
    val ls = ListBuffer(ts, GraphUtil.fromOp(op), "e",
      srcVertex.innerId, tgtVertex.innerId, label.label)
    //    if (!propsWithName.isEmpty) ls += Json.toJson(propsWithName ++ Map("version" -> version.toString))
    if (!props.isEmpty) ls += props
    ls.mkString("\t")
  }
  def isSame(other: Any): Boolean = {
    val ret = this.toString == other.toString
    logger.debug(s"EdgeWithIndex\n$this\n$other\n$ret")
    ret
  }
}
/**
 *
 */
case class Edge(srcVertex: Vertex, tgtVertex: Vertex, labelWithDir: LabelWithDirection, op: Byte,
  ts: Long, version: Long, propsWithTs: Map[Byte, InnerValWithTs]) extends GraphElement with JSONParser {

  val logger = Edge.logger
  implicit val ex = Graph.executionContext

  lazy val props =
    if (op == GraphUtil.operations("delete")) Map(LabelMeta.timeStampSeq -> InnerVal.withLong(ts))
    else for ((k, v) <- propsWithTs) yield (k -> v.innerVal)

  lazy val relatedEdges = {
    labelWithDir.dir match {
      case 2 => //undirected
        val out = LabelWithDirection(labelWithDir.labelId, 0)
        val base = Edge(srcVertex, tgtVertex, out, op, ts, version, propsWithTs)
        List(base, base.duplicateEdge, base.reverseDirEdge, base.reverseSrcTgtEdge)
      case 0 | 1 =>
        List(this, duplicateEdge)
    }
  }

  lazy val srcForVertex = Vertex(srcVertex.id.updateIsEdge(false), srcVertex.ts, srcVertex.props)
  lazy val tgtForVertex = Vertex(tgtVertex.id.updateIsEdge(false), tgtVertex.ts, tgtVertex.props)

  lazy val duplicateEdge = Edge(tgtVertex, srcVertex, labelWithDir.dirToggled, op, ts, version, propsWithTs)
  lazy val reverseDirEdge = Edge(srcVertex, tgtVertex, labelWithDir.dirToggled, op, ts, version, propsWithTs)
  lazy val reverseSrcTgtEdge = Edge(tgtVertex, srcVertex, labelWithDir, op, ts, version, propsWithTs)

  lazy val label = Label.findById(labelWithDir.labelId)
  lazy val labelOrders = LabelIndex.findByLabelIdAll(labelWithDir.labelId)
  override lazy val serviceName = label.serviceName
  override lazy val queueKey = Seq(ts.toString, tgtVertex.serviceName).mkString("|")
  override lazy val queuePartitionKey = Seq(srcVertex.innerId, tgtVertex.innerId).mkString("|")

  lazy val propsPlusTs = propsWithTs.get(LabelMeta.timeStampSeq) match {
    case Some(_) => props
    case None => props ++ Map(LabelMeta.timeStampSeq -> InnerVal.withLong(ts))
  }

  lazy val propsPlusTsValid = propsPlusTs.filter(kv => kv._1 >= 0)

  lazy val edgesWithIndex = {
    for (labelOrder <- labelOrders) yield {
      EdgeWithIndex(srcVertex, tgtVertex, labelWithDir, op, version, labelOrder.seq, propsPlusTs)
    }
  }
  lazy val edgesWithIndexValid = {
    for (labelOrder <- labelOrders) yield {
      EdgeWithIndex(srcVertex, tgtVertex, labelWithDir, op, version, labelOrder.seq, propsPlusTsValid)
    }
  }
  def edgesWithIndex(newOp: Byte) = {
    for (labelOrder <- labelOrders) yield {
      EdgeWithIndex(srcVertex, tgtVertex, labelWithDir, newOp, version, labelOrder.seq, propsPlusTs)
    }
  }
  def edgesWithIndexValid(newOp: Byte) = {
    for (labelOrder <- labelOrders) yield {
      EdgeWithIndex(srcVertex, tgtVertex, labelWithDir, newOp, version, labelOrder.seq, propsPlusTsValid)
    }
  }

  lazy val edgesWithInvertedIndex = {
    EdgeWithIndexInverted(srcVertex, tgtVertex, labelWithDir, op, version, propsWithTs ++
      Map(LabelMeta.timeStampSeq -> InnerValWithTs(InnerVal.withLong(ts), ts)))
  }
  def edgesWithInvertedIndex(newOp: Byte) = {
    EdgeWithIndexInverted(srcVertex, tgtVertex, labelWithDir, newOp, version, propsWithTs ++
      Map(LabelMeta.timeStampSeq -> InnerValWithTs(InnerVal.withLong(ts), ts)))
  }
  lazy val propsWithName = for {
    (seq, v) <- props
    name <- label.metaPropNamesMap.get(seq) if seq > 0
  } yield (name -> innerValToJsValue(v))

  lazy val canBeBatched =
    op == GraphUtil.operations("insertBulk") ||
      (op == GraphUtil.operations("insert") && label.consistencyLevel != "strong")

  def updateTgtVertex(id: InnerVal) = {
    val newCompositeId = CompositeId(tgtVertex.id.colId, id, isEdge = tgtVertex.id.isEdge, useHash = tgtVertex.id.useHash)
    val newTgtVertex = Vertex(newCompositeId, tgtVertex.ts, tgtVertex.props)
    Edge(srcVertex, newTgtVertex, labelWithDir, op, ts, version, propsWithTs)
  }

  /**
   * methods for build mutations.
   */
  def buildVertexPuts(): List[Put] = srcForVertex.buildPuts ++ tgtForVertex.buildPuts
//
  def buildVertexPutsAsync(): List[PutRequest] = srcForVertex.buildPutsAsync() ++ tgtForVertex.buildPutsAsync()

  def buildPutsAll(): List[HBaseRpc] = {
    val edgePuts =
      relatedEdges.flatMap { edge =>
        if (edge.op == GraphUtil.operations("insert")) {
          if (label.consistencyLevel == "strong") {
            edge.upsert()
            List.empty[PutRequest]
          } else {
            edge.insert()
          }
        } else if (edge.op == GraphUtil.operations("delete")) {
          edge.delete(label.consistencyLevel == "strong")
          List.empty[PutRequest]
        } else if (edge.op == GraphUtil.operations("update")) {
          edge.update()
          List.empty[PutRequest]
        } else if (edge.op == GraphUtil.operations("increment")) {
          edge.increment()
          List.empty[PutRequest]
        } else if (edge.op == GraphUtil.operations("insertBulk")) {
          edge.insert()
        } else {
          throw new Exception(s"operation[${edge.op}] is not supported on edge.")
        }

      }
    edgePuts
  }
  def insertBulk() = {
    edgesWithInvertedIndex.buildPut :: edgesWithIndex.flatMap(e => e.buildPuts())
  }
  def insert() = {
    edgesWithInvertedIndex.buildPutAsync :: edgesWithIndex.flatMap(e => e.buildPutsAsync)
  }
  /**
   * write must be synchronous to keep consistencyLevel
   */
//  def fetchInverted() = {
//    Graph.getEdgeSync(srcVertex, tgtVertex, label, labelWithDir.dir)
//  }
  def fetchInvertedAsync() = {
    Graph.getEdge(srcVertex, tgtVertex, label, labelWithDir.dir)
  }
  /**
   *
   */

//  def mutate(f: (Option[Edge], Edge) => EdgeUpdate): Unit = {
//    val edgeUpdate = f(fetchInverted().headOption, this)
//    withWriteTable(label.hbaseZkAddr, label.hbaseTableName) { table =>
//      val size = edgeUpdate.mutations.size
//      val rets: Array[Object] = Array.fill(size)(null)
//      table.batch(edgeUpdate.mutations, rets)
//      for (ret <- rets if ret == null) {
//        Logger.error(s"mutation tryed but failed.")
//      }
//    } {
//      Logger.error(s"mutation failed.")
//    }
//  }

  def mutate(f: (Option[Edge], Edge) => EdgeUpdate): Unit = {
    try {
      val client = Graph.getClient(label.hbaseZkAddr)
      for {
        edges <- fetchInvertedAsync()
        invertedEdgeOpt = edges.headOption
        edgeUpdate = f(invertedEdgeOpt, this)
      } yield {
        /**
         * we can use CAS operation on inverted Edge checkAndSet() and if it return false
         * then re-read and build update and write recursively.
         */
        Graph.writeAsync(label.hbaseZkAddr, edgeUpdate.mutations)
      }
//      client.flush()
    } catch {
      case e: Throwable =>
        logger.error(s"mutate failed. $e", e)
    }
  }

  def upsert(): Unit = {
    mutate(Edge.buildUpsert)
  }

  def delete(mutateInPlace: Boolean = true) = {
    mutate(Edge.buildDelete)
  }

  def update() = {
    mutate(Edge.buildUpdate)
  }

  def increment() = {
    mutate(Edge.buildIncrement)
  }

  def toJson() = {}

  def rank(r: RankParam): Double = {

    if (r.keySeqAndWeights.size <= 0) 1.0f
    else {
      //      Logger.warn(s"orderMapping : ${orderByKeyValues} keyIdAndWeights : ${r.keyIdAndWeights}")

      var sum: Double = 0
      for {
        (seq, w) <- r.keySeqAndWeights
      } {
        seq match {
          case -1 => { //case key == "count"
            sum += 1
          }
          case _ => {
            props.get(seq) match {
              case None =>
              //                Logger.error(s"Not Found SortKeyType : ${seq} for rank in Label(${labelWithDir.labelId}})'s OrderByKeys(${orderByKey.typeIds}})")
              case Some(innerVal) => {
                sum += w * innerVal.longV.getOrElse(1L).toDouble
              }
            }
          }
        }
      }
      sum
    }
  }
  override def toString(): String = {
    val ls = ListBuffer(ts, GraphUtil.fromOp(op), "e",
      srcVertex.innerId, tgtVertex.innerId, label.label)
    if (!propsWithName.isEmpty) ls += Json.toJson(propsWithName)
    //    if (!propsPlusTs.isEmpty) ls += propsPlusTs
    ls.mkString("\t")
  }
  def toStringRaw(): String = {
    val ls = ListBuffer(ts, GraphUtil.fromOp(op), "e",
      srcVertex.innerId, tgtVertex.innerId, label.label)
    if (!propsPlusTs.isEmpty) ls += propsPlusTs
    ls.mkString("\t")
  }
}

case class EdgeUpdate(mutations: List[HBaseRpc] = List.empty[HBaseRpc],
                      edgesToDelete: List[EdgeWithIndex] = List.empty[EdgeWithIndex],
                      edgesToInsert: List[EdgeWithIndex] = List.empty[EdgeWithIndex],
                      newInvertedEdge: Option[EdgeWithIndexInverted] = None) {

  override def toString(): String = {
    val size = s"mutations: ${mutations.size}"
    val deletes = s"deletes: ${edgesToDelete.map(e => e.toString).mkString("\n")}"
    val inserts = s"inserts: ${edgesToInsert.map(e => e.toString).mkString("\n")}"
    val updates = s"snapshot: $newInvertedEdge"
    List(size, deletes, inserts, updates).mkString("\n")
  }
}

object Edge {
  val logger = LoggerFactory.getLogger(classOf[Edge])
  //  val initialVersion = 2L
  val incrementVersion = 1L
  val minOperationTs = 1L
  val minTsVal = 0L
  type PropsPairWithTs = (Map[Byte, InnerValWithTs], Map[Byte, InnerValWithTs], Long)

  def buildUpsert(invertedEdge: Option[Edge], requestEdge: Edge): EdgeUpdate = {
    assert(requestEdge.op == GraphUtil.operations("insert"))
    buildOperation(invertedEdge, requestEdge)(buildUpsert)
  }
  def buildUpdate(invertedEdge: Option[Edge], requestEdge: Edge): EdgeUpdate = {
    assert(requestEdge.op == GraphUtil.operations("update"))
    buildOperation(invertedEdge, requestEdge)(buildUpdate)
  }
  def buildDelete(invertedEdge: Option[Edge], requestEdge: Edge): EdgeUpdate = {
    assert(requestEdge.op == GraphUtil.operations("delete"))
    buildOperation(invertedEdge, requestEdge)(buildDelete)
  }
  def buildIncrement(invertedEdge: Option[Edge], requestEdge: Edge): EdgeUpdate = {
    assert(requestEdge.op == GraphUtil.operations("increment"))
    buildOperation(invertedEdge, requestEdge)(buildIncrement)
  }
  def buildOperation(invertedEdge: Option[Edge], requestEdge: Edge)(f: PropsPairWithTs => (Map[Byte, InnerValWithTs], Boolean)) = {
//        Logger.debug(s"oldEdge: ${invertedEdge.map(_.toStringRaw)}")
//        Logger.debug(s"requestEdge: ${requestEdge.toStringRaw}")

    val oldPropsWithTs = if (invertedEdge.isEmpty) Map.empty[Byte, InnerValWithTs] else invertedEdge.get.propsWithTs
    val minTsInOldProps = if (invertedEdge.isEmpty) minTsVal else invertedEdge.get.propsWithTs.map(kv => kv._2.ts).min

    val oldTs = invertedEdge.map(e => e.ts).getOrElse(minTsVal)

    if (oldTs == requestEdge.ts) {
      logger.error(s"duplicate timestamp on same edge. $requestEdge")
      EdgeUpdate()
    } else {
      val (newPropsWithTs, shouldReplace) = f(oldPropsWithTs, requestEdge.propsWithTs, requestEdge.ts)

      if (!shouldReplace) {
        logger.error(s"drop request $requestEdge becaseu shouldReplace is $shouldReplace")
        EdgeUpdate()
      } else {

        val maxTsInNewProps = newPropsWithTs.map(kv => kv._2.ts).max
        val newOp = if (maxTsInNewProps > requestEdge.ts) {
          invertedEdge match {
            case None => requestEdge.op
            case Some(old) => old.op
          }
        } else {
          requestEdge.op
        }
        val newEdgeVersion = invertedEdge.map(e => e.version + incrementVersion).getOrElse(requestEdge.ts)

        val maxTs = if (oldTs > requestEdge.ts) oldTs else requestEdge.ts
        val newEdge = Edge(requestEdge.srcVertex, requestEdge.tgtVertex, requestEdge.labelWithDir,
          newOp, maxTs, newEdgeVersion, newPropsWithTs)

        buildReplace(invertedEdge, newEdge, newPropsWithTs)
      }
    }

  }
  def buildUpsert(propsPairWithTs: PropsPairWithTs) = {
    var shouldReplace = false
    val (oldPropsWithTs, propsWithTs, requestTs) = propsPairWithTs
    val lastDeletedAt = oldPropsWithTs.get(LabelMeta.lastDeletedAt).map(v => v.ts).getOrElse(minTsVal)
    val existInOld = for ((k, oldValWithTs) <- oldPropsWithTs) yield {
      propsWithTs.get(k) match {
        case Some(newValWithTs) =>
          assert(oldValWithTs.ts >= lastDeletedAt)
          val v = if (oldValWithTs.ts >= newValWithTs.ts) oldValWithTs else {
            shouldReplace = true
            newValWithTs
          }
          Some(k -> v)
        //          if (v.ts > lastDeletedAt) Some(k -> v)
        //          else {
        //            shouldReplace = true
        //            None
        //          }

        case None =>
          assert(oldValWithTs.ts >= lastDeletedAt)
          if (oldValWithTs.ts >= requestTs || k < 0) Some(k -> oldValWithTs)
          else {
            shouldReplace = true
            None
          }
      }
    }
    val existInNew =
      for {
        (k, newValWithTs) <- propsWithTs if !oldPropsWithTs.contains(k) && newValWithTs.ts > lastDeletedAt
      } yield {
        shouldReplace = true
        Some(k -> newValWithTs)
      }

    ((existInOld.flatten ++ existInNew.flatten).toMap, shouldReplace)
  }
  def buildUpdate(propsPairWithTs: PropsPairWithTs) = {
    var shouldReplace = false
    val (oldPropsWithTs, propsWithTs, requestTs) = propsPairWithTs
    val lastDeletedAt = oldPropsWithTs.get(LabelMeta.lastDeletedAt).map(v => v.ts).getOrElse(minTsVal)
    val existInOld = for ((k, oldValWithTs) <- oldPropsWithTs) yield {
      propsWithTs.get(k) match {
        case Some(newValWithTs) =>
          assert(oldValWithTs.ts >= lastDeletedAt)
          val v = if (oldValWithTs.ts >= newValWithTs.ts) oldValWithTs else {
            shouldReplace = true
            newValWithTs
          }
          Some(k -> v)
        //          if (v.ts > lastDeletedAt) Some(k -> v)
        //          else None

        case None =>
          // important: update need to merge previous valid values.
          assert(oldValWithTs.ts >= lastDeletedAt)
          Some(k -> oldValWithTs)
        //          if (oldValWithTs.ts >= lastDeletedAt) Some(k -> oldValWithTs) else None
      }
    }
    val existInNew = for {
      (k, newValWithTs) <- propsWithTs if !oldPropsWithTs.contains(k) && newValWithTs.ts > lastDeletedAt
    } yield {
      shouldReplace = true
      Some(k -> newValWithTs)
    }

    ((existInOld.flatten ++ existInNew.flatten).toMap, shouldReplace)
  }
  def buildIncrement(propsPairWithTs: PropsPairWithTs) = {
    var shouldReplace = false
    val (oldPropsWithTs, propsWithTs, requestTs) = propsPairWithTs
    val lastDeletedAt = oldPropsWithTs.get(LabelMeta.lastDeletedAt).map(v => v.ts).getOrElse(minTsVal)
    val existInOld = for ((k, oldValWithTs) <- oldPropsWithTs) yield {
      propsWithTs.get(k) match {
        case Some(newValWithTs) =>
          if (k == LabelMeta.timeStampSeq) {
            val v = if (oldValWithTs.ts >= newValWithTs.ts) oldValWithTs else {
              shouldReplace = true
              newValWithTs
            }
            Some(k -> v)
          } else {
            //            assert(oldValWithTs.ts >= lastDeletedAt)
            //            if (oldValWithTs.ts < newValWithTs.ts) {
            //              shouldReplace = true
            //              Some(k -> InnerValWithTs(oldValWithTs.innerVal + newValWithTs.innerVal, oldValWithTs.ts))
            //            } else if (newValWithTs.ts >= lastDeletedAt){
            //              shouldReplace = true
            //              Some(k -> InnerValWithTs(oldValWithTs.innerVal + newValWithTs.innerVal, newValWithTs.ts))
            //            } else {
            //              Some(k -> oldValWithTs)
            //            }
            if (oldValWithTs.ts >= newValWithTs.ts) {
              Some(k -> oldValWithTs)
              //              if (newValWithTs.ts > lastDeletedAt) {
              //                shouldReplace = true
              //                Some(k -> InnerValWithTs(oldValWithTs.innerVal + newValWithTs.innerVal, newValWithTs.ts))
              //              } else {
              //                Some(k -> oldValWithTs)
              //              }

              //              if (oldValWithTs.ts > lastDeletedAt) Some(k -> oldValWithTs)
              //              else None
            } else {
              assert(oldValWithTs.ts < newValWithTs.ts && oldValWithTs.ts >= lastDeletedAt)
              shouldReplace = true
              // incr(t0), incr(t2), d(t1) => deleted
              Some(k -> InnerValWithTs(oldValWithTs.innerVal + newValWithTs.innerVal, oldValWithTs.ts))

              //              if (oldValWithTs.ts > lastDeletedAt) {
              //                
              //                Some(k -> InnerValWithTs(oldValWithTs.innerVal + newValWithTs.innerVal, newValWithTs.ts))
              //              } else if (oldValWithTs.ts <= lastDeletedAt && lastDeletedAt < newValWithTs.ts) {
              //                Some(k -> newValWithTs)
              //              } else None
            }
          }

        case None =>
          assert(oldValWithTs.ts >= lastDeletedAt)
          Some(k -> oldValWithTs)
        //          if (oldValWithTs.ts >= lastDeletedAt) Some(k -> oldValWithTs) else None
      }
    }
    val existInNew = for {
      (k, newValWithTs) <- propsWithTs if !oldPropsWithTs.contains(k) && newValWithTs.ts > lastDeletedAt
    } yield {
      shouldReplace = true
      Some(k -> newValWithTs)
    }

    ((existInOld.flatten ++ existInNew.flatten).toMap, shouldReplace)
  }
  def buildDelete(propsPairWithTs: PropsPairWithTs) = {
    var shouldReplace = false
    val (oldPropsWithTs, propsWithTs, requestTs) = propsPairWithTs
    val lastDeletedAt = oldPropsWithTs.get(LabelMeta.lastDeletedAt) match {
      case Some(prevDeletedAt) =>
        if (prevDeletedAt.ts >= requestTs) prevDeletedAt.ts
        else {
          shouldReplace = true
          requestTs
        }
      case None => {
        shouldReplace = true
        requestTs
      }
    }
    val existInOld = for ((k, oldValWithTs) <- oldPropsWithTs) yield {
      if (k == LabelMeta.timeStampSeq) {
        if (oldValWithTs.ts >= requestTs) Some(k -> oldValWithTs)
        else {
          shouldReplace = true
          Some(k -> InnerValWithTs.withLong(requestTs, requestTs))
        }
      } else {
        if (oldValWithTs.ts >= lastDeletedAt) Some(k -> oldValWithTs)
        else {
          shouldReplace = true
          None
        }
      }
    }
    val mustExistInNew = Map(LabelMeta.lastDeletedAt -> InnerValWithTs.withLong(lastDeletedAt, lastDeletedAt))
    ((existInOld.flatten ++ mustExistInNew).toMap, shouldReplace)
  }

  private def allPropsDeleted(props: Map[Byte, InnerValWithTs]) = {
    val maxTsProp = props.toList.sortBy(kv => (kv._2.ts, -1 * kv._1)).reverse.head
    maxTsProp._1 == LabelMeta.lastDeletedAt
  }
  /**
   * delete invertedEdge.edgesWithIndex
   * insert requestEdge.edgesWithIndex
   * update requestEdge.edgesWithIndexInverted
   */
  def buildReplace(invertedEdge: Option[Edge], requestEdge: Edge, newPropsWithTs: Map[Byte, InnerValWithTs]): EdgeUpdate = {

    val edgesToDelete = invertedEdge match {
      //      case Some(e) if e.op != GraphUtil.operations("delete") => e.edgesWithIndexValid
      case Some(e) => e.edgesWithIndexValid
      case _ =>
        // nothing to remove on indexed.
        List.empty[EdgeWithIndex]
    }

    val edgesToInsert = {
      if (newPropsWithTs.isEmpty) List.empty[EdgeWithIndex]
      else {
        //        val maxTsProp = newPropsWithTs.toList.sortBy(kv => (kv._2.ts, -1 * kv._1)).reverse.head
        //        if (maxTsProp._1 == LabelMeta.lastDeletedAt) {
        if (allPropsDeleted(newPropsWithTs)) {
          // all props is older than lastDeletedAt so nothing to insert on indexed.
          List.empty[EdgeWithIndex]
        } else {
          requestEdge.edgesWithIndexValid
        }
      }
    }
    val edgeInverted = if (newPropsWithTs.isEmpty) None else Some(requestEdge.edgesWithInvertedIndex)

    val deleteMutations = edgesToDelete.flatMap(edge => edge.buildDeletesAsync).toList
    val insertMutations = edgesToInsert.flatMap(edge => edge.buildPutsAsync).toList
    val invertMutations = edgeInverted.map(e => List(e.buildPutAsync)).getOrElse(List.empty[PutRequest])
    val mutations = deleteMutations ++ insertMutations ++ invertMutations

    val update = EdgeUpdate(mutations, edgesToDelete, edgesToInsert, edgeInverted)

//        Logger.debug(s"UpdatedProps: ${newPropsWithTs}\n")
//        Logger.debug(s"EdgeUpdate: $update\n")
    update
  }
  def fromString(s: String): Option[Edge] = Graph.toEdge(s)

  def toEdge(kv: org.hbase.async.KeyValue, param: QueryParam): Option[Edge] = {
    val version = kv.timestamp()

    val rowKey = EdgeRowKey(kv.key(), 0)
    val srcVertexId = rowKey.srcVertexId
    val (tgtVertexId, props, op, ts) = rowKey.isInverted match {
      case true =>
        val qualifier = EdgeQualifierInverted(kv.qualifier(), 0)
        val value = EdgeValueInverted(kv.value(), 0)
        val kvsMap = value.props.toMap
        val ts = kvsMap.get(LabelMeta.timeStampSeq) match {
          case None => version
          case Some(v) => v.innerVal.longV.get
        }
        (qualifier.tgtVertexId, kvsMap, value.op, ts)
      case false =>
        val kvQual = kv.qualifier()
        val qualifier = EdgeQualifier(kvQual, 0, kvQual.length)
        val value = EdgeValue(kv.value(), 0)
        val kvs = qualifier.propsKVs(rowKey.labelWithDir.labelId, rowKey.labelOrderSeq) ::: value.props.toList
        val kvsMap = kvs.toMap

        val ts = kvsMap.get(LabelMeta.timeStampSeq) match {
          case None => version
          case Some(v) => v.longV.get
        }
        val kvsMapWithoutTs = kvsMap.map(kv => (kv._1 -> InnerValWithTs(kv._2, ts)))
        (qualifier.tgtVertexId, kvsMapWithoutTs, qualifier.op, ts)
    }
    //    Logger.error(s"toEdge: $rowKey, $tgtVertexId")
    val labelMetas = LabelMeta.findAllByLabelId(rowKey.labelWithDir.labelId, useCache = true)
    val propsWithDefault = (for (meta <- labelMetas) yield {
      props.get(meta.seq) match {
        case Some(v) => (meta.seq -> v)
        case None => (meta.seq -> InnerValWithTs(meta.defaultInnerVal, minTsVal))
      }
    }).toMap
    //    play.api.Logger.debug(s"$rowKey, $tgtVertexId, $props, $op, $ts => ${param.hasFilters}")
    /**
     * TODO: backward compatibility only. deprecate has field
     */
    val edge = Edge(Vertex(srcVertexId, ts), Vertex(tgtVertexId, ts), rowKey.labelWithDir, op, ts, version, props)
    //    if (param.propsFilters.filter(edge)) Some(edge) else None
    val matches =
      for {
        (k, v) <- param.hasFilters
        edgeVal <- propsWithDefault.get(k) if edgeVal.innerVal.value == v.value
      } yield (k -> v)

    val ret = if (matches.size == param.hasFilters.size && param.where.map(_.filter(edge)).getOrElse(true)) {
      //      val edge = Edge(Vertex(srcVertexId, ts), Vertex(tgtVertexId, ts), rowKey.labelWithDir, op, ts, version, props)
      //      Logger.debug(s"fetchedEdge: $edge")
      Some(edge)
    } else {
      None
    }
    //    Logger.debug(s"${cell.getQualifier().toList}, ${ret.map(x => x.toStringRaw)}")
    ret
  }
  def toEdge(kv: org.hbase.async.KeyValue): Edge = {
    val version = kv.timestamp()

    val rowKey = EdgeRowKey(kv.key(), 0)
    val srcVertexId = rowKey.srcVertexId
    val (tgtVertexId, props, op, ts) = rowKey.isInverted match {
      case true =>
        val qualifier = EdgeQualifierInverted(kv.qualifier(), 0)
        val value = EdgeValueInverted(kv.value(), 0)
        val kvsMap = value.props.toMap
        val ts = kvsMap.get(LabelMeta.timeStampSeq) match {
          case None => version
          case Some(v) => v.innerVal.longV.get
        }
        (qualifier.tgtVertexId, kvsMap, value.op, ts)
      case false =>
        val kvQual = kv.qualifier()
        val qualifier = EdgeQualifier(kvQual, 0, kvQual.length)
        val value = EdgeValue(kv.value(), 0)
        val kvs = qualifier.propsKVs(rowKey.labelWithDir.labelId, rowKey.labelOrderSeq) ::: value.props.toList
        val kvsMap = kvs.toMap

        val ts = kvsMap.get(LabelMeta.timeStampSeq) match {
          case None => version
          case Some(v) => v.longV.get
        }
        val kvsMapWithoutTs = kvsMap.map(kv => (kv._1 -> InnerValWithTs(kv._2, ts)))
        (qualifier.tgtVertexId, kvsMapWithoutTs, qualifier.op, ts)
    }
    //    Logger.error(s"toEdge: $rowKey, $tgtVertexId")
    val labelMetas = LabelMeta.findAllByLabelId(rowKey.labelWithDir.labelId, useCache = true)
    val propsWithDefault = (for (meta <- labelMetas) yield {
      props.get(meta.seq) match {
        case Some(v) => (meta.seq -> v)
        case None => (meta.seq -> InnerValWithTs(meta.defaultInnerVal, minTsVal))
      }
    }).toMap
    //    play.api.Logger.debug(s"$rowKey, $tgtVertexId, $props, $op, $ts => ${param.hasFilters}")
    /**
     * TODO: backward compatibility only. deprecate has field
     */
    Edge(Vertex(srcVertexId, ts), Vertex(tgtVertexId, ts), rowKey.labelWithDir, op, ts, version, props)
  }

}