/*
 * Copyright (C) 2003-2010 eXo Platform SAS.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Affero General Public License
 * as published by the Free Software Foundation; either version 3
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, see<http://www.gnu.org/licenses/>.
 */
package org.exoplatform.services.jcr.impl.core.query;

import org.exoplatform.services.jcr.config.RepositoryConfigurationException;
import org.exoplatform.services.jcr.impl.core.query.lucene.OfflinePersistentIndex;
import org.exoplatform.services.jcr.impl.util.io.DirectoryHelper;
import org.exoplatform.services.log.ExoLogger;
import org.exoplatform.services.log.Log;
import org.exoplatform.services.rpc.RPCException;
import org.exoplatform.services.rpc.RPCService;
import org.exoplatform.services.rpc.RemoteCommand;
import org.exoplatform.services.rpc.TopologyChangeEvent;
import org.exoplatform.services.rpc.TopologyChangeListener;
import org.exoplatform.services.rpc.jgv3.RPCServiceImpl;

import org.apache.commons.lang3.StringUtils;
import org.jboss.util.file.Files;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
import java.io.Serializable;
import java.util.*;

import javax.jcr.RepositoryException;

/**
 * Created by The eXo Platform SAS.
 *
 * Date: 16.02.2011
 * 
 * @author <a href="mailto:anatoliy.bazko@exoplatform.com.ua">Anatoliy Bazko</a>
 * @version $Id: IndexRetrievalImpl.java 34360 2010-11-11 11:11:11Z tolusha $
 */
public class IndexRecoveryImpl implements IndexRecovery, TopologyChangeListener
{

   /**
    * Logger instance for this class.
    */
   private static final Log log = ExoLogger.getLogger("exo.jcr.component.core.IndexRecoveryImpl");

   /**
    * Buffer size.
    */
   public static final int BUFFER_SIZE = 1024 * 1024;

   /**
    * The service for executing commands on all nodes of cluster.
    */
   protected final RPCService rpcService;

   /**
    * Remote command responsible for getting the list of relative paths of all files from index directory.
    */
   private RemoteCommand getIndexList;


   /**
    * Remote command responsible for getting the coordinator address.
    */
   private RemoteCommand getCoordinatorAddress;

   /**
    * Remote command responsible for getting data of target file.
    */
   private RemoteCommand getIndexFile;

   /**
    * Remote command responsible for getting the whole workspace index file
    */
   private RemoteCommand getAllIndexes;
   
   /**
    * Create one single indexes zip file to prepare it to be transfered 
    */
   private RemoteCommand prepareIndexesForCopy;
   
   /**
    * Deletes indexes zip file to prepare it to be transfered 
    */
   private RemoteCommand cleanupIndexesAfterCopy;
   
   /**
    * List of compressed index folders to send to coordinators
    */
   private List<File> indexesZipFiles = Collections.synchronizedList(new ArrayList<>());

   /**
    * Remote command to switch index between RO/RW state.
    */
   private RemoteCommand changeIndexMode;

   /**
    * Remote command to check if index can be retrieved.
    */
   private RemoteCommand checkIndexReady;

   /**
    * Remote command to check if node responsible for set index online leave the cluster. 
    */
   private RemoteCommand requestForResponsibleToSetIndexOnline;

   /**
    * Indicates that node keep responsible to set index online.
    */
   protected Boolean isResponsibleToSetIndexOnline = false;

   /**
    * Indicates whether current node is in online or offline mode
    */
   protected Boolean isOnline = true;

   protected final SearchManager searchManager;

