package com.easyagents.document.core.mineru;

import com.alibaba.fastjson2.JSONObject;
import com.easyagents.document.core.entity.ParseFile;
import com.easyagents.document.core.entity.ParseRequest;
import com.easyagents.document.core.entity.ParseResponse;
import com.easyagents.document.core.entity.ParseTaskInfo;
import com.easyagents.document.core.entity.ParseTaskStatus;
import com.easyagents.document.core.exception.DocumentParseException;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.Protocol;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;
import okio.Buffer;
import okio.BufferedSource;
import okio.Okio;
import okio.Source;
import okio.Timeout;
import org.junit.Assert;
import org.junit.Test;

import java.io.ByteArrayOutputStream;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

/**
 * MinerU 通用文档解析服务测试。
 *
 * @author Codex
 * @since 2026-04-16
 */
public class MineruDocumentParseServiceTest {

    @Test
    public void shouldForceAsyncResultArtifacts() {
        RecordingClient client = new RecordingClient(defaultProperties());
        MineruMapper mapper = new MineruMapper(defaultProperties());
        MineruDocumentParseService service = new MineruDocumentParseService(defaultProperties(), client, mapper);

        ParseRequest request = buildRequest();
        request.setReturnMarkdown(false);
        request.setReturnMiddleJson(false);
        request.setReturnContentList(false);
        request.setReturnModelOutput(false);
        request.setReturnImages(false);

        ParseTaskStatus status = service.submit(request);

        Assert.assertEquals("task-1", status.getTaskId());
        Assert.assertTrue(client.lastSubmitRequest.getReturnMarkdown());
        Assert.assertTrue(client.lastSubmitRequest.getReturnMiddleJson());
        Assert.assertTrue(client.lastSubmitRequest.getReturnContentList());
        Assert.assertTrue(client.lastSubmitRequest.getReturnModelOutput());
        Assert.assertTrue(client.lastSubmitRequest.getReturnImages());
    }

    @Test
    public void shouldUseTaskMetadataWhenQueryingAsyncZipResult() {
        RecordingClient client = new RecordingClient(defaultProperties());
        MineruMapper mapper = new MineruMapper(defaultProperties());
        MineruDocumentParseService service = new MineruDocumentParseService(defaultProperties(), client, mapper);

        ParseResponse response = service.queryResult("task-1");

        Assert.assertEquals("vlm-http-client", response.getBackend());
        Assert.assertEquals("3.0.9", response.getVersion());
        Assert.assertEquals(1, response.getResults().size());
        Assert.assertEquals("demo", response.getResults().get(0).getFileName());
    }

    @Test
    public void shouldReturnCompletedResultInTaskInfo() {
        RecordingClient client = new RecordingClient(defaultProperties());
        MineruMapper mapper = new MineruMapper(defaultProperties());
        MineruDocumentParseService service = new MineruDocumentParseService(defaultProperties(), client, mapper);

        ParseTaskInfo taskInfo = service.queryTaskInfo("task-1");

        Assert.assertEquals("completed", taskInfo.getStatus());
        Assert.assertNotNull(taskInfo.getResult());
        Assert.assertEquals(1, taskInfo.getResult().getResults().size());
        Assert.assertEquals(1, client.queryResultZipCount);
    }

    @Test
    public void shouldSendRepeatedLangListFields() {
        InspectingMultipartClient client = new InspectingMultipartClient(defaultProperties());
        ParseRequest request = buildRequest();
        request.setLanguages(java.util.Arrays.asList("zh", "en"));

        client.parse(request);

        Assert.assertEquals(2, countOccurrences(client.lastMultipartBody, "name=\"lang_list\""));
        Assert.assertTrue(client.lastMultipartBody.contains("\r\nzh\r\n"));
        Assert.assertTrue(client.lastMultipartBody.contains("\r\nen\r\n"));
    }

    /**
     * 验证异步任务提交总超时会取消无响应的 HTTP 调用。
     *
     * @throws Exception 本地测试套接字异常
     */
    @Test
    public void submitShouldCancelUnresponsiveCallAtConfiguredTimeout() throws Exception {
        AtomicReference<Socket> acceptedSocket = new AtomicReference<Socket>();
        try (ServerSocket serverSocket = new ServerSocket(0)) {
            Thread serverThread = new Thread(() -> {
                try {
                    Socket socket = serverSocket.accept();
                    acceptedSocket.set(socket);
                    while (socket.getInputStream().read() >= 0) {
                        // 持续读取请求但不返回响应，模拟 MinerU 提交接口失去响应。
                    }
                } catch (IOException ignore) {
                    // 客户端超时取消或测试关闭套接字后结束服务线程。
                }
            }, "mineru-submit-timeout-test-server");
            serverThread.setDaemon(true);
            serverThread.start();

            MineruProperties properties = defaultProperties();
            properties.setBaseUrl("http://127.0.0.1:" + serverSocket.getLocalPort());
            properties.setSubmitTimeoutMs(200);
            properties.setConnectTimeoutMs(5000);
            properties.setReadTimeoutMs(5000);
            properties.setWriteTimeoutMs(5000);
            MineruClient client = new MineruClient(properties, new MineruMapper(properties));

            long startedAt = System.currentTimeMillis();
            try {
                client.submit(buildRequest());
                Assert.fail("Expected MinerU submit timeout");
            } catch (DocumentParseException expected) {
                long elapsed = System.currentTimeMillis() - startedAt;
                Assert.assertTrue("Submit call should be cancelled promptly, elapsed=" + elapsed, elapsed < 2000);
            } finally {
                Socket socket = acceptedSocket.get();
                if (socket != null) {
                    socket.close();
                }
                serverThread.join(1000);
            }
        }
    }

