Merge branch 'persistence/manager' into persistence/module-states

Author: rubenthoms

backend_py/primary/poetry.lock

@@ -11,6 +11,7 @@ def to_api_session_metadata_summary(metadata: SessionMetadataWithId) -> schemas.
         createdAt=metadata.created_at.isoformat(),
         updatedAt=metadata.updated_at.isoformat(),
         version=metadata.version,
+        hash=metadata.hash,
     )
 
 

backend_py/primary/primary/routers/persistence/sessions/__init__.py

@@ -5,12 +5,12 @@
 
 from primary.middleware.add_browser_cache import no_cache
 from primary.services.database_access.session_access.session_access import SessionAccess
+from primary.services.database_access.query_collation_options import SortDirection
 from primary.auth.auth_helper import AuthHelper, AuthenticatedUser
 from primary.services.database_access.session_access.types import (
     NewSession,
     SessionUpdate,
-    SortBy,
-    SortDirection,
+    SessionSortBy,
 )
 from primary.routers.persistence.sessions.converters import (
     to_api_session_metadata_summary,
@@ -28,21 +28,28 @@
 @no_cache
 async def get_sessions_metadata(
     user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user),
-    sort_by: Optional[SortBy] = Query(None, description="Sort the result by"),
+    sort_by: Optional[SessionSortBy] = Query(None, description="Sort the result by"),
     sort_direction: Optional[SortDirection] = Query(SortDirection.ASC, description="Sort direction: 'asc' or 'desc'"),
-    limit: Optional[int] = Query(10, ge=1, le=100, description="Limit the number of results"),
-):
+    limit: int = Query(10, ge=1, le=100, description="Limit the number of results"),
+    page: int = Query(0, ge=0),
+) -> list[schemas.SessionMetadataWithId]:
     access = SessionAccess.create(user.get_user_id())
     async with access:
         items = await access.get_filtered_sessions_metadata_for_user_async(
-            sort_by=sort_by, sort_direction=sort_direction, limit=limit
+            sort_by=sort_by,
+            sort_direction=sort_direction,
+            limit=limit,
+            offset=limit * page,
         )
+
         return [to_api_session_metadata_summary(item) for item in items]
 
 
 @router.get("/sessions/{session_id}", response_model=schemas.SessionDocument)
 @no_cache
-async def get_session(session_id: str, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)):
+async def get_session(
+    session_id: str, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)
+) -> schemas.SessionDocument:
     access = SessionAccess.create(user.get_user_id())
     async with access:
         session = await access.get_session_by_id_async(session_id)
@@ -53,7 +60,9 @@ async def get_session(session_id: str, user: AuthenticatedUser = Depends(AuthHel
 
 @router.get("/sessions/metadata/{session_id}", response_model=schemas.SessionMetadata)
 @no_cache
-async def get_session_metadata(session_id: str, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)):
+async def get_session_metadata(
+    session_id: str, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)
+) -> schemas.SessionMetadata:
     access = SessionAccess.create(user.get_user_id())
     async with access:
         metadata = await access.get_session_metadata_async(session_id)
@@ -63,26 +72,29 @@ async def get_session_metadata(session_id: str, user: AuthenticatedUser = Depend
 
 
 @router.post("/sessions", response_model=str)
-async def create_session(session: NewSession, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)):
+async def create_session(
+    session: NewSession, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)
+) -> str:
     access = SessionAccess.create(user.get_user_id())
     async with access:
         session_id = await access.insert_session_async(session)
         return session_id
 
 
-@router.put("/sessions/{session_id}")
+@router.put("/sessions/{session_id}", description="Updates a session object. Allows for partial update objects")
 async def update_session(
     session_id: str,
     session_update: SessionUpdate,
     user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user),
-):
+) -> schemas.SessionDocument:
     access = SessionAccess.create(user.get_user_id())
     async with access:
