traits.rs

// Copyright Motia LLC and/or licensed to Motia LLC under one or more
// contributor license agreements. Licensed under the Elastic License 2.0;
// you may not use this file except in compliance with the Elastic License 2.0.
// This software is patent protected. We welcome discussions - reach out at team@iii.dev
// See LICENSE and PATENTS files for details.

use std::{
    collections::HashMap,
    future::Future,
    pin::Pin,
    sync::{Arc, RwLock},
};

use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde_json::Value;

use crate::{
    engine::Engine,
    workers::registry::{AdapterRegistrationEntry, WorkerFuture},
};

// use across modules
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct AdapterEntry {
    /// Adapter class name (e.g. `kv`, `redis`, `bridge`).
    pub name: String,
    /// Adapter-specific configuration. Free-form so adapters stay pluggable —
    /// the schema does not constrain its shape. `skip_serializing_if` omits the
    /// key when absent rather than emitting `config: null`: a worker that
    /// advertises a closed per-adapter schema (a `oneOf` whose branches `$ref`
    /// an object `config`) would otherwise reject its own `{name}`-only seed,
    /// because `null` matches no branch.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub config: Option<Value>,
}

pub(crate) fn bind_address_error(
    addr: impl std::fmt::Display,
    err: std::io::Error,
) -> anyhow::Error {
    let addr = addr.to_string();

    if err.kind() == std::io::ErrorKind::AddrInUse {
        tracing::error!("address {} is already in use", addr);
        anyhow::anyhow!("address {} is already in use", addr)
    } else {
        tracing::error!(address = %addr, error = %err, "failed to bind address");
        anyhow::Error::new(err).context(format!("failed to bind to {}", addr))
    }
}

#[async_trait::async_trait]
pub trait Worker: Send + Sync {
    fn name(&self) -> &'static str;
    async fn create(engine: Arc<Engine>, config: Option<Value>) -> anyhow::Result<Box<dyn Worker>>
    where
        Self: Sized;

    fn make_worker(engine: Arc<Engine>, config: Option<Value>) -> WorkerFuture
    where
        Self: Sized + 'static,
    {
        Self::create(engine, config)
    }

    /// Initializes the module
    async fn initialize(&self) -> anyhow::Result<()>;

    async fn start_background_tasks(
        &self,
        _shutdown_rx: tokio::sync::watch::Receiver<bool>,
        _shutdown_tx: tokio::sync::watch::Sender<bool>,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    async fn destroy(&self) -> anyhow::Result<()> {
        tracing::info!("Destroying module: {}", self.name());
        Ok(())
    }

    /// Reports whether the backing process/state for this worker is still
    /// alive. Used by the reloader to detect workers whose external VM or
    /// child process died without the engine noticing — those get promoted
    /// from `unchanged` to `changed` on the next config reload so they get
    /// rebooted instead of being skipped.
    ///
    /// Default is `true`: built-in, in-process workers are alive as long as
    /// the engine is alive. Override for workers that track out-of-process
    /// state (e.g. detached VMs via pidfile).
    async fn is_alive(&self) -> bool {
        true
    }

    fn is_external_process(&self) -> bool {
        false
    }

    /// Registers functions to the engine
    #[allow(unused_variables)]
    fn register_functions(&self, engine: Arc<Engine>) {
        // blank implementation since it going to be overriden by the macros
    }
}

pub type AdapterFactory<A> = Arc<
    dyn Fn(
            Arc<Engine>,
            Option<Value>,
        ) -> Pin<Box<dyn Future<Output = anyhow::Result<Arc<A>>> + Send>>
        + Send
        + Sync,
>;

#[async_trait::async_trait]
pub trait ConfigurableWorker: Worker + Sized + 'static {
    type Config: DeserializeOwned + Default + Send;
    type Adapter: Send + Sync + 'static + ?Sized;
    type AdapterRegistration: AdapterRegistrationEntry<Self::Adapter> + inventory::Collect;
    const DEFAULT_ADAPTER_NAME: &'static str;

    async fn register_adapter(
        name: impl Into<String> + Send,
        factory: AdapterFactory<Self::Adapter>,
    ) {
        let registry = Self::registry().await;
        let mut reg = registry.write().unwrap();
        reg.insert(name.into(), factory);
    }

