Merge pull request eclipse-ditto#1872 from eclipse-ditto/bugfix/1869-resolving-imported-policies-in-ditto-search

eclipse-ditto#1869 use cache in order to load imported policies when policies are …

Author: thjaeckle

internal/utils/cache/src/main/java/org/eclipse/ditto/internal/utils/cache/Cache.java

@@ -17,6 +17,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Function;
+import java.util.function.Predicate;
 
 /**
  * A general purpose cache for items which are associated with a key.
@@ -75,6 +76,16 @@ public interface Cache<K, V> {
      */
     boolean invalidate(K key);
 
+    /**
+     * Invalidates the passed key from the cache if present and the passed {@code valueCondition} evaluates to
+     * {@code true}.
+     *
+     * @param key the key to invalidate.
+     * @param valueCondition the condition which has to pass in order to invalidate the key from the cache.
+     * @return {@code true} if the entry was cached and is now invalidated, {@code false} otherwise.
+     */
+    boolean invalidateConditionally(K key, Predicate<V> valueCondition);
+
     /**
      * Associates the {@code value} with the {@code key} in this cache.
      * <p>

internal/utils/cache/src/main/java/org/eclipse/ditto/internal/utils/cache/CaffeineCache.java

@@ -22,6 +22,7 @@
 import java.util.concurrent.Executor;
 import java.util.function.BiFunction;
 import java.util.function.Function;
+import java.util.function.Predicate;
 
 import javax.annotation.Nullable;
 
@@ -230,6 +231,25 @@ public boolean invalidate(final K key) {
         return currentlyExisting;
     }
 
+    @Override
+    public boolean invalidateConditionally(final K key, final Predicate<V> valueCondition) {
+        requireNonNull(key);
+
+        V value = synchronousCacheView.getIfPresent(key);
+        if (value != null) {
+            synchronized (value) {
+                value = synchronousCacheView.getIfPresent(key);
+                if (value != null && valueCondition.test(value)) {
+                    return invalidate(key);
+                } else {
+                    return false;
+                }
+            }
+        } else {
+            return false;
+        }
+    }
+
     // optimized batch invalidation method for caffeine
     @Override
     public void invalidateAll(final Collection<K> keys) {

internal/utils/cache/src/main/java/org/eclipse/ditto/internal/utils/cache/ProjectedCache.java

@@ -17,6 +17,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Function;
+import java.util.function.Predicate;
 
 /**
  * A cache working on an embedded passed {@code cache} with values of type {@code <V>}, representing a projected cache
@@ -68,6 +69,11 @@ public boolean invalidate(final K key) {
         return cache.invalidate(key);
     }
 
+    @Override
+    public boolean invalidateConditionally(final K key, final Predicate<U> valueCondition) {
+        return cache.invalidateConditionally(key, value -> valueCondition.test(project.apply(value)));
+    }
+
     @Override
     public void put(final K key, final U value) {
         cache.put(key, embed.apply(value));

policies/enforcement/src/main/java/org/eclipse/ditto/policies/enforcement/PolicyEnforcerCache.java

@@ -21,6 +21,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Function;
+import java.util.function.Predicate;
 
 import org.eclipse.ditto.internal.utils.cache.Cache;
 import org.eclipse.ditto.internal.utils.cache.CacheFactory;
@@ -102,6 +103,23 @@ public boolean invalidate(final PolicyId policyId) {
         return directlyCached || indirectlyCachedViaImport;
     }
 
+    @Override
+    public boolean invalidateConditionally(final PolicyId policyId,
+            final Predicate<Entry<PolicyEnforcer>> valueCondition) {
+        // Invalidate the changed policy
+        final boolean directlyCached = delegate.invalidateConditionally(policyId, valueCondition);
+
+        // Invalidate all policies that import the changed policy
+        final boolean indirectlyCachedViaImport = Optional.ofNullable(policyIdToImportingMap.remove(policyId))
+                .stream()
+                .flatMap(Collection::stream)
+                .map(p -> delegate.invalidateConditionally(p, valueCondition))
+                .reduce((previous, next) -> previous || next)
+                .orElse(false);
+
+        return directlyCached || indirectlyCachedViaImport;
+    }
+
     @Override
     public void put(final PolicyId key, final Entry<PolicyEnforcer> value) {
         delegate.put(key, value);

policies/model/src/main/java/org/eclipse/ditto/policies/model/PolicyImport.java

@@ -14,6 +14,8 @@
 
 import java.util.Optional;
 
+import javax.annotation.Nullable;
+
 import org.eclipse.ditto.base.model.json.FieldType;
 import org.eclipse.ditto.base.model.json.JsonSchemaVersion;
 import org.eclipse.ditto.base.model.json.Jsonifiable;
@@ -35,9 +37,9 @@ public interface PolicyImport extends Jsonifiable.WithFieldSelectorAndPredicate<
      * @param importedPolicyId the {@link PolicyId} of the imported policy.
      * @param effectedImports the EffectedImports of the new PolicyImport to create.
      * @return the new {@code PolicyImport}.
-     * @throws NullPointerException if any argument is {@code null}.
+     * @throws NullPointerException if {@code importedPolicyId} is {@code null}.
      */