-        await access.update_session_async(session_id, session_update)
+        updated_session = await access.update_session_async(session_id, session_update)
+        return to_api_session_record(updated_session)
 
 
 @router.delete("/sessions/{session_id}")
-async def delete_session(session_id: str, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)):
+async def delete_session(session_id: str, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)) -> None:
     access = SessionAccess.create(user.get_user_id())
     async with access:
         await access.delete_session_async(session_id)

backend_py/primary/primary/routers/persistence/sessions/converters.py

@@ -2,15 +2,6 @@
 from pydantic import BaseModel
 
 
-class SessionMetadataWithId(BaseModel):
-    id: str
-    title: str
-    description: Optional[str]
-    createdAt: str
-    updatedAt: str
-    version: int
-
-
 class SessionMetadata(BaseModel):
     title: str
     description: Optional[str]
@@ -20,6 +11,10 @@ class SessionMetadata(BaseModel):
     hash: str
 
 
+class SessionMetadataWithId(SessionMetadata):
+    id: str
+
+
 class SessionDocument(BaseModel):
     id: str
     ownerId: str

backend_py/primary/primary/routers/persistence/sessions/router.py

@@ -1,4 +1,4 @@
-from primary.services.database_access.snapshot_access.models import SnapshotAccessLog
+from primary.services.database_access.snapshot_access.models import SnapshotAccessLogDocument
 from primary.services.database_access.snapshot_access.types import Snapshot, SnapshotMetadata, SnapshotMetadataWithId
 
 from . import schemas
@@ -35,12 +35,13 @@ def to_api_snapshot(snapshot: Snapshot) -> schemas.Snapshot:
     )
 
 
-def to_api_snapshot_access_log(access_log: SnapshotAccessLog, metadata: SnapshotMetadata) -> schemas.SnapshotAccessLog:
+def to_api_snapshot_access_log(access_log: SnapshotAccessLogDocument) -> schemas.SnapshotAccessLog:
     return schemas.SnapshotAccessLog(
         visitorId=access_log.visitor_id,
         snapshotId=access_log.snapshot_id,
         visits=access_log.visits,
         firstVisitedAt=access_log.first_visited_at.isoformat() if access_log.first_visited_at else None,
         lastVisitedAt=access_log.last_visited_at.isoformat() if access_log.last_visited_at else None,
-        snapshotMetadata=to_api_snapshot_metadata(metadata),
+        snapshotDeleted=access_log.snapshot_deleted,
+        snapshotMetadata=to_api_snapshot_metadata(access_log.snapshot_metadata),
     )

backend_py/primary/primary/routers/persistence/sessions/schemas.py

@@ -1,18 +1,18 @@
 import logging
 from typing import List, Optional
 
-from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import APIRouter, BackgroundTasks, Depends, Query
 
 from primary.services.database_access.snapshot_access.types import (
     NewSnapshot,
-    SnapshotUpdate,
-    SortBy,
-    SortDirection,
+    SnapshotSortBy,
+    SnapshotAccessLogSortBy,
 )
 from primary.middleware.add_browser_cache import no_cache
 from primary.services.database_access.snapshot_access.snapshot_access import SnapshotAccess
-from primary.services.database_access.snapshot_access.snapshot_logs_access import SnapshotLogsAccess
-from primary.services.database_access.snapshot_access.query_collation_options import QueryCollationOptions
+from primary.services.database_access.snapshot_access.snapshot_log_access import SnapshotLogAccess
+from primary.services.database_access.query_collation_options import QueryCollationOptions, SortDirection
+from primary.services.database_access.workers.mark_logs_deleted import mark_logs_deleted_worker
 
 
 from primary.auth.auth_helper import AuthHelper, AuthenticatedUser
@@ -33,41 +33,31 @@
 @router.get("/recent_snapshots", response_model=list[schemas.SnapshotAccessLog])
 async def get_recent_snapshots(
     user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user),
