cufflink_types/manifest/

service.rs

use super::{
    AuthorizationConfig, CustomRouteDefinition, HandlerType, MigrationHistory, ScheduleDefinition,
    SqlMigration, StreamDefinition, SubscriptionDefinition, TableDefinition,
};
use crate::features;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::BTreeMap;

/// 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, Default, Serialize, Deserialize)]
pub struct ServiceManifest {
    pub name: String,
    pub version: Option<String>,
    pub tables: Vec<TableDefinition>,
    /// Tables only the service's own SQL reaches. The platform checks them
    /// against the database as it does `tables`, and serves none of them.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub private_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>,
    /// NATS subject patterns (with `*`/`>` wildcards) this service is allowed to
    /// publish or request outside its own `dw.{tenant}.*` namespace. `None` marks
    /// a manifest built before publish authorization existed: the platform then
    /// runs in legacy mode (allow + warn). `Some` — always emitted by the current
    /// SDK — enables enforcement.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub publishes: Option<Vec<String>>,
    /// Extensions the service's handlers may call through `ext_call`.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub extensions: Vec<String>,
    /// Extensions a native service serves, and who may call each.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub provides: Vec<ProvidedExtension>,
    /// Platform capabilities this service requires to run safely. The platform
    /// rejects a deploy whose required features it does not advertise at
    /// `/api/version`. The SDK auto-derives this (e.g. `route_posture` when any
    /// route declares a posture) so a security-relevant manifest cannot be
    /// silently deployed to a platform too old to enforce it.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub required_platform_features: Vec<String>,
    /// Serve the auto-generated GraphQL endpoint. Off unless the service
    /// opts in.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub graphql: bool,
    /// How custom routes are reached: `legacy` serves each only by handler
    /// name under `_fn/`; `strict` also matches their declared paths.
    #[serde(default, skip_serializing_if = "RoutingMode::is_legacy")]
    pub routing: RoutingMode,
    /// The JSON key the platform's own error responses for this service
    /// carry their message under, instead of `error`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub errors: Option<ErrorProfile>,
    /// The service's OpenAPI document settings; its presence (or any
    /// documented route) makes the spec server-relative.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub openapi: Option<OpenApiOptions>,
    /// Shared JSON Schemas the route docs refer to as
    /// `#/components/schemas/{name}`.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub components: BTreeMap<String, Value>,
    /// How the durable consumers of the service's subscriptions are named.
    #[serde(default, skip_serializing_if = "ConsumerNaming::is_legacy")]
    pub consumers: ConsumerNaming,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub streams: Vec<StreamDefinition>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub schedules: Vec<ScheduleDefinition>,
    /// Who owns the DDL of the service's schema.
    #[serde(default, skip_serializing_if = "SchemaOwnership::is_managed")]
    pub schema: SchemaOwnership,
    /// The service's Postgres schema; derived from its name when unset.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub schema_name: Option<String>,
    /// The service's SQL migrations, in version order.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub migrations: Vec<SqlMigration>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub migration_history: Option<MigrationHistory>,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ConsumerNaming {
    /// `{service}_{handler}`.
    #[default]
    Legacy,
    /// Named for the tenant, the service, the handler and what it consumes.
    Scoped,
}

impl ConsumerNaming {
    pub fn is_legacy(&self) -> bool {
        *self == ConsumerNaming::Legacy
    }
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SchemaOwnership {
    /// The platform creates and alters the schema from the manifest.
    #[default]
    Managed,
    /// The schema exists already; the platform never emits DDL for it.
    Adopted,
}

impl SchemaOwnership {
    pub fn is_managed(&self) -> bool {
        *self == SchemaOwnership::Managed
    }
}

#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct OpenApiOptions {
    /// Defaults to the service name.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub title: Option<String>,
    /// Defaults to the manifest version.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub version: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    /// Defaults to the service's own `/svc/{tenant}/{service}` base.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub servers: Vec<ServerDoc>,
    /// The document's tag list.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub tags: Vec<TagDoc>,
    /// The HTTP bearer scheme authenticated routes name; `bearer_auth`
    /// when unset.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub security_scheme: Option<String>,
}

impl OpenApiOptions {
    pub const DEFAULT_SECURITY_SCHEME: &'static str = "bearer_auth";

