//! Schema Registry Service (IDL-15)
//!
//! Provides REST endpoints for schema versioning, compatibility checking,
//! and runtime schema negotiation between peers.
//!
//! ## Endpoints
//!
//! - `POST /api/v1/schemas`       — Register a new schema version
//! - `GET  /api/v1/schemas`       — List all registered schemas
//! - `GET  /api/v1/schemas/:id`   — Get a specific schema by ID
//! - `GET  /api/v1/schemas/:id/versions` — List all versions of a schema
//! - `POST /api/v1/schemas/:id/check` — Check compatibility of a new version
//! - `DELETE /api/v1/schemas/:id` — Delete a schema (soft delete)

use crate::service::{ServiceRequest, ServiceResponse};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

// ─── Types ───────────────────────────────────────────────────────────────────

/// Compatibility mode for schema evolution.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum CompatibilityMode {
    /// No compatibility checks (allow any change).
    None,
    /// New schema can read data written by old schema.
    Backward,
    /// Old schema can read data written by new schema.
    Forward,
    /// Both backward and forward compatible.
    Full,
}

impl Default for CompatibilityMode {
    fn default() -> Self {
        CompatibilityMode::Backward
    }
}

/// A registered schema subject (e.g., "chat.v1.SendRequest").
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaSubject {
    /// Unique identifier (auto-generated UUID)
    pub id: String,
    /// Human-readable subject name (e.g., "chat.v1.SendRequest")
    pub subject: String,
    /// Compatibility mode for this subject
    pub compatibility: CompatibilityMode,
    /// Ordered list of schema versions (latest last)
    pub versions: Vec<SchemaVersion>,
    /// Whether this schema has been soft-deleted
    pub deleted: bool,
    /// ISO 8601 creation timestamp
    pub created_at: String,
    /// ISO 8601 last-updated timestamp
    pub updated_at: String,
}

/// A single version of a schema.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaVersion {
    /// Auto-incrementing version number (1-based)
    pub version: u32,
    /// The PSL schema source text
    pub schema: String,
    /// SHA-256 fingerprint of the schema text
    pub fingerprint: String,
    /// ISO 8601 registration timestamp
    pub registered_at: String,
}

/// Shared state for the schema registry.
pub type SchemaStore = Arc<RwLock<SchemaRegistry>>;

#[derive(Debug, Clone, Default)]
pub struct SchemaRegistry {
    pub subjects: HashMap<String, SchemaSubject>,
}

impl SchemaRegistry {
    pub fn new() -> Self {
        Self {
            subjects: HashMap::new(),
        }
    }
}

// ─── Request / Response DTOs ─────────────────────────────────────────────────

#[derive(Debug, Deserialize)]
pub struct RegisterSchemaRequest {
    /// Subject name (e.g., "chat.v1.SendRequest")
    pub subject: String,
    /// PSL schema source text
    pub schema: String,
    /// Optional compatibility mode override
    pub compatibility: Option<CompatibilityMode>,
}

#[derive(Debug, Serialize)]
pub struct RegisterSchemaResponse {
    pub id: String,
    pub subject: String,
    pub version: u32,
    pub fingerprint: String,
    pub compatible: bool,
}

#[derive(Debug, Deserialize)]
pub struct ListSchemasQuery {
    /// Filter by subject prefix
    pub prefix: Option<String>,
    /// Include soft-deleted schemas
    pub include_deleted: Option<bool>,
}

#[derive(Debug, Serialize)]
pub struct SchemaListResponse {
    pub schemas: Vec<SchemaSubjectSummary>,
    pub total: usize,
}

#[derive(Debug, Serialize)]
pub struct SchemaSubjectSummary {
    pub id: String,
    pub subject: String,
    pub latest_version: u32,
    pub compatibility: CompatibilityMode,
    pub version_count: usize,
    pub created_at: String,
    pub updated_at: String,
}

#[derive(Debug, Deserialize)]
pub struct CompatibilityCheckRequest {
    /// New schema to check against existing versions
    pub schema: String,
}

