/**
 * This file is part of the Meeds project (https://meeds.io/).
 *
 * Copyright (C) 2020 - 2025 Meeds Association contact@meeds.io
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 3 of the License, or (at your option) any later version.
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 */
package org.exoplatform.social.core.jpa.storage;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.stream.Collectors;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;

import org.exoplatform.commons.api.persistence.ExoTransactional;
import org.exoplatform.commons.persistence.impl.EntityManagerHolder;
import org.exoplatform.commons.utils.CommonsUtils;
import org.exoplatform.commons.utils.PropertyManager;
import org.exoplatform.container.ExoContainerContext;
import org.exoplatform.services.log.ExoLogger;
import org.exoplatform.services.log.Log;
import org.exoplatform.services.security.ConversationState;
import org.exoplatform.social.core.ActivityProcessor;
import org.exoplatform.social.core.BaseActivityProcessorPlugin;
import org.exoplatform.social.core.activity.ActivityFilter;
import org.exoplatform.social.core.activity.model.ActivityFile;
import org.exoplatform.social.core.activity.model.ActivityShareAction;
import org.exoplatform.social.core.activity.model.ActivityStream;
import org.exoplatform.social.core.activity.model.ActivityStreamImpl;
import org.exoplatform.social.core.activity.model.ExoSocialActivity;
import org.exoplatform.social.core.activity.model.ExoSocialActivityImpl;
import org.exoplatform.social.core.identity.model.Identity;
import org.exoplatform.social.core.identity.provider.OrganizationIdentityProvider;
import org.exoplatform.social.core.identity.provider.SpaceIdentityProvider;
import org.exoplatform.social.core.jpa.storage.dao.ActivityDAO;
import org.exoplatform.social.core.jpa.storage.dao.ActivityShareActionDAO;
import org.exoplatform.social.core.jpa.storage.dao.ConnectionDAO;
import org.exoplatform.social.core.jpa.storage.entity.ActivityEntity;
import org.exoplatform.social.core.jpa.storage.entity.ActivityShareActionEntity;
import org.exoplatform.social.core.jpa.storage.entity.StreamItemEntity;
import org.exoplatform.social.core.jpa.storage.entity.StreamType;
import org.exoplatform.social.core.manager.IdentityManager;
import org.exoplatform.social.core.space.SpaceFilter;
import org.exoplatform.social.core.space.model.Space;
import org.exoplatform.social.core.storage.ActivityFileStoragePlugin;
import org.exoplatform.social.core.storage.ActivityStorageException;
import org.exoplatform.social.core.storage.ActivityStorageException.Type;
import org.exoplatform.social.core.storage.api.ActivityStorage;
import org.exoplatform.social.core.storage.api.IdentityStorage;
import org.exoplatform.social.core.storage.impl.StorageUtils;
import org.exoplatform.social.core.utils.MentionUtils;
import org.exoplatform.social.metadata.favorite.FavoriteService;
import org.exoplatform.social.metadata.model.MetadataItem;
import org.exoplatform.social.notification.service.SpaceWebNotificationService;

import io.meeds.social.space.constant.SpaceMembershipStatus;

import jakarta.persistence.LockModeType;

public class RDBMSActivityStorageImpl implements ActivityStorage {

  private static final Log                       LOG                         =
                                                     ExoLogger.getLogger(RDBMSActivityStorageImpl.class);

  public static final String                     COMMENT_PREFIX              = "comment";

  private final ActivityShareActionDAO           activityShareActionDAO;

  private final ActivityDAO                      activityDAO;

  private final ConnectionDAO                    connectionDAO;

  private IdentityStorage                        identityStorage;

  private final SpaceStorage                     spaceStorage;

  private final SpaceWebNotificationService      spaceWebNotificationService;

  private final SortedSet<ActivityProcessor>     activityProcessors;

  private ActivityStorage                        activityStorage;

  /**
   * Most prior storage plugin that will handle attachments storage
   */
  private ActivityFileStoragePlugin              activityFileStorage;

  /**
   * Most prior storage plugin that will handle attachments storage
   */
  private Map<String, ActivityFileStoragePlugin> activityFileStorageByDSName = new HashMap<>();

  public RDBMSActivityStorageImpl(IdentityStorage identityStorage,
                                  SpaceStorage spaceStorage,
                                  ActivityShareActionDAO activityShareActionDAO,
                                  ActivityDAO activityDAO,
                                  SpaceWebNotificationService spaceWebNotificationService,
                                  ConnectionDAO connectionDAO) {
    this.identityStorage = identityStorage;
    this.activityProcessors = new TreeSet<>(processorComparator());
    this.activityDAO = activityDAO;
    this.activityShareActionDAO = activityShareActionDAO;
    this.connectionDAO = connectionDAO;
    this.spaceStorage = spaceStorage;
    this.spaceWebNotificationService = spaceWebNotificationService;
  }

  private static Comparator<ActivityProcessor> processorComparator() {
    return new Comparator<ActivityProcessor>() {
      public int compare(ActivityProcessor p1, ActivityProcessor p2) {
        if (p1 == null || p2 == null) {
          throw new IllegalArgumentException("Cannot compare null ActivityProcessor");
        }
        int diff = p1.getPriority() - p2.getPriority();
        return diff == 0 ? p1.getName().compareTo(p2.getName()) : diff;
      }
    };
  }

  private ExoSocialActivity fillActivityFromEntity(ActivityEntity activityEntity, ExoSocialActivity activity) {
    if (activity == null) {
      activity = new ExoSocialActivityImpl(activityEntity.getPosterId(),
                                           activityEntity.getType(),
                                           activityEntity.getTitle(),
                                           activityEntity.getBody(),
                                           false);
    } else {
      activity.setPosterId(activityEntity.getPosterId());
      activity.setType(activityEntity.getType());
      activity.setTitle(activity.getTitle());
      activity.setBody(activity.getBody());
      activity.isComment(false);
    }

    activity.setId(String.valueOf(activityEntity.getId()));
    activity.setLikeIdentityIds(activityEntity.getLikerIds().toArray(new String[] {}));

    List<ActivityShareActionEntity> shareActions = activityShareActionDAO.getShareActionsByActivityId(activityEntity.getId());
    if (CollectionUtils.isNotEmpty(shareActions)) {
      activity.setShareActions(shareActions.stream()
                                           .map(this::fromEntity)
                                           .collect(Collectors.toSet()));
    }
    activity.setTemplateParams(activityEntity.getTemplateParams()
        != null ? new LinkedHashMap<String, String>(activityEntity.getTemplateParams()) : new HashMap<String, String>());

    String ownerIdentityId = activityEntity.getOwnerId();
    ActivityStream stream = new ActivityStreamImpl();
    Identity owner = identityStorage.findIdentityById(ownerIdentityId);
    if (owner != null) {
      stream.setType(owner.getProviderId());
      stream.setPrettyId(owner.getRemoteId());
      stream.setId(owner.getId());
      activity.setStreamOwner(owner.getRemoteId());
      if (stream.isSpace()) {
        Space space = spaceStorage.getSpaceByPrettyName(owner.getRemoteId());
        if (space != null) {
          activity.setSpaceId(space.getId());
        }
      }
    } else {
      LOG.warn("Cannot find stream of activity " + activityEntity.getId() + " since identity " + ownerIdentityId +
          " does not exist");
    }
    //
    activity.setActivityStream(stream);
    activity.setPosterId(activityEntity.getPosterId());
    //
    activity.isLocked(activityEntity.getLocked());
    activity.isHidden(activityEntity.getHidden());
    activity.setTitleId(activityEntity.getTitleId());
    activity.setPostedTime(activityEntity.getPosted() != null ? activityEntity.getPosted().getTime() : 0);
    activity.setUpdated(activityEntity.getUpdatedDate().getTime());
    //
    activity.setPinned(activityEntity.getPinned());
    activity.setPinDate(StorageUtils.toRFC3339Date(activityEntity.getPinDate()));
    activity.setPinAuthorId(activityEntity.getPinAuthorId());
    activity.setCategoryIds(activityEntity.getCategoryIds());
    //
    List<String> commentPosterIds = new ArrayList<>();
    List<String> replyToIds = new ArrayList<>();
    fillCommentsIdsAndPosters(activityEntity, commentPosterIds, replyToIds, false);
    activity.setCommentedIds(commentPosterIds.toArray(new String[commentPosterIds.size()]));
    activity.setReplyToId(replyToIds.toArray(new String[replyToIds.size()]));
    activity.setMentionedIds(activityEntity.getMentionerIds().toArray(new String[activityEntity.getMentionerIds().size()]));

    return activity;
  }

  private void fillCommentsIdsAndPosters(ActivityEntity activity,
                                         List<String> commentPosterIds,
                                         List<String> replyToIds,
                                         boolean isSubComment) {

    List<ActivityEntity> comments = activityDAO.findCommentsAndSubCommentsOfActivity(activity.getId());
    List<Long> commentIds = new ArrayList<>();
    for (ActivityEntity comment : comments) {
      if (!commentPosterIds.contains(comment.getPosterId())) {
        commentPosterIds.add(comment.getPosterId());
      }
      replyToIds.add(getExoCommentID(comment.getId()));
      commentIds.add(comment.getId());
    }
  }

  private ExoSocialActivity convertActivityEntityToActivity(ActivityEntity activityEntity) {
    if (activityEntity == null)
      return null;
    //
    ExoSocialActivity activity = fillActivityFromEntity(activityEntity, null);
    //

    //
    processActivity(activity);

    return activity;
  }

  private ActivityEntity convertActivityToActivityEntity(ExoSocialActivity activity, String ownerId) {
    preSaveProcessActivity(activity);

    ActivityEntity activityEntity = new ActivityEntity();
    if (activity.getId() != null) {
      activityEntity = activityDAO.find(Long.valueOf(activity.getId()));
    }
    activityEntity.setTitle(activity.getTitle());
    activityEntity.setTitleId(activity.getTitleId());
    activityEntity.setType(activity.getType());
    activityEntity.setBody(activity.getBody());
    if (ownerId != null) {
      activityEntity.setPosterId(activity.getUserId() != null ? activity.getUserId() : ownerId);
    }
    if (activity.getLikeIdentityIds() != null) {
      activityEntity.setLikerIds(new HashSet<String>(Arrays.asList(activity.getLikeIdentityIds())));
    }
    Map<String, String> params = activity.getTemplateParams();
    if (params != null) {
      activityEntity.setTemplateParams(params);
    }

    processDates(activityEntity);
    activityEntity.setLocked(activity.isLocked());
    activityEntity.setHidden(activity.isHidden());
    activityEntity.setPinned(activity.isPinned());
    activityEntity.setPinDate(StorageUtils.parseRFC3339Date(activity.getPinDate()));
    activityEntity.setPinAuthorId(activity.getPinAuthorId());
    activityEntity.setCategoryIds(activity.getCategoryIds());
    activityEntity.setMentionerIds(new HashSet<>(Arrays.asList(processMentionsByIdentityId(ownerId,
                                                                                           activity.getTitle(),
                                                                                           activity.getTemplateParams()))));
    //
    return activityEntity;
  }

