Merge pull request #34 from OpenMined/feat/sync-endpoints

feat: implement asynchronous endpoint synchronization to marketplaces…

Author: shubham3121

backend/syft_space/components/endpoints/handlers.py

@@ -1,5 +1,6 @@
 """Endpoint handlers for business logic."""
 
+from typing import Any
 from uuid import UUID
 
 from fastapi import HTTPException
@@ -725,41 +726,7 @@ async def _publish_to_marketplace(
                     username=marketplace.email, password=marketplace.password
                 )
 
-                endpoint_type = (
-                    "model" if endpoint.model_id is not None else "data_source"
-                )
-                policies = [
-                    {
-                        "type": policy.policy_type,
-                        "version": "1.0",
-                        "enabled": True,
-                        "description": policy.name,
-                        "config": policy.configuration,
-                    }
-                    for policy in endpoint.policies
-                ]
-                connection_config = {
-                    "path": f"/api/v1/endpoints/{endpoint.slug}/query",
-                }
-
-                payload = {
-                    "name": endpoint.name,
-                    "description": endpoint.summary or "",
-                    "type": endpoint_type,
-                    "visibility": "public",
-                    "version": "0.1.0",
-                    "readme": endpoint.description or "",
-                    "slug": endpoint.slug,
-                    "policies": policies,
-                    "connect": [
-                        {
-                            "type": "https",
-                            "enabled": True,
-                            "description": "",
-                            "config": connection_config,
-                        }
-                    ],
-                }
+                payload = self._build_publish_payload(endpoint)
                 await client.publish_endpoint(payload, overwrite=True)
 
         except SyftHubError as e:
@@ -920,3 +887,125 @@ async def _check_marketplace_availability(
                 available=None,
                 error=str(e),
             )
+
+    def _build_publish_payload(self, endpoint: Endpoint) -> dict[str, Any]:
+        """Build the publish payload for an endpoint.
+
+        Args:
+            endpoint: Endpoint entity
+
+        Returns:
+            Dict payload for publish/sync APIs
+        """
+        endpoint_type = "model" if endpoint.model_id is not None else "data_source"
+
+        policies = [
+            {
+                "type": policy.policy_type,
+                "version": "1.0",
+                "enabled": True,
+                "description": policy.name,
+                "config": policy.configuration,
+            }
+            for policy in endpoint.policies
+        ]
+
+        connection_config = {
+            "path": f"/api/v1/endpoints/{endpoint.slug}/query",
+        }
+
+        return {
+            "name": endpoint.name,
+            "description": endpoint.summary or "",
+            "type": endpoint_type,
+            "visibility": "public",
+            "version": "0.1.0",
+            "readme": endpoint.description or "",
+            "slug": endpoint.slug,
+            "policies": policies,
+            "connect": [
+                {
+                    "type": "https",
+                    "enabled": True,
+                    "description": "",
+                    "config": connection_config,
+                }
+            ],
+        }
+
+    async def sync_endpoints_to_marketplaces(
+        self, tenant: Tenant
+    ) -> dict[str, list[str]]:
+        """Sync all published endpoints to their respective marketplaces.
+
+        Groups endpoints by marketplace and calls sync_endpoints API for each.
+
+        Args:
+            tenant: Tenant context
+
+        Returns:
+            Dict mapping marketplace_id -> list of synced endpoint slugs
+        """
+        if not self.marketplace_repository:
+            logger.warning("Marketplace repository not configured, skipping sync")
+            return {}
+
+        # Get all published endpoints
+        endpoints = await self.endpoint_repository.get_published_endpoints(tenant.id)
+        if not endpoints:
+            logger.debug("No published endpoints to sync")
+            return {}
+
+        # Group endpoints by marketplace
+        marketplace_endpoints: dict[UUID, list[Endpoint]] = {}
+        for endpoint in endpoints:
+            for marketplace_id in endpoint.published_to:
+                marketplace_endpoints.setdefault(marketplace_id, []).append(endpoint)
+
+        results: dict[str, list[str]] = {}
+
+        # Sync to each marketplace
+        for marketplace_id, eps in marketplace_endpoints.items():
+            try:
+                marketplace = await self.marketplace_repository.get_by_id(
+                    UUID(marketplace_id), tenant.id
+                )
+                if not marketplace:
+                    logger.warning(f"Marketplace {marketplace_id} not found, skipping")
+                    continue
+
+                if not marketplace.is_active:
+                    logger.warning(f"Marketplace {marketplace_id} not active, skipping")
+                    continue
+
+                if not marketplace.email or not marketplace.password:
+                    logger.warning(
+                        f"Marketplace {marketplace_id} missing credentials, skipping"
+                    )
+                    continue
+
+                # Build payloads
+                payloads = [self._build_publish_payload(ep) for ep in eps]
+
+                # Call sync API
+                async with SyftHubClient(base_url=marketplace.url) as client:
+                    await client.login(
+                        username=marketplace.email, password=marketplace.password
+                    )
+                    await client.sync_endpoints(payloads)
+
+                results[marketplace_id] = [ep.slug for ep in eps]
+                logger.info(
+                    f"Synced {len(eps)} endpoints to marketplace {marketplace.name}"
+                )
+
+            except SyftHubError as e:
+                logger.warning(
+                    f"Failed to sync to marketplace {marketplace_id}: {e.message}"
+                )
+            except Exception as e:
+                logger.error(
+                    f"Unexpected error syncing to marketplace {marketplace_id}: {e}"
+                )
+
+        return results

