// Copyright (c) 2023-2026 ParadeDB, Inc.
//
// This file is part of ParadeDB - Postgres for Search and Analytics
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

use std::cell::{Ref, RefCell};
use std::ffi::CStr;
use std::num::NonZeroUsize;
use std::rc::Rc;

use crate::api::{FieldName, HashMap};
use crate::gucs;
use crate::postgres::utils::{extract_field_attributes, ExtractedFieldAttribute};
use crate::schema::IndexRecordOption;
use crate::schema::{SearchFieldConfig, SearchFieldType};

use crate::api::tokenizers::search_field_config_from_type;
use crate::gucs::{global_enable_background_merging, global_target_segment_count};
use anyhow::Result;
use memoffset::*;
use pgrx::pg_sys::AsPgCStr;
use pgrx::*;
use serde::{Deserialize, Serialize};
use serde_json::Map;
use tokenizers::manager::SearchTokenizerFilters;
use tokenizers::{SearchNormalizer, SearchTokenizer};
/* ADDING OPTIONS
 * in init(), call pg_sys::add_{type}_reloption (check postgres docs for what args you need)
 * add the corresponding entries to SearchIndexOptionsData struct definition
 * in amoptions(), add a relopt_parse_elt entry to the options array and change NUM_REL_OPTS
 * Note that for string options, postgres will give you the offset of the string, and you have to read the string
 * yourself (see get_tokenizer)
*/

/* READING OPTIONS
 * options are placed in relation.rd_options
 * As in ambuild(), cast relation.rd_options into SearchIndexOptionsData using PgBox
 * (because SearchIndexOptionsData is a postgres-allocated object) and use getters and setters
*/

// We set the target segment count equal to the CPU count by default
// However for machines with only a few CPUs, we set the target segment count to 4
// to avoid cramming everything into a single segment
const MIN_TARGET_SEGMENT_COUNT: usize = 4;

static mut RELOPT_KIND_PDB: pg_sys::relopt_kind::Type = 0;

#[allow(clippy::identity_op)]
pub(crate) const DEFAULT_FOREGROUND_LAYER_SIZES: &[u64] = &[];

#[allow(clippy::identity_op)]
pub(crate) const DEFAULT_BACKGROUND_LAYER_SIZES: &[u64] = &[
    100 * 1024,          // 100KB
    1 * 1024 * 1024,     // 1MB
    10 * 1024 * 1024,    // 10MB
    100 * 1024 * 1024,   // 100MB
    1000 * 1024 * 1024,  // 1GB
    10000 * 1024 * 1024, // 10GB
];

pub(crate) const DEFAULT_MUTABLE_SEGMENT_ROWS: usize = 1000;
pub(crate) const MAX_MUTABLE_SEGMENT_ROWS: usize = 10000;

#[pg_guard]
extern "C-unwind" fn validate_text_fields(value: *const std::os::raw::c_char) {
    let json_str = cstr_to_rust_str(value);
    if json_str.is_empty() {
        return;
    }
    deserialize_config_fields(json_str, &SearchFieldConfig::text_from_json);
}

#[pg_guard]
extern "C-unwind" fn validate_inet_fields(value: *const std::os::raw::c_char) {
    let json_str = cstr_to_rust_str(value);
    if json_str.is_empty() {
        return;
    }
    deserialize_config_fields(json_str, &SearchFieldConfig::inet_from_json);
}

#[pg_guard]
extern "C-unwind" fn validate_numeric_fields(value: *const std::os::raw::c_char) {
    let json_str = cstr_to_rust_str(value);
    if json_str.is_empty() {
        return;
    }
    deserialize_config_fields(json_str, &SearchFieldConfig::numeric_from_json);
}

#[pg_guard]
extern "C-unwind" fn validate_boolean_fields(value: *const std::os::raw::c_char) {
    let json_str = cstr_to_rust_str(value);
    if json_str.is_empty() {
        return;
    }
    deserialize_config_fields(json_str, &SearchFieldConfig::boolean_from_json);
}

#[pg_guard]
extern "C-unwind" fn validate_json_fields(value: *const std::os::raw::c_char) {
    let json_str = cstr_to_rust_str(value);
    if json_str.is_empty() {
        return;
    }
    deserialize_config_fields(json_str, &SearchFieldConfig::json_from_json);
}

#[pg_guard]
extern "C-unwind" fn validate_range_fields(value: *const std::os::raw::c_char) {
    let json_str = cstr_to_rust_str(value);
    if json_str.is_empty() {
        return;
    }
    deserialize_config_fields(json_str, &SearchFieldConfig::range_from_json);
}

#[pg_guard]
extern "C-unwind" fn validate_datetime_fields(value: *const std::os::raw::c_char) {
    let json_str = cstr_to_rust_str(value);
    if json_str.is_empty() {
        return;
    }
    deserialize_config_fields(json_str, &SearchFieldConfig::date_from_json);
}

#[pg_guard]
extern "C-unwind" fn validate_fields(value: *const std::os::raw::c_char) {
    let json_str = cstr_to_rust_str(value);
    if json_str.is_empty() {
        return;
    }

    // Just ensure the config can be deserialized as json.
    let _: HashMap<String, serde_json::Value> = json5::from_str(&json_str)
        .unwrap_or_else(|err| panic!("failed to deserialize field config: {err:?}"));
}

