Merge pull request #160 from arenaxr/mongodb

feat(persist): add direct pymongo persist db connection for all reads, rm REST persist reads

Author: mwfarb

requirements.txt

@@ -13,3 +13,4 @@ drf-yasg==1.21.14
 python-dotenv==1.2.1
 requests==2.32.5
 django-autocomplete-light==3.12.1
+pymongo==4.10.1

users/apps.py

@@ -5,7 +5,7 @@
 def post_migration_callback(sender, **kwargs):
     from . import startup
 
-    startup.setup_socialapps()
+    startup.setup_databases()
 
 
 class UsersConfig(AppConfig):

users/forms.py

@@ -5,8 +5,7 @@
 from django.contrib.auth.models import User
 
 from .models import Device, Namespace, Scene
-from .mqtt import TOPIC_SUPPORTED_API_VERSIONS, all_scenes_read_token
-from .persistence import get_persist_scenes_ns
+from .persistence import read_persist_scenes_by_namespace
 
 
 class SocialSignupForm(_SocialSignupForm):
@@ -30,9 +29,7 @@ def clean(self):
             self.add_error("username", msg)
         # reject usernames in form on signup: Namespace used in persist db
         else:
-            version = TOPIC_SUPPORTED_API_VERSIONS[0]  # TODO (mwfarb): resolve missing request.version
-            token = all_scenes_read_token(version)
-            if len(get_persist_scenes_ns(token, username)) > 0:
+            if len(read_persist_scenes_by_namespace([username])) > 0:
                 msg = f"Sorry, '{username}' is a persistence namespace."
                 self.add_error("username", msg)
 

users/models.py

@@ -19,6 +19,23 @@
     RE_NS_SLASH_ID, "Only alphanumeric, underscore, hyphen, in namespace/idname format allowed."
 )
 
+# assign accessible model for persist collection
+def get_arenaobjects_collection():
+    from .persist_db import get_persist_db
+    return get_persist_db()['arenaobjects']
+
+# arenaobjects schema reference:
+# https://github.com/arenaxr/arena-persist/blob/master/server.js#L28-L42
+# object_id: {type: String, required: true, index: true},
+# type: {type: String, required: true, index: true},
+# attributes: {type: Object, required: true, default: {}},
+# expireAt: {type: Date, expires: 0},
+# realm: {type: String, required: true, index: true},
+# namespace: {type: String, required: true, index: true, default: 'public'},
+# sceneId: {type: String, required: true, index: true},
+# private: {type: Boolean},
+# program_id: {type: String},
+
 
 class NamespaceDefault:
     def __init__(self, name=""):

users/mqtt.py

@@ -29,31 +29,6 @@
 TOPIC_SUPPORTED_API_VERSIONS = [API_V2]
 
 
