// Copyright 2025 RisingWave Labs
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use pgwire::pg_response::StatementType;
use risingwave_common::catalog::AlterDatabaseParam;
use risingwave_common::system_param::{NOTICE_BARRIER_INTERVAL_MS, NOTICE_CHECKPOINT_FREQUENCY};
use risingwave_sqlparser::ast::{ObjectName, SetVariableValue};

use super::{HandlerArgs, RwPgResponse};
use crate::Binder;
use crate::error::Result;
use crate::handler::alter_resource_group::resolve_resource_group;

pub async fn handle_alter_database_param(
    handler_args: HandlerArgs,
    database_name: ObjectName,
    param: AlterDatabaseParam,
) -> Result<RwPgResponse> {
    let mut builder = RwPgResponse::builder(StatementType::ALTER_DATABASE);

    let session = handler_args.session;

    let database_name = Binder::resolve_database_name(database_name)?;
    let database_id = {
        let catalog_reader = session.env().catalog_reader().read_guard();
        let database = catalog_reader.get_database_by_name(&database_name)?;

        // The user should be super user or owner to alter the database.
        session.check_privilege_for_drop_alter_db_schema(database)?;

        database.id()
    };

    match param {
        AlterDatabaseParam::BarrierIntervalMs(Some(interval)) => {
            if !cfg!(test) {
                risingwave_common::license::Feature::ResourceGroup.check_available()?;
            }
            if interval >= NOTICE_BARRIER_INTERVAL_MS {
                builder = builder.notice(
                    format!("Barrier interval is set to {} ms >= {} ms. This can hurt freshness and potentially cause OOM.",
                             interval, NOTICE_BARRIER_INTERVAL_MS));
            }
        }
        AlterDatabaseParam::CheckpointFrequency(Some(frequency)) => {
            if !cfg!(test) {
                risingwave_common::license::Feature::ResourceGroup.check_available()?;
            }
            if frequency >= NOTICE_CHECKPOINT_FREQUENCY {
                builder = builder.notice(
                    format!("Checkpoint frequency is set to {} >= {}. This can hurt freshness and potentially cause OOM.",
                             frequency, NOTICE_CHECKPOINT_FREQUENCY));
            }
        }
        _ => {}
    }

    let catalog_writer = session.catalog_writer()?;
    catalog_writer
        .alter_database_param(database_id, param)
        .await?;

    Ok(builder.into())
}

pub async fn handle_alter_database_resource_group(
    handler_args: HandlerArgs,
    database_name: ObjectName,
    resource_group: Option<SetVariableValue>,
    _deferred: bool,
) -> Result<RwPgResponse> {
    let mut builder = RwPgResponse::builder(StatementType::ALTER_DATABASE);

    let session = handler_args.session;

    let database_name = Binder::resolve_database_name(database_name)?;
    let database_id = {
        let catalog_reader = session.env().catalog_reader().read_guard();
        let database = catalog_reader.get_database_by_name(&database_name)?;

        // The user should be super user or owner to alter the database.
        session.check_privilege_for_drop_alter_db_schema(database)?;

        database.id()
    };

    let resource_group = resource_group
        .map(resolve_resource_group)
        .transpose()?
        .flatten();

    if resource_group.is_some() {
        risingwave_common::license::Feature::ResourceGroup.check_available()?;
    }

    let catalog_writer = session.catalog_writer()?;
    catalog_writer
        .alter_database_resource_group(database_id, resource_group, _deferred)
        .await?;

    builder = builder.notice(
        "The database resource group metadata has been updated, but it will not take effect immediately. Manually trigger recovery for inherited streaming jobs to use the new database resource group."
            .to_owned(),
    );

    Ok(builder.into())
}

#[cfg(test)]
mod tests {
    use risingwave_common::util::worker_util::DEFAULT_RESOURCE_GROUP;

    use crate::test_utils::LocalFrontend;

    #[tokio::test]
    async fn test_alter_barrier() {
        let frontend = LocalFrontend::new(Default::default()).await;
        let session = frontend.session_ref();
        let catalog_reader = session.env().catalog_reader();

        frontend.run_sql("CREATE DATABASE test_db").await.unwrap();
        {
            let reader = catalog_reader.read_guard();
            let db = reader.get_database_by_name("test_db").unwrap();
            assert!(db.barrier_interval_ms.is_none());
            assert!(db.checkpoint_frequency.is_none());
        }

        frontend
            .run_sql("ALTER DATABASE test_db SET BARRIER_INTERVAL_MS = 1000")
            .await
            .unwrap();
        {
            let reader = catalog_reader.read_guard();
            let db = reader.get_database_by_name("test_db").unwrap();
            assert_eq!(db.barrier_interval_ms, Some(1000));
            assert!(db.checkpoint_frequency.is_none());
        }

        frontend
            .run_sql("ALTER DATABASE test_db SET CHECKPOINT_FREQUENCY = 10")
            .await
            .unwrap();
        {
            let reader = catalog_reader.read_guard();
            let db = reader.get_database_by_name("test_db").unwrap();
            assert_eq!(db.barrier_interval_ms, Some(1000));
            assert_eq!(db.checkpoint_frequency, Some(10));
        }

        frontend
            .run_sql("ALTER DATABASE test_db SET BARRIER_INTERVAL_MS = DEFAULT")
            .await
            .unwrap();
        {
            let reader = catalog_reader.read_guard();
            let db = reader.get_database_by_name("test_db").unwrap();
            assert!(db.barrier_interval_ms.is_none());
            assert_eq!(db.checkpoint_frequency, Some(10));
        }

        frontend
            .run_sql("ALTER DATABASE test_db SET CHECKPOINT_FREQUENCY = DEFAULT")
            .await
            .unwrap();
        {
            let reader = catalog_reader.read_guard();
            let db = reader.get_database_by_name("test_db").unwrap();
            assert!(db.barrier_interval_ms.is_none());
            assert!(db.checkpoint_frequency.is_none());
        }
    }

    #[tokio::test]
    async fn test_alter_resource_group() {
        let frontend = LocalFrontend::new(Default::default()).await;
        let session = frontend.session_ref();
        let catalog_reader = session.env().catalog_reader();

        frontend.run_sql("CREATE DATABASE test_db").await.unwrap();

        frontend
            .run_sql("ALTER DATABASE test_db SET RESOURCE_GROUP = DEFAULT DEFERRED")
            .await
            .unwrap();
        {
            let reader = catalog_reader.read_guard();
            let db = reader.get_database_by_name("test_db").unwrap();
            assert_eq!(db.resource_group, DEFAULT_RESOURCE_GROUP);
        }

        frontend
            .run_sql("ALTER DATABASE test_db RESET RESOURCE_GROUP DEFERRED")
            .await
            .unwrap();
        {
            let reader = catalog_reader.read_guard();
            let db = reader.get_database_by_name("test_db").unwrap();
            assert_eq!(db.resource_group, DEFAULT_RESOURCE_GROUP);
        }

        let err = frontend
            .run_sql("ALTER DATABASE test_db SET RESOURCE_GROUP = 1 DEFERRED")
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("target resource group must be a valid string or default")
        );
    }
}