backend/syft_space/components/endpoints/repository.py

@@ -4,7 +4,7 @@
 from uuid import UUID
 
 from sqlalchemy.orm import selectinload
-from sqlmodel import select
+from sqlmodel import or_, select
 
 from syft_space.components.endpoints.entities import Endpoint
 from syft_space.components.shared.database import AsyncBaseRepository, AsyncDatabase
@@ -254,3 +254,28 @@ async def remove_publication(
                 await session.refresh(endpoint)
 
             return endpoint
+
+    async def get_published_endpoints(self, tenant_id: UUID) -> list[Endpoint]:
+        """Get all endpoints that are published to at least one marketplace.
+
+        Args:
+            tenant_id: Tenant ID
+
+        Returns:
+            List of published endpoints with policies eagerly loaded
+        """
+        async with self.db.get_session() as session:
+            statement = (
+                select(Endpoint)
+                .where(
+                    Endpoint.tenant_id == tenant_id,
+                    or_(Endpoint.published_to.isnot(None), Endpoint.published_to != []),
+                )
+                .options(
+                    selectinload(Endpoint.model),
+                    selectinload(Endpoint.dataset),
+                    selectinload(Endpoint.policies),
+                )
+            )
+            result = await session.exec(statement)
+            return list(result.all())

backend/syft_space/components/shared/syfthub_client.py

@@ -554,6 +554,23 @@ async def verify_satellite_token(self, token: str) -> SatelliteToken:
         response = await self._client.post("/api/v1/verify", json={"token": token})  # type: ignore
         return _handle_response(response, SatelliteToken)
 
+    async def sync_endpoints(self, payload: list[dict[str, Any]]) -> dict[str, Any]:
+        """Sync endpoints to SyftHub.
+
+        It is used to sync endpoints from the database to SyftHub.
+        This is a destructive operation that will overwrite the existing endpoints with the new ones.
+
+        Args:
+            payload: List of endpoints to sync
+        Returns:
+            dict[str, Any]: Sync endpoints response
+        """
+        self._require_auth()
+        response = await self._client.post(
+            "/api/v1/endpoints/sync", json={"endpoints": payload}
+        )  # type: ignore
+        return _handle_response_raw(response)
+
     def _require_auth(self) -> None:
         if self._client is None:
             raise NotAuthenticatedError()

backend/syft_space/main.py

@@ -123,6 +123,35 @@ async def _sync_public_url_safe(
         logger.warning(f"Marketplace sync failed: {e} - server will continue starting")
 
 
+async def _sync_endpoints_safe(handler: EndpointHandler, tenant: Tenant) -> None:
+    """Sync published endpoints to marketplaces without blocking startup.
+
+    This is a fire-and-forget helper that syncs all published endpoints to their
+    respective marketplaces. If the sync fails or times out, the server continues
+    operating - eventual consistency is achieved via frontend publish calls.
+
+    Args:
+        handler: Endpoint handler instance
+        tenant: Tenant context
+    """
+    try:
+        results = await asyncio.wait_for(
+            handler.sync_endpoints_to_marketplaces(tenant),
+            timeout=60.0,  # 60 second timeout for multi-marketplace sync
+        )
+        if results:
+            total = sum(len(slugs) for slugs in results.values())
+            logger.info(
+                f"Startup endpoint sync completed: {total} endpoints to {len(results)} marketplaces"
+            )
+        else:
+            logger.debug("No published endpoints to sync on startup")
+    except asyncio.TimeoutError:
+        logger.warning("Endpoint sync timed out - server will continue running")
+    except Exception as e:
+        logger.warning(f"Endpoint sync failed: {e} - server will continue running")
+
+
 async def _setup_tenant_and_settings(
     tenant_repo: TenantRepository,
     settings_hdlr: SettingsHandler,
@@ -202,9 +231,13 @@ async def lifespan(app: FastAPI):
                 _sync_public_url_safe(settings_handler, default_tenant, public_url)
             )
 
+    # 7. Fire-and-forget: Sync published endpoints to marketplaces
+    if default_tenant:
+        asyncio.create_task(_sync_endpoints_safe(endpoint_handler, default_tenant))
+
     yield  # Application runs here
 
-    # 7. Shutdown all services (reverse order: ingestion → provisioner → proxy)
+    # 8. Shutdown all services (reverse order: ingestion → provisioner → proxy)
     for name, service in reversed(services):
         try:
             await service.shutdown()