package com.ankurm.semanticcache;

import static org.assertj.core.api.Assertions.assertThat;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Random;

import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.embedding.EmbeddingModel;
import redis.clients.jedis.RedisClient;

/**
 * Every test writes one transcript under output/ and asserts the same numbers it prints. Needs a Redis with the
 * RediSearch module on 127.0.0.1:6393 (scripts/services-up.sh starts one).
 */
class SemanticCacheTest {

    static final int REDIS_PORT = 6393;
    static final String TENANT = "shop-a";

    static EmbeddingModel embeddings;
    static RedisClient jedis;
    static Dataset data;
    static Map<String, String> intentByText;

    @BeforeAll
    static void setUp() {
        embeddings = Embedder.miniLm();
        jedis = RedisClient.create("127.0.0.1", REDIS_PORT);
        data = new Dataset();
        intentByText = data.intentByText();
    }

    static SemanticCache freshCache(double threshold) {
        jedis.flushAll();
        return new SemanticCache(jedis, embeddings, threshold);
    }

    static SemanticCache seededCache(double threshold) {
        SemanticCache c = freshCache(threshold);
        data.seeds().forEach(r -> c.put(TENANT, r.text(), ScriptedModel.answerFor(r.intent())));
        return c;
    }

    static String intentOfAnswer(String answer) {
        return answer.substring(1, answer.indexOf(']'));
    }

    // ---------------------------------------------------------------- 1. what a similarity score looks like

    @Test
    void similarityScoresOfParaphrasesAndLookAlikes() {
        List<Double> paraphrase = new ArrayList<>(), lookAlike = new ArrayList<>(), offTopic = new ArrayList<>();
        Map<String, float[]> seedVec = new java.util.LinkedHashMap<>();
        data.seeds().forEach(r -> seedVec.put(r.intent(), embeddings.embed(r.text())));
        String worstLookAlike = "";
        double worstScore = -1;
        String lowestParaphrase = "";
        double lowestScore = 2;
        for (var r : data.paraphrases()) {
            float[] v = embeddings.embed(r.text());
            double own = Embedder.cosine(v, seedVec.get(r.intent()));
            paraphrase.add(own);
            if (own < lowestScore) {
                lowestScore = own;
                lowestParaphrase = r.text() + "  ->  " + seedText(r.intent());
            }
            for (var e : seedVec.entrySet()) {
                if (e.getKey().equals(r.intent())) continue;
                double s = Embedder.cosine(v, e.getValue());
                if (s > worstScore) {
                    worstScore = s;
                    worstLookAlike = r.text() + "  ->  " + seedText(e.getKey());
                }
            }
            double best = seedVec.entrySet().stream().filter(e -> !e.getKey().equals(r.intent()))
                    .mapToDouble(e -> Embedder.cosine(v, e.getValue())).max().orElse(0);
            lookAlike.add(best);
        }
        for (var r : data.offTopic()) {
            float[] v = embeddings.embed(r.text());
            offTopic.add(seedVec.values().stream().mapToDouble(s -> Embedder.cosine(v, s)).max().orElse(0));
        }

        // Which different intents are closest to each other? Every pair of cached seeds, 30 x 29 / 2 = 435 pairs.
        record Pair(String a, String b, double score) { }
        List<Pair> pairs = new ArrayList<>();
        List<String> seedIntents = new ArrayList<>(seedVec.keySet());
        for (int i = 0; i < seedIntents.size(); i++)
            for (int j = i + 1; j < seedIntents.size(); j++)
                pairs.add(new Pair(seedIntents.get(i), seedIntents.get(j),
                        (1 + Embedder.cosine(seedVec.get(seedIntents.get(i)), seedVec.get(seedIntents.get(j)))) / 2));
        pairs.sort(Comparator.comparingDouble(Pair::score).reversed());

        // Does Redis report the same number? Ask for one question and compare with the cosine computed in Java.
        SemanticCache cache = seededCache(0.0);
        var probe = data.paraphrases().get(0);
        var redisHit = cache.nearest(TENANT, probe.text()).orElseThrow();
        double javaCosine = Embedder.cosine(embeddings.embed(probe.text()), seedVec.get(probe.intent()));
        var exact = cache.nearest(TENANT, data.seeds().get(0).text()).orElseThrow();

        try (var t = new Transcript("01-similarity-scores.txt", "What a similarity score looks like (all-MiniLM-L6-v2, cosine)")) {
            t.line("Plain cosine similarity (NOT the Spring AI score, see the end) of each later phrasing with the cached phrasing of the SAME question (%d pairs):", paraphrase.size());
            t.line("  %s", summary(paraphrase));
            t.line("  lowest: %.3f   %s", lowestScore, lowestParaphrase);
            t.blank();
            t.line("Best score each of those phrasings gets against a cached question of a DIFFERENT intent (look-alikes like reset password / reset router):");
            t.line("  %s", summary(lookAlike));
            t.line("  highest: %.3f   %s", worstScore, worstLookAlike);
            t.blank();
            t.line("Best score of %d off-topic questions against any cached question:", offTopic.size());
            t.line("  %s", summary(offTopic));
            t.blank();
            t.line("Does Redis report the same number? \"%s\"", probe.text());
            t.line("  cosine computed in Java        : %.6f", javaCosine);
            t.line("  Document.getScore() from Redis : %.6f   (matched stored question: \"%s\")", redisHit.score(), redisHit.storedQuestion());
            t.line("  (1 + cosine) / 2               : %.6f   <- the score is this, not the cosine", (1 + javaCosine) / 2);
            t.line("  same question as the cached one -> score %.6f", exact.score());
            t.blank();
            t.line("The closest pairs of DIFFERENT cached questions (Spring AI score; 435 pairs in all). Above the threshold, one answers for the other:");
            for (int i = 0; i < 8; i++)
                t.line("  %.3f  \"%s\"  /  \"%s\"", pairs.get(i).score(), seedText(pairs.get(i).a()), seedText(pairs.get(i).b()));
            t.blank();
            t.line("Spring AI's similarityThreshold is applied to that score. Threshold -> the cosine it really means (cosine = 2 x score - 1):");
            for (double th : new double[]{0.70, 0.80, 0.85, 0.90, 0.95}) t.line("  similarityThreshold %.2f  =  cosine %.2f", th, 2 * th - 1);
        }
        assertThat(redisHit.score()).isCloseTo((1 + javaCosine) / 2, org.assertj.core.data.Offset.offset(1e-3));
        assertThat(exact.score()).isGreaterThan(0.999);
        // the two populations overlap: the reason a threshold cannot be both safe and generous
        assertThat(max(lookAlike)).isGreaterThan(min(paraphrase));
        assertThat(pairs.get(0).score()).isGreaterThan(0.80);
    }