#[derive(Debug, Serialize)]
pub struct CompatibilityCheckResponse {
    pub compatible: bool,
    pub mode: CompatibilityMode,
    pub errors: Vec<String>,
    pub warnings: Vec<String>,
}

// ─── Handlers ────────────────────────────────────────────────────────────────

// ─── Handlers ────────────────────────────────────────────────────────────────

pub async fn handle_schemas_router(req: &ServiceRequest, registry: &SchemaStore) -> ServiceResponse {
    match req.method.as_str() {
        "register" => {
            if let Ok(reg_req) = serde_json::from_value::<RegisterSchemaRequest>(req.params.clone()) {
                register_schema(req.id.clone(), reg_req, registry).await
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("invalid params".into()),
                }
            }
        }
        "list" => {
            if let Ok(query) = serde_json::from_value::<ListSchemasQuery>(req.params.clone()) {
                list_schemas(req.id.clone(), query, registry).await
            } else {
                list_schemas(req.id.clone(), ListSchemasQuery { prefix: None, include_deleted: None }, registry).await
            }
        }
        "get" => {
            if let Some(subject) = req.params.get("subject").and_then(|s| s.as_str()) {
                get_schema(req.id.clone(), subject.to_string(), registry).await
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("missing subject".into()),
                }
            }
        }
        "versions" => {
            if let Some(subject) = req.params.get("subject").and_then(|s| s.as_str()) {
                get_schema_versions(req.id.clone(), subject.to_string(), registry).await
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("missing subject".into()),
                }
            }
        }
        "check" => {
            if let Some(subject) = req.params.get("subject").and_then(|s| s.as_str()) {
                if let Ok(check_req) = serde_json::from_value::<CompatibilityCheckRequest>(req.params.clone()) {
                    check_compatibility(req.id.clone(), subject.to_string(), check_req, registry).await
                } else {
                    ServiceResponse {
                        id: req.id.clone(),
                        status: 400,
                        data: serde_json::Value::Null,
                        error: Some("invalid params".into()),
                    }
                }
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("missing subject".into()),
                }
            }
        }
        "delete" => {
            if let Some(subject) = req.params.get("subject").and_then(|s| s.as_str()) {
                delete_schema(req.id.clone(), subject.to_string(), registry).await
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("missing subject".into()),
                }
            }
        }
        _ => ServiceResponse {
            id: req.id.clone(),
            status: 400,
            data: serde_json::Value::Null,
            error: Some(format!("unknown schemas method: {}", req.method)),
        },
    }
}

