/*
 *   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.directory.server.ldap.replication.provider;


import java.io.File;
import java.io.IOException;

import jdbm.RecordManager;
import jdbm.recman.BaseRecordManager;
import jdbm.recman.TransactionManager;

import org.apache.directory.api.ldap.model.constants.Loggers;
import org.apache.directory.api.ldap.model.constants.SchemaConstants;
import org.apache.directory.api.ldap.model.exception.LdapException;
import org.apache.directory.api.ldap.model.name.Dn;
import org.apache.directory.api.ldap.model.schema.SchemaManager;
import org.apache.directory.api.ldap.model.schema.comparators.SerializableComparator;
import org.apache.directory.server.core.api.DirectoryService;
import org.apache.directory.server.core.api.event.EventType;
import org.apache.directory.server.core.api.event.NotificationCriteria;
import org.apache.directory.server.core.api.partition.PartitionTxn;
import org.apache.directory.server.core.partition.impl.btree.jdbm.JdbmTable;
import org.apache.directory.server.core.partition.impl.btree.jdbm.StringSerializer;
import org.apache.directory.server.ldap.replication.ReplicaEventMessage;
import org.apache.directory.server.ldap.replication.ReplicaEventMessageSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * A structure storing the configuration on each consumer registered on a producer. It stores 
 * the following informations :
 * <ul>
 * <li>replicaId : the internal ID associated with the consumer on the provider</li>
 * <li>hostname : the consumer's host</li>
 * <li>searchFilter : the filter</li>
 * <li>lastSentCsn : the last CSN sent by the consumer</li>
 * <li>refreshNPersist : a flag indicating that the consumer is processing in Refresh and persist mode</li>
 * <li></li>
 * </ul>
 * A separate log is maintained for each syncrepl consumer.<br>
 * We also associate a Queue with each structure, which will store the messages to send to the consumer.
 *
 * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
 */
public class ReplicaEventLog implements Comparable<ReplicaEventLog>
{
    /** The logger */
    private static final Logger LOG = LoggerFactory.getLogger( ReplicaEventLog.class );

    /** A logger for the replication provider */
    private static final Logger PROVIDER_LOG = LoggerFactory.getLogger( Loggers.PROVIDER_LOG.getName() );

    /** hostname of the syncrepl consumer */
    private String hostName;

    /** the unmodified search filter as it was when received from the client */
    private String searchFilter;

    /** the csn that was sent to the client during the last sync session*/
    private String lastSentCsn;

    /** the persistent listener */
    private SyncReplSearchListener persistentListener;

    /** notification criteria used by the persistent search */
    private NotificationCriteria searchCriteria;

    /** the replica id */
    private int replicaId;

    /** flag indicating refreshAndPersist mode */
    private boolean refreshNPersist;

    /** the duration(in seconds) of consumer inactivity after which this log will be deleted. Defaults to 172800 seconds (i.e. 2 days) */
    private long maxIdlePeriod = DEFAULT_MAX_IDLE_PERIOD;

    /** the minimum number of entries to be present for beginning purging entries older than the last sent CSN. Default is 10000 */
    private int purgeThresholdCount = DEFAULT_PURGE_THRESHOLD_COUNT;

    // fields that won't be serialized
    /** The Journal of modifications */
    private JdbmTable<String, ReplicaEventMessage> journal;

    /** the underlying file  */
    private File journalFile;

    /** The record manager*/
    private RecordManager recman;

    /** A flag used to indicate that the consumer is not up to date */
    private volatile boolean dirty;

    /** the DN of the entry where this event log details are stored */
    private Dn consumerEntryDn;

    public static final String REPLICA_EVENT_LOG_NAME_PREFIX = "REPL_EVENT_LOG.";

    public static final int DEFAULT_PURGE_THRESHOLD_COUNT = 10000;

    /** The max delay for an idle replication log with no activity, by default the logs have no idle time period */
    public static final int DEFAULT_MAX_IDLE_PERIOD = -1;
    
    /** The partition transaction */
    private PartitionTxn partitionTxn;


