/*
    * 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.query;
  
  import static com.entwinemedia.fn.Stream.$;
  
  import org.opencastproject.assetmanager.api.Property;
  import org.opencastproject.assetmanager.api.Snapshot;
  import org.opencastproject.assetmanager.api.query.ARecord;
  import org.opencastproject.assetmanager.api.query.AResult;
  import org.opencastproject.assetmanager.api.query.ASelectQuery;
  import org.opencastproject.assetmanager.api.query.Order;
  import org.opencastproject.assetmanager.api.query.Predicate;
  import org.opencastproject.assetmanager.impl.AssetManagerImpl;
  import org.opencastproject.assetmanager.impl.RuntimeTypes;
  import org.opencastproject.assetmanager.impl.persistence.EntityPaths;
  import org.opencastproject.assetmanager.impl.persistence.PropertyDto;
  import org.opencastproject.assetmanager.impl.persistence.QPropertyDto;
  import org.opencastproject.assetmanager.impl.persistence.SnapshotDto;
  import org.opencastproject.util.RequireUtil;
  import org.opencastproject.util.data.Function;
  
  import com.entwinemedia.fn.Fn;
  import com.entwinemedia.fn.Fn2;
  import com.entwinemedia.fn.Stream;
  import com.entwinemedia.fn.data.Opt;
  import com.entwinemedia.fn.data.SetB;
  import com.entwinemedia.fn.fns.Booleans;
  import com.mysema.query.Tuple;
  import com.mysema.query.jpa.impl.JPAQuery;
  import com.mysema.query.jpa.impl.JPAQueryFactory;
  import com.mysema.query.types.EntityPath;
  import com.mysema.query.types.Expression;
  import com.mysema.query.types.OrderSpecifier;
  import com.mysema.query.types.expr.BooleanExpression;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.LinkedHashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  import java.util.stream.Collectors;
  
  public abstract class AbstractASelectQuery implements ASelectQuery, SelectQueryContributor, EntityPaths {
    protected static final Logger logger = LoggerFactory.getLogger(AbstractASelectQuery.class);
  
    private final AbstractASelectQuery self = this;
    private final AssetManagerImpl am;
  
    public AbstractASelectQuery(AssetManagerImpl am) {
      this.am = am;
    }
  
    @Override public ASelectQuery where(final Predicate predicate) {
      return new AbstractASelectQuery(am) {
        @Override public SelectQueryContribution contributeSelect(JPAQueryFactory f) {
          final SelectQueryContribution predicateContrib = RuntimeTypes.convert(predicate).contributeSelect(f);
          return self.contributeSelect(f)
                  .addFrom(predicateContrib.from)
                  .addJoin(predicateContrib.join)
                  .andWhere(predicateContrib.where);
        }
  
        @Override public String toString() {
          return "where " + predicate;
        }
      };
    }
  
    @Override public ASelectQuery page(final int offset, final int size) {
      return new AbstractASelectQuery(am) {
        @Override public SelectQueryContribution contributeSelect(JPAQueryFactory f) {
          return self.contributeSelect(f).offset(offset).limit(size);
        }
      };
    }
 
   @Override public ASelectQuery orderBy(final Order order) {
     return new AbstractASelectQuery(am) {
       @Override public SelectQueryContribution contributeSelect(JPAQueryFactory f) {
         final SelectQueryContribution orderContrib = RuntimeTypes.convert(order).contributeSelect(f);
         return self.contributeSelect(f).addOrder(orderContrib.order).andWhere(orderContrib.where);
       }
     };
   }
 
   @Override public AResult run() {
     return am.getDatabase().run(new Function<JPAQueryFactory, AResult>() {
       @Override public AResult apply(JPAQueryFactory f) {
         return run(f);
       }
     });
   }
 
   private AResult run(JPAQueryFactory f) {
     // run query and map the result to records
     final long startTime = System.nanoTime();
     // resolve AST
     final SelectQueryContribution r = contributeSelect(f);
     final boolean toFetchProperties = r.fetch.exists(Booleans.<Expression<?>>eq(QPropertyDto.propertyDto));
     // # create Querydsl query
     final JPAQuery q = f.query();
     // # from
     {
       // Make sure that the snapshotDto is always contained in the from clause because the media package ID and
       //   the ID are always selected.
       // Use a mutable hash set to be able to use the removeAll operation.
       final Set<EntityPath<?>> from = Stream.<EntityPath<?>>mk(Q_SNAPSHOT)
               .append(r.from) // all collected from clauses
               .append(r.join.map(Join.getFrom)) // all from clauses from the joins
               .toSet(SetB.MH);
       // Now remove everything that will be joined. Adding them in both the from and a join
       //   clause is not allowed.
       from.removeAll(r.join.map(Join.getJoin).toSet());
       q.from(JpaFns.toEntityPathArray(from));
     }
     // # join
     if (!r.join.isEmpty()) {
       // Group joins by entity and combine all "on" clauses with "or" expressions.
       // This way there is only one join clause per distinct entity which eliminates the need to alias entities
       //   like this `new QPropertyDto("alias")`.
       // Entity aliasing produces many issues which seem to cause a huge rewrite of the query building mechanism
       //   so it should be prevented at all costs.
       final Map<EntityPath<?>, BooleanExpression> joins = r.join.foldl(
           new HashMap<EntityPath<?>, BooleanExpression>(),
           new Fn2<Map<EntityPath<?>, BooleanExpression>, Join, Map<EntityPath<?>, BooleanExpression>>() {
             @Override
             public Map<EntityPath<?>, BooleanExpression> apply(Map<EntityPath<?>, BooleanExpression> sum, Join join) {
               // get the on expression saved with the join, may be null
               final BooleanExpression existing = sum.get(join.join);
               final BooleanExpression combined;
               // combine the existing and the current expression
               if (existing == null) {
                 combined = join.on;
               } else if (existing.equals(join.on)) {
                 // if both expressions are equal there is no need to combine them
                 combined = existing;
               } else {
                 // if different combine with logical "or"
                 combined = existing.or(join.on);
               }
               sum.put(join.join, combined);
               return sum;
             }
           });
       for (final Map.Entry<EntityPath<?>, BooleanExpression> j : joins.entrySet()) {
         q.leftJoin(j.getKey()).on(j.getValue());
       }
     }
     // # where
     q.where(r.where.orNull());
     // # paging
     for (Integer a : r.offset) {
       q.offset(a);
     }
     for (Integer a : r.limit) {
       q.limit(a);
     }
     // # order
     for (OrderSpecifier<?> a : r.order) {
       q.orderBy(a);
     }
     // # distinct
     if (!toFetchProperties) {
       // if no properties shall be fetched the result set can be distinct
       q.distinct();
     }
     // # fetch
     // create parameters for fetch clause, i.e. Querydsl's list() method
     final List<Expression<?>> fetch;
     {
       // check if the media package ID needs to be selected separately
       if (r.fetch.exists(MandatoryFetch.exists)) {
         fetch = r.fetch.toList();
       } else {
         fetch = r.fetch.append(MandatoryFetch.fetch).toList();
       }
     }
     // Run the query and transform the result into records
     final LinkedHashSet<ARecordImpl> records;
     {
       // run query
       am.getDatabase().logQuery(q);
       final List<Tuple> result = q.list(JpaFns.toExpressionArray(fetch));
       logger.debug("Pure query ms " + (System.nanoTime() - startTime) / 1000000);
       // map result based on the fact whether properties have been fetched or not
       if (!toFetchProperties) {
         // No properties have been fetched -> each result row (tuple) is a distinct record (snapshot).
         records = result.stream()
             .map(tuple -> toARecord(tuple, r))
             .map(record -> {
               Optional<Snapshot> snapshotOpt = record.getSnapshot();
               Snapshot snapshot = null;
               if (snapshotOpt.isPresent()) {
                 // make sure the delivered media package has valid URIs
                 snapshot = am.getHttpAssetProvider().prepareForDelivery(snapshotOpt.get());
               }
               return new ARecordImpl(
                   record.getSnapshotId(),
                   record.getMediaPackageId(),
                   record.getProperties(),
                   snapshot);
             }).collect(Collectors.toCollection(LinkedHashSet::new));
       } else {
         logger.trace("Fetched properties");
         // Properties have been fetched -> there may be multiple rows (tuples)
         // per snapshot because of the join with the property table. Extract
         // records and properties and link them together.
 
         // group properties after their media package ID and make sure that no duplicate properties occur
         final Map<String, Set<Property>> propertiesPerMp = $(result).bind(toProperty).foldl(
             new HashMap<String, Set<Property>>(),
             new Fn2<Map<String, Set<Property>>, Property, Map<String, Set<Property>>>() {
               @Override
               public Map<String, Set<Property>> apply(Map<String, Set<Property>> sum, Property p) {
                 final String mpId = p.getId().getMediaPackageId();
                 final Set<Property> props = sum.get(mpId);
                 if (props != null) {
                   props.add(p);
                 } else {
                   sum.put(mpId, SetB.MH.mk(p));
                 }
                 return sum;
               }
             });
         // group records after their media package ID
         final Map<String, List<ARecordImpl>> distinctRecords = result.stream()
             .map(tuple -> toARecord(tuple, r))
             .collect(Collectors.groupingBy(record -> record.getMediaPackageId()));
         records = distinctRecords.values().stream()
             .flatMap(List::stream)
             .map(record -> {
               final Set<Property> properties = propertiesPerMp.get(record.getMediaPackageId());
               final List<Property> p = properties != null
                   ? properties.stream().collect(Collectors.toList()) : new ArrayList<>();
               Snapshot snapshot = null;
               Optional<Snapshot> snapshotOpt = record.getSnapshot();
               if (snapshotOpt.isPresent()) {
                 // make sure the delivered media package has valid URIs
                 snapshot = am.getHttpAssetProvider().prepareForDelivery(snapshotOpt.get());
               }
               return new ARecordImpl(record.getSnapshotId(), record.getMediaPackageId(), p, snapshot);
             })
             .collect(Collectors.toCollection(LinkedHashSet::new));
       }
     }
     final long searchTime = (System.nanoTime() - startTime) / 1000000;
     logger.debug("Complete query ms " + searchTime);
     LinkedHashSet<ARecord> narrowRecords = new LinkedHashSet<>();
     for (ARecordImpl recordImpl : records) {
       narrowRecords.add(recordImpl);
     }
     return new AResultImpl(
         narrowRecords,
         narrowRecords.size(),
         r.offset.getOr(0),
         r.limit.getOr(-1),
         searchTime
     );
   }
 
   /**
    * Transform a Querydsl result {@link Tuple} into an {@link ARecord}.
    * To do the transformation I need to know what targets have been selected.
    */
   private Fn<Tuple, ARecordImpl> toARecord(final SelectQueryContribution c) {
     return new Fn<Tuple, ARecordImpl>() {
       @Override public ARecordImpl apply(Tuple tuple) {
         final String mediaPackageId;
         SnapshotDto snapshotDto = null;
         final long id;
         // Only fetch the snapshot if it is in the fetch list.
         if (c.fetch.exists(Booleans.<Expression<?>>eq(Q_SNAPSHOT))) {
           snapshotDto = RequireUtil.notNull(tuple.get(Q_SNAPSHOT), "[BUG] snapshot table data");
           id = snapshotDto.getId();
           mediaPackageId = snapshotDto.getMediaPackageId();
         } else {
           // The media package ID and the snapshot's database ID must always be fetched.
           id = RequireUtil.notNull(tuple.get(Q_SNAPSHOT.id), "[BUG] snapshot table id");
           mediaPackageId = RequireUtil.notNull(
               tuple.get(Q_SNAPSHOT.mediaPackageId),
               "[BUG] snapshot table media package id"
           );
         }
         return new ARecordImpl(id, mediaPackageId, new ArrayList<>(), snapshotDto);
       }
     };
   }
 
   private ARecordImpl toARecord(Tuple tuple, final SelectQueryContribution c) {
     final String mediaPackageId;
     SnapshotDto snapshotDto = null;
     final long id;
     // Only fetch the snapshot if it is in the fetch list.
     if (c.fetch.exists(Booleans.<Expression<?>>eq(Q_SNAPSHOT))) {
       snapshotDto = RequireUtil.notNull(tuple.get(Q_SNAPSHOT), "[BUG] snapshot table data");
       id = snapshotDto.getId();
       mediaPackageId = snapshotDto.getMediaPackageId();
     } else {
       // The media package ID and the snapshot's database ID must always be fetched.
       id = RequireUtil.notNull(tuple.get(Q_SNAPSHOT.id), "[BUG] snapshot table id");
       mediaPackageId = RequireUtil.notNull(
           tuple.get(Q_SNAPSHOT.mediaPackageId),
           "[BUG] snapshot table media package id"
       );
     }
     return new ARecordImpl(id, mediaPackageId, new ArrayList<>(), snapshotDto);
   }
 
   private static Fn<Tuple, Opt<Property>> toProperty = new Fn<Tuple, Opt<Property>>() {
     @Override public Opt<Property> apply(Tuple tuple) {
       final PropertyDto dto = tuple.get(Q_PROPERTY);
       return dto != null ? Opt.some(dto.toProperty()) : Opt.<Property>none();
     }
   };
 
   /**
    * Specification of fields whose fetch is mandatory.
    */
   private static final class MandatoryFetch {
     static final Fn<Expression<?>, Boolean> exists =
             Booleans.<Expression<?>>eq(Q_SNAPSHOT)
                     .or(Booleans.<Expression<?>>eq(Q_SNAPSHOT.mediaPackageId))
                     .or(Booleans.<Expression<?>>eq(Q_SNAPSHOT.id));
 
     static final Stream<Expression<?>> fetch = Stream.<Expression<?>>mk(Q_SNAPSHOT.mediaPackageId, Q_SNAPSHOT.id);
   }
 
   private static <A> Stream<A> vary(Stream<? extends A> a) {
     return (Stream<A>) a;
   }
 
   private static <A> int sizeOf(Stream<A> stream) {
     int count = 0;
     for (A ignore : stream) {
       count++;
     }
     return count;
   }
 }