    /**
     * 验证 MinerU 网关返回 5xx 时会在上传文件前立即失败，并复用短期不可用缓存。
     *
     * @throws Exception 本地测试套接字异常
     */
    @Test
    public void submitShouldFailBeforeMultipartUploadWhenProbeReturnsServerError() throws Exception {
        AtomicInteger requestCount = new AtomicInteger();
        AtomicReference<String> requestLine = new AtomicReference<String>();
        Thread serverThread;
        try (ServerSocket serverSocket = new ServerSocket(0)) {
            serverThread = new Thread(() -> {
                while (!serverSocket.isClosed()) {
                    try (Socket socket = serverSocket.accept();
                         BufferedReader reader = new BufferedReader(
                             new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8))) {
                        requestCount.incrementAndGet();
                        requestLine.set(reader.readLine());
                        String header;
                        while ((header = reader.readLine()) != null && !header.isEmpty()) {
                            // 读取完整请求头后再返回网关错误。
                        }
                        byte[] body = "Service Unavailable".getBytes(StandardCharsets.UTF_8);
                        String responseHeaders = "HTTP/1.1 503 Service Unavailable\r\n"
                            + "Content-Type: text/plain\r\n"
                            + "Content-Length: " + body.length + "\r\n"
                            + "Connection: close\r\n\r\n";
                        socket.getOutputStream().write(responseHeaders.getBytes(StandardCharsets.UTF_8));
                        socket.getOutputStream().write(body);
                        socket.getOutputStream().flush();
                    } catch (IOException ignore) {
                        // 测试结束关闭 ServerSocket 后退出服务线程。
                    }
                }
            }, "mineru-availability-probe-test-server");
            serverThread.setDaemon(true);
            serverThread.start();

            MineruProperties properties = defaultProperties();
            properties.setBaseUrl("http://127.0.0.1:" + serverSocket.getLocalPort());
            MineruClient client = new MineruClient(properties, new MineruMapper(properties));

            long startedAt = System.currentTimeMillis();
            for (int attempt = 0; attempt < 2; attempt++) {
                try {
                    client.submit(buildRequest());
                    Assert.fail("Expected MinerU availability failure");
                } catch (DocumentParseException expected) {
                    Assert.assertTrue(expected.getMessage().contains("status=503"));
                }
            }
            long elapsed = System.currentTimeMillis() - startedAt;
            Assert.assertTrue("Server error should fail promptly, elapsed=" + elapsed, elapsed < 2000);
        }
        serverThread.join(1000);

