// Copyright 2024 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 std::collections::{BTreeMap, HashMap};
use std::fmt;

use anyhow::{Context, anyhow};
use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
use postgres_openssl::MakeTlsConnector;
use risingwave_common::bail;
use risingwave_common::catalog::{ColumnDesc, ColumnId};
use risingwave_common::types::{DataType, ScalarImpl, StructType};
use sea_schema::postgres::def::{ColumnType as SeaType, TableDef, TableInfo};
use sea_schema::postgres::discovery::SchemaDiscovery;
use sea_schema::sea_query::{Alias, IntoIden};
use serde::Deserialize;
use sqlx::postgres::{PgConnectOptions, PgSslMode};
use sqlx::{PgPool, Row};
use thiserror_ext::AsReport;
use tokio_postgres::types::Kind as PgKind;
use tokio_postgres::{Client as PgClient, NoTls};

#[cfg(not(madsim))]
use super::maybe_tls_connector::MaybeMakeTlsConnector;
use crate::error::ConnectorResult;
use crate::sink::postgres::TcpKeepaliveConfig;

/// SQL query to discover primary key columns directly from PostgreSQL system tables.
/// This bypasses querying `information_schema.table_constraints` to avoid permission issues.
const DISCOVER_PRIMARY_KEY_QUERY: &str = r#"
    SELECT a.attname as column_name
    FROM pg_index i
    JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
    WHERE i.indrelid = ($1 || '.' || $2)::regclass
      AND i.indisprimary = true
    ORDER BY array_position(i.indkey, a.attnum)
"#;

/// Discover pgvector columns with both `atttypmod` (dimension) and `format_type` text.
/// `vector(n)` is stored as `atttypmod = n`, while dimension-less `vector` uses `-1`.
/// We rely on this to keep user-defined type modifiers that are not preserved by sea-schema.
const DISCOVER_PGVECTOR_COLUMNS_QUERY: &str = r#"
    SELECT
      a.attname as column_name,
      a.atttypmod as atttypmod,
      format_type(a.atttypid, a.atttypmod) as formatted_type
    FROM pg_attribute a
    JOIN pg_class c ON c.oid = a.attrelid
    JOIN pg_namespace n ON n.oid = c.relnamespace
    JOIN pg_type t ON t.oid = a.atttypid
    WHERE n.nspname = $1
      AND c.relname = $2
      AND t.typname = 'vector'
      AND a.attnum > 0
      AND NOT a.attisdropped
    ORDER BY a.attnum
"#;

pub struct PgConnectionConfig {
    pub host: String,
    pub port: String,
    pub user: String,
    pub password: String,
    pub database: String,
    pub ssl_mode: SslMode,
    pub ssl_root_cert: Option<String>,
}

pub fn pg_connection_config_from_properties(
    props: &BTreeMap<String, String>,
) -> ConnectorResult<PgConnectionConfig> {
    Ok(PgConnectionConfig {
        host: props
            .get("hostname")
            .context("missing `hostname` in postgres-cdc properties")?
            .clone(),
        port: props
            .get("port")
            .context("missing `port` in postgres-cdc properties")?
            .clone(),
        user: props
            .get("username")
            .context("missing `username` in postgres-cdc properties")?
            .clone(),
        password: props.get("password").cloned().unwrap_or_default(),
        database: props
            .get("database.name")
            .context("missing `database.name` in postgres-cdc properties")?
            .clone(),
        ssl_mode: props
            .get("ssl.mode")
            .and_then(|v| v.parse::<SslMode>().ok())
            .unwrap_or_default(),
        ssl_root_cert: props.get("ssl.root.cert").cloned(),
    })
}

pub async fn create_pg_client_from_properties(
    props: &BTreeMap<String, String>,
    tcp_keepalive: Option<TcpKeepaliveConfig>,
) -> ConnectorResult<PgClient> {
    let config = pg_connection_config_from_properties(props)?;
    create_pg_client(
        &config.user,
        &config.password,
        &config.host,
        &config.port,
        &config.database,
        &config.ssl_mode,
        &config.ssl_root_cert,
        tcp_keepalive,
    )
    .await
    .map_err(Into::into)
}

