fixed relation update event and added checks for entity type in relation for CFs processing

application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java

@@ -27,9 +27,11 @@ import org.thingsboard.server.actors.service.DefaultActorService;
 import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor;
 import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.ProfileEntityIdInfo;
 import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.asset.AssetProfile;
 import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
@@ -74,6 +76,7 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ScheduledFuture;
@@ -292,17 +295,33 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     }
 
     private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) {
+        // 2 = 1 for entity processing + 1 for relation processing
+        MultipleTbCallback multiCallback = new MultipleTbCallback(2, callback);
+        deleteEntityRelations(msg, multiCallback);
         switch (msg.getEntityId().getEntityType()) {
             case DEVICE, ASSET -> entityProfileCache.removeEntityId(msg.getEntityId());
             case CUSTOMER -> ownerEntities.remove(msg.getEntityId());
         }
         ownerEntities.values().forEach(entities -> entities.remove(msg.getEntityId()));
-        if (isMyPartition(msg.getEntityId(), callback)) {
+        if (isMyPartition(msg.getEntityId(), multiCallback)) {
             log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId());
-            getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback));
+            getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), multiCallback));
         }
     }
 
+    private void deleteEntityRelations(ComponentLifecycleMsg msg, TbCallback callback) {
+        List<EntityRelation> entityRelations = relationService.findEntityRelations(tenantId, msg.getEntityId());
+        if (entityRelations.isEmpty()) {
+            callback.onSuccess();
+        }
+        entityRelations.forEach(entityRelation -> {
+            Function<EntityId, TriConsumer<EntityId, CalculatedFieldCtx, TbCallback>> deleteAction =
+                    relatedId -> (entityId, ctx, cb) -> deleteRelatedEntity(entityId, relatedId, ctx, cb);
+
+            processRelationIfSupported(entityRelation, callback, deleteAction);
+        });
+    }
+
     private void onRelationChangedEvent(ComponentLifecycleMsg msg, TbCallback callback) {
         Function<EntityId, TriConsumer<EntityId, CalculatedFieldCtx, TbCallback>> relationAction = switch (msg.getEvent()) {
             case RELATION_UPDATED -> relatedId -> (entityId, ctx, cb) -> initRelatedEntity(entityId, relatedId, ctx, cb);
@@ -316,16 +335,26 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         }
 
         EntityRelation entityRelation = JacksonUtil.treeToValue(msg.getInfo(), EntityRelation.class);
+        processRelationIfSupported(entityRelation, callback, relationAction);
+    }
+
+    private void processRelationIfSupported(EntityRelation entityRelation,
+                                            TbCallback callback,
+                                            Function<EntityId, TriConsumer<EntityId, CalculatedFieldCtx, TbCallback>> relationAction) {
         EntityId toId = entityRelation.getTo();
         EntityId fromId = entityRelation.getFrom();
         String relationType = entityRelation.getType();
 
+        if (!(CalculatedField.isSupportedRefEntity(toId) || CalculatedField.isSupportedRefEntity(fromId))) {
+            callback.onSuccess();
+            return;
+        }
+
         MultipleTbCallback callbackForToAndFrom = new MultipleTbCallback(2, callback);
         processRelationByDirection(EntitySearchDirection.TO, relationType, toId, callbackForToAndFrom, relationAction.apply(fromId));
         processRelationByDirection(EntitySearchDirection.FROM, relationType, fromId, callbackForToAndFrom, relationAction.apply(toId));
     }
 
