/*
 * 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.engine.modification;

import org.apache.iotdb.db.engine.modification.io.LocalTextModificationAccessor;
import org.apache.iotdb.db.engine.modification.io.ModificationReader;
import org.apache.iotdb.db.engine.modification.io.ModificationWriter;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Random;

/**
 * ModificationFile stores the Modifications of a TsFile or unseq file in another file in the same
 * directory. Methods in this class are highly synchronized for concurrency safety.
 */
public class ModificationFile implements AutoCloseable {

  private static final Logger logger = LoggerFactory.getLogger(ModificationFile.class);
  public static final String FILE_SUFFIX = ".mods";
  public static final String COMPACTION_FILE_SUFFIX = ".compaction.mods";

  // lazy loaded, set null when closed
  private List<Modification> modifications;
  private ModificationWriter writer;
  private ModificationReader reader;
  private String filePath;
  private Random random = new Random();

  /**
   * Construct a ModificationFile using a file as its storage.
   *
   * @param filePath the path of the storage file.
   */
  public ModificationFile(String filePath) {
    LocalTextModificationAccessor accessor = new LocalTextModificationAccessor(filePath);
    this.writer = accessor;
    this.reader = accessor;
    this.filePath = filePath;
  }

  private void init() {
    synchronized (this) {
      modifications = (List<Modification>) reader.read();
    }
  }

  private void checkInit() {
    if (modifications == null) {
      init();
    }
  }

  /** Release resources such as streams and caches. */
  @Override
  public void close() throws IOException {
    synchronized (this) {
      writer.close();
      modifications = null;
    }
  }

  public void abort() throws IOException {
    synchronized (this) {
      writer.abort();
      if (modifications != null && !modifications.isEmpty()) {
        modifications.remove(modifications.size() - 1);
      }
    }
  }

  /**
   * Write a modification in this file. The modification will first be written to the persistent
   * store then the memory cache.
   *
   * @param mod the modification to be written.
   * @throws IOException if IOException is thrown when writing the modification to the store.
   */
  public void write(Modification mod) throws IOException {
    synchronized (this) {
      writer.write(mod);
      if (modifications != null) {
        modifications.add(mod);
      }
    }
  }

  /**
   * Get all modifications stored in this file.
   *
   * @return an ArrayList of modifications.
   */
  public Collection<Modification> getModifications() {
    synchronized (this) {
      checkInit();
      return new ArrayList<>(modifications);
    }
  }

  public String getFilePath() {
    return filePath;
  }

  public void setFilePath(String filePath) {
    this.filePath = filePath;
  }

  public void remove() throws IOException {
    close();
    FSFactoryProducer.getFSFactory().getFile(filePath).delete();
  }

  public boolean exists() {
    return new File(filePath).exists();
  }

  /**
   * Create a hardlink for the modification file. The hardlink with have a suffix like
   * ".{sysTime}_{randomLong}"
   *
   * @return a new ModificationFile with its path changed to the hardlink, or null if the origin
   *     file does not exist or the hardlink cannot be created.
   */
  public ModificationFile createHardlink() {
    if (!exists()) {
      return null;
    }

    while (true) {
      String hardlinkSuffix =
          TsFileConstant.PATH_SEPARATOR + System.currentTimeMillis() + "_" + random.nextLong();
      File hardlink = new File(filePath + hardlinkSuffix);

      try {
        Files.createLink(Paths.get(hardlink.getAbsolutePath()), Paths.get(filePath));
        return new ModificationFile(hardlink.getAbsolutePath());
      } catch (FileAlreadyExistsException e) {
        // retry a different name if the file is already created
      } catch (IOException e) {
        logger.error("Cannot create hardlink for {}", filePath, e);
        return null;
      }
    }
  }

  public static ModificationFile getNormalMods(TsFileResource tsFileResource) {
    return new ModificationFile(tsFileResource.getTsFilePath() + ModificationFile.FILE_SUFFIX);
  }

  public static ModificationFile getCompactionMods(TsFileResource tsFileResource) {
    return new ModificationFile(
        tsFileResource.getTsFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
  }
}