#[pg_guard]
extern "C-unwind" fn validate_key_field(value: *const std::os::raw::c_char) {
    cstr_to_rust_str(value);
}

#[pg_guard]
extern "C-unwind" fn validate_sort_by(value: *const std::os::raw::c_char) {
    let sort_by_str = cstr_to_rust_str(value);
    if sort_by_str.is_empty() {
        return;
    }
    // Parse and validate the sort_by string (panics on invalid input)
    let _ = parse_sort_by_string(&sort_by_str);
}

#[pg_guard]
extern "C-unwind" fn validate_search_tokenizer(value: *const std::os::raw::c_char) {
    let s = cstr_to_rust_str(value);
    if s.is_empty() {
        return;
    }
    crate::api::tokenizers::tokenizer_from_expression(&s)
        .unwrap_or_else(|| panic!("invalid search_tokenizer: '{s}'"));
}

#[pg_guard]
extern "C-unwind" fn validate_layer_sizes(value: *const std::os::raw::c_char) {
    if value.is_null() {
        // a NULL value means we're to use whatever our defaults are
        return;
    }
    let cstr = unsafe { CStr::from_ptr(value) };
    let _ = get_layer_sizes(cstr.to_str().expect("`layer_sizes` must be valid UTF-8"))
        .collect::<Vec<_>>();
}

fn get_layer_sizes(s: &str) -> impl Iterator<Item = u64> + use<'_> {
    s.split(",").filter_map(|part| unsafe {
        let size = u64::try_from(
            direct_function_call::<i64>(pg_sys::pg_size_bytes, &[part.into_datum()])
                .expect("`pg_size_bytes()` should not return NULL"),
        )
        .ok();

        match size {
            Some(b) if b > 0 => Some(b),
            Some(0) => None,
            _ => panic!("a single layer size must be non-negative"),
        }
    })
}

#[inline]
fn cstr_to_rust_str(value: *const std::os::raw::c_char) -> String {
    if value.is_null() {
        return "".to_string();
    }

    unsafe { CStr::from_ptr(value) }
        .to_str()
        .expect("failed to parse fields as utf-8")
        .to_string()
}

