package com.bigdata.cluster.tools

import java.util.concurrent.atomic.AtomicInteger
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase._
import org.apache.hadoop.hbase.client.{Admin, Connection, ConnectionFactory, Put}
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm
import org.apache.hadoop.hbase.util.{Bytes, MD5Hash}
import org.apache.hadoop.hbase.util.RegionSplitter.HexStringSplit

/**
  * HBase 리소스 생명주기와 행키 생성기를 담당하는 유틸리티 클래스
  */
object ClusterManager {
  // 설정 인스턴스
  private val settings: Configuration = new Configuration()
  settings.set("hbase.zookeeper.quorum", "zk_host_1,zk_host_2")
  settings.set("hbase.master", "hmaster_host")
  settings.set("hbase.zookeeper.property.clientPort", "2181")
  
  // 커넥션 풀 관리
  private val connectionInstance: Connection = ConnectionFactory.createConnection(settings)
  private val adminInstance: Admin = connectionInstance.getAdmin
  
  // 작업 결과 카운터
  private val operationCounter = new AtomicInteger(0)

  /**
    * 테이블 초기화 및 사전 분할 실행
    * @param schemaName 데이터베이스 스키마 이름
    * @param partitions 분할할 리전 개수
    * @param columnFamilies 열 가족 목록
    */
  def bootstrapTable(schemaName: String, partitions: Int, columnFamilies: Array[String]): Unit = {
    val tableNameObj = TableName.valueOf(schemaName)
    
    // 테이블이 존재하지 않는 경우에만 생성
    if (!adminInstance.tableExists(tableNameObj)) {
      val descriptor = new HTableDescriptor(tableNameObj)
      
      // 열 가족 설정
      columnFamilies.foreach(family => {
        val colDesc = new HColumnDescriptor(family)
        colDesc.setCompressionType(Algorithm.SNAPPY)
        colDesc.setMaxVersions(1)
        descriptor.addFamily(colDesc)
      })
      
      // HexStringSplit 을 이용한 분할 키 자동 생성
      val splitter = new HexStringSplit()
      val boundaryKeys = splitter.split(partitions)
      
      println(s"$schemaName 테이블 생성 완료, 총 $partitions 개 리전 분할됨.")
      adminInstance.createTable(descriptor, boundaryKeys)
    }
  }

  /**
    * MD5 해시값을 사용하여 RowKey 전위자를 생성
    * 데이터 편중 방지를 위해 해시값의 일부를 RowKey 앞에 붙임
    * @param saltLength 사용될 해시 길이 (기본값 4 자)
    * @param rawKey 원래 비즈니스 키 값들
    */
  def createHashedRowKey(saltLength: Int, rawKey: String*): Array[Byte] = {
    val concatenatedKey = rawKey.mkString("-")
    val md5Hex = MD5Hash.getMD5AsHex(Bytes.toBytes(concatenatedKey))
    
    // 지정된 길이의 해시 접두사 추출
    val hashPrefix = if (saltLength < 8 && saltLength > 0) {
      md5Hex.substring(0, saltLength)
    } else {
      md5Hex.substring(0, 4)
    }
    
    Bytes.toBytes(s"$hashPrefix$concatenatedKey")
  }

  /**
    * 데이터 저장용 PUT 명령어 실행
    */
  def insertRecord(tableId: String, keyValue: String, family: String, column: String, value: String): Boolean = {
    try {
      val table = connectionInstance.getTable(TableName.valueOf(tableId))
      val rowBytes = createHashedRowKey(4, keyValue)
      
      val putOp = new Put(rowBytes)
      putOp.addColumn(
        Bytes.toBytes(family),
        Bytes.toBytes(column),
        Bytes.toBytes(value)
      )
      
      table.put(putOp)
      operationCounter.incrementAndGet()
      true
    } catch {
      case e: Exception =>
        println(s"데이터 저장 실패: ${e.getMessage}")
        operationCounter.decrementAndGet()
        false
    } finally {
      // 리소스 누수 방지를 위한 명시적 닫기 또는 커넥션 풀 반환 권장
      println("작업 처리 완료.")
    }
  }
}