async fn register_schema(
    req_id: String,
    req: RegisterSchemaRequest,
    store: &SchemaStore,
) -> ServiceResponse {
    let mut registry = store.write().await;
    let now = chrono::Utc::now().to_rfc3339();
    let fingerprint = sha256_hex(&req.schema);

    if let Some(existing) = registry.subjects.get_mut(&req.subject) {
        if let Some(v) = existing.versions.iter().find(|v| v.fingerprint == fingerprint) {
            return ServiceResponse {
                id: req_id,
                status: 200,
                data: serde_json::to_value(RegisterSchemaResponse {
                    id: existing.id.clone(),
                    subject: existing.subject.clone(),
                    version: v.version,
                    fingerprint,
                    compatible: true,
                }).unwrap(),
                error: None,
            };
        }

        let compat_mode = req.compatibility.unwrap_or(existing.compatibility);
        let latest = existing.versions.last().map(|v| &v.schema);
        let compat_result = if let Some(prev_schema) = latest {
            check_schema_compatibility(prev_schema, &req.schema, compat_mode)
        } else {
            CompatibilityResult { compatible: true, errors: vec![], warnings: vec![] }
        };

        if !compat_result.compatible {
            return ServiceResponse {
                id: req_id,
                status: 409,
                data: serde_json::to_value(RegisterSchemaResponse {
                    id: existing.id.clone(),
                    subject: existing.subject.clone(),
                    version: 0,
                    fingerprint,
                    compatible: false,
                }).unwrap(),
                error: Some(compat_result.errors.join(", ")),
            };
        }

        let new_version = existing.versions.last().map(|v| v.version).unwrap_or(0) + 1;
        existing.versions.push(SchemaVersion {
            version: new_version,
            schema: req.schema,
            fingerprint: fingerprint.clone(),
            registered_at: now.clone(),
        });
        existing.updated_at = now;
        if let Some(mode) = req.compatibility {
            existing.compatibility = mode;
        }

        ServiceResponse {
            id: req_id,
            status: 200,
            data: serde_json::to_value(RegisterSchemaResponse {
                id: existing.id.clone(),
                subject: existing.subject.clone(),
                version: new_version,
                fingerprint,
                compatible: true,
            }).unwrap(),
            error: None,
        }
    } else {
        let id = uuid::Uuid::new_v4().to_string();
        let mode = req.compatibility.unwrap_or_default();

        let subject = SchemaSubject {
            id: id.clone(),
            subject: req.subject.clone(),
            compatibility: mode,
            versions: vec![SchemaVersion {
                version: 1,
                schema: req.schema,
                fingerprint: fingerprint.clone(),
                registered_at: now.clone(),
            }],
            deleted: false,
            created_at: now.clone(),
            updated_at: now,
        };

        registry.subjects.insert(req.subject.clone(), subject);

        ServiceResponse {
            id: req_id,
            status: 201,
            data: serde_json::to_value(RegisterSchemaResponse {
                id,
                subject: req.subject,
                version: 1,
                fingerprint,
                compatible: true,
            }).unwrap(),
            error: None,
        }
    }
}

async fn list_schemas(
    req_id: String,
    query: ListSchemasQuery,
    store: &SchemaStore,
) -> ServiceResponse {
    let registry = store.read().await;
    let include_deleted = query.include_deleted.unwrap_or(false);

    let mut schemas: Vec<SchemaSubjectSummary> = registry
        .subjects
        .values()
        .filter(|s| include_deleted || !s.deleted)
        .filter(|s| if let Some(ref prefix) = query.prefix { s.subject.starts_with(prefix) } else { true })
        .map(|s| SchemaSubjectSummary {
            id: s.id.clone(),
            subject: s.subject.clone(),
            latest_version: s.versions.last().map(|v| v.version).unwrap_or(0),
            compatibility: s.compatibility,
            version_count: s.versions.len(),
            created_at: s.created_at.clone(),
            updated_at: s.updated_at.clone(),
        })
        .collect();

    schemas.sort_by(|a, b| a.subject.cmp(&b.subject));
    let total = schemas.len();

    ServiceResponse {
        id: req_id,
        status: 200,
        data: serde_json::to_value(SchemaListResponse { schemas, total }).unwrap(),
        error: None,
    }
}

async fn get_schema(req_id: String, subject: String, store: &SchemaStore) -> ServiceResponse {
    let registry = store.read().await;
    match registry.subjects.get(&subject) {
        Some(s) if !s.deleted => ServiceResponse {
            id: req_id,
            status: 200,
            data: serde_json::to_value(s).unwrap(),
            error: None,
        },
        _ => ServiceResponse {
            id: req_id,
            status: 404,
            data: serde_json::Value::Null,
            error: Some(format!("Schema '{}' not found", subject)),
        },
    }
}

async fn get_schema_versions(req_id: String, subject: String, store: &SchemaStore) -> ServiceResponse {
    let registry = store.read().await;
    match registry.subjects.get(&subject) {
        Some(s) if !s.deleted => ServiceResponse {
            id: req_id,
            status: 200,
            data: serde_json::json!({ "subject": s.subject, "compatibility": s.compatibility, "versions": s.versions }),
            error: None,
        },
        _ => ServiceResponse {
            id: req_id,
            status: 404,
            data: serde_json::Value::Null,
            error: Some(format!("Schema '{}' not found", subject)),
        },
    }
}