const NUM_REL_OPTS: usize = 14;
#[pg_guard]
pub unsafe extern "C-unwind" fn amoptions(
    reloptions: pg_sys::Datum,
    validate: bool,
) -> *mut pg_sys::bytea {
    let options: [pg_sys::relopt_parse_elt; NUM_REL_OPTS] = [
        pg_sys::relopt_parse_elt {
            optname: "text_fields".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, text_fields_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "inet_fields".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, inet_fields_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "numeric_fields".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, numeric_fields_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "boolean_fields".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, boolean_fields_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "json_fields".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, json_fields_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "range_fields".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, range_fields_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "datetime_fields".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, datetime_fields_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "key_field".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, key_field_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "layer_sizes".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, layer_sizes_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "target_segment_count".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_INT,
            offset: offset_of!(BM25IndexOptionsData, target_segment_count) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "background_layer_sizes".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, background_layer_sizes_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "mutable_segment_rows".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_INT,
            offset: offset_of!(BM25IndexOptionsData, mutable_segment_rows) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "sort_by".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, sort_by_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
        pg_sys::relopt_parse_elt {
            optname: "search_tokenizer".as_pg_cstr(),
            opttype: pg_sys::relopt_type::RELOPT_TYPE_STRING,
            offset: offset_of!(BM25IndexOptionsData, search_tokenizer_offset) as i32,
            #[cfg(feature = "pg18")]
            isset_offset: 0,
        },
    ];
    build_relopts(reloptions, validate, options)
}

unsafe fn build_relopts(
    reloptions: pg_sys::Datum,
    validate: bool,
    options: [pg_sys::relopt_parse_elt; NUM_REL_OPTS],
) -> *mut pg_sys::bytea {
    let rdopts = pg_sys::build_reloptions(
        reloptions,
        validate,
        RELOPT_KIND_PDB,
        std::mem::size_of::<BM25IndexOptionsData>(), // TODO: proper size calculator
        options.as_ptr(),
        NUM_REL_OPTS as i32,
    );

    rdopts as *mut pg_sys::bytea
}

#[derive(Debug, Clone, Default)]
struct LazyInfo {
    // these are ordered in an order that's likely most common to least common
    text: Rc<RefCell<Option<HashMap<FieldName, SearchFieldConfig>>>>,
    numeric: Rc<RefCell<Option<HashMap<FieldName, SearchFieldConfig>>>>,
    datetime: Rc<RefCell<Option<HashMap<FieldName, SearchFieldConfig>>>>,
    boolean: Rc<RefCell<Option<HashMap<FieldName, SearchFieldConfig>>>>,
    json: Rc<RefCell<Option<HashMap<FieldName, SearchFieldConfig>>>>,
    range: Rc<RefCell<Option<HashMap<FieldName, SearchFieldConfig>>>>,
    inet: Rc<RefCell<Option<HashMap<FieldName, SearchFieldConfig>>>>,

    attributes: Rc<RefCell<HashMap<FieldName, ExtractedFieldAttribute>>>,
}

#[derive(Clone, Debug)]
pub struct BM25IndexOptions {
    indexrel: pg_sys::Relation,
    lazy: LazyInfo,
}

impl BM25IndexOptions {
    pub const MISSING_KEY_FIELD_CONFIG: &'static str =
        "index should have a `WITH (key_field='...')` option";

    pub unsafe fn from_relation(indexrel: pg_sys::Relation) -> Self {
        assert!(!indexrel.is_null());
        Self {
            indexrel,
            lazy: LazyInfo::default(),
        }
    }

    pub fn foreground_layer_sizes(&self) -> Vec<u64> {
        self.options_data().foreground_layer_sizes()
    }

    pub fn background_layer_sizes(&self) -> Vec<u64> {
        if !global_enable_background_merging() {
            return vec![];
        }

        self.options_data().background_layer_sizes()
    }

    pub fn target_segment_count(&self) -> usize {
        let global_tsc = global_target_segment_count();
        if global_tsc != 0 {
            return global_tsc as usize;
        }

        self.options_data()
            .target_segment_count()
            .map(|count| count as usize)
            .unwrap_or(crate::available_parallelism().max(MIN_TARGET_SEGMENT_COUNT))
    }

    pub fn mutable_segment_rows(&self) -> Option<NonZeroUsize> {
        match gucs::global_mutable_segment_rows() {
            Some(rows) if rows > 0 => NonZeroUsize::new(rows),
            Some(0) => None,
            _ => self.options_data().mutable_segment_rows(),
        }
    }

    pub fn key_field_name(&self) -> FieldName {
        self.options_data()
            .key_field_name()
            .expect(Self::MISSING_KEY_FIELD_CONFIG)
    }

    /// Returns the sort_by configuration.
    /// - Not specified: defaults to none (no segment sorting)
    /// - "none": returns empty vec (no sorting)
    /// - Otherwise: returns parsed sort fields
    pub fn sort_by(&self) -> Vec<SortByField> {
        self.options_data().sort_by()
    }

    pub fn search_tokenizer(&self) -> Option<SearchTokenizer> {
        self.options_data().search_tokenizer()
    }

    pub fn key_field_type(&self) -> SearchFieldType {
        self.get_field_type(&self.key_field_name())
            .expect(Self::MISSING_KEY_FIELD_CONFIG)
    }

    /// Returns either the config explicitly set in the CREATE INDEX WITH options,
    /// falling back to the default config for the field type.
    pub fn field_config_or_default(&self, field_name: &FieldName) -> SearchFieldConfig {
        match self.field_config(field_name) {
            Some(config) => config,
            None => {
                let field_type = self.get_field_type(field_name).unwrap_or_else(|| {
                    panic!(
                        "field `{field_name}` is not configured in the CREATE INDEX WITH options"
                    )
                });
                field_type.default_config()
            }
        }
    }

    pub fn text_config(&self) -> Ref<'_, Option<HashMap<FieldName, SearchFieldConfig>>> {
        if self.lazy.text.borrow().is_none() {
            *self.lazy.text.borrow_mut() = Some(self.options_data().text_configs());
        }
        self.lazy.text.borrow()
    }

    pub fn numeric_config(&self) -> Ref<'_, Option<HashMap<FieldName, SearchFieldConfig>>> {
        if self.lazy.numeric.borrow().is_none() {
            *self.lazy.numeric.borrow_mut() = Some(self.options_data().numeric_configs());
        }
        self.lazy.numeric.borrow()
    }

    pub fn datetime_config(&self) -> Ref<'_, Option<HashMap<FieldName, SearchFieldConfig>>> {
        if self.lazy.datetime.borrow().is_none() {
            *self.lazy.datetime.borrow_mut() = Some(self.options_data().datetime_configs());
        }
        self.lazy.datetime.borrow()
    }

    pub fn boolean_config(&self) -> Ref<'_, Option<HashMap<FieldName, SearchFieldConfig>>> {
        if self.lazy.boolean.borrow().is_none() {
            *self.lazy.boolean.borrow_mut() = Some(self.options_data().boolean_configs());
        }
        self.lazy.boolean.borrow()
    }

    pub fn json_config(&self) -> Ref<'_, Option<HashMap<FieldName, SearchFieldConfig>>> {
        if self.lazy.json.borrow().is_none() {
            *self.lazy.json.borrow_mut() = Some(self.options_data().json_configs());
        }
        self.lazy.json.borrow()
    }

    pub fn range_config(&self) -> Ref<'_, Option<HashMap<FieldName, SearchFieldConfig>>> {
        if self.lazy.range.borrow().is_none() {
            *self.lazy.range.borrow_mut() = Some(self.options_data().range_configs());
        }
        self.lazy.range.borrow()
    }

    pub fn inet_config(&self) -> Ref<'_, Option<HashMap<FieldName, SearchFieldConfig>>> {
        if self.lazy.inet.borrow().is_none() {
            *self.lazy.inet.borrow_mut() = Some(self.options_data().inet_configs());
        }
        self.lazy.inet.borrow()
    }

    /// Returns the config only if it is explicitly set in the CREATE INDEX WITH options
    fn field_config(&self, field_name: &FieldName) -> Option<SearchFieldConfig> {
        let data = self.options_data();
        if field_name.is_ctid() {
            return Some(SearchFieldConfig::Numeric {
                indexed: true,
                fast: true,
                scale: None,
            });
        }

        if field_name.root() == data.key_field_name()?.root() {
            return match self.text_config().as_ref().unwrap().get(field_name) {
                // if the key_field is TEXT then we'll use the config for it
                config @ Some(SearchFieldConfig::Text { .. }) => config.cloned(),

                // otherwise we'll use the default config for key_fields in general
                _ => self.get_field_type(field_name).map(key_field_config),
            };
        }

        let field_type = self.get_field_type(field_name);

        if field_name.root() == data.key_field_name()?.root() {
            return field_type.map(key_field_config);
        } else if let Some(SearchFieldType::Tokenized(tokenizer_oid, typmod, inner_typoid)) =
            field_type
        {
            return search_field_config_from_type(tokenizer_oid, typmod, inner_typoid);
        }

        self.text_config()
            .as_ref()
            .unwrap()
            .get(field_name)
            .cloned()
            .or_else(|| {
                self.numeric_config()
                    .as_ref()
                    .unwrap()
                    .get(field_name)
                    .cloned()
            })
            .or_else(|| {
                self.datetime_config()
                    .as_ref()
                    .unwrap()
                    .get(field_name)
                    .cloned()
            })
            .or_else(|| {
                self.boolean_config()
                    .as_ref()
                    .unwrap()
                    .get(field_name)
                    .cloned()
            })
            .or_else(|| {
                self.json_config()
                    .as_ref()
                    .unwrap()
                    .get(field_name)
                    .cloned()
            })
            .or_else(|| {
                self.range_config()
                    .as_ref()
                    .unwrap()
                    .get(field_name)
                    .cloned()
            })
            .or_else(|| {
                self.inet_config()
                    .as_ref()
                    .unwrap()
                    .get(field_name)
                    .cloned()
            })
    }

    /// Returns a `Vec` of aliased text field names and their configs.
    pub fn aliased_text_configs(&self) -> Vec<(FieldName, SearchFieldConfig)> {
        if self.lazy.text.borrow().is_none() {
            *self.lazy.text.borrow_mut() = Some(self.options_data().text_configs());
        }

        self.lazy
            .text
            .borrow()
            .iter()
            .flatten()
            .filter_map(|(field_name, config)| {
                let alias = config.alias()?;
                match self
                    .attributes()
                    .get(&FieldName::from(alias.to_string())) {
                    None => None,
                    Some(ExtractedFieldAttribute { tantivy_type, ..}) => {
                        assert!(
                            matches!(tantivy_type, SearchFieldType::Text(_)),
                            "unexpected type `{tantivy_type:?}` for `{field_name}` aliasing `{alias}`.  Expecting `SearchFieldType::Text`"
                        );
                        Some((field_name.clone(), config.clone()))
                    }
                }
            })
            .collect()
    }

    /// Returns a `Vec` of aliased JSON field names and their configs.
    pub fn aliased_json_configs(&self) -> Vec<(FieldName, SearchFieldConfig)> {
        if self.lazy.json.borrow().is_none() {
            *self.lazy.json.borrow_mut() = Some(self.options_data().json_configs());
        }
        self.lazy
            .json
            .borrow()
            .iter()
            .flatten()
            .filter_map(|(field_name, config)| {
                let alias = config.alias()?;
                match self
                    .attributes()
                    .get(&FieldName::from(alias.to_string())) {
                    None => None,
                    Some(ExtractedFieldAttribute { tantivy_type, ..}) => {
                        assert!(
                            matches!(tantivy_type, SearchFieldType::Json(_)),
                            "unexpected type `{tantivy_type:?}` for `{field_name}` aliasing `{alias}`.  Expecting `SearchFieldType::Json`"
                        );
                        Some((field_name.clone(), config.clone()))
                    }
                }
            })
            .collect()
    }

    pub fn get_field_type(&self, field_name: &FieldName) -> Option<SearchFieldType> {
        if field_name.is_ctid() {
            // the "ctid" field isn't an attribute, per se, in the index itself
            // it's one we add directly, so we need to account for it here
            return Some(SearchFieldType::U64(pg_sys::TIDOID));
        }
        self.attributes()
            .get(field_name)
            .map(|ExtractedFieldAttribute { tantivy_type, .. }| *tantivy_type)
            // account for aliased fields
            .or_else(|| self.get_aliased_field_type(field_name))
    }

    fn get_aliased_field_type(&self, field_name: &FieldName) -> Option<SearchFieldType> {
        // this is an expensive thing to do
        for (aliased_field_name, config) in self
            .aliased_text_configs()
            .into_iter()
            .chain(self.aliased_json_configs().into_iter())
        {
            if &aliased_field_name == field_name {
                if let Some(alias) = config.alias() {
                    return self.get_field_type(&FieldName::from(alias.to_string()));
                }
            }
        }

        None
    }

    pub fn attributes(&self) -> Ref<'_, HashMap<FieldName, ExtractedFieldAttribute>> {
        if self.lazy.attributes.borrow().is_empty() {
            *self.lazy.attributes.borrow_mut() = unsafe { extract_field_attributes(self.indexrel) };
        }
        self.lazy.attributes.borrow()
    }

    #[inline(always)]
    fn options_data(&self) -> &BM25IndexOptionsData {
        unsafe {
            assert!(!(*self.indexrel).rd_options.is_null());
            &*((*self.indexrel).rd_options as *const BM25IndexOptionsData)
        }
    }
}

