sochdb_kernel/

plugin.rs

// SPDX-License-Identifier: AGPL-3.0-or-later
// SochDB - LLM-Optimized Embedded Database
// Copyright (C) 2026 Sushanth Reddy Vanagala (https://github.com/sushanthpy)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.

//! Plugin Architecture
//!
//! SochDB uses a plugin architecture to keep the kernel minimal while
//! allowing rich functionality through extensions.
//!
//! ## Design Philosophy
//!
//! 1. **Core is Minimal**: The kernel contains only ACID-critical code
//! 2. **Extensions Add Features**: Storage backends, indices, observability are plugins
//! 3. **No Bloat**: Users only pay for what they use
//! 4. **Vendor Neutral**: No lock-in to specific monitoring/storage systems
//!
//! ## Plugin Categories
//!
//! - `StorageExtension`: Alternative storage backends (LSCS, RocksDB, etc.)
//! - `IndexExtension`: Custom index types (vector, learned, etc.)
//! - `ObservabilityExtension`: Metrics, tracing, logging backends
//! - `CompressionExtension`: Compression algorithms
//!
//! ## Example: Adding Prometheus Metrics
//!
//! ```ignore
//! // In a separate crate: sochdb-prometheus-plugin
//! struct PrometheusPlugin { /* ... */ }
//!
//! impl ObservabilityExtension for PrometheusPlugin {
//!     fn record_metric(&self, name: &str, value: f64, tags: &[(&str, &str)]) {
//!         // Push to Prometheus
//!     }
//! }
//!
//! // Usage:
//! let db = KernelDB::open(path)?;
//! db.plugins().register_observability(Box::new(PrometheusPlugin::new()))?;
//! ```

use crate::error::{KernelError, KernelResult};
use crate::kernel_api::{HealthInfo, RowId, TableId};
use crate::transaction::TransactionId;
use parking_lot::{Mutex, RwLock};
use std::any::Any;
use std::collections::HashMap;
use std::sync::Arc;

// ============================================================================
// Extension Trait Definitions
// ============================================================================

/// Information about an extension
#[derive(Debug, Clone)]
pub struct ExtensionInfo {
    /// Unique extension name (e.g., "prometheus-metrics")
    pub name: String,
    /// Extension version
    pub version: String,
    /// Human-readable description
    pub description: String,
    /// Extension author
    pub author: String,
    /// Capabilities provided
    pub capabilities: Vec<ExtensionCapability>,
}

/// Capabilities an extension can provide
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum ExtensionCapability {
    /// Alternative storage backend
    Storage,
    /// Custom index type
    Index,
    /// Metrics/tracing/logging
    Observability,
    /// Compression algorithm
    Compression,
    /// Query optimization
    QueryOptimizer,
    /// Authentication/Authorization
    Auth,
    /// Custom - for third-party extensions
    Custom(String),
}

/// Base trait for all extensions
pub trait Extension: Send + Sync {
    /// Get extension information
    fn info(&self) -> ExtensionInfo;

    /// Initialize the extension
    fn init(&mut self) -> KernelResult<()> {
        Ok(())
    }

    /// Shutdown the extension gracefully
    fn shutdown(&mut self) -> KernelResult<()> {
        Ok(())
    }

    /// Cast to Any for downcasting
    fn as_any(&self) -> &dyn Any;

    /// Cast to mutable Any for downcasting
    fn as_any_mut(&mut self) -> &mut dyn Any;
}

// ============================================================================
// Storage Extension
// ============================================================================

/// Storage backend extension
///
/// Implement this to provide alternative storage engines (LSCS, RocksDB, etc.)
pub trait StorageExtension: Extension {
    /// Read data for a key
    fn get(&self, table_id: TableId, key: &[u8]) -> KernelResult<Option<Vec<u8>>>;

    /// Write data for a key
    fn put(
        &self,
        table_id: TableId,
        key: &[u8],
        value: &[u8],
        txn_id: TransactionId,
    ) -> KernelResult<()>;

    /// Delete a key
    fn delete(&self, table_id: TableId, key: &[u8], txn_id: TransactionId) -> KernelResult<()>;

    /// Scan a range of keys
    fn scan(
        &self,
        table_id: TableId,
        start: &[u8],
        end: &[u8],
        limit: usize,
    ) -> KernelResult<Vec<(Vec<u8>, Vec<u8>)>>;