async fn check_compatibility(
    req_id: String,
    subject: String,
    req: CompatibilityCheckRequest,
    store: &SchemaStore,
) -> ServiceResponse {
    let registry = store.read().await;
    match registry.subjects.get(&subject) {
        Some(s) if !s.deleted => {
            let latest = s.versions.last().map(|v| &v.schema);
            let result = if let Some(prev) = latest {
                check_schema_compatibility(prev, &req.schema, s.compatibility)
            } else {
                CompatibilityResult { compatible: true, errors: vec![], warnings: vec![] }
            };
            ServiceResponse {
                id: req_id,
                status: 200,
                data: serde_json::to_value(CompatibilityCheckResponse {
                    compatible: result.compatible,
                    mode: s.compatibility,
                    errors: result.errors,
                    warnings: result.warnings,
                }).unwrap(),
                error: None,
            }
        }
        _ => ServiceResponse {
            id: req_id,
            status: 404,
            data: serde_json::Value::Null,
            error: Some(format!("Schema '{}' not found", subject)),
        },
    }
}

async fn delete_schema(req_id: String, subject: String, store: &SchemaStore) -> ServiceResponse {
    let mut registry = store.write().await;
    match registry.subjects.get_mut(&subject) {
        Some(s) if !s.deleted => {
            s.deleted = true;
            s.updated_at = chrono::Utc::now().to_rfc3339();
            ServiceResponse {
                id: req_id,
                status: 200,
                data: serde_json::json!({ "deleted": true, "subject": subject }),
                error: None,
            }
        }
        _ => ServiceResponse {
            id: req_id,
            status: 404,
            data: serde_json::Value::Null,
            error: Some(format!("Schema '{}' not found", subject)),
        },
    }
}

// ─── Compatibility Checking ──────────────────────────────────────────────────

struct CompatibilityResult {
    compatible: bool,
    errors: Vec<String>,
    warnings: Vec<String>,
}