    /**
     * Creates a new instance of EventLog for a replica
     * 
     * @param partitionTxn The Transaction to use
     * @param directoryService The DirectoryService instance
     * @param replicaId The replica ID
     * @throws IOException if we weren't able to log the event
     */
    public ReplicaEventLog( PartitionTxn partitionTxn, DirectoryService directoryService, int replicaId ) throws IOException
    {
        PROVIDER_LOG.debug( "Creating the replication queue for replica {}", replicaId );
        SchemaManager schemaManager = directoryService.getSchemaManager();
        this.replicaId = replicaId;
        this.searchCriteria = new NotificationCriteria( schemaManager );
        this.searchCriteria.setEventMask( EventType.ALL_EVENT_TYPES_MASK );

        // Create the journal file, or open if already exists
        File replDir = directoryService.getInstanceLayout().getReplDirectory();
        journalFile = new File( replDir, REPLICA_EVENT_LOG_NAME_PREFIX + replicaId );
        recman = new BaseRecordManager( journalFile.getAbsolutePath() );
        TransactionManager transactionManager = ( ( BaseRecordManager ) recman ).getTransactionManager();
        transactionManager.setMaximumTransactionsInLog( 200 );

        SerializableComparator<String> comparator = new SerializableComparator<>(
            SchemaConstants.CSN_ORDERING_MATCH_MR_OID );
        comparator.setSchemaManager( schemaManager );

        journal = new JdbmTable<>( schemaManager, journalFile.getName(), recman, comparator,
            StringSerializer.INSTANCE, new ReplicaEventMessageSerializer( schemaManager ) );
        
        this.partitionTxn = partitionTxn;
    }


    /**
     * Stores the given message in the queue 
     *
     * @param message The message to store
     */
    public synchronized void log( ReplicaEventMessage message )
    {
        try
        {
            LOG.debug( "logging entry with Dn {} with the event {}", message.getEntry().getDn(),
                message.getChangeType() );
            PROVIDER_LOG.debug( "logging entry with Dn {} with the event {}", message.getEntry().getDn(),
                message.getChangeType() );

            String entryCsn = message.getEntry().get( SchemaConstants.ENTRY_CSN_AT ).getString();
            journal.put( partitionTxn, entryCsn, message );
        }
        catch ( Exception e )
        {
            LOG.warn( "Failed to insert the entry into syncrepl log", e );
            PROVIDER_LOG.error( "Failed to insert the entry into syncrepl log", e );
        }
    }


    /**
     * Deletes the queue (to remove the log) and recreates a new queue instance
     * with the same queue name. Also creates the corresponding message producer
     *
     * @throws Exception If the queue can't be deleted
     */
    public void truncate() throws Exception
    {
    }


    /**
     * Re-create the queue
     * @throws Exception If the creation has failed
     */
    public void recreate() throws Exception
    {
        LOG.debug( "recreating the queue for the replica id {}", replicaId );
    }


    /**
     * Stop the EventLog
     * 
     * @throws Exception If the stop failed
     */
    public void stop() throws Exception
    {
        PROVIDER_LOG.debug( "Stopping the EventLog for replicaId {}", replicaId );

        // Close the producer and session, DO NOT close connection 
        if ( journal != null )
        {
            journal.close( partitionTxn );
        }

        journal = null;

        if ( recman != null )
        {
            recman.close();
        }

        recman = null;
    }


    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals( Object obj )
    {
        if ( !( obj instanceof ReplicaEventLog ) )
        {
            return false;
        }

        ReplicaEventLog other = ( ReplicaEventLog ) obj;

        return replicaId == other.getId();
    }


    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode()
    {
        int result = 17;
        result = 31 * result + searchFilter.hashCode();
        result = 31 * result + hostName.hashCode();

        return result;
    }


    /**
     * {@inheritDoc}
     */
    public int compareTo( ReplicaEventLog o )
    {
        if ( this.equals( o ) )
        {
            return 0;
        }

        return 1;
    }


    /**
     * @return The listener
     */
    public SyncReplSearchListener getPersistentListener()
    {
        return persistentListener;
    }


    /**
     * Set the listener
     * @param persistentListener The listener
     */
    public void setPersistentListener( SyncReplSearchListener persistentListener )
    {
        this.persistentListener = persistentListener;
    }


    /**
     * @return The search criteria
     */
    public NotificationCriteria getSearchCriteria()
    {
        return searchCriteria;
    }


    /**
     * Stores the search criteria
     * @param searchCriteria The search criteria
     */
    public void setSearchCriteria( NotificationCriteria searchCriteria )
    {
        this.searchCriteria = searchCriteria;
    }


