//
// Name
//  $RCSfile: PartitionManager.java,v $
// 
// Copyright
//      Copyright 2010-2012 Cloud Software Group, Inc. ALL RIGHTS RESERVED. 
//      Cloud Software Group, Inc. Confidential Information
//
// History
//  $Revision: 1.1.2.32 $ $Date: 2012/04/25 21:02:03 $
//
package com.kabira.platform.highavailability;

import com.kabira.platform.ManagedObject;
import com.kabira.platform.ManagedClassError;
import com.kabira.platform.ResourceUnavailableException;
import com.kabira.platform.Transaction;
import com.kabira.platform.Transaction.Rollback;
import com.kabira.platform.disteng.DEPartitionManager;
import com.kabira.platform.disteng.DEPartitionNotifier;
import com.kabira.platform.disteng.DEPartition;
import com.kabira.platform.disteng.DEProperties;
import com.kabira.platform.disteng.DEPartitionNotFound;
import com.kabira.platform.disteng.DENodeMismatch;
import com.kabira.platform.disteng.DENotActiveNode;
import com.kabira.platform.disteng.DEEnableAction;
import com.kabira.platform.disteng.DEDisableAction;
import com.kabira.platform.disteng.DEMapperAudit;
import com.kabira.platform.disteng.PartitionState;
import com.kabira.platform.disteng.DEReplicationType;

import java.util.HashSet;
import java.util.logging.LogManager;
import java.util.logging.Logger;
import java.util.logging.Level;

/**
 * The PartitionManager class. This is the main interface for managing
 * partitions. 
 */
public final class PartitionManager 
{
    /**
     * An enumeration of the possible actions that can be taken when
     * enabling high availability for a node.
     */
    public enum EnableAction
    {
        /**
         * Perform any object migrations needed to activate the
         * partitions defined for this node. This option should be used
         * for newly added nodes.
         */
        JOIN_CLUSTER,
        /**
         * Before migrating the partition data, remove any instances that
         * exist on the local node. This option should be used for nodes
         * where the local data doesn't need to be preserved.
         */
        JOIN_CLUSTER_PURGE,
        /**
         * For all partitions where the node joining the cluster is the
         * current active node, access the remote node where the
         * partition should be restored from, and compare the objects
         * between the local node and remote node. If an object on the
         * local node has been updated, or a new object created, push the
         * update to the remote node. If the remote node has been
         * updated, replicate the change to the local node. If a state
         * conflict is detected, or a duplicate key is found, remove the
         * local object and copy the remote node object to the local
         * node.
         * <p>
         * For all partitions where the node joining the cluster is
         * acting as a replica, the replica objects are deleted and
         * instances migrated from the remote node.
         * <p>
         * Warning: This is performance hostile, and should only be done
         * when resolving a split-brain scenario in a cluster.
         * 
         * @see Partition.Properties#restoreFromNode(String)
         */
        JOIN_CLUSTER_RESTORE
    }

    /**
     * An enumeration of the possible actions that can be taken when
     * disabling high availability for a node.
     */
    public enum DisableAction
    {
        /**
         * Perform any object migrations needed to migrate
         * partitions owned by this node to the first order
         * replica node, and remove this node from all replica
         * lists. Before the disable is executed, an audit is
         * done to insure that no partitions will be moved to the
         * UNAVAILABLE state.
         */
        LEAVE_CLUSTER,
        /**
         * Same as LEAVE_CLUSTER, but with audits disabled.
         */
        LEAVE_CLUSTER_FORCE,
    }

    /**
     * Find all partitions on the local node.
     * 
     * @return Array of Partition instances.
     */
    public static final Partition [] getPartitions()
    {
        DEPartitionManager pm = new DEPartitionManager();
        DEPartition [] dp = pm.getPartitions();

        Partition [] pa = new Partition[dp.length];

        for (int i = 0; i < dp.length; i++)
        {
            pa[i] = new Partition(dp[i].name, dp[i].nodes,
                dp[i].replicationTypeList,
                mapState(dp[i].state), dp[i].lastUpdated,
                dp[i].forceReplication, dp[i].objectChunks);
        }
        return pa;
    }

