#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use ducklake_catalog::FileScheduledForDeletion;

use crate::{Result, StorageError};

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn renamed_table_path_for_target_schema(
    data_path: &str,
    source_schema: &ducklake_catalog::SchemaEntry,
    target_schema: &ducklake_catalog::SchemaEntry,
    table: &ducklake_catalog::TableEntry,
) -> (Option<String>, bool) {
    if source_schema.schema_id == target_schema.schema_id {
        return (table.path.clone(), table.path_is_relative);
    }

    let source_base = resolve_table_base_path(data_path, source_schema, table);
    let target_base = resolve_schema_base_path(data_path, target_schema);
    if table.path.is_none() && source_base == target_base {
        return (None, table.path_is_relative);
    }

    let (path, path_is_relative) = store_path_relative_to_parent(&source_base, &target_base);
    (Some(path), path_is_relative)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn resolve_table_base_path(
    data_path: &str,
    schema: &ducklake_catalog::SchemaEntry,
    table: &ducklake_catalog::TableEntry,
) -> String {
    let mut base = resolve_schema_base_path(data_path, schema);
    if let Some(table_path) = &table.path {
        base = resolve_path_component(&base, table_path, table.path_is_relative);
    }
    base
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn resolve_schema_base_path(data_path: &str, schema: &ducklake_catalog::SchemaEntry) -> String {
    let mut base = data_path.to_string();
    if let Some(schema_path) = &schema.path {
        base = resolve_path_component(&base, schema_path, schema.path_is_relative);
    }
    base
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn resolve_path_component(base: &str, component: &str, is_relative: bool) -> String {
    if !is_relative || path_is_absoluteish(component) {
        component.to_string()
    } else {
        join_path_components(base, component)
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn scheduled_cleanup_path_for_table_file(
    data_path: &str,
    schema_path: Option<&str>,
    schema_path_is_relative: bool,
    table_path: Option<&str>,
    table_path_is_relative: bool,
    file_path: &str,
    file_path_is_relative: bool,
) -> (String, bool) {
    if !file_path_is_relative || path_is_absoluteish(file_path) {
        return (file_path.to_string(), false);
    }

    let relative_to_data_path = schema_path
        .is_none_or(|path| schema_path_is_relative && !path_is_absoluteish(path))
        && table_path.is_none_or(|path| table_path_is_relative && !path_is_absoluteish(path));
    if relative_to_data_path {
        let mut path = String::new();
        if let Some(schema_path) = schema_path {
            path = join_path_components(&path, schema_path);
        }
        if let Some(table_path) = table_path {
            path = join_path_components(&path, table_path);
        }
        return (join_path_components(&path, file_path), true);
    }

    (
        resolve_catalog_file_path(
            data_path,
            schema_path,
            schema_path_is_relative,
            table_path,
            table_path_is_relative,
            file_path,
            file_path_is_relative,
        ),
        false,
    )
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn resolve_catalog_file_path(
    data_path: &str,
    schema_path: Option<&str>,
    schema_path_is_relative: bool,
    table_path: Option<&str>,
    table_path_is_relative: bool,
    file_path: &str,
    file_path_is_relative: bool,
) -> String {
    if !file_path_is_relative || path_is_absoluteish(file_path) {
        return file_path.to_string();
    }

    let mut base = data_path.to_string();
    if let Some(schema_path) = schema_path {
        base = resolve_path_component(&base, schema_path, schema_path_is_relative);
    }
    if let Some(table_path) = table_path {
        base = resolve_path_component(&base, table_path, table_path_is_relative);
    }
    join_path_components(&base, file_path)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn store_path_relative_to_parent(path: &str, parent: &str) -> (String, bool) {
    if let Some(relative) = path.strip_prefix(parent) {
        (relative.to_string(), true)
    } else {
        (path.to_string(), false)
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn path_is_absoluteish(path: &str) -> bool {
    path.starts_with('/') || path.contains("://")
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn join_path_components(base: &str, child: &str) -> String {
    if base.is_empty() {
        return child.to_string();
    }
    if child.is_empty() {
        return base.to_string();
    }
    match (base.ends_with('/'), child.starts_with('/')) {
        (true, true) => format!("{}{}", base, child.trim_start_matches('/')),
        (true, false) | (false, true) => format!("{base}{child}"),
        (false, false) => format!("{base}/{child}"),
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn resolve_cleanup_file_path(
    data_path: &str,
    file: &FileScheduledForDeletion,
) -> String {
    if !file.path_is_relative {
        return file.path.clone();
    }
    join_cleanup_path_components(data_path, &file.path)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn join_cleanup_path_components(base: &str, child: &str) -> String {
    join_path_components(base, child)
}

pub fn ducklake_store_data_path(raw: &str) -> String {
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return String::new();
    }
    let mut value = trimmed.trim_end_matches(&['/', '\\'][..]).to_string();
    value.push('/');
    value
}

pub const LOCAL_OBJECT_STORE_URL: &str = "file://";

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResolvedObjectStorePath {
    pub object_store_url: String,
    pub object_path: String,
    pub is_local: bool,
}

pub fn resolve_object_store_path(path: &str) -> Result<ResolvedObjectStorePath> {
    let Some(scheme_end) = path.find("://") else {
        return Ok(ResolvedObjectStorePath {
            object_store_url: LOCAL_OBJECT_STORE_URL.to_string(),
            object_path: path.to_string(),
            is_local: true,
        });
    };

    let scheme = &path[..scheme_end];
    let rest = &path[scheme_end + 3..];
    if scheme.eq_ignore_ascii_case("file") {
        if rest.starts_with('/') {
            return Ok(ResolvedObjectStorePath {
                object_store_url: LOCAL_OBJECT_STORE_URL.to_string(),
                object_path: rest.to_string(),
                is_local: true,
            });
        }
        return Err(StorageError::Unsupported(format!(
            "file URL {path} is not an absolute local path"
        )));
    }

    if rest.is_empty() {
        return Err(StorageError::Unsupported(format!(
            "object store URL {path} has no authority"
        )));
    }
    if rest.contains('?') || rest.contains('#') {
        return Err(StorageError::Unsupported(format!(
            "object store URL {path} includes a query or fragment"
        )));
    }

    let (authority, object_path) = match rest.find('/') {
        Some(path_start) => (&rest[..path_start], &rest[path_start + 1..]),
        None => (rest, ""),
    };
    if authority.is_empty() {
        return Err(StorageError::Unsupported(format!(
            "object store URL {path} has no authority"
        )));
    }

    Ok(ResolvedObjectStorePath {
        object_store_url: format!("{scheme}://{authority}"),
        object_path: object_path.to_string(),
        is_local: false,
    })
}

pub fn local_filesystem_path(path: &str) -> Result<String> {
    let resolved = resolve_object_store_path(path)?;
    if !resolved.is_local {
        return Err(StorageError::Unsupported(format!(
            "Materialized object-store file access is not supported for {path}"
        )));
    }
    Ok(resolved.object_path)
}

/// Returns true when `resolved` refers to a location contained within `root`
/// (typically the catalog `data_path`).
///
/// This is a purely logical check (no filesystem access, correct for object
/// stores): both paths must share the same object-store scheme and authority,
/// and after collapsing `.`/`..` segments the resolved path must equal `root` or
/// live under it at a segment boundary. It is fail-closed: unparseable paths, a
/// differing scheme/bucket, or `..` that escapes the store root all return false.
/// It backs the opt-in `ducklake.contain_paths` guard against untrusted catalogs
/// pointing file paths outside `data_path`.
pub fn path_is_within(resolved: &str, root: &str) -> bool {
    let (Ok(resolved), Ok(root)) = (
        resolve_object_store_path(resolved),
        resolve_object_store_path(root),
    ) else {
        return false;
    };
    if resolved.object_store_url != root.object_store_url {
        return false;
    }
    let (Some(resolved_path), Some(root_path)) = (
        normalize_object_path(&resolved.object_path),
        normalize_object_path(&root.object_path),
    ) else {
        return false;
    };
    if root_path.is_empty() || resolved_path == root_path {
        return true;
    }
    match resolved_path.strip_prefix(&root_path) {
        Some(rest) => rest.starts_with('/'),
        None => false,
    }
}

/// Collapse `.` and `..` segments in an object path. Returns `None` if `..`
/// escapes above the store root (more `..` than preceding segments).
fn normalize_object_path(path: &str) -> Option<String> {
    let is_absolute = path.starts_with('/');
    let mut segments: Vec<&str> = Vec::new();
    for segment in path.split('/') {
        match segment {
            "" | "." => continue,
            ".." => {
                segments.pop()?;
            }
            other => segments.push(other),
        }
    }
    let joined = segments.join("/");
    Some(if is_absolute {
        format!("/{joined}")
    } else {
        joined
    })
}

#[cfg(test)]
mod tests {
    use super::{normalize_object_path, path_is_within};

    #[test]
    fn path_within_allows_paths_under_root() {
        let root = "s3://bucket/lake/";
        assert!(path_is_within("s3://bucket/lake/data/part-1.parquet", root));
        assert!(path_is_within("s3://bucket/lake/", root)); // exact root
        assert!(path_is_within("s3://bucket/lake", root));
    }

    #[test]
    fn path_within_rejects_escapes() {
        let root = "s3://bucket/lake/";
        // Parent traversal that escapes the root prefix.
        assert!(!path_is_within(
            "s3://bucket/lake/../../other/secret.parquet",
            root
        ));
        // Different bucket / authority.
        assert!(!path_is_within("s3://other-bucket/lake/x.parquet", root));
        // Prefix spoofing must be rejected at the segment boundary.
        assert!(!path_is_within("s3://bucket/lake-evil/x.parquet", root));
    }

    #[test]
    fn path_within_rejects_scheme_and_absolute_changes() {
        let root = "s3://bucket/lake/";
        assert!(!path_is_within("/etc/passwd", root)); // local absolute vs remote root
        assert!(!path_is_within("file:///etc/passwd", root));

        let local_root = "/var/lib/ducklake/lake/";
        assert!(path_is_within(
            "/var/lib/ducklake/lake/data/x.parquet",
            local_root
        ));
        assert!(!path_is_within("/etc/passwd", local_root));
        assert!(!path_is_within(
            "/var/lib/ducklake/lake/../../etc/x",
            local_root
        ));
        assert!(!path_is_within("s3://bucket/x.parquet", local_root));
    }

    #[test]
    fn normalize_collapses_and_detects_escape() {
        assert_eq!(normalize_object_path("a/./b/../c").as_deref(), Some("a/c"));
        assert_eq!(normalize_object_path("/a//b/").as_deref(), Some("/a/b"));
        assert_eq!(normalize_object_path("a/../.."), None);
    }
}