    static String seedText(String intent) {
        return data.seeds().stream().filter(r -> r.intent().equals(intent)).findFirst().orElseThrow().text();
    }

    static double min(List<Double> v) { return v.stream().mapToDouble(Double::doubleValue).min().orElse(0); }

    static double max(List<Double> v) { return v.stream().mapToDouble(Double::doubleValue).max().orElse(0); }

    static String summary(List<Double> v) {
        List<Double> s = v.stream().sorted().toList();
        return String.format("min %.3f   p10 %.3f   median %.3f   p90 %.3f   max %.3f", s.get(0), s.get((int) (s.size() * 0.1)),
                s.get(s.size() / 2), s.get((int) (s.size() * 0.9) - 1), s.get(s.size() - 1));
    }

    // ---------------------------------------------------------------- 2. the advisor in a ChatClient

    @Test
    void theAdvisorServesAParaphraseWithoutCallingTheModel() {
        SemanticCache cache = freshCache(0.80);
        ScriptedModel model = new ScriptedModel(intentByText, 0);
        SemanticCacheAdvisor advisor = new SemanticCacheAdvisor(cache, 0);
        ChatClient client = ChatClient.builder(model).defaultAdvisors(advisor).build();

        String[][] asks = {
                {"reset-password", "How do I reset my account password?"},
                {"reset-password", "How do I reset my account password?"},
                {"reset-password", "I forgot my password, how can I get back into my account?"},
                {"reset-router", "How do I reset my router to factory settings?"},
                {"ood", "What is the capital of Australia?"},
        };
        try (var t = new Transcript("02-advisor-in-a-chat-client.txt", "SemanticCacheAdvisor in front of a ChatClient (threshold 0.80)")) {
            for (String[] a : asks) {
                long before = model.calls.get();
                var resp = client.prompt().user(a[1]).call().chatClientResponse();
                boolean hit = Boolean.TRUE.equals(resp.context().get(SemanticCacheAdvisor.HIT));
                String answer = resp.chatResponse().getResult().getOutput().getText();
                t.line("ask : %s", a[1]);
                t.line("  -> %s%s, model calls so far %d, answer starts \"%s\"", hit ? "CACHE HIT, score " : "model call",
                        hit ? String.format("%.3f", (Double) resp.context().get(SemanticCacheAdvisor.SCORE)) : "", model.calls.get(),
                        answer.substring(0, answer.indexOf(']') + 1));
                assertThat(model.calls.get() - before).isEqualTo(hit ? 0 : 1);
            }
            t.blank();
            t.line("advisor counters: %d hits, %d misses; model received %d calls for %d questions; entries in Redis: %d",
                    advisor.hits.get(), advisor.misses.get(), model.calls.get(), asks.length, cache.size());
        }
        assertThat(advisor.hits.get()).isEqualTo(2);
        assertThat(model.calls.get()).isEqualTo(3);
        assertThat(cache.size()).isEqualTo(3);
    }