    /// Flush pending writes
    fn flush(&self) -> KernelResult<()>;

    /// Compact storage (if applicable)
    fn compact(&self) -> KernelResult<()> {
        Ok(()) // Default: no-op
    }

    /// Get storage statistics
    fn stats(&self) -> StorageStats {
        StorageStats::default()
    }
}

/// Storage statistics
#[derive(Debug, Clone, Default)]
pub struct StorageStats {
    /// Total bytes stored
    pub bytes_stored: u64,
    /// Number of keys
    pub key_count: u64,
    /// Pending compaction bytes
    pub pending_compaction_bytes: u64,
    /// Write amplification factor
    pub write_amplification: f64,
}

// ============================================================================
// Index Extension
// ============================================================================

/// Index extension
///
/// Implement this for custom index types (vector, learned, full-text, etc.)
pub trait IndexExtension: Extension {
    /// Index type name (e.g., "hnsw", "learned", "btree")
    fn index_type(&self) -> &str;

    /// Build index on existing data
    fn build(
        &mut self,
        table_id: TableId,
        column_id: u16,
        data: &[(RowId, Vec<u8>)],
    ) -> KernelResult<()>;

    /// Insert a key-value pair into the index
    fn insert(&mut self, key: &[u8], row_id: RowId) -> KernelResult<()>;

    /// Delete a key from the index
    fn delete(&mut self, key: &[u8], row_id: RowId) -> KernelResult<()>;

    /// Point lookup
    fn lookup(&self, key: &[u8]) -> KernelResult<Vec<RowId>>;

    /// Range scan
    fn range(&self, start: &[u8], end: &[u8], limit: usize) -> KernelResult<Vec<RowId>>;

    /// Nearest neighbor search (for vector indices)
    fn nearest(&self, _query: &[u8], _k: usize) -> KernelResult<Vec<(RowId, f32)>> {
        Err(KernelError::Plugin {
            message: "nearest neighbor not supported by this index type".into(),
        })
    }

    /// Get index size in bytes
    fn size_bytes(&self) -> u64;
}

// ============================================================================
// Observability Extension (PLUGIN ARCHITECTURE)
// ============================================================================

/// Observability extension
///
/// Implement this for metrics, tracing, and logging backends.
///
/// ## Why Plugin Architecture?
///
/// 1. **No Dependency Bloat**: Core doesn't pull in Prometheus, DataDog, etc.
/// 2. **Vendor Neutral**: Users choose their observability stack
/// 3. **Flexible**: Can run without any observability in embedded scenarios
///
/// ## Available Plugins (separate crates):
///
/// - `sochdb-prometheus`: Prometheus metrics
/// - `sochdb-datadog`: DataDog integration  
/// - `sochdb-opentelemetry`: OpenTelemetry support
/// - `sochdb-logging-json`: JSON structured logging
/// - `sochdb-logging-logfmt`: logfmt style logging
pub trait ObservabilityExtension: Extension {
    // -------------------------------------------------------------------------
    // Metrics
    // -------------------------------------------------------------------------

    /// Record a counter increment
    fn counter_inc(&self, name: &str, value: u64, labels: &[(&str, &str)]);

    /// Record a gauge value
    fn gauge_set(&self, name: &str, value: f64, labels: &[(&str, &str)]);

    /// Record a histogram observation
    fn histogram_observe(&self, name: &str, value: f64, labels: &[(&str, &str)]);

    // -------------------------------------------------------------------------
    // Tracing
    // -------------------------------------------------------------------------

    /// Start a new span
    fn span_start(&self, name: &str, parent: Option<u64>) -> u64;

    /// End a span
    fn span_end(&self, span_id: u64);

    /// Add an event to a span
    fn span_event(&self, span_id: u64, name: &str, attributes: &[(&str, &str)]);

    // -------------------------------------------------------------------------
    // Logging
    // -------------------------------------------------------------------------

    /// Log at debug level
    fn log_debug(&self, message: &str, fields: &[(&str, &str)]) {
        let _ = (message, fields); // Default: no-op
    }

    /// Log at info level
    fn log_info(&self, message: &str, fields: &[(&str, &str)]) {
        let _ = (message, fields); // Default: no-op
    }

