Feat/socket snapshot update required

backend/pyproject.toml

@@ -18,6 +18,7 @@ tomli = { version = "^2.0.1", python = "<3.11" }
 tomli-w = "^1.0.0"
 requests = "^2.32.4"
 jsonpatch = "^1.33"
+websockets = "^15.0.1"
 qualibrate-config = "^0.1.8,<0.2.0"
 # TODO: replace after publishing timeline-db
 # json-timeline-database = { path = "../../json-timeline-database", optional = true, develop = true }

backend/qualibrate_app/api/core/lifespan.py

@@ -0,0 +1,15 @@
+import asyncio
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from qualibrate_app.api.sockets.tasks import update_snapshot_history_required
+
+__all__ = ["app_lifespan"]
+
+
+@asynccontextmanager
+async def app_lifespan(app: FastAPI) -> AsyncIterator[None]:
+    await asyncio.gather(update_snapshot_history_required())
+    yield

backend/qualibrate_app/api/core/schemas/update_history.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel, computed_field
+
+from qualibrate_app.api.core.types import IdType
+
+
+class UpdateHistoryRequired(BaseModel):
+    saved_id: IdType | None
+    latest_id: IdType | None
+
+    @computed_field
+    def update_required(self) -> bool:
+        return self.latest_id != self.saved_id

backend/qualibrate_app/api/core/socket/periodic_tasks.py

@@ -0,0 +1,81 @@
+import asyncio
+from collections.abc import Callable, Coroutine
+from functools import wraps
+from typing import Any, TypeAlias
+
+from fastapi.concurrency import run_in_threadpool
+
+NoArgsNoReturnFuncT: TypeAlias = Callable[[], None]
+NoArgsNoReturnAsyncFuncT: TypeAlias = Callable[[], Coroutine[Any, Any, None]]
+ExcArgNoReturnFuncT: TypeAlias = Callable[[Exception], None]
+ExcArgNoReturnAsyncFuncT: TypeAlias = Callable[
+    [Exception], Coroutine[Any, Any, None]
+]
+NoArgsNoReturnAnyFuncT: TypeAlias = (
+    NoArgsNoReturnFuncT | NoArgsNoReturnAsyncFuncT
+)
+ExcArgNoReturnAnyFuncT: TypeAlias = (
+    ExcArgNoReturnFuncT | ExcArgNoReturnAsyncFuncT
+)
+
+__all__ = ["repeat_every"]
+
+
+async def _handle_func(func: NoArgsNoReturnAnyFuncT) -> None:
+    if asyncio.iscoroutinefunction(func):
+        await func()
+    else:
+        await run_in_threadpool(func)
+
+
+async def _handle_exc(
+    exc: Exception, on_exception: ExcArgNoReturnAnyFuncT | None
+) -> None:
+    if on_exception is None:
+        return
+    if asyncio.iscoroutinefunction(on_exception):
+        await on_exception(exc)
+    else:
+        await run_in_threadpool(on_exception, exc)
+
+
+def repeat_every(
+    *,
+    seconds: float,
+    on_exception: Callable[[Exception], None] | None = None,
+) -> Callable[[NoArgsNoReturnAnyFuncT], NoArgsNoReturnAsyncFuncT]:
+    """
+    This function returns a decorator that modifies a function so it is
+    periodically re-executed after its first call.
+    The function it decorates should accept no arguments and return nothing.
+
+    Parameters
+    ----------
+    seconds: float
+        The number of seconds to wait between repeated calls
+    on_exception: Optional[Callable[[Exception], None]] (default None)
+        A function to call when an exception is raised by the decorated
+        function.
+    """
+
+    def decorator(func: NoArgsNoReturnAnyFuncT) -> NoArgsNoReturnAsyncFuncT:
+        """
+        Converts the decorated function into a repeated,
+        periodically-called version of itself.
+        """
+
+        @wraps(func)
+        async def wrapped() -> None:
+            async def loop() -> None:
+                while True:
+                    try:
+                        await _handle_func(func)
+                    except Exception as exc:
+                        await _handle_exc(exc, on_exception)
+                    await asyncio.sleep(seconds)
+
+            asyncio.ensure_future(loop())
+
+        return wrapped
+
+    return decorator

backend/qualibrate_app/api/core/socket/ws_manager.py

