package at.procon.eventhub.processing.eventprocessing.module.epl;

import com.espertech.esper.common.client.EPCompiled;
import com.espertech.esper.common.client.EventBean;
import com.espertech.esper.common.client.configuration.Configuration;
import com.espertech.esper.compiler.client.CompilerArguments;
import com.espertech.esper.compiler.client.EPCompileException;
import com.espertech.esper.compiler.client.EPCompilerProvider;
import com.espertech.esper.runtime.client.EPDeployException;
import com.espertech.esper.runtime.client.EPDeployment;
import com.espertech.esper.runtime.client.EPRuntime;
import com.espertech.esper.runtime.client.EPRuntimeProvider;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Service;
import org.springframework.util.StreamUtils;

@Service
public class RuntimeEplModuleExecutor {

    private static final AtomicLong RUNTIME_COUNTER = new AtomicLong();

    public RuntimeEplModuleExecutionResult execute(RuntimeEplModuleDefinition definition) {
        EPRuntime runtime = null;
        try {
            Configuration configuration = new Configuration();
            definition.eventTypes().forEach((eventTypeName, eventTypeDefinition) ->
                    configuration.getCommon().addEventType(eventTypeName, eventTypeDefinition));

            String runtimeUri = "eventhub-runtime-epl-module-"
                    + definition.moduleKey().replaceAll("[^A-Za-z0-9_-]", "-")
                    + "-" + RUNTIME_COUNTER.incrementAndGet();
            runtime = EPRuntimeProvider.getRuntime(runtimeUri, configuration);

            EPCompiled compiled = EPCompilerProvider.getCompiler().compile(renderEpl(definition.eplResources()), new CompilerArguments(configuration));
            EPDeployment deployment = runtime.getDeploymentService().deploy(compiled);

            Map<String, List<Map<String, Object>>> outputs = new LinkedHashMap<>();
            for (String statementName : definition.outputStatementNames()) {
                outputs.put(statementName, new ArrayList<>());
                var statement = runtime.getDeploymentService().getStatement(deployment.getDeploymentId(), statementName);
                if (statement == null) {
                    throw new IllegalStateException("EPL module " + definition.moduleKey()
                            + " did not deploy expected statement '" + statementName + "'.");
                }
                statement.addListener((newData, oldData, stmt, rt) -> collect(newData, outputs.get(statementName)));
            }

            for (RuntimeEplInputEventStream inputStream : definition.inputStreams()) {
                for (Map<String, Object> event : inputStream.events()) {
                    runtime.getEventService().sendEventMap(event, inputStream.eventTypeName());
                }
            }

            Map<String, Object> metadata = new LinkedHashMap<>();
            metadata.put("engine", "EPL");
            metadata.put("eplResources", definition.eplResources());
            Map<String, Integer> inputCounts = new LinkedHashMap<>();
            for (RuntimeEplInputEventStream inputStream : definition.inputStreams()) {
                inputCounts.merge(inputStream.eventTypeName(), inputStream.events().size(), Integer::sum);
            }
            metadata.put("inputCounts", inputCounts);
            Map<String, Integer> outputCounts = new LinkedHashMap<>();
            outputs.forEach((statement, events) -> outputCounts.put(statement, events.size()));
            metadata.put("outputCounts", outputCounts);
            return new RuntimeEplModuleExecutionResult(outputs, metadata);
        } catch (EPCompileException | EPDeployException e) {
            throw new IllegalStateException("Cannot compile/deploy runtime EPL module " + definition.moduleKey(), e);
        } finally {
            if (runtime != null) {
                runtime.destroy();
            }
        }
    }

    private String renderEpl(List<String> resources) {
        StringBuilder builder = new StringBuilder();
        for (String resource : resources) {
            if (resource == null || resource.isBlank()) {
                continue;
            }
            if (!builder.isEmpty()) {
                builder.append("\n\n");
            }
            builder.append(loadResource(resource.trim()));
        }
        return builder.toString();
    }

    private static String loadResource(String path) {
        try {
            ClassPathResource resource = new ClassPathResource(path);
            return StreamUtils.copyToString(resource.getInputStream(), StandardCharsets.UTF_8);
        } catch (IOException e) {
            throw new IllegalStateException("Cannot load EPL resource: " + path, e);
        }
    }

    private void collect(EventBean[] newData, List<Map<String, Object>> target) {
        if (newData == null || target == null) {
            return;
        }
        for (EventBean eventBean : newData) {
            LinkedHashMap<String, Object> values = new LinkedHashMap<>();
            for (String propertyName : eventBean.getEventType().getPropertyNames()) {
                if (propertyName == null) {
                    continue;
                }
                values.put(propertyName, eventBean.get(propertyName));
            }
            target.add(values);
        }
    }
}
