/**
 * 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.storm.blobstore;

import static org.apache.storm.blobstore.BlobStoreAclHandler.ADMIN;
import static org.apache.storm.blobstore.BlobStoreAclHandler.READ;
import static org.apache.storm.blobstore.BlobStoreAclHandler.WRITE;
import static org.apache.storm.daemon.nimbus.Nimbus.NIMBUS_SUBJECT;
import static org.apache.storm.daemon.nimbus.Nimbus.getVersionForKey;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.NoSuchFileException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import javax.security.auth.Subject;
import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
import org.apache.storm.cluster.ClusterStateContext;
import org.apache.storm.cluster.ClusterUtils;
import org.apache.storm.cluster.DaemonType;
import org.apache.storm.cluster.IStormClusterState;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.KeyAlreadyExistsException;
import org.apache.storm.generated.KeyNotFoundException;
import org.apache.storm.generated.ReadableBlobMeta;
import org.apache.storm.generated.SettableBlobMeta;
import org.apache.storm.nimbus.ILeaderElector;
import org.apache.storm.nimbus.NimbusInfo;
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.shade.org.apache.curator.framework.CuratorFramework;
import org.apache.storm.shade.org.apache.zookeeper.KeeperException;
import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Utils;
import org.apache.storm.utils.WrappedKeyAlreadyExistsException;
import org.apache.storm.utils.WrappedKeyNotFoundException;

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

/**
 * Provides a local file system backed blob store implementation for Nimbus.
 *
 * <p>For a local blob store the user and the supervisor use NimbusBlobStore Client API in order to talk to nimbus through thrift.
 * The authentication and authorization here is based on the subject.
 * We currently have NIMBUS_ADMINS and SUPERVISOR_ADMINS configuration. NIMBUS_ADMINS are given READ, WRITE and ADMIN
 * access whereas the SUPERVISOR_ADMINS are given READ access in order to read and download the blobs form the nimbus.
 *
 * <p>The ACLs for the blob store are validated against whether the subject is a NIMBUS_ADMIN, SUPERVISOR_ADMIN or USER
 * who has read, write or admin privileges in order to perform respective operations on the blob.
 *
 * <p>For local blob store
 * 1. The USER interacts with nimbus to upload and access blobs through NimbusBlobStore Client API.
 * 2. The USER sets the ACLs, and the blob access is validated against these ACLs.
 * 3. The SUPERVISOR interacts with nimbus through the NimbusBlobStore Client API to download the blobs.
 * The supervisors principal should match the set of users configured into SUPERVISOR_ADMINS.
 * Here, the PrincipalToLocalPlugin takes care of mapping the principal to user name before the ACL validation.
 */
public class LocalFsBlobStore extends BlobStore {
    public static final Logger LOG = LoggerFactory.getLogger(LocalFsBlobStore.class);
    private static final String DATA_PREFIX = "data_";
    private static final String META_PREFIX = "meta_";
    private static final String BLOBSTORE_SUBTREE = "/blobstore/";
    private final int allPermissions = READ | WRITE | ADMIN;
    protected BlobStoreAclHandler aclHandler;
    private NimbusInfo nimbusInfo;
    private FileBlobStoreImpl fbs;
    private Map<String, Object> conf;
    private CuratorFramework zkClient;
    private IStormClusterState stormClusterState;
    private Timer timer;
    private ILeaderElector leaderElector;

