/*
 * Copyright (C) 2003-2011 eXo Platform SAS.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero 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 Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

package org.exoplatform.social.core.storage.cache;

import java.util.*;
import java.util.Map.Entry;

import org.apache.commons.lang.StringUtils;

import org.exoplatform.commons.cache.future.FutureExoCache;
import org.exoplatform.services.cache.ExoCache;
import org.exoplatform.services.log.ExoLogger;
import org.exoplatform.services.log.Log;
import org.exoplatform.social.core.identity.model.Identity;
import org.exoplatform.social.core.identity.provider.OrganizationIdentityProvider;
import org.exoplatform.social.core.jpa.storage.RDBMSRelationshipStorageImpl;
import org.exoplatform.social.core.profile.ProfileFilter;
import org.exoplatform.social.core.relationship.model.Relationship;
import org.exoplatform.social.core.storage.RelationshipStorageException;
import org.exoplatform.social.core.storage.api.IdentityStorage;
import org.exoplatform.social.core.storage.api.RelationshipStorage;
import org.exoplatform.social.core.storage.cache.loader.ServiceContext;
import org.exoplatform.social.core.storage.cache.model.data.*;
import org.exoplatform.social.core.storage.cache.model.key.*;
import org.exoplatform.social.core.storage.cache.selector.*;

/**
 * Cache support for RelationshipStorage.
 *
 * @author <a href="mailto:alain.defrance@exoplatform.com">Alain Defrance</a>
 * @version $Revision$
 */
public class CachedRelationshipStorage implements RelationshipStorage {

  /** Logger */
  private static final Log                                                                                   LOG                    =
                                                                                                                 ExoLogger.getLogger(CachedRelationshipStorage.class);

  //
  private final ExoCache<RelationshipKey, RelationshipData>                                                  exoRelationshipCache;

  private final ExoCache<RelationshipIdentityKey, RelationshipKey>                                           exoRelationshipByIdentityCache;

  private final ExoCache<RelationshipCountKey, IntegerData>                                                  exoRelationshipCountCache;

  private final ExoCache<ListRelationshipsKey, ListIdentitiesData>                                           exoRelationshipsCache;

  private final ExoCache<SuggestionKey, SuggestionsData>                                                     exoSuggestionCache;

  private final ExoCache<ActivityCountKey, IntegerData>                                                      exoActivitiesCountCache;

  private final ExoCache<ListActivitiesKey, ListActivitiesData>                                              exoActivitiesCache;

  //
  private final FutureExoCache<RelationshipKey, RelationshipData, ServiceContext<RelationshipData>>          relationshipCache;

  private final FutureExoCache<RelationshipIdentityKey, RelationshipKey, ServiceContext<RelationshipKey>>    relationshipCacheIdentity;

  private final FutureExoCache<RelationshipCountKey, IntegerData, ServiceContext<IntegerData>>               relationshipsCount;

  private final FutureExoCache<ListRelationshipsKey, ListIdentitiesData, ServiceContext<ListIdentitiesData>> relationshipsCache;

  private final FutureExoCache<SuggestionKey, SuggestionsData, ServiceContext<SuggestionsData>>              suggestionCache;

  //
  private final ExoCache<IdentityKey, IdentityData>                                                          exoIdentityCache;

  //
  private final RelationshipStorage                                                                          storage;

  private final IdentityStorage                                                                              identityStorage;

  //
  private static RelationshipKey                                                                             RELATIONSHIP_NOT_FOUND =
                                                                                                                                    new RelationshipKey(null);

  void clearCacheFor(Relationship r) {

    List<String> identities = new ArrayList<String>();

    if (r.getSender() != null) {
      identities.add(r.getSender().getId());
    }
    if (r.getReceiver() != null) {
      identities.add(r.getReceiver().getId());
    }

    try {
      exoRelationshipsCache.select(new RelationshipCacheSelector(identities.toArray(new String[] {})));
      exoRelationshipCountCache.select(new RelationshipCacheSelector(identities.toArray(new String[] {})));
      exoSuggestionCache.select(new SuggestionCacheSelector(identities.toArray(new String[] {})));
      exoActivitiesCache.select(new CacheSelector<ListActivitiesKey, ListActivitiesData>());
      exoActivitiesCountCache.select(new CacheSelector<ActivityCountKey, IntegerData>());
    } catch (Exception e) {
      LOG.error(e);
    }

  }

