Notification topics refactorization using a db table (#680)

Change completely the structure of topics in Hyperion. Instead of using
an Enum difficult to edit, we have a table in db with all existing
topics.

This allows to ensure that users can only subscribe to *real* topics. 
Topics can be created either in the module definition (ex: `cinema`), or
using a utils from an endpoint (creation of a new topic each time we
create an advertiser in `announce`). Topics can now be restricted to a
specific group.

By default, all users are subscribed to each topic, they can unsubscribe
manually if they want.

A change in the frontend is required:
 - using `/notification/topics` the client will get all available topics
 - it could be nice to group topics by `module_root`

This PR will reset existing subscriptions. All users will be subscribed
to all topics, except for `advert`

Fix #479
Requires #768

---------

Co-authored-by: Foucauld Bellanger <[email protected]>
Co-authored-by: armanddidierjean <[email protected]>

Author: foucblg

app/app.py

@@ -5,7 +5,7 @@
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from contextlib import asynccontextmanager
 from pathlib import Path
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING
 
 import alembic.command as alembic_command
 import alembic.config as alembic_config
@@ -28,6 +28,7 @@
 from app.core.google_api.google_api import GoogleAPI
 from app.core.groups import models_groups
 from app.core.groups.groups_type import GroupType
+from app.core.notification.cruds_notification import get_notification_topic
 from app.core.schools import models_schools
 from app.core.schools.schools_type import SchoolType
 from app.core.utils.config import Settings
@@ -36,17 +37,19 @@
     disconnect_state,
     get_app_state,
     get_db,
+    get_notification_manager,
     get_redis_client,
     init_app_state,
 )
-from app.module import module_list
+from app.module import all_modules, module_list
 from app.types.exceptions import (
     ContentHTTPException,
     GoogleAPIInvalidCredentialsError,
     MultipleWorkersWithoutRedisInitializationError,
 )
 from app.types.sqlalchemy import Base
 from app.utils import initialization
+from app.utils.communication.notifications import NotificationManager
 from app.utils.redis import limiter
 from app.utils.state import LifespanState
 
@@ -293,6 +296,32 @@ def initialize_module_visibility(
             )
 
 
+async def initialize_notification_topics(
+    db: AsyncSession,
+    hyperion_error_logger: logging.Logger,
+    notification_manager: NotificationManager,
+) -> None:
+    existing_topics = await get_notification_topic(db=db)
+    existing_topics_id = [topic.id for topic in existing_topics]
+    for module in all_modules:
+        if module.registred_topics:
+            for registred_topic in module.registred_topics:
+                if registred_topic.id not in existing_topics_id:
+                    # We want to register this new topic
+                    hyperion_error_logger.info(
+                        f"Registering topic {registred_topic.name} ({registred_topic.id})",
+                    )
+                    await notification_manager.register_new_topic(
+                        topic_id=registred_topic.id,
+                        name=registred_topic.name,
+                        module_root=registred_topic.module_root,
+                        topic_identifier=registred_topic.topic_identifier,
+                        restrict_to_group_id=registred_topic.restrict_to_group_id,
+                        restrict_to_members=registred_topic.restrict_to_members,
+                        db=db,
+                    )
+
+
 def use_route_path_as_operation_ids(app: FastAPI) -> None:
     """
     Simplify operation IDs so that generated API clients have simpler function names.
@@ -404,17 +433,16 @@ async def init_lifespan(
 
     # We get `init_app_state` as a dependency, as tests
     # should override it to provide their own state
-    state = await app.dependency_overrides.get(
+    state: LifespanState = await app.dependency_overrides.get(
         init_app_state,
         init_app_state,
     )(
         app=app,
         settings=settings,
         hyperion_error_logger=hyperion_error_logger,
     )
-    state = cast("LifespanState", state)
 
-    redis_client = app.dependency_overrides.get(
+    redis_client: Redis | None = app.dependency_overrides.get(
         get_redis_client,
         get_redis_client,
     )(state=state)
@@ -452,10 +480,15 @@ async def init_lifespan(
         hyperion_error_logger=hyperion_error_logger,
     )
 
-    async for db in app.dependency_overrides.get(
+    get_db_dependency: Callable[
+        [LifespanState],
+        AsyncGenerator[AsyncSession, None],
+    ] = app.dependency_overrides.get(
         get_db,
         get_db,
-    )(state=state):
+    )
+
+    async for db in get_db_dependency(state):
         await initialization.use_lock_for_workers(
             init_google_API,
             "init_google_API",
@@ -466,6 +499,22 @@ async def init_lifespan(
             settings=settings,
         )
 
+    async for db in get_db_dependency(state):
+        notification_manager = app.dependency_overrides.get(
+            get_notification_manager,
+            get_notification_manager,
+        )(state)
+        await initialization.use_lock_for_workers(
+            initialize_notification_topics,
+            "initialize_notification_topics",
+            redis_client,
+            number_of_workers,
+            hyperion_error_logger,
+            db=db,
+            hyperion_error_logger=hyperion_error_logger,
+            notification_manager=notification_manager,
+        )
+
     return state
 
 

app/core/groups/endpoints_groups.py

@@ -12,14 +12,17 @@
 
 from app.core.groups import cruds_groups, models_groups, schemas_groups
 from app.core.groups.groups_type import GroupType
+from app.core.notification.utils_notification import get_topics_restricted_to_group_id
 from app.core.users import cruds_users
 from app.dependencies import (
     get_db,
+    get_notification_manager,
     get_request_id,
     is_user_an_ecl_member,
     is_user_in,
 )
 from app.types.module import CoreModule
+from app.utils.communication.notifications import NotificationManager
 
 router = APIRouter(tags=["Groups"])
 
@@ -166,15 +169,12 @@ async def create_membership(
         f"Create_membership: Admin user {user.id} ({user.name}) added user {user_db.id} ({user_db.email}) to group {group_db.id} ({group_db.name}) ({request_id})",
     )
 
-    try:
-        membership_db = models_groups.CoreMembership(
-            user_id=membership.user_id,
-            group_id=membership.group_id,
-            description=membership.description,
-        )
-        return await cruds_groups.create_membership(db=db, membership=membership_db)
-    except ValueError as error:
-        raise HTTPException(status_code=422, detail=str(error))
+    membership_db = models_groups.CoreMembership(
+        user_id=membership.user_id,
+        group_id=membership.group_id,
+        description=membership.description,
+    )
+    return await cruds_groups.create_membership(db=db, membership=membership_db)
 
 
 @router.post(
@@ -232,6 +232,7 @@ async def delete_membership(
     db: AsyncSession = Depends(get_db),
     user=Depends(is_user_in(GroupType.admin)),
     request_id: str = Depends(get_request_id),
+    notification_manager: NotificationManager = Depends(get_notification_manager),
 ):
     """
     Delete a membership using the user and group ids.
