product_os_connector/

lib.rs

//! # Product OS Connector
//!
//! A flexible, configuration-driven framework for defining API servers, workflows, and authentication.
//!
//! ## Overview
//!
//! The `product-os-connector` crate provides a powerful abstraction for building API integration
//! systems using declarative configuration rather than imperative code. It supports multiple
//! protocol types (REST, GraphQL, WebSocket) and various authentication methods (`OAuth2`, JWT,
//! API keys, etc.).
//!
//! ## Features
//!
//! - **`definition`**: Core flow definitions and routing (depends on `matchit`, `serde_json`, `regex`)
//! - **`connectors`**: Full connector functionality with async support (depends on `async-trait`, `parking_lot`)
//! - **`openapi`**: OpenAPI/Swagger integration for automatic API definition import
//! - **`default`**: Enables all features with standard library support
//!
//! ## Architecture
//!
//! The crate is organized around several key concepts:
//!
//! - **Definition**: Declarative API specification (protocols, paths, auth, flows)
//! - **Connector**: Runtime handler that processes inward/outward requests
//! - **Interface**: Protocol-specific implementations (REST, GraphQL, WebSocket)
//! - **Flows**: Inward (server endpoints) and Outward (client calls) request handling
//! - **Authentication**: Pluggable auth methods with lock/unlock semantics
//!
//! ## Quick Start
//!
//! ```rust,no_run
//! use product_os_connector::{Definition, ProductOSConnectors, ConnectorKind};
//! use std::collections::BTreeMap;
//!
//! // Load or create connector definitions
//! let mut definitions = BTreeMap::new();
//! // definitions.insert("my-api".to_string(), my_definition);
//!
//! // Initialize connectors
//! let connectors = ProductOSConnectors::new(definitions);
//!
//! // Register with router (requires product-os-router)
//! // let mut router = ProductOSRouter::new();
//! // connectors.setup_handlers(&mut router).await;
//! ```
//!
//! ## Configuration-Driven Workflows
//!
//! Instead of writing code for each API endpoint, you define them in configuration:
//!
//! ```json
//! {
//!   "info": {
//!     "identifier": "my-service",
//!     "version": "1.0.0"
//!   },
//!   "kind": "rest",
//!   "protocol": "https",
//!   "addresses": ["api.example.com"],
//!   "inward_root": "/api/v1",
//!   "flows_inward": {
//!     "/users": {
//!       "methods": {
//!         "get": {
//!           "auth_method": "jwt",
//!           "output": { /* mapping */ }
//!         }
//!       }
//!     }
//!   }
//! }
//! ```
//!
//! ## `no_std` Support
//!
//! This crate is `no_std` compatible via the `no-std-compat` facade, though async operations
//! and networking features require an allocator.
//!
//! ## See Also
//!
//! - [`Definition`]: The core configuration structure
//! - [`ProductOSConnectors`]: Main entry point for connector runtime
//! - [`ConnectorKind`]: Supported protocol types

#![no_std]
#![warn(missing_docs)]
#![warn(rust_2018_idioms)]
#![warn(clippy::all)]
#![warn(clippy::pedantic)]
#![allow(clippy::module_name_repetitions)]
#![allow(clippy::missing_errors_doc)]
#![allow(clippy::missing_panics_doc)]

extern crate no_std_compat as std;

use std::prelude::v1::*;

#[cfg(feature = "definition")]
mod definition;

#[cfg(all(feature = "connectors", feature = "definition"))]
mod authentication;
#[cfg(all(feature = "connectors", feature = "definition"))]
mod rest;
#[cfg(all(feature = "connectors", feature = "definition"))]
pub mod connector;
#[cfg(all(feature = "connectors", feature = "definition"))]
mod ws;
#[cfg(all(feature = "connectors", feature = "definition"))]
mod graphql;
#[cfg(all(feature = "connectors", feature = "definition"))]
mod interface;


#[cfg(feature = "connectors")]
use std::collections::BTreeMap;

#[cfg(feature = "connectors")]
use std::sync::Arc;

#[cfg(feature = "connectors")]
use std::time::Duration;

#[cfg(feature = "connectors")]
use product_os_capabilities::{Feature, RegistryFeature, What};
use serde::{Deserialize, Serialize};

#[cfg(feature = "definition")]
pub use crate::definition::Definition;