    /**
     * @return true if the consumer is in Refresh And Persist mode
     */
    public boolean isRefreshNPersist()
    {
        return refreshNPersist;
    }


    /**
     * @param refreshNPersist if true, set the EventLog in Refresh and Persist mode
     */
    public void setRefreshNPersist( boolean refreshNPersist )
    {
        this.refreshNPersist = refreshNPersist;
    }


    /**
     * @return The replica ID
     */
    public int getId()
    {
        return replicaId;
    }


    /**
     * @return The last CSN sent by the consumer
     */
    public String getLastSentCsn()
    {
        return lastSentCsn;
    }


    /**
     * Update the last Sent CSN. If it's different from the present one, we
     * will set the dirty flag to true, and it will be stored in DIT.
     *  
     * @param lastSentCsn The new Sent CSN
     */
    public void setLastSentCsn( String lastSentCsn )
    {
        // set only if there is a change in cookie value
        // this will avoid setting the dirty flag which eventually is used for
        // storing the details of this log
        if ( !lastSentCsn.equals( this.lastSentCsn ) )
        {
            this.lastSentCsn = lastSentCsn;
            dirty = true;
        }
    }


    /**
     * @return The consumer Hostname
     */
    public String getHostName()
    {
        return hostName;
    }


    /**
     * Set the consumer hostname
     * @param hostName The consumer hostname
     */
    public void setHostName( String hostName )
    {
        this.hostName = hostName;
    }


    /**
     * @return The searchFilter
     */
    public String getSearchFilter()
    {
        return searchFilter;
    }


    /**
     * Set the searchFilter
     * @param searchFilter The searchFilter
     */
    public void setSearchFilter( String searchFilter )
    {
        this.searchFilter = searchFilter;
    }


    /**
     * @return True if the consumer is not up to date
     */
    public boolean isDirty()
    {
        return dirty;
    }


    /**
     * Set the dirty flag
     * @param dirty The current consumer status
     */
    public void setDirty( boolean dirty )
    {
        this.dirty = dirty;
    }


    /**
     * @return The queue name
     */
    public String getQueueName()
    {
        return "replicaId=" + replicaId;
    }


    /**
     * @param consumerCsn the consumer's CSN extracted from cookie
     * @return A cursor on top of the queue
     * @throws Exception If the cursor can't be created
     */
    public ReplicaJournalCursor getCursor( String consumerCsn ) throws Exception
    {
        return new ReplicaJournalCursor( partitionTxn, journal, consumerCsn );
    }


    /**
     * @return the name of this replica log
     */
    public String getName()
    {
        return journal.getName();
    }


    /**
     * @return the number of entries present in the replica log
     */
    public synchronized long count()
    {
        try
        {
            return journal.count( partitionTxn );
        }
        catch ( LdapException e )
        {
            throw new RuntimeException( e );
        }
    }


    public long getMaxIdlePeriod()
    {
        return maxIdlePeriod;
    }


    public void setMaxIdlePeriod( long maxIdlePeriod )
    {
        if ( maxIdlePeriod <= 0 )
        {
            maxIdlePeriod = DEFAULT_MAX_IDLE_PERIOD;
        }

        this.maxIdlePeriod = maxIdlePeriod;
    }


    public int getPurgeThresholdCount()
    {
        return purgeThresholdCount;
    }


    public void setPurgeThresholdCount( int purgeThresholdCount )
    {
        if ( purgeThresholdCount <= 0 )
        {
            purgeThresholdCount = DEFAULT_PURGE_THRESHOLD_COUNT;
        }

        this.purgeThresholdCount = purgeThresholdCount;
    }


    public Dn getConsumerEntryDn()
    {
        return consumerEntryDn;
    }


    public void setConsumerEntryDn( Dn consumerEntryDn )
    {
        this.consumerEntryDn = consumerEntryDn;
    }


    @Override
    public String toString()
    {
        return "ReplicaEventLog [hostName=" + hostName + ", searchFilter=" + searchFilter + ", lastSentCsn="
            + lastSentCsn + ", searchCriteria=" + searchCriteria + ", replicaId=" + replicaId
            + ", refreshNPersist=" + refreshNPersist + ", maxInactivePeriod=" + maxIdlePeriod
            + ", purgeThresholdCount=" + purgeThresholdCount + ", journalFile=" + journalFile
            + ", dirty=" + dirty + ", consumerEntryDn=" + consumerEntryDn + "]";
    }
}