  private ExoSocialActivity convertCommentEntityToComment(ActivityEntity comment) {
    ExoSocialActivity exoComment = new ExoSocialActivityImpl(comment.getPosterId(),
                                                             null,
                                                             comment.getTitle(),
                                                             comment.getBody(),
                                                             false);
    exoComment.setId(getExoCommentID(comment.getId()));
    exoComment.setTitle(comment.getTitle());
    exoComment.setType(comment.getType());
    exoComment.setTitleId(comment.getTitleId());
    exoComment.setBody(comment.getBody());
    exoComment.setTemplateParams(comment.getTemplateParams()
        != null ? new LinkedHashMap<String, String>(comment.getTemplateParams()) : new HashMap<String, String>());
    exoComment.setPosterId(comment.getPosterId());
    exoComment.isComment(true);
    //
    exoComment.isLocked(comment.getLocked() != null && comment.getLocked());
    exoComment.isHidden(comment.getHidden() != null && comment.getHidden());
    exoComment.setUpdated(comment.getUpdatedDate() != null ? comment.getUpdatedDate().getTime() : null);
    //
    ActivityEntity parentActivity = getTopParentActivity(comment);
    exoComment.setParentId(parentActivity == null ? null : String.valueOf(parentActivity.getId()));
    exoComment.setParentCommentId(isSubComment(comment) ? getExoCommentID(comment.getParent().getId()) : null);
    //
    exoComment.setPostedTime(comment.getPosted() != null ? comment.getPosted().getTime() : 0);
    exoComment.setUpdated(comment.getUpdatedDate() != null ? comment.getUpdatedDate().getTime() : null);
    //
    Set<String> mentioned = comment.getMentionerIds();
    if (mentioned != null && !mentioned.isEmpty()) {
      exoComment.setMentionedIds(comment.getMentionerIds().toArray(new String[mentioned.size()]));
    }

    //
    Set<String> likers = comment.getLikerIds();
    if (likers != null && !likers.isEmpty()) {
      exoComment.setLikeIdentityIds(comment.getLikerIds().toArray(new String[likers.size()]));
    }
    processActivity(exoComment);

    return exoComment;
  }

  private ActivityEntity getTopParentActivity(ActivityEntity activity) {
    ActivityEntity parent = activity.getParent();
    if (parent == null || (parent.getId() == activity.getId())) {
      return activity;
    } else {
      return getTopParentActivity(parent);
    }
  }

  private boolean isSubComment(ActivityEntity comment) {
    return comment.getParent() != null && comment.getParent().isComment();
  }

  private List<ExoSocialActivity> convertCommentEntitiesToComments(List<ActivityEntity> comments) {
    return convertCommentEntitiesToComments(comments, false, false);
  }

  private List<ExoSocialActivity> convertCommentEntitiesToComments(List<ActivityEntity> comments, boolean loadSubComments) {
    return convertCommentEntitiesToComments(comments, false, false);
  }

  private List<ExoSocialActivity> convertCommentEntitiesToComments(List<ActivityEntity> comments,
                                                                   boolean loadSubComments,
                                                                   boolean sortDescending) {
    if (comments == null || comments.isEmpty())
      return Collections.emptyList();
    if (loadSubComments) {
      // Add subComments to the list
      List<Long> ids = new ArrayList<>();
      for (ActivityEntity activityEntity : comments) {
        ids.add(activityEntity.getId());
      }
      List<ActivityEntity> subComments = activityDAO.findCommentsOfActivities(ids);

      if (subComments != null && !subComments.isEmpty()) {
        comments.addAll(subComments);
      }

      // Sort sub comments just after the comment
      Collections.sort(comments, new CommentComparator(sortDescending));
    }

    return comments.stream().map(comment -> convertCommentEntityToComment(comment)).collect(Collectors.toList());
  }

  private ActivityEntity convertCommentToCommentEntity(ActivityEntity activityEntity, ExoSocialActivity comment) {
    if (comment.getTemplateParams() != null) {
      preSaveProcessActivity(comment);
    }

    ActivityEntity commentEntity = new ActivityEntity();
    if (comment.getId() != null) {
      commentEntity = activityDAO.find(getCommentID(comment.getId()));
    }
    if (comment.getParentCommentId() != null) {
      ActivityEntity parentCommentEntity = activityDAO.find(getCommentID(comment.getParentCommentId()));
      parentCommentEntity.addComment(commentEntity);
    } else {
      activityEntity.addComment(commentEntity);
    }
    commentEntity.setComment(true);
    commentEntity.setTitle(comment.getTitle());
    commentEntity.setTitleId(comment.getTitleId());
    commentEntity.setType(comment.getType());
    commentEntity.setBody(comment.getBody());
    commentEntity.setPosterId(comment.getPosterId() != null ? comment.getPosterId() : comment.getUserId());
    if (comment.getTemplateParams() != null) {
      commentEntity.setTemplateParams(comment.getTemplateParams());
    }
    //
    commentEntity.setLocked(comment.isLocked());
    commentEntity.setHidden(comment.isHidden());
    //
    processDates(commentEntity);
    commentEntity.setMentionerIds(new HashSet<>(Arrays.asList(processMentionsByIdentityId(activityEntity.getOwnerId(),
                                                                                          comment.getTitle(),
                                                                                          comment.getTemplateParams()))));
    //
    return commentEntity;
  }

  /**
   * Help to update mentionIds of activity Help to update corresponding
   * StreamItem This method must be called before updating title and template
   * param of activity entity activity's mentionIds include : - mentioned ids
   * processed from its title and template param - mentioned ids from its
   * comments
   *
   * @param activityEntity
   * @param activity
   */
  private void updateActivityMentions(ActivityEntity activityEntity, ExoSocialActivity activity) {
    Set<String> commentMentions = new HashSet<>();
    if (activityEntity.getComments() != null) {
      activityEntity.getComments().forEach(comment -> {
        String[] mentions = processMentionsByIdentityId(activityEntity.getOwnerId(),
                                                        comment.getTitle(),
                                                        comment.getTemplateParams());
        commentMentions.addAll(Arrays.asList(mentions));
      });
    }
    Set<String> mentionsToRemove = new HashSet<>(Arrays.asList(processMentionsByIdentityId(activityEntity.getOwnerId(),
                                                                                           activityEntity.getTitle(),
                                                                                           activityEntity.getTemplateParams())));
    Set<String> mentionToAdd = new HashSet<>(Arrays.asList(processMentionsByIdentityId(activityEntity.getOwnerId(),
                                                                                       activity.getTitle(),
                                                                                       activity.getTemplateParams())));

    mentionsToRemove.forEach(mentionedId -> {
      if (!commentMentions.contains(mentionedId) && !mentionToAdd.contains(mentionedId)) {
        StreamItemEntity item = new StreamItemEntity(StreamType.MENTIONER);
        item.setOwnerId(Long.parseLong(mentionedId));
        activityEntity.removeStreamItem(item);
      }
    });

    mentionToAdd.forEach(mentionedId -> {
      if (!commentMentions.contains(mentionedId) && !mentionsToRemove.contains(mentionedId)) {
        mention(activityEntity, new String[] { mentionedId });
      }
    });

    mentionToAdd.addAll(commentMentions);
    activityEntity.setMentionerIds(mentionToAdd);
  }

  private List<ExoSocialActivity> convertActivityIdsToActivities(List<Long> activityIds) {
    if (activityIds == null)
      return Collections.emptyList();
    // Use getActivityStorage to benifit from Cached Storage
    return activityIds.stream()
                      .map(activityId -> getActivityStorage().getActivity(String.valueOf(activityId)))
                      .toList();
  }

  private List<ExoSocialActivity> convertActivityEntitiesToActivities(List<ActivityEntity> activities) {
    if (CollectionUtils.isNotEmpty(activities)) {
      return activities.stream().map(this::convertActivityEntityToActivity).toList();
    } else {
      return Collections.emptyList();
    }
  }

  @Override
  @ExoTransactional
  public ExoSocialActivity getActivity(String activityId) {
    if (activityId == null || activityId.isEmpty()) {
      return null;
    }
    if (activityId.startsWith(COMMENT_PREFIX)) {
      return getComment(activityId);
    }
    try {
      ActivityEntity entity = activityDAO.find(Long.valueOf(activityId));
      if (entity != null && entity.isComment()) {
        return convertCommentEntityToComment(entity);
      } else {
        return convertActivityEntityToActivity(entity);
      }
    } catch (Exception e) {
      if (PropertyManager.isDevelopping()) {
        throw new ActivityStorageException(Type.FAILED_TO_GET_ACTIVITY, e.getMessage(), e);
      }
      return null;
    }
  }

  @Override
  public boolean isActivityExists(String activityId) {
    if (activityId == null || activityId.isEmpty()) {
      return false;
    }
    Long id = null;
    if (activityId.startsWith(COMMENT_PREFIX)) {
      id = getCommentID(activityId);
    } else {
      id = Long.valueOf(activityId);
    }
    return activityDAO.find(id) != null;
  }

  public ExoSocialActivity getComment(String commentId) {
    try {
      ActivityEntity entity = activityDAO.find(getCommentID(commentId));
      if (entity != null && entity.isComment()) {
        return convertCommentEntityToComment(entity);
      } else {
        return null;
      }
    } catch (Exception e) {
      if (PropertyManager.isDevelopping()) {
        throw new ActivityStorageException(Type.FAILED_TO_GET_ACTIVITY, e.getMessage(), e);
      }
      return null;
    }
  }

  @Override
  public List<ExoSocialActivity> getUserActivities(Identity owner) {
    return getUserActivities(owner, 0, -1);
  }

  @Override
  public List<ExoSocialActivity> getUserActivities(Identity owner, long offset, long limit) {
    return getUserActivitiesForUpgrade(owner, offset, limit);
  }

  @Override
  public List<String> getUserSpacesActivityIds(Identity ownerIdentity, int offset, int limit) {
    return activityDAO.getUserSpacesActivityIds(ownerIdentity, offset, limit, memberOfSpaceIds(ownerIdentity));
  }