        Assert.assertEquals("Cached failure should avoid repeated probes", 1, requestCount.get());
        Assert.assertTrue(requestLine.get().startsWith("GET /health HTTP/1.1"));
    }

    /**
     * 验证任务接口返回 5xx 后不会等待或读取异常响应体。
     */
    @Test
    public void submitShouldThrowServerErrorWithoutReadingResponseBody() {
        AtomicInteger requestCount = new AtomicInteger();
        AtomicReference<Boolean> errorBodyRead = new AtomicReference<Boolean>(false);
        OkHttpClient httpClient = new OkHttpClient.Builder()
            .addInterceptor(chain -> {
                int currentRequest = requestCount.incrementAndGet();
                Response.Builder responseBuilder = new Response.Builder()
                    .request(chain.request())
                    .protocol(Protocol.HTTP_1_1);
                if (currentRequest == 1) {
                    return responseBuilder
                        .code(200)
                        .message("OK")
                        .body(ResponseBody.create((MediaType) null, new byte[0]))
                        .build();
                }
                ResponseBody trackingBody = new ResponseBody() {

                    private final BufferedSource source = Okio.buffer(new Source() {

                        @Override
                        public long read(Buffer sink, long byteCount) {
                            errorBodyRead.set(true);
                            return -1;
                        }

                        @Override
                        public Timeout timeout() {
                            return Timeout.NONE;
                        }

                        @Override
                        public void close() {
                            // 无底层资源需要关闭。
                        }
                    });

                    @Override
                    public MediaType contentType() {
                        return MediaType.parse("text/plain");
                    }

                    @Override
                    public long contentLength() {
                        return 19;
                    }

                    @Override
                    public BufferedSource source() {
                        return source;
                    }
                };
                return responseBuilder
                    .code(500)
                    .message("Internal Server Error")
                    .body(trackingBody)
                    .build();
            })
            .build();
        MineruProperties properties = defaultProperties();
        MineruClient client = new MineruClient(properties, httpClient, new MineruMapper(properties));

        try {
            client.submit(buildRequest());
            Assert.fail("Expected MinerU server error");
        } catch (DocumentParseException expected) {
            Assert.assertTrue(expected.getMessage().contains("status=500"));
        }

        Assert.assertEquals(2, requestCount.get());
        Assert.assertFalse("5xx response body should not be read", errorBodyRead.get());
    }

    private ParseRequest buildRequest() {
        ParseRequest request = new ParseRequest();
        request.addFile(ParseFile.of("demo.pptx", "ppt".getBytes(StandardCharsets.UTF_8)));
        return request;
    }

    private MineruProperties defaultProperties() {
        MineruProperties properties = new MineruProperties();
        properties.setBaseUrl("http://127.0.0.1:8000");
        properties.setResultTimeoutMs(50);
        properties.setPollIntervalMs(1);
        return properties;
    }

    private int countOccurrences(String source, String token) {
        int count = 0;
        int index = 0;
        while (source != null && token != null && !token.isEmpty() && (index = source.indexOf(token, index)) >= 0) {
            count++;
            index += token.length();
        }
        return count;
    }

    private static class RecordingClient extends MineruClient {

        private ParseRequest lastSubmitRequest;
        private int queryResultZipCount;

        private RecordingClient(MineruProperties properties) {
            super(properties, new MineruMapper(properties));
        }

        @Override
        public MineruTaskStatus submit(ParseRequest request) {
            this.lastSubmitRequest = request;
            MineruTaskStatus taskStatus = new MineruTaskStatus();
            taskStatus.setTaskId("task-1");
            taskStatus.setStatus("pending");
            return taskStatus;
        }

        @Override
        public MineruTaskStatus queryTask(String taskId) {
            MineruTaskStatus taskStatus = new MineruTaskStatus();
            taskStatus.setTaskId(taskId);
            taskStatus.setStatus("completed");
            taskStatus.setBackend("vlm-http-client");
            taskStatus.setVersion("3.0.9");
            return taskStatus;
        }

        @Override
        public byte[] queryResultZip(String taskId) {
            queryResultZipCount++;
            try {
                return buildZipResult();
            } catch (IOException exception) {
                throw new IllegalStateException("Failed to build test ZIP", exception);
            }
        }

        private static byte[] buildZipResult() throws IOException {
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            try (ZipOutputStream zipOutputStream = new ZipOutputStream(outputStream)) {
                addEntry(zipOutputStream, "demo/vlm/demo.md", "# title");
                addEntry(zipOutputStream, "demo/vlm/demo_middle.json", middleJson().toJSONString());
                addEntry(zipOutputStream, "demo/vlm/demo_content_list.json", contentList().toJSONString());
            }
            return outputStream.toByteArray();
        }

        private static void addEntry(ZipOutputStream zipOutputStream, String name, String content) throws IOException {
            zipOutputStream.putNextEntry(new ZipEntry(name));
            zipOutputStream.write(content.getBytes(StandardCharsets.UTF_8));
            zipOutputStream.closeEntry();
        }

        private static JSONObject middleJson() {
            JSONObject middleJson = new JSONObject();
            middleJson.put("_backend", "vlm");
            middleJson.put("_version_name", "3.0.9");
            middleJson.put("pdf_info", new com.alibaba.fastjson2.JSONArray());
            return middleJson;
        }

        private static com.alibaba.fastjson2.JSONArray contentList() {
            com.alibaba.fastjson2.JSONArray contentList = new com.alibaba.fastjson2.JSONArray();
            JSONObject text = new JSONObject();
            text.put("type", "text");
            text.put("text", "title");
            text.put("page_idx", 0);
            text.put("bbox", new com.alibaba.fastjson2.JSONArray());
            contentList.add(text);
            return contentList;
        }
    }

    private static class InspectingMultipartClient extends MineruClient {

        private String lastMultipartBody;

        private InspectingMultipartClient(MineruProperties properties) {
            super(properties, new MineruMapper(properties));
        }

        @Override
        protected JSONObject executeJsonRequest(String path, Request request) {
            try {
                Buffer buffer = new Buffer();
                request.body().writeTo(buffer);
                this.lastMultipartBody = buffer.readUtf8();
            } catch (IOException exception) {
                throw new IllegalStateException("Failed to inspect multipart body", exception);
            }
            return new JSONObject();
        }
    }
}