@@ -0,0 +1,72 @@
+from collections import defaultdict
+from collections.abc import Hashable
+from typing import Any, Generic, TypeVar
+
+from fastapi import WebSocket
+
+
+class SocketConnectionManagerList:
+    def __init__(self) -> None:
+        self.active_connections: list[WebSocket] = []
+
+    async def connect(self, websocket: WebSocket) -> None:
+        await websocket.accept()
+        self.active_connections.append(websocket)
+
+    def disconnect(self, websocket: WebSocket) -> None:
+        if websocket in self.active_connections:
+            self.active_connections.remove(websocket)
+
+    async def broadcast(self, message: Any) -> None:
+        disconnected: list[WebSocket] = []
+        for connection in self.active_connections:
+            try:
+                await connection.send_json(message)
+            except Exception:
+                disconnected.append(connection)
+        for conn in disconnected:
+            self.disconnect(conn)
+
+    @property
+    def any_subscriber(self) -> bool:
+        return len(self.active_connections) > 0
+
+
+KT = TypeVar("KT", bound=Hashable)
+
+
+class SocketConnectionManagerMapping(Generic[KT]):
+    def __init__(self) -> None:
+        self.active_connections: dict[KT, list[WebSocket]] = defaultdict(list)
+
+    async def connect(self, key: KT, websocket: WebSocket) -> None:
+        await websocket.accept()
+        self.active_connections[key].append(websocket)
+
+    def disconnect(self, key: KT, websocket: WebSocket) -> None:
+        if key in self.active_connections and websocket in self.active_connections[key]:
+            self.active_connections[key].remove(websocket)
+
+    async def broadcast(self, key: KT, message: Any) -> None:
+        if not self.any_subscriber_for(key):
+            return
+        disconnected: list[WebSocket] = []
+        for connection in self.active_connections[key]:
+            try:
+                await connection.send_json(message)
+            except Exception:
+                disconnected.append(connection)
+        for conn in disconnected:
+            self.disconnect(key, conn)
+
+    @property
+    def any_subscriber(self) -> bool:
+        return len(self.active_connections) > 0 and any(
+            map(len, self.active_connections.values())
+        )
+
+    def any_subscriber_for(self, key: KT) -> bool:
+        return (
+            key in self.active_connections
+            and len(self.active_connections[key]) > 0
+        )

backend/qualibrate_app/api/core/socket/ws_managers.py

@@ -0,0 +1,14 @@
+from functools import cache
+
+from qualibrate_app.api.core.socket.ws_manager import (
+    SocketConnectionManagerList,
+)
+
+__all__ = ["get_need_to_update_snapshots_history_socket_manager"]
+
+
+@cache
+def get_need_to_update_snapshots_history_socket_manager() -> (
+    SocketConnectionManagerList
+):
+    return SocketConnectionManagerList()

backend/qualibrate_app/api/sockets/__init__.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+
+from .common import common_ws_router
+
+base_ws_router = APIRouter()
+base_ws_router.include_router(common_ws_router)

backend/qualibrate_app/api/sockets/common.py

@@ -0,0 +1,30 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+from starlette.websockets import WebSocket, WebSocketDisconnect
+
+from qualibrate_app.api.core.socket.ws_manager import (
+    SocketConnectionManagerList,
+)
+from qualibrate_app.api.core.socket.ws_managers import (
+    get_need_to_update_snapshots_history_socket_manager,
+)
+
+common_ws_router = APIRouter()
+
+
+@common_ws_router.websocket("/update_snapshots_history_required")
+async def run_status_subscribe(
+    websocket: WebSocket,
+    *,
+    manager: Annotated[
+        SocketConnectionManagerList,
+        Depends(get_need_to_update_snapshots_history_socket_manager),
+    ],
+) -> None:
+    await manager.connect(websocket)
+    try:
+        while True:
+            await websocket.receive_text()
+    except WebSocketDisconnect:
+        manager.disconnect(websocket)

backend/qualibrate_app/api/sockets/tasks.py

@@ -0,0 +1,46 @@
+import logging
+
+from qualibrate_app.api.core.domain.local_storage.utils.node_utils import (
+    find_nodes_ids_by_filter,
+)
+from qualibrate_app.api.core.schemas.update_history import UpdateHistoryRequired
+from qualibrate_app.api.core.socket.periodic_tasks import repeat_every
+from qualibrate_app.api.core.socket.ws_managers import (
+    get_need_to_update_snapshots_history_socket_manager,
+)
+from qualibrate_app.api.core.types import IdType
+from qualibrate_app.config import get_config_path, get_settings
+
+__all__ = ["update_snapshot_history_required"]
+
+
+def _on_exc(exc: Exception) -> None:
+    logging.exception(
+        "Exception occurred while running periodic task", exc_info=exc
+    )
+
+
+previous_snapshot_id: IdType | None = None
+
+
+@repeat_every(seconds=1, on_exception=_on_exc)
+async def update_snapshot_history_required() -> None:
+    manager = get_need_to_update_snapshots_history_socket_manager()
+    if not manager.any_subscriber:
+        return
+    settings = get_settings(get_config_path())
+    id = next(
+        find_nodes_ids_by_filter(
+            settings.storage.location,
+            project_name=settings.project,
+            descending=True,
+        ),
+        None,
+    )
+    global previous_snapshot_id
+    update_required = UpdateHistoryRequired(
+        latest_id=id,
+        saved_id=previous_snapshot_id,
+    )
+    previous_snapshot_id = id
+    await manager.broadcast(update_required.model_dump(mode="json"))

