package com.easyagents.scheduler.quartz;

import com.easyagents.scheduler.ConcurrencyPolicy;
import com.easyagents.scheduler.CronSchedulePlan;
import com.easyagents.scheduler.MisfirePolicy;
import com.easyagents.scheduler.OnceSchedulePlan;
import com.easyagents.scheduler.ScheduleDefinition;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleFireContext;
import com.easyagents.scheduler.ScheduleHandler;
import com.easyagents.scheduler.ScheduleId;
import com.easyagents.scheduler.ScheduleRefireException;
import com.easyagents.scheduler.ScheduleStatus;
import org.junit.After;
import org.junit.Test;
import org.quartz.JobDetail;
import org.quartz.Scheduler;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.impl.matchers.GroupMatcher;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.util.Base64;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

/**
 * {@link QuartzScheduleService} 的无数据库行为测试。
 */
public class QuartzScheduleServiceTest {

    private QuartzScheduleService service;

    /**
     * 关闭每个测试创建的 Scheduler。
     */
    @After
    public void tearDown() {
        if (service != null) {
            service.close();
        }
    }

    /**
     * 验证创建幂等、冲突检测、暂停恢复、原子替换和删除。
     */
    @Test
    public void shouldManageScheduleLifecycle() throws Exception {
        service = newRamService(context -> { });
        ScheduleDefinition original = cronDefinition("handler", "0 0 0 1 1 ? 2099");

        assertEquals(original, service.create(original).definition());
        assertEquals(original, service.create(original).definition());

        ScheduleDefinition conflicting = cronDefinition("handler", "0 0 0 2 1 ? 2099");
        try {
            service.create(conflicting);
            fail("expected schedule conflict");
        } catch (ScheduleException exception) {
            assertEquals(ScheduleErrorCode.SCHEDULE_CONFLICT, exception.errorCode());
        }

        assertEquals(ScheduleStatus.PAUSED, service.pause(original.id()).status());
        assertEquals(ScheduleStatus.SCHEDULED, service.resume(original.id()).status());
        assertEquals(conflicting, service.replace(conflicting).definition());
        assertTrue(service.delete(original.id()));
        assertFalse(service.delete(original.id()));
    }

