/*
 * (c) 2020 Pluto Inc.
 */
package tv.pluto.mongo.utils;

import static com.mongodb.MongoClient.getDefaultCodecRegistry;
import static com.mongodb.client.model.Projections.fields;
import static com.mongodb.client.model.Projections.include;
import static com.mongodb.client.model.Sorts.ascending;
import static com.mongodb.client.model.Sorts.descending;
import static com.mongodb.client.model.Sorts.orderBy;
import static org.bson.codecs.configuration.CodecRegistries.fromProviders;
import static org.bson.codecs.configuration.CodecRegistries.fromRegistries;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.mongodb.BasicDBList;
import com.mongodb.BasicDBObject;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Aggregates;
import com.mongodb.client.model.Collation;
import com.mongodb.client.model.CollationStrength;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.Projections;
import com.mongodb.client.result.DeleteResult;
import com.mongodb.client.result.InsertManyResult;
import com.mongodb.client.result.InsertOneResult;
import com.plutotv.test.lib.LogUtils;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Type;
import java.nio.file.FileVisitOption;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Random;
import java.util.Set;
import java.util.TimeZone;
import java.util.stream.Collectors;
import org.bson.Document;
import org.bson.codecs.configuration.CodecConfigurationException;
import org.bson.codecs.configuration.CodecProvider;
import org.bson.codecs.configuration.CodecRegistries;
import org.bson.codecs.configuration.CodecRegistry;
import org.bson.codecs.pojo.PojoCodecProvider;
import org.bson.conversions.Bson;
import org.bson.types.ObjectId;
import org.testng.Assert;
import tv.pluto.core.MongoDbManager;
import tv.pluto.mongo.dto.ContentFeedQuery;
import tv.pluto.mongo.dto.DocumentQuery;
import tv.pluto.mongo.entities.activeregions.ActiveRegions;
import tv.pluto.mongo.entities.appnames.AppNames;
import tv.pluto.mongo.entities.campaigns.Campaigns;
import tv.pluto.mongo.entities.clips.AvailableVodContentRecommendation;
import tv.pluto.mongo.entities.clips.Clips;
import tv.pluto.mongo.entities.common.ObjectIdContentIdSet;
import tv.pluto.mongo.entities.episodes.AgeRange;
import tv.pluto.mongo.entities.episodes.AgeRangeEpisode;
import tv.pluto.mongo.entities.episodes.Episodes;
import tv.pluto.mongo.entities.featuredgroup.FeaturedGroup;
import tv.pluto.mongo.entities.gracenoruns.GracenoteRuns;
import tv.pluto.mongo.entities.gracenotelogs.GracenoteLogs;
import tv.pluto.mongo.entities.gracenotemetadata.GracenoteMetadata;
import tv.pluto.mongo.entities.partners.Partners;
import tv.pluto.mongo.entities.placeholder.Clip;
import tv.pluto.mongo.entities.timelinedrafts.TimelineDrafts;
import tv.pluto.mongo.entities.timelines.Timelines;
import tv.pluto.mongo.entities.timelinespast.TimelinesPast;
import tv.pluto.mongo.entities.users.User;
import tv.pluto.mongo.entities.vod.VodCategories;
import tv.pluto.mongo.entities.vod.VodCategoryEntries;
import tv.pluto.mongo.utils.json.mongocodec.DateCodec;
import tv.pluto.mongo.utils.json.mongocodec.IntegerCodec;
import tv.pluto.mongo.utils.json.mongocodec.TimelinesPastCodec;
import tv.pluto.utils.CommonUtils;

public class MongoUtils {

  protected static CodecRegistry pojoCodecRegistry;
  protected MongoCollection<ActiveRegions> collectionActiveRegions;
  public MongoCollection<AgeRange> collectionAgeRanges;
  public MongoCollection<AgeRangeEpisode> collectionAgeRangeEpisode;
  protected MongoCollection<AppNames> collectionAppNames;
  protected MongoCollection<Clips> collectionClips;
  protected MongoCollection<Clip> collectionClip;
  protected MongoCollection<Campaigns> collectionCampaigns;
  protected MongoCollection<Episodes> collectionEpisodes;
  protected MongoCollection<GracenoteLogs> collectionGracenoteLogs;
  protected MongoCollection<GracenoteMetadata> collectionGracenoteMetadata;
  protected MongoCollection<GracenoteRuns> collectionGracenoteRuns;
  protected MongoCollection<FeaturedGroup> collectionFeaturedGroup;
  protected MongoCollection<Partners> collectionPartners;
  public MongoCollection<Timelines> collectionTimelines;
  public MongoCollection<TimelineDrafts> collectionTimelineDrafts;
  public MongoCollection<TimelinesPast> collectionTimelinesPast;
  public MongoCollection<User> collectionUser;
  protected MongoCollection<VodCategoryEntries> collectionVodCategoryEntries;
  protected MongoCollection<tv.pluto.mongo.entities.vodcategoryentries.VodCategoryEntries>
      collectionVodCategoriesEntries;
  protected MongoCollection<VodCategories> collectionVodCategories;
  protected MongoCollection<AvailableVodContentRecommendation> availableVodContentRecommendations;
  protected MongoCollection<?> genericCollection;

  private final MongoDatabase mongoDatabase;

  public LocalDateTime updatedAtMin;
  public LocalDateTime updatedAtMax;
  public String updatedAtMinString;
  public String updatedAtMaxString;

  protected Gson gson = new GsonBuilder().setPrettyPrinting().create();

  private static final String ENTITIES_PATH = "src/main/java/tv/pluto/mongo/entities";

