package org.apache.commons.jcs3.auxiliary.disk.block;

/*
 * 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.
 */

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;

import org.apache.commons.jcs3.auxiliary.AuxiliaryCacheAttributes;
import org.apache.commons.jcs3.auxiliary.disk.AbstractDiskCache;
import org.apache.commons.jcs3.engine.behavior.ICacheElement;
import org.apache.commons.jcs3.engine.behavior.IElementSerializer;
import org.apache.commons.jcs3.engine.behavior.IRequireScheduler;
import org.apache.commons.jcs3.engine.control.group.GroupAttrName;
import org.apache.commons.jcs3.engine.control.group.GroupId;
import org.apache.commons.jcs3.engine.stats.StatElement;
import org.apache.commons.jcs3.engine.stats.Stats;
import org.apache.commons.jcs3.engine.stats.behavior.IStatElement;
import org.apache.commons.jcs3.engine.stats.behavior.IStats;
import org.apache.commons.jcs3.log.Log;
import org.apache.commons.jcs3.log.LogManager;
import org.apache.commons.jcs3.utils.serialization.StandardSerializer;

/**
 * There is one BlockDiskCache per region. It manages the key and data store.
 */
public class BlockDiskCache<K, V>
    extends AbstractDiskCache<K, V>
    implements IRequireScheduler
{
    /** The logger. */
    private static final Log log = LogManager.getLog( BlockDiskCache.class );

    /** The name to prefix all log messages with. */
    private final String logCacheName;

    /** The name of the file to store data. */
    private final String fileName;

    /** The data access object */
    private BlockDisk dataFile;

    /** Attributes governing the behavior of the block disk cache. */
    private final BlockDiskCacheAttributes blockDiskCacheAttributes;

    /** The root directory for keys and data. */
    private final File rootDirectory;

    /** Store, loads, and persists the keys */
    private BlockDiskKeyStore<K> keyStore;

    /**
     * Use this lock to synchronize reads and writes to the underlying storage mechanism. We don't
     * need a reentrant lock, since we only lock one level.
     */
    private final ReentrantReadWriteLock storageLock = new ReentrantReadWriteLock();

    private ScheduledFuture<?> future;

    /**
     * Constructs the BlockDisk after setting up the root directory.
     * <p>
     * @param cacheAttributes
     */
    public BlockDiskCache( final BlockDiskCacheAttributes cacheAttributes )
    {
        this( cacheAttributes, new StandardSerializer() );
    }

    /**
     * Constructs the BlockDisk after setting up the root directory.
     * <p>
     * @param cacheAttributes
     * @param elementSerializer used if supplied, the super's super will not set a null
     */
    public BlockDiskCache( final BlockDiskCacheAttributes cacheAttributes, final IElementSerializer elementSerializer )
    {
        super( cacheAttributes );
        setElementSerializer( elementSerializer );

        this.blockDiskCacheAttributes = cacheAttributes;
        this.logCacheName = "Region [" + getCacheName() + "] ";

        log.info("{0}: Constructing BlockDiskCache with attributes {1}", logCacheName, cacheAttributes );

        // Make a clean file name
        this.fileName = getCacheName().replaceAll("[^a-zA-Z0-9-_\\.]", "_");
        this.rootDirectory = cacheAttributes.getDiskPath();

        log.info("{0}: Cache file root directory: [{1}]", logCacheName, rootDirectory);

        try
        {
            if ( this.blockDiskCacheAttributes.getBlockSizeBytes() > 0 )
            {
                this.dataFile = new BlockDisk( new File( rootDirectory, fileName + ".data" ),
                                               this.blockDiskCacheAttributes.getBlockSizeBytes(),
                                               getElementSerializer() );
            }
            else
            {
                this.dataFile = new BlockDisk( new File( rootDirectory, fileName + ".data" ),
                                               getElementSerializer() );
            }

            keyStore = new BlockDiskKeyStore<>( this.blockDiskCacheAttributes, this );

            final boolean alright = verifyDisk();

            if ( keyStore.isEmpty() || !alright )
            {
                this.reset();
            }

            // Initialization finished successfully, so set alive to true.
            setAlive(true);
            log.info("{0}: Block Disk Cache is alive.", logCacheName);
        }
        catch ( final IOException e )
        {
            log.error("{0}: Failure initializing for fileName: {1} and root directory: {2}",
                    logCacheName, fileName, rootDirectory, e);
        }
    }

    /**
     * @see org.apache.commons.jcs3.engine.behavior.IRequireScheduler#setScheduledExecutorService(java.util.concurrent.ScheduledExecutorService)
     */
    @Override
    public void setScheduledExecutorService(final ScheduledExecutorService scheduledExecutor)
    {
        // add this region to the persistence thread.
        // TODO we might need to stagger this a bit.
        if ( this.blockDiskCacheAttributes.getKeyPersistenceIntervalSeconds() > 0 )
        {
            future = scheduledExecutor.scheduleAtFixedRate(keyStore::saveKeys,
                    this.blockDiskCacheAttributes.getKeyPersistenceIntervalSeconds(),
                    this.blockDiskCacheAttributes.getKeyPersistenceIntervalSeconds(),
                    TimeUnit.SECONDS);
        }
    }

    /**
     * We need to verify that the file on disk uses the same block size and that the file is the
     * proper size.
     * <p>
     * @return true if it looks ok
     */
    protected boolean verifyDisk()
    {
        boolean alright = false;
        // simply try to read a few. If it works, then the file is probably ok.
        // TODO add more.

        storageLock.readLock().lock();

        try
        {
            this.keyStore.entrySet().stream()
                .limit(100)
                .forEach(entry -> {
                    try
                    {
                        final Object data = this.dataFile.read(entry.getValue());
                        if ( data == null )
                        {
                            throw new IOException("Data is null");
                        }
                    }
                    catch (final IOException | ClassNotFoundException e)
                    {
                        throw new RuntimeException(logCacheName
                                + " Couldn't find data for key [" + entry.getKey() + "]", e);
                    }
                });
            alright = true;
        }
        catch ( final Exception e )
        {
            log.warn("{0}: Problem verifying disk.", logCacheName, e);
            alright = false;
        }
        finally
        {
            storageLock.readLock().unlock();
        }

        return alright;
    }

    /**
     * Return the keys in this cache.
     * <p>
     * @see org.apache.commons.jcs3.auxiliary.disk.AbstractDiskCache#getKeySet()
     */
    @Override
    public Set<K> getKeySet() throws IOException
    {
        final HashSet<K> keys = new HashSet<>();

        storageLock.readLock().lock();

        try
        {
            keys.addAll(this.keyStore.keySet());
        }
        finally
        {
            storageLock.readLock().unlock();
        }

        return keys;
    }

    /**
     * Gets matching items from the cache.
     * <p>
     * @param pattern
     * @return a map of K key to ICacheElement&lt;K, V&gt; element, or an empty map if there is no
     *         data in cache matching keys
     */
    @Override
    public Map<K, ICacheElement<K, V>> processGetMatching( final String pattern )
    {
        Set<K> keyArray = null;
        storageLock.readLock().lock();
        try
        {
            keyArray = new HashSet<>(keyStore.keySet());
        }
        finally
        {
            storageLock.readLock().unlock();
        }

        final Set<K> matchingKeys = getKeyMatcher().getMatchingKeysFromArray( pattern, keyArray );

        return matchingKeys.stream()
            .collect(Collectors.toMap(
                    key -> key,
                    this::processGet)).entrySet().stream()
                .filter(entry -> entry.getValue() != null)
                .collect(Collectors.toMap(
                        Entry::getKey,
                        Entry::getValue));
    }

    /**
     * Returns the number of keys.
     * <p>
     * (non-Javadoc)
     * @see org.apache.commons.jcs3.auxiliary.disk.AbstractDiskCache#getSize()
     */
    @Override
    public int getSize()
    {
        return this.keyStore.size();
    }

    /**
     * Gets the ICacheElement&lt;K, V&gt; for the key if it is in the cache. The program flow is as follows:
     * <ol>
     * <li>Make sure the disk cache is alive.</li> <li>Get a read lock.</li> <li>See if the key is
     * in the key store.</li> <li>If we found a key, ask the BlockDisk for the object at the
     * blocks..</li> <li>Release the lock.</li>
     * </ol>
     * @param key
     * @return ICacheElement
     * @see org.apache.commons.jcs3.auxiliary.disk.AbstractDiskCache#get(Object)
     */
    @Override
    protected ICacheElement<K, V> processGet( final K key )
    {
        if ( !isAlive() )
        {
            log.debug("{0}: No longer alive so returning null for key = {1}", logCacheName, key );
            return null;
        }

        log.debug("{0}: Trying to get from disk: {1}", logCacheName, key );

        ICacheElement<K, V> object = null;


        try
        {
            storageLock.readLock().lock();
            try {
                final int[] ded = this.keyStore.get( key );
                if ( ded != null )
                {
                    object = this.dataFile.read( ded );
                }
            } finally {
                storageLock.readLock().unlock();
            }

        }
        catch ( final IOException ioe )
        {
            log.error("{0}: Failure getting from disk--IOException, key = {1}", logCacheName, key, ioe );
            reset();
        }
        catch ( final Exception e )
        {
            log.error("{0}: Failure getting from disk, key = {1}", logCacheName, key, e );
        }
        return object;
    }

    /**
     * Writes an element to disk. The program flow is as follows:
     * <ol>
     * <li>Acquire write lock.</li> <li>See id an item exists for this key.</li> <li>If an item
     * already exists, add its blocks to the remove list.</li> <li>Have the Block disk write the
     * item.</li> <li>Create a descriptor and add it to the key map.</li> <li>Release the write
     * lock.</li>
     * </ol>
     * @param element
     * @see org.apache.commons.jcs3.auxiliary.disk.AbstractDiskCache#update(ICacheElement)
     */
    @Override
    protected void processUpdate( final ICacheElement<K, V> element )
    {
        if ( !isAlive() )
        {
            log.debug("{0}: No longer alive; aborting put of key = {1}",
                    () -> logCacheName, element::getKey);
            return;
        }

        int[] old = null;

        // make sure this only locks for one particular cache region
        storageLock.writeLock().lock();

        try
        {
            old = this.keyStore.get( element.getKey() );

            if ( old != null )
            {
                this.dataFile.freeBlocks( old );
            }

            final int[] blocks = this.dataFile.write( element );

            this.keyStore.put( element.getKey(), blocks );

            log.debug("{0}: Put to file [{1}] key [{2}]", () -> logCacheName,
                    () -> fileName, element::getKey);
        }
        catch ( final IOException e )
        {
            log.error("{0}: Failure updating element, key: {1} old: {2}",
                    logCacheName, element.getKey(), Arrays.toString(old), e);
        }
        finally
        {
            storageLock.writeLock().unlock();
        }

        log.debug("{0}: Storing element on disk, key: {1}", () -> logCacheName,
                element::getKey);
    }

    /**
     * Returns true if the removal was successful; or false if there is nothing to remove. Current
     * implementation always result in a disk orphan.
     * <p>
     * @param key
     * @return true if removed anything
     * @see org.apache.commons.jcs3.auxiliary.disk.AbstractDiskCache#remove(Object)
     */
    @Override
    protected boolean processRemove( final K key )
    {
        if ( !isAlive() )
        {
            log.debug("{0}: No longer alive so returning false for key = {1}", logCacheName, key );
            return false;
        }

        boolean reset = false;
        boolean removed = false;

        storageLock.writeLock().lock();

        try
        {
            if (key instanceof String && key.toString().endsWith(NAME_COMPONENT_DELIMITER))
            {
                removed = performPartialKeyRemoval((String) key);
            }
            else if (key instanceof GroupAttrName && ((GroupAttrName<?>) key).attrName == null)
            {
                removed = performGroupRemoval(((GroupAttrName<?>) key).groupId);
            }
            else
            {
                removed = performSingleKeyRemoval(key);
            }
        }
        catch ( final Exception e )
        {
            log.error("{0}: Problem removing element.", logCacheName, e );
            reset = true;
        }
        finally
        {
            storageLock.writeLock().unlock();
        }

        if ( reset )
        {
            reset();
        }

        return removed;
    }

    /**
     * Remove all elements from the group. This does not use the iterator to remove. It builds a
     * list of group elements and then removes them one by one.
     * <p>
     * This operates under a lock obtained in doRemove().
     * <p>
     *
     * @param key
     * @return true if an element was removed
     */
    private boolean performGroupRemoval(final GroupId key)
    {
        // remove all keys of the same name group.
        final List<K> itemsToRemove = keyStore.keySet()
                .stream()
                .filter(k -> k instanceof GroupAttrName && ((GroupAttrName<?>) k).groupId.equals(key))
                .collect(Collectors.toList());

        // remove matches.
        // Don't add to recycle bin here
        // https://issues.apache.org/jira/browse/JCS-67
        itemsToRemove.forEach(this::performSingleKeyRemoval);
        // TODO this needs to update the remove count separately

        return !itemsToRemove.isEmpty();
    }

    /**
     * Iterates over the keyset. Builds a list of matches. Removes all the keys in the list. Does
     * not remove via the iterator, since the map impl may not support it.
     * <p>
     * This operates under a lock obtained in doRemove().
     * <p>
     *
     * @param key
     * @return true if there was a match
     */
    private boolean performPartialKeyRemoval(final String key)
    {
        // remove all keys of the same name hierarchy.
        final List<K> itemsToRemove = keyStore.keySet()
                .stream()
                .filter(k -> k instanceof String && k.toString().startsWith(key))
                .collect(Collectors.toList());

        // remove matches.
        // Don't add to recycle bin here
        // https://issues.apache.org/jira/browse/JCS-67
        itemsToRemove.forEach(this::performSingleKeyRemoval);
        // TODO this needs to update the remove count separately

        return !itemsToRemove.isEmpty();
    }


        private boolean performSingleKeyRemoval(final K key) {
                final boolean removed;
                // remove single item.
                final int[] ded = this.keyStore.remove( key );
                removed = ded != null;
                if ( removed )
                {
                    this.dataFile.freeBlocks( ded );
                }

            log.debug("{0}: Disk removal: Removed from key hash, key [{1}] removed = {2}",
                    logCacheName, key, removed);
                return removed;
        }

    /**
     * Resets the keyfile, the disk file, and the memory key map.
     * <p>
     * @see org.apache.commons.jcs3.auxiliary.disk.AbstractDiskCache#removeAll()
     */
    @Override
    protected void processRemoveAll()
    {
        reset();
    }

    /**
     * Dispose of the disk cache in a background thread. Joins against this thread to put a cap on
     * the disposal time.
     * <p>
     * TODO make dispose window configurable.
     */
    @Override
    public void processDispose()
    {
        final Thread t = new Thread(this::disposeInternal, "BlockDiskCache-DisposalThread" );
        t.start();
        // wait up to 60 seconds for dispose and then quit if not done.
        try
        {
            t.join( 60 * 1000 );
        }
        catch ( final InterruptedException ex )
        {
            log.error("{0}: Interrupted while waiting for disposal thread to finish.",
                    logCacheName, ex );
        }
    }

    /**
     * Internal method that handles the disposal.
     */
    protected void disposeInternal()
    {
        if ( !isAlive() )
        {
            log.error("{0}: Not alive and dispose was called, filename: {1}", logCacheName, fileName);
            return;
        }
        storageLock.writeLock().lock();
        try
        {
            // Prevents any interaction with the cache while we're shutting down.
            setAlive(false);
            this.keyStore.saveKeys();

            if (future != null)
            {
                future.cancel(true);
            }

            try
            {
                log.debug("{0}: Closing files, base filename: {1}", logCacheName, fileName );
                dataFile.close();
                // dataFile = null;
            }
            catch ( final IOException e )
            {
                log.error("{0}: Failure closing files in dispose, filename: {1}",
                        logCacheName, fileName, e );
            }
        }
        finally
        {
            storageLock.writeLock().unlock();
        }

        log.info("{0}: Shutdown complete.", logCacheName);
    }

    /**
     * Returns the attributes.
     * <p>
     * @see org.apache.commons.jcs3.auxiliary.AuxiliaryCache#getAuxiliaryCacheAttributes()
     */
    @Override
    public AuxiliaryCacheAttributes getAuxiliaryCacheAttributes()
    {
        return this.blockDiskCacheAttributes;
    }

    /**
     * Reset effectively clears the disk cache, creating new files, recycle bins, and keymaps.
     * <p>
     * It can be used to handle errors by last resort, force content update, or remove all.
     */
    private void reset()
    {
        log.info("{0}: Resetting cache", logCacheName);

        try
        {
            storageLock.writeLock().lock();

            this.keyStore.reset();

            if ( dataFile != null )
            {
                dataFile.reset();
            }
        }
        catch ( final IOException e )
        {
            log.error("{0}: Failure resetting state", logCacheName, e );
        }
        finally
        {
            storageLock.writeLock().unlock();
        }
    }

    /**
     * Add these blocks to the emptyBlock list.
     * <p>
     * @param blocksToFree
     */
    protected void freeBlocks( final int[] blocksToFree )
    {
        this.dataFile.freeBlocks( blocksToFree );
    }

    /**
     * Returns info about the disk cache.
     * <p>
     * @see org.apache.commons.jcs3.auxiliary.AuxiliaryCache#getStatistics()
     */
    @Override
    public IStats getStatistics()
    {
        final IStats stats = new Stats();
        stats.setTypeName( "Block Disk Cache" );

        final ArrayList<IStatElement<?>> elems = new ArrayList<>();

        elems.add(new StatElement<>( "Is Alive", Boolean.valueOf(isAlive()) ) );
        elems.add(new StatElement<>( "Key Map Size", Integer.valueOf(this.keyStore.size()) ) );

        if (this.dataFile != null)
        {
            try
            {
                elems.add(new StatElement<>( "Data File Length", Long.valueOf(this.dataFile.length()) ) );
            }
            catch ( final IOException e )
            {
                log.error( e );
            }

            elems.add(new StatElement<>( "Block Size Bytes",
                    Integer.valueOf(this.dataFile.getBlockSizeBytes()) ) );
            elems.add(new StatElement<>( "Number Of Blocks",
                    Integer.valueOf(this.dataFile.getNumberOfBlocks()) ) );
            elems.add(new StatElement<>( "Average Put Size Bytes",
                    Long.valueOf(this.dataFile.getAveragePutSizeBytes()) ) );
            elems.add(new StatElement<>( "Empty Blocks",
                    Integer.valueOf(this.dataFile.getEmptyBlocks()) ) );
        }

        // get the stats from the super too
        final IStats sStats = super.getStatistics();
        elems.addAll(sStats.getStatElements());

        stats.setStatElements( elems );

        return stats;
    }

    /**
     * This is used by the event logging.
     * <p>
     * @return the location of the disk, either path or ip.
     */
    @Override
    protected String getDiskLocation()
    {
        return dataFile.getFilePath();
    }
}