Merge pull request #119 from nebulabroadcast/coalesce-requests

Coalesce requests

Author: martastain

backend/api/order/set_rundown_order.py

@@ -52,10 +52,6 @@ async def set_rundown_order(
                         )
                         continue
 
-                    if not item:
-                        nebula.log.trace(f"Skipping {item}", user=user.name)
-                        continue
-
                 # Shut-up mypy
                 assert item is not None, "Item should not be None at this point"
                 assert item["id_bin"] is not None, "Item w/o bin. Shouldn't happen."

backend/api/rundown/get_rundown.py

@@ -4,19 +4,19 @@
 from nebula.enum import ObjectStatus, RunMode
 from nebula.helpers.scheduling import get_pending_assets, parse_rundown_date
 
-from .models import RundownRequestModel, RundownResponseModel, RundownRow
+from .models import RundownResponseModel, RundownRow
 
 
-async def get_rundown(request: RundownRequestModel) -> RundownResponseModel:
+async def get_rundown(id_channel: int, date: str | None = None) -> RundownResponseModel:
     """Get a rundown"""
-    if not (channel := nebula.settings.get_playout_channel(request.id_channel)):
-        raise nebula.BadRequestException(f"No such channel: {request.id_channel}")
+    if not (channel := nebula.settings.get_playout_channel(id_channel)):
+        raise nebula.BadRequestException(f"No such channel: {id_channel}")
 
     request_start_time = time.monotonic()
-    start_time = parse_rundown_date(request.date, channel)
+    start_time = parse_rundown_date(date, channel)
     end_time = start_time + (3600 * 24)
     pending_assets = await get_pending_assets(channel.send_action)