pub async fn discover_pgvector_dimensions(
    client: &PgClient,
    schema: &str,
    table: &str,
) -> ConnectorResult<HashMap<String, usize>> {
    let rows = client
        .query(DISCOVER_PGVECTOR_COLUMNS_QUERY, &[&schema, &table])
        .await?;

    let mut dims = HashMap::new();
    for row in rows {
        let col_name: String = row.get("column_name");
        let atttypmod: i32 = row.get("atttypmod");
        if atttypmod > 0
            && let Ok(dim) = usize::try_from(atttypmod)
        {
            dims.insert(col_name, dim);
        }
    }
    Ok(dims)
}

#[derive(Debug, Clone, PartialEq, Deserialize, Default)]
#[serde(rename_all = "lowercase")]
pub enum SslMode {
    #[serde(alias = "disable")]
    Disabled,
    #[serde(alias = "prefer")]
    #[default]
    Preferred,
    #[serde(alias = "require")]
    Required,
    /// verify that the server is trustworthy by checking the certificate chain
    /// up to the root certificate stored on the client.
    #[serde(alias = "verify-ca")]
    VerifyCa,
    /// Besides verify the certificate, will also verify that the serverhost name
    /// matches the name stored in the server certificate.
    #[serde(alias = "verify-full")]
    VerifyFull,
}

pub struct PostgresExternalTable {
    column_descs: Vec<ColumnDesc>,
    pk_names: Vec<String>,
}

impl PostgresExternalTable {
    /// Discover primary key columns directly from PostgreSQL system tables.
    /// This bypasses querying `information_schema.table_constraints` to avoid requiring table owner permissions.
    async fn discover_primary_key(
        connection: &PgPool,
        schema_name: &str,
        table_name: &str,
    ) -> ConnectorResult<Vec<String>> {
        let rows = sqlx::query(DISCOVER_PRIMARY_KEY_QUERY)
            .bind(schema_name)
            .bind(table_name)
            .fetch_all(connection)
            .await
            .context("Failed to discover primary key columns")?;

        let pk_columns = rows
            .into_iter()
            .map(|row| row.get::<String, _>("column_name"))
            .collect();

        Ok(pk_columns)
    }

    /// Discover schema with workaround for primary key discovery
    /// This method uses direct PostgreSQL system table queries for primary keys
    /// to avoid permission issues when querying `information_schema.table_constraints`
    async fn discover_pk_and_full_columns(
        username: &str,
        password: &str,
        host: &str,
        port: u16,
        database: &str,
        schema: &str,
        table: &str,
        ssl_mode: &SslMode,
        ssl_root_cert: &Option<String>,
    ) -> ConnectorResult<(Vec<sea_schema::postgres::def::ColumnInfo>, Vec<String>)> {
        let mut options = PgConnectOptions::new()
            .username(username)
            .password(password)
            .host(host)
            .port(port)
            .database(database)
            .ssl_mode(match ssl_mode {
                SslMode::Disabled => PgSslMode::Disable,
                SslMode::Preferred => PgSslMode::Prefer,
                SslMode::Required => PgSslMode::Require,
                SslMode::VerifyCa => PgSslMode::VerifyCa,
                SslMode::VerifyFull => PgSslMode::VerifyFull,
            });

        if (*ssl_mode == SslMode::VerifyCa || *ssl_mode == SslMode::VerifyFull)
            && let Some(root_cert) = ssl_root_cert
        {
            options = options.ssl_root_cert(root_cert.as_str());
        }

        let connection = PgPool::connect_with(options).await?;

        // Use sea-schema only for column discovery (no permission issues)
        let schema_discovery = SchemaDiscovery::new(connection.clone(), schema);
        let empty_map: HashMap<String, Vec<String>> = HashMap::new();
        let columns = schema_discovery
            .discover_columns(
                Alias::new(schema).into_iden(),
                Alias::new(table).into_iden(),
                &empty_map,
            )
            .await?;

        let pgvector_columns = sqlx::query(DISCOVER_PGVECTOR_COLUMNS_QUERY)
            .bind(schema)
            .bind(table)
            .fetch_all(&connection)
            .await
            .context("Failed to discover PostgreSQL pgvector columns")?;
        let formatted_type_by_column: HashMap<String, String> = pgvector_columns
            .into_iter()
            .map(|row| {
                (
                    row.get::<String, _>("column_name"),
                    row.get::<String, _>("formatted_type"),
                )
            })
            .collect();

        // sea-schema reports pgvector as `Unknown("vector")` and drops the dimension.
        // Patch it with PostgreSQL's formatted type text so we can derive vector(n).
        let mut columns = columns;
        for col in &mut columns {
            if let SeaType::Unknown(name) = &col.col_type
                && name.eq_ignore_ascii_case("vector")
                && let Some(formatted_type) = formatted_type_by_column.get(&col.name)
            {
                col.col_type = SeaType::Unknown(formatted_type.clone());
            }
        }

        // Use direct system table query for primary key discovery
        let pk_columns = Self::discover_primary_key(&connection, schema, table).await?;

        Ok((columns, pk_columns))
    }