   /**
    * Constructor IndexRetrieveImpl.
    * 
    * @throws RepositoryConfigurationException 
    */
   public IndexRecoveryImpl(RPCService rpcService, final SearchManager searchManager)
      throws RepositoryConfigurationException
   {
      this.rpcService = rpcService;
      this.searchManager = searchManager;

      final String commandSuffix = searchManager.getWsId() + "-" + (searchManager.parentSearchManager == null);
      final File indexDirectory = searchManager.getIndexDirectory();

      changeIndexMode = rpcService.registerCommand(new RemoteCommand()
      {
         public String getId()
         {
            return "org.exoplatform.services.jcr.impl.core.query.IndexRecoveryImpl-changeIndexMode-" + commandSuffix;
         }

         public Serializable execute(Serializable[] args) throws Throwable
         {
            boolean isOnline = (Boolean)args[0];
            searchManager.setOnline(isOnline, false, false);
            IndexRecoveryImpl.this.isOnline = isOnline;
            return null;
         }
      });

      getIndexList = rpcService.registerCommand(new RemoteCommand()
      {
         public String getId()
         {
            return "org.exoplatform.services.jcr.impl.core.query.IndexRecoveryImpl-getIndexList-" + commandSuffix;
         }

         public Serializable execute(Serializable[] args) throws Throwable
         {
            int indexDirLen = indexDirectory.getAbsolutePath().length();

            ArrayList<String> result = new ArrayList<String>();
            for (File file : DirectoryHelper.listFiles(indexDirectory))
            {
               if (!file.isDirectory())
               {
                  // if parent directory is not "offline" then add this file. Otherwise skip it.
                  if (!file.getParent().endsWith(OfflinePersistentIndex.NAME))
                  {
                     result.add(file.getAbsolutePath().substring(indexDirLen));
                  }
               }
            }
            return result;
         }
      });

      getIndexFile = rpcService.registerCommand(new RemoteCommand()
      {
         public String getId()
         {
            return "org.exoplatform.services.jcr.impl.core.query.IndexRecoveryImpl-getIndexFile-" + commandSuffix;
         }

         public Serializable execute(Serializable[] args) throws Throwable
         {
            String filePath = (String)args[0];
            long offset = (Long)args[1];

            RandomAccessFile file = new RandomAccessFile(new File(indexDirectory, filePath), "r");
            try
            {
               file.seek(offset);
               byte[] buffer = new byte[BUFFER_SIZE];
               int len = file.read(buffer);
               if (len == -1)
               {
                  return null;
               }
               else
               {
                  byte[] data = new byte[len];
                  System.arraycopy(buffer, 0, data, 0, len);

                  return data;
               }
            }
            finally
            {
               try
               {
                  file.close();
               }
               catch (IOException e)
               {
                  log.debug("Could not close the file", e);
               }
            }
         }
      });

      prepareIndexesForCopy = rpcService.registerCommand(new RemoteCommand() {
        public String getId() {
          return "org.exoplatform.services.jcr.impl.core.query.IndexRecoveryImpl-prepareIndexesForCopy-" + commandSuffix;
        }
  
        public Serializable execute(Serializable[] args) throws Throwable {
          File indexesZipFile = File.createTempFile("jcr", "indexes");
          DirectoryHelper.compressDirectory(indexDirectory, indexesZipFile);
          indexesZipFiles.add(indexesZipFile);
          return indexesZipFile.getName();
        }
      });

      cleanupIndexesAfterCopy = rpcService.registerCommand(new RemoteCommand() {
        public String getId() {
          return "org.exoplatform.services.jcr.impl.core.query.IndexRecoveryImpl-cleanupIndexesAfterCopy-" + commandSuffix;
        }
  
        public Serializable execute(Serializable[] args) throws Throwable {
          String fileName = (String) args[0];
          File indexesZipFile = indexesZipFiles.stream().filter(file -> file.getName().equals(fileName)).findFirst().orElse(null);
          if (indexesZipFile == null || !indexesZipFile.exists()) {
            return false;
          }
          try {
            if (!Files.delete(indexesZipFile)) {
              indexesZipFile.deleteOnExit();
            }
            return true;
          } catch (Exception e) {
            log.debug("Could not delete file {}", indexesZipFile, e);
            indexesZipFile.deleteOnExit();
          } finally {
            indexesZipFiles.remove(indexesZipFile);
          }
          return false;
        }
      });

      getAllIndexes = rpcService.registerCommand(new RemoteCommand() {
        public String getId() {
          return "org.exoplatform.services.jcr.impl.core.query.IndexRecoveryImpl-getAllIndexes-" + commandSuffix;
        }
  
        public Serializable execute(Serializable[] args) throws Throwable {
          String fileName = (String) args[0];
          long offset = (Long) args[1];
          File indexesZipFile = indexesZipFiles.stream().filter(file -> file.getName().equals(fileName)).findFirst().orElse(null);
          if (indexesZipFile == null) {
            return new byte[0];
          }
  
          RandomAccessFile file = new RandomAccessFile(indexesZipFile, "r");
          try {
            file.seek(offset);
            byte[] buffer = new byte[BUFFER_SIZE];
            int len = file.read(buffer);
            if (len == -1) {
              return null;
            } else {
              byte[] data = new byte[len];
              System.arraycopy(buffer, 0, data, 0, len);
              return data;
            }
          } finally {
            try {
              file.close();
            } catch (IOException e) {
              log.debug("Could not close the file", e);
            }
          }
        }
      });

      getCoordinatorAddress = rpcService.registerCommand(new RemoteCommand()
      {
         @Override
         public Serializable execute(Serializable[] serializables) throws Throwable
         {
           return ((RPCServiceImpl) rpcService).getHostAddress();
         }

         @Override
         public String getId()
         {
            return "org.exoplatform.services.jcr.impl.core.query.IndexRecoveryImpl-getCoordinatorAddress-" + commandSuffix;
         }
      });

      requestForResponsibleToSetIndexOnline = rpcService.registerCommand(new RemoteCommand()
      {

         public String getId()
         {
            return "org.exoplatform.services.jcr.impl.core.query.IndexRecoveryImpl-requestForResponsibleToSetIndexOnline-"
               + commandSuffix;
         }

         public Serializable execute(Serializable[] args) throws Throwable
         {
            return isResponsibleToSetIndexOnline;
         }
      });

      checkIndexReady = rpcService.registerCommand(new RemoteCommand()
      {
         public String getId()
         {
            return "org.exoplatform.services.jcr.impl.core.query.IndexRecoveryImpl-checkIndexIsReady-" + commandSuffix;
         }

         public Serializable execute(Serializable[] args) throws Throwable
         {
            // if index is currently online, then it can be retrieved 
            return searchManager.isOnline();
         }
      });

      rpcService.registerTopologyChangeListener(this);
   }

