use super::mix::MixQuery;
use crate::error::QdrantError::*;
use crate::error::Result;
use crate::protocol::index::PayloadIndexInfo;
use crate::protocol::index::PayloadSchemaParams;
use crate::protocol::index::PayloadSchemaType;
use crate::protocol::points::*;
use crate::protocol::Payload;
use crate::protocol::UpdateStatus::Completed;
use crate::protocol::{Record, UpdateResult};
use crate::state::ApiState;
use lazy_static::lazy_static;
use log::*;
use model::collection::delete::CollectionRecordDeleted;
use model::collection::{CollectionRecord, CollectionRecordQueryBuilder};
use model::error::column_not_found;
use model::error::Error;
use model::meta::tenant::TenantCollectionMeta;
use model::meta::CollectionMeta;
use sqlx::mysql::MySqlArguments;
use sqlx::query::QueryAs;
use sqlx::{MySql, MySqlPool};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::spawn;
use tokio::try_join;

lazy_static! {
    static ref COLLECTION_ID_CACHE: moka::sync::Cache<String, u64> = moka::sync::Cache::new(1000);
}

fn get_tokens_map(
    ctx: Arc<context::Context>,
    payload: Option<Payload>,
    payload_schema: Option<String>,
) -> Option<HashMap<String, Vec<String>>> {
    let (payload, payload_schema) = match (payload, payload_schema) {
        (Some(payload), Some(payload_schema)) => (payload, payload_schema),
        _ => return None,
    };

    let index_info: HashMap<String, PayloadIndexInfo> = match serde_json::from_str(&payload_schema)
    {
        Ok(index_info) => index_info,
        Err(e) => {
            error!("[{}] Failed to parse payload schema: {:?}", ctx.id(), e);
            return None;
        }
    };

    let mut tokens_map = HashMap::new();
    for (path, index) in index_info {
        if index.data_type != PayloadSchemaType::Text {
            continue;
        }

        if let Some(content) = payload.0.get(&path) {
            let min_len = index.params.clone().map_or(2, |p| {
                if let PayloadSchemaParams::Text(p) = p {
                    p.min_token_len.map_or(2, |v| v)
                } else {
                    2
                }
            });
            let max_len = index.params.map_or(20, |p| {
                if let PayloadSchemaParams::Text(p) = p {
                    p.max_token_len.map_or(20, |v| v)
                } else {
                    20
                }
            });
            let tokens = fts_index::tokenizer::tokenize(&content.to_string(), min_len, max_len);
            tokens_map.insert(path, tokens);
        }
    }

    Some(tokens_map)
}

pub(crate) async fn upsert_points_list(
    ctx: Arc<context::Context>,
    meta_db: Arc<MySqlPool>,
    db: Arc<MySqlPool>,
    tenant_token: &str,
    collection_name: &str,
    points: Vec<PointStruct>,
) -> Result<UpdateResult> {
    let db_clone = db.clone();
    let (version, schema) = try_join!(
        CollectionMeta::bump_version(meta_db.as_ref(), tenant_token, collection_name),
        TenantCollectionMeta::get_by_collection_name(
            ctx.clone(),
            &db_clone,
            collection_name,
            false
        )
    )?;

    let schema = schema.ok_or(CollectionNotFoundWithTraceId(ctx.id().into()))?;

    let mut records = Vec::new();
    let mut tokens = Vec::new();
    for p in &points {
        records.push(CollectionRecord::new(
            p.id(),
            p.vector(),
            p.payload()?,
            version,
        ));
        tokens.push(get_tokens_map(
            ctx.clone(),
            p.payload.clone(),
            schema.payload_schema.clone(),
        ));
    }

    match CollectionRecord::upsert(ctx.clone(), &db, collection_name, records, tokens).await {
        Err(e) => {
            error!(
                "[{}] Upsert points error for tenant-token[{}], collection_name[{}]: {:?}",
                ctx.id(),
                tenant_token,
                collection_name,
                e
            );
            Err(e.into())
        }
        Ok(c) => Ok({
            // update data_count in collection with async way
            let tenant_token_clone = tenant_token.to_string();
            let collection_name_clone = collection_name.to_string();
            let meta_db_clone = meta_db.clone();
            tokio::spawn(async move {
                if let Err(e) = CollectionMeta::update_data_count(
                    &meta_db_clone,
                    &tenant_token_clone,
                    &collection_name_clone,
                    c as i64,
                )
                .await
                {
                    error!(
                        "Failed to update data_count for collection {}: {}",
                        &collection_name_clone, e
                    );
                }
            });

            UpdateResult {
                operation_id: Some(version),
                status: Completed,
            }
        }),
    }
}

pub(super) async fn upsert_points_batch(
    ctx: Arc<context::Context>,
    meta_db: Arc<MySqlPool>,
    db: Arc<MySqlPool>,
    tenant_token: &str,
    collection_name: &str,
    batch: PointsBatch,
) -> Result<UpdateResult> {
    upsert_points_list(
        ctx.clone(),
        meta_db,
        db,
        tenant_token,
        collection_name,
        batch
            .batch
            .ids
            .into_iter()
            .zip(batch.batch.vectors.into_iter())
            .zip(batch.batch.payloads.into_iter())
            .map(|((id, vector), payload)| PointStruct {
                id,
                vector,
                payload,
            })
            .collect(),
    )
    .await
}

