/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile;

import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.schema.SchemaConstant;
import org.apache.iotdb.commons.schema.node.role.IMeasurementMNode;
import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory;
import org.apache.iotdb.commons.schema.view.LogicalViewSchema;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.ICachedMNode;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.container.ICachedMNodeContainer;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.loader.MNodeFactoryLoader;

import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Map;

/**
 * This class translate an ICacheMNode into a bytebuffer, or vice versa. Expected to support record
 * as entry of segment-level index further. Coupling with ICacheMNode structure.<br>
 * <br>
 * TODO: Guardian statements on higher stack NEEDED. The longest ALIAS is limited to 0x7fff(32767)
 * bytes for that a Short is used to record the length, hence a colossal record may collapse the
 * stack.
 */
public class RecordUtils {
  // Offsets of ICacheMNode infos in a record buffer
  private static final short INTERNAL_NODE_LENGTH =
      (short) 1 + 2 + 8 + 4 + 1; // always fixed length record
  private static final short MEASUREMENT_BASIC_LENGTH =
      (short) 1 + 2 + 8 + 8 + 4; // final length depends on its alias and props
  private static final short VIEW_BASIC_LENGTH =
      (short) 1 + 2 + 8 + 1; // final length depends on its view expression

  /** These offset rather than magic number may also be used to track usage of related field. */
  private static final short LENGTH_OFFSET = 1;

  private static final short ALIAS_OFFSET = 19;
  private static final short SEG_ADDRESS_OFFSET = 3;
  private static final short SCHEMA_OFFSET = 11;
  private static final short VIEW_OFFSET = 12;
  private static final short INTERNAL_BITFLAG_OFFSET = 15;

  private static final byte INTERNAL_TYPE = 0;
  private static final byte ENTITY_TYPE = 1;
  private static final byte MEASUREMENT_TYPE = 4;
  private static final byte VIEW_TYPE = 5;

  private static final IMNodeFactory<ICachedMNode> nodeFactory =
      MNodeFactoryLoader.getInstance().getCachedMNodeIMNodeFactory();

  public static ByteBuffer node2Buffer(ICachedMNode node) {
    if (node.isMeasurement()) {
      IMeasurementMNode<ICachedMNode> measurementMNode = node.getAsMeasurementMNode();
      if (measurementMNode.isLogicalView()) {
        return view2Buffer(measurementMNode);
      } else {
        return measurement2Buffer(measurementMNode);
      }
    } else {
      return internal2Buffer(node);
    }
  }

  public static int getRecordLength(ICachedMNode node) {
    if (node.isMeasurement()) {
      return getRecordLength(node.getAsMeasurementMNode());
    } else {
      return INTERNAL_NODE_LENGTH;
    }
  }

  private static int getRecordLength(IMeasurementMNode<ICachedMNode> node) {
    if (node.isLogicalView()) {
      return VIEW_BASIC_LENGTH
          + ViewExpression.getSerializeSize(((LogicalViewSchema) node.getSchema()).getExpression());
    } else {
      // consider props and alias
      int bufferLength =
          node.getAlias() == null
              ? 4 + MEASUREMENT_BASIC_LENGTH
              : (node.getAlias().getBytes().length + 4 + MEASUREMENT_BASIC_LENGTH);
      if (node.getSchema().getProps() != null) {
        for (Map.Entry<String, String> e : node.getSchema().getProps().entrySet()) {
          bufferLength += 8 + e.getKey().getBytes().length + e.getValue().length();
        }
      }
      return bufferLength;
    }
  }

  /**
   * Internal/Entity MNode Record Structure (in bytes): <br>
   * (fixed length record)
   *
   * <ul>
   *   <li>1 byte: nodeType, 0 for internal, 1 for entity, 4 for measurement
   *   <li>1 short (2 bytes): recLen, length of record (remove it may reduce space overhead while a
   *       bit slower)
   *   <li>1 long (8 bytes): glbIndex, combined index to its children records
   *   <li>1 int (4 byte): templateId, id of template, occupies only 1 byte before
   * </ul>
   *
   * -- bitwise flags (1 byte) --
   *
   * <ul>
   *   <li>1 bit : usingTemplate, whether using template
   *   <li>2 bit : isAligned (00 for not aligned, 01 for aligned, 10 for null)
   * </ul>
   *
   * @param node
   * @return
   */
  private static ByteBuffer internal2Buffer(ICachedMNode node) {
    byte nodeType = INTERNAL_TYPE;
    Boolean isAligned = null;
    int schemaTemplateIdWithState = SchemaConstant.NON_TEMPLATE;
    boolean isUseTemplate = false;

    if (node.isDevice()) {
      nodeType = ENTITY_TYPE;
      isAligned = node.getAsDeviceMNode().isAlignedNullable();
      schemaTemplateIdWithState = node.getAsDeviceMNode().getSchemaTemplateIdWithState();
      isUseTemplate = node.getAsDeviceMNode().isUseTemplate();
    }

    ByteBuffer buffer = ByteBuffer.allocate(INTERNAL_NODE_LENGTH);
    ReadWriteIOUtils.write(nodeType, buffer);
    ReadWriteIOUtils.write(INTERNAL_NODE_LENGTH, buffer);
    ReadWriteIOUtils.write(
        ICachedMNodeContainer.getCachedMNodeContainer(node).getSegmentAddress(), buffer);
    ReadWriteIOUtils.write(schemaTemplateIdWithState, buffer);

    // encode bitwise flag
    byte useAndAligned = encodeInternalStatus(isUseTemplate, isAligned);

    ReadWriteIOUtils.write(useAndAligned, buffer);
    return buffer;
  }