    /// Build the registry map with adapter factories from the inventory registry.
    fn build_registry() -> HashMap<String, AdapterFactory<Self::Adapter>> {
        let mut registry = HashMap::new();
        for registration in inventory::iter::<Self::AdapterRegistration> {
            let factory = registration.factory();
            let adapter_factory: AdapterFactory<Self::Adapter> =
                Arc::new(move |engine, config| (factory)(engine, config));
            registry.insert(registration.name().to_string(), adapter_factory);
        }
        registry
    }

    /// Get the static registry. This method should be implemented by creating a static Lazy
    /// that calls `Self::build_registry()`. Example:
    /// ```ignore
    /// async fn registry() -> &'static RwLock<HashMap<String, AdapterFactory<Self::Adapter>>> {
    ///     static REGISTRY: Lazy<RwLock<HashMap<String, AdapterFactory<dyn MyAdapter>>>> =
    ///         Lazy::new(|| RwLock::new(MyWorker::build_registry()));
    ///     &REGISTRY
    /// }
    /// ```
    async fn registry() -> &'static RwLock<HashMap<String, AdapterFactory<Self::Adapter>>>;
    /// Build the module from parsed config and adapter
    fn build(engine: Arc<Engine>, config: Self::Config, adapter: Arc<Self::Adapter>) -> Self;