    /// Log at warn level
    fn log_warn(&self, message: &str, fields: &[(&str, &str)]) {
        let _ = (message, fields); // Default: no-op
    }

    /// Log at error level
    fn log_error(&self, message: &str, fields: &[(&str, &str)]) {
        let _ = (message, fields); // Default: no-op
    }

    // -------------------------------------------------------------------------
    // Health Reporting
    // -------------------------------------------------------------------------

    /// Report health status (called periodically by kernel)
    fn report_health(&self, health: &HealthInfo) {
        let _ = health; // Default: no-op
    }
}

// ============================================================================
// Compression Extension
// ============================================================================

/// Compression algorithm extension
pub trait CompressionExtension: Extension {
    /// Algorithm name (e.g., "lz4", "zstd", "snappy")
    fn algorithm(&self) -> &str;

    /// Compress data
    fn compress(&self, input: &[u8]) -> KernelResult<Vec<u8>>;

    /// Decompress data
    fn decompress(&self, input: &[u8]) -> KernelResult<Vec<u8>>;

    /// Compression level (if applicable)
    fn set_level(&mut self, _level: i32) -> KernelResult<()> {
        Ok(())
    }
}

// ============================================================================
// Plugin Manager
// ============================================================================

/// Plugin manager - registry for all extensions
///
/// The kernel uses this to discover and invoke extensions.
pub struct PluginManager {
    /// Storage extensions by name
    storage: RwLock<HashMap<String, Arc<dyn StorageExtension>>>,
    /// Index extensions by name
    indices: RwLock<HashMap<String, Arc<RwLock<dyn IndexExtension>>>>,
    /// Observability extensions (can have multiple)
    observability: RwLock<Vec<Arc<dyn ObservabilityExtension>>>,
    /// Compression extensions by algorithm name
    compression: RwLock<HashMap<String, Arc<dyn CompressionExtension>>>,
    /// Active storage backend name
    active_storage: RwLock<Option<String>>,
    /// Shutdown handles for storage extensions (wraps extension in Mutex for interior mutability)
    storage_shutdown: RwLock<HashMap<String, Mutex<Arc<dyn StorageExtension>>>>,
    /// Shutdown handles for index extensions
    index_shutdown: RwLock<HashMap<String, Arc<Mutex<Arc<RwLock<dyn IndexExtension>>>>>>,
    /// Shutdown handles for observability
    observability_shutdown: RwLock<Vec<Mutex<Arc<dyn ObservabilityExtension>>>>,
    /// Shutdown handles for compression
    compression_shutdown: RwLock<HashMap<String, Mutex<Arc<dyn CompressionExtension>>>>,
}

impl Default for PluginManager {
    fn default() -> Self {
        Self::new()
    }
}

impl PluginManager {
    /// Create a new plugin manager
    pub fn new() -> Self {
        Self {
            storage: RwLock::new(HashMap::new()),
            indices: RwLock::new(HashMap::new()),
            observability: RwLock::new(Vec::new()),
            compression: RwLock::new(HashMap::new()),
            active_storage: RwLock::new(None),
            storage_shutdown: RwLock::new(HashMap::new()),
            index_shutdown: RwLock::new(HashMap::new()),
            observability_shutdown: RwLock::new(Vec::new()),
            compression_shutdown: RwLock::new(HashMap::new()),
        }
    }

    // -------------------------------------------------------------------------
    // Storage Extensions
    // -------------------------------------------------------------------------

    /// Register a storage extension
    pub fn register_storage(&self, ext: Arc<dyn StorageExtension>) -> KernelResult<()> {
        let name = ext.info().name.clone();
        let mut storage = self.storage.write();

        if storage.contains_key(&name) {
            return Err(KernelError::Plugin {
                message: format!("storage extension '{}' already registered", name),
            });
        }

        // Also add to shutdown storage - need to wrap the Arc<dyn StorageExtension> in Mutex
        let mut shutdown = self.storage_shutdown.write();
        shutdown.insert(name.clone(), Mutex::new(Arc::clone(&ext)));

        storage.insert(name.clone(), ext);

        // Set as active if it's the first one
        let mut active = self.active_storage.write();
        if active.is_none() {
            *active = Some(name);
        }

        Ok(())
    }

