//! [`Warehouse`] — an Iceberg warehouse in one directory: `catalog.redb`
//! (the embedded [`RedbCatalog`]) beside a `warehouse/` data tree, opened with
//! one call. The catalog-from-path constructor znippy asked for.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use arrow_schema::Schema as ArrowSchema;
use iceberg::io::LocalFsStorageFactory;
use iceberg::spec::{FormatVersion, PartitionSpec, Transform};

/// Iceberg table format version skade writes for new tables. **V3** is the
/// newest the spec (and iceberg-rust 0.9.1) supports — it unlocks the v3
/// formats (binary deletion vectors, variant, native geometry/geography,
/// nanosecond timestamps, row lineage, default column values) and is the
/// on-ramp to v4 (Parquet metadata). Our stack (iceberg-rust + iceberg-datafusion)
/// reads V3; manifests are still Avro until v4 lands. See ROADMAP.md.
const NEW_TABLE_FORMAT: FormatVersion = FormatVersion::V3;
use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation, TableIdent};
use skade_katalog::{RedbCatalog, RedbCatalogBuilder};

use crate::bridge::arrow_to_iceberg;
use crate::error::{Result, SkadeError};
use crate::object_store::{ObjectStore, ObjectStoreConfig, ObjectStoreFactory};
use crate::table::Table;

/// Namespace used for table names given without an explicit `ns.` prefix.
pub const DEFAULT_NAMESPACE: &str = "main";

/// An embedded Iceberg warehouse rooted at one local directory:
///
/// ```text
/// <root>/catalog.redb   — the RedbCatalog (namespaces + table pointers, ACID)
/// <root>/warehouse/     — table metadata JSON, Avro manifests, Parquet data
/// ```
///
/// Single-process (redb file lock); drop the `Warehouse` before reopening the
/// same directory. Table names are `"table"` (→ namespace [`DEFAULT_NAMESPACE`])
/// or `"ns.table"` / `"a.b.table"`.
pub struct Warehouse {
    catalog: Arc<RedbCatalog>,
    root: PathBuf,
}

impl Warehouse {
    /// Open (creating if absent) the warehouse at `dir`.
    pub async fn open(dir: impl AsRef<Path>) -> Result<Self> {
        let root = dir.as_ref().to_path_buf();
        std::fs::create_dir_all(root.join("warehouse"))?;
        // file:// URIs must be absolute.
        let root = root.canonicalize()?;

        let catalog = RedbCatalogBuilder::default()
            .db_path(root.join("catalog.redb"))
            .warehouse_location(format!("file://{}", root.join("warehouse").display()))
            .with_storage_factory(Arc::new(LocalFsStorageFactory))
            .load("skade", HashMap::new())
            .await?;
        let wh = Warehouse {
            catalog: Arc::new(catalog),
            root,
        };
        wh.ensure_namespace(&NamespaceIdent::new(DEFAULT_NAMESPACE.to_string()))
            .await?;
        Ok(wh)
    }

    /// Open a warehouse whose Iceberg data/metadata blobs live in a pluggable
    /// [`ObjectStore`] (e.g. a shared MinIO), while the redb catalog file stays
    /// on a local disk at `db_path`.
    ///
    /// `warehouse_uri` is the location iceberg stamps into table metadata and
    /// passes back to the store on every read/write (e.g.
    /// `s3://mybucket/warehouse` for an S3 store, or `file:///…` for a local
    /// store). The supplied `store` handles the bytes-IO behind that URI.
    ///
    /// This is the constructor a consumer (Njord) uses to run skade → Iceberg on
    /// a shared MinIO with the S3 SDK of its choice — see
    /// [`crate::object_store`]. [`Warehouse::open`] is unchanged and keeps using
    /// the local filesystem; nothing here affects existing callers.
    pub async fn open_with_store(
        db_path: impl AsRef<Path>,
        warehouse_uri: impl Into<String>,
        store: Arc<dyn ObjectStore>,
    ) -> Result<Self> {
        Self::open_with_factory(
            db_path,
            warehouse_uri,
            Arc::new(ObjectStoreFactory::from_store(store)),
        )
        .await
    }

