// 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_base::readable_size::ReadableSize;
use common_config::{Configurable, KvBackendConfig};
use common_event_recorder::EventRecorderOptions;
use common_memory_manager::OnExhaustedPolicy;
use common_options::memory::MemoryOptions;
use common_telemetry::logging::{LoggingOptions, SlowQueryOptions, TracingOptions};
use common_wal::config::DatanodeWalConfig;
use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, StorageConfig};
use file_engine::config::EngineConfig as FileEngineConfig;
use flow::FlowConfig;
use frontend::frontend::FrontendOptions;
use frontend::service_config::{
    InfluxdbOptions, JaegerOptions, MysqlOptions, OpentsdbOptions, OtlpOptions, PostgresOptions,
    PromStoreOptions,
};
use mito2::config::MitoConfig;
use query::options::QueryOptions;
use serde::{Deserialize, Serialize};
use servers::grpc::GrpcOptions;
use servers::http::HttpOptions;

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
#[serde(default)]
pub struct StandaloneOptions {
    pub enable_telemetry: bool,
    pub default_timezone: Option<String>,
    pub default_column_prefix: Option<String>,
    /// Server-side global switch for auto table creation on write.
    /// Upper bound: when `false`, missing tables are never auto-created even if a
    /// request sets the `auto_create_table` hint to `true`. Default: `true`.
    pub auto_create_table: bool,
    /// Maximum total memory for all concurrent write request bodies and messages (HTTP, gRPC, Flight).
    /// Set to 0 to disable the limit. Default: "0" (unlimited)
    pub max_in_flight_write_bytes: ReadableSize,
    /// Policy when write bytes quota is exhausted.
    /// Options: "wait" (default, 10s), "wait(<duration>)", "fail"
    pub write_bytes_exhausted_policy: OnExhaustedPolicy,
    pub http: HttpOptions,
    pub grpc: GrpcOptions,
    pub mysql: MysqlOptions,
    pub postgres: PostgresOptions,
    pub opentsdb: OpentsdbOptions,
    pub influxdb: InfluxdbOptions,
    pub jaeger: JaegerOptions,
    pub otlp: OtlpOptions,
    pub prom_store: PromStoreOptions,
    pub wal: DatanodeWalConfig,
    pub storage: StorageConfig,
    pub metadata_store: KvBackendConfig,
    pub procedure: ProcedureConfig,
    pub flow: FlowConfig,
    pub logging: LoggingOptions,
    pub user_provider: Option<String>,
    /// Options for different store engines.
    pub region_engine: Vec<RegionEngineConfig>,
    pub tracing: TracingOptions,
    pub init_regions_in_background: bool,
    pub init_regions_parallelism: usize,
    pub slow_query: SlowQueryOptions,
    pub query: QueryOptions,
    pub memory: MemoryOptions,
    /// The event recorder options.
    pub event_recorder: EventRecorderOptions,
    /// Environment variable keys to read and report in heartbeat messages.
    pub heartbeat_env_vars: Vec<String>,
}

impl Default for StandaloneOptions {
    fn default() -> Self {
        Self {
            enable_telemetry: true,
            default_timezone: None,
            default_column_prefix: None,
            auto_create_table: true,
            max_in_flight_write_bytes: ReadableSize(0),
            write_bytes_exhausted_policy: OnExhaustedPolicy::default(),
            http: HttpOptions::default(),
            grpc: GrpcOptions::default(),
            mysql: MysqlOptions::default(),
            postgres: PostgresOptions::default(),
            opentsdb: OpentsdbOptions::default(),
            influxdb: InfluxdbOptions::default(),
            jaeger: JaegerOptions::default(),
            otlp: OtlpOptions::default(),
            prom_store: PromStoreOptions::default(),
            wal: DatanodeWalConfig::default(),
            storage: StorageConfig::default(),
            metadata_store: KvBackendConfig::default(),
            procedure: ProcedureConfig::default(),
            flow: FlowConfig::default(),
            logging: LoggingOptions::default(),
            user_provider: None,
            region_engine: vec![
                RegionEngineConfig::Mito(MitoConfig::default()),
                RegionEngineConfig::File(FileEngineConfig::default()),
            ],
            tracing: TracingOptions::default(),
            init_regions_in_background: false,
            init_regions_parallelism: 16,
            slow_query: SlowQueryOptions::default(),
            query: QueryOptions::default(),
            memory: MemoryOptions::default(),
            event_recorder: EventRecorderOptions::default(),
            heartbeat_env_vars: vec![],
        }
    }
}