pub(super) async fn delete_points(
    ctx: Arc<context::Context>,
    meta_db: Arc<MySqlPool>,
    tenant_db: Arc<MySqlPool>,
    tenant_token: &str,
    collection_name: &str,
    point_selector: PointsSelector,
    fts_version: u64,
) -> Result<UpdateResult> {
    let (version, payload_schema) = try_join!(
        CollectionMeta::bump_version(meta_db.as_ref(), tenant_token, collection_name),
        TenantCollectionMeta::get_by_collection_name(
            ctx.clone(),
            &tenant_db,
            collection_name,
            false
        ),
    )
    .map_err(|e| e.ctx(ctx.clone()))?;

    let payload_schema = payload_schema.ok_or(CollectionNotFoundWithTraceId(ctx.id().into()))?;

    CollectionRecordDeleted::create_table(tenant_db.as_ref(), collection_name, version)
        .await
        .map_err(|e| e.ctx(ctx.clone()))?;

    {
        let pool_clone = tenant_db.clone();
        let collection_name_clone = collection_name.to_string();
        spawn(async move {
            CollectionRecordDeleted::deleted_gc(&pool_clone, &collection_name_clone, fts_version)
                .await;
        });
    }

    let query_params: PointQueryParameters =
        (point_selector, payload_schema.payload_schema).try_into()?;
    let mut builder = CollectionRecordQueryBuilder::delete(collection_name, version);
    query_params.apply_predicates(builder.predicate(), false);
    let (move_sql, delete_sql) = builder.build(ctx.clone()).map_err(|e| e.ctx(ctx.clone()))?;

    let mut tx = tenant_db
        .begin()
        .await
        .map_err(|e| Error::from(e).ctx(ctx.clone()))?;
    let mut retry = true;
    loop {
        match bind_query_params!(query_params, sqlx::query(&move_sql), false)
            .execute(&mut *tx)
            .await
        {
            Ok(_) => break,
            Err(e) => {
                if retry
                    && column_not_found(&e)
                    && e.to_string().contains("__payload__metadata.doc_id")
                {
                    tx.rollback().await?;
                    tx = tenant_db
                        .begin()
                        .await
                        .map_err(|e| Error::from(e).ctx(ctx.clone()))?;
                    super::try_repair_field_schema(
                        ctx.clone(),
                        tenant_db.clone(),
                        tenant_token.to_string(),
                        collection_name.to_string(),
                    )
                    .await;
                    retry = false;
                    continue;
                }
                return Err(Error::from(e).ctx(ctx.clone()).into());
            }
        }
    }
    let result = bind_query_params!(query_params, sqlx::query(&delete_sql), false)
        .execute(&mut *tx)
        .await
        .map_err(|e| Error::from(e).ctx(ctx.clone()))?;
    tx.commit()
        .await
        .map_err(|e| Error::from(e).ctx(ctx.clone()))?;

    debug!(
        "[{}] Move deleted points {} -- {} -- {} -- query params: {:?}",
        ctx.id(),
        result.rows_affected(),
        move_sql,
        delete_sql,
        query_params
    );

    // update data_count in collection with async way
    let tenant_token_clone = tenant_token.to_string();
    let collection_name_clone = collection_name.to_string();
    let meta_db_clone = meta_db.clone();
    let c = result.rows_affected();
    tokio::spawn(async move {
        if let Err(e) = CollectionMeta::update_data_count(
            &meta_db_clone,
            &tenant_token_clone,
            &collection_name_clone,
            -(c as i64),
        )
        .await
        {
            error!(
                "Failed to update data_count for collection {}: {}",
                &collection_name_clone, e
            );
        }
    });

    Ok(UpdateResult {
        operation_id: Some(version),
        status: Completed,
    })
}

pub(super) async fn count_points(
    ctx: Arc<context::Context>,
    tenant_db: Arc<MySqlPool>,
    collection_name: &str,
) -> Result<PointsCount> {
    let points_count: PointsCount = CollectionRecord::count(ctx, &tenant_db, collection_name)
        .await?
        .into();
    Ok(points_count)
}

pub(super) async fn get_point(
    tenant_db: Arc<MySqlPool>,
    collection_name: &str,
    id: &str,
) -> Result<Record> {
    let mut builder = CollectionRecordQueryBuilder::select(collection_name);
    builder.with_payload(true).with_vector(true);
    builder.predicate().id_in(1).not_deleted();
    let record: Record = sqlx::query_as::<_, GenericPointModel>(&builder.build())
        .bind(id)
        .fetch_optional(tenant_db.as_ref())
        .await?
        .ok_or(PointNotFound)?
        .into();
    Ok(record)
}

pub(super) async fn get_points(
    tenant_db: Arc<MySqlPool>,
    collection_name: &str,
    point_request: PointRequest,
) -> Result<Vec<Record>> {
    let mut builder = CollectionRecordQueryBuilder::select(collection_name);
    builder
        .with_payload(point_request.with_payload)
        .with_vector(point_request.with_vector);
    builder
        .predicate()
        .id_in(point_request.ids.len())
        .not_deleted();
    let sql = builder.build();
    let mut query: QueryAs<MySql, GenericPointModel, MySqlArguments> =
        sqlx::query_as::<_, GenericPointModel>(&sql);
    for id in point_request.ids {
        query = query.bind(id.to_string());
    }
    let result: Vec<Record> = query
        .fetch_all(tenant_db.as_ref())
        .await?
        .into_iter()
        .map(Into::into)
        .collect();
    Ok(result)
}