@@ -243,6 +244,19 @@ async def delete_membership(
         f"Create_membership: Admin user {user.id} ({user.name}) removed user {membership.user_id} from group {membership.group_id} ({request_id})",
     )
 
+    # To remove a user from a group, we should unsubscribe this user from all
+    # topic that required to be a member of this group
+    restricted_topics = await get_topics_restricted_to_group_id(
+        group_id=membership.group_id,
+        db=db,
+    )
+    for topic in restricted_topics:
+        await notification_manager.unsubscribe_user_to_topic(
+            topic_id=topic.id,
+            user_id=membership.user_id,
+            db=db,
+        )
+
     await cruds_groups.delete_membership_by_group_and_user_id(
         group_id=membership.group_id,
         user_id=membership.user_id,
@@ -259,6 +273,7 @@ async def delete_batch_membership(
     db: AsyncSession = Depends(get_db),
     user=Depends(is_user_in(GroupType.admin)),
     request_id: str = Depends(get_request_id),
+    notification_manager: NotificationManager = Depends(get_notification_manager),
 ):
     """
     This endpoint removes all users from a given group.
@@ -273,6 +288,20 @@ async def delete_batch_membership(
     if group_db is None:
         raise HTTPException(status_code=400, detail="Invalid group_id")
 
+    # To remove a user from a group, we should unsubscribe this user from all
+    # topic that required to be a member of this group
+    restricted_topics = await get_topics_restricted_to_group_id(
+        group_id=batch_membership.group_id,
+        db=db,
+    )
+    for topic in restricted_topics:
+        for user_from_group in group_db.members:
+            await notification_manager.unsubscribe_user_to_topic(
+                topic_id=topic.id,
+                user_id=user_from_group.id,
+                db=db,
+            )
+
     hyperion_security_logger.warning(
         f"Create_batch_membership: Admin user {user.id} ({user.name}) removed all users from group {group_db.id} ({group_db.name}) in batch ({request_id})",
     )

app/core/notification/cruds_notification.py

@@ -1,11 +1,66 @@
 from collections.abc import Sequence
 from datetime import date
+from uuid import UUID
 
 from sqlalchemy import delete, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.notification import models_notification
-from app.core.notification.notification_types import CustomTopic, Topic
+
+
+async def get_notification_topic(
+    db: AsyncSession,
+) -> Sequence[models_notification.NotificationTopic]:
+    result = await db.execute(select(models_notification.NotificationTopic))
+    return result.scalars().all()
+
+
+async def get_notification_topic_by_id(
+    topic_id: UUID,
+    db: AsyncSession,
+) -> models_notification.NotificationTopic | None:
+    result = await db.execute(
+        select(models_notification.NotificationTopic).where(
+            models_notification.NotificationTopic.id == topic_id,
+        ),
+    )
+    return result.scalars().first()
+
+
+async def get_topics_restricted_to_group_id(
+    group_id: str,
+    db: AsyncSession,
+) -> Sequence[models_notification.NotificationTopic]:
+    result = await db.execute(
+        select(models_notification.NotificationTopic).where(
+            models_notification.NotificationTopic.restrict_to_group_id == group_id,
+        ),
+    )
+    return result.scalars().all()
+
+
+async def get_notification_topic_by_root_and_identifier(
+    module_root: str,
+    topic_identifier: str | None,
+    db: AsyncSession,
+) -> models_notification.NotificationTopic | None:
+    result = await db.execute(
+        select(models_notification.NotificationTopic).where(
+            models_notification.NotificationTopic.module_root == module_root,
+            models_notification.NotificationTopic.topic_identifier == topic_identifier,
+        ),
+    )
+    return result.scalars().first()
+
+
+async def create_notification_topic(
+    notification_topic: models_notification.NotificationTopic,
+    db: AsyncSession,
+) -> None:
+    """Register a new topic in database and return it"""
+
+    db.add(notification_topic)
+    await db.flush()
 
 
 async def get_firebase_devices_by_user_id(
@@ -111,29 +166,25 @@ async def create_topic_membership(
 
 async def delete_topic_membership(
     user_id: str,
-    custom_topic: CustomTopic,
+    topic_id: UUID,
     db: AsyncSession,
 ):
     await db.execute(
         delete(models_notification.TopicMembership).where(
             models_notification.TopicMembership.user_id == user_id,
-            models_notification.TopicMembership.topic == custom_topic.topic,
-            models_notification.TopicMembership.topic_identifier
-            == custom_topic.topic_identifier,
+            models_notification.TopicMembership.topic_id == topic_id,
         ),
     )
     await db.flush()
 
 
-async def get_topic_memberships_by_topic(
-    custom_topic: CustomTopic,
+async def get_topic_memberships_by_topic_id(
+    topic_id: str,
     db: AsyncSession,
 ) -> Sequence[models_notification.TopicMembership]:
     result = await db.execute(
         select(models_notification.TopicMembership).where(
-            models_notification.TopicMembership.topic == custom_topic.topic,
-            models_notification.TopicMembership.topic_identifier
-            == custom_topic.topic_identifier,
+            models_notification.TopicMembership.topic_id == topic_id,
         ),
     )
     return result.scalars().all()
@@ -151,46 +202,41 @@ async def get_topic_memberships_by_user_id(
     return result.scalars().all()
 
 
-async def get_topic_memberships_with_identifiers_by_user_id_and_topic(
+async def get_topic_memberships_with_identifiers_by_user_id_and_topic_id(
     user_id: str,
-    topic: Topic,
+    topic_id: str,
     db: AsyncSession,
 ) -> Sequence[models_notification.TopicMembership]:
     result = await db.execute(
         select(models_notification.TopicMembership).where(
             models_notification.TopicMembership.user_id == user_id,
-            models_notification.TopicMembership.topic == topic,
-            models_notification.TopicMembership.topic_identifier != "",
+            models_notification.TopicMembership.topic_id == topic_id,
         ),
     )
     return result.scalars().all()
 
 
-async def get_topic_membership_by_user_id_and_custom_topic(
+async def get_topic_membership_by_user_id_and_topic_id(
     user_id: str,
-    custom_topic: CustomTopic,
+    topic_id: UUID,
     db: AsyncSession,
 ) -> models_notification.TopicMembership | None:
     result = await db.execute(
         select(models_notification.TopicMembership).where(
             models_notification.TopicMembership.user_id == user_id,
-            models_notification.TopicMembership.topic == custom_topic.topic,
-            models_notification.TopicMembership.topic_identifier
-            == custom_topic.topic_identifier,
+            models_notification.TopicMembership.topic_id == topic_id,
         ),
     )
     return result.scalars().first()
 
 
-async def get_user_ids_by_topic(
-    custom_topic: CustomTopic,
+async def get_user_ids_by_topic_id(
+    topic_id: UUID,
     db: AsyncSession,
 ) -> list[str]:
     result = await db.execute(
         select(models_notification.TopicMembership.user_id).where(
-            models_notification.TopicMembership.topic == custom_topic.topic,
-            models_notification.TopicMembership.topic_identifier
-            == custom_topic.topic_identifier,
+            models_notification.TopicMembership.topic_id == topic_id,
         ),
     )
     return list(result.scalars().all())