  /**
   * It is convenient to expand the semantic of the statusBytes for further status of a measurement,
   * e.g., preDelete, since 8 bytes are far more sufficient to represent the schema of it.
   *
   * <p>Measurement MNode Record Structure: <br>
   * (var length record, with length member)
   *
   * <ul>
   *   <li>1 byte: nodeType, as above
   *   <li>1 short (2 bytes): recLength, length of whole record
   *   <li>1 long (8 bytes): tagIndex, value of the offset within a measurement
   *   <li>1 long (8 bytes): statusBytes, including datatype/compressionType/encoding and so on
   *   <li>var length string (4+var_length bytes): alias
   *   <li>var length map (4+var_length bytes): props, serialized by {@link ReadWriteIOUtils}
   * </ul>
   *
   * <p>It doesn't use MeasurementSchema.serializeTo for duplication of measurementId
   */
  private static ByteBuffer measurement2Buffer(IMeasurementMNode<ICachedMNode> node) {
    int bufferLength = getRecordLength(node);
    // normal measurement
    ByteBuffer buffer = ByteBuffer.allocate(bufferLength);
    ReadWriteIOUtils.write(MEASUREMENT_TYPE, buffer);
    ReadWriteIOUtils.write((short) bufferLength, buffer);
    ReadWriteIOUtils.write(convertTags2Long(node), buffer);
    ReadWriteIOUtils.write(convertMeasStat2Long(node), buffer);
    ReadWriteIOUtils.write(node.getAlias(), buffer);
    ReadWriteIOUtils.write(node.getSchema().getProps(), buffer);
    return buffer;
  }

  /**
   * LogicalView MNode Record Structure: <br>
   * (var length record, with length member)
   *
   * <ul>
   *   <li>1 byte: nodeType, as above
   *   <li>1 short (2 bytes): recLength, length of whole record
   *   <li>1 long (8 bytes): tagIndex, value of the offset within a measurement
   *   <li>1 boolean (1 bytes): statusBytes, isPreDeleted
   *   <li>var length ViewExpression
   * </ul>
   */
  private static ByteBuffer view2Buffer(IMeasurementMNode<ICachedMNode> node) {
    ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
    try {
      ViewExpression.serialize(
          ((LogicalViewSchema) node.getSchema()).getExpression(), byteArrayOutputStream);
    } catch (IOException e) {
      // ByteArrayOutputStream is a memory-based output stream that does not involve disk IO and
      // will not throw an IOException except for OOM.
      throw new RuntimeException(e);
    }
    byte[] expressionData = byteArrayOutputStream.toByteArray();

    int bufferLength = VIEW_BASIC_LENGTH + expressionData.length;

    ByteBuffer buffer = ByteBuffer.allocate(bufferLength);
    ReadWriteIOUtils.write(VIEW_TYPE, buffer);
    ReadWriteIOUtils.write((short) bufferLength, buffer);
    ReadWriteIOUtils.write(convertTags2Long(node), buffer);
    ReadWriteIOUtils.write(node.isPreDeleted(), buffer);
    buffer.put(expressionData);
    return buffer;
  }

