cufflink-types 0.9.0

use serde::{Deserialize, Serialize};
use std::collections::HashMap;

pub mod openapi;

/// A service manifest describes the complete definition of a cufflink service.
/// This is generated by the SDK macros from the user's Rust struct definitions.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServiceManifest {
    pub name: String,
    pub version: Option<String>,
    pub tables: Vec<TableDefinition>,
    #[serde(default)]
    pub cells: Vec<CellDefinition>,
    #[serde(default)]
    pub events: Vec<EventDefinition>,
    #[serde(default)]
    pub subscriptions: Vec<SubscriptionDefinition>,
    #[serde(default)]
    pub custom_routes: Vec<CustomRouteDefinition>,
    #[serde(default)]
    pub mode: ServiceMode,
    /// Authorization configuration (permission areas and default roles).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub authorization: Option<AuthorizationConfig>,
    /// Optional WASM handler invoked between additive and destructive schema
    /// phases of a deploy. Used to backfill or restructure data when columns
    /// are split, renamed, or have their semantics changed. The handler must
    /// be idempotent — it may be re-invoked on retried deploys.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub on_migrate: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum ServiceMode {
    #[default]
    Crud,
    Wasm,
    Container,
    Web,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableDefinition {
    pub name: String,
    pub columns: Vec<ColumnDefinition>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub indexes: Vec<IndexDefinition>,
    #[serde(default)]
    pub soft_delete: bool,
    /// Column name that stores the owner's user ID for row-level security.
    /// When set, CRUD operations are scoped to the authenticated user's records.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub owner_field: Option<String>,
    /// When true, all CRUD operations on this table require authentication.
    /// Unauthenticated requests receive a 401 Unauthorized response.
    #[serde(default)]
    pub auth_required: bool,
    /// Permission area that controls access to this table's CRUD operations.
    /// When set, platform checks "{area}:view" for GET, "{area}:create" for POST,
    /// "{area}:edit" for PUT, "{area}:delete" for DELETE.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub permission_area: Option<String>,
    /// CRUD lifecycle hooks — WASM handler names called before/after operations.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub hooks: Option<TableHooks>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ColumnDefinition {
    pub name: String,
    pub column_type: ColumnType,
    #[serde(default)]
    pub primary_key: bool,
    #[serde(default)]
    pub nullable: bool,
    #[serde(default)]
    pub auto_generate: bool,
    pub default_value: Option<String>,
    /// Foreign key reference in "table.column" format
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub references: Option<String>,
    /// Whether to cascade deletes on the FK
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub on_delete: Option<ForeignKeyAction>,
    /// Whether this column has a unique constraint
    #[serde(default)]
    pub unique: bool,
    /// Validation rules for this column
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub validations: Vec<ValidationRule>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "rule", rename_all = "snake_case")]
// Note: ValidationRule contains f64 fields (Min/Max), so Eq cannot be derived
pub enum ValidationRule {
    /// Regex pattern match
    Regex { pattern: String },
    /// Minimum numeric value
    Min { value: f64 },
    /// Maximum numeric value
    Max { value: f64 },
    /// Minimum string length
    MinLength { value: usize },
    /// Maximum string length
    MaxLength { value: usize },
    /// Must be one of the listed values
    OneOf { values: Vec<String> },
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct IndexDefinition {
    pub name: String,
    pub columns: Vec<String>,
    #[serde(default)]
    pub unique: bool,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ForeignKeyAction {
    Cascade,
    SetNull,
    Restrict,
    NoAction,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum ColumnType {
    Uuid,
    Text,
    Integer,
    BigInteger,
    Float,
    Double,
    Boolean,
    Timestamp,
    Date,
    Jsonb,
}

impl ColumnType {
    pub fn to_sql(&self) -> &str {
        match self {
            ColumnType::Uuid => "UUID",
            ColumnType::Text => "TEXT",
            ColumnType::Integer => "INTEGER",
            ColumnType::BigInteger => "BIGINT",
            ColumnType::Float => "REAL",
            ColumnType::Double => "DOUBLE PRECISION",
            ColumnType::Boolean => "BOOLEAN",
            ColumnType::Timestamp => "TIMESTAMPTZ",
            ColumnType::Date => "DATE",
            ColumnType::Jsonb => "JSONB",
        }
    }

    /// Map to the filter column type for query param parsing
    pub fn filter_type(&self) -> &str {
        match self {
            ColumnType::Uuid => "uuid",
            ColumnType::Text => "text",
            ColumnType::Integer | ColumnType::BigInteger => "integer",
            ColumnType::Float | ColumnType::Double => "double",
            ColumnType::Boolean => "boolean",
            ColumnType::Timestamp => "timestamp",
            ColumnType::Date => "date",
            ColumnType::Jsonb => "jsonb",
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CellDefinition {
    pub name: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventDefinition {
    pub name: String,
    pub table: String,
    pub action: EventAction,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum EventAction {
    Create,
    Update,
    Delete,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SubscriptionDefinition {
    pub subject: String,
    pub handler: String,
    pub handler_type: HandlerType,
    #[serde(default)]
    pub config: HashMap<String, String>,
    /// Explicit JetStream stream to host this subscription's consumer. When
    /// `None`, the platform uses the tenant stream (`DW_{tenant}`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum HandlerType {
    DeleteCascade,
    UpdateField,
    Webhook,
    Wasm,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CustomRouteDefinition {
    pub method: String,
    pub path: String,
    pub handler: String,
    /// When present, the HTTP route accepts the request, persists a
    /// `wasm_jobs` row, publishes to JetStream, and returns `202 Accepted`
    /// with `{job_id}`. The handler executes asynchronously on a worker
    /// with no inline 30s wall.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub job: Option<JobConfig>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct JobConfig {
    /// Per-handler wall-clock ceiling enforced by the worker.
    pub timeout_secs: u32,
    /// Maximum delivery attempts before the job is permanently failed.
    /// `1` disables retry-on-crash.
    pub max_attempts: u32,
}

impl JobConfig {
    pub const DEFAULT_TIMEOUT_SECS: u32 = 600;
    pub const DEFAULT_MAX_ATTEMPTS: u32 = 3;
}

impl Default for JobConfig {
    fn default() -> Self {
        Self {
            timeout_secs: Self::DEFAULT_TIMEOUT_SECS,
            max_attempts: Self::DEFAULT_MAX_ATTEMPTS,
        }
    }
}

/// CRUD lifecycle hooks for a table. Each field names a WASM handler function.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct TableHooks {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub before_create: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after_create: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub before_update: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after_update: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub before_delete: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after_delete: Option<String>,
}

/// Authorization configuration declared in the service manifest.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthorizationConfig {
    /// Permission areas declared by this service.
    #[serde(default)]
    pub areas: Vec<PermissionAreaDef>,
    /// Default roles to seed on first deployment.
    #[serde(default)]
    pub default_roles: Vec<DefaultRoleDef>,
}

/// A permission area with its supported operations.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PermissionAreaDef {
    pub name: String,
    #[serde(default)]
    pub operations: Vec<String>,
}

/// A default role seeded on deployment.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DefaultRoleDef {
    pub name: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    #[serde(default)]
    pub permissions: Vec<String>,
}

/// Description of the changes a deploy is about to apply, passed as the
/// payload to the `on_migrate` hook so user code can decide which backfill
/// blocks should fire.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct SchemaDiff {
    #[serde(default)]
    pub added_tables: Vec<String>,
    #[serde(default)]
    pub dropped_tables: Vec<String>,
    /// (table, column) pairs for newly added columns. In phase 2 these
    /// already exist in the database (added in phase 1) but may be NULL on
    /// existing rows; the hook is the place to backfill them.
    #[serde(default)]
    pub added_columns: Vec<(String, String)>,
    /// (table, column) pairs for columns that will be dropped in phase 3.
    /// During phase 2 these are still present in the database, so the hook
    /// can read them to populate new columns before they disappear.
    #[serde(default)]
    pub dropped_columns: Vec<(String, String)>,
    /// (table, column, old_type, new_type)
    #[serde(default)]
    pub type_changes: Vec<(String, String, String, String)>,
    /// (table, column, new_nullable)
    #[serde(default)]
    pub nullability_changes: Vec<(String, String, bool)>,
}

impl SchemaDiff {
    pub fn added_column(&self, table: &str, col: &str) -> bool {
        self.added_columns
            .iter()
            .any(|(t, c)| t == table && c == col)
    }

    pub fn dropped_column(&self, table: &str, col: &str) -> bool {
        self.dropped_columns
            .iter()
            .any(|(t, c)| t == table && c == col)
    }

    pub fn added_table(&self, table: &str) -> bool {
        self.added_tables.iter().any(|t| t == table)
    }

    pub fn dropped_table(&self, table: &str) -> bool {
        self.dropped_tables.iter().any(|t| t == table)
    }

    pub fn is_empty(&self) -> bool {
        self.added_tables.is_empty()
            && self.dropped_tables.is_empty()
            && self.added_columns.is_empty()
            && self.dropped_columns.is_empty()
            && self.type_changes.is_empty()
            && self.nullability_changes.is_empty()
    }
}

impl ServiceManifest {
    /// Get table definition by name
    pub fn get_table(&self, name: &str) -> Option<&TableDefinition> {
        self.tables.iter().find(|t| t.name == name)
    }

    /// Whether this manifest needs a WASM artifact bundled at deploy time.
    ///
    /// A WASM module is required when the service runs user code: explicit
    /// `mode: wasm`, an `on_migrate` schema hook, table CRUD lifecycle hooks,
    /// or any `Wasm` subscription/custom-route handler. The CLI uses this to
    /// decide whether to compile a WASM artifact; the platform uses it to
    /// reject deploys that arrive without one.
    pub fn requires_wasm(&self) -> bool {
        if self.mode == ServiceMode::Wasm {
            return true;
        }
        if self.on_migrate.is_some() {
            return true;
        }
        if self.tables.iter().any(|t| t.hooks.is_some()) {
            return true;
        }
        if !self.custom_routes.is_empty() {
            return true;
        }
        if self
            .subscriptions
            .iter()
            .any(|s| matches!(s.handler_type, HandlerType::Wasm))
        {
            return true;
        }
        false
    }

    /// All handler names this manifest expects the WASM module to export.
    /// Used at deploy time to reject manifests whose declared handlers are
    /// missing from the bundled binary.
    pub fn declared_wasm_handlers(&self) -> Vec<&str> {
        let mut handlers: Vec<&str> = Vec::new();
        if let Some(ref h) = self.on_migrate {
            handlers.push(h.as_str());
        }
        for route in &self.custom_routes {
            handlers.push(route.handler.as_str());
        }
        for sub in &self.subscriptions {
            if matches!(sub.handler_type, HandlerType::Wasm) {
                handlers.push(sub.handler.as_str());
            }
        }
        for table in &self.tables {
            if let Some(ref hooks) = table.hooks {
                for h in [
                    hooks.before_create.as_deref(),
                    hooks.after_create.as_deref(),
                    hooks.before_update.as_deref(),
                    hooks.after_update.as_deref(),
                    hooks.before_delete.as_deref(),
                    hooks.after_delete.as_deref(),
                ]
                .iter()
                .flatten()
                {
                    handlers.push(h);
                }
            }
        }
        handlers.sort();
        handlers.dedup();
        handlers
    }

    /// Compute a SHA-256 hash of the manifest for change detection
    pub fn schema_hash(&self) -> String {
        use sha2::{Digest, Sha256};
        let mut hasher = Sha256::new();
        hasher.update(
            serde_json::to_string(&self.tables)
                .unwrap_or_default()
                .as_bytes(),
        );
        hasher.update(
            serde_json::to_string(&self.custom_routes)
                .unwrap_or_default()
                .as_bytes(),
        );
        hasher.update(
            serde_json::to_string(&self.subscriptions)
                .unwrap_or_default()
                .as_bytes(),
        );
        hasher.update(
            serde_json::to_string(&self.authorization)
                .unwrap_or_default()
                .as_bytes(),
        );
        hasher.update(
            serde_json::to_string(&self.on_migrate)
                .unwrap_or_default()
                .as_bytes(),
        );
        hex::encode(hasher.finalize())
    }
}

impl TableDefinition {
    /// Get the primary key column
    pub fn primary_key(&self) -> Option<&ColumnDefinition> {
        self.columns.iter().find(|c| c.primary_key)
    }

    /// Get all non-primary-key, non-auto columns (for INSERT/UPDATE)
    pub fn writable_columns(&self) -> Vec<&ColumnDefinition> {
        self.columns
            .iter()
            .filter(|c| !c.auto_generate || !c.primary_key)
            .collect()
    }
}

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

    fn sample_manifest() -> ServiceManifest {
        ServiceManifest {
            name: "test-service".into(),
            version: Some("1.0.0".into()),
            tables: vec![TableDefinition {
                name: "todos".into(),
                columns: vec![
                    ColumnDefinition {
                        name: "id".into(),
                        column_type: ColumnType::Uuid,
                        primary_key: true,
                        nullable: false,
                        auto_generate: true,
                        default_value: None,
                        references: None,
                        on_delete: None,
                        unique: false,
                        validations: vec![],
                    },
                    ColumnDefinition {
                        name: "title".into(),
                        column_type: ColumnType::Text,
                        primary_key: false,
                        nullable: false,
                        auto_generate: false,
                        default_value: None,
                        references: None,
                        on_delete: None,
                        unique: false,
                        validations: vec![],
                    },
                    ColumnDefinition {
                        name: "user_id".into(),
                        column_type: ColumnType::Uuid,
                        primary_key: false,
                        nullable: false,
                        auto_generate: false,
                        default_value: None,
                        references: Some("users.id".into()),
                        on_delete: Some(ForeignKeyAction::Cascade),
                        unique: false,
                        validations: vec![],
                    },
                ],
                indexes: vec![],
                soft_delete: false,
                owner_field: None,
                auth_required: false,
                permission_area: None,
                hooks: None,
            }],
            cells: vec![],
            events: vec![],
            subscriptions: vec![],
            custom_routes: vec![],
            mode: ServiceMode::Crud,
            authorization: None,
            on_migrate: None,
        }
    }

    #[test]
    fn test_manifest_serialization_roundtrip() {
        let manifest = sample_manifest();
        let json = serde_json::to_string(&manifest).unwrap();
        let parsed: ServiceManifest = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.name, "test-service");
        assert_eq!(parsed.tables.len(), 1);
        assert_eq!(parsed.tables[0].columns.len(), 3);
    }

    #[test]
    fn test_get_table() {
        let manifest = sample_manifest();
        assert!(manifest.get_table("todos").is_some());
        assert!(manifest.get_table("nonexistent").is_none());
    }

    #[test]
    fn test_primary_key() {
        let manifest = sample_manifest();
        let table = manifest.get_table("todos").unwrap();
        let pk = table.primary_key().unwrap();
        assert_eq!(pk.name, "id");
        assert!(pk.auto_generate);
    }

    #[test]
    fn test_schema_hash_deterministic() {
        let m1 = sample_manifest();
        let m2 = sample_manifest();
        assert_eq!(m1.schema_hash(), m2.schema_hash());
    }

    #[test]
    fn test_schema_hash_changes() {
        let mut m1 = sample_manifest();
        let m2 = sample_manifest();
        m1.tables[0].columns.push(ColumnDefinition {
            name: "extra".into(),
            column_type: ColumnType::Text,
            primary_key: false,
            nullable: true,
            auto_generate: false,
            default_value: None,
            references: None,
            on_delete: None,
            unique: false,
            validations: vec![],
        });
        assert_ne!(m1.schema_hash(), m2.schema_hash());
    }

    #[test]
    fn test_column_type_to_sql() {
        assert_eq!(ColumnType::Uuid.to_sql(), "UUID");
        assert_eq!(ColumnType::Text.to_sql(), "TEXT");
        assert_eq!(ColumnType::Integer.to_sql(), "INTEGER");
        assert_eq!(ColumnType::BigInteger.to_sql(), "BIGINT");
        assert_eq!(ColumnType::Boolean.to_sql(), "BOOLEAN");
        assert_eq!(ColumnType::Timestamp.to_sql(), "TIMESTAMPTZ");
        assert_eq!(ColumnType::Date.to_sql(), "DATE");
        assert_eq!(ColumnType::Jsonb.to_sql(), "JSONB");
    }

    #[test]
    fn test_fk_serialization() {
        let manifest = sample_manifest();
        let json = serde_json::to_string(&manifest).unwrap();
        assert!(json.contains("\"references\":\"users.id\""));
        assert!(json.contains("\"on_delete\":\"cascade\""));

        let parsed: ServiceManifest = serde_json::from_str(&json).unwrap();
        let user_id_col = &parsed.tables[0].columns[2];
        assert_eq!(user_id_col.references.as_deref(), Some("users.id"));
        assert_eq!(user_id_col.on_delete, Some(ForeignKeyAction::Cascade));
    }

    #[test]
    fn test_fk_not_serialized_when_none() {
        let col = ColumnDefinition {
            name: "title".into(),
            column_type: ColumnType::Text,
            primary_key: false,
            nullable: false,
            auto_generate: false,
            default_value: None,
            references: None,
            on_delete: None,
            unique: false,
            validations: vec![],
        };
        let json = serde_json::to_string(&col).unwrap();
        assert!(!json.contains("references"));
        assert!(!json.contains("on_delete"));
    }

    #[test]
    fn test_unique_column_serialization() {
        let col = ColumnDefinition {
            name: "email".into(),
            column_type: ColumnType::Text,
            primary_key: false,
            nullable: false,
            auto_generate: false,
            default_value: None,
            references: None,
            on_delete: None,
            unique: true,
            validations: vec![],
        };
        let json = serde_json::to_string(&col).unwrap();
        assert!(json.contains("\"unique\":true"));
        let parsed: ColumnDefinition = serde_json::from_str(&json).unwrap();
        assert!(parsed.unique);
    }

    #[test]
    fn test_index_serialization() {
        let table = TableDefinition {
            name: "users".into(),
            columns: vec![],
            indexes: vec![IndexDefinition {
                name: "idx_users_email".into(),
                columns: vec!["email".into()],
                unique: true,
            }],
            soft_delete: false,
            owner_field: None,
            auth_required: false,
            permission_area: None,
            hooks: None,
        };
        let json = serde_json::to_string(&table).unwrap();
        assert!(json.contains("idx_users_email"));
        let parsed: TableDefinition = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.indexes.len(), 1);
        assert!(parsed.indexes[0].unique);
    }

    #[test]
    fn test_indexes_not_serialized_when_empty() {
        let table = TableDefinition {
            name: "users".into(),
            columns: vec![],
            indexes: vec![],
            soft_delete: false,
            owner_field: None,
            auth_required: false,
            permission_area: None,
            hooks: None,
        };
        let json = serde_json::to_string(&table).unwrap();
        assert!(!json.contains("indexes"));
    }

    #[test]
    fn test_service_mode_default() {
        let json = r#"{"name":"svc","tables":[]}"#;
        let m: ServiceManifest = serde_json::from_str(json).unwrap();
        assert_eq!(m.mode, ServiceMode::Crud);
    }

    #[test]
    fn test_owner_field_serialization() {
        let table = TableDefinition {
            name: "notes".into(),
            columns: vec![],
            indexes: vec![],
            soft_delete: false,
            owner_field: Some("user_id".into()),
            auth_required: false,
            permission_area: None,
            hooks: None,
        };
        let json = serde_json::to_string(&table).unwrap();
        assert!(json.contains("\"owner_field\":\"user_id\""));
        let parsed: TableDefinition = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.owner_field.as_deref(), Some("user_id"));
    }

    #[test]
    fn test_owner_field_not_serialized_when_none() {
        let table = TableDefinition {
            name: "notes".into(),
            columns: vec![],
            indexes: vec![],
            soft_delete: false,
            owner_field: None,
            auth_required: false,
            permission_area: None,
            hooks: None,
        };
        let json = serde_json::to_string(&table).unwrap();
        assert!(!json.contains("owner_field"));
    }

    #[test]
    fn test_owner_field_defaults_to_none() {
        let json = r#"{"name":"notes","columns":[]}"#;
        let table: TableDefinition = serde_json::from_str(json).unwrap();
        assert!(table.owner_field.is_none());
    }

    #[test]
    fn test_auth_required_serialization() {
        let table = TableDefinition {
            name: "orders".into(),
            columns: vec![],
            indexes: vec![],
            soft_delete: false,
            owner_field: None,
            auth_required: true,
            permission_area: None,
            hooks: None,
        };
        let json = serde_json::to_string(&table).unwrap();
        assert!(json.contains("\"auth_required\":true"));
        let parsed: TableDefinition = serde_json::from_str(&json).unwrap();
        assert!(parsed.auth_required);
    }

    #[test]
    fn test_auth_required_defaults_to_false() {
        let json = r#"{"name":"orders","columns":[]}"#;
        let table: TableDefinition = serde_json::from_str(json).unwrap();
        assert!(!table.auth_required);
    }

    #[test]
    fn test_on_migrate_defaults_to_none() {
        let json = r#"{"name":"svc","tables":[]}"#;
        let m: ServiceManifest = serde_json::from_str(json).unwrap();
        assert!(m.on_migrate.is_none());
    }

    #[test]
    fn test_on_migrate_serialization_roundtrip() {
        let mut m = sample_manifest();
        m.on_migrate = Some("handle_on_migrate".into());
        let json = serde_json::to_string(&m).unwrap();
        assert!(json.contains("\"on_migrate\":\"handle_on_migrate\""));
        let parsed: ServiceManifest = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.on_migrate.as_deref(), Some("handle_on_migrate"));
    }