    // ---------------------------------------------------------------- 3. the threshold sweep

    record Probe(String intent, String text, boolean offTopic, SemanticCache.Hit nearest) {
        String nearestIntent() { return nearest == null ? "-" : intentOfAnswer(nearest.answer()); }
    }

    static List<Probe> probes(SemanticCache cache) {
        List<Probe> out = new ArrayList<>();
        for (var r : data.paraphrases()) out.add(new Probe(r.intent(), r.text(), false, cache.nearest(TENANT, r.text()).orElse(null)));
        for (var r : data.offTopic()) out.add(new Probe(r.intent(), r.text(), true, cache.nearest(TENANT, r.text()).orElse(null)));
        return out;
    }

    @Test
    void thresholdSweepHitRateAgainstWrongAnswers() {
        SemanticCache cache = seededCache(0.0);
        List<Probe> probes = probes(cache);
        long nPara = probes.stream().filter(p -> !p.offTopic()).count();
        long nOff = probes.size() - nPara;
        double[] thresholds = {0.50, 0.60, 0.70, 0.75, 0.80, 0.85, 0.90, 0.95};
        List<float[]> sv = data.seeds().stream().map(r -> embeddings.embed(r.text())).toList();
        int[] confusable = new int[thresholds.length];
        for (int i = 0; i < sv.size(); i++)
            for (int j = i + 1; j < sv.size(); j++) {
                double sc = (1 + Embedder.cosine(sv.get(i), sv.get(j))) / 2;
                for (int k = 0; k < thresholds.length; k++) if (sc >= thresholds[k]) confusable[k]++;
            }
        try (var t = new Transcript("03-threshold-sweep.txt", "Threshold sweep: 30 cached questions, 90 rewordings and 20 off-topic questions asked")) {
            t.line("threshold = the value given to SearchRequest.similarityThreshold (Spring AI score); cosine = 2 x threshold - 1.");
            t.line("A served answer is RIGHT if it belongs to the intent of the question asked, WRONG if it belongs to another intent.");
            t.line("An off-topic question has no right answer in the cache, so any hit on it is WRONG.");
            t.blank();
            t.line("%9s | %6s | %8s | %9s | %14s | %15s | %s | %s", "threshold", "cosine", "hit rate", "right", "wrong (reworded)", "wrong (off-topic)", "wrong / all hits", "cached pairs of different intents that answer for each other");
            for (int ti = 0; ti < thresholds.length; ti++) {
                double th = thresholds[ti];
                long hits = 0, right = 0, wrongPara = 0, wrongOff = 0, paraHits = 0;
                for (Probe p : probes) {
                    if (p.nearest() == null || p.nearest().score() < th) continue;
                    hits++;
                    boolean ok = !p.offTopic() && p.nearestIntent().equals(p.intent());
                    if (!p.offTopic()) paraHits++;
                    if (ok) right++;
                    else if (p.offTopic()) wrongOff++;
                    else wrongPara++;
                }
                t.line("%9.2f | %6.2f | %7.0f%% | %5d/%-3d | %8d/%-6d | %9d/%-7d | %s | %s", th, 2 * th - 1, 100.0 * paraHits / nPara, right, nPara, wrongPara, nPara,
                        wrongOff, nOff, hits == 0 ? "-" : String.format("%.1f%%", 100.0 * (wrongPara + wrongOff) / hits), confusable[ti] + " of 435");
            }
            t.blank();
            t.line("Questions that were answered WRONGLY at threshold 0.80 (asked -> served from): %s", probes.stream().anyMatch(p -> p.nearest().score() >= 0.80 && (p.offTopic() || !p.nearestIntent().equals(p.intent()))) ? "" : "none among these 110 probes");
            for (Probe p : probes) {
                if (p.nearest() != null && p.nearest().score() >= 0.80 && (p.offTopic() || !p.nearestIntent().equals(p.intent())))
                    t.line("  %.3f  \"%s\"  ->  \"%s\"", p.nearest().score(), p.text(), p.nearest().storedQuestion());
            }
            t.blank();
            t.line("Rewordings the cache MISSED at threshold 0.80 (best stored question and its score):");
            for (Probe p : probes) {
                if (!p.offTopic() && (p.nearest() == null || p.nearest().score() < 0.80))
                    t.line("  %.3f  \"%s\"  (nearest: \"%s\")", p.nearest().score(), p.text(), p.nearest().storedQuestion());
            }
        }
        // the two ends of the dial
        long hitsAt50 = probes.stream().filter(p -> p.nearest().score() >= 0.50).count();
        long offHitsAt50 = probes.stream().filter(p -> p.offTopic() && p.nearest().score() >= 0.50).count();
        long wrongAt95 = probes.stream().filter(p -> p.nearest().score() >= 0.95 && (p.offTopic() || !p.nearestIntent().equals(p.intent()))).count();
        assertThat(hitsAt50).isGreaterThan(nPara);
        assertThat(offHitsAt50).isGreaterThan(0);
        assertThat(wrongAt95).isZero();
    }