  /**
   * NOTICE: Make sure that buffer has set its position and limit clearly before pass to this
   * method.<br>
   * <br>
   * FIXME: recLen is not futile although 'never used', since when decode from a buffer, this field
   * alleviate an extra mem read for length of alias. BUT it is indeed redundant with length of
   * alias which flushed by {@linkplain ReadWriteIOUtils#write(String, ByteBuffer)}, and error-prone
   * since it only contains 2 bytes while the latter contains 4.
   *
   * @param nodeName name of the constructed node
   * @param buffer content of the node
   * @return node constructed from buffer
   */
  public static ICachedMNode buffer2Node(String nodeName, ByteBuffer buffer)
      throws MetadataException {
    ICachedMNode resNode;

    byte nodeType = ReadWriteIOUtils.readByte(buffer);
    if (nodeType < 2) {
      // internal or entity node

      short recLen = ReadWriteIOUtils.readShort(buffer);
      long segAddr = ReadWriteIOUtils.readLong(buffer);
      int templateId = ReadWriteIOUtils.readInt(buffer);
      byte bitFlag = ReadWriteIOUtils.readByte(buffer);

      boolean usingTemplate = usingTemplate(bitFlag);
      Boolean isAligned = isAligned(bitFlag);

      if (nodeType == 0) {
        resNode = nodeFactory.createInternalMNode(null, nodeName);
      } else {
        resNode = nodeFactory.createDeviceMNode(null, nodeName).getAsMNode();
        resNode.getAsDeviceMNode().setAligned(isAligned);
        resNode.getAsDeviceMNode().setUseTemplate(usingTemplate);
        resNode.getAsDeviceMNode().setSchemaTemplateId(templateId);
      }

      ICachedMNodeContainer.getCachedMNodeContainer(resNode).setSegmentAddress(segAddr);

      return resNode;
    } else if (nodeType == MEASUREMENT_TYPE) {
      // measurement node
      short recLenth = ReadWriteIOUtils.readShort(buffer);
      long tagIndex = ReadWriteIOUtils.readLong(buffer);
      long schemaByte = ReadWriteIOUtils.readLong(buffer);
      String alias = ReadWriteIOUtils.readString(buffer);
      Map<String, String> props = ReadWriteIOUtils.readMap(buffer);
      return paddingMeasurement(nodeName, tagIndex, schemaByte, alias, props);
    } else if (nodeType == VIEW_TYPE) {
      short recLenth = ReadWriteIOUtils.readShort(buffer);
      long tagIndex = ReadWriteIOUtils.readLong(buffer);
      boolean isPreDeleted = ReadWriteIOUtils.readBool(buffer);
      ViewExpression viewExpression = ViewExpression.deserialize(buffer);
      return paddingLogicalView(nodeName, tagIndex, isPreDeleted, viewExpression);
    }
    throw new MetadataException("Unrecognized node type: " + nodeType);
  }

  // region Getter and Setter to Record Buffer
  /** These methods need a buffer whose position is ready to read. */
  public static short getRecordLength(ByteBuffer recBuf) {
    int oriPos = recBuf.position();
    recBuf.position(oriPos + LENGTH_OFFSET);
    short len = ReadWriteIOUtils.readShort(recBuf);
    recBuf.position(oriPos);
    return len;
  }

  public static byte getRecordType(ByteBuffer recBuf) {
    int oriPos = recBuf.position();
    recBuf.position(oriPos);
    byte type = ReadWriteIOUtils.readByte(recBuf);
    recBuf.position(oriPos);
    return type;
  }

  public static long getRecordSegAddr(ByteBuffer recBuf) {
    int oriPos = recBuf.position();
    recBuf.position(oriPos + SEG_ADDRESS_OFFSET);
    long addr = ReadWriteIOUtils.readLong(recBuf);
    recBuf.position(oriPos);
    return addr;
  }

  /** return as: [dataType, encoding, compression, preDelete] */
  public static byte[] getMeasStatsBytes(ByteBuffer recBuf) {
    byte[] res = new byte[4];
    int oriPos = recBuf.position();
    recBuf.position(oriPos + SCHEMA_OFFSET);
    long statusBytes = ReadWriteIOUtils.readLong(recBuf);
    res[0] = (byte) (statusBytes >>> 16 & 0xffL);
    res[1] = (byte) (statusBytes >>> 8 & 0xffL);
    res[2] = (byte) (statusBytes & 0xffL);
    res[3] = (byte) (statusBytes >>> 24 & 0xffL);
    recBuf.position(oriPos);
    return res;
  }

  /** return as: [dataType, encoding, compression, preDelete] */
  public static ViewExpression getViewExpression(ByteBuffer recBuf) {
    int oriPos = recBuf.position();
    recBuf.position(oriPos + VIEW_OFFSET);
    return ViewExpression.deserialize(recBuf);
  }

  public static Boolean getAlignment(ByteBuffer recBuf) {
    int oriPos = recBuf.position();
    recBuf.position(oriPos + INTERNAL_BITFLAG_OFFSET);
    byte flag = ReadWriteIOUtils.readByte(recBuf);
    recBuf.position(oriPos);
    return isAligned(flag);
  }

  public static String getRecordAlias(ByteBuffer recBuf) {
    int oriPos = recBuf.position();
    if (ReadWriteIOUtils.readByte(recBuf) != MEASUREMENT_TYPE) {
      recBuf.position(oriPos);
      return null;
    }
    recBuf.position(oriPos + ALIAS_OFFSET);
    String alias = ReadWriteIOUtils.readString(recBuf);
    recBuf.position(oriPos);
    return alias;
  }