    pub fn security_scheme(&self) -> &str {
        self.security_scheme
            .as_deref()
            .unwrap_or(Self::DEFAULT_SECURITY_SCHEME)
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ServerDoc {
    pub url: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TagDoc {
    pub name: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ErrorProfile {
    pub key: String,
}

impl ErrorProfile {
    /// Whether `key` is a valid error key: `[a-z_][a-z0-9_]{0,63}`.
    pub fn is_valid_key(key: &str) -> bool {
        let mut bytes = key.bytes();
        key.len() <= 64
            && bytes
                .next()
                .is_some_and(|b| b.is_ascii_lowercase() || b == b'_')
            && bytes.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
    }
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RoutingMode {
    #[default]
    Legacy,
    Strict,
}

impl RoutingMode {
    pub fn is_legacy(&self) -> bool {
        *self == RoutingMode::Legacy
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum ServiceMode {
    #[default]
    Crud,
    Wasm,
    Container,
    Web,
    /// A Rust binary supervised by the extension runner that serves the
    /// extensions it `provides`.
    Native,
}

impl ServiceMode {
    /// The mode as a manifest names it.
    pub fn as_str(&self) -> &'static str {
        match self {
            ServiceMode::Crud => "crud",
            ServiceMode::Wasm => "wasm",
            ServiceMode::Container => "container",
            ServiceMode::Web => "web",
            ServiceMode::Native => "native",
        }
    }

    /// Whether the platform runs the service itself: its schema, module,
    /// database role, subscriptions and schedules. A web app and a native
    /// service run in a runner of their own and have none of them.
    pub fn runs_on_platform(&self) -> bool {
        match self {
            ServiceMode::Crud | ServiceMode::Wasm | ServiceMode::Container => true,
            ServiceMode::Web | ServiceMode::Native => false,
        }
    }
}

/// An extension a native service serves.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProvidedExtension {
    pub name: String,
    /// Service names, or [`ANY_CALLER`](crate::ext::ANY_CALLER).
    pub callers: Vec<String>,
}

#[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,
}

impl ServiceManifest {
    /// The served table named `name`; a private table is never found.
    pub fn get_table(&self, name: &str) -> Option<&TableDefinition> {
        self.tables.iter().find(|t| t.name == name)
    }

    /// Every table the service declares, served or not.
    pub fn declared_tables(&self) -> impl Iterator<Item = &TableDefinition> {
        self.tables.iter().chain(&self.private_tables)
    }

    /// 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,
    /// a schedule, 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 {
        match self.mode {
            ServiceMode::Wasm => true,
            ServiceMode::Native => false,
            ServiceMode::Crud | ServiceMode::Container | ServiceMode::Web => {
                self.on_migrate.is_some()
                    || self.tables.iter().any(|t| t.hooks.is_some())
                    || !self.custom_routes.is_empty()
                    || !self.schedules.is_empty()
                    || self
                        .subscriptions
                        .iter()
                        .any(|s| matches!(s.handler_type, HandlerType::Wasm))
            }
        }
    }

    /// 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());
            if let Some(reducer) = route.sse.as_ref().and_then(|sse| sse.reducer.as_deref()) {
                handlers.push(reducer);
            }
        }
        for sub in &self.subscriptions {
            if matches!(sub.handler_type, HandlerType::Wasm) {
                handlers.push(sub.handler.as_str());
            }
        }
        handlers.extend(self.schedules.iter().map(|s| s.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
    }

    /// The `custom_routes` entry exposing `handler` over the HTTP `_fn/`
    /// surface, if any. Only custom routes are HTTP-invocable; subscription,
    /// `on_migrate`, and table-hook handlers run solely via their internal
    /// triggers, so a handler absent here must not be dispatched from an
    /// inbound request. Matched by name alone — the `_fn/` route accepts every
    /// method, so a verb-aware match would demote a job handler to inline
    /// execution on the wrong verb.
    pub fn http_route(&self, handler: &str) -> Option<&CustomRouteDefinition> {
        self.http_route_index(handler)
            .and_then(|index| self.custom_routes.get(index))
    }

    /// The position of [`http_route`](Self::http_route) in `custom_routes`.
    pub fn http_route_index(&self, handler: &str) -> Option<usize> {
        self.custom_routes.iter().position(|r| r.handler == handler)
    }

    /// Whether this manifest was built by a posture-aware SDK. True when any
    /// custom route carries a posture — the current SDK emits a posture on
    /// every route, so a posture-aware manifest is all-or-nothing and a lone
    /// undeclared route among declared ones signals tampering or a bug.
    pub fn is_posture_aware(&self) -> bool {
        self.custom_routes.iter().any(|r| r.posture.is_some())
    }

    /// The service's Postgres schema: `schema_name`, or the service name
    /// with `-` as `_`. Every other derivation of it is a bug.
    pub fn schema(&self) -> String {
        service_schema(&self.name, self.schema_name.as_deref())
    }

    /// Where the service's applied SQL migrations are recorded.
    pub fn migration_history(&self) -> MigrationHistory {
        self.migration_history.clone().unwrap_or_default()
    }

    /// The manifest without the text of its SQL migrations, which only a
    /// deploy reads.
    pub fn without_migration_bodies(mut self) -> Self {
        self.migrations
            .iter_mut()
            .for_each(|migration| migration.sql = String::new());
        self
    }

    /// Platform features required by service-level fields.
    pub(crate) fn service_features(&self) -> impl Iterator<Item = &'static str> {
        let adopted = !self.schema.is_managed() || self.schema_name.is_some();
        [
            (self.routing == RoutingMode::Strict).then_some(features::ROUTE_MATCHING),
            self.errors.is_some().then_some(features::ERROR_PROFILE),
            (!self.consumers.is_legacy()).then_some(features::NATS_CONSUMER_CONFIG),
            (!self.streams.is_empty()).then_some(features::STREAMS),
            (!self.schedules.is_empty()).then_some(features::SCHEDULES),
            adopted.then_some(features::SCHEMA_ADOPTED),
            (!self.migrations.is_empty()).then_some(features::SQL_MIGRATIONS),
            (!self.extensions.is_empty()).then_some(features::EXT_CALL),
            (self.mode == ServiceMode::Native).then_some(features::NATIVE_SERVICES),
            (!self.private_tables.is_empty()).then_some(features::PRIVATE_TABLES),
        ]
        .into_iter()
        .flatten()
    }

    /// SHA-256 of the whole manifest: the deploy skip check compares this, so
    /// a change to any field (`publishes`, `events`, `mode`, …) deploys.
    pub fn deploy_hash(&self) -> serde_json::Result<String> {
        use sha2::{Digest, Sha256};
        serde_json::to_vec(self).map(|bytes| hex::encode(Sha256::digest(bytes)))
    }

    /// SHA-256 of the schema-bearing fields, stored on schema snapshots. A
    /// field added since contributes only when it is set, so the hash of a
    /// manifest without it never changes.
    pub fn schema_hash(&self) -> serde_json::Result<String> {
        use sha2::{Digest, Sha256};
        let ownership = [
            (!self.schema.is_managed())
                .then(|| serde_json::to_string(&self.schema))
                .transpose()?,
            self.schema_name
                .as_ref()
                .map(serde_json::to_string)
                .transpose()?,
            (!self.migrations.is_empty())
                .then(|| serde_json::to_string(&self.migrations))
                .transpose()?,
            self.migration_history
                .as_ref()
                .map(serde_json::to_string)
                .transpose()?,
        ];
        let private_tables = (!self.private_tables.is_empty())
            .then(|| serde_json::to_string(&self.private_tables))
            .transpose()?;
        let digest = [
            serde_json::to_string(&self.tables)?,
            serde_json::to_string(&self.custom_routes)?,
            serde_json::to_string(&self.subscriptions)?,
            serde_json::to_string(&self.authorization)?,
            serde_json::to_string(&self.on_migrate)?,
        ]
        .into_iter()
        .chain(ownership.into_iter().flatten())
        .chain(private_tables)
        .fold(Sha256::new(), |hasher, part| hasher.chain_update(part))
        .finalize();
        Ok(hex::encode(digest))
    }
}

/// The Postgres schema of the service `name`: the `schema_name` it
/// declares, or its name with `-` as `_`.
pub fn service_schema(name: &str, schema_name: Option<&str>) -> String {
    schema_name.map_or_else(|| name.replace('-', "_"), str::to_string)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::manifest::fixtures::sample_manifest;
    use crate::manifest::{
        ColumnDefinition, ColumnType, ForeignKeyAction, JobConfig, RoutePosture, TableHooks,
    };

    #[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 routing_is_serialized_only_when_strict() {
        let mut manifest = sample_manifest();
        let json = serde_json::to_value(&manifest).unwrap();
        assert!(json.get("routing").is_none(), "{json}");
        manifest.routing = RoutingMode::Strict;
        let json = serde_json::to_value(&manifest).unwrap();
        assert_eq!(json["routing"], "strict");
        let parsed: ServiceManifest = serde_json::from_value(json).unwrap();
        assert_eq!(parsed.routing, RoutingMode::Strict);
    }

    #[test]
    fn error_profile_keys_are_snake_case_identifiers() {
        for key in ["error_message", "_e", "e1", &"e".repeat(64)] {
            assert!(ErrorProfile::is_valid_key(key), "{key}");
        }
        for key in [
            "",
            "1e",
            "Error",
            "error-message",
            "error message",
            &"e".repeat(65),
        ] {
            assert!(!ErrorProfile::is_valid_key(key), "{key}");
        }
        let mut manifest = sample_manifest();
        assert!(serde_json::to_value(&manifest)
            .unwrap()
            .get("errors")
            .is_none());
        manifest.errors = Some(ErrorProfile {
            key: "error_message".into(),
        });
        assert_eq!(
            serde_json::to_value(&manifest).unwrap()["errors"],
            serde_json::json!({"key": "error_message"})
        );
    }

    #[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_schema_hash_deterministic() {
        let m1 = sample_manifest();
        let m2 = sample_manifest();
        assert_eq!(
            m1.schema_hash().expect("schema hash"),
            m2.schema_hash().expect("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().expect("schema hash"),
            m2.schema_hash().expect("schema hash")
        );
    }

    #[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_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_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 deploy_hash_covers_fields_schema_hash_ignores() {
        let m1 = sample_manifest();
        let mut m2 = sample_manifest();
        let hash = |m: &ServiceManifest| m.deploy_hash().expect("hash");
        assert_eq!(hash(&m1), hash(&m2));
        m2.publishes = Some(vec!["acme.orders.>".into()]);
        assert_ne!(hash(&m1), hash(&m2));
        assert_eq!(
            m1.schema_hash().expect("schema hash"),
            m2.schema_hash().expect("schema hash")
        );
    }