    /// Like [`open_with_store`](Self::open_with_store) but from a serializable
    /// [`ObjectStoreConfig`] (the backend is built lazily, and the config
    /// survives an iceberg metadata round-trip). Use this for S3/MinIO backends
    /// configured from a settings file or env.
    pub async fn open_with_object_store_config(
        db_path: impl AsRef<Path>,
        warehouse_uri: impl Into<String>,
        config: ObjectStoreConfig,
    ) -> Result<Self> {
        Self::open_with_factory(
            db_path,
            warehouse_uri,
            Arc::new(ObjectStoreFactory::from_config(config)),
        )
        .await
    }

    async fn open_with_factory(
        db_path: impl AsRef<Path>,
        warehouse_uri: impl Into<String>,
        factory: Arc<dyn iceberg::io::StorageFactory>,
    ) -> Result<Self> {
        let db_path = db_path.as_ref().to_path_buf();
        if let Some(parent) = db_path.parent() {
            if !parent.as_os_str().is_empty() {
                std::fs::create_dir_all(parent)?;
            }
        }
        let warehouse_uri = warehouse_uri.into();
        let catalog = RedbCatalogBuilder::default()
            .db_path(&db_path)
            .warehouse_location(warehouse_uri)
            .with_storage_factory(factory)
            .load("skade", HashMap::new())
            .await?;
        // The "root" for an object-store warehouse is the catalog file's dir
        // (the only thing on local disk); the data tree lives in the store.
        let root = db_path
            .parent()
            .map(|p| p.to_path_buf())
            .unwrap_or_else(|| PathBuf::from("."));
        let wh = Warehouse { catalog: Arc::new(catalog), root };
        wh.ensure_namespace(&NamespaceIdent::new(DEFAULT_NAMESPACE.to_string()))
            .await?;
        Ok(wh)
    }

    /// The warehouse root directory.
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// The underlying embedded catalog (an [`iceberg::Catalog`]).
    pub fn catalog(&self) -> Arc<RedbCatalog> {
        self.catalog.clone()
    }

    /// Parse `"table"` / `"ns.table"` / `"a.b.table"` into a [`TableIdent`].
    pub fn table_ident(&self, name: &str) -> Result<TableIdent> {
        let parts: Vec<&str> = name.split('.').filter(|s| !s.is_empty()).collect();
        match parts.as_slice() {
            [] => Err(SkadeError::Other(format!("invalid table name '{name}'"))),
            [t] => Ok(TableIdent::new(
                NamespaceIdent::new(DEFAULT_NAMESPACE.to_string()),
                (*t).to_string(),
            )),
            [ns @ .., t] => {
                let ns = NamespaceIdent::from_strs(ns.iter().copied())?;
                Ok(TableIdent::new(ns, (*t).to_string()))
            }
        }
    }

    async fn ensure_namespace(&self, ns: &NamespaceIdent) -> Result<()> {
        if !self.catalog.namespace_exists(ns).await? {
            self.catalog.create_namespace(ns, HashMap::new()).await?;
        }
        Ok(())
    }

    /// Create `name` with an Iceberg schema derived from `schema`
    /// (field ids `1..N`, see [`arrow_to_iceberg`]). The namespace is created
    /// if missing.
    pub async fn create_table(&self, name: &str, schema: &ArrowSchema) -> Result<Table> {
        let ident = self.table_ident(name)?;
        self.ensure_namespace(ident.namespace()).await?;
        let creation = TableCreation::builder()
            .name(ident.name().to_string())
            .schema(arrow_to_iceberg(schema)?)
            .format_version(NEW_TABLE_FORMAT)
            .build();
        let inner = self
            .catalog
            .create_table(ident.namespace(), creation)
            .await?;
        Ok(Table::new(self.catalog.clone(), inner))
    }

