package com.daumkakao.s2graph.core
import GraphUtil._
import KGraphExceptions._
import org.apache.hadoop.hbase.util.Bytes
import scala.collection.mutable.ListBuffer
import org.apache.hadoop.hbase.Cell
import scala.collection.mutable.ArrayBuffer


object HBaseElement {
  val ttsForActivity = 60 * 60 * 24 * 30
  val delimiter = "|"
  val seperator = ":"
  val bytesForMurMur = 2
  val bitsForDir = 2
  val bytesForOp = 3
  val bitsForLenWithDir = 5
  val bitsForDirWithLen = 2
  val bitsForOp = 3

  val bitForPropMode = 1
  val bitForByte = 7
  /**
   * id between application and Graph Instance.
   */
  object CompositeId {
    val defaultColId = 0
    val defaultInnerId = 0
    val isDescOrder = false
    val emptyCompositeId = CompositeId(defaultColId, InnerVal.withLong(defaultInnerId), isEdge = true, useHash = true)
    def apply(bytes: Array[Byte], offset: Int, isEdge: Boolean, useHash: Boolean): CompositeId = {
      var pos = offset
      if (useHash) {
        // skip over murmur hash
        pos += 2
      }
      val innerId = InnerVal(bytes, pos)
      pos += innerId.bytes.length
      if (isEdge) {
        CompositeId(defaultColId, innerId, true, useHash)
      } else {
        val cId = Bytes.toInt(bytes, pos, 4)
        CompositeId(cId, innerId, false, useHash)
      }
    }
  }
  // TODO: colId range < (1<<15??) id length??
  case class CompositeId(colId: Int, innerId: InnerVal, isEdge: Boolean, useHash: Boolean) {
    //    play.api.Logger.debug(s"$this")
    lazy val hash = murmur3(innerId.value.toString)
    lazy val bytes = {
      var ret = if (useHash) Bytes.toBytes(hash) else Array.empty[Byte]
      isEdge match {
        case false =>
          Bytes.add(ret, innerId.bytes, Bytes.toBytes(colId))
        case true => Bytes.add(ret, innerId.bytes)
      }
    }
    lazy val bytesInUse = bytes.length
    def updateIsEdge(otherIsEdge: Boolean) = CompositeId(colId, innerId, otherIsEdge, useHash)
    def updateUseHash(otherUseHash: Boolean) = CompositeId(colId, innerId, isEdge, otherUseHash)
    override def equals(obj: Any) = {
      obj match {
        case other: CompositeId => colId == other.colId && innerId == other.innerId
        case _ => false
      }
    }
  }

  /**
   * label + direction
   */

  object LabelWithDirection {
    val maxBytes = Bytes.toBytes(Int.MaxValue)
    def apply(compositeInt: Int): LabelWithDirection = {
      //      play.api.Logger.debug(s"CompositeInt: $compositeInt")

      val dir = compositeInt & ((1 << bitsForDir) - 1)
      val labelId = compositeInt >> bitsForDir
      LabelWithDirection(labelId, dir)
    }
  }
  case class LabelWithDirection(labelId: Int, dir: Int) {
    assert(dir < (1 << bitsForDir))
    assert(labelId < (Int.MaxValue >> bitsForDir))

    val labelBits = labelId << bitsForDir

    lazy val compositeInt = labelBits | dir
    lazy val bytes = Bytes.toBytes(compositeInt)
    lazy val dirToggled = LabelWithDirection(labelId, toggleDir(dir))
    def updateDir(newDir: Int) = LabelWithDirection(labelId, newDir)

  }

  object InnerVal {
    val defaultVal = InnerVal(None, None, None)
    val stringLenOffset = 7.toByte
    val maxStringLen = Byte.MaxValue - stringLenOffset
    val maxMetaByte = Byte.MaxValue
    val minMetaByte = 0.toByte
    /**
     * first byte encoding rule.
     * 0 => default
     * 1 => long
     * 2 => int
     * 3 => short
     * 4 => byte
     * 5 => true
     * 6 => false
     * 7 ~ 127 => string len + 7
     */
    val metaByte = Map("default" -> 0, "long" -> 1, "int" -> 2, "short" -> 3,
      "byte" -> 4, "true" -> 5, "false" -> 6).map {
        case (k, v) => (k, v.toByte)
      }
    val metaByteRev = metaByte.map { case (k, v) => (v.toByte, k) } ++ metaByte.map { case (k, v) => ((-v).toByte, k) }

