package com.xly.service;

import com.xly.agent.EventLogChatMemory;
import com.fasterxml.jackson.databind.ObjectMapper;
import dev.langchain4j.data.message.AiMessage;
import dev.langchain4j.data.message.ToolExecutionResultMessage;
import org.junit.jupiter.api.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * 「流式输出中途点【确认】」的并发不变量：对话流（EventLogChatMemory 落模型消息）与确认端点
 * （recordOutcome 落 confirm 事件）同时写同一会话时，**所有事件都在、各写者内部有序**。
 * 旧 chat:mem 整包读改写在此场景必丢更新；事件日志逐条追加从机制上消除该竞态。
 */
class EventLogConcurrencyTest {

    private static final String CONV = "u1:c-conc";

    @Test
    void concurrentStreamAndConfirmLoseNothing() throws Exception {
        InMemoryLedger log = new InMemoryLedger();
        EventLogChatMemory mem = new EventLogChatMemory(
                CONV, log, new EventProjectionService(new ObjectMapper()), null, false);

        int writers = 4;
        int perWriter = 100;
        ExecutorService pool = Executors.newFixedThreadPool(writers);
        CountDownLatch start = new CountDownLatch(1);
        CountDownLatch done = new CountDownLatch(writers);

        for (int w = 0; w < writers; w++) {
            final int id = w;
            pool.submit(() -> {
                try {
                    start.await();
                    for (int i = 0; i < perWriter; i++) {
                        if (id % 2 == 0) { // 对话流：模型消息经记忆适配器落账
                            if (i % 2 == 0) {
                                mem.add(AiMessage.from("w" + id + "-answer-" + i));
                            } else {
                                mem.add(ToolExecutionResultMessage.from(
                                        "t" + id + "-" + i, "readFormData", "w" + id + "-result-" + i));
                            }
                        } else { // 确认端点：按钮事件直接落账
                            log.append(CONV, "confirm", Map.of(
                                    "opId", "w" + id + "-" + i, "status", "executed",
                                    "msg", "", "description", "op w" + id + "-" + i));
                        }
                    }
                } catch (InterruptedException ignored) {
                } finally {
                    done.countDown();
                }
            });
        }
        start.countDown();
        assertTrue(done.await(30, TimeUnit.SECONDS));
        pool.shutdown();

        List<Map<String, Object>> events = log.events(CONV);
        assertEquals(writers * perWriter, events.size(), "并发写者的事件一条不丢");

        for (int w = 0; w < writers; w++) {
            List<Integer> seq = new ArrayList<>();
            for (Map<String, Object> ev : events) {
                String marker = String.valueOf(
                        ev.containsKey("opId") ? ev.get("opId")
                                : ev.containsKey("tcId") ? ev.get("tcId") : ev.get("text"));
                String prefix = marker.startsWith("t" + w + "-") ? "t" + w + "-" : "w" + w + "-";
                if (marker.startsWith(prefix)) {
                    String tail = marker.substring(marker.lastIndexOf('-') + 1);
                    try {
                        seq.add(Integer.parseInt(tail));
                    } catch (NumberFormatException ignore) {
                    }
                }
            }
            for (int i = 1; i < seq.size(); i++) {
                assertTrue(seq.get(i) > seq.get(i - 1), "写者 " + w + " 自身顺序保持");
            }
        }
    }
}