    async fn discover_schema(
        username: &str,
        password: &str,
        host: &str,
        port: u16,
        database: &str,
        schema: &str,
        table: &str,
        ssl_mode: &SslMode,
        ssl_root_cert: &Option<String>,
    ) -> ConnectorResult<TableDef> {
        let mut options = PgConnectOptions::new()
            .username(username)
            .password(password)
            .host(host)
            .port(port)
            .database(database)
            .ssl_mode(match ssl_mode {
                SslMode::Disabled => PgSslMode::Disable,
                SslMode::Preferred => PgSslMode::Prefer,
                SslMode::Required => PgSslMode::Require,
                SslMode::VerifyCa => PgSslMode::VerifyCa,
                SslMode::VerifyFull => PgSslMode::VerifyFull,
            });

        if (*ssl_mode == SslMode::VerifyCa || *ssl_mode == SslMode::VerifyFull)
            && let Some(root_cert) = ssl_root_cert
        {
            options = options.ssl_root_cert(root_cert.as_str());
        }

        let connection = PgPool::connect_with(options).await?;
        let schema_discovery = SchemaDiscovery::new(connection, schema);
        // fetch column schema and primary key
        let empty_map = HashMap::new();
        let table_schema = schema_discovery
            .discover_table(
                TableInfo {
                    name: table.to_owned(),
                    of_type: None,
                },
                &empty_map,
            )
            .await?;
        Ok(table_schema)
    }

    pub async fn connect(
        username: &str,
        password: &str,
        host: &str,
        port: u16,
        database: &str,
        schema: &str,
        table: &str,
        ssl_mode: &SslMode,
        ssl_root_cert: &Option<String>,
        is_append_only: bool,
    ) -> ConnectorResult<Self> {
        tracing::debug!("connect to postgres external table");

        let (columns, pk_names) = Self::discover_pk_and_full_columns(
            username,
            password,
            host,
            port,
            database,
            schema,
            table,
            ssl_mode,
            ssl_root_cert,
        )
        .await?;

        let mut column_descs = vec![];
        for col in &columns {
            let rw_data_type = sea_type_to_rw_type(&col.col_type)?;
            let column_desc = if let Some(ref default_expr) = col.default {
                // parse the value of "column_default" field in information_schema.columns,
                // non number data type will be stored as "'value'::type"
                let val_text = default_expr
                    .0
                    .split("::")
                    .map(|s| s.trim_matches('\''))
                    .next()
                    .expect("default value expression");

                match ScalarImpl::from_text(val_text, &rw_data_type) {
                    Ok(scalar) => ColumnDesc::named_with_default_value(
                        col.name.clone(),
                        ColumnId::placeholder(),
                        rw_data_type.clone(),
                        Some(scalar),
                    ),
                    Err(err) => {
                        tracing::warn!(error=%err.as_report(), "failed to parse postgres default value expression, only constant is supported");
                        ColumnDesc::named(col.name.clone(), ColumnId::placeholder(), rw_data_type)
                    }
                }
            } else {
                ColumnDesc::named(col.name.clone(), ColumnId::placeholder(), rw_data_type)
            };
            column_descs.push(column_desc);
        }

        // Check primary key existence using the directly discovered pk_names
        if !is_append_only && pk_names.is_empty() {
            return Err(anyhow!(
                "Postgres table should define the primary key for non-append-only tables"
            )
            .into());
        }

        Ok(Self {
            column_descs,
            pk_names,
        })
    }

