/**
 * 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.smartdata.metastore;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.smartdata.metaservice.BackupMetaService;
import org.smartdata.metaservice.CmdletMetaService;
import org.smartdata.metaservice.CopyMetaService;
import org.smartdata.metastore.dao.AccessCountDao;
import org.smartdata.metastore.dao.AccessCountTable;
import org.smartdata.metastore.dao.ActionDao;
import org.smartdata.metastore.dao.BackUpInfoDao;
import org.smartdata.metastore.dao.CacheFileDao;
import org.smartdata.metastore.dao.ClusterConfigDao;
import org.smartdata.metastore.dao.ClusterInfoDao;
import org.smartdata.metastore.dao.CmdletDao;
import org.smartdata.metastore.dao.CompressionFileDao;
import org.smartdata.metastore.dao.DataNodeInfoDao;
import org.smartdata.metastore.dao.DataNodeStorageInfoDao;
import org.smartdata.metastore.dao.ErasureCodingPolicyDao;
import org.smartdata.metastore.dao.FileDiffDao;
import org.smartdata.metastore.dao.FileInfoDao;
import org.smartdata.metastore.dao.FileStateDao;
import org.smartdata.metastore.dao.GeneralDao;
import org.smartdata.metastore.dao.GlobalConfigDao;
import org.smartdata.metastore.dao.MetaStoreHelper;
import org.smartdata.metastore.dao.RuleDao;
import org.smartdata.metastore.dao.SmallFileDao;
import org.smartdata.metastore.dao.StorageDao;
import org.smartdata.metastore.dao.StorageHistoryDao;
import org.smartdata.metastore.dao.SystemInfoDao;
import org.smartdata.metastore.dao.UserInfoDao;
import org.smartdata.metastore.dao.WhitelistDao;
import org.smartdata.metastore.dao.XattrDao;
import org.smartdata.metastore.utils.MetaStoreUtils;
import org.smartdata.metrics.FileAccessEvent;
import org.smartdata.model.ActionInfo;
import org.smartdata.model.BackUpInfo;
import org.smartdata.model.CachedFileStatus;
import org.smartdata.model.ClusterConfig;
import org.smartdata.model.ClusterInfo;
import org.smartdata.model.CmdletInfo;
import org.smartdata.model.CmdletState;
import org.smartdata.model.CompactFileState;
import org.smartdata.model.CompressionFileState;
import org.smartdata.model.DataNodeInfo;
import org.smartdata.model.DataNodeStorageInfo;
import org.smartdata.model.DetailedFileAction;
import org.smartdata.model.DetailedRuleInfo;
import org.smartdata.model.ErasureCodingPolicyInfo;
import org.smartdata.model.FileAccessInfo;
import org.smartdata.model.FileDiff;
import org.smartdata.model.FileDiffState;
import org.smartdata.model.FileInfo;
import org.smartdata.model.FileState;
import org.smartdata.model.GlobalConfig;
import org.smartdata.model.NormalFileState;
import org.smartdata.model.RuleInfo;
import org.smartdata.model.RuleState;
import org.smartdata.model.S3FileState;
import org.smartdata.model.StorageCapacity;
import org.smartdata.model.StoragePolicy;
import org.smartdata.model.SystemInfo;
import org.smartdata.model.UserInfo;
import org.smartdata.model.XAttribute;

import org.springframework.dao.EmptyResultDataAccessException;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Operations supported for upper functions.
 */
public class MetaStore implements CopyMetaService, CmdletMetaService, BackupMetaService {
  static final Logger LOG = LoggerFactory.getLogger(MetaStore.class);

  private DBPool pool = null;
  private DBType dbType;

