Emit events for collaborative sessions

jupyter_collaboration/app.py

@@ -7,6 +7,7 @@
 
 from .handlers import DocSessionHandler, YDocWebSocketHandler
 from .stores import SQLiteYStore
+from .utils import EVENTS_SCHEMA_PATH
 
 
 class YDocExtension(ExtensionApp):
@@ -45,6 +46,10 @@ class YDocExtension(ExtensionApp):
         directory.""",
     )
 
+    def initialize(self):
+        super().initialize()
+        self.serverapp.event_logger.register_event_schema(EVENTS_SCHEMA_PATH)
+
     def initialize_settings(self):
         self.settings.update(
             {

jupyter_collaboration/events/session.yaml

@@ -0,0 +1,59 @@
+"$id": https://schema.jupyter.org/jupyter_collaboration/session/v1
+version: 1
+title: Collaborative session events
+personal-data: true
+description: |
+  Events emitted server-side during a collaborative session.
+type: object
+required:
+  - level
+  - room
+  - path
+properties:
+  level:
+    enum:
+      - INFO
+      - DEBUG
+      - WARNING
+      - ERROR
+      - CRITICAL
+    description: |
+      Message type.
+  room:
+    type: string
+    description: |
+      Room ID. Usually composed by the file type, format and ID.
+  path:
+    type: string
+    description: |
+      File path.
+  store:
+    type: string
+    description: |
+      The store used to track the document history.
+  action:
+    enum:
+      - initialize
+      - load
+      - save
+      - overwrite
+      - clean
+    description: |
+      Action performed in a room during a collaborative session.
+      Possible values:
+      1. initialize
+         Initialize a room by loading the content from the contents manager or a store.
+      2. load
+         Load the content from the contents manager.
+      3. save
+         Save the content with the contents manager.
+      4. overwrite
+         Overwrite the content in a room with content from the contents manager.
+         This can happen when multiple rooms access the same file or when a user
+         modifies the file outside Jupyter Server (e.g. using a different app).
+      5. clean
+         Clean the room once is empty (aka there is no more users connected to it).
+  msg:
+    type: string
+    description: |
+      Event message.

jupyter_collaboration/handlers.py

@@ -20,7 +20,7 @@
 
 from .loaders import FileLoader
 from .rooms import DocumentRoom, TransientRoom
-from .utils import decode_file_path
+from .utils import JUPYTER_COLLABORATION_EVENTS_URI, LogLevel, decode_file_path
 
 YFILE = YDOCS["file"]
 
@@ -131,7 +131,7 @@ def __init__(
         super().__init__(app, request, **kwargs)
 
         # CONFIG
-        file_id_manager = self.settings["file_id_manager"]
+        self._file_id_manager = self.settings["file_id_manager"]
         ystore_class = self.settings["collaborative_ystore_class"]
         self._cleanup_delay = self.settings["collaborative_document_cleanup_delay"]
         # self.settings["collaborative_file_poll_interval"]
@@ -163,7 +163,7 @@ def __init__(
             if self._room_id.count(":") >= 2:
                 # DocumentRoom
                 file_format, file_type, file_id = decode_file_path(self._room_id)
-                path = file_id_manager.get_path(file_id)
+                path = self._file_id_manager.get_path(file_id)
 
                 # Instantiate the FileLoader if it doesn't exist yet
                 file = YDocWebSocketHandler.files.get(file_id)
@@ -173,13 +173,20 @@ def __init__(
                         file_id,
                         file_format,
                         file_type,
-                        file_id_manager,
+                        self._file_id_manager,
                         self.contents_manager,
                         self.log,
                         self.settings["collaborative_file_poll_interval"],
                     )
                     self.files[file_id] = file
 
+                else:
+                    self._emit(
+                        LogLevel.WARNING,
+                        None,
+                        "There is another collaborative session accessing the same file.\nThe synchronization between rooms is not supported and you might lose some of your changes.",
+                    )
+
                 path = Path(path)
                 updates_file_path = str(path.parent / f".{file_type}:{path.name}.y")
                 ystore = ystore_class(path=updates_file_path, log=self.log)
@@ -188,6 +195,7 @@ def __init__(
                     file_format,
                     file_type,
                     file,
+                    self.event_logger,
                     ystore,
                     self.log,
                     self.settings["collaborative_document_save_delay"],
@@ -258,6 +266,8 @@ async def open(self, room_id):
             # Initialize the room
             await self.room.initialize()
 
+            self._emit(LogLevel.INFO, "initialize", "New client connected.")
+
     async def send(self, message):
         """
         Send a message to the client.
@@ -321,6 +331,18 @@ def on_close(self) -> None:
             self.log.info("Cleaning room: %s", self._room_id)
             self.room.cleaner = asyncio.create_task(self._clean_room())
 
+    def _emit(self, level: LogLevel, action: str | None = None, msg: str | None = None) -> None:
+        _, _, file_id = decode_file_path(self._room_id)
+        path = self._file_id_manager.get_path(file_id)
+
+        data = {"level": level.value, "room": self._room_id, "path": path}
+        if action:
+            data["action"] = action
+        if msg:
+            data["msg"] = msg
+
+        self.event_logger.emit(schema_id=JUPYTER_COLLABORATION_EVENTS_URI, data=data)
+
     async def _clean_room(self) -> None:
         """
         Async task for cleaning up the resources.
@@ -348,6 +370,7 @@ async def _clean_room(self) -> None:
         # Clean room
         del self.room
         self.log.info("Room %s deleted", self._room_id)
+        self._emit(LogLevel.INFO, "clean", "Room deleted.")
 
         # Clean the file loader if there are not rooms using it
         _, _, file_id = decode_file_path(self._room_id)
@@ -356,6 +379,7 @@ async def _clean_room(self) -> None:
             self.log.info("Deleting file %s", file.path)
             file.clean()
             del self.files[file_id]
+            self._emit(LogLevel.INFO, "clean", "Loader deleted.")
 
     def check_origin(self, origin):
         """