  /**
   * When enable/disable an user, we need to clear all cache associated to
   * suggestion and relationship
   */
  public void clearAllRelationshipCache() {
    try {
      exoRelationshipsCache.clearCache();
      exoRelationshipCountCache.clearCache();
      exoSuggestionCache.clearCache();
    } catch (Exception e) {
      LOG.error(e);
    }
  }

  /**
   * Build the identity list from the caches Ids.
   *
   * @param data ids
   * @return identities
   */
  private List<Identity> buildRelationships(ListIdentitiesData data) {

    List<Identity> identities = new ArrayList<Identity>();
    for (IdentityKey k : data.getIds()) {
      Identity gotIdentity = identityStorage.findIdentityById(k.getId());
      identities.add(gotIdentity);
    }
    return identities;

  }

  /**
   * Build the ids from the identitiy list.
   *
   * @param identities identities
   * @return ids
   */
  private ListIdentitiesData buildIds(List<Identity> identities) {

    List<IdentityKey> data = new ArrayList<IdentityKey>();
    for (Identity i : identities) {
      IdentityKey k = new IdentityKey(i);
      if (exoIdentityCache.get(k) == null) {
        // data from db no need to replicate on other cluster nodes
        exoIdentityCache.putLocal(k, new IdentityData(i));
      }
      data.add(new IdentityKey(i));
    }
    return new ListIdentitiesData(data);

  }

  /**
   * Build the suggestions from the identity map.
   *
   * @param map map of identity
   */
  private SuggestionsData buildIdMap(Map<Identity, Integer> map) {

    Map<String, Integer> data = new LinkedHashMap<String, Integer>();
    for (Entry<Identity, Integer> item : map.entrySet()) {
      data.put(item.getKey().getId(), item.getValue());
    }

    return new SuggestionsData(data);
  }

  /**
   * Build the suggestions map from the caches Ids.
   *
   * @param data map of identities
   * @return suggestions
   */
  private Map<Identity, Integer> buildSuggestions(SuggestionsData data) {
    Map<Identity, Integer> suggestions = new LinkedHashMap<Identity, Integer>();
    for (Entry<String, Integer> item : data.getMap().entrySet()) {
      Identity gotIdentity = identityStorage.findIdentityById(item.getKey());
      suggestions.put(gotIdentity, item.getValue());
    }
    return suggestions;

  }

  /**
   * Remove identity cache.
   *
   * @param remoteId
   */
  private void removeIdentityCache(String remoteId) {
    if (identityStorage instanceof CachedIdentityStorage) {
      ((CachedIdentityStorage) identityStorage).clearIdentityCache(OrganizationIdentityProvider.NAME, remoteId, false);
    }
  }

  public CachedRelationshipStorage(final RDBMSRelationshipStorageImpl storage,
                                   final IdentityStorage identityStorage,
                                   final SocialStorageCacheService cacheService) {

    //
    this.storage = storage;
    this.identityStorage = identityStorage;

    //
    this.exoRelationshipCache = cacheService.getRelationshipCache();
    this.exoRelationshipByIdentityCache = cacheService.getRelationshipCacheByIdentity();
    this.exoRelationshipCountCache = cacheService.getRelationshipsCount();
    this.exoRelationshipsCache = cacheService.getRelationshipsCache();
    this.exoSuggestionCache = cacheService.getSuggestionCache();
    this.exoActivitiesCountCache = cacheService.getActivitiesCountCache();
    this.exoActivitiesCache = cacheService.getActivitiesCache();

    //
    this.relationshipCache = CacheType.RELATIONSHIP.createFutureCache(exoRelationshipCache);
    this.relationshipCacheIdentity = CacheType.RELATIONSHIP_FROM_IDENTITY.createFutureCache(exoRelationshipByIdentityCache);
    this.relationshipsCount = CacheType.RELATIONSHIPS_COUNT.createFutureCache(exoRelationshipCountCache);
    this.relationshipsCache = CacheType.RELATIONSHIPS.createFutureCache(exoRelationshipsCache);
    this.suggestionCache = CacheType.SUGGESTIONS.createFutureCache(exoSuggestionCache);

    //
    this.exoIdentityCache = cacheService.getIdentityCache();

  }

