distributed 4.2.0

use serde::Deserialize;
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::fs::{self, File};
use std::io::{Read, Write};
use std::path::{Component, Path, PathBuf};

const INVENTORY_PATH: &str = "migrations/inventory.json";
const INVENTORY_VERSION: u32 = 1;
const MAX_INVENTORY_BYTES: usize = 1024 * 1024;
const MAX_SQL_BYTES: usize = 4 * 1024 * 1024;
const MAX_MIGRATIONS: usize = 256;
const MAX_JSON_DEPTH: usize = 24;
const MAX_TOTAL_ENTRIES: usize = MAX_MIGRATIONS * 4;
const MAX_TOP_LEVEL_ENTRIES: usize = 64;
const MAX_DIRECTORIES: usize = 4_096;
const MAX_SQL_FILES: usize = MAX_MIGRATIONS * 2;
const REDACTED_MIGRATION_PATH: &str = "<redacted-migration-path>";

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct Inventory {
    schema_version: u32,
    migrations: Vec<Migration>,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct Migration {
    version: u64,
    description: String,
    sqlite: MigrationFile,
    postgres: MigrationFile,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct MigrationFile {
    path: String,
    sha256: String,
}

#[derive(Clone, Copy)]
enum Dialect {
    Sqlite,
    Postgres,
}

impl Dialect {
    const ALL: [Self; 2] = [Self::Sqlite, Self::Postgres];

    const fn name(self) -> &'static str {
        match self {
            Self::Sqlite => "sqlite",
            Self::Postgres => "postgres",
        }
    }

    const fn directory(self) -> &'static str {
        match self {
            Self::Sqlite => "migrations/sqlite",
            Self::Postgres => "migrations/postgres",
        }
    }

    fn file(self, migration: &Migration) -> &MigrationFile {
        match self {
            Self::Sqlite => &migration.sqlite,
            Self::Postgres => &migration.postgres,
        }
    }
}

fn main() {
    emit_migration_inventory();

    // Only run gRPC codegen when the "grpc" feature is enabled.
    // Cargo sets CARGO_FEATURE_GRPC when compiling with --features grpc.
    if std::env::var("CARGO_FEATURE_GRPC").is_ok() {
        let service = tonic_build::manual::Service::builder()
            .name("CommandService")
            .package("sourced.microsvc")
            .method(
                tonic_build::manual::Method::builder()
                    .name("dispatch")
                    .route_name("Dispatch")
                    .input_type("crate::microsvc::grpc::GrpcRequest")
                    .output_type("crate::microsvc::grpc::GrpcResponse")
                    .codec_path("tonic_prost::ProstCodec")
                    .build(),
            )
            .method(
                tonic_build::manual::Method::builder()
                    .name("health")
                    .route_name("Health")
                    .input_type("crate::microsvc::grpc::HealthRequest")
                    .output_type("crate::microsvc::grpc::HealthResponse")
                    .codec_path("tonic_prost::ProstCodec")
                    .build(),
            )
            .build();

        tonic_build::manual::Builder::new().compile(&[service]);
    }
}

fn emit_migration_inventory() {
    println!("cargo:rerun-if-changed={INVENTORY_PATH}");
    for dialect in Dialect::ALL {
        // Watching the dialect roots also notices an unregistered SQL file
        // being added; the explicit file lines below keep declared inputs
        // visible in Cargo's build explanation.
        println!("cargo:rerun-if-changed={}", dialect.directory());
    }
    let manifest_dir = PathBuf::from(
        std::env::var_os("CARGO_MANIFEST_DIR")
            .expect("Cargo must provide CARGO_MANIFEST_DIR to the build script"),
    );
    let root = fs::canonicalize(&manifest_dir)
        .unwrap_or_else(|error| panic!("resolve repository root for migrations: {error}"));
    let inventory_path = root.join(INVENTORY_PATH);
    let bytes = read_bounded_file(
        &root,
        &inventory_path,
        MAX_INVENTORY_BYTES,
        "migration inventory",
    );
    validate_json_nesting(&bytes).unwrap_or_else(|error| panic!("parse {INVENTORY_PATH}: {error}"));
    let inventory: Inventory = serde_json::from_slice(&bytes)
        .unwrap_or_else(|error| panic!("parse {INVENTORY_PATH}: {error}"));
    validate_inventory(&root, &inventory);

    for migration in &inventory.migrations {
        for dialect in Dialect::ALL {
            println!("cargo:rerun-if-changed={}", dialect.file(migration).path);
        }
    }

    let out_dir = PathBuf::from(
        std::env::var_os("OUT_DIR").expect("Cargo must provide OUT_DIR to the build script"),
    );
    let generated_path = out_dir.join("migration_inventory.rs");
    let mut generated =
        String::from("// Generated by build.rs from migrations/inventory.json; do not edit.\n");
    emit_dialect(
        &mut generated,
        "SQLITE_MIGRATIONS",
        Dialect::Sqlite,
        &inventory,
    );
    emit_dialect(
        &mut generated,
        "POSTGRES_MIGRATIONS",
        Dialect::Postgres,
        &inventory,
    );
    let mut file = File::create(&generated_path)
        .unwrap_or_else(|error| panic!("create generated migration registration: {error}"));
    file.write_all(generated.as_bytes())
        .unwrap_or_else(|error| panic!("write generated migration registration: {error}"));
}