jupyter_collaboration/rooms.py

@@ -4,11 +4,13 @@
 from logging import Logger
 from typing import Any
 
+from jupyter_events import EventLogger
 from jupyter_ydoc import ydocs as YDOCS
 from ypy_websocket.websocket_server import YRoom
 from ypy_websocket.ystore import BaseYStore, YDocNotFound
 
 from .loaders import FileLoader, OutOfBandChanges
+from .utils import JUPYTER_COLLABORATION_EVENTS_URI, LogLevel
 
 YFILE = YDOCS["file"]
 
@@ -22,6 +24,7 @@ def __init__(
         file_format: str,
         file_type: str,
         file: FileLoader,
+        logger: EventLogger,
         ystore: BaseYStore | None,
         log: Logger | None,
         save_delay: int | None = None,
@@ -35,6 +38,7 @@ def __init__(
         self._file: FileLoader = file
         self._document = YDOCS.get(self._file_type, YFILE)(self.ydoc)
 
+        self._logger = logger
         self._save_delay = save_delay
 
         self._update_lock = asyncio.Lock()
@@ -95,6 +99,13 @@ async def initialize(self) -> None:
                 if self.ystore is not None:
                     try:
                         await self.ystore.apply_updates(self.ydoc)
+                        self._emit(
+                            LogLevel.INFO,
+                            "load",
+                            "Content loaded from the store {}".format(
+                                self.ystore.__class__.__qualname__
+                            ),
+                        )
                         self.log.info(
                             "Content in room %s loaded from the ystore %s",
                             self._room_id,
@@ -109,6 +120,9 @@ async def initialize(self) -> None:
                     # if YStore updates and source file are out-of-sync, resync updates with source
                     if self._document.source != model["content"]:
                         # TODO: Delete document from the store.
+                        self._emit(
+                            LogLevel.INFO, "initialize", "The file is out-of-sync with the ystore."
+                        )
                         self.log.info(
                             "Content in file %s is out-of-sync with the ystore %s",
                             self._file.path,
@@ -117,6 +131,7 @@ async def initialize(self) -> None:
                         read_from_source = True
 
                 if read_from_source:
+                    self._emit(LogLevel.INFO, "load", "Content loaded from disk.")
                     self.log.info(
                         "Content in room %s loaded from file %s", self._room_id, self._file.path
                     )
@@ -128,6 +143,16 @@ async def initialize(self) -> None:
                 self._last_modified = model["last_modified"]
                 self._document.dirty = False
                 self.ready = True
+                self._emit(LogLevel.INFO, "initialize", "Room initialized")
+
+    def _emit(self, level: LogLevel, action: str | None = None, msg: str | None = None) -> None:
+        data = {"level": level.value, "room": self._room_id, "path": self._file.path}
+        if action:
+            data["action"] = action
+        if msg:
+            data["msg"] = msg
+
+        self._logger.emit(schema_id=JUPYTER_COLLABORATION_EVENTS_URI, data=data)
 
     def _clean(self) -> None:
         """
@@ -155,6 +180,7 @@ async def _on_content_change(self, event: str, args: dict[str, Any]) -> None:
             model = await self._file.load_content(self._file_format, self._file_type, True)
 
             self.log.info("Out-of-band changes. Overwriting the content in room %s", self._room_id)
+            self._emit(LogLevel.INFO, "overwrite", "Out-of-band changes. Overwriting the room.")
 
             async with self._update_lock:
                 self._document.source = model["content"]
@@ -215,6 +241,8 @@ async def _maybe_save_document(self) -> None:
             async with self._update_lock:
                 self._document.dirty = False
 
+            self._emit(LogLevel.INFO, "save", "Content saved.")
+
         except OutOfBandChanges:
             self.log.info("Out-of-band changes. Overwriting the content in room %s", self._room_id)
             model = await self._file.load_content(self._file_format, self._file_type, True)
@@ -223,6 +251,8 @@ async def _maybe_save_document(self) -> None:
                 self._last_modified = model["last_modified"]
                 self._document.dirty = False
 
+            self._emit(LogLevel.INFO, "overwrite", "Out-of-band changes while saving.")
+
 
 class TransientRoom(YRoom):
     """A Y room for sharing state (e.g. awareness)."""

jupyter_collaboration/utils.py

@@ -1,5 +1,20 @@
+import pathlib
+from enum import Enum
 from typing import Tuple
 
+JUPYTER_COLLABORATION_EVENTS_URI = (
+    "https://events.jupyter.org/jupyter_server/jupyter_collaboration/v1"
+)
+EVENTS_SCHEMA_PATH = pathlib.Path(__file__).parent / "events" / "session.yaml"
+
+
+class LogLevel(Enum):
+    INFO = "INFO"
+    DEBUG = "DEBUG"
+    WARNING = "WARNING"
+    ERROR = "ERROR"
+    CRITICAL = "CRITICAL"
+
 
 def decode_file_path(path: str) -> Tuple[str, str, str]:
     """