  @Override
  public List<String> getUserIdsActivities(Identity owner, long offset, long limit) {
    return activityDAO.getUserIdsActivities(owner, offset, limit);
  }

  @Override
  public List<ExoSocialActivity> getActivitiesByFilter(Identity viewerIdentity,
                                                       ActivityFilter activityFilter,
                                                       long offset,
                                                       long limit) {
    List<String> streamIdentityIds = null;
    switch (activityFilter.getStreamType()) {
    case USER_STREAM:
      activityFilter.setUserId(viewerIdentity.getIdentityId());
      if (activityFilter.getSpaceIdentityId() == 0) {
        streamIdentityIds = spaceStorage.getSpaceIdentityIdsByUserRole(viewerIdentity.getRemoteId(),
                                                                       String.valueOf(SpaceMembershipStatus.MEMBER),
                                                                       0,
                                                                       -1);
      }
      break;
    case USER_FAVORITE_STREAM:
      List<ExoSocialActivity> activities = getFavoriteActivities(viewerIdentity, activityFilter.getSpaceIdentityId());
      return StorageUtils.subList(activities, (int) offset, (int) limit);
    case FAVORITE_SPACES_STREAM:
      streamIdentityIds = spaceStorage.getFavoriteSpaceIdentityIds(viewerIdentity.getRemoteId(), new SpaceFilter(), 0, -1);
      if (CollectionUtils.isEmpty(streamIdentityIds)) {
        return Collections.emptyList();
      }
      break;
    case UNREAD_SPACES_STREAM:
      List<Long> activityIds;
      if (activityFilter.getSpaceIdentityId() == 0) {
        activityIds = spaceWebNotificationService.getUnreadActivityIds(viewerIdentity.getRemoteId(), offset, limit);
      } else {
        try {
          activityIds = spaceWebNotificationService.getUnreadActivityIdsBySpace(viewerIdentity.getRemoteId(),
                                                                                activityFilter.getSpaceIdentityId(),
                                                                                offset,
                                                                                limit);
        } catch (Exception e) {
          throw new IllegalStateException(String.format("Unable to retrieve activities for user %s with filtered space %s",
                                                        viewerIdentity.getRemoteId(),
                                                        activityFilter.getSpaceIdentityId()),
                                          e);
        }
      }
      if (CollectionUtils.isEmpty(activityIds)) {
        return Collections.emptyList();
      } else {
        return convertActivityIdsToActivities(activityIds);
      }
    case MANAGE_SPACES_STREAM:
      streamIdentityIds = spaceStorage.getSpaceIdentityIdsByUserRole(viewerIdentity.getRemoteId(),
                                                                     String.valueOf(SpaceMembershipStatus.MANAGER),
                                                                     0,
                                                                     -1);
      if (CollectionUtils.isEmpty(streamIdentityIds)) {
        return Collections.emptyList();
      }
      break;
    case ANY_SPACE_ACTIVITY:
      if (activityFilter.getSpaceIdentityId() == 0) {
        return Collections.emptyList();
      }
      activityFilter.setShowPinned(true);
      break;
    case ALL_STREAM:
      activityFilter.setPosterId(viewerIdentity.getIdentityId());
      streamIdentityIds = spaceStorage.getSpaceIdentityIdsByUserRole(viewerIdentity.getRemoteId(),
                                                                     String.valueOf(SpaceMembershipStatus.MEMBER),
                                                                     0,
                                                                     -1);
      streamIdentityIds = new ArrayList<>(streamIdentityIds);
      streamIdentityIds.add(viewerIdentity.getId());
      Set<Long> connectionIds =
                              connectionDAO.getConnectionIds(Long.parseLong(viewerIdentity.getId()),
                                                             org.exoplatform.social.core.relationship.model.Relationship.Type.CONFIRMED);
      if (connectionIds != null) {
        streamIdentityIds.addAll(connectionIds.stream().map(String::valueOf).toList());
      }
      break;
    default:
      throw new UnsupportedOperationException();
    }
    return convertActivityIdsToActivities(activityDAO.getActivityByFilter(activityFilter,
                                                                          streamIdentityIds,
                                                                          (int) offset,
                                                                          (int) limit));
  }

  @Override
  public List<String> getActivityIdsByFilter(Identity viewerIdentity, ActivityFilter activityFilter, long offset, long limit) {
    List<String> streamIdentityIds = null;
    switch (activityFilter.getStreamType()) {
    case USER_STREAM:
      activityFilter.setUserId(viewerIdentity.getIdentityId());
      if (activityFilter.getSpaceIdentityId() == 0) {
        streamIdentityIds = spaceStorage.getSpaceIdentityIdsByUserRole(viewerIdentity.getRemoteId(),
                                                                       String.valueOf(SpaceMembershipStatus.MEMBER),
                                                                       0,
                                                                       -1);
      }
      break;
    case USER_FAVORITE_STREAM:
      List<String> activityIds = getFavoriteActivities(viewerIdentity, activityFilter.getSpaceIdentityId()).stream()
                                                                                                   .map(ExoSocialActivity::getId)
                                                                                                   .collect(Collectors.toList());
      return StorageUtils.subList(activityIds, (int) offset, (int) limit);
    case FAVORITE_SPACES_STREAM:
      streamIdentityIds = spaceStorage.getFavoriteSpaceIdentityIds(viewerIdentity.getRemoteId(), new SpaceFilter(), 0, -1);
      if (CollectionUtils.isEmpty(streamIdentityIds)) {
        return Collections.emptyList();
      }
      break;
    case UNREAD_SPACES_STREAM:
      List<Long> unreadActivityIds;
      if (activityFilter.getSpaceIdentityId() == 0) {
        unreadActivityIds = spaceWebNotificationService.getUnreadActivityIds(viewerIdentity.getRemoteId(), offset, limit);
      } else {
        try {
          Identity spaceIdentity = identityStorage.findIdentityById(String.valueOf(activityFilter.getSpaceIdentityId()));
          if (spaceIdentity == null || !spaceIdentity.isEnable() || spaceIdentity.isDeleted() || !spaceIdentity.isSpace()) {
            return Collections.emptyList();
          }
          Space space = spaceStorage.getSpaceByPrettyName(spaceIdentity.getRemoteId());
          unreadActivityIds = spaceWebNotificationService.getUnreadActivityIdsBySpace(viewerIdentity.getRemoteId(),
                                                                                      Long.parseLong(space.getId()),
                                                                                      offset,
                                                                                      limit);
        } catch (Exception e) {
          throw new IllegalStateException(String.format("Unable to retrieve activities for user %s with filtered space %s",
                                                        viewerIdentity.getRemoteId(),
                                                        activityFilter.getSpaceIdentityId()),
                                          e);
        }
      }
      if (CollectionUtils.isEmpty(unreadActivityIds)) {
        return Collections.emptyList();
      } else {
        return unreadActivityIds.stream().map(String::valueOf).toList();
      }
    case MANAGE_SPACES_STREAM:
      streamIdentityIds = spaceStorage.getSpaceIdentityIdsByUserRole(viewerIdentity.getRemoteId(),
                                                                     String.valueOf(SpaceMembershipStatus.MANAGER),
                                                                     0,
                                                                     -1);
      if (CollectionUtils.isEmpty(streamIdentityIds)) {
        return Collections.emptyList();
      }
      break;
    case ANY_SPACE_ACTIVITY:
      if (activityFilter.getSpaceIdentityId() == 0) {
        return Collections.emptyList();
      }
      activityFilter.setShowPinned(true);
      break;
    case ALL_STREAM:
      activityFilter.setPosterId(viewerIdentity.getIdentityId());
      streamIdentityIds = spaceStorage.getSpaceIdentityIdsByUserRole(viewerIdentity.getRemoteId(),
                                                                     String.valueOf(SpaceMembershipStatus.MEMBER),
                                                                     0,
                                                                     -1);
      streamIdentityIds = new ArrayList<>(streamIdentityIds);
      streamIdentityIds.add(viewerIdentity.getId());
      Set<Long> connectionIds =
                              connectionDAO.getConnectionIds(Long.parseLong(viewerIdentity.getId()),
                                                             org.exoplatform.social.core.relationship.model.Relationship.Type.CONFIRMED);
      if (connectionIds != null) {
        streamIdentityIds.addAll(connectionIds.stream().map(String::valueOf).toList());
      }
      break;

    case PIN_STREAM:
      streamIdentityIds = spaceStorage.getSpaceIdentityIdsByUserRole(viewerIdentity.getRemoteId(),
                                                                     String.valueOf(SpaceMembershipStatus.MEMBER),
                                                                     0,
                                                                     -1);
      if (CollectionUtils.isEmpty(streamIdentityIds)) {
        return Collections.emptyList();
      }
      activityFilter.setPinned(true);
      activityFilter.setShowPinned(true);
      break;
    default:
      throw new UnsupportedOperationException();
    }
    return activityDAO.getActivityIdsByFilter(activityFilter, streamIdentityIds, (int) offset, (int) limit);
  }