-    sort_by: Optional[SortBy] = Query(SortBy.LAST_VISIT, description="Sort the result by"),
-    sort_direction: Optional[SortDirection] = Query(SortDirection.DESC, description="Sort direction: 'asc' or 'desc'"),
-    limit: Optional[int] = Query(5, ge=1, le=100, description="Limit the number of results"),
-    offset: Optional[int] = Query(0, ge=0, description="The offset of the results"),
+    sort_by: Optional[SnapshotAccessLogSortBy] = Query(None, description="Sort the result by"),
+    sort_direction: Optional[SortDirection] = Query(None, description="Sort direction: 'asc' or 'desc'"),
+    limit: Optional[int] = Query(None, ge=1, le=100, description="Limit the number of results"),
+    offset: Optional[int] = Query(None, ge=0, description="The offset of the results"),
 ) -> list[schemas.SnapshotAccessLog]:
-    async with (
-        SnapshotAccess.create(user.get_user_id()) as snapshot_access,
-        SnapshotLogsAccess.create(user.get_user_id()) as log_access,
-    ):
+    async with SnapshotLogAccess.create(user.get_user_id()) as log_access:
         collation_options = QueryCollationOptions(sort_by=sort_by, sort_dir=sort_direction, limit=limit, offset=offset)
 
         recent_logs = await log_access.get_access_logs_for_user_async(collation_options)
 
-        payload: list[schemas.SnapshotAccessLog] = []
-
-        for log in recent_logs:
-            metadata = await snapshot_access.get_snapshot_metadata_async(log.snapshot_id, log.snapshot_owner_id)
-
-            payload.append(to_api_snapshot_access_log(log, metadata))
-
-        return payload
+        return [to_api_snapshot_access_log(log) for log in recent_logs]
 
 
 @router.get("/snapshots", response_model=List[schemas.SnapshotMetadata])
 @no_cache
 async def get_snapshots_metadata(
     user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user),
-    sort_by: Optional[SortBy] = Query(SortBy.LAST_VISIT, description="Sort the result by"),
+    sort_by: Optional[SnapshotSortBy] = Query(SnapshotSortBy.UPDATED_AT, description="Sort the result by"),
     sort_direction: Optional[SortDirection] = Query(SortDirection.DESC, description="Sort direction: 'asc' or 'desc'"),
     limit: Optional[int] = Query(10, ge=1, le=100, description="Limit the number of results"),
 ) -> List[schemas.SnapshotMetadata]:
     access = SnapshotAccess.create(user.get_user_id())
     async with access:
         items = await access.get_filtered_snapshots_metadata_for_user_async(
-            sort_by=sort_by, sort_direction=sort_direction, limit=limit
+            sort_by=sort_by, sort_direction=sort_direction, limit=limit, offset=0
         )
         return [to_api_snapshot_metadata_summary(item) for item in items]
 
@@ -77,58 +67,54 @@ async def get_snapshots_metadata(
 async def get_snapshot(
     snapshot_id: str, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)
 ) -> schemas.Snapshot:
-    access = SnapshotAccess.create(user.get_user_id())
-    logs_access = SnapshotLogsAccess.create(user_id=user.get_user_id())
-
-    async with access, logs_access:
-        snapshot = await access.get_snapshot_by_id_async(snapshot_id)
-        if not snapshot:
-            raise HTTPException(status_code=404, detail="Snapshot not found")
-
-        await logs_access.log_snapshot_visit_async(snapshot_id, snapshot.owner_id)
-
+    snapshot_access = SnapshotAccess.create(user.get_user_id())
+    log_access = SnapshotLogAccess.create(user_id=user.get_user_id())
+
+    async with snapshot_access, log_access:
+        snapshot = await snapshot_access.get_snapshot_by_id_async(snapshot_id)
+        # Should we clear the log if a snapshot was not found? This could mean that the snapshot was
+        # deleted but deletion of logs has failed
+        await log_access.log_snapshot_visit_async(snapshot_id, snapshot.owner_id)
         return to_api_snapshot(snapshot)
 
 
 @router.get("/snapshots/metadata/{snapshot_id}", response_model=schemas.SnapshotMetadata)
 @no_cache