    def maxIdVal(dataType: String) = {
      dataType match {
        case "string" => InnerVal.withStr((0 until (Byte.MaxValue - stringLenOffset)).map("~").mkString)
        case "long" => InnerVal.withLong(Long.MaxValue)
        case "bool" => InnerVal.withBoolean(true)
        case _ => throw IllegalDataTypeException(dataType)
      }
    }
    def minIdVal(dataType: String) = {
      dataType match {
        case "string" => InnerVal.withStr("")
        case "long" => InnerVal.withLong(1)
        case "bool" => InnerVal.withBoolean(false)
        case _ => throw IllegalDataTypeException(dataType)
      }
    }
    def apply(bytes: Array[Byte], offset: Int): InnerVal = {
      var pos = offset
      //      
      val len = bytes(pos)
      //      play.api.Logger.debug(s"${bytes(offset)}: ${bytes.toList.slice(pos, bytes.length)}")
      pos += 1

      val (longV, strV, boolV) = metaByteRev.get(len) match {
        case Some(s) =>
          s match {
            case "default" => (None, None, None)
            case "true" => (None, None, Some(true))
            case "false" => (None, None, Some(false))
            case "byte" =>
              val b = bytes(pos)
              val value = if (b >= 0) Byte.MaxValue - b else Byte.MinValue - b - 1
              (Some(value.toLong), None, None)
            case "short" =>
              val b = Bytes.toShort(bytes, pos, 2)
              val value = if (b >= 0) Short.MaxValue - b else Short.MinValue - b - 1
              (Some(value.toLong), None, None)
            case "int" =>
              val b = Bytes.toInt(bytes, pos, 4)
              val value = if (b >= 0) Int.MaxValue - b else Int.MinValue - b - 1
              (Some(value.toLong), None, None)
            case "long" =>
              val b = Bytes.toLong(bytes, pos, 8)
              val value = if (b >= 0) Long.MaxValue - b else Long.MinValue - b - 1
              (Some(value.toLong), None, None)
          }
        case _ => // string
          val strLen = len - stringLenOffset
          (None, Some(Bytes.toString(bytes, pos, strLen)), None)
      }

      InnerVal(longV, strV, boolV)
    }

    def withLong(l: Long): InnerVal = {
      //      if (l < 0) throw new IllegalDataRangeException("value should be >= 0")
      InnerVal(Some(l), None, None)
    }
    def withStr(s: String): InnerVal = {
      InnerVal(None, Some(s), None)
    }
    def withBoolean(b: Boolean): InnerVal = {
      InnerVal(None, None, Some(b))
    }
    /**
     * In natural order
     * -129, -128 , -2, -1 < 0 < 1, 2, 127, 128
     *
     * In byte order
     * 0 < 1, 2, 127, 128 < -129, -128, -2, -1
     *
     */
    def transform(l: Long): (Byte, Array[Byte]) = {
      if (Byte.MinValue < l && l <= Byte.MaxValue) {
        //        val value = if (l < 0) l - Byte.MinValue else l + Byte.MinValue
        val key = if (l >= 0) metaByte("byte") else -metaByte("byte")
        val value = if (l >= 0) Byte.MaxValue - l else Byte.MinValue - l - 1
        val valueBytes = Array.fill(1)(value.toByte)
        (key.toByte, valueBytes)
      } else if (Short.MinValue < l && l <= Short.MaxValue) {
        val key = if (l >= 0) metaByte("short") else -metaByte("short")
        val value = if (l >= 0) Short.MaxValue - l else Short.MinValue - l - 1
        val valueBytes = Bytes.toBytes(value.toShort)
        (key.toByte, valueBytes)
      } else if (Int.MinValue < l && l <= Int.MaxValue) {
        val key = if (l >= 0) metaByte("int") else -metaByte("int")
        val value = if (l >= 0) Int.MaxValue - l else Int.MinValue - l - 1
        val valueBytes = Bytes.toBytes(value.toInt)
        (key.toByte, valueBytes)
      } else if (Long.MinValue < l && l <= Long.MaxValue) {
        val key = if (l >= 0) metaByte("long") else -metaByte("long")
        val value = if (l >= 0) Long.MaxValue - l else Long.MinValue - l - 1
        val valueBytes = Bytes.toBytes(value.toLong)
        (key.toByte, valueBytes)
      } else {
        throw new Exception(s"InnerVal range is out: $l")
      }
    }
  }