// Postgres handles string options by placing each option offset bytes from the start of rdopts and
// plops the offset in the struct
#[repr(C)]
struct BM25IndexOptionsData {
    // varlena header (needed bc postgres treats this as bytea)
    vl_len_: i32,
    text_fields_offset: i32,
    numeric_fields_offset: i32,
    boolean_fields_offset: i32,
    json_fields_offset: i32,
    range_fields_offset: i32,
    datetime_fields_offset: i32,
    key_field_offset: i32,
    layer_sizes_offset: i32,
    inet_fields_offset: i32,
    target_segment_count: i32,
    background_layer_sizes_offset: i32,
    mutable_segment_rows: i32,
    sort_by_offset: i32,
    search_tokenizer_offset: i32,
}

impl BM25IndexOptionsData {
    /// Returns the configured `layer_sizes`, split into a [`Vec<u64>`] of byte sizes.
    ///
    /// If none is applied to the index, the specified `default` sizes are used.
    pub fn foreground_layer_sizes(&self) -> Vec<u64> {
        let foreground_layer_sizes_str = self.get_str(self.layer_sizes_offset, Default::default());
        if foreground_layer_sizes_str.trim().is_empty() {
            return DEFAULT_FOREGROUND_LAYER_SIZES.to_vec();
        }
        get_layer_sizes(&foreground_layer_sizes_str).collect()
    }

