/*
 * 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.sync.receiver.recover;

import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.directories.DirectoryManager;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.DiskSpaceInsufficientException;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.sync.conf.SyncConstant;
import org.apache.iotdb.db.sync.receiver.load.FileLoader;
import org.apache.iotdb.db.sync.receiver.load.FileLoaderManager;
import org.apache.iotdb.db.sync.receiver.load.FileLoaderTest;
import org.apache.iotdb.db.sync.receiver.load.IFileLoader;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.*;

import static org.junit.Assert.*;

public class SyncReceiverLogAnalyzerTest {

  private static final Logger LOGGER = LoggerFactory.getLogger(FileLoaderTest.class);
  private static final String SG_NAME = "root.sg";
  private String dataDir;
  private IFileLoader fileLoader;
  private ISyncReceiverLogAnalyzer logAnalyze;
  private ISyncReceiverLogger receiverLogger;

  @Before
  public void setUp()
      throws IOException, InterruptedException, StartupException, DiskSpaceInsufficientException, MetadataException {
    EnvironmentUtils.closeStatMonitor();
    EnvironmentUtils.envSetUp();
    dataDir = new File(DirectoryManager.getInstance().getNextFolderForSequenceFile())
        .getParentFile().getAbsolutePath();
    logAnalyze = SyncReceiverLogAnalyzer.getInstance();
    initMetadata();
  }

  private void initMetadata() throws MetadataException {
    MManager mmanager = MManager.getInstance();
    mmanager.init();
    mmanager.setStorageGroup("root.sg0");
    mmanager.setStorageGroup("root.sg1");
    mmanager.setStorageGroup("root.sg2");
  }

  @After
  public void tearDown() throws InterruptedException, IOException, StorageEngineException {
    EnvironmentUtils.cleanEnv();
  }

  @Test
  public void recover() throws IOException, StorageEngineException, InterruptedException {
    receiverLogger = new SyncReceiverLogger(
        new File(getReceiverFolderFile(), SyncConstant.SYNC_LOG_NAME));
    fileLoader = FileLoader.createFileLoader(getReceiverFolderFile());
    Map<String, Set<File>> allFileList = new HashMap<>();
    Map<String, Set<File>> correctSequenceLoadedFileMap = new HashMap<>();

    // add some new tsfiles
    Random r = new Random(0);
    receiverLogger.startSyncTsFiles();
    Set<String> toBeSyncedFiles = new HashSet<>();
    for (int i = 0; i < 3; i++) {
      for (int j = 0; j < 10; j++) {
        allFileList.putIfAbsent(SG_NAME + i, new HashSet<>());
        correctSequenceLoadedFileMap.putIfAbsent(SG_NAME + i, new HashSet<>());
        String rand = String.valueOf(r.nextInt(10000) + i * j);
        String fileName =
            getSnapshotFolder() + File.separator + SG_NAME + i + File.separator + System
                .currentTimeMillis() + IoTDBConstant.TSFILE_NAME_SEPARATOR + rand
                + IoTDBConstant.TSFILE_NAME_SEPARATOR + "0.tsfile";
        Thread.sleep(1);
        File syncFile = new File(fileName);
        receiverLogger
            .finishSyncTsfile(syncFile);
        toBeSyncedFiles.add(syncFile.getAbsolutePath());
        File dataFile = new File(
            DirectoryManager.getInstance().getNextFolderForSequenceFile(),
            syncFile.getParentFile().getName() + File.separatorChar
                + syncFile.getName());
        correctSequenceLoadedFileMap.get(SG_NAME + i).add(dataFile);
        allFileList.get(SG_NAME + i).add(syncFile);
        if (!syncFile.getParentFile().exists()) {
          syncFile.getParentFile().mkdirs();
        }
        if (!syncFile.exists() && !syncFile.createNewFile()) {
          LOGGER.error("Can not create new file {}", syncFile.getPath());
        }
        if (!new File(syncFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).exists()
            && !new File(syncFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX)
            .createNewFile()) {
          LOGGER.error("Can not create new file {}", syncFile.getPath());
        }
        TsFileResource tsFileResource = new TsFileResource(syncFile);
        tsFileResource.putStartTime(String.valueOf(i), (long) j * 10);
        tsFileResource.putEndTime(String.valueOf(i), (long) j * 10 + 5);
        tsFileResource.serialize();
      }
    }

    for (int i = 0; i < 3; i++) {
      StorageGroupProcessor processor = StorageEngine.getInstance().getProcessor(SG_NAME + i);
      assertTrue(processor.getSequenceFileTreeSet().isEmpty());
      assertTrue(processor.getUnSequenceFileList().isEmpty());
    }

    assertTrue(getReceiverFolderFile().exists());
    for (Set<File> set : allFileList.values()) {
      for (File newTsFile : set) {
        if (!newTsFile.getName().endsWith(TsFileResource.RESOURCE_SUFFIX)) {
          fileLoader.addTsfile(newTsFile);
        }
      }
    }

    receiverLogger.close();
    assertTrue(new File(getReceiverFolderFile(), SyncConstant.LOAD_LOG_NAME).exists());
    assertTrue(new File(getReceiverFolderFile(), SyncConstant.SYNC_LOG_NAME).exists());
    assertTrue(
        FileLoaderManager.getInstance().containsFileLoader(getReceiverFolderFile().getName()));
    int mode = 0;
    Set<String> toBeSyncedFilesTest = new HashSet<>();
    try (BufferedReader br = new BufferedReader(
        new FileReader(new File(getReceiverFolderFile(), SyncConstant.SYNC_LOG_NAME)))) {
      String line;
      while ((line = br.readLine()) != null) {
        if (line.equals(SyncReceiverLogger.SYNC_DELETED_FILE_NAME_START)) {
          mode = -1;
        } else if (line.equals(SyncReceiverLogger.SYNC_TSFILE_START)) {
          mode = 1;
        } else {
          if (mode == 1) {
            toBeSyncedFilesTest.add(line);
          }
        }
      }
    }
    assertEquals(toBeSyncedFilesTest.size(), toBeSyncedFiles.size());
    assertTrue(toBeSyncedFilesTest.containsAll(toBeSyncedFiles));

    logAnalyze.recover(getReceiverFolderFile().getName());

    try {
      long waitTime = 0;
      while (FileLoaderManager.getInstance()
          .containsFileLoader(getReceiverFolderFile().getName())) {
        Thread.sleep(100);
        waitTime += 100;
        LOGGER.info("Has waited for loading new tsfiles {}ms", waitTime);
      }
    } catch (InterruptedException e) {
      LOGGER.error("Fail to wait for loading new tsfiles", e);
    }

    assertFalse(new File(getReceiverFolderFile(), SyncConstant.LOAD_LOG_NAME).exists());
    assertFalse(new File(getReceiverFolderFile(), SyncConstant.SYNC_LOG_NAME).exists());
  }


  private File getReceiverFolderFile() {
    return new File(dataDir + File.separatorChar + SyncConstant.SYNC_RECEIVER + File.separatorChar
        + "127.0.0.1_5555");
  }

  private File getSnapshotFolder() {
    return new File(getReceiverFolderFile(), SyncConstant.RECEIVER_DATA_FOLDER_NAME);
  }
}