/*
 * 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.carbondata.core.locks;

import java.util.Collections;
import java.util.List;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.path.CarbonTablePath;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

/**
 * For Handling the zookeeper locking implementation
 */
public class ZooKeeperLocking extends AbstractCarbonLock {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(ZooKeeperLocking.class.getName());

  /**
   * zk is the zookeeper client instance
   */
  private static ZooKeeper zk;

  /**
   * zooKeeperLocation is the location in the zoo keeper file system where the locks will be
   * maintained.
   */
  private static final String zooKeeperLocation = CarbonCommonConstants.ZOOKEEPER_LOCATION;

  /**
   * Unique folder for each table with DatabaseName_TableName
   */
  private final String tableIdFolder;

  /**
   * lockName is the name of the lock to use. This name should be same for every process that want
   * to share the same lock
   */
  private String lockName;

  /**
   * lockPath is the unique path created for the each instance of the carbon lock.
   */
  private String lockPath;

  private String lockTypeFolder;

  public ZooKeeperLocking(AbsoluteTableIdentifier absoluteTableIdentifier, String lockFile) {
    this(absoluteTableIdentifier.getCarbonTableIdentifier().getDatabaseName()
        + CarbonCommonConstants.FILE_SEPARATOR + absoluteTableIdentifier.getCarbonTableIdentifier()
        .getTableName(), lockFile);
  }

  public static void initialize() {
    String zooKeeperUrl =
        CarbonProperties.getInstance().getProperty(CarbonCommonConstants.ZOOKEEPER_URL);
    if (null == zk) {
      zk = ZookeeperInit.getInstance(zooKeeperUrl).getZookeeper();
    }
  }

  /**
   * @param lockLocation
   * @param lockFile
   */
  public ZooKeeperLocking(String lockLocation, String lockFile) {
    this.lockName = lockFile;
    this.tableIdFolder = zooKeeperLocation + CarbonCommonConstants.FILE_SEPARATOR
        + CarbonTablePath.getLockFilesDirPath(lockLocation);

    initialize();

    this.lockTypeFolder = tableIdFolder + CarbonCommonConstants.FILE_SEPARATOR + lockFile;
    try {
      createBaseNode();
      // if exists returns null then path doesnt exist. so creating.
      if (null == zk.exists(this.tableIdFolder, true)) {
        createRecursively(this.tableIdFolder);
      }
      // if exists returns null then path doesnt exist. so creating.
      if (null == zk.exists(this.lockTypeFolder, true)) {
        zk.create(this.lockTypeFolder, new byte[1], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      }
    } catch (KeeperException | InterruptedException e) {
      LOGGER.error(e.getMessage(), e);
    }
    initRetry();
  }

  /**
   * Creating a znode in which all the znodes (lock files )are maintained.
   */
  private void createBaseNode() throws KeeperException, InterruptedException {
    if (null == zk.exists(zooKeeperLocation, true)) {
      // creating a znode in which all the znodes (lock files )are maintained.
      zk.create(zooKeeperLocation, new byte[1], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }
  }

  /**
   * Create zookeeper node if not exist
   * @param path
   * @throws KeeperException
   * @throws InterruptedException
   */
  private void createRecursively(String path) throws KeeperException, InterruptedException {
    if (zk.exists(path, true) == null && path.length() > 0) {
      String temp = path.substring(0, path.lastIndexOf(CarbonCommonConstants.FILE_SEPARATOR));
      createRecursively(temp);
      zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }
  }

  /**
   * Handling of the locking mechanism using zoo keeper.
   */
  @Override
  public boolean lock() {
    try {
      // create the lock file with lockName.
      lockPath =
          zk.create(this.lockTypeFolder + CarbonCommonConstants.FILE_SEPARATOR + lockName, null,
              Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

      // get the children present in zooKeeperLocation.
      List<String> nodes = zk.getChildren(this.lockTypeFolder, null);

      // sort the children
      Collections.sort(nodes);

      // here the logic is , for each lock request zookeeper will create a file ending with
      // incremental digits.
      // so first request will be 00001 next is 00002 and so on.
      // if the current request is 00002 and already one previous request(00001) is present then get
      // children will give both nodes.
      // after the sort we are checking if the lock path is first or not .if it is first then lock
      // has been acquired.

      if (lockPath.endsWith(nodes.get(0))) {
        return true;
      } else {
        // if locking failed then deleting the created lock as next time again new lock file will be
        // created.
        zk.delete(lockPath, -1);
        return false;
      }
    } catch (KeeperException | InterruptedException e) {
      LOGGER.error(e.getMessage(), e);
      return false;
    }
  }

  /**
   * @return status where lock file is unlocked or not.
   */
  @Override
  public boolean unlock() {
    try {
      // exists will return null if the path doesn't exists.
      if (lockPath != null && null != zk.exists(lockPath, true)) {
        zk.delete(lockPath, -1);
        lockPath = null;
      }
    } catch (KeeperException | InterruptedException e) {
      LOGGER.error(e.getMessage(), e);
      return false;
    }
    return true;
  }

}