    pub fn background_layer_sizes(&self) -> Vec<u64> {
        let background_layer_sizes_str =
            self.get_str(self.background_layer_sizes_offset, Default::default());
        if background_layer_sizes_str.trim().is_empty() {
            return DEFAULT_BACKGROUND_LAYER_SIZES.to_vec();
        }
        get_layer_sizes(&background_layer_sizes_str).collect()
    }

    pub fn target_segment_count(&self) -> Option<i32> {
        if self.target_segment_count == 0 {
            None
        } else {
            Some(self.target_segment_count)
        }
    }

    pub fn mutable_segment_rows(&self) -> Option<NonZeroUsize> {
        if self.mutable_segment_rows > 0 {
            NonZeroUsize::new(self.mutable_segment_rows as usize)
        } else {
            None
        }
    }

    pub fn key_field_name(&self) -> Option<FieldName> {
        let key_field_name = self.get_str(self.key_field_offset, "".to_string());
        if key_field_name.is_empty() {
            return None;
        }
        Some(key_field_name.into())
    }

    /// Returns the sort_by configuration.
    /// - Empty string (not specified): defaults to none (no segment sorting)
    /// - "none": returns empty vec (no sorting)
    /// - Otherwise: returns parsed sort fields
    ///
    /// Note: This is only called during index build, so the allocation for the
    /// default case is acceptable and not worth optimizing with a static.
    pub fn sort_by(&self) -> Vec<SortByField> {
        let sort_by_str = self.get_str(self.sort_by_offset, "".to_string());
        if sort_by_str.is_empty() {
            // Default: no segment sorting
            return vec![];
        }
        parse_sort_by_string(&sort_by_str)
    }

    pub fn search_tokenizer(&self) -> Option<SearchTokenizer> {
        let expr = self.get_str(self.search_tokenizer_offset, "".to_string());
        if expr.is_empty() {
            return None;
        }
        crate::api::tokenizers::tokenizer_from_expression(&expr)
    }

    pub fn text_configs(&self) -> HashMap<FieldName, SearchFieldConfig> {
        self.deserialize_configs(self.text_fields_offset, &SearchFieldConfig::text_from_json)
    }

    fn inet_configs(&self) -> HashMap<FieldName, SearchFieldConfig> {
        self.deserialize_configs(self.inet_fields_offset, &SearchFieldConfig::inet_from_json)
    }

    pub fn numeric_configs(&self) -> HashMap<FieldName, SearchFieldConfig> {
        self.deserialize_configs(
            self.numeric_fields_offset,
            &SearchFieldConfig::numeric_from_json,
        )
    }

    pub fn boolean_configs(&self) -> HashMap<FieldName, SearchFieldConfig> {
        self.deserialize_configs(
            self.boolean_fields_offset,
            &SearchFieldConfig::boolean_from_json,
        )
    }

    pub fn json_configs(&self) -> HashMap<FieldName, SearchFieldConfig> {
        self.deserialize_configs(self.json_fields_offset, &SearchFieldConfig::json_from_json)
    }

