#[macro_use]
pub mod actions;
mod mix;

use std::sync::Arc;

use crate::error::QdrantError::{self, *};
use crate::error::{validate_collection_name, Result};
use crate::metrics;
use crate::protocol::points::*;
use crate::protocol::{ApiResponse, Record, UpdateResult};
use crate::server::ExtractApiKey;
use crate::state::ApiState;
use axum::extract::{Path, Query, State};
use axum::Json;
use log::*;
use model::collection::CollectionRecord;
use model::meta::tenant::{IndexType, PayloadIndex, TenantCollectionMeta};
use model::meta::DEFAULT_SHARD_NAME;
use sqlx::MySqlPool;

pub(crate) async fn upsert_points(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
    Query(_points_api_query_params): Query<PointsApiQueryParams>,
    Json(point_insert_operation): Json<PointInsertOperations>,
) -> Result<Json<ApiResponse<UpdateResult>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] PUT /collections/:collection_name/points with: collection_name:[ {} ] point_insert_operation:[ {:?} ]", ctx.id(), &collection_name, &point_insert_operation);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points", "put"])
        .inc();
    let (meta_db, tenant_db, tenant_token) = state.get_meta_and_tenant_db(&ctx, &api_key).await?;
    let meta_db = meta_db.ok_or(QdrantError::InternalServerErrorWithMsg(
        "meta-db not configured".to_string(),
    ))?;
    let res = match point_insert_operation {
        PointInsertOperations::PointsBatch(batch) => {
            actions::upsert_points_batch(
                ctx.clone(),
                meta_db,
                tenant_db,
                tenant_token,
                &collection_name,
                batch,
            )
            .await
        }
        PointInsertOperations::PointsList(points) => {
            actions::upsert_points_list(
                ctx.clone(),
                meta_db,
                tenant_db,
                tenant_token,
                &collection_name,
                points.points,
            )
            .await
        }
    }?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points", "put"])
        .observe(metrics::since(&ctx.started()));

    Ok(Json(ApiResponse::ok(ctx.started(), res)))
}

pub(crate) async fn delete_points(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
    Query(_points_api_query_params): Query<PointsApiQueryParams>,
    Json(point_selector): Json<PointsSelector>,
) -> Result<Json<ApiResponse<UpdateResult>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] POST /collections/:collection_name/points/delete with: collection_name:[ {} ] point_selector:[ {:?} ]",ctx.id(), &collection_name, &point_selector);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/delete", "post"])
        .inc();

    let (meta_db, tenant_db, tenant_token) = state.get_meta_and_tenant_db(&ctx, &api_key).await?;
    let meta_db = meta_db.ok_or(QdrantError::InternalServerErrorWithMsg(
        "meta-db not configured".to_string(),
    ))?;
    let fts_version = state.get_index_version(DEFAULT_SHARD_NAME).await?;
    let result = actions::delete_points(
        ctx.clone(),
        meta_db,
        tenant_db,
        tenant_token,
        &collection_name,
        point_selector,
        fts_version,
    )
    .await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points/delete", "post"])
        .observe(metrics::since(&ctx.started()));

    Ok(Json(ApiResponse::ok(ctx.started(), result)))
}

pub(crate) async fn count_points(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
) -> Result<Json<ApiResponse<PointsCount>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!(
        "[{}] POST /collections/:collection_name/points/count with: collection_name:[ {} ]",
        ctx.id(),
        &collection_name
    );
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/count", "post"])
        .inc();

    let (tenant_db, _) = state.get_tenant_db(&ctx, &api_key).await?;
    let points_count: PointsCount =
        actions::count_points(ctx.clone(), tenant_db, &collection_name).await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points/count", "post"])
        .observe(metrics::since(&ctx.started()));
    Ok(Json(ApiResponse::ok(ctx.started(), points_count)))
}