  /**
   * {@inheritDoc}
   */
  public Relationship saveRelationship(final Relationship relationship) throws RelationshipStorageException {

    Relationship r = storage.saveRelationship(relationship);

    RelationshipIdentityKey identityKey1 = new RelationshipIdentityKey(r.getSender().getId(), r.getReceiver().getId());
    RelationshipIdentityKey identityKey2 = new RelationshipIdentityKey(r.getReceiver().getId(), r.getSender().getId());
    RelationshipKey key = new RelationshipKey(relationship.getId());

    exoRelationshipCache.put(key, new RelationshipData(r));
    exoRelationshipByIdentityCache.put(identityKey1, key);
    exoRelationshipByIdentityCache.put(identityKey2, key);

    removeIdentityCache(r.getSender().getRemoteId());
    removeIdentityCache(r.getReceiver().getRemoteId());

    clearCacheFor(relationship);

    return r;

  }

  /**
   * {@inheritDoc}
   */
  public void removeRelationship(final Relationship relationship) throws RelationshipStorageException {

    storage.removeRelationship(relationship);

    //
    exoRelationshipCache.remove(new RelationshipKey(relationship.getId()));

    // clear caching.
    if (relationship.getSender() != null && relationship.getReceiver() != null) {
      RelationshipIdentityKey identityKey = new RelationshipIdentityKey(relationship.getSender().getId(),
                                                                        relationship.getReceiver().getId());
      relationshipCacheIdentity.remove(identityKey);

      identityKey = new RelationshipIdentityKey(relationship.getReceiver().getId(), relationship.getSender().getId());
      relationshipCacheIdentity.remove(identityKey);

      removeIdentityCache(relationship.getSender().getRemoteId());
      removeIdentityCache(relationship.getReceiver().getRemoteId());
    }

    //
    clearCacheFor(relationship);

  }

  /**
   * {@inheritDoc}
   */
  public Relationship getRelationship(final String uuid) throws RelationshipStorageException {

    //
    RelationshipKey key = new RelationshipKey(uuid);

    //
    RelationshipData data = relationshipCache.get(
                                                  new ServiceContext<RelationshipData>() {
                                                    public RelationshipData execute() {
                                                      Relationship got = storage.getRelationship(uuid);
                                                      if (got != null) {
                                                        return new RelationshipData(storage.getRelationship(uuid));
                                                      }
                                                      return null;
                                                    }
                                                  },
                                                  key);

    //
    if (data != null) {
      return data.build();
    }
    return null;

  }

  /**
   * {@inheritDoc}
   */
  public List<Relationship> getSenderRelationships(
                                                   final Identity sender,
                                                   final Relationship.Type type,
                                                   final List<Identity> listCheckIdentity)
                                                                                           throws RelationshipStorageException {
    return storage.getSenderRelationships(sender, type, listCheckIdentity);
  }

  /**
   * {@inheritDoc}
   */
  public List<Relationship> getSenderRelationships(
                                                   final String senderId,
                                                   final Relationship.Type type,
                                                   final List<Identity> listCheckIdentity)
                                                                                           throws RelationshipStorageException {
    return storage.getSenderRelationships(senderId, type, listCheckIdentity);
  }

  /**
   * {@inheritDoc}
   */
  public List<Relationship> getReceiverRelationships(
                                                     final Identity receiver,
                                                     final Relationship.Type type,
                                                     final List<Identity> listCheckIdentity)
                                                                                             throws RelationshipStorageException {
    return storage.getReceiverRelationships(receiver, type, listCheckIdentity);
  }

  /**
   * {@inheritDoc}
   */
  public Relationship getRelationship(Identity identity1, Identity identity2)
                                                                              throws RelationshipStorageException {

    // We make sure to check the Relationship in the same order to improve
    // efficiency of the cache
    final Identity idFirst, idLast;
    if (identity1 == null || identity2 == null || StringUtils.isBlank(identity1.getId()) || StringUtils.isBlank(identity2.getId())
        || identity1.getId().trim().equals(identity2.getId().trim())) {
      return null;
    } else if (identity1.getId().compareTo(identity2.getId()) > 0) {
      idFirst = identity1;
      idLast = identity2;
    } else {
      idFirst = identity2;
      idLast = identity1;
    }

    final RelationshipIdentityKey key = new RelationshipIdentityKey(idFirst.getId(), idLast.getId());

    //
    RelationshipKey gotKey = relationshipCacheIdentity.get(
                                                           new ServiceContext<RelationshipKey>() {
                                                             public RelationshipKey execute() {
                                                               Relationship got = storage.getRelationship(idFirst, idLast);
                                                               if (got != null) {
                                                                 RelationshipKey k = new RelationshipKey(got.getId());
                                                                 return k;
                                                               } else {
                                                                 return RELATIONSHIP_NOT_FOUND;
                                                               }
                                                             }
                                                           },
                                                           key);

    //
    if (gotKey != null && !gotKey.equals(RELATIONSHIP_NOT_FOUND)) {
      return getRelationship(gotKey.getId());
    } else {
      return null;
    }

  }