  private static final String[] ENTITIES_PACKAGES =
      Arrays.stream(listDirectories(ENTITIES_PATH))
          .map(
              path ->
                  path.replaceAll(".+?(?=tv)", "") // remove all before tv
                      .replace("\\", ".") // replace \ with . - windows
                      .replace("/", ".")) // replace / with . - linux
          .toArray(String[]::new);

  public static final int LIMIT =
      System.getProperty("limit") == null ? 10 : Integer.parseInt(System.getProperty("limit"));
  public static final String INVALID_ID_VALUE = "099999999999999999999999";
  public static final String NOT_ID_VALUE = "notId";

  // Constructor accepting MongoDatabase
  public MongoUtils(MongoDatabase database) {
    this.mongoDatabase = database;
  }

  // Default Constructor
  public MongoUtils() {
    this(MongoDbManager.getDatabase()); // Use shared database instance
    setCodecRegistry();
  }

  // Constructor for specific entity class
  public <T> MongoUtils(T entityClass) {
    this(MongoDbManager.getDatabase());
    setCodecRegistry(entityClass);
  }

  // Constructor for specific collection and entity class
  public <T> MongoUtils(String collectionName, T entityClass) {
    this(MongoDbManager.getDatabase());
    setCodecRegistry(collectionName, entityClass);
  }

  /**
   * Generic method for getting random item from list
   *
   * @param list list of objects
   * @param <T> generic param
   * @return random item from list
   */
  public <T> T getRandomItem(List<T> list) {
    return list.get(new Random().nextInt(list.size()));
  }

  /**
   * Transforms list of ObjectIds into list of strings
   *
   * @param ids list of ids to transform
   * @return list ids
   */
  public static List<String> transformIdsToStrings(List<ObjectId> ids) {
    if (!ids.isEmpty()) {
      return ids.stream()
          .map(
              id -> {
                if (id == null) {
                  return "null";
                } else {
                  return id.toString();
                }
              })
          .collect(Collectors.toList());
    } else {
      return Collections.emptyList();
    }
  }

  /**
   * Transforms list of strings into list of ObjectIds
   *
   * @param ids list of ids to transform
   * @return list ids
   */
  public static List<ObjectId> transformStringsToIds(List<String> ids) {
    if (!ids.isEmpty()) {
      return ids.stream().map(ObjectId::new).collect(Collectors.toList());
    } else {
      return Collections.emptyList();
    }
  }

  /**
   * Transforms date to ISO format which is used in database
   *
   * @param date date to transform
   * @return date in string format
   */
  public static String toMongoDate(Date date) {
    if (date != null) {
      SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", Locale.US);
      format.setTimeZone(TimeZone.getTimeZone("UTC"));
      return format.format(date);
    } else {
      return "";
    }
  }

  public static Date stringToDate(String str) {
    if (str == null || str.isEmpty()) {
      return null;
    } else {
      Instant parse = Instant.parse(str);
      return Date.from(parse);
    }
  }

  /**
   * Add/remove years to date
   *
   * @param date date to update
   * @param years to add. Plus years if value positive, minus years if negative
   * @return updated date
   */
  public static Date addYearsToDate(Date date, int years) {
    Calendar calendar = Calendar.getInstance();
    calendar.setTime(date);
    calendar.add(Calendar.YEAR, years);
    return calendar.getTime();
  }