    // ---------------------------------------------------------------- 4. replay a workload

    record Replay(double threshold, int requests, long modelCalls, long hits, long wrong, long promptTokens, long completionTokens,
                  double avgLookupMillis) { }

    static Replay replay(double threshold, List<Data> workload) {
        SemanticCache cache = freshCache(threshold);
        ScriptedModel model = new ScriptedModel(intentByText, 0);
        SemanticCacheAdvisor advisor = new SemanticCacheAdvisor(cache, 0);
        ChatClient client = ChatClient.builder(model).defaultAdvisors(advisor).build();
        long wrong = 0;
        long lookupNanos = 0;
        for (Data d : workload) {
            long t0 = System.nanoTime();
            var resp = client.prompt().user(d.text()).call().chatClientResponse();
            long elapsed = System.nanoTime() - t0;
            boolean hit = Boolean.TRUE.equals(resp.context().get(SemanticCacheAdvisor.HIT));
            if (hit) {
                lookupNanos += elapsed;
                String served = intentOfAnswer(resp.chatResponse().getResult().getOutput().getText());
                if (!served.equals(d.intent())) wrong++;
            }
        }
        double avgHitMillis = advisor.hits.get() == 0 ? 0 : lookupNanos / 1e6 / advisor.hits.get();
        return new Replay(threshold, workload.size(), model.calls.get(), advisor.hits.get(), wrong, model.promptTokens.get(),
                model.completionTokens.get(), avgHitMillis);
    }

    record Data(String intent, String text) { }

    /** 600 requests: popular intents asked far more often, each time in a random one of its four phrasings, 15% off-topic. */
    static List<Data> workload() {
        Random rnd = new Random(42);
        List<String> intents = data.seeds().stream().map(Dataset.Row::intent).toList();
        Map<String, List<String>> phrasings = new java.util.LinkedHashMap<>();
        data.rows.stream().filter(r -> !r.role().equals("ood")).forEach(r -> phrasings.computeIfAbsent(r.intent(), k -> new ArrayList<>()).add(r.text()));
        double[] weight = new double[intents.size()];
        double total = 0;
        for (int i = 0; i < weight.length; i++) {
            weight[i] = 1.0 / (i + 1);          // Zipf: the first intent is asked 30 times as often as the last
            total += weight[i];
        }
        List<Data> out = new ArrayList<>();
        List<Dataset.Row> off = data.offTopic();
        for (int n = 0; n < 600; n++) {
            if (rnd.nextDouble() < 0.15) {
                var r = off.get(rnd.nextInt(off.size()));
                out.add(new Data(r.intent(), r.text()));
                continue;
            }
            double x = rnd.nextDouble() * total;
            int i = 0;
            while (i < weight.length - 1 && (x -= weight[i]) > 0) i++;
            List<String> ph = phrasings.get(intents.get(i));
            out.add(new Data(intents.get(i), ph.get(rnd.nextInt(ph.size()))));
        }
        return out;
    }