    // return the mapping from column name to pg type, the pg type is used for writing data to postgres
    pub async fn type_mapping(
        username: &str,
        password: &str,
        host: &str,
        port: u16,
        database: &str,
        schema: &str,
        table: &str,
        ssl_mode: &SslMode,
        ssl_root_cert: &Option<String>,
        is_append_only: bool,
    ) -> ConnectorResult<HashMap<String, tokio_postgres::types::Type>> {
        tracing::debug!("connect to postgres external table to get type mapping");
        let table_schema = Self::discover_schema(
            username,
            password,
            host,
            port,
            database,
            schema,
            table,
            ssl_mode,
            ssl_root_cert,
        )
        .await?;
        let mut column_name_to_pg_type = HashMap::new();
        for col in &table_schema.columns {
            let pg_type = sea_type_to_pg_type(&col.col_type)?;
            column_name_to_pg_type.insert(col.name.clone(), pg_type);
        }
        if !is_append_only && table_schema.primary_key_constraints.is_empty() {
            return Err(anyhow!(
                "Postgres table should define the primary key for non-append-only tables"
            )
            .into());
        }
        Ok(column_name_to_pg_type)
    }

    pub fn column_descs(&self) -> &Vec<ColumnDesc> {
        &self.column_descs
    }

    pub fn pk_names(&self) -> &Vec<String> {
        &self.pk_names
    }
}

impl fmt::Display for SslMode {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            SslMode::Disabled => "disabled",
            SslMode::Preferred => "preferred",
            SslMode::Required => "required",
            SslMode::VerifyCa => "verify-ca",
            SslMode::VerifyFull => "verify-full",
        })
    }
}

impl std::str::FromStr for SslMode {
    type Err = serde_json::Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        serde_json::from_value(serde_json::Value::String(s.to_owned()))
    }
}

