/*
    * Licensed to The Apereo Foundation under one or more contributor license
    * agreements. See the NOTICE file distributed with this work for additional
    * information regarding copyright ownership.
    *
    *
    * The Apereo Foundation licenses this file to you under the Educational
    * Community 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://opensource.org/licenses/ecl2.txt
   *
   * 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.opencastproject.search.impl;
  
  import static org.opencastproject.security.api.Permissions.Action.WRITE;
  import static org.opencastproject.security.util.SecurityUtil.getEpisodeRoleId;
  
  import org.opencastproject.elasticsearch.index.ElasticsearchIndex;
  import org.opencastproject.elasticsearch.index.rebuild.AbstractIndexProducer;
  import org.opencastproject.elasticsearch.index.rebuild.IndexProducer;
  import org.opencastproject.elasticsearch.index.rebuild.IndexRebuildException;
  import org.opencastproject.elasticsearch.index.rebuild.IndexRebuildService;
  import org.opencastproject.list.api.ListProviderException;
  import org.opencastproject.list.api.ListProvidersService;
  import org.opencastproject.list.api.ResourceListQuery;
  import org.opencastproject.list.impl.ResourceListQueryImpl;
  import org.opencastproject.mediapackage.MediaPackage;
  import org.opencastproject.metadata.dublincore.DublinCore;
  import org.opencastproject.metadata.dublincore.DublinCoreCatalog;
  import org.opencastproject.metadata.dublincore.DublinCoreUtil;
  import org.opencastproject.metadata.dublincore.DublinCoreValue;
  import org.opencastproject.metadata.dublincore.DublinCores;
  import org.opencastproject.search.api.SearchException;
  import org.opencastproject.search.api.SearchResult;
  import org.opencastproject.search.api.SearchService;
  import org.opencastproject.search.impl.persistence.SearchServiceDatabase;
  import org.opencastproject.search.impl.persistence.SearchServiceDatabaseException;
  import org.opencastproject.security.api.AccessControlEntry;
  import org.opencastproject.security.api.AccessControlList;
  import org.opencastproject.security.api.AccessControlUtil;
  import org.opencastproject.security.api.AuthorizationService;
  import org.opencastproject.security.api.Organization;
  import org.opencastproject.security.api.OrganizationDirectoryService;
  import org.opencastproject.security.api.SecurityService;
  import org.opencastproject.security.api.UnauthorizedException;
  import org.opencastproject.security.api.User;
  import org.opencastproject.security.util.SecurityUtil;
  import org.opencastproject.series.api.SeriesException;
  import org.opencastproject.series.api.SeriesService;
  import org.opencastproject.util.NotFoundException;
  import org.opencastproject.util.data.Tuple;
  import org.opencastproject.workspace.api.Workspace;
  
  import com.google.gson.Gson;
  import com.google.gson.JsonElement;
  
  import org.apache.commons.io.IOUtils;
  import org.elasticsearch.ElasticsearchStatusException;
  import org.elasticsearch.action.DocWriteResponse;
  import org.elasticsearch.action.index.IndexRequest;
  import org.elasticsearch.action.search.SearchRequest;
  import org.elasticsearch.action.search.SearchResponse;
  import org.elasticsearch.action.update.UpdateRequest;
  import org.elasticsearch.action.update.UpdateResponse;
  import org.elasticsearch.client.RequestOptions;
  import org.elasticsearch.client.indices.CreateIndexRequest;
  import org.elasticsearch.common.xcontent.XContentType;
  import org.elasticsearch.rest.RestStatus;
  import org.elasticsearch.search.builder.SearchSourceBuilder;
  import org.osgi.service.component.ComponentContext;
  import org.osgi.service.component.annotations.Activate;
  import org.osgi.service.component.annotations.Component;
  import org.osgi.service.component.annotations.Reference;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.io.InputStream;
  import java.nio.charset.StandardCharsets;
  import java.time.Instant;
  import java.time.format.DateTimeFormatter;
  import java.util.Collections;
  import java.util.Date;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
 /**
  * A Elasticsearch-based {@link SearchService} implementation.
  */
 @Component(
         immediate = true,
         service = { SearchServiceIndex.class, IndexProducer.class },
         property = {
                 "service.description=Search Service Index",
                 "service.pid=org.opencastproject.search.impl.SearchServiceIndex"
         }
 )
 public final class SearchServiceIndex extends AbstractIndexProducer implements IndexProducer {
 
   @Override
   public IndexRebuildService.Service getService() {
     return IndexRebuildService.Service.Search;
   }
 
   /** Log facility */
   private static final Logger logger = LoggerFactory.getLogger(SearchServiceIndex.class);
 
   public static final String INDEX_NAME = "opencast_search";
 
   private final Gson gson = new Gson();
 
   private ElasticsearchIndex esIndex;
 
   private SeriesService seriesService;
 
   /** The local workspace */
   private Workspace workspace;
 
   /** The security service */
   private SecurityService securityService;
 
   /** The authorization service */
   private AuthorizationService authorizationService;
 
   /** Persistent storage */
   private SearchServiceDatabase persistence;
 
   /** The organization directory service */
   private OrganizationDirectoryService organizationDirectory = null;
 
   private ListProvidersService listProvidersService;
 
   private String systemUserName = null;
 
 
   /**
    * Creates a new instance of the search service index.
    */
   public SearchServiceIndex() {
   }
 
   /**
    * Service activator, called via declarative services configuration.
    *
    * @param cc
    *          the component context
    */
   @Activate
   public void activate(final ComponentContext cc) throws IllegalStateException {
     createIndex();
     systemUserName = SecurityUtil.getSystemUserName(cc);
   }
 
   private void createIndex() {
     var mapping = "";
     try (var in = this.getClass().getResourceAsStream("/search-mapping.json")) {
       mapping = IOUtils.toString(in, StandardCharsets.UTF_8);
     } catch (IOException e) {
       throw new SearchException("Could not read mapping.", e);
     }
     try {
       logger.debug("Trying to create index for '{}'", INDEX_NAME);
       InputStream is = getClass().getResourceAsStream("/elasticsearch/indexSettings.json");
       String indexSettings = IOUtils.toString(is, StandardCharsets.UTF_8);
       final CreateIndexRequest request = new CreateIndexRequest(INDEX_NAME)
           .settings(indexSettings, XContentType.JSON)
           .mapping(mapping, XContentType.JSON);
       var response = esIndex.getClient().indices().create(request, RequestOptions.DEFAULT);
       if (!response.isAcknowledged()) {
         throw new SearchException("Unable to create index for '" + INDEX_NAME + "'");
       }
     } catch (ElasticsearchStatusException e) {
       if (e.getDetailedMessage().contains("already_exists_exception")) {
         logger.info("Detected existing index '{}'", INDEX_NAME);
       } else {
         throw e;
       }
     } catch (IOException e) {
       throw new SearchException(e);
     }
   }
 
   @Reference
   public void setEsIndex(ElasticsearchIndex esIndex) {
     this.esIndex = esIndex;
   }
 
 
   public SearchResponse search(SearchSourceBuilder searchSource) throws SearchException {
     SearchRequest searchRequest = new SearchRequest(INDEX_NAME);
     logger.debug("Sending for query: {}", searchSource.query());
     searchRequest.source(searchSource);
     try {
       return esIndex.getClient().search(searchRequest, RequestOptions.DEFAULT);
     } catch (IOException e) {
       throw new SearchException(e);
     }
   }
 
   /**
    * Immediately adds the mediapackage to the search index.
    *
    * @param mediaPackage
    *          the media package
    * @throws SearchException
    *           if the media package cannot be added to the search index
    * @throws IllegalArgumentException
    *           if the media package is <code>null</code>
    * @throws UnauthorizedException
    *           if the user does not have the rights to add the mediapackage
    */
   public void addSynchronously(MediaPackage mediaPackage)
           throws SearchException, IllegalArgumentException, UnauthorizedException, SearchServiceDatabaseException {
     if (mediaPackage == null) {
       throw new IllegalArgumentException("Unable to add a null mediapackage");
     }
     var mediaPackageId = mediaPackage.getIdentifier().toString();
 
     checkSearchEntityWritePermission(mediaPackageId);
 
     logger.debug("Attempting to add media package {} to search index", mediaPackageId);
     final var acls = new AccessControlList[1];
     final var org = securityService.getOrganization();
     final var systemUser = SecurityUtil.createSystemUser(systemUserName, org);
     // Ensure we always get the actual acl by forcing access
     SecurityUtil.runAs(securityService, org, systemUser, () -> {
       acls[0] = authorizationService.getActiveAcl(mediaPackage).getA();
     });
     var acl = acls[0] == null ? new AccessControlList() : acls[0];
     var now = new Date();
 
     try {
       persistence.storeMediaPackage(mediaPackage, acl, now);
     } catch (SearchServiceDatabaseException e) {
       throw new SearchException(String.format("Could not store media package to search database %s", mediaPackageId),
           e);
     }
 
     indexMediaPackage(mediaPackage, acl);
   }
 
   public void indexMediaPackage(String mediaPackageId)
           throws SearchException, SearchServiceDatabaseException, UnauthorizedException, NotFoundException {
     if (!securityService.getUser().hasRole("ROLE_ADMIN")) {
       throw new UnauthorizedException("Only global administrators may trigger manual event updates.");
     }
     try {
       MediaPackage mp = persistence.getMediaPackage(mediaPackageId);
       AccessControlList acl = persistence.getAccessControlList(mediaPackageId);
       Date modificationDate = persistence.getModificationDate(mediaPackageId);
       Date deletionDate = persistence.getDeletionDate(mediaPackageId);
       indexMediaPackage(mp, acl, modificationDate, deletionDate);
     } catch (RuntimeException e) {
       logSkippingElement(logger, "event", mediaPackageId, e);
     }
   }
 
   private void indexMediaPackage(MediaPackage mediaPackage, AccessControlList acl)
           throws SearchException, SearchServiceDatabaseException {
     indexMediaPackage(mediaPackage, acl, null, null);
   }
 
   private void indexMediaPackage(MediaPackage mediaPackage, AccessControlList acl, Date modDate, Date delDate)
           throws SearchException, SearchServiceDatabaseException {
     String mediaPackageId = mediaPackage.getIdentifier().toString();
     String orgId = securityService.getOrganization().getId();
 
     //If the entry has been deleted then there's *probably* no dc file to load.
     DublinCoreCatalog dc = null == delDate
         ? DublinCoreUtil.loadEpisodeDublinCore(workspace, mediaPackage).orElse(DublinCores.mkSimple())
         : DublinCores.mkSimple();
 
     List<DublinCoreCatalog> seriesList = Collections.emptyList();
     if (dc.hasValue(DublinCore.PROPERTY_IS_PART_OF)) {
       //Find the series (if any), filter for those which exist to prevent linking non-existent series
       seriesList = dc.get(DublinCore.PROPERTY_IS_PART_OF).stream().map(DublinCoreValue::getValue).map(s -> {
         try {
           return seriesService.getSeries(s);
         } catch (NotFoundException e) {
           logger.warn("Series {} not found during index of event {}, omitting the link from the indexed data", s,
               mediaPackageId);
         } catch (UnauthorizedException e) {
           logger.warn("Not authorized for series {} during index of event {}, omitting the link from the indexed data",
               s, mediaPackageId);
         } catch (SeriesException e) {
           throw new SearchException(e);
         }
         return null;
       }).filter(Objects::nonNull).collect(Collectors.toList());
     }
 
     // Add custom roles if enabled
     acl = addCustomAclRoles(mediaPackageId, acl);
 
     SearchResult item = new SearchResult(SearchService.IndexEntryType.Episode, dc, acl, orgId, mediaPackage,
         null != modDate ? modDate.toInstant() : Instant.now(),
         null != delDate ? delDate.toInstant() : null);
     Map<String, Object> metadata = item.dehydrateForIndex();
     try {
       var request = new IndexRequest(INDEX_NAME);
       request.id(mediaPackageId);
       request.source(metadata);
       esIndex.getClient().index(request, RequestOptions.DEFAULT);
       logger.debug("Indexed episode {}", mediaPackageId);
     } catch (IOException e) {
       throw new SearchException(e);
     }
 
     // Elasticsearch series
     for (DublinCoreCatalog seriesDc : seriesList) {
       String seriesId = seriesDc.getFirst(DublinCore.PROPERTY_IDENTIFIER);
       AccessControlList seriesAcl = persistence.getAccessControlLists(seriesId, mediaPackageId).stream()
           .map(aclPair -> addCustomAclRoles(aclPair.getKey(), aclPair.getValue()))
           .reduce(new AccessControlList(acl.getEntries()), AccessControlList::mergeActions);
       item = new SearchResult(SearchService.IndexEntryType.Series, seriesDc, seriesAcl, orgId,
           null, Instant.now(), null);
 
       Map<String, Object> seriesData = item.dehydrateForIndex();
       try {
         var request = new IndexRequest(INDEX_NAME);
         request.id(seriesId);
         request.source(seriesData);
         esIndex.getClient().index(request, RequestOptions.DEFAULT);
         logger.debug("Indexed series {} related to episode {}", seriesId, mediaPackageId);
       } catch (IOException e) {
         throw new SearchException(e);
       }
     }
   }
 
   /**
    * Add custom roles of the media package to the passed ACL
    *
    * @param mediaPackageId
    *          the media package
    * @param acl
    *          the existing access control list
    * @return {@link AccessControlList} containing the passed and the custom roles merged together
    *
    */
   private AccessControlList addCustomAclRoles(String mediaPackageId, AccessControlList acl) {
     // This allows users with a role of the form ROLE_EPISODE_<ID>_<ACTION> to access the event through the index
     Set<AccessControlEntry> customEntries = new HashSet<>();
     customEntries.add(new AccessControlEntry(getEpisodeRoleId(mediaPackageId, "READ"), "read", true));
     customEntries.add(new AccessControlEntry(getEpisodeRoleId(mediaPackageId, "WRITE"), "write", true));
 
     ResourceListQuery query = new ResourceListQueryImpl();
     if (listProvidersService.hasProvider("ACL.ACTIONS")) {
       Map<String, String> actions = new HashMap<>();
       try {
         actions = listProvidersService.getList("ACL.ACTIONS", query, true);
       } catch (ListProviderException e) {
         throw new SearchException("Listproviders not loaded. " + e);
       }
       for (String action : actions.keySet()) {
         customEntries.add(
             new AccessControlEntry(getEpisodeRoleId(mediaPackageId, action), action, true));
       }
     }
 
     return acl;
   }
 
   private void checkSearchEntityWritePermission(final String mediaPackageId) throws SearchException {
     User user = securityService.getUser();
     try {
       if (!persistence.isAvailable(mediaPackageId)) {
         throw new NotFoundException();
       }
       AccessControlList acl = persistence.getAccessControlList(mediaPackageId);
       if (!AccessControlUtil.isAuthorized(acl, user, securityService.getOrganization(), WRITE.toString(),
           mediaPackageId)) {
         throw new UnauthorizedException(user, "Write permission denied for " + mediaPackageId, acl);
       }
     } catch (NotFoundException e) {
       logger.debug("Mediapackage {} does not exist or was deleted, allowing writes for user {}", mediaPackageId, user);
     } catch (SearchServiceDatabaseException | UnauthorizedException e) {
       throw new SearchException(e);
     }
   }
 
   /**
    * Immediately removes the given mediapackage from the search service.
    *
    * @param mediaPackageId
    *          the media package identifier
    * @return <code>true</code> if the mediapackage was deleted
    * @throws SearchException
    *           if deletion failed
    */
   public boolean deleteSynchronously(final String mediaPackageId) throws SearchException {
 
     checkSearchEntityWritePermission(mediaPackageId);
 
     String deletionString = DateTimeFormatter.ISO_INSTANT.format(Instant.now());
 
     try {
       logger.info("Marking media package {} as deleted in search index", mediaPackageId);
       JsonElement json = gson.toJsonTree(Map.of(
           SearchResult.DELETED_DATE, deletionString,
           SearchResult.MODIFIED_DATE, deletionString));
       var updateRequst = new UpdateRequest(INDEX_NAME, mediaPackageId)
           .doc(gson.toJson(json), XContentType.JSON);
       esIndex.getClient().update(updateRequst, RequestOptions.DEFAULT);
     } catch (ElasticsearchStatusException e) {
       if (e.status().getStatus() != RestStatus.NOT_FOUND.getStatus()) {
         throw e;
       }
       logger.warn("Event {} is not in the search index. Skipping deletion", mediaPackageId);
     } catch (IOException e) {
       throw new SearchException("Could not delete episode " + mediaPackageId + " from index", e);
     }
 
     try {
       logger.info("Marking media package {} as deleted in search database", mediaPackageId);
 
       String seriesId = null;
       Date now = new Date();
       try {
         seriesId = persistence.getMediaPackage(mediaPackageId).getSeries();
         persistence.deleteMediaPackage(mediaPackageId, now);
         logger.info("Removed media package {} from search persistence", mediaPackageId);
       } catch (NotFoundException e) {
         // even if mp not found in persistence, it might still exist in search index.
         logger.info("Could not find media package with id {} in persistence, but will try remove it from index anyway.",
             mediaPackageId);
       } catch (SearchServiceDatabaseException | UnauthorizedException e) {
         throw new SearchException(
             String.format("Could not delete media package with id %s from persistence storage", mediaPackageId), e);
       }
 
       // Update series
       if (seriesId != null) {
         try {
           if (!persistence.getSeries(seriesId).isEmpty()) {
             // Update series acl if there are still episodes in the series
             final AccessControlList seriesAcl = persistence.getAccessControlLists(seriesId).stream()
                 .map(aclPair -> addCustomAclRoles(aclPair.getKey(), aclPair.getValue()))
                 .reduce(new AccessControlList(), AccessControlList::mergeActions);
             JsonElement json = gson.toJsonTree(Map.of(
                 SearchResult.INDEX_ACL, SearchResult.dehydrateAclForIndex(seriesAcl),
                 SearchResult.MODIFIED_DATE, deletionString));
             var updateRequest = new UpdateRequest(INDEX_NAME, seriesId).doc(gson.toJson(json), XContentType.JSON);
             try {
               esIndex.getClient().update(updateRequest, RequestOptions.DEFAULT);
             } catch (ElasticsearchStatusException e) {
               if (RestStatus.NOT_FOUND == e.status()) {
                 logger.warn("Attempted to modify {}, but that series does not exist in the index.", seriesId);
               }
             }
           } else {
             // Remove series if there are no episodes in the series any longer
             deleteSeriesSynchronously(seriesId);
           }
         } catch (IOException e) {
           throw new SearchException(e);
         }
       }
 
       return true;
     } catch (SearchServiceDatabaseException e) {
       logger.info("Could not delete media package with id {} from search index", mediaPackageId);
       throw new SearchException(e);
     }
   }
 
   /**
    * Immediately removes the given series from the search service.
    *
    * @param seriesId
    *          the series
    * @throws SearchException
    */
   public boolean deleteSeriesSynchronously(String seriesId) throws SearchException {
     try {
       logger.info("Marking {} as deleted in the search index", seriesId);
       JsonElement json = gson.toJsonTree(Map.of(
           "deleted", Instant.now().getEpochSecond(),
           "modified", Instant.now().toString()));
       var updateRequest = new UpdateRequest(INDEX_NAME, seriesId).doc(gson.toJson(json), XContentType.JSON);
       try {
         UpdateResponse response = esIndex.getClient().update(updateRequest, RequestOptions.DEFAULT);
         //NB: We're marking things as deleted but *not actually deleting them**
         return DocWriteResponse.Result.UPDATED == response.getResult();
       } catch (ElasticsearchStatusException e) {
         if (RestStatus.NOT_FOUND == e.status()) {
           logger.debug("Attempted to delete {}, but that series does not exist in the index.", seriesId);
           return true;
         }
         throw new SearchException(e);
       }
     } catch (IOException e) {
       throw new SearchException("Could not delete series " + seriesId + " from index", e);
     }
   }
 
   @Override
   public void repopulate(IndexRebuildService.DataType type) throws IndexRebuildException {
     final Organization originalOrg = securityService.getOrganization();
     final User originalUser = securityService.getUser();
 
     try {
       int total = persistence.countMediaPackages();
       int pageSize = 50;
       int pageOffset = 0;
       AtomicInteger current = new AtomicInteger(1);
       logIndexRebuildBegin(logger, total, "search");
       List<Tuple<MediaPackage, String>> page = null;
 
       do {
         page = persistence.getAllMediaPackages(pageSize, pageOffset).collect(Collectors.toList());
         page.forEach(tuple -> {
           try {
             MediaPackage mediaPackage = tuple.getA();
             Organization organization = organizationDirectory.getOrganization(tuple.getB());
             final var systemUser = SecurityUtil.createSystemUser(systemUserName, organization);
             securityService.setUser(systemUser);
             securityService.setOrganization(organization);
 
             String mediaPackageId = mediaPackage.getIdentifier().toString();
 
             AccessControlList acl = persistence.getAccessControlList(mediaPackageId);
             Date modificationDate = persistence.getModificationDate(mediaPackageId);
             Date deletionDate = persistence.getDeletionDate(mediaPackageId);
 
             current.getAndIncrement();
 
             indexMediaPackage(mediaPackage, acl, modificationDate, deletionDate);
           } catch (SearchServiceDatabaseException e) {
             logIndexRebuildError(logger, total, current.get(), e);
             //NB: Runtime exception thrown to escape the functional interfacing
             throw new RuntimeException("Internal Index Rebuild Failure", e);
           } catch (RuntimeException | NotFoundException e) {
             logSkippingElement(logger, "event", tuple.getA().getIdentifier().toString(), e);
           }
         });
         //Current is the *page* index, so we remove one since each page only has pageSize entries
         logIndexRebuildProgress(logger, total, current.get() - 1, pageSize);
         pageOffset += 1;
       } while (pageOffset * pageSize <= total);
       //NB: Catching RuntimeException since it can be thrown inside the functional forEach here
     } catch (SearchServiceDatabaseException | RuntimeException e) {
       logIndexRebuildError(logger, e);
       throw new IndexRebuildException("Index Rebuild Failure", e);
     } finally {
       securityService.setUser(originalUser);
       securityService.setOrganization(originalOrg);
     }
   }
 
   @Reference
   public void setPersistence(SearchServiceDatabase persistence) {
     this.persistence = persistence;
   }
 
   @Reference
   public void setSeriesService(SeriesService seriesService) {
     this.seriesService = seriesService;
   }
 
   @Reference
   public void setWorkspace(Workspace workspace) {
     this.workspace = workspace;
   }
 
   @Reference
   public void setAuthorizationService(AuthorizationService authorizationService) {
     this.authorizationService = authorizationService;
   }
 
   /**
    * Callback for setting the security service.
    *
    * @param securityService
    *          the securityService to set
    */
   @Reference
   public void setSecurityService(SecurityService securityService) {
     this.securityService = securityService;
   }
 
   /**
    * Sets a reference to the organization directory service.
    *
    * @param organizationDirectory
    *          the organization directory
    */
   @Reference
   public void setOrganizationDirectoryService(OrganizationDirectoryService organizationDirectory) {
     this.organizationDirectory = organizationDirectory;
   }
 
   @Reference
   public void setListProvidersService(ListProvidersService listProvidersService) {
     this.listProvidersService = listProvidersService;
   }
 }