pub(crate) async fn get_point(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path((collection_name, id)): Path<(String, String)>,
) -> Result<Json<ApiResponse<Record>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!(
        "[{}] GET /collections/:collection_name/points/:id with: collection_name:[ {} ] id:[ {} ]",
        ctx.id(),
        &collection_name,
        &id
    );
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/id", "get"])
        .inc();

    let (tenant_db, _) = state.get_tenant_db(&ctx, &api_key).await?;
    let record = actions::get_point(tenant_db, &collection_name, &id).await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points/id", "get"])
        .observe(metrics::since(&ctx.started()));

    Ok(Json(ApiResponse::ok(ctx.started(), record)))
}

pub(crate) async fn get_points(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
    Query(_points_api_query_params): Query<PointsApiQueryParams>,
    Json(point_request): Json<PointRequest>,
) -> Result<Json<ApiResponse<Vec<Record>>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] POST /collections/:collection_name/points with: collection_name:[ {} ] point_request:[ {:?} ]", ctx.id(), &collection_name, &point_request);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points", "post"])
        .inc();

    let (tenant_db, _) = state.get_tenant_db(&ctx, &api_key).await?;
    let result = actions::get_points(tenant_db, &collection_name, point_request).await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points", "post"])
        .observe(metrics::since(&ctx.started()));

    Ok(Json(ApiResponse::ok(ctx.started(), result)))
}

pub(crate) async fn search_points(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
    Query(_points_api_query_params): Query<PointsApiQueryParams>,
    Json(search_request): Json<SearchRequest>,
) -> Result<Json<ApiResponse<Vec<ScoredPoint>>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] POST /collections/:collection_name/points/search with: collection_name:[ {} ] search_request:[ {:?} ]", ctx.id(), &collection_name, &search_request);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/search", "post"])
        .inc();

    if search_request.score_threshold >= 1.0
        || search_request.limit.map(|l| l == 0).unwrap_or(false)
    {
        warn!(
            "[{}] Invalid search request: {:?}",
            ctx.id(),
            search_request
        );
        return Err(BadRequest);
    }

    let (meta_db, tenant_db, tenant_token) = state.get_meta_and_tenant_db(&ctx, &api_key).await?;
    let result = actions::search_points(
        ctx.clone(),
        state,
        meta_db,
        tenant_db,
        tenant_token,
        &collection_name,
        search_request,
    )
    .await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points/search", "post"])
        .observe(metrics::since(&ctx.started()));

    Ok(Json(ApiResponse::ok(ctx.started(), result)))
}

pub(crate) async fn scroll_points(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
    Query(_points_api_query_params): Query<PointsApiQueryParams>,
    Json(scroll_params): Json<ScrollRequest>,
) -> Result<Json<ApiResponse<ScrollResult>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] POST /collections/:collection_name/points/scroll with: collection_name:[ {} ] scroll_params:[ {:?} ]", ctx.id(), &collection_name, &scroll_params);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/scroll", "post"])
        .inc();

    if scroll_params.limit.map(|l| l == 0).unwrap_or(false) {
        warn!("[{}] Invalid scroll request: {:?}", ctx.id(), scroll_params);
        return Err(BadRequest);
    }

    let (meta_db, tenant_db, tenant_token) = state.get_meta_and_tenant_db(&ctx, &api_key).await?;
    let result = actions::scroll_points(
        ctx.clone(),
        state,
        meta_db,
        tenant_db,
        tenant_token,
        &collection_name,
        scroll_params,
    )
    .await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points/scroll", "post"])
        .observe(metrics::since(&ctx.started()));

    Ok(Json(ApiResponse::ok(ctx.started(), result)))
}

pub async fn set_payload(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
    Json(set_payload): Json<SetPayloadRequest>,
) -> Result<Json<ApiResponse<UpdateResult>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] POST /collections/:collection_name/points/payload with: collection_name:[ {} ] set_payload:[ {:?} ]",ctx.id(), &collection_name, &set_payload);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/payload", "post"])
        .inc();

    let (meta_db, tenant_db, tenant_token) = state.get_meta_and_tenant_db(&ctx, &api_key).await?;
    let meta_db = meta_db.ok_or(QdrantError::InternalServerErrorWithMsg(
        "meta-db not configured".to_string(),
    ))?;
    let result = actions::set_payload(
        ctx.clone(),
        meta_db,
        tenant_db,
        tenant_token,
        &collection_name,
        set_payload,
    )
    .await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points/payload", "post"])
        .observe(metrics::since(&ctx.started()));
    Ok(Json(ApiResponse::ok(ctx.started(), result)))
}