/// Check compatibility between two PSL schema strings.
///
/// Rules by mode:
/// - **BACKWARD**: New schema can read old data → fields added must be optional,
///   fields removed are breaking.
/// - **FORWARD**: Old schema can read new data → fields removed must not be
///   required in old, fields added are OK.
/// - **FULL**: Both backward and forward.
/// - **NONE**: No checks, always compatible.
fn check_schema_compatibility(
    old_schema: &str,
    new_schema: &str,
    mode: CompatibilityMode,
) -> CompatibilityResult {
    if mode == CompatibilityMode::None {
        return CompatibilityResult {
            compatible: true,
            errors: vec![],
            warnings: vec![],
        };
    }

    let old_parsed = match pulse_idl::parser::parse(old_schema) {
        Ok(s) => s,
        Err(_) => {
            return CompatibilityResult {
                compatible: true,
                errors: vec![],
                warnings: vec!["Previous schema version failed to parse; skipping compatibility check".into()],
            };
        }
    };

    let new_parsed = match pulse_idl::parser::parse(new_schema) {
        Ok(s) => s,
        Err(e) => {
            return CompatibilityResult {
                compatible: false,
                errors: vec![format!("New schema parse error: {}", e)],
                warnings: vec![],
            };
        }
    };

    let mut errors = Vec::new();
    let mut warnings = Vec::new();

    // Extract messages from both schemas
    let old_messages = extract_messages(&old_parsed);
    let new_messages = extract_messages(&new_parsed);

    // Check backward compatibility (new reads old)
    let check_backward = mode == CompatibilityMode::Backward || mode == CompatibilityMode::Full;
    // Check forward compatibility (old reads new)
    let check_forward = mode == CompatibilityMode::Forward || mode == CompatibilityMode::Full;

    if check_backward {
        // Removing a required field is backward-incompatible
        for (name, old_fields) in &old_messages {
            if let Some(new_fields) = new_messages.get(name) {
                for (field_name, field_info) in old_fields {
                    if !new_fields.contains_key(field_name) && !field_info.optional {
                        errors.push(format!(
                            "BACKWARD: Required field '{}.{}' was removed",
                            name, field_name
                        ));
                    }
                }
            } else {
                errors.push(format!("BACKWARD: Message '{}' was removed", name));
            }
        }

        // Added required fields are backward-incompatible (old data won't have them)
        for (name, new_fields) in &new_messages {
            if let Some(old_fields) = old_messages.get(name) {
                for (field_name, field_info) in new_fields {
                    if !old_fields.contains_key(field_name) && !field_info.optional {
                        errors.push(format!(
                            "BACKWARD: New required field '{}.{}' added (must be optional)",
                            name, field_name
                        ));
                    }
                    if !old_fields.contains_key(field_name) && field_info.optional {
                        warnings.push(format!(
                            "BACKWARD: New optional field '{}.{}' added (safe)",
                            name, field_name
                        ));
                    }
                }
            }
            // New messages are always backward-compatible
        }
    }

    if check_forward {
        // New required fields are forward-incompatible (old code won't know about them)
        for (name, new_fields) in &new_messages {
            if let Some(old_fields) = old_messages.get(name) {
                for (field_name, field_info) in new_fields {
                    if !old_fields.contains_key(field_name) && !field_info.optional {
                        errors.push(format!(
                            "FORWARD: New required field '{}.{}' added (old code can't handle it)",
                            name, field_name
                        ));
                    }
                }
            }
        }

        // Removing optional fields is forward-compatible; removing required is not
        for (name, old_fields) in &old_messages {
            if let Some(new_fields) = new_messages.get(name) {
                for (field_name, field_info) in old_fields {
                    if !new_fields.contains_key(field_name) && !field_info.optional {
                        errors.push(format!(
                            "FORWARD: Required field '{}.{}' was removed",
                            name, field_name
                        ));
                    }
                }
            } else {
                errors.push(format!("FORWARD: Message '{}' was removed", name));
            }
        }
    }

    CompatibilityResult {
        compatible: errors.is_empty(),
        errors,
        warnings,
    }
}

struct FieldInfo {
    optional: bool,
}

fn extract_messages(
    schema: &pulse_idl::ast::Schema,
) -> HashMap<String, HashMap<String, FieldInfo>> {
    let mut result = HashMap::new();
    for item in &schema.items {
        if let pulse_idl::ast::Item::Message(msg) = item {
            let mut fields = HashMap::new();
            for field in &msg.fields {
                fields.insert(
                    field.name.clone(),
                    FieldInfo {
                        optional: field.optional,
                    },
                );
            }
            result.insert(msg.name.clone(), fields);
        }
    }
    result
}

fn sha256_hex(data: &str) -> String {
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};
    // Simple fingerprint (in production would use SHA-256 crate)
    let mut hasher = DefaultHasher::new();
    data.hash(&mut hasher);
    format!("{:016x}", hasher.finish())
}