-def all_scenes_read_token(version):
-    config = settings.PUBSUB
-    privkeyfile = settings.MQTT_TOKEN_PRIVKEY
-    if not os.path.exists(privkeyfile):
-        print("Error: keyfile not found" + privkeyfile)
-        return None
-    with open(privkeyfile) as privatefile:
-        private_key = privatefile.read()
-
-    realm = config["mqtt_realm"]
-    username = config["mqtt_username"]
-
-    payload = {}
-    payload["sub"] = username
-    payload["exp"] = datetime.datetime.utcnow() + DEF_JWT_DURATION
-
-    if version == API_V2:
-        payload["subs"] = [f"{realm}/s/+/+/o/+/+"]  # v2
-    else:
-        payload["subs"] = [f"{realm}/s/#"]  # v1
-
-    token = jwt.encode(payload, private_key, algorithm="RS256")
-    return token
-
-
 def generate_arena_token(
     *,
     user,

users/persist_db.py

@@ -0,0 +1,40 @@
+
+import logging
+
+from pymongo import MongoClient
+from pymongo.database import Database
+
+'''
+persist_db.py: This Mongo connection manager is triggered at startup and checked
+before each database call, this allows a lazy connection to Mongo. There is no
+explicit disconnect from Mongo. PyMongo manages its own connection pooling and the OS
+handles socket cleanup on process exit. Plus, Django shutdown is hard to detect.
+'''
+
+client: MongoClient = None
+db: Database = None
+
+logging.getLogger("pymongo").setLevel(logging.WARNING)
+
+def get_persist_db():
+    global client, db
+    if db is not None:
+        return db
+
+    # connect to mongodb, read-only
+    print("arena_persist: connecting...")
+    client = MongoClient("mongodb://mongodb/arena_persist?readPreference=primaryPreferred")
+
+    try:
+        dba = client.admin
+        server_status = dba.command('serverStatus')
+        current_connections = server_status['connections']['current']
+        print(f"arena_persist: current connections: {current_connections}")
+        total_connections = server_status['connections']['totalCreated']
+        print(f"arena_persist: total connections created: {total_connections}")
+
+    except Exception as e:
+        print(f"arena_persist: error: {e}")
+
+    db = client.arena_persist
+    return db

users/persistence.py

@@ -1,59 +1,102 @@
 import json
+from datetime import datetime
 
 import requests
+from bson import ObjectId
 from requests.exceptions import HTTPError
 
+from .models import get_arenaobjects_collection
 from .utils import get_rest_host
 
 PERSIST_TIMEOUT = 30  # 30 seconds
 
 
-def get_scene_objects(token, scene):
-    # get scene from persist
-    verify, host = get_rest_host()
-    url = f"https://{host}/persist/{scene}"
-    result = _urlopen(url, token, "GET", verify)
-    if result:
-        return json.loads(result)
-    return []
+# Mongo DB PyMongo queries for Persist:
+# https://pymongo.readthedocs.io/en/stable/index.html
 
 
-def delete_scene_objects(token, scene):
-    # delete scene from persist
-    verify, host = get_rest_host()
-    url = f"https://{host}/persist/{scene}"
-    result = _urlopen(url, token, "DELETE", verify)
-    return result
+class MongoJSONEncoder(json.JSONEncoder):
+    def default(self, o):
+        if isinstance(o, ObjectId):
+            return str(o)  # Convert ObjectId to string
+        if isinstance(o, datetime):
+            return o.isoformat()  # Convert datetime to ISO format
+        return super().default(o)  # Call the default method for other types
 
 
-def get_persist_ns_all(token):
-    # request all scenes from persist
-    verify, host = get_rest_host()
-    url = f"https://{host}/persist/!allnamespaces"
-    result = _urlopen(url, token, "GET", verify)
-    if result:
-        return json.loads(result)
-    return []
+def read_persist_ns_all():
+    arenaobjects = get_arenaobjects_collection().aggregate([{
+        "$group": {
+            "_id": {
+                "namespace": "$namespace",
+                }
+            }
+        }
+    ])
+    return [doc['_id']['namespace'] for doc in arenaobjects]
 
 
-def get_persist_scenes_all(token):
-    # request all scenes from persist
-    verify, host = get_rest_host()
-    url = f"https://{host}/persist/!allscenes"
-    result = _urlopen(url, token, "GET", verify)
-    if result:
-        return json.loads(result)
-    return []
+def read_persist_scenes_all():
+    arenaobjects = get_arenaobjects_collection().aggregate([
+        {
+            "$group": {
+                "_id": {
+                    "namespace": "$namespace",
+                    "sceneId": "$sceneId",
+                }
+            }
+        }
+    ])
+    unique_scenes = []
+    for doc in arenaobjects:
+        ns = doc['_id']['namespace']
+        sc = doc['_id']['sceneId']
+        unique_scenes.append(f"{ns}/{sc}")
+
+    return unique_scenes
+
 
+def read_persist_scenes_by_namespace(namespaces):
+    arenaobjects = get_arenaobjects_collection().aggregate([
+        {
+            "$match": {
+                "namespace": {"$in": namespaces}
+            }
+        },
+        {
+            "$group": {
+                "_id": {
+                    "namespace": "$namespace",
+                    "sceneId": "$sceneId",
+                }
+            }
+        }
+    ])
+    unique_scenes = []
+    for doc in arenaobjects:
+        ns = doc['_id']['namespace']
+        sc = doc['_id']['sceneId']
+        unique_scenes.append(f"{ns}/{sc}")
 
-def get_persist_scenes_ns(token, namespace):
-    # request all namespace scenes from persist
+    return unique_scenes
+
+
+def read_persist_scene_objects(namespace, scene):
+    query = {"namespace": namespace, "sceneId": scene}
+    arenaobjects = get_arenaobjects_collection().find(query)
+    json_str = MongoJSONEncoder().encode(list(arenaobjects))
+    return json.loads(json_str)
+
+
+# Mongo DB REST queries for Persist:
+
+
+def delete_scene_objects(token, scene):
+    # delete scene objects from persist
     verify, host = get_rest_host()
-    url = f"https://{host}/persist/{namespace}/!allscenes"
-    result = _urlopen(url, token, "GET", verify)
-    if result:
-        return json.loads(result)
-    return []
+    url = f"https://{host}/persist/{scene}"
+    result = _urlopen(url, token, "DELETE", verify)
+    return result
 
 
 def _urlopen(url, token, method, verify):

users/serializers.py

@@ -13,7 +13,6 @@ class Meta:
             "name",
             "editors",
             "viewers",
-            "is_default",
         ]
 
 