    /// Default adapter class name
    fn default_adapter_name() -> &'static str {
        Self::DEFAULT_ADAPTER_NAME
    }

    /// Extract adapter class from config (optional override)
    fn adapter_name_from_config(_config: &Self::Config) -> Option<String> {
        None
    }

    /// Extract adapter config from module config (optional override)
    fn adapter_config_from_config(_config: &Self::Config) -> Option<Value> {
        None
    }

    async fn get_adapter(name: &str) -> Option<AdapterFactory<Self::Adapter>> {
        let registry = Self::registry().await;
        let map = registry.read().unwrap();
        map.get(name).cloned()
    }

    /// Helper function to create an adapter factory from a closure
    fn make_adapter_factory<F, Fut>(create_fn: F) -> AdapterFactory<Self::Adapter>
    where
        F: Fn(Arc<Engine>, Option<Value>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = anyhow::Result<Arc<Self::Adapter>>> + Send + 'static,
    {
        Arc::new(move |engine, config| Box::pin(create_fn(engine, config)))
    }

    /// Helper function to register a new adapter factory from a closure.
    /// This is a convenience method that combines `make_adapter_factory` and `register_adapter`.
    ///
    /// # Example
    /// ```ignore
    /// QueueCoreWorker::add_adapter("my_adapter", |engine, config| async move {
    ///     Ok(Arc::new(MyAdapter::new(engine).await?) as Arc<dyn QueueAdapter>)
    /// }).await;
    /// ```
    async fn add_adapter<F, Fut>(name: impl Into<String> + Send, create_fn: F) -> anyhow::Result<()>
    where
        F: Fn(Arc<Engine>, Option<Value>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = anyhow::Result<Arc<Self::Adapter>>> + Send + 'static,
    {
        let factory = Self::make_adapter_factory(create_fn);
        Self::register_adapter(name, factory).await;
        Ok(())
    }
    /// Create with typed adapters
    async fn create_with_adapters(
        engine: Arc<Engine>,
        config: Option<Value>,
    ) -> anyhow::Result<Box<dyn Worker>> {
        // 1. Parse config
        let parsed_config: Self::Config = config
            .map(serde_json::from_value)
            .transpose()?
            .unwrap_or_default();

        // 2. Determine which adapter to use
        let adapter_name = match Self::adapter_name_from_config(&parsed_config) {
            Some(class) => class,
            None => {
                tracing::debug!(
                    "No adapter class specified in config, using default: '{}'",
                    Self::default_adapter_name()
                );
                Self::default_adapter_name().to_string()
            }
        };
        // 3. Get the factory
        let factory = match Self::get_adapter(&adapter_name).await {
            Some(factory) => factory,
            None => {
                let registry = Self::registry().await;
                let available: Vec<String> = registry.read().unwrap().keys().cloned().collect();
                return Err(anyhow::anyhow!(
                    "Adapter factory '{}' not found. Available: {:?}",
                    adapter_name,
                    available
                ));
            }
        };

        // 4. Create adapter
        let adapter_config = Self::adapter_config_from_config(&parsed_config);
        tracing::debug!(
            "Using adapter class '{}' with config: {:?}",
            adapter_name,
            &adapter_config
        );
        let adapter = factory(engine.clone(), adapter_config).await?;

        // 5. Build module
        Ok(Box::new(Self::build(engine, parsed_config, adapter)))
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Mutex;

    use once_cell::sync::Lazy;
    use serial_test::serial;

    use super::*;
    use crate::workers::registry::{AdapterFuture, AdapterRegistration};

    trait TestAdapter: Send + Sync {
        fn adapter_name(&self) -> String;
        fn adapter_config(&self) -> Option<Value>;
    }

    type TestAdapterRegistration = AdapterRegistration<dyn TestAdapter>;
    inventory::collect!(TestAdapterRegistration);

    fn inventory_test_factory(
        _engine: Arc<Engine>,
        config: Option<Value>,
    ) -> AdapterFuture<dyn TestAdapter> {
        Box::pin(async move {
            Ok(Arc::new(RecordingAdapter {
                name: "inventory".to_string(),
                config,
            }) as Arc<dyn TestAdapter>)
        })
    }

    inventory::submit! {
        TestAdapterRegistration {
            name: "test::inventory",
            factory: inventory_test_factory,
        }
    }

    #[derive(Debug)]
    struct RecordingAdapter {
        name: String,
        config: Option<Value>,
    }

    impl TestAdapter for RecordingAdapter {
        fn adapter_name(&self) -> String {
            self.name.clone()
        }

        fn adapter_config(&self) -> Option<Value> {
            self.config.clone()
        }
    }

    #[derive(Clone, Debug, Default, Deserialize, Serialize)]
    struct TestWorkerConfig {
        #[serde(default)]
        adapter: Option<AdapterEntry>,
    }

    #[derive(Clone)]
    struct TestConfigurableWorker {
        _engine: Arc<Engine>,
        _config: TestWorkerConfig,
        _adapter: Arc<dyn TestAdapter>,
    }

    type FactoryCall = (String, Option<Value>);

    static FACTORY_CALLS: Lazy<Mutex<Vec<FactoryCall>>> = Lazy::new(|| Mutex::new(Vec::new()));

    #[async_trait::async_trait]
    impl Worker for TestConfigurableWorker {
        fn name(&self) -> &'static str {
            "TestConfigurableWorker"
        }

        async fn create(
            engine: Arc<Engine>,
            config: Option<Value>,
        ) -> anyhow::Result<Box<dyn Worker>> {
            Self::create_with_adapters(engine, config).await
        }

        async fn initialize(&self) -> anyhow::Result<()> {
            Ok(())
        }
    }

    #[async_trait::async_trait]
    impl ConfigurableWorker for TestConfigurableWorker {
        type Config = TestWorkerConfig;
        type Adapter = dyn TestAdapter;
        type AdapterRegistration = TestAdapterRegistration;
        const DEFAULT_ADAPTER_NAME: &'static str = "test::default";

        async fn registry() -> &'static RwLock<HashMap<String, AdapterFactory<Self::Adapter>>> {
            static REGISTRY: Lazy<RwLock<HashMap<String, AdapterFactory<dyn TestAdapter>>>> =
                Lazy::new(|| RwLock::new(HashMap::new()));
            &REGISTRY
        }

        fn build(engine: Arc<Engine>, config: Self::Config, adapter: Arc<Self::Adapter>) -> Self {
            Self {
                _engine: engine,
                _config: config,
                _adapter: adapter,
            }
        }

        fn adapter_name_from_config(config: &Self::Config) -> Option<String> {
            config.adapter.as_ref().map(|entry| entry.name.clone())
        }

        fn adapter_config_from_config(config: &Self::Config) -> Option<Value> {
            config
                .adapter
                .as_ref()
                .and_then(|entry| entry.config.clone())
        }
    }

    struct SimpleWorker;

    #[async_trait::async_trait]
    impl Worker for SimpleWorker {
        fn name(&self) -> &'static str {
            "SimpleWorker"
        }

        async fn create(
            _engine: Arc<Engine>,
            _config: Option<Value>,
        ) -> anyhow::Result<Box<dyn Worker>> {
            Ok(Box::new(SimpleWorker))
        }

        async fn initialize(&self) -> anyhow::Result<()> {
            Ok(())
        }
    }

    #[derive(Clone, Debug, Default, Deserialize, Serialize)]
    struct DefaultHooksConfig {
        enabled: bool,
    }

    #[derive(Clone)]
    struct DefaultHooksWorker {
        _engine: Arc<Engine>,
        _config: DefaultHooksConfig,
        _adapter: Arc<dyn TestAdapter>,
    }

    #[async_trait::async_trait]
    impl Worker for DefaultHooksWorker {
        fn name(&self) -> &'static str {
            "DefaultHooksWorker"
        }

        async fn create(
            engine: Arc<Engine>,
            config: Option<Value>,
        ) -> anyhow::Result<Box<dyn Worker>> {
            Self::create_with_adapters(engine, config).await
        }

        async fn initialize(&self) -> anyhow::Result<()> {
            Ok(())
        }
    }

    #[async_trait::async_trait]
    impl ConfigurableWorker for DefaultHooksWorker {
        type Config = DefaultHooksConfig;
        type Adapter = dyn TestAdapter;
        type AdapterRegistration = TestAdapterRegistration;
        const DEFAULT_ADAPTER_NAME: &'static str = "test::inventory";

        async fn registry() -> &'static RwLock<HashMap<String, AdapterFactory<Self::Adapter>>> {
            static REGISTRY: Lazy<RwLock<HashMap<String, AdapterFactory<dyn TestAdapter>>>> =
                Lazy::new(|| RwLock::new(DefaultHooksWorker::build_registry()));
            &REGISTRY
        }

        fn build(engine: Arc<Engine>, config: Self::Config, adapter: Arc<Self::Adapter>) -> Self {
            Self {
                _engine: engine,
                _config: config,
                _adapter: adapter,
            }
        }
    }

    async fn clear_test_registry() {
        TestConfigurableWorker::registry()
            .await
            .write()
            .unwrap()
            .clear();
        FACTORY_CALLS.lock().expect("lock factory calls").clear();
    }

    fn recording_factory(name: &'static str) -> AdapterFactory<dyn TestAdapter> {
        TestConfigurableWorker::make_adapter_factory(move |_engine, config| async move {
            FACTORY_CALLS
                .lock()
                .expect("lock factory calls")
                .push((name.to_string(), config.clone()));
            Ok(Arc::new(RecordingAdapter {
                name: name.to_string(),
                config,
            }) as Arc<dyn TestAdapter>)
        })
    }

    // =========================================================================
    // AdapterEntry serialization/deserialization
    // =========================================================================

    #[test]
    fn adapter_entry_deserialize_with_config() {
        let json = r#"{"name": "my::Adapter", "config": {"key": "value"}}"#;
        let entry: AdapterEntry = serde_json::from_str(json).unwrap();
        assert_eq!(entry.name, "my::Adapter");
        assert!(entry.config.is_some());
        assert_eq!(entry.config.unwrap()["key"], "value");
    }

    #[test]
    fn adapter_entry_deserialize_without_config() {
        let json = r#"{"name": "my::Adapter"}"#;
        let entry: AdapterEntry = serde_json::from_str(json).unwrap();
        assert_eq!(entry.name, "my::Adapter");
        assert!(entry.config.is_none());
    }

    #[test]
    fn adapter_entry_serialize_roundtrip() {
        let entry = AdapterEntry {
            name: "test::Adapter".to_string(),
            config: Some(serde_json::json!({"port": 8080})),
        };
        let json_str = serde_json::to_string(&entry).unwrap();
        let deserialized: AdapterEntry = serde_json::from_str(&json_str).unwrap();
        assert_eq!(deserialized.name, "test::Adapter");
        assert_eq!(deserialized.config.unwrap()["port"], 8080);
    }

    #[test]
    fn adapter_entry_serialize_no_config() {
        let entry = AdapterEntry {
            name: "test::Adapter".to_string(),
            config: None,
        };
        let json_str = serde_json::to_string(&entry).unwrap();
        let deserialized: AdapterEntry = serde_json::from_str(&json_str).unwrap();
        assert_eq!(deserialized.name, "test::Adapter");
        assert!(deserialized.config.is_none());
    }

    #[test]
    fn adapter_entry_clone() {
        let entry = AdapterEntry {
            name: "test::Adapter".to_string(),
            config: Some(serde_json::json!({"a": 1})),
        };
        let cloned = entry.clone();
        assert_eq!(cloned.name, "test::Adapter");
        assert_eq!(cloned.config, Some(serde_json::json!({"a": 1})));
    }

    #[test]
    fn adapter_entry_debug() {
        let entry = AdapterEntry {
            name: "test::Adapter".to_string(),
            config: None,
        };
        let debug_str = format!("{:?}", entry);
        assert!(debug_str.contains("test::Adapter"));
    }

    #[tokio::test]
    async fn module_default_trait_methods_work() {
        let engine = Arc::new(Engine::new());
        let module = SimpleWorker::make_worker(engine, None)
            .await
            .expect("make module should succeed");

        assert_eq!(module.name(), "SimpleWorker");

        let (tx, rx) = tokio::sync::watch::channel(false);
        module
            .start_background_tasks(rx, tx)
            .await
            .expect("default background tasks should succeed");
        module
            .destroy()
            .await
            .expect("default destroy should succeed");
    }

    #[tokio::test]
    async fn module_register_functions_default_is_noop() {
        let engine = Arc::new(Engine::new());
        let module = SimpleWorker;
        module.register_functions(engine.clone());
        assert!(engine.functions.get("missing").is_none());
    }

    #[tokio::test]
    async fn module_default_trait_methods_on_concrete_type_work() {
        let (tx, rx) = tokio::sync::watch::channel(false);
        let module = SimpleWorker;

        module
            .start_background_tasks(rx, tx)
            .await
            .expect("default background tasks should succeed");
        module
            .destroy()
            .await
            .expect("default destroy should succeed");
    }

    #[tokio::test]
    async fn make_adapter_factory_wraps_custom_closure() {
        let engine = Arc::new(Engine::new());
        let factory = TestConfigurableWorker::make_adapter_factory(|_engine, config| async move {
            Ok(Arc::new(RecordingAdapter {
                name: "inline".to_string(),
                config,
            }) as Arc<dyn TestAdapter>)
        });

        let adapter = factory(engine, Some(serde_json::json!({ "from": "factory" })))
            .await
            .expect("factory should create adapter");
        assert_eq!(adapter.adapter_name(), "inline");
        assert_eq!(
            adapter.adapter_config(),
            Some(serde_json::json!({ "from": "factory" }))
        );
    }

    #[tokio::test]
    #[serial]
    async fn register_and_get_adapter_work() {
        clear_test_registry().await;

        TestConfigurableWorker::register_adapter("test::default", recording_factory("default"))
            .await;

        assert_eq!(
            TestConfigurableWorker::default_adapter_name(),
            "test::default"
        );
        assert!(
            TestConfigurableWorker::get_adapter("test::default")
                .await
                .is_some()
        );
        assert!(
            TestConfigurableWorker::get_adapter("missing")
                .await
                .is_none()
        );
    }

    #[tokio::test]
    #[serial]
    async fn configurable_module_helper_defaults_are_accessible_directly() {
        clear_test_registry().await;

        let inventory_registry = DefaultHooksWorker::build_registry();
        assert!(inventory_registry.contains_key("test::inventory"));
        assert_eq!(
            DefaultHooksWorker::default_adapter_name(),
            "test::inventory"
        );
        assert!(
            DefaultHooksWorker::get_adapter("test::inventory")
                .await
                .is_some()
        );

        TestConfigurableWorker::add_adapter("test::default", |_engine, config| async move {
            Ok(Arc::new(RecordingAdapter {
                name: "added-direct".to_string(),
                config,
            }) as Arc<dyn TestAdapter>)
        })
        .await
        .expect("add adapter should succeed");

        let factory = TestConfigurableWorker::get_adapter("test::default")
            .await
            .expect("registered adapter factory");
        let adapter = factory(
            Arc::new(Engine::new()),
            Some(serde_json::json!({ "ok": true })),
        )
        .await
        .expect("factory should create adapter");
        assert_eq!(adapter.adapter_name(), "added-direct");
        assert_eq!(
            adapter.adapter_config(),
            Some(serde_json::json!({ "ok": true }))
        );
    }

    #[tokio::test]
    #[serial]
    async fn create_with_adapters_uses_default_and_custom_configured_adapter() {
        clear_test_registry().await;
        let engine = Arc::new(Engine::new());

        TestConfigurableWorker::register_adapter("test::default", recording_factory("default"))
            .await;
        TestConfigurableWorker::register_adapter("test::custom", recording_factory("custom")).await;

        let default_module = TestConfigurableWorker::create_with_adapters(engine.clone(), None)
            .await
            .expect("create with default adapter");
        assert_eq!(default_module.name(), "TestConfigurableWorker");

        let configured = TestWorkerConfig {
            adapter: Some(AdapterEntry {
                name: "test::custom".to_string(),
                config: Some(serde_json::json!({ "mode": "custom" })),
            }),
        };
        assert_eq!(
            TestConfigurableWorker::adapter_name_from_config(&configured).as_deref(),
            Some("test::custom")
        );
        assert_eq!(
            TestConfigurableWorker::adapter_config_from_config(&configured),
            Some(serde_json::json!({ "mode": "custom" }))
        );

        let custom_module = TestConfigurableWorker::create_with_adapters(
            engine,
            Some(serde_json::to_value(configured).expect("serialize test config")),
        )
        .await
        .expect("create with custom adapter");
        assert_eq!(custom_module.name(), "TestConfigurableWorker");

        let calls = FACTORY_CALLS.lock().expect("lock factory calls").clone();
        assert_eq!(
            calls,
            vec![
                ("default".to_string(), None),
                (
                    "custom".to_string(),
                    Some(serde_json::json!({ "mode": "custom" }))
                ),
            ]
        );
    }

    #[tokio::test]
    #[serial]
    async fn add_adapter_and_missing_adapter_error_are_reported() {
        clear_test_registry().await;
        let engine = Arc::new(Engine::new());

        TestConfigurableWorker::add_adapter("test::default", |_engine, config| async move {
            FACTORY_CALLS
                .lock()
                .expect("lock factory calls")
                .push(("added".to_string(), config.clone()));
            Ok(Arc::new(RecordingAdapter {
                name: "added".to_string(),
                config,
            }) as Arc<dyn TestAdapter>)
        })
        .await
        .expect("add adapter should succeed");

        let created = TestConfigurableWorker::create_with_adapters(engine.clone(), None)
            .await
            .expect("create with added adapter");
        assert_eq!(created.name(), "TestConfigurableWorker");

        let missing = TestConfigurableWorker::create_with_adapters(
            engine,
            Some(serde_json::json!({
                "adapter": { "name": "test::missing" }
            })),
        )
        .await;
        let error = missing.err().expect("missing adapter should fail");
        assert!(error.to_string().contains("test::missing"));
        assert!(error.to_string().contains("test::default"));
    }

    #[tokio::test]
    async fn build_registry_and_default_hooks_cover_inventory_path() {
        let registry = DefaultHooksWorker::build_registry();
        assert!(registry.contains_key("test::inventory"));

        let module = DefaultHooksWorker::create_with_adapters(
            Arc::new(Engine::new()),
            Some(serde_json::json!({ "enabled": true })),
        )
        .await
        .expect("create module from inventory-backed registry");
        assert_eq!(module.name(), "DefaultHooksWorker");
        assert_eq!(
            DefaultHooksWorker::adapter_name_from_config(&DefaultHooksConfig::default()),
            None
        );
        assert_eq!(
            DefaultHooksWorker::adapter_config_from_config(&DefaultHooksConfig::default()),
            None
        );
    }

    #[tokio::test]
    #[serial]
    async fn create_with_adapters_reports_parse_and_factory_errors() {
        clear_test_registry().await;

        TestConfigurableWorker::add_adapter("test::failing", |_engine, _config| async move {
            Err(anyhow::anyhow!("adapter factory failed"))
        })
        .await
        .expect("register failing adapter");

        let parse_error = match TestConfigurableWorker::create_with_adapters(
            Arc::new(Engine::new()),
            Some(serde_json::json!(["not", "an", "object"])),
        )
        .await
        {
            Ok(_) => panic!("invalid config shape should fail"),
            Err(err) => err,
        };
        assert!(parse_error.to_string().contains("invalid type"));

        let factory_error = match TestConfigurableWorker::create_with_adapters(
            Arc::new(Engine::new()),
            Some(serde_json::json!({
                "adapter": { "name": "test::failing" }
            })),
        )
        .await
        {
            Ok(_) => panic!("failing adapter should bubble up"),
            Err(err) => err,
        };
        assert!(factory_error.to_string().contains("adapter factory failed"));
    }
}