    static final double PRICE_IN_PER_MILLION = 2.50, PRICE_OUT_PER_MILLION = 10.00;   // illustrative, not a vendor's price sheet

    static double dollars(long in, long out) {
        return in / 1e6 * PRICE_IN_PER_MILLION + out / 1e6 * PRICE_OUT_PER_MILLION;
    }

    @Test
    void replayingSixHundredRequests() {
        List<Data> workload = workload();
        long distinctTexts = workload.stream().map(Data::text).distinct().count();
        long exactRepeats = workload.size() - distinctTexts;
        ScriptedModel baseline = new ScriptedModel(intentByText, 0);
        ChatClient plain = ChatClient.builder(baseline).build();
        workload.forEach(d -> plain.prompt().user(d.text()).call().content());
        double baseCost = dollars(baseline.promptTokens.get(), baseline.completionTokens.get());

        List<Replay> runs = new ArrayList<>();
        for (double th : new double[]{0.70, 0.80, 0.90, 0.95, 0.9999}) runs.add(replay(th, workload));

        try (var t = new Transcript("04-replay-600-requests.txt", "600 requests through the advisor, by threshold")) {
            t.line("Workload: 600 requests (seed 42), 30 intents asked with Zipf popularity in random phrasings, 15% off-topic.");
            t.line("  %d distinct texts, so a plain exact-match cache could answer at most %d of 600 (%.0f%%).", distinctTexts, exactRepeats, 100.0 * exactRepeats / 600);
            t.line("Cost model (ILLUSTRATIVE): $%.2f per million input tokens, $%.2f per million output tokens, tokens = characters / 4.", PRICE_IN_PER_MILLION, PRICE_OUT_PER_MILLION);
            t.blank();
            t.line("%-22s | %10s | %8s | %6s | %13s | %9s | %s", "setup", "model calls", "hit rate", "wrong", "wrong of hits", "model $", "saved");
            t.line("%-22s | %10d | %8s | %6s | %13s | %9.4f | %s", "no cache", baseline.calls.get(), "-", "-", "-", baseCost, "-");
            for (Replay r : runs) {
                double cost = dollars(r.promptTokens, r.completionTokens);
                t.line("%-22s | %10d | %7.1f%% | %6d | %12.1f%% | %9.4f | %.0f%%",
                        r.threshold >= 0.999 ? "cache, exact match only" : String.format("cache, threshold %.2f", r.threshold), r.modelCalls,
                        100.0 * r.hits / r.requests, r.wrong, r.hits == 0 ? 0 : 100.0 * r.wrong / r.hits, cost, 100.0 * (1 - cost / baseCost));
            }
            t.blank();
            t.line("Average time to serve a cache hit (embed the question, search Redis, build the response): %s",
                    runs.stream().filter(r -> r.threshold < 0.999).map(r -> String.format("%.0f ms at %.2f", r.avgLookupMillis, r.threshold)).reduce((a, b) -> a + ", " + b).orElse(""));
            t.line("Saved dollars exclude the cost of embedding (local model here, CPU only) and of running Redis.");
        }
        Replay at80 = runs.get(1), exact = runs.get(4);
        assertThat(baseline.calls.get()).isEqualTo(600);
        assertThat(exact.hits).isEqualTo(exactRepeats);
        assertThat(at80.hits).isGreaterThan(exact.hits);
        assertThat(runs.get(0).wrong).isGreaterThanOrEqualTo(at80.wrong);
    }

    // ---------------------------------------------------------------- 5. the traps