impl Configurable for StandaloneOptions {
    fn env_list_keys() -> Option<&'static [&'static str]> {
        Some(&[
            "heartbeat_env_vars",
            "wal.broker_endpoints",
            "event_recorder.event_types",
        ])
    }
}

/// The [`StandaloneOptions`] is only defined in `standalone` crate,
/// we don't want to make `frontend` depends on it, so impl [`Into`]
/// rather than [`From`].
#[allow(clippy::from_over_into)]
impl Into<FrontendOptions> for StandaloneOptions {
    fn into(self) -> FrontendOptions {
        self.frontend_options()
    }
}

impl StandaloneOptions {
    pub fn frontend_options(&self) -> FrontendOptions {
        let cloned_opts = self.clone();
        FrontendOptions {
            default_timezone: cloned_opts.default_timezone,
            auto_create_table: cloned_opts.auto_create_table,
            max_in_flight_write_bytes: cloned_opts.max_in_flight_write_bytes,
            write_bytes_exhausted_policy: cloned_opts.write_bytes_exhausted_policy,
            http: cloned_opts.http,
            grpc: cloned_opts.grpc,
            mysql: cloned_opts.mysql,
            postgres: cloned_opts.postgres,
            opentsdb: cloned_opts.opentsdb,
            influxdb: cloned_opts.influxdb,
            jaeger: cloned_opts.jaeger,
            otlp: cloned_opts.otlp,
            prom_store: cloned_opts.prom_store,
            meta_client: None,
            logging: cloned_opts.logging,
            user_provider: cloned_opts.user_provider,
            query: cloned_opts.query,
            slow_query: cloned_opts.slow_query,
            event_recorder: cloned_opts.event_recorder,
            heartbeat_env_vars: cloned_opts.heartbeat_env_vars.clone(),
            ..Default::default()
        }
    }

    pub fn datanode_options(&self) -> DatanodeOptions {
        let cloned_opts = self.clone();
        DatanodeOptions {
            node_id: Some(0),
            enable_telemetry: cloned_opts.enable_telemetry,
            wal: cloned_opts.wal,
            storage: cloned_opts.storage,
            region_engine: cloned_opts.region_engine,
            grpc: cloned_opts.grpc,
            init_regions_in_background: cloned_opts.init_regions_in_background,
            init_regions_parallelism: cloned_opts.init_regions_parallelism,
            query: cloned_opts.query,
            heartbeat_env_vars: cloned_opts.heartbeat_env_vars,
            ..Default::default()
        }
    }

    /// Sanitize the `StandaloneOptions` to ensure the config is valid.
    pub fn sanitize(&mut self) {
        if self.storage.is_object_storage() {
            self.storage
                .store
                .cache_config_mut()
                .unwrap()
                .sanitize(&self.storage.data_home);
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use common_event_recorder::EventTypeFilter;

    use super::*;

    #[test]
    fn test_event_recorder_event_types_preserve_filter_semantics() {
        let all: StandaloneOptions = toml::from_str("").unwrap();
        let none: StandaloneOptions = toml::from_str("[event_recorder]\nevent_types = []").unwrap();
        let selected: StandaloneOptions =
            toml::from_str("[event_recorder]\nevent_types = ['create_database']").unwrap();

        assert!(all.event_recorder.event_types.allows("future_event"));
        assert_eq!(
            none.event_recorder.event_types.as_ref(),
            &EventTypeFilter::Only(Default::default())
        );
        assert!(
            selected
                .event_recorder
                .event_types
                .allows("create_database")
        );
        assert!(!selected.event_recorder.event_types.allows("drop_database"));

        let frontend_options = selected.frontend_options();
        assert!(Arc::ptr_eq(
            &selected.event_recorder.event_types,
            &frontend_options.event_recorder.event_types,
        ));
    }

    #[test]
    fn test_query_options_propagated_to_components() {
        let mut options = StandaloneOptions::default();
        options.query.parallelism = 4;

        assert_eq!(options.frontend_options().query.parallelism, 4);
        assert_eq!(options.datanode_options().query.parallelism, 4);
    }
}
