DuplicateEventWorkflowOperationHandler.java

/*
 * 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.workflow.handler.workflow;

import static org.apache.commons.lang3.StringUtils.split;
import static org.apache.commons.lang3.StringUtils.trimToEmpty;

import org.opencastproject.assetmanager.api.AssetManager;
import org.opencastproject.assetmanager.api.Property;
import org.opencastproject.assetmanager.api.PropertyId;
import org.opencastproject.distribution.api.DistributionService;
import org.opencastproject.job.api.Job;
import org.opencastproject.job.api.JobContext;
import org.opencastproject.mediapackage.Catalog;
import org.opencastproject.mediapackage.MediaPackage;
import org.opencastproject.mediapackage.MediaPackageBuilderFactory;
import org.opencastproject.mediapackage.MediaPackageElement;
import org.opencastproject.mediapackage.MediaPackageElementBuilder;
import org.opencastproject.mediapackage.MediaPackageElementBuilderFactory;
import org.opencastproject.mediapackage.MediaPackageElementFlavor;
import org.opencastproject.mediapackage.MediaPackageElements;
import org.opencastproject.mediapackage.MediaPackageException;
import org.opencastproject.mediapackage.Publication;
import org.opencastproject.mediapackage.PublicationImpl;
import org.opencastproject.mediapackage.identifier.IdImpl;
import org.opencastproject.mediapackage.selector.SimpleElementSelector;
import org.opencastproject.metadata.dublincore.DCMIPeriod;
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.EncodingSchemeUtils;
import org.opencastproject.metadata.dublincore.OpencastMetadataCodec;
import org.opencastproject.metadata.dublincore.Precision;
import org.opencastproject.security.api.AccessControlList;
import org.opencastproject.security.api.AclScope;
import org.opencastproject.security.api.AuthorizationService;
import org.opencastproject.security.api.UnauthorizedException;
import org.opencastproject.series.api.SeriesException;
import org.opencastproject.series.api.SeriesService;
import org.opencastproject.serviceregistry.api.ServiceRegistry;
import org.opencastproject.util.JobUtil;
import org.opencastproject.util.NotFoundException;
import org.opencastproject.workflow.api.AbstractWorkflowOperationHandler;
import org.opencastproject.workflow.api.ConfiguredTagsAndFlavors;
import org.opencastproject.workflow.api.WorkflowInstance;
import org.opencastproject.workflow.api.WorkflowOperationException;
import org.opencastproject.workflow.api.WorkflowOperationHandler;
import org.opencastproject.workflow.api.WorkflowOperationInstance;
import org.opencastproject.workflow.api.WorkflowOperationResult;
import org.opencastproject.workflow.api.WorkflowOperationResult.Action;
import org.opencastproject.workflow.handler.distribution.InternalPublicationChannel;
import org.opencastproject.workspace.api.Workspace;

import org.apache.commons.io.FilenameUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
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.net.URI;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;


/**
 * This WOH duplicates an input event.
 */
@Component(
    immediate = true,
    service = WorkflowOperationHandler.class,
    property = {
        "service.description=Duplicate Event Workflow Handler",
        "workflow.operation=duplicate-event"
    }
)
public class DuplicateEventWorkflowOperationHandler extends AbstractWorkflowOperationHandler {
  /**
   * If a target series is given, bundle all the information about it in a class
   */
  private static final class SeriesInformation {
    private final String id;
    private final DublinCoreCatalog dc;
    private final String title;

    private SeriesInformation(String id, DublinCoreCatalog dc, String title) {
      this.id = id;
      this.dc = dc;
      this.title = title;
    }
  }

  private static final Logger logger = LoggerFactory.getLogger(DuplicateEventWorkflowOperationHandler.class);
  private static final String PLUS = "+";
  private static final String MINUS = "-";

  private static final DateFormat ADMIN_UI_DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");

  /** Name of the configuration option that provides the source flavors we are looking for */
  public static final String SOURCE_FLAVORS_PROPERTY = "source-flavors";