    @Test
    void lookAlikesThatShareWordsButNotMeaning() {
        String[][] pairs = {
                {"How do I cancel my order?", "How do I keep my order and not cancel it?"},
                {"Is shipping free for orders over $50?", "Is shipping free for orders over $500?"},
                {"Do you ship to Canada?", "Do you ship to Cuba?"},
                {"What is the warranty on the laptop?", "What is the warranty on the monitor?"},
                {"Where is order 1001?", "Where is order 2002?"},
                {"Can I get a refund within 30 days?", "Can I get a refund after 30 days?"},
                {"I want to delete my account", "I do not want to delete my account"},
        };
        try (var t = new Transcript("05-lookalikes.txt", "Pairs that read alike and mean different things (cosine similarity)")) {
            t.line("%-45s | %-45s | %6s | %s", "cached question", "new question", "cosine", "Spring AI score = (1 + cosine) / 2");
            for (String[] p : pairs) {
                double c = Embedder.cosine(embeddings.embed(p[0]), embeddings.embed(p[1]));
                double s = (1 + c) / 2;
                t.line("%-45s | %-45s | %6.3f | %.3f %s", p[0], p[1], c, s, s >= 0.90 ? "<- served at threshold 0.90 and 0.80" : s >= 0.80 ? "<- served at threshold 0.80" : "");
                assertThat(s).isBetween(0.0, 1.0);
            }
        }
        long servedAt90 = java.util.Arrays.stream(pairs).filter(p -> (1 + Embedder.cosine(embeddings.embed(p[0]), embeddings.embed(p[1]))) / 2 >= 0.90).count();
        assertThat(servedAt90).isGreaterThan(0);
    }

    @Test
    void tenantsDoNotShareAnswers() {
        SemanticCache cache = freshCache(0.80);
        ScriptedModel model = new ScriptedModel(intentByText, 0);
        SemanticCacheAdvisor advisor = new SemanticCacheAdvisor(cache, 0);
        ChatClient client = ChatClient.builder(model).defaultAdvisors(advisor).build();
        String q = "How do I reset my account password?";
        client.prompt().user(q).advisors(a -> a.param(SemanticCacheAdvisor.TENANT, "shop-a")).call().content();
        var other = client.prompt().user(q).advisors(a -> a.param(SemanticCacheAdvisor.TENANT, "shop-b")).call().chatClientResponse();
        var same = client.prompt().user(q).advisors(a -> a.param(SemanticCacheAdvisor.TENANT, "shop-a")).call().chatClientResponse();
        // the same lookup without the tenant filter, to show what the filter prevents
        var unfiltered = cache.nearest("shop-a", q);
        try (var t = new Transcript("06-tenant-isolation.txt", "Same question, two tenants")) {
            t.line("shop-a asks \"%s\"  -> model call (calls so far: 1)", q);
            t.line("shop-b asks the identical question -> %s (model calls so far: %d)",
                    Boolean.TRUE.equals(other.context().get(SemanticCacheAdvisor.HIT)) ? "CACHE HIT (leak)" : "model call, no hit", model.calls.get());
            t.line("shop-a asks again               -> %s", Boolean.TRUE.equals(same.context().get(SemanticCacheAdvisor.HIT)) ? "cache hit" : "model call");
            t.line("entries in the index: %d (one per tenant)", cache.size());
            t.line("The filter is a tag on the document and an expression on the search: tenant == 'shop-b'.");
            t.line("nearest(shop-a, q) without asking as shop-b still finds: %s", unfiltered.map(h -> "\"" + h.storedQuestion() + "\"").orElse("nothing"));
        }
        assertThat(Boolean.TRUE.equals(other.context().get(SemanticCacheAdvisor.HIT))).isFalse();
        assertThat(Boolean.TRUE.equals(same.context().get(SemanticCacheAdvisor.HIT))).isTrue();
        assertThat(cache.size()).isEqualTo(2);
    }

    @Test
    void anExpiredEntryStopsBeingServed() throws Exception {
        SemanticCache cache = freshCache(0.80);
        String id = cache.put(TENANT, "How do I reset my account password?", ScriptedModel.answerFor("reset-password"));
        boolean before = cache.lookup(TENANT, "How do I reset my account password?").isPresent();
        long sizeBefore = cache.size();
        cache.expireAfterSeconds(id, 1);
        Thread.sleep(1500);
        boolean after = cache.lookup(TENANT, "How do I reset my account password?").isPresent();
        try (var t = new Transcript("07-expiry.txt", "Expiry with EXPIRE on the document's key")) {
            t.line("stored one entry, index holds %d document(s); lookup -> %s", sizeBefore, before ? "hit" : "miss");
            t.line("EXPIRE sc:<id> 1, wait 1.5 s; index now holds %d document(s); lookup -> %s", cache.size(), after ? "hit" : "miss");
        }
        assertThat(before).isTrue();
        assertThat(after).isFalse();
        assertThat(cache.size()).isZero();
    }