  case class InnerVal(longV: Option[Long], strV: Option[String], boolV: Option[Boolean]) {
    import InnerVal._

    lazy val bytes = {
      val (meta, valBytes) = (longV, strV, boolV) match {
        case (None, None, None) =>
          (metaByte("default"), Array.empty[Byte])
        case (Some(l), None, None) =>
          transform(l)
        case (None, None, Some(b)) =>
          val meta = if (b) metaByte("true") else metaByte("false")
          (meta, Array.empty[Byte])
        case (None, Some(s), None) =>
          val sBytes = Bytes.toBytes(s)
          if (sBytes.length > maxStringLen) throw new IllegalDataTypeException(s"string in innerVal maxSize is $maxStringLen, given ${sBytes.length}")
          assert(sBytes.length <= maxStringLen)
          val meta = (stringLenOffset + sBytes.length).toByte
          (meta, sBytes)
        case _ => throw new IllegalDataTypeException("innerVal data type should be [long/string/bool]")
      }
      Bytes.add(Array.fill(1)(meta.toByte), valBytes)
    }

    //    lazy val bytesInUse = bytes.length
    lazy val isDefault = longV.isEmpty && strV.isEmpty && boolV.isEmpty
    lazy val value = (longV, strV, boolV) match {
      case (Some(l), None, None) => l
      case (None, Some(s), None) => s
      case (None, None, Some(b)) => b
      case _ => throw new Exception(s"InnerVal should be [long/integeer/short/byte/string/boolean]")
    }
    lazy val valueType = (longV, strV, boolV) match {
      case (Some(l), None, None) => "long"
      case (None, Some(s), None) => "string"
      case (None, None, Some(b)) => "boolean"
      case _ => throw new Exception(s"InnerVal should be [long/integeer/short/byte/string/boolean]")
    }
    override def toString(): String = {
      value.toString
    }

    def compare(other: InnerVal) = {
      (value, other.value) match {
        case (v1: Long, v2: Long) => v1.compare(v2)
        case (b1: Boolean, b2: Boolean) => b1.compare(b2)
        case (s1: String, s2: String) => s1.compare(s2)
        case _ => throw new Exception("Please check a type of the compare operands")
      }
    }
    def +(other: InnerVal) = {
      (value, other.value) match {
        case (v1: Long, v2: Long) => InnerVal.withLong(v1 + v2)
        case (b1: Boolean, b2: Boolean) => InnerVal.withBoolean(if (b2) !b1 else b1)
        case _ => throw new Exception("Please check a type of the incr operands")
      }
    }
    def <(other: InnerVal) = this.compare(other) < 0
    def <=(other: InnerVal) = this.compare(other) <= 0
    def >(other: InnerVal) = this.compare(other) > 0
    def >=(other: InnerVal) = this.compare(other) >= 0
    
  }
  object InnerValWithTs {
    def apply(bytes: Array[Byte], offset: Int): InnerValWithTs = {
      val innerVal = InnerVal(bytes, offset)
      var pos = offset + innerVal.bytes.length
      val ts = Bytes.toLong(bytes, pos, 8)
      InnerValWithTs(innerVal, ts)
    }
    def withLong(value: Long, ts: Long) = InnerValWithTs(InnerVal.withLong(value), ts)
    def withStr(value: String, ts: Long) = InnerValWithTs(InnerVal.withStr(value), ts)
    def withBoolean(value: Boolean, ts: Long) = InnerValWithTs(InnerVal.withBoolean(value), ts)
    def withInnerVal(value: InnerVal, ts: Long) = InnerValWithTs(value, ts)
  }
  case class InnerValWithTs(innerVal: InnerVal, ts: Long) {
    lazy val bytes = Bytes.add(innerVal.bytes, Bytes.toBytes(ts))
  }
  def propsToBytes(props: Seq[(Byte, InnerVal)]): Array[Byte] = {
    val len = props.length
    assert(len < Byte.MaxValue)
    var bytes = Array.fill(1)(len.toByte)
    for ((k, v) <- props) bytes = Bytes.add(bytes, v.bytes)

    //    Logger.debug(s"propsToBytes: $props => ${bytes.toList}")
    bytes
  }
  def propsToKeyValues(props: Seq[(Byte, InnerVal)]): Array[Byte] = {
    val len = props.length
    assert(len < Byte.MaxValue)
    var bytes = Array.fill(1)(len.toByte)
    for ((k, v) <- props) bytes = Bytes.add(bytes, Array.fill(1)(k), v.bytes)
    //    Logger.debug(s"propsToBytes: $props => ${bytes.toList}")
    bytes
  }
  def propsToKeyValuesWithTs(props: Seq[(Byte, InnerValWithTs)]): Array[Byte] = {
    val len = props.length
    assert(len < Byte.MaxValue)
    var bytes = Array.fill(1)(len.toByte)
    for ((k, v) <- props) bytes = Bytes.add(bytes, Array.fill(1)(k), v.bytes)
    //    Logger.debug(s"propsToBytes: $props => ${bytes.toList}")
    bytes
  }
  def bytesToKeyValues(bytes: Array[Byte], offset: Int): (Seq[(Byte, InnerVal)], Int) = {
    var pos = offset
    val len = bytes(pos)
    pos += 1
    val kvs = new ArrayBuffer[(Byte, InnerVal)]
    for (i <- (0 until len)) {
      val k = bytes(pos)
      pos += 1
      val v = InnerVal(bytes, pos)
      pos += v.bytes.length
      kvs += (k -> v)
    }
    val ret = (kvs.toList, pos)
    //    Logger.debug(s"bytesToProps: $ret")
    ret
  }
  def bytesToKeyValuesWithTs(bytes: Array[Byte], offset: Int): (Seq[(Byte, InnerValWithTs)], Int) = {
    var pos = offset
    val len = bytes(pos)
    pos += 1
    val kvs = new ArrayBuffer[(Byte, InnerValWithTs)]
    for (i <- (0 until len)) {
      val k = bytes(pos)
      pos += 1
      val v = InnerValWithTs(bytes, pos)
      pos += v.bytes.length
      kvs += (k -> v)
    }
    val ret = (kvs.toList, pos)
    //    Logger.debug(s"bytesToProps: $ret")
    ret
  }
  def bytesToProps(bytes: Array[Byte], offset: Int): (Seq[(Byte, InnerVal)], Int) = {
    var pos = offset
    val len = bytes(pos)
    pos += 1
    val kvs = new ArrayBuffer[(Byte, InnerVal)]
    for (i <- (0 until len)) {
      val k = LabelMeta.emptyValue
      val v = InnerVal(bytes, pos)
      pos += v.bytes.length
      kvs += (k -> v)
    }
    val ret = (kvs.toList, pos)
    //    Logger.debug(s"bytesToProps: $ret")
    ret
  }
  def labelOrderSeqWithIsInverted(labelOrderSeq: Byte, isInverted: Boolean): Array[Byte] = {
    assert(labelOrderSeq < (1 << 6))
    val byte = labelOrderSeq << 1 | (if (isInverted) 1 else 0)
    Array.fill(1)(byte.toByte)
  }
  def bytesToLabelIndexSeqWithIsInverted(bytes: Array[Byte], offset: Int): (Byte, Boolean) = {
    val byte = bytes(offset)
    val isInverted = if ((byte & 1) != 0) true else false
    val labelOrderSeq = byte >> 1
    (labelOrderSeq.toByte, isInverted)
  }
  /**
   * hbase specific classes
   */