    #[test]
    fn test_on_migrate_skipped_when_none() {
        let m = sample_manifest();
        let json = serde_json::to_string(&m).unwrap();
        assert!(!json.contains("on_migrate"));
    }

    #[test]
    fn test_on_migrate_changes_schema_hash() {
        let m1 = sample_manifest();
        let mut m2 = sample_manifest();
        m2.on_migrate = Some("handle_on_migrate".into());
        assert_ne!(m1.schema_hash(), m2.schema_hash());
    }

    #[test]
    fn test_subscription_stream_defaults_to_none_in_legacy_json() {
        let json = r#"{"subject":"x","handler":"h","handler_type":"wasm"}"#;
        let sub: SubscriptionDefinition = serde_json::from_str(json).unwrap();
        assert!(sub.stream.is_none());
    }

    #[test]
    fn test_subscription_stream_roundtrip_when_set() {
        let sub = SubscriptionDefinition {
            subject: "cufflink.tick.5min".into(),
            handler: "handle_tick_resync".into(),
            handler_type: HandlerType::Wasm,
            config: HashMap::new(),
            stream: Some("CUFFLINK_TICKS".into()),
        };
        let json = serde_json::to_string(&sub).unwrap();
        assert!(json.contains("\"stream\":\"CUFFLINK_TICKS\""));
        let parsed: SubscriptionDefinition = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.stream.as_deref(), Some("CUFFLINK_TICKS"));
    }

    #[test]
    fn test_subscription_stream_skipped_when_none() {
        let sub = SubscriptionDefinition {
            subject: "x".into(),
            handler: "h".into(),
            handler_type: HandlerType::Wasm,
            config: HashMap::new(),
            stream: None,
        };
        let json = serde_json::to_string(&sub).unwrap();
        assert!(!json.contains("stream"));
    }

    #[test]
    fn test_schema_diff_helpers() {
        let diff = SchemaDiff {
            added_tables: vec!["batches".into()],
            dropped_tables: vec!["legacy_batches".into()],
            added_columns: vec![("pickups".into(), "min".into())],
            dropped_columns: vec![("pickups".into(), "midpoint".into())],
            type_changes: vec![],
            nullability_changes: vec![],
        };
        assert!(!diff.is_empty());
        assert!(diff.added_table("batches"));
        assert!(!diff.added_table("nope"));
        assert!(diff.dropped_table("legacy_batches"));
        assert!(diff.added_column("pickups", "min"));
        assert!(!diff.added_column("pickups", "max"));
        assert!(diff.dropped_column("pickups", "midpoint"));
    }

    #[test]
    fn test_schema_diff_is_empty_default() {
        assert!(SchemaDiff::default().is_empty());
    }

    #[test]
    fn test_schema_diff_serialization_roundtrip() {
        let diff = SchemaDiff {
            added_tables: vec![],
            dropped_tables: vec![],
            added_columns: vec![("t".into(), "c".into())],
            dropped_columns: vec![],
            type_changes: vec![("t".into(), "c".into(), "TEXT".into(), "INTEGER".into())],
            nullability_changes: vec![("t".into(), "c".into(), false)],
        };
        let json = serde_json::to_string(&diff).unwrap();
        let parsed: SchemaDiff = serde_json::from_str(&json).unwrap();
        assert!(parsed.added_column("t", "c"));
        assert_eq!(parsed.type_changes.len(), 1);
        assert_eq!(parsed.nullability_changes.len(), 1);
    }

    #[test]
    fn test_requires_wasm_false_for_plain_crud() {
        let m = sample_manifest();
        assert!(!m.requires_wasm());
    }

    #[test]
    fn test_requires_wasm_true_for_wasm_mode() {
        let mut m = sample_manifest();
        m.mode = ServiceMode::Wasm;
        assert!(m.requires_wasm());
    }

    #[test]
    fn test_requires_wasm_true_for_on_migrate() {
        let mut m = sample_manifest();
        m.on_migrate = Some("handle_on_migrate".into());
        assert!(m.requires_wasm());
    }

    #[test]
    fn test_requires_wasm_true_for_table_hooks() {
        let mut m = sample_manifest();
        m.tables[0].hooks = Some(TableHooks {
            before_create: Some("validate".into()),
            ..Default::default()
        });
        assert!(m.requires_wasm());
    }

    #[test]
    fn test_requires_wasm_true_for_custom_routes() {
        let mut m = sample_manifest();
        m.custom_routes.push(CustomRouteDefinition {
            method: "GET".into(),
            path: "/hello".into(),
            handler: "hello".into(),
            job: None,
        });
        assert!(m.requires_wasm());
    }

    #[test]
    fn test_requires_wasm_true_for_wasm_subscription() {
        let mut m = sample_manifest();
        m.subscriptions.push(SubscriptionDefinition {
            subject: "events.foo".into(),
            handler: "on_foo".into(),
            handler_type: HandlerType::Wasm,
            config: HashMap::new(),
            stream: None,
        });
        assert!(m.requires_wasm());
    }

    #[test]
    fn test_requires_wasm_false_for_non_wasm_subscription() {
        let mut m = sample_manifest();
        m.subscriptions.push(SubscriptionDefinition {
            subject: "events.foo".into(),
            handler: "on_foo".into(),
            handler_type: HandlerType::Webhook,
            config: HashMap::new(),
            stream: None,
        });
        assert!(!m.requires_wasm());
    }

    #[test]
    fn test_declared_wasm_handlers_collects_all_sources() {
        let mut m = sample_manifest();
        m.on_migrate = Some("on_mig".into());
        m.custom_routes.push(CustomRouteDefinition {
            method: "GET".into(),
            path: "/h".into(),
            handler: "route_h".into(),
            job: None,
        });
        m.subscriptions.push(SubscriptionDefinition {
            subject: "x".into(),
            handler: "sub_h".into(),
            handler_type: HandlerType::Wasm,
            config: HashMap::new(),
            stream: None,
        });
        m.subscriptions.push(SubscriptionDefinition {
            subject: "y".into(),
            handler: "wh_ignored".into(),
            handler_type: HandlerType::Webhook,
            config: HashMap::new(),
            stream: None,
        });
        m.tables[0].hooks = Some(TableHooks {
            before_create: Some("bc".into()),
            after_create: Some("ac".into()),
            before_delete: Some("bc".into()), // duplicate handler name
            ..Default::default()
        });
        let handlers = m.declared_wasm_handlers();
        assert!(handlers.contains(&"on_mig"));
        assert!(handlers.contains(&"route_h"));
        assert!(handlers.contains(&"sub_h"));
        assert!(handlers.contains(&"bc"));
        assert!(handlers.contains(&"ac"));
        assert!(!handlers.contains(&"wh_ignored"));
        // Deduplicated
        assert_eq!(handlers.iter().filter(|h| **h == "bc").count(), 1);
    }

    #[test]
    fn job_config_defaults() {
        let cfg = JobConfig::default();
        assert_eq!(cfg.timeout_secs, JobConfig::DEFAULT_TIMEOUT_SECS);
        assert_eq!(cfg.max_attempts, JobConfig::DEFAULT_MAX_ATTEMPTS);
    }

    #[test]
    fn custom_route_serde_round_trip_with_job() {
        let route = CustomRouteDefinition {
            method: "POST".into(),
            path: "/run".into(),
            handler: "do_thing".into(),
            job: Some(JobConfig {
                timeout_secs: 900,
                max_attempts: 2,
            }),
        };
        let json = serde_json::to_string(&route).unwrap();
        let round: CustomRouteDefinition = serde_json::from_str(&json).unwrap();
        assert_eq!(round.method, "POST");
        let job = round.job.unwrap();
        assert_eq!(job.timeout_secs, 900);
        assert_eq!(job.max_attempts, 2);
    }

    #[test]
    fn custom_route_without_job_serializes_without_field() {
        let route = CustomRouteDefinition {
            method: "GET".into(),
            path: "/x".into(),
            handler: "x".into(),
            job: None,
        };
        let json = serde_json::to_value(&route).unwrap();
        assert!(json.get("job").is_none(), "absent job should be omitted");
    }

    #[test]
    fn custom_route_legacy_json_without_job_deserializes() {
        let json = r#"{"method":"GET","path":"/x","handler":"x"}"#;
        let route: CustomRouteDefinition = serde_json::from_str(json).unwrap();
        assert!(route.job.is_none());
    }
}