   /**
    * {@inheritDoc}
    */
   @SuppressWarnings("unchecked")
   public List<String> getIndexList() throws RepositoryException
   {
      try
      {
         return (List<String>)rpcService.executeCommandOnCoordinator(getIndexList, true);
      }
      catch (SecurityException e)
      {
         throw new RepositoryException(e);
      }
      catch (RPCException e)
      {
         throw new RepositoryException(e);
      }
   }

   @Override
   public String getCoordinatorAddress() throws RepositoryException
   {
      try
      {
         if(!(rpcService instanceof RPCServiceImpl))
         {
            throw new RepositoryException("failed get coordinator address");
         }
         String address=  (String) rpcService.executeCommandOnCoordinator(getCoordinatorAddress, true);
         if(address == null)
         {
            throw new RepositoryException("Error to get coordinator address. Unsupported address object");
         }
         return address;
      }
      catch (SecurityException e)
      {
         throw new RepositoryException(e);
      }
      catch (RPCException e)
      {
         throw new RepositoryException(e);
      }
   }

   /**
    * {@inheritDoc}
    */
   public void setIndexOffline() throws RepositoryException
   {
      try
      {
         isResponsibleToSetIndexOnline = true;
         rpcService.executeCommandOnCoordinator(changeIndexMode, true, false);
      }
      catch (SecurityException e)
      {
         throw new RepositoryException(e);
      }
      catch (RPCException e)
      {
         throw new RepositoryException(e);
      }
   }

   /**
    * {@inheritDoc}
    */
   public void setIndexOnline() throws RepositoryException
   {
      try
      {
         rpcService.executeCommandOnCoordinator(changeIndexMode, true, true);
         isResponsibleToSetIndexOnline = false;
      }
      catch (SecurityException e)
      {
         throw new RepositoryException(e);
      }
      catch (RPCException e)
      {
         throw new RepositoryException(e);
      }
   }

   /**
    * {@inheritDoc}
    */
   public InputStream getIndexFile(String filePath) throws RepositoryException
   {
      try
      {
         return new RemoteInputStream(filePath);
      }
      catch (SecurityException e)
      {
         throw new RepositoryException(e);
      }
      catch (RPCException e)
      {
         throw new RepositoryException(e);
      }
   }

    @Override
    public InputStream getIndexFolderInZip() throws RepositoryException {
      try {
        return new RemoteInputStream(true);
      } catch (RPCException e) {
        throw new RepositoryException(e);
      }
    }

   /**
    * @see org.exoplatform.services.jcr.impl.core.query.IndexRecovery#checkIndexReady()
    */
   public boolean checkIndexReady() throws RepositoryException
   {
      try
      {
         return (Boolean)rpcService.executeCommandOnCoordinator(checkIndexReady, true);
      }
      catch (SecurityException e)
      {
         throw new RepositoryException(e);
      }
      catch (RPCException e)
      {
         throw new RepositoryException(e);
      }
   }

   /**
    * Allows to read data from remote machine.
    */
   class RemoteInputStream extends InputStream
   {
      private String filePath;