  /** Name of the configuration option that provides the source tags we are looking for */
  public static final String SOURCE_TAGS_PROPERTY = "source-tags";

  /** Name of the configuration option that provides the target tags we should apply */
  public static final String TARGET_TAGS_PROPERTY = "target-tags";

  /** Name of the configuration option that provides the number of events to create */
  public static final String NUMBER_PROPERTY = "number-of-events";

  /** Name of the configuration option that provides the maximum number of events to create */
  public static final String MAX_NUMBER_PROPERTY = "max-number-of-events";

  /** Whether to actually use the number suffix (makes sense in conjunction with "set-series-id" */
  public static final String NO_SUFFIX = "no-suffix";

  /** The series ID that should be set on the copies (if unset, uses the same series) */
  public static final String SET_SERIES_ID = "set-series-id";

  /** The new title that should be set on the copies (if unset, uses the old title (copy-number-prefix)) */
  public static final String SET_TITLE = "set-title";

  /** The new startDate that should be set on the copies (if unset, uses the old startDate) */
  public static final String SET_START_DATE = "set-start-date-time";

  /** The default maximum number of events to create. Can be overridden. */
  public static final int MAX_NUMBER_DEFAULT = 25;

  /** The namespaces of the asset manager properties to copy. */
  public static final String PROPERTY_NAMESPACES_PROPERTY = "property-namespaces";

  /** The prefix to use for the number which is appended to the original title of the event. */
  public static final String COPY_NUMBER_PREFIX_PROPERTY = "copy-number-prefix";

  /** AssetManager to use for creating new media packages. */
  private AssetManager assetManager;

  /** The distribution service */
  protected DistributionService distributionService;

  /** The series service */
  private SeriesService seriesService;

  /** The authorization service */
  private AuthorizationService authorizationService;

  /**
   * OSGi setter
   * @param authorizationService
   */
  @Reference
  public void setAuthorizationService(AuthorizationService authorizationService) {
    this.authorizationService = authorizationService;
  }

  /**
   * OSGi setter
   * @param seriesService
   */
  @Reference
  public void setSeriesService(SeriesService seriesService) {
    this.seriesService = seriesService;
  }

  /**
   * Callback for the OSGi declarative services configuration.
   *
   * @param assetManager
   *          the asset manager
   */
  @Reference
  public void setAssetManager(AssetManager assetManager) {
    this.assetManager = assetManager;
  }

  /**
   * Callback for the OSGi declarative services configuration.
   *
   * @param workspace
   *          the workspace
   */
  @Reference
  public void setWorkspace(Workspace workspace) {
    this.workspace = workspace;
  }

  /**
   * Callback for the OSGi declarative services configuration.
   *
   * @param distributionService
   *          the distributionService to set
   */
  @Reference(target = "(distribution.channel=download)")
  public void setDistributionService(DistributionService distributionService) {
    this.distributionService = distributionService;
  }