-    static PolicyImport newInstance(final PolicyId importedPolicyId, final EffectedImports effectedImports) {
+    static PolicyImport newInstance(final PolicyId importedPolicyId, @Nullable final EffectedImports effectedImports) {
         return PoliciesModelFactory.newPolicyImport(importedPolicyId, effectedImports);
     }
 

thingsearch/service/src/main/java/org/eclipse/ditto/thingsearch/service/persistence/read/MongoThingsSearchPersistence.java

@@ -27,6 +27,13 @@
 
 import javax.annotation.Nullable;
 
+import org.apache.pekko.NotUsed;
+import org.apache.pekko.actor.ActorSystem;
+import org.apache.pekko.event.Logging;
+import org.apache.pekko.event.LoggingAdapter;
+import org.apache.pekko.japi.pf.PFBuilder;
+import org.apache.pekko.stream.SystemMaterializer;
+import org.apache.pekko.stream.javadsl.Source;
 import org.bson.BsonDocument;
 import org.bson.Document;
 import org.bson.conversions.Bson;
@@ -71,13 +78,6 @@
 import com.mongodb.reactivestreams.client.MongoCollection;
 import com.mongodb.reactivestreams.client.MongoDatabase;
 
-import org.apache.pekko.NotUsed;
-import org.apache.pekko.actor.ActorSystem;
-import org.apache.pekko.event.Logging;
-import org.apache.pekko.event.LoggingAdapter;
-import org.apache.pekko.japi.pf.PFBuilder;
-import org.apache.pekko.stream.SystemMaterializer;
-import org.apache.pekko.stream.javadsl.Source;
 import scala.PartialFunction;
 
 /**
@@ -402,7 +402,7 @@ private static Metadata readAsMetadata(final Document document) {
                         .map(dittoBsonJson::serialize)
                         .map(PolicyTag::fromJson)
                         .collect(Collectors.toSet());
-        return Metadata.of(thingId, thingRevision, thingPolicyTag, referencedPolicies, modified, null);
+        return Metadata.of(thingId, thingRevision, thingPolicyTag, null, referencedPolicies, modified, null);
     }
 
     private static AbstractWriteModel documentToWriteModel(final Document document) {

thingsearch/service/src/main/java/org/eclipse/ditto/thingsearch/service/persistence/write/mapping/EnforcedThingMapper.java

@@ -24,7 +24,7 @@
 import static org.eclipse.ditto.thingsearch.service.persistence.PersistenceConstants.FIELD_REVISION;
 import static org.eclipse.ditto.thingsearch.service.persistence.PersistenceConstants.FIELD_THING;
 
-import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
@@ -82,7 +82,7 @@ public static ThingWriteModel toWriteModel(final JsonObject thing,
         final long thingRevision = thing.getValueOrThrow(Thing.JsonFields.REVISION);
         final var optionalPolicyId = thing.getValue(Thing.JsonFields.POLICY_ID).map(PolicyId::of);
 
-        final HashSet<PolicyTag> allReferencedPolicies = new HashSet<>(referencedPolicies);
+        final Set<PolicyTag> allReferencedPolicies = new LinkedHashSet<>(referencedPolicies);
         final List<PolicyTag> policyTagsOfDeletedButStillImportedPolicies =
                 Optional.ofNullable(oldMetadata).map(Metadata::getAllReferencedPolicyTags).orElseGet(Set::of).stream()
                         .filter(oldReferencedPolicyTag -> policy.getPolicyImports()
@@ -99,7 +99,7 @@ public static ThingWriteModel toWriteModel(final JsonObject thing,
                 .orElse(null);
 
         final var metadata =
-                Metadata.of(thingId, thingRevision, thingPolicyTag, allReferencedPolicies,
+                Metadata.of(thingId, thingRevision, thingPolicyTag, null, allReferencedPolicies,
                         Optional.ofNullable(oldMetadata).flatMap(Metadata::getModified).orElse(null),
                         Optional.ofNullable(oldMetadata).map(Metadata::getEvents).orElse(List.of()),
                         Optional.ofNullable(oldMetadata).map(Metadata::getTimers).orElse(List.of()),