pf-sdn/src/main/java/ru/kirillius/pf/sdn/service/FlowService.java

package ru.kirillius.pf.sdn.service;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import ru.kirillius.java.utils.events.EventListener;
import ru.kirillius.pf.sdn.api.Networking.Subscriptions.Subscription;
import ru.kirillius.pf.sdn.api.flow.Flow;
import ru.kirillius.pf.sdn.api.flow.FlowAction;
import ru.kirillius.pf.sdn.api.flow.FlowFunction;
import ru.kirillius.pf.sdn.api.flow.TriggerType;
import ru.kirillius.pf.sdn.entity.ActionConfig;
import ru.kirillius.pf.sdn.entity.FlowConfig;
import ru.kirillius.pf.sdn.entity.SubscriptionSetEntry;
import ru.kirillius.pf.sdn.flow.DummyFunction;
import ru.kirillius.pf.sdn.repository.PipelineConfigRepository;

import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.regex.Pattern;

@Slf4j
@Service
@RequiredArgsConstructor
public class FlowService {

    private final PipelineConfigRepository configRepository;
    private final FlowExecutorService flowExecutorService;
    private final SubscriptionService subscriptionService;
    private final ExecutorService executorService;

    private final Map<FlowConfig, Flow> pipelines = new ConcurrentHashMap<>();
    private final Map<String, Class<? extends FlowFunction>> functions = new ConcurrentHashMap<>();
    private EventListener<Flow> executedListener;
    private EventListener<Subscription> subscriptionUpdateListener;
    private EventListener<SubscriptionSetEntry> subscriptionSetUpdateListener;

    public void registerFunction(Class<? extends FlowFunction> functionClass, String id) {
        if (functions.containsKey(id)) {
            throw new IllegalStateException("Function with id '" + id + "' already exists");
        }
        functions.put(id, functionClass);
    }

    public void trigger(long id) {
        pipelines.keySet().stream().filter(config -> config.getId().equals(id)).findFirst().ifPresent(this::trigger);
    }

    public void trigger(FlowConfig flowConfig) {
        if (pipelines.containsKey(flowConfig)) {
            trigger(pipelines.get(flowConfig));
        }
    }

    public void trigger(Flow pipeline) {
        var matchedConditions = pipeline.getConfig().getStartConditions().stream().filter(c -> c.getTrigger() == TriggerType.Manual).toList();
        if (matchedConditions.isEmpty()) {
            log.error("Unable to manual start pipeline {} because it has no manual trigger", pipeline.getConfig().getName());
            return;
        }
        matchedConditions.forEach(c -> {
            if (c.isDontStartIfRunning() && pipeline.isRunning()) {
                return;
            }
            flowExecutorService.triggerExecute(pipeline);
        });

    }

    public void update(FlowConfig flowConfig) {
        if (configRepository.existsByGuid(flowConfig.getGuid())) {
            load(flowConfig);
        } else {
            flowExecutorService.cancel(pipelines.get(flowConfig));
            pipelines.remove(flowConfig);
        }
    }

    private void load(FlowConfig flowConfig) {
        if (pipelines.containsKey(flowConfig)) {
            flowExecutorService.cancel(pipelines.get(flowConfig));
        }
        pipelines.put(flowConfig, new Flow(flowConfig, this::buildAction));
    }

    private FlowAction buildAction(ActionConfig config) {
        return new FlowAction(functions.getOrDefault(config.getFunctionId(), DummyFunction.class), config);
    }

    private void checkForIntervalExecution(Flow pipeline) {

        var matchedConditions = pipeline.getConfig().getStartConditions().stream().filter(c -> c.getTrigger() == TriggerType.Interval).toList();

        if (matchedConditions.isEmpty()) {
            return;
        }

        matchedConditions.forEach(c -> {
            var properties = c.getProperties();
            var timer = properties.getOrDefault("interval", null);
            if (timer == null) {
                return;
            }
            var delay = Long.parseLong(timer);
            if (delay <= 0) {
                return;
            }

            executorService.submit(() -> {
                try {
                    Thread.sleep(delay * 60000L);
                } catch (InterruptedException e) {
                    return;
                }
                if (c.isDontStartIfRunning() && pipeline.isRunning()) {
                    return;
                }
                flowExecutorService.triggerExecute(pipeline);
            });
        });

    }

    @PostConstruct
    private void initialize() {
        subscriptionUpdateListener = subscriptionService.getUpdateEvent().add(this::subscriptionUpdate);
        subscriptionSetUpdateListener = subscriptionService.getSetUpdateEvent().add(this::subscriptionSetUpdate);
        executedListener = flowExecutorService.getOnExecuted().add(this::checkForIntervalExecution);
        registerFunction(DummyFunction.class, "Error:fallback");
        configRepository.findAll().forEach(this::load);
        pipelines.values()
                .stream()
                .filter(p -> p
                        .getConfig()
                        .getStartConditions()
                        .stream()
                        .anyMatch(c -> c.getTrigger() == TriggerType.OnStart))
                .forEach(flowExecutorService::triggerExecute);
    }

    private void subscriptionSetUpdate(SubscriptionSetEntry setEntry) {
        pipelines.values().forEach(pipeline -> {
            var matchedConditions = pipeline.getConfig().getStartConditions().stream().filter(c -> c.getTrigger() == TriggerType.OnSubscriptionSetUpdate).toList();
            if (matchedConditions.isEmpty()) {
                return;
            }

            matchedConditions.forEach(c -> {
                if (c.isDontStartIfRunning() && pipeline.isRunning()) {
                    return;
                }

                var properties = c.getProperties();
                var names = properties.getOrDefault("names", null);
                if (names == null || names.isEmpty()) {
                    flowExecutorService.triggerExecute(pipeline);
                    return;
                }

                if (Arrays.stream(names.split(Pattern.quote(","))).anyMatch(s -> setEntry.getName().equals(s))) {
                    flowExecutorService.triggerExecute(pipeline);
                }
            });
        });
    }

    private void subscriptionUpdate(Subscription subscription) {

        pipelines.values().forEach(pipeline -> {
            var matchedConditions = pipeline.getConfig().getStartConditions().stream().filter(c -> c.getTrigger() == TriggerType.OnSubscriptionUpdate).toList();
            if (matchedConditions.isEmpty()) {
                return;
            }

            matchedConditions.forEach(c -> {
                if (c.isDontStartIfRunning() && pipeline.isRunning()) {
                    return;
                }

                var properties = c.getProperties();
                var names = properties.getOrDefault("names", null);
                if (names == null || names.isEmpty()) {
                    flowExecutorService.triggerExecute(pipeline);
                    return;
                }

                if (Arrays.stream(names.split(Pattern.quote(","))).anyMatch(s -> subscription.getName().equals(s))) {
                    flowExecutorService.triggerExecute(pipeline);
                }
            });
        });

    }

    @PreDestroy
    private void destroy() {
        if (executedListener != null) {
            flowExecutorService.getOnExecuted().remove(executedListener);
            executedListener = null;
        }
        if (subscriptionUpdateListener != null) {
            subscriptionService.getUpdateEvent().remove(subscriptionUpdateListener);
            subscriptionUpdateListener = null;
        }
        if (subscriptionSetUpdateListener != null) {
            subscriptionService.getSetUpdateEvent().remove(subscriptionSetUpdateListener);
            subscriptionSetUpdateListener = null;
        }
    }

}