    /// Set the active storage backend
    pub fn set_active_storage(&self, name: &str) -> KernelResult<()> {
        let storage = self.storage.read();
        if !storage.contains_key(name) {
            return Err(KernelError::Plugin {
                message: format!("storage extension '{}' not found", name),
            });
        }
        *self.active_storage.write() = Some(name.to_string());
        Ok(())
    }

    /// Get the active storage backend
    pub fn storage(&self) -> Option<Arc<dyn StorageExtension>> {
        let active = self.active_storage.read();
        active
            .as_ref()
            .and_then(|name| self.storage.read().get(name).cloned())
    }

    // -------------------------------------------------------------------------
    // Index Extensions
    // -------------------------------------------------------------------------

    /// Register an index extension
    pub fn register_index(&self, ext: Arc<RwLock<dyn IndexExtension>>) -> KernelResult<()> {
        let name = ext.read().info().name.clone();
        let mut indices = self.indices.write();

        if indices.contains_key(&name) {
            return Err(KernelError::Plugin {
                message: format!("index extension '{}' already registered", name),
            });
        }

        indices.insert(name.clone(), ext.clone());

        // Also add to shutdown storage
        let mut shutdown = self.index_shutdown.write();
        shutdown.insert(name, Arc::new(Mutex::new(ext)));
        Ok(())
    }

    /// Get an index extension by name
    pub fn index(&self, name: &str) -> Option<Arc<RwLock<dyn IndexExtension>>> {
        self.indices.read().get(name).cloned()
    }

    /// List registered index types
    pub fn list_index_types(&self) -> Vec<String> {
        self.indices.read().keys().cloned().collect()
    }

    // -------------------------------------------------------------------------
    // Observability Extensions
    // -------------------------------------------------------------------------

    /// Register an observability extension
    ///
    /// Multiple observability extensions can be registered (fan-out to all)
    pub fn register_observability(&self, ext: Arc<dyn ObservabilityExtension>) -> KernelResult<()> {
        self.observability.write().push(ext.clone());

        // Also add to shutdown storage
        let mut shutdown = self.observability_shutdown.write();
        shutdown.push(Mutex::new(Arc::clone(&ext)));
        Ok(())
    }

    /// Record a counter across all observability extensions
    pub fn counter_inc(&self, name: &str, value: u64, labels: &[(&str, &str)]) {
        for ext in self.observability.read().iter() {
            ext.counter_inc(name, value, labels);
        }
    }

    /// Record a gauge across all observability extensions
    pub fn gauge_set(&self, name: &str, value: f64, labels: &[(&str, &str)]) {
        for ext in self.observability.read().iter() {
            ext.gauge_set(name, value, labels);
        }
    }

    /// Record a histogram observation across all observability extensions
    pub fn histogram_observe(&self, name: &str, value: f64, labels: &[(&str, &str)]) {
        for ext in self.observability.read().iter() {
            ext.histogram_observe(name, value, labels);
        }
    }

    /// Report health to all observability extensions
    pub fn report_health(&self, health: &HealthInfo) {
        for ext in self.observability.read().iter() {
            ext.report_health(health);
        }
    }

    /// Check if any observability is configured
    pub fn has_observability(&self) -> bool {
        !self.observability.read().is_empty()
    }

    // -------------------------------------------------------------------------
    // Compression Extensions
    // -------------------------------------------------------------------------

    /// Register a compression extension
    pub fn register_compression(&self, ext: Arc<dyn CompressionExtension>) -> KernelResult<()> {
        let algo = ext.algorithm().to_string();
        let mut compression = self.compression.write();

        if compression.contains_key(&algo) {
            return Err(KernelError::Plugin {
                message: format!("compression '{}' already registered", algo),
            });
        }

        compression.insert(algo.clone(), ext.clone());

        // Also add to shutdown storage
        let mut shutdown = self.compression_shutdown.write();
        shutdown.insert(algo.clone(), Mutex::new(Arc::clone(&ext)));
        Ok(())
    }

    /// Get a compression extension by algorithm name
    pub fn compression(&self, algorithm: &str) -> Option<Arc<dyn CompressionExtension>> {
        self.compression.read().get(algorithm).cloned()
    }

    /// List available compression algorithms
    pub fn list_compression(&self) -> Vec<String> {
        self.compression.read().keys().cloned().collect()
    }

    // -------------------------------------------------------------------------
    // Lifecycle
    // -------------------------------------------------------------------------

