CF: Current customer dynamic source support

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -116,6 +116,7 @@ import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 import org.thingsboard.server.service.cf.CalculatedFieldProcessingService;
 import org.thingsboard.server.service.cf.CalculatedFieldQueueService;
 import org.thingsboard.server.service.cf.CalculatedFieldStateService;
+import org.thingsboard.server.service.cf.OwnerService;
 import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
 import org.thingsboard.server.service.component.ComponentDiscoveryService;
 import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
@@ -566,6 +567,10 @@ public class ActorSystemContext {
     @Getter
     private JobManager jobManager;
 
+    @Autowired
+    @Getter
+    private OwnerService ownerService;
+
     @Value("${actors.session.max_concurrent_sessions_per_device:1}")
     @Getter
     private int maxConcurrentSessionsPerDevice;

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

@@ -0,0 +1,37 @@
+/**
+ * Copyright © 2016-2025 The Thingsboard Authors
+ *
+ * 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.thingsboard.server.actors.calculatedField;
+
+import lombok.Data;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.MsgType;
+import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
+import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
+
+@Data
+public class CalculatedFieldArgumentResetMsg implements ToCalculatedFieldSystemMsg {
+
+    private final TenantId tenantId;
+    private final CalculatedFieldCtx ctx;
+    private final TbCallback callback;
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.CF_ARGUMENT_RESET_MSG;
+    }
+
+}

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

@@ -84,6 +84,9 @@ public class CalculatedFieldEntityActor extends AbstractCalculatedFieldActor {
             case CF_ALARM_ACTION_MSG:
                 processor.process((CalculatedFieldAlarmActionMsg) msg);
                 break;
+            case CF_ARGUMENT_RESET_MSG:
+                processor.process((CalculatedFieldArgumentResetMsg) msg);
+                break;
             default:
                 return false;
         }

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

@@ -155,6 +155,24 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
         }
     }
 
+    public void process(CalculatedFieldArgumentResetMsg msg) throws CalculatedFieldException {
+        log.debug("[{}] Processing CF argument reset msg.", entityId);
+        var ctx = msg.getCtx();
+        var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback());
+        try {
+            Map<String, Argument> dynamicSourceArgs = ctx.getArguments().entrySet().stream()
+                    .filter(entry -> entry.getValue().hasOwnerSource())
+                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+            Map<String, ArgumentEntry> fetchedArgs = cfService.fetchArgsFromDb(tenantId, entityId, dynamicSourceArgs);
+            fetchedArgs.values().forEach(arg -> arg.setForceResetPrevious(true));
+
+            processArgumentValuesUpdate(ctx, Collections.singletonList(ctx.getCfId()), callback, fetchedArgs, null, null);
+        } catch (Exception e) {
+            throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
+        }
+    }
+
     public void process(CalculatedFieldEntityDeleteMsg msg) {
         log.debug("[{}] Processing CF entity delete msg.", msg.getEntityId());
         if (this.entityId.equals(msg.getEntityId())) {
@@ -422,11 +440,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
     }
 
     private Map<String, ArgumentEntry> mapToArguments(CalculatedFieldCtx ctx, EntityId entityId, List<TsKvProto> data) {
-        var argNames = ctx.getLinkedEntityArguments().get(entityId);
+        return mapToArguments(ctx.getLinkedAndDynamicArgs(entityId), data);
-        if (argNames.isEmpty()) {
-            return Collections.emptyMap();
-        }
-        return mapToArguments(argNames, data);
     }
 
     private Map<String, ArgumentEntry> mapToArguments(Map<ReferencedEntityKey, String> argNames, List<TsKvProto> data) {
@@ -454,11 +468,11 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
     }
 
     private Map<String, ArgumentEntry> mapToArguments(CalculatedFieldCtx ctx, EntityId entityId, AttributeScopeProto scope, List<AttributeValueProto> attrDataList) {
-        var argNames = ctx.getLinkedEntityArguments().get(entityId);
+        var argNames = ctx.getLinkedAndDynamicArgs(entityId);
         if (argNames.isEmpty()) {
             return Collections.emptyMap();
         }
-        List<String> geofencingArgumentNames = ctx.getLinkedEntityGeofencingArgumentNames();
+        List<String> geofencingArgumentNames = ctx.getLinkedEntityAndCurrentOwnerGeofencingArgumentNames();
         return mapToArguments(entityId, argNames, geofencingArgumentNames, scope, attrDataList);
     }
 
@@ -480,11 +494,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
     }
 
     private Map<String, ArgumentEntry> mapToArgumentsWithDefaultValue(CalculatedFieldCtx ctx, EntityId entityId, AttributeScopeProto scope, List<String> removedAttrKeys) {
-        var argNames = ctx.getLinkedEntityArguments().get(entityId);
+        return mapToArgumentsWithDefaultValue(ctx.getLinkedAndDynamicArgs(entityId), ctx.getArguments(), scope, removedAttrKeys);
-        if (argNames.isEmpty()) {
-            return Collections.emptyMap();
-        }
-        return mapToArgumentsWithDefaultValue(argNames, ctx.getArguments(), scope, removedAttrKeys);
     }
 
     private Map<String, ArgumentEntry> mapToArgumentsWithDefaultValue(CalculatedFieldCtx ctx, AttributeScopeProto scope, List<String> removedAttrKeys) {

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

@@ -49,6 +49,7 @@ import org.thingsboard.server.dao.device.DeviceService;
 import org.thingsboard.server.queue.settings.TbQueueCalculatedFieldSettings;
 import org.thingsboard.server.service.cf.CalculatedFieldProcessingService;
 import org.thingsboard.server.service.cf.CalculatedFieldStateService;
+import org.thingsboard.server.service.cf.OwnerService;
 import org.thingsboard.server.service.cf.cache.TenantEntityProfileCache;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
@@ -58,8 +59,10 @@ import org.thingsboard.server.service.profile.TbDeviceProfileCache;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ScheduledFuture;
@@ -77,6 +80,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     private final Map<CalculatedFieldId, CalculatedFieldCtx> calculatedFields = new HashMap<>();
     private final Map<EntityId, List<CalculatedFieldCtx>> entityIdCalculatedFields = new HashMap<>();
     private final Map<EntityId, List<CalculatedFieldLink>> entityIdCalculatedFieldLinks = new HashMap<>();
+    private final Map<EntityId, Set<EntityId>> ownerEntities = new HashMap<>();
     private final Map<CalculatedFieldId, ScheduledFuture<?>> cfDynamicArgumentsRefreshTasks = new ConcurrentHashMap<>();
     private ScheduledFuture<?> cfsReevaluationTask;
 
@@ -88,6 +92,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     private final TbAssetProfileCache assetProfileCache;
     private final TbDeviceProfileCache deviceProfileCache;
     private final TenantEntityProfileCache entityProfileCache;
+    private final OwnerService ownerService;
     private final TbQueueCalculatedFieldSettings cfSettings;
     protected final TenantId tenantId;
 
@@ -103,6 +108,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         this.assetProfileCache = systemContext.getAssetProfileCache();
         this.deviceProfileCache = systemContext.getDeviceProfileCache();
         this.entityProfileCache = new TenantEntityProfileCache();
+        this.ownerService = systemContext.getOwnerService();
         this.cfSettings = systemContext.getCalculatedFieldSettings();
         this.tenantId = tenantId;
     }
@@ -128,7 +134,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
 
     public void onCacheInitMsg(CalculatedFieldCacheInitMsg msg) {
         log.debug("[{}] Processing CF actor init message.", msg.getTenantId().getId());
-        initEntityProfileCache();
+        initEntitiesCache();
         initCalculatedFields();
         scheduleCfsReevaluation();
         msg.getCallback().onSuccess();
@@ -251,6 +257,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         if (profileId != null) {
             entityProfileCache.add(profileId, entityId);
         }
+        updateEntityOwner(entityId);
         if (!isMyPartition(entityId, callback)) {
             return;
         }
@@ -283,11 +290,16 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
             } else {
                 callback.onSuccess();
             }
+        } else if (msg.isOwnerChanged()) {
+            onEntityOwnerChanged(msg, callback);
+        } else {
+            callback.onSuccess();
         }
     }
 
     private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) {
         entityProfileCache.removeEntityId(msg.getEntityId());
+        ownerEntities.values().forEach(entities -> entities.remove(msg.getEntityId()));
         if (isMyPartition(msg.getEntityId(), callback)) {
             log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId());
             getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback));
@@ -415,8 +427,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     public void onTelemetryMsg(CalculatedFieldTelemetryMsg msg) {
         EntityId entityId = msg.getEntityId();
         log.debug("Received telemetry msg from entity [{}]", entityId);
-        // 2 = 1 for CF processing + 1 for links processing
+        // 3 = 1 for CF processing + 1 for links processing + 1 for owner entity processing
-        MultipleTbCallback callback = new MultipleTbCallback(2, msg.getCallback());
+        MultipleTbCallback callback = new MultipleTbCallback(3, msg.getCallback());
         // process all cfs related to entity, or it's profile;
         var entityIdFields = getCalculatedFieldsByEntityId(entityId);
         var profileIdFields = getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId));
@@ -434,6 +446,17 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         } else {
             callback.onSuccess();
         }
+        // process all cfs related to owner entity
+        if (entityId.getEntityType().isOneOf(EntityType.TENANT, EntityType.CUSTOMER)) {
+            List<CalculatedFieldEntityCtxId> ownerCFs = filterOwnerEntitiesCFs(msg);
+            if (!ownerCFs.isEmpty()) {
+                cfExecService.pushMsgToLinks(msg, ownerCFs, callback);
+            } else {
+                callback.onSuccess();
+            }
+        } else {
+            callback.onSuccess();
+        }
     }
 
     public void onLinkedTelemetryMsg(CalculatedFieldLinkedTelemetryMsg msg) {
@@ -456,6 +479,31 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         }
     }
 
+    private void onEntityOwnerChanged(ComponentLifecycleMsg msg, TbCallback msgCallback) {
+        EntityId entityId = msg.getEntityId();
+        log.debug("Received changed owner msg from entity [{}]", entityId);
+        updateEntityOwner(entityId);
+        List<CalculatedFieldCtx> cfs = new ArrayList<>();
+        cfs.addAll(getCalculatedFieldsByEntityId(entityId));
+        cfs.addAll(getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId)));
+        if (cfs.isEmpty()) {
+            msgCallback.onSuccess();
+            return;
+        }
+        MultipleTbCallback callback = new MultipleTbCallback(cfs.size(), msgCallback);
+        cfs.forEach(cf -> {
+            if (isMyPartition(entityId, callback)) {
+                if (cf.hasCurrentOwnerSourceArguments()) {
+                    CalculatedFieldArgumentResetMsg argResetMsg = new CalculatedFieldArgumentResetMsg(tenantId, cf, callback);
+                    log.debug("Pushing CF argument reset msg to specific actor [{}]", entityId);
+                    getOrCreateActor(entityId).tell(argResetMsg);
+                } else {
+                    callback.onSuccess();
+                }
+            }
+        });
+    }
+
     private List<CalculatedFieldEntityCtxId> filterCalculatedFieldLinks(CalculatedFieldTelemetryMsg msg) {
         EntityId entityId = msg.getEntityId();
         var proto = msg.getProto();
@@ -469,6 +517,27 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         return result;
     }
 
+    private List<CalculatedFieldEntityCtxId> filterOwnerEntitiesCFs(CalculatedFieldTelemetryMsg msg) {
+        Set<EntityId> entities = getOwnerEntities(msg.getEntityId());
+        var proto = msg.getProto();
+        List<CalculatedFieldEntityCtxId> result = new ArrayList<>();
+        for (var entityId : entities) {
+            var ownerEntityCFs = getCalculatedFieldsByEntityId(entityId);
+            for (var ctx : ownerEntityCFs) {
+                if (ctx.dynamicSourceMatches(proto)) {
+                    result.add(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId));
+                }
+            }
+            var ownerEntityProfileCFs = getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId));
+            for (var ctx : ownerEntityProfileCFs) {
+                if (ctx.dynamicSourceMatches(proto)) {
+                    result.add(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId));
+                }
+            }
+        }
+        return result;
+    }
+
     private List<CalculatedFieldCtx> getCalculatedFieldsByEntityId(EntityId entityId) {
         if (entityId == null) {
             return Collections.emptyList();
@@ -491,6 +560,17 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         return result;
     }
 
+    private Set<EntityId> getOwnerEntities(EntityId entityId) {
+        if (entityId == null) {
+            return Collections.emptySet();
+        }
+        var result = ownerEntities.get(entityId);
+        if (result == null) {
+            result = Collections.emptySet();
+        }
+        return result;
+    }
+
     private void scheduleDynamicArgumentsRefreshTaskForCfIfNeeded(CalculatedFieldCtx cfCtx) {
         CalculatedField cf = cfCtx.getCalculatedField();
         if (!(cf.getConfiguration() instanceof ScheduledUpdateSupportedCalculatedFieldConfiguration scheduledCfConfig)) {
@@ -623,12 +703,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(link);
     }
 
-    private void initEntityProfileCache() {
+    private void initEntitiesCache() {
         PageDataIterable<ProfileEntityIdInfo> deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findProfileEntityIdInfosByTenantId(tenantId, pageLink), cfSettings.getInitTenantFetchPackSize());
         for (ProfileEntityIdInfo idInfo : deviceIdInfos) {
             log.trace("Processing device record: {}", idInfo);
             try {
                 entityProfileCache.add(idInfo.getProfileId(), idInfo.getEntityId());
+                ownerEntities.computeIfAbsent(idInfo.getOwnerId(), ownerId -> new HashSet<>()).add(idInfo.getEntityId());
             } catch (Exception e) {
                 log.error("Failed to process device record: {}", idInfo, e);
             }
@@ -638,12 +719,19 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
             log.trace("Processing asset record: {}", idInfo);
             try {
                 entityProfileCache.add(idInfo.getProfileId(), idInfo.getEntityId());
+                ownerEntities.computeIfAbsent(idInfo.getOwnerId(), ownerId -> new HashSet<>()).add(idInfo.getEntityId());
             } catch (Exception e) {
                 log.error("Failed to process asset record: {}", idInfo, e);
             }
         }
     }
 
+    private void updateEntityOwner(EntityId entityId) {
+        ownerEntities.values().forEach(entities -> entities.remove(entityId));
+        EntityId owner = ownerService.getOwner(tenantId, entityId);
+        ownerEntities.computeIfAbsent(owner, ownerId -> new HashSet<>()).add(entityId);
+    }
+
     private void applyToTargetCfEntityActors(CalculatedFieldCtx ctx,
                                              TbCallback callback,
                                              BiConsumer<EntityId, TbCallback> action) {

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

@@ -19,11 +19,13 @@ import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
+import jakarta.annotation.Nullable;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.common.util.ThingsBoardExecutors;
+import org.thingsboard.server.common.data.EntityType;
 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.RelationQueryDynamicSourceConfiguration;
@@ -45,6 +47,7 @@ import org.thingsboard.server.dao.usagerecord.ApiLimitService;
 import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -55,6 +58,7 @@ import java.util.stream.Collectors;
 
 import static org.thingsboard.server.common.data.cf.configuration.geofencing.EntityCoordinates.ENTITY_ID_LATITUDE_ARGUMENT_KEY;
 import static org.thingsboard.server.common.data.cf.configuration.geofencing.EntityCoordinates.ENTITY_ID_LONGITUDE_ARGUMENT_KEY;
+import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultAttributeEntry;
 import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultKvEntry;
 import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformSingleValueArgument;
 
@@ -66,6 +70,7 @@ public abstract class AbstractCalculatedFieldProcessingService {
     protected final TimeseriesService timeseriesService;
     protected final ApiLimitService apiLimitService;
     protected final RelationService relationService;
+    protected final OwnerService ownerService;
 
     protected ListeningExecutorService calculatedFieldCallbackExecutor;
 
@@ -84,14 +89,14 @@ public abstract class AbstractCalculatedFieldProcessingService {
 
     protected abstract String getExecutorNamePrefix();
 
-    protected ListenableFuture<Map<String, ArgumentEntry>> fetchArguments(CalculatedFieldCtx ctx, EntityId entityId) {
+    protected ListenableFuture<Map<String, ArgumentEntry>> fetchArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) {
         Map<String, ListenableFuture<ArgumentEntry>> argFutures = switch (ctx.getCalculatedField().getType()) {
-            case GEOFENCING -> fetchGeofencingCalculatedFieldArguments(ctx, entityId, false);
+            case GEOFENCING -> fetchGeofencingCalculatedFieldArguments(ctx, entityId, false, ts);
             case SIMPLE, SCRIPT, ALARM -> {
                 Map<String, ListenableFuture<ArgumentEntry>> futures = new HashMap<>();
                 for (var entry : ctx.getArguments().entrySet()) {
-                    var argEntityId = resolveEntityId(entityId, entry.getValue());
+                    var argEntityId = resolveEntityId(ctx.getTenantId(), entityId, entry.getValue());
-                    var argValueFuture = fetchArgumentValue(ctx.getTenantId(), argEntityId, entry.getValue(), System.currentTimeMillis());
+                    var argValueFuture = fetchArgumentValue(ctx.getTenantId(), argEntityId, entry.getValue(), ts);
                     futures.put(entry.getKey(), argValueFuture);
                 }
                 yield futures;
@@ -102,8 +107,14 @@ public abstract class AbstractCalculatedFieldProcessingService {
                         MoreExecutors.directExecutor());
     }
 
-    protected EntityId resolveEntityId(EntityId entityId, Argument argument) {
+    protected EntityId resolveEntityId(TenantId tenantId, EntityId entityId, Argument argument) {
-        return argument.getRefEntityId() != null ? argument.getRefEntityId() : entityId;
+        if (argument.getRefEntityId() != null) {
+            return argument.getRefEntityId();
+        }
+        if (!argument.hasOwnerSource()) {
+            return entityId;
+        }
+        return resolveOwnerArgument(tenantId, entityId, argument);
     }
 
     protected Map<String, ArgumentEntry> resolveArgumentFutures(Map<String, ListenableFuture<ArgumentEntry>> argFutures) {
@@ -123,18 +134,18 @@ public abstract class AbstractCalculatedFieldProcessingService {
                 ));
     }
 
-    protected Map<String, ListenableFuture<ArgumentEntry>> fetchGeofencingCalculatedFieldArguments(CalculatedFieldCtx ctx, EntityId entityId, boolean dynamicArgumentsOnly) {
+    protected Map<String, ListenableFuture<ArgumentEntry>> fetchGeofencingCalculatedFieldArguments(CalculatedFieldCtx ctx, EntityId entityId, boolean dynamicArgumentsOnly, long startTs) {
         Map<String, ListenableFuture<ArgumentEntry>> argFutures = new HashMap<>();
         Set<Map.Entry<String, Argument>> entries = ctx.getArguments().entrySet();
         if (dynamicArgumentsOnly) {
             entries = entries.stream()
-                    .filter(entry -> entry.getValue().hasDynamicSource())
+                    .filter(entry -> entry.getValue().hasRelationQuerySource())
                     .collect(Collectors.toSet());
         }
         for (var entry : entries) {
             switch (entry.getKey()) {
                 case ENTITY_ID_LATITUDE_ARGUMENT_KEY, ENTITY_ID_LONGITUDE_ARGUMENT_KEY ->
-                        argFutures.put(entry.getKey(), fetchArgumentValue(ctx.getTenantId(), entityId, entry.getValue(), System.currentTimeMillis()));
+                        argFutures.put(entry.getKey(), fetchArgumentValue(ctx.getTenantId(), entityId, entry.getValue(), startTs));
                 default -> {
                     var resolvedEntityIdsFuture = resolveGeofencingEntityIds(ctx.getTenantId(), entityId, entry);
                     argFutures.put(entry.getKey(), Futures.transformAsync(resolvedEntityIdsFuture, resolvedEntityIds ->
@@ -155,6 +166,14 @@ public abstract class AbstractCalculatedFieldProcessingService {
         }
         var refDynamicSourceConfiguration = value.getRefDynamicSourceConfiguration();
         return switch (refDynamicSourceConfiguration.getType()) {
+            case CURRENT_CUSTOMER -> {
+                EntityId resolved = resolveOwnerArgument(tenantId, entityId, value);
+                if (resolved != null) {
+                    yield Futures.immediateFuture(List.of(resolved));
+                } else {
+                    yield Futures.immediateFuture(Collections.emptyList());
+                }
+            }
             case RELATION_QUERY -> {
                 var configuration = (RelationQueryDynamicSourceConfiguration) refDynamicSourceConfiguration;
                 if (configuration.isSimpleRelation()) {
@@ -170,7 +189,23 @@ public abstract class AbstractCalculatedFieldProcessingService {
                 yield Futures.transform(relationService.findByQuery(tenantId, configuration.toEntityRelationsQuery(entityId)),
                         configuration::resolveEntityIds, calculatedFieldCallbackExecutor);
             }
-            case CURRENT_CUSTOMER -> throw new UnsupportedOperationException(); // fixme implement
+        };
+    }
+
+    @Nullable
+    private EntityId resolveOwnerArgument(TenantId tenantId, EntityId entityId, Argument argument) {
+        return switch (argument.getRefDynamicSourceConfiguration().getType()) {
+            case CURRENT_CUSTOMER -> {
+                EntityId ownerId = ownerService.getOwner(tenantId, entityId);
+                if (ownerId.getEntityType() == EntityType.TENANT) {
+                    // todo: if inherit is true - use customer id
+                    // fixme: WTF do we need it at all?
+                    yield null;
+                } else {
+                    yield ownerId;
+                }
+            }
+            default -> throw new UnsupportedOperationException();
         };
     }
 
@@ -187,8 +222,7 @@ public abstract class AbstractCalculatedFieldProcessingService {
                             argument.getRefEntityKey().getKey()
                     );
                     return Futures.transform(attributesFuture, resultOpt ->
-                                    Map.entry(entityId, resultOpt.orElseGet(() ->
+                                    Map.entry(entityId, resultOpt.orElseGet(() -> createDefaultAttributeEntry(argument, System.currentTimeMillis()))),
-                                            new BaseAttributeKvEntry(createDefaultKvEntry(argument), System.currentTimeMillis(), 0L))),
                             calculatedFieldCallbackExecutor
                     );
                 }).collect(Collectors.toList());
@@ -200,6 +234,9 @@ public abstract class AbstractCalculatedFieldProcessingService {
     }
 
     protected ListenableFuture<ArgumentEntry> fetchArgumentValue(TenantId tenantId, EntityId entityId, Argument argument, long startTs) {
+        if (entityId == null) {
+            return Futures.immediateFuture(transformSingleValueArgument(Optional.empty()));
+        }
         return switch (argument.getRefEntityKey().getType()) {
             case TS_ROLLING -> fetchTsRolling(tenantId, entityId, argument, startTs);
             case ATTRIBUTE -> fetchAttribute(tenantId, entityId, argument, startTs);

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

@@ -23,6 +23,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 
 import java.util.List;
+import java.util.Set;
 import java.util.function.Predicate;
 
 public interface CalculatedFieldCache {
@@ -47,4 +48,14 @@ public interface CalculatedFieldCache {
 
     EntityId getProfileId(TenantId tenantId, EntityId entityId);
 
+    Set<EntityId> getDynamicEntities(TenantId tenantId, EntityId entityId);
+
+    void updateOwnerEntity(TenantId tenantId, EntityId entityId);
+
+    void addOwnerEntity(TenantId tenantId, EntityId entityId);
+
+    void evictEntity(EntityId entityId);
+
+    void evictOwner(EntityId owner);
+
 }

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

@@ -23,6 +23,7 @@ import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 import org.springframework.util.ConcurrentReferenceHashMap;
 import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
 import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration;
@@ -40,6 +41,7 @@ import org.thingsboard.server.service.profile.TbDeviceProfileCache;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -59,6 +61,7 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
     private final TbDeviceProfileCache deviceProfileCache;
     @Lazy
     private final ActorSystemContext systemContext;
+    private final OwnerService ownerService;
 
     private final ConcurrentMap<CalculatedFieldId, CalculatedField> calculatedFields = new ConcurrentHashMap<>();
     private final ConcurrentMap<EntityId, List<CalculatedField>> entityIdCalculatedFields = new ConcurrentHashMap<>();
@@ -66,6 +69,8 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
     private final ConcurrentMap<EntityId, List<CalculatedFieldLink>> entityIdCalculatedFieldLinks = new ConcurrentHashMap<>();
     private final ConcurrentMap<CalculatedFieldId, CalculatedFieldCtx> calculatedFieldsCtx = new ConcurrentHashMap<>();
 
+    private final ConcurrentMap<EntityId, Set<EntityId>> ownerEntities = new ConcurrentHashMap<>();
+
     @Value("${queue.calculated_fields.init_fetch_pack_size:50000}")
     @Getter
     private int initFetchPackSize;
@@ -220,6 +225,44 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
         };
     }
 
+    @Override
+    public Set<EntityId> getDynamicEntities(TenantId tenantId, EntityId entityId) {
+        if (entityId != null && entityId.getEntityType().isOneOf(EntityType.CUSTOMER, EntityType.TENANT)) {
+            return getOwnedEntities(tenantId, entityId);
+        }
+        return Collections.emptySet();
+    }
+
+    @Override
+    public void addOwnerEntity(TenantId tenantId, EntityId entityId) {
+        EntityId owner = ownerService.getOwner(tenantId, entityId);
+        getOwnedEntities(tenantId, owner).add(entityId);
+    }
+
+    @Override
+    public void updateOwnerEntity(TenantId tenantId, EntityId entityId) {
+        evictEntity(entityId);
+        addOwnerEntity(tenantId, entityId);
+    }
+
+    @Override
+    public void evictEntity(EntityId entityId) {
+        ownerEntities.values().forEach(entities -> entities.remove(entityId));
+    }
+
+    @Override
+    public void evictOwner(EntityId owner) {
+        ownerEntities.remove(owner);
+    }
+
+    private Set<EntityId> getOwnedEntities(TenantId tenantId, EntityId ownerId) {
+        return ownerEntities.computeIfAbsent(ownerId, owner -> {
+            Set<EntityId> entities = ConcurrentHashMap.newKeySet();
+            entities.addAll(ownerService.getOwnedEntities(tenantId, ownerId));
+            return entities;
+        });
+    }
+
     private Lock getFetchLock(CalculatedFieldId id) {
         return calculatedFieldFetchLocks.computeIfAbsent(id, __ -> new ReentrantLock());
     }

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

@@ -69,9 +69,10 @@ public class DefaultCalculatedFieldProcessingService extends AbstractCalculatedF
                                                    TimeseriesService timeseriesService,
                                                    ApiLimitService apiLimitService,
                                                    RelationService relationService,
+                                                   OwnerService ownerService,
                                                    TbClusterService clusterService,
                                                    PartitionService partitionService) {
-        super(attributesService, timeseriesService, apiLimitService, relationService);
+        super(attributesService, timeseriesService, apiLimitService, relationService, ownerService);
         this.clusterService = clusterService;
         this.partitionService = partitionService;
     }
@@ -83,26 +84,26 @@ public class DefaultCalculatedFieldProcessingService extends AbstractCalculatedF
 
     @Override
     public ListenableFuture<Map<String, ArgumentEntry>> fetchArguments(CalculatedFieldCtx ctx, EntityId entityId) {
-        return super.fetchArguments(ctx, entityId);
+        return super.fetchArguments(ctx, entityId, System.currentTimeMillis());
     }
 
     @Override
     public Map<String, ArgumentEntry> fetchDynamicArgsFromDb(CalculatedFieldCtx ctx, EntityId entityId) {
-        // only geofencing calculated fields supports dynamic arguments scheduled updates
+        // only scheduledSupported CF instances supports dynamic arguments scheduled updates
         if (!ctx.getCalculatedField().getType().equals(CalculatedFieldType.GEOFENCING)) {
             return Map.of();
         }
-        return resolveArgumentFutures(fetchGeofencingCalculatedFieldArguments(ctx, entityId, true));
+        return resolveArgumentFutures(fetchGeofencingCalculatedFieldArguments(ctx, entityId, true, System.currentTimeMillis()));
     }
 
     @Override
     public Map<String, ArgumentEntry> fetchArgsFromDb(TenantId tenantId, EntityId entityId, Map<String, Argument> arguments) {
         Map<String, ListenableFuture<ArgumentEntry>> argFutures = new HashMap<>();
         for (var entry : arguments.entrySet()) {
-            if (entry.getValue().hasDynamicSource()) {
+            if (entry.getValue().hasRelationQuerySource()) {
                 continue;
             }
-            var argEntityId = resolveEntityId(entityId, entry.getValue());
+            var argEntityId = resolveEntityId(tenantId, entityId, entry.getValue());
             var argValueFuture = fetchArgumentValue(tenantId, argEntityId, entry.getValue(), System.currentTimeMillis());
             argFutures.put(entry.getKey(), argValueFuture);
         }

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

@@ -85,6 +85,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
         checkEntityAndPushToQueue(tenantId, entityId,
                 cf -> cf.matches(entries),
                 cf -> cf.linkMatches(entityId, entries),
+                cf -> cf.dynamicSourceMatches(request.getEntries()),
                 () -> toCalculatedFieldTelemetryMsgProto(request, result), callback);
     }
 
@@ -102,6 +103,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
         checkEntityAndPushToQueue(tenantId, entityId,
                 cf -> cf.matches(entries, scope),
                 cf -> cf.linkMatches(entityId, entries, scope),
+                cf -> cf.dynamicSourceMatches(request.getEntries(), request.getScope()),
                 () -> toCalculatedFieldTelemetryMsgProto(request, result), callback);
     }
 
@@ -118,6 +120,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
         checkEntityAndPushToQueue(tenantId, entityId,
                 cf -> cf.matchesKeys(result, scope),
                 cf -> cf.linkMatchesAttrKeys(entityId, result, scope),
+                cf -> cf.matchesDynamicSourceKeys(result, request.getScope()),
                 () -> toCalculatedFieldTelemetryMsgProto(request, result), callback);
     }
 
@@ -128,16 +131,19 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
         checkEntityAndPushToQueue(tenantId, entityId,
                 cf -> cf.matchesKeys(result),
                 cf -> cf.linkMatchesTsKeys(entityId, result),
+                cf -> cf.matchesDynamicSourceKeys(result),
                 () -> toCalculatedFieldTelemetryMsgProto(request, result), callback);
     }
 
     private void checkEntityAndPushToQueue(TenantId tenantId, EntityId entityId,
-                                           Predicate<CalculatedFieldCtx> mainEntityFilter, Predicate<CalculatedFieldCtx> linkedEntityFilter,
+                                           Predicate<CalculatedFieldCtx> mainEntityFilter,
+                                           Predicate<CalculatedFieldCtx> linkedEntityFilter,
+                                           Predicate<CalculatedFieldCtx> dynamicSourceFilter,
                                            Supplier<ToCalculatedFieldMsg> msg, FutureCallback<Void> callback) {
         if (EntityType.TENANT.equals(entityId.getEntityType())) {
             tenantId = (TenantId) entityId;
         }
-        boolean send = checkEntityForCalculatedFields(tenantId, entityId, mainEntityFilter, linkedEntityFilter);
+        boolean send = checkEntityForCalculatedFields(tenantId, entityId, mainEntityFilter, linkedEntityFilter, dynamicSourceFilter);
         if (send) {
             ToCalculatedFieldMsg calculatedFieldMsg = msg.get();
             clusterService.pushMsgToCalculatedFields(tenantId, entityId, calculatedFieldMsg, wrap(callback));
@@ -148,7 +154,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
         }
     }
 
-    private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate<CalculatedFieldCtx> filter, Predicate<CalculatedFieldCtx> linkedEntityFilter) {
+    private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate<CalculatedFieldCtx> filter, Predicate<CalculatedFieldCtx> linkedEntityFilter, Predicate<CalculatedFieldCtx> dynamicSourceFilter) {
         if (!supportedReferencedEntities.contains(entityId.getEntityType())) {
             return false;
         }
@@ -165,6 +171,16 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
             }
         }
 
+        for (EntityId dynamicEntity : calculatedFieldCache.getDynamicEntities(tenantId, entityId)) {
+            if (calculatedFieldCache.getCalculatedFieldCtxsByEntityId(dynamicEntity).stream().anyMatch(dynamicSourceFilter)) {
+                return true;
+            }
+            EntityId dynamicEntityProfileId = calculatedFieldCache.getProfileId(tenantId, dynamicEntity);
+            if (calculatedFieldCache.getCalculatedFieldCtxsByEntityId(dynamicEntityProfileId).stream().anyMatch(dynamicSourceFilter)) {
+                return true;
+            }
+        }
+
         return false;
     }
 

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

@@ -0,0 +1,67 @@
+/**
+ * Copyright © 2016-2025 The Thingsboard Authors
+ *
+ * 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.thingsboard.server.service.cf;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.common.data.DeviceInfo;
+import org.thingsboard.server.common.data.DeviceInfoFilter;
+import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.id.AssetId;
+import org.thingsboard.server.common.data.id.CustomerId;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.page.PageDataIterable;
+import org.thingsboard.server.dao.asset.AssetService;
+import org.thingsboard.server.dao.device.DeviceService;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Service
+@RequiredArgsConstructor
+public class OwnerService {
+
+    private final DeviceService deviceService;
+    private final AssetService assetService;
+
+    public EntityId getOwner(TenantId tenantId, EntityId entityId) {
+        return switch (entityId.getEntityType()) {
+            case DEVICE -> deviceService.findDeviceById(tenantId, (DeviceId) entityId).getOwnerId();
+            case ASSET -> assetService.findAssetById(tenantId, (AssetId) entityId).getOwnerId();
+            default -> throw new UnsupportedOperationException();
+        };
+    }
+
+    public Set<EntityId> getOwnedEntities(TenantId tenantId, EntityId ownerId) {
+        Set<EntityId> ownerEntities = new HashSet<>();
+        if (EntityType.CUSTOMER.equals(ownerId.getEntityType())) {
+            PageDataIterable<DeviceInfo> deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findDeviceInfosByFilter(DeviceInfoFilter.builder().tenantId(tenantId).customerId((CustomerId) ownerId).build(), pageLink), 1000);
+            deviceIdInfos.forEach(deviceInfo -> ownerEntities.add(deviceInfo.getId()));
+            PageDataIterable<Asset> assets = new PageDataIterable<>(pageLink -> assetService.findAssetsByTenantIdAndCustomerId(tenantId, (CustomerId) ownerId, pageLink), 1000);
+            assets.forEach(asset -> ownerEntities.add(asset.getId()));
+        } else if (EntityType.TENANT.equals(ownerId.getEntityType())) {
+            PageDataIterable<DeviceInfo> deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findDeviceInfosByFilter(DeviceInfoFilter.builder().tenantId((TenantId) ownerId).customerId(new CustomerId(CustomerId.NULL_UUID)).build(), pageLink), 1000);
+            deviceIdInfos.forEach(deviceInfo -> ownerEntities.add(deviceInfo.getId()));
+            PageDataIterable<Asset> assets = new PageDataIterable<>(pageLink -> assetService.findAssetsByTenantIdAndCustomerId((TenantId) ownerId, new CustomerId(CustomerId.NULL_UUID), pageLink), 1000);
+            assets.forEach(asset -> ownerEntities.add(asset.getId()));
+        }
+        return ownerEntities;
+    }
+
+}

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java

@@ -73,6 +73,7 @@ public class CalculatedFieldCtx {
     private final Map<String, Argument> arguments;
     private final Map<ReferencedEntityKey, String> mainEntityArguments;
     private final Map<EntityId, Map<ReferencedEntityKey, String>> linkedEntityArguments;
+    private final Map<ReferencedEntityKey, String> dynamicEntityArguments;
     private final List<String> argNames;
     private Output output;
     private String expression;
@@ -93,7 +94,7 @@ public class CalculatedFieldCtx {
     private long maxSingleValueArgumentSize;
 
     private List<String> mainEntityGeofencingArgumentNames;
-    private List<String> linkedEntityGeofencingArgumentNames;
+    private List<String> linkedEntityAndCurrentOwnerGeofencingArgumentNames;
 
     public CalculatedFieldCtx(CalculatedField calculatedField,
                               ActorSystemContext systemContext) {
@@ -106,19 +107,27 @@ public class CalculatedFieldCtx {
         this.arguments = new HashMap<>();
         this.mainEntityArguments = new HashMap<>();
         this.linkedEntityArguments = new HashMap<>();
+        this.dynamicEntityArguments = new HashMap<>();
         this.argNames = new ArrayList<>();
         this.mainEntityGeofencingArgumentNames = new ArrayList<>();
-        this.linkedEntityGeofencingArgumentNames = new ArrayList<>();
+        this.linkedEntityAndCurrentOwnerGeofencingArgumentNames = new ArrayList<>();
         this.output = calculatedField.getConfiguration().getOutput();
         if (calculatedField.getConfiguration() instanceof ArgumentsBasedCalculatedFieldConfiguration argBasedConfig) {
             this.arguments.putAll(argBasedConfig.getArguments());
             for (Map.Entry<String, Argument> entry : arguments.entrySet()) {
                 var refId = entry.getValue().getRefEntityId();
                 var refKey = entry.getValue().getRefEntityKey();
-                if (refId == null && entry.getValue().hasDynamicSource()) {
+                if (refId == null) {
-                    continue;
+                    // TODO: no matchers for this type of source exists yet, so no reason to add to dynamicEntityArguments map.
-                }
+                    if (entry.getValue().hasRelationQuerySource()) {
-                if (refId == null || refId.equals(calculatedField.getEntityId())) {
+                        continue;
+                    }
+                    if (entry.getValue().hasOwnerSource()) {
+                        dynamicEntityArguments.put(refKey, entry.getKey());
+                    } else {
+                        mainEntityArguments.put(refKey, entry.getKey());
+                    }
+                } else if (refId.equals(calculatedField.getEntityId())) {
                     mainEntityArguments.put(refKey, entry.getKey());
                 } else {
                     linkedEntityArguments.computeIfAbsent(refId, key -> new HashMap<>()).put(refKey, entry.getKey());
@@ -135,8 +144,8 @@ public class CalculatedFieldCtx {
                         mainEntityGeofencingArgumentNames.add(zoneGroupName);
                         return;
                     }
-                    if (config.isLinkedCfEntitySource(entityId)) {
+                    if (config.isLinkedCfEntitySource(entityId) || config.hasCurrentOwnerSource()) {
-                        linkedEntityGeofencingArgumentNames.add(zoneGroupName);
+                        linkedEntityAndCurrentOwnerGeofencingArgumentNames.add(zoneGroupName);
                     }
                 });
             }
@@ -292,6 +301,14 @@ public class CalculatedFieldCtx {
         return map != null && matchesTimeSeries(map, values);
     }
 
+    public boolean dynamicSourceMatches(List<TsKvEntry> values) {
+        return matchesTimeSeries(dynamicEntityArguments, values);
+    }
+
+    public boolean dynamicSourceMatches(List<AttributeKvEntry> values, AttributeScope scope) {
+        return matchesAttributes(dynamicEntityArguments, values, scope);
+    }
+
     private boolean matchesAttributes(Map<ReferencedEntityKey, String> argMap, List<AttributeKvEntry> values, AttributeScope scope) {
         if (argMap.isEmpty() || values.isEmpty()) {
             return false;
@@ -335,6 +352,14 @@ public class CalculatedFieldCtx {
         return matchesTimeSeriesKeys(mainEntityArguments, keys);
     }
 
+    public boolean matchesDynamicSourceKeys(List<String> keys, AttributeScope scope) {
+        return matchesAttributesKeys(dynamicEntityArguments, keys, scope);
+    }
+
+    public boolean matchesDynamicSourceKeys(List<String> keys) {
+        return matchesTimeSeriesKeys(dynamicEntityArguments, keys);
+    }
+
     private boolean matchesAttributesKeys(Map<ReferencedEntityKey, String> argMap, List<String> keys, AttributeScope scope) {
         if (argMap.isEmpty() || keys.isEmpty()) {
             return false;
@@ -381,6 +406,25 @@ public class CalculatedFieldCtx {
         return map != null && matchesTimeSeriesKeys(map, keys);
     }
 
+    public boolean dynamicSourceMatches(CalculatedFieldTelemetryMsgProto proto) {
+        if (!proto.getTsDataList().isEmpty()) {
+            List<TsKvEntry> updatedTelemetry = proto.getTsDataList().stream()
+                    .map(ProtoUtils::fromProto)
+                    .toList();
+            return dynamicSourceMatches(updatedTelemetry);
+        } else if (!proto.getAttrDataList().isEmpty()) {
+            AttributeScope scope = AttributeScope.valueOf(proto.getScope().name());
+            List<AttributeKvEntry> updatedTelemetry = proto.getAttrDataList().stream()
+                    .map(ProtoUtils::fromProto)
+                    .toList();
+            return dynamicSourceMatches(updatedTelemetry, scope);
+        } else if (!proto.getRemovedTsKeysList().isEmpty()) {
+            return matchesDynamicSourceKeys(proto.getRemovedTsKeysList());
+        } else {
+            return matchesDynamicSourceKeys(proto.getRemovedAttrKeysList(), AttributeScope.valueOf(proto.getScope().name()));
+        }
+    }
+
     public boolean linkMatches(EntityId entityId, CalculatedFieldTelemetryMsgProto proto) {
         if (!proto.getTsDataList().isEmpty()) {
             List<TsKvEntry> updatedTelemetry = proto.getTsDataList().stream()
@@ -400,6 +444,18 @@ public class CalculatedFieldCtx {
         }
     }
 
+    public Map<ReferencedEntityKey, String> getLinkedAndDynamicArgs(EntityId entityId) {
+        var argNames = new HashMap<ReferencedEntityKey, String>();
+        var linkedArgNames = linkedEntityArguments.get(entityId);
+        if (linkedArgNames != null && !linkedArgNames.isEmpty()) {
+            argNames.putAll(linkedArgNames);
+        }
+        if (dynamicEntityArguments != null && !dynamicEntityArguments.isEmpty()) {
+            argNames.putAll(dynamicEntityArguments);
+        }
+        return argNames;
+    }
+
     public CalculatedFieldEntityCtxId toCalculatedFieldEntityCtxId() {
         return new CalculatedFieldEntityCtxId(tenantId, cfId, entityId);
     }
@@ -457,6 +513,10 @@ public class CalculatedFieldCtx {
         return "Failed to init CF state. State size exceeds limit of " + (maxStateSize / 1024) + "Kb!";
     }
 
+    public boolean hasCurrentOwnerSourceArguments() {
+        return !dynamicEntityArguments.isEmpty();
+    }
+
     @Override
     public String toString() {
         return "CalculatedFieldCtx{" +

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java

@@ -668,6 +668,7 @@ public class DefaultTbClusterService implements TbClusterService {
             }
             msg.event(ComponentLifecycleEvent.UPDATED)
                     .oldProfileId(old.getDeviceProfileId())
+                    .ownerChanged(!entity.getOwnerId().equals(old.getOwnerId()))
                     .oldName(old.getName());
         }
         broadcast(msg.build());
@@ -688,6 +689,7 @@ public class DefaultTbClusterService implements TbClusterService {
         } else {
             msg.event(ComponentLifecycleEvent.UPDATED)
                     .oldProfileId(old.getAssetProfileId())
+                    .ownerChanged(!entity.getOwnerId().equals(old.getOwnerId()))
                     .oldName(old.getName());
         }
         broadcast(msg.build());

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java

@@ -87,8 +87,7 @@ public class DefaultTbEdgeConsumerService extends AbstractConsumerService<ToEdge
 
     public DefaultTbEdgeConsumerService(TbCoreQueueFactory tbCoreQueueFactory, ActorSystemContext actorContext,
                                         StatsFactory statsFactory, EdgeContextComponent edgeCtx) {
-        super(actorContext, null, null, null, null, null, null,
+        super(actorContext, null, null, null, null, null, null, null, null);
-                null, null);
         this.edgeCtx = edgeCtx;
         this.stats = new EdgeConsumerStats(statsFactory);
         this.queueFactory = tbCoreQueueFactory;

application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java

@@ -175,6 +175,7 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
                     apiUsageStateService.onTenantUpdate(tenantId);
                 } else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.DELETED)) {
                     apiUsageStateService.onTenantDelete(tenantId);
+                    calculatedFieldCache.evictOwner(tenantId);
                     partitionService.removeTenant(tenantId);
                 }
             }
@@ -182,10 +183,24 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
             deviceProfileCache.evict(tenantId, new DeviceProfileId(componentLifecycleMsg.getEntityId().getId()));
         } else if (EntityType.DEVICE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             deviceProfileCache.evict(tenantId, new DeviceId(componentLifecycleMsg.getEntityId().getId()));
+            if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.CREATED)) {
+                calculatedFieldCache.addOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
+            } else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.UPDATED) && componentLifecycleMsg.isOwnerChanged()) {
+                calculatedFieldCache.updateOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
+            } else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.DELETED)) {
+                calculatedFieldCache.evictEntity(componentLifecycleMsg.getEntityId());
+            }
         } else if (EntityType.ASSET_PROFILE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             assetProfileCache.evict(tenantId, new AssetProfileId(componentLifecycleMsg.getEntityId().getId()));
         } else if (EntityType.ASSET.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             assetProfileCache.evict(tenantId, new AssetId(componentLifecycleMsg.getEntityId().getId()));
+            if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.CREATED)) {
+                calculatedFieldCache.addOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
+            } else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.UPDATED) && componentLifecycleMsg.isOwnerChanged()) {
+                calculatedFieldCache.updateOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
+            } else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.DELETED)) {
+                calculatedFieldCache.evictEntity(componentLifecycleMsg.getEntityId());
+            }
         } else if (EntityType.ENTITY_VIEW.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             actorContext.getTbEntityViewService().onComponentLifecycleMsg(componentLifecycleMsg);
         } else if (EntityType.API_USAGE_STATE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
@@ -193,6 +208,7 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
         } else if (EntityType.CUSTOMER.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             if (componentLifecycleMsg.getEvent() == ComponentLifecycleEvent.DELETED) {
                 apiUsageStateService.onCustomerDelete((CustomerId) componentLifecycleMsg.getEntityId());
+                calculatedFieldCache.evictOwner(componentLifecycleMsg.getEntityId());
             }
         } else if (EntityType.CALCULATED_FIELD.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             if (componentLifecycleMsg.getEvent() == ComponentLifecycleEvent.CREATED) {

application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java

@@ -22,6 +22,8 @@ import org.apache.commons.lang3.math.NumberUtils;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.cf.configuration.Argument;
 import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BooleanDataEntry;
 import org.thingsboard.server.common.data.kv.DoubleDataEntry;
 import org.thingsboard.server.common.data.kv.KvEntry;
@@ -66,6 +68,11 @@ public class CalculatedFieldArgumentUtils {
         return new StringDataEntry(key, defaultValue);
     }
 
+    public static AttributeKvEntry createDefaultAttributeEntry(Argument argument, long ts) {
+        KvEntry kvEntry = createDefaultKvEntry(argument);
+        return new BaseAttributeKvEntry(kvEntry, ts, 0L);
+    }
+
     public static CalculatedFieldState createStateByType(CalculatedFieldCtx ctx, EntityId entityId) {
         return switch (ctx.getCfType()) {
             case SIMPLE -> new SimpleCalculatedFieldState(entityId);

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

@@ -25,6 +25,7 @@ import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.rule.engine.action.TbAlarmResult;
 import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.common.data.AttributeScope;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmSeverity;
@@ -40,6 +41,7 @@ import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.cf.configuration.AlarmCalculatedFieldConfiguration;
 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.CurrentCustomerDynamicSourceConfiguration;
 import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey;
 import org.thingsboard.server.common.data.debug.DebugSettings;
 import org.thingsboard.server.common.data.event.CalculatedFieldDebugEvent;
@@ -74,13 +76,14 @@ public class AlarmRulesTest extends AbstractControllerTest {
     @Autowired
     private EventDao eventDao;
 
+    private Device device;
     private DeviceId deviceId;
     private EventId latestEventId;
 
     @Before
     public void beforeEach() throws Exception {
         loginTenantAdmin();
-        Device device = createDevice("Device A", "aaa");
+        device = createDevice("Device A", "aaa");
         deviceId = device.getId();
     }
 
@@ -207,6 +210,40 @@ public class AlarmRulesTest extends AbstractControllerTest {
         });
     }
 
+    @Test
+    public void testCreateAlarm_currentOwnerArgument() throws Exception {
+        Argument temperatureArgument = new Argument();
+        temperatureArgument.setRefEntityKey(new ReferencedEntityKey("temperature", ArgumentType.TS_LATEST, null));
+        temperatureArgument.setDefaultValue("0");
+
+        Argument temperatureThresholdArgument = new Argument();
+        temperatureThresholdArgument.setRefEntityKey(new ReferencedEntityKey("temperatureThreshold", ArgumentType.ATTRIBUTE, AttributeScope.SERVER_SCOPE));
+        temperatureThresholdArgument.setRefDynamicSourceConfiguration(new CurrentCustomerDynamicSourceConfiguration());
+        temperatureThresholdArgument.setDefaultValue("1000");
+
+        Map<String, Argument> arguments = Map.of(
+                "temperature", temperatureArgument,
+                "temperatureThreshold", temperatureThresholdArgument
+        );
+
+        Map<AlarmSeverity, Condition> createRules = Map.of(
+                AlarmSeverity.CRITICAL, new Condition("return temperature >= temperatureThreshold;", null, null)
+        );
+
+        device.setCustomerId(customerId);
+        device = doPost("/api/device", device, Device.class);
+        CalculatedField calculatedField = createAlarmCf(deviceId, "High Temperature Alarm",
+                arguments, createRules, null);
+        postAttributes(customerId, AttributeScope.SERVER_SCOPE, "{\"temperatureThreshold\":50}");
+
+        postTelemetry(deviceId, "{\"temperature\":51}");
+        checkAlarmResult(calculatedField, alarmResult -> {
+            assertThat(alarmResult.isCreated()).isTrue();
+            assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.CRITICAL);
+            assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.ACTIVE_UNACK);
+        });
+    }
+
     private void checkAlarmResult(CalculatedField calculatedField, Consumer<TbAlarmResult> assertion) {
         await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> {
             TbAlarmResult alarmResult = getLatestAlarmResult(calculatedField.getId());

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

@@ -19,7 +19,6 @@ import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import org.junit.Test;
-import org.junit.jupiter.api.BeforeEach;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.AttributeScope;
 import org.thingsboard.server.common.data.DataConstants;
@@ -67,11 +66,6 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
     public static final int TIMEOUT = 60;
     public static final int POLL_INTERVAL = 1;
 
-    @BeforeEach
-    void setUp() throws Exception {
-        loginTenantAdmin();
-    }
-
     @Test
     public void testSimpleCalculatedFieldWhenAllTelemetryPresent() throws Exception {
         Device testDevice = createDevice("Test device", "1234567890");

application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java

@@ -77,6 +77,7 @@ import org.thingsboard.server.actors.device.DeviceActorMessageProcessor;
 import org.thingsboard.server.actors.device.SessionInfo;
 import org.thingsboard.server.actors.device.ToDeviceRpcRequestMetadata;
 import org.thingsboard.server.actors.service.DefaultActorService;
+import org.thingsboard.server.common.data.AttributeScope;
 import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.Device;
@@ -1317,8 +1318,13 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
     }
 
     protected void postTelemetry(EntityId entityId, String payload) throws Exception {
-        doPost("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() +
+        doPostAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() +
-               "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(payload)).andExpect(status().isOk());
+                    "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(payload), 30_000L).andExpect(status().isOk());
+    }
+
+    protected void postAttributes(EntityId entityId, AttributeScope scope, String payload) throws Exception {
+        doPostAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() +
+                    "/attributes/" + scope, JacksonUtil.toJsonNode(payload), 30_000L).andExpect(status().isOk());
     }
 
     protected CalculatedField saveCalculatedField(CalculatedField calculatedField) {

common/data/src/main/java/org/thingsboard/server/common/data/Device.java

@@ -28,6 +28,7 @@ import org.thingsboard.server.common.data.device.data.DeviceData;
 import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.DeviceProfileId;
+import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.OtaPackageId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.validation.Length;
@@ -142,6 +143,11 @@ public class Device extends BaseDataWithAdditionalInfo<DeviceId> implements HasL
         this.customerId = customerId;
     }
 
+    @JsonIgnore
+    public EntityId getOwnerId() {
+        return customerId != null && !customerId.isNullUid() ? customerId : tenantId;
+    }
+
     @Schema(requiredMode = Schema.RequiredMode.REQUIRED, description = "Unique Device Name in scope of Tenant", example = "A4B72CCDFF33")
     @Override
     public String getName() {

common/data/src/main/java/org/thingsboard/server/common/data/ProfileEntityIdInfo.java

@@ -34,21 +34,23 @@ public class ProfileEntityIdInfo implements Serializable, HasTenantId {
     private static final long serialVersionUID = 8532058281983868003L;
 
     private final TenantId tenantId;
+    private final EntityId ownerId;
     private final EntityId profileId;
     private final EntityId entityId;
 
-    private ProfileEntityIdInfo(UUID tenantId, EntityId profileId, EntityId entityId) {
+    private ProfileEntityIdInfo(UUID tenantId, EntityId ownerId, EntityId profileId, EntityId entityId) {
         this.tenantId = TenantId.fromUUID(tenantId);
+        this.ownerId = ownerId;
         this.profileId = profileId;
         this.entityId = entityId;
     }
 
-    public static ProfileEntityIdInfo create(UUID tenantId, DeviceProfileId profileId, DeviceId entityId) {
+    public static ProfileEntityIdInfo create(UUID tenantId, EntityId ownerId, DeviceProfileId profileId, DeviceId entityId) {
-        return new ProfileEntityIdInfo(tenantId, profileId, entityId);
+        return new ProfileEntityIdInfo(tenantId, ownerId, profileId, entityId);
     }
 
-    public static ProfileEntityIdInfo create(UUID tenantId, AssetProfileId profileId, AssetId entityId) {
+    public static ProfileEntityIdInfo create(UUID tenantId, EntityId ownerId, AssetProfileId profileId, AssetId entityId) {
-        return new ProfileEntityIdInfo(tenantId, profileId, entityId);
+        return new ProfileEntityIdInfo(tenantId, ownerId, profileId, entityId);
     }
 
 }

common/data/src/main/java/org/thingsboard/server/common/data/asset/Asset.java

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.common.data.asset;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.databind.JsonNode;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.EqualsAndHashCode;
@@ -29,6 +30,7 @@ import org.thingsboard.server.common.data.HasVersion;
 import org.thingsboard.server.common.data.id.AssetId;
 import org.thingsboard.server.common.data.id.AssetProfileId;
 import org.thingsboard.server.common.data.id.CustomerId;
+import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.validation.Length;
 import org.thingsboard.server.common.data.validation.NoXss;
@@ -125,6 +127,11 @@ public class Asset extends BaseDataWithAdditionalInfo<AssetId> implements HasLab
         this.customerId = customerId;
     }
 
+    @JsonIgnore
+    public EntityId getOwnerId() {
+        return customerId != null && !customerId.isNullUid() ? customerId : tenantId;
+    }
+
     @Schema(requiredMode = Schema.RequiredMode.REQUIRED, description = "Unique Asset Name in scope of Tenant", example = "Empire State Building")
     @Override
     public String getName() {

common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java

@@ -37,4 +37,12 @@ public class Argument {
         return refDynamicSourceConfiguration != null;
     }
 
+    public boolean hasRelationQuerySource() {
+        return hasDynamicSource() && CFArgumentDynamicSourceType.RELATION_QUERY.equals(refDynamicSourceConfiguration.getType());
+    }
+
+    public boolean hasOwnerSource() {
+        return hasDynamicSource() && refDynamicSourceConfiguration.getType() == CFArgumentDynamicSourceType.CURRENT_CUSTOMER;
+    }
+
 }

common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/BaseCalculatedFieldConfiguration.java

@@ -31,8 +31,8 @@ public abstract class BaseCalculatedFieldConfiguration implements ExpressionBase
         if (arguments.containsKey("ctx")) {
             throw new IllegalArgumentException("Argument name 'ctx' is reserved and cannot be used.");
         }
-        if (arguments.values().stream().anyMatch(Argument::hasDynamicSource)) {
+        if (arguments.values().stream().anyMatch(Argument::hasRelationQuerySource)) {
-            throw new IllegalArgumentException("Calculated field with type: '" + getType() + "' doesn't support dynamic source configuration!");
+            throw new IllegalArgumentException("Calculated field with type: '" + getType() + "' doesn't support relation query source configuration!");
         }
     }
 

common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/geofencing/GeofencingCalculatedFieldConfiguration.java

@@ -63,7 +63,7 @@ public class GeofencingCalculatedFieldConfiguration implements ArgumentsBasedCal
 
     @Override
     public boolean isScheduledUpdateEnabled() {
-        return scheduledUpdateInterval > 0 && zoneGroups.values().stream().anyMatch(ZoneGroupConfiguration::hasDynamicSource);
+        return scheduledUpdateInterval > 0 && zoneGroups.values().stream().anyMatch(ZoneGroupConfiguration::hasRelationQuerySource);
     }
 
     @Override

common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/geofencing/ZoneGroupConfiguration.java

@@ -54,7 +54,7 @@ public class ZoneGroupConfiguration {
         if (reportStrategy == null) {
             throw new IllegalArgumentException("Report strategy must be specified for '" + name + "' zone group!");
         }
-        if (hasDynamicSource()) {
+        if (refDynamicSourceConfiguration != null) {
             refDynamicSourceConfiguration.validate();
         }
         if (!createRelationsWithMatchedZones) {
@@ -68,8 +68,12 @@ public class ZoneGroupConfiguration {
         }
     }
 
-    public boolean hasDynamicSource() {
+    public boolean hasRelationQuerySource() {
-        return refDynamicSourceConfiguration != null;
+        return toArgument().hasRelationQuerySource();
+    }
+
+    public boolean hasCurrentOwnerSource() {
+        return toArgument().hasOwnerSource();
     }
 
     @JsonIgnore
@@ -92,4 +96,5 @@ public class ZoneGroupConfiguration {
         argument.setRefEntityKey(new ReferencedEntityKey(perimeterKeyName, ArgumentType.ATTRIBUTE, AttributeScope.SERVER_SCOPE));
         return argument;
     }
+
 }

common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/ArgumentTest.java

@@ -29,10 +29,21 @@ public class ArgumentTest {
     }
 
     @Test
-    void validateShouldReturnTrueIfDynamicSourceConfigurationIsNotNull() {
+    void validateWhenRelationQuerySourceConfigurationIsNotNull() {
         var argument = new Argument();
         argument.setRefDynamicSourceConfiguration(new RelationQueryDynamicSourceConfiguration());
         assertThat(argument.hasDynamicSource()).isTrue();
+        assertThat(argument.hasRelationQuerySource()).isTrue();
+        assertThat(argument.hasOwnerSource()).isFalse();
+    }
+
+    @Test
+    void validateWhenCurrentCustomerSourceConfigurationIsNotNull() {
+        var argument = new Argument();
+        argument.setRefDynamicSourceConfiguration(new CurrentCustomerDynamicSourceConfiguration());
+        assertThat(argument.hasDynamicSource()).isTrue();
+        assertThat(argument.hasOwnerSource()).isTrue();
+        assertThat(argument.hasRelationQuerySource()).isFalse();
     }
 
 }

common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/geofencing/GeofencingCalculatedFieldConfigurationTest.java

@@ -24,14 +24,12 @@ 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.ReferencedEntityKey;
 
-import java.util.List;
 import java.util.Map;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.thingsboard.server.common.data.cf.configuration.geofencing.EntityCoordinates.ENTITY_ID_LATITUDE_ARGUMENT_KEY;
@@ -112,7 +110,7 @@ public class GeofencingCalculatedFieldConfigurationTest {
     void scheduledUpdateDisabledWhenIntervalIsGreaterThanZeroButNoZonesWithDynamicArguments() {
         var cfg = new GeofencingCalculatedFieldConfiguration();
         var zoneGroupConfigurationMock = mock(ZoneGroupConfiguration.class);
-        when(zoneGroupConfigurationMock.hasDynamicSource()).thenReturn(false);
+        when(zoneGroupConfigurationMock.hasRelationQuerySource()).thenReturn(false);
         cfg.setZoneGroups(Map.of("someGroupName", zoneGroupConfigurationMock));
         cfg.setScheduledUpdateInterval(60);
         assertThat(cfg.isScheduledUpdateEnabled()).isFalse();
@@ -122,7 +120,7 @@ public class GeofencingCalculatedFieldConfigurationTest {
     void scheduledUpdateEnabledWhenIntervalIsGreaterThanZeroAndDynamicArgumentsPresent() {
         var cfg = new GeofencingCalculatedFieldConfiguration();
         var zoneGroupConfigurationMock = mock(ZoneGroupConfiguration.class);
-        when(zoneGroupConfigurationMock.hasDynamicSource()).thenReturn(true);
+        when(zoneGroupConfigurationMock.hasRelationQuerySource()).thenReturn(true);
         cfg.setZoneGroups(Map.of("someGroupName", zoneGroupConfigurationMock));
         cfg.setScheduledUpdateInterval(60);
         assertThat(cfg.isScheduledUpdateEnabled()).isTrue();

common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/geofencing/ZoneGroupConfigurationTest.java

@@ -22,6 +22,7 @@ import org.junit.jupiter.params.provider.ValueSource;
 import org.thingsboard.server.common.data.AttributeScope;
 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.CurrentCustomerDynamicSourceConfiguration;
 import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey;
 import org.thingsboard.server.common.data.cf.configuration.RelationQueryDynamicSourceConfiguration;
 import org.thingsboard.server.common.data.relation.EntityRelation;
@@ -98,19 +99,25 @@ public class ZoneGroupConfigurationTest {
     }
 
     @Test
-    void whenHasDynamicSourceCalled_shouldReturnTrueIfDynamicSourceConfigurationIsNotNull() {
+    void whenHasRelationQuerySourceCalled_shouldReturnTrueIfRelationQuerySourceConfigurationIsNotNull() {
         var zoneGroupConfiguration = new ZoneGroupConfiguration("perimeter", REPORT_TRANSITION_EVENTS_AND_PRESENCE_STATUS, false);
         zoneGroupConfiguration.setRefDynamicSourceConfiguration(new RelationQueryDynamicSourceConfiguration());
-        assertThat(zoneGroupConfiguration.hasDynamicSource()).isTrue();
+        assertThat(zoneGroupConfiguration.hasRelationQuerySource()).isTrue();
     }
 
     @Test
-    void whenHasDynamicSourceCalled_shouldReturnTrueIfDynamicSourceConfigurationIsNull() {
+    void whenHasRelationQuerySourceCalled_shouldReturnFalseIfRelationQuerySourceConfigurationIsNull() {
         var zoneGroupConfiguration = mock(ZoneGroupConfiguration.class);
         assertThat(zoneGroupConfiguration.getRefDynamicSourceConfiguration()).isNull();
-        assertThat(zoneGroupConfiguration.hasDynamicSource()).isFalse();
+        assertThat(zoneGroupConfiguration.hasRelationQuerySource()).isFalse();
     }
 
+    @Test
+    void whenHasRelationQuerySourceCalled_shouldReturnFalseIfCurrentCustomerSourceConfigured() {
+        var zoneGroupConfiguration = mock(ZoneGroupConfiguration.class);
+        zoneGroupConfiguration.setRefDynamicSourceConfiguration(new CurrentCustomerDynamicSourceConfiguration());
+        assertThat(zoneGroupConfiguration.hasRelationQuerySource()).isFalse();
+    }
 
     @Test
     void validateToArgumentsMethodCallWithoutRefEntityId() {

common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java

@@ -151,6 +151,7 @@ public enum MsgType {
     CF_ENTITY_INIT_CF_MSG,
     CF_ENTITY_DELETE_MSG,
 
+    CF_ARGUMENT_RESET_MSG, // Sent to reset argument;
     CF_DYNAMIC_ARGUMENTS_REFRESH_MSG,
     CF_ENTITY_DYNAMIC_ARGUMENTS_REFRESH_MSG,
     CF_REEVALUATE_MSG;

common/message/src/main/java/org/thingsboard/server/common/msg/plugin/ComponentLifecycleMsg.java

@@ -46,14 +46,15 @@ public class ComponentLifecycleMsg implements TenantAwareMsg, ToAllNodesMsg {
     private final String name;
     private final EntityId oldProfileId;
     private final EntityId profileId;
+    private final boolean ownerChanged;
     private final JsonNode info;
 
     public ComponentLifecycleMsg(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent event) {
-        this(tenantId, entityId, event, null, null, null, null, null);
+        this(tenantId, entityId, event, null, null, null, null, false, null);
     }
 
     @Builder
-    private ComponentLifecycleMsg(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent event, String oldName, String name, EntityId oldProfileId, EntityId profileId, JsonNode info) {
+    private ComponentLifecycleMsg(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent event, String oldName, String name, EntityId oldProfileId, EntityId profileId, boolean ownerChanged, JsonNode info) {
         this.tenantId = tenantId;
         this.entityId = entityId;
         this.event = event;
@@ -61,6 +62,7 @@ public class ComponentLifecycleMsg implements TenantAwareMsg, ToAllNodesMsg {
         this.name = name;
         this.oldProfileId = oldProfileId;
         this.profileId = profileId;
+        this.ownerChanged = ownerChanged;
         this.info = info;
     }
 

common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java

@@ -129,6 +129,7 @@ public class ProtoUtils {
             builder.setOldProfileIdMSB(msg.getOldProfileId().getId().getMostSignificantBits());
             builder.setOldProfileIdLSB(msg.getOldProfileId().getId().getLeastSignificantBits());
         }
+        builder.setOwnerChanged(msg.isOwnerChanged());
         if (msg.getName() != null) {
             builder.setName(msg.getName());
         }
@@ -165,6 +166,7 @@ public class ProtoUtils {
             var profileType = EntityType.DEVICE.equals(entityId.getEntityType()) ? EntityType.DEVICE_PROFILE : EntityType.ASSET_PROFILE;
             builder.oldProfileId(EntityIdFactory.getByTypeAndUuid(profileType, new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB())));
         }
+        builder.ownerChanged(proto.getOwnerChanged());
         if (proto.hasInfo()) {
             builder.info(JacksonUtil.toJsonNode(proto.getInfo()));
         }

common/proto/src/main/proto/queue.proto

@@ -1291,6 +1291,7 @@ message ComponentLifecycleMsgProto {
   int64 profileIdMSB = 11;
   int64 profileIdLSB = 12;
   optional string info = 13;
+  bool ownerChanged = 100;
 }
 
 message EdgeEventMsgProto {

dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java

@@ -100,7 +100,7 @@ public class CalculatedFieldDataValidator extends DataValidator<CalculatedField>
         }
         Map<String, RelationQueryDynamicSourceConfiguration> relationQueryBasedArguments = argumentsBasedCfg.getArguments().entrySet()
                 .stream()
-                .filter(entry -> entry.getValue().hasDynamicSource())
+                .filter(entry -> entry.getValue().hasRelationQuerySource())
                 .collect(Collectors.toMap(Map.Entry::getKey, entry -> (RelationQueryDynamicSourceConfiguration) entry.getValue().getRefDynamicSourceConfiguration()));
         if (relationQueryBasedArguments.isEmpty()) {
             return;

dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeAssetRepository.java

@@ -23,9 +23,12 @@ import org.springframework.transaction.support.TransactionTemplate;
 import org.thingsboard.server.common.data.ProfileEntityIdInfo;
 import org.thingsboard.server.common.data.id.AssetId;
 import org.thingsboard.server.common.data.id.AssetProfileId;
+import org.thingsboard.server.common.data.id.CustomerId;
+import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 
+import java.util.Map;
 import java.util.UUID;
 
 @Repository
@@ -40,23 +43,24 @@ public class DefaultNativeAssetRepository extends AbstractNativeRepository imple
 
     @Override
     public PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(Pageable pageable) {
-        String PROFILE_ASSET_ID_INFO_QUERY = "SELECT tenant_id as tenantId, asset_profile_id as profileId, id as id FROM asset ORDER BY created_time ASC LIMIT %s OFFSET %s";
+        String PROFILE_ASSET_ID_INFO_QUERY = "SELECT tenant_id as tenantId, customer_id as customerId, asset_profile_id as profileId, id as id FROM asset ORDER BY created_time ASC LIMIT %s OFFSET %s";
-        return find(COUNT_QUERY, PROFILE_ASSET_ID_INFO_QUERY, pageable, row -> {
+        return find(COUNT_QUERY, PROFILE_ASSET_ID_INFO_QUERY, pageable, DefaultNativeAssetRepository::toInfo);
-            AssetId id = new AssetId((UUID) row.get("id"));
-            AssetProfileId profileId = new AssetProfileId((UUID) row.get("profileId"));
-            var tenantIdObj = row.get("tenantId");
-            return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id);
-        });
     }
 
     @Override
     public PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(UUID tenantId, Pageable pageable) {
-        String PROFILE_ASSET_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, asset_profile_id as profileId, id as id FROM asset WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId);
+        String PROFILE_ASSET_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, customer_id as customerId, asset_profile_id as profileId, id as id FROM asset WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId);
-        return find(COUNT_QUERY, PROFILE_ASSET_ID_INFO_QUERY, pageable, row -> {
+        return find(COUNT_QUERY, PROFILE_ASSET_ID_INFO_QUERY, pageable, DefaultNativeAssetRepository::toInfo);
-            AssetId id = new AssetId((UUID) row.get("id"));
-            AssetProfileId profileId = new AssetProfileId((UUID) row.get("profileId"));
-            var tenantIdObj = row.get("tenantId");
-            return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id);
-        });
     }
+
+    private static ProfileEntityIdInfo toInfo(Map<String, Object> row) {
+        var tenantIdObj = row.get("tenantId");
+        UUID tenantId = tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId();
+        AssetId id = new AssetId((UUID) row.get("id"));
+        CustomerId customerId = new CustomerId((UUID) row.get("customerId"));
+        EntityId ownerId = !customerId.isNullUid() ? customerId : TenantId.fromUUID(tenantId);
+        AssetProfileId profileId = new AssetProfileId((UUID) row.get("profileId"));
+        return ProfileEntityIdInfo.create(tenantId, ownerId, profileId, id);
+    }
+
 }

dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeDeviceRepository.java

@@ -22,11 +22,14 @@ import org.springframework.stereotype.Repository;
 import org.springframework.transaction.support.TransactionTemplate;
 import org.thingsboard.server.common.data.DeviceIdInfo;
 import org.thingsboard.server.common.data.ProfileEntityIdInfo;
+import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.DeviceProfileId;
+import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 
+import java.util.Map;
 import java.util.UUID;
 
 @Repository
@@ -52,24 +55,24 @@ public class DefaultNativeDeviceRepository extends AbstractNativeRepository impl
 
     @Override
     public PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(Pageable pageable) {
-        String PROFILE_DEVICE_ID_INFO_QUERY = "SELECT tenant_id as tenantId, device_profile_id as profileId, id as id FROM device ORDER BY created_time ASC LIMIT %s OFFSET %s";
+        String PROFILE_DEVICE_ID_INFO_QUERY = "SELECT tenant_id as tenantId, customer_id as customerId, device_profile_id as profileId, id as id FROM device ORDER BY created_time ASC LIMIT %s OFFSET %s";
-        return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, row -> {
+        return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, DefaultNativeDeviceRepository::toInfo);
-            DeviceId id = new DeviceId((UUID) row.get("id"));
-            DeviceProfileId profileId = new DeviceProfileId((UUID) row.get("profileId"));
-            var tenantIdObj = row.get("tenantId");
-            return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id);
-        });
     }
 
     @Override
     public PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(UUID tenantId, Pageable pageable) {
-        String PROFILE_DEVICE_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, device_profile_id as profileId, id as id FROM device WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId);
+        String PROFILE_DEVICE_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, customer_id as customerId, device_profile_id as profileId, id as id FROM device WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId);
-        return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, row -> {
+        return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, DefaultNativeDeviceRepository::toInfo);
-            DeviceId id = new DeviceId((UUID) row.get("id"));
+    }
-            DeviceProfileId profileId = new DeviceProfileId((UUID) row.get("profileId"));
+
-            var tenantIdObj = row.get("tenantId");
+    private static ProfileEntityIdInfo toInfo(Map<String, Object> row) {
-            return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id);
+        var tenantIdObj = row.get("tenantId");
-        });
+        UUID tenantId = tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId();
+        DeviceId id = new DeviceId((UUID) row.get("id"));
+        CustomerId customerId = new CustomerId((UUID) row.get("customerId"));
+        EntityId ownerId = !customerId.isNullUid() ? customerId : TenantId.fromUUID(tenantId);
+        DeviceProfileId profileId = new DeviceProfileId((UUID) row.get("profileId"));
+        return ProfileEntityIdInfo.create(tenantId, ownerId, profileId, id);
     }
 
 }