#[cfg(feature = "connectors")]
use async_trait::async_trait;

#[cfg(feature = "connectors")]
use parking_lot::Mutex;

#[cfg(feature = "connectors")]
use product_os_router::{Body, IntoResponse, Response, StatusCode};

#[cfg(all(feature = "connectors", feature = "definition"))]
use crate::graphql::GraphQL;
#[cfg(all(feature = "connectors", feature = "definition"))]
use crate::interface::Interface;
#[cfg(all(feature = "connectors", feature = "definition"))]
use crate::rest::Rest;
#[cfg(all(feature = "connectors", feature = "definition"))]
use crate::ws::WebSocket;



/// The type of connector protocol to use for API communication.
///
/// Each variant represents a different network protocol with its own semantics:
///
/// - **REST**: Traditional HTTP/HTTPS `RESTful` APIs with standard HTTP methods
/// - **GraphQL**: Graph Query Language APIs with schema-based queries
/// - **WebSocket**: Bidirectional, persistent connection protocols
///
/// # Examples
///
/// ```
/// use product_os_connector::ConnectorKind;
///
/// let kind = ConnectorKind::Rest;
/// assert_eq!(format!("{:?}", kind), "Rest");
/// ```
///
/// # Serialization
///
/// Serializes as lowercase strings in JSON/YAML:
/// ```json
/// { "kind": "rest" }
/// { "kind": "graphql" }
/// { "kind": "websocket" }
/// ```
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum ConnectorKind {
    /// REST/HTTP API connector
    Rest,
    /// GraphQL API connector
    GraphQL,
    /// WebSocket bidirectional connector
    WebSocket,
    // EventStream,
    // MQTT,
    // Kafka,
    // Soap,
}


/// Central coordinator for managing multiple API connectors.
///
/// `ProductOSConnectors` handles the lifecycle of multiple connector instances, each with
/// its own protocol type and configuration. It manages interface registration, request routing,
/// and cross-connector communication.
///
/// # Architecture
///
/// Each connector is wrapped in an `Arc<Mutex<dyn Interface>>` to allow:
/// - Thread-safe shared access
/// - Dynamic dispatch across protocol types
/// - Safe concurrent request handling
///
/// # Examples
///
/// ```no_run
/// use product_os_connector::{ProductOSConnectors, Definition};
/// use std::collections::BTreeMap;
///
/// # async fn example() {
/// let definitions = BTreeMap::new();
/// let connectors = ProductOSConnectors::new(definitions);
///
/// // Setup routing
/// // let mut router = product_os_router::ProductOSRouter::new();
/// // connectors.setup_handlers(&mut router).await;
/// # }
/// ```
///
/// # Inter-Connector Communication
///
/// Connectors can reference each other in workflow steps, enabling complex
/// integration patterns like:
/// - API Gateway → Backend Service
/// - Service Mesh Communication
/// - Webhook Forwarding
#[cfg(all(feature = "connectors", feature = "definition"))]
pub struct ProductOSConnectors {
    /// Map of connector identifier to interface implementation
    interfaces: BTreeMap<String, Arc<Mutex<dyn Interface>>>
    //relational_store: Option<Arc<ProductOSRelationalStore>>,
}