fn emit_dialect(generated: &mut String, name: &str, dialect: Dialect, inventory: &Inventory) {
    generated.push_str("#[cfg(feature = \"");
    generated.push_str(dialect.name());
    generated.push_str("\")]\n");
    generated.push_str("pub(crate) const ");
    generated.push_str(name);
    generated.push_str(": &[EmbeddedMigration] = &[\n");
    for migration in &inventory.migrations {
        let file = dialect.file(migration);
        generated.push_str("    EmbeddedMigration { version: ");
        generated.push_str(&migration.version.to_string());
        generated.push_str(", description: ");
        generated.push_str(&format!("{:?}", migration.description));
        generated.push_str(", sql: include_str!(concat!(env!(\"CARGO_MANIFEST_DIR\"), \"/\", ");
        generated.push_str(&format!("{:?}", file.path));
        generated.push_str(")), },\n");
    }
    generated.push_str("];\n");
}

fn validate_inventory(root: &Path, inventory: &Inventory) {
    if inventory.schema_version != INVENTORY_VERSION {
        panic!(
            "{INVENTORY_PATH} schema version {} is unsupported; expected {INVENTORY_VERSION}",
            inventory.schema_version
        );
    }
    if inventory.migrations.is_empty() || inventory.migrations.len() > MAX_MIGRATIONS {
        panic!("{INVENTORY_PATH} must contain 1..={MAX_MIGRATIONS} migrations");
    }
    let mut paths = BTreeMap::new();
    for (index, migration) in inventory.migrations.iter().enumerate() {
        let expected = (index + 1) as u64;
        if migration.version != expected {
            panic!(
                "{INVENTORY_PATH} versions must be consecutive: expected {expected}, observed {}",
                migration.version
            );
        }
        if migration.version > i64::MAX as u64
            || migration.description.is_empty()
            || migration.description.trim() != migration.description
            || migration.description.len() > 4 * 1024
            || migration.description.contains('\0')
            || is_secret_like(&migration.description)
        {
            panic!(
                "{INVENTORY_PATH} migration {} has an invalid description or version",
                migration.version
            );
        }
        for dialect in Dialect::ALL {
            let file = dialect.file(migration);
            let display_path = declared_path_display(&file.path);
            validate_path(root, dialect, file);
            if file.sha256.len() != 64
                || !file.sha256.bytes().all(|byte| byte.is_ascii_hexdigit())
                || file
                    .sha256
                    .chars()
                    .any(|character| character.is_ascii_uppercase())
            {
                panic!(
                    "{INVENTORY_PATH} {} migration `{}` has an invalid SHA-256",
                    dialect.name(),
                    display_path
                );
            }
            if paths
                .insert(file.path.clone(), (migration.version, dialect.name()))
                .is_some()
            {
                panic!(
                    "{INVENTORY_PATH} migration path `{}` is declared more than once",
                    display_path
                );
            }
            let sql_path = root.join(&file.path);
            let sql = read_bounded_file(root, &sql_path, MAX_SQL_BYTES, "migration SQL");
            if std::str::from_utf8(&sql).is_err() {
                panic!("migration SQL `{display_path}` is not UTF-8");
            }
            let observed = sha256_hex(&sql);
            if observed != file.sha256 {
                panic!(
                    "{INVENTORY_PATH} {} migration `{}` checksum mismatch: expected {}, observed {}",
                    dialect.name(),
                    display_path,
                    file.sha256,
                    observed
                );
            }
        }
    }
    for dialect in Dialect::ALL {
        let actual = collect_sql_files(root, dialect);
        for path in actual.keys() {
            if !paths.contains_key(path) {
                let display_path = declared_path_display(path);
                panic!(
                    "{INVENTORY_PATH} extra {} migration file `{display_path}` is not registered",
                    dialect.name()
                );
            }
        }
    }
    validate_dialect_directories(root);
}

