Merge branch 'master' into is5646/use-project-nodes-table-instead-of-workbench

Author: giancarloromeo

packages/models-library/src/models_library/api_schemas_webserver/projects.py

@@ -302,6 +302,9 @@ class ProjectDocument(OutputSchema):
     model_config = ConfigDict(from_attributes=True, arbitrary_types_allowed=True)
 
 
+ProjectDocumentVersion: TypeAlias = int
+
+
 __all__: tuple[str, ...] = (
     "EmptyModel",
     "ProjectCopyOverride",

packages/service-library/src/servicelib/rest_constants.py

@@ -23,3 +23,4 @@ class PydanticExportParametersDict(TypedDict):
 
 # Headers keys
 X_PRODUCT_NAME_HEADER: Final[str] = "X-Simcore-Products-Name"
+X_CLIENT_SESSION_ID_HEADER: Final[str] = "X-Client-Session-Id"

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/notifier/_notifier.py

@@ -11,7 +11,7 @@
 )
 from models_library.api_schemas_webserver.projects_nodes import NodeGet, NodeGetIdle
 from models_library.api_schemas_webserver.socketio import SocketIORoomStr
-from models_library.users import UserID
+from models_library.projects import ProjectID
 from servicelib.fastapi.app_state import SingletonInAppStateMixin
 from servicelib.services_utils import get_status_as_dict
 
@@ -23,20 +23,22 @@ def __init__(self, sio_manager: socketio.AsyncAioPikaManager):
         self._sio_manager = sio_manager
 
     async def notify_service_status(
-        self, user_id: UserID, status: NodeGet | DynamicServiceGet | NodeGetIdle
+        self, project_id: ProjectID, status: NodeGet | DynamicServiceGet | NodeGetIdle
     ) -> None:
         await self._sio_manager.emit(
             SOCKET_IO_SERVICE_STATUS_EVENT,
             data=jsonable_encoder(get_status_as_dict(status)),
-            room=SocketIORoomStr.from_user_id(user_id),
+            room=SocketIORoomStr.from_project_id(project_id),
         )
 
 
 async def notify_service_status_change(
-    app: FastAPI, user_id: UserID, status: NodeGet | DynamicServiceGet | NodeGetIdle
+    app: FastAPI,
+    project_id: ProjectID,
+    status: NodeGet | DynamicServiceGet | NodeGetIdle,
 ) -> None:
     notifier: Notifier = Notifier.get_from_app_state(app)
-    await notifier.notify_service_status(user_id=user_id, status=status)
+    await notifier.notify_service_status(project_id=project_id, status=status)
 
 
 async def lifespan(app: FastAPI) -> AsyncIterator[State]:

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/service_tracker/__init__.py