// ─── Tests ───────────────────────────────────────────────────────────────────

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

    fn new_store() -> SchemaStore {
        Arc::new(RwLock::new(SchemaRegistry::new()))
    }

    #[tokio::test]
    async fn test_register_and_retrieve_schema() {
        let store = new_store();
        let mut reg = store.write().await;

        let fingerprint = sha256_hex("message Msg { string text = 1; }");
        let now = chrono::Utc::now().to_rfc3339();

        reg.subjects.insert(
            "chat.v1.Msg".into(),
            SchemaSubject {
                id: "test-id-1".into(),
                subject: "chat.v1.Msg".into(),
                compatibility: CompatibilityMode::Backward,
                versions: vec![SchemaVersion {
                    version: 1,
                    schema: "message Msg { string text = 1; }".into(),
                    fingerprint: fingerprint.clone(),
                    registered_at: now.clone(),
                }],
                deleted: false,
                created_at: now.clone(),
                updated_at: now,
            },
        );

        assert_eq!(reg.subjects.len(), 1);
        let sub = reg.subjects.get("chat.v1.Msg").unwrap();
        assert_eq!(sub.versions.len(), 1);
        assert_eq!(sub.versions[0].version, 1);
        assert_eq!(sub.compatibility, CompatibilityMode::Backward);
    }

    #[tokio::test]
    async fn test_duplicate_fingerprint_detection() {
        let store = new_store();
        let schema = "message Dup { string a = 1; }";
        let fp = sha256_hex(schema);

        let mut reg = store.write().await;
        let now = chrono::Utc::now().to_rfc3339();

        reg.subjects.insert(
            "test.Dup".into(),
            SchemaSubject {
                id: "dup-id".into(),
                subject: "test.Dup".into(),
                compatibility: CompatibilityMode::None,
                versions: vec![SchemaVersion {
                    version: 1,
                    schema: schema.into(),
                    fingerprint: fp.clone(),
                    registered_at: now.clone(),
                }],
                deleted: false,
                created_at: now.clone(),
                updated_at: now,
            },
        );

        let sub = reg.subjects.get("test.Dup").unwrap();
        let dup = sub.versions.iter().find(|v| v.fingerprint == fp);
        assert!(dup.is_some());
        assert_eq!(dup.unwrap().version, 1);
    }

    #[tokio::test]
    async fn test_schema_versioning() {
        let store = new_store();
        let now = chrono::Utc::now().to_rfc3339();

        let mut reg = store.write().await;
        reg.subjects.insert(
            "ver.Msg".into(),
            SchemaSubject {
                id: "ver-id".into(),
                subject: "ver.Msg".into(),
                compatibility: CompatibilityMode::None,
                versions: vec![
                    SchemaVersion {
                        version: 1,
                        schema: "message Msg { string name = 1; }".into(),
                        fingerprint: sha256_hex("v1"),
                        registered_at: now.clone(),
                    },
                    SchemaVersion {
                        version: 2,
                        schema: "message Msg { string name = 1; string email? = 2; }".into(),
                        fingerprint: sha256_hex("v2"),
                        registered_at: now.clone(),
                    },
                ],
                deleted: false,
                created_at: now.clone(),
                updated_at: now,
            },
        );

        let sub = reg.subjects.get("ver.Msg").unwrap();
        assert_eq!(sub.versions.len(), 2);
        assert_eq!(sub.versions.last().unwrap().version, 2);
    }

    #[tokio::test]
    async fn test_list_with_prefix_filter() {
        let store = new_store();
        let now = chrono::Utc::now().to_rfc3339();

        let mut reg = store.write().await;
        for name in &["alpha.Msg", "alpha.Event", "beta.Cmd"] {
            reg.subjects.insert(
                name.to_string(),
                SchemaSubject {
                    id: format!("id-{}", name),
                    subject: name.to_string(),
                    compatibility: CompatibilityMode::Backward,
                    versions: vec![SchemaVersion {
                        version: 1,
                        schema: format!("message {} {{ string id = 1; }}", name.replace('.', "_")),
                        fingerprint: sha256_hex(name),
                        registered_at: now.clone(),
                    }],
                    deleted: false,
                    created_at: now.clone(),
                    updated_at: now.clone(),
                },
            );
        }

        let alpha: Vec<_> = reg
            .subjects
            .values()
            .filter(|s| s.subject.starts_with("alpha"))
            .collect();
        assert_eq!(alpha.len(), 2);

        let all: Vec<_> = reg.subjects.values().filter(|s| !s.deleted).collect();
        assert_eq!(all.len(), 3);
    }

    #[tokio::test]
    async fn test_soft_delete() {
        let store = new_store();
        let now = chrono::Utc::now().to_rfc3339();

        let mut reg = store.write().await;
        reg.subjects.insert(
            "to.Delete".into(),
            SchemaSubject {
                id: "del-id".into(),
                subject: "to.Delete".into(),
                compatibility: CompatibilityMode::None,
                versions: vec![SchemaVersion {
                    version: 1,
                    schema: "message Del { string id = 1; }".into(),
                    fingerprint: sha256_hex("del"),
                    registered_at: now.clone(),
                }],
                deleted: false,
                created_at: now.clone(),
                updated_at: now,
            },
        );

        reg.subjects.get_mut("to.Delete").unwrap().deleted = true;
        assert!(reg.subjects.get("to.Delete").unwrap().deleted);

        let visible: Vec<_> = reg.subjects.values().filter(|s| !s.deleted).collect();
        assert_eq!(visible.len(), 0);
    }

    #[tokio::test]
    async fn test_backward_compatibility_check() {
        let old = "message Msg {\n    string name = 1;\n    i32 age = 2;\n}";
        let new_compat =
            "message Msg {\n    string name = 1;\n    i32 age = 2;\n    string email? = 3;\n}";
        let new_breaking = "message Msg {\n    string name = 1;\n}";

        let result = check_schema_compatibility(old, new_compat, CompatibilityMode::Backward);
        assert!(
            result.compatible,
            "Adding optional field should be backward-compatible"
        );

        let result = check_schema_compatibility(old, new_breaking, CompatibilityMode::Backward);
        assert!(
            !result.compatible,
            "Removing required field should break backward compat"
        );
        assert!(result.errors.iter().any(|e| e.contains("age")));
    }

    #[tokio::test]
    async fn test_forward_compatibility_check() {
        let old = "message Msg {\n    string name = 1;\n}";
        let new_with_required = "message Msg {\n    string name = 1;\n    i32 age = 2;\n}";

        let result =
            check_schema_compatibility(old, new_with_required, CompatibilityMode::Forward);
        assert!(
            !result.compatible,
            "Adding required field breaks forward compat"
        );
        assert!(result
            .errors
            .iter()
            .any(|e| e.contains("FORWARD") && e.contains("age")));
    }

    #[tokio::test]
    async fn test_full_compatibility_check() {
        let old = "message Msg {\n    string name = 1;\n}";
        let new_with_required = "message Msg {\n    string name = 1;\n    i32 age = 2;\n}";

        let result = check_schema_compatibility(old, new_with_required, CompatibilityMode::Full);
        assert!(
            !result.compatible,
            "Adding required field breaks both forward and backward compat"
        );
    }

    #[tokio::test]
    async fn test_compatibility_none_allows_anything() {
        let old = "message A { string x = 1; }";
        let new = "enum B { FOO = 0; }";

        let result = check_schema_compatibility(old, new, CompatibilityMode::None);
        assert!(result.compatible, "NONE mode should allow any change");
    }

    #[tokio::test]
    async fn test_message_removal_is_breaking() {
        let old = "message A { string x = 1; }\nmessage B { i32 y = 1; }";
        let new = "message A { string x = 1; }";

        let result = check_schema_compatibility(old, new, CompatibilityMode::Backward);
        assert!(
            !result.compatible,
            "Removing a message should break backward compat"
        );
        assert!(result.errors.iter().any(|e| e.contains("B")));
    }

    #[tokio::test]
    async fn test_adding_new_message_is_safe() {
        let old = "message A { string x = 1; }";
        let new = "message A { string x = 1; }\nmessage B { i32 y = 1; }";

        let result = check_schema_compatibility(old, new, CompatibilityMode::Backward);
        assert!(
            result.compatible,
            "Adding a new message should be backward-compatible"
        );
    }

    #[test]
    fn test_fingerprint_determinism() {
        let a = sha256_hex("message Foo { string bar = 1; }");
        let b = sha256_hex("message Foo { string bar = 1; }");
        let c = sha256_hex("message Foo { string baz = 1; }");
        assert_eq!(a, b, "Same input should produce same fingerprint");
        assert_ne!(a, c, "Different input should produce different fingerprint");
    }
}