pub async fn overwrite_payload(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
    Json(set_payload): Json<SetPayloadRequest>,
) -> Result<Json<ApiResponse<UpdateResult>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] PUT /collections/:collection_name/points/payload with: collection_name:[ {} ] set_payload:[ {:?} ]", ctx.id(), &collection_name, &set_payload);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/payload", "put"])
        .inc();

    let (meta_db, tenant_db, tenant_token) = state.get_meta_and_tenant_db(&ctx, &api_key).await?;
    let meta_db = meta_db.ok_or(QdrantError::InternalServerErrorWithMsg(
        "meta-db not configured".to_string(),
    ))?;
    let result = actions::overwrite_payload(
        ctx.clone(),
        meta_db,
        tenant_db,
        tenant_token,
        &collection_name,
        set_payload,
    )
    .await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points/payload", "put"])
        .observe(metrics::since(&ctx.started()));
    Ok(Json(ApiResponse::ok(ctx.started(), result)))
}

pub async fn delete_payload(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
    Json(delete_payload): Json<DeletePayloadRequest>,
) -> Result<Json<ApiResponse<UpdateResult>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] POST /collections/:collection_name/points/payload/delete with: collection_name:[ {} ] delete_payload:[ {:?} ]", ctx.id(), &collection_name, &delete_payload);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/payload/delete", "post"])
        .inc();

    let (meta_db, tenant_db, tenant_token) = state.get_meta_and_tenant_db(&ctx, &api_key).await?;
    let meta_db = meta_db.ok_or(QdrantError::InternalServerErrorWithMsg(
        "meta-db not configured".to_string(),
    ))?;
    let result = actions::delete_payload(
        meta_db,
        tenant_db,
        tenant_token,
        &collection_name,
        delete_payload,
    )
    .await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points/payload/delete", "post"])
        .observe(metrics::since(&ctx.started()));
    Ok(Json(ApiResponse::ok(ctx.started(), result)))
}

pub async fn clear_payload(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
    Json(clear_payload): Json<PointsSelector>,
) -> Result<Json<ApiResponse<UpdateResult>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] POST /collections/:collection_name/points/payload/clear with: collection_name:[ {} ] clear_payload:[ {:?} ]", ctx.id(), &collection_name, &clear_payload);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/payload/clear", "post"])
        .inc();

    let (tenant_db, _) = state.get_tenant_db(&ctx, &api_key).await?;
    let result = actions::clear_payload(tenant_db, &collection_name, clear_payload).await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points/payload/clear", "post"])
        .observe(metrics::since(&ctx.started()));
    Ok(Json(ApiResponse::ok(ctx.started(), result)))
}

pub async fn update_vectors(
    State(state): State<ApiState>,
    ExtractApiKey(api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
    Json(update_vectors): Json<UpdateVectors>,
) -> Result<Json<ApiResponse<UpdateResult>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] PUT /collections/:collection_name/points/vectors with: collection_name:[ {} ] update_vectors:[ {:?} ]", ctx.id(), &collection_name, &update_vectors);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/vectors", "put"])
        .inc();

    let (tenant_db, _) = state.get_tenant_db(&ctx, &api_key).await?;
    let result =
        actions::update_vectors(ctx.clone(), tenant_db, &collection_name, update_vectors).await?;

    metrics::ROUTER_COLLECTION_WITH_PATH_HISTOGRAM
        .with_label_values(&[&collection_name, "/points/vectors", "put"])
        .observe(metrics::since(&ctx.started()));

    Ok(Json(ApiResponse::ok(ctx.started(), result)))
}