  @Override
  public WorkflowOperationResult start(final WorkflowInstance workflowInstance, final JobContext context)
          throws WorkflowOperationException {

    ConfiguredTagsAndFlavors tagsAndFlavors = getTagsAndFlavors(workflowInstance,
        Configuration.many, Configuration.many, Configuration.many, Configuration.none);
    final MediaPackage mediaPackage = workflowInstance.getMediaPackage();
    final WorkflowOperationInstance operation = workflowInstance.getCurrentOperation();
    final List<MediaPackageElementFlavor> configuredSourceFlavors = tagsAndFlavors.getSrcFlavors();
    final List<String> configuredSourceTags = tagsAndFlavors.getSrcTags();
    final ConfiguredTagsAndFlavors.TargetTags configuredTargetTags = tagsAndFlavors.getTargetTags();
    final boolean noSuffix = Boolean.parseBoolean(trimToEmpty(operation.getConfiguration(NO_SUFFIX)));
    final String startDateString = trimToEmpty(operation.getConfiguration(SET_START_DATE));
    final String seriesId = trimToEmpty(operation.getConfiguration(SET_SERIES_ID));
    final String title = trimToEmpty(operation.getConfiguration(SET_TITLE));
    final int numberOfEvents = Integer.parseInt(operation.getConfiguration(NUMBER_PROPERTY));
    final String configuredPropertyNamespaces = trimToEmpty(operation.getConfiguration(PROPERTY_NAMESPACES_PROPERTY));
    int maxNumberOfEvents = MAX_NUMBER_DEFAULT;

    if (operation.getConfiguration(MAX_NUMBER_PROPERTY) != null) {
      maxNumberOfEvents = Integer.parseInt(operation.getConfiguration(MAX_NUMBER_PROPERTY));
    }

    if (numberOfEvents > maxNumberOfEvents) {
      throw new WorkflowOperationException("Number of events to create exceeds the maximum of "
          + maxNumberOfEvents + ". Aborting.");
    }

    SeriesInformation series = null;
    AccessControlList seriesAccessControl = null;
    if (!seriesId.isEmpty() && !seriesId.startsWith("${") && !seriesId.endsWith("}")) {
      try {
        final DublinCoreCatalog dc = seriesService.getSeries(seriesId);
        series = new SeriesInformation(seriesId, dc, dc.get(DublinCore.PROPERTY_TITLE).get(0).getValue());
        seriesAccessControl = seriesService.getSeriesAccessControl(seriesId);
      } catch (SeriesException e) {
        throw new WorkflowOperationException(e);
      } catch (NotFoundException e) {
        throw new WorkflowOperationException("couldn't find series for ID \"" + seriesId + "\"");
      } catch (UnauthorizedException e) {
        throw new WorkflowOperationException("not allowed to access series \"" + seriesId + "\"");
      }
    }

    logger.info("Creating {} new media packages from media package with id {}.", numberOfEvents,
        mediaPackage.getIdentifier());

    final String[] propertyNamespaces = split(configuredPropertyNamespaces, ",");
    final String copyNumberPrefix = trimToEmpty(operation.getConfiguration(COPY_NUMBER_PREFIX_PROPERTY));

    final SimpleElementSelector elementSelector = new SimpleElementSelector();
    for (MediaPackageElementFlavor flavor : configuredSourceFlavors) {
      elementSelector.addFlavor(flavor);
    }

    for (String tag : configuredSourceTags) {
      elementSelector.addTag(tag);
    }

    // Filter elements to copy based on input tags and input flavors
    final Collection<MediaPackageElement> elements = elementSelector.select(mediaPackage, false);
    final Collection<Publication> internalPublications = new HashSet<>();

    final List<String> seriesAclTags = new ArrayList<>();
    for (MediaPackageElement e : mediaPackage.getElements()) {
      if (e instanceof Publication) {
        if (InternalPublicationChannel.CHANNEL_ID.equals(((Publication) e).getChannel())) {
          internalPublications.add((Publication) e);
        }
        elements.remove(e); // don't duplicate publications
      }
      if (MediaPackageElements.EPISODE.equals(e.getFlavor())) {
        // Remove episode DC since we will add a new one (with changed title)
        elements.remove(e);
      }
      // The series DC changes
      if (series != null && MediaPackageElements.SERIES.equals(e.getFlavor())) {
        // Remove episode DC since we will add a new one
        elements.remove(e);
      }
      if (series != null && MediaPackageElements.XACML_POLICY_SERIES.equals(e.getFlavor())) {
        seriesAclTags.addAll(Arrays.asList(e.getTags()));
        elements.remove(e);
      }
    }

    final MediaPackageElement[] originalEpisodeDc = mediaPackage.getElementsByFlavor(MediaPackageElements.EPISODE);
    if (originalEpisodeDc.length != 1) {
      throw new WorkflowOperationException("Media package " + mediaPackage.getIdentifier() + " has "
          + originalEpisodeDc.length + " episode dublin cores while it is expected to have exactly 1. Aborting.");
    }

    String mpIds = "";
    String sep = "";
    Map<String, String> properties = new HashMap<>();
    for (int i = 0; i < numberOfEvents; i++) {
      final List<URI> temporaryFiles = new ArrayList<>();
      MediaPackage newMp = null;
      try {
        String newMpId = workflowInstance.getConfiguration("newMpId");
        if (newMpId == null) {
          newMpId = UUID.randomUUID().toString();
        }
        // Clone the media package (without its elements)
        String useTitle;
        if (title.isEmpty() || (title.startsWith("${") && (title.endsWith("}")))) {
          final DublinCoreCatalog dublinCore = DublinCoreUtil.loadEpisodeDublinCore(workspace, mediaPackage).get();
          useTitle = dublinCore.getFirst(DublinCore.PROPERTY_TITLE);
        } else {
          useTitle = title;
        }
        if (!noSuffix) {
          useTitle = String.format("%s (%s %d)", useTitle, copyNumberPrefix, i + 1);
        }
        Date mpDate;
        if (!startDateString.isEmpty() || (!startDateString.startsWith("${") && (!startDateString.endsWith("}")))) {
          try {
            mpDate = ADMIN_UI_DATE_FORMAT.parse(startDateString);
            logger.info("Setting StartDate to {}", mpDate);
          } catch (ParseException ex) {
            logger.warn("Could not parse: {} as date time", startDateString);
            mpDate = mediaPackage.getDate();
            logger.warn("Using original event date {} as default", mpDate);
          }
        } else {
          mpDate = mediaPackage.getDate();
          logger.warn("No date set, using original event date {} as default", mpDate);
        }
        newMp = copyMediaPackage(mediaPackage, series, newMpId, useTitle, mpDate);

        if (series != null) {
          URI newSeriesURI = null;
          String newSeriesId = UUID.randomUUID().toString();
          try (InputStream seriesDCInputStream = IOUtils.toInputStream(series.dc.toXmlString(), "UTF-8")) {
            newSeriesURI = workspace.put(newMpId, newSeriesId, "dublincore.xml", seriesDCInputStream);
          }
          MediaPackageElementBuilder elementBuilder = MediaPackageElementBuilderFactory.newInstance()
                  .newElementBuilder();
          MediaPackageElement newSeriesMpElement = elementBuilder.elementFromURI(newSeriesURI,
                  Catalog.TYPE, MediaPackageElements.SERIES);
          newSeriesMpElement.setIdentifier(newSeriesId);
          newMp.add(newSeriesMpElement);

          if (seriesAccessControl != null) {
            newMp = authorizationService.setAcl(newMp, AclScope.Series, seriesAccessControl).getA();
            for (MediaPackageElement seriesAclMpe
                : newMp.getElementsByFlavor(MediaPackageElements.XACML_POLICY_SERIES)) {
              for (final String tag : seriesAclTags) {
                seriesAclMpe.addTag(tag);
              }
            }
          }
        }

        // Create and add new episode dublin core with changed title
        newMp = copyDublinCore(mediaPackage, originalEpisodeDc[0],
              newMp, series, configuredTargetTags,
              temporaryFiles, mpDate);

        // Clone regular elements
        for (final MediaPackageElement e : elements) {
          final MediaPackageElement element = (MediaPackageElement) e.clone();
          updateTags(element, configuredTargetTags);
          newMp.add(element);
        }

        // Clone internal publications
        for (final Publication originalPub : internalPublications) {
          copyPublication(originalPub, mediaPackage, newMp, configuredTargetTags, temporaryFiles);
        }

        assetManager.takeSnapshot(AssetManager.DEFAULT_OWNER, newMp);

        // Clone properties of media package
        for (String namespace : propertyNamespaces) {
          copyProperties(namespace, mediaPackage, newMp);
        }

        // Store media package ID as workflow property
        properties.put("duplicate_media_package_" + (i + 1) + "_id", newMp.getIdentifier().toString());
        mpIds += sep + newMp.getIdentifier().toString();
        sep = ", ";
      } catch (IOException | MediaPackageException e) {
        throw new WorkflowOperationException(e);
      } finally {
        cleanup(temporaryFiles, Optional.ofNullable(newMp));
      }
    }
    properties.put("duplicate_media_package_ids", mpIds);
    return createResult(mediaPackage, properties, Action.CONTINUE, 0);
  }