pub(super) async fn search_points(
    ctx: Arc<context::Context>,
    state: ApiState,
    meta_db: Option<Arc<MySqlPool>>,
    tenant_db: Arc<MySqlPool>,
    tenant_token: &str,
    collection_name: &str,
    search_request: SearchRequest,
) -> Result<Vec<ScoredPoint>> {
    let payload_schema = TenantCollectionMeta::get_by_collection_name(
        ctx.clone(),
        &tenant_db,
        collection_name,
        false,
    )
    .await?
    .ok_or(CollectionNotFoundWithTraceId(ctx.id().into()))?;

    let mut mix_query = MixQuery::new(
        ctx.clone(),
        state,
        tenant_token.into(),
        meta_db.clone(),
        tenant_db.clone(),
        collection_name.to_string(),
        payload_schema.collection_id,
    );

    mix_query
        .with_version(true)
        .with_payload(search_request.with_payload)
        .with_vector(search_request.with_vector)
        .vector(search_request.vector, search_request.score_threshold)
        .offset(search_request.offset)
        .limit(search_request.limit)
        .point_query_param(
            search_request
                .filter
                .map(|f| (f, payload_schema.payload_schema).try_into())
                .transpose()?,
        );

    if mix_query.need_filter_by_tokens() {
        mix_query.execute().await
    } else {
        mix_query.query_points(tenant_db, 0).await
    }
}

pub(super) async fn scroll_points(
    ctx: Arc<context::Context>,
    state: ApiState,
    meta_db: Option<Arc<MySqlPool>>,
    tenant_db: Arc<MySqlPool>,
    tenant_token: &str,
    collection_name: &str,
    scroll_params: ScrollRequest,
) -> Result<ScrollResult> {
    let payload_schema = TenantCollectionMeta::get_by_collection_name(
        ctx.clone(),
        &tenant_db,
        collection_name,
        false,
    )
    .await?
    .ok_or(CollectionNotFoundWithTraceId(ctx.id().into()))?;

    let mut mix_query = MixQuery::new(
        ctx,
        state,
        tenant_token.into(),
        meta_db.clone(),
        tenant_db.clone(),
        collection_name.to_string(),
        payload_schema.collection_id,
    );

    mix_query
        .with_payload(scroll_params.with_payload)
        .with_vector(scroll_params.with_vector)
        .limit(scroll_params.limit)
        .point_query_param(
            scroll_params
                .filter
                .map(|f| (f, payload_schema.payload_schema).try_into())
                .transpose()?,
        );

    let result = if mix_query.need_filter_by_tokens() {
        mix_query.execute().await?
    } else {
        mix_query.query_points(tenant_db, 0).await?
    };

    Ok(ScrollResult {
        points: result,
        next_page_offset: None,
    })
}

pub(super) async fn set_payload(
    ctx: Arc<context::Context>,
    meta_db: Arc<MySqlPool>,
    tenant_db: Arc<MySqlPool>,
    tenant_token: &str,
    collection_name: &str,
    set_payload: SetPayloadRequest,
) -> Result<UpdateResult> {
    let (version, payload_schema) = try_join!(
        CollectionMeta::bump_version(meta_db.as_ref(), tenant_token, collection_name),
        TenantCollectionMeta::get_by_collection_name(
            ctx.clone(),
            tenant_db.as_ref(),
            collection_name,
            false,
        ),
    )?;
    let payload_schema = payload_schema.ok_or(CollectionNotFoundWithTraceId(ctx.id().into()))?;
    let payload: String = set_payload.payload.clone().try_into()?;
    let query_params: PointQueryParameters = set_payload.clone().try_into()?;
    let mut update_builder = CollectionRecordQueryBuilder::update(collection_name, version);
    query_params.apply_predicates(update_builder.predicate(), false);
    if let Some(ref path) = set_payload.key {
        update_builder.set_payload_with_path(path);
    } else {
        update_builder.set_payload();
    }

    // text fields
    let tokens = get_tokens_map(
        context::new(),
        Some(set_payload.payload.clone()),
        payload_schema.payload_schema,
    );
    if let Some(tokens) = tokens.clone() {
        tokens.iter().for_each(|(path, _)| {
            update_builder.set_tokens(path);
        });
    }

    let sql = update_builder.build();
    let mut query = sqlx::query(&sql);
    query = query.bind(payload);

    // bind tokens
    if let Some(tokens) = tokens {
        for (_, tokens) in tokens {
            query = query.bind(serde_json::to_string(&tokens).unwrap());
        }
    }

    let query = bind_query_params!(query_params, query, false);
    query.execute(tenant_db.as_ref()).await?;
    Ok(UpdateResult {
        operation_id: Some(version),
        status: Completed,
    })
}

pub(super) async fn overwrite_payload(
    ctx: Arc<context::Context>,
    meta_db: Arc<MySqlPool>,
    tenant_db: Arc<MySqlPool>,
    tenant_token: &str,
    collection_name: &str,
    set_payload: SetPayloadRequest,
) -> Result<UpdateResult> {
    let (version, payload_schema) = try_join!(
        CollectionMeta::bump_version(meta_db.as_ref(), tenant_token, collection_name),
        TenantCollectionMeta::get_by_collection_name(
            ctx.clone(),
            tenant_db.as_ref(),
            collection_name,
            false,
        ),
    )
    .map_err(|e| e.ctx(ctx.clone()))?;
    let payload_schema = payload_schema.ok_or(CollectionNotFoundWithTraceId(ctx.id().into()))?;

    let tokens = get_tokens_map(
        ctx.clone(),
        Some(set_payload.payload.clone()),
        payload_schema.payload_schema,
    );

    let payload = serde_json::to_string(&set_payload.payload)?;
    let query_params: PointQueryParameters = set_payload.try_into()?;
    let mut update_builder = CollectionRecordQueryBuilder::update(collection_name, version);
    query_params.apply_predicates(update_builder.predicate(), false);
    update_builder.overwrite_payload();

    // text fields
    if let Some(tokens) = tokens.clone() {
        tokens.iter().for_each(|(path, _)| {
            update_builder.set_tokens(path);
        });
    }

    let sql = update_builder.build();
    let mut query = sqlx::query(&sql);
    query = query.bind(payload);

    // bind tokens
    if let Some(tokens) = tokens {
        for (_, tokens) in tokens {
            query = query.bind(serde_json::to_string(&tokens).unwrap());
        }
    }

    let query = bind_query_params!(query_params, query, false);
    query
        .execute(tenant_db.as_ref())
        .await
        .map_err(|e| Error::from(e).ctx(ctx.clone()))?;

    Ok(UpdateResult {
        operation_id: Some(version),
        status: Completed,
    })
}

pub(super) async fn delete_payload(
    meta_db: Arc<MySqlPool>,
    tenant_db: Arc<MySqlPool>,
    tenant_token: &str,
    collection_name: &str,
    delete_payload: DeletePayloadRequest,
) -> Result<UpdateResult> {
    let version =
        CollectionMeta::bump_version(meta_db.as_ref(), tenant_token, collection_name).await?;
    let (points_or_filter, keys) = (
        (delete_payload.points, delete_payload.filter),
        delete_payload.keys,
    );
    let query_params: PointQueryParameters = points_or_filter.try_into()?;
    let mut update_builder = CollectionRecordQueryBuilder::update(collection_name, version);
    query_params.apply_predicates(update_builder.predicate(), false);
    update_builder.delete_payload(keys.len());
    let sql = update_builder.build();
    let mut query = sqlx::query(&sql);
    for key in keys {
        query = query.bind(key);
    }
    let query = bind_query_params!(query_params, query, false);
    query.execute(tenant_db.as_ref()).await?;
    Ok(UpdateResult {
        operation_id: Some(version),
        status: Completed,
    })
}

pub(super) async fn clear_payload(
    tenant_db: Arc<MySqlPool>,
    collection_name: &str,
    clear_payload: PointsSelector,
) -> Result<UpdateResult> {
    let version = CollectionMeta::bump_version(tenant_db.as_ref(), "", collection_name).await?;
    let query_params: PointQueryParameters = clear_payload.try_into()?;
    let mut update_builder = CollectionRecordQueryBuilder::update(collection_name, version);
    query_params.apply_predicates(update_builder.predicate(), false);
    update_builder.clear_payload();
    let sql = update_builder.build();
    let query = sqlx::query(&sql);
    let query = bind_query_params!(query_params, query, false);
    query.execute(tenant_db.as_ref()).await?;
    Ok(UpdateResult {
        operation_id: Some(version),
        status: Completed,
    })
}

pub(super) async fn update_vectors(
    ctx: Arc<context::Context>,
    tenant_db: Arc<MySqlPool>,
    collection_name: &str,
    vectors: UpdateVectors,
) -> Result<UpdateResult> {
    for vector in vectors.points {
        let mut query_builder = CollectionRecordQueryBuilder::update(collection_name, 0);
        query_builder.set_vector().predicate().id();
        let sql = query_builder.build();
        let mut query = sqlx::query(&sql);
        query = query.bind(vector.vector());
        query = query.bind(vector.id());
        query
            .execute(tenant_db.as_ref())
            .await
            .map_err(|e| Error::from(e).ctx(ctx.clone()))?;
    }

    Ok(UpdateResult {
        operation_id: Some(0),
        status: Completed,
    })
}

#[cfg(test)]
mod tests {
    use index::{CreateFieldIndex, PayloadFieldSchema};

    use super::*;
    use crate::collection::{self, actions::*};
    use crate::protocol::*;
    use crate::{state, Args};

    const COLLECTION_FOR_TEST_POINTS: &str = "collection_test_for_points";

    fn get_api_key() -> String {
        std::env::var("QDRANT_GATEWAY_APIKEY").unwrap()
    }

    async fn get_state() -> ApiState {
        state::new(Args::default()).await.unwrap()
    }