    /**
     * Find a partition by name.
     * 
     * @param partitionName Name of partition to find.
     *
     * @return Partition instance.
     *
     * @exception PartitionNotFound
     *  The given partition doesn't exist.
     */
    public static final Partition getPartition(
        final String partitionName) throws PartitionNotFound
    {
        DEPartitionManager pm = new DEPartitionManager();

        try
        {
            DEPartition p = pm.getPartition(partitionName);
            return new Partition(p.name, p.nodes, p.replicationTypeList,
                mapState(p.state), p.lastUpdated,
                p.forceReplication, p.objectChunks);
        }
        catch (DEPartitionNotFound ex)
        {
            throw new PartitionNotFound(ex.errMsg);
        }
    }

    /**
     * Get the partition an object is in.
     * 
     * @param obj Object to examine.
     *
     * @return Partition instance.
     *
     * @exception PartitionNotFound
     *  The given object isn't in a partition. This can happen if
     *  object instances were created before a {@link PartitionMapper} was
     *  installed for a managed class, or no mapper was ever defined.
     */
    public static final Partition getObjectPartition(
        final Object obj) throws PartitionNotFound
    {
        DEPartitionManager pm = new DEPartitionManager();

        String  partitionName = pm.getObjectPartition(obj);
        if (partitionName.length() == 0)
        {
            throw new PartitionNotFound(
                "No Partition exists for object '" + obj + "'");
        }

        return getPartition(partitionName);
    }

    /**
     * Use {@link #definePartition(String, Partition.Properties, String, ReplicaNode [])}
     */
    @Deprecated
    public static final void definePartition(
        final String partitionName,
        final String [] nodes) throws IllegalArgumentException
    {
        DEPartitionManager pm = new DEPartitionManager();

        if (partitionName.isEmpty())
        {
            throw new IllegalArgumentException("partitionName cannot be empty");
        }
        validateNodeList(nodes);

        DEReplicationType [] rtl = new DEReplicationType[nodes.length];
        for (int i = 0; i < rtl.length; i++)
        {
            rtl[i] = DEReplicationType.DataSynchronous;
        }

        DEProperties props = new DEProperties();
        props.forceReplication = false;
        props.objectChunks = Partition.DefaultObjectsLockedPerTransaction;

        pm.definePartition(partitionName, props, nodes, rtl);
    }

    /**
     * Use {@link #definePartition(String, Partition.Properties, String, ReplicaNode [])}
     * instead.
     */
    @Deprecated
    public static final void definePartition(
        final String partitionName,
        final String restoreFromNode,
        final String [] nodes) throws IllegalArgumentException, NotActiveNode
    {
        DEPartitionManager pm = new DEPartitionManager();

        if (restoreFromNode.isEmpty())
        {
            throw new IllegalArgumentException(
                "restoreFromNode cannot be empty");
        }

        try
        {
            pm.restorePartition(partitionName, restoreFromNode);
        }
        catch (DENotActiveNode ex)
        {
            throw new NotActiveNode(ex.errMsg);
        }

        definePartition(partitionName, nodes);
    }