    /// Like [`create_table`](Self::create_table) but partitioned by
    /// `partition_cols` with **identity** transforms (each must be a string
    /// column present in `schema`). This is what lets [`append`](crate::append)
    /// tag data files with a partition value and what enables partition pruning
    /// on filtered reads. Empty `partition_cols` is equivalent to
    /// [`create_table`](Self::create_table).
    pub async fn create_partitioned_table(
        &self,
        name: &str,
        schema: &ArrowSchema,
        partition_cols: &[&str],
    ) -> Result<Table> {
        let ident = self.table_ident(name)?;
        self.ensure_namespace(ident.namespace()).await?;
        let ice_schema = arrow_to_iceberg(schema)?;
        let creation = if partition_cols.is_empty() {
            TableCreation::builder()
                .name(ident.name().to_string())
                .schema(ice_schema)
                .format_version(NEW_TABLE_FORMAT)
                .build()
        } else {
            let mut b = PartitionSpec::builder(Arc::new(ice_schema.clone()));
            for c in partition_cols {
                b = b.add_partition_field(c, (*c).to_string(), Transform::Identity)?;
            }
            let spec = b.build()?.into_unbound();
            TableCreation::builder()
                .name(ident.name().to_string())
                .schema(ice_schema)
                .partition_spec(spec)
                .format_version(NEW_TABLE_FORMAT)
                .build()
        };
        let inner = self
            .catalog
            .create_table(ident.namespace(), creation)
            .await?;
        Ok(Table::new(self.catalog.clone(), inner))
    }

    /// Load the existing table `name`.
    pub async fn table(&self, name: &str) -> Result<Table> {
        let ident = self.table_ident(name)?;
        let inner = self.catalog.load_table(&ident).await?;
        Ok(Table::new(self.catalog.clone(), inner))
    }

    /// Load `name`, creating it (schema derived from `schema`) if absent.
    pub async fn table_or_create(&self, name: &str, schema: &ArrowSchema) -> Result<Table> {
        let ident = self.table_ident(name)?;
        if self.catalog.table_exists(&ident).await? {
            self.table(name).await
        } else {
            self.create_table(name, schema).await
        }
    }

    /// All table idents in the warehouse (every namespace).
    pub async fn table_idents(&self) -> Result<Vec<TableIdent>> {
        let mut out = Vec::new();
        for ns in self.catalog.list_namespaces(None).await? {
            out.extend(self.catalog.list_tables(&ns).await?);
        }
        Ok(out)
    }
}

#[cfg(feature = "sql")]
mod sql {
    use super::*;
    use arrow_array::RecordBatch;
    use datafusion::prelude::SessionContext;
    use iceberg_datafusion::{IcebergCatalogProvider, IcebergStaticTableProvider};

    impl Warehouse {
        /// A fresh DataFusion session over the warehouse's *current* state.
        ///
        /// Every table is reachable as `skade.<ns>.<table>`; tables in
        /// [`DEFAULT_NAMESPACE`] are additionally registered under their bare
        /// name, so `SELECT … FROM t JOIN u …` works unqualified. The session
        /// snapshots the catalog at creation — build a new one (or call
        /// [`Warehouse::sql`], which does) to see later commits.
        pub async fn session(&self) -> Result<SessionContext> {
            let ctx = SessionContext::new();
            let provider =
                IcebergCatalogProvider::try_new(self.catalog.clone() as Arc<dyn Catalog>).await?;
            ctx.register_catalog("skade", Arc::new(provider));

            // Register every table (all namespaces) under its bare name so
            // unqualified `SELECT … FROM <table>` resolves. Cross-namespace name
            // collisions are last-wins (rare in an embedded warehouse); use the
            // `skade.<ns>.<table>` catalog path to disambiguate.
            for ns in self.catalog.list_namespaces(None).await? {
                for ident in self.catalog.list_tables(&ns).await? {
                    let table = self.catalog.load_table(&ident).await?;
                    let provider = IcebergStaticTableProvider::try_new_from_table(table).await?;
                    ctx.register_table(ident.name(), Arc::new(provider))?;
                }
            }
            Ok(ctx)
        }

        /// Run one SQL statement over the warehouse (any number of tables) and
        /// collect the result batches. Convenience for
        /// `self.session().await?.sql(query).await?.collect()`.
        pub async fn sql(&self, query: &str) -> Result<Vec<RecordBatch>> {
            let ctx = self.session().await?;
            Ok(ctx.sql(query).await?.collect().await?)
        }
    }
}