pub async fn create_pg_client(
    user: &str,
    password: &str,
    host: &str,
    port: &str,
    database: &str,
    ssl_mode: &SslMode,
    ssl_root_cert: &Option<String>,
    tcp_keepalive: Option<TcpKeepaliveConfig>,
) -> anyhow::Result<PgClient> {
    let mut pg_config = tokio_postgres::Config::new();
    pg_config
        .user(user)
        .password(password)
        .host(host)
        .port(port.parse::<u16>().unwrap())
        .dbname(database);

    // Configure TCP keepalive if provided
    if let Some(keepalive) = tcp_keepalive {
        pg_config.keepalives(true);
        pg_config.keepalives_idle(std::time::Duration::from_secs(
            keepalive.tcp_keepalive_idle as u64,
        ));
        #[cfg(not(target_os = "windows"))]
        {
            pg_config.keepalives_interval(std::time::Duration::from_secs(
                keepalive.tcp_keepalive_interval as u64,
            ));
            pg_config.keepalives_retries(keepalive.tcp_keepalive_count);
        }
        tracing::info!(
            "TCP keepalive enabled: idle={}s, interval={}s, retries={}",
            keepalive.tcp_keepalive_idle,
            keepalive.tcp_keepalive_interval,
            keepalive.tcp_keepalive_count
        );
    }

    let (_verify_ca, verify_hostname) = match ssl_mode {
        SslMode::VerifyCa => (true, false),
        SslMode::VerifyFull => (true, true),
        _ => (false, false),
    };

    #[cfg(not(madsim))]
    let connector = match ssl_mode {
        SslMode::Disabled => {
            pg_config.ssl_mode(tokio_postgres::config::SslMode::Disable);
            MaybeMakeTlsConnector::NoTls(NoTls)
        }
        SslMode::Preferred => {
            pg_config.ssl_mode(tokio_postgres::config::SslMode::Prefer);
            match SslConnector::builder(SslMethod::tls()) {
                Ok(mut builder) => {
                    // disable certificate verification for `prefer`
                    builder.set_verify(SslVerifyMode::NONE);
                    MaybeMakeTlsConnector::Tls(MakeTlsConnector::new(builder.build()))
                }
                Err(e) => {
                    tracing::warn!(error = %e.as_report(), "SSL connector error");
                    MaybeMakeTlsConnector::NoTls(NoTls)
                }
            }
        }
        SslMode::Required => {
            pg_config.ssl_mode(tokio_postgres::config::SslMode::Require);
            let mut builder = SslConnector::builder(SslMethod::tls())?;
            // disable certificate verification for `require`
            builder.set_verify(SslVerifyMode::NONE);
            MaybeMakeTlsConnector::Tls(MakeTlsConnector::new(builder.build()))
        }

        SslMode::VerifyCa | SslMode::VerifyFull => {
            pg_config.ssl_mode(tokio_postgres::config::SslMode::Require);
            let mut builder = SslConnector::builder(SslMethod::tls())?;
            if let Some(ssl_root_cert) = ssl_root_cert {
                builder.set_ca_file(ssl_root_cert).map_err(|e| {
                    anyhow!(format!("bad ssl root cert error: {}", e.to_report_string()))
                })?;
            }
            let mut connector = MakeTlsConnector::new(builder.build());
            if !verify_hostname {
                connector.set_callback(|config, _| {
                    config.set_verify_hostname(false);
                    Ok(())
                });
            }
            MaybeMakeTlsConnector::Tls(connector)
        }
    };
    #[cfg(madsim)]
    let connector = NoTls;

    let (client, connection) = pg_config.connect(connector).await?;

    tokio::spawn(async move {
        if let Err(e) = connection.await {
            tracing::error!(error = %e.as_report(), "postgres connection error");
        }
    });

    Ok(client)
}