  private Map<Integer, String> mapStoragePolicyIdName = null;
  private Map<String, Integer> mapStoragePolicyNameId = null;
  private Map<String, StorageCapacity> mapStorageCapacity = null;
  private Set<String> setBackSrc = null;
  private RuleDao ruleDao;
  private CmdletDao cmdletDao;
  private ActionDao actionDao;
  private FileInfoDao fileInfoDao;
  private CacheFileDao cacheFileDao;
  private StorageDao storageDao;
  private StorageHistoryDao storageHistoryDao;
  private XattrDao xattrDao;
  private FileDiffDao fileDiffDao;
  private AccessCountDao accessCountDao;
  private MetaStoreHelper metaStoreHelper;
  private ClusterConfigDao clusterConfigDao;
  private GlobalConfigDao globalConfigDao;
  private DataNodeInfoDao dataNodeInfoDao;
  private DataNodeStorageInfoDao dataNodeStorageInfoDao;
  private BackUpInfoDao backUpInfoDao;
  private ClusterInfoDao clusterInfoDao;
  private SystemInfoDao systemInfoDao;
  private UserInfoDao userInfoDao;
  private FileStateDao fileStateDao;
  private CompressionFileDao compressionFileDao;
  private GeneralDao generalDao;
  private SmallFileDao smallFileDao;
  private ErasureCodingPolicyDao ecDao;
  private WhitelistDao whitelistDao;
  private final ReentrantLock accessCountLock;

  public MetaStore(DBPool pool) throws MetaStoreException {
    this.pool = pool;
    initDbInfo();
    ruleDao = new RuleDao(pool.getDataSource());
    cmdletDao = new CmdletDao(pool.getDataSource());
    actionDao = new ActionDao(pool.getDataSource());
    fileInfoDao = new FileInfoDao(pool.getDataSource());
    xattrDao = new XattrDao(pool.getDataSource());
    cacheFileDao = new CacheFileDao(pool.getDataSource());
    storageDao = new StorageDao(pool.getDataSource());
    storageHistoryDao = new StorageHistoryDao(pool.getDataSource());
    accessCountDao = new AccessCountDao(pool.getDataSource());
    fileDiffDao = new FileDiffDao(pool.getDataSource());
    metaStoreHelper = new MetaStoreHelper(pool.getDataSource());
    clusterConfigDao = new ClusterConfigDao(pool.getDataSource());
    globalConfigDao = new GlobalConfigDao(pool.getDataSource());
    dataNodeInfoDao = new DataNodeInfoDao(pool.getDataSource());
    dataNodeStorageInfoDao = new DataNodeStorageInfoDao(pool.getDataSource());
    backUpInfoDao = new BackUpInfoDao(pool.getDataSource());
    clusterInfoDao = new ClusterInfoDao(pool.getDataSource());
    systemInfoDao = new SystemInfoDao(pool.getDataSource());
    userInfoDao = new UserInfoDao(pool.getDataSource());
    fileStateDao = new FileStateDao(pool.getDataSource());
    compressionFileDao = new CompressionFileDao(pool.getDataSource());
    generalDao = new GeneralDao(pool.getDataSource());
    smallFileDao = new SmallFileDao(pool.getDataSource());
    ecDao = new ErasureCodingPolicyDao(pool.getDataSource());
    whitelistDao = new WhitelistDao(pool.getDataSource());
    accessCountLock = new ReentrantLock();
  }

  private void initDbInfo() throws MetaStoreException {
    Connection conn = null;
    try {
      try {
        conn = getConnection();
        String driver = conn.getMetaData().getDriverName();
        driver = driver.toLowerCase();
        if (driver.contains("sqlite")) {
          dbType = DBType.SQLITE;
        } else if (driver.contains("mysql")) {
          dbType = DBType.MYSQL;
        } else {
          throw new MetaStoreException("Unknown database: " + driver);
        }
      } finally {
        if (conn != null) {
          closeConnection(conn);
        }
      }
    } catch (SQLException e) {
      throw new MetaStoreException(e);
    }
  }

  public Connection getConnection() throws MetaStoreException {
    if (pool != null) {
      try {
        return pool.getConnection();
      } catch (SQLException e) {
        throw new MetaStoreException(e);
      }
    }
    return null;
  }

  private void closeConnection(Connection conn) throws MetaStoreException {
    if (pool != null) {
      try {
        pool.closeConnection(conn);
      } catch (SQLException e) {
        throw new MetaStoreException(e);
      }
    }
  }

  public DBType getDbType() {
    return dbType;
  }