    pub fn range_configs(&self) -> HashMap<FieldName, SearchFieldConfig> {
        self.deserialize_configs(
            self.range_fields_offset,
            &SearchFieldConfig::range_from_json,
        )
    }

    pub fn datetime_configs(&self) -> HashMap<FieldName, SearchFieldConfig> {
        self.deserialize_configs(
            self.datetime_fields_offset,
            &SearchFieldConfig::date_from_json,
        )
    }

    fn deserialize_configs(
        &self,
        offset: i32,
        parser: &dyn Fn(serde_json::Value) -> Result<SearchFieldConfig>,
    ) -> HashMap<FieldName, SearchFieldConfig> {
        let mut configs = HashMap::default();
        let config = self.get_str(offset, "".to_string());
        if !config.is_empty() {
            let mut deserialized = deserialize_config_fields(config, parser);
            for (field_name, config) in deserialized.drain() {
                configs.insert(field_name, config);
            }
        }
        configs
    }

    fn get_str(&self, offset: i32, default: String) -> String {
        if offset == 0 {
            default
        } else {
            let opts = self as *const _ as void_ptr as usize;
            let value =
                unsafe { CStr::from_ptr((opts + offset as usize) as *const std::os::raw::c_char) };

            value
                .to_str()
                .expect("value should be valid utf-8")
                .to_owned()
        }
    }
}

// it adds the tokenizer option to the list of relation options so we can parse it in amoptions
pub unsafe fn init() {
    // adding our own relopt type because zombodb does, but one of the built-in Postgres ones might be more appropriate
    RELOPT_KIND_PDB = pg_sys::add_reloption_kind();
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "text_fields".as_pg_cstr(),
        "JSON string specifying how text fields should be indexed".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_text_fields),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "numeric_fields".as_pg_cstr(),
        "JSON string specifying how numeric fields should be indexed".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_numeric_fields),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "boolean_fields".as_pg_cstr(),
        "JSON string specifying how boolean fields should be indexed".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_boolean_fields),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "json_fields".as_pg_cstr(),
        "JSON string specifying how JSON fields should be indexed".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_json_fields),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "range_fields".as_pg_cstr(),
        "JSON string specifying how range fields should be indexed".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_range_fields),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "datetime_fields".as_pg_cstr(),
        "JSON string specifying how date fields should be indexed".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_datetime_fields),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "key_field".as_pg_cstr(),
        "Column name as a string specify the unique identifier for a row".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_key_field),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "layer_sizes".as_pg_cstr(),
        "The sizes of each layer to merge in the foreground".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_layer_sizes),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_int_reloption(
        RELOPT_KIND_PDB,
        "target_segment_count".as_pg_cstr(),
        "When creating or reindexing, how many segments should be created".as_pg_cstr(),
        0,
        0,
        i32::MAX,
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_int_reloption(
        RELOPT_KIND_PDB,
        "mutable_segment_rows".as_pg_cstr(),
        "The size of mutable segments.".as_pg_cstr(),
        DEFAULT_MUTABLE_SEGMENT_ROWS as i32,
        0,
        MAX_MUTABLE_SEGMENT_ROWS as i32,
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "background_layer_sizes".as_pg_cstr(),
        "The sizes of each layer to merge in the background".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_layer_sizes),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "sort_by".as_pg_cstr(),
        "Comma-separated list of fields to sort segments by (e.g., 'field1 ASC, field2 DESC NULLS LAST')".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_sort_by),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
    pg_sys::add_string_reloption(
        RELOPT_KIND_PDB,
        "search_tokenizer".as_pg_cstr(),
        "Default search-time tokenizer for text/JSON fields".as_pg_cstr(),
        std::ptr::null(),
        Some(validate_search_tokenizer),
        pg_sys::AccessExclusiveLock as pg_sys::LOCKMODE,
    );
}

/// As a SearchFieldConfig is an enum, for it to be correctly serialized the variant needs
/// to be present on the json object. This helper method will "wrap" the json object in
/// another object with the variant key, which is passed into the function. For example:
///
/// {"Text": { <actual_config> }}
///
/// This way, serde will know to deserialize the config as SearchFieldConfig::Text.
fn deserialize_config_fields(
    serialized: String,
    parser: &dyn Fn(serde_json::Value) -> Result<SearchFieldConfig>,
) -> HashMap<FieldName, SearchFieldConfig> {
    let config_map: Map<String, serde_json::Value> = serde_json::from_str(&serialized)
        .unwrap_or_else(|err| panic!("failed to deserialize field config: {err:?}"));

    config_map
        .into_iter()
        .map(|(field_name, field_config)| {
            (
                field_name.clone().into(),
                parser(field_config).unwrap_or_else(|err| {
                    panic!(
                        "field config should be valid for SearchFieldConfig::{field_name}: {err}"
                    )
                }),
            )
        })
        .collect()
}

/// Represents a single field in the sort_by specification.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SortByField {
    pub field_name: FieldName,
    pub direction: SortByDirection,
}

/// Sort direction for sort_by fields
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
pub enum SortByDirection {
    #[default]
    Asc,
    Desc,
}