  @Override
  public int getActivitiesCountByFilter(Identity viewerIdentity, ActivityFilter activityFilter) {
    List<String> streamIdentityIds = null;
    switch (activityFilter.getStreamType()) {
    case USER_STREAM:
      activityFilter.setUserId(viewerIdentity.getIdentityId());
      if (activityFilter.getSpaceIdentityId() == 0) {
        streamIdentityIds = spaceStorage.getSpaceIdentityIdsByUserRole(viewerIdentity.getRemoteId(),
                                                                       String.valueOf(SpaceMembershipStatus.MEMBER),
                                                                       0,
                                                                       -1);
      }
      break;
    case USER_FAVORITE_STREAM:
      return getFavoriteActivities(viewerIdentity, activityFilter.getSpaceIdentityId()).size();
    case FAVORITE_SPACES_STREAM:
      streamIdentityIds = spaceStorage.getFavoriteSpaceIdentityIds(viewerIdentity.getRemoteId(), new SpaceFilter(), 0, -1);
      if (CollectionUtils.isEmpty(streamIdentityIds)) {
        return 0;
      }
      break;
    case UNREAD_SPACES_STREAM:
      if (activityFilter.getSpaceIdentityId() == 0) {
        return (int) spaceWebNotificationService.countUnreadActivities(viewerIdentity.getRemoteId());
      } else {
        try {
          return (int) spaceWebNotificationService.countUnreadActivitiesBySpace(viewerIdentity.getRemoteId(),
                                                                                activityFilter.getSpaceIdentityId());
        } catch (Exception e) {
          throw new IllegalStateException(String.format("Unable to retrieve activities for user %s with filtered space %s",
                                                        viewerIdentity.getRemoteId(),
                                                        activityFilter.getSpaceIdentityId()),
                                          e);
        }
      }
    case MANAGE_SPACES_STREAM:
      streamIdentityIds = spaceStorage.getSpaceIdentityIdsByUserRole(viewerIdentity.getRemoteId(),
                                                                     String.valueOf(SpaceMembershipStatus.MANAGER),
                                                                     0,
                                                                     -1);
      if (CollectionUtils.isEmpty(streamIdentityIds)) {
        return 0;
      }
      break;
    case ALL_STREAM:
      activityFilter.setPosterId(viewerIdentity.getIdentityId());
      streamIdentityIds = spaceStorage.getSpaceIdentityIdsByUserRole(viewerIdentity.getRemoteId(),
                                                                     String.valueOf(SpaceMembershipStatus.MEMBER),
                                                                     0,
                                                                     -1);
      streamIdentityIds = new ArrayList<>(streamIdentityIds);
      streamIdentityIds.add(viewerIdentity.getId());
      Set<Long> connectionIds = connectionDAO.getConnectionIds(Long.parseLong(viewerIdentity.getId()),
                                                               org.exoplatform.social.core.relationship.model.Relationship.Type.CONFIRMED);
      if (connectionIds != null) {
        streamIdentityIds.addAll(connectionIds.stream().map(String::valueOf).toList());
      }
      break;
    case ANY_SPACE_ACTIVITY:
      if (activityFilter.getSpaceIdentityId() == 0) {
        return 0;
      }
      break;
    default:
      throw new UnsupportedOperationException();
    }
    return activityDAO.getActivitiesCountByFilter(activityFilter, streamIdentityIds);
  }