  public Long queryForLong(String sql) throws MetaStoreException {
    try {
      return generalDao.queryForLong(sql);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }


  /**
   * Store a single file info into database.
   *
   * @param file
   */
  public void insertFile(FileInfo file)
    throws MetaStoreException {
    updateCache();
    fileInfoDao.insert(file);
  }


  /**
   * Store files info into database.
   *
   * @param files
   */
  public void insertFiles(FileInfo[] files)
    throws MetaStoreException {
    updateCache();
    fileInfoDao.insert(files);
  }

  public int updateFileStoragePolicy(String path, String policyName)
    throws MetaStoreException {
    if (mapStoragePolicyIdName == null) {
      updateCache();
    }
    if (!mapStoragePolicyNameId.containsKey(policyName)) {
      throw new MetaStoreException("Unknown storage policy name '"
        + policyName + "'");
    }
    try {
      return storageDao.updateFileStoragePolicy(path, mapStoragePolicyNameId.get(policyName));
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public FileInfo getFile(long fid) throws MetaStoreException {
    updateCache();
    try {
      return fileInfoDao.getById(fid);
    } catch (EmptyResultDataAccessException e) {
      return null;
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public FileInfo getFile(String path) throws MetaStoreException {
    updateCache();
    try {
      return fileInfoDao.getByPath(path);
    } catch (EmptyResultDataAccessException e) {
      return null;
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<FileInfo> getFile() throws MetaStoreException {
    updateCache();
    try {
      return fileInfoDao.getAll();
    } catch (EmptyResultDataAccessException e) {
      return new ArrayList<>();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<FileInfo> getFilesByPrefix(String path) throws MetaStoreException {
    updateCache();
    try {
      return fileInfoDao.getFilesByPrefix(path);
    } catch (EmptyResultDataAccessException e) {
      return new ArrayList<>();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<FileInfo> getFilesByPrefixInOrder(String path) throws MetaStoreException {
    updateCache();
    try {
      return fileInfoDao.getFilesByPrefixInOrder(path);
    } catch (EmptyResultDataAccessException e) {
      return new ArrayList<>();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<FileInfo> getFilesByPaths(Collection<String> paths)
    throws MetaStoreException {
    try {
      return fileInfoDao.getFilesByPaths(paths);
    } catch (EmptyResultDataAccessException e) {
      return new ArrayList<>();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public Map<String, Long> getFileIDs(Collection<String> paths)
    throws MetaStoreException {
    try {
      return fileInfoDao.getPathFids(paths);
    } catch (EmptyResultDataAccessException e) {
      return new HashMap<>();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public Map<Long, String> getFilePaths(Collection<Long> ids)
    throws MetaStoreException {
    try {
      return fileInfoDao.getFidPaths(ids);
    } catch (EmptyResultDataAccessException e) {
      return new HashMap<>();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<FileAccessInfo> getHotFiles(
    List<AccessCountTable> tables,
    int topNum) throws MetaStoreException {
    Iterator<AccessCountTable> tableIterator = tables.iterator();
    if (tableIterator.hasNext()) {
      try {
        Map<Long, Integer> accessCounts =
          accessCountDao.getHotFiles(tables, topNum);
        if (accessCounts.size() == 0) {
          return new ArrayList<>();
        }
        Map<Long, String> idToPath = getFilePaths(accessCounts.keySet());
        List<FileAccessInfo> result = new ArrayList<>();
        for (Map.Entry<Long, Integer> entry : accessCounts.entrySet()) {
          Long fid = entry.getKey();
          if (idToPath.containsKey(fid) && entry.getValue() > 0) {
            result.add(
              new FileAccessInfo(fid, idToPath.get(fid), entry.getValue()));
          }
        }
        return result;
      } catch (EmptyResultDataAccessException e) {
        return new ArrayList<>();
      } catch (Exception e) {
        throw new MetaStoreException(e);
      } finally {
        for (AccessCountTable accessCountTable : tables) {
          if (accessCountTable.isEphemeral()) {
            this.dropTable(accessCountTable.getTableName());
          }
        }
      }
    } else {
      return new ArrayList<>();
    }
  }

  public ReentrantLock getAccessCountLock() {
    return accessCountLock;
  }

  /**
   * @param fidSrc the fid of old file.
   * @param fidDest the fid of new file that will take over the access
   *                count of old file.
   * @throws MetaStoreException
   */
  public void updateAccessCountTableFid(long fidSrc, long fidDest)
      throws MetaStoreException {
    if (fidSrc == fidDest) {
      LOG.warn("No need to update fid for access count table "
          + "with same fid: " + fidDest);
      return;
    }
    accessCountLock.lock();
    try {
      accessCountDao.updateFid(fidSrc, fidDest);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    } finally {
      accessCountLock.unlock();
    }
  }

  public void deleteAllFileInfo() throws MetaStoreException {
    try {
      fileInfoDao.deleteAll();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void deleteAllEcPolicies() throws MetaStoreException {
    try {
      ecDao.deleteAll();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void insertEcPolicies(List<ErasureCodingPolicyInfo> ecInfos) throws MetaStoreException {
    try {
      ecDao.insert(ecInfos);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<ErasureCodingPolicyInfo> getAllEcPolicies() throws MetaStoreException {
    try {
      return ecDao.getAllEcPolicies();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void deleteFileByPath(String path) throws MetaStoreException {
    try {
      fileInfoDao.deleteByPath(path);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<AccessCountTable> getAllSortedTables() throws MetaStoreException {
    try {
      return accessCountDao.getAllSortedTables();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void deleteAccessCountTable(
    AccessCountTable table) throws MetaStoreException {
    accessCountLock.lock();
    try {
      accessCountDao.delete(table);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    } finally {
      accessCountLock.unlock();
    }
  }

  public void insertAccessCountTable(
    AccessCountTable accessCountTable) throws MetaStoreException {
    try {
      if (accessCountDao.getAccessCountTableByName(accessCountTable.getTableName()).isEmpty()) {
        accessCountDao.insert(accessCountTable);
      }
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void insertUpdateStoragesTable(StorageCapacity[] storages)
    throws MetaStoreException {
    mapStorageCapacity = null;
    try {
      storageDao.insertUpdateStoragesTable(storages);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void insertUpdateStoragesTable(List<StorageCapacity> storages)
    throws MetaStoreException {
    mapStorageCapacity = null;
    try {
      storageDao.insertUpdateStoragesTable(
        storages.toArray(new StorageCapacity[storages.size()]));
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void insertUpdateStoragesTable(StorageCapacity storage)
    throws MetaStoreException {
    insertUpdateStoragesTable(new StorageCapacity[]{storage});
  }

  public Map<String, StorageCapacity> getStorageCapacity() throws MetaStoreException {
    updateCache();

    Map<String, StorageCapacity> ret = new HashMap<>();
    Map<String, StorageCapacity> currentMapStorageCapacity = mapStorageCapacity;
    if (currentMapStorageCapacity != null) {
      for (String key : currentMapStorageCapacity.keySet()) {
        ret.put(key, currentMapStorageCapacity.get(key));
      }
    }
    return ret;
  }

  public void deleteStorage(String storageType) throws MetaStoreException {
    try {
      mapStorageCapacity = null;
      storageDao.deleteStorage(storageType);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public StorageCapacity getStorageCapacity(
    String type) throws MetaStoreException {
    updateCache();
    Map<String, StorageCapacity> currentMapStorageCapacity = mapStorageCapacity;
    while (currentMapStorageCapacity == null) {
      try {
        Thread.sleep(100);
      } catch (InterruptedException ex) {
        LOG.error(ex.getMessage());
      }
      currentMapStorageCapacity = mapStorageCapacity;
    }
    try {
      return currentMapStorageCapacity.get(type);
    } catch (EmptyResultDataAccessException e) {
      return null;
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public boolean updateStoragesTable(String type,
      Long capacity, Long free) throws MetaStoreException {
    try {
      mapStorageCapacity = null;
      return storageDao.updateStoragesTable(type, capacity, free);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void insertStorageHistTable(StorageCapacity[] storages, long interval)
    throws MetaStoreException {
    try {
      storageHistoryDao.insertStorageHistTable(storages, interval);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<StorageCapacity> getStorageHistoryData(String type, long interval,
      long startTime, long endTime) {
    return storageHistoryDao.getStorageHistoryData(type, interval, startTime, endTime);
  }

  public void deleteStorageHistoryOldRecords(String type, long interval, long beforTimeStamp)
    throws MetaStoreException {
    try {
      storageHistoryDao.deleteOldRecords(type, interval, beforTimeStamp);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  private void updateCache() throws MetaStoreException {
    if (mapStoragePolicyIdName == null) {
      mapStoragePolicyNameId = null;
      try {
        mapStoragePolicyIdName = storageDao.getStoragePolicyIdNameMap();
      } catch (Exception e) {
        throw new MetaStoreException(e);
      }
      mapStoragePolicyNameId = new HashMap<>();
      for (Integer key : mapStoragePolicyIdName.keySet()) {
        mapStoragePolicyNameId.put(mapStoragePolicyIdName.get(key), key);
      }
    }

    if (mapStorageCapacity == null) {
      try {
        mapStorageCapacity = storageDao.getStorageTablesItem();
      } catch (Exception e) {
        throw new MetaStoreException(e);
      }
    }
  }

  public void insertCachedFiles(long fid, String path,
      long fromTime,
      long lastAccessTime, int numAccessed) throws MetaStoreException {
    try {
      cacheFileDao.insert(fid, path, fromTime, lastAccessTime, numAccessed);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void insertCachedFiles(List<CachedFileStatus> s)
    throws MetaStoreException {
    try {
      cacheFileDao.insert(s.toArray(new CachedFileStatus[s.size()]));
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void deleteAllCachedFile() throws MetaStoreException {
    try {
      cacheFileDao.deleteAll();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public boolean updateCachedFiles(Long fid,
      Long lastAccessTime,
      Integer numAccessed) throws MetaStoreException {
    try {
      return cacheFileDao.update(fid, lastAccessTime, numAccessed) >= 0;
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void updateCachedFiles(Map<String, Long> pathToIds,
      List<FileAccessEvent> events)
    throws MetaStoreException {
    try {
      cacheFileDao.update(pathToIds, events);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void deleteCachedFile(long fid) throws MetaStoreException {
    try {
      cacheFileDao.deleteById(fid);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<CachedFileStatus> getCachedFileStatus() throws MetaStoreException {
    try {
      return cacheFileDao.getAll();
    } catch (EmptyResultDataAccessException e) {
      return new ArrayList<>();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<Long> getCachedFids() throws MetaStoreException {
    try {
      return cacheFileDao.getFids();
    } catch (EmptyResultDataAccessException e) {
      return new ArrayList<>();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public CachedFileStatus getCachedFileStatus(
    long fid) throws MetaStoreException {
    try {
      return cacheFileDao.getById(fid);
    } catch (EmptyResultDataAccessException e) {
      return null;
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void createProportionTable(AccessCountTable dest,
      AccessCountTable source)
    throws MetaStoreException {
    try {
      accessCountDao.createProportionTable(dest, source);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void dropTable(String tableName) throws MetaStoreException {
    try {
      LOG.debug("Drop table = {}", tableName);
      metaStoreHelper.dropTable(tableName);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public void execute(String sql) throws MetaStoreException {
    try {
      LOG.debug("Execute sql = {}", sql);
      metaStoreHelper.execute(sql);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  //Todo: optimize
  public void execute(List<String> statements) throws MetaStoreException {
    for (String statement : statements) {
      execute(statement);
    }
  }

  public List<String> executeFilesPathQuery(
    String sql) throws MetaStoreException {
    try {
      LOG.debug("ExecuteFilesPathQuery sql = {}", sql);
      return metaStoreHelper.getFilesPath(sql);
    } catch (EmptyResultDataAccessException e) {
      return new ArrayList<>();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<DetailedFileAction> listFileActions(long rid,
      int size) throws MetaStoreException {
    if (mapStoragePolicyIdName == null) {
      updateCache();
    }
    List<ActionInfo> actionInfos = getActions(rid, size);
    List<DetailedFileAction> detailedFileActions = new ArrayList<>();

    for (ActionInfo actionInfo : actionInfos) {
      DetailedFileAction detailedFileAction = new DetailedFileAction(actionInfo);
      String filePath = actionInfo.getArgs().get("-file");
      FileInfo fileInfo = getFile(filePath);
      if (fileInfo == null) {
        // LOG.debug("Namespace is not sync! File {} not in file table!", filePath);
        // Add a mock fileInfo
        fileInfo = new FileInfo(filePath, 0L, 0L, false,
          (short) 0, 0L, 0L, 0L, (short) 0,
          "root", "root", (byte) 0, (byte) 0);
      }
      detailedFileAction.setFileLength(fileInfo.getLength());
      detailedFileAction.setFilePath(filePath);
      if (actionInfo.getActionName().contains("allssd")
        || actionInfo.getActionName().contains("onessd")
        || actionInfo.getActionName().contains("archive")
        || actionInfo.getActionName().contains("alldisk")
        || actionInfo.getActionName().contains("onedisk")
        || actionInfo.getActionName().contains("ramdisk")) {
        detailedFileAction.setTarget(actionInfo.getActionName());
        detailedFileAction.setSrc(mapStoragePolicyIdName.get((int) fileInfo.getStoragePolicy()));
      } else {
        detailedFileAction.setSrc(actionInfo.getArgs().get("-src"));
        detailedFileAction.setTarget(actionInfo.getArgs().get("-dest"));
      }
      detailedFileActions.add(detailedFileAction);
    }
    return detailedFileActions;
  }

  public List<DetailedFileAction> listFileActions(long rid, long start, long offset)
    throws MetaStoreException {
    if (mapStoragePolicyIdName == null) {
      updateCache();
    }
    List<ActionInfo> actionInfos = getActions(rid, start, offset);
    List<DetailedFileAction> detailedFileActions = new ArrayList<>();
    for (ActionInfo actionInfo : actionInfos) {
      DetailedFileAction detailedFileAction = new DetailedFileAction(actionInfo);
      String filePath = actionInfo.getArgs().get("-file");
      FileInfo fileInfo = getFile(filePath);
      if (fileInfo == null) {
        // LOG.debug("Namespace is not sync! File {} not in file table!", filePath);
        // Add a mock fileInfo
        fileInfo = new FileInfo(filePath, 0L, 0L, false,
          (short) 0, 0L, 0L, 0L, (short) 0,
          "root", "root", (byte) 0, (byte) 0);
      }
      detailedFileAction.setFileLength(fileInfo.getLength());
      detailedFileAction.setFilePath(filePath);
      if (actionInfo.getActionName().contains("allssd")
        || actionInfo.getActionName().contains("onessd")
        || actionInfo.getActionName().contains("archive")
        || actionInfo.getActionName().contains("alldisk")
        || actionInfo.getActionName().contains("onedisk")
        || actionInfo.getActionName().contains("ramdisk")) {
        detailedFileAction.setTarget(actionInfo.getActionName());
        detailedFileAction.setSrc(mapStoragePolicyIdName.get((int) fileInfo.getStoragePolicy()));
      } else {
        detailedFileAction.setSrc(actionInfo.getArgs().get("-src"));
        detailedFileAction.setTarget(actionInfo.getArgs().get("-dest"));
      }
      detailedFileActions.add(detailedFileAction);
    }
    return detailedFileActions;
  }

  public long getNumFileAction(long rid) throws MetaStoreException {
    return listFileActions(rid, 0).size();
  }

  public List<DetailedRuleInfo> listMoveRules() throws MetaStoreException {
    List<RuleInfo> ruleInfos = getRuleInfo();
    List<DetailedRuleInfo> detailedRuleInfos = new ArrayList<>();
    for (RuleInfo ruleInfo : ruleInfos) {
      int lastIndex = ruleInfo.getRuleText().lastIndexOf("|");
      String lastPart = ruleInfo.getRuleText().substring(lastIndex + 1);
      if (lastPart.contains("sync")) {
        continue;
      } else if (lastPart.contains("allssd")
              || lastPart.contains("onessd")
              || lastPart.contains("archive")
              || lastPart.contains("alldisk")
              || lastPart.contains("onedisk")
              || lastPart.contains("ramdisk")) {
        DetailedRuleInfo detailedRuleInfo = new DetailedRuleInfo(ruleInfo);
        // Add mover progress
        List<CmdletInfo> cmdletInfos = cmdletDao.getByRid(ruleInfo.getId());
        int currPos = 0;
        for (CmdletInfo cmdletInfo : cmdletInfos) {
          if (cmdletInfo.getState().getValue() <= 4) {
            break;
          }
          currPos += 1;
        }
        int countRunning = 0;
        for (CmdletInfo cmdletInfo : cmdletInfos) {
          if (cmdletInfo.getState().getValue() <= 4) {
            countRunning++;
          }
        }
        detailedRuleInfo
          .setBaseProgress(cmdletInfos.size() - currPos);
        detailedRuleInfo.setRunningProgress(countRunning);
        if (detailedRuleInfo.getState() != RuleState.DELETED){
          detailedRuleInfos.add(detailedRuleInfo);
        }
      }
    }
    return detailedRuleInfos;
  }


  public List<DetailedRuleInfo> listSyncRules() throws MetaStoreException {
    List<RuleInfo> ruleInfos = getRuleInfo();
    List<DetailedRuleInfo> detailedRuleInfos = new ArrayList<>();
    for (RuleInfo ruleInfo : ruleInfos) {
      if (ruleInfo.getState() == RuleState.DELETED) {
        continue;
      }
      int lastIndex = ruleInfo.getRuleText().lastIndexOf("|");
      String lastPart = ruleInfo.getRuleText().substring(lastIndex + 1);
      if (lastPart.contains("sync")) {
        DetailedRuleInfo detailedRuleInfo = new DetailedRuleInfo(ruleInfo);
        // Add sync progress
        BackUpInfo backUpInfo = getBackUpInfo(ruleInfo.getId());
        // Get total matched files
        if (backUpInfo != null) {
          detailedRuleInfo
              .setBaseProgress(getFilesByPrefix(backUpInfo.getSrc()).size());
          long count = fileDiffDao.getPendingDiff(backUpInfo.getSrc()).size();
          count += fileDiffDao.getByState(backUpInfo.getSrc(), FileDiffState.RUNNING).size();
          if (count > detailedRuleInfo.baseProgress) {
            count = detailedRuleInfo.baseProgress;
          }
          detailedRuleInfo.setRunningProgress(count);
        } else {
          detailedRuleInfo
            .setBaseProgress(0);
          detailedRuleInfo.setRunningProgress(0);
        }
        if (detailedRuleInfo.getState() != RuleState.DELETED){
          detailedRuleInfos.add(detailedRuleInfo);
        }
      }
    }
    return detailedRuleInfos;
  }

  public boolean insertNewRule(RuleInfo info)
    throws MetaStoreException {
    try {
      return ruleDao.insert(info) >= 0;
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public boolean updateRuleInfo(long ruleId, RuleState rs,
      long lastCheckTime, long checkedCount, int commandsGen)
    throws MetaStoreException {
    try {
      if (rs == null) {
        return ruleDao.update(ruleId,
          lastCheckTime, checkedCount, commandsGen) >= 0;
      }
      return ruleDao.update(ruleId,
        rs.getValue(), lastCheckTime, checkedCount, commandsGen) >= 0;
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public boolean updateRuleState(long ruleId, RuleState rs)
    throws MetaStoreException {
    if (rs == null) {
      throw new MetaStoreException("Rule state can not be null, ruleId = " + ruleId);
    }
    try {
      return ruleDao.update(ruleId, rs.getValue()) >= 0;
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public RuleInfo getRuleInfo(long ruleId) throws MetaStoreException {
    try {
      return ruleDao.getById(ruleId);
    } catch (EmptyResultDataAccessException e) {
      return null;
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<RuleInfo> listPageRule(long start, long offset, List<String> orderBy,
      List<Boolean> desc)
    throws MetaStoreException {
    LOG.debug("List Rule, start {}, offset {}", start, offset);
    try {
      if (orderBy.size() == 0) {
        return ruleDao.getAPageOfRule(start, offset);
      } else {
        return ruleDao.getAPageOfRule(start, offset, orderBy, desc);
      }
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }


  public List<RuleInfo> getRuleInfo() throws MetaStoreException {
    try {
      return ruleDao.getAll();
    } catch (EmptyResultDataAccessException e) {
      return new ArrayList<>();
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public List<CmdletInfo> listPageCmdlets(long rid, long start, long offset,
      List<String> orderBy, List<Boolean> desc)
    throws MetaStoreException {
    LOG.debug("List cmdlet, start {}, offset {}", start, offset);
    try {
      if (orderBy.size() == 0) {
        return cmdletDao.getByRid(rid, start, offset);
      } else {
        return cmdletDao.getByRid(rid, start, offset, orderBy, desc);
      }
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public long getNumCmdletsByRid(long rid) {
    try {
      return cmdletDao.getNumByRid(rid);
    } catch (Exception e) {
      return 0;
    }
  }

  public List<CmdletInfo> listPageCmdlets(long start, long offset,
