Send notification content through Firebase and add an Arq scheduler

app/app.py

@@ -30,21 +30,22 @@
 from app.dependencies import (
     get_db,
     get_redis_client,
+    get_scheduler,
     get_websocket_connection_manager,
     init_and_get_db_engine,
 )
 from app.modules.module_list import module_list
-from app.types.exceptions import (
-    ContentHTTPException,
-    GoogleAPIInvalidCredentialsError,
-)
+from app.types.exceptions import ContentHTTPException, GoogleAPIInvalidCredentialsError
 from app.types.sqlalchemy import Base
 from app.utils import initialization
 from app.utils.redis import limiter
 
 if TYPE_CHECKING:
     import redis
 
+    from app.types.scheduler import Scheduler
+    from app.types.websocket import WebsocketConnectionManager
+
 # NOTE: We can not get loggers at the top of this file like we do in other files
 # as the loggers are not yet initialized
 
@@ -352,14 +353,27 @@ async def lifespan(app: FastAPI) -> AsyncGenerator:
                     # We expect this error to be raised if the credentials were never set before
                     pass
 
-        ws_manager = app.dependency_overrides.get(
+        ws_manager: WebsocketConnectionManager = app.dependency_overrides.get(
             get_websocket_connection_manager,
             get_websocket_connection_manager,
         )(settings=settings)
 
+        arq_scheduler: Scheduler = app.dependency_overrides.get(
+            get_scheduler,
+            get_scheduler,
+        )(settings=settings)
+
         await ws_manager.connect_broadcaster()
+        await arq_scheduler.start(
+            redis_host=settings.REDIS_HOST,
+            redis_port=settings.REDIS_PORT,
+            redis_password=settings.REDIS_PASSWORD,
+            _dependency_overrides=app.dependency_overrides,
+        )
+
         yield
         hyperion_error_logger.info("Shutting down")
+        await arq_scheduler.close()
         await ws_manager.disconnect_broadcaster()
 
     # Initialize app

app/core/endpoints_core.py

@@ -28,7 +28,9 @@
     response_model=schemas_core.CoreInformation,
     status_code=200,
 )
-async def read_information(settings: Settings = Depends(get_settings)):
+async def read_information(
+    settings: Settings = Depends(get_settings),
+):
     """
     Return information about Hyperion. This endpoint can be used to check if the API is up.
     """

app/core/log.py

@@ -234,6 +234,7 @@ def get_config_dict(self, settings: Settings):
                     ],
                     "level": MINIMUM_LOG_LEVEL,
                 },
+                "scheduler": {"handlers": ["console"], "level": MINIMUM_LOG_LEVEL},
                 # We disable "uvicorn.access" to replace it with our custom "hyperion.access" which add custom information like the request_id
                 "uvicorn.access": {"handlers": []},
                 "uvicorn.error": {
@@ -245,6 +246,15 @@ def get_config_dict(self, settings: Settings):
                     "level": MINIMUM_LOG_LEVEL,
                     "propagate": False,
                 },
+                "arq.worker": {
+                    "handlers": [
+                        "console",
+                        "file_errors",
+                        "matrix_errors",
+                    ],
+                    "level": MINIMUM_LOG_LEVEL,
+                    "propagate": False,
+                },
             },
         }
 

app/core/notification/cruds_notification.py

@@ -9,96 +9,6 @@
 from app.core.notification.notification_types import CustomTopic, Topic
 
 
