/**
 * Copyright © 2016-2026 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.dao.relation;

import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.event.TransactionalEventListener;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.CollectionUtils;
import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.server.cache.TbTransactionalCache;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.audit.ActionType;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UUIDBased;
import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.data.relation.EntityRelationInfo;
import org.thingsboard.server.common.data.relation.EntityRelationPathQuery;
import org.thingsboard.server.common.data.relation.EntityRelationsQuery;
import org.thingsboard.server.common.data.relation.EntitySearchDirection;
import org.thingsboard.server.common.data.relation.RelationEntityTypeFilter;
import org.thingsboard.server.common.data.relation.RelationPathLevel;
import org.thingsboard.server.common.data.relation.RelationTypeGroup;
import org.thingsboard.server.common.data.relation.RelationsSearchParameters;
import org.thingsboard.server.common.data.rule.RuleChainType;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
import org.thingsboard.server.dao.entity.EntityService;
import org.thingsboard.server.dao.eventsourcing.RelationActionEvent;
import org.thingsboard.server.dao.service.ConstraintValidator;
import org.thingsboard.server.dao.sql.JpaExecutorService;
import org.thingsboard.server.dao.sql.relation.JpaRelationQueryExecutorService;
import org.thingsboard.server.dao.usagerecord.ApiLimitService;
import org.thingsboard.server.exception.DataValidationException;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Predicate;

import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static org.thingsboard.server.dao.service.Validator.validateId;
import static org.thingsboard.server.dao.service.Validator.validatePositiveNumber;

@Slf4j
@Service
class BaseRelationService implements RelationService {

    private final RelationDao relationDao;
    private final EntityService entityService;
    private final TbTransactionalCache<RelationCacheKey, RelationCacheValue> cache;
    private final ApplicationEventPublisher eventPublisher;
    private final JpaExecutorService executor;
    private final JpaRelationQueryExecutorService relationsExecutor;
    private final ApiLimitService apiLimitService;

    private ScheduledExecutorService timeoutExecutorService;

    @Value("${sql.relations.query_timeout:20}")
    private Integer relationQueryTimeout;

    public BaseRelationService(RelationDao relationDao, @Lazy EntityService entityService,
                               TbTransactionalCache<RelationCacheKey, RelationCacheValue> cache,
                               ApplicationEventPublisher eventPublisher, JpaExecutorService executor,
                               JpaRelationQueryExecutorService relationsExecutor, ApiLimitService apiLimitService) {
        this.relationDao = relationDao;
        this.entityService = entityService;
        this.cache = cache;
        this.eventPublisher = eventPublisher;
        this.executor = executor;
        this.relationsExecutor = relationsExecutor;
        this.apiLimitService = apiLimitService;
    }

    @PostConstruct
    public void init() {
        timeoutExecutorService = ThingsBoardExecutors.newSingleThreadScheduledExecutor("relations-query-timeout");
    }

    @PreDestroy
    public void destroy() {
        if (timeoutExecutorService != null) {
            timeoutExecutorService.shutdownNow();
        }
    }

    @TransactionalEventListener(classes = EntityRelationEvent.class)
    public void handleEvictEvent(EntityRelationEvent event) {
        List<RelationCacheKey> keys = new ArrayList<>(5);
        keys.add(new RelationCacheKey(event.getFrom(), event.getTo(), event.getType(), event.getTypeGroup()));
        keys.add(new RelationCacheKey(event.getFrom(), null, event.getType(), event.getTypeGroup(), EntitySearchDirection.FROM));
        keys.add(new RelationCacheKey(event.getFrom(), null, null, event.getTypeGroup(), EntitySearchDirection.FROM));
        keys.add(new RelationCacheKey(null, event.getTo(), event.getType(), event.getTypeGroup(), EntitySearchDirection.TO));
        keys.add(new RelationCacheKey(null, event.getTo(), null, event.getTypeGroup(), EntitySearchDirection.TO));
        cache.evict(keys);
        log.debug("Processed evict event: {}", event);
    }

    @Override
    public ListenableFuture<Boolean> checkRelationAsync(TenantId tenantId, EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) {
        log.trace("Executing checkRelationAsync [{}][{}][{}][{}]", from, to, relationType, typeGroup);
        validate(from, to, relationType, typeGroup);
        return relationDao.checkRelationAsync(tenantId, from, to, relationType, typeGroup);
    }

    @Override
    public boolean checkRelation(TenantId tenantId, EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) {
        log.trace("Executing checkRelation [{}][{}][{}][{}]", from, to, relationType, typeGroup);
        validate(from, to, relationType, typeGroup);
        return relationDao.checkRelation(tenantId, from, to, relationType, typeGroup);
    }

    @Override
    public EntityRelation getRelation(TenantId tenantId, EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) {
        log.trace("Executing EntityRelation [{}][{}][{}][{}]", from, to, relationType, typeGroup);
        validate(from, to, relationType, typeGroup);
        RelationCacheKey cacheKey = new RelationCacheKey(from, to, relationType, typeGroup);
        return cache.getAndPutInTransaction(cacheKey,
                () -> {
                    log.trace("FETCH EntityRelation [{}][{}][{}][{}]", from, to, relationType, typeGroup);
                    return relationDao.getRelation(tenantId, from, to, relationType, typeGroup);
                },
                RelationCacheValue::getRelation,
                relation -> RelationCacheValue.builder().relation(relation).build(), false);
    }

    @Override
    public EntityRelation saveRelation(TenantId tenantId, EntityRelation relation) {
        log.trace("Executing saveRelation [{}]", relation);
        validate(relation);
        var result = relationDao.saveRelation(tenantId, relation);
        publishEvictEvent(EntityRelationEvent.from(result));
        eventPublisher.publishEvent(new RelationActionEvent(tenantId, result, ActionType.RELATION_ADD_OR_UPDATE));
        return result;
    }

    @Override
    public void saveRelations(TenantId tenantId, List<EntityRelation> relations) {
        log.trace("Executing saveRelations [{}]", relations);
        for (EntityRelation relation : relations) {
            validate(relation);
        }
        List<EntityRelation> savedRelations = new ArrayList<>(relations.size());
        for (List<EntityRelation> partition : Lists.partition(relations, 1024)) {
            savedRelations.addAll(relationDao.saveRelations(tenantId, partition));
        }
        for (EntityRelation relation : savedRelations) {
            publishEvictEvent(EntityRelationEvent.from(relation));
            eventPublisher.publishEvent(new RelationActionEvent(tenantId, relation, ActionType.RELATION_ADD_OR_UPDATE));
        }
    }

    @Override
    public ListenableFuture<Boolean> saveRelationAsync(TenantId tenantId, EntityRelation relation) {
        log.trace("Executing saveRelationAsync [{}]", relation);
        try {
            validate(relation);
        } catch (DataValidationException e) {
            return Futures.immediateFailedFuture(e);
        }
        var future = relationDao.saveRelationAsync(tenantId, relation);
        return Futures.transform(future, savedRelation -> {
            if (savedRelation != null) {
                handleEvictEvent(EntityRelationEvent.from(savedRelation));
                eventPublisher.publishEvent(new RelationActionEvent(tenantId, savedRelation, ActionType.RELATION_ADD_OR_UPDATE));
            }
            return savedRelation != null;
        }, directExecutor());
    }

    @Override
    public boolean deleteRelation(TenantId tenantId, EntityRelation relation) {
        log.trace("Executing DeleteRelation [{}]", relation);
        validate(relation);
        var result = relationDao.deleteRelation(tenantId, relation);
        if (result != null) {
            publishEvictEvent(EntityRelationEvent.from(result));
            eventPublisher.publishEvent(new RelationActionEvent(tenantId, result, ActionType.RELATION_DELETED));
        }
        return result != null;
    }

    @Override
    public ListenableFuture<Boolean> deleteRelationAsync(TenantId tenantId, EntityRelation relation) {
        log.trace("Executing deleteRelationAsync [{}]", relation);
        try {
            validate(relation);
        } catch (DataValidationException e) {
            return Futures.immediateFailedFuture(e);
        }
        var future = relationDao.deleteRelationAsync(tenantId, relation);
        return Futures.transform(future, deletedRelation -> {
            if (deletedRelation != null) {
                handleEvictEvent(EntityRelationEvent.from(deletedRelation));
                eventPublisher.publishEvent(new RelationActionEvent(tenantId, deletedRelation, ActionType.RELATION_DELETED));
            }
            return deletedRelation != null;
        }, directExecutor());
    }

    @Override
    public EntityRelation deleteRelation(TenantId tenantId, EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) {
        log.trace("Executing deleteRelation [{}][{}][{}][{}]", from, to, relationType, typeGroup);
        validate(from, to, relationType, typeGroup);
        var result = relationDao.deleteRelation(tenantId, from, to, relationType, typeGroup);
        if (result != null) {
            publishEvictEvent(EntityRelationEvent.from(result));
            eventPublisher.publishEvent(new RelationActionEvent(tenantId, result, ActionType.RELATION_DELETED));
        }
        return result;
    }

    @Override
    public ListenableFuture<Boolean> deleteRelationAsync(TenantId tenantId, EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) {
        log.trace("Executing deleteRelationAsync [{}][{}][{}][{}]", from, to, relationType, typeGroup);
        validate(from, to, relationType, typeGroup);
        var future = relationDao.deleteRelationAsync(tenantId, from, to, relationType, typeGroup);
        return Futures.transform(future, deletedEvent -> {
            if (deletedEvent != null) {
                handleEvictEvent(EntityRelationEvent.from(deletedEvent));
                eventPublisher.publishEvent(new RelationActionEvent(tenantId, deletedEvent, ActionType.RELATION_DELETED));
            }
            return deletedEvent != null;
        }, directExecutor());
    }

    @Transactional
    @Override
    public void deleteEntityCommonRelations(TenantId tenantId, EntityId entityId) {
        deleteEntityRelations(tenantId, entityId, RelationTypeGroup.COMMON);
    }

    @Transactional
    @Override
    public void deleteEntityRelations(TenantId tenantId, EntityId entityId) {
        deleteEntityRelations(tenantId, entityId, null);
    }

    @Transactional
    public void deleteEntityRelations(TenantId tenantId, EntityId entityId, RelationTypeGroup relationTypeGroup) {
        log.trace("Executing deleteEntityRelations [{}]", entityId);
        validate(entityId);

        List<EntityRelation> inboundRelations;
        if (relationTypeGroup == null) {
            inboundRelations = relationDao.deleteInboundRelations(tenantId, entityId);
        } else {
            inboundRelations = relationDao.deleteInboundRelations(tenantId, entityId, relationTypeGroup);
        }

        for (EntityRelation relation : inboundRelations) {
            eventPublisher.publishEvent(EntityRelationEvent.from(relation));
            eventPublisher.publishEvent(new RelationActionEvent(tenantId, relation, ActionType.RELATION_DELETED));
        }

        List<EntityRelation> outboundRelations;
        if (relationTypeGroup == null) {
            outboundRelations = relationDao.deleteOutboundRelations(tenantId, entityId);
        } else {
            outboundRelations = relationDao.deleteOutboundRelations(tenantId, entityId, relationTypeGroup);
        }

        for (EntityRelation relation : outboundRelations) {
            eventPublisher.publishEvent(EntityRelationEvent.from(relation));
            eventPublisher.publishEvent(new RelationActionEvent(tenantId, relation, ActionType.RELATION_DELETED));
        }
    }

    @Override
    public List<EntityRelation> findByFrom(TenantId tenantId, EntityId from, RelationTypeGroup typeGroup) {
        validate(from);
        validateTypeGroup(typeGroup);
        RelationCacheKey cacheKey = RelationCacheKey.builder().from(from).typeGroup(typeGroup).direction(EntitySearchDirection.FROM).build();
        return cache.getAndPutInTransaction(cacheKey,
                () -> relationDao.findAllByFrom(tenantId, from, typeGroup),
                RelationCacheValue::getRelations,
                relations -> RelationCacheValue.builder().relations(relations).build(), false);
    }

    @Override
    public ListenableFuture<List<EntityRelation>> findByFromAsync(TenantId tenantId, EntityId from, RelationTypeGroup typeGroup) {
        log.trace("Executing findByFrom [{}][{}]", from, typeGroup);
        validate(from);
        validateTypeGroup(typeGroup);

        var cacheValue = cache.get(RelationCacheKey.builder().from(from).typeGroup(typeGroup).direction(EntitySearchDirection.FROM).build());

        if (cacheValue != null && cacheValue.get() != null) {
            return Futures.immediateFuture(cacheValue.get().getRelations());
        } else {
            //Disabled cache put for the async requests due to limitations of the cache implementation (Redis lib does not support thread-safe transactions)
            return executor.submit(() -> findByFrom(tenantId, from, typeGroup));
        }
    }

    @Override
    public ListenableFuture<List<EntityRelationInfo>> findInfoByFrom(TenantId tenantId, EntityId from, RelationTypeGroup typeGroup) {
        log.trace("Executing findInfoByFrom [{}][{}]", from, typeGroup);
        validate(from);
        validateTypeGroup(typeGroup);
        return Futures.transform(executor.submit(() -> relationDao.findAllByFrom(tenantId, from, typeGroup)),
                relations -> relations.stream()
                        .map(relation -> fetchRelationInfo(tenantId, relation, EntityRelation::getTo, EntityRelationInfo::setToName))
                        .toList(), directExecutor());
    }

    @Override
    public List<EntityRelation> findByFromAndType(TenantId tenantId, EntityId from, String relationType, RelationTypeGroup typeGroup) {
        RelationCacheKey cacheKey = RelationCacheKey.builder().from(from).type(relationType).typeGroup(typeGroup).direction(EntitySearchDirection.FROM).build();
        return cache.getAndPutInTransaction(cacheKey,
                () -> relationDao.findAllByFromAndType(tenantId, from, relationType, typeGroup),
                RelationCacheValue::getRelations,
                relations -> RelationCacheValue.builder().relations(relations).build(), false);
    }

    @Override
    public ListenableFuture<List<EntityRelation>> findByFromAndTypeAsync(TenantId tenantId, EntityId from, String relationType, RelationTypeGroup typeGroup) {
        log.trace("Executing findByFromAndType [{}][{}][{}]", from, relationType, typeGroup);
        validate(from);
        validateType(relationType);
        validateTypeGroup(typeGroup);
        return executor.submit(() -> findByFromAndType(tenantId, from, relationType, typeGroup));
    }

    @Override
    public List<EntityRelation> findByTo(TenantId tenantId, EntityId to, RelationTypeGroup typeGroup) {
        validate(to);
        validateTypeGroup(typeGroup);
        RelationCacheKey cacheKey = RelationCacheKey.builder().to(to).typeGroup(typeGroup).direction(EntitySearchDirection.TO).build();
        return cache.getAndPutInTransaction(cacheKey,
                () -> relationDao.findAllByTo(tenantId, to, typeGroup),
                RelationCacheValue::getRelations,
                relations -> RelationCacheValue.builder().relations(relations).build(), false);

    }

    @Override
    public ListenableFuture<List<EntityRelation>> findByToAsync(TenantId tenantId, EntityId to, RelationTypeGroup typeGroup) {
        log.trace("Executing findByToAsync [{}][{}]", to, typeGroup);
        validate(to);
        validateTypeGroup(typeGroup);
        return executor.submit(() -> findByTo(tenantId, to, typeGroup));
    }

    @Override
    public ListenableFuture<List<EntityRelationInfo>> findInfoByTo(TenantId tenantId, EntityId to, RelationTypeGroup typeGroup) {
        log.trace("Executing findInfoByTo [{}][{}]", to, typeGroup);
        validate(to);
        validateTypeGroup(typeGroup);
        return Futures.transform(findByToAsync(tenantId, to, typeGroup),
                relations -> relations.stream()
                        .map(relation -> fetchRelationInfo(tenantId, relation, EntityRelation::getFrom, EntityRelationInfo::setFromName))
                        .toList(), directExecutor());
    }

    @Override
    public List<EntityRelation> findByToAndType(TenantId tenantId, EntityId to, String relationType, RelationTypeGroup typeGroup) {
        log.trace("Executing findByToAndType [{}][{}][{}]", to, relationType, typeGroup);
        validate(to);
        validateType(relationType);
        validateTypeGroup(typeGroup);
        RelationCacheKey cacheKey = RelationCacheKey.builder().to(to).type(relationType).typeGroup(typeGroup).direction(EntitySearchDirection.TO).build();
        return cache.getAndPutInTransaction(cacheKey,
                () -> relationDao.findAllByToAndType(tenantId, to, relationType, typeGroup),
                RelationCacheValue::getRelations,
                relations -> RelationCacheValue.builder().relations(relations).build(), false);

    }

    @Override
    public ListenableFuture<List<EntityRelation>> findByToAndTypeAsync(TenantId tenantId, EntityId to, String relationType, RelationTypeGroup typeGroup) {
        log.trace("Executing findByToAndTypeAsync [{}][{}][{}]", to, relationType, typeGroup);
        validate(to);
        validateType(relationType);
        validateTypeGroup(typeGroup);
        return executor.submit(() -> findByToAndType(tenantId, to, relationType, typeGroup));
    }

    @Override
    public ListenableFuture<List<EntityRelation>> findByQuery(TenantId tenantId, EntityRelationsQuery query) {
        log.trace("Executing findByQuery [{}]", query);
        RelationsSearchParameters params = query.getParameters();
        final List<RelationEntityTypeFilter> filters = query.getFilters();
        if (filters == null || filters.isEmpty()) {
            log.debug("Filters are not set [{}]", query);
        }

        int maxLvl = params.getMaxLevel() > 0 ? params.getMaxLevel() : Integer.MAX_VALUE;

        try {
            ListenableFuture<Set<EntityRelation>> relationSet = findRelationsRecursively(tenantId, params.getEntityId(), params.getDirection(),
                    params.getRelationTypeGroup(), maxLvl, params.isFetchLastLevelOnly(), new ConcurrentHashMap<>());
            return Futures.transform(relationSet, input -> {
                List<EntityRelation> relations = new ArrayList<>();
                if (filters == null || filters.isEmpty()) {
                    relations.addAll(input);
                    return relations;
                }
                for (EntityRelation relation : input) {
                    if (matchFilters(filters, relation, params.getDirection())) {
                        relations.add(relation);
                    }
                }
                return relations;
            }, directExecutor());
        } catch (Exception e) {
            log.warn("Failed to query relations: [{}]", query, e);
            throw new RuntimeException(e);
        }
    }

    @Override
    public ListenableFuture<List<EntityRelationInfo>> findInfoByQuery(TenantId tenantId, EntityRelationsQuery query) {
        log.trace("Executing findInfoByQuery [{}]", query);

        EntitySearchDirection direction = query.getParameters().getDirection();

        Function<EntityRelation, EntityId> entityIdGetter = relation -> direction == EntitySearchDirection.FROM ? relation.getTo() : relation.getFrom();

        BiConsumer<EntityRelationInfo, String> entityNameSetter = (EntityRelationInfo relationInfo, String entityName) -> {
            if (direction == EntitySearchDirection.FROM) {
                relationInfo.setToName(entityName);
            } else {
                relationInfo.setFromName(entityName);
            }
        };

        return Futures.transform(findByQuery(tenantId, query),
                relations -> relations.stream()
                        .map(relation -> fetchRelationInfo(tenantId, relation, entityIdGetter, entityNameSetter))
                        .toList(), directExecutor());
    }

    private EntityRelationInfo fetchRelationInfo(TenantId tenantId, EntityRelation relation,
                                                 Function<EntityRelation, EntityId> entityIdGetter,
                                                 BiConsumer<EntityRelationInfo, String> entityNameSetter) {
        var relationInfo = new EntityRelationInfo(relation);
        entityNameSetter.accept(relationInfo, entityService.fetchEntityName(tenantId, entityIdGetter.apply(relation)).orElse("N/A"));
        return relationInfo;
    }

    @Override
    public void removeRelations(TenantId tenantId, EntityId entityId) {
        log.trace("removeRelations {}", entityId);

        List<EntityRelation> relations = new ArrayList<>();
        for (RelationTypeGroup relationTypeGroup : RelationTypeGroup.values()) {
            relations.addAll(findByFrom(tenantId, entityId, relationTypeGroup));
            relations.addAll(findByTo(tenantId, entityId, relationTypeGroup));
        }

        for (EntityRelation relation : relations) {
            deleteRelation(tenantId, relation);
        }
    }

    @Override
    public List<EntityRelation> findRuleNodeToRuleChainRelations(TenantId tenantId, RuleChainType ruleChainType, int limit) {
        log.trace("Executing findRuleNodeToRuleChainRelations, tenantId [{}], ruleChainType {} and limit {}", tenantId, ruleChainType, limit);
        validateId(tenantId, id -> "Invalid tenant id: " + id);
        return relationDao.findRuleNodeToRuleChainRelations(ruleChainType, limit);
    }

    @Override
    public ListenableFuture<List<EntityRelation>> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery) {
        return findFilteredRelationsByPathQueryAsync(tenantId, relationPathQuery, null);
    }

    @Override
    public ListenableFuture<List<EntityRelation>> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate<EntityRelation> relationFilter) {
        log.trace("Executing findByRelationPathQuery, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery);
        validateId(tenantId, id -> "Invalid tenant id: " + id);
        validate(relationPathQuery);
        int limit = (int) apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxRelatedEntitiesToReturnPerCfArgument);
        validatePositiveNumber(limit, "Max related entities limit for relation path query must be positive!");
        if (relationPathQuery.levels().size() == 1) {
            RelationPathLevel relationPathLevel = relationPathQuery.levels().get(0);
            var relationsFuture = switch (relationPathLevel.direction()) {
                case FROM -> findByFromAndTypeAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON);
                case TO -> findByToAndTypeAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON);
            };
            return Futures.transform(relationsFuture, entityRelations -> {
                if (entityRelations == null || entityRelations.isEmpty()) {
                    return Collections.emptyList();
                }
                List<EntityRelation> relations = relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations;
                if (relations.size() > limit) {
                    List<EntityRelation> limitedRelations = new ArrayList<>(relations);
                    limitedRelations.sort(Comparator.comparing(r -> r.getFrom().getId()));
                    return limitedRelations.subList(0, limit);
                } else {
                    return relations;
                }
            }, directExecutor());
        }
        return executor.submit(() -> {
            List<EntityRelation> entityRelations = relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit);
            return relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations;
        });
    }

    private List<EntityRelation> filterRelations(List<EntityRelation> entityRelations,  Predicate<EntityRelation> relationFilter) {
        return entityRelations.stream()
                .filter(relationFilter)
                .toList();
    }

    @Override
    public List<EntityRelation> findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery) {
        log.trace("Executing findByRelationPathQuery, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery);
        validateId(tenantId, id -> "Invalid tenant id: " + id);
        validate(relationPathQuery);
        int limit = (int) apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxRelatedEntitiesToReturnPerCfArgument);
        if (relationPathQuery.levels().size() == 1) {
            RelationPathLevel relationPathLevel = relationPathQuery.levels().get(0);
            var relations = switch (relationPathLevel.direction()) {
                case FROM -> findByFromAndType(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON);
                case TO -> findByToAndType(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON);
            };
            if (relations.size() > limit) {
                List<EntityRelation> limitedRelations = new ArrayList<>(relations);
                limitedRelations.sort(Comparator.comparing(r -> r.getFrom().getId()));
                return limitedRelations.subList(0, limit);
            } else {
                return relations;
            }
        }
        return relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit);
    }

    private void validate(EntityRelationPathQuery relationPathQuery) {
        validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id);
        List<RelationPathLevel> levels = relationPathQuery.levels();
        if (CollectionUtils.isEmpty(levels)) {
            throw new DataValidationException("Validation error: relation path levels should be specified!");
        }
        levels.forEach(RelationPathLevel::validate);
    }

    private static void validate(EntityRelation relation) {
        if (relation == null) {
            throw new DataValidationException("Validation error: relation must not be null");
        }
        ConstraintValidator.validateFields(relation);
    }

    private static void validate(EntityId from, EntityId to, String type, RelationTypeGroup typeGroup) {
        validateType(type);
        validateTypeGroup(typeGroup);
        if (from == null) {
            throw new DataValidationException("Relation should contain from entity!");
        }
        if (to == null) {
            throw new DataValidationException("Relation should contain to entity!");
        }
    }

    private static void validateType(String type) {
        if (type == null) {
            throw new DataValidationException("Relation type should be specified!");
        }
    }

    private static void validateTypeGroup(RelationTypeGroup typeGroup) {
        if (typeGroup == null) {
            throw new DataValidationException("Relation type group should be specified!");
        }
    }

    private static void validate(EntityId entity) {
        if (entity == null) {
            throw new DataValidationException("Entity should be specified!");
        }
    }

    private static boolean matchFilters(List<RelationEntityTypeFilter> filters, EntityRelation relation, EntitySearchDirection direction) {
        for (RelationEntityTypeFilter filter : filters) {
            if (match(filter, relation, direction)) {
                return true;
            }
        }
        return false;
    }

    private static boolean match(RelationEntityTypeFilter filter, EntityRelation relation, EntitySearchDirection direction) {
        if (StringUtils.isEmpty(filter.getRelationType()) || filter.getRelationType().equals(relation.getType())) {
            if (filter.getEntityTypes() == null || filter.getEntityTypes().isEmpty()) {
                return true;
            } else {
                EntityId entityId = direction == EntitySearchDirection.FROM ? relation.getTo() : relation.getFrom();
                return filter.getEntityTypes().contains(entityId.getEntityType());
            }
        } else {
            return false;
        }
    }

    @RequiredArgsConstructor
    private static class RelationQueueCtx {

        final SettableFuture<Set<EntityRelation>> future = SettableFuture.create();
        final Set<EntityRelation> result = ConcurrentHashMap.newKeySet();
        final Queue<RelationTask> tasks = new ConcurrentLinkedQueue<>();

        final TenantId tenantId;
        final EntitySearchDirection direction;
        final RelationTypeGroup relationTypeGroup;
        final boolean fetchLastLevelOnly;
        final int maxLvl;
        final ConcurrentHashMap<EntityId, Boolean> uniqueMap;

    }

    private record RelationTask(int currentLvl, EntityId root, List<EntityRelation> prevRelations) {}

    private void processQueue(RelationQueueCtx ctx) {
        RelationTask task = ctx.tasks.poll();
        while (task != null) {
            List<EntityRelation> relations = findRelations(ctx.tenantId, task.root, ctx.direction, ctx.relationTypeGroup);
            Map<EntityId, List<EntityRelation>> newChildrenRelations = new HashMap<>();
            for (EntityRelation childRelation : relations) {
                log.trace("Found Relation: {}", childRelation);
                EntityId childId = ctx.direction == EntitySearchDirection.FROM ? childRelation.getTo() : childRelation.getFrom();
                if (ctx.uniqueMap.putIfAbsent(childId, Boolean.TRUE) == null) {
                    log.trace("Adding Relation: {}", childId);
                    newChildrenRelations.put(childId, new ArrayList<>());
                }
                if (ctx.fetchLastLevelOnly) {
                    var list = newChildrenRelations.get(childId);
                    if (list != null) {
                        list.add(childRelation);
                    }
                }
            }
            if (ctx.fetchLastLevelOnly) {
                if (relations.isEmpty()) {
                    ctx.result.addAll(task.prevRelations);
                } else if (task.currentLvl == ctx.maxLvl) {
                    ctx.result.addAll(relations);
                }
            } else {
                ctx.result.addAll(relations);
            }
            var finalTask = task;
            newChildrenRelations.forEach((child, childRelations) -> {
                var newLvl = finalTask.currentLvl + 1;
                if (newLvl <= ctx.maxLvl)
                    ctx.tasks.add(new RelationTask(newLvl, child, childRelations));
            });
            task = ctx.tasks.poll();
        }
        ctx.future.set(ctx.result);
    }

    private ListenableFuture<Set<EntityRelation>> findRelationsRecursively(final TenantId tenantId, final EntityId rootId, final EntitySearchDirection direction,
                                                                           RelationTypeGroup relationTypeGroup, int lvl, boolean fetchLastLevelOnly,
                                                                           final ConcurrentHashMap<EntityId, Boolean> uniqueMap) {
        if (lvl == 0) {
            return Futures.immediateFuture(Collections.emptySet());
        }
        var relationQueueCtx = new RelationQueueCtx(tenantId, direction, relationTypeGroup, fetchLastLevelOnly, lvl, uniqueMap);
        relationQueueCtx.tasks.add(new RelationTask(1, rootId, Collections.emptyList()));
        relationsExecutor.submit(() -> processQueue(relationQueueCtx));
        return Futures.withTimeout(relationQueueCtx.future, relationQueryTimeout, TimeUnit.SECONDS, timeoutExecutorService);
    }


    private List<EntityRelation> findRelations(final TenantId tenantId, final EntityId rootId, final EntitySearchDirection direction, RelationTypeGroup relationTypeGroup) {
        List<EntityRelation> relations;
        if (relationTypeGroup == null) {
            relationTypeGroup = RelationTypeGroup.COMMON;
        }
        if (direction == EntitySearchDirection.FROM) {
            relations = findByFrom(tenantId, rootId, relationTypeGroup);
        } else {
            relations = findByTo(tenantId, rootId, relationTypeGroup);
        }
        return relations;
    }

    private void publishEvictEvent(EntityRelationEvent event) {
        if (TransactionSynchronizationManager.isActualTransactionActive()) {
            eventPublisher.publishEvent(event);
        } else {
            handleEvictEvent(event);
        }
    }

}
