braze-sync 0.3.0

//! Catalog Schema file I/O.
//!
//! Layout (IMPLEMENTATION.md §9.1):
//!
//! ```text
//! <catalogs_root>/
//! ├── cardiology/
//! │   ├── schema.yaml
//! │   └── items.csv          (Phase B3 — not handled here)
//! └── dermatology/
//!     ├── schema.yaml
//!     └── items.csv
//! ```
//!
//! v0.1.0 implements schema only. Items I/O is not yet handled here.

use crate::error::{Error, Result};
use crate::fs::{validate_resource_name, write_atomic};
use crate::resource::Catalog;
use std::path::Path;

/// Header prepended to every `schema.yaml` written by braze-sync. Editable
/// — the comment is just a hint to humans browsing the file tree.
const SCHEMA_HEADER: &str =
    "# Generated by braze-sync. Edit and run `braze-sync apply` to sync to Braze.\n";

const SCHEMA_FILE_NAME: &str = "schema.yaml";

/// Load every `schema.yaml` under `catalogs_root` into a `Vec<Catalog>`,
/// sorted by catalog name for deterministic output.
///
/// Behaviour:
/// - Missing root → `Ok(vec![])`. A fresh project with no catalogs yet is a
///   valid state.
/// - Root exists but is a file → `Err(InvalidFormat)`.
/// - Subdirectory without `schema.yaml` → silently skipped (e.g. a
///   future-Phase-B layout where items exist but schema is being added).
/// - `schema.yaml` whose `name:` field doesn't match its directory name →
///   `Err(InvalidFormat)`. Stopping here makes diff / apply correctness
///   trivially debuggable.
pub fn load_all_schemas(catalogs_root: &Path) -> Result<Vec<Catalog>> {
    let read_dir = match std::fs::read_dir(catalogs_root) {
        Ok(rd) => rd,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(e) => {
            // read_dir on a file (not a dir) returns an OS-specific error;
            // surface it as InvalidFormat for a clearer message.
            if catalogs_root.is_file() {
                return Err(Error::InvalidFormat {
                    path: catalogs_root.to_path_buf(),
                    message: "expected a directory for the catalogs root".into(),
                });
            }
            return Err(e.into());
        }
    };

    let mut schemas = Vec::new();
    for entry in read_dir {
        let entry = entry?;
        if !entry.file_type()?.is_dir() {
            tracing::debug!(path = %entry.path().display(), "skipping non-directory entry");
            continue;
        }
        let dir = entry.path();
        let schema_path = dir.join(SCHEMA_FILE_NAME);
        if !schema_path.is_file() {
            continue;
        }
        let cat = read_schema_file(&schema_path)?;

        let dir_name = entry.file_name().to_string_lossy().into_owned();
        if cat.name != dir_name {
            return Err(Error::InvalidFormat {
                path: schema_path,
                message: format!(
                    "schema name '{}' does not match its catalog directory '{}'",
                    cat.name, dir_name
                ),
            });
        }
        schemas.push(cat);
    }

    schemas.sort_by(|a, b| a.name.cmp(&b.name));
    Ok(schemas)
}

/// Read a single `schema.yaml` by absolute path. Does not validate that
/// the parent directory name matches `cat.name` — the caller (typically
/// [`load_all_schemas`]) is responsible for that.
///
/// Resource files are forward-compat permissive: unknown fields are
/// ignored, in line with IMPLEMENTATION.md §2.5.
pub fn read_schema_file(path: &Path) -> Result<Catalog> {
    let bytes = std::fs::read_to_string(path)?;
    let cat: Catalog = serde_norway::from_str(&bytes).map_err(|source| Error::YamlParse {
        path: path.to_path_buf(),
        source,
    })?;
    Ok(cat)
}