    #[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().expect("schema hash"),
            m2.schema_hash().expect("schema hash")
        );
    }

    #[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 a_mode_is_named_as_a_manifest_names_it() {
        for (mode, on_platform) in [
            (ServiceMode::Crud, true),
            (ServiceMode::Wasm, true),
            (ServiceMode::Container, true),
            (ServiceMode::Web, false),
            (ServiceMode::Native, false),
        ] {
            assert_eq!(serde_json::to_value(&mode).unwrap(), mode.as_str());
            assert_eq!(mode.runs_on_platform(), on_platform, "{}", mode.as_str());
        }
    }

    #[test]
    fn a_native_service_runs_no_wasm_and_demands_native_services() {
        let mut m = sample_manifest();
        m.on_migrate = Some("handle_on_migrate".into());
        m.mode = ServiceMode::Native;
        assert!(!m.requires_wasm());
        assert!(m.derived_features().contains(features::NATIVE_SERVICES));
        let json = serde_json::to_value(&m).unwrap();
        assert_eq!(json["mode"], "native");
        let parsed: ServiceManifest = serde_json::from_value(json).unwrap();
        assert_eq!(parsed.mode, ServiceMode::Native);
    }

    #[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,
            posture: None,
            forward_auth: false,
            ..Default::default()
        });
        assert!(m.requires_wasm());
    }

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

    #[test]
    fn test_requires_wasm_false_for_non_wasm_subscription() {
        let mut m = sample_manifest();
        m.subscriptions.push(SubscriptionDefinition::new(
            "events.foo",
            "on_foo",
            HandlerType::Webhook,
        ));
        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,
            posture: None,
            forward_auth: false,
            ..Default::default()
        });
        m.subscriptions
            .push(SubscriptionDefinition::new("x", "sub_h", HandlerType::Wasm));
        m.subscriptions.push(SubscriptionDefinition::new(
            "y",
            "wh_ignored",
            HandlerType::Webhook,
        ));
        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 http_route_exposes_only_custom_routes() {
        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,
            posture: None,
            forward_auth: false,
            ..Default::default()
        });
        m.subscriptions.push(SubscriptionDefinition::new(
            "evt.sub",
            "sub_h",
            HandlerType::Wasm,
        ));
        m.tables[0].hooks = Some(TableHooks {
            after_create: Some("hook_h".into()),
            ..Default::default()
        });

        assert_eq!(
            m.http_route("route_h").map(|r| r.handler.as_str()),
            Some("route_h")
        );
        assert!(m.http_route("sub_h").is_none());
        assert!(m.http_route("on_mig").is_none());
        assert!(m.http_route("hook_h").is_none());
        assert!(m.http_route("unknown").is_none());
    }

    #[test]
    fn http_route_allows_handler_declared_as_both_route_and_subscription() {
        let mut m = sample_manifest();
        m.custom_routes.push(CustomRouteDefinition {
            method: "POST".into(),
            path: "/dual".into(),
            handler: "dual".into(),
            job: Some(JobConfig::default()),
            posture: None,
            forward_auth: false,
            ..Default::default()
        });
        m.subscriptions.push(SubscriptionDefinition::new(
            "evt.dual",
            "dual",
            HandlerType::Wasm,
        ));

        let route = m
            .http_route("dual")
            .expect("dual-purpose handler stays HTTP-invocable");
        assert!(route.job.is_some());
    }

    #[test]
    fn posture_aware_detects_any_declared_route() {
        let mut m = sample_manifest();
        assert!(!m.is_posture_aware());
        m.custom_routes.push(CustomRouteDefinition {
            method: "GET".into(),
            path: "/a".into(),
            handler: "a".into(),
            job: None,
            posture: None,
            forward_auth: false,
            ..Default::default()
        });
        assert!(!m.is_posture_aware());
        m.custom_routes.push(CustomRouteDefinition {
            method: "GET".into(),
            path: "/b".into(),
            handler: "b".into(),
            job: None,
            posture: Some(RoutePosture::Authenticated),
            forward_auth: false,
            ..Default::default()
        });
        assert!(m.is_posture_aware());
    }

    #[test]
    fn manifest_publishes_and_features_default_absent() {
        let json = serde_json::to_value(sample_manifest()).unwrap();
        assert!(json.get("publishes").is_none());
        assert!(json.get("required_platform_features").is_none());
        let legacy = r#"{"name":"s","tables":[],"mode":"crud"}"#;
        let parsed: ServiceManifest = serde_json::from_str(legacy).unwrap();
        assert!(parsed.publishes.is_none());
        assert!(parsed.required_platform_features.is_empty());
    }

    #[test]
    fn the_schema_is_the_override_or_the_name_with_underscores() {
        let mut m = sample_manifest();
        for (name, schema) in [
            ("asset-service", "asset_service"),
            ("note_service", "note_service"),
            ("Blog-Service", "Blog_Service"),
            ("a-b-c", "a_b_c"),
        ] {
            m.name = name.into();
            assert_eq!(m.schema(), schema);
            assert_eq!(service_schema(name, None), schema);
            assert_eq!(service_schema(name, Some("notes")), "notes");
        }
        m.schema_name = Some("notes".into());
        assert_eq!(m.schema(), "notes");
    }

    #[test]
    fn migration_bodies_are_stripped_and_nothing_else() {
        let mut m = sample_manifest();
        m.migrations = vec![
            SqlMigration {
                version: 1,
                name: "initial_tables".into(),
                sql: "CREATE TABLE notes ();".into(),
                no_transaction: false,
            },
            SqlMigration {
                version: 6,
                name: "add_note_kind".into(),
                sql: "-- cufflink:no-transaction\nSELECT 1;".into(),
                no_transaction: true,
            },
        ];
        let stripped = m.clone().without_migration_bodies();
        assert!(stripped.migrations.iter().all(|m| m.sql.is_empty()));
        let shape = |m: &ServiceManifest| -> Vec<(u32, String, bool)> {
            m.migrations
                .iter()
                .map(|m| (m.version, m.name.clone(), m.no_transaction))
                .collect()
        };
        assert_eq!(shape(&stripped), shape(&m));
        assert_eq!(stripped.name, m.name);
        assert_eq!(stripped.tables.len(), m.tables.len());
    }

    #[test]
    fn the_history_defaults_to_refinerys_table() {
        let mut m = sample_manifest();
        assert_eq!(m.migration_history(), MigrationHistory::default());
        let declared = MigrationHistory {
            table: "schema_history".into(),
            baseline: Some(6),
            ..Default::default()
        };
        m.migration_history = Some(declared.clone());
        assert_eq!(m.migration_history(), declared);
    }

    #[test]
    fn schema_ownership_changes_the_schema_hash_only_when_set() {
        let base = sample_manifest();
        let hash = |m: &ServiceManifest| m.schema_hash().expect("schema hash");
        let changed: Vec<ServiceManifest> = vec![
            ServiceManifest {
                schema: SchemaOwnership::Adopted,
                ..base.clone()
            },
            ServiceManifest {
                schema_name: Some("notes".into()),
                ..base.clone()
            },
            ServiceManifest {
                migrations: vec![SqlMigration {
                    version: 1,
                    name: "a".into(),
                    sql: "SELECT 1;".into(),
                    no_transaction: false,
                }],
                ..base.clone()
            },
            ServiceManifest {
                migration_history: Some(MigrationHistory::default()),
                ..base.clone()
            },
        ];
        let hashes: std::collections::BTreeSet<String> = changed.iter().map(hash).collect();
        assert_eq!(hashes.len(), changed.len());
        assert!(!hashes.contains(&hash(&base)));
    }

    #[test]
    fn a_schedule_runs_user_code() {
        let mut m = sample_manifest();
        m.schedules = serde_json::from_value(serde_json::json!([
            {"name": "order_retry", "handler": "retry_orders", "spec": {"kind": "every", "secs": 30}}
        ]))
        .unwrap();
        assert!(m.requires_wasm());
        assert_eq!(m.declared_wasm_handlers(), ["retry_orders"]);
        assert!(m.http_route("retry_orders").is_none());
    }

    fn private(name: &str) -> TableDefinition {
        TableDefinition {
            name: name.into(),
            columns: vec![ColumnDefinition {
                name: "body".into(),
                column_type: ColumnType::Text,
                ..Default::default()
            }],
            managed: false,
            ..Default::default()
        }
    }

    #[test]
    fn declared_tables_chains_both_lists() {
        let mut m = sample_manifest();
        let names = |m: &ServiceManifest| -> Vec<String> {
            m.declared_tables().map(|t| t.name.clone()).collect()
        };
        assert_eq!(names(&m), ["todos"]);
        m.private_tables = vec![private("notes"), private("audit_log")];
        assert_eq!(names(&m), ["todos", "notes", "audit_log"]);
        m.tables.clear();
        assert_eq!(names(&m), ["notes", "audit_log"]);
    }

    #[test]
    fn get_table_does_not_find_a_private_table() {
        let mut m = sample_manifest();
        m.private_tables = vec![private("notes")];
        assert!(m.get_table("todos").is_some());
        assert!(m.get_table("notes").is_none());
    }

    #[test]
    fn a_reader_without_the_key_sees_no_private_table() {
        let mut m = sample_manifest();
        m.private_tables = vec![private("notes")];
        let mut wire = serde_json::to_value(&m).unwrap();
        assert_eq!(wire["private_tables"][0]["name"], "notes");
        wire.as_object_mut().unwrap().remove("private_tables");
        let older: ServiceManifest = serde_json::from_value(wire).unwrap();
        assert!(older.private_tables.is_empty());
        assert!(older.get_table("notes").is_none());
        assert_eq!(older.declared_tables().count(), 1);
    }

    #[test]
    fn private_tables_join_the_schema_hash_last_and_only_when_declared() {
        use sha2::{Digest, Sha256};
        let base = sample_manifest();
        let mut declaring = base.clone();
        declaring.private_tables = vec![private("notes")];
        let parts = |m: &ServiceManifest| {
            [
                serde_json::to_string(&m.tables).unwrap(),
                serde_json::to_string(&m.custom_routes).unwrap(),
                serde_json::to_string(&m.subscriptions).unwrap(),
                serde_json::to_string(&m.authorization).unwrap(),
                serde_json::to_string(&m.on_migrate).unwrap(),
            ]
            .concat()
        };
        let hash = |text: String| hex::encode(Sha256::digest(text));
        assert_eq!(base.schema_hash().unwrap(), hash(parts(&base)));
        assert_eq!(
            declaring.schema_hash().unwrap(),
            hash(parts(&declaring) + &serde_json::to_string(&declaring.private_tables).unwrap())
        );
        let mut moved = declaring.clone();
        moved.private_tables[0].name = "memos".into();
        assert_ne!(
            moved.schema_hash().unwrap(),
            declaring.schema_hash().unwrap()
        );
    }
}