// 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::regex_pattern::NAME_PATTERN_REG;
pub use datafusion_common::ScalarValue;
use once_cell::sync::OnceCell;
use snafu::ensure;

pub use crate::columnar_value::ColumnarValue;
use crate::error::{InvalidColumnPrefixSnafu, Result};
use crate::native_histogram::NATIVE_HISTOGRAM_FIELD;

/// Default time index column name.
static GREPTIME_TIMESTAMP_CELL: OnceCell<String> = OnceCell::new();

/// Default value column name.
static GREPTIME_VALUE_CELL: OnceCell<String> = OnceCell::new();

/// Default native histogram column name.
static GREPTIME_NATIVE_HISTOGRAM_CELL: OnceCell<String> = OnceCell::new();

pub fn set_default_prefix(prefix: Option<&str>) -> Result<()> {
    // Strip surrounding double quotes as a defensive measure against upstream
    // sources (scripts, CI, template engines, incorrect shell escaping) that may
    // pass literal `""` as the value instead of an empty string.
    let stripped = prefix.map(|s| {
        s.strip_prefix('"')
            .and_then(|s| s.strip_suffix('"'))
            .unwrap_or(s)
    });

    match stripped {
        None => {
            // use default greptime prefix
            GREPTIME_TIMESTAMP_CELL.get_or_init(|| GREPTIME_TIMESTAMP.to_string());
            GREPTIME_VALUE_CELL.get_or_init(|| GREPTIME_VALUE.to_string());
            GREPTIME_NATIVE_HISTOGRAM_CELL.get_or_init(|| NATIVE_HISTOGRAM_FIELD.to_string());
        }
        Some(s) if s.trim().is_empty() => {
            // use "" to disable prefix
            GREPTIME_TIMESTAMP_CELL.get_or_init(|| "timestamp".to_string());
            GREPTIME_VALUE_CELL.get_or_init(|| "value".to_string());
            GREPTIME_NATIVE_HISTOGRAM_CELL.get_or_init(|| "native_histogram".to_string());
        }
        Some(x) => {
            ensure!(
                NAME_PATTERN_REG.is_match(x),
                InvalidColumnPrefixSnafu { prefix: x }
            );
            GREPTIME_TIMESTAMP_CELL.get_or_init(|| format!("{}_timestamp", x));
            GREPTIME_VALUE_CELL.get_or_init(|| format!("{}_value", x));
            GREPTIME_NATIVE_HISTOGRAM_CELL.get_or_init(|| format!("{}_native_histogram", x));
        }
    }
    Ok(())
}

/// Get the default timestamp column name.
/// Returns the configured value, or `greptime_timestamp` if not set.
pub fn greptime_timestamp() -> &'static str {
    GREPTIME_TIMESTAMP_CELL.get_or_init(|| GREPTIME_TIMESTAMP.to_string())
}

/// Get the default value column name.
/// Returns the configured value, or `greptime_value` if not set.
pub fn greptime_value() -> &'static str {
    GREPTIME_VALUE_CELL.get_or_init(|| GREPTIME_VALUE.to_string())
}

/// Get the default native histogram column name.
/// Returns the configured value, or `greptime_native_histogram` if not set.
#[inline]
pub fn greptime_native_histogram() -> &'static str {
    GREPTIME_NATIVE_HISTOGRAM_CELL
        .get()
        .map_or(NATIVE_HISTOGRAM_FIELD, String::as_str)
}

/// Default timestamp column name constant for backward compatibility.
const GREPTIME_TIMESTAMP: &str = "greptime_timestamp";
/// Default value column name constant for backward compatibility.
const GREPTIME_VALUE: &str = "greptime_value";
/// Default counter column name for OTLP metrics (legacy mode).
pub const GREPTIME_COUNT: &str = "greptime_count";
/// Default physical table name
pub const GREPTIME_PHYSICAL_TABLE: &str = "greptime_physical_table";

#[cfg(test)]
mod tests {
    use super::*;

    // Each test runs in a separate process via `cargo nextest`, so OnceCell
    // state does not leak between tests.

    #[test]
    fn test_set_default_prefix_none() {
        set_default_prefix(None).unwrap();
        assert_eq!(greptime_timestamp(), "greptime_timestamp");
        assert_eq!(greptime_value(), "greptime_value");
        assert_eq!(greptime_native_histogram(), "greptime_native_histogram");
    }

    #[test]
    fn test_set_default_prefix_empty_string() {
        set_default_prefix(Some("")).unwrap();
        assert_eq!(greptime_timestamp(), "timestamp");
        assert_eq!(greptime_value(), "value");
        assert_eq!(greptime_native_histogram(), "native_histogram");
    }

    #[test]
    fn test_set_default_prefix_quoted_empty() {
        // Handles upstream sources that pass literal `""` instead of an empty string
        set_default_prefix(Some("\"\"")).unwrap();
        assert_eq!(greptime_timestamp(), "timestamp");
        assert_eq!(greptime_value(), "value");
        assert_eq!(greptime_native_histogram(), "native_histogram");
    }

    #[test]
    fn test_set_default_prefix_custom() {
        set_default_prefix(Some("mydb")).unwrap();
        assert_eq!(greptime_timestamp(), "mydb_timestamp");
        assert_eq!(greptime_value(), "mydb_value");
        assert_eq!(greptime_native_histogram(), "mydb_native_histogram");
    }

    #[test]
    fn test_set_default_prefix_invalid() {
        assert!(set_default_prefix(Some("invalid prefix!")).is_err());
    }
}