    /**
     * Define a partition.
     * <p>
     * This method defines a partition. This method can be called on any
     * node in the cluster to define a new partition, or change the   
     * definition of a pre-existing partition. Once all partitions have
     * been defined for a node, the enablePartitions() method should be
     * called to move the partitions to the active state. It is a runtime
     * error to attempt to create or modify instances in a newly defined
     * partition until the enablePartitions() method is called.
     * <p>
     * The activeNode is the node that will be the active node for the
     * partition after it is enabled.
     * <p>
     * Optional partition properties can be used by passing in an
     * instance of the {@link Partition.Properties} class. User defined
     * properties such as forceReplication are saved and subsequently 
     * used during failover processing. 
     * <p>
     * The replicas array contains an ordered list of replica nodes for
     * the partition. A failure of the active node results in the
     * partition automatically migrating to the the next entry in the
     * replicas array. Objects are sent to each replica node based on
     * the configuration in the ReplicaNode entry.
     *
     * @param partitionName Name of partition to create.
     * @param partitionProperties Optional properties for the partition.
     *  If null, the default property values are used.
     * @param activeNode The active node for the partition
     * @param replicas An ordered list of replica nodes for the partition.
     *  Should be passed in as a null instance or a zero length array if
     *  no replicas are desired.
     *
     * @exception IllegalArgumentException
     *  The partitionName or activeNode was empty, or the replicas array
     *  was invalid.
     * @exception NotActiveNode
     *  The current node is not the active node for the partition and 
     *  restoreFromNode was defined in partitionProperties.
     */
    public static final void definePartition(
        final String partitionName,
        final Partition.Properties partitionProperties,
        final String activeNode,
        final ReplicaNode [] replicas)
            throws IllegalArgumentException, NotActiveNode
    {
        DEPartitionManager pm = new DEPartitionManager();

        if (partitionName.isEmpty())
        {
            throw new IllegalArgumentException("partitionName cannot be empty");
        }

        if (activeNode.isEmpty())
        {
            throw new IllegalArgumentException("activeNode cannot be empty");
        }

        validateReplicaList(replicas);

        int nodeLength = (replicas == null) ? 1 : replicas.length + 1;

        String [] nodes = new String[nodeLength];

        nodes[0] = activeNode;
        if (replicas != null)
        {
            for (int i = 0; i < replicas.length; i++)
            {
                assert( i + 1 < nodeLength );
                nodes[i + 1] = replicas[i].nodeName;
            }
        }
        validateNodeList(nodes);

        DEReplicationType [] rtl = new DEReplicationType[nodes.length];
        rtl[0] = DEReplicationType.DataSynchronous;
        if (replicas != null)
        {
            for (int i = 0; i < replicas.length; i++)
            {
                rtl[i + 1] = mapReplicationType(replicas[i].replicationType);
            }
        }

        if (partitionProperties != null &&
            partitionProperties.m_restoreFromNode != null)
        {
            try
            {
                pm.restorePartition(partitionName,
                    partitionProperties.m_restoreFromNode);
            }
            catch (DENotActiveNode ex)
            {
                throw new NotActiveNode(ex.errMsg);
            }
        }

        DEProperties props = new DEProperties();
        if (partitionProperties == null)
        {
            props.forceReplication = false;
            props.objectChunks = Partition.DefaultObjectsLockedPerTransaction;
        }
        else
        {
            props.forceReplication = partitionProperties.m_forceReplication;
            props.objectChunks =
                partitionProperties.m_objectsLockedPerTransaction;
        }

        pm.definePartition(partitionName, props, nodes, rtl);
    }

    /**
     * Associate a mapper to a given managed class.
     *
     * This method uses {@link PartitionMapper.Properties.Audit#IGNORE_PARTITIONING}
     * for audit.
     * 
     * @param managedClass Name of a Managed object.
     * @param partitionMapper User defined PartitionMapper instance.
     *
     * @exception ManagedClassError
     *  The managedClass is not a Managed object, or has a non-abstract
     *  parent class with a key defined but no mapper installed.
     *
     * @see PartitionManager#setMapper(Class, PartitionMapper, PartitionMapper.Properties)
     */
    public static final void setMapper(
        final Class managedClass,
        final PartitionMapper partitionMapper) throws ManagedClassError
    {
        setMapper(managedClass, partitionMapper, null);
    }