impl SortByField {
    pub fn new(field_name: FieldName, direction: SortByDirection) -> Self {
        Self {
            field_name,
            direction,
        }
    }
}

/// Parse a sort_by string like "field1 ASC, field2 DESC"
///
/// Grammar:
///   sort_by_value  ::= 'none' | sort_key { ',' sort_key }
///   sort_key       ::= field_name [ 'ASC' | 'DESC' ]
fn parse_sort_by_string(input: &str) -> Vec<SortByField> {
    let input = input.trim();

    // Handle 'none' case
    if input.eq_ignore_ascii_case("none") {
        return vec![];
    }

    let mut fields = Vec::new();

    for part in input.split(',') {
        let part = part.trim();
        if part.is_empty() {
            continue;
        }

        let field = parse_sort_key(part);
        fields.push(field);
    }

    if fields.is_empty() {
        panic!("invalid sort_by value: must specify at least one field or 'none'");
    }

    // Multi-field sorting is not yet supported
    if fields.len() > 1 {
        panic!(
            "sort_by specifies {} fields, but only single-field sorting is currently supported",
            fields.len()
        );
    }

    fields
}

/// Parse a single sort key like "field1 ASC NULLS FIRST" or "field1 DESC NULLS LAST"
///
/// NULLS ordering is REQUIRED to be explicit (to avoid confusion with PostgreSQL defaults).
/// Tantivy's NULL ordering is fixed:
/// - ASC: only NULLS FIRST is supported
/// - DESC: only NULLS LAST is supported
fn parse_sort_key(input: &str) -> SortByField {
    let input = input.trim();
    let mut tokens = input.split_whitespace();

    let field_name = tokens
        .next()
        .map(|s| FieldName::from(s.to_string()))
        .unwrap_or_else(|| panic!("invalid sort_by value: empty sort key"));

    let mut direction = SortByDirection::Asc; // Default: ASC
    let mut nulls_first: Option<bool> = None;

    while let Some(token) = tokens.next() {
        match token.to_uppercase().as_str() {
            "ASC" => direction = SortByDirection::Asc,
            "DESC" => direction = SortByDirection::Desc,
            "NULLS" => {
                let nulls_order = tokens.next().map(|s| s.to_uppercase()).unwrap_or_else(|| {
                    panic!("invalid sort_by value: expected FIRST or LAST after NULLS")
                });
                match nulls_order.as_str() {
                    "FIRST" => nulls_first = Some(true),
                    "LAST" => nulls_first = Some(false),
                    _ => panic!(
                        "invalid sort_by value: expected FIRST or LAST after NULLS, got: {}",
                        nulls_order
                    ),
                }
            }
            _ => panic!(
                "invalid sort_by value: unexpected token in sort key: {}",
                token
            ),
        }
    }

    // NULLS ordering is required to be explicit
    let Some(is_nulls_first) = nulls_first else {
        panic!(
            "invalid sort_by value: NULLS FIRST or NULLS LAST is required. \
             Supported: 'field ASC NULLS FIRST' or 'field DESC NULLS LAST'"
        )
    };

    // Validate NULLS ordering matches Tantivy's fixed behavior
    match (direction, is_nulls_first) {
        (SortByDirection::Asc, true) => {}   // ASC NULLS FIRST - OK
        (SortByDirection::Desc, false) => {} // DESC NULLS LAST - OK
        (SortByDirection::Asc, false) => {
            panic!(
                "invalid sort_by value: ASC only supports NULLS FIRST currently. \
                 Use 'field ASC NULLS FIRST'"
            )
        }
        (SortByDirection::Desc, true) => {
            panic!(
                "invalid sort_by value: DESC only supports NULLS LAST currently. \
                 Use 'field DESC NULLS LAST'"
            )
        }
    }

    SortByField::new(field_name, direction)
}

fn key_field_config(field_type: SearchFieldType) -> SearchFieldConfig {
    match field_type {
        SearchFieldType::I64(_) | SearchFieldType::U64(_) | SearchFieldType::F64(_) => {
            SearchFieldConfig::Numeric {
                indexed: true,
                fast: true,
                scale: None,
            }
        }
        SearchFieldType::Numeric64(_, scale) => SearchFieldConfig::Numeric {
            indexed: true,
            fast: true,
            scale: Some(scale),
        },
        SearchFieldType::NumericBytes(_, scale) => SearchFieldConfig::Numeric {
            indexed: true,
            fast: true,
            scale,
        },
        SearchFieldType::Text(_) | SearchFieldType::Uuid(_) => SearchFieldConfig::Text {
            indexed: true,
            fast: true,
            fieldnorms: false,

            // NB:  This should use the `SearchTokenizer::Keyword` tokenizer but for historical
            // reasons it uses the `SearchTokenizer::Raw` tokenizer but with the same filters
            // configuration as the `SearchTokenizer::Keyword` tokenizer.
            #[allow(deprecated)]
            tokenizer: SearchTokenizer::Raw(SearchTokenizerFilters::keyword().clone()),
            search_tokenizer: None,
            record: IndexRecordOption::Basic,
            normalizer: SearchNormalizer::Raw,
            column: None,
            k1: None,
            b: None,
        },
        SearchFieldType::Tokenized(..) => {
            panic!("the key_field cannot use a custom tokenizer configuration")
        }
        SearchFieldType::Inet(_) => SearchFieldConfig::Inet {
            indexed: true,
            fast: true,
        },
        SearchFieldType::Json(_) => SearchFieldConfig::Json {
            indexed: true,
            fast: true,
            fieldnorms: false,
            expand_dots: false,
            #[allow(deprecated)]
            tokenizer: SearchTokenizer::Raw(SearchTokenizerFilters::default()),
            search_tokenizer: None,
            record: IndexRecordOption::Basic,
            normalizer: SearchNormalizer::Raw,
            column: None,
            k1: None,
            b: None,
        },
        SearchFieldType::Range(_) => SearchFieldConfig::Range { fast: true },
        SearchFieldType::Bool(_) => SearchFieldConfig::Boolean {
            indexed: true,
            fast: true,
        },
        SearchFieldType::Date(_) => SearchFieldConfig::Date {
            indexed: true,
            fast: true,
        },
    }
}