    /// Shutdown all extensions gracefully
    ///
    /// Calls `shutdown()` on each registered extension in reverse registration order.
    /// This allows extensions to release resources, close connections, flush buffers, etc.
    pub fn shutdown_all(&self) -> KernelResult<()> {
        let mut errors = Vec::new();

        // Shutdown storage extensions (in reverse order)
        {
            let storage = self.storage_shutdown.read();
            let names: Vec<_> = storage.keys().cloned().collect();
            for name in names.into_iter().rev() {
                if let Some(ext) = storage.get(&name) {
                    let mut ext_guard = ext.lock();
                    if let Some(ext) = Arc::get_mut(&mut ext_guard) {
                        if let Err(e) = ext.shutdown() {
                            errors.push(format!("storage '{}': {}", name, e));
                        }
                    }
                }
            }
        }

        // Shutdown index extensions (in reverse order)
        {
            let indices = self.index_shutdown.read();
            let names: Vec<_> = indices.keys().cloned().collect();
            for name in names.into_iter().rev() {
                if let Some(ext) = indices.get(&name) {
                    let ext_guard = ext.lock();
                    let mut inner = ext_guard.write();
                    if let Err(e) = inner.shutdown() {
                        errors.push(format!("index '{}': {}", name, e));
                    }
                }
            }
        }

        // Shutdown observability extensions (in reverse order)
        {
            let observability = self.observability_shutdown.read();
            for ext in observability.iter().rev() {
                let mut ext_guard = ext.lock();
                if let Some(ext) = Arc::get_mut(&mut ext_guard) {
                    if let Err(e) = ext.shutdown() {
                        errors.push(format!("observability: {}", e));
                    }
                }
            }
        }

        // Shutdown compression extensions (in reverse order)
        {
            let compression = self.compression_shutdown.read();
            let names: Vec<_> = compression.keys().cloned().collect();
            for name in names.into_iter().rev() {
                if let Some(ext) = compression.get(&name) {
                    let mut ext_guard = ext.lock();
                    if let Some(ext) = Arc::get_mut(&mut ext_guard) {
                        if let Err(e) = ext.shutdown() {
                            errors.push(format!("compression '{}': {}", name, e));
                        }
                    }
                }
            }
        }

        if errors.is_empty() {
            Ok(())
        } else {
            Err(KernelError::Plugin {
                message: format!("shutdown errors: {}", errors.join("; ")),
            })
        }
    }

    /// Get information about all registered extensions
    pub fn list_extensions(&self) -> Vec<ExtensionInfo> {
        let mut result = Vec::new();

        for ext in self.storage.read().values() {
            result.push(ext.info());
        }

        for ext in self.indices.read().values() {
            result.push(ext.read().info());
        }

        for ext in self.observability.read().iter() {
            result.push(ext.info());
        }

        for ext in self.compression.read().values() {
            result.push(ext.info());
        }

        result
    }
}

// ============================================================================
// Null Observability (Default - No-Op)
// ============================================================================

/// Null observability extension - does nothing
///
/// Used when no observability is configured. Zero overhead.
pub struct NullObservability;

impl Extension for NullObservability {
    fn info(&self) -> ExtensionInfo {
        ExtensionInfo {
            name: "null-observability".into(),
            version: "0.0.0".into(),
            description: "No-op observability (default)".into(),
            author: "SochDB".into(),
            capabilities: vec![ExtensionCapability::Observability],
        }
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn as_any_mut(&mut self) -> &mut dyn Any {
        self
    }
}

impl ObservabilityExtension for NullObservability {
    fn counter_inc(&self, _name: &str, _value: u64, _labels: &[(&str, &str)]) {}
    fn gauge_set(&self, _name: &str, _value: f64, _labels: &[(&str, &str)]) {}
    fn histogram_observe(&self, _name: &str, _value: f64, _labels: &[(&str, &str)]) {}
    fn span_start(&self, _name: &str, _parent: Option<u64>) -> u64 {
        0
    }
    fn span_end(&self, _span_id: u64) {}
    fn span_event(&self, _span_id: u64, _name: &str, _attributes: &[(&str, &str)]) {}
}

// ============================================================================
// Dynamic Plugin Loading (Optional Feature)
// ============================================================================

#[cfg(feature = "dynamic-plugins")]
pub mod dynamic {
    //! Dynamic plugin loading support
    //!
    //! Enabled with the `dynamic-plugins` feature.
    //! Allows loading plugins from shared libraries at runtime.
    //!
    //! # Security Warning
    //!
    //! Loading a native shared library (`dlopen`) executes arbitrary code
    //! in the host process — including the library's `_init()` constructors
    //! which run *before* any symbol lookup.  A malicious `.so/.dylib` can:
    //!
    //! - Exfiltrate data from the database process
    //! - Spawn background threads or open network connections
    //! - Corrupt kernel memory (no isolation boundary)
    //!
    //! **Only load libraries you have built from audited source code.**
    //! For untrusted extensions, use the WASM plugin sandbox instead.