  @Override
  public boolean hasRelationship(Identity identity1,
                                 Identity identity2,
                                 String relationshipPath) throws RelationshipStorageException {
    RelationshipIdentityKey key = new RelationshipIdentityKey(identity2.getId(), identity1.getId());
    RelationshipKey gotKey = exoRelationshipByIdentityCache.get(key);
    if (gotKey != null && !gotKey.equals(RELATIONSHIP_NOT_FOUND)
        && getRelationship(identity1, identity2).getStatus().equals(Relationship.Type.CONFIRMED)) {
      return true;
    }

    key = new RelationshipIdentityKey(identity1.getId(), identity2.getId());
    gotKey = exoRelationshipByIdentityCache.get(key);
    if (gotKey != null && !gotKey.equals(RELATIONSHIP_NOT_FOUND)
        && getRelationship(identity1, identity2).getStatus().equals(Relationship.Type.CONFIRMED)) {
      return true;
    }

    return storage.hasRelationship(identity1, identity2, relationshipPath);
  }

  /**
   * {@inheritDoc}
   */
  public List<Relationship> getRelationships(
                                             final Identity identity,
                                             final Relationship.Type type,
                                             final List<Identity> listCheckIdentity)
                                                                                     throws RelationshipStorageException {
    return storage.getRelationships(identity, type, listCheckIdentity);
  }

  /**
   * {@inheritDoc}
   */
  public List<Identity> getRelationships(final Identity identity, final long offset, final long limit)
                                                                                                       throws RelationshipStorageException {

    //
    IdentityKey key = new IdentityKey(identity);
    ListRelationshipsKey<IdentityKey> listKey =
                                              new ListRelationshipsKey<IdentityKey>(key,
                                                                                    RelationshipType.RELATIONSHIP,
                                                                                    offset,
                                                                                    limit);
    ListIdentitiesData keys = relationshipsCache.get(
                                                     new ServiceContext<ListIdentitiesData>() {
                                                       public ListIdentitiesData execute() {
                                                         List<Identity> got = storage.getRelationships(identity, offset, limit);
                                                         return buildIds(got);
                                                       }
                                                     },
                                                     listKey);

    //
    return buildRelationships(keys);

  }

  /**
   * {@inheritDoc}
   */
  public List<Identity> getIncomingRelationships(final Identity receiver, final long offset, final long limit)
                                                                                                               throws RelationshipStorageException {

    //
    IdentityKey key = new IdentityKey(receiver);
    ListRelationshipsKey<IdentityKey> listKey =
                                              new ListRelationshipsKey<IdentityKey>(key,
                                                                                    RelationshipType.INCOMMING,
                                                                                    offset,
                                                                                    limit);
    ListIdentitiesData keys = relationshipsCache.get(
                                                     new ServiceContext<ListIdentitiesData>() {
                                                       public ListIdentitiesData execute() {
                                                         List<Identity> got = storage.getIncomingRelationships(receiver,
                                                                                                               offset,
                                                                                                               limit);
                                                         return buildIds(got);
                                                       }
                                                     },
                                                     listKey);

    //
    return buildRelationships(keys);

  }

  /**
   * {@inheritDoc}
   */
  public int getIncomingRelationshipsCount(final Identity receiver) throws RelationshipStorageException {

    //
    IdentityKey iKey = new IdentityKey(receiver);
    RelationshipCountKey<IdentityKey> key = new RelationshipCountKey<IdentityKey>(iKey, RelationshipType.INCOMMING);

    //
    return relationshipsCount.get(
                                  new ServiceContext<IntegerData>() {
                                    public IntegerData execute() {
                                      return new IntegerData(storage.getIncomingRelationshipsCount(receiver));
                                    }
                                  },
                                  key)
                             .build();

  }