#[cfg(any(test, feature = "pg_test"))]
#[pgrx::pg_schema]
mod tests {
    use super::*;
    use pgrx::pg_test;

    #[pg_test]
    #[should_panic(expected = "NULLS FIRST or NULLS LAST is required")]
    fn test_parse_sort_by_without_nulls_error() {
        // NULLS is required - omitting it should fail
        parse_sort_by_string("created_at ASC");
    }

    #[pg_test]
    fn test_parse_sort_by_asc_nulls_first() {
        // ASC NULLS FIRST is valid (matches Tantivy's behavior)
        let result = parse_sort_by_string("created_at ASC NULLS FIRST");
        assert_eq!(result.len(), 1);
        assert_eq!(
            result[0].field_name,
            FieldName::from("created_at".to_string())
        );
        assert_eq!(result[0].direction, SortByDirection::Asc);
    }

    #[pg_test]
    fn test_parse_sort_by_desc_nulls_last() {
        // DESC NULLS LAST is valid (matches Tantivy's behavior)
        let result = parse_sort_by_string("score DESC NULLS LAST");
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].field_name, FieldName::from("score".to_string()));
        assert_eq!(result[0].direction, SortByDirection::Desc);
    }

    #[pg_test]
    #[should_panic(expected = "ASC only supports NULLS FIRST")]
    fn test_parse_sort_by_asc_nulls_last_error() {
        // ASC NULLS LAST is invalid (opposite of Tantivy's behavior)
        parse_sort_by_string("created_at ASC NULLS LAST");
    }

    #[pg_test]
    #[should_panic(expected = "DESC only supports NULLS LAST")]
    fn test_parse_sort_by_desc_nulls_first_error() {
        // DESC NULLS FIRST is invalid (opposite of Tantivy's behavior)
        parse_sort_by_string("score DESC NULLS FIRST");
    }

    #[pg_test]
    #[should_panic(expected = "only single-field sorting is currently supported")]
    fn test_parse_sort_by_multiple_fields_error() {
        // Multi-field sorting is not yet supported
        parse_sort_by_string("category ASC NULLS FIRST, created_at DESC NULLS LAST");
    }

    #[pg_test]
    fn test_parse_sort_by_none() {
        assert!(parse_sort_by_string("none").is_empty());
        assert!(parse_sort_by_string("NONE").is_empty());
        assert!(parse_sort_by_string("None").is_empty());
    }

    #[pg_test]
    fn test_parse_sort_by_case_insensitive_keywords() {
        let result = parse_sort_by_string("field asc nulls first");
        assert_eq!(result[0].direction, SortByDirection::Asc);

        let result = parse_sort_by_string("field Desc Nulls Last");
        assert_eq!(result[0].direction, SortByDirection::Desc);
    }

    #[pg_test]
    fn test_parse_sort_by_whitespace_handling() {
        // Test that extra whitespace is handled correctly
        let result = parse_sort_by_string("  field1  ASC  NULLS  FIRST  ");
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].field_name, FieldName::from("field1".to_string()));
        assert_eq!(result[0].direction, SortByDirection::Asc);
    }

    #[pg_test]
    #[should_panic(expected = "must specify at least one field")]
    fn test_parse_sort_by_empty_error() {
        parse_sort_by_string("");
    }

    #[pg_test]
    #[should_panic(expected = "must specify at least one field")]
    fn test_parse_sort_by_only_whitespace_error() {
        parse_sort_by_string("   ");
    }

    #[pg_test]
    #[should_panic(expected = "unexpected token")]
    fn test_parse_sort_by_unexpected_token() {
        parse_sort_by_string("field ASC INVALID");
    }

    #[pg_test]
    fn test_sort_by_field_new() {
        let field = SortByField::new(FieldName::from("test".to_string()), SortByDirection::Desc);
        assert_eq!(field.field_name, FieldName::from("test".to_string()));
        assert_eq!(field.direction, SortByDirection::Desc);
    }

    #[pg_test]
    fn test_sort_by_direction_default() {
        let dir = SortByDirection::default();
        assert_eq!(dir, SortByDirection::Asc);
    }
}