-async def create_message(
-    message: models_notification.Message,
-    db: AsyncSession,
-) -> None:
-    db.add(message)
-    try:
-        await db.commit()
-    except IntegrityError:
-        await db.rollback()
-        raise
-
-
-async def create_batch_messages(
-    messages: list[models_notification.Message],
-    db: AsyncSession,
-) -> None:
-    db.add_all(messages)
-    try:
-        await db.commit()
-    except IntegrityError:
-        await db.rollback()
-        raise
-
-
-async def get_messages_by_firebase_token(
-    firebase_token: str,
-    db: AsyncSession,
-) -> Sequence[models_notification.Message]:
-    result = await db.execute(
-        select(models_notification.Message).where(
-            models_notification.Message.firebase_device_token == firebase_token,
-        ),
-    )
-    return result.scalars().all()
-
-
-async def get_messages_by_context_and_firebase_tokens(
-    context: str,
-    firebase_tokens: list[str],
-    db: AsyncSession,
-) -> Sequence[models_notification.Message]:
-    result = await db.execute(
-        select(models_notification.Message).where(
-            models_notification.Message.context == context,
-            models_notification.Message.firebase_device_token.in_(firebase_tokens),
-        ),
-    )
-    return result.scalars().all()
-
-
-async def remove_message_by_context_and_firebase_device_token(
-    context: str,
-    firebase_device_token: str,
-    db: AsyncSession,
-):
-    await db.execute(
-        delete(models_notification.Message).where(
-            models_notification.Message.context == context,
-            models_notification.Message.firebase_device_token == firebase_device_token,
-        ),
-    )
-    await db.commit()
-
-
-async def remove_message_by_firebase_device_token(
-    firebase_device_token: str,
-    db: AsyncSession,
-):
-    await db.execute(
-        delete(models_notification.Message).where(
-            models_notification.Message.firebase_device_token == firebase_device_token,
-        ),
-    )
-    await db.commit()
-
-
-async def remove_messages_by_context_and_firebase_device_tokens_list(
-    context: str,
-    tokens: list[str],
-    db: AsyncSession,
-):
-    await db.execute(
-        delete(models_notification.Message).where(
-            models_notification.Message.context == context,
-            models_notification.Message.firebase_device_token.in_(tokens),
-        ),
-    )
-    await db.commit()
-
-
 async def get_firebase_devices_by_user_id(
     user_id: str,
     db: AsyncSession,

app/core/notification/endpoints_notification.py

@@ -15,9 +15,11 @@
     get_db,
     get_notification_manager,
     get_notification_tool,
+    get_scheduler,
     is_user,
     is_user_in,
 )
+from app.types.scheduler import Scheduler
 from app.utils.communication.notifications import NotificationManager, NotificationTool
 
 router = APIRouter(tags=["Notifications"])
@@ -61,6 +63,16 @@ async def register_firebase_device(
             db=db,
         )
 