    async fn get_collection_id(
        ctx: Arc<context::Context>,
        tenant_token: &str,
        collection_name: &str,
        db: Arc<MySqlPool>,
    ) -> Result<u64> {
        if let Some(id) = COLLECTION_ID_CACHE.get(collection_name) {
            return Ok(id);
        }

        let collection =
            match CollectionMeta::get(ctx.clone(), &db, tenant_token, collection_name).await {
                Ok(Some(collection)) => collection,
                Ok(None) => return Err(CollectionNotFoundWithTraceId(ctx.id().to_string())),
                Err(e) => {
                    error!(
                        "[{}] Get collection-id tenant-token[{}] collection-name[{}] error: {:?}",
                        ctx.id(),
                        tenant_token,
                        collection_name,
                        e
                    );
                    return Err(e.into());
                }
            };
        COLLECTION_ID_CACHE.insert(collection_name.to_string(), collection.collection_id);
        Ok(collection.collection_id)
    }

    async fn setup_with_state(
        collection_name: &str,
        state: ApiState,
    ) -> (Arc<MySqlPool>, Arc<MySqlPool>, String) {
        let api_key = get_api_key();
        let ctx = context::new();
        let (meta, tenant, token) = state.get_meta_and_tenant_db(&ctx, &api_key).await.unwrap();
        let meta = meta.unwrap();
        clean_up(
            ctx.clone(),
            meta.clone(),
            tenant.clone(),
            &token,
            &collection_name,
        )
        .await;
        create_collection(
            ctx.clone(),
            meta.clone(),
            tenant.clone(),
            &token,
            collection_name,
            CreateCollection {
                vectors: VectorsConfig {
                    size: 2,
                    distance: Distance::Cosine,
                },
                hnsw_config: None,
            },
        )
        .await
        .unwrap();
        collection::actions::create_field_index(
            ctx.clone(),
            tenant.clone(),
            collection_name,
            CreateFieldIndex {
                field_name: "page_content".to_string(),
                field_schema: Some(PayloadFieldSchema::FieldType(PayloadSchemaType::Text)),
            },
        )
        .await
        .unwrap();
        collection::actions::create_field_index(
            ctx,
            tenant.clone(),
            collection_name,
            CreateFieldIndex {
                field_name: "group_id".to_string(),
                field_schema: Some(PayloadFieldSchema::FieldType(PayloadSchemaType::Keyword)),
            },
        )
        .await
        .unwrap();
        (meta, tenant, token.to_string())
    }

    async fn setup(collection_name: &str) -> (Arc<MySqlPool>, Arc<MySqlPool>, String) {
        let ctx = context::new();
        let s = get_state().await;
        let api_key = get_api_key();
        let (meta, tenant, token) = s.get_meta_and_tenant_db(&ctx, &api_key).await.unwrap();
        let meta = meta.unwrap();
        clean_up(
            ctx.clone(),
            meta.clone(),
            tenant.clone(),
            &token,
            &collection_name,
        )
        .await;
        create_collection(
            ctx,
            meta.clone(),
            tenant.clone(),
            &token,
            collection_name,
            CreateCollection {
                vectors: VectorsConfig {
                    size: 2,
                    distance: Distance::Cosine,
                },
                hnsw_config: None,
            },
        )
        .await
        .unwrap();
        (meta, tenant, token.to_string())
    }

    async fn clean_up(
        ctx: Arc<context::Context>,
        meta_db: Arc<MySqlPool>,
        tenant_db: Arc<MySqlPool>,
        tenant_token: &str,
        collection_name: &str,
    ) {
        match delete_collection(ctx, meta_db, tenant_db, &tenant_token, collection_name).await {
            Ok(_) => {}
            Err(_) => {
                error!("Failed to clean up collection {}", collection_name);
            }
        };
    }

    #[tokio::test]
    async fn test_get_collection_id() {
        let ctx = context::new();
        let collection_name = format!("{}_01", COLLECTION_FOR_TEST_POINTS);
        let (meta_db, tenant_db, tenant_token) = setup(&collection_name).await;

        let result = get_collection_id(
            ctx.clone(),
            &tenant_token,
            &collection_name,
            meta_db.clone(),
        )
        .await
        .unwrap();
        assert!(result > 0);

        clean_up(ctx, meta_db, tenant_db, &tenant_token, &collection_name).await;
    }