  public List<ExoSocialActivity> getFavoriteActivities(Identity viewerIdentity, long spaceIdentityId) {
    long userIdentityId = Long.parseLong(viewerIdentity.getId());
    FavoriteService favoriteService = ExoContainerContext.getService(FavoriteService.class);
    List<MetadataItem> metadataItems = new ArrayList<>();
    if (spaceIdentityId > 0) {
      Identity spaceIdentity = identityStorage.findIdentityById(String.valueOf(spaceIdentityId));
      if (spaceIdentity != null) {
        Space space = spaceStorage.getSpaceByPrettyName(spaceIdentity.getRemoteId());
        metadataItems =
                      favoriteService.getFavoriteItemsByCreatorAndTypeAndSpaceId(ExoSocialActivityImpl.DEFAULT_ACTIVITY_METADATA_OBJECT_TYPE,
                                                                                 userIdentityId,
                                                                                 Long.parseLong(space.getId()),
                                                                                 0,
                                                                                 -1);
      }

    } else {
      metadataItems =
                    favoriteService.getFavoriteItemsByCreatorAndType(ExoSocialActivityImpl.DEFAULT_ACTIVITY_METADATA_OBJECT_TYPE,
                                                                     userIdentityId,
                                                                     0,
                                                                     -1);
    }

    return metadataItems.stream()
                        .map(metadataItem -> getActivityStorage().getActivity(String.valueOf(metadataItem.getObjectId())))
                        .sorted(Comparator.comparing(ExoSocialActivity::getUpdated).reversed())
                        .collect(Collectors.toList());
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getUserActivitiesForUpgrade(Identity owner,
                                                             long offset,
                                                             long limit) {
    return convertActivityIdsToActivities(activityDAO.getUserActivities(owner, offset, limit));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getActivities(Identity owner,
                                               Identity viewer,
                                               long offset,
                                               long limit) {
    return convertActivityIdsToActivities(activityDAO.getActivities(owner, viewer, offset, limit));
  }

  @Override
  public List<ExoSocialActivity> getAllActivities(int index, int limit) {
    return convertActivityEntitiesToActivities(activityDAO.getAllActivities());
  }

  @Override
  @ExoTransactional
  public void saveComment(ExoSocialActivity activity, ExoSocialActivity comment) {
    ActivityEntity activityEntity = activityDAO.find(Long.valueOf(activity.getId()));
    EntityManagerHolder.get().lock(activityEntity, LockModeType.PESSIMISTIC_WRITE);
    try {
      if (CollectionUtils.isNotEmpty(comment.getFiles())) {
        String ownerIdentityId = activityEntity.getOwnerId();
        try {
          Identity owner = identityStorage.findIdentityById(ownerIdentityId);
          storeFile(owner, comment);
        } catch (Exception e) {
          throw new ActivityStorageException(Type.FAILED_TO_ATTACH_FILES_TO_ACTIVITY,
                                             "Failed to attach files into activity " + comment.getId(),
                                             e);
        }
      }

      ActivityEntity commentEntity = convertCommentToCommentEntity(activityEntity, comment);
      commentEntity = activityDAO.create(commentEntity);

      comment.setId(getExoCommentID(commentEntity.getId()));
      comment.setPosterId(commentEntity.getPosterId());
      comment.setParentId(String.valueOf(activity.getId()));
      comment.isComment(true);

      Set<String> mentioned = commentEntity.getMentionerIds();
      if (mentioned != null && !mentioned.isEmpty()) {
        comment.setMentionedIds(mentioned.toArray(new String[mentioned.size()]));
      }

      //
      Identity commenter = identityStorage.findIdentityById(commentEntity.getPosterId());
      saveStreamItemForCommenter(commenter, activityEntity);

      //
      String[] mentioners = processMentionsByIdentityId(activityEntity.getOwnerId(),
                                                        comment.getTitle(),
                                                        comment.getTemplateParams());
      if (mentioners != null && mentioners.length > 0) {
        mention(activityEntity, mentioners);
        activityEntity.setMentionerIds(processMentionOfComment(activityEntity,
                                                               commentEntity,
                                                               activityEntity.getMentionerIds()
                                                                             .toArray(new String[activityEntity.getMentionerIds()
                                                                                                               .size()]),
                                                               mentioners,
                                                               true));
      }

      //
      processActivityStreamUpdatedTime(activityEntity);
      activityDAO.update(activityEntity);

    } finally {
      if (EntityManagerHolder.get().isOpen() && EntityManagerHolder.get().getLockMode(activityEntity) != null
          && EntityManagerHolder.get().getLockMode(activityEntity) != LockModeType.NONE) {
        EntityManagerHolder.get().lock(activityEntity, LockModeType.NONE);
      }
    }

  }

  /**
   * Creates the StreamItem for commenter
   * 
   * @param commenter
   * @param activityEntity
   */
  private void saveStreamItemForCommenter(Identity commenter, ActivityEntity activityEntity) {
    Identity ownerActivity = identityStorage.findIdentityById(activityEntity.getOwnerId());
    if (!SpaceIdentityProvider.NAME.equals(ownerActivity.getProviderId())) {
      createStreamItem(StreamType.COMMENTER, activityEntity, Long.parseLong(commenter.getId()));
    }
  }

  private Set<String> processMentionOfComment(ActivityEntity activityEntity,
                                              ActivityEntity commentEntity,
                                              String[] activityMentioners,
                                              String[] commentMentioners,
                                              boolean isAdded) {
    Set<String> mentioners = new HashSet<>(Arrays.asList(activityMentioners));
    if (commentMentioners.length == 0)
      return mentioners;
    //
    for (String mentioner : commentMentioners) {
      if (!mentioners.contains(mentioner) && isAdded) {
        mentioners.add(mentioner);
      }
      if (mentioners.contains(mentioner) && !isAdded
          && isAllowedToRemove(activityEntity, commentEntity, mentioner)) {
        mentioners.remove(mentioner);
        // remove stream item
        StreamItemEntity item = new StreamItemEntity(StreamType.MENTIONER);
        item.setOwnerId(Long.parseLong(mentioner));
        activityEntity.removeStreamItem(item);
      }
    }
    return mentioners;

  }

  private boolean isAllowedToRemove(ActivityEntity activity, ActivityEntity comment, String mentioner) {
    if (ArrayUtils.contains(processMentions((Space) null, activity.getTitle(), activity.getTemplateParams()), mentioner)) {
      return false;
    }
    List<ActivityEntity> comments = activity.getComments();
    comments.remove(comment);
    for (ActivityEntity cmt : comments) {
      if (ArrayUtils.contains(processMentions((Space) null, cmt.getTitle(), cmt.getTemplateParams()), mentioner)) {
        return false;
      }
    }
    return true;
  }

  @Override
  public ExoSocialActivity saveActivity(Identity owner, ExoSocialActivity activity) {
    boolean isNew = (activity.getId() == null);

    if (CollectionUtils.isNotEmpty(activity.getFiles())) {
      try {
        storeFile(owner, activity);
      } catch (Exception e) {
        throw new ActivityStorageException(Type.FAILED_TO_ATTACH_FILES_TO_ACTIVITY,
                                           "Failed to attach files into activity " + activity.getId(),
                                           e);
      }
    }

    ActivityEntity entity = convertActivityToActivityEntity(activity, owner.getId());
    //
    entity.setOwnerId(owner.getId());
    entity.setProviderId(owner.getProviderId());
    saveStreamItem(owner, entity);
    //
    entity = activityDAO.create(entity);
    activity.setId(Long.toString(entity.getId()));
    //

    if (isNew) {
      fillActivityFromEntity(entity, activity);
    }

    return activity;
  }

  public void addFileStoragePlugin(ActivityFileStoragePlugin fileStoragePlugin) {
    if (this.activityFileStorage == null || this.activityFileStorage.getPriority() < fileStoragePlugin.getPriority()) {
      this.activityFileStorage = fileStoragePlugin;
    }
    this.activityFileStorageByDSName.put(fileStoragePlugin.getDatasourceName(), fileStoragePlugin);
  }

  private void storeFile(Identity streamOwner, ExoSocialActivity activity) throws Exception {
    if (this.activityFileStorage == null) {
      throw new ActivityStorageException(Type.FAILED_TO_ATTACH_FILES_TO_ACTIVITY, "No configured file storage plugin");
    }
    List<ActivityFile> files = activity.getFiles();
    if (CollectionUtils.isNotEmpty(files)) {
      for (ActivityFile activityFile : files) {
        ActivityFileStoragePlugin activityFileStoragePlugin = this.activityFileStorage;
        if (StringUtils.isNotBlank(activityFile.getStorage())) {
          if (this.activityFileStorageByDSName.containsKey(activityFile.getStorage())) {
            activityFileStoragePlugin = this.activityFileStorageByDSName.get(activityFile.getStorage());
          } else {
            LOG.warn("No file storage plugin for datasource with name '{}'", activityFile.getStorage());
          }
        }
        if (StringUtils.isEmpty(activityFile.getId()) && StringUtils.isNotEmpty(activityFile.getUploadId())) {
          try {
            activityFileStoragePlugin.storeAttachments(activity, streamOwner, activityFile);
          } catch (Exception e) {
            LOG.error("Error while adding attachment to activity " + activityFile.getName(), e);
          }
        }
        // Attachments from existing resource
        else {
          try {
            activityFileStoragePlugin.attachExistingFile(activity, streamOwner, activityFile);
          } catch (Exception e) {
            LOG.error("Error while adding attachment to activity " + activityFile.getId(), e);
          }
        }
      }
      if (activity.getTemplateParams() != null) {
        activity.getTemplateParams().put("MESSAGE", activity.getTitle());
      }
    }
  }

  /**
   * Creates the StreamIteam for Space Member
   * 
   * @param spaceOwner
   * @param activity
   */
  private void spaceMembers(Identity spaceOwner, ActivityEntity activity) {
    createStreamItem(StreamType.SPACE, activity, Long.parseLong(spaceOwner.getId()));
    createStreamItem(StreamType.SPACE, activity, Long.parseLong(activity.getPosterId()));
  }

  private void saveStreamItem(Identity owner, ActivityEntity activityEntity) {
    // create StreamItem
    if (OrganizationIdentityProvider.NAME.equals(owner.getProviderId())) {
      // poster
      poster(owner, activityEntity);

    } else {
      // for SPACE
      spaceMembers(owner, activityEntity);
    }
    // mention
    mention(activityEntity,
            processMentionsByIdentityId(activityEntity.getOwnerId(),
                                        activityEntity.getTitle(),
                                        activityEntity.getTemplateParams()));
  }

  /**
   * Creates the StreamItem for poster and stream owner
   * 
   * @param owner
   * @param activity
   */
  private void poster(Identity owner, ActivityEntity activity) {
    createStreamItem(StreamType.POSTER, activity, Long.parseLong(activity.getPosterId()));
    // User A posts a new activity on user B stream (A connected B)
    if (!owner.getId().equals(activity.getPosterId())) {
      createStreamItem(StreamType.POSTER, activity, Long.parseLong(owner.getId()));
    }
  }

  /**
   * Creates StreamItem for each user who has mentioned on the activity
   * 
   * @param owner
   * @param activity
   */
  private void mention(ActivityEntity activity, String[] mentions) {
    for (String mentioner : mentions) {
      Identity identity = identityStorage.findIdentityById(mentioner);
      if (identity != null) {
        createStreamItem(StreamType.MENTIONER, activity, Long.parseLong(identity.getId()));
      }
    }
  }

  private void createStreamItem(StreamType streamType, ActivityEntity activity, Long ownerId) {
    StreamItemEntity streamItem = new StreamItemEntity(streamType);
    streamItem.setOwnerId(ownerId);
    if (streamType == StreamType.POSTER || streamType == StreamType.SPACE
        || streamType == StreamType.MENTIONER
        || streamType == StreamType.COMMENTER) {
      streamItem.setUpdatedDate(activity.getUpdatedDate());
    } else {
      streamItem.setUpdatedDate(null);
    }
    boolean isExist = activity.getId() != null
                      && activity.getStreamItems()
                                 .stream()
                                 .anyMatch(item -> item.getOwnerId().equals(ownerId) && streamType.equals(item.getStreamType()));
    if (!isExist) {
      activity.addStreamItem(streamItem);
    }
  }

  /**
   * Processes Mentioners who has been mentioned via the Activity.
   * 
   * @param identityId
   * @param title
   * @param templateParams
   */
  private String[] processMentionsByIdentityId(String identityId, String title, Map<String, String> templateParams) {
    Identity identity = StringUtils.isBlank(identityId) ? null : identityStorage.findIdentityById(identityId);
    if (identity != null && identity.isSpace()) {
      Space space = spaceStorage.getSpaceByPrettyName(identity.getRemoteId());
      return processMentions(space, title, templateParams);
    } else {
      return processMentions(null, title, templateParams);
    }
  }

  /**
   * Processes Mentioners who has been mentioned via the Activity.
   * 
   * @param space
   * @param title
   * @param templateParams
   */
  private String[] processMentions(Space space, String title, Map<String, String> templateParams) {
    Set<String> mentions = new HashSet<>();
    mentions.addAll(MentionUtils.getMentionedUsernames(title));
    getTemplateParamToProcess(templateParams).forEach(value -> mentions.addAll(MentionUtils.getMentionedUsernames(value)));

    if (space != null) {
      String spaceIdentityId = identityStorage.findIdentityId(SpaceIdentityProvider.NAME, space.getPrettyName());
      Set<String> mentionedRoles = MentionUtils.getMentionedRoles(title, spaceIdentityId);
      mentionedRoles.forEach(role -> {
        if (StringUtils.equals("member", role) && space.getMembers() != null) {
          mentions.addAll(Arrays.asList(space.getMembers()));
        } else if (StringUtils.equals("manager", role) && space.getManagers() != null) {
          mentions.addAll(Arrays.asList(space.getManagers()));
        } else if (StringUtils.equals("redactor", role) && space.getRedactors() != null) {
          mentions.addAll(Arrays.asList(space.getRedactors()));
        } else if (StringUtils.equals("publisher", role) && space.getPublishers() != null) {
          mentions.addAll(Arrays.asList(space.getPublishers()));
        }
      });
    }

    return mentions.stream()
                   .map(remoteId -> {
                     Identity identity = identityStorage.findIdentity(OrganizationIdentityProvider.NAME, remoteId);
                     return identity == null ? null : identity.getId();
                   })
                   .filter(Objects::nonNull)
                   .collect(Collectors.toSet())
                   .toArray(new String[mentions.size()]);
  }

  private List<String> getTemplateParamToProcess(Map<String, String> templateParams) {
    List<String> params = new ArrayList<>();

    if (templateParams != null && templateParams.containsKey(BaseActivityProcessorPlugin.TEMPLATE_PARAM_TO_PROCESS)) {
      String[] templateParamKeys = templateParams
                                                 .get(BaseActivityProcessorPlugin.TEMPLATE_PARAM_TO_PROCESS)
                                                 .split(BaseActivityProcessorPlugin.TEMPLATE_PARAM_LIST_DELIM);
      for (String key : templateParamKeys) {
        if (templateParams.containsKey(key)) {
          params.add(templateParams.get(key));
        }
      }
    }
    return params;
  }

  @Override
  @ExoTransactional
  public ExoSocialActivity getParentActivity(ExoSocialActivity comment) {
    try {
      Long commentId = getCommentID(comment.getId());
      return convertActivityEntityToActivity(activityDAO.getParentActivity(commentId));
    } catch (NumberFormatException e) {
      LOG.warn("The input ExoSocialActivity is not comment, it is Activity");
      return null;
    }
  }

  @Override
  public void deleteActivity(String activityId) {
    ActivityEntity a = activityDAO.find(Long.valueOf(activityId));
    if (a != null) {
      activityDAO.delete(a);
    } else {
      LOG.warn("The activity's " + activityId + " is not found!");
    }
  }

  @Override
  public ExoSocialActivity hideActivity(String activityId) {
    ActivityEntity activityEntity = activityDAO.find(Long.valueOf(activityId));
    activityEntity.setHidden(true);
    activityEntity = activityDAO.update(activityEntity);
    return convertActivityEntityToActivity(activityEntity);
  }

  @Override
  public ExoSocialActivity pinActivity(String activityId, String userId) {
    ActivityEntity activityEntity = activityDAO.find(Long.valueOf(activityId));
    activityEntity.setPinned(true);
    activityEntity.setPinDate(new Date());
    activityEntity.setPinAuthorId(Long.valueOf(userId));
    activityEntity = activityDAO.update(activityEntity);
    return convertActivityEntityToActivity(activityEntity);
  }

  @Override
  public ExoSocialActivity unpinActivity(String activityId) {
    ActivityEntity activityEntity = activityDAO.find(Long.valueOf(activityId));
    activityEntity.setPinned(false);
    activityEntity.setPinDate(null);
    activityEntity.setPinAuthorId(null);
    activityEntity = activityDAO.update(activityEntity);
    return convertActivityEntityToActivity(activityEntity);
  }

  @Override
  @ExoTransactional
  public void deleteComment(String activityId, String commentId) {
    ActivityEntity comment = activityDAO.find(getCommentID(commentId));
    activityDAO.delete(comment);
    //
    ActivityEntity activity = activityDAO.find(Long.valueOf(activityId));
    activity.getComments().remove(comment);
    //
    activity.setMentionerIds(processMentionOfComment(activity,
                                                     comment,
                                                     activity.getMentionerIds()
                                                             .toArray(new String[activity.getMentionerIds().size()]),
                                                     processMentionsByIdentityId(activity.getOwnerId(),
                                                                                 comment.getTitle(),
                                                                                 comment.getTemplateParams()),
                                                     false));
    //
    if (!hasOtherComment(activity, comment.getPosterId())) {
      StreamItemEntity item = new StreamItemEntity(StreamType.COMMENTER);
      item.setOwnerId(Long.parseLong(comment.getPosterId()));
      activity.removeStreamItem(item);
    }
    //
    activityDAO.update(activity);
  }

  @Override
  public List<Long> getActivityCategoryIds(long spaceIdentityId) {
    return activityDAO.getActivityCategoryIds(spaceIdentityId);
  }

  @Override
  public List<Long> getActivityCategoryIds(Set<Long> streamFeedOwnerIds) {
    return activityDAO.getActivityCategoryIds(streamFeedOwnerIds);
  }

  private boolean hasOtherComment(ActivityEntity activity, String poster) {
    for (ActivityEntity comment : activity.getComments()) {
      if (poster.equals(comment.getPosterId())) {
        return true;
      }
    }
    return false;
  }

  @Override
  public int getNumberOfUserActivities(Identity owner) {
    return getNumberOfUserActivitiesForUpgrade(owner);
  }

  @Override
  public int getNumberOfUserActivitiesForUpgrade(Identity owner) {
    return activityDAO.getNumberOfUserActivities(owner);
  }

  @Override
  public int getNumberOfNewerOnUserActivities(Identity ownerIdentity, ExoSocialActivity baseActivity) {
    return getNumberOfNewerOnUserActivities(ownerIdentity, baseActivity.getUpdated().getTime());
  }

  @Override
  public List<ExoSocialActivity> getNewerOnUserActivities(Identity ownerIdentity, ExoSocialActivity baseActivity, int limit) {
    return getNewerUserActivities(ownerIdentity, baseActivity.getUpdated().getTime(), limit);
  }

  @Override
  public int getNumberOfOlderOnUserActivities(Identity ownerIdentity, ExoSocialActivity baseActivity) {
    return getNumberOfOlderOnUserActivities(ownerIdentity, baseActivity.getUpdated().getTime());
  }

  @Override
  public List<ExoSocialActivity> getOlderOnUserActivities(Identity ownerIdentity, ExoSocialActivity baseActivity, int limit) {
    return getOlderUserActivities(ownerIdentity, baseActivity.getUpdated().getTime(), limit);
  }

  @Override
  public List<ExoSocialActivity> getActivityFeed(Identity ownerIdentity, int offset, int limit) {
    return getActivityFeedForUpgrade(ownerIdentity, offset, limit);
  }

  @Override
  public List<String> getActivityIdsFeed(Identity ownerIdentity, int offset, int limit) {
    return activityDAO.getActivityIdsFeed(ownerIdentity, offset, limit, memberOfSpaceIds(ownerIdentity));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getActivityFeedForUpgrade(Identity ownerIdentity, int offset, int limit) {
    return convertActivityIdsToActivities(activityDAO.getActivityFeed(ownerIdentity,
                                                                      offset,
                                                                      limit,
                                                                      memberOfSpaceIds(ownerIdentity)));
  }

  @Override
  public int getNumberOfActivitesOnActivityFeed(Identity ownerIdentity) {
    return getNumberOfActivitesOnActivityFeedForUpgrade(ownerIdentity);
  }

  @Override
  public int getNumberOfActivitesOnActivityFeedForUpgrade(Identity ownerIdentity) {
    return activityDAO.getNumberOfActivitesOnActivityFeed(ownerIdentity, memberOfSpaceIds(ownerIdentity));
  }

  @Override
  public int getNumberOfNewerOnActivityFeed(Identity ownerIdentity, ExoSocialActivity baseActivity) {
    return getNumberOfNewerOnActivityFeed(ownerIdentity, baseActivity.getUpdated().getTime());
  }

  @Override
  public List<ExoSocialActivity> getNewerOnActivityFeed(Identity ownerIdentity, ExoSocialActivity baseActivity, int limit) {
    return getNewerFeedActivities(ownerIdentity, baseActivity.getUpdated().getTime(), limit);
  }

  @Override
  public int getNumberOfOlderOnActivityFeed(Identity ownerIdentity, ExoSocialActivity baseActivity) {
    return getNumberOfOlderOnActivityFeed(ownerIdentity, baseActivity.getUpdated().getTime());
  }

  @Override
  public List<ExoSocialActivity> getOlderOnActivityFeed(Identity ownerIdentity, ExoSocialActivity baseActivity, int limit) {
    return getOlderFeedActivities(ownerIdentity, baseActivity.getUpdated().getTime(), limit);
  }

  @Override
  public List<ExoSocialActivity> getActivitiesOfConnections(Identity ownerIdentity, int offset, int limit) {
    return getActivitiesOfConnectionsForUpgrade(ownerIdentity, offset, limit);
  }

  @Override
  public List<String> getActivityIdsOfConnections(Identity ownerIdentity, int offset, int limit) {
    return activityDAO.getActivityIdsOfConnections(ownerIdentity, offset, limit);
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getActivitiesOfConnectionsForUpgrade(Identity ownerIdentity, int offset, int limit) {
    return convertActivityIdsToActivities(activityDAO.getActivitiesOfConnections(ownerIdentity, offset, limit));
  }

  @Override
  public int getNumberOfActivitiesOfConnections(Identity ownerIdentity) {
    return getNumberOfActivitiesOfConnectionsForUpgrade(ownerIdentity);
  }

  @Override
  public int getNumberOfActivitiesOfConnectionsForUpgrade(Identity ownerIdentity) {
    return activityDAO.getNumberOfActivitiesOfConnections(ownerIdentity);
  }

  @Override
  public List<ExoSocialActivity> getActivitiesOfIdentity(Identity ownerIdentity, long offset, long limit) {
    return getUserActivities(ownerIdentity, offset, limit);
  }

  @Override
  public int getNumberOfNewerOnActivitiesOfConnections(Identity ownerIdentity, ExoSocialActivity baseActivity) {
    return getNumberOfNewerOnActivitiesOfConnections(ownerIdentity, baseActivity.getUpdated().getTime());
  }

  @Override
  public List<ExoSocialActivity> getNewerOnActivitiesOfConnections(Identity ownerIdentity,
                                                                   ExoSocialActivity baseActivity,
                                                                   long limit) {
    return getNewerActivitiesOfConnections(ownerIdentity, baseActivity.getUpdated().getTime(), (int) limit);
  }

  @Override
  public int getNumberOfOlderOnActivitiesOfConnections(Identity ownerIdentity, ExoSocialActivity baseActivity) {
    return getNumberOfOlderOnActivitiesOfConnections(ownerIdentity, baseActivity.getUpdated().getTime());
  }

  @Override
  public List<ExoSocialActivity> getOlderOnActivitiesOfConnections(Identity ownerIdentity,
                                                                   ExoSocialActivity baseActivity,
                                                                   int limit) {
    return getOlderActivitiesOfConnections(ownerIdentity, baseActivity.getUpdated().getTime(), limit);
  }

  @Override
  public List<ExoSocialActivity> getUserSpacesActivities(Identity ownerIdentity, int offset, int limit) {
    return getUserSpacesActivitiesForUpgrade(ownerIdentity, offset, limit);
  }

  @Override
  public List<String> getSpaceActivityIds(Identity spaceIdentity, int offset, int limit) {
    return activityDAO.getSpaceActivityIds(spaceIdentity, offset, limit);
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getUserSpacesActivitiesForUpgrade(Identity ownerIdentity, int offset, int limit) {
    return convertActivityIdsToActivities(activityDAO.getUserSpacesActivities(ownerIdentity,
                                                                              offset,
                                                                              limit,
                                                                              memberOfSpaceIds(ownerIdentity)));
  }

  @Override
  public int getNumberOfUserSpacesActivities(Identity ownerIdentity) {
    return getNumberOfUserSpacesActivitiesForUpgrade(ownerIdentity);
  }

  @Override
  public int getNumberOfUserSpacesActivitiesForUpgrade(Identity ownerIdentity) {
    return activityDAO.getNumberOfUserSpacesActivities(ownerIdentity, memberOfSpaceIds(ownerIdentity));
  }

  @Override
  public int getNumberOfNewerOnUserSpacesActivities(Identity ownerIdentity, ExoSocialActivity baseActivity) {
    return getNumberOfNewerOnUserSpacesActivities(ownerIdentity, baseActivity.getUpdated().getTime());
  }

  @Override
  public List<ExoSocialActivity> getNewerOnUserSpacesActivities(Identity ownerIdentity,
                                                                ExoSocialActivity baseActivity,
                                                                int limit) {
    return getNewerUserSpacesActivities(ownerIdentity, baseActivity.getUpdated().getTime(), limit);
  }

  @Override
  public int getNumberOfOlderOnUserSpacesActivities(Identity ownerIdentity, ExoSocialActivity baseActivity) {
    return getNumberOfOlderOnUserSpacesActivities(ownerIdentity, baseActivity.getUpdated().getTime());
  }

  @Override
  public List<ExoSocialActivity> getOlderOnUserSpacesActivities(Identity ownerIdentity,
                                                                ExoSocialActivity baseActivity,
                                                                int limit) {
    return getOlderUserSpacesActivities(ownerIdentity, baseActivity.getUpdated().getTime(), limit);
  }

  @Override
  public List<ExoSocialActivity> getComments(ExoSocialActivity existingActivity, boolean loadSubComments, int offset, int limit) {
    return getComments(existingActivity, loadSubComments, offset, limit, false);
  }

  @Override
  public List<ExoSocialActivity> getComments(ExoSocialActivity existingActivity,
                                             boolean loadSubComments,
                                             int offset,
                                             int limit,
                                             boolean sortDescending) {
    long activityId = 0;
    try {
      activityId = Long.parseLong(existingActivity.getId());
    } catch (NumberFormatException ex) {
      activityId = 0;
    }

    List<ActivityEntity> comments;
    if (activityId > 0) {
      comments = activityDAO.getComments(activityId, offset, limit, sortDescending);
    } else {
      comments = null;
    }

    return convertCommentEntitiesToComments(comments, loadSubComments, sortDescending);
  }

  @Override
  public int getNumberOfComments(ExoSocialActivity existingActivity) {
    return (int) activityDAO.getNumberOfComments(Long.valueOf(existingActivity.getId()));
  }

  @Override
  public int getNumberOfAllComments(String activityId) {
    return activityDAO.getNumberOfAllComments(Long.parseLong(activityId));
  }

  @Override
  public int getNumberOfNewerComments(ExoSocialActivity existingActivity, ExoSocialActivity baseComment) {
    return getNewerComments(existingActivity, baseComment, 0).size();
  }

  @Override
  public List<ExoSocialActivity> getNewerComments(ExoSocialActivity existingActivity, ExoSocialActivity baseComment, int limit) {
    return getNewerComments(existingActivity, baseComment.getPostedTime(), limit);
  }

  @Override
  public int getNumberOfOlderComments(ExoSocialActivity existingActivity, ExoSocialActivity baseComment) {
    return getOlderComments(existingActivity, baseComment, 0).size();
  }

  @Override
  public List<ExoSocialActivity> getOlderComments(ExoSocialActivity existingActivity, ExoSocialActivity baseComment, int limit) {
    return getOlderComments(existingActivity, baseComment.getPostedTime(), limit);
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getNewerComments(ExoSocialActivity existingActivity, Long sinceTime, int limit) {
    List<ActivityEntity> comments = activityDAO.getNewerComments(Long.valueOf(existingActivity.getId()),
                                                                 sinceTime > 0 ? new Date(sinceTime) : null,
                                                                 0,
                                                                 limit);
    //
    return convertCommentEntitiesToComments(comments);
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getOlderComments(ExoSocialActivity existingActivity, Long sinceTime, int limit) {
    List<ActivityEntity> comments = activityDAO.getOlderComments(Long.valueOf(existingActivity.getId()),
                                                                 sinceTime > 0 ? new Date(sinceTime) : null,
                                                                 0,
                                                                 limit);
    return convertCommentEntitiesToComments(comments);
  }

  @Override
  public int getNumberOfNewerComments(ExoSocialActivity existingActivity, Long sinceTime) {
    return getNewerComments(existingActivity, sinceTime, 0).size();
  }

  @Override
  public int getNumberOfOlderComments(ExoSocialActivity existingActivity, Long sinceTime) {
    return getOlderComments(existingActivity, sinceTime, 0).size();
  }

  @Override
  public SortedSet<ActivityProcessor> getActivityProcessors() {
    return activityProcessors;
  }

  @Override
  public void updateActivity(ExoSocialActivity existingActivity) {
    if (existingActivity == null) {
      throw new IllegalArgumentException("Activity to update cannot be null");
    }
    if (existingActivity.getTemplateParams() != null) {
      existingActivity.getTemplateParams().put("id", "");
      existingActivity.getTemplateParams().put("DOCPATH", "");
      existingActivity.getTemplateParams().put("docTitle", "");
      existingActivity.getTemplateParams().put("mimeType", "");
    }
    if (CollectionUtils.isNotEmpty(existingActivity.getFiles())) {
      try {
        ConversationState conversationstate = ConversationState.getCurrent();
        String currentUsername = conversationstate.getIdentity().getUserId();
        IdentityManager identityManager = CommonsUtils.getService(IdentityManager.class);
        Identity identity = identityManager.getOrCreateIdentity(OrganizationIdentityProvider.NAME, currentUsername);
        storeFile(identity, existingActivity);
      } catch (Exception e) {
        throw new ActivityStorageException(Type.FAILED_TO_ATTACH_FILES_TO_ACTIVITY,
                                           "Failed to attach files into activity " + existingActivity.getId(),
                                           e);
      }
    }
    ActivityEntity parentActivity = null;
    ActivityEntity updatedActivity = null;
    boolean isComment = existingActivity.getId().startsWith(COMMENT_PREFIX);
    if (isComment) {
      long id = getCommentID(existingActivity.getId());
      updatedActivity = activityDAO.find(id);
      parentActivity = getTopParentActivity(updatedActivity);
    } else {
      parentActivity = updatedActivity = activityDAO.find(Long.valueOf(existingActivity.getId()));
    }

    if (updatedActivity != null) {
      if (existingActivity.getTemplateParams() != null) {
        preSaveProcessActivity(existingActivity);
      }

      // only raise the activity in stream when activity date updated
      if (existingActivity.getUpdated() != null && updatedActivity.getUpdatedDate() != null
          && existingActivity.getUpdated().getTime() != updatedActivity.getUpdatedDate().getTime()) {
        processActivityStreamUpdatedTime(updatedActivity);
      }
      // create or remove liker if exist
      processLikerActivityInStreams(new HashSet<>(Arrays.asList(existingActivity.getLikeIdentityIds())),
                                    new HashSet<>(updatedActivity.getLikerIds()),
                                    parentActivity,
                                    isComment);
      // this method must be called before updating title and template params
      updateActivityMentions(updatedActivity, existingActivity);

      if (existingActivity.getTitleId() != null)
        updatedActivity.setTitleId(existingActivity.getTitleId());
      if (existingActivity.getType() != null)
        updatedActivity.setType(existingActivity.getType());
      if (existingActivity.getTitle() != null)
        updatedActivity.setTitle(existingActivity.getTitle());
      if (existingActivity.getBody() != null)
        updatedActivity.setBody(existingActivity.getBody());
      if (existingActivity.getLikeIdentityIds() != null)
        updatedActivity.setLikerIds(new HashSet<>(Arrays.asList(existingActivity.getLikeIdentityIds())));
      if (existingActivity.getPermaLink() != null)
        updatedActivity.setPermaLink(existingActivity.getPermaLink());
      if (existingActivity.getTemplateParams() != null) {
        Map<String, String> updatedActivityTemplateParams = updatedActivity.getTemplateParams();
        updatedActivityTemplateParams.putAll(existingActivity.getTemplateParams());
        updatedActivity.setTemplateParams(updatedActivityTemplateParams);
      }
      if (existingActivity.getUpdated() != null)
        updatedActivity.setUpdatedDate(existingActivity.getUpdated());
      updatedActivity.setHidden(existingActivity.isHidden());
      updatedActivity.setComment(existingActivity.isComment());
      updatedActivity.setLocked(existingActivity.isLocked());
      updatedActivity.setPinned(existingActivity.isPinned());
      updatedActivity.setPinDate(StorageUtils.parseRFC3339Date(existingActivity.getPinDate()));
      updatedActivity.setPinAuthorId(existingActivity.getPinAuthorId());
      updatedActivity.setCategoryIds(existingActivity.getCategoryIds());

      activityDAO.update(updatedActivity);
    } else {
      throw new ActivityStorageException(Type.FAILED_TO_UPDATE_ACTIVITY,
                                         "Cannot find activity with id=" + existingActivity.getId());
    }
  }

  private void processLikerActivityInStreams(Set<String> newLikerList,
                                             Set<String> oldLikerList,
                                             ActivityEntity activity,
                                             boolean commentLike) {
    for (String id : newLikerList) {
      if (!oldLikerList.contains(id)) {// new like ==> create stream item
        createStreamItem(commentLike ? StreamType.COMMENT_LIKER : StreamType.LIKER, activity, Long.parseLong(id));
      } else {
        oldLikerList.remove(id);
      }
    }
    if (!oldLikerList.isEmpty()) {// unlike ==> remove stream item
      for (String id : oldLikerList) {
        StreamItemEntity item = new StreamItemEntity(StreamType.LIKER);
        item.setOwnerId(Long.parseLong(id));
        activity.removeStreamItem(item);
        item.setStreamType(StreamType.COMMENT_LIKER);
        activity.removeStreamItem(item);
      }
    }
  }

  @Override
  public int getNumberOfNewerOnActivityFeed(Identity ownerIdentity, Long sinceTime) {
    return activityDAO.getNumberOfNewerOnActivityFeed(ownerIdentity, sinceTime, memberOfSpaceIds(ownerIdentity));
  }

  @Override
  public int getNumberOfNewerOnUserActivities(Identity ownerIdentity, Long sinceTime) {
    return activityDAO.getNumberOfNewerOnUserActivities(ownerIdentity, sinceTime);
  }

  @Override
  public int getNumberOfNewerOnActivitiesOfConnections(Identity ownerIdentity, Long sinceTime) {
    return activityDAO.getNumberOfNewerOnActivitiesOfConnections(ownerIdentity, sinceTime);
  }

  @Override
  public int getNumberOfNewerOnUserSpacesActivities(Identity ownerIdentity, Long sinceTime) {
    return activityDAO.getNumberOfNewerOnUserSpacesActivities(ownerIdentity, sinceTime, memberOfSpaceIds(ownerIdentity));
  }

  @Override
  public int getNumberOfSpaceActivities(Identity spaceIdentity) {
    return getNumberOfSpaceActivitiesForUpgrade(spaceIdentity);
  }

  @Override
  public int getNumberOfSpaceActivitiesForUpgrade(Identity spaceIdentity) {
    return activityDAO.getNumberOfSpaceActivities(spaceIdentity);
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getSpaceActivities(Identity spaceIdentity, int offset, int limit) {
    return convertActivityIdsToActivities(activityDAO.getSpaceActivities(spaceIdentity, offset, limit));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getSpaceActivitiesForUpgrade(Identity spaceIdentity, int offset, int limit) {
    return convertActivityIdsToActivities(activityDAO.getSpaceActivities(spaceIdentity, offset, limit));
  }

  @Override
  public List<ExoSocialActivity> getActivitiesByPoster(Identity posterIdentity, int offset, int limit) {
    return getActivitiesByPoster(posterIdentity, offset, limit, new String[0]);
  }

  @Override
  public List<ExoSocialActivity> getActivitiesByPoster(Identity posterIdentity, int offset, int limit, String... activityTypes) {
    return convertActivityIdsToActivities(activityDAO.getActivitiesByPoster(posterIdentity, offset, limit, activityTypes));
  }

  @Override
  public int getNumberOfActivitiesByPoster(Identity posterIdentity) {
    return activityDAO.getNumberOfActivitiesByPoster(posterIdentity, new String[0]);
  }

  @Override
  public int getNumberOfActivitiesByPoster(Identity ownerIdentity, Identity viewerIdentity) {
    return 0;
  }

  @Override
  public List<ExoSocialActivity> getNewerOnSpaceActivities(Identity spaceIdentity, ExoSocialActivity baseActivity, int limit) {
    return getNewerSpaceActivities(spaceIdentity, baseActivity.getUpdated().getTime(), limit);
  }

  @Override
  public int getNumberOfNewerOnSpaceActivities(Identity spaceIdentity, ExoSocialActivity baseActivity) {
    return getNumberOfNewerOnSpaceActivities(spaceIdentity, baseActivity.getUpdated().getTime());
  }

  @Override
  public List<ExoSocialActivity> getOlderOnSpaceActivities(Identity spaceIdentity, ExoSocialActivity baseActivity, int limit) {
    return getOlderSpaceActivities(spaceIdentity, baseActivity.getUpdated().getTime(), limit);
  }

  @Override
  public int getNumberOfOlderOnSpaceActivities(Identity spaceIdentity, ExoSocialActivity baseActivity) {
    return getNumberOfOlderOnSpaceActivities(spaceIdentity, baseActivity.getUpdated().getTime());
  }

  @Override
  public int getNumberOfNewerOnSpaceActivities(Identity spaceIdentity, Long sinceTime) {
    return activityDAO.getNumberOfNewerOnSpaceActivities(spaceIdentity, sinceTime);
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getNewerFeedActivities(Identity owner, Long sinceTime, int limit) {
    return convertActivityIdsToActivities(activityDAO.getNewerOnActivityFeed(owner, sinceTime, limit, memberOfSpaceIds(owner)));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getNewerUserActivities(Identity owner, Long sinceTime, int limit) {
    return convertActivityIdsToActivities(activityDAO.getNewerOnUserActivities(owner, sinceTime, limit));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getNewerUserSpacesActivities(Identity owner, Long sinceTime, int limit) {
    return convertActivityIdsToActivities(activityDAO.getNewerOnUserSpacesActivities(owner,
                                                                                     sinceTime,
                                                                                     limit,
                                                                                     memberOfSpaceIds(owner)));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getNewerActivitiesOfConnections(Identity owner, Long sinceTime, int limit) {
    return convertActivityIdsToActivities(activityDAO.getNewerOnActivitiesOfConnections(owner, sinceTime, limit));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getNewerSpaceActivities(Identity owner, Long sinceTime, int limit) {
    return convertActivityIdsToActivities(activityDAO.getNewerOnSpaceActivities(owner, sinceTime, limit));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getOlderFeedActivities(Identity owner, Long sinceTime, int limit) {
    return convertActivityIdsToActivities(activityDAO.getOlderOnActivityFeed(owner, sinceTime, limit, memberOfSpaceIds(owner)));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getOlderUserActivities(Identity owner, Long sinceTime, int limit) {
    return convertActivityIdsToActivities(activityDAO.getOlderOnUserActivities(owner, sinceTime, limit));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getOlderUserSpacesActivities(Identity owner, Long sinceTime, int limit) {
    return convertActivityIdsToActivities(activityDAO.getOlderOnUserSpacesActivities(owner,
                                                                                     sinceTime,
                                                                                     limit,
                                                                                     memberOfSpaceIds(owner)));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getOlderActivitiesOfConnections(Identity owner, Long sinceTime, int limit) {
    return convertActivityIdsToActivities(activityDAO.getOlderOnActivitiesOfConnections(owner, sinceTime, limit));
  }

  @Override
  @ExoTransactional
  public List<ExoSocialActivity> getOlderSpaceActivities(Identity owner, Long sinceTime, int limit) {
    return convertActivityIdsToActivities(activityDAO.getOlderOnSpaceActivities(owner, sinceTime, limit));
  }

  @Override
  public int getNumberOfOlderOnActivityFeed(Identity ownerIdentity, Long sinceTime) {
    return activityDAO.getNumberOfOlderOnActivityFeed(ownerIdentity, sinceTime, memberOfSpaceIds(ownerIdentity));
  }

  @Override
  public int getNumberOfOlderOnUserActivities(Identity ownerIdentity, Long sinceTime) {
    return activityDAO.getNumberOfOlderOnUserActivities(ownerIdentity, sinceTime);
  }

  @Override
  public int getNumberOfOlderOnActivitiesOfConnections(Identity ownerIdentity, Long sinceTime) {
    return activityDAO.getNumberOfOlderOnActivitiesOfConnections(ownerIdentity, sinceTime);
  }

  @Override
  public int getNumberOfOlderOnUserSpacesActivities(Identity ownerIdentity, Long sinceTime) {
    return activityDAO.getNumberOfOlderOnUserSpacesActivities(ownerIdentity, sinceTime, memberOfSpaceIds(ownerIdentity));
  }

  @Override
  public int getNumberOfOlderOnSpaceActivities(Identity ownerIdentity, Long sinceTime) {
    return activityDAO.getNumberOfOlderOnSpaceActivities(ownerIdentity, sinceTime);
  }

  @Override
  public List<ExoSocialActivity> getSubComments(ExoSocialActivity comment) {
    long commentId = getCommentID(comment.getId());
    List<ActivityEntity> subComments = activityDAO.getComments(commentId, 0, -1, false);
    return convertCommentEntitiesToComments(subComments, false);
  }

  @Override
  public Set<Long> getStreamFeedOwnerIds(Identity identity) {
    Set<Long> owners = new HashSet<>();
    long identityId = Long.parseLong(identity.getId());
    owners.add(identityId);
    List<String> spaceIds = memberOfSpaceIds(identity);
    if (spaceIds != null) {
      owners.addAll(spaceIds.stream().map(spaceId -> Long.parseLong(spaceId)).collect(Collectors.toList()));
    }
    Set<Long> connectionIds =
                            connectionDAO.getConnectionIds(identityId,
                                                           org.exoplatform.social.core.relationship.model.Relationship.Type.CONFIRMED);
    if (connectionIds != null) {
      owners.addAll(connectionIds);
    }
    return owners;
  }

  @Override
  public ActivityShareAction createShareActivityAction(ActivityShareAction activityShareAction) {
    ActivityShareActionEntity actionEntity = toEntity(activityShareAction);
    actionEntity.setId(null);
    actionEntity.setShareDate(new Date());
    actionEntity = activityShareActionDAO.create(actionEntity);
    return fromEntity(actionEntity);
  }

  private Long getCommentID(String commentId) {
    return (commentId == null || commentId.trim().isEmpty()) ? null : Long.valueOf(commentId.replace(COMMENT_PREFIX, ""));
  }

  private String getExoCommentID(Long commentId) {
    return String.valueOf(COMMENT_PREFIX + commentId);
  }

  private void preSaveProcessActivity(ExoSocialActivity existingActivity) {
    processActivity(existingActivity, true);
  }

  private void processActivity(ExoSocialActivity existingActivity) {
    processActivity(existingActivity, false);
  }

  private void processActivity(ExoSocialActivity existingActivity, boolean preSave) {
    Map<String, String> templateParams = existingActivity.getTemplateParams();
    if (templateParams == null) {
      templateParams = new HashMap<>();
      existingActivity.setTemplateParams(templateParams);
    }
    Iterator<ActivityProcessor> it = activityProcessors.iterator();
    while (it.hasNext()) {
      ActivityProcessor processor = it.next();
      if ((preSave && !processor.isPreActivityProcessor())
          || (!preSave && !processor.isReadActivityProcessor())) {
        continue;
      }
      try {
        processor.processActivity(existingActivity);
      } catch (Exception e) {
        LOG.warn("Activity processing failed", e);
      }
    }
  }

  /**
   * This method will update StreamItem updatedDate field, which is used to sort
   * activies in Activity Stream We only update this in 3 cases: 1. edit
   * activity message 2. edit activity's comment message 3. add new comment
   * 
   * @param activityEntity
   * @return
   */
  private ActivityEntity processActivityStreamUpdatedTime(ActivityEntity activityEntity) {
    if (activityEntity.getStreamItems() != null) {
      List<StreamItemEntity> items = activityEntity.getStreamItems()
                                                   .stream()
                                                   .filter(item -> item.getStreamType() == StreamType.POSTER
                                                                   || item.getStreamType() == StreamType.SPACE)
                                                   .toList();
      if (!items.isEmpty()) {
        Date now = new Date();
        items.stream().forEach(item -> item.setUpdatedDate(now));
      }
    }
    return activityEntity;
  }

  /**
   * Gets the list of spaceIds what the given identify is member
   * 
   * @param ownerIdentity
   * @return
   */
  private List<String> memberOfSpaceIds(Identity ownerIdentity) {
    return spaceStorage.getMemberRoleSpaceIdentityIds(ownerIdentity.getId(), 0, -1);

  }

  public void setIdentityStorage(IdentityStorage identityStorage) {
    this.identityStorage = identityStorage;
  }

  public static final class CommentComparator implements Comparator<ActivityEntity> {

    private final boolean sortDescending;

    public CommentComparator(boolean sortDescending) {
      this.sortDescending = sortDescending;
    }

    public int compare(ActivityEntity o1, ActivityEntity o2) {
      ActivityEntity parent1 = o1.getParent();
      ActivityEntity parent2 = o2.getParent();

      boolean isParentActivity1 = parent1 == null || !parent1.isComment();
      boolean isParentActivity2 = parent2 == null || !parent2.isComment();

      if (isParentActivity1 && isParentActivity2) {
        return sortDescending ? o2.getPosted().compareTo(o1.getPosted()) : o1.getPosted().compareTo(o2.getPosted());
      } else if (isParentActivity1) {
        return compare(o1, parent2);
      } else if (isParentActivity2) {
        return compare(parent1, o2);
      } else if (parent1.getId() == parent2.getId()) {
        return o1.getPosted().compareTo(o2.getPosted());
      } else {
        return compare(parent1, parent2);
      }
    }
  }

  public ActivityStorage getActivityStorage() {
    if (activityStorage == null) {
      activityStorage = CommonsUtils.getService(ActivityStorage.class);
      // This may happen in Test context
      if (activityStorage == null) {
        activityStorage = this;
      }
    }
    return activityStorage;
  }

  @Override
  public List<ExoSocialActivity> getActivities(List<String> activityIdList) {
    if (activityIdList == null || activityIdList.isEmpty()) {
      return Collections.emptyList();
    }
    List<Long> activityIds = new ArrayList<>();
    for (String activityId : activityIdList) {
      if (activityId == null || activityId.isEmpty()) {
        continue;
      }

      if (activityId.startsWith(COMMENT_PREFIX)) {
        activityIds.add(getCommentID(activityId));
      } else {
        activityIds.add(Long.valueOf(activityId));
      }
    }
    List<ActivityEntity> activityEntities = activityDAO.findActivities(activityIds);
    if (activityEntities == null || activityEntities.isEmpty()) {
      return Collections.emptyList();
    }
    List<ExoSocialActivity> activityDTOs = new ArrayList<>();
    for (ActivityEntity activityEntity : activityEntities) {
      activityDTOs.add(convertActivityEntityToActivity(activityEntity));
    }
    return activityDTOs;
  }

  private void processDates(ActivityEntity activityEntity) {
    Date newPosted = new Date();
    // post time should never change once added
    if (activityEntity.getPosted() == null) {
      activityEntity.setPosted(newPosted);
    }
    // update time have to be same as post time when activity not updated
    activityEntity.setUpdatedDate(newPosted);
  }

  private ActivityShareActionEntity toEntity(ActivityShareAction activityShareAction) {
    ActivityShareActionEntity actionEntity = new ActivityShareActionEntity();
    actionEntity.setActivityId(activityShareAction.getActivityId());
    actionEntity.setUserId(activityShareAction.getUserIdentityId());
    actionEntity.setShareDate(new Date(activityShareAction.getShareDate()));
    actionEntity.setSharedActivityIds(activityShareAction.getSharedActivityIds());
    actionEntity.setSharedSpaceIds(activityShareAction.getSpaceIds());
    actionEntity.setTitle(activityShareAction.getMessage());
    actionEntity.setId(activityShareAction.getId());
    return actionEntity;
  }

  private ActivityShareAction fromEntity(ActivityShareActionEntity actionEntity) {
    ActivityShareAction activityShareAction = new ActivityShareAction();
    activityShareAction.setActivityId(actionEntity.getActivityId());
    activityShareAction.setUserIdentityId(actionEntity.getUserId());
    activityShareAction.setShareDate(actionEntity.getShareDate().getTime());
    activityShareAction.setSharedActivityIds(actionEntity.getSharedActivityIds());
    activityShareAction.setSpaceIds(actionEntity.getSharedSpaceIds());
    activityShareAction.setMessage(actionEntity.getTitle());
    activityShareAction.setId(actionEntity.getId());
    return activityShareAction;
  }

}