/// Write `catalog` to `<catalogs_root>/<catalog.name>/schema.yaml`,
/// creating directories as needed. Field ordering is normalized (sorted
/// alphabetically) for deterministic output and diff stability.
///
/// Rejects catalog names that contain path separators or `..` to prevent
/// path traversal — defense in depth on top of the validate command's
/// naming-pattern check (§7.6 / §10).
pub fn save_schema(catalogs_root: &Path, catalog: &Catalog) -> Result<()> {
    validate_resource_name("catalog", &catalog.name)?;

    let dir = catalogs_root.join(&catalog.name);
    let path = dir.join(SCHEMA_FILE_NAME);

    let normalized = catalog.normalized();
    let yaml = serde_norway::to_string(&normalized).map_err(|e| Error::InvalidFormat {
        path: path.clone(),
        message: format!("yaml serialization failed: {e}"),
    })?;

    let mut content = String::with_capacity(SCHEMA_HEADER.len() + yaml.len());
    content.push_str(SCHEMA_HEADER);
    content.push_str(&yaml);

    write_atomic(&path, content.as_bytes())?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::resource::{CatalogField, CatalogFieldType};

    fn field(name: &str, t: CatalogFieldType) -> CatalogField {
        CatalogField {
            name: name.into(),
            field_type: t,
        }
    }

    fn cat(name: &str, fields: Vec<CatalogField>) -> Catalog {
        Catalog {
            name: name.into(),
            description: Some(format!("{name} catalog")),
            fields,
        }
    }

    #[test]
    fn round_trip_single_catalog() {
        let dir = tempfile::tempdir().unwrap();
        let c = cat(
            "cardiology",
            vec![
                field("condition_id", CatalogFieldType::String),
                field("display_order", CatalogFieldType::Number),
                field("is_active", CatalogFieldType::Boolean),
            ],
        );
        save_schema(dir.path(), &c).unwrap();
        let loaded = load_all_schemas(dir.path()).unwrap();
        assert_eq!(loaded.len(), 1);
        assert_eq!(loaded[0], c.normalized());
    }

    #[test]
    fn round_trip_all_field_types() {
        let dir = tempfile::tempdir().unwrap();
        let c = cat(
            "everything",
            vec![
                field("a_string", CatalogFieldType::String),
                field("b_number", CatalogFieldType::Number),
                field("c_boolean", CatalogFieldType::Boolean),
                field("d_time", CatalogFieldType::Time),
                field("e_object", CatalogFieldType::Object),
                field("f_array", CatalogFieldType::Array),
            ],
        );
        save_schema(dir.path(), &c).unwrap();
        let loaded = load_all_schemas(dir.path()).unwrap();
        assert_eq!(loaded[0], c);
    }

    #[test]
    fn save_sorts_fields_alphabetically() {
        let dir = tempfile::tempdir().unwrap();
        let c = Catalog {
            name: "x".into(),
            description: None,
            fields: vec![
                field("z", CatalogFieldType::String),
                field("a", CatalogFieldType::String),
                field("m", CatalogFieldType::String),
            ],
        };
        save_schema(dir.path(), &c).unwrap();
        let loaded = load_all_schemas(dir.path()).unwrap();
        assert_eq!(loaded[0].fields[0].name, "a");
        assert_eq!(loaded[0].fields[1].name, "m");
        assert_eq!(loaded[0].fields[2].name, "z");
    }

    #[test]
    fn load_multiple_catalogs_sorted_alphabetically() {
        let dir = tempfile::tempdir().unwrap();
        save_schema(
            dir.path(),
            &cat("zebra", vec![field("id", CatalogFieldType::String)]),
        )
        .unwrap();
        save_schema(
            dir.path(),
            &cat("apple", vec![field("id", CatalogFieldType::String)]),
        )
        .unwrap();
        save_schema(
            dir.path(),
            &cat("mango", vec![field("id", CatalogFieldType::String)]),
        )
        .unwrap();
        let loaded = load_all_schemas(dir.path()).unwrap();
        assert_eq!(
            loaded.iter().map(|c| c.name.as_str()).collect::<Vec<_>>(),
            vec!["apple", "mango", "zebra"]
        );
    }

    #[test]
    fn missing_catalogs_root_returns_empty() {
        let dir = tempfile::tempdir().unwrap();
        let nonexistent = dir.path().join("not_here");
        let loaded = load_all_schemas(&nonexistent).unwrap();
        assert!(loaded.is_empty());
    }

    #[test]
    fn empty_catalogs_root_returns_empty() {
        let dir = tempfile::tempdir().unwrap();
        let loaded = load_all_schemas(dir.path()).unwrap();
        assert!(loaded.is_empty());
    }

    #[test]
    fn catalogs_root_pointing_at_a_file_is_rejected() {
        let dir = tempfile::tempdir().unwrap();
        let file_path = dir.path().join("not_a_dir");
        std::fs::write(&file_path, "x").unwrap();
        let err = load_all_schemas(&file_path).unwrap_err();
        assert!(matches!(err, Error::InvalidFormat { .. }), "got: {err:?}");
    }

    #[test]
    fn dir_without_schema_yaml_is_silently_skipped() {
        let dir = tempfile::tempdir().unwrap();
        // dir for a catalog with only items.csv (Phase B layout)
        let lone = dir.path().join("lonely");
        std::fs::create_dir_all(&lone).unwrap();
        std::fs::write(lone.join("items.csv"), "id\n").unwrap();
        // a real catalog
        save_schema(
            dir.path(),
            &cat("real", vec![field("id", CatalogFieldType::String)]),
        )
        .unwrap();

        let loaded = load_all_schemas(dir.path()).unwrap();
        assert_eq!(loaded.len(), 1);
        assert_eq!(loaded[0].name, "real");
    }

    #[test]
    fn schema_name_mismatch_with_dir_name_is_an_error() {
        let dir = tempfile::tempdir().unwrap();
        let cat_dir = dir.path().join("on_disk_name");
        std::fs::create_dir_all(&cat_dir).unwrap();
        std::fs::write(
            cat_dir.join("schema.yaml"),
            "name: in_yaml_name\nfields: []\n",
        )
        .unwrap();

        let err = load_all_schemas(dir.path()).unwrap_err();
        match err {
            Error::InvalidFormat { message, .. } => {
                assert!(message.contains("on_disk_name"));
                assert!(message.contains("in_yaml_name"));
            }
            other => panic!("expected InvalidFormat, got {other:?}"),
        }
    }

    #[test]
    fn unknown_field_in_schema_yaml_is_ignored_for_forward_compat() {
        let dir = tempfile::tempdir().unwrap();
        let cat_dir = dir.path().join("future");
        std::fs::create_dir_all(&cat_dir).unwrap();
        // future_v1_3_field is something a v1.3 binary might emit. v1.0
        // should silently accept it per IMPLEMENTATION.md §2.5.
        let yaml = "\
name: future
description: hi
future_v1_3_field: surprise
fields:
  - name: id
    type: string
";
        std::fs::write(cat_dir.join("schema.yaml"), yaml).unwrap();
        let loaded = load_all_schemas(dir.path()).unwrap();
        assert_eq!(loaded.len(), 1);
        assert_eq!(loaded[0].name, "future");
        assert_eq!(loaded[0].fields.len(), 1);
    }

    #[test]
    fn schema_yaml_with_header_comment_parses() {
        let dir = tempfile::tempdir().unwrap();
        let cat_dir = dir.path().join("commented");
        std::fs::create_dir_all(&cat_dir).unwrap();
        let yaml = "\
# Generated by braze-sync. Edit and run `braze-sync apply` to sync to Braze.
name: commented
fields:
  - name: id
    type: string
";
        std::fs::write(cat_dir.join("schema.yaml"), yaml).unwrap();
        let loaded = load_all_schemas(dir.path()).unwrap();
        assert_eq!(loaded.len(), 1);
        assert_eq!(loaded[0].name, "commented");
    }

    #[test]
    fn save_writes_header_comment() {
        let dir = tempfile::tempdir().unwrap();
        let c = cat("hdr", vec![field("id", CatalogFieldType::String)]);
        save_schema(dir.path(), &c).unwrap();
        let raw = std::fs::read_to_string(dir.path().join("hdr/schema.yaml")).unwrap();
        assert!(
            raw.starts_with("# Generated by braze-sync."),
            "missing header in: {raw}"
        );
    }

    #[test]
    fn save_creates_nested_directories() {
        let dir = tempfile::tempdir().unwrap();
        let nested = dir.path().join("braze").join("catalogs");
        // nested doesn't exist yet
        save_schema(
            &nested,
            &cat("derm", vec![field("id", CatalogFieldType::String)]),
        )
        .unwrap();
        assert!(nested.join("derm").join("schema.yaml").is_file());
    }

    #[test]
    fn save_rejects_path_traversal_in_name() {
        let dir = tempfile::tempdir().unwrap();
        for bad in ["../evil", "..", ".", "", "a/b", "a\\b"] {
            let c = Catalog {
                name: bad.into(),
                description: None,
                fields: vec![],
            };
            let err = save_schema(dir.path(), &c).unwrap_err();
            assert!(
                matches!(err, Error::InvalidFormat { .. }),
                "name {bad:?} should be rejected; got {err:?}"
            );
        }
    }

    #[test]
    fn atomic_save_leaves_no_temp_files() {
        let dir = tempfile::tempdir().unwrap();
        let c = cat("clean", vec![field("id", CatalogFieldType::String)]);
        save_schema(dir.path(), &c).unwrap();
        let cat_dir = dir.path().join("clean");
        let entries: Vec<_> = std::fs::read_dir(&cat_dir)
            .unwrap()
            .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
            .collect();
        assert_eq!(entries, vec!["schema.yaml".to_string()]);
    }

    #[test]
    fn save_overwrites_existing_schema() {
        let dir = tempfile::tempdir().unwrap();
        let v1 = cat("ovr", vec![field("a", CatalogFieldType::String)]);
        save_schema(dir.path(), &v1).unwrap();
        let v2 = cat(
            "ovr",
            vec![
                field("a", CatalogFieldType::String),
                field("b", CatalogFieldType::Number),
            ],
        );
        save_schema(dir.path(), &v2).unwrap();
        let loaded = load_all_schemas(dir.path()).unwrap();
        assert_eq!(loaded.len(), 1);
        assert_eq!(loaded[0].fields.len(), 2);
    }
}