fix(DATAGO-120949): Move migrations to __init__ for sequential execution (#741)

Both Platform and WebUI migrations now run in __init__() during component
creation, ensuring sequential execution and preventing Alembic/SQLAlchemy
race conditions.

Changes:
- Platform: Migrations in __init__() via _run_database_migrations()
- WebUI: Migrations in __init__() via _run_database_migrations()
- Removed unused component parameter from _setup_database()
- Simplified comments and docstrings
- Added Alembic log prefixing for better debugging

Author: mo-radwan1

src/solace_agent_mesh/gateway/http_sse/component.py

@@ -308,8 +308,27 @@ def __init__(self, **kwargs):
                 "%s Data retention is disabled via configuration.", self.log_identifier
             )
 
+        if self.database_url:
+            log.info("%s Running database migrations...", self.log_identifier)
+            self._run_database_migrations()
+            log.info("%s Database migrations completed", self.log_identifier)
+
         log.info("%s Web UI Backend Component initialized.", self.log_identifier)
 
+    def _run_database_migrations(self):
+        """Run database migrations synchronously during __init__."""
+        try:
+            from ...gateway.http_sse.main import _setup_database
+            _setup_database(self.database_url)
+        except Exception as e:
+            log.error(
+                "%s Failed to run database migrations: %s",
+                self.log_identifier,
+                e,
+                exc_info=True
+            )
+            raise RuntimeError(f"Database migration failed during component initialization: {e}") from e
+
     def process_event(self, event: Event):
         if event.event_type == EventType.TIMER:
             timer_id = event.data.get("timer_id")
@@ -1283,7 +1302,7 @@ def _start_fastapi_server(self):
 
             self.fastapi_app = fastapi_app_instance
 
-            setup_dependencies(self, self.database_url)
+            setup_dependencies(self)
 
             # Instantiate services that depend on the database session factory.
             # This must be done *after* setup_dependencies has run.

src/solace_agent_mesh/gateway/http_sse/main.py

@@ -48,14 +48,13 @@
 
 log = logging.getLogger(__name__)
 
+
 app = FastAPI(
     title="A2A Web UI Backend",
     version="1.0.0",  # Updated to reflect simplified architecture
     description="Backend API and SSE server for the A2A Web UI, hosted by Solace AI Connector.",
 )
 
-# Global flag to track if dependencies have been initialized
-_dependencies_initialized = False
 
 
 
@@ -77,58 +76,44 @@ def _run_community_migrations(database_url: str) -> None:
     try:
         from sqlalchemy import create_engine
 
-        log.info("Starting community migrations...")
+        log.info("[WebUI Gateway] Starting community migrations...")
         engine = create_engine(database_url)
         inspector = sa.inspect(engine)
         existing_tables = inspector.get_table_names()
 
+        alembic_cfg = _setup_alembic_config(database_url)
         if not existing_tables or "sessions" not in existing_tables:
-            log.info("Running initial community database setup")
-            alembic_cfg = _setup_alembic_config(database_url)
-            command.upgrade(alembic_cfg, "head")
-            log.info("Community database migrations completed")
+            log.info("[WebUI Gateway] Running initial database setup")
         else:
-            log.info("Checking for community schema updates")
-            alembic_cfg = _setup_alembic_config(database_url)
-            command.upgrade(alembic_cfg, "head")
-            log.info("Community database schema is current")
+            log.info("[WebUI Gateway] Checking for schema updates")
+
+        command.upgrade(alembic_cfg, "head")
+        log.info("[WebUI Gateway] Community migrations completed")
     except Exception as e:
-        log.warning(
-            "Community migration check failed: %s - attempting to run migrations",
-            e,
-        )
+        log.warning("[WebUI Gateway] Migration check failed: %s - attempting to run migrations", e)
         try:
             alembic_cfg = _setup_alembic_config(database_url)
             command.upgrade(alembic_cfg, "head")
-            log.info("Community database migrations completed")
+            log.info("[WebUI Gateway] Community migrations completed")
         except Exception as migration_error:
-            log.error("Community migration failed: %s", migration_error)
-            log.error("Check database connectivity and permissions")
+            log.error("[WebUI Gateway] Migration failed: %s", migration_error)
+            log.error("[WebUI Gateway] Check database connectivity and permissions")
             raise RuntimeError(
                 f"Community database migration failed: {migration_error}"
             ) from migration_error
 
 
 
 
-def _setup_database(
-    component: "WebUIBackendComponent",
-    database_url: str,
-) -> None:
+def _setup_database(database_url: str) -> None:
     """
-    Initialize database and run migrations for WebUI Gateway (chat only).
-
-    Platform database is no longer used by WebUI Gateway.
-    Platform migrations are handled by Platform Service.
+    Initialize database and run migrations for WebUI Gateway.
 
     Args:
-        component: WebUIBackendComponent instance
-        database_url: Chat database URL (sessions, tasks, feedback) - REQUIRED
+        database_url: Chat database URL (sessions, tasks, feedback)
     """
     dependencies.init_database(database_url)
-    log.info("Persistence enabled - sessions will be stored in database")
-    log.info("Running database migrations...")
-
+    log.info("[WebUI Gateway] Running database migrations...")
     _run_community_migrations(database_url)
 
 
@@ -164,49 +149,24 @@ def _create_api_config(app_config: dict, database_url: str) -> dict:
     }
 
 
-def setup_dependencies(
-    component: "WebUIBackendComponent",
-    database_url: str = None,
-):
+def setup_dependencies(component: "WebUIBackendComponent"):
     """
-    Initialize dependencies for WebUI Gateway (chat only).
+    Initialize FastAPI dependencies (middleware, routers, static files).
+    Database migrations are handled in component.__init__().
 
     Args:
         component: WebUIBackendComponent instance
-        database_url: Chat database URL (sessions, tasks, feedback).
-                     If None, runs in compatibility mode with in-memory sessions.
-
-    This function is idempotent and safe to call multiple times.
     """
-    global _dependencies_initialized
-
-    if _dependencies_initialized:
-        log.debug("[setup_dependencies] Dependencies already initialized, skipping")
-        return
-
     dependencies.set_component_instance(component)
 
-    if database_url:
-        _setup_database(component, database_url)
-    else:
-        log.warning(
-            "No database URL provided - using in-memory session storage (data not persisted across restarts)"
-        )
-        log.info("This maintains backward compatibility for existing SAM installations")
-
     app_config = _get_app_config(component)
-    api_config_dict = _create_api_config(app_config, database_url)
-
+    api_config_dict = _create_api_config(app_config, component.database_url)
     dependencies.set_api_config(api_config_dict)
-    log.debug("API configuration extracted and stored.")
 
     _setup_middleware(component)
     _setup_routers()
     _setup_static_files()
 
-    _dependencies_initialized = True
-    log.debug("[setup_dependencies] Dependencies initialization complete")
-
 
 def _setup_middleware(component: "WebUIBackendComponent") -> None:
     allowed_origins = component.get_cors_origins()

src/solace_agent_mesh/services/platform/api/main.py

@@ -23,8 +23,6 @@
     description="Platform configuration management API (agents, connectors, toolsets, deployments)",
 )
 
-# Global flag to track initialization (idempotent)
-_dependencies_initialized = False
 
 
 def _setup_alembic_config(database_url: str) -> Config:
@@ -96,90 +94,45 @@ def _run_enterprise_migrations(database_url: str) -> None:
     This is optional and only runs if the enterprise package is available.
 
     Args:
-        component: PlatformServiceComponent instance.
         database_url: Database connection string.
     """
     try:
         from solace_agent_mesh_enterprise.platform_service.migration_runner import run_migrations
 
-        log.info("Starting enterprise platform migrations...")
+        log.info("[Platform Service] Starting enterprise migrations...")
         run_migrations(database_url)
-        log.info("Enterprise platform migrations completed")
+        log.info("[Platform Service] Enterprise migrations completed successfully")
     except ImportError:
-        log.debug("Enterprise platform module not found - skipping enterprise migrations")
+        log.debug("[Platform Service] Enterprise module not found - skipping enterprise migrations")
     except Exception as e:
-        log.error("Enterprise platform migration failed: %s", e)
-        log.error("Enterprise platform features may be unavailable")
+        log.error("[Platform Service] Enterprise migration failed: %s", e)
+        log.error("[Platform Service] Enterprise features may be unavailable")
         raise RuntimeError(f"Enterprise platform database migration failed: {e}") from e
 
 
 def _setup_database(database_url: str) -> None:
-    """
-    Initialize database connection and run all required migrations.
-    Sets up both community and enterprise platform database schemas.
-
-    This follows the same pattern as gateway/http_sse:
-    1. Initialize database (create engine and SessionLocal)
-    2. Run community migrations
-    3. Run enterprise migrations (if available)
-
-    Args:
-        component: PlatformServiceComponent instance.
-        database_url: Platform database URL (agents, connectors, deployments, toolsets) - REQUIRED
-    """
-    from . import dependencies
-
-    # MIGRATION PHASE 1: DB is initialized in enterprise repo. In next phase, platform db will be initialized here
-    # dependencies.init_database(database_url)
-    # log.info("Platform database initialized - running migrations...")
-
-    # MIGRATION PHASE 1: No community migrations yet - only enterprise migrations
-    # _run_community_migrations(database_url)
+    """Initialize database and run migrations."""
+    log.info("[Platform Service] Initializing database and running migrations...")
     _run_enterprise_migrations(database_url)
+    log.info("[Platform Service] Database initialization complete")
 
 
-def setup_dependencies(component: "PlatformServiceComponent", database_url: str):
+def setup_dependencies(component: "PlatformServiceComponent"):
     """
-    Initialize dependencies for the Platform Service.
-
-    This function is idempotent and safe to call multiple times.
-    It sets up:
-    1. Component instance reference
-    2. Database connection
-    3. Middleware (CORS, OAuth2)
-    4. Routers (community and enterprise)
+    Initialize FastAPI dependencies (middleware, routers).
+    Database migrations are handled in component.__init__().
 
     Args:
-        component: PlatformServiceComponent instance.
-        database_url: Database connection string.
+        component: PlatformServiceComponent instance
     """
-    global _dependencies_initialized
-
-    if _dependencies_initialized:
-        log.debug("Platform service dependencies already initialized, skipping")
-        return
-
     log.info("Initializing Platform Service dependencies...")
 
-    # Store component reference for dependency injection
     from . import dependencies
-
     dependencies.set_component_instance(component)
 
-    # Initialize database and run migrations
-    if database_url:
-        _setup_database(database_url)
-        log.info("Platform database initialized with migrations")
-    else:
-        log.warning("No database URL provided - platform service will not function")
-
-    # Setup middleware
     _setup_middleware(component)
-
-    # Setup routers
     _setup_routers()
 
-    _dependencies_initialized = True
     log.info("Platform Service dependencies initialized successfully")
 
 

src/solace_agent_mesh/services/platform/component.py

@@ -175,13 +175,27 @@ def __init__(self, **kwargs):
         self.background_scheduler = None
         self.background_tasks_thread = None
 
-        # Direct message publisher for deployer commands
         self.direct_publisher = None
 
+        log.info("%s Running database migrations...", self.log_identifier)
+        self._run_database_migrations()
+        log.info("%s Database migrations completed", self.log_identifier)
+
         log.info("%s Platform Service Component initialized.", self.log_identifier)
 
-        # Note: FastAPI server, direct publisher, and background tasks are started
-        # in _late_init() after SamComponentBase.run() is called and broker is ready
+    def _run_database_migrations(self):
+        """Run database migrations synchronously during __init__."""
+        try:
+            from .api.main import _setup_database
+            _setup_database(self.database_url)
+        except Exception as e:
+            log.error(
+                "%s Failed to run database migrations: %s",
+                self.log_identifier,
+                e,
+                exc_info=True
+            )
+            raise RuntimeError(f"Database migration failed during component initialization: {e}") from e
 
     def _late_init(self):
         """
@@ -234,8 +248,7 @@ def _start_fastapi_server(self):
 
             self.fastapi_app = fastapi_app_instance
 
-            # Setup dependencies (idempotent - safe to call multiple times)
-            setup_dependencies(self, self.database_url)
+            setup_dependencies(self)
 
             # Register startup event for background tasks
             @self.fastapi_app.on_event("startup")