@@ -42,7 +41,6 @@ class Meta:
             "anonymous_users",
             "video_conference",
             "users",
-            "is_default",
         ]
 
 

users/startup.py

@@ -3,9 +3,22 @@
 from allauth.socialaccount.models import SocialApp
 from django.conf import settings
 from django.contrib.sites.models import Site
+from users.models import (
+    SCENE_ANON_USERS_DEF,
+    SCENE_PUBLIC_READ_DEF,
+    SCENE_PUBLIC_WRITE_DEF,
+    SCENE_USERS_DEF,
+    SCENE_VIDEO_CONF_DEF,
+    Scene,
+)
 
 
-def setup_socialapps():
+from users.persist_db import get_persist_db
+
+def setup_databases():
+    # Force db connection
+    get_persist_db()
+
     # Sites db must have a SITE_ID row that equals our host
     host = os.getenv("HOSTNAME")
     site_id = settings.SITE_ID
@@ -17,7 +30,9 @@ def setup_socialapps():
             # check that another site is not using our host, remove
             host_inst = Site.objects.filter(domain=host)
             if host_inst.exists():
-                host_inst.delete()
+                count, _ = host_inst.delete()
+                if count > 0:
+                    print(f"Deleted Site entries matching host: {count}")
             # update proper site_id with host
             id_inst.name = host
             id_inst.domain = host
@@ -27,4 +42,19 @@ def setup_socialapps():
         id_inst.save()
 
     # social apps are defined in settings.py, so remove legacy allauth entires that might conflict
-    SocialApp.objects.filter(provider='google').delete()
+    count, _ = SocialApp.objects.filter(provider='google').delete()
+    if count > 0:
+        print(f"Deleted legacy SocialApp entries: {count}")
+
+    # check for scene permissions where Scene.is_default is True and remove them
+    count, _ = Scene.objects.filter(
+        public_read=SCENE_PUBLIC_READ_DEF,
+        public_write=SCENE_PUBLIC_WRITE_DEF,
+        anonymous_users=SCENE_ANON_USERS_DEF,
+        video_conference=SCENE_VIDEO_CONF_DEF,
+        users=SCENE_USERS_DEF,
+        editors__isnull=True,
+        viewers__isnull=True,
+    ).delete()
+    if count > 0:
+        print(f"Deleted redundant default Scene permission entries: {count}")

users/templates/users/user_profile.html

@@ -135,9 +135,6 @@ <h5>Namespace Permissions
                         <tr>
                             <!--name-->
                             <td><b>{{ namespace.name }}</b>
-                                {% if namespace.is_default %}
-                                <b> (default)</b>
-                                {% endif %}
                             </td>
                             <!--user account-->
                             <td>
@@ -215,9 +212,6 @@ <h5>Scene Permissions <button id="button_expand_add_scene" type="button" title="
                             <!--name-->
                             <td><a
                                     href="{{ request.scheme }}://{{ request.META.HTTP_HOST }}/{{ scene.name }}"><b>{{ scene.name }}</b></a>
-                                {% if scene.is_default %}
-                                <b> (default)</b>
-                                {% endif %}
                             </td>
                             {% if user.is_staff %}
                             <!--persisted in db-->