/*
    * 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.assetmanager.impl;
  
  import org.opencastproject.assetmanager.api.AssetManager;
  import org.opencastproject.assetmanager.api.AssetManagerException;
  import org.opencastproject.assetmanager.api.Snapshot;
  import org.opencastproject.assetmanager.api.Version;
  import org.opencastproject.assetmanager.api.query.ARecord;
  import org.opencastproject.assetmanager.api.query.RichAResult;
  import org.opencastproject.assetmanager.api.storage.AssetStore;
  import org.opencastproject.assetmanager.api.storage.RemoteAssetStore;
  import org.opencastproject.job.api.AbstractJobProducer;
  import org.opencastproject.job.api.Job;
  import org.opencastproject.security.api.OrganizationDirectoryService;
  import org.opencastproject.security.api.SecurityService;
  import org.opencastproject.security.api.UserDirectoryService;
  import org.opencastproject.serviceregistry.api.ServiceRegistry;
  import org.opencastproject.serviceregistry.api.ServiceRegistryException;
  import org.opencastproject.util.NotFoundException;
  import org.opencastproject.util.RequireUtil;
  
  import com.google.gson.Gson;
  
  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.util.Date;
  import java.util.HashMap;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.function.Consumer;
  
  @Component(
      immediate = true,
      service = AssetManagerJobProducer.class,
      property = {
          "service.description=Opencast Asset Manager Job Producer"
      }
  )
  public class AssetManagerJobProducer extends AbstractJobProducer {
  
    /** The logging facility */
    private static final Logger logger = LoggerFactory.getLogger(AssetManagerJobProducer.class);
  
    public static final String JOB_TYPE = "org.opencastproject.assetmanager";
    public static final Float JOB_LOAD = 0.1f;
    public static final Float NONTERMINAL_JOB_LOAD = 0.1f;
  
    public enum Operation {
      MoveById, MoveByIdAndVersion, MoveByIdAndDate, MoveByDate, MoveRecords
    }
  
    private static final String OK = "OK";
  
    private AssetManager tsam = null;
    private ServiceRegistry serviceRegistry = null;
    private SecurityService securityService = null;
    private UserDirectoryService userDirectoryService = null;
    private OrganizationDirectoryService organizationDirectoryService = null;
  
    public AssetManagerJobProducer() {
      super(JOB_TYPE);
    }
  
    /**
     * OSGi callback on component activation.
     *
     * @param cc
     *          the component context
     */
    @Override
    @Activate
    public void activate(ComponentContext cc) {
      logger.info("Activating assetmanager job service");
     super.activate(cc);
   }
 
   public boolean datastoreExists(String storeId) {
     Optional<AssetStore> store = tsam.getAssetStore(storeId);
     return store.isPresent();
   }
 
   /** Utility class to collect RecordInformation for moving larger 
    * groups of mediapackages in combined jobs.
    */
   private class MoveRecordInfo {
     private final Gson gson = new Gson();
     private int success = 0;
     private int failed = 0;
     private String currentMpId = "";
     public void addSuccess() {
       success++;
     };
     public void addFailed() {
       failed ++;
     }
 
     public boolean isNewMpId(String mpId) {
       if (currentMpId.equals(mpId)) {
         return false;
       }
       currentMpId = mpId;
       return true;
     }
 
     @Override
     public String toString() {
       Map<String,Integer> result = new HashMap<>();
       if (success > 0) {
         result.put("OK", success);
       }
       if (failed > 0) {
         result.put("FAIL", failed);
       }
       return gson.toJson(result);
     }
   };
 
   @Override
   protected String process(Job job) throws ServiceRegistryException {
     Operation op = null;
     String operation = job.getOperation();
     List<String> arguments = job.getArguments();
     String id;
     String targetStore = arguments.get(0);
     VersionImpl version;
     Date start;
     Date end;
     try {
       op = Operation.valueOf(operation);
       switch (op) {
         case MoveById:
           id = arguments.get(1);
           return internalMoveById(id, targetStore);
         case MoveByIdAndVersion:
           id = arguments.get(1);
           version = VersionImpl.mk(Long.parseLong(arguments.get(2)));
           return internalMoveByIdAndVersion(version, id, targetStore);
         case MoveByDate:
           start = new Date(Long.parseLong(arguments.get(1)));
           end = new Date(Long.parseLong(arguments.get(2)));
           return internalMoveByDate(start, end, targetStore);
         case MoveByIdAndDate:
           id = arguments.get(1);
           start = new Date(Long.parseLong(arguments.get(2)));
           end = new Date(Long.parseLong(arguments.get(3)));
           return internalMoveByIdAndDate(id, start, end, targetStore);
         default:
           throw new IllegalArgumentException("Unknown operation '" + operation + "'");
       }
     } catch (NotFoundException e) {
       throw new ServiceRegistryException("Error running job", e);
     } catch (Exception e) {
       throw new ServiceRegistryException("Error handling operation '" + op + "'", e);
     }
   }
 
   /**
    * Spawns a job to move a single snapshot from its current storage to a new target storage location
    *
    * @param version
    *  The {@link Version} to move
    * @param mpId
    *  The mediapackage ID of the snapshot to move
    * @param targetStorage
    *  The {@link RemoteAssetStore} ID where the snapshot should be moved
    * @return
    */
   public Job moveByIdAndVersion(final Version version, final String mpId, final String targetStorage) {
     RequireUtil.notNull(version, "version");
     RequireUtil.notEmpty(mpId, "mpId");
     RequireUtil.notEmpty(targetStorage, "targetStorage");
     List<String> args = new LinkedList<>();
     args.add(targetStorage);
     args.add(mpId);
     args.add(version.toString());
 
     try {
       return serviceRegistry.createJob(JOB_TYPE, Operation.MoveByIdAndVersion.toString(), args, null, true, JOB_LOAD);
     } catch (ServiceRegistryException e) {
       throw new AssetManagerException("Unable to create a job", e);
     }
   }
 
   /**
    * Triggers the move operation inside the {@link AssetManager}
    *
    * @param version
    *  The {@link Version} to move
    * @param mpId
    *  The mediapackage ID of the snapshot to move
    * @param targetStorage
    *  The {@link RemoteAssetStore} ID where the snapshot should be moved
    * @return
    *  The string "OK"
    * @throws NotFoundException
    */
   protected String internalMoveByIdAndVersion(
       final Version version,
       final String mpId,
       final String targetStorage
   ) throws NotFoundException {
     tsam.moveSnapshotToStore(version, mpId, targetStorage);
     return OK;
   }
 
   /**
    * Spawns a job to move a all snapshots of a mediapackage from their current storage to a new target storage location
    *
    * @param mpId
    *  The mediapackage ID of the snapshot to move
    * @param targetStorage
    *  The {@link RemoteAssetStore} ID where the snapshot should be moved
    * @return
    *  The {@link Job}
    */
   public Job moveById(final String mpId, final String targetStorage) {
     RequireUtil.notEmpty(mpId, "mpId");
     RequireUtil.notEmpty(targetStorage, "targetStorage");
     List<String> args = new LinkedList<>();
     args.add(targetStorage);
     args.add(mpId);
 
     try {
       return serviceRegistry.createJob(JOB_TYPE, Operation.MoveById.toString(), args, null, true, NONTERMINAL_JOB_LOAD);
     } catch (ServiceRegistryException e) {
       throw new AssetManagerException("Unable to create a job", e);
     }
   }
 
   /**
    * Moves all the appropriate snapshots to their new home
    *
    * @param mpId
    *  The mediapackage ID of the snapshot to move
    * @param targetStorage
    *  The {@link RemoteAssetStore} ID where the snapshot should be moved
    * @return
    *  The String containing the number of successful and failed moves
    *  [0 OK ][0 FAILED ]
    */
   protected String internalMoveById(final String mpId, final String targetStorage) {
     RichAResult results = tsam.getSnapshotsByIdOrderedByVersion(mpId, true);
     MoveRecordInfo result = moveSnapshots(results, targetStorage);
     return result.toString();
   }
 
 
   /**
    * Spawns a job to move a all snapshots taken between two points from their
    * current storage to a new target storage location
    *
    * @param start
    *  The start {@link Date}
    * @param end
    *  The end {@link Date}
    * @param targetStorage
    *  The {@link RemoteAssetStore} ID where the snapshot should be moved
    * @return
    *  The {@link Job}
    */
   public Job moveByDate(final Date start, final Date end, final String targetStorage) {
     RequireUtil.notNull(start, "start");
     RequireUtil.notNull(end, "end");
     RequireUtil.notNull(targetStorage, "targetStorage");
     List<String> args = new LinkedList<>();
     args.add(targetStorage);
     args.add(Long.toString(start.getTime()));
     args.add(Long.toString(end.getTime()));
 
     try {
       return serviceRegistry.createJob(
           JOB_TYPE, Operation.MoveByDate.toString(), args, null, true, NONTERMINAL_JOB_LOAD);
     } catch (ServiceRegistryException e) {
       throw new AssetManagerException("Unable to create a job", e);
     }
   }
 
   /**
    * Spawns subjobs on a per-snapshot level to move the appropriate snapshots to their new home
    * Moves all the appropriate snapshots to their new home
    *
    * @param start
    *  The start {@link Date}
    * @param end
    *  The end {@link Date}
    * @param targetStorage
    *  The {@link RemoteAssetStore} ID where the snapshot should be moved
    * @return
    *  The number of subjobs spawned
    */
   protected String internalMoveByDate(final Date start, final Date end, final String targetStorage) {
     RichAResult results = tsam.getSnapshotsByDateOrderedById(start, end);
     List<Job> subjobs = spawnSubjobs(results, start, end, targetStorage);
     return Integer.toString(subjobs.size());
   }
 
   /**
    * Spawns a job to move a all snapshots of a given mediapackage taken between
    * two points from their current storage to a new target storage location
    *
    * @param mpId
    *  The mediapackage ID of the snapshot to move
    * @param start
    *  The start {@link Date}
    * @param end
    *  The end {@link Date}
    * @param targetStorage
    *  The {@link RemoteAssetStore} ID where the snapshot should be moved
    * @return
    *  The {@link Job}
    */
   public Job moveByIdAndDate(final String mpId, final Date start, final Date end, final String targetStorage) {
     RequireUtil.notNull(mpId, "mpId");
     RequireUtil.notNull(start, "start");
     RequireUtil.notNull(end, "end");
     RequireUtil.notNull(targetStorage, "targetStorage");
     List<String> args = new LinkedList<>();
     args.add(targetStorage);
     args.add(mpId);
     args.add(Long.toString(start.getTime()));
     args.add(Long.toString(end.getTime()));
 
     try {
       return serviceRegistry.createJob(
           JOB_TYPE, Operation.MoveByIdAndDate.toString(), args, null, true, NONTERMINAL_JOB_LOAD);
     } catch (ServiceRegistryException e) {
       throw new AssetManagerException("Unable to create a job", e);
     }
   }
 
   /**
    * Moves all the appropriate snapshots to their new home
    *
    * @param mpId
    *  The mediapackage ID of the snapshot to move
    * @param start
    *  The start {@link Date}
    * @param end
    *  The end {@link Date}
    * @param targetStorage
    *  The {@link RemoteAssetStore} ID where the snapshot should be moved
    * @return
    *  The JSON String containing the number of successful and failed moves
    *  {"OK": 0,"FAIL": 0}
    */
   protected String internalMoveByIdAndDate(
       final String mpId,
       final Date start,
       final Date end,
       final String targetStorage
   ) {
     RichAResult results = tsam.getSnapshotsByIdAndDateOrderedByVersion(mpId, start, end, true);
     MoveRecordInfo result = moveSnapshots(results, targetStorage);
     return result.toString();
   }
 
   /**
    * Spawns the subjobs based on the stream of records
    *
    * @param records
    *  The stream of records containing the snapshots to move to the new target storage
    * @param targetStorage
    *  The {@link RemoteAssetStore} ID where the snapshot should be moved
    * @return
    *  The set of subjobs
    */
 
   private List<Job> spawnSubjobs(
       final RichAResult records,
       final Date start,
       final Date end,
       final String targetStorage
   ) {
     List<Job> jobs = new LinkedList<>();
     MoveRecordInfo recordInfo = new MoveRecordInfo();
     records.forEach(new Consumer<ARecord>() {
       @Override
       public void accept(ARecord record) {
         Snapshot snap = record.getSnapshot().get();
         String mediaPackageId = snap.getMediaPackage().getIdentifier().toString();
         if (recordInfo.isNewMpId(mediaPackageId)) {
           jobs.add(moveByIdAndDate(mediaPackageId,start,end,targetStorage));
         }
       }
     });
     return jobs;
   }
 
   /**
    * Moves all snapshot based on the stream of records from its current storage to a new target storage location
    *
    * @param records
    *  The stream of records containing the snapshots to move to the new target storage
    * @param targetStorage
    *  The {@link RemoteAssetStore} ID where the snapshot should be moved
    * @return
    *  The {@link MoveRecordInfo}
    */
   private MoveRecordInfo moveSnapshots(final RichAResult records, final String targetStorage) {
     final MoveRecordInfo result = new MoveRecordInfo();
     records.forEach(new Consumer<ARecord>() {
       @Override
       public void accept(ARecord record) {
         Snapshot snap = record.getSnapshot().get();
           try {
             logger.debug("moving Mediapackage {} Version {} from {} to {}",
                 snap.getMediaPackage().getIdentifier().toString(),
                 snap.getVersion().toString(),
                 snap.getStorageId(),
                 targetStorage
             );
             internalMoveByIdAndVersion(snap.getVersion(),
                 snap.getMediaPackage().getIdentifier().toString(),
                 targetStorage
             );
             result.addSuccess();
           } catch (NotFoundException e) {
             result.addFailed();
             logger.warn(e.getMessage());
           }
       }
     });
     return result;
   }
 
   @Reference
   protected void setServiceRegistry(ServiceRegistry serviceRegistry) {
     this.serviceRegistry = serviceRegistry;
   }
 
   @Override
   protected ServiceRegistry getServiceRegistry() {
     return this.serviceRegistry;
   }
 
   @Reference
   protected void setAssetManager(AssetManager assetManager) {
     this.tsam = assetManager;
   }
 
   @Reference
   protected void setSecurityService(SecurityService securityService) {
     this.securityService = securityService;
   }
 
   @Override
   protected SecurityService getSecurityService() {
     return this.securityService;
   }
 
   @Reference
   protected void setUserDirectoryService(UserDirectoryService uds) {
     this.userDirectoryService = uds;
   }
 
   @Override
   protected UserDirectoryService getUserDirectoryService() {
     return this.userDirectoryService;
   }
 
   @Reference
   protected void setOrganizationDirectoryService(OrganizationDirectoryService os) {
     this.organizationDirectoryService = os;
   }
 
   @Override
   protected OrganizationDirectoryService getOrganizationDirectoryService() {
     return this.organizationDirectoryService;
   }
 }