-
     private void processRelationByDirection(EntitySearchDirection direction,
                                             String relationType,
                                             EntityId mainId,
@@ -339,7 +368,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
 
         List<CalculatedFieldCtx> matchingCfs = cfsByEntityIdAndProfile.stream()
                 .filter(cf -> {
-                    if (cf.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config ) {
+                    if (cf.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config) {
                         RelationPathLevel relation = config.getRelation();
                         return direction.equals(relation.direction()) && relationType.equals(relation.relationType());
                     } else {
@@ -717,8 +746,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
 
     private EntityId getProfileId(TenantId tenantId, EntityId entityId) {
         return switch (entityId.getEntityType()) {
-            case ASSET -> assetProfileCache.get(tenantId, (AssetId) entityId).getId();
-            case DEVICE -> deviceProfileCache.get(tenantId, (DeviceId) entityId).getId();
+            case ASSET -> Optional.ofNullable(assetProfileCache.get(tenantId, (AssetId) entityId)).map(AssetProfile::getId).orElse(null);
+            case DEVICE -> Optional.ofNullable(deviceProfileCache.get(tenantId, (DeviceId) entityId)).map(DeviceProfile::getId).orElse(null);
             default -> null;
         };
     }

application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java

@@ -24,6 +24,7 @@ import jakarta.annotation.PreDestroy;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.common.util.ThingsBoardExecutors;
+import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.configuration.Argument;
 import org.thingsboard.server.common.data.cf.configuration.ArgumentType;
 import org.thingsboard.server.common.data.cf.configuration.RelationPathQueryDynamicSourceConfiguration;
@@ -194,8 +195,14 @@ public abstract class AbstractCalculatedFieldProcessingService {
             return switch (relation.direction()) {
                 case FROM -> relations.stream()
                         .map(EntityRelation::getTo)
+                        .filter(CalculatedField::isSupportedRefEntity)
                         .toList();
-                case TO -> relations.isEmpty() ? List.of() : List.of(relations.get(0).getFrom());
+                case TO -> relations.stream()
+                        .map(EntityRelation::getFrom)
+                        .filter(CalculatedField::isSupportedRefEntity)
+                        .findFirst()
+                        .map(List::of)
+                        .orElseGet(Collections::emptyList);
             };
         }, calculatedFieldCallbackExecutor);
     }

application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java

@@ -162,7 +162,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
     }
 
     private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate<CalculatedFieldCtx> filter, Predicate<CalculatedFieldCtx> linkedEntityFilter, Predicate<CalculatedFieldCtx> dynamicSourceFilter, Predicate<CalculatedFieldCtx> relatedEntityFilter) {
-        if (!CalculatedField.SUPPORTED_REFERENCED_ENTITIES.contains(entityId.getEntityType())) {
+        if (!CalculatedField.isSupportedRefEntity(entityId)) {
             return false;
         }
 

application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java

@@ -29,7 +29,6 @@ import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.EntityType;
-import org.thingsboard.server.common.data.ObjectType;
 import org.thingsboard.server.common.data.TbResource;
 import org.thingsboard.server.common.data.TbResourceInfo;
 import org.thingsboard.server.common.data.Tenant;
@@ -49,6 +48,7 @@ import org.thingsboard.server.common.data.job.Job;
 import org.thingsboard.server.common.data.msg.TbMsgType;
 import org.thingsboard.server.common.data.notification.NotificationRequest;
 import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
+import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.data.rule.RuleChain;
 import org.thingsboard.server.common.data.rule.RuleChainType;
 import org.thingsboard.server.common.data.security.DeviceCredentials;
@@ -274,10 +274,13 @@ public class EntityStateSourcingListener {
 
     @TransactionalEventListener(fallbackExecution = true)
     public void handleEvent(RelationActionEvent relationEvent) {
-        if (relationEvent.getActionType() == ActionType.RELATION_ADD_OR_UPDATE) {
-            tbClusterService.onRelationUpdated(relationEvent.getTenantId(), relationEvent.getRelation(), TbQueueCallback.EMPTY);
-        } else if (relationEvent.getActionType() == ActionType.RELATION_DELETED) {
-            tbClusterService.onRelationDeleted(relationEvent.getTenantId(), relationEvent.getRelation(), TbQueueCallback.EMPTY);
+        EntityRelation relation = relationEvent.getRelation();
+        if (CalculatedField.isSupportedRefEntity(relation.getFrom()) && CalculatedField.isSupportedRefEntity(relation.getTo())) {
+            if (relationEvent.getActionType() == ActionType.RELATION_ADD_OR_UPDATE) {
+                tbClusterService.onRelationUpdated(relationEvent.getTenantId(), relation, TbQueueCallback.EMPTY);
+            } else if (relationEvent.getActionType() == ActionType.RELATION_DELETED) {
+                tbClusterService.onRelationDeleted(relationEvent.getTenantId(), relation, TbQueueCallback.EMPTY);
+            }
         }
     }
 

application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java

@@ -51,6 +51,7 @@ import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.data.relation.EntitySearchDirection;
 import org.thingsboard.server.common.data.relation.RelationPathLevel;
 import org.thingsboard.server.common.data.relation.RelationTypeGroup;
+import org.thingsboard.server.common.data.rule.RuleChain;
 import org.thingsboard.server.common.data.security.Authority;
 import org.thingsboard.server.controller.AbstractControllerTest;
 import org.thingsboard.server.dao.service.DaoSqlTest;
@@ -299,6 +300,45 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
                 });
     }
 
+    @Test
+    public void testCreateCfAndRelationToRuleChain_checkAggregation() throws Exception {
+        Asset asset2 = createAsset("Asset 2", assetProfile.getId());
+        Device device3 = createDevice("Device 3", "1234567890333");
+        postTelemetry(device3.getId(), "{\"occupied\":true}");
+
+        RuleChain ruleChain = new RuleChain();
+        ruleChain.setName("RuleChain");
+        ruleChain = doPost("/api/ruleChain", ruleChain, RuleChain.class);
+        postTelemetry(ruleChain.getId(), "{\"occupied\":true}");
+
+        createEntityRelation(asset2.getId(), device3.getId(), "Contains");
+        createEntityRelation(asset2.getId(), ruleChain.getId(), "Contains");
+
+        createOccupancyCF(asset2.getId());
+
+        await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    verifyTelemetry(asset2.getId(), Map.of(
+                            "freeSpaces", "0",
+                            "occupiedSpaces", "1",
+                            "totalSpaces", "1"
+                    ));
+                });
+
+        postTelemetry(ruleChain.getId(), "{\"occupied\":true}");
+
+        await().alias("update telemetry on rule chain and no aggregation performed").atMost(deduplicationInterval, TimeUnit.SECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    verifyTelemetry(asset2.getId(), Map.of(
+                            "freeSpaces", "0",
+                            "occupiedSpaces", "1",
+                            "totalSpaces", "1"
+                    ));
+                });
+    }
+
     @Test
     public void testDeleteCf_checkNoAggregation() throws Exception {
         CalculatedField cf = createOccupancyCF(asset.getId());
@@ -466,6 +506,24 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
                 });
     }
 
