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

/*
 * 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.IOException;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import javax.sql.DataSource;

import org.apache.commons.jcs3.auxiliary.AuxiliaryCacheAttributes;
import org.apache.commons.jcs3.auxiliary.disk.AbstractDiskCache;
import org.apache.commons.jcs3.auxiliary.disk.jdbc.dsfactory.DataSourceFactory;
import org.apache.commons.jcs3.engine.behavior.ICache;
import org.apache.commons.jcs3.engine.behavior.ICacheElement;
import org.apache.commons.jcs3.engine.logging.behavior.ICacheEvent;
import org.apache.commons.jcs3.engine.logging.behavior.ICacheEventLogger;
import org.apache.commons.jcs3.engine.stats.StatElement;
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;

/**
 * This is the jdbc disk cache plugin.
 * <p>
 * It expects a table created by the following script. The table name is configurable.
 * </p>
 * <pre>
 *                       drop TABLE JCS_STORE;
 *                       CREATE TABLE JCS_STORE
 *                       (
 *                       CACHE_KEY                  VARCHAR(250)          NOT NULL,
 *                       REGION                     VARCHAR(250)          NOT NULL,
 *                       ELEMENT                    BLOB,
 *                       CREATE_TIME                TIMESTAMP,
 *                       UPDATE_TIME_SECONDS        BIGINT,
 *                       MAX_LIFE_SECONDS           BIGINT,
 *                       SYSTEM_EXPIRE_TIME_SECONDS BIGINT,
 *                       IS_ETERNAL                 CHAR(1),
 *                       PRIMARY KEY (CACHE_KEY, REGION)
 *                       );
 * </pre>
 * <p>
 * The cleanup thread will delete non eternal items where (now - create time) &gt; max life seconds *
 * 1000
 * </p>
 * <p>
 * To speed up the deletion the SYSTEM_EXPIRE_TIME_SECONDS is used instead. It is recommended that
 * an index be created on this column is you will have over a million records.
 * </p>
 */