    use super::*;
    use libloading::{Library, Symbol};
    use std::path::Path;

    /// Environment variable that opts in to native (`dlopen`) plugin loading.
    ///
    /// Set to `1`/`true`/`yes` to allow a default-constructed
    /// [`DynamicPluginLoader`] to load native shared libraries. Absent or any
    /// other value keeps native loading disabled and steers callers to the WASM
    /// sandbox.
    pub const ALLOW_NATIVE_ENV: &str = "SOCHDB_ALLOW_NATIVE_PLUGINS";

    fn env_opts_in_to_native() -> bool {
        std::env::var(ALLOW_NATIVE_ENV)
            .map(|v| {
                let v = v.trim().to_ascii_lowercase();
                v == "1" || v == "true" || v == "yes" || v == "on"
            })
            .unwrap_or(false)
    }

    /// Dynamic plugin loader
    ///
    /// Native loading is **untrusted by default**. Even with the
    /// `dynamic-plugins` feature compiled in, a loader will refuse to `dlopen`
    /// a shared library unless it was explicitly marked trusted — either by
    /// constructing it with [`DynamicPluginLoader::new_trusted`] or by setting
    /// the [`ALLOW_NATIVE_ENV`] environment variable. Untrusted extensions
    /// should run in the WASM sandbox instead.
    pub struct DynamicPluginLoader {
        /// Loaded libraries (kept alive)
        _libraries: Vec<Library>,
        /// Whether native `dlopen` loading is permitted for this loader.
        trusted: bool,
    }

    impl DynamicPluginLoader {
        /// Create a new dynamic plugin loader.
        ///
        /// Native loading is **disabled** unless the [`ALLOW_NATIVE_ENV`]
        /// environment variable opts in. Use [`new_trusted`](Self::new_trusted)
        /// to enable native loading programmatically. This safe-by-default
        /// stance means an attacker who can drop a `.so` on disk cannot get it
        /// loaded without an explicit operator decision.
        pub fn new() -> Self {
            Self {
                _libraries: Vec::new(),
                trusted: env_opts_in_to_native(),
            }
        }

        /// Create a loader that is explicitly permitted to load native
        /// shared libraries.
        ///
        /// # Safety / Trust
        ///
        /// Calling this is an assertion by the operator that any library passed
        /// to [`load_observability`](Self::load_observability) comes from
        /// audited, trusted source code. Native libraries run with **no
        /// isolation** and can execute arbitrary code in the database process
        /// (see the module-level security warning). Prefer the WASM sandbox for
        /// anything that is not fully trusted.
        pub fn new_trusted() -> Self {
            Self {
                _libraries: Vec::new(),
                trusted: true,
            }
        }

        /// Whether this loader is permitted to load native libraries.
        pub fn is_trusted(&self) -> bool {
            self.trusted
        }

