package com.ankurm.redis;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import com.ankurm.redis.hash.Session;
import com.ankurm.redis.hash.SessionRepository;

import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.core.RedisKeyExpiredEvent;
import org.springframework.data.redis.core.RedisKeyValueAdapter.EnableKeyspaceEvents;
import org.springframework.data.redis.repository.configuration.EnableRedisRepositories;

/**
 * The same entity with {@code enableKeyspaceEvents = ON_STARTUP}. Spring Data Redis turns the
 * server setting on, keeps a phantom copy of the hash for five more minutes, and cleans up the
 * index entries when Redis announces the expiry. Writes docs/output/10-hash-ttl-events-on.txt.
 */
@RedisTest
class HashTtlEventsOnTest {

    static List<String> before;

    @Autowired
    SessionRepository sessions;

    @Autowired
    ExpiryLog expiryLog;

    @TestConfiguration
    @EnableRedisRepositories(basePackageClasses = SessionRepository.class, enableKeyspaceEvents = EnableKeyspaceEvents.ON_STARTUP)
    static class Config {

        @Bean
        ExpiryLog expiryLog() {
            return new ExpiryLog();
        }
    }

    static class ExpiryLog {

        final List<String> events = new CopyOnWriteArrayList<>();

        @EventListener
        void on(RedisKeyExpiredEvent<?> event) {
            events.add("keyspace=" + event.getKeyspace() + " id=" + new String(event.getId()) + " value=" + event.getValue());
        }
    }

    @BeforeAll
    static void serverDefaults() {
        // runs before the Spring context is created, so this is the setting Spring Data Redis finds
        LocalRedis.reset();
        before = Transcript.run(List.of("redis-cli", "-p", String.valueOf(LocalRedis.PORT), "--no-raw", "CONFIG", "GET", "notify-keyspace-events"));
    }

    @AfterAll
    static void restoreServerDefaults() {
        LocalRedis.reset();
    }

    @BeforeEach
    void clean() {
        LocalRedis.flush();
    }

    @Test
    void springCleansUpWhenRedisAnnouncesTheExpiry() {
        try (Transcript t = new Transcript("10-hash-ttl-events-on.txt",
                "@RedisHash(timeToLive = 2) with enableKeyspaceEvents = ON_STARTUP")) {

            t.section("Server setting before the application context started");
            t.line("$ redis-cli -p %d --no-raw CONFIG GET notify-keyspace-events", LocalRedis.PORT);
            before.forEach(l -> t.line("%s", l));

            t.section("Server setting now");
            t.cli("--no-raw", "CONFIG", "GET", "notify-keyspace-events");

            t.section("Save a session that lives for two seconds");
            sessions.save(new Session("s1", "ankur"));
            t.line("sessions.save(new Session(\"s1\", \"ankur\"))");
            t.cliSorted("KEYS", "*");
            List<String> ttl = t.cli("--no-raw", "TTL", "session:s1");
            assertThat(ttl.get(0)).isEqualTo("(integer) 2");
            List<String> phantom = t.cli("--no-raw", "TTL", "session:s1:phantom");
            assertThat(phantom.get(0)).isEqualTo("(integer) 302");

            t.section("Wait for Redis to expire it, and for Spring to react");
            await().atMost(Duration.ofSeconds(10)).until(() -> !expiryLog.events.isEmpty());
            await().atMost(Duration.ofSeconds(5)).until(() ->
                    Transcript.run(List.of("redis-cli", "-p", String.valueOf(LocalRedis.PORT), "DBSIZE")).get(0).equals("0"));
            expiryLog.events.forEach(e -> t.line("RedisKeyExpiredEvent: %s", e));

            t.section("What is left behind");
            t.cli("--no-raw", "DBSIZE");
            t.line("sessions.count() = %d", sessions.count());
            assertThat(sessions.count()).isZero();
        }
    }
}