    @Test
    void anotherEmbeddingModelIsAnotherIndex() {
        SemanticCache cache = freshCache(0.80);
        cache.put(TENANT, "How do I reset my account password?", "a");
        EmbeddingModel eight = new EmbeddingModel() {
            @Override
            public org.springframework.ai.embedding.EmbeddingResponse call(org.springframework.ai.embedding.EmbeddingRequest r) {
                List<org.springframework.ai.embedding.Embedding> out = new ArrayList<>();
                for (int i = 0; i < r.getInstructions().size(); i++) out.add(new org.springframework.ai.embedding.Embedding(embed(r.getInstructions().get(i)), i));
                return new org.springframework.ai.embedding.EmbeddingResponse(out);
            }

            @Override
            public float[] embed(org.springframework.ai.document.Document d) {
                return new float[8];
            }

            @Override
            public float[] embed(String text) {
                float[] v = new float[8];
                v[Math.abs(text.hashCode()) % 8] = 1f;
                return v;
            }

            @Override
            public List<float[]> embed(List<String> texts) {
                return texts.stream().map(this::embed).toList();
            }

            @Override
            public int dimensions() { return 8; }
        };
        String outcome;
        try {
            // same index name and prefix, a model with 8 dimensions instead of 384; afterPropertiesSet() finds the index already exists
            var swapped = new org.springframework.ai.vectorstore.redis.RedisVectorStore[1];
            swapped[0] = org.springframework.ai.vectorstore.redis.RedisVectorStore.builder(jedis, eight)
                    .indexName(SemanticCache.INDEX).prefix(SemanticCache.PREFIX)
                    .metadataFields(org.springframework.ai.vectorstore.redis.RedisVectorStore.MetadataField.tag("tenant"), org.springframework.ai.vectorstore.redis.RedisVectorStore.MetadataField.text("answer"))
                    .initializeSchema(true).build();
            swapped[0].afterPropertiesSet();
            swapped[0].add(List.of(new org.springframework.ai.document.Document("x", "hello", Map.of("tenant", TENANT, "answer", "b"))));
            var found = swapped[0].similaritySearch(org.springframework.ai.vectorstore.SearchRequest.builder().query("hello").topK(1)
                    .filterExpression("tenant == '" + TENANT + "'").build());
            outcome = "no exception; search returned " + found.size() + " document(s): " + found.stream().map(d -> d.getText()).toList();
        } catch (Exception e) {
            Throwable root = e;
            while (root.getCause() != null) root = root.getCause();
            outcome = "exception " + root.getClass().getSimpleName() + ": " + root.getMessage();
        }
        try (var t = new Transcript("08-embedding-model-change.txt", "An index built for 384 dimensions, written by a model with 8")) {
            t.line("index created by MiniLM (384 dimensions); a new application version configures an 8-dimension model against the same index name.");
            t.line("add + search -> %s", outcome);
        }
        assertThat(outcome).isNotBlank();
    }

    // ---------------------------------------------------------------- 9. where the milliseconds go

    static double pct(List<Double> v, double p) {
        List<Double> s = v.stream().sorted().toList();
        return s.get(Math.min(s.size() - 1, (int) Math.ceil(p * s.size()) - 1));
    }

    @Test
    void whereTheTimeGoesOnAHit() {
        SemanticCache cache = seededCache(0.80);
        List<String> questions = new ArrayList<>();
        data.paraphrases().forEach(r -> questions.add(r.text()));
        data.offTopic().forEach(r -> questions.add(r.text()));
        for (int i = 0; i < 10; i++) embeddings.embed(questions.get(i));   // warm-up, not measured
        List<Double> embed = new ArrayList<>(), search = new ArrayList<>();
        for (String q : questions) {
            long t0 = System.nanoTime();
            embeddings.embed(q);
            long t1 = System.nanoTime();
            cache.lookup(TENANT, q);           // embeds again inside the store, then searches
            long t2 = System.nanoTime();
            embed.add((t1 - t0) / 1e6);
            search.add((t2 - t1) / 1e6 - (t1 - t0) / 1e6);
        }
        try (var t = new Transcript("09-hit-latency.txt", "Where the milliseconds go on a lookup (110 questions, 30 entries cached, 2 vCPUs)")) {
            t.line("embed the question (MiniLM in-process) : median %.1f ms, p95 %.1f ms", pct(embed, 0.5), pct(embed, 0.95));
            t.line("Redis KNN search with the tenant filter : median %.1f ms, p95 %.1f ms   (lookup time minus one embedding)", pct(search, 0.5), pct(search, 0.95));
            t.line("Timings drift run to run; the shape (embedding dominates, Redis is small) does not.");
        }
        assertThat(pct(embed, 0.5)).isGreaterThan(pct(search, 0.5));
    }