    /**
     * 验证立即触发会覆盖同名基础参数并透传调用标识。
     */
    @Test
    public void shouldDispatchImmediateTriggerWithOverrideParameters() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);
        AtomicReference<ScheduleFireContext> captured = new AtomicReference<>();
        service = newRamService(context -> {
            captured.set(context);
            latch.countDown();
        });
        ScheduleDefinition definition = new ScheduleDefinition(
            new ScheduleId("test", "immediate"),
            "handler",
            new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW,
            true,
            Map.of("same", "base", "stable", "value"),
            "immediate test"
        );
        service.create(definition);

        service.triggerNow(definition.id(), "invoke-1", Map.of("same", "override"));

        assertTrue("handler did not execute", latch.await(5, TimeUnit.SECONDS));
        assertNotNull(captured.get());
        assertEquals("invoke-1", captured.get().invocationId());
        assertEquals("override", captured.get().parameters().get("same"));
        assertEquals("value", captured.get().parameters().get("stable"));
    }

    /** 验证可恢复失败会持久化延迟重试，并保留原始 fire 上下文。 */
    @Test
    public void shouldRefireRetryableHandlerFailure() throws Exception {
        AtomicInteger attempts = new AtomicInteger();
        CountDownLatch completed = new CountDownLatch(1);
        CopyOnWriteArrayList<ScheduleFireContext> contexts = new CopyOnWriteArrayList<>();
        service = newRamService(context -> {
            contexts.add(context);
            if (attempts.incrementAndGet() == 1) {
                throw new ScheduleRefireException("temporary registration failure",
                    new IllegalStateException("database unavailable"), 2,
                    Duration.ofMillis(100));
            }
            completed.countDown();
        });
        ScheduleDefinition definition = onceDefinition(
            "retryable-immediate",
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW,
            Instant.parse("2099-01-01T00:00:00Z")
        );
        service.create(definition);

        service.triggerNow(definition.id(), "retryable-invocation", Map.of());

        assertTrue("retryable handler was not refired", completed.await(5, TimeUnit.SECONDS));
        assertEquals(2, attempts.get());
        assertEquals(contexts.get(0).scheduledFireTime(), contexts.get(1).scheduledFireTime());
        assertEquals(contexts.get(0).actualFireTime(), contexts.get(1).actualFireTime());
        assertEquals(contexts.get(0).fireInstanceId(), contexts.get(1).fireInstanceId());
        assertEquals("retryable-invocation", contexts.get(1).invocationId());
    }

    /** 持久化延迟重试不得占用当前 Quartz worker。 */
    @Test
    public void shouldReleaseWorkerWhilePersistentRetryIsDelayed() throws Exception {
        CountDownLatch retryScheduled = new CountDownLatch(1);
        CountDownLatch healthyCompleted = new CountDownLatch(1);
        CountDownLatch retryCompleted = new CountDownLatch(1);
        AtomicInteger retryAttempts = new AtomicInteger();
        service = newRamService(context -> {
            if (context.scheduleId().name().equals("delayed-retry")) {
                if (retryAttempts.incrementAndGet() == 1) {
                    retryScheduled.countDown();
                    throw new ScheduleRefireException("temporary registration failure",
                        new IllegalStateException("database unavailable"), 1,
                        Duration.ofSeconds(1));
                }
                retryCompleted.countDown();
            } else {
                healthyCompleted.countDown();
            }
        }, 30_000L, 1);
        ScheduleDefinition retry = onceDefinition(
            "delayed-retry", MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW, Instant.parse("2099-01-01T00:00:00Z"));
        ScheduleDefinition healthy = onceDefinition(
            "healthy-during-retry", MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW, Instant.parse("2099-01-01T00:00:00Z"));
        service.create(retry);
        service.create(healthy);

        service.triggerNow(retry.id(), "retry-1", Map.of());
        assertTrue(retryScheduled.await(5, TimeUnit.SECONDS));
        service.triggerNow(healthy.id(), "healthy-1", Map.of());

        assertTrue("single Quartz worker remained occupied by delayed retry",
            healthyCompleted.await(750, TimeUnit.MILLISECONDS));
        assertTrue(retryCompleted.await(5, TimeUnit.SECONDS));
    }

    /** 已持久化 fire 的重试不得因 replace 而切换到新 Handler。 */
    @Test
    public void shouldKeepOriginalHandlerWhenScheduleIsReplacedDuringRetry() throws Exception {
        AtomicInteger oldAttempts = new AtomicInteger();
        AtomicInteger newAttempts = new AtomicInteger();
        CountDownLatch oldRetryCompleted = new CountDownLatch(1);
        service = newRamServiceWithHandlers(List.of(
            handler("old-handler", context -> {
                if (oldAttempts.incrementAndGet() == 1) {
                    throw new ScheduleRefireException("temporary registration failure",
                        new IllegalStateException("database unavailable"), 1,
                        Duration.ofMillis(400));
                }
                oldRetryCompleted.countDown();
            }),
            handler("new-handler", context -> newAttempts.incrementAndGet())
        ), 30_000L, 2);
        ScheduleId id = new ScheduleId("test", "replace-during-retry");
        ScheduleDefinition original = definition(id, "old-handler");
        service.create(original);
        service.triggerNow(id, "replace-retry-1", Map.of());
        awaitPersistentRetryTrigger();

        service.replace(definition(id, "new-handler"));

        assertTrue("old handler retry did not complete",
            oldRetryCompleted.await(5, TimeUnit.SECONDS));
        assertEquals(2, oldAttempts.get());
        assertEquals(0, newAttempts.get());
    }

    /**
     * 验证立即触发会在返回回执前校验基础参数与覆盖参数的合并结果。
     */
    @Test
    public void shouldRejectOversizedMergedImmediateParameters() throws Exception {
        AtomicInteger executions = new AtomicInteger();
        service = newRamService(context -> executions.incrementAndGet());
        ScheduleDefinition definition = new ScheduleDefinition(
            new ScheduleId("test", "oversized-immediate"),
            "handler",
            new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW,
            true,
            Map.of("base", "x".repeat(9_000)),
            "oversized immediate test"
        );
        service.create(definition);

        try {
            service.triggerNow(
                definition.id(),
                "oversized-invocation",
                Map.of("override", "y".repeat(9_000))
            );
            fail("expected oversized merged parameters");
        } catch (ScheduleException exception) {
            assertEquals(ScheduleErrorCode.INVALID_DEFINITION, exception.errorCode());
        }
        assertEquals(0, executions.get());
    }

    /**
     * 验证立即触发读取到损坏持久参数时返回 Provider 错误。
     */
    @Test
    public void shouldClassifyCorruptedPersistedParametersBeforeImmediateTrigger()
        throws Exception {
        service = newRamService(context -> { });
        ScheduleDefinition definition = onceDefinition(
            "corrupted-immediate",
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW,
            Instant.parse("2099-01-01T00:00:00Z")
        );
        service.create(definition);
        JobDetail job = service.quartzScheduler().getJobDetail(
            QuartzScheduleMapper.jobKey(definition.id())
        );
        String encodedKey = Base64.getUrlEncoder().withoutPadding().encodeToString(
            "payload".getBytes(StandardCharsets.UTF_8)
        );
        job.getJobDataMap().put(
            QuartzScheduleMapper.PARAMETER_PREFIX + encodedKey,
            "x".repeat(ScheduleDefinition.MAX_PARAMETER_BYTES)
        );
        service.quartzScheduler().addJob(job, true);

        try {
            service.triggerNow(definition.id(), "corrupted-invocation", Map.of());
            fail("expected corrupted persisted parameters");
        } catch (ScheduleException exception) {
            assertEquals(ScheduleErrorCode.PROVIDER_FAILURE, exception.errorCode());
        }
    }

    /**
     * 验证预览不会落库且非法 Cron 返回稳定错误码。
     */
    @Test
    public void shouldPreviewFireTimesWithoutPersisting() throws Exception {
        service = newRamService(context -> { });

        assertEquals(
            3,
            service.nextFireTimes(
                new CronSchedulePlan("0 0/5 * * * ?", ZoneId.of("Asia/Shanghai")),
                3
            ).size()
        );
        try {
            service.nextFireTimes(new CronSchedulePlan("bad cron", ZoneId.of("UTC")), 1);
            fail("expected invalid cron");
        } catch (ScheduleException exception) {
            assertEquals(ScheduleErrorCode.INVALID_DEFINITION, exception.errorCode());
        }
    }

    /**
     * 验证 DISALLOW 策略会串行执行同一 Schedule 的立即触发。
     */
    @Test
    public void shouldDisallowConcurrentExecutionForSameSchedule() throws Exception {
        CountDownLatch firstStarted = new CountDownLatch(1);
        CountDownLatch release = new CountDownLatch(1);
        CountDownLatch completed = new CountDownLatch(2);
        AtomicInteger active = new AtomicInteger();
        AtomicInteger maximum = new AtomicInteger();
        service = newRamService(context -> {
            int current = active.incrementAndGet();
            maximum.accumulateAndGet(current, Math::max);
            firstStarted.countDown();
            await(release);
            active.decrementAndGet();
            completed.countDown();
        });
        ScheduleDefinition definition = onceDefinition(
            "serial",
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW,
            Instant.parse("2099-01-01T00:00:00Z")
        );
        service.create(definition);

        service.triggerNow(definition.id(), "serial-1", Map.of());
        assertTrue(firstStarted.await(5, TimeUnit.SECONDS));
        service.triggerNow(definition.id(), "serial-2", Map.of());
        Thread.sleep(250L);
        assertEquals(1, maximum.get());
        release.countDown();
        assertTrue(completed.await(5, TimeUnit.SECONDS));
        assertEquals(1, maximum.get());
    }

    /**
     * 验证 ALLOW 策略允许同一 Schedule 按 Worker 容量并发执行。
     */
    @Test
    public void shouldAllowConcurrentExecutionForSameSchedule() throws Exception {
        CountDownLatch bothStarted = new CountDownLatch(2);
        CountDownLatch release = new CountDownLatch(1);
        CountDownLatch completed = new CountDownLatch(2);
        AtomicInteger active = new AtomicInteger();
        AtomicInteger maximum = new AtomicInteger();
        service = newRamService(context -> {
            int current = active.incrementAndGet();
            maximum.accumulateAndGet(current, Math::max);
            bothStarted.countDown();
            await(release);
            active.decrementAndGet();
            completed.countDown();
        });
        ScheduleDefinition definition = onceDefinition(
            "parallel",
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.ALLOW,
            Instant.parse("2099-01-01T00:00:00Z")
        );
        service.create(definition);

        service.triggerNow(definition.id(), "parallel-1", Map.of());
        service.triggerNow(definition.id(), "parallel-2", Map.of());
        assertTrue("both handlers did not start", bothStarted.await(5, TimeUnit.SECONDS));
        assertEquals(2, maximum.get());
        release.countDown();
        assertTrue(completed.await(5, TimeUnit.SECONDS));
    }

    /**
     * 验证删除运行中任务不会与 Quartz 临时 Trigger 的完成清理发生竞态。
     */
    @Test
    public void shouldDeleteScheduleWhileImmediateTriggerIsRunning() throws Exception {
        CountDownLatch started = new CountDownLatch(1);
        CountDownLatch release = new CountDownLatch(1);
        CountDownLatch completed = new CountDownLatch(1);
        service = newRamService(context -> {
            started.countDown();
            await(release);
            completed.countDown();
        });
        ScheduleDefinition definition = onceDefinition(
            "delete-running",
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW,
            Instant.parse("2099-01-01T00:00:00Z")
        );
        service.create(definition);
        service.triggerNow(definition.id(), "delete-running-1", Map.of());
        assertTrue(started.await(5, TimeUnit.SECONDS));

        try {
            assertTrue(service.delete(definition.id()));
            assertTrue(service.get(definition.id()).isEmpty());
        } finally {
            release.countDown();
        }
        assertTrue(completed.await(5, TimeUnit.SECONDS));
    }

    /**
     * 验证一次性任务的 SKIP 与 FIRE_ONCE_NOW Misfire 映射。
     */
    @Test
    public void shouldApplyOnceMisfirePolicies() throws Exception {
        CountDownLatch fired = new CountDownLatch(1);
        AtomicInteger executions = new AtomicInteger();
        service = newRamService(context -> {
            executions.incrementAndGet();
            if (context.scheduleId().name().equals("fire-misfire")) {
                fired.countDown();
            }
        });
        Instant missedAt = Instant.now().minusSeconds(2);

        service.create(onceDefinition(
            "skip-misfire",
            MisfirePolicy.SKIP,
            ConcurrencyPolicy.DISALLOW,
            missedAt
        ));
        service.create(onceDefinition(
            "fire-misfire",
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW,
            missedAt
        ));

        assertTrue("FIRE_ONCE_NOW did not execute", fired.await(5, TimeUnit.SECONDS));
        Thread.sleep(250L);
        assertEquals(1, executions.get());
        assertEquals(
            ScheduleStatus.COMPLETE,
            service.get(new ScheduleId("test", "skip-misfire")).orElseThrow().status()
        );
    }

    /**
     * 验证关闭等待超过上限后会有界返回并中断内部 Handler 线程。
     */
    @Test
    public void shouldBoundShutdownWaitAndInterruptHandler() throws Exception {
        CountDownLatch started = new CountDownLatch(1);
        CountDownLatch interrupted = new CountDownLatch(1);
        service = newRamService(context -> {
            started.countDown();
            try {
                Thread.sleep(60_000L);
            } catch (InterruptedException exception) {
                interrupted.countDown();
                Thread.currentThread().interrupt();
            }
        }, 200L);
        ScheduleDefinition definition = onceDefinition(
            "bounded-close",
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW,
            Instant.parse("2099-01-01T00:00:00Z")
        );
        service.create(definition);
        service.triggerNow(definition.id(), "bounded-close-1", Map.of());
        assertTrue(started.await(5, TimeUnit.SECONDS));

        long startedAt = System.nanoTime();
        try {
            service.close();
            fail("expected bounded shutdown timeout");
        } catch (ScheduleException exception) {
            assertEquals(ScheduleErrorCode.PROVIDER_FAILURE, exception.errorCode());
        }
        long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAt);
        assertTrue("shutdown exceeded bound: " + elapsedMillis, elapsedMillis < 2_000L);
        assertTrue("handler thread was not interrupted", interrupted.await(2, TimeUnit.SECONDS));
    }

    private QuartzScheduleService newRamService(Consumer<ScheduleFireContext> handler)
        throws Exception {
        return newRamService(handler, 30_000L);
    }

    private QuartzScheduleService newRamService(
        Consumer<ScheduleFireContext> handler,
        long shutdownWaitTimeoutMillis
    ) throws Exception {
        return newRamService(handler, shutdownWaitTimeoutMillis, 2);
    }

    private QuartzScheduleService newRamService(
        Consumer<ScheduleFireContext> handler,
        long shutdownWaitTimeoutMillis,
        int threadCount
    ) throws Exception {
        return newRamServiceWithHandlers(List.of(handler("handler", handler)),
            shutdownWaitTimeoutMillis, threadCount);
    }

    private QuartzScheduleService newRamServiceWithHandlers(
        List<ScheduleHandler> handlers,
        long shutdownWaitTimeoutMillis,
        int threadCount
    ) throws Exception {
        Properties properties = new Properties();
        properties.setProperty(
            "org.quartz.scheduler.instanceName",
            "ram-scheduler-" + UUID.randomUUID()
        );
        properties.setProperty("org.quartz.scheduler.instanceId", "NON_CLUSTERED");
        properties.setProperty("org.quartz.scheduler.interruptJobsOnShutdown", "true");
        properties.setProperty("org.quartz.threadPool.class", "org.quartz.simpl.SimpleThreadPool");
        properties.setProperty("org.quartz.threadPool.threadCount", Integer.toString(threadCount));
        properties.setProperty("org.quartz.jobStore.class", "org.quartz.simpl.RAMJobStore");
        properties.setProperty("org.quartz.jobStore.misfireThreshold", "100");
        Scheduler scheduler = new StdSchedulerFactory(properties).getScheduler();
        QuartzScheduleService result = QuartzSchedulerFactory.attach(
            scheduler,
            true,
            shutdownWaitTimeoutMillis,
            handlers,
            java.util.List.of()
        );
        result.start();
        assertSame(scheduler, result.quartzScheduler());
        return result;
    }

    private void awaitPersistentRetryTrigger() throws Exception {
        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
        while (System.nanoTime() < deadline) {
            boolean found = service.quartzScheduler()
                .getTriggerKeys(GroupMatcher.anyTriggerGroup())
                .stream()
                .anyMatch(key -> key.getGroup().startsWith(
                    QuartzScheduleMapper.GROUP_PREFIX + "retry."));
            if (found) return;
            Thread.sleep(10L);
        }
        fail("persistent retry trigger was not created");
    }

    private static ScheduleHandler handler(
        String code,
        Consumer<ScheduleFireContext> consumer
    ) {
        return new ScheduleHandler() {
            @Override
            public String code() {
                return code;
            }

            @Override
            public void execute(ScheduleFireContext context) {
                consumer.accept(context);
            }
        };
    }

    private static ScheduleDefinition definition(ScheduleId id, String handlerCode) {
        return new ScheduleDefinition(
            id,
            handlerCode,
            new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW,
            true,
            Map.of(),
            id.name()
        );
    }

    private static ScheduleDefinition cronDefinition(String handlerCode, String expression) {
        return new ScheduleDefinition(
            new ScheduleId("test", "lifecycle"),
            handlerCode,
            new CronSchedulePlan(expression, ZoneId.of("UTC")),
            MisfirePolicy.SKIP,
            ConcurrencyPolicy.DISALLOW,
            true,
            Map.of("key", "value"),
            "lifecycle test"
        );
    }

    private static ScheduleDefinition onceDefinition(
        String name,
        MisfirePolicy misfirePolicy,
        ConcurrencyPolicy concurrencyPolicy,
        Instant fireAt
    ) {
        return new ScheduleDefinition(
            new ScheduleId("test", name),
            "handler",
            new OnceSchedulePlan(fireAt),
            misfirePolicy,
            concurrencyPolicy,
            true,
            Map.of(),
            name
        );
    }

    private static void await(CountDownLatch latch) {
        boolean interrupted = false;
        try {
            while (true) {
                try {
                    latch.await();
                    return;
                } catch (InterruptedException exception) {
                    interrupted = true;
                }
            }
        } finally {
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        }
    }
}