#[cfg(all(feature = "connectors", feature = "definition"))]
impl ProductOSConnectors {
    /// Creates a new connector manager from a map of definitions.
    ///
    /// Each definition is instantiated into its appropriate protocol handler (REST, GraphQL,
    /// or WebSocket) and stored for request routing. Connectors are cross-registered so they
    /// can call each other in workflow steps.
    ///
    /// # Arguments
    ///
    /// * `predefined_definitions` - Map of connector identifiers to their definitions
    ///
    /// # Panics
    ///
    /// May panic if interface registration fails due to lock timeout (10 seconds).
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use product_os_connector::{ProductOSConnectors, Definition};
    /// use std::collections::BTreeMap;
    ///
    /// let mut defs = BTreeMap::new();
    /// // defs.insert("api-1".to_string(), definition1);
    /// // defs.insert("api-2".to_string(), definition2);
    ///
    /// let connectors = ProductOSConnectors::new(defs);
    /// ```
    pub fn new(predefined_definitions: BTreeMap<String, Definition>, /*relational_store: Option<Arc<ProductOSRelationalStore>>*/) -> Self {
        let mut interfaces: BTreeMap<String, Arc<Mutex<dyn Interface>>> = BTreeMap::new();

        tracing::debug!("Definitions: {:?}", predefined_definitions);

        for definition in predefined_definitions.values() {
            match definition.kind {
                ConnectorKind::Rest => {
                    let rest = Rest::new(definition);
                    interfaces.insert(definition.info.identifier.clone(), Arc::new(Mutex::new(rest)));
                }
                ConnectorKind::GraphQL => {
                    let graph_ql = GraphQL::new(definition);
                    interfaces.insert(definition.info.identifier.clone(), Arc::new(Mutex::new(graph_ql)));
                }
                ConnectorKind::WebSocket => {
                    let web_socket = WebSocket::new(definition);
                    interfaces.insert(definition.info.identifier.clone(), Arc::new(Mutex::new(web_socket)));
                }
            }
        }

        let interfaces_to_register = Arc::new(interfaces.clone());
        for interface in interfaces.values() {
            match interface.try_lock_for(Duration::from_secs(10)) {
                None => {}
                Some(mut interface) => {
                    interface.register_interfaces(Some(interfaces_to_register.clone()));
                }
            }
        }

        Self {
            interfaces,
            //relational_store
        }
    }

    /// Registers all connector handlers with the provided router.
    ///
    /// This method iterates through all managed connectors and registers their inward
    /// endpoints with the router. Each connector adds its own routes based on its
    /// configuration (inward root path, HTTP methods, etc.).
    ///
    /// # Arguments
    ///
    /// * `router` - The router instance to register handlers with
    ///
    /// # Async
    ///
    /// This is an async method because connectors may need to fetch remote definitions
    /// (e.g., `OpenAPI` specs) during registration.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use product_os_connector::ProductOSConnectors;
    /// # use std::collections::BTreeMap;
    /// # async fn example(connectors: ProductOSConnectors) {
    /// let mut router = product_os_router::ProductOSRouter::new();
    /// connectors.setup_handlers(&mut router).await;
    /// # }
    /// ```
    pub async fn setup_handlers(&self, router: &mut product_os_router::ProductOSRouter) {
        for interface in self.interfaces.values() {
            match interface.try_lock_for(Duration::from_secs(10)) {
                None => {}
                Some(mut interface) => {
                    interface.register(router).await;
                }
            }
        }
    }
}



/*
#[async_trait]
impl Feature for ProductOSAuthentication {
    fn identifier(&self) -> String {
        "Authentication".to_string()
    }

    fn register(&self, feature: Arc<dyn Feature>, base_path: String, router: &mut product_os_router::ProductOSRouter) -> RegistryFeature {

    }

    async fn request(&self, request: Request<Body>, version: String) -> Response {

    }

    async fn request_mut(&mut self, request: Request<Body>, version: String) -> Response {

    }
}
*/





#[cfg(all(feature = "connectors", feature = "definition"))]
#[async_trait]
impl Feature for ProductOSConnectors {
    fn identifier(&self) -> String {
        "Connectors".to_string()
    }

    async fn register(&self, feature: Arc<dyn Feature>, base_path: String, router: &mut product_os_router::ProductOSRouter) -> RegistryFeature {
        let shared_base_path = base_path.clone();

        self.setup_handlers(router).await;

        let mut path = shared_base_path;
        path.push_str("/*sub_path");

        RegistryFeature {
            identifier: "Connectors".to_string(),
            paths: vec!(path),
            feature: Some(feature),
            feature_mut: None
        }
    }

    async fn register_mut(&self, _feature: Arc<Mutex<dyn Feature>>, _base_path: String, _router: &mut product_os_router::ProductOSRouter) -> RegistryFeature {
        panic!("Mutable connector server not allowed to be registered")
    }

    async fn request(&self, _action: &What, _input: &Option<serde_json::Value>, _semver: &str) -> Response<Body> {
        Response::builder()
            .status(StatusCode::NOT_IMPLEMENTED)
            .body(Body::from("{}"))
            .unwrap().into_response()
    }

    async fn request_mut(&mut self, action: &What, input: &Option<serde_json::Value>, semver: &str) -> Response<Body> {
        self.request(action, input, semver).await
    }
}