  public static void updateSegAddr(ByteBuffer recBuf, long newSegAddr) {
    int oriPos = recBuf.position();
    recBuf.position(oriPos + SEG_ADDRESS_OFFSET);
    ReadWriteIOUtils.write(newSegAddr, recBuf);
    recBuf.position(oriPos);
  }

  // endregion

  @TestOnly
  public static String buffer2String(ByteBuffer buffer) throws MetadataException {
    StringBuilder builder = new StringBuilder("[");
    ICachedMNode node = buffer2Node("unspecified", buffer);
    if (node.isMeasurement()) {
      builder.append("measurementNode, ");
      builder.append(
          String.format(
              "alias: %s, ",
              node.getAsMeasurementMNode().getAlias() == null
                  ? ""
                  : node.getAsMeasurementMNode().getAlias()));
      builder.append(
          String.format("type: %s, ", node.getAsMeasurementMNode().getDataType().toString()));
      builder.append(
          String.format(
              "encoding: %s, ",
              node.getAsMeasurementMNode().getSchema().getEncodingType().toString()));
      builder.append(
          String.format(
              "compressor: %s]",
              node.getAsMeasurementMNode().getSchema().getCompressor().toString()));
      return builder.toString();
    } else if (node.isDevice()) {
      builder.append("entityNode, ");

      if (node.getAsDeviceMNode().isAlignedNullable() == null) {
        builder.append("aligned is null, ");
      } else if (node.getAsDeviceMNode().isAligned()) {
        builder.append("aligned, ");
      } else {
        builder.append("not aligned, ");
      }

      if (node.getAsDeviceMNode().isUseTemplate()) {
        builder.append("using template.]");
      } else {
        builder.append("not using template.]");
      }

    } else {
      builder.append("internalNode, ");
    }

    return builder.toString();
  }

  // region padding with ICacheMNode
  /** These 2 convert methods are coupling with tag, template module respectively. */
  private static long convertTags2Long(IMeasurementMNode<ICachedMNode> node) {
    return node.getOffset();
  }

  /** Including schema and pre-delete flag of a measurement, could be expanded further. */
  private static long convertMeasStat2Long(IMeasurementMNode<ICachedMNode> node) {
    byte dataType = node.getSchema().getTypeInByte();
    byte encoding = node.getSchema().getEncodingType().serialize();
    byte compressor = node.getSchema().getCompressor().serialize();
    byte preDelete = (byte) (node.getAsMeasurementMNode().isPreDeleted() ? 0x01 : 0x00);

    return (preDelete << 24 | dataType << 16 | encoding << 8 | compressor);
  }

  private static ICachedMNode paddingMeasurement(
      String nodeName, long tagIndex, long statsBytes, String alias, Map<String, String> props) {
    byte preDel = (byte) (statsBytes >>> 24);
    byte dataType = (byte) (statsBytes >>> 16);
    byte encoding = (byte) ((statsBytes >>> 8) & 0xffL);
    byte compressor = (byte) (statsBytes & 0xffL);

    IMeasurementSchema schema =
        new MeasurementSchema(
            nodeName,
            TSDataType.values()[dataType],
            TSEncoding.values()[encoding],
            CompressionType.deserialize(compressor),
            props);

    ICachedMNode res =
        nodeFactory.createMeasurementMNode(null, nodeName, schema, alias).getAsMNode();
    res.getAsMeasurementMNode().setOffset(tagIndex);

    if (preDel > 0) {
      res.getAsMeasurementMNode().setPreDeleted(true);
    }

    return res;
  }

  private static ICachedMNode paddingLogicalView(
      String nodeName, long tagIndex, boolean isPreDeleted, ViewExpression viewExpression) {
    IMeasurementSchema schema = new LogicalViewSchema(nodeName, viewExpression);
    IMeasurementMNode<ICachedMNode> res =
        nodeFactory.createLogicalViewMNode(null, nodeName, schema);
    res.setOffset(tagIndex);
    res.setPreDeleted(isPreDeleted);
    return res.getAsMNode();
  }

  // endregion

  // region codex for bit flag

  private static byte encodeInternalStatus(boolean usingTemplate, Boolean isAligned) {
    byte flag = 0;
    if (usingTemplate) {
      flag |= 0x01;
    }
    if (isAligned == null) {
      flag |= 0x04;
    } else if (isAligned) {
      flag |= 0x02;
    }
    return flag;
  }

  private static Boolean isAligned(byte flag) {
    if ((flag & 0x04) != 0) {
      return null;
    } else {
      return (flag & 0x02) == 2;
    }
  }

  private static boolean usingTemplate(byte flag) {
    return (flag & 0x01) == 1;
  }

  // endregion
}