    /**
     * Associate a mapper to a given managed class.
     * <p>
     * This method is used to associate a user defined PartitionMapper
     * instance with a managed class. All managed types that are replicated 
     * must have a PartitionMapper associated with the class. This method must
     * be called on all nodes that has the partition definition.
     * <p>
     * The same PartitionMapper instance can be associated with multiple
     * classes. If a PartitionMapper is already associated with a class,
     * the new value overwrites the previous one.
     * <p>
     * Applications should only delete a PartitionMapper instance after a
     * new instance is installed. If a PartitionMapper instance is
     * deleted, any subsequent object creates for the class will throw a
     * runtime exception.
     * <p>
     * Mapper instances are inherited by all classes that extend the
     * managed class. If the managed class extends another class, that
     * class must not have any keys defined.
     * <p>
     * If auditing is enabled in the mapperProperties instance, this routine
     * will verify that no unpartitioned instances of the given class
     * already exist in shared memory. This can happen if instances are
     * created before a mapper is installed.
     * 
     * @param managedClass Name of a Managed object.
     * @param partitionMapper User defined PartitionMapper instance.
     * @param mapperProperties Optional mapper properties. If null,
     * {@link PartitionMapper.Properties.Audit#IGNORE_PARTITIONING} is used.
     *
     * @exception ManagedClassError
     *  The managedClass is not a Managed object, or has a non-abstract
     *  parent class with a key defined but no mapper installed.
     */
    public static final void setMapper(
        final Class managedClass,
        final PartitionMapper partitionMapper,
        final PartitionMapper.Properties mapperProperties)
            throws ManagedClassError
    {
        DEPartitionManager pm = new DEPartitionManager();
        try
        {

            DEMapperAudit audit = (mapperProperties == null)
                ? DEMapperAudit.MapperIgnore
                : mapAudit(mapperProperties.getAudit());
            pm.setMapperAudit(managedClass.getName(), partitionMapper, audit);
        }
        catch (ResourceUnavailableException ex)
        {
            throw new ManagedClassError(ex.getMessage());
        }
    }

    /**
     * Determine if a class has a partition mapper installed.
     * <p>
     * This method determines if a user defined PartitionMapper
     * instance is associated with a managed class.
     * 
     * @param managedClass Name of a Managed object.
     *
     * @return True if class has a mapper instance installed, otherwise false.
     *
     * @exception ManagedClassError
     *  The managedClass is not a Managed object.
     */
    public static final boolean isPartitioned(final Class managedClass)
    {
        DEPartitionManager pm = new DEPartitionManager();
        try
        {
            return pm.isPartitioned(managedClass.getName());
        }
        catch (ResourceUnavailableException ex)
        {
            throw new ManagedClassError(ex.getMessage());
        }
    }

    /**
     * Enable high availability for the all partitions defined on the node.
     * <p>
     * This method is used to enable all partitions that have been
     * defined for this node. This method should be called each time new
     * partitions are defined.
     *
     * @param enableAction The action to take when enabling partitions.
     *
     * @exception ResourceUnavailableException
     *  The minimum number of nodes needed to establish a quorum has not been
     *  seen.
     */
    public static final void enablePartitions(
        final EnableAction enableAction) throws ResourceUnavailableException
    {
        DEPartitionManager pm = new DEPartitionManager();
        pm.enablePartitions(mapEnable(enableAction));
    }

    /**
     * Disable high availability for the all partitions defined on the node.
     * <p>
     * This method is used to remove the local node from the node list of
     * all partitions that have been defined for this node. If called
     * multiple times, the additional calls have no effect.
     *
     * @param disableAction  The action to take when disabling partitions.
     *
     * @exception ResourceUnavailableException
     *  One or more partitions would be in the UNAVAILABLE state after the
     *  disable executes, not thrown if LEAVE_CLUSTER_FORCE is used.
     */
    public static final void disablePartitions(
        final DisableAction disableAction) throws ResourceUnavailableException
    {
        DEPartitionManager pm = new DEPartitionManager();
        pm.disablePartitions(mapDisable(disableAction));
    }

    /**
     * Wait for a remote node to become available.
     *
     * @param remoteNode name of remote node to wait for.
     *
     * @exception ResourceUnavailableException
     *  The operation timed out waiting for the remote node to become
     *  available.
     */
    public static final void waitForNode(
        final String remoteNode) throws ResourceUnavailableException
    {
        DEPartitionManager pm = new DEPartitionManager();
        pm.waitForNode(remoteNode);
    }

    //
    // Package private methods. These don't bother with handling
    // PartitionNotFound exceptions, since they can only be called after
    // getPartition() was executed, and we never delete partitions.
    //

