# Integration Patterns\n\n## Overview\n\nProvisioning implements sophisticated integration patterns to coordinate between its hybrid Rust/Nushell architecture, manage multi-provider\nworkflows, and enable extensible functionality. This document outlines the key integration patterns, their implementations, and best practices.\n\n## Core Integration Patterns\n\n### 1. Hybrid Language Integration\n\n#### Rust-to-Nushell Communication Pattern\n\n**Use Case**: Orchestrator invoking business logic operations\n\n**Implementation**:\n\n```\nuse tokio::process::Command;\nuse serde_json;\n\npub async fn execute_nushell_workflow(\n    workflow: &str,\n    args: &[String]\n) -> Result<WorkflowResult, Error> {\n    let mut cmd = Command::new("nu");\n    cmd.arg("-c")\n       .arg(format!("use core/nulib/workflows/{}.nu *; {}", workflow, args.join(" ")));\n\n    let output = cmd.output().await?;\n    let result: WorkflowResult = serde_json::from_slice(&output.stdout)?;\n    Ok(result)\n}\n```\n\n**Data Exchange Format**:\n\n```\n{\n    "status": "success" | "error" | "partial",\n    "result": {\n        "operation": "server_create",\n        "resources": ["server-001", "server-002"],\n        "metadata": { ... }\n    },\n    "error": null | { "code": "ERR001", "message": "..." },\n    "context": { "workflow_id": "wf-123", "step": 2 }\n}\n```\n\n#### Nushell-to-Rust Communication Pattern\n\n**Use Case**: Business logic submitting workflows to orchestrator\n\n**Implementation**:\n\n```\ndef submit-workflow [workflow: record] -> record {\n    let payload = $workflow | to json\n\n    http post "http://localhost:9090/workflows/submit" {\n        headers: { "Content-Type": "application/json" }\n        body: $payload\n    }\n    | from json\n}\n```\n\n**API Contract**:\n\n```\n{\n    "workflow_id": "wf-456",\n    "name": "multi_cloud_deployment",\n    "operations": [...],\n    "dependencies": { ... },\n    "configuration": { ... }\n}\n```\n\n### 2. Provider Abstraction Pattern\n\n#### Standard Provider Interface\n\n**Purpose**: Uniform API across different cloud providers\n\n**Interface Definition**:\n\n```\n# Standard provider interface that all providers must implement\nexport def list-servers [] -> table {\n    # Provider-specific implementation\n}\n\nexport def create-server [config: record] -> record {\n    # Provider-specific implementation\n}\n\nexport def delete-server [id: string] -> nothing {\n    # Provider-specific implementation\n}\n\nexport def get-server [id: string] -> record {\n    # Provider-specific implementation\n}\n```\n\n**Configuration Integration**:\n\n```\n[providers.aws]\nregion = "us-west-2"\ncredentials_profile = "default"\ntimeout = 300\n\n[providers.upcloud]\nzone = "de-fra1"\napi_endpoint = "https://api.upcloud.com"\ntimeout = 180\n\n[providers.local]\ndocker_socket = "/var/run/docker.sock"\nnetwork_mode = "bridge"\n```\n\n#### Provider Discovery and Loading\n\n```\ndef load-providers [] -> table {\n    let provider_dirs = glob "providers/*/nulib"\n\n    $provider_dirs\n    | each { |dir|\n        let provider_name = $dir | path basename | path dirname | path basename\n        let provider_config = get-provider-config $provider_name\n\n        {\n            name: $provider_name,\n            path: $dir,\n            config: $provider_config,\n            available: (test-provider-connectivity $provider_name)\n        }\n    }\n}\n```\n\n### 3. Configuration Resolution Pattern\n\n#### Hierarchical Configuration Loading\n\n**Implementation**:\n\n```\ndef resolve-configuration [context: record] -> record {\n    let base_config = open config.defaults.toml\n    let user_config = if ("config.user.toml" | path exists) {\n        open config.user.toml\n    } else { {} }\n\n    let env_config = if ($env.PROVISIONING_ENV? | is-not-empty) {\n        let env_file = $"config.($env.PROVISIONING_ENV).toml"\n        if ($env_file | path exists) { open $env_file } else { {} }\n    } else { {} }\n\n    let merged_config = $base_config\n    | merge $user_config\n    | merge $env_config\n    | merge ($context.runtime_config? | default {})\n\n    interpolate-variables $merged_config\n}\n```\n\n#### Variable Interpolation Pattern\n\n```\ndef interpolate-variables [config: record] -> record {\n    let interpolations = {\n        "{{paths.base}}": ($env.PWD),\n        "{{env.HOME}}": ($env.HOME),\n        "{{now.date}}": (date now | format date "%Y-%m-%d"),\n        "{{git.branch}}": (git branch --show-current | str trim)\n    }\n\n    $config\n    | to json\n    | str replace --all "{{paths.base}}" $interpolations."{{paths.base}}"\n    | str replace --all "{{env.HOME}}" $interpolations."{{env.HOME}}"\n    | str replace --all "{{now.date}}" $interpolations."{{now.date}}"\n    | str replace --all "{{git.branch}}" $interpolations."{{git.branch}}"\n    | from json\n}\n```\n\n### 4. Workflow Orchestration Patterns\n\n#### Dependency Resolution Pattern\n\n**Use Case**: Managing complex workflow dependencies\n\n**Implementation (Rust)**:\n\n```\nuse petgraph::{Graph, Direction};\nuse std::collections::HashMap;\n\npub struct DependencyResolver {\n    graph: Graph<String, ()>,\n    node_map: HashMap<String, petgraph::graph::NodeIndex>,\n}\n\nimpl DependencyResolver {\n    pub fn resolve_execution_order(&self) -> Result<Vec<String>, Error> {\n        let mut topo = petgraph::algo::toposort(&self.graph, None)\n            .map_err(|_| Error::CyclicDependency)?;\n\n        Ok(topo.into_iter()\n            .map(|idx| self.graph[idx].clone())\n            .collect())\n    }\n\n    pub fn add_dependency(&mut self, from: &str, to: &str) {\n        let from_idx = self.get_or_create_node(from);\n        let to_idx = self.get_or_create_node(to);\n        self.graph.add_edge(from_idx, to_idx, ());\n    }\n}\n```\n\n#### Parallel Execution Pattern\n\n```\nuse tokio::task::JoinSet;\nuse futures::stream::{FuturesUnordered, StreamExt};\n\npub async fn execute_parallel_batch(\n    operations: Vec<Operation>,\n    parallelism_limit: usize\n) -> Result<Vec<OperationResult>, Error> {\n    let semaphore = tokio::sync::Semaphore::new(parallelism_limit);\n    let mut join_set = JoinSet::new();\n\n    for operation in operations {\n        let permit = semaphore.clone();\n        join_set.spawn(async move {\n            let _permit = permit.acquire().await?;\n            execute_operation(operation).await\n        });\n    }\n\n    let mut results = Vec::new();\n    while let Some(result) = join_set.join_next().await {\n        results.push(result??);\n    }\n\n    Ok(results)\n}\n```\n\n### 5. State Management Patterns\n\n#### Checkpoint-Based Recovery Pattern\n\n**Use Case**: Reliable state persistence and recovery\n\n**Implementation**:\n\n```\n#[derive(Serialize, Deserialize)]\npub struct WorkflowCheckpoint {\n    pub workflow_id: String,\n    pub step: usize,\n    pub completed_operations: Vec<String>,\n    pub current_state: serde_json::Value,\n    pub metadata: HashMap<String, String>,\n    pub timestamp: chrono::DateTime<chrono::Utc>,\n}\n\npub struct CheckpointManager {\n    checkpoint_dir: PathBuf,\n}\n\nimpl CheckpointManager {\n    pub fn save_checkpoint(&self, checkpoint: &WorkflowCheckpoint) -> Result<(), Error> {\n        let checkpoint_file = self.checkpoint_dir\n            .join(&checkpoint.workflow_id)\n            .with_extension("json");\n\n        let checkpoint_data = serde_json::to_string_pretty(checkpoint)?;\n        std::fs::write(checkpoint_file, checkpoint_data)?;\n        Ok(())\n    }\n\n    pub fn restore_checkpoint(&self, workflow_id: &str) -> Result<Option<WorkflowCheckpoint>, Error> {\n        let checkpoint_file = self.checkpoint_dir\n            .join(workflow_id)\n            .with_extension("json");\n\n        if checkpoint_file.exists() {\n            let checkpoint_data = std::fs::read_to_string(checkpoint_file)?;\n            let checkpoint = serde_json::from_str(&checkpoint_data)?;\n            Ok(Some(checkpoint))\n        } else {\n            Ok(None)\n        }\n    }\n}\n```\n\n#### Rollback Pattern\n\n```\npub struct RollbackManager {\n    rollback_stack: Vec<RollbackAction>,\n}\n\n#[derive(Clone, Debug)]\npub enum RollbackAction {\n    DeleteResource { provider: String, resource_id: String },\n    RestoreFile { path: PathBuf, content: String },\n    RevertConfiguration { key: String, value: serde_json::Value },\n    CustomAction { command: String, args: Vec<String> },\n}\n\nimpl RollbackManager {\n    pub async fn execute_rollback(&self) -> Result<(), Error> {\n        // Execute rollback actions in reverse order\n        for action in self.rollback_stack.iter().rev() {\n            match action {\n                RollbackAction::DeleteResource { provider, resource_id } => {\n                    self.delete_resource(provider, resource_id).await?;\n                }\n                RollbackAction::RestoreFile { path, content } => {\n                    tokio::fs::write(path, content).await?;\n                }\n                // ... handle other rollback actions\n            }\n        }\n        Ok(())\n    }\n}\n```\n\n### 6. Event and Messaging Patterns\n\n#### Event-Driven Architecture Pattern\n\n**Use Case**: Decoupled communication between components\n\n**Event Definition**:\n\n```\n#[derive(Serialize, Deserialize, Clone, Debug)]\npub enum SystemEvent {\n    WorkflowStarted { workflow_id: String, name: String },\n    WorkflowCompleted { workflow_id: String, result: WorkflowResult },\n    WorkflowFailed { workflow_id: String, error: String },\n    ResourceCreated { provider: String, resource_type: String, resource_id: String },\n    ResourceDeleted { provider: String, resource_type: String, resource_id: String },\n    ConfigurationChanged { key: String, old_value: serde_json::Value, new_value: serde_json::Value },\n}\n```\n\n**Event Bus Implementation**:\n\n```\nuse tokio::sync::broadcast;\n\npub struct EventBus {\n    sender: broadcast::Sender<SystemEvent>,\n}\n\nimpl EventBus {\n    pub fn new(capacity: usize) -> Self {\n        let (sender, _) = broadcast::channel(capacity);\n        Self { sender }\n    }\n\n    pub fn publish(&self, event: SystemEvent) -> Result<(), Error> {\n        self.sender.send(event)\n            .map_err(|_| Error::EventPublishFailed)?;\n        Ok(())\n    }\n\n    pub fn subscribe(&self) -> broadcast::Receiver<SystemEvent> {\n        self.sender.subscribe()\n    }\n}\n```\n\n### 7. Extension Integration Patterns\n\n#### Extension Discovery and Loading\n\n```\ndef discover-extensions [] -> table {\n    let extension_dirs = glob "extensions/*/extension.toml"\n\n    $extension_dirs\n    | each { |manifest_path|\n        let extension_dir = $manifest_path | path dirname\n        let manifest = open $manifest_path\n\n        {\n            name: $manifest.extension.name,\n            version: $manifest.extension.version,\n            type: $manifest.extension.type,\n            path: $extension_dir,\n            manifest: $manifest,\n            valid: (validate-extension $manifest),\n            compatible: (check-compatibility $manifest.compatibility)\n        }\n    }\n    | where valid and compatible\n}\n```\n\n#### Extension Interface Pattern\n\n```\n# Standard extension interface\nexport def extension-info [] -> record {\n    {\n        name: "custom-provider",\n        version: "1.0.0",\n        type: "provider",\n        description: "Custom cloud provider integration",\n        entry_points: {\n            cli: "nulib/cli.nu",\n            provider: "nulib/provider.nu"\n        }\n    }\n}\n\nexport def extension-validate [] -> bool {\n    # Validate extension configuration and dependencies\n    true\n}\n\nexport def extension-activate [] -> nothing {\n    # Perform extension activation tasks\n}\n\nexport def extension-deactivate [] -> nothing {\n    # Perform extension cleanup tasks\n}\n```\n\n### 8. API Design Patterns\n\n#### REST API Standardization\n\n**Base API Structure**:\n\n```\nuse axum::{\n    extract::{Path, State},\n    response::Json,\n    routing::{get, post, delete},\n    Router,\n};\n\npub fn create_api_router(state: AppState) -> Router {\n    Router::new()\n        .route("/health", get(health_check))\n        .route("/workflows", get(list_workflows).post(create_workflow))\n        .route("/workflows/:id", get(get_workflow).delete(delete_workflow))\n        .route("/workflows/:id/status", get(workflow_status))\n        .route("/workflows/:id/logs", get(workflow_logs))\n        .with_state(state)\n}\n```\n\n**Standard Response Format**:\n\n```\n{\n    "status": "success" | "error" | "pending",\n    "data": { ... },\n    "metadata": {\n        "timestamp": "2025-09-26T12:00:00Z",\n        "request_id": "req-123",\n        "version": "3.1.0"\n    },\n    "error": null | {\n        "code": "ERR001",\n        "message": "Human readable error",\n        "details": { ... }\n    }\n}\n```\n\n## Error Handling Patterns\n\n### Structured Error Pattern\n\n```\n#[derive(thiserror::Error, Debug)]\npub enum ProvisioningError {\n    #[error("Configuration error: {message}")]\n    Configuration { message: String },\n\n    #[error("Provider error [{provider}]: {message}")]\n    Provider { provider: String, message: String },\n\n    #[error("Workflow error [{workflow_id}]: {message}")]\n    Workflow { workflow_id: String, message: String },\n\n    #[error("Resource error [{resource_type}/{resource_id}]: {message}")]\n    Resource { resource_type: String, resource_id: String, message: String },\n}\n```\n\n### Error Recovery Pattern\n\n```\ndef with-retry [operation: closure, max_attempts: int = 3] {\n    mut attempts = 0\n    mut last_error = null\n\n    while $attempts < $max_attempts {\n        try {\n            return (do $operation)\n        } catch { |error|\n            $attempts = $attempts + 1\n            $last_error = $error\n\n            if $attempts < $max_attempts {\n                let delay = (2 ** ($attempts - 1)) * 1000  # Exponential backoff\n                sleep $"($delay)ms"\n            }\n        }\n    }\n\n    error make { msg: $"Operation failed after ($max_attempts) attempts: ($last_error)" }\n}\n```\n\n## Performance Optimization Patterns\n\n### Caching Strategy Pattern\n\n```\nuse std::sync::Arc;\nuse tokio::sync::RwLock;\nuse std::collections::HashMap;\nuse chrono::{DateTime, Utc, Duration};\n\n#[derive(Clone)]\npub struct CacheEntry<T> {\n    pub value: T,\n    pub expires_at: DateTime<Utc>,\n}\n\npub struct Cache<T> {\n    store: Arc<RwLock<HashMap<String, CacheEntry<T>>>>,\n    default_ttl: Duration,\n}\n\nimpl<T: Clone> Cache<T> {\n    pub async fn get(&self, key: &str) -> Option<T> {\n        let store = self.store.read().await;\n        if let Some(entry) = store.get(key) {\n            if entry.expires_at > Utc::now() {\n                Some(entry.value.clone())\n            } else {\n                None\n            }\n        } else {\n            None\n        }\n    }\n\n    pub async fn set(&self, key: String, value: T) {\n        let expires_at = Utc::now() + self.default_ttl;\n        let entry = CacheEntry { value, expires_at };\n\n        let mut store = self.store.write().await;\n        store.insert(key, entry);\n    }\n}\n```\n\n### Streaming Pattern for Large Data\n\n```\ndef process-large-dataset [source: string] -> nothing {\n    # Stream processing instead of loading entire dataset\n    open $source\n    | lines\n    | each { |line|\n        # Process line individually\n        $line | process-record\n    }\n    | save output.json\n}\n```\n\n## Testing Integration Patterns\n\n### Integration Test Pattern\n\n```\n#[cfg(test)]\nmod integration_tests {\n    use super::*;\n    use tokio_test;\n\n    #[tokio::test]\n    async fn test_workflow_execution() {\n        let orchestrator = setup_test_orchestrator().await;\n        let workflow = create_test_workflow();\n\n        let result = orchestrator.execute_workflow(workflow).await;\n\n        assert!(result.is_ok());\n        assert_eq!(result.unwrap().status, WorkflowStatus::Completed);\n    }\n}\n```\n\nThese integration patterns provide the foundation for the system's sophisticated multi-component architecture, enabling reliable, scalable, and\nmaintainable infrastructure automation.