-async def get_snapshot_metadata(snapshot_id: str, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)):
+async def get_snapshot_metadata(
+    snapshot_id: str, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)
+) -> schemas.SnapshotMetadata:
     access = SnapshotAccess.create(user.get_user_id())
     async with access:
         metadata = await access.get_snapshot_metadata_async(snapshot_id)
-        if not metadata:
-            raise HTTPException(status_code=404, detail="Session metadata not found")
         return to_api_snapshot_metadata(metadata)
 
 
 @router.post("/snapshots", response_model=str)
 async def create_snapshot(
     session: NewSnapshot, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)
 ) -> str:
-    access = SnapshotAccess.create(user.get_user_id())
-    logs_access = SnapshotLogsAccess.create(user.get_user_id())
+    snapshot_access = SnapshotAccess.create(user.get_user_id())
+    log_access = SnapshotLogAccess.create(user.get_user_id())
 
-    async with access, logs_access:
-        snapshot_id = await access.insert_snapshot_async(session)
+    async with snapshot_access, log_access:
+        snapshot_id = await snapshot_access.insert_snapshot_async(session)
 
-        # We count snapshot creation as implicit visit. This also makes it so
-        await logs_access.log_snapshot_visit_async(snapshot_id=snapshot_id, snapshot_owner_id=user.get_user_id())
+        # We count snapshot creation as implicit visit. This also makes it so we can get recently created ones alongside other shared screenshots
+        await log_access.log_snapshot_visit_async(snapshot_id=snapshot_id, snapshot_owner_id=user.get_user_id())
         return snapshot_id
 
 
-@router.put("/snapshots/{snapshot_id}")
-async def update_snapshot(
+@router.delete("/snapshots/{snapshot_id}")
+async def delete_snapshot(
     snapshot_id: str,
-    snapshot_update: SnapshotUpdate,
+    background_tasks: BackgroundTasks,
     user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user),
-):
-    access = SnapshotAccess.create(user.get_user_id())
-    async with access:
-        await access.update_snapshot_metadata_async(snapshot_id, snapshot_update)
-
-
-@router.delete("/snapshots/{snapshot_id}")
-async def delete_snapshot(snapshot_id: str, user: AuthenticatedUser = Depends(AuthHelper.get_authenticated_user)):
-    access = SnapshotAccess.create(user.get_user_id())
-    async with access:
-        await access.delete_snapshot_async(snapshot_id)
+) -> None:
+    snapshot_access = SnapshotAccess.create(user.get_user_id())
+    async with snapshot_access:
+        await snapshot_access.delete_snapshot_async(snapshot_id)
+
+    # This is the fastest solution for the moment. As we are expecting <= 150 logs per snapshot
+    # and consistency is not critical, we can afford to do this in the background and without
+    # a safety net. We can later consider adding this to a queue for better reliability.
+    background_tasks.add_task(mark_logs_deleted_worker, snapshot_id=snapshot_id)

backend_py/primary/primary/routers/persistence/snapshots/__init__.py

@@ -29,6 +29,7 @@ class SnapshotAccessLog(BaseModel):
     visits: int
     firstVisitedAt: str | None
     lastVisitedAt: str | None
+    snapshotDeleted: bool
 
     snapshotMetadata: SnapshotMetadata
 

backend_py/primary/primary/routers/persistence/snapshots/converters.py

@@ -1,4 +1,5 @@
 import hashlib
+from typing import Any, cast
 
 
 # Utility function to hash a JSON string using SHA-256
@@ -11,3 +12,7 @@ def hash_json_string(json_string: str) -> str:
     hash_bytes = hashlib.sha256(data).digest()
     hash_hex = "".join(f"{b:02x}" for b in hash_bytes)
     return hash_hex
+
+
+def cast_query_params(params: list[dict[str, Any]]) -> list[dict[str, object]]:
+    return cast(list[dict[str, object]], params)