// Used for both source and sink connector
pub fn sea_type_to_rw_type(col_type: &SeaType) -> ConnectorResult<DataType> {
    let dtype = match col_type {
        SeaType::SmallInt | SeaType::SmallSerial => DataType::Int16,
        SeaType::Integer | SeaType::Serial => DataType::Int32,
        SeaType::BigInt | SeaType::BigSerial => DataType::Int64,
        SeaType::Money | SeaType::Decimal(_) | SeaType::Numeric(_) => DataType::Decimal,
        SeaType::Real => DataType::Float32,
        SeaType::DoublePrecision => DataType::Float64,
        SeaType::Varchar(_) | SeaType::Char(_) | SeaType::Text => DataType::Varchar,
        SeaType::Bytea => DataType::Bytea,
        SeaType::Timestamp(_) => DataType::Timestamp,
        SeaType::TimestampWithTimeZone(_) => DataType::Timestamptz,
        SeaType::Date => DataType::Date,
        SeaType::Time(_) | SeaType::TimeWithTimeZone(_) => DataType::Time,
        SeaType::Interval(_) => DataType::Interval,
        SeaType::Boolean => DataType::Boolean,
        SeaType::Point => DataType::Struct(StructType::new(vec![
            ("x", DataType::Float32),
            ("y", DataType::Float32),
        ])),
        SeaType::Uuid => DataType::Varchar,
        SeaType::Xml => DataType::Varchar,
        SeaType::Json => DataType::Jsonb,
        SeaType::JsonBinary => DataType::Jsonb,
        SeaType::Array(def) => {
            let item_type = match def.col_type.as_ref() {
                Some(ty) => sea_type_to_rw_type(ty.as_ref())?,
                None => {
                    return Err(anyhow!("ARRAY type missing element type").into());
                }
            };

            DataType::list(item_type)
        }
        SeaType::PgLsn => DataType::Int64,
        SeaType::Cidr
        | SeaType::Inet
        | SeaType::MacAddr
        | SeaType::MacAddr8
        | SeaType::Int4Range
        | SeaType::Int8Range
        | SeaType::NumRange
        | SeaType::TsRange
        | SeaType::TsTzRange
        | SeaType::DateRange
        | SeaType::Enum(_) => DataType::Varchar,
        SeaType::Line
        | SeaType::Lseg
        | SeaType::Box
        | SeaType::Path
        | SeaType::Polygon
        | SeaType::Circle
        | SeaType::Bit(_)
        | SeaType::VarBit(_)
        | SeaType::TsVector
        | SeaType::TsQuery => {
            bail!("{:?} type not supported", col_type);
        }
        SeaType::Unknown(name) => {
            if let Some(dim) = parse_pgvector_dimension(name)? {
                DataType::Vector(dim)
            } else {
                // NOTES: user-defined enum type is classified as `Unknown`
                tracing::warn!("Unknown Postgres data type: {name}, map to varchar");
                DataType::Varchar
            }
        }
    };

    Ok(dtype)
}

fn parse_pgvector_dimension(type_name: &str) -> ConnectorResult<Option<usize>> {
    let normalized = type_name.trim().to_ascii_lowercase();
    if normalized == "vector" {
        bail!("pgvector type `vector` is missing dimension, expected `vector(n)`")
    }
    if !normalized.starts_with("vector(") || !normalized.ends_with(')') {
        return Ok(None);
    }

    let dim_text = normalized
        .trim_start_matches("vector(")
        .trim_end_matches(')')
        .trim();
    let dim = dim_text
        .parse::<usize>()
        .map_err(|_| anyhow!("invalid pgvector dimension in type `{type_name}`"))?;

    if !(1..=DataType::VEC_MAX_SIZE).contains(&dim) {
        bail!(
            "pgvector dimension out of range in type `{}`: expect 1..={}",
            type_name,
            DataType::VEC_MAX_SIZE
        );
    }

    Ok(Some(dim))
}