    #[tokio::test]
    async fn test_upsert_points_list() {
        let ctx = context::new();
        let collection_name = format!("{}_02", COLLECTION_FOR_TEST_POINTS);
        let state = get_state().await;
        let (meta_db, tenant_db, tenant_token) =
            setup_with_state(&collection_name, state.clone()).await;
        let point = serde_json::from_str(
            r#"
            {
                "id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E" },
                    "group_id": "group_id",
                    "page_content": "青山不改，绿水长流"
                }
            }
        "#,
        )
        .unwrap();

        let points = vec![point];
        let result = upsert_points_list(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            points,
        )
        .await
        .unwrap();
        assert!(matches!(result.status, UpdateStatus::Completed));

        clean_up(ctx, meta_db, tenant_db, &tenant_token, &collection_name).await;
    }

    #[tokio::test]
    async fn test_update_vectors() {
        let ctx = context::new();
        let collection_name = format!("{}_09", COLLECTION_FOR_TEST_POINTS);
        let state = get_state().await;
        let (meta_db, tenant_db, tenant_token) =
            setup_with_state(&collection_name, state.clone()).await;
        let point = serde_json::from_str(
            r#"
            {
                "id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E" },
                    "group_id": "group_id",
                    "page_content": "青山不改，绿水长流"
                }
            }
        "#,
        )
        .unwrap();

        let points = vec![point];
        let _ = upsert_points_list(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            points,
        )
        .await;

        let result = update_vectors(
            ctx.clone(),
            tenant_db.clone(),
            &collection_name,
            UpdateVectors {
                points: vec![PointStruct {
                    id: "FC7747EE-D483-4380-8CB1-AD268ACEE40E".into(),
                    vector: vec![3.0, 4.0],
                    payload: None,
                }],
            },
        )
        .await
        .unwrap();
        assert!(matches!(result.status, UpdateStatus::Completed));

        clean_up(ctx, meta_db, tenant_db, &tenant_token, &collection_name).await;
    }

    #[tokio::test]
    async fn test_get_point() {
        let ctx = context::new();
        let collection_name = format!("{}_03", COLLECTION_FOR_TEST_POINTS);
        let state = get_state().await;
        let (meta_db, tenant_db, tenant_token) =
            setup_with_state(&collection_name, state.clone()).await;
        let point = serde_json::from_str(
            r#"
            {
                "id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E" },
                    "group_id": "group_id",
                    "page_content": "青山不改，绿水长流"
                }
            }
        "#,
        )
        .unwrap();

        let points = vec![point];
        let _ = upsert_points_list(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            points,
        )
        .await;

        let result = get_point(
            tenant_db.clone(),
            &collection_name,
            "FC7747EE-D483-4380-8CB1-AD268ACEE40E",
        )
        .await
        .unwrap();
        assert!(result.id.to_string() == "FC7747EE-D483-4380-8CB1-AD268ACEE40E");

        clean_up(ctx, meta_db, tenant_db, &tenant_token, &collection_name).await;
    }

    #[tokio::test]
    async fn test_get_points() {
        let ctx = context::new();
        let collection_name = format!("{}_04", COLLECTION_FOR_TEST_POINTS);
        let state = get_state().await;
        let (meta_db, tenant_db, tenant_token) =
            setup_with_state(&collection_name, state.clone()).await;

        let points = vec![
            serde_json::from_str(
                r#"
            {
                "id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E" },
                    "group_id": "group_id",
                    "page_content": "青山不改，绿水长流"
                }
            }
        "#,
            )
            .unwrap(),
            serde_json::from_str(
                r#"
            {
                "id": "FC7747EE-D483-4380-8CB1-AD268ACEE41E",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "FC7747EE-D483-4380-8CB1-AD268ACEE41E" },
                    "group_id": "group_id",
                    "page_content": "青山不改，绿水长流"
                }
            }
        "#,
            )
            .unwrap(),
        ];
        let _ = upsert_points_list(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            points,
        )
        .await;

        let count = count_points(ctx.clone(), tenant_db.clone(), &collection_name)
            .await
            .unwrap();
        let result = get_points(
            tenant_db.clone(),
            &collection_name,
            PointRequest {
                ids: vec![
                    PointIdType::Uuid("FC7747EE-D483-4380-8CB1-AD268ACEE40E".into()),
                    PointIdType::Uuid("FC7747EE-D483-4380-8CB1-AD268ACEE41E".into()),
                ],
                with_payload: true,
                with_vector: true,
            },
        )
        .await
        .unwrap();
        assert!(result.len() == u64::from(count) as usize);

        clean_up(ctx, meta_db, tenant_db, &tenant_token, &collection_name).await;
    }

    #[tokio::test]
    async fn test_search_points() {
        let ctx = context::new();
        let collection_name = format!("{}_05", COLLECTION_FOR_TEST_POINTS);
        let state = get_state().await;
        let (meta_db, tenant_db, tenant_token) =
            setup_with_state(&collection_name, state.clone()).await;
        let point = serde_json::from_str(
            r#"
            {
                "id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E" },
                    "group_id": "group_id",
                    "page_content": "青山不改，绿水长流"
                }
            }
        "#,
        )
        .unwrap();

        let points = vec![point];
        let _ = upsert_points_list(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            points,
        )
        .await;

        let search_request = SearchRequest {
            vector: vec![1.0, 2.0],
            score_threshold: 0.0,
            filter: Some(Filter {
                must: Some(vec![FieldCondition {
                    key: "group_id".to_string(),
                    r#match: Match::Value(MatchValue {
                        value: ValueVariants::String("group_id".to_string()),
                    }),
                    without_index: false,
                }]),
                should: None,
                must_not: None,
            }),
            with_payload: true,
            with_vector: true,
            offset: Some(0),
            limit: Some(10),
        };
        let result = search_points(
            ctx.clone(),
            state.clone(),
            Some(meta_db.clone()),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            search_request,
        )
        .await
        .unwrap();
        assert!(result.len() == 1);

        clean_up(ctx, meta_db, tenant_db, &tenant_token, &collection_name).await;
    }

    #[tokio::test]
    async fn test_delete_points() {
        let ctx = context::new();
        let collection_name = format!("{}_09", COLLECTION_FOR_TEST_POINTS);
        let state = get_state().await;
        let (meta_db, tenant_db, tenant_token) =
            setup_with_state(&collection_name, state.clone()).await;

        let points = vec![
            serde_json::from_str(
                r#"
            {
                "id": "point-1",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "point-1" },
                    "group_id": "group_id",
                    "page_content": "青山不改"
                }
            }
        "#,
            )
            .unwrap(),
            serde_json::from_str(
                r#"
            {
                "id": "point-2",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "point-2" },
                    "group_id": "group_id",
                    "page_content": "绿水长流"
                }
            }
        "#,
            )
            .unwrap(),
        ];
        let _ = upsert_points_list(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            points,
        )
        .await;

        let result = delete_points(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            PointsSelector::FilterSelector(FilterSelector {
                filter: Filter {
                    must: Some(vec![FieldCondition {
                        key: "group_id".to_string(),
                        r#match: Match::Value(MatchValue {
                            value: ValueVariants::String("group_id".to_string()),
                        }),
                        without_index: false,
                    }]),
                    should: None,
                    must_not: None,
                },
            }),
            0,
        )
        .await
        .unwrap();
        assert!(matches!(result.status, UpdateStatus::Completed));

        clean_up(ctx, meta_db, tenant_db, &tenant_token, &collection_name).await;
    }

    #[tokio::test]
    async fn test_scroll_points() {
        let ctx = context::new();
        let collection_name = format!("{}_06", COLLECTION_FOR_TEST_POINTS);
        let state = get_state().await;
        let (meta_db, tenant_db, tenant_token) =
            setup_with_state(&collection_name, state.clone()).await;

        let points = vec![
            serde_json::from_str(
                r#"
            {
                "id": "point-1",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "point-1" },
                    "group_id": "group_id",
                    "page_content": "青山不改"
                }
            }
        "#,
            )
            .unwrap(),
            serde_json::from_str(
                r#"
            {
                "id": "point-2",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "point-2" },
                    "group_id": "group_id",
                    "page_content": "绿水长流"
                }
            }
        "#,
            )
            .unwrap(),
        ];
        let _ = upsert_points_list(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            points,
        )
        .await;

        // scroll points by text
        let scroll_request = ScrollRequest {
            filter: Some(Filter {
                must: Some(vec![FieldCondition {
                    key: "page_content".to_string(),
                    r#match: Match::Text(MatchText {
                        text: "青山".to_string(),
                    }),
                    without_index: false,
                }]),
                should: None,
                must_not: None,
            }),
            with_payload: true,
            with_vector: true,
            limit: Some(10),
            ..Default::default()
        };
        let result = scroll_points(
            ctx.clone(),
            state.clone(),
            Some(meta_db.clone()),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            scroll_request,
        )
        .await
        .unwrap();
        assert!(result.points.len() == 1);

        // scroll points by id
        let scroll_request = ScrollRequest {
            filter: Some(Filter {
                must: Some(vec![FieldCondition {
                    key: "id".to_string(),
                    r#match: Match::Value(MatchValue {
                        value: ValueVariants::String("point-1".into()),
                    }),
                    without_index: false,
                }]),
                should: None,
                must_not: None,
            }),
            with_payload: true,
            with_vector: true,
            limit: Some(10),
            ..Default::default()
        };
        let result = scroll_points(
            ctx.clone(),
            state.clone(),
            Some(meta_db.clone()),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            scroll_request,
        )
        .await
        .unwrap();
        assert!(result.points.len() == 1);

        // scroll points by id in shoulds
        let scroll_request = ScrollRequest {
            filter: Some(Filter {
                must: Some(vec![FieldCondition {
                    key: "group_id".to_string(),
                    r#match: Match::Any(MatchAny {
                        any: AnyVariants::String(vec!["group_id".into()]),
                    }),
                    without_index: false,
                }]),
                should: Some(vec![
                    FieldCondition {
                        key: "id".to_string(),
                        r#match: Match::Value(MatchValue {
                            value: ValueVariants::String("point-1".into()),
                        }),
                        without_index: false,
                    },
                    FieldCondition {
                        key: "id".to_string(),
                        r#match: Match::Value(MatchValue {
                            value: ValueVariants::String("point-2".into()),
                        }),
                        without_index: false,
                    },
                ]),
                must_not: None,
            }),
            with_payload: true,
            with_vector: true,
            limit: Some(10),
            ..Default::default()
        };
        let result = scroll_points(
            ctx.clone(),
            state.clone(),
            Some(meta_db.clone()),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            scroll_request,
        )
        .await
        .unwrap();
        assert!(result.points.len() == 2);

        // scroll points by id in must_not
        let scroll_request = ScrollRequest {
            filter: Some(Filter {
                must: None,
                should: None,
                must_not: Some(vec![FieldCondition {
                    key: "id".to_string(),
                    r#match: Match::Value(MatchValue {
                        value: ValueVariants::String("point-1".into()),
                    }),
                    without_index: false,
                }]),
            }),
            with_payload: true,
            with_vector: true,
            limit: Some(10),
            ..Default::default()
        };
        let result = scroll_points(
            ctx.clone(),
            state.clone(),
            Some(meta_db.clone()),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            scroll_request,
        )
        .await
        .unwrap();
        assert!(result.points.len() == 1);

        // scroll points by id MatchAny
        let scroll_request = ScrollRequest {
            filter: Some(Filter {
                must: Some(vec![FieldCondition {
                    key: "id".to_string(),
                    r#match: Match::Any(MatchAny {
                        any: AnyVariants::String(vec!["point-1".into(), "point-2".into()]),
                    }),
                    without_index: false,
                }]),
                should: None,
                must_not: None,
            }),
            with_payload: true,
            with_vector: true,
            limit: Some(10),
            ..Default::default()
        };
        let result = scroll_points(
            ctx.clone(),
            state.clone(),
            Some(meta_db.clone()),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            scroll_request,
        )
        .await
        .unwrap();
        assert!(result.points.len() == 2);

        // scroll points by id MatchAny
        let scroll_request = ScrollRequest {
            filter: Some(Filter {
                must: Some(vec![FieldCondition {
                    key: "id".to_string(),
                    r#match: Match::Except(MatchExcept {
                        except: AnyVariants::String(vec!["point-2".into()]),
                    }),
                    without_index: false,
                }]),
                should: None,
                must_not: None,
            }),
            with_payload: true,
            with_vector: true,
            limit: Some(10),
            ..Default::default()
        };
        let result = scroll_points(
            ctx.clone(),
            state.clone(),
            Some(meta_db.clone()),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            scroll_request,
        )
        .await
        .unwrap();
        assert!(result.points.len() == 1);

        clean_up(ctx, meta_db, tenant_db, &tenant_token, &collection_name).await;
    }

    #[tokio::test]
    async fn test_set_payload() {
        let ctx = context::new();
        let collection_name = format!("{}_07", COLLECTION_FOR_TEST_POINTS);
        let state = get_state().await;
        let (meta_db, tenant_db, tenant_token) =
            setup_with_state(&collection_name, state.clone()).await;
        let point = serde_json::from_str(
            r#"
            {
                "id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E" },
                    "group_id": "group_id",
                    "page_content": "青山不改，绿水长流"
                }
            }
        "#,
        )
        .unwrap();

        let points = vec![point];
        let _ = upsert_points_list(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            points,
        )
        .await;

        // set payload
        let request = SetPayloadRequest {
            points: Some(vec![PointIdType::Uuid(
                "FC7747EE-D483-4380-8CB1-AD268ACEE40E".to_string(),
            )]),
            key: Some("page_content".to_string()),
            payload: Payload::try_from(("page_content".to_string(), "推测".to_string())).unwrap(),
            filter: None,
        };
        let result = set_payload(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            request,
        )
        .await
        .unwrap();
        assert!(matches!(result.status, UpdateStatus::Completed));

        // get point
        let scroll_request = ScrollRequest {
            filter: Some(Filter {
                must: Some(vec![FieldCondition {
                    key: "page_content".to_string(),
                    r#match: Match::Text(MatchText {
                        text: "推测".to_string(),
                    }),
                    without_index: false,
                }]),
                should: None,
                must_not: None,
            }),
            with_payload: true,
            with_vector: true,
            limit: Some(10),
            ..Default::default()
        };
        let result = scroll_points(
            ctx.clone(),
            state.clone(),
            Some(meta_db.clone()),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            scroll_request,
        )
        .await
        .unwrap();
        assert!(result.points.len() == 1);

        clean_up(ctx, meta_db, tenant_db, &tenant_token, &collection_name).await;
    }

    #[tokio::test]
    async fn test_overwrite_payload() {
        let ctx = context::new();
        let collection_name = format!("{}_08", COLLECTION_FOR_TEST_POINTS);
        let state = get_state().await;
        let (meta_db, tenant_db, tenant_token) =
            setup_with_state(&collection_name, state.clone()).await;
        let point = serde_json::from_str(
            r#"
            {
                "id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E",
                "vector": [1.0, 2.0],
                "payload": {
                    "abc": "def",
                    "metadata": { "doc_id": "FC7747EE-D483-4380-8CB1-AD268ACEE40E" },
                    "group_id": "group_id",
                    "page_content": "青山不改，绿水长流"
                }
            }
        "#,
        )
        .unwrap();

        let points = vec![point];
        let _ = upsert_points_list(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            points,
        )
        .await;

        // set payload
        let request = SetPayloadRequest {
            points: Some(vec![PointIdType::Uuid(
                "FC7747EE-D483-4380-8CB1-AD268ACEE40E".to_string(),
            )]),
            key: Some("page_content".to_string()),
            payload: Payload::try_from(("page_content".to_string(), "推测".to_string())).unwrap(),
            filter: None,
        };
        let result = overwrite_payload(
            ctx.clone(),
            meta_db.clone(),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            request,
        )
        .await
        .unwrap();
        assert!(matches!(result.status, UpdateStatus::Completed));

        // get point
        let scroll_request = ScrollRequest {
            filter: Some(Filter {
                must: Some(vec![FieldCondition {
                    key: "page_content".to_string(),
                    r#match: Match::Text(MatchText {
                        text: "推测".to_string(),
                    }),
                    without_index: false,
                }]),
                should: None,
                must_not: None,
            }),
            with_payload: true,
            with_vector: true,
            limit: Some(10),
            ..Default::default()
        };
        let result = scroll_points(
            ctx.clone(),
            state.clone(),
            Some(meta_db.clone()),
            tenant_db.clone(),
            &tenant_token,
            &collection_name,
            scroll_request,
        )
        .await
        .unwrap();
        assert!(result.points.len() == 1);

        clean_up(ctx, meta_db, tenant_db, &tenant_token, &collection_name).await;
    }
}