  /** Performs mapping using pojoCodecProvider */
  private void setCodecRegistry() {
    CodecProvider pojoCodecProvider =
        PojoCodecProvider.builder().register(ENTITIES_PACKAGES).build();

    pojoCodecRegistry =
        fromRegistries(
            CodecRegistries.fromCodecs(new IntegerCodec()),
            CodecRegistries.fromCodecs(new DateCodec()),
            getDefaultCodecRegistry(),
            fromProviders(pojoCodecProvider));

    CodecRegistry pojoCodecRegistryPast =
        fromRegistries(
            CodecRegistries.fromCodecs(new IntegerCodec()),
            CodecRegistries.fromCodecs(new DateCodec()),
            CodecRegistries.fromCodecs(new TimelinesPastCodec()),
            getDefaultCodecRegistry(),
            fromProviders(pojoCodecProvider));

    collectionActiveRegions =
        mongoDatabase
            .getCollection("activeregions", ActiveRegions.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionAppNames =
        mongoDatabase
            .getCollection("appnames", AppNames.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionEpisodes =
        mongoDatabase
            .getCollection("episodes", Episodes.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionAgeRangeEpisode =
        mongoDatabase
            .getCollection("episodes", AgeRangeEpisode.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionAgeRanges =
        mongoDatabase
            .getCollection("ageranges", AgeRange.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionGracenoteLogs =
        mongoDatabase
            .getCollection("gracenote-logs", GracenoteLogs.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionGracenoteMetadata =
        mongoDatabase
            .getCollection("gracenote-metadata", GracenoteMetadata.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionGracenoteRuns =
        mongoDatabase
            .getCollection("gracenote-runs", GracenoteRuns.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionFeaturedGroup =
        mongoDatabase
            .getCollection("featuredgroups", FeaturedGroup.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionPartners =
        mongoDatabase
            .getCollection("partners", Partners.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionClips =
        mongoDatabase.getCollection("clips", Clips.class).withCodecRegistry(pojoCodecRegistry);
    availableVodContentRecommendations =
        mongoDatabase
            .getCollection("clips", AvailableVodContentRecommendation.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionClip =
        mongoDatabase.getCollection("clips", Clip.class).withCodecRegistry(pojoCodecRegistry);
    collectionCampaigns =
        mongoDatabase
            .getCollection("campaigns", Campaigns.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionVodCategories =
        mongoDatabase
            .getCollection("vodcategories", VodCategories.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionVodCategoryEntries =
        mongoDatabase
            .getCollection("vodcategoryentries", VodCategoryEntries.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionVodCategoriesEntries =
        mongoDatabase
            .getCollection(
                "vodcategoryentries",
                tv.pluto.mongo.entities.vodcategoryentries.VodCategoryEntries.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionTimelines =
        mongoDatabase
            .getCollection("timelines", Timelines.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionTimelineDrafts =
        mongoDatabase
            .getCollection("timelinedrafts", TimelineDrafts.class)
            .withCodecRegistry(pojoCodecRegistry);
    collectionTimelinesPast =
        mongoDatabase
            .getCollection("timelines-past", TimelinesPast.class)
            .withCodecRegistry(pojoCodecRegistryPast);
    collectionUser =
        mongoDatabase.getCollection("users", User.class).withCodecRegistry(pojoCodecRegistry);
  }

  /** Performs mapping using pojoCodecProvider */
  private <T> void setCodecRegistry(T entityClass) {
    CodecProvider pojoCodecProvider =
        PojoCodecProvider.builder().register(ENTITIES_PACKAGES).build();

    pojoCodecRegistry =
        fromRegistries(
            CodecRegistries.fromCodecs(new IntegerCodec()),
            CodecRegistries.fromCodecs(new DateCodec()),
            getDefaultCodecRegistry(),
            fromProviders(pojoCodecProvider));

    genericCollection =
        mongoDatabase
            .getCollection(
                entityClass.getClass().getSimpleName().toLowerCase(), entityClass.getClass())
            .withCodecRegistry(pojoCodecRegistry);
  }

  /** Performs mapping using pojoCodecProvider */
  private <T> void setCodecRegistry(String collectionName, T entityClass) {
    CodecProvider pojoCodecProvider =
        PojoCodecProvider.builder().register(ENTITIES_PACKAGES).build();
    pojoCodecRegistry = fromRegistries(getDefaultCodecRegistry(), fromProviders(pojoCodecProvider));

    genericCollection =
        mongoDatabase
            .getCollection(collectionName, entityClass.getClass())
            .withCodecRegistry(pojoCodecRegistry);
  }

  /**
   * Map grpc object to POJO class
   *
   * @param grpcEntity grpc response
   * @param pojoClass POJO class for mapping
   * @param <T> generic type of grpc response
   * @param <U> generic type of method response
   * @return generic POJO type
   */
  public <T, U> U mapGrpcObjectToPojo(T grpcEntity, Type pojoClass) {
    return gson.fromJson(gson.toJson(grpcEntity).replaceAll("_\":", "\":"), pojoClass);
  }

  /**
   * Map list of grpc objects to list of POJO class type
   *
   * @param grpcEntities list of grpc objects
   * @param pojoClass POJO class for mapping
   * @return list of generic POJO type
   */
  @SuppressWarnings("unchecked")
  public <T> List<T> mapGrpcObjectsToPojo(List<?> grpcEntities, Type pojoClass) {
    return grpcEntities.stream()
        .map(entity -> (T) mapGrpcObjectToPojo(entity, pojoClass))
        .collect(Collectors.toList());
  }

  public <T> Set<T> mapGrpcObjectsToPojoSet(List<?> grpcEntities, Type pojoClass) {
    return grpcEntities.stream()
        .map(entity -> (T) mapGrpcObjectToPojo(entity, pojoClass))
        .collect(Collectors.toSet());
  }

  /**
   * Map map to POJO class type
   *
   * @param initMap initial map
   * @param pojoClass POJO class for mapping
   * @return map with POJO objects as values
   */
  public Map<String, ?> mapMapToPojo(Map<String, ?> initMap, Type pojoClass) {
    Map<String, ?> result = new HashMap<>();

    initMap.forEach((k, v) -> result.put(k, this.mapGrpcObjectToPojo(v, pojoClass)));

    return result;
  }

  /**
   * Get a document by id.
   *
   * @param id id of the document
   * @return document
   */
  @SuppressWarnings("unchecked")
  public <T> T getDocumentById(String id) {
    if (id == null) {
      throw new NullPointerException("ObjectId is null.");
    } else if (id.isEmpty()) {
      throw new IllegalArgumentException("ObjectId is empty.");
    }

    BasicDBObject query = new BasicDBObject();
    query.put("_id", new ObjectId(id));

    return (T) genericCollection.find(query).first();
  }

  public List getDocumentsByFieldAndId(String field, String id) {
    BasicDBObject query = new BasicDBObject();
    query.put(field, new ObjectId(id));
    Iterator iterator = genericCollection.find(query).iterator();
    return CommonUtils.iteratorToList(iterator);
  }

  public <T> T getDocumentByFieldAndId(String field, String id) {
    BasicDBObject query = new BasicDBObject();
    query.put(field, new ObjectId(id));
    return (T) genericCollection.find(query).first();
  }

  public <T> T getDocumentsByFieldAndValue(String field, String value) {
    BasicDBObject query = new BasicDBObject();
    query.put(field, value);
    Iterator<?> iterator = genericCollection.find(query).iterator();
    return (T) CommonUtils.iteratorToList(iterator);
  }

  public <T> T getDocumentByFieldAndValue(String field, int value) {
    BasicDBObject query = new BasicDBObject();
    query.put(field, value);
    return (T) genericCollection.find(query).first();
  }

  public <T> T getDocumentByFieldAndValue(String field, String value) {
    BasicDBObject query = new BasicDBObject();
    query.put(field, value);
    return (T) genericCollection.find(query).first();
  }

  /**
   * Get a document by id.
   *
   * @param id id of the document
   * @param fields array of fields for projection
   * @return document
   */
  @SuppressWarnings("unchecked")
  public <T> T getDocumentById(String id, String[] fields) {
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new ObjectId(id));
    return (T) genericCollection.find(query).projection(include(fields)).first();
  }

  /**
   * Get a list of documents by ids.
   *
   * @param ids list of documents ids
   * @return List of documents
   */
  @SuppressWarnings("unchecked")
  public <T> List<T> getDocumentsByIds(List<String> ids) {
    List<T> result = new ArrayList<>();
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new BasicDBObject("$in", transformStringsToIds(ids)));
    genericCollection.find(query).forEach(e -> result.add((T) e));
    return result;
  }

  public <T> List<T> getDocumentsByIdsProjection(List<String> ids, List<String> projection) {
    Bson projectionFields =
        projection != null && !projection.isEmpty() ? include(projection) : null;

    List<T> result = new ArrayList<>();
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new BasicDBObject("$in", transformStringsToIds(ids)));
    genericCollection.find(query).projection(projectionFields).forEach(e -> result.add((T) e));
    return result;
  }

  public <T> List<T> getDocumentsByIds(List<String> ids, int sort) {
    Bson bSort = sort == 1 ? orderBy(ascending("_id")) : orderBy(descending("_id"));
    List<T> result = new ArrayList<>();
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new BasicDBObject("$in", transformStringsToIds(ids)));
    genericCollection.find(query).sort(bSort).forEach(e -> result.add((T) e));
    return result;
  }

  public <T> List<T> getDocumentsByIds(String field, List<String> ids) {
    List<T> result = new ArrayList<>();
    BasicDBObject query = new BasicDBObject();
    query.put(field, new BasicDBObject("$in", transformStringsToIds(ids)));
    genericCollection.find(query).forEach(e -> result.add((T) e));
    return result;
  }

  /**
   * Get a list of documents by ids.
   *
   * @param ids list of documents ids
   * @return List of documents
   */
  @SuppressWarnings("unchecked")
  public <T> List<T> getDocumentsByIds(List<String> ids, String[] fields) {
    List<T> result = new ArrayList<>();
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new BasicDBObject("$in", transformStringsToIds(ids)));
    genericCollection
        .find(query)
        .projection(fields(include(fields)))
        .forEach(e -> result.add((T) e));
    return result;
  }

  @SuppressWarnings("unchecked")
  public <T> List<T> getDocumentsByIds(List<String> ids, Bson filter) {
    List<T> result = new ArrayList<>();
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new BasicDBObject("$in", transformStringsToIds(ids)));
    genericCollection.find(query).filter(filter).forEach(e -> result.add((T) e));
    return result;
  }

  public <T> List<T> getDocumentsByKeyListValues(String key, List<String> listValues) {
    List<T> result = new ArrayList<>();
    BasicDBObject query = new BasicDBObject();
    query.put(key, new BasicDBObject("$in", listValues));
    genericCollection.find(query).forEach(e -> result.add((T) e));
    return result;
  }

  /**
   * Get limited list of documents
   *
   * @param limit limit for query
   * @return List of documents
   */
  @SuppressWarnings("unchecked")
  public <T> List<T> getDocuments(int limit) {
    List<T> result = new ArrayList<>();
    BasicDBObject query = new BasicDBObject("archived", new BasicDBObject("$ne", true));
    genericCollection.find(query).limit(limit).forEach(e -> result.add((T) e));
    return result;
  }

  public <T> List<T> getAllDocuments() {
    List<T> result = new ArrayList<>();
    genericCollection.find().forEach(e -> result.add((T) e));
    return result;
  }

  @SuppressWarnings("unchecked")
  public <T> List<T> getDocuments(int limit, String[] projectionFields) {
    BasicDBObject projection = new BasicDBObject();
    for (String projectionField : projectionFields) {
      projection.put(projectionField, 1);
    }

    List<T> result = new ArrayList<>();
    genericCollection
        .find()
        .projection(projection)
        .limit(limit)
        .forEach(document -> result.add((T) document));

    return result;
  }

  public <T> List<T> getRandomDocuments(int limit) {
    List<T> result = new ArrayList<>();
    genericCollection.find().limit(999).forEach(e -> result.add((T) e));
    Collections.shuffle(result);
    return result.stream().limit(limit).collect(Collectors.toList());
  }

  @SuppressWarnings("unchecked")
  public <T> List<T> getDocuments(BasicDBObject query, int limit) {
    List<T> result = new ArrayList<>();
    genericCollection.find(query).limit(limit).forEach(e -> result.add((T) e));
    return result;
  }

  /** Get count all documents in the collection */
  public long getDocumentsCountsInTheCollection() {
    long totalObjects = genericCollection.countDocuments();
    LogUtils.logInfoMessage("Total objects in the collection: " + totalObjects);
    return totalObjects;
  }

  /**
   * Get limited list of documents with offset
   *
   * @param limit limit for query
   * @param offset offset for query
   * @return List of documents
   */
  @SuppressWarnings("unchecked")
  public <T> List<T> getDocuments(int limit, int offset, Bson sort) {
    List<T> result = new ArrayList<>();
    genericCollection.find().limit(limit).skip(offset).sort(sort).forEach(e -> result.add((T) e));
    return result;
  }

  /**
   * Get limited list of documents with offset
   *
   * @param documentQuery query object
   * @return List of documents
   */
  @SuppressWarnings("unchecked")
  public <T> List<T> getDocumentsWithCollation(DocumentQuery documentQuery) {
    List<T> result = new ArrayList<>();
    var findIterable = genericCollection.find();

    if (documentQuery.getProjections() != null) {
      findIterable = findIterable.projection(documentQuery.getProjections());
    }

    findIterable
        .limit(documentQuery.getLimit())
        .skip(documentQuery.getOffset())
        .sort(documentQuery.getSort())
        .collation(documentQuery.getCollation())
        .forEach(e -> result.add((T) e));

    return result;
  }

  /**
   * Ger random document
   *
   * @return random document
   */
  public <T> T getRandomDocument() {
    return getRandomItem(getDocuments(MongoUtils.LIMIT));
  }

  public <T> T getRandomDocument(String[] projectionFields) {
    return getRandomItem(getDocuments(MongoUtils.LIMIT, projectionFields));
  }

  @SuppressWarnings("unchecked")
  public <T> List<T> getDocumentsForContentFeeds(ContentFeedQuery contentFeedQuery) {
    List<T> result = new ArrayList<>();

    BasicDBObject query =
        new BasicDBObject(
            "updatedAt",
            new BasicDBObject("$gte", contentFeedQuery.getUpdatedAtMin())
                .append("$lte", contentFeedQuery.getUpdatedAtMax()));

    var findIterable = genericCollection.find(query);

    if (contentFeedQuery.getProjections() != null) {
      findIterable.projection(contentFeedQuery.getProjections());
    }

    findIterable
        .sort(new BasicDBObject("updatedAt", 1))
        .limit(contentFeedQuery.getLimit())
        .forEach(e -> result.add((T) e));

    return result;
  }

  protected <T> LocalDateTime getUpdatedAtMin(List<T> documents) {
    LocalDateTime updatedAtMin = null;

    T randomItem = getRandomItem(documents);

    try {
      updatedAtMin =
          (LocalDateTime) randomItem.getClass().getMethod("getUpdatedAt").invoke(randomItem);
    } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
      LogUtils.logSevereMessageThenFail(e.getMessage());
    }

    return updatedAtMin;
  }

  @SuppressWarnings("unchecked")
  protected <T> List<T> getDocumentsContentFeeds() {
    List<T> result = new ArrayList<>();
    Bson notNullFilter = Filters.not(Filters.eq("updatedAt", null));

    genericCollection
        .find()
        .sort(new BasicDBObject("updatedAt", 1))
        .limit(LIMIT)
        .filter(notNullFilter)
        .forEach(e -> result.add((T) e));

    return result;
  }

  public <T> List<T> getDocumentsForFeedsSyncStartTime(LocalDateTime startDateTime) {
    List<T> result = new ArrayList<>();
    BasicDBObject query = new BasicDBObject("updatedAt", new BasicDBObject("$gte", startDateTime));

    genericCollection.find(query).forEach(e -> result.add((T) e));

    return result;
  }

  public <T> List<T> getDocumentsForFeedsSyncStartTime(
      Instant starTime, String value, String channel) {
    List<T> result = new ArrayList<>();
    BasicDBObject query =
        new BasicDBObject("start", new BasicDBObject(value, starTime))
            .append("channel", new ObjectId(channel));

    genericCollection.find(query).forEach(e -> result.add((T) e));
    return result;
  }

  @SuppressWarnings("unchecked")
  protected <T> List<T> getChannelsDocumentsContentFeeds() {
    List<T> result = new ArrayList<>();

    Bson notNullFilter = Filters.not(Filters.eq("updatedAt", null));

    List<BasicDBObject> mongoOrList =
        Arrays.asList(
            new BasicDBObject("plutoOfficeOnly", false),
            new BasicDBObject("plutoOfficeOnly", new BasicDBObject("$exists", false)));

    BasicDBObject query = new BasicDBObject().append("archived", false).append("$or", mongoOrList);

    try {
      genericCollection
          .find(query)
          .sort(new BasicDBObject("updatedAt", 1))
          .filter(notNullFilter)
          .filter(Filters.ne("_id", new ObjectId("545943f1c9f133a519bbac92")))
          .forEach(e -> result.add((T) e));
    } catch (CodecConfigurationException e) {
      LogUtils.logSevereMessageThenFail(e.getMessage());
    }

    return result;
  }

  /** Get a full list of all objects from the collection */
  @SuppressWarnings("unchecked")
  public <T> List<T> getListOfAllObjects() {
    List<T> result = new ArrayList<>();
    genericCollection.find().forEach(e -> result.add((T) e));
    return result;
  }

  public <T> List<T> getDistinctListOfTheField(String field, Class resultClass) {
    List<T> result = new ArrayList<>();
    genericCollection.distinct(field, resultClass).forEach(e -> result.add((T) e));
    return result;
  }

  public <T> T getRandomDocumentWithTheField(String field, boolean exist) {
    BasicDBObject query = new BasicDBObject(field, new BasicDBObject("$exists", exist));

    return (T) genericCollection.find(query).limit(-1).skip(5).first();
  }

  public <T> T getRandomDocumentWithTheFieldInSeries(String field, boolean exist) {
    List<Bson> pipeline =
        Arrays.asList(
            Aggregates.match(Filters.exists(field, exist)),
            Aggregates.lookup("series", field, "_id", "matchedSeries"),
            Aggregates.match(
                Filters.expr(
                    new Document(
                        "$gt", Arrays.asList(new Document("$size", "$matchedSeries"), 0)))),
            Aggregates.skip(5),
            Aggregates.limit(1));
    return (T) genericCollection.aggregate(pipeline).first();
  }

  public <T> T getRandomDocumentWithTheField(String field, boolean exist, int skip) {
    BasicDBObject query = new BasicDBObject(field, new BasicDBObject("$exists", exist));

    return (T) genericCollection.find(query).skip(skip).first();
  }

  public <T> T getRandomDocumentWithTheField(
      String field, boolean exist, String secondField, boolean secondExist) {
    BasicDBObject query =
        new BasicDBObject(field, new BasicDBObject("$exists", exist))
            .append(secondField, new BasicDBObject("$exists", secondExist));

    return (T) genericCollection.find(query).limit(-1).skip(5).first();
  }

  public <T> T getDocumentsWithTheField(String field, boolean exist) {
    BasicDBObject query = new BasicDBObject();
    query.put(field, new BasicDBObject("$exists", exist));
    query.put("isArchived", new BasicDBObject("$ne", true));

    return (T) genericCollection.find(query).iterator();
  }

  public <T> T getRandomDocumentWithTheFieldSize(String field, int size) {
    BasicDBObject query = new BasicDBObject(field, new BasicDBObject("$size", size));
    return (T) genericCollection.find(query).first();
  }

  public <T> T getRandomDocumentsWithTheFieldSize(String field, int size, int limit) {
    BasicDBObject query = new BasicDBObject(field, new BasicDBObject("$size", size));
    return (T) genericCollection.find(query).limit(limit).iterator();
  }

  public Iterator getRandomDocumentWithTheFieldGreaterSize(String field) {
    BasicDBObject sizeCondition =
        new BasicDBObject(
            "$gt",
            Arrays.asList(
                new BasicDBObject(
                    "$size",
                    new BasicDBObject("$ifNull", Arrays.asList("$" + field, Arrays.asList()))),
                0));

    BasicDBObject query = new BasicDBObject("$expr", sizeCondition);
    LogUtils.logInfoMessage("Query: " + query);

    return genericCollection.find(query).iterator();
  }

  public <T> T getSortedDocumentByField(String sortField, int sortValue) {
    BasicDBObject query = new BasicDBObject(sortField, sortValue);
    return (T) genericCollection.find().sort(query).limit(1).first();
  }

  public String getDistinctRandomValue(String field) {
    List<String> listObjects =
        CommonUtils.iteratorToListOf(genericCollection.distinct(field, String.class).iterator());
    int index = new Random().nextInt(listObjects.size() - 1);
    return listObjects.get(index);
  }

  protected void setContentFeedsDates() {
    updatedAtMax = Objects.requireNonNull(getUpdatedAtMin(getDocumentsContentFeeds()));
    updatedAtMin = updatedAtMax.minusHours(1);

    updatedAtMaxString = updatedAtMax + "Z";
    updatedAtMinString = updatedAtMin + "Z";
  }

  protected void setChannelsContentFeedsDates() {
    updatedAtMax = Objects.requireNonNull(getUpdatedAtMin(getChannelsDocumentsContentFeeds()));
    updatedAtMin = updatedAtMax.minusHours(1);

    updatedAtMaxString = updatedAtMax + "Z";
    updatedAtMinString = updatedAtMin + "Z";
  }

  public Map<String, String> getIdsFromContentSet(List<ObjectIdContentIdSet> contentIdSets) {
    Map<String, String> result = new HashMap<>();
    contentIdSets.forEach(content -> result.put(content.getId(), content.getContentUUID()));
    return result;
  }

  public <T> T getDocumentsByKeyValue(String key, String value) {
    BasicDBObject query = new BasicDBObject(key, value);
    Iterator iterator = genericCollection.find(query).iterator();
    return (T) CommonUtils.iteratorToList(iterator);
  }

  public <T> T getDocumentsByKeyValueId(String key, String value) {
    BasicDBObject query = new BasicDBObject(key, new ObjectId(value));
    Iterator iterator = genericCollection.find(query).iterator();
    return (T) CommonUtils.iteratorToList(iterator);
  }

  public <T> List<T> getSortedDocumentsByKeyValue(
      String key, String id, String activeRegion, String sortKey, int sortValue) {
    BasicDBObject query =
        new BasicDBObject(key, new ObjectId(id)).append("activeRegion", activeRegion);
    List<T> result = new ArrayList<>();
    Collation collation =
        Collation.builder()
            .locale("en_US")
            .collationStrength(CollationStrength.PRIMARY)
            .numericOrdering(true)
            .build();
    genericCollection
        .find(query)
        .sort(new BasicDBObject(sortKey, sortValue))
        .collation(collation)
        .forEach(e -> result.add((T) e));
    return result;
  }

  public <T> T getDocumentByKeyValue(String key, String value) {
    BasicDBObject query = new BasicDBObject(key, value);
    return (T) genericCollection.find(query).first();
  }

  public <T> T getDocumentsByKeyValue(String key, boolean value, int limit) {
    BasicDBObject query = new BasicDBObject(key, value);
    Iterator iterator = genericCollection.find(query).limit(limit).iterator();
    return (T) CommonUtils.iteratorToList(iterator);
  }

  public <T> T getDocumentsByKeyValue(String key, boolean value) {
    BasicDBObject query = new BasicDBObject(key, value);
    Iterator iterator = genericCollection.find(query).iterator();
    return (T) CommonUtils.iteratorToList(iterator);
  }

  public <T> T getDocumentByKeyValue(String key, boolean value) {
    BasicDBObject query = new BasicDBObject(key, value);
    return (T) genericCollection.find(query).first();
  }

  public <T> T getDocumentsByKeyObjectIdValue(String key, String id) {
    BasicDBObject query = new BasicDBObject(key, new ObjectId(id));
    Iterator iterator = genericCollection.find(query).iterator();
    return (T) CommonUtils.iteratorToList(iterator);
  }

  public <T> T getDocumentByKeyValueObject(String key, BasicDBObject value) {
    BasicDBObject query = new BasicDBObject(key, value);
    return (T) genericCollection.find(query).first();
  }

  public <T> T getDocumentByKeyValueList(String key, BasicDBList value) {
    BasicDBObject query = new BasicDBObject(key, value);
    return (T) genericCollection.find(query).first();
  }

  public <T> T getExistsAndSizeNotZero(String key) {
    BasicDBObject query = new BasicDBObject(key, existsNotZeroBasicObject());
    return (T) genericCollection.find(query).first();
  }

  public <T> T getExistsAndNotEmpty(String key) {
    BasicDBObject query =
        new BasicDBObject(key, new BasicDBObject("$exists", true).append("$ne", ""));
    return (T) genericCollection.find(query).first();
  }

  public <T> T getExistsAndSizeNotZeroIterator(String key, int skip) {
    BasicDBObject query = new BasicDBObject(key, existsNotZeroBasicObject());

    return (T) genericCollection.find(query).skip(skip).iterator();
  }

  public <T> T getExistsAndSizeNotZero(String key, int skip) {
    BasicDBObject query = new BasicDBObject(key, existsNotZeroBasicObject());
    return (T) genericCollection.find(query).skip(skip).first();
  }

  public <T> List<T> getDocumentsInList(String key, List<String> listIds) {
    List<T> result = new ArrayList<>();
    BasicDBObject query =
        new BasicDBObject(key, new BasicDBObject("$in", transformStringsToIds(listIds)));
    genericCollection.find(query).forEach(e -> result.add((T) e));
    return result;
  }

  public static String[] listDirectories(String path) {
    List<String> directoryPaths = new ArrayList<>();

    try {
      Files.walkFileTree(
          Paths.get(path),
          EnumSet.of(FileVisitOption.FOLLOW_LINKS),
          Integer.MAX_VALUE,
          new SimpleFileVisitor<Path>() {
            @Override
            public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs)
                throws IOException {
              directoryPaths.add(
                  dir.toString()); // Add the path of the current directory to the list
              return FileVisitResult.CONTINUE;
            }
          });
    } catch (IOException e) {
      LogUtils.logSevereMessageThenFail(e.getLocalizedMessage());
    }

    return directoryPaths.toArray(new String[0]);
  }

  public long getKeyValueCount(String field, boolean value) {
    BasicDBObject query = new BasicDBObject(field, value);
    return genericCollection.countDocuments(query);
  }

  public static BasicDBObject existsNotZeroBasicObject() {
    return new BasicDBObject("$exists", true).append("$ne", new ArrayList<>());
  }

  public <T> T aggregatePublishedChannelsEpisodesTimelines(boolean isPublished) {
    List<Bson> query =
        Arrays.asList(
            // Step 1: Match published channels
            Aggregates.match(Filters.eq("published", isPublished)),
            // Step 2: Lookup episodes related to these channels
            Aggregates.lookup("episodes", "_id", "channels", "episodeDetails"),
            // Step 3: Unwind episodes to work with individual documents
            Aggregates.unwind("$episodeDetails"),
            // Step 4: Match episodes with non-empty sources array
            Aggregates.match(
                Filters.and(
                    Filters.exists("episodeDetails.sources", true),
                    Filters.not(Filters.size("episodeDetails.sources", 0)))),
            // Step 5: Lookup timelines for these episodes
            Aggregates.lookup("timelines", "episodeDetails._id", "episode", "timelineDetails"),
            // Step 6: Unwind timelines to work with individual documents
            Aggregates.unwind("$timelineDetails"),
            // Step 7: Project the required fields
            Aggregates.project(
                Projections.fields(
                    Projections.computed("channel", "$_id"),
                    Projections.include("episodeDetails._id", "timelineDetails"))));

    return (T) genericCollection.aggregate(query).first();
  }

  public <T> T getEpisodesWithChannels() {
    List<Bson> query =
        Arrays.asList(
            Aggregates.lookup("channels", "channels", "_id", "matchedChannels"),
            Aggregates.match(Filters.ne("matchedChannels", Collections.emptyList())));
    return (T) genericCollection.aggregate(query).iterator();
  }

  public void deleteDocumentById(String id) {
    long deletedIds =
        genericCollection.deleteOne(new BasicDBObject("_id", new ObjectId(id))).getDeletedCount();
    LogUtils.logInfoMessage("Removed document with id: " + id);
    Assert.assertEquals(deletedIds, 1);
  }

  public void deleteDocuments(List<String> listIds) {
    LogUtils.logInfoMessage("List ids to delete: " + listIds);
    List<ObjectId> objectIds = listIds.stream().map(ObjectId::new).collect(Collectors.toList());
    long deletedIds =
        genericCollection
            .deleteMany(new BasicDBObject("_id", new BasicDBObject("$in", objectIds)))
            .getDeletedCount();
    Assert.assertEquals(listIds.size(), deletedIds);
  }

  public void deleteDocument(String id) {
    LogUtils.logInfoMessage("Delete id: " + id);
    long deletedIds =
        genericCollection.deleteOne(new BasicDBObject("_id", new ObjectId(id))).getDeletedCount();
    Assert.assertEquals(deletedIds, 1);
  }

  public static void cleanUpDbTestData() {
    // Episodes
    BasicDBObject query =
        new BasicDBObject()
            .append("sources.adPods.startAt", new BasicDBObject().append("$type", "double"));
    BasicDBObject queryAllotment =
        new BasicDBObject().append("allotment", new BasicDBObject().append("$type", "double"));
    BasicDBObject queryDuration =
        new BasicDBObject().append("duration", new BasicDBObject().append("$type", "double"));
    getResult("episodes", List.of(query, queryAllotment, queryDuration));

    // VodCategoryEntries
    BasicDBObject queryVodCategoryEntriesType =
        new BasicDBObject().append("createdAt", new BasicDBObject().append("$type", "string"));
    getResult("vodcategoryentries", List.of(queryVodCategoryEntriesType));

    // Series
    BasicDBObject querySeriesType =
        new BasicDBObject().append("createdAt", new BasicDBObject().append("$type", "string"));
    getResult("series", List.of(querySeriesType));

    // Channels
    BasicDBObject queryChannelsRegionType =
        new BasicDBObject().append("regionFilter", new BasicDBObject().append("$type", "array"));
    getResult("channels", List.of(queryChannelsRegionType));

    // VodCategories
    BasicDBObject queryVodCategoriesRegionType =
        new BasicDBObject().append("regionFilter", new BasicDBObject().append("$type", "array"));
    getResult("vodcategories", List.of(queryVodCategoriesRegionType));

    BasicDBObject queryVodCategoriesCategoriesType =
        new BasicDBObject()
            .append(
                "categories",
                new BasicDBObject().append("$not", new BasicDBObject().append("$type", "array")));
    getResult("vodcategories", List.of(queryVodCategoriesCategoriesType));

    // Clips
    BasicDBObject queryClipsUpdatedAtType =
        new BasicDBObject().append("updatedAt", new BasicDBObject().append("$type", "string"));
    getResult("clips", List.of(queryClipsUpdatedAtType));

    // Timelines
    BasicDBObject queryTimelinesUpdatedChannel =
        new BasicDBObject().append("channel", new BasicDBObject().append("$type", "string"));
    getResult("timelines", List.of(queryTimelinesUpdatedChannel));
  }

  private static void getResult(String collection, List<BasicDBObject> listQuery) {
    MongoCollection<Document> dbCollection = MongoDbManager.getDatabase().getCollection(collection);

    for (BasicDBObject query : listQuery) {
      LogUtils.logInfoMessage("Ran query: " + query + " for " + collection + " collection.");
      DeleteResult deleteResult = dbCollection.deleteMany(query);
      if (deleteResult.getDeletedCount() > 0)
        LogUtils.logInfoMessage(
            "Deleted files for collection : " + collection + ", num: " + deleteResult);
    }
  }

  public <T> List<T> getDocumentsByKeyIds(String field, List<String> ids) {
    List<T> result = new ArrayList<>();
    BasicDBObject query = new BasicDBObject();
    query.put(field, new BasicDBObject("$in", transformStringsToIds(ids)));
    genericCollection.find(query).forEach(e -> result.add((T) e));
    return result;
  }

  public void updateDocument(String key, Object value, Document update) {
    Document query = new Document(key, value);
    genericCollection.updateOne(query, new Document("$set", update));
    LogUtils.logInfoMessage("Document updated successfully");
  }

  public InsertManyResult insertDocuments(List<Document> listDocuments) {
    InsertManyResult insertManyResult =
        ((MongoCollection<Document>) genericCollection).insertMany(listDocuments);
    LogUtils.logInfoMessage("Documents inserted successfully");
    return insertManyResult;
  }

  public String insertDocument(Document document) {
    InsertOneResult insertOneResult =
        ((MongoCollection<Document>) genericCollection).insertOne(document);
    String insertedId = insertOneResult.getInsertedId().asObjectId().getValue().toString();
    LogUtils.logInfoMessage("Document inserted successfully: " + insertedId);
    return insertedId;
  }

  public Document getChannelDoc(List<Document> listCategories, int number, String activeRegion) {
    return new Document("categories", listCategories)
        .append("number", number)
        .append("activeRegion", activeRegion);
  }

  public <T> T getChannelWithMainCategory(String field, int skip, boolean isArray) {
    List<Bson> pipeline =
        Arrays.asList(
            Aggregates.lookup("maincategories", "categories.catId", "_id", "matchedCategories"),
            Aggregates.match(
                Filters.and(
                    Filters.exists("categories", true),
                    Filters.ne("categories", 0),
                    isArray
                        ? Filters.and(
                            Filters.exists(field, true), Filters.not(Filters.size(field, 0)))
                        : Filters.exists(field, true),
                    Filters.expr(
                        new BasicDBObject(
                            "$gt",
                            Arrays.asList(new BasicDBObject("$size", "$matchedCategories"), 0))))),
            Aggregates.skip(skip));

    return (T) genericCollection.aggregate(pipeline).first();
  }

  public <T> T getChannelsEpisodesQueueInEpisodes() {
    Document lookup =
        new Document(
            "$lookup",
            new Document("from", "episodes")
                .append("localField", "episodesQueue.episode")
                .append("foreignField", "_id")
                .append("as", "matchedEpisodes"));

    Document match =
        new Document(
            "$match", new Document("matchedEpisodes", new Document("$ne", Arrays.asList())));

    return (T) genericCollection.aggregate(Arrays.asList(lookup, match)).first();
  }
}