-    pskey = f"playout_status/{request.id_channel}"
+    pskey = f"playout_status/{id_channel}"
 
     query = """
         SELECT
@@ -75,9 +75,7 @@ async def get_rundown(request: RundownRequestModel) -> RundownResponseModel:
     last_event = None
     ts_broadcast = ts_scheduled = 0.0
 
-    async for record in nebula.db.iterate(
-        query, request.id_channel, start_time, end_time
-    ):
+    async for record in nebula.db.iterate(query, id_channel, start_time, end_time):
         id_event = record["id_event"]
         id_item = record["id_item"]
         id_bin = record["id_bin"]

backend/api/rundown/rundown_request.py

@@ -1,4 +1,5 @@
 import nebula
+from nebula.helpers.coalescer import Coalescer
 from server.dependencies import CurrentUser
 from server.request import APIRequest
 
@@ -34,4 +35,6 @@ async def handle(
         if not user.can("rundown_view", request.id_channel):
             raise nebula.ForbiddenException("You are not allowed to view this rundown")
 
-        return await get_rundown(request)
+        coalesce = Coalescer()
+        rundown = await coalesce(get_rundown, request.id_channel, request.date)
+        return rundown

backend/api/scheduler/scheduler.py

@@ -1,28 +1,33 @@
 import nebula
-from nebula.helpers.create_new_event import create_new_event
+from nebula.helpers.create_new_event import EventData, create_new_event
 from nebula.helpers.scheduling import parse_rundown_date
 
-from .models import SchedulerRequestModel, SchedulerResponseModel
+from .models import SchedulerResponseModel
 from .utils import delete_events, get_event_at_time, get_events_in_range
 
-
 async def scheduler(
-    request: SchedulerRequestModel,
+    id_channel: int,
+    date: str | None = None,
+    days: int = 7,
+    delete: list[int] | None = None,
+    events: list[EventData] | None = None,
     editable: bool = True,
     user: nebula.User | None = None,
 ) -> SchedulerResponseModel:
     """Modify and display channel schedule"""
-    start_time: float | None = None
-    end_time: float | None = None
 
     username = user.name if user else None
+    start_time: float | None = None
+    end_time: float | None = None
+    delete = delete or []
+    events = events or []
 
-    if not (channel := nebula.settings.get_playout_channel(request.id_channel)):
-        raise nebula.BadRequestException(f"No such channel {request.id_channel}")
+    if not (channel := nebula.settings.get_playout_channel(id_channel)):
+        raise nebula.BadRequestException(f"No such channel {id_channel}")
 
-    if request.date:
-        start_time = parse_rundown_date(request.date, channel)
-        end_time = start_time + (request.days * 86400)
+    if date:
+        start_time = parse_rundown_date(date, channel)
+        end_time = start_time + (days * 86400)
 
     affected_events: list[int] = []
     affected_bins: list[int] = []
@@ -31,14 +36,14 @@ async def scheduler(
     # Delete events
     #
 
-    if request.delete and editable:
-        deleted_event_ids = await delete_events(request.delete, user=user)
+    if delete and editable:
+        deleted_event_ids = await delete_events(delete, user=user)
         affected_events.extend(deleted_event_ids)
     #
     # Create / update events
     #
 
-    for event_data in request.events:
+    for event_data in events:
         if not editable:
             # weird syntax, but keeps indentation level low
             break
@@ -127,11 +132,16 @@ async def scheduler(
     # Return existing events
 
     if (start_time is not None) and (end_time is not None):
-        events = await get_events_in_range(channel.id, start_time, end_time, user=user)
+        c_events = await get_events_in_range(
+            channel.id,
+            start_time,
+            end_time,
+            user=user,
+        )
     else:
-        events = []
+        c_events = []
     return SchedulerResponseModel(
-        events=[e.meta for e in events],
+        events=[e.meta for e in c_events],
         affected_events=affected_events,
         affected_bins=affected_bins,
     )

backend/api/scheduler/scheduler_request.py

@@ -1,4 +1,5 @@
 import nebula
+from nebula.helpers.coalescer import Coalescer
 from nebula.helpers.scheduling import bin_refresh
 from server.dependencies import CurrentUser, RequestInitiator
 from server.request import APIRequest
@@ -29,8 +30,32 @@ async def handle(
         if not user.can("scheduler_view", request.id_channel):
             raise nebula.ForbiddenException("You are not allowed to view this channel")
 
+        if not (request.events or request.delete):
+            # Read-only request. coalesce the requests and
+            # Return directly
+            coalesce = Coalescer()
+            result = await coalesce(
+                scheduler,
+                request.id_channel,
+                date=request.date,
+                days=request.days,
+                user=user,
+            )
+            return result
+
+        # Write request. Do not coalesce, and send notifications
+
         editable = user.can("scheduler_edit", request.id_channel)
-        result = await scheduler(request, editable, user=user)
+
+        result = await scheduler(
+            request.id_channel,
+            date=request.date,
+            days=request.days,
+            editable=editable,
+            events=request.events,
+            delete=request.delete,
+            user=user,
+        )
 
         if result.affected_bins:
             await bin_refresh(

backend/nebula/helpers/coalescer.py

@@ -0,0 +1,83 @@
+"""
+This module provides a Coalescer class that prevents multiple identical,
+concurrent asynchronous operations from being executed simultaneously.
+
+If you invoke an async function (e.g., fetch_data(id=1)) through the Coalescer instance,
+and another call to fetch_data(id=1) is already in progress, the Coalescer ensures
+that the second (and any subsequent identical) call will simply await the result
+of the first, ongoing operation, rather than initiating a new, redundant execution.
+
+In essence: It's a smart wrapper for async functions that says:
+
+> If I'm already doing this exact same thing,  don't start it again.
+> Just wait for the one already running to finish and share its result.
+
+## Example usage:
+
+```
+async def fetch_data(item_id: int):
+    print(f"Fetching data for {item_id}...")
+    await asyncio.sleep(2) # Simulate network call
+    return f"Data for {item_id}"
+
+coalescer = Coalescer()
+
+# These calls will be coalesced if made close together
+
+task1 = asyncio.create_task(coalescer(fetch_data, item_id=1))
+task2 = asyncio.create_task(coalescer(fetch_data, item_id=1)) # Will use task1's future
+task3 = asyncio.create_task(coalescer(fetch_data, item_id=2)) # New actual call
+
+result1 = await task1
+result2 = await task2
+result3 = await task3
+print(result1, result2, result3)
+```
+"""
+
+import asyncio
+import hashlib
+from collections.abc import Callable, Coroutine
+from typing import Any, Generic, TypeVar
+
+
+def _hash_args(func: Callable[..., Any], *args: Any, **kwargs: Any) -> str:
+    """Generates a hash from the function arguments and keyword arguments."""
+    func_id = str(id(func))
+    arg_str = str(args)
+    kwarg_str = str(sorted(kwargs.items()))
+    combined_str = arg_str + kwarg_str + func_id
+    return hashlib.md5(combined_str.encode()).hexdigest()  # noqa: S324
+
+
+T = TypeVar("T")
+
+
+class Coalescer(Generic[T]):
+    _instance: "Coalescer[Any] | None" = None
+    lock: asyncio.Lock
+    current_futures: dict[str, asyncio.Task[T]]
+
+    def __new__(cls) -> "Coalescer[Any]":
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance.current_futures = {}
+            cls._instance.lock = asyncio.Lock()
+        return cls._instance
+
+    async def __call__(
+        self,
+        func: Callable[..., Coroutine[Any, Any, T]],
+        *args: Any,
+        **kwargs: Any,
+    ) -> T:
+        key = _hash_args(func, *args, **kwargs)
+        async with self.lock:
+            if key not in self.current_futures:
+                self.current_futures[key] = asyncio.create_task(func(*args, **kwargs))
+
+        try:
+            return await self.current_futures[key]
+        finally:
+            async with self.lock:
+                self.current_futures.pop(key, None)

backend/nebula/settings/models.py

@@ -98,7 +98,7 @@ class SSOProvider(SettingsModel):
         str,
         Field(
             title="Name",
-            example="myoauth",
+            examples=["myoauth"],
         ),
     ]
 
@@ -107,7 +107,7 @@ class SSOProvider(SettingsModel):
         Field(
             title="Title",
             description="Used on the SSO button on the login page",
-            example="Log in using MyOauth",
+            examples=["Log in using MyOauth"],
         ),
     ]
 
@@ -124,23 +124,23 @@ class SSOProvider(SettingsModel):
         Field(
             title="Entrypoint",
             description="URL to the SSO provider configuration endpoint",
-            example="https://iam.example.com/realms/nebula/.well-known/openid-configuration",
+            examples=["https://iam.example.com/realms/nebula/.well-known/openid-configuration"],
         ),
     ] = None
 
     client_id: Annotated[
         str,
         Field(
             title="Client ID",
-            example="myclientid",
+            examples=["myclientid"],
         ),
     ]
 
     client_secret: Annotated[
         str,
         Field(
             title="Client secret",
-            example="myclientsecret",
+            examples=["myclientsecret"],
         ),
     ]