  /**
   * {@inheritDoc}
   */
  public List<Identity> getOutgoingRelationships(final Identity sender, final long offset, final long limit)
                                                                                                             throws RelationshipStorageException {

    //
    IdentityKey key = new IdentityKey(sender);
    ListRelationshipsKey<IdentityKey> listKey =
                                              new ListRelationshipsKey<IdentityKey>(key,
                                                                                    RelationshipType.OUTGOING,
                                                                                    offset,
                                                                                    limit);
    ListIdentitiesData keys = relationshipsCache.get(
                                                     new ServiceContext<ListIdentitiesData>() {
                                                       public ListIdentitiesData execute() {
                                                         List<Identity> got = storage.getOutgoingRelationships(sender,
                                                                                                               offset,
                                                                                                               limit);
                                                         return buildIds(got);
                                                       }
                                                     },
                                                     listKey);

    //
    return buildRelationships(keys);

  }

  /**
   * {@inheritDoc}
   */
  public int getOutgoingRelationshipsCount(final Identity sender) throws RelationshipStorageException {

    //
    IdentityKey iKey = new IdentityKey(sender);
    RelationshipCountKey<IdentityKey> key = new RelationshipCountKey<IdentityKey>(iKey, RelationshipType.OUTGOING);

    //
    return relationshipsCount.get(
                                  new ServiceContext<IntegerData>() {
                                    public IntegerData execute() {
                                      return new IntegerData(storage.getOutgoingRelationshipsCount(sender));
                                    }
                                  },
                                  key)
                             .build();

  }

  /**
   * {@inheritDoc}
   */
  public int getRelationshipsCount(final Identity identity) throws RelationshipStorageException {

    //
    IdentityKey iKey = new IdentityKey(identity);
    RelationshipCountKey<IdentityKey> key = new RelationshipCountKey<IdentityKey>(iKey, RelationshipType.RELATIONSHIP);

    //
    return relationshipsCount.get(
                                  new ServiceContext<IntegerData>() {
                                    public IntegerData execute() {
                                      return new IntegerData(storage.getRelationshipsCount(identity));
                                    }
                                  },
                                  key)
                             .build();

  }

  /**
   * {@inheritDoc}
   */
  public List<Identity> getConnections(final Identity identity, final long offset, final long limit)
                                                                                                     throws RelationshipStorageException {

    //
    IdentityKey key = new IdentityKey(identity);
    ListRelationshipsKey<IdentityKey> listKey = new ListRelationshipsKey<IdentityKey>(key,
                                                                                      RelationshipType.CONNECTION,
                                                                                      offset,
                                                                                      limit);
    ListIdentitiesData keys = relationshipsCache.get(
                                                     new ServiceContext<ListIdentitiesData>() {
                                                       public ListIdentitiesData execute() {
                                                         List<Identity> got = storage.getConnections(identity, offset, limit);
                                                         return buildIds(got);
                                                       }
                                                     },
                                                     listKey);

    //
    return buildRelationships(keys);

  }

  /**
   * {@inheritDoc}
   */
  public List<Identity> getConnections(final Identity identity) throws RelationshipStorageException {
    return storage.getConnections(identity);
  }

  /**
   * {@inheritDoc}
   */
  public int getConnectionsCount(final Identity identity) throws RelationshipStorageException {

    //
    IdentityKey iKey = new IdentityKey(identity);
    RelationshipCountKey<IdentityKey> key = new RelationshipCountKey<IdentityKey>(iKey, RelationshipType.CONNECTION);

    //
    return relationshipsCount.get(
                                  new ServiceContext<IntegerData>() {
                                    public IntegerData execute() {
                                      return new IntegerData(storage.getConnectionsCount(identity));
                                    }
                                  },
                                  key)
                             .build();

  }

