package com.rj.qdrant;

import io.qdrant.client.QdrantClient;
import io.qdrant.client.QdrantGrpcClient;
import io.qdrant.client.grpc.Collections.CollectionInfo;
import io.qdrant.client.grpc.Collections.Distance;
import io.qdrant.client.grpc.Collections.VectorParams;
import io.qdrant.client.grpc.Points.Filter;
import io.qdrant.client.grpc.Points.GetPoints;
import io.qdrant.client.grpc.Points.PointId;
import io.qdrant.client.grpc.Points.PointStruct;
import io.qdrant.client.grpc.Points.Range;
import io.qdrant.client.grpc.Points.ScoredPoint;
import io.qdrant.client.grpc.Points.ScrollPoints;
import io.qdrant.client.grpc.Points.ScrollResponse;
import io.qdrant.client.grpc.Points.SearchPoints;
import io.qdrant.client.grpc.Points.UpdateResult;
import io.qdrant.client.grpc.Points.WithPayloadSelector;
import io.qdrant.client.grpc.Points.WithVectorsSelector;
import io.qdrant.client.grpc.QdrantOuterClass.HealthCheckReply;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.parallel.Execution;
import org.junit.jupiter.api.parallel.ExecutionMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import static io.qdrant.client.ConditionFactory.matchKeyword;
import static io.qdrant.client.ConditionFactory.range;
import static io.qdrant.client.PointIdFactory.id;
import static io.qdrant.client.ValueFactory.value;
import static io.qdrant.client.VectorsFactory.vectors;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * 使用官方 {@code io.qdrant:client}（gRPC）对 Qdrant 做端到端验证。
 * <p>
 * 连接参数（可选）：{@code -Dqdrant.host=127.0.0.1 -Dqdrant.grpc.port=6334}
 * </p>
 * <p>
 * 若本机未启动 Qdrant，{@link org.junit.jupiter.api.Assumptions} 会跳过全部用例。
 * </p>
 * <p>
 * 本类在 {@link BeforeAll} 中即创建空集合，且强制同线程顺序执行，避免「只跑后面几条用例」或并行执行时
 * 出现 {@code Collection doesn't exist}。
 * </p>
 */
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
@Execution(ExecutionMode.SAME_THREAD)
class QdrantClientVerificationTest {

    private static final Logger log = LoggerFactory.getLogger(QdrantClientVerificationTest.class);

    private static final String HOST = System.getProperty("qdrant.host", "192.168.1.44");
    private static final int GRPC_PORT = Integer.getInteger("qdrant.grpc.port", 6335);
    private static final Duration RPC_TIMEOUT = Duration.ofSeconds(35);

    private static final int VECTOR_SIZE = 4;
    private static final String COLLECTION = "java_verify_" + UUID.randomUUID().toString().replace("-", "");

    private static QdrantClient client;

    @BeforeAll
    static void connectOrAbort() throws Exception {
        log.info("======== Qdrant 验证开始 ========");
        log.info("目标: gRPC {}:{} ，集合名: {} ，向量维度: {}", HOST, GRPC_PORT, COLLECTION, VECTOR_SIZE);
        try {
            log.info("步骤: 创建 QdrantClient（明文 gRPC，无 TLS）…");
            client =
                    new QdrantClient(
                            QdrantGrpcClient.newBuilder(HOST, GRPC_PORT, false).build());
            log.info("步骤: healthCheck（超时 {}）…", RPC_TIMEOUT);
            HealthCheckReply health =
                    client.healthCheckAsync(RPC_TIMEOUT).get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
            assertNotNull(health);
            assertFalse(health.getTitle().isBlank(), "health title should be non-empty");
            log.info(
                    "[成功] 健康检查通过 | title={} | version={} | commit={}",
                    health.getTitle(),
                    health.getVersion(),
                    health.hasCommit() ? health.getCommit() : "(无)");

            // recreateCollectionAsync 会先 delete：集合不存在时部分 Qdrant 版本/客户端会报「无法删除」并中止，导致从未执行 create。
            VectorParams params =
                    VectorParams.newBuilder()
                            .setSize(VECTOR_SIZE)
                            .setDistance(Distance.Cosine)
                            .build();
            long rpcMs = RPC_TIMEOUT.toMillis();
            boolean exists =
                    client
                            .collectionExistsAsync(COLLECTION, RPC_TIMEOUT)
                            .get(rpcMs, TimeUnit.MILLISECONDS);
            if (exists) {
                log.info("步骤: 集合 '{}' 已存在，先 delete 再 create，得到干净空集合…", COLLECTION);
                client.deleteCollectionAsync(COLLECTION, RPC_TIMEOUT).get(rpcMs, TimeUnit.MILLISECONDS);
                log.info("[成功] 已删除旧集合");
            } else {
                log.info("步骤: 集合 '{}' 尚不存在，跳过 delete，直接 create…", COLLECTION);
            }
            log.info("步骤: createCollection '{}'（Cosine, size={}）…", COLLECTION, VECTOR_SIZE);
            client.createCollectionAsync(COLLECTION, params, RPC_TIMEOUT).get(rpcMs, TimeUnit.MILLISECONDS);
            log.info("[成功] @BeforeAll 已创建空集合 {}", COLLECTION);
        } catch (ExecutionException | TimeoutException | InterruptedException e) {
            log.warn(
                    "[失败/跳过] 无法连接 Qdrant gRPC {}:{} — {}: {}",
                    HOST,
                    GRPC_PORT,
                    e.getClass().getSimpleName(),
                    e.getMessage());
            Assumptions.abort(
                    "Qdrant gRPC 不可达 (%s:%d)：%s".formatted(HOST, GRPC_PORT, e.getMessage()));
        }
    }

    @AfterAll
    static void tearDown() throws Exception {
        if (client == null) {
            log.info("======== Qdrant 验证结束（未建立 client，无清理）========");
            return;
        }
        try {
            log.info("清理: 删除测试集合 '{}' …", COLLECTION);
            client.deleteCollectionAsync(COLLECTION, RPC_TIMEOUT).get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
            log.info("[成功] 已删除集合 {}", COLLECTION);
        } catch (Exception e) {
            log.info("清理: 删除集合跳过或失败（可能已不存在）— {}: {}", e.getClass().getSimpleName(), e.getMessage());
        }
        client.close();
        log.info("======== Qdrant 验证结束 ========");
    }

    @Test
    @Order(1)
    void listCollections_and_recreateCollection() throws Exception {
        log.info("--- 测试1: 列出集合并校验元数据（集合已在 @BeforeAll 创建）---");
        List<String> names =
                client.listCollectionsAsync(RPC_TIMEOUT).get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertNotNull(names);
        log.info("[成功] listCollections | 当前共有 {} 个集合: {}", names.size(), names);
        assertTrue(names.contains(COLLECTION), "listCollections 应包含本测试集合: " + COLLECTION);

        boolean exists =
                client.collectionExistsAsync(COLLECTION, RPC_TIMEOUT).get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertTrue(exists);
        log.info("[成功] collectionExists | '{}' = {}", COLLECTION, exists);

        CollectionInfo info =
                client.getCollectionInfoAsync(COLLECTION, RPC_TIMEOUT).get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertNotNull(info);
        long reportedSize = info.getConfig().getParams().getVectorsConfig().getParams().getSize();
        assertEquals(VECTOR_SIZE, reportedSize);
        log.info(
                "[成功] getCollectionInfo | 向量维度={}（期望 {}）| points_count={} | status={}",
                reportedSize,
                VECTOR_SIZE,
                info.getPointsCount(),
                info.getStatus());
        log.info("--- 测试1: 全部通过 ---");
    }

    @Test
    @Order(2)
    void upsert_count_retrieve() throws Exception {
        log.info("--- 测试2: upsert / count / retrieve ---");
        List<PointStruct> points =
                List.of(
                        PointStruct.newBuilder()
                                .setId(id(1L))
                                .setVectors(vectors(0.32f, 0.52f, 0.21f, 0.52f))
                                .putAllPayload(
                                        Map.of(
                                                "city", value("shanghai"),
                                                "score", value(10L)))
                                .build(),
                        PointStruct.newBuilder()
                                .setId(id(2L))
                                .setVectors(vectors(0.42f, 0.52f, 0.67f, 0.63f))
                                .putAllPayload(
                                        Map.of(
                                                "city", value("beijing"),
                                                "score", value(20L)))
                                .build(),
                        PointStruct.newBuilder()
                                .setId(id(3L))
                                .setVectors(vectors(0.12f, 0.82f, 0.37f, 0.23f))
                                .putAllPayload(
                                        Map.of(
                                                "city", value("shanghai"),
                                                "score", value(5L)))
                                .build());

        log.info("步骤: upsert 3 个点（id=1,2,3）…");
        UpdateResult upsert =
                client.upsertAsync(COLLECTION, points, RPC_TIMEOUT).get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertNotNull(upsert);
        log.info("[成功] upsert | status={} | operation_id={}", upsert.getStatus(), upsert.getOperationId());

        long total =
                client.countAsync(COLLECTION, RPC_TIMEOUT).get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertEquals(3L, total);
        log.info("[成功] count | 集合中点数 = {}（期望 3）", total);

        log.info("步骤: retrieve 点 id 1、2（含 payload 与向量）…");
        var retrieved =
                client
                        .retrieveAsync(
                                GetPoints.newBuilder()
                                        .setCollectionName(COLLECTION)
                                        .addIds(id(1L))
                                        .addIds(id(2L))
                                        .setWithPayload(WithPayloadSelector.newBuilder().setEnable(true).build())
                                        .setWithVectors(WithVectorsSelector.newBuilder().setEnable(true).build())
                                        .build(),
                                RPC_TIMEOUT)
                        .get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertEquals(2, retrieved.size());
        for (var p : retrieved) {
            log.info(
                    "[成功] retrieve 到点 id={} | payload keys={}",
                    p.getId().getNum(),
                    p.getPayloadMap().keySet());
        }
        log.info("--- 测试2: 全部通过 ---");
    }

    @Test
    @Order(3)
    void search_withFilter() throws Exception {
        log.info("--- 测试3: 向量检索 + 条件过滤 ---");
        List<ScoredPoint> shanghai =
                client
                        .searchAsync(
                                SearchPoints.newBuilder()
                                        .setCollectionName(COLLECTION)
                                        .addAllVector(List.of(0.32f, 0.52f, 0.21f, 0.52f))
                                        .setLimit(10)
                                        .setFilter(
                                                Filter.newBuilder().addMust(matchKeyword("city", "shanghai")).build())
                                        .setWithPayload(WithPayloadSelector.newBuilder().setEnable(true).build())
                                        .build(),
                                RPC_TIMEOUT)
                        .get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertEquals(2, shanghai.size());
        log.info("[成功] search + filter city=shanghai | 命中 {} 条", shanghai.size());
        for (ScoredPoint p : shanghai) {
            log.info("  · id={} score={} payload={}", p.getId().getNum(), p.getScore(), p.getPayloadMap());
        }

        log.info("步骤: search + filter score >= 15 …");
        List<ScoredPoint> byScore =
                client
                        .searchAsync(
                                SearchPoints.newBuilder()
                                        .setCollectionName(COLLECTION)
                                        .addAllVector(List.of(0.5f, 0.5f, 0.5f, 0.5f))
                                        .setLimit(10)
                                        .setFilter(
                                                Filter.newBuilder()
                                                        .addMust(range("score", Range.newBuilder().setGte(15L).build()))
                                                        .build())
                                        .build(),
                                RPC_TIMEOUT)
                        .get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertEquals(1, byScore.size());
        assertEquals(2L, byScore.get(0).getId().getNum());
        log.info(
                "[成功] search + range(score>=15) | 命中 id={} score={}",
                byScore.get(0).getId().getNum(),
                byScore.get(0).getScore());
        log.info("--- 测试3: 全部通过 ---");
    }

    @Test
    @Order(4)
    void setPayload_scroll_delete() throws Exception {
        log.info("--- 测试4: setPayload / scroll / delete ---");
        log.info("步骤: 对 id=1 设置 payload note=updated …");
        client
                .setPayloadAsync(
                        COLLECTION,
                        Map.of("note", value("updated")),
                        id(1L),
                        false,
                        null,
                        RPC_TIMEOUT)
                .get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        log.info("[成功] setPayload 已提交");

        log.info("步骤: retrieve id=1 校验 note …");
        var one =
                client
                        .retrieveAsync(
                                GetPoints.newBuilder()
                                        .setCollectionName(COLLECTION)
                                        .addIds(id(1L))
                                        .setWithPayload(
                                                WithPayloadSelector.newBuilder().setEnable(true).build())
                                        .setWithVectors(
                                                WithVectorsSelector.newBuilder().setEnable(false).build())
                                        .build(),
                                RPC_TIMEOUT)
                        .get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertEquals(1, one.size());
        assertEquals("updated", one.get(0).getPayloadMap().get("note").getStringValue());
        log.info("[成功] retrieve id=1 | note={}", one.get(0).getPayloadMap().get("note").getStringValue());

        log.info("步骤: scroll（limit=10, withPayload）…");
        ScrollResponse scroll =
                client
                        .scrollAsync(
                                ScrollPoints.newBuilder()
                                        .setCollectionName(COLLECTION)
                                        .setLimit(10)
                                        .setWithPayload(WithPayloadSelector.newBuilder().setEnable(true).build())
                                        .build(),
                                RPC_TIMEOUT)
                        .get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertFalse(scroll.getResultList().isEmpty());
        log.info(
                "[成功] scroll | 本页 {} 条 | next_page_offset present={}",
                scroll.getResultList().size(),
                scroll.hasNextPageOffset());

        log.info("步骤: delete 点 id=3 …");
        client
                .deleteAsync(COLLECTION, List.<PointId>of(id(3L)), RPC_TIMEOUT)
                .get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        log.info("[成功] delete 完成");

        long after =
                client.countAsync(COLLECTION, RPC_TIMEOUT).get(RPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertEquals(2L, after);
        log.info("[成功] count 删除后 = {}（期望 2）", after);
        log.info("--- 测试4: 全部通过 ---");
    }
}