  private void cleanup(List<URI> temporaryFiles, Optional<MediaPackage> newMp) {
    // Remove temporary files of new media package
    for (URI temporaryFile : temporaryFiles) {
      try {
        workspace.delete(temporaryFile);
      } catch (NotFoundException e) {
        logger.debug("{} could not be found in the workspace and hence, cannot be deleted.", temporaryFile);
      } catch (IOException e) {
        logger.warn("Failed to delete {} from workspace.", temporaryFile);
      }
    }
    newMp.ifPresent(mp -> {
      try {
        workspace.cleanup(mp.getIdentifier());
      } catch (IOException e) {
        logger.warn("Failed to cleanup the workspace for media package {}", mp.getIdentifier());
      }
    });
  }

  private void updateTags(MediaPackageElement element, ConfiguredTagsAndFlavors.TargetTags targetTags) {
    element.generateIdentifier();
    applyTargetTagsToElement(targetTags, element);
  }

  private MediaPackage copyMediaPackage(
      final MediaPackage source,
      final SeriesInformation series,
      final String newMpId,
      final String title,
      final Date startDate
  ) throws WorkflowOperationException {
    // We are not using MediaPackage.clone() here, since it does "too much" for us (e.g. copies all the attachments)
    MediaPackage destination;
    try {
      destination = MediaPackageBuilderFactory.newInstance().newMediaPackageBuilder().createNew(new IdImpl(newMpId));
    } catch (MediaPackageException e) {
      logger.error("Failed to create media package " + e.getLocalizedMessage());
      throw new WorkflowOperationException(e);
    }
    logger.info("Created mediapackage {}", destination);
    destination.setDate(source.getDate());
    if (series != null) {
      destination.setSeries(series.id);
      destination.setSeriesTitle(series.title);
    } else {
      destination.setSeries(source.getSeries());
      destination.setSeriesTitle(source.getSeriesTitle());
    }
    destination.setDuration(source.getDuration());
    destination.setLanguage(source.getLanguage());
    destination.setLicense(source.getLicense());
    destination.setDate(startDate);
    destination.setTitle(title);
    return destination;
  }