// Used for sink connector
// We use `sea-schema` for table schema discovery.
// So we have to map `sea-schema` pg types
// to `tokio-postgres` pg types (which we use for query binding).
fn sea_type_to_pg_type(sea_type: &SeaType) -> ConnectorResult<tokio_postgres::types::Type> {
    use tokio_postgres::types::Type as PgType;
    match sea_type {
        SeaType::SmallInt => Ok(PgType::INT2),
        SeaType::Integer => Ok(PgType::INT4),
        SeaType::BigInt => Ok(PgType::INT8),
        SeaType::Decimal(_) => Ok(PgType::NUMERIC),
        SeaType::Numeric(_) => Ok(PgType::NUMERIC),
        SeaType::Real => Ok(PgType::FLOAT4),
        SeaType::DoublePrecision => Ok(PgType::FLOAT8),
        SeaType::Varchar(_) => Ok(PgType::VARCHAR),
        SeaType::Char(_) => Ok(PgType::CHAR),
        SeaType::Text => Ok(PgType::TEXT),
        SeaType::Bytea => Ok(PgType::BYTEA),
        SeaType::Timestamp(_) => Ok(PgType::TIMESTAMP),
        SeaType::TimestampWithTimeZone(_) => Ok(PgType::TIMESTAMPTZ),
        SeaType::Date => Ok(PgType::DATE),
        SeaType::Time(_) => Ok(PgType::TIME),
        SeaType::TimeWithTimeZone(_) => Ok(PgType::TIMETZ),
        SeaType::Interval(_) => Ok(PgType::INTERVAL),
        SeaType::Boolean => Ok(PgType::BOOL),
        SeaType::Point => Ok(PgType::POINT),
        SeaType::Uuid => Ok(PgType::UUID),
        SeaType::Json => Ok(PgType::JSON),
        SeaType::JsonBinary => Ok(PgType::JSONB),
        SeaType::Array(t) => {
            let Some(t) = t.col_type.as_ref() else {
                bail!("missing array type")
            };
            match t.as_ref() {
                // RW only supports 1 level of nesting.
                SeaType::SmallInt => Ok(PgType::INT2_ARRAY),
                SeaType::Integer => Ok(PgType::INT4_ARRAY),
                SeaType::BigInt => Ok(PgType::INT8_ARRAY),
                SeaType::Decimal(_) => Ok(PgType::NUMERIC_ARRAY),
                SeaType::Numeric(_) => Ok(PgType::NUMERIC_ARRAY),
                SeaType::Real => Ok(PgType::FLOAT4_ARRAY),
                SeaType::DoublePrecision => Ok(PgType::FLOAT8_ARRAY),
                SeaType::Varchar(_) => Ok(PgType::VARCHAR_ARRAY),
                SeaType::Char(_) => Ok(PgType::CHAR_ARRAY),
                SeaType::Text => Ok(PgType::TEXT_ARRAY),
                SeaType::Bytea => Ok(PgType::BYTEA_ARRAY),
                SeaType::Timestamp(_) => Ok(PgType::TIMESTAMP_ARRAY),
                SeaType::TimestampWithTimeZone(_) => Ok(PgType::TIMESTAMPTZ_ARRAY),
                SeaType::Date => Ok(PgType::DATE_ARRAY),
                SeaType::Time(_) => Ok(PgType::TIME_ARRAY),
                SeaType::TimeWithTimeZone(_) => Ok(PgType::TIMETZ_ARRAY),
                SeaType::Interval(_) => Ok(PgType::INTERVAL_ARRAY),
                SeaType::Boolean => Ok(PgType::BOOL_ARRAY),
                SeaType::Point => Ok(PgType::POINT_ARRAY),
                SeaType::Uuid => Ok(PgType::UUID_ARRAY),
                SeaType::Json => Ok(PgType::JSON_ARRAY),
                SeaType::JsonBinary => Ok(PgType::JSONB_ARRAY),
                SeaType::Array(_) => bail!("nested array type is not supported"),
                SeaType::Unknown(name) => {
                    // Treat as enum type
                    Ok(PgType::new(
                        name.clone(),
                        0,
                        PgKind::Array(PgType::new(
                            name.clone(),
                            0,
                            PgKind::Enum(vec![]),
                            "".into(),
                        )),
                        "".into(),
                    ))
                }
                _ => bail!("unsupported array type: {:?}", t),
            }
        }
        SeaType::Unknown(name) => {
            // Treat as enum type
            Ok(PgType::new(
                name.clone(),
                0,
                PgKind::Enum(vec![]),
                "".into(),
            ))
        }
        _ => bail!("unsupported type: {:?}", sea_type),
    }
}

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

    #[test]
    fn test_parse_pgvector_dimension() {
        assert_eq!(parse_pgvector_dimension("vector(3)").unwrap(), Some(3));
        assert_eq!(parse_pgvector_dimension("VECTOR(768)").unwrap(), Some(768));
        assert_eq!(parse_pgvector_dimension("varchar").unwrap(), None);
    }

    #[test]
    fn test_parse_pgvector_dimension_requires_size() {
        let err = parse_pgvector_dimension("vector").unwrap_err();
        assert!(err.to_string().contains("missing dimension"));
    }
}
