/***********************************************************************
 * Copyright (c) 2013-2025 Commonwealth Computer Research, Inc.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Apache License, Version 2.0
 * which accompanies this distribution and is available at
 * http://www.opensource.org/licenses/apache2.0.php.
 ***********************************************************************/

package org.locationtech.geomesa.hbase.data

import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes
import org.locationtech.geomesa.hbase.utils.HBaseVersions
import org.locationtech.geomesa.index.metadata.{KeyValueStoreMetadata, MetadataSerializer}
import org.locationtech.geomesa.utils.collection.CloseableIterator
import org.locationtech.geomesa.utils.io.WithClose

import scala.collection.JavaConverters._

class HBaseBackedMetadata[T](connection: Connection, catalog: TableName, val serializer: MetadataSerializer[T])
    extends KeyValueStoreMetadata[T] {

  import HBaseBackedMetadata._
  import state.table

  // state object to allow table val to be instantiated before superclass initializes
  private object state {
    val table: Table = connection.getTable(catalog)
  }

  override protected def checkIfTableExists: Boolean = WithClose(connection.getAdmin)(_.tableExists(catalog))

  override protected def createTable(): Unit = {
    WithClose(connection.getAdmin) { admin =>
      if (!admin.tableExists(catalog)) {
        HBaseVersions.createTableAsync(admin, catalog, Seq(ColumnFamily), None, None, None, Some(true), None, Seq.empty)
        HBaseIndexAdapter.waitForTable(admin, catalog)
      }
    }
  }

  override protected def createEmptyBackup(timestamp: String): HBaseBackedMetadata[T] =
    new HBaseBackedMetadata(connection, TableName.valueOf(s"${catalog}_${timestamp}_bak"), serializer)

  override protected def write(rows: Seq[(Array[Byte], Array[Byte])]): Unit =
    table.put(rows.map { case (r, v) => new Put(r).addColumn(ColumnFamily, ColumnQualifier, v) }.toList.asJava)

  override protected def delete(rows: Seq[Array[Byte]]): Unit =
    // note: list passed in must be mutable
    table.delete(rows.map(r => new Delete(r)).toBuffer.asJava)

  override protected def scanValue(row: Array[Byte]): Option[Array[Byte]] = {
    val result = table.get(new Get(row).addColumn(ColumnFamily, ColumnQualifier))
    if (result.isEmpty) { None } else { Option(result.getValue(ColumnFamily, ColumnQualifier)) }
  }

  override protected def scanRows(prefix: Option[Array[Byte]]): CloseableIterator[(Array[Byte], Array[Byte])] = {
    val scan = new Scan().addColumn(ColumnFamily, ColumnQualifier)
    prefix.foreach(scan.setRowPrefixFilter)
    val scanner = table.getScanner(scan)
    val results = scanner.iterator.asScala.map(s => (s.getRow, s.getValue(ColumnFamily, ColumnQualifier)))
    CloseableIterator(results, scanner.close())
  }

  override def close(): Unit = table.close()
}

object HBaseBackedMetadata {
  val ColumnFamily: Array[Byte] = Bytes.toBytes("m")
  val ColumnQualifier: Array[Byte] = Bytes.toBytes("v")
}