fn validate_path(root: &Path, dialect: Dialect, file: &MigrationFile) {
    let path = Path::new(&file.path);
    let display_path = declared_path_display(&file.path);
    if file.path.is_empty()
        || file.path.trim() != file.path
        || file.path.len() > 4 * 1024
        || file.path.contains('\0')
        || file.path.contains('\\')
        || !file.path.ends_with(".sql")
        || path.is_absolute()
        || path
            .components()
            .any(|component| !matches!(component, Component::Normal(_)))
        || !path.starts_with(dialect.directory())
        || is_secret_like(&file.path)
    {
        panic!(
            "{INVENTORY_PATH} {} migration path `{}` is outside `{}`",
            dialect.name(),
            display_path,
            dialect.directory()
        );
    }
    let mut current = root.to_path_buf();
    for component in path.components() {
        let Component::Normal(component) = component else {
            unreachable!("validated migration path components");
        };
        current.push(component);
        let metadata = fs::symlink_metadata(&current)
            .unwrap_or_else(|error| panic!("inspect migration path `{display_path}`: {error}"));
        if metadata.file_type().is_symlink() {
            panic!("migration path `{display_path}` must not be a symlink");
        }
    }
}

fn declared_path_display(path: &str) -> String {
    let path_value = Path::new(path);
    if is_secret_like(path)
        || path_value.is_absolute()
        || path.contains('\\')
        || path_value
            .components()
            .any(|component| !matches!(component, Component::Normal(_)))
    {
        REDACTED_MIGRATION_PATH.to_string()
    } else {
        path.to_string()
    }
}

fn is_secret_like(value: &str) -> bool {
    let lower = value.to_ascii_lowercase();
    lower.contains("postgres://")
        || lower.contains("postgresql://")
        || lower.contains("mysql://")
        || lower.contains("mongodb://")
        || lower.contains("bearer ")
        || lower.contains("password=")
        || lower.contains("token=")
        || lower.contains("secret=")
        || lower.contains("-----begin ")
}

fn validate_json_nesting(input: &[u8]) -> Result<(), &'static str> {
    let mut depth = 0usize;
    let mut escaped = false;
    let mut in_string = false;

    for byte in input {
        if in_string {
            if escaped {
                escaped = false;
            } else if *byte == b'\\' {
                escaped = true;
            } else if *byte == b'"' {
                in_string = false;
            }
            continue;
        }

        match *byte {
            b'"' => in_string = true,
            b'{' | b'[' => {
                depth = depth.saturating_add(1);
                if depth > MAX_JSON_DEPTH {
                    return Err("migration inventory exceeds maximum JSON nesting depth");
                }
            }
            b'}' | b']' => depth = depth.saturating_sub(1),
            _ => {}
        }
    }

    Ok(())
}

fn relative_path_display(root: &Path, path: &Path) -> String {
    let relative = path
        .strip_prefix(root)
        .map(|relative| {
            relative
                .to_string_lossy()
                .replace(std::path::MAIN_SEPARATOR, "/")
        })
        .unwrap_or_else(|_| "<outside-repository>".to_string());
    declared_path_display(&relative)
}

fn read_bounded_file(root: &Path, path: &Path, limit: usize, label: &str) -> Vec<u8> {
    let relative = relative_path_display(root, path);
    let metadata = fs::symlink_metadata(path)
        .unwrap_or_else(|error| panic!("read {label} `{relative}`: {error}"));
    if metadata.file_type().is_symlink() {
        panic!("{label} `{relative}` must not be a symlink");
    }
    if !metadata.is_file() {
        panic!("{label} `{relative}` is not a regular file");
    }
    if metadata.len() > limit as u64 {
        panic!("{label} `{relative}` exceeds {limit} bytes");
    }
    let file =
        File::open(path).unwrap_or_else(|error| panic!("read {label} `{relative}`: {error}"));
    let opened_metadata = file
        .metadata()
        .unwrap_or_else(|error| panic!("inspect opened {label} `{relative}`: {error}"));
    if !opened_metadata.is_file() {
        panic!("opened {label} `{relative}` is not a regular file");
    }
    let opened_size = opened_metadata.len();
    if opened_size > limit as u64 {
        panic!("opened {label} `{relative}` exceeds {limit} bytes");
    }
    let mut bytes = Vec::with_capacity(opened_size as usize);
    file.take(limit as u64 + 1)
        .read_to_end(&mut bytes)
        .unwrap_or_else(|error| panic!("read {label} `{relative}`: {error}"));
    if bytes.len() > limit {
        panic!("{label} `{relative}` exceeds {limit} bytes");
    }
    bytes
}