  public List<Identity> getConnectionsByFilter(final Identity existingIdentity,
                                               final ProfileFilter profileFilter,
                                               final long offset,
                                               final long limit) throws RelationshipStorageException {

    //
    IdentityFilterKey key = new IdentityFilterKey(existingIdentity.getProviderId(),
                                                  existingIdentity.getRemoteId(),
                                                  profileFilter);
    ListRelationshipsKey<IdentityFilterKey> listKey =
                                                    new ListRelationshipsKey<IdentityFilterKey>(key,
                                                                                                RelationshipType.CONNECTION_WITH_FILTER,
                                                                                                offset,
                                                                                                limit);

    //
    ListIdentitiesData keys = relationshipsCache.get(
                                                     new ServiceContext<ListIdentitiesData>() {
                                                       public ListIdentitiesData execute() {
                                                         List<Identity> got = storage.getConnectionsByFilter(existingIdentity,
                                                                                                             profileFilter,
                                                                                                             offset,
                                                                                                             limit);
                                                         return buildIds(got);
                                                       }
                                                     },
                                                     listKey);

    //
    return buildRelationships(keys);

  }

  public List<Identity> getIncomingByFilter(final Identity existingIdentity,
                                            final ProfileFilter profileFilter,
                                            final long offset,
                                            final long limit) throws RelationshipStorageException {

    //
    IdentityFilterKey key =
                          new IdentityFilterKey(existingIdentity.getProviderId(), existingIdentity.getRemoteId(), profileFilter);
    ListRelationshipsKey<IdentityFilterKey> listKey =
                                                    new ListRelationshipsKey<IdentityFilterKey>(key,
                                                                                                RelationshipType.INCOMMING_WITH_FILTER,
                                                                                                offset,
                                                                                                limit);

    //
    ListIdentitiesData keys = relationshipsCache.get(
                                                     new ServiceContext<ListIdentitiesData>() {
                                                       public ListIdentitiesData execute() {
                                                         List<Identity> got = storage.getIncomingByFilter(existingIdentity,
                                                                                                          profileFilter,
                                                                                                          offset,
                                                                                                          limit);
                                                         return buildIds(got);
                                                       }
                                                     },
                                                     listKey);

    //
    return buildRelationships(keys);

  }

  public List<Identity> getOutgoingByFilter(final Identity existingIdentity,
                                            final ProfileFilter profileFilter,
                                            final long offset,
                                            final long limit) throws RelationshipStorageException {

    //
    IdentityFilterKey key =
                          new IdentityFilterKey(existingIdentity.getProviderId(), existingIdentity.getRemoteId(), profileFilter);
    ListRelationshipsKey<IdentityFilterKey> listKey =
                                                    new ListRelationshipsKey<IdentityFilterKey>(key,
                                                                                                RelationshipType.OUTGOING_WITH_FILTER,
                                                                                                offset,
                                                                                                limit);

    //
    ListIdentitiesData keys = relationshipsCache.get(
                                                     new ServiceContext<ListIdentitiesData>() {
                                                       public ListIdentitiesData execute() {
                                                         List<Identity> got = storage.getOutgoingByFilter(existingIdentity,
                                                                                                          profileFilter,
                                                                                                          offset,
                                                                                                          limit);
                                                         return buildIds(got);
                                                       }
                                                     },
                                                     listKey);

    //
    return buildRelationships(keys);

  }

  public int getConnectionsCountByFilter(
                                         final Identity existingIdentity,
                                         final ProfileFilter profileFilter) throws RelationshipStorageException {

    //
    IdentityFilterKey iKey = new IdentityFilterKey(existingIdentity.getProviderId(),
                                                   existingIdentity.getRemoteId(),
                                                   profileFilter);
    RelationshipCountKey<IdentityFilterKey> key =
                                                new RelationshipCountKey<IdentityFilterKey>(iKey,
                                                                                            RelationshipType.CONNECTION_WITH_FILTER);

    //
    return relationshipsCount.get(
                                  new ServiceContext<IntegerData>() {
                                    public IntegerData execute() {
                                      return new IntegerData(storage.getConnectionsCountByFilter(existingIdentity,
                                                                                                 profileFilter));
                                    }
                                  },
                                  key)
                             .build();

  }

  public int getIncomingCountByFilter(
                                      final Identity existingIdentity,
                                      final ProfileFilter profileFilter) throws RelationshipStorageException {

    //
    IdentityFilterKey iKey =
                           new IdentityFilterKey(existingIdentity.getProviderId(), existingIdentity.getRemoteId(), profileFilter);
    RelationshipCountKey<IdentityFilterKey> key =
                                                new RelationshipCountKey<IdentityFilterKey>(iKey,
                                                                                            RelationshipType.INCOMMING_WITH_FILTER);

    //
    return relationshipsCount.get(
                                  new ServiceContext<IntegerData>() {
                                    public IntegerData execute() {
                                      return new IntegerData(storage.getIncomingCountByFilter(existingIdentity, profileFilter));
                                    }
                                  },
                                  key)
                             .build();

  }