    static final void updatePartition(
        final String existingPartition,
        DEProperties props) throws NodeMismatch, NotActiveNode
    {
        DEPartitionManager pm = new DEPartitionManager();
        try
        {
            pm.updatePartition(existingPartition, props);
        }
        catch (DENodeMismatch ex)
        {
            throw new NodeMismatch(ex.errMsg);
        }
        catch (DENotActiveNode ex)
        {
            throw new NotActiveNode(ex.errMsg);
        }
    }

    static final void migratePartition(
        final String partitionName,
        DEProperties props,
        final String [] nodes,
        DEReplicationType [] rtl) throws NotActiveNode
    {
        DEPartitionManager pm = new DEPartitionManager();
        try
        {
            pm.migratePartition(partitionName, props, nodes, rtl);
        }
        catch (DENotActiveNode ex)
        {
            throw new NotActiveNode(ex.errMsg);
        }
    }

    static final void setNotifier(
        final String partitionName,
        final PartitionNotifier partitionNotifier) throws PartitionNotFound
    {
        Notifier notifier = new Notifier(partitionNotifier);
        DEPartitionManager pm = new DEPartitionManager();
        pm.setNotifier(partitionName, notifier);
    }

    //
    // Validate the replica list. We just check for null replicas here, the
    // rest of the checks are done in validateNodeList()
    //
    static void validateReplicaList(
        final ReplicaNode [] replicas) throws IllegalArgumentException
    {
        if (replicas != null)
        {
            for (int i = 0; i < replicas.length; i++)
            {
                ReplicaNode r = replicas[i];
                if (r == null)
                {
                    throw new IllegalArgumentException(
                        "Replica entry at index " + i + " cannot be empty");
                }
            }
        }
    }

    //
    // Validate the node list
    //
    static void validateNodeList(
        final String [] nodes) throws IllegalArgumentException
    {
        HashSet<String>     nodeSet = new HashSet<String>();
        if (nodes.length == 0)
        {
            throw new IllegalArgumentException(
                "The node array must have at least one entry");
        }
        for (int i = 0; i < nodes.length; i++)
        {
            String node = nodes[i];
            if (node == null || node.isEmpty())
            {
                throw new IllegalArgumentException("Node entry at index " +
                    i + " cannot be empty");
            }
            if (nodeSet.contains(node))
            {
                throw new IllegalArgumentException("Node entry " + node +
                    " at index " + i + " is a duplicate. " +
                    "All nodes must be unique");
            }
            nodeSet.add(node);
        }
    }

    //
    // Wrapper class to map notifiers.
    //
    private static class Notifier extends DEPartitionNotifier
    {
        public void stateChange(
                final String partitionName,
                final PartitionState oldState,
                final PartitionState newState)
        {
            if (ManagedObject.isEmpty(m_partitionNotifier))
            {
                ManagedObject.delete(this);
            }
            else
            {
                m_partitionNotifier.stateChange(
                    partitionName,
                    mapState(oldState),
                    mapState(newState));
            }
        }

        Notifier(PartitionNotifier partitionNotifier)
        {
            m_partitionNotifier = partitionNotifier;
        }

        PartitionNotifier   m_partitionNotifier;
    }

    //
    // Utility function to map PartitionState to Partition.State
    //
    private static Partition.State mapState(PartitionState state)
    {
        Partition.State s;

        switch (state)
        {
        case PartitionInitial:
            s = Partition.State.INITIAL;
            break;
        case PartitionActive:
            s = Partition.State.ACTIVE;
            break;
        case PartitionUpdating:
            s = Partition.State.UPDATING;
            break;
        case PartitionMigrating:
            s = Partition.State.MIGRATING;
            break;
        case PartitionReplicating:
            s = Partition.State.REPLICATING;
            break;
        default:
            assert( state == PartitionState.PartitionUnavailable );
            s = Partition.State.UNAVAILABLE;
            break;
        }
        return s;
    }