backend/qualibrate_app/app.py

@@ -4,9 +4,11 @@
 from starlette.middleware.cors import CORSMiddleware
 
 from qualibrate_app.api.__main__ import api_router
+from qualibrate_app.api.core.lifespan import app_lifespan
 from qualibrate_app.api.exceptions.classes.base import QualibrateException
 from qualibrate_app.api.exceptions.handler import qualibrate_exception_handler
 from qualibrate_app.api.middleware.process_time import ProcessTimeMiddleware
+from qualibrate_app.api.sockets import base_ws_router
 from qualibrate_app.config.resolvers import (
     get_config_path,
     get_default_static_files_path,
@@ -20,6 +22,7 @@
 
 
 app = FastAPI(
+    lifespan=app_lifespan,
     title="Qualibrate",
     openapi_url="/app_openapi.json",
     docs_url="/app_docs",
@@ -44,6 +47,7 @@
 app.add_middleware(ProcessTimeMiddleware)
 
 app.include_router(api_router, prefix="/api")
+app.include_router(base_ws_router, prefix="/ws")
 
 static_files_path = (
     _settings.app.static_site_files

frontend/src/modules/Project/components/Project/ProjectActions.tsx

@@ -7,8 +7,7 @@ import cyKeys from "../../../../utils/cyKeys";
 import { NODES_KEY } from "../../../AppRoutes";
 import { useRootDispatch } from "../../../../stores";
 import { setActivePage } from "../../../../stores/NavigationStore";
-import { useSelector } from "react-redux";
-import { getReset, clearData } from "../../../../stores/SnapshotsStore";
+import { clearData, fetchGitgraphSnapshots } from "../../../../stores/SnapshotsStore";
 
 interface ProjectActionsProps {
   isCurrentProject: boolean;
@@ -18,15 +17,15 @@ interface ProjectActionsProps {
 
 const ProjectActions: React.FC<ProjectActionsProps> = ({ isCurrentProject, projectName, selectedProject }) => {
   const dispatch = useRootDispatch();
-  const reset = useSelector(getReset);
 
   const handleSubmit = useCallback(() => {
     if (!selectedProject) return;
 
     dispatch(selectActiveProject(selectedProject));
     dispatch(clearData());
+    dispatch(fetchGitgraphSnapshots(true));
     dispatch(setActivePage(NODES_KEY));
-  }, [ selectedProject, reset ]);
+  }, [ selectedProject ]);
 
   return (
     <div className={styles.pageActions}>

frontend/src/services/webSocketRoutes.ts

@@ -1,2 +1,3 @@
 export const WS_GET_STATUS = "execution/ws/run_status";
 export const WS_EXECUTION_HISTORY = "execution/ws/workflow_execution_history";
+export const WS_SNAPSHOT_INFO = "ws/update_snapshots_history_required";

frontend/src/stores/SnapshotsStore/SnapshotsStore.ts

@@ -16,14 +16,13 @@ interface SnapshotsState {
   result: object | undefined;
   firstId: string;
   secondId: string;
-  reset: boolean;
 }
 
 const initialState: SnapshotsState = {
   trackLatestSidePanel: true,
   trackPreviousSnapshot: true,
   totalPages: 0,
-  pageNumber: 0,
+  pageNumber: 1,
   allSnapshots: [],
   selectedSnapshotId: undefined,
   latestSnapshotId: undefined,
@@ -34,7 +33,6 @@ const initialState: SnapshotsState = {
   result: {},
   firstId: "0",
   secondId: "0",
-  reset: false,
 };
 
 export const SnapshotsSlice = createSlice({
@@ -79,7 +77,6 @@ export const SnapshotsSlice = createSlice({
       state.jsonData = undefined;
       state.result = undefined;
       state.diffData = undefined;
-      state.reset = true;
     },
     setResult: (state, action) => {
       state.result = action.payload;
@@ -90,9 +87,6 @@ export const SnapshotsSlice = createSlice({
     setSecondId: (state, action) => {
       state.secondId = action.payload;
     },
-    setReset: (state, action) => {
-      state.reset = action.payload;
-    },
   }
 });