  object VertexRowKey {
    val isEdge = false
    def apply(bytes: Array[Byte], offset: Int): VertexRowKey = {
      VertexRowKey(CompositeId(bytes, offset, isEdge, true))
    }
  }

  case class VertexRowKey(id: CompositeId) {
    lazy val bytes = id.bytes
  }

  object VertexQualifier {
    def apply(bytes: Array[Byte], offset: Int, len: Int): VertexQualifier = {
      VertexQualifier(bytes(offset))
    }
  }
  case class VertexQualifier(propKey: Byte) {
    lazy val bytes = Array.fill(1)(propKey)
  }

  object EdgeRowKey {
    val propMode = 0
    val isEdge = true
    def apply(bytes: Array[Byte], offset: Int): EdgeRowKey = {
      var pos = offset
      val copmositeId = CompositeId(bytes, pos, isEdge, true)
      pos += copmositeId.bytesInUse
      val labelWithDir = LabelWithDirection(Bytes.toInt(bytes, pos, 4))
      pos += 4
      val (labelOrderSeq, isInverted) = bytesToLabelIndexSeqWithIsInverted(bytes, pos)
      EdgeRowKey(copmositeId, labelWithDir, labelOrderSeq, isInverted)
    }
  }
  //TODO: split inverted table? cf?
  case class EdgeRowKey(srcVertexId: CompositeId, labelWithDir: LabelWithDirection, labelOrderSeq: Byte, isInverted: Boolean) {
    //    play.api.Logger.debug(s"$this")
    lazy val innerSrcVertexId = srcVertexId.updateUseHash(true)
    lazy val bytes = Bytes.add(innerSrcVertexId.bytes, labelWithDir.bytes, labelOrderSeqWithIsInverted(labelOrderSeq, isInverted))
  }