public class JDBCDiskCache<K, V>
    extends AbstractDiskCache<K, V>
{
    /** The local logger. */
    private static final Log log = LogManager.getLog( JDBCDiskCache.class );

    /** configuration */
    private JDBCDiskCacheAttributes jdbcDiskCacheAttributes;

    /** # of times update was called */
    private final AtomicInteger updateCount = new AtomicInteger(0);

    /** # of times get was called */
    private final AtomicInteger getCount = new AtomicInteger(0);

    /** # of times getMatching was called */
    private final AtomicInteger getMatchingCount = new AtomicInteger(0);

    /** db connection pool */
    private final DataSourceFactory dsFactory;

    /** tracks optimization */
    private TableState tableState;

    /**
     * Constructs a JDBC Disk Cache for the provided cache attributes. The table state object is
     * used to mark deletions.
     * <p>
     * @param cattr the configuration object for this cache
     * @param dsFactory the DataSourceFactory for this cache
     * @param tableState an object to track table operations
     */
    public JDBCDiskCache(final JDBCDiskCacheAttributes cattr, final DataSourceFactory dsFactory, final TableState tableState)
    {
        super( cattr );

        setTableState( tableState );
        setJdbcDiskCacheAttributes( cattr );

        log.info( "jdbcDiskCacheAttributes = {0}", this::getJdbcDiskCacheAttributes);

        // This initializes the pool access.
        this.dsFactory = dsFactory;

        // Initialization finished successfully, so set alive to true.
        setAlive(true);
    }

    /**
     * Inserts or updates. By default it will try to insert. If the item exists we will get an
     * error. It will then update. This behavior is configurable. The cache can be configured to
     * check before inserting.
     * <p>
     * @param ce
     */
    @Override
    protected void processUpdate( final ICacheElement<K, V> ce )
    {
        updateCount.incrementAndGet();

        log.debug( "updating, ce = {0}", ce );

        try (Connection con = getDataSource().getConnection())
        {
            log.debug( "Putting [{0}] on disk.", ce::getKey);

            try
            {
                final byte[] element = getElementSerializer().serialize( ce );
                insertOrUpdate( ce, con, element );
            }
            catch ( final IOException e )
            {
                log.error( "Could not serialize element", e );
            }
        }
        catch ( final SQLException e )
        {
            log.error( "Problem getting connection.", e );
        }
    }

    /**
     * If test before insert it true, we check to see if the element exists. If the element exists
     * we will update. Otherwise, we try inserting.  If this fails because the item exists, we will
     * update.
     * <p>
     * @param ce
     * @param con
     * @param element
     */
    private void insertOrUpdate( final ICacheElement<K, V> ce, final Connection con, final byte[] element )
    {
        boolean exists = false;

        // First do a query to determine if the element already exists
        if ( this.getJdbcDiskCacheAttributes().isTestBeforeInsert() )
        {
            exists = doesElementExist( ce, con );
        }

        // If it doesn't exist, insert it, otherwise update
        if ( !exists )
        {
            exists = insertRow( ce, con, element );
        }

        // update if it exists.
        if ( exists )
        {
            updateRow( ce, con, element );
        }
    }

    /**
     * This inserts a new row in the database.
     * <p>
     * @param ce
     * @param con
     * @param element
     * @return true if the insertion fails because the record exists.
     */
    private boolean insertRow( final ICacheElement<K, V> ce, final Connection con, final byte[] element )
    {
        boolean exists = false;
        final String sqlI = String.format("insert into %s"
                + " (CACHE_KEY, REGION, ELEMENT, MAX_LIFE_SECONDS, IS_ETERNAL, CREATE_TIME, UPDATE_TIME_SECONDS,"
                + " SYSTEM_EXPIRE_TIME_SECONDS) "
                + " values (?, ?, ?, ?, ?, ?, ?, ?)", getJdbcDiskCacheAttributes().getTableName());

        try (PreparedStatement psInsert = con.prepareStatement( sqlI ))
        {
            psInsert.setString( 1, ce.getKey().toString() );
            psInsert.setString( 2, this.getCacheName() );
            psInsert.setBytes( 3, element );
            psInsert.setLong( 4, ce.getElementAttributes().getMaxLife() );
            psInsert.setString( 5, ce.getElementAttributes().getIsEternal() ? "T" : "F" );

            final Timestamp createTime = new Timestamp( ce.getElementAttributes().getCreateTime() );
            psInsert.setTimestamp( 6, createTime );

            final long now = System.currentTimeMillis() / 1000;
            psInsert.setLong( 7, now );

            final long expireTime = now + ce.getElementAttributes().getMaxLife();
            psInsert.setLong( 8, expireTime );

            psInsert.execute();
        }
        catch ( final SQLException e )
        {
            if ("23000".equals(e.getSQLState()))
            {
                exists = true;
            }
            else
            {
                log.error( "Could not insert element", e );
            }

            // see if it exists, if we didn't already
            if ( !exists && !this.getJdbcDiskCacheAttributes().isTestBeforeInsert() )
            {
                exists = doesElementExist( ce, con );
            }
        }

        return exists;
    }

    /**
     * This updates a row in the database.
     * <p>
     * @param ce
     * @param con
     * @param element
     */
    private void updateRow( final ICacheElement<K, V> ce, final Connection con, final byte[] element )
    {
        final String sqlU = String.format("update %s"
                + " set ELEMENT  = ?, CREATE_TIME = ?, UPDATE_TIME_SECONDS = ?, " + " SYSTEM_EXPIRE_TIME_SECONDS = ? "
                + " where CACHE_KEY = ? and REGION = ?", getJdbcDiskCacheAttributes().getTableName());

        try (PreparedStatement psUpdate = con.prepareStatement( sqlU ))
        {
            psUpdate.setBytes( 1, element );

            final Timestamp createTime = new Timestamp( ce.getElementAttributes().getCreateTime() );
            psUpdate.setTimestamp( 2, createTime );

            final long now = System.currentTimeMillis() / 1000;
            psUpdate.setLong( 3, now );

            final long expireTime = now + ce.getElementAttributes().getMaxLife();
            psUpdate.setLong( 4, expireTime );

            psUpdate.setString( 5, (String) ce.getKey() );
            psUpdate.setString( 6, this.getCacheName() );
            psUpdate.execute();

            log.debug( "ran update {0}", sqlU );
        }
        catch ( final SQLException e )
        {
            log.error( "Error executing update sql [{0}]", sqlU, e );
        }
    }

    /**
     * Does an element exist for this key?
     * <p>
     * @param ce the cache element
     * @param con a database connection
     * @return boolean
     */
    protected boolean doesElementExist( final ICacheElement<K, V> ce, final Connection con )
    {
        boolean exists = false;
        // don't select the element, since we want this to be fast.
        final String sqlS = String.format("select CACHE_KEY from %s where REGION = ? and CACHE_KEY = ?",
                getJdbcDiskCacheAttributes().getTableName());

        try (PreparedStatement psSelect = con.prepareStatement( sqlS ))
        {
            psSelect.setString( 1, this.getCacheName() );
            psSelect.setString( 2, (String) ce.getKey() );

            try (ResultSet rs = psSelect.executeQuery())
            {
                exists = rs.next();
            }

            log.debug( "[{0}] existing status is {1}", ce.getKey(), exists );
        }
        catch ( final SQLException e )
        {
            log.error( "Problem looking for item before insert.", e );
        }

        return exists;
    }

    /**
     * Queries the database for the value. If it gets a result, the value is deserialized.
     * <p>
     * @param key
     * @return ICacheElement
     * @see org.apache.commons.jcs3.auxiliary.disk.AbstractDiskCache#get(Object)
     */
    @Override
    protected ICacheElement<K, V> processGet( final K key )
    {
        getCount.incrementAndGet();

        log.debug( "Getting [{0}] from disk", key );

        if ( !isAlive() )
        {
            return null;
        }

        ICacheElement<K, V> obj = null;

        // region, key
        final String selectString = String.format("select ELEMENT from %s where REGION = ? and CACHE_KEY = ?",
                getJdbcDiskCacheAttributes().getTableName());

        try (Connection con = getDataSource().getConnection())
        {
            try (PreparedStatement psSelect = con.prepareStatement( selectString ))
            {
                psSelect.setString( 1, this.getCacheName() );
                psSelect.setString( 2, key.toString() );

                try (ResultSet rs = psSelect.executeQuery())
                {
                    byte[] data = null;

                    if ( rs.next() )
                    {
                        data = rs.getBytes( 1 );
                    }

                    if ( data != null )
                    {
                        try
                        {
                            // USE THE SERIALIZER
                            obj = getElementSerializer().deSerialize( data, null );
                        }
                        catch ( final IOException | ClassNotFoundException e )
                        {
                            log.error( "Problem getting item for key [{0}]", key, e );
                        }
                    }
                }
            }
        }
        catch ( final SQLException sqle )
        {
            log.error( "Caught a SQL exception trying to get the item for key [{0}]",
                    key, sqle );
        }

        return obj;
    }

    /**
     * This will run a like query. It will try to construct a usable query but different
     * implementations will be needed to adjust the syntax.
     * <p>
     * @param pattern
     * @return key,value map
     */
    @Override
    protected Map<K, ICacheElement<K, V>> processGetMatching( final String pattern )
    {
        getMatchingCount.incrementAndGet();

        log.debug( "Getting [{0}] from disk", pattern);

        if ( !isAlive() )
        {
            return null;
        }

        final Map<K, ICacheElement<K, V>> results = new HashMap<>();

        // region, key
        final String selectString = String.format("select ELEMENT from %s where REGION = ? and CACHE_KEY like ?",
                getJdbcDiskCacheAttributes().getTableName());

        try (Connection con = getDataSource().getConnection())
        {
            try (PreparedStatement psSelect = con.prepareStatement( selectString ))
            {
                psSelect.setString( 1, this.getCacheName() );
                psSelect.setString( 2, constructLikeParameterFromPattern( pattern ) );

                try (ResultSet rs = psSelect.executeQuery())
                {
                    while ( rs.next() )
                    {
                        final byte[] data = rs.getBytes(1);
                        if ( data != null )
                        {
                            try
                            {
                                // USE THE SERIALIZER
                                final ICacheElement<K, V> value = getElementSerializer().deSerialize( data, null );
                                results.put( value.getKey(), value );
                            }
                            catch ( final IOException | ClassNotFoundException e )
                            {
                                log.error( "Problem getting items for pattern [{0}]", pattern, e );
                            }
                        }
                    }
                }
            }
        }
        catch ( final SQLException sqle )
        {
            log.error( "Caught a SQL exception trying to get items for pattern [{0}]",
                    pattern, sqle );
        }

        return results;
    }

    /**
     * @param pattern
     * @return String to use in the like query.
     */
    public String constructLikeParameterFromPattern( final String pattern )
    {
        String likePattern = pattern.replace( ".+", "%" );
        likePattern = likePattern.replace( ".", "_" );

        log.debug( "pattern = [{0}]", likePattern );

        return likePattern;
    }

    /**
     * Returns true if the removal was successful; or false if there is nothing to remove. Current
     * implementation always results in a disk orphan.
     * <p>
     * @param key
     * @return boolean
     */
    @Override
    protected boolean processRemove( final K key )
    {
        // remove single item.
        final String sqlSingle = String.format("delete from %s where REGION = ? and CACHE_KEY = ?",
                getJdbcDiskCacheAttributes().getTableName());
        // remove all keys of the same name group.
        final String sqlPartial = String.format("delete from %s where REGION = ? and CACHE_KEY like ?",
                getJdbcDiskCacheAttributes().getTableName());

        try (Connection con = getDataSource().getConnection())
        {
            boolean partial = key.toString().endsWith(ICache.NAME_COMPONENT_DELIMITER);
            String sql = partial ? sqlPartial : sqlSingle;

            try (PreparedStatement psSelect = con.prepareStatement(sql))
            {
                psSelect.setString( 1, this.getCacheName() );
                if ( partial )
                {
                    psSelect.setString( 2, key.toString() + "%" );
                }
                else
                {
                    psSelect.setString( 2, key.toString() );
                }

                psSelect.executeUpdate();

                setAlive(true);
            }
            catch ( final SQLException e )
            {
                log.error( "Problem creating statement. sql [{0}]", sql, e );
                setAlive(false);
            }
        }
        catch ( final SQLException e )
        {
            log.error( "Problem updating cache.", e );
            reset();
        }
        return false;
    }

    /**
     * This should remove all elements. The auxiliary can be configured to forbid this behavior. If
     * remove all is not allowed, the method balks.
     */
    @Override
    protected void processRemoveAll()
    {
        // it should never get here from the abstract disk cache.
        if ( this.jdbcDiskCacheAttributes.isAllowRemoveAll() )
        {
            final String sql = String.format("delete from %s where REGION = ?",
                    getJdbcDiskCacheAttributes().getTableName());

            try (Connection con = getDataSource().getConnection())
            {
                try (PreparedStatement psDelete = con.prepareStatement( sql ))
                {
                    psDelete.setString( 1, this.getCacheName() );
                    setAlive(true);
                    psDelete.executeUpdate();
                }
                catch ( final SQLException e )
                {
                    log.error( "Problem creating statement.", e );
                    setAlive(false);
                }
            }
            catch ( final SQLException e )
            {
                log.error( "Problem removing all.", e );
                reset();
            }
        }
        else
        {
            log.info( "RemoveAll was requested but the request was not fulfilled: "
                    + "allowRemoveAll is set to false." );
        }
    }

    /**
     * Removed the expired. (now - create time) &gt; max life seconds * 1000
     * <p>
     * @return the number deleted
     */
    protected int deleteExpired()
    {
        int deleted = 0;

        try (Connection con = getDataSource().getConnection())
        {
            // The shrinker thread might kick in before the table is created
            // So check if the table exists first
            final DatabaseMetaData dmd = con.getMetaData();
            final ResultSet result = dmd.getTables(null, null,
                    getJdbcDiskCacheAttributes().getTableName(), null);

            if (result.next())
            {
                getTableState().setState( TableState.DELETE_RUNNING );
                final long now = System.currentTimeMillis() / 1000;

                final String sql = String.format("delete from %s where IS_ETERNAL = ? and REGION = ?"
                        + " and ? > SYSTEM_EXPIRE_TIME_SECONDS", getJdbcDiskCacheAttributes().getTableName());

                try (PreparedStatement psDelete = con.prepareStatement( sql ))
                {
                    psDelete.setString( 1, "F" );
                    psDelete.setString( 2, this.getCacheName() );
                    psDelete.setLong( 3, now );

                    setAlive(true);

                    deleted = psDelete.executeUpdate();
                }
                catch ( final SQLException e )
                {
                    log.error( "Problem creating statement.", e );
                    setAlive(false);
                }

                logApplicationEvent( getAuxiliaryCacheAttributes().getName(), "deleteExpired",
                                     "Deleted expired elements.  URL: " + getDiskLocation() );
            }
            else
            {
                log.warn( "Trying to shrink non-existing table [{0}]",
                        getJdbcDiskCacheAttributes().getTableName() );
            }
        }
        catch ( final SQLException e )
        {
            logError( getAuxiliaryCacheAttributes().getName(), "deleteExpired",
                    e.getMessage() + " URL: " + getDiskLocation() );
            log.error( "Problem removing expired elements from the table.", e );
            reset();
        }
        finally
        {
            getTableState().setState( TableState.FREE );
        }

        return deleted;
    }

    /**
     * Typically this is used to handle errors by last resort, force content update, or removeall
     */
    public void reset()
    {
        // nothing
    }

    /** Shuts down the pool */
    @Override
    public void processDispose()
    {
        final ICacheEvent<K> cacheEvent = createICacheEvent( getCacheName(), null, ICacheEventLogger.DISPOSE_EVENT );

        try
        {
                dsFactory.close();
        }
        catch ( final SQLException e )
        {
            log.error( "Problem shutting down.", e );
        }
        finally
        {
            logICacheEvent( cacheEvent );
        }
    }

    /**
     * Returns the current cache size. Just does a count(*) for the region.
     * <p>
     * @return The size value
     */
    @Override
    public int getSize()
    {
        int size = 0;

        // region, key
        final String selectString = String.format("select count(*) from %s where REGION = ?",
                getJdbcDiskCacheAttributes().getTableName());

        try (Connection con = getDataSource().getConnection())
        {
            try (PreparedStatement psSelect = con.prepareStatement( selectString ))
            {
                psSelect.setString( 1, this.getCacheName() );

                try (ResultSet rs = psSelect.executeQuery())
                {
                    if ( rs.next() )
                    {
                        size = rs.getInt( 1 );
                    }
                }
            }
        }
        catch ( final SQLException e )
        {
            log.error( "Problem getting size.", e );
        }

        return size;
    }

    /**
     * Return the keys in this cache.
     * <p>
     * @see org.apache.commons.jcs3.auxiliary.disk.AbstractDiskCache#getKeySet()
     */
    @Override
    public Set<K> getKeySet() throws IOException
    {
        throw new UnsupportedOperationException( "Groups not implemented." );
        // return null;
    }

    /**
     * @param jdbcDiskCacheAttributes The jdbcDiskCacheAttributes to set.
     */
    protected void setJdbcDiskCacheAttributes( final JDBCDiskCacheAttributes jdbcDiskCacheAttributes )
    {
        this.jdbcDiskCacheAttributes = jdbcDiskCacheAttributes;
    }

    /**
     * @return Returns the jdbcDiskCacheAttributes.
     */
    protected JDBCDiskCacheAttributes getJdbcDiskCacheAttributes()
    {
        return jdbcDiskCacheAttributes;
    }

    /**
     * @return Returns the AuxiliaryCacheAttributes.
     */
    @Override
    public AuxiliaryCacheAttributes getAuxiliaryCacheAttributes()
    {
        return this.getJdbcDiskCacheAttributes();
    }

    /**
     * Extends the parent stats.
     * <p>
     * @return IStats
     */
    @Override
    public IStats getStatistics()
    {
        final IStats stats = super.getStatistics();
        stats.setTypeName( "JDBC/Abstract Disk Cache" );

        final List<IStatElement<?>> elems = stats.getStatElements();

        elems.add(new StatElement<>( "Update Count", updateCount ) );
        elems.add(new StatElement<>( "Get Count", getCount ) );
        elems.add(new StatElement<>( "Get Matching Count", getMatchingCount ) );
        elems.add(new StatElement<>( "DB URL", getJdbcDiskCacheAttributes().getUrl()) );

        stats.setStatElements( elems );

        return stats;
    }

    /**
     * Returns the name of the table.
     * <p>
     * @return the table name or UNDEFINED
     */
    protected String getTableName()
    {
        String name = "UNDEFINED";
        if ( this.getJdbcDiskCacheAttributes() != null )
        {
            name = this.getJdbcDiskCacheAttributes().getTableName();
        }
        return name;
    }

    /**
     * @param tableState The tableState to set.
     */
    public void setTableState( final TableState tableState )
    {
        this.tableState = tableState;
    }

    /**
     * @return Returns the tableState.
     */
    public TableState getTableState()
    {
        return tableState;
    }

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

    /**
     * Public so managers can access it.
     * @return the dsFactory
     * @throws SQLException if getting a data source fails
     */
    public DataSource getDataSource() throws SQLException
    {
        return dsFactory.getDataSource();
    }

    /**
     * For debugging.
     * <p>
     * @return this.getStats();
     */
    @Override
    public String toString()
    {
        return this.getStats();
    }
}