/*
    *  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.core.partition.impl.btree.jdbm;
  
  
  import java.io.File;
  import java.io.IOException;
  import java.net.URI;
  
  import jdbm.RecordManager;
  import jdbm.helper.ByteArraySerializer;
  
  import org.apache.directory.api.ldap.model.cursor.Cursor;
  import org.apache.directory.api.ldap.model.cursor.CursorException;
  import org.apache.directory.api.ldap.model.cursor.EmptyCursor;
  import org.apache.directory.api.ldap.model.cursor.Tuple;
  import org.apache.directory.api.ldap.model.exception.LdapException;
  import org.apache.directory.api.ldap.model.exception.LdapOtherException;
  import org.apache.directory.api.ldap.model.schema.AttributeType;
  import org.apache.directory.api.ldap.model.schema.MatchingRule;
  import org.apache.directory.api.ldap.model.schema.SchemaManager;
  import org.apache.directory.api.ldap.model.schema.comparators.SerializableComparator;
  import org.apache.directory.api.ldap.model.schema.comparators.UuidComparator;
  import org.apache.directory.server.core.api.partition.PartitionTxn;
  import org.apache.directory.server.core.partition.impl.btree.IndexCursorAdaptor;
  import org.apache.directory.server.i18n.I18n;
  import org.apache.directory.server.xdbm.AbstractIndex;
  import org.apache.directory.server.xdbm.IndexEntry;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  
  /**
   * A Jdbm based index implementation. It creates an Index for a give AttributeType.
   *
   * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
   */
  public class JdbmIndex<K> extends AbstractIndex<K, String>
  {
      /** A logger for this class */
      private static final Logger LOG = LoggerFactory.getLogger( JdbmIndex.class );
  
      /** default duplicate limit before duplicate keys switch to using a btree for values */
      public static final int DEFAULT_DUPLICATE_LIMIT = 512;
  
      /**  the key used for the forward btree name */
      public static final String FORWARD_BTREE = "_forward";
  
      /**  the key used for the reverse btree name */
      public static final String REVERSE_BTREE = "_reverse";
  
      /**
       * the forward btree where the btree key is the value of the indexed attribute and
       * the value of the btree is the entry id of the entry containing an attribute with
       * that value
       */
      protected JdbmTable<K, String> forward;
  
      /**
       * the reverse btree where the btree key is the entry id of the entry containing a
       * value for the indexed attribute, and the btree value is the value of the indexed
       * attribute
       */
      protected JdbmTable<String, K> reverse;
  
      /**
       * the JDBM record manager for the file containing this index
       */
      protected RecordManager recMan;
  
      /**
       * duplicate limit before duplicate keys switch to using a btree for values
       */
      protected int numDupLimit = DEFAULT_DUPLICATE_LIMIT;
  
      /** a custom working directory path when specified in configuration */
      protected File wkDirPath;
  
  
      /*
       * NOTE: Duplicate Key Limit
       *
      * Jdbm cannot store duplicate keys: meaning it cannot have more than one value
      * for the same key in the btree.  Thus as a workaround we stuff values for the
      * same key into a TreeSet.  This is only effective up to some threshold after
      * which we run into problems with serialization on and off disk.  A threshold
      * is used to determine when to switch from using a TreeSet to start using another
      * btree in the same index file just for the values.  This value only btree just
      * has keys populated without a value for it's btree entries. When the switch
      * occurs the value for the key in the index btree contains a pointer to the
      * btree containing it's values.
      *
      * This numDupLimit is the threshold at which we switch from using in memory
      * containers for values of the same key to using a btree for those values
      * instead with indirection.
      */
 
     // ------------------------------------------------------------------------
     // C O N S T R U C T O R S
     // ----------------------------------------------------------------------
     /**
      * Creates a JdbmIndex instance for a give AttributeId
      * 
      * @param attributeId The Attribute ID
      * @param withReverse If we have to create a reverse index
      */
     public JdbmIndex( String attributeId, boolean withReverse )
     {
         super( attributeId, withReverse );
 
         initialized = false;
     }
 
 
     /**
      * Initialize the index for an Attribute, with a specific working directory (may be null).
      * 
      * @param recMan The RecordManager
      * @param schemaManager The schemaManager to use to get back the Attribute
      * @param attributeType The attributeType this index is created for
      * @throws LdapException If the initialization failed
      * @throws IOException If the initialization failed
      */
     public void init( RecordManager recMan, SchemaManager schemaManager, AttributeType attributeType ) 
             throws LdapException, IOException
     {
         LOG.debug( "Initializing an Index for attribute '{}'", attributeType.getName() );
 
         this.attributeType = attributeType;
 
         if ( attributeId == null )
         {
             setAttributeId( attributeType.getName() );
         }
 
         // see DIRSERVER-2002
         // prevent the OOM when more than 50k users are loaded at a stretch
         // adding this system property to make it configurable till JDBM gets replaced by Mavibot
         String cacheSizeVal = System.getProperty( "jdbm.recman.cache.size", "100" );
         
         int recCacheSize = Integer.parseInt( cacheSizeVal );
         
         LOG.info( "Setting CacheRecondManager's cache size to {}", recCacheSize );
         
         this.recMan = recMan;
 
         try
         {
             initTables( schemaManager );
         }
         catch ( IOException e )
         {
             // clean up
             close( null );
             throw e;
         }
 
         initialized = true;
     }
 
 
     /**
      * Initializes the forward and reverse tables used by this Index.
      * 
      * @param schemaManager The server schemaManager
      * @throws IOException if we cannot initialize the forward and reverse
      * tables
      */
     private void initTables( SchemaManager schemaManager ) throws IOException
     {
         SerializableComparator<K> comp;
 
         MatchingRule mr = attributeType.getEquality();
 
         if ( mr == null )
         {
             throw new IOException( I18n.err( I18n.ERR_574, attributeType.getName() ) );
         }
 
         comp = new SerializableComparator<>( mr.getOid() );
 
         /*
          * The forward key/value map stores attribute values to master table
          * primary keys.  A value for an attribute can occur several times in
          * different entries so the forward map can have more than one value.
          */
         UuidComparator.INSTANCE.setSchemaManager( schemaManager );
         comp.setSchemaManager( schemaManager );
 
         if ( mr.getSyntax().isHumanReadable() )
         {
             forward = new JdbmTable<>( schemaManager, attributeType.getOid() + FORWARD_BTREE, numDupLimit,
                 recMan,
                 comp, UuidComparator.INSTANCE, StringSerializer.INSTANCE, UuidSerializer.INSTANCE );
         }
         else
         {
             forward = new JdbmTable<>( schemaManager, attributeType.getOid() + FORWARD_BTREE, numDupLimit,
                 recMan,
                 comp, UuidComparator.INSTANCE, new ByteArraySerializer(), UuidSerializer.INSTANCE );
         }
 
         /*
          * Now the reverse map stores the primary key into the master table as
          * the key and the values of attributes as the value.  If an attribute
          * is single valued according to its specification based on a schema
          * then duplicate keys should not be allowed within the reverse table.
          */
         if ( withReverse )
         {
             if ( attributeType.isSingleValued() )
             {
                 reverse = new JdbmTable<>( schemaManager, attributeType.getOid() + REVERSE_BTREE, recMan,
                     UuidComparator.INSTANCE, UuidSerializer.INSTANCE, null );
             }
             else
             {
                 reverse = new JdbmTable<>( schemaManager, attributeType.getOid() + REVERSE_BTREE, numDupLimit,
                     recMan,
                     UuidComparator.INSTANCE, comp, UuidSerializer.INSTANCE, null );
             }
         }
     }
 
 
     // ------------------------------------------------------------------------
     // C O N F I G U R A T I O N   M E T H O D S
     // ------------------------------------------------------------------------
     /**
      * Gets the threshold at which point duplicate keys use btree indirection to store
      * their values.
      *
      * @return the threshold for storing a keys values in another btree
      */
     public int getNumDupLimit()
     {
         return numDupLimit;
     }
 
 
     /**
      * Sets the threshold at which point duplicate keys use btree indirection to store
      * their values.
      *
      * @param numDupLimit the threshold for storing a keys values in another btree
      */
     public void setNumDupLimit( int numDupLimit )
     {
         protect( "numDupLimit" );
         this.numDupLimit = numDupLimit;
     }
 
 
     /**
      * Sets the working directory path to something other than the default. Sometimes more
      * performance is gained by locating indices on separate disk spindles.
      *
      * @param wkDirPath optional working directory path
      */
     public void setWkDirPath( URI wkDirPath )
     {
         protect( "wkDirPath" );
         this.wkDirPath = new File( wkDirPath );
     }
 
 
     /**
      * Gets the working directory path to something other than the default. Sometimes more
      * performance is gained by locating indices on separate disk spindles.
      *
      * @return optional working directory path
      */
     public URI getWkDirPath()
     {
         return wkDirPath != null ? wkDirPath.toURI() : null;
     }
 
 
     // ------------------------------------------------------------------------
     // Scan Count Methods
     // ------------------------------------------------------------------------
     /**
      * {@inheritDoc}
      */
     public long count( PartitionTxn partitionTxn ) throws LdapException
     {
         return forward.count( partitionTxn );
     }
 
 
     /**
      * {@inheritDoc}
      */
     public long count( PartitionTxn partitionTxn, K attrVal ) throws LdapException
     {
         return forward.count( partitionTxn, attrVal );
     }
 
 
     /**
      * {@inheritDoc}
      */
     @Override
     public long greaterThanCount( PartitionTxn partitionTxn, K attrVal ) throws LdapException
     {
         return forward.greaterThanCount( partitionTxn, attrVal );
     }
 
 
     /**
      * {@inheritDoc}
      */
     @Override
     public long lessThanCount( PartitionTxn partitionTxn, K attrVal ) throws LdapException
     {
         return forward.lessThanCount( partitionTxn, attrVal );
     }
 
 
     // ------------------------------------------------------------------------
     // Forward and Reverse Lookups
     // ------------------------------------------------------------------------
 
     /**
      * {@inheritDoc}
      */
     public String forwardLookup( PartitionTxn partitionTxn, K attrVal ) throws LdapException
     {
         return forward.get( partitionTxn, attrVal );
     }
 
 
     /**
      * {@inheritDoc}
      */
     public K reverseLookup( PartitionTxn partitionTxn, String id ) throws LdapException
     {
         if ( withReverse )
         {
             return reverse.get( partitionTxn, id );
         }
         else
         {
             return null;
         }
     }
 
 
     // ------------------------------------------------------------------------
     // Add/Drop Methods
     // ------------------------------------------------------------------------
 
     /**
      * {@inheritDoc}
      */
     public synchronized void add( PartitionTxn partitionTxn,  K attrVal, String id ) throws LdapException
     {
         // The pair to be added must exists
         forward.put( partitionTxn, attrVal, id );
 
         if ( withReverse )
         {
             reverse.put( partitionTxn, id, attrVal );
         }
     }
 
 
     /**
      * {@inheritDoc}
      */
     public synchronized void drop( PartitionTxn partitionTxn, K attrVal, String id ) throws LdapException
     {
         // The pair to be removed must exists
         if ( forward.has( partitionTxn, attrVal, id ) )
         {
             forward.remove( partitionTxn, attrVal, id );
 
             if ( withReverse )
             {
                 reverse.remove( partitionTxn, id, attrVal );
             }
         }
     }
 
 
     /**
      * {@inheritDoc}
      */
     public void drop( PartitionTxn partitionTxn, String entryId ) throws LdapException
     {
         if ( withReverse )
         {
             if ( isDupsEnabled() )
             {
                 // Build a cursor to iterate on all the keys referencing
                 // this entryId
                 Cursor<Tuple<String, K>> values = reverse.cursor( partitionTxn, entryId );
 
                 try
                 {
                     while ( values.next() )
                     {
                         // Remove the Key -> entryId from the index
                         forward.remove( partitionTxn, values.get().getValue(), entryId );
                     }
     
                     values.close();
                 }
                 catch ( CursorException | IOException e )
                 {
                     throw new LdapOtherException( e.getMessage(), e );
                 }
             }
             else
             {
                 K key = reverse.get( partitionTxn, entryId );
 
                 forward.remove( partitionTxn, key );
             }
 
             // Remove the id -> key from the reverse index
             reverse.remove( partitionTxn, entryId );
         }
     }
 
 
     // ------------------------------------------------------------------------
     // Index Cursor Operations
     // ------------------------------------------------------------------------
     @SuppressWarnings("unchecked")
     public Cursor<IndexEntry<K, String>> forwardCursor( PartitionTxn partitionTxn ) throws LdapException
     {
         return new IndexCursorAdaptor<>( partitionTxn, ( Cursor ) forward.cursor(), true );
     }
 
 
     public Cursor<IndexEntry<K, String>> forwardCursor( PartitionTxn partitionTxn, K key ) throws LdapException
     {
         return new IndexCursorAdaptor<>( partitionTxn, ( Cursor ) forward.cursor( partitionTxn, key ), true );
     }
 
 
     /**
      * {@inheritDoc}
      */
     @Override
     public Cursor<K> reverseValueCursor( PartitionTxn partitionTxn, String id ) throws LdapException
     {
         if ( withReverse )
         {
             return reverse.valueCursor( partitionTxn, id );
         }
         else
         {
             return new EmptyCursor<>();
         }
     }
 
 
     public Cursor<String> forwardValueCursor( PartitionTxn partitionTxn, K key ) throws LdapException
     {
         return forward.valueCursor( partitionTxn, key );
     }
 
 
     // ------------------------------------------------------------------------
     // Value Assertion (a.k.a Index Lookup) Methods //
     // ------------------------------------------------------------------------
     /**
      * {@inheritDoc}
      */
     public boolean forward( PartitionTxn partitionTxn, K attrVal ) throws LdapException
     {
         return forward.has( partitionTxn, attrVal );
     }
 
 
     /**
      * {@inheritDoc}
      */
     public boolean forward( PartitionTxn partitionTxn, K attrVal, String id ) throws LdapException
     {
         return forward.has( partitionTxn, attrVal, id );
     }
 
 
     /**
      * {@inheritDoc}
      */
     public boolean reverse( PartitionTxn partitionTxn, String id ) throws LdapException
     {
         if ( withReverse )
         {
             return reverse.has( partitionTxn, id );
         }
         else
         {
             return false;
         }
     }
 
 
     /**
      * {@inheritDoc}
      */
     public boolean reverse( PartitionTxn partitionTxn, String id, K attrVal ) throws LdapException
     {
         return forward.has( partitionTxn, attrVal, id );
     }
 
 
     // ------------------------------------------------------------------------
     // Maintenance Methods
     // ------------------------------------------------------------------------
     /**
      * {@inheritDoc}
      */
     @Override
     public synchronized void close( PartitionTxn partitionTxn ) throws LdapException, IOException
     {
         if ( forward != null )
         {
             forward.close( partitionTxn );
         }
 
         if ( reverse != null )
         {
             reverse.close( partitionTxn );
         }
     }
 
     
     /**
      * {@inheritDoc}
      */
     @Override
     public boolean isDupsEnabled()
     {
         if ( withReverse )
         {
             return reverse.isDupsEnabled();
         }
         else
         {
             return false;
         }
     }
 
 
     /**
      * @see Object#toString()
      */
     public String toString()
     {
         return "Index<" + attributeId + ">";
     }
 }