pub async fn delete_vectors(
    State(_state): State<ApiState>,
    ExtractApiKey(_api_key): ExtractApiKey,
    Path(collection_name): Path<String>,
) -> Result<Json<ApiResponse<UpdateResult>>> {
    let ctx = context::with_metrics(metrics::INTERFACE_PERFORMANCE_HISTOGRAM.clone());
    info!("[{}] POST /collections/:collection_name/points/vectors/delete with: collection_name:[ {} ]", ctx.id(), &collection_name);
    let collection_name = validate_collection_name(collection_name)?;
    metrics::ROUTER_COLLECTION_WITH_PATH
        .with_label_values(&[&collection_name, "/points/vector/delete", "post"])
        .inc();

    Err(NotImplemented)
}

// a temporary solution to repair field schema for Dify
pub(crate) async fn try_repair_field_schema(
    ctx: Arc<context::Context>,
    tenant_db: Arc<MySqlPool>,
    tenant_token: String,
    collection_name: String,
) {
    info!(
        "[{}] Alter payload schema for tenant: {}",
        ctx.id(),
        tenant_token
    );
    if let Ok(mut conn) = tenant_db.begin().await {
        TenantCollectionMeta::add_field_payload_schema(&mut conn).await;
        if let Err(e) = conn.commit().await {
            error!("[{}] Failed to commit connection: {:?}", ctx.id(), e);
        };
    };

    info!(
        "[{}] Drop columns[meta,content] for collection: {}",
        ctx.id(),
        collection_name
    );
    if let Ok(mut conn) = tenant_db.begin().await {
        CollectionRecord::drop_columns(&mut conn, &collection_name).await;
        if let Err(e) = conn.commit().await {
            error!("[{}] Failed to commit connection: {:?}", ctx.id(), e);
        };
    }

    info!(
        "[{}] Alter page_content for collection: {}",
        ctx.id(),
        collection_name
    );
    if let Err(e) = CollectionRecord::add_tokens_column(
        ctx.clone(),
        &tenant_db,
        &collection_name,
        "page_content",
    )
    .await
    {
        error!(
            "[{}] Failed to alter page_content for collection {}: {:?}",
            ctx.id(),
            collection_name,
            e
        );
    };

    info!(
        "[{}] Alter group_id for collection: {}",
        ctx.id(),
        collection_name
    );
    if let Err(e) = TenantCollectionMeta::add_payload_idx(
        ctx.clone(),
        &tenant_db,
        &collection_name,
        PayloadIndex {
            path: "group_id".to_string(),
            length: Some(255),
            index_type: IndexType::Varchar,
        },
    )
    .await
    {
        error!(
            "[{}] Failed to alter group_id for collection {}: {:?}",
            ctx.id(),
            collection_name,
            e
        );
    };

    info!(
        "[{}] Alter metadata.doc_id for collection: {}",
        ctx.id(),
        collection_name
    );
    if let Err(e) = TenantCollectionMeta::add_payload_idx(
        ctx.clone(),
        &tenant_db,
        &collection_name,
        PayloadIndex {
            path: "metadata.doc_id".to_string(),
            length: Some(255),
            index_type: IndexType::Varchar,
        },
    )
    .await
    {
        error!(
            "[{}] Failed to alter metadata.doc_id for collection {}: {:?}",
            ctx.id(),
            collection_name,
            e
        );
    };

    info!(
        "[{}] Update payload_index for collection: {}",
        ctx.id(),
        &collection_name
    );
    if let Err(e) = TenantCollectionMeta::update_payload_index(
            ctx.clone(),
            &tenant_db,
            &collection_name,
            r#"
{"group_id": {"data_type": "keyword", "points": 0}, "metadata.doc_id": {"data_type": "keyword", "points": 0}, "page_content": {"data_type": "text", "params": {"lowercase": true, "max_token_len": 20, "min_token_len": 2, "tokenizer": "multilingual", "type": "text"}, "points": 0}}
            "#.to_string(),
        )
        .await
        {
            error!(
                "[{}] Failed to update payload_index for collection {}: {:?}",
                ctx.id(), collection_name, e
            );
        };
}