        /// Load an observability plugin from a shared library.
        ///
        /// # Safety
        ///
        /// This function is `unsafe` because loading a native shared library
        /// can execute arbitrary code.  The caller MUST ensure:
        ///
        /// 1. The library at `path` was built from audited, trusted source code.
        /// 2. The path is an absolute, canonicalized path (no symlink races).
        /// 3. The file permissions prevent modification by unprivileged users.
        ///
        /// The library must export:
        /// ```c
        /// extern "C" fn create_observability_plugin() -> *mut dyn ObservabilityExtension
        /// ```
        ///
        /// Returns an error (without performing any `dlopen`) if this loader is
        /// not trusted — see [`new_trusted`](Self::new_trusted) and
        /// [`ALLOW_NATIVE_ENV`].
        pub unsafe fn load_observability(
            &mut self,
            path: &Path,
        ) -> KernelResult<Arc<dyn ObservabilityExtension>> {
            // Untrusted-by-default gate: refuse native loading unless explicitly
            // enabled. This check happens BEFORE any filesystem access or
            // `dlopen`, so a malicious library is never even opened.
            if !self.trusted {
                return Err(KernelError::Plugin {
                    message: format!(
                        "native plugin loading is disabled (untrusted by default). \
                         Load untrusted extensions via the WASM sandbox, or opt in \
                         explicitly with DynamicPluginLoader::new_trusted() or by \
                         setting {ALLOW_NATIVE_ENV}=1. Refused: {}",
                        path.display()
                    ),
                });
            }

            // Validate the path is absolute to prevent relative-path hijacking
            if !path.is_absolute() {
                return Err(KernelError::Plugin {
                    message: format!(
                        "plugin path must be absolute to prevent path hijacking: {}",
                        path.display()
                    ),
                });
            }

            // Canonicalize to resolve symlinks and detect TOCTOU races
            let canonical = path.canonicalize().map_err(|e| KernelError::Plugin {
                message: format!("failed to canonicalize plugin path: {}", e),
            })?;

            unsafe {
                let lib = Library::new(&canonical).map_err(|e| KernelError::Plugin {
                    message: format!("failed to load library: {}", e),
                })?;

                let create_fn: Symbol<fn() -> Box<dyn ObservabilityExtension>> = lib
                    .get(b"create_observability_plugin")
                    .map_err(|e| KernelError::Plugin {
                        message: format!("symbol not found: {}", e),
                    })?;

                let plugin = create_fn();
                self._libraries.push(lib);

                Ok(Arc::from(plugin))
            }
        }
    }

    impl Default for DynamicPluginLoader {
        fn default() -> Self {
            Self::new()
        }
    }

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

        #[test]
        fn test_untrusted_loader_refuses_native_load() {
            // Default loader (no env opt-in) must refuse to load a native lib,
            // returning an error WITHOUT touching the filesystem.
            let mut loader = DynamicPluginLoader::new_untrusted_for_test();
            assert!(!loader.is_trusted());
            let path = Path::new("/nonexistent/evil.so");
            let result = unsafe { loader.load_observability(path) };
            assert!(result.is_err(), "untrusted loader must refuse native load");
            let msg = format!("{:?}", result.err().unwrap());
            assert!(
                msg.contains("untrusted") || msg.contains("disabled"),
                "error should explain the untrusted-by-default policy, got: {msg}"
            );
        }

        #[test]
        fn test_trusted_loader_passes_gate_then_fails_on_missing_file() {
            // A trusted loader passes the trust gate; with a missing file it
            // fails later (canonicalize), proving the gate itself is not what
            // rejects it.
            let mut loader = DynamicPluginLoader::new_trusted();
            assert!(loader.is_trusted());
            let path = Path::new("/nonexistent/trusted.so");
            let result = unsafe { loader.load_observability(path) };
            assert!(result.is_err());
            let msg = format!("{:?}", result.err().unwrap());
            assert!(
                !msg.contains("untrusted"),
                "trusted loader must fail past the trust gate, got: {msg}"
            );
        }
    }