fn collect_sql_files(root: &Path, dialect: Dialect) -> BTreeMap<String, ()> {
    let mut pending = vec![root.join(dialect.directory())];
    let mut files = BTreeMap::new();
    let mut directories = 0usize;
    let mut entries_seen = 0usize;
    while let Some(directory) = pending.pop() {
        directories += 1;
        if directories > MAX_DIRECTORIES {
            panic!(
                "{} migration directory tree exceeds {MAX_DIRECTORIES} directories",
                dialect.name()
            );
        }
        let directory_metadata = fs::symlink_metadata(&directory).unwrap_or_else(|error| {
            panic!(
                "inspect migration directory `{}`: {error}",
                relative_path_display(root, &directory)
            )
        });
        if directory_metadata.file_type().is_symlink() {
            panic!(
                "migration directory `{}` must not be a symlink",
                relative_path_display(root, &directory)
            );
        }
        if !directory_metadata.is_dir() {
            panic!(
                "migration directory `{}` is not a directory",
                relative_path_display(root, &directory)
            );
        }
        let mut read_entries = fs::read_dir(&directory).unwrap_or_else(|error| {
            panic!(
                "read migration directory `{}`: {error}",
                relative_path_display(root, &directory)
            )
        });
        let remaining_entries = MAX_TOTAL_ENTRIES - entries_seen;
        let mut entries = Vec::with_capacity(remaining_entries);
        loop {
            let Some(entry) = read_entries.next() else {
                break;
            };
            if entries_seen >= MAX_TOTAL_ENTRIES {
                panic!(
                    "migration directory tree exceeds {MAX_TOTAL_ENTRIES} entries for {}",
                    dialect.name()
                );
            }
            entries_seen += 1;
            entries.push(entry.unwrap_or_else(|error| {
                panic!(
                    "read migration directory entry for {}: {error}",
                    dialect.name()
                )
            }));
        }
        entries.sort_by_key(|entry| entry.file_name());
        for entry in entries {
            let path = entry.path();
            let metadata = fs::symlink_metadata(&path).unwrap_or_else(|error| {
                panic!(
                    "inspect migration path `{}`: {error}",
                    relative_path_display(root, &path)
                )
            });
            if metadata.file_type().is_symlink() {
                panic!(
                    "migration path `{}` must not be a symlink",
                    relative_path_display(root, &path)
                );
            }
            if metadata.is_dir() {
                pending.push(path);
                continue;
            }
            if !metadata.is_file() {
                panic!(
                    "migration path `{}` is not a regular file",
                    relative_path_display(root, &path)
                );
            }
            if path.extension().and_then(|extension| extension.to_str()) != Some("sql") {
                continue;
            }
            let relative = path
                .strip_prefix(root)
                .unwrap_or_else(|_| panic!("migration path escaped repository root"))
                .to_string_lossy()
                .replace(std::path::MAIN_SEPARATOR, "/");
            files.insert(relative, ());
            if files.len() > MAX_SQL_FILES {
                panic!(
                    "{} migration directory tree contains more than {MAX_SQL_FILES} SQL files",
                    dialect.name()
                );
            }
        }
    }
    files
}

fn validate_dialect_directories(root: &Path) {
    let migrations = root.join("migrations");
    let migrations_metadata = fs::symlink_metadata(&migrations)
        .unwrap_or_else(|error| panic!("inspect migrations directory: {error}"));
    if migrations_metadata.file_type().is_symlink() {
        panic!("migrations directory must not be a symlink");
    }
    if !migrations_metadata.is_dir() {
        panic!("migrations path is not a directory");
    }
    let mut read_entries = fs::read_dir(&migrations)
        .unwrap_or_else(|error| panic!("read migrations directory: {error}"));
    let mut entries_seen = 0usize;
    let mut entries = Vec::with_capacity(MAX_TOP_LEVEL_ENTRIES);
    loop {
        let Some(entry) = read_entries.next() else {
            break;
        };
        if entries_seen >= MAX_TOP_LEVEL_ENTRIES {
            panic!("migrations directory exceeds {MAX_TOP_LEVEL_ENTRIES} entries");
        }
        entries_seen += 1;
        entries.push(entry.unwrap_or_else(|error| panic!("read migrations entry: {error}")));
    }
    entries.sort_by_key(|entry| entry.file_name());
    for entry in entries {
        let path = entry.path();
        let metadata = fs::symlink_metadata(&path)
            .unwrap_or_else(|error| panic!("inspect migrations entry: {error}"));
        if metadata.file_type().is_symlink() {
            panic!(
                "migration path `{}` must not be a symlink",
                relative_path_display(root, &path)
            );
        }
        if !metadata.is_dir() {
            continue;
        }
        let name = path
            .file_name()
            .and_then(|value| value.to_str())
            .unwrap_or_default();
        if !matches!(name, "sqlite" | "postgres") {
            panic!(
                "unsupported migration dialect directory `{}`",
                relative_path_display(root, &path)
            );
        }
    }
}

fn sha256_hex(bytes: &[u8]) -> String {
    Sha256::digest(bytes)
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}