/*
    *   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.util.Map;
  
  import org.apache.directory.api.ldap.model.constants.SchemaConstants;
  import org.apache.directory.api.ldap.model.csn.Csn;
  import org.apache.directory.api.ldap.model.exception.LdapException;
  import org.apache.directory.server.core.api.DirectoryService;
  import org.apache.directory.server.ldap.replication.ReplicaEventMessage;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  
  /**
   * Deletes old entries from the replication event logs that are configured in refreshNPersist mode.
   * 
   * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
   */
  public class ReplicaEventLogJanitor extends Thread
  {
      private static final Logger LOG = LoggerFactory.getLogger( ReplicaEventLogJanitor.class );
  
      private DirectoryService directoryService;
  
      private Map<Integer, ReplicaEventLog> replicaLogMap;
  
      private volatile boolean stop = false;
  
      /** A lock used to wait */
      final Object lock = new Object();
  
      /** time the janitor thread sleeps before successive cleanup attempts. Default value is 5 minutes */
      private long sleepTime = 5 * 60 * 1000L;
  
      private long thresholdTime = 2 * 60 * 60 * 1000L;
  
  
      public ReplicaEventLogJanitor( final DirectoryService directoryService,
          final Map<Integer, ReplicaEventLog> replicaLogMap )
      {
          // if log is in refreshNpersist mode, has more entries than the log's threshold count then 
          // all the entries before the last sent CSN and older than 2 hours will be purged
          this.directoryService = directoryService;
          this.replicaLogMap = replicaLogMap;
          setDaemon( true );
      }
  
  
      @Override
      public void run()
      {
          while ( !stop )
          {
              for ( ReplicaEventLog log : replicaLogMap.values() )
              {
                  synchronized ( log ) // lock the log and clean
                  {
                      try
                      {
                          String lastSentCsn = log.getLastSentCsn();
  
                          if ( lastSentCsn == null )
                          {
                              LOG.debug( "last sent CSN is null for the replica {}, skipping cleanup", log.getName() );
                              return;
                          }
  
                          long now = directoryService.getTimeProvider().currentIimeMillis();
  
                          long maxIdleTime = log.getMaxIdlePeriod() * 1000L;
  
                          long lastUpdatedTime = new Csn( lastSentCsn ).getTimestamp();
  
                          LOG.debug( "checking log idle time now={} lastUpdatedTime={} maxIdleTime={}", now,
                              lastUpdatedTime, maxIdleTime );
  
                          // DO NOT delete those with maxIdleTime <= 0
                          if ( ( maxIdleTime > 0 ) && ( now - lastUpdatedTime ) >= maxIdleTime )
                         {
                             //max idle time of the event log reached, delete it
                             removeEventLog( log );
 
                             // delete the associated entry from DiT, note that ConsumerLogEntryDeleteListener 
                             // will get called eventually but removeEventLog() will not be called cause by 
                             // that time this log will not be present in replicaLogMap
                             // The reason we don't call this method first is to guard against any rename
                             // operation performed on the log's entry in DiT
                             try
                             {
                                 directoryService.getAdminSession().delete( log.getConsumerEntryDn() );
                             }
                             catch ( LdapException e )
                             {
                                 LOG.warn( "Failed to delete the entry {} of replica event log {}",
                                     log.getConsumerEntryDn(), log.getName(), e );
                             }
 
                             continue;
                         }
 
                         long thresholdCount = log.getPurgeThresholdCount();
 
                         if ( log.count() < thresholdCount )
                         {
                             continue;
                         }
 
                         LOG.debug( "starting to purge the log entries that are older than {} milliseconds",
                             thresholdTime );
 
                         long deleteCount = 0;
 
                         ReplicaJournalCursor cursor = log.getCursor( null ); // pass no CSN
                         cursor.skipQualifyingWhileFetching();
 
                         while ( cursor.next() )
                         {
                             ReplicaEventMessage message = cursor.get();
                             String csnVal = message.getEntry().get( SchemaConstants.ENTRY_CSN_AT ).getString();
 
                             // skip if we reach the lastSentCsn or got past it
                             if ( csnVal.compareTo( lastSentCsn ) >= 0 )
                             {
                                 break;
                             }
 
                             Csn csn = new Csn( csnVal );
 
                             if ( ( now - csn.getTimestamp() ) >= thresholdTime )
                             {
                                 cursor.delete();
                                 deleteCount++;
                             }
                         }
 
                         cursor.close();
 
                         LOG.debug( "purged {} messages from the log {}", deleteCount, log.getName() );
                     }
                     catch ( Exception e )
                     {
                         LOG.warn( "Failed to purge old entries from the log {}", log.getName(), e );
                     }
                 }
             }
 
             try
             {
                 synchronized ( lock )
                 {
                     lock.wait( sleepTime );
                 }
             }
             catch ( InterruptedException e )
             {
                 LOG.warn( "ReplicaEventLogJanitor thread was interrupted, processing logs for cleanup", e );
             }
         }
     }
 
 
     public synchronized void removeEventLog( ReplicaEventLog replicaEventLog )
     {
         directoryService.getEventService().removeListener( replicaEventLog.getPersistentListener() );
         String name = replicaEventLog.getName();
         LOG.debug( "removed the persistent listener for replication event log {}", name );
 
         replicaLogMap.remove( replicaEventLog.getId() );
 
         try
         {
             replicaEventLog.stop();
 
             new File( directoryService.getInstanceLayout().getReplDirectory(), name + ".db" ).delete();
             new File( directoryService.getInstanceLayout().getReplDirectory(), name + ".lg" ).delete();
             LOG.info( "successfully removed replication event log {}", name );
         }
         catch ( Exception e )
         {
             LOG.warn(
                 "Closing the replication event log of the entry {} was not successful, will be removed anyway",
                 name, e );
         }
     }
 
 
     public void setSleepTime( long sleepTime )
     {
         this.sleepTime = sleepTime;
     }
 
 
     public long getSleepTime()
     {
         return sleepTime;
     }
 
 
     public void stopCleaning()
     {
         stop = true;
 
         synchronized ( lock )
         {
             lock.notify();
         }
     }
 }