Merge pull request #42 from BrainDriveAI/feature/plugin-services-runtime

Feature/plugin services runtime

Author: bekmuradov

.gitignore

@@ -47,6 +47,8 @@ backend/alembic/versions/
 *.log
 backend/logs/
 
+backend/services_runtime/
+
 # Environment variables
 .env
 .env.local

backend/app/dto/__init__.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel
+from typing import List, Optional
+from datetime import datetime
+
+# This schema is used for returning data from the repository.
+# It ensures that JSON fields are correctly converted to Python types.
+class PluginServiceRuntimeDTO(BaseModel):
+    """
+    A Pydantic model to represent a PluginServiceRuntime object,
+    with required_env_vars as a list of strings.
+    """
+    id: str
+    plugin_id: str
+    plugin_slug: str
+    name: str
+    source_url: Optional[str] = None
+    type: Optional[str] = None
+    install_command: Optional[str] = None
+    start_command: Optional[str] = None
+    healthcheck_url: Optional[str] = None
+    required_env_vars: List[str] = []
+    status: Optional[str] = None
+    user_id: str
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None

backend/app/dto/plugin.py

@@ -5,6 +5,7 @@
 from app.api.v1.api import api_router
 from app.core.config import settings
 from app.routers.plugins import plugin_manager
+from backend.app.plugins.service_installler.start_stop_plugin_services import start_plugin_services, stop_plugin_services
 import logging
 import time
 import structlog
@@ -29,8 +30,21 @@ async def startup_event():
     logger.info("Initializing application settings...")
     from app.init_settings import init_ollama_settings
     await init_ollama_settings()
+    # Start plugin services
+    await start_plugin_services()
     logger.info("Settings initialization completed")
 
+
+# Add shutdown event to gracefully stop services
+@app.on_event("shutdown")
+async def shutdown_event():
+    """Gracefully stop all plugin services on application shutdown."""
+    logger.info("Shutting down application and stopping plugin services...")
+    # Stop all plugin services gracefully
+    await stop_plugin_services()
+    logger.info("Application shutdown completed.")
+
+
 # Add middleware to log all requests
 logger = structlog.get_logger()
 
@@ -94,5 +108,3 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
 
 # Include API routers
 app.include_router(api_router)
-
-

backend/app/main.py

@@ -1,10 +1,13 @@
-from sqlalchemy import Column, String, Integer, Boolean, ForeignKey, Text, JSON, UniqueConstraint, TIMESTAMP
+from sqlalchemy import Column, String, Integer, Boolean, ForeignKey, Text, JSON, UniqueConstraint, TIMESTAMP, DateTime
 import sqlalchemy
 from sqlalchemy.orm import relationship
 from sqlalchemy.sql import func
+from datetime import datetime, UTC
+import json
 
 from app.models.base import Base
 
+
 class Plugin(Base):
     """SQLAlchemy model for plugins."""
 
@@ -44,6 +47,7 @@ class Plugin(Base):
     config_fields = Column(Text)  # Stored as JSON string
     messages = Column(Text)       # Stored as JSON string
     dependencies = Column(Text)   # Stored as JSON string
+    required_services_runtime = Column(Text, nullable=True)
 
     # Timestamps
     created_at = Column(String, default=func.now())
@@ -60,6 +64,7 @@ class Plugin(Base):
 
     # Relationships
     modules = relationship("Module", back_populates="plugin", cascade="all, delete-orphan")
+    service_runtimes = relationship("PluginServiceRuntime", back_populates="plugin", cascade="all, delete-orphan")
 
     def to_dict(self):
         """Convert model to dictionary."""
@@ -118,6 +123,11 @@ def to_dict(self):
         else:
             result["permissions"] = []
 
+        if self.required_services_runtime:
+            result["requiredServicesRuntime"] = json.loads(self.required_services_runtime)
+        else:
+            result["requiredServicesRuntime"] = []
+
         return result
 
     @classmethod
@@ -162,10 +172,81 @@ def from_dict(cls, data):
         # Remove modules from data as they are handled separately
         if "modules" in db_data:
             db_data.pop("modules")
+
+        # Handle service runtimes (only store names in plugin table)
+        if "requiredServicesRuntime" in db_data and db_data["requiredServicesRuntime"] is not None:
+            db_data["required_services_runtime"] = json.dumps([
+                r["name"] for r in db_data["requiredServicesRuntime"]
+            ])
+            db_data.pop("requiredServicesRuntime")
 
         return cls(**db_data)
 
 