  private void copyPublication(
      Publication sourcePublication,
      MediaPackage source,
      MediaPackage destination,
      ConfiguredTagsAndFlavors.TargetTags targetTags,
      List<URI> temporaryFiles) throws WorkflowOperationException {
    final String newPublicationId = UUID.randomUUID().toString();
    final Publication newPublication = PublicationImpl.publication(newPublicationId,
        InternalPublicationChannel.CHANNEL_ID, null, null);

    // re-distribute elements of publication to internal publication channel
    final Collection<MediaPackageElement> sourcePubElements = new HashSet<>();
    sourcePubElements.addAll(Arrays.asList(sourcePublication.getAttachments()));
    sourcePubElements.addAll(Arrays.asList(sourcePublication.getCatalogs()));
    sourcePubElements.addAll(Arrays.asList(sourcePublication.getTracks()));
    for (final MediaPackageElement e : sourcePubElements) {
      try {
        // We first have to copy the media package element into the workspace
        final MediaPackageElement element = (MediaPackageElement) e.clone();
        try (InputStream inputStream = workspace.read(element.getURI())) {
          final URI tmpUri = workspace.put(destination.getIdentifier().toString(), element.getIdentifier(),
              FilenameUtils.getName(element.getURI().toString()), inputStream);
          temporaryFiles.add(tmpUri);
          element.generateIdentifier();
          element.setURI(tmpUri);
        }

        // Now we can distribute it to the new media package
        destination.add(element); // Element has to be added before it can be distributed
        final Job job = distributionService.distribute(InternalPublicationChannel.CHANNEL_ID, destination,
            element.getIdentifier());
        final MediaPackageElement distributedElement =
            JobUtil.payloadAsMediaPackageElement(serviceRegistry, job);
        destination.remove(element);

        updateTags(distributedElement, targetTags);

        PublicationImpl.addElementToPublication(newPublication, distributedElement);
      } catch (Exception exception) {
        throw new WorkflowOperationException(exception);
      }
    }

    // Using an altered copy of the source publication's URI is a bit hacky,
    // but it works without knowing the URI pattern...
    String publicationUri = sourcePublication.getURI().toString();
    publicationUri = publicationUri.replace(source.getIdentifier().toString(), destination.getIdentifier().toString());
    publicationUri = publicationUri.replace(sourcePublication.getIdentifier(), newPublicationId);
    newPublication.setURI(URI.create(publicationUri));
    destination.add(newPublication);
  }

