// Copyright 2023 Greptime Team
//
// 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 common_error::ext::BoxedError;
use common_meta::election::ElectionRef;
use common_meta::election::rds::postgres::{ElectionPgClient, PgElection};
use common_meta::kv_backend::KvBackendRef;
use common_meta::kv_backend::rds::PgStore;
use common_meta::kv_backend::rds::postgres::{
    TlsMode as PgTlsMode, TlsOption as PgTlsOption, create_postgres_tls_connector,
};
use deadpool_postgres::{Config, Runtime};
use servers::tls::TlsOption;
use snafu::{OptionExt, ResultExt};
use tokio_postgres::NoTls;

use crate::error::{self, Result};

/// Converts [`TlsOption`] to [`PgTlsOption`] to avoid circular dependencies
fn convert_tls_option(tls_option: &TlsOption) -> PgTlsOption {
    let mode = match tls_option.mode {
        servers::tls::TlsMode::Disable => PgTlsMode::Disable,
        servers::tls::TlsMode::Prefer => PgTlsMode::Prefer,
        servers::tls::TlsMode::Require => PgTlsMode::Require,
        servers::tls::TlsMode::VerifyCa => PgTlsMode::VerifyCa,
        servers::tls::TlsMode::VerifyFull => PgTlsMode::VerifyFull,
    };

    PgTlsOption {
        mode,
        cert_path: tls_option.cert_path.clone(),
        key_path: tls_option.key_path.clone(),
        ca_cert_path: tls_option.ca_cert_path.clone(),
        watch: tls_option.watch,
    }
}

/// Creates a pool for the Postgres backend with config and optional TLS.
///
/// It only use first store addr to create a pool, and use the given config to create a pool.
pub async fn create_postgres_pool(
    store_addrs: &[String],
    cfg: Option<Config>,
    tls_config: Option<TlsOption>,
) -> Result<deadpool_postgres::Pool> {
    let mut cfg = cfg.unwrap_or_default();
    let postgres_url = store_addrs.first().context(error::InvalidArgumentsSnafu {
        err_msg: "empty store addrs",
    })?;
    cfg.url = Some(postgres_url.clone());

    let pool = if let Some(tls_config) = tls_config {
        let pg_tls_config = convert_tls_option(&tls_config);
        let tls_connector =
            create_postgres_tls_connector(&pg_tls_config).map_err(|e| error::Error::Other {
                source: BoxedError::new(e),
                location: snafu::Location::new(file!(), line!(), 0),
            })?;
        cfg.create_pool(Some(Runtime::Tokio1), tls_connector)
            .context(error::CreatePostgresPoolSnafu)?
    } else {
        cfg.create_pool(Some(Runtime::Tokio1), NoTls)
            .context(error::CreatePostgresPoolSnafu)?
    };

    Ok(pool)
}

/// Builds a Postgres-backed metadata [`KvBackendRef`].
///
/// * `store_addrs` - Postgres connection URLs; only the first address is used.
/// * `cfg` - optional deadpool config to customize pool/session behavior.
/// * `tls_config` - optional TLS settings for the Postgres connection.
/// * `schema_name` - optional schema containing the metadata table.
/// * `table_name` - metadata KV table name.
/// * `max_txn_ops` - maximum operations allowed in one metadata transaction.
/// * `auto_create_schema` - whether to create `schema_name` when it is missing.
#[allow(clippy::too_many_arguments)]
pub async fn build_postgres_kv_backend(
    store_addrs: &[String],
    cfg: Option<Config>,
    tls_config: Option<TlsOption>,
    schema_name: Option<&str>,
    table_name: &str,
    max_txn_ops: usize,
    auto_create_schema: bool,
) -> Result<KvBackendRef> {
    let pool = create_postgres_pool(store_addrs, cfg, tls_config).await?;
    PgStore::with_pg_pool(
        pool,
        schema_name,
        table_name,
        max_txn_ops,
        auto_create_schema,
    )
    .await
    .context(error::KvBackendSnafu)
}

/// Builds a Postgres-backed election implementation.
///
/// * `store_addrs` - Postgres connection URLs; only the first address is used.
/// * `cfg` - optional deadpool config to customize pool/session behavior.
/// * `tls_config` - optional TLS settings for the Postgres connection.
/// * `leader_value` - advertised address of this election candidate.
/// * `store_key_prefix` - prefix for election and candidate keys.
/// * `candidate_lease_ttl` - TTL for registered candidate metadata.
/// * `meta_lease_ttl` - TTL for the elected leader metadata.
/// * `schema_name` - optional schema containing the metadata table.
/// * `table_name` - metadata KV table name used for election records.
/// * `lock_id` - Postgres advisory lock id used by the election.
#[allow(clippy::too_many_arguments)]
pub async fn build_postgres_election(
    store_addrs: &[String],
    cfg: Option<Config>,
    tls_config: Option<TlsOption>,
    leader_value: String,
    store_key_prefix: String,
    candidate_lease_ttl: std::time::Duration,
    meta_lease_ttl: std::time::Duration,
    schema_name: Option<&str>,
    table_name: &str,
    lock_id: u64,
) -> Result<ElectionRef> {
    let pool = create_postgres_pool(store_addrs, cfg, tls_config).await?;
    let election_client =
        ElectionPgClient::new(pool, meta_lease_ttl, meta_lease_ttl, meta_lease_ttl)
            .context(error::KvBackendSnafu)?;
    PgElection::with_pg_client(
        leader_value,
        election_client,
        store_key_prefix,
        candidate_lease_ttl,
        meta_lease_ttl,
        schema_name,
        table_name,
        lock_id,
    )
    .await
    .context(error::KvBackendSnafu)
}