    //
    // Utility function to map PartitionManager.EnableAction to EnableAction
    //
    private static DEEnableAction mapEnable(
        final PartitionManager.EnableAction enableAction)
    {
        DEEnableAction e;

        switch (enableAction)
        {
        case JOIN_CLUSTER:
            e = DEEnableAction.JoinCluster;
            break;
        case JOIN_CLUSTER_PURGE:
            e = DEEnableAction.JoinClusterPurge;
            break;
        default:
            assert( enableAction == EnableAction.JOIN_CLUSTER_RESTORE );
            e = DEEnableAction.JoinClusterRestore;
            break;
        }
        return e;
    }

    //
    // Utility function to map PartitionManager.DisableAction to DisableAction
    //
    private static DEDisableAction mapDisable(
        final PartitionManager.DisableAction disableAction)
    {
        DEDisableAction e;

        switch (disableAction)
        {
        case LEAVE_CLUSTER:
            e = DEDisableAction.LeaveCluster;
            break;
        default:
            assert( disableAction == DisableAction.LEAVE_CLUSTER_FORCE );
            e = DEDisableAction.LeaveClusterForce;
            break;
        }
        return e;
    }

    //
    // Utility function to map ReplicaNode.ReplicationType to DEReplicationType
    //
    static DEReplicationType mapReplicationType(
        final ReplicaNode.ReplicationType replicationType)
    {
        DEReplicationType rt;

        switch (replicationType)
        {
        case SYNCHRONOUS:
            rt = DEReplicationType.DataSynchronous;
            break;
        default:
            assert( replicationType ==
                ReplicaNode.ReplicationType.ASYNCHRONOUS );
            rt = DEReplicationType.DataAsynchronous;
            break;
        }
        return rt;
    }

    //
    // Utility function to map DEReplicationType to ReplicaNode.ReplicationType
    //
    static ReplicaNode.ReplicationType mapDEReplicationType(
        final DEReplicationType rt)
    {
        ReplicaNode.ReplicationType replicationType;

        switch (rt)
        {
        case DataSynchronous:
            replicationType = ReplicaNode.ReplicationType.SYNCHRONOUS;
            break;
        default:
            assert( rt == DEReplicationType.DataAsynchronous );
            replicationType = ReplicaNode.ReplicationType.ASYNCHRONOUS;
            break;
        }
        return replicationType;
    }

    //
    // Utility function to map MapperProperties.Audit to DEMapperAudit
    //
    private static DEMapperAudit mapAudit(
        final PartitionMapper.Properties.Audit audit)
    {
        DEMapperAudit e;

        switch (audit)
        {
        case VERIFY_PARTIONING:
            e = DEMapperAudit.MapperVerify;
            break;
        default:
            assert( audit == PartitionMapper.Properties.Audit.IGNORE_PARTITIONING );
            e = DEMapperAudit.MapperIgnore;
            break;
        }
        return e;
    }

    //
    // Never should be called
    //
    private PartitionManager()
    {
    }

    //
    // Private logger class
    //
    private static class DistributionLogger extends Logger
    {
        DistributionLogger()
        {
            super("com.kabira.platform.highavailability", null);
            m_enabled = false;
        }

        public void setLevel(Level level)
        {
            super.setLevel(level);
            if (level != null)
            {
                m_enabled = (level.intValue() <= Level.FINEST.intValue())
                    ? true : false;
                if (Transaction.isActive())
                {
                    debugTrace();
                }
                else
                {
                    new Transaction("DistributionLogger")
                    {
                        @Override
                        protected void run() throws Rollback
                        {
                            debugTrace();
                        }
                    }.execute();
                }
            }
        }

        private boolean m_enabled;

        private void debugTrace()
        {
                DEPartitionManager pm = new DEPartitionManager();
                pm.debugTrace(m_enabled);
        }

        @Override
        public String toString()
        {
            return "DistributionLogger";
        }
    }

    //
    // Create and register logger at start of world.
    //
    private static DistributionLogger    m_logger;

    static
    {
        m_logger = new DistributionLogger();
        LogManager lm = LogManager.getLogManager();
        lm.addLogger(m_logger);
    }
}