      private long fileOffset = 0;

      private boolean allIndexData = false;

      private int bufferOffset = 0;

      private byte[] buffer;

      RemoteInputStream(boolean allIndexData) throws SecurityException, RPCException
      {
         this.allIndexData = allIndexData;
         if (this.allIndexData) {
           this.filePath = (String) rpcService.executeCommandOnCoordinator(prepareIndexesForCopy, true);
         }
         readNext();
      }

      RemoteInputStream(String filePath) throws SecurityException, RPCException
      {
        this.filePath = filePath;
        readNext();
      }

      /**
       * {@inheritDoc}
       */
      @Override
      public int read() throws IOException
      {
         throw new UnsupportedOperationException("RemoteStream.read() method is not supported");
      }

      /**
       * {@inheritDoc}
       */
      @Override
      public int available() throws IOException
      {
         return buffer == null ? 0 : buffer.length - bufferOffset;
      }

      /**
       * {@inheritDoc}
       */
      @Override
      public int read(byte b[]) throws IOException
      {
         if (buffer == null)
         {
            return -1;
         }
         else if (available() == 0)
         {
            try
            {
               readNext();

               if (buffer == null)
               {
                  return -1;
               }
            }
            catch (SecurityException | RPCException e) {
              throw new IOException(e);
            }
         }

         int len = Math.min(b.length, available());
         System.arraycopy(buffer, bufferOffset, b, 0, len);
         bufferOffset += len;

         return len;
      }

      /**
       * {@inheritDoc}
       */
      @Override
      public int read(byte b[], int off, int len) throws IOException {
         throw new UnsupportedOperationException(
            "RemoteStream.read(byte b[], int off, int len) method is not supported");
      }

      private void readNext() throws RPCException {
        if (this.allIndexData) {
          this.buffer = (byte[]) rpcService.executeCommandOnCoordinator(getAllIndexes, true, filePath, fileOffset);
        } else if (StringUtils.isNotBlank(filePath)) {
          this.buffer = (byte[]) rpcService.executeCommandOnCoordinator(getIndexFile, true, filePath, fileOffset);
        }
  
        if (buffer != null) {
          this.fileOffset += this.buffer.length;
          this.bufferOffset = 0;
        }
      }

      @Override
      public void close() throws IOException {
        super.close();
        if (this.allIndexData) {
          try {
            rpcService.executeCommandOnCoordinator(cleanupIndexesAfterCopy, true, filePath);
          } catch (RPCException e) {
            throw new IOException("Error cleaning recovered file index from coordinator", e);
          }
        }
      }
   }

   /**
    * {@inheritDoc}
    */
   public void onChange(TopologyChangeEvent event)
   {
      try
      {
         if (rpcService.isCoordinator() && !isOnline)
         {
            new Thread()
            {
               @Override
               public synchronized void run()
               {
                  try
                  {
                     List<Object> results =
                        rpcService.executeCommandOnAllNodes(requestForResponsibleToSetIndexOnline, true);

                     for (Object result : results)
                     {
                        if (result instanceof Boolean)
                        {
                           if ((Boolean)result)
                           {
                              return;
                           }
                        }
                        else
                        {
                           log.error("Result is not an instance of Boolean" + result);
                        }
                     }
                     // node which was responsible for resuming leave the cluster, so resume component
                     log
                        .error("Node responsible for setting index back online seems to leave the cluster. Setting back online.");
                     searchManager.setOnline(true, false, false);
                  }
                  catch (SecurityException e1)
                  {
                     log.error("You haven't privileges to execute remote command", e1);
                  }
                  catch (RPCException e1)
                  {
                     log.error("Exception during command execution", e1);
                  }
                  catch (IOException e2)
                  {
                     log.error("Exception during setting index back online", e2);
                  }
               }
            }.start();
         }
      }
      catch (RPCException e)
      {
         log.error("Can't check if node coordinator or not.");
      }
   }

   /**
    * {@inheritDoc}
    */
   public void close()
   {
      rpcService.unregisterCommand(changeIndexMode);
      rpcService.unregisterCommand(getIndexList);
      rpcService.unregisterCommand(getCoordinatorAddress);
      rpcService.unregisterCommand(getIndexFile);
      rpcService.unregisterCommand(requestForResponsibleToSetIndexOnline);
      rpcService.unregisterCommand(checkIndexReady);

      rpcService.unregisterTopologyChangeListener(this);
   }

}
