package at.procon.eventhub.importing;

import at.procon.eventhub.config.EventHubProperties;
import at.procon.eventhub.dto.SchedulerTriggerMode;
import java.time.OffsetDateTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.support.CronExpression;

public abstract class AbstractImportScheduler<R extends ImportRunRequest> {

    private final Logger log = LoggerFactory.getLogger(getClass());
    private final Map<String, ZonedDateTime> nextRunByPlan = new ConcurrentHashMap<>();
    private final Map<String, AtomicBoolean> runningByPlan = new ConcurrentHashMap<>();

    protected void triggerInitialPlansOnStartup() {
        if (!schedulerEnabled()) {
            return;
        }
        for (EventHubProperties.ConfiguredImportPlan plan : configuredPlans()) {
            if (plan.isEnabled() && hasPlanKey(plan) && plan.isRunInitialOnStartup()) {
                triggerPlan(plan, true, schedulerTriggerMode());
            }
        }
    }

    protected void pollConfiguredPlans() {
        if (!schedulerEnabled()) {
            return;
        }
        ZonedDateTime now = ZonedDateTime.now(ZoneId.systemDefault());
        for (EventHubProperties.ConfiguredImportPlan plan : configuredPlans()) {
            if (!plan.isEnabled() || !hasPlanKey(plan) || plan.getCron() == null || plan.getCron().isBlank()) {
                continue;
            }
            String key = plan.getPlanKey();
            ZonedDateTime next = nextRunByPlan.computeIfAbsent(key, ignored -> CronExpression.parse(plan.getCron()).next(now));
            if (next != null && !next.isAfter(now)) {
                triggerPlan(plan, false, schedulerTriggerMode());
                nextRunByPlan.put(key, CronExpression.parse(plan.getCron()).next(now.plusSeconds(1)));
            }
        }
    }

    protected abstract boolean schedulerEnabled();

    protected abstract List<EventHubProperties.ConfiguredImportPlan> configuredPlans();

    protected abstract SchedulerTriggerMode schedulerTriggerMode();

    protected abstract R createInitialRequest(EventHubProperties.ConfiguredImportPlan plan);

    protected abstract R createScheduledRequest(EventHubProperties.ConfiguredImportPlan plan);

    protected abstract void startImport(R request);

    protected abstract void startAndExecuteImport(R request);

    protected String providerName() {
        return "provider";
    }

    private void triggerPlan(EventHubProperties.ConfiguredImportPlan plan, boolean initial, SchedulerTriggerMode triggerMode) {
        String key = plan.getPlanKey();
        AtomicBoolean running = runningByPlan.computeIfAbsent(key, ignored -> new AtomicBoolean(false));
        if (!running.compareAndSet(false, true)) {
            log.info("Skipping {} import plan={} because a previous run is still active", providerName(), key);
            return;
        }
        try {
            R request = initial ? createInitialRequest(plan) : createScheduledRequest(plan);
            log.info("Triggering {} import plan={} initial={} mode={} strategy={} triggerMode={} at={}",
                    providerName(), key, initial, request.mode(), request.acquisitionStrategy(), triggerMode, OffsetDateTime.now());
            if (triggerMode == SchedulerTriggerMode.EXECUTE) {
                startAndExecuteImport(request);
            } else {
                startImport(request);
            }
        } catch (RuntimeException ex) {
            log.error("{} import plan={} failed to trigger", providerName(), key, ex);
        } finally {
            running.set(false);
        }
    }

    private boolean hasPlanKey(EventHubProperties.ConfiguredImportPlan plan) {
        return plan.getPlanKey() != null && !plan.getPlanKey().isBlank();
    }
}