@@ -1,8 +1,8 @@
 from ._api import (
     NORMAL_RATE_POLL_INTERVAL,
     get_all_tracked_services,
+    get_project_id_for_service,
     get_tracked_service,
-    get_user_id_for_service,
     remove_tracked_service,
     set_frontend_notified_for_service,
     set_if_status_changed_for_service,
@@ -17,11 +17,11 @@
 
 __all__: tuple[str, ...] = (
     "get_all_tracked_services",
+    "get_project_id_for_service",
     "get_tracked_service",
-    "get_user_id_for_service",
-    "service_tracker_lifespan",
     "NORMAL_RATE_POLL_INTERVAL",
     "remove_tracked_service",
+    "service_tracker_lifespan",
     "set_frontend_notified_for_service",
     "set_if_status_changed_for_service",
     "set_request_as_running",

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/service_tracker/_api.py

@@ -11,12 +11,16 @@
     DynamicServiceStop,
 )
 from models_library.api_schemas_webserver.projects_nodes import NodeGet, NodeGetIdle
+from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.services_enums import ServiceState
-from models_library.users import UserID
 from servicelib.deferred_tasks import TaskUID
 
-from ._models import SchedulerServiceState, TrackedServiceModel, UserRequestedState
+from ._models import (
+    SchedulerServiceState,
+    TrackedServiceModel,
+    UserRequestedState,
+)
 from ._setup import get_tracker
 
 _logger = logging.getLogger(__name__)
@@ -242,7 +246,7 @@ async def get_all_tracked_services(app: FastAPI) -> dict[NodeID, TrackedServiceM
     return await get_tracker(app).all()
 
 
-async def get_user_id_for_service(app: FastAPI, node_id: NodeID) -> UserID | None:
-    """returns user_id for the service"""
+async def get_project_id_for_service(app: FastAPI, node_id: NodeID) -> ProjectID | None:
+    """returns project_id for the service"""
     model: TrackedServiceModel | None = await get_tracker(app).load(node_id)
-    return model.user_id if model else None
+    return model.project_id if model else None

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/status_monitor/_deferred_get_status.py

@@ -7,8 +7,8 @@
     RunningDynamicServiceDetails,
 )
 from models_library.api_schemas_webserver.projects_nodes import NodeGet, NodeGetIdle
+from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
-from models_library.users import UserID
 from servicelib.deferred_tasks import BaseDeferredHandler, TaskUID
 from servicelib.deferred_tasks._base_deferred_handler import DeferredContext
 
@@ -69,15 +69,15 @@ async def on_result(
         if await service_tracker.should_notify_frontend_for_service(
             app, node_id, status_changed=status_changed
         ):
-            user_id: UserID | None = await service_tracker.get_user_id_for_service(
-                app, node_id
+            project_id: ProjectID | None = (
+                await service_tracker.get_project_id_for_service(app, node_id)
             )
-            if user_id:
-                await notify_service_status_change(app, user_id, result)
+            if project_id:
+                await notify_service_status_change(app, project_id, result)
                 await service_tracker.set_frontend_notified_for_service(app, node_id)
             else:
                 _logger.info(
-                    "Did not find a user for '%s', skipping status delivery of: %s",
+                    "Did not find a project for '%s', skipping status delivery of: %s",
                     node_id,
                     result,
                 )

services/web/server/src/simcore_service_webserver/db_listener/_db_comp_tasks_listening_task.py

@@ -52,7 +52,12 @@ async def _update_project_state(
     node_errors: list[ErrorDict] | None,
 ) -> None:
     project = await _projects_service.update_project_node_state(
-        app, user_id, project_uuid, node_uuid, new_state
+        app,
+        user_id,
+        project_uuid,
+        node_uuid,
+        new_state,
+        client_session_id=None,  # <-- The trigger for this update is not from the UI (its db listener)
     )
 
     await _projects_service.notify_project_node_update(
@@ -95,6 +100,7 @@ async def _handle_db_notification(
                 changed_row.run_hash,
                 node_errors=changed_row.errors,
                 ui_changed_keys=None,
+                client_session_id=None,  # <-- The trigger for this update is not from the UI (its db listener)
             )
 
         if "state" in payload.changes and (changed_row.state is not None):

services/web/server/src/simcore_service_webserver/folders/_workspaces_repository.py

@@ -8,6 +8,7 @@
 from simcore_postgres_database.utils_repos import transaction_context
 
 from ..db.plugin import get_asyncpg_engine
+from ..models import ClientSessionID
 from ..projects import _folders_repository as projects_folders_repository
 from ..projects import _groups_repository as projects_groups_repository
 from ..projects._access_rights_service import check_user_project_permission
@@ -26,6 +27,7 @@ async def move_folder_into_workspace(
     folder_id: FolderID,
     workspace_id: WorkspaceID | None,
     product_name: ProductName,
+    client_session_id: ClientSessionID | None = None,
 ) -> None:
     # 1. User needs to have delete permission on source folder
     folder_db = await _folders_repository.get(
@@ -84,6 +86,7 @@ async def move_folder_into_workspace(
                 project_uuid=project_id,
                 patch_project_data={"workspace_id": workspace_id},
                 user_primary_gid=user["primary_gid"],
+                client_session_id=client_session_id,
             )
 
         # 5. BATCH update of folders with workspace_id

services/web/server/src/simcore_service_webserver/folders/_workspaces_rest.py

@@ -2,10 +2,14 @@
 
 from aiohttp import web
 from servicelib.aiohttp import status
-from servicelib.aiohttp.requests_validation import parse_request_path_parameters_as
+from servicelib.aiohttp.requests_validation import (
+    parse_request_headers_as,
+    parse_request_path_parameters_as,
+)
 
 from .._meta import api_version_prefix as VTAG
 from ..login.decorators import login_required
+from ..models import ClientSessionHeaderParams
 from ..security.decorators import permission_required
 from . import _workspaces_repository
 from ._common.exceptions_handlers import handle_plugin_requests_exceptions
@@ -27,12 +31,14 @@
 async def move_folder_to_workspace(request: web.Request):
     req_ctx = FoldersRequestContext.model_validate(request)
     path_params = parse_request_path_parameters_as(FolderWorkspacesPathParams, request)
+    header_params = parse_request_headers_as(ClientSessionHeaderParams, request)
 
     await _workspaces_repository.move_folder_into_workspace(
         app=request.app,
         user_id=req_ctx.user_id,
         folder_id=path_params.folder_id,
         workspace_id=path_params.workspace_id,
         product_name=req_ctx.product_name,
+        client_session_id=header_params.client_session_id,
     )
     return web.json_response(status=status.HTTP_204_NO_CONTENT)

services/web/server/src/simcore_service_webserver/models.py

@@ -3,9 +3,10 @@
 from models_library.products import ProductName
 from models_library.rest_base import RequestParameters
 from models_library.users import UserID
-from pydantic import ConfigDict, Field
+from pydantic import ConfigDict, Field, StringConstraints
 from pydantic_extra_types.phone_numbers import PhoneNumberValidator
 from servicelib.request_keys import RQT_USERID_KEY
+from servicelib.rest_constants import X_CLIENT_SESSION_ID_HEADER
 
 from .constants import RQ_PRODUCT_KEY
 
@@ -16,6 +17,18 @@
 ]
 
 
+ClientSessionID: TypeAlias = Annotated[
+    str,
+    StringConstraints(
+        strip_whitespace=True,
+        min_length=36,
+        max_length=36,
+        pattern=r"^[0-9a-fA-F\-]{36}$",  # UUID format
+        strict=True,
+    ),
+]
+
+
 class AuthenticatedRequestContext(RequestParameters):
     """Fields expected in the request context for authenticated endpoints"""
 
@@ -25,3 +38,20 @@ class AuthenticatedRequestContext(RequestParameters):
     model_config = ConfigDict(
         frozen=True  # prevents modifications after middlewares creates this model
     )
+
+
+assert X_CLIENT_SESSION_ID_HEADER
+
+
+class ClientSessionHeaderParams(RequestParameters):
+    """Header parameters for client session tracking in collaborative features."""
+
+    client_session_id: ClientSessionID | None = Field(
+        default=None,
+        alias="X-Client-Session-Id",  # X_CLIENT_SESSION_ID_HEADER,
+        description="Client session identifier for collaborative features (UUID string)",
+    )
+
+    model_config = ConfigDict(
+        validate_by_name=True,
+    )