    @Override
    public void prepare(Map<String, Object> conf, String overrideBase, NimbusInfo nimbusInfo, ILeaderElector leaderElector) {
        this.conf = conf;
        this.nimbusInfo = nimbusInfo;
        zkClient = BlobStoreUtils.createZKClient(conf, DaemonType.NIMBUS);
        if (overrideBase == null) {
            overrideBase = ConfigUtils.absoluteStormBlobStoreDir(conf);
        }
        File baseDir = new File(overrideBase, BASE_BLOBS_DIR_NAME);
        try {
            fbs = new FileBlobStoreImpl(baseDir, conf);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        aclHandler = new BlobStoreAclHandler(conf);
        try {
            this.stormClusterState = ClusterUtils.mkStormClusterState(conf, new ClusterStateContext(DaemonType.NIMBUS, conf));
        } catch (Exception e) {
            e.printStackTrace();
        }
        timer = new Timer("BLOB-STORE-TIMER", true);
        this.leaderElector = leaderElector;
    }

    /**
     * Sets up blobstore state for all current keys.
     */
    private void setupBlobstore() throws AuthorizationException, KeyNotFoundException {
        IStormClusterState state = stormClusterState;
        BlobStore store = this;
        Set<String> localKeys = new HashSet<>();
        for (Iterator<String> it = store.listKeys(); it.hasNext();) {
            localKeys.add(it.next());
        }
        Set<String> activeKeys = new HashSet<>(state.activeKeys());
        Set<String> activeLocalKeys = new HashSet<>(localKeys);
        activeLocalKeys.retainAll(activeKeys);
        Set<String> keysToDelete = new HashSet<>(localKeys);
        keysToDelete.removeAll(activeKeys);
        NimbusInfo nimbusInfo = this.nimbusInfo;
        LOG.debug("Deleting keys not on the zookeeper {}", keysToDelete);
        for (String toDelete: keysToDelete) {
            store.deleteBlob(toDelete, NIMBUS_SUBJECT);
        }
        LOG.debug("Creating list of key entries for blobstore inside zookeeper {} local {}", activeKeys, activeLocalKeys);
        for (String key: activeLocalKeys) {
            try {
                state.setupBlob(key, nimbusInfo, getVersionForKey(key, nimbusInfo, zkClient));
            } catch (KeyNotFoundException e) {
                // invalid key, remove it from blobstore
                store.deleteBlob(key, NIMBUS_SUBJECT);
            }
        }
    }


    private void blobSync() throws Exception {
        if ("distributed".equals(conf.get(Config.STORM_CLUSTER_MODE))) {
            if (!this.leaderElector.isLeader()) {
                IStormClusterState state = stormClusterState;
                NimbusInfo nimbusInfo = this.nimbusInfo;
                BlobStore store = this;
                Set<String> allKeys = new HashSet<>();
                for (Iterator<String> it = store.listKeys(); it.hasNext();) {
                    allKeys.add(it.next());
                }
                Set<String> zkKeys = new HashSet<>(state.blobstore(() -> {
                    try {
                        this.blobSync();
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }));
                LOG.debug("blob-sync blob-store-keys {} zookeeper-keys {}", allKeys, zkKeys);
                LocalFsBlobStoreSynchronizer sync = new LocalFsBlobStoreSynchronizer(store, conf);
                sync.setNimbusInfo(nimbusInfo);
                sync.setBlobStoreKeySet(allKeys);
                sync.setZookeeperKeySet(zkKeys);
                sync.setZkClient(zkClient);
                sync.syncBlobs();
            } //else leader (NOOP)
        } //else local (NOOP)
    }


    @Override
    public  void startSyncBlobs() throws KeyNotFoundException, AuthorizationException {
        //register call back for blob-store
        this.stormClusterState.blobstore(() -> {
            try {
                blobSync();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        setupBlobstore();

        //Schedule nimbus code sync thread to sync code from other nimbuses.
        this.timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                try {
                    blobSync();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }, 0, ObjectReader.getInt(conf.get(DaemonConfig.NIMBUS_CODE_SYNC_FREQ_SECS)) * 1000);

    }

    @Override
    public AtomicOutputStream createBlob(String key, SettableBlobMeta meta, Subject who) throws AuthorizationException,
        KeyAlreadyExistsException {
        LOG.debug("Creating Blob for key {}", key);
        validateKey(key);
        aclHandler.normalizeSettableBlobMeta(key, meta, who, allPermissions);
        BlobStoreAclHandler.validateSettableACLs(key, meta.get_acl());
        aclHandler.hasPermissions(meta.get_acl(), allPermissions, who, key);
        if (fbs.exists(DATA_PREFIX + key)) {
            throw new WrappedKeyAlreadyExistsException(key);
        }
        BlobStoreFileOutputStream outputStream = null;
        try {
            outputStream = new BlobStoreFileOutputStream(fbs.write(META_PREFIX + key, true));
            outputStream.write(Utils.thriftSerialize(meta));
            outputStream.close();
            outputStream = null;
            this.stormClusterState.setupBlob(key, this.nimbusInfo, getVersionForKey(key, this.nimbusInfo, zkClient));
            return new BlobStoreFileOutputStream(fbs.write(DATA_PREFIX + key, true));
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (KeyNotFoundException e) {
            throw new RuntimeException(e);
        } finally {
            if (outputStream != null) {
                try {
                    outputStream.cancel();
                } catch (IOException e) {
                    //Ignored
                }
            }
        }
    }

    @Override
    public AtomicOutputStream updateBlob(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
        validateKey(key);
        checkPermission(key, who, WRITE);
        try {
            return new BlobStoreFileOutputStream(fbs.write(DATA_PREFIX + key, false));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private SettableBlobMeta getStoredBlobMeta(String key) throws KeyNotFoundException {
        InputStream in = null;
        try {
            LocalFsBlobStoreFile pf = fbs.read(META_PREFIX + key);
            try {
                in = pf.getInputStream();
            } catch (FileNotFoundException fnf) {
                throw new WrappedKeyNotFoundException(key);
            }
            ByteArrayOutputStream out = new ByteArrayOutputStream();
            byte[] buffer = new byte[2048];
            int len;
            while ((len = in.read(buffer)) > 0) {
                out.write(buffer, 0, len);
            }
            in.close();
            in = null;
            return Utils.thriftDeserialize(SettableBlobMeta.class, out.toByteArray());
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    //Ignored
                }
            }
        }
    }

    @Override
    public ReadableBlobMeta getBlobMeta(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
        validateKey(key);
        if (!checkForBlobOrDownload(key)) {
            checkForBlobUpdate(key);
        }
        SettableBlobMeta meta = getStoredBlobMeta(key);
        aclHandler.validateUserCanReadMeta(meta.get_acl(), who, key);
        ReadableBlobMeta rbm = new ReadableBlobMeta();
        rbm.set_settable(meta);
        try {
            LocalFsBlobStoreFile pf = fbs.read(DATA_PREFIX + key);
            rbm.set_version(pf.getModTime());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return rbm;
    }

    /**
     * Sets leader elector (only used by LocalFsBlobStore to help sync blobs between Nimbi.
     */
    @Override
    public void setLeaderElector(ILeaderElector leaderElector) {
        this.leaderElector = leaderElector;
    }

    @Override
    public void setBlobMeta(String key, SettableBlobMeta meta, Subject who) throws AuthorizationException, KeyNotFoundException {
        validateKey(key);
        checkForBlobOrDownload(key);
        aclHandler.normalizeSettableBlobMeta(key, meta, who, ADMIN);
        BlobStoreAclHandler.validateSettableACLs(key, meta.get_acl());
        SettableBlobMeta orig = getStoredBlobMeta(key);
        aclHandler.hasPermissions(orig.get_acl(), ADMIN, who, key);
        BlobStoreFileOutputStream outputStream = null;
        try {
            outputStream = new BlobStoreFileOutputStream(fbs.write(META_PREFIX + key, false));
            outputStream.write(Utils.thriftSerialize(meta));
            outputStream.close();
            outputStream = null;
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            if (outputStream != null) {
                try {
                    outputStream.cancel();
                } catch (IOException e) {
                    //Ignored
                }
            }
        }
    }

    @Override
    public void deleteBlob(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
        validateKey(key);

        if (!aclHandler.checkForValidUsers(who, WRITE)) {
            // need to get ACL from meta
            LOG.debug("Retrieving meta to get ACL info... key: {} subject: {}", key, who);

            try {
                checkPermission(key, who, WRITE);
            } catch (KeyNotFoundException e) {
                LOG.error("Error while retrieving meta from ZK or local... key: {} subject: {}", key, who);
                throw e;
            }
        } else {
            // able to delete the blob without checking meta's ACL
            // skip checking everything and continue deleting local files
            LOG.debug("Given subject is eligible to delete key without checking ACL, skipping... key: {} subject: {}",
                      key, who);
        }

        try {
            deleteKeyIgnoringFileNotFound(DATA_PREFIX + key);
            deleteKeyIgnoringFileNotFound(META_PREFIX + key);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        this.stormClusterState.removeBlobstoreKey(key);
        this.stormClusterState.removeKeyVersion(key);
    }

    private void checkPermission(String key, Subject who, int mask) throws KeyNotFoundException, AuthorizationException {
        checkForBlobOrDownload(key);
        SettableBlobMeta meta = getStoredBlobMeta(key);
        aclHandler.hasPermissions(meta.get_acl(), mask, who, key);
    }

    private void deleteKeyIgnoringFileNotFound(String key) throws IOException {
        try {
            fbs.deleteKey(key);
        } catch (IOException e) {
            if (e instanceof FileNotFoundException || e instanceof NoSuchFileException) {
                LOG.debug("Ignoring FileNotFoundException since we're about to delete such key... key: {}", key);
            } else {
                throw e;
            }
        }
    }

    @Override
    public InputStreamWithMeta getBlob(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
        validateKey(key);
        if (!checkForBlobOrDownload(key)) {
            checkForBlobUpdate(key);
        }
        SettableBlobMeta meta = getStoredBlobMeta(key);
        aclHandler.hasPermissions(meta.get_acl(), READ, who, key);
        try {
            return new BlobStoreFileInputStream(fbs.read(DATA_PREFIX + key));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public Iterator<String> listKeys() {
        try {
            return new KeyTranslationIterator(fbs.listKeys(), DATA_PREFIX);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void shutdown() {
        if (zkClient != null) {
            zkClient.close();
        }
        if (timer != null) {
            timer.cancel();;
        }
        stormClusterState.disconnect();
    }

    @Override
    public int getBlobReplication(String key, Subject who) throws Exception {
        int replicationCount = 0;
        validateKey(key);
        SettableBlobMeta meta = getStoredBlobMeta(key);
        aclHandler.hasPermissions(meta.get_acl(), READ, who, key);
        if (zkClient.checkExists().forPath(BLOBSTORE_SUBTREE + key) == null) {
            return 0;
        }
        try {
            replicationCount = zkClient.getChildren().forPath(BLOBSTORE_SUBTREE + key).size();
        } catch (KeeperException.NoNodeException e) {
            //Race with delete
            //If it is not here the replication is 0 
        }
        return replicationCount;
    }

    @Override
    public int updateBlobReplication(String key, int replication, Subject who) throws AuthorizationException, KeyNotFoundException {
        throw new UnsupportedOperationException("For local file system blob store the update blobs function does not work. "
                                                + "Please use HDFS blob store to make this feature available.");
    }

    //This additional check and download is for nimbus high availability in case you have more than one nimbus
    public synchronized boolean checkForBlobOrDownload(String key) throws KeyNotFoundException {
        boolean checkBlobDownload = false;
        try {
            List<String> keyList = BlobStoreUtils.getKeyListFromBlobStore(this);
            if (!keyList.contains(key)) {
                if (zkClient.checkExists().forPath(BLOBSTORE_SUBTREE + key) != null) {
                    Set<NimbusInfo> nimbusSet = BlobStoreUtils.getNimbodesWithLatestSequenceNumberOfBlob(zkClient, key);
                    nimbusSet.remove(this.nimbusInfo);
                    if (BlobStoreUtils.downloadMissingBlob(conf, this, key, nimbusSet)) {
                        LOG.debug("Updating blobs state");
                        BlobStoreUtils.createStateInZookeeper(conf, key, nimbusInfo);
                        checkBlobDownload = true;
                    }
                }
            }
        } catch (KeyNotFoundException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return checkBlobDownload;
    }

    public synchronized void checkForBlobUpdate(String key) {
        BlobStoreUtils.updateKeyForBlobStore(conf, this, zkClient, key, nimbusInfo);
    }

    public void fullCleanup(long age) throws IOException {
        fbs.fullCleanup(age);
    }

    @VisibleForTesting
    File getKeyDataDir(String key) {
        return fbs.getKeyDir(DATA_PREFIX + key);
    }
}