feat: Optionally store analytics data

The analytics data is just stored on disk but not made available or processed in any way yet.

Author: defnull

bbblb/services/__init__.py

@@ -215,9 +215,9 @@ def configure_logging(config: BBBLBConfig):
 async def bootstrap(
     config: BBBLBConfig, autostart=True, logging=True
 ) -> ServiceRegistry:
-    import bbblb.settings
     import bbblb.services.poller
     import bbblb.services.recording
+    import bbblb.services.analytics
     import bbblb.services.locks
     import bbblb.services.db
     import bbblb.services.bbb
@@ -255,6 +255,10 @@ def watch_debug_level(name, old, new):
         "importer",
         bbblb.services.recording.RecordingManager(config),
     )
+    ctx.register(
+        "analytics",
+        bbblb.services.analytics.AnalyticsHandler(config),
+    )
 
     if autostart:
         for name, service in ctx.services.items():

bbblb/services/analytics.py

@@ -0,0 +1,29 @@
+import asyncio
+import json
+import logging
+
+from bbblb import model
+from bbblb.services import ManagedService
+from bbblb.settings import BBBLBConfig
+
+LOG = logging.getLogger(__name__)
+
+
+class AnalyticsHandler(ManagedService):
+    """Store analytics data to {PATH_DATA}/{tenant}/{meetingID}.json"""
+
+    def __init__(self, config: BBBLBConfig):
+        self.store_path = (config.PATH_DATA / "analytics").resolve()
+
+    async def on_start(self):
+        await super().on_start()
+
+    async def on_shutdown(self):
+        return await super().on_shutdown()
+
+    async def store(self, tenant: model.Tenant, data: dict):
+        """Store the payload of an analytics callback"""
+        internal_meeting_id = data["internal_meeting_id"]
+        dst = self.store_path / tenant.name / f"{internal_meeting_id}.json"
+        await asyncio.to_thread(dst.parent.mkdir, parents=True, exist_ok=True)
+        await asyncio.to_thread(dst.write_text, json.dumps(data))

bbblb/services/bbb.py

@@ -52,10 +52,11 @@ async def trigger_callback(
                     continue
 
     async def fire_callback(self, callback: model.Callback, payload: dict, clear=True):
-        url = callback.forward
-        key = callback.tenant.secret
-        data = {"signed_parameters": jwt.encode(payload, key, "HS256")}
-        await self.trigger_callback("POST", url, data=data)
+        forward_url = callback.forward
+        if forward_url:
+            key = callback.tenant.secret
+            data = {"signed_parameters": jwt.encode(payload, key, "HS256")}
+            await self.trigger_callback("POST", forward_url, data=data)
         if clear:
             async with self.db.session() as session, session.begin():
                 await session.delete(callback)

bbblb/settings.py

@@ -235,6 +235,12 @@ class BBBLBConfig(BaseConfig):
     #: by LOAD_PENALTY. The applied penalty decreases linearly over time.
     LOAD_COOLDOWN: float = 30
 
+    #: If true, BBBLB will intercept the analytics-callback-url webhook
+    #: and dump json files into the {PATH_DATA}/analytics/{tenant}/
+    #: folder for later analysis (WIP). The callback is only fired if
+    #: BBB is configured with defaultKeepEvents=true in bbb-web.properties.
+    ANALYTICS_STORE: bool = False
+
     #: Maximum number of meetings or recordings to return from APIs that
     #: potentially return an unlimited amount of data.
     MAX_ITEMS: int = 1000

bbblb/web/bbbapi.py

@@ -254,20 +254,27 @@ async def _intercept_callbacks(
     # For all other known callbacks (other than meta_endCallbackUrl) we assume
     # that they follow the JWT model and can be proxied immediately. They still
     # need to be intercepted because we have to re-sign their payload.
-    for meta in ("meta_analytics-callback-url",):
+
+    forward = set(("meta_analytics-callback-url",))
+    intercept = set()
+    if cxt.config.ANALYTICS_STORE:
+        intercept.add("meta_analytics-callback-url")
+
+    for meta in forward:
         orig_url = params.pop(meta, None)
-        if orig_url:
-            typename = meta[5:-14]  # Just the middle part
-            if is_new:
-                callbacks.append(
-                    model.Callback(
-                        uuid=meeting.uuid,
-                        type=typename,
-                        tenant=meeting.tenant,
-                        server=meeting.server,
-                        forward=orig_url,
-                    )
+        typename = meta[5:-14]  # Just the middle part
+        if is_new:
+            callbacks.append(
+                model.Callback(
+                    uuid=meeting.uuid,
+                    type=typename,
+                    tenant=meeting.tenant,
+                    server=meeting.server,
+                    forward=orig_url,  # can be None
                 )
+            )
+
+        if orig_url or meta in intercept:
             url = cxt.request.url_for(
                 "bbblb:callback_proxy",
                 uuid=meeting.uuid,

bbblb/web/bbblbapi.py

@@ -6,6 +6,7 @@
 import logging
 import jwt
 
+from bbblb.services.analytics import AnalyticsHandler
 from bbblb.web import bbbapi
 from bbblb import model
 
@@ -257,6 +258,7 @@ async def handle_callback_proxy(ctx: BBBLBApiRequest):
     except (UnicodeDecodeError, KeyError, IndexError):
         return Response("Invalid request", 400)
 
+    # Forward callbacks requested by the original client
     stmt = model.Callback.select(uuid=meeting_uuid, type=callback_type)
     callbacks = (await ctx.session.execute(stmt)).scalars().all()
     if not callbacks:
@@ -269,8 +271,15 @@ async def handle_callback_proxy(ctx: BBBLBApiRequest):
     except BaseException:
         return Response("Access denied, signature check failed", 401)
 
+    # Intercept callbacks we are interested in
+    if ctx.config.ANALYTICS_STORE and callback_type == "analytics":
+        analytics = await ctx.services.use("analytics", AnalyticsHandler)
+        # Fire and forget
+        asyncio.create_task(analytics.store(callbacks[0].tenant, payload))
+
     # Find and trigger callbacks
     for callback in callbacks:
+        # Fire and forget
         asyncio.create_task(ctx.bbb.fire_callback(callback, payload, clear=True))
 
     return Response("OK", 200)

docs/_config.rst

@@ -103,6 +103,13 @@ meeting age. See LOAD_COOLDOWN.
 Number of minutes after which new meetings are no longer impacted
 by LOAD_PENALTY. The applied penalty decreases linearly over time.
 
+``ANALYTICS_STORE`` (type: ``bool``, default: ``False``)
+
+If true, BBBLB will intercept the analytics-callback-url webhook
+and dump json files into the {PATH_DATA}/analytics/{tenant}/
+folder for later analysis (WIP). The callback is only fired if
+BBB is configured with defaultKeepEvents=true in bbb-web.properties.
+
 ``MAX_ITEMS`` (type: ``int``, default: ``1000``)
 
 Maximum number of meetings or recordings to return from APIs that

examples/bbblb-compose/bbblb.env.example

@@ -103,6 +103,13 @@
 # (default: 30; type: float)
 #BBBLB_LOAD_COOLDOWN=
 
+# If true, BBBLB will intercept the analytics-callback-url webhook
+# and dump json files into the {PATH_DATA}/analytics/{tenant}/
+# folder for later analysis (WIP). The callback is only fired if
+# BBB is configured with defaultKeepEvents=true in bbb-web.properties.
+# (default: False; type: bool)
+#BBBLB_ANALYTICS_STORE=
+
 # Maximum number of meetings or recordings to return from APIs that
 # potentially return an unlimited amount of data.
 # (default: 1000; type: int)