+class PluginServiceRuntime(Base):
+    """SQLAlchemy model for plugin service runtimes."""
+
+    __tablename__ = "plugin_service_runtime"
+
+    id = Column(String, primary_key=True, index=True)
+    plugin_id = Column(String, ForeignKey("plugin.id"), nullable=False, index=True)
+    plugin_slug = Column(String, nullable=False, index=True)
+
+    name = Column(String, nullable=False)
+    source_url = Column(String)
+    type = Column(String)
+    install_command = Column(Text)
+    start_command = Column(Text)
+    healthcheck_url = Column(String)
+    required_env_vars = Column(Text)  # store as JSON string
+    status = Column(String, default="pending")
+
+    created_at = Column(DateTime, default=datetime.now(UTC))
+    updated_at = Column(DateTime, default=datetime.now(UTC), onupdate=datetime.now(UTC))
+
+    user_id = Column(String(32), ForeignKey("users.id", name="fk_plugin_service_runtime_user_id"), nullable=False)
+    user = relationship("User")
+
+    # Relationship back to plugin
+    plugin = relationship("Plugin", back_populates="service_runtimes")
+
+    def to_dict(self):
+        """
+        Convert the model instance to a dictionary, handling JSON fields.
+        """
+        return {
+            "id": self.id,
+            "plugin_id": self.plugin_id,
+            "plugin_slug": self.plugin_slug,
+            "name": self.name,
+            "source_url": self.source_url,
+            "type": self.type,
+            "install_command": self.install_command,
+            "start_command": self.start_command,
+            "healthcheck_url": self.healthcheck_url,
+            "required_env_vars": json.loads(self.required_env_vars) if self.required_env_vars else [],
+            "status": self.status,
+            "user_id": self.user_id,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        """
+        Create a new instance from a dictionary, serializing JSON fields.
+        """
+        db_data = data.copy()
+        
+        if "required_env_vars" in db_data and db_data["required_env_vars"] is not None:
+            db_data["required_env_vars"] = json.dumps(db_data["required_env_vars"])
+            
+        # Handle conversion from camelCase to snake_case if necessary
+        # For simplicity, we are assuming keys in the incoming dict match model attributes
+        
+        return cls(**db_data)
+
+
 class Module(Base):
     """SQLAlchemy model for plugin modules."""
 

backend/app/models/plugin.py

@@ -20,6 +20,8 @@
 from typing import Dict, Any, Optional, List, Tuple
 from urllib.parse import urlparse
 import structlog
+from .service_installler.plugin_service_manager import install_and_start_docker_service
+from .service_installler.service_runtime_extractor import extract_required_services_runtime
 
 logger = structlog.get_logger()
 
@@ -237,6 +239,15 @@ async def install_from_url(self, repo_url: str, user_id: str, version: str = "la
             if install_result['success']:
                 logger.info(f"Plugin installation successful: {install_result}")
 
+                service_runtime: list = validation_result.get("service_runtime", [])
+                logger.info(f"\n\n>>>>>>>>SERVICE RUNTIME\n\n: {service_runtime}\n\n>>>>>>>>>>")
+                if service_runtime:
+                    plugin_slug = validation_result["plugin_info"].get("plugin_slug")
+                    await install_and_start_docker_service(
+                        service_runtime,
+                        plugin_slug
+                    )
+
                 # Store installation metadata
                 try:
                     await self._store_installation_metadata(
@@ -718,6 +729,7 @@ async def _validate_plugin_structure(self, plugin_dir: Path) -> Dict[str, Any]:
 
             # Try to load plugin metadata
             plugin_info = {}
+            service_runtime = []
 
             # Check package.json
             package_json_path = plugin_dir / 'package.json'
@@ -828,6 +840,12 @@ async def _validate_plugin_structure(self, plugin_dir: Path) -> Dict[str, Any]:
                                     extracted_slug = slug_match.group(1)
                                     plugin_info['plugin_slug'] = extracted_slug
                                     logger.info(f"Extracted plugin_slug from source: {extracted_slug}")
+
+                                # Extract services using the dedicated function
+                                services = extract_required_services_runtime(content, plugin_info.get('plugin_slug'))
+                                if services:
+                                    plugin_info['required_services_runtime'] = services
+                                    service_runtime.extend(services)
                     except Exception as extract_error:
                         logger.warning(f"Could not extract plugin_slug from source: {extract_error}")
 
@@ -839,7 +857,8 @@ async def _validate_plugin_structure(self, plugin_dir: Path) -> Dict[str, Any]:
 
             return {
                 'valid': True,
-                'plugin_info': plugin_info
+                'plugin_info': plugin_info,
+                'service_runtime': service_runtime
             }
 
         except Exception as e:

backend/app/plugins/remote_installer.py

@@ -6,7 +6,8 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 import structlog
 
-from app.models.plugin import Plugin, Module
+from app.models.plugin import Plugin, Module, PluginServiceRuntime
+from app.dto.plugin import PluginServiceRuntimeDTO
 
 logger = structlog.get_logger()
 
@@ -85,7 +86,7 @@ async def get_all_plugins_with_modules(self, user_id: str = None) -> List[Dict[s
             plugin_dicts = []
             for plugin in plugins:
                 plugin_dict = plugin.to_dict()
-                
+
                 # Get modules for this plugin
                 modules_query = select(Module).where(Module.plugin_id == plugin.id)
 
@@ -104,6 +105,25 @@ async def get_all_plugins_with_modules(self, user_id: str = None) -> List[Dict[s
         except Exception as e:
             logger.error("Error getting plugins with modules", error=str(e))
             raise
+
+    async def get_all_service_runtimes(self) -> List[PluginServiceRuntimeDTO]:
+        """
+        Get all plugin service runtimes for startup and return them as DTOs.
+        """
+        try:
+            query = select(PluginServiceRuntime).where(
+                PluginServiceRuntime.status.in_(["pending", "stopped", "running"])
+            )
+            
+            result = await self.db.execute(query)
+            services = result.scalars().all()
+            
+            # Convert SQLAlchemy models to Pydantic DTOs for a typed return
+            return [PluginServiceRuntimeDTO(**service.to_dict()) for service in services]
+            
+        except Exception as e:
+            logger.error("Error getting service runtimes", error=str(e))
+            raise
 
     async def get_plugin(self, plugin_id: str) -> Optional[Dict[str, Any]]:
         """Get a specific plugin by ID."""