use super::*;
use crate::error::*;

pub(crate) const COLLECTION_DELETED_TABLE: &str = "_deleted";
pub struct CollectionRecordDeleted {}

impl CollectionRecordDeleted {
    pub async fn create_table(
        pool: &sqlx::MySqlPool,
        collection_name: &str,
        version: u64,
    ) -> Result<()> {
        let deleted_table_name = format!("{}{}", collection_name, COLLECTION_DELETED_TABLE);
        match CollectionRecord::try_create_table(pool, &deleted_table_name).await {
            Ok(_) => {
                Self::move_deleted_points(pool, collection_name, version).await?;
            }
            Err(e) => {
                debug!("Create deleted table error: {:?}", e);
            }
        };

        Ok(())
    }

    async fn move_deleted_points(
        pool: &sqlx::MySqlPool,
        collection_name: &str,
        version: u64,
    ) -> Result<()> {
        let deleted_table_name = format!("{}{}", collection_name, COLLECTION_DELETED_TABLE);

        let move_sql = format!(
            r#"
            REPLACE INTO `{}`(`id`, `version`) SELECT `id`, {} as `version` FROM `{}` WHERE `deleted_at` IS NOT NULL
            "#,
            deleted_table_name, version, collection_name
        );
        let delete_sql = format!(
            r#"
            DELETE FROM `{}` WHERE `deleted_at` IS NOT NULL
            "#,
            collection_name
        );

        let mut tx = pool.begin().await?;
        sqlx::query(&move_sql).execute(&mut *tx).await?;
        let result = sqlx::query(&delete_sql).execute(&mut *tx).await?;
        tx.commit().await?;
        debug!(
            "Move deleted points {}: {} && {}",
            result.rows_affected(),
            move_sql,
            delete_sql
        );

        Ok(())
    }

    pub async fn deleted_gc(pool: &sqlx::MySqlPool, collection_name: &str, version: u64) {
        let deleted_table_name = format!("{}{}", collection_name, COLLECTION_DELETED_TABLE);
        let delete_sql = format!(
            r#"
            DELETE FROM `{}` WHERE `version` <= {}
            "#,
            deleted_table_name, version
        );
        match sqlx::query(&delete_sql).execute(pool).await {
            Ok(result) => debug!(
                "GC {} deleted points {}: {}",
                collection_name,
                result.rows_affected(),
                delete_sql
            ),
            Err(e) => error!("GC {} deleted points error: {}", collection_name, e),
        };
    }
}

pub struct DeleteCollectionRecordBuilder {
    move_sql: String,
    delete_sql: String,
    predicate: Option<CollectionRecordPredicateBuilder>,
}

impl DeleteCollectionRecordBuilder {
    // move sql, delete sql, must execute in transaction
    pub(crate) fn new(table_name: String, version: u64) -> Self {
        let move_sql = format!(
            r#"
            REPLACE INTO `{}{}`(`id`, `version`) SELECT `id`, {} as `version` FROM `{}`
            "#,
            table_name, COLLECTION_DELETED_TABLE, version, table_name
        );
        let delete_sql = format!(
            r#"
            DELETE FROM `{}`
            "#,
            table_name
        );

        Self {
            move_sql,
            delete_sql,
            predicate: None,
        }
    }

    pub fn predicate(&mut self) -> &mut CollectionRecordPredicateBuilder {
        self.predicate.get_or_insert_with(Default::default)
    }

    fn build_move(&self, ctx: Arc<context::Context>) -> Result<String> {
        let predicate = match self.predicate {
            Some(ref predicate) => predicate,
            None => {
                return Err(Error::new(
                    ErrorKind::BadRequest,
                    "Predicate is required in move".to_string(),
                )
                .ctx(ctx));
            }
        };
        let mut sql = self.move_sql.clone();
        if !predicate.sql.is_empty() {
            sql.push_str(" WHERE ");
            sql.push_str(&predicate.sql);
        } else {
            return Err(Error::new(
                ErrorKind::BadRequest,
                "Predicate's SQL is required in move-sql".to_string(),
            )
            .ctx(ctx));
        }
        Ok(sql)
    }

    fn build_delete(&self, ctx: Arc<context::Context>) -> Result<String> {
        let predicate = match self.predicate {
            Some(ref predicate) => predicate,
            None => {
                return Err(Error::new(
                    ErrorKind::BadRequest,
                    "Predicate is required in delete".to_string(),
                )
                .ctx(ctx));
            }
        };
        let mut sql = self.delete_sql.clone();
        if !predicate.sql.is_empty() {
            sql.push_str(" WHERE ");
            sql.push_str(&predicate.sql);
        } else {
            return Err(Error::new(
                ErrorKind::BadRequest,
                "Predicate's SQL is required in delete-sql".to_string(),
            )
            .ctx(ctx));
        }
        Ok(sql)
    }

    pub fn build(self, ctx: Arc<context::Context>) -> Result<(String, String)> {
        Ok((self.build_move(ctx.clone())?, self.build_delete(ctx)?))
    }
}