  public int getOutgoingCountByFilter(
                                      final Identity existingIdentity,
                                      final ProfileFilter profileFilter) throws RelationshipStorageException {

    //
    IdentityFilterKey iKey =
                           new IdentityFilterKey(existingIdentity.getProviderId(), existingIdentity.getRemoteId(), profileFilter);
    RelationshipCountKey<IdentityFilterKey> key =
                                                new RelationshipCountKey<IdentityFilterKey>(iKey,
                                                                                            RelationshipType.OUTGOING_WITH_FILTER);

    //
    return relationshipsCount.get(
                                  new ServiceContext<IntegerData>() {
                                    public IntegerData execute() {
                                      return new IntegerData(storage.getOutgoingCountByFilter(existingIdentity, profileFilter));
                                    }
                                  },
                                  key)
                             .build();

  }

  public Map<Identity, Integer> getSuggestions(final Identity identity,
                                               final int maxConnections,
                                               final int maxConnectionsToLoad,
                                               final int maxSuggestions)
                                                                         throws RelationshipStorageException {
    //
    IdentityKey key = new IdentityKey(identity);
    SuggestionKey<IdentityKey> suggestKey = new SuggestionKey<IdentityKey>(key,
                                                                           maxConnections,
                                                                           maxConnectionsToLoad,
                                                                           maxSuggestions);

    SuggestionsData keys = suggestionCache.get(
                                               new ServiceContext<SuggestionsData>() {
                                                 public SuggestionsData execute() {
                                                   Map<Identity, Integer> got = storage.getSuggestions(identity,
                                                                                                       maxConnections,
                                                                                                       maxConnectionsToLoad,
                                                                                                       maxSuggestions);
                                                   return buildIdMap(got);
                                                 }
                                               },
                                               suggestKey);

    //
    return buildSuggestions(keys);
  }

  @Override
  public List<Identity> getLastConnections(final Identity identity, final int limit) throws RelationshipStorageException {
    //
    IdentityKey key = new IdentityKey(identity);
    ListRelationshipsKey<IdentityKey> listKey =
                                              new ListRelationshipsKey<IdentityKey>(key,
                                                                                    RelationshipType.LAST_CONNECTIONS,
                                                                                    0,
                                                                                    limit);
    ListIdentitiesData keys = relationshipsCache.get(
                                                     new ServiceContext<ListIdentitiesData>() {
                                                       public ListIdentitiesData execute() {
                                                         List<Identity> got = storage.getLastConnections(identity, limit);
                                                         return buildIds(got);
                                                       }
                                                     },
                                                     listKey);

    //
    return buildRelationships(keys);
  }

  public List<Relationship> getRelationshipsByStatus(Identity identity, Relationship.Type type, long offset, long limit) {
    return storage.getRelationshipsByStatus(identity, type, offset, limit);
  }

  public int getRelationshipsCountByStatus(Identity identity, Relationship.Type type) {
    return storage.getRelationshipsCountByStatus(identity, type);
  }

  @Override
  public int getConnectionsInCommonCount(Identity identity1, Identity identity2) {
    // We make sure to check the Relationship in the same order to improve
    // efficiency of the cache
    final Identity idFirst;
    final Identity idLast;
    if (identity1 == null || identity2 == null || StringUtils.isBlank(identity1.getId()) || StringUtils.isBlank(identity2.getId())
        || identity1.getId().trim().equals(identity2.getId().trim())) {
      return 0;
    } else if (identity1.getId().compareTo(identity2.getId()) > 0) {
      idFirst = identity1;
      idLast = identity2;
    } else {
      idFirst = identity2;
      idLast = identity1;
    }

    RelationshipIdentityKey identitiesKey = new RelationshipIdentityKey(idFirst.getId(), idLast.getId());
    RelationshipCountKey<RelationshipIdentityKey> key = new RelationshipCountKey<>(identitiesKey, RelationshipType.CONNECTION);

    return relationshipsCount.get(new ServiceContext<IntegerData>() {
      public IntegerData execute() {
        return new IntegerData(storage.getConnectionsInCommonCount(idFirst, idLast));
      }
    }, key).build();
  }
}