    @Test
    void againstARealLocalModel() throws Exception {
        org.junit.jupiter.api.Assumptions.assumeTrue(isUp(11555), "no Ollama on 127.0.0.1:11555");
        var api = org.springframework.ai.ollama.api.OllamaApi.builder().baseUrl("http://127.0.0.1:11555").build();
        var model = org.springframework.ai.ollama.OllamaChatModel.builder().ollamaApi(api)
                .options(org.springframework.ai.ollama.api.OllamaChatOptions.builder().model("tl").temperature(0.0).numPredict(60).build()).build();
        SemanticCache cache = freshCache(0.80);
        SemanticCacheAdvisor advisor = new SemanticCacheAdvisor(cache, 0);
        ChatClient client = ChatClient.builder(model).defaultAdvisors(advisor).build();
        client.prompt().user("Say hello").call().content();     // load the model, not measured
        cache = freshCache(0.80);
        advisor = new SemanticCacheAdvisor(cache, 0);
        client = ChatClient.builder(model).defaultAdvisors(advisor).build();

        List<Dataset.Row> seeds = data.seeds().subList(0, 6);
        List<Dataset.Row> paras = new ArrayList<>();
        for (var s : seeds) paras.add(data.paraphrases().stream().filter(p -> p.intent().equals(s.intent())).findFirst().orElseThrow());
        List<Double> missMs = new ArrayList<>(), hitMs = new ArrayList<>();
        int wrongHits = 0;
        try (var t = new Transcript("10-real-model-latency.txt", "A real local model (TinyLlama 1.1B, Q4_0, 60 tokens) behind the advisor, 2 vCPUs")) {
            for (var s : seeds) {
                long t0 = System.nanoTime();
                var resp = client.prompt().user(s.text()).call().chatClientResponse();
                double ms = (System.nanoTime() - t0) / 1e6;
                boolean hit = Boolean.TRUE.equals(resp.context().get(SemanticCacheAdvisor.HIT));
                if (hit) {
                    t.line("first ask   %6.0f ms  CACHE    \"%s\"  <- served the answer to \"%s\" (score %.3f), a DIFFERENT question", ms, s.text(),
                            resp.context().get(SemanticCacheAdvisor.MATCHED), (Double) resp.context().get(SemanticCacheAdvisor.SCORE));
                    wrongHits++;
                } else {
                    missMs.add(ms);
                    t.line("first ask   %6.0f ms  model    \"%s\"", ms, s.text());
                }
            }
            for (var p : paras) {
                long t0 = System.nanoTime();
                var resp = client.prompt().user(p.text()).call().chatClientResponse();
                double ms = (System.nanoTime() - t0) / 1e6;
                boolean hit = Boolean.TRUE.equals(resp.context().get(SemanticCacheAdvisor.HIT));
                if (hit) hitMs.add(ms);
                t.line("reworded    %6.0f ms  %s  \"%s\"%s", ms, hit ? "CACHE   " : "model   ", p.text(),
                        hit ? "  <- " + resp.context().get(SemanticCacheAdvisor.MATCHED) : "");
            }
            t.blank();
            t.line("median model call %.0f ms; median cache hit %.0f ms (%d of %d rewordings hit); first-time questions wrongly answered from the cache: %d of %d",
                    pct(missMs, 0.5), hitMs.isEmpty() ? 0 : pct(hitMs, 0.5), hitMs.size(), paras.size(), wrongHits, seeds.size());
        }
        assertThat(hitMs).isNotEmpty();
        assertThat(pct(hitMs, 0.5)).isLessThan(pct(missMs, 0.5));
    }

    static boolean isUp(int port) {
        try (var s = new java.net.Socket("127.0.0.1", port)) {
            return true;
        } catch (Exception e) {
            return false;
        }
    }
}