+    user_topics = await cruds_notification.get_topic_memberships_by_user_id(
+        user_id=user.id,
+        db=db,
+    )
+    for topic in user_topics:
+        await notification_manager.subscribe_tokens_to_topic(
+            custom_topic=CustomTopic(topic.topic),
+            tokens=[firebase_token],
+        )
+
     firebase_device = models_notification.FirebaseDevice(
         user_id=user.id,
         firebase_device_token=firebase_token,
@@ -84,7 +96,7 @@ async def unregister_firebase_device(
     notification_manager: NotificationManager = Depends(get_notification_manager),
 ):
     """
-    Unregister a new firebase device for the user
+    Unregister a firebase device for the user
 
     **The user must be authenticated to use this endpoint**
     """
@@ -95,46 +107,15 @@ async def unregister_firebase_device(
         db=db,
     )
 
-
-@router.get(
-    "/notification/messages/{firebase_token}",
-    response_model=list[schemas_notification.Message],
-    status_code=200,
-)
-async def get_messages(
-    firebase_token: str,
-    db: AsyncSession = Depends(get_db),
-    # If we want to enable authentification for /messages/{firebase_token} endpoint, we may to uncomment the following line
-    # user: models_core.CoreUser = Depends(is_user()),
-):
-    """
-    Get all messages for a specific device from the user
-
-    **The user must be authenticated to use this endpoint**
-    """
-    firebase_device = await cruds_notification.get_firebase_devices_by_user_id_and_firebase_token(
-        # If we want to enable authentification for /messages/{firebase_token} endpoint, we may to uncomment the following line
-        firebase_token=firebase_token,
-        db=db,  # user_id=user.id,
-    )
-
-    if firebase_device is None:
-        raise HTTPException(
-            status_code=404,
-            detail="Device not found for user",  # {user.id}"
-        )
-
-    messages = await cruds_notification.get_messages_by_firebase_token(
-        firebase_token=firebase_token,
-        db=db,
-    )
-
-    await cruds_notification.remove_message_by_firebase_device_token(
-        firebase_device_token=firebase_token,
+    user_topics = await cruds_notification.get_topic_memberships_by_user_id(
+        user_id=user.id,
         db=db,
     )
-
-    return messages
+    for topic in user_topics:
+        await notification_manager.unsubscribe_tokens_to_topic(
+            custom_topic=CustomTopic(topic.topic),
+            tokens=[firebase_token],
+        )
 
 
 @router.post(
@@ -268,12 +249,9 @@ async def send_notification(
     **Only admins can use this endpoint**
     """
     message = schemas_notification.Message(
-        context="notification-test",
-        is_visible=True,
         title="Test notification",
         content="Ceci est un test de notification",
-        # The notification will expire in 3 days
-        expire_on=datetime.now(UTC) + timedelta(days=3),
+        action_module="test",
     )
     await notification_tool.send_notification_to_user(
         user_id=user.id,
@@ -288,24 +266,76 @@ async def send_notification(
 async def send_future_notification(
     user: models_core.CoreUser = Depends(is_user_in(GroupType.admin)),
     notification_tool: NotificationTool = Depends(get_notification_tool),
+    scheduler: Scheduler = Depends(get_scheduler),
 ):
     """
     Send ourself a test notification.
 
     **Only admins can use this endpoint**
     """
     message = schemas_notification.Message(
-        context="future-notification-test",
-        is_visible=True,
         title="Test notification future",
         content="Ceci est un test de notification future",
-        # The notification will expire in 3 days
-        expire_on=datetime.now(UTC) + timedelta(days=3),
-        delivery_datetime=datetime.now(UTC) + timedelta(minutes=3),
+        action_module="test",
     )
-    await notification_tool.send_notification_to_user(
-        user_id=user.id,
+    await notification_tool.send_notification_to_users(
+        user_ids=[user.id],
+        message=message,
+        defer_date=datetime.now(UTC) + timedelta(seconds=10),
+        scheduler=scheduler,
+        job_id="testtt",
+    )
+
+
+@router.post(
+    "/notification/send/topic",
+    status_code=201,
+)
+async def send_notification_topic(
+    user: models_core.CoreUser = Depends(is_user_in(GroupType.admin)),
+    notification_tool: NotificationTool = Depends(get_notification_tool),
+):
+    """
+    Send ourself a test notification.
+
+    **Only admins can use this endpoint**
+    """
+    message = schemas_notification.Message(
+        title="Test notification topic",
+        content="Ceci est un test de notification topic",
+        action_module="test",
+    )
+    await notification_tool.send_notification_to_topic(
+        custom_topic=CustomTopic.from_str("test"),
+        message=message,
+    )
+
+
+@router.post(
+    "/notification/send/topic/future",
+    status_code=201,
+)
+async def send_future_notification_topic(
+    user: models_core.CoreUser = Depends(is_user_in(GroupType.admin)),
+    notification_tool: NotificationTool = Depends(get_notification_tool),
+    scheduler: Scheduler = Depends(get_scheduler),
+):
+    """
+    Send ourself a test notification.
+
+    **Only admins can use this endpoint**
+    """
+    message = schemas_notification.Message(
+        title="Test notification future topic",
+        content="Ceci est un test de notification future topic",
+        action_module="test",
+    )
+    await notification_tool.send_notification_to_topic(
+        custom_topic=CustomTopic.from_str("test"),
         message=message,
+        defer_date=datetime.now(UTC) + timedelta(seconds=10),
+        job_id="test26",
+        scheduler=scheduler,
     )
 
 

app/core/notification/notification_types.py

@@ -15,6 +15,7 @@ class Topic(str, Enum):
     raffle = "raffle"
     vote = "vote"
     ph = "ph"
+    test = "test"
 
 
 class CustomTopic: