/*
    * This file is part of ***  M y C o R e  ***
    * See http://www.mycore.de/ for details.
    *
    * MyCoRe is free software: you can redistribute it and/or modify
    * it under the terms of the GNU General Public License as published by
    * the Free Software Foundation, either version 3 of the License, or
    * (at your option) any later version.
    *
   * MyCoRe 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 MyCoRe.  If not, see <http://www.gnu.org/licenses/>.
   */
  
  package org.mycore.oai;
  
  import java.time.Instant;
  import java.util.Arrays;
  import java.util.Date;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.Timer;
  import java.util.TimerTask;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  
  import org.apache.logging.log4j.LogManager;
  import org.apache.logging.log4j.Logger;
  import org.mycore.common.MCRException;
  import org.mycore.common.MCRSession;
  import org.mycore.common.MCRSessionMgr;
  import org.mycore.common.config.MCRConfiguration2;
  import org.mycore.common.events.MCRShutdownHandler;
  import org.mycore.oai.pmh.BadResumptionTokenException;
  import org.mycore.oai.pmh.DefaultResumptionToken;
  import org.mycore.oai.pmh.Header;
  import org.mycore.oai.pmh.MetadataFormat;
  import org.mycore.oai.pmh.OAIDataList;
  import org.mycore.oai.pmh.Record;
  import org.mycore.oai.set.MCRSet;
  import org.mycore.util.concurrent.MCRTransactionableRunnable;
  
  /**
   * Search manager of the mycore OAI-PMH implementation. Creates a new
   * {@link MCROAISearcher} instance for each
   * {@link #searchHeader(MetadataFormat, MCRSet, Instant, Instant)}
   * and {@link #searchRecord(MetadataFormat, MCRSet, Instant, Instant)} call.
   * The resumption token created by those methods can be reused for
   * later calls to the same searcher. A searcher is dropped after an
   * expiration time. The time increases for each query call.
   *
   * <p>Due to token based querying it is not possible to set a current
   * position for the resumption token. Its always set to -1.</p>
   *
   * @author Matthias Eichner
   */
  public class MCROAISearchManager {
  
      protected static final Logger LOGGER = LogManager.getLogger(MCROAISearchManager.class);
  
      protected static final String TOKEN_DELIMITER = "@";
  
      protected static int MAX_AGE;
  
      protected Map<String, MCROAISearcher> resultMap;
  
      protected MCROAIIdentify identify;
  
      protected MCROAIObjectManager objManager;
  
      protected MCROAISetManager setManager;
  
      protected int partitionSize;
  
      private ExecutorService executorService;
  
      private boolean runListRecordsParallel;
  
      static {
          String prefix = MCROAIAdapter.PREFIX + "ResumptionTokens.";
          MAX_AGE = MCRConfiguration2.getInt(prefix + "MaxAge").orElse(30) * 60 * 1000;
      }
  
      public MCROAISearchManager() {
          this.resultMap = new ConcurrentHashMap<>();
          TimerTask tt = new TimerTask() {
              @Override
              public void run() {
                  for (Map.Entry<String, MCROAISearcher> entry : resultMap.entrySet()) {
                      String searchId = entry.getKey();
                      MCROAISearcher searcher = entry.getValue();
                     if ((searcher != null) && searcher.isExpired()) {
                         LOGGER.info("Removing expired resumption token {}", searchId);
                         resultMap.remove(searchId);
                     }
                 }
             }
         };
         new Timer().schedule(tt, new Date(System.currentTimeMillis() + MAX_AGE), MAX_AGE);
         runListRecordsParallel = MCRConfiguration2
             .getOrThrow(MCROAIAdapter.PREFIX + "RunListRecordsParallel", Boolean::parseBoolean);
         if (runListRecordsParallel) {
             executorService = Executors.newWorkStealingPool();
             MCRShutdownHandler.getInstance().addCloseable(executorService::shutdownNow);
         }
     }
 
     public void init(MCROAIIdentify identify, MCROAIObjectManager objManager, MCROAISetManager setManager,
         int partitionSize) {
         this.identify = identify;
         this.objManager = objManager;
         this.setManager = setManager;
         this.partitionSize = partitionSize;
     }
 
     public Optional<Header> getHeader(String oaiId) {
         MCROAISearcher searcher = getSearcher(this.identify, null, 1, setManager, objManager);
         return searcher.getHeader(objManager.getMyCoReId(oaiId));
     }
 
     public OAIDataList<Header> searchHeader(String resumptionToken) throws BadResumptionTokenException {
         String searchId = getSearchId(resumptionToken);
         String tokenCursor = getTokenCursor(resumptionToken);
         MCROAISearcher searcher = this.resultMap.get(searchId);
         if (searcher == null || tokenCursor == null || tokenCursor.length() <= 0) {
             throw new BadResumptionTokenException(resumptionToken);
         }
         MCROAIResult result = searcher.query(tokenCursor);
         return getHeaderList(searcher, result);
     }
 
     public OAIDataList<Record> searchRecord(String resumptionToken) throws BadResumptionTokenException {
         String searchId = getSearchId(resumptionToken);
         String tokenCursor = getTokenCursor(resumptionToken);
         MCROAISearcher searcher = this.resultMap.get(searchId);
         if (searcher == null || tokenCursor == null || tokenCursor.length() <= 0) {
             throw new BadResumptionTokenException(resumptionToken);
         }
         MCROAIResult result = searcher.query(tokenCursor);
         return getRecordList(searcher, result);
     }
 
     public OAIDataList<Header> searchHeader(MetadataFormat format, MCRSet set, Instant from, Instant until) {
         MCROAISearcher searcher = getSearcher(this.identify, format, getPartitionSize(), setManager, objManager);
         this.resultMap.put(searcher.getID(), searcher);
         MCROAIResult result = searcher.query(set, from, until);
         return getHeaderList(searcher, result);
     }
 
     public OAIDataList<Record> searchRecord(MetadataFormat format, MCRSet set, Instant from, Instant until) {
         MCROAISearcher searcher = getSearcher(this.identify, format, getPartitionSize(), setManager, objManager);
         this.resultMap.put(searcher.getID(), searcher);
         MCROAIResult result = searcher.query(set, from, until);
         return getRecordList(searcher, result);
     }
 
     protected OAIDataList<Record> getRecordList(MCROAISearcher searcher, MCROAIResult result) {
         OAIDataList<Record> recordList = runListRecordsParallel ? getRecordListParallel(searcher, result)
             : getRecordListSequential(searcher, result);
         if (recordList.contains(null)) {
             if (MCRConfiguration2.getBoolean("MCR.OAIDataProvider.FailOnErrorRecords").orElse(false)) {
                 throw new MCRException(
                     "An internal error occur. Some of the following records are invalid and cannot be processed."
                         + " Please inform the system administrator. " + result.list());
             }
             recordList.removeIf(Objects::isNull);
         }
         this.setResumptionToken(recordList, searcher, result);
         return recordList;
     }
 
     private OAIDataList<Record> getRecordListSequential(MCROAISearcher searcher, MCROAIResult result) {
         OAIDataList<Record> recordList = new OAIDataList<>();
         result.list().forEach(header -> {
             Record record = this.objManager.getRecord(header, searcher.getMetadataFormat());
             recordList.add(record);
         });
         return recordList;
     }
 
     private OAIDataList<Record> getRecordListParallel(MCROAISearcher searcher, MCROAIResult result) {
         List<Header> headerList = result.list();
         int listSize = headerList.size();
         Record[] records = new Record[listSize];
         @SuppressWarnings("rawtypes")
         CompletableFuture[] futures = new CompletableFuture[listSize];
         MetadataFormat metadataFormat = searcher.getMetadataFormat();
         MCRSession mcrSession = MCRSessionMgr.getCurrentSession();
         for (int i = 0; i < listSize; i++) {
             Header header = headerList.get(i);
             int resultIndex = i;
             MCRTransactionableRunnable r = new MCRTransactionableRunnable(
                 () -> records[resultIndex] = this.objManager.getRecord(header, metadataFormat), mcrSession);
             CompletableFuture<Void> future = CompletableFuture.runAsync(r, executorService);
             futures[i] = future;
         }
         CompletableFuture.allOf(futures).join();
         OAIDataList<Record> recordList = new OAIDataList<>();
         recordList.addAll(Arrays.asList(records));
         return recordList;
     }
 
     protected OAIDataList<Header> getHeaderList(MCROAISearcher searcher, MCROAIResult result) {
         OAIDataList<Header> headerList = new OAIDataList<>();
         headerList.addAll(result.list());
         this.setResumptionToken(headerList, searcher, result);
         return headerList;
     }
 
     public String getSearchId(String token) throws BadResumptionTokenException {
         try {
             return token.split(TOKEN_DELIMITER)[0];
         } catch (Exception exc) {
             throw new BadResumptionTokenException(token);
         }
     }
 
     public String getTokenCursor(String token) throws BadResumptionTokenException {
         try {
             String[] tokenParts = token.split(TOKEN_DELIMITER);
             return tokenParts[tokenParts.length - 1];
         } catch (Exception exc) {
             throw new BadResumptionTokenException(token);
         }
     }
 
     protected void setResumptionToken(OAIDataList<?> dataList, MCROAISearcher searcher, MCROAIResult result) {
         result.nextCursor().map(cursor -> {
             DefaultResumptionToken rsToken = new DefaultResumptionToken();
             rsToken.setToken(searcher.getID() + TOKEN_DELIMITER + cursor);
             rsToken.setCompleteListSize(result.getNumHits());
             rsToken.setExpirationDate(searcher.getExpirationTime());
             return rsToken;
         }).ifPresent(dataList::setResumptionToken);
     }
 
     public int getPartitionSize() {
         return partitionSize;
     }
 
     public static MCROAISearcher getSearcher(MCROAIIdentify identify, MetadataFormat format, int partitionSize,
         MCROAISetManager setManager, MCROAIObjectManager objectManager) {
         String className = identify.getConfigPrefix() + "Searcher";
         MCROAISearcher searcher = MCRConfiguration2.<MCROAISearcher>getInstanceOf(className)
             .orElseGet(MCROAICombinedSearcher::new);
         searcher.init(identify, format, MAX_AGE, partitionSize, setManager, objectManager);
         return searcher;
     }
 
 }