  private MediaPackage copyDublinCore(
      final MediaPackage source,
      final MediaPackageElement sourceDublinCore,
      final MediaPackage destination,
      final SeriesInformation series,
      final ConfiguredTagsAndFlavors.TargetTags targetTags,
      final List<URI> temporaryFiles,
      final Date creationDate
  ) throws WorkflowOperationException {
    final DublinCoreCatalog destinationDublinCore = DublinCoreUtil.loadEpisodeDublinCore(workspace, source).get();
    destinationDublinCore.generateIdentifier();
    destinationDublinCore.setURI(sourceDublinCore.getURI());
    destinationDublinCore.set(DublinCore.PROPERTY_CREATED,
        OpencastMetadataCodec.encodeDate(creationDate, Precision.Second));
    destinationDublinCore.set(DublinCore.PROPERTY_TITLE, destination.getTitle());
    if (StringUtils.isNotBlank(destinationDublinCore.getFirst(DublinCore.PROPERTY_TEMPORAL))) {
      DublinCoreValue eventTime = EncodingSchemeUtils.encodePeriod(
          new DCMIPeriod(creationDate, creationDate), Precision.Second);
      destinationDublinCore.set(DublinCore.PROPERTY_TEMPORAL, eventTime);
    }
    if (series != null) {
      destinationDublinCore.set(DublinCore.PROPERTY_IS_PART_OF, series.id);
    }
    try (InputStream inputStream = IOUtils.toInputStream(destinationDublinCore.toXmlString(), "UTF-8")) {
      final String elementId = UUID.randomUUID().toString();
      final URI newUrl = workspace.put(destination.getIdentifier().toString(), elementId, "dublincore.xml",
          inputStream);
      temporaryFiles.add(newUrl);
      final MediaPackageElement mpe = destination.add(newUrl, MediaPackageElement.Type.Catalog,
          MediaPackageElements.EPISODE);
      for (String tag : sourceDublinCore.getTags()) {
        mpe.addTag(tag);
      }
      updateTags(mpe, targetTags);
      mpe.setIdentifier(elementId);
    } catch (IOException e) {
      throw new WorkflowOperationException(e);
    }
    return destination;
  }

  private void copyProperties(String namespace, MediaPackage source, MediaPackage destination) {
    List<Property> properties = assetManager.selectProperties(source.getIdentifier().toString(), namespace);
    if (properties.isEmpty()) {
      logger.info("No properties to copy for media package {}, namespace {}.", source.getIdentifier(), namespace);
      return;
    }
    for (final Property p : properties) {
      final PropertyId newPropId = PropertyId.mk(destination.getIdentifier().toString(), namespace, p.getId()
          .getName());
      assetManager.setProperty(Property.mk(newPropId, p.getValue()));
    }
  }

  @Reference
  @Override
  public void setServiceRegistry(ServiceRegistry serviceRegistry) {
    super.setServiceRegistry(serviceRegistry);
  }

}