/*
 * Copyright (C) 2021 Ignite Realtime Foundation. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.jivesoftware.openfire.spi;

import org.jivesoftware.openfire.cluster.ClusteredCacheEntryListener;
import org.jivesoftware.openfire.cluster.NodeID;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;

/**
 * Cache entry listener implementation that maintains a reverse lookup map for the cache that is being observed.
 */
public class ReverseLookupUpdatingCacheEntryListener<K, V> implements ClusteredCacheEntryListener<K, V> {
    private final ConcurrentMap<NodeID, Set<K>> reverseCacheRepresentation;

    public ReverseLookupUpdatingCacheEntryListener(@Nonnull final ConcurrentMap<NodeID, Set<K>> reverseCacheRepresentation) {
        this.reverseCacheRepresentation = reverseCacheRepresentation;
    }

    @Override
    public void entryAdded(@Nonnull final K key, @Nullable final V value, @Nonnull final NodeID nodeID) {
        reverseCacheRepresentation.computeIfAbsent(nodeID, k -> new HashSet<>()).add(key);
    }

    @Override
    public void entryRemoved(@Nonnull final K key, @Nullable final V oldValue, @Nonnull final NodeID nodeID) {
        reverseCacheRepresentation.computeIfPresent(nodeID, (k, v) -> {
            v.remove(key);
            return v;
        });
    }

    @Override
    public void entryUpdated(@Nonnull final K key, @Nullable final V oldValue, @Nullable final V newValue, @Nonnull final NodeID nodeID) {
        // Although we only care about keys, we do need to process this for caches in which values are not uniquely 'owned' by a particular node, such as the #componentsCache.
        reverseCacheRepresentation.computeIfAbsent(nodeID, k -> new HashSet<>()).add(key);
    }

    @Override
    public void entryEvicted(@Nonnull final K key, @Nullable final V oldValue, @Nonnull final NodeID nodeID) {
        entryRemoved(key, oldValue, nodeID);
    }

    @Override
    public void mapCleared(@Nonnull final NodeID nodeID) {
        reverseCacheRepresentation.remove(nodeID);
    }

    @Override
    public void mapEvicted(@Nonnull final NodeID nodeID) {
        reverseCacheRepresentation.remove(nodeID);
    }
}

@@ -946,16 +946,52 @@
 rentMap;
+%0Aimport java.util.function.Function;
 %0A%0A/**%0A *
@@ -1288,16 +1288,85 @@
 ation;%0A%0A
+    private final Function%3CV, Set%3CNodeID%3E%3E nodeIDsFromValueDeducer;%0A%0A
     publ
@@ -1553,16 +1553,393 @@
 tation;%0A
+        this.nodeIDsFromValueDeducer = null;%0A    %7D%0A%0A    public ReverseLookupUpdatingCacheEntryListener(@Nonnull final ConcurrentMap%3CNodeID, Set%3CK%3E%3E reverseCacheRepresentation, @Nonnull final Function%3CV, Set%3CNodeID%3E%3E nodeIDsFromValueDeducer) %7B%0A        this.reverseCacheRepresentation = reverseCacheRepresentation;%0A        this.nodeIDsFromValueDeducer = nodeIDsFromValueDeducer;%0A
     %7D%0A%0A 
@@ -2696,16 +2696,27 @@
  'owned'
+%0A        //
  by a pa
@@ -2763,99 +2763,748 @@
 che.
-%0A        reverseCacheRepresentation.computeIfAbsent(nodeID, k -%3E new HashSet%3C%3E()).add(key);
+ Such utility must define a function to identify the current%0A        // nodes for which the entry is valid.%0A        if (nodeIDsFromValueDeducer != null) %7B%0A            // FIXME this implementation is untested, and likely incorrect. Also, there's an argument to be made that the%0A            //       implementation of the other methods should need to use this Function.%0A            final Set%3CNodeID%3E newNodes = nodeIDsFromValueDeducer.apply(newValue);%0A            if (newNodes.contains(nodeID)) %7B%0A                reverseCacheRepresentation.computeIfAbsent(nodeID, k -%3E new HashSet%3C%3E()).add(key);%0A            %7D else %7B%0A                reverseCacheRepresentation.computeIfAbsent(nodeID, k -%3E new HashSet%3C%3E()).remove(key);%0A            %7D%0A        %7D
 %0A   

Support caches for which entries are not directly relatable to exactly one cluster node.