  object EdgeQualifier {
    val isEdge = true
    def apply(bytes: Array[Byte], offset: Int, len: Int): EdgeQualifier = {
      var pos = offset
      val op = bytes(offset + len - 1)

      val (props, tgtVertexId) = {
        val (props, endAt) = bytesToProps(bytes, pos)
        val tgtVertexId = CompositeId(bytes, endAt, true, false)
        (props, tgtVertexId)
      }
      EdgeQualifier(props, tgtVertexId, op)
    }
  }
  case class EdgeQualifier(props: Seq[(Byte, InnerVal)], tgtVertexId: CompositeId, op: Byte) {

    val opBytes = Array.fill(1)(op)
    val innerTgtVertexId = tgtVertexId.updateUseHash(false)
    lazy val propsBytes = propsToBytes(props)
    lazy val bytes = Bytes.add(propsBytes, innerTgtVertexId.bytes, opBytes)
    //TODO:
    def propsKVs(labelId: Int, labelOrderSeq: Byte): List[(Byte, InnerVal)] = {
      val filtered = props.filter(kv => kv._1 != LabelMeta.emptyValue)
      if (filtered.isEmpty) {
        val opt = for (index <- LabelIndex.findByLabelIdAndSeq(labelId, labelOrderSeq)) yield {
          val v = index.metaSeqs.zip(props.map(_._2))
          v
        }
        opt.getOrElse(List.empty[(Byte, InnerVal)])
      } else {
        filtered.toList
      }
    }
  }
  object EdgeQualifierInverted {
    def apply(bytes: Array[Byte], offset: Int): EdgeQualifierInverted = {
      val tgtVertexId = CompositeId(bytes, offset, true, false)
      EdgeQualifierInverted(tgtVertexId)
    }
  }
  case class EdgeQualifierInverted(tgtVertexId: CompositeId) {
    //    play.api.Logger.debug(s"$this")
    val innerTgtVertexId = tgtVertexId.updateUseHash(false)
    lazy val bytes = innerTgtVertexId.bytes
  }
  object EdgeValue {
    def apply(bytes: Array[Byte], offset: Int): EdgeValue = {
      val (props, endAt) = bytesToKeyValues(bytes, offset)
      EdgeValue(props)
    }
  }
  case class EdgeValue(props: Seq[(Byte, InnerVal)]) {
    lazy val bytes = propsToKeyValues(props)
  }
  object EdgeValueInverted {
    def apply(bytes: Array[Byte], offset: Int): EdgeValueInverted = {
      var pos = offset
      val op = bytes(pos)
      pos += 1
      val (props, endAt) = bytesToKeyValuesWithTs(bytes, pos)
      EdgeValueInverted(op, props)
    }
  }
  case class EdgeValueInverted(op: Byte, props: Seq[(Byte, InnerValWithTs)]) {
    lazy val bytes = Bytes.add(Array.fill(1)(op), propsToKeyValuesWithTs(props))
  }

}