    impl DynamicPluginLoader {
        /// Test-only constructor forcing the untrusted state regardless of the
        /// ambient environment (so the env var cannot make the test flaky).
        #[cfg(test)]
        fn new_untrusted_for_test() -> Self {
            Self {
                _libraries: Vec::new(),
                trusted: false,
            }
        }
    }
}

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

    #[test]
    fn test_plugin_manager_creation() {
        let pm = PluginManager::new();
        assert!(!pm.has_observability());
        assert!(pm.storage().is_none());
    }

    #[test]
    fn test_null_observability() {
        let null = NullObservability;
        // Should not panic
        null.counter_inc("test", 1, &[]);
        null.gauge_set("test", 1.0, &[]);
        null.histogram_observe("test", 1.0, &[]);
        let span = null.span_start("test", None);
        null.span_event(span, "event", &[]);
        null.span_end(span);
    }

    #[test]
    fn test_register_observability() {
        let pm = PluginManager::new();
        let null = Arc::new(NullObservability);

        assert!(!pm.has_observability());
        pm.register_observability(null).unwrap();
        assert!(pm.has_observability());
    }

    struct TestExtension {
        shutdown_called: Arc<std::sync::atomic::AtomicBool>,
    }

    impl TestExtension {
        fn new_with_flag(flag: Arc<std::sync::atomic::AtomicBool>) -> Self {
            Self {
                shutdown_called: flag,
            }
        }
    }

    impl ObservabilityExtension for TestExtension {
        fn counter_inc(&self, _name: &str, _value: u64, _labels: &[(&str, &str)]) {}
        fn gauge_set(&self, _name: &str, _value: f64, _labels: &[(&str, &str)]) {}
        fn histogram_observe(&self, _name: &str, _value: f64, _labels: &[(&str, &str)]) {}
        fn span_start(&self, _name: &str, _parent: Option<u64>) -> u64 {
            0
        }
        fn span_event(&self, _span_id: u64, _name: &str, _labels: &[(&str, &str)]) {}
        fn span_end(&self, _span_id: u64) {}
    }

    impl Extension for TestExtension {
        fn info(&self) -> ExtensionInfo {
            ExtensionInfo {
                name: "test-extension".into(),
                version: "0.0.1".into(),
                description: "Test extension".into(),
                author: "test".into(),
                capabilities: vec![],
            }
        }

        fn shutdown(&mut self) -> KernelResult<()> {
            self.shutdown_called
                .store(true, std::sync::atomic::Ordering::SeqCst);
            Ok(())
        }

        fn as_any(&self) -> &dyn Any {
            self
        }

        fn as_any_mut(&mut self) -> &mut dyn Any {
            self
        }
    }

    #[test]
    fn test_shutdown_all_calls_shutdown() {
        let pm = PluginManager::new();
        let flag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        {
            let ext = TestExtension::new_with_flag(flag.clone());
            let arc_ext: Arc<dyn ObservabilityExtension> = Arc::new(ext);

            let mut shutdown = pm.observability_shutdown.write();
            shutdown.push(Mutex::new(arc_ext));
        }

        pm.shutdown_all().unwrap();

        assert!(
            flag.load(std::sync::atomic::Ordering::SeqCst),
            "shutdown() should have been called on the extension"
        );
    }

    #[test]
    fn test_shutdown_all_with_single_ref() {
        use std::sync::atomic::Ordering;

        struct SingleRefExtension {
            flag: Arc<std::sync::atomic::AtomicBool>,
        }

        impl SingleRefExtension {
            fn new(flag: Arc<std::sync::atomic::AtomicBool>) -> Self {
                Self { flag }
            }
        }

        impl ObservabilityExtension for SingleRefExtension {
            fn counter_inc(&self, _: &str, _: u64, _: &[(&str, &str)]) {}
            fn gauge_set(&self, _: &str, _: f64, _: &[(&str, &str)]) {}
            fn histogram_observe(&self, _: &str, _: f64, _: &[(&str, &str)]) {}
            fn span_start(&self, _: &str, _: Option<u64>) -> u64 {
                0
            }
            fn span_event(&self, _: u64, _: &str, _: &[(&str, &str)]) {}
            fn span_end(&self, _: u64) {}
        }

        impl Extension for SingleRefExtension {
            fn info(&self) -> ExtensionInfo {
                ExtensionInfo {
                    name: "single-ref".into(),
                    version: "0.0.1".into(),
                    description: "Test".into(),
                    author: "test".into(),
                    capabilities: vec![],
                }
            }
            fn shutdown(&mut self) -> KernelResult<()> {
                self.flag.store(true, Ordering::SeqCst);
                Ok(())
            }
            fn as_any(&self) -> &dyn Any {
                self
            }
            fn as_any_mut(&mut self) -> &mut dyn Any {
                self
            }
        }

        let pm = PluginManager::new();
        let flag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        {
            let ext = SingleRefExtension::new(flag.clone());
            let boxed: Box<dyn ObservabilityExtension> = Box::new(ext);
            let arc_ext: Arc<dyn ObservabilityExtension> = Arc::from(boxed);

            let mut shutdown = pm.observability_shutdown.write();
            shutdown.push(Mutex::new(arc_ext));
        }

        pm.shutdown_all().unwrap();

        assert!(
            flag.load(Ordering::SeqCst),
            "shutdown should have been called"
        );
    }
}