+    @Test
+    public void testDeleteEntityByRelation_checkAggregation() throws Exception {
+        createOccupancyCF(asset.getId());
+        checkInitialCalculation();
+
+        doDelete("/api/device/" + device1.getId()).andExpect(status().isOk());
+
+        await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    verifyTelemetry(asset.getId(), Map.of(
+                            "freeSpaces", "1",
+                            "occupiedSpaces", "0",
+                            "totalSpaces", "1"
+                    ));
+                });
+    }
+
     @Test
     public void testUpdateRelationPath_checkAggregation() throws Exception {
         CalculatedField cf = createOccupancyCF(asset.getId());

common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java

@@ -88,6 +88,8 @@ public interface RelationService {
 
     List<EntityRelation> findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery);
 
+    List<EntityRelation> findEntityRelations(TenantId tenantId, EntityId entityId);
+
 //    TODO: This method may be useful for some validations in the future
 //    ListenableFuture<Boolean> checkRecursiveRelation(EntityId from, EntityId to);
 

common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java

@@ -65,6 +65,10 @@ public class CalculatedField extends BaseData<CalculatedFieldId> implements HasN
             EntityType.DEVICE, EntityType.ASSET, EntityType.CUSTOMER, EntityType.TENANT
     ));
 
+    public static boolean isSupportedRefEntity(EntityId entity) {
+        return SUPPORTED_REFERENCED_ENTITIES.contains(entity.getEntityType());
+    }
+
     private TenantId tenantId;
     private EntityId entityId;
 

dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java

@@ -541,6 +541,16 @@ public class BaseRelationService implements RelationService {
         return relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit);
     }
 
+    @Override
+    public List<EntityRelation> findEntityRelations(TenantId tenantId, EntityId entityId) {
+        List<EntityRelation> relations = new ArrayList<>();
+        for (RelationTypeGroup relationTypeGroup : RelationTypeGroup.values()) {
+            relations.addAll(findByFrom(tenantId, entityId, relationTypeGroup));
+            relations.addAll(findByTo(tenantId, entityId, relationTypeGroup));
+        }
+        return relations;
+    }
+
     private void validate(EntityRelationPathQuery relationPathQuery) {
         validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id);
         List<RelationPathLevel> levels = relationPathQuery.levels();