Fix active/all token issues

.gitea/workflows/release.yml

@@ -0,0 +1,58 @@
+name: release-image
+
+on:
+  push:
+    branches:
+      - master
+
+jobs:
+  release-image:
+    runs-on: ubuntu-latest
+    env:
+      DOCKER_LATEST: nightly
+      CONTAINER_NAME: drb-core-server
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v3
+
+      - name: Set up Docker BuildX
+        uses: docker/setup-buildx-action@v3
+        with: # replace it with your local IP
+          config-inline: |
+            [registry."git.vpn.cusano.net"] 
+              http = false
+              insecure = false            
+
+      - name: Login to DockerHub
+        uses: docker/login-action@v3
+        with:
+          registry: git.vpn.cusano.net # replace it with your local IP
+          username: ${{ secrets.GIT_REPO_USERNAME }}
+          password: ${{ secrets.GIT_REPO_PASSWORD }}
+          
+      - name: Get Meta
+        id: meta
+        run: |
+          echo REPO_NAME=$(echo ${GITHUB_REPOSITORY} | awk -F"/" '{print $2}') >> $GITHUB_OUTPUT
+          echo REPO_VERSION=$(git describe --tags --always | sed 's/^v//') >> $GITHUB_OUTPUT                
+
+      - name: Validate build configuration
+        uses: docker/build-push-action@v6
+        with:
+          call: check
+
+      - name: Build and push
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: ./Dockerfile
+          platforms: |
+            linux/arm64
+            linux/amd64
+          push: true
+          tags: | # replace it with your local IP and tags
+            git.vpn.cusano.net/logan/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.REPO_VERSION }}
+            git.vpn.cusano.net/logan/${{ env.CONTAINER_NAME }}:${{ env.DOCKER_LATEST }}

.gitignore

@@ -1,3 +1,4 @@
 *.venv
 *__pycache__/
-*.bat
+*.bat
+.DS_Store

Makefile

@@ -15,5 +15,7 @@ run: build
 		--name $(SERVER_CONTAINER_NAME) \
 		-e DB_NAME=$(DB_NAME) \
 		-e MONGO_URL=$(MONGO_URL) \
-		--network=host \
+		-e JWT_SECRET_KEY=$(JWT_SECRET_KEY) \
+		-p 5000:5000 \
+		-p 8765:8765 \
 		$(SERVER_IMAGE)

app/config/jwt_config.py

@@ -0,0 +1,16 @@
+import os
+from quart_jwt_extended import JWTManager
+
+# Initialize JWTManager outside of any function to ensure it's a singleton
+# It will be initialized with the app object later in server.py
+jwt = JWTManager()
+
+def configure_jwt(app):
+    """Configures JWT settings for the Quart app."""
+    app.config["JWT_SECRET_KEY"] = os.getenv("JWT_SECRET_KEY", "your-super-secret-key-that-should-be-in-env")
+    app.config["JWT_ACCESS_TOKEN_EXPIRES"] = 3600 # 1 hour
+    app.config["JWT_ALGORITHM"] = "HS256"
+    # You might need to set a custom error handler for unauthorized access
+    # @app.errorhandler(401)
+    # def unauthorized(error):
+    #     return jsonify({"msg": "Missing or invalid token"}), 401

app/internal/auth_wrappers.py

@@ -0,0 +1,77 @@
+# internal/auth_wrappers.py
+import os
+import asyncio
+from uuid import uuid4
+from typing import Optional, List, Dict, Any
+from internal.db_handler import MongoHandler #
+from internal.types import User, UserRoles
+
+DB_NAME = os.getenv("DB_NAME", "default_db")
+MONGO_URL = os.getenv("MONGO_URL", "mongodb://10.10.202.4:27017/")
+
+USER_DB_COLLECTION_NAME = "users"
+
+class UserDbController:
+    def __init__(self):
+        self.db_h = MongoHandler(DB_NAME, USER_DB_COLLECTION_NAME, MONGO_URL) #
+
+    async def close_db_connection(self):
+        """Closes the underlying MongoDB connection."""
+        if self.db_h:
+            await self.db_h.close_client() #
+
+    async def create_user(self, user_data: Dict[str, Any]) -> Optional[User]:
+        try:
+            if not user_data.get("_id"):
+                user_data['_id'] = str(uuid4())
+
+            inserted_id = None
+            async with self.db_h as db: #
+                insert_result = await db.insert_one(user_data) #
+                inserted_id = insert_result.inserted_id
+
+            if inserted_id:
+                query = {"_id": inserted_id}
+                inserted_doc = None
+                async with self.db_h as db: #
+                    inserted_doc = await db.find_one(query) #
+                if inserted_doc:
+                    return User.from_dict(inserted_doc)
+            return None
+        except Exception as e:
+            print(f"Create user failed: {e}")
+            return None
+
+    async def find_user(self, query: Dict[str, Any]) -> Optional[User]:
+        try:
+            found_doc = None
+            async with self.db_h as db: #
+                # The 'db' here is self.db_h, which is an instance of MongoHandler.
+                # MongoHandler.find_one will be called.
+                found_doc = await db.find_one(query) #
+            if found_doc:
+                return User.from_dict(found_doc)
+            return None
+        except Exception as e:
+            print(f"Find user failed: {e}") # This error should be less frequent or indicate actual DB issues now
+            return None
+
+    async def update_user(self, query: Dict[str, Any], update_data: Dict[str, Any]) -> Optional[int]:
+        try:
+            update_result = None
+            async with self.db_h as db: #
+                update_result = await db.update_one(query, update_data) #
+            return update_result.modified_count
+        except Exception as e:
+            print(f"Update user failed: {e}")
+            return None
+
+    async def delete_user(self, query: Dict[str, Any]) -> Optional[int]:
+        try:
+            delete_result = None
+            async with self.db_h as db: #
+                delete_result = await db.delete_one(query) #
+            return delete_result.deleted_count
+        except Exception as e:
+            print(f"Delete user failed: {e}")
+            return None

app/internal/db_handler.py

@@ -4,8 +4,9 @@ from typing import Optional, Dict, Any, List
 
 class MongoHandler:
     """
-    A basic asynchronous handler for MongoDB operations using motor.
+    An asynchronous handler for MongoDB operations using motor.
-    Designed to be used with asyncio.
+    The client connection is established on first use (via async with or direct call to connect())
+    and should be explicitly closed by calling close_client() at application shutdown.
     """
     def __init__(self, db_name: str, collection_name: str, mongo_uri: str = "mongodb://localhost:27017/"):
         """
@@ -15,7 +16,6 @@ class MongoHandler:
             db_name (str): The name of the database to connect to.
             collection_name (str): The name of the collection to use.
             mongo_uri (str): The MongoDB connection string URI.
-                             Defaults to the standard local URI.
         """
         self.mongo_uri = mongo_uri
         self.db_name = db_name
@@ -23,229 +23,142 @@ class MongoHandler:
         self._client: Optional[motor.motor_asyncio.AsyncIOMotorClient] = None
         self._db: Optional[motor.motor_asyncio.AsyncIOMotorDatabase] = None
         self._collection: Optional[motor.motor_asyncio.AsyncIOMotorCollection] = None
+        self._lock = asyncio.Lock() # Lock for serializing client creation
 
     async def connect(self):
-        """Establishes an asynchronous connection to MongoDB."""
+        """
+        Establishes an asynchronous connection to MongoDB if not already established.
+        This method is idempotent.
+        """
         if self._client is None:
-            try:
+            async with self._lock: # Ensure only one coroutine attempts to initialize the client
-                self._client = motor.motor_asyncio.AsyncIOMotorClient(self.mongo_uri)
+                if self._client is None: # Double-check after acquiring lock
-                # The ismaster command is cheap and does not require auth.
+                    try:
-                # It is used to confirm that the client can connect to the deployment.
+                        print(f"Initializing MongoDB client for: DB '{self.db_name}', Collection '{self.collection_name}' URI: {self.mongo_uri.split('@')[-1]}") # Avoid logging credentials
-                await self._client.admin.command('ismaster')
+                        self._client = motor.motor_asyncio.AsyncIOMotorClient(self.mongo_uri)
+                        # The ismaster command is cheap and does not require auth.
+                        await self._client.admin.command('ismaster')
+                        self._db = self._client[self.db_name]
+                        self._collection = self._db[self.collection_name]
+                        print(f"MongoDB client initialized and connected: Database '{self.db_name}', Collection '{self.collection_name}'")
+                    except Exception as e:
+                        print(f"Failed to initialize MongoDB client at {self.mongo_uri.split('@')[-1]} for {self.db_name}/{self.collection_name}: {e}")
+                        self._client = None # Ensure client is None if connection fails
+                        self._db = None
+                        self._collection = None
+                        raise # Re-raise the exception after printing
+        
+        if self._collection is None and self._client is not None:
+             # This can happen if connect was called, client was set, but then an error occurred before collection was set
+             # Or if connect logic needs to re-establish db/collection objects without re-creating client (less common with motor)
+            if self._db is None:
                 self._db = self._client[self.db_name]
-                self._collection = self._db[self.collection_name]
+            self._collection = self._db[self.collection_name]
-                print(f"Connected to MongoDB: Database '{self.db_name}', Collection '{self.collection_name}'")
+            if self._collection is None:
-            except Exception as e:
+                 raise RuntimeError(f"MongoDB collection '{self.collection_name}' could not be established even though client exists.")
-                print(f"Failed to connect to MongoDB at {self.mongo_uri}: {e}")
-                self._client = None # Ensure client is None if connection fails
-                raise # Re-raise the exception after printing
 
-    async def close(self):
+
-        """Closes the MongoDB connection."""
+    async def close_client(self):
-        if self._client:
+        """Closes the MongoDB client connection. Should be called on application shutdown."""
-            self._client.close()
+        async with self._lock:
-            self._client = None
+            if self._client:
-            self._db = None
+                print(f"Closing MongoDB client for: Database '{self.db_name}', Collection '{self.collection_name}'")
-            self._collection = None
+                self._client.close()
-            print("MongoDB connection closed.")
+                self._client = None
+                self._db = None
+                self._collection = None
+                print(f"MongoDB client closed for: Database '{self.db_name}', Collection '{self.collection_name}'.")
 
     async def __aenter__(self):
-        """Allows using the handler with async with."""
+        """Allows using the handler with async with. Ensures connection is active."""
         await self.connect()
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        """Ensures the connection is closed when exiting async with."""
+        """Ensures the connection is NOT closed when exiting async with."""
-        await self.close()
+        # The connection is managed and closed at the application level by calling close_client()
+        pass
 
     async def insert_one(self, document: Dict[str, Any]) -> Any:
-        """
-        Inserts a single document into the collection.
-
-        Args:
-            document (Dict[str, Any]): The document to insert.
-
-        Returns:
-            Any: The result of the insert operation (InsertOneResult).
-        """
         if self._collection is None:
-            raise RuntimeError("MongoDB connection not established. Call connect() first or use async with.")
+            await self.connect() # Try to connect if collection is None
+            if self._collection is None: # Check again after attempting to connect
+                 raise RuntimeError("MongoDB collection not available. Call connect() or use async with.")
         print(f"Inserting document into '{self.collection_name}'...")
         result = await self._collection.insert_one(document)
         print(f"Inserted document with ID: {result.inserted_id}")
         return result
 
     async def find_one(self, query: Dict[str, Any]) -> Optional[Dict[str, Any]]:
-        """
-        Finds a single document matching the query.
-
-        Args:
-            query (Dict[str, Any]): The query document.
-
-        Returns:
-            Optional[Dict[str, Any]]: The found document, or None if not found.
-        """
         if self._collection is None:
-            raise RuntimeError("MongoDB connection not established. Call connect() first or use async with.")
+            await self.connect()
+            if self._collection is None:
+                 raise RuntimeError("MongoDB collection not available. Call connect() or use async with.")
         print(f"Finding one document in '{self.collection_name}' with query: {query}")
         document = await self._collection.find_one(query)
         return document
 
     async def find(self, query: Dict[str, Any] = None) -> List[Dict[str, Any]]:
-        """
-        Finds multiple documents matching the query.
-
-        Args:
-            query (Dict[str, Any], optional): The query document. Defaults to None (find all).
-
-        Returns:
-            List[Dict[str, Any]]: A list of matching documents.
-        """
         if self._collection is None:
-            raise RuntimeError("MongoDB connection not established. Call connect() first or use async with.")
+            await self.connect()
+            if self._collection is None:
+                 raise RuntimeError("MongoDB collection not available. Call connect() or use async with.")
         if query is None:
             query = {}
         print(f"Finding documents in '{self.collection_name}' with query: {query}")
-        # Use list comprehension to iterate through the cursor asynchronously
         documents = [doc async for doc in self._collection.find(query)]
         print(f"Found {len(documents)} documents.")
         return documents
 
     async def update_one(self, query: Dict[str, Any], update: Dict[str, Any], upsert: bool = False) -> Any:
-        """
-        Updates a single document matching the query.
-
-        Args:
-            query (Dict[str, Any]): The query document.
-            update (Dict[str, Any]): The update operations to apply.
-            upsert (bool): If True, insert a new document if no match is found.
-
-        Returns:
-            Any: The result of the update operation (UpdateResult).
-        """
         if self._collection is None:
-            raise RuntimeError("MongoDB connection not established. Call connect() first or use async with.")
+            await self.connect()
+            if self._collection is None:
+                 raise RuntimeError("MongoDB collection not available. Call connect() or use async with.")
         print(f"Updating one document in '{self.collection_name}' with query: {query}, update: {update}")
         result = await self._collection.update_one(query, update, upsert=upsert)
         print(f"Matched {result.matched_count}, Modified {result.modified_count}, Upserted ID: {result.upserted_id}")
         return result
 
     async def delete_one(self, query: Dict[str, Any]) -> Any:
-        """
-        Deletes a single document matching the query.
-
-        Args:
-            query (Dict[str, Any]): The query document.
-
-        Returns:
-            Any: The result of the delete operation (DeleteResult).
-        """
         if self._collection is None:
-            raise RuntimeError("MongoDB connection not established. Call connect() first or use async with.")
+            await self.connect()
+            if self._collection is None:
+                 raise RuntimeError("MongoDB collection not available. Call connect() or use async with.")
         print(f"Deleting one document from '{self.collection_name}' with query: {query}")
         result = await self._collection.delete_one(query)
         print(f"Deleted count: {result.deleted_count}")
         return result
 
     async def delete_many(self, query: Dict[str, Any]) -> Any:
-        """
-        Deletes multiple documents matching the query.
-
-        Args:
-            query (Dict[str, Any]): The query document.
-
-        Returns:
-            Any: The result of the delete operation (DeleteResult).
-        """
         if self._collection is None:
-            raise RuntimeError("MongoDB connection not established. Call connect() first or use async with.")
+            await self.connect()
+            if self._collection is None:
+                 raise RuntimeError("MongoDB collection not available. Call connect() or use async with.")
         print(f"Deleting many documents from '{self.collection_name}' with query: {query}")
         result = await self._collection.delete_many(query)
         print(f"Deleted count: {result.deleted_count}")
         return result
 
-
+# --- Example Usage (no change needed here, but behavior of MongoHandler is different) ---
-# --- Example Usage ---
 async def example_mongo_usage():
-    """Demonstrates how to use the MongoHandler."""
-    # Ensure you have a MongoDB server running, default is localhost:27017
     db_name = "radio_app_db"
-    collection_name = "channels"
+    collection_name = "channels_example" # Use a different collection for example
+    handler = MongoHandler(db_name, collection_name) # MONGO_URL defaults to localhost
 
-    # Using async with ensures the connection is closed automatically
+    try:
-    async with MongoHandler(db_name, collection_name) as mongo:
+        async with handler: # Connects on enter (if not already connected)
-        # --- Insert Example ---
+            print("\n--- Inserting a document ---")
-        print("\n--- Inserting a document ---")
+            # ... (rest of example usage)
-        channel_data = {
+            channel_data = { "_id": "example_channel", "name": "Example" }
-            "_id": "channel_3", # You can specify _id or let MongoDB generate one
+            await handler.insert_one(channel_data)
-            "name": "Emergency Services",
+            found = await handler.find_one({"_id": "example_channel"})
-            "frequencies": 453000,
+            print(f"Found: {found}")
-            "location": "Countywide",
+            await handler.delete_one({"_id": "example_channel"})
-            "avail_on_nodes": ["client-xyz987"],
+            print("Example completed.")
-            "description": "Monitor for emergency broadcasts."
-        }
-        try:
-            insert_result = await mongo.insert_one(channel_data)
-            print(f"Insert successful: {insert_result.inserted_id}")
-        except Exception as e:
-            print(f"Insert failed: {e}")
 
+    finally:
+        await handler.close_client() # Explicitly close client at the end of usage
 
-        # --- Find One Example ---
-        print("\n--- Finding one document ---")
-        query = {"_id": "channel_3"}
-        found_channel = await mongo.find_one(query)
-        if found_channel:
-            print("Found document:", found_channel)
-        else:
-            print("Document not found.")
-
-        # --- Find Many Example ---
-        print("\n--- Finding all documents ---")
-        all_channels = await mongo.find() # Empty query finds all
-        print("All documents:", all_channels)
-
-        # --- Update Example ---
-        print("\n--- Updating a document ---")
-        update_query = {"_id": "channel_3"}
-        update_data = {"$set": {"location": "Statewide", "avail_on_nodes": ["client-xyz987", "client-newnode1"]}}
-        update_result = await mongo.update_one(update_query, update_data)
-        print(f"Update successful: Matched {update_result.matched_count}, Modified {update_result.modified_count}")
-
-        print("\n--- Finding the updated document ---")
-        updated_channel = await mongo.find_one(update_query)
-        print("Updated document:", updated_channel)
-
-        # --- Delete Example ---
-        print("\n--- Deleting a document ---")
-        delete_query = {"_id": "channel_3"}
-        delete_result = await mongo.delete_one(delete_query)
-        print(f"Delete successful: Deleted count {delete_result.deleted_count}")
-
-        print("\n--- Verifying deletion ---")
-        deleted_channel = await mongo.find_one(delete_query)
-        if deleted_channel:
-            print("Document still found (deletion failed).")
-        else:
-            print("Document successfully deleted.")
-
-        # --- Insert another for delete_many example ---
-        temp_doc1 = {"_id": "temp_1", "tag": "temp"}
-        temp_doc2 = {"_id": "temp_2", "tag": "temp"}
-        await mongo.insert_one(temp_doc1)
-        await mongo.insert_one(temp_doc2)
-
-        # --- Delete Many Example ---
-        print("\n--- Deleting many documents ---")
-        delete_many_query = {"tag": "temp"}
-        delete_many_result = await mongo.delete_many(delete_many_query)
-        print(f"Delete many successful: Deleted count {delete_many_result.deleted_count}")
-
-
-# To run the example usage:
-# 1. Ensure you have a MongoDB server running locally on the default port (27017).
-# 2. Save the code as mongodb_handler.py.
-# 3. Run from your terminal: python -m asyncio mongodb_handler.py
 if __name__ == "__main__":
-    # Running the example directly requires running within an asyncio loop
+    asyncio.run(example_mongo_usage())
-    asyncio.run(example_mongo_usage())

app/internal/db_wrappers.py

@@ -3,7 +3,7 @@ import asyncio
 from uuid import uuid4
 from typing import Optional, List, Dict, Any
 from enum import Enum
-from internal.db_handler import MongoHandler
+from internal.db_handler import MongoHandler #
 from internal.types import System, DiscordId
 
 # Init vars
@@ -16,70 +16,47 @@ DISCORD_ID_DB_COLLECTION_NAME = "discord_bot_ids"
 # --- System class ---
 class SystemDbController():
     def __init__(self):
-        # Init the handler
+        self.db_h = MongoHandler(DB_NAME, SYSTEM_DB_COLLECTION_NAME, MONGO_URL) #
-        self.db_h = MongoHandler(DB_NAME, SYSTEM_DB_COLLECTION_NAME, MONGO_URL)
+
+    async def close_db_connection(self):
+        """Closes the underlying MongoDB connection."""
+        if self.db_h:
+            await self.db_h.close_client() #
 
     async def create_system(self, system_data: Dict[str, Any]) -> Optional[System]:
-        """
-        Creates a new system entry in the database.
-
-        Args:
-            system_data: A dictionary containing the data for the new system.
-
-        Returns:
-            The created System object if successful, None otherwise.
-        """
         print("\n--- Creating a document ---")
         try:
-            # Check if the data to be inserted has an ID
             if not system_data.get("_id"):
-                system_data['_id'] = uuid4()
+                system_data['_id'] = str(uuid4())
 
-            inserted_result = None
             inserted_id = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                insert_result = await self.db_h.insert_one(system_data)
+                insert_result = await db.insert_one(system_data) #
                 inserted_id = insert_result.inserted_id
 
             if inserted_id:
                 print(f"Insert successful with ID: {inserted_id}")
-                # Fetch the inserted document to get the complete data including the generated _id
                 query = {"_id": inserted_id}
-
                 inserted_doc = None
-                async with self.db_h as db:
+                async with self.db_h as db: #
-                    inserted_doc = await db.find_one(query)
+                    inserted_doc = await db.find_one(query) #
-
                 if inserted_doc:
-                    # Convert the fetched dictionary back to a System object
                     return System.from_dict(inserted_doc)
             else:
                 print("Insert acknowledged but no ID returned.")
                 return None
-
         except Exception as e:
             print(f"Create failed: {e}")
             return None
 
     async def find_system(self, query: Dict[str, Any]) -> Optional[System]:
-        """
-        Finds a single system entry in the database.
-
-        Args:
-            query: A dictionary representing the query criteria.
-
-        Returns:
-            A System object if found, None otherwise.
-        """
         print("\n--- Finding one document ---")
         try:
             found_doc = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                found_doc = await db.find_one(query)
+                found_doc = await db.find_one(query) #
-
             if found_doc:
                 print("Found document (raw dict):", found_doc)
-                # Convert the dictionary result to a System object
                 return System.from_dict(found_doc)
             else:
                 print("Document not found.")
@@ -89,30 +66,15 @@ class SystemDbController():
             return None
 
     async def find_systems(self, query: Dict[str, Any]) -> Optional[List[System]]:
-        """
-        Finds one or more system entries in the database.
-
-        Args:
-            query: A dictionary representing the query criteria.
-
-        Returns:
-            A list of System object(s) if found, None otherwise.
-        """
         print("\n--- Finding documents ---")
         try:
             found_docs = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                found_docs = await db.find(query)
+                found_docs = await db.find(query) #
-
             if found_docs:
                 print("Found document (raw dict):", found_docs)
-                # Convert the dictionary results to a System object
+                converted_systems = [System.from_dict(doc) for doc in found_docs]
-                converted_systems = []
+                print("Found document (converted):", converted_systems)
-                for doc in found_docs:
-                    converted_systems.append(System.from_dict(doc))
-
-                print("YURB", found_docs, converted_systems)
-                
                 return converted_systems if len(converted_systems) > 0 else None
             else:
                 print("Document not found.")
@@ -121,25 +83,19 @@ class SystemDbController():
             print(f"Find failed: {e}")
             return None
 
-    async def find_all_systems(self, query: Dict[str, Any] = {}) -> List[System]:
+    async def find_all_systems(self, query: Optional[Dict[str, Any]] = None) -> List[System]:
-        """
-        Finds multiple system entries in the database.
-
-        Args:
-            query: A dictionary representing the query criteria (default is empty to find all).
-
-        Returns:
-            A list of System objects.
-        """
         print("\n--- Finding multiple documents ---")
         try:
+            # Initialize an empty dictionary if no query is provided
+            if query is None:
+                query = {}
+
             found_docs = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                found_docs = await db.find(query)
+                found_docs = await db.find(query) #
 
             if found_docs:
                 print(f"Found {len(found_docs)} documents (raw dicts).")
-                # Convert the list of dictionaries to a list of System objects
                 return [System.from_dict(doc) for doc in found_docs]
             else:
                 print("No documents found.")
@@ -149,22 +105,11 @@ class SystemDbController():
             return []
 
     async def update_system(self, query: Dict[str, Any], update_data: Dict[str, Any]) -> Optional[int]:
-        """
-        Updates a single system entry in the database.
-
-        Args:
-            query: A dictionary representing the query criteria to find the document.
-            update_data: A dictionary representing the update operations (e.g., using $set).
-
-        Returns:
-            The number of modified documents if successful, None otherwise.
-        """
         print("\n--- Updating a document ---")
         try:
             update_result = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                update_result = await db.update_one(query, update_data)
+                update_result = await db.update_one(query, update_data) #
-
             print(f"Update result: Matched {update_result.matched_count}, Modified {update_result.modified_count}")
             return update_result.modified_count
         except Exception as e:
@@ -172,19 +117,11 @@ class SystemDbController():
             return None
 
     async def delete_system(self, query: Dict[str, Any]) -> Optional[int]:
-        """
-        Deletes a single system entry from the database.
-        Args:
-            query: A dictionary representing the query criteria to find the document to delete.
-        Returns:
-            The number of deleted documents if successful, None otherwise.
-        """
         print("\n--- Deleting a document ---")
         try:
             delete_result = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                delete_result = await self.db_h.delete_one(query)
+                delete_result = await db.delete_one(query) #
-
             print(f"Delete result: Deleted count {delete_result.deleted_count}")
             return delete_result.deleted_count
         except Exception as e:
@@ -195,66 +132,45 @@ class SystemDbController():
 # --- DiscordIdDbController class ---
 class DiscordIdDbController():
     def __init__(self):
-        # Init the handler for Discord IDs
+        self.db_h = MongoHandler(DB_NAME, DISCORD_ID_DB_COLLECTION_NAME, MONGO_URL) #
-        self.db_h = MongoHandler(DB_NAME, DISCORD_ID_DB_COLLECTION_NAME, MONGO_URL)
+
+    async def close_db_connection(self):
+        """Closes the underlying MongoDB connection."""
+        if self.db_h:
+            await self.db_h.close_client() #
 
     async def create_discord_id(self, discord_id_data: Dict[str, Any]) -> Optional[DiscordId]:
-        """
-        Creates a new Discord ID entry in the database.
-
-        Args:
-            discord_id_data: A dictionary containing the data for the new Discord ID.
-
-        Returns:
-            The created DiscordId object if successful, None otherwise.
-        """
         print("\n--- Creating a Discord ID document ---")
         try:
             if not discord_id_data.get("_id"):
-                discord_id_data['_id'] = str(uuid4()) # Ensure _id is a string
+                discord_id_data['_id'] = str(uuid4())
 
             inserted_id = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                insert_result = await self.db_h.insert_one(discord_id_data)
+                insert_result = await db.insert_one(discord_id_data) #
                 inserted_id = insert_result.inserted_id
 
             if inserted_id:
                 print(f"Discord ID insert successful with ID: {inserted_id}")
                 query = {"_id": inserted_id}
                 inserted_doc = None
-                async with self.db_h as db:
+                async with self.db_h as db: #
-                    inserted_doc = await db.find_one(query)
+                    inserted_doc = await db.find_one(query) #
-
                 if inserted_doc:
                     return DiscordId.from_dict(inserted_doc)
             else:
                 print("Discord ID insert acknowledged but no ID returned.")
                 return None
-
         except Exception as e:
             print(f"Discord ID create failed: {e}")
             return None
 
-    async def find_discord_id(self, query: Dict[str, Any], active_only: bool = False) -> Optional[DiscordId]:
+    async def find_discord_id(self, query: Dict[str, Any]) -> Optional[DiscordId]:
-        """
-        Finds a single Discord ID entry in the database.
-
-        Args:
-            query: A dictionary representing the query criteria.
-            active_only: If True, only returns active Discord IDs.
-
-        Returns:
-            A DiscordId object if found, None otherwise.
-        """
         print("\n--- Finding one Discord ID document ---")
         try:
-            if active_only:
-                query["active"] = True
-
             found_doc = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                found_doc = await db.find_one(query)
+                found_doc = await db.find_one(query) #
-
             if found_doc:
                 print("Found Discord ID document (raw dict):", found_doc)
                 return DiscordId.from_dict(found_doc)
@@ -265,38 +181,25 @@ class DiscordIdDbController():
             print(f"Discord ID find failed: {e}")
             return None
 
-    async def find_discord_ids(self, query: Dict[str, Any] = {}, guild_id: Optional[str] = None, active_only: bool = False) -> Optional[List[DiscordId]]:
+    async def find_discord_ids(self, query: Optional[Dict[str, Any]] = None, guild_id: Optional[str] = None, active_only: bool = False) -> Optional[List[DiscordId]]:
-        """
-        Finds one or more Discord ID entries in the database.
-
-        Args:
-            query: A dictionary representing the query criteria.
-            guild_id: Optional. If provided, filters Discord IDs that belong to this guild.
-            active_only: If True, only returns active Discord IDs.
-
-        Returns:
-            A list of DiscordId object(s) if found, None otherwise.
-        """
         print("\n--- Finding multiple Discord ID documents ---")
         try:
-            # Add active filter if requested
+            if query is None:
-            if active_only:
+                query = {}
+
+            if active_only == True:
+                print("Searching active IDs")
                 query["active"] = True
-
-            # Add guild_id filter if provided
             if guild_id:
+                print(f"Searching for IDs in {guild_id}")
                 query["guild_ids"] = {"$in": [guild_id]}
-
             found_docs = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                found_docs = await db.find(query)
+                print(f"Query: {query}")
-
+                found_docs = await db.find(query) #
             if found_docs:
                 print(f"Found {len(found_docs)} Discord ID documents (raw dicts).")
-                converted_discord_ids = []
+                converted_discord_ids = [DiscordId.from_dict(doc) for doc in found_docs]
-                for doc in found_docs:
-                    converted_discord_ids.append(DiscordId.from_dict(doc))
-
                 return converted_discord_ids if len(converted_discord_ids) > 0 else None
             else:
                 print("Discord ID documents not found.")
@@ -306,22 +209,11 @@ class DiscordIdDbController():
             return None
 
     async def update_discord_id(self, query: Dict[str, Any], update_data: Dict[str, Any]) -> Optional[int]:
-        """
-        Updates a single Discord ID entry in the database.
-
-        Args:
-            query: A dictionary representing the query criteria to find the document.
-            update_data: A dictionary representing the update operations (e.g., using $set).
-
-        Returns:
-            The number of modified documents if successful, None otherwise.
-        """
         print("\n--- Updating a Discord ID document ---")
         try:
             update_result = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                update_result = await db.update_one(query, update_data)
+                update_result = await db.update_one(query, update_data) #
-
             print(f"Discord ID update result: Matched {update_result.matched_count}, Modified {update_result.modified_count}")
             return update_result.modified_count
         except Exception as e:
@@ -329,19 +221,11 @@ class DiscordIdDbController():
             return None
 
     async def delete_discord_id(self, query: Dict[str, Any]) -> Optional[int]:
-        """
-        Deletes a single Discord ID entry from the database.
-        Args:
-            query: A dictionary representing the query criteria to find the document to delete.
-        Returns:
-            The number of deleted documents if successful, None otherwise.
-        """
         print("\n--- Deleting a Discord ID document ---")
         try:
             delete_result = None
-            async with self.db_h as db:
+            async with self.db_h as db: #
-                delete_result = await self.db_h.delete_one(query)
+                delete_result = await db.delete_one(query) #
-
             print(f"Discord ID delete result: Deleted count {delete_result.deleted_count}")
             return delete_result.deleted_count
         except Exception as e:

app/internal/types.py

@@ -1,3 +1,4 @@
+# internal/types.py
 from typing import Optional, List, Dict, Any
 from enum import Enum
 
@@ -10,6 +11,10 @@ class DemodTypes(str, Enum):
 class NodeCommands(str, Enum):
     JOIN = "join_server"
     LEAVE = "leave_server"
+    STATUS = "get_status"
+    OP25_START = "op25_start"
+    OP25_STOP = "op25_stop"
+    OP25_SET = "op25_set"
 
 
 class TalkgroupTag:
@@ -20,7 +25,7 @@ class TalkgroupTag:
 
     # Add a method to convert to a dictionary, useful for sending as JSON
     def to_dict(self) -> Dict[str, Any]:
-        return {"talkgroup": self.talkgroup, "tagDec": self.tagDec}
+        return {"talkgroup": self.talkgroup, "tagDec": self.talkgroup}
 
 
 class DiscordId:
@@ -33,7 +38,8 @@ class DiscordId:
                  name: str,
                  token: str,
                  active: bool,
-                 guild_ids: List[str]):
+                 guild_ids: List[str],
+                 fixed_node: Optional[str]=None):
         """
         Initializes a DiscordId object.
 
@@ -44,6 +50,7 @@ class DiscordId:
             token: The authentication token.
             active: Boolean indicating if the ID is active.
             guild_ids: A list of guild IDs the Discord user is part of.
+            fixed_node: The node ID this DiscordId must use.
         """
         self._id: str = str(_id)
         self.discord_id: str = discord_id
@@ -51,6 +58,7 @@ class DiscordId:
         self.token: str = token
         self.active: bool = active
         self.guild_ids: List[str] = guild_ids
+        self.fixed_node: Optional[str] = fixed_node
 
     def __repr__(self) -> str:
         """
@@ -70,6 +78,7 @@ class DiscordId:
             "token": self.token,
             "active": self.active,
             "guild_ids": self.guild_ids,
+            "fixed_node": self.fixed_node,
         }
 
     @classmethod
@@ -84,6 +93,7 @@ class DiscordId:
             token=data.get("token", ""),
             active=data.get("active", False), # Default to False if not present
             guild_ids=data.get("guild_ids", []), # Default to empty list if not present
+            fixed_node=data.get("fixed_node", None), # Default to empty if not present
         )
 
 
@@ -182,11 +192,80 @@ class System:
 
 class ActiveClient:
     """
-        The active client model in memory for quicker access
+    The active client model in memory for quicker access
     """
+    client_id: str = None
     websocket = None
     active_token: DiscordId = None
+    nickname: str = None
+    access_token: str = None
 
-    def __init__(self, websocket= None, active_token:DiscordId=None):
+    def __init__(self, client_id: str=None, websocket=None, active_token: DiscordId = None, nickname: str = None, access_token: str = None):
+        self.client_id = client_id
         self.active_token = active_token
-        self.websocket = websocket
+        self.websocket = websocket
+        self.nickname = nickname
+        self.access_token = access_token
+
+    def to_dict(self):
+        return {
+            "client_id": self.client_id,
+            "nickname": self.nickname,
+            "active_token": self.active_token.to_dict() if self.active_token else None
+        }
+
+    def __str__(self):
+        """
+        Returns a neatly formatted string representation of the ActiveClient object.
+        """
+        return (f"--- Active Client ---\n"
+                f"Active Token: {self.active_token if self.active_token else 'N/A'}\n"
+                f"Nickname: {self.nickname if self.nickname else 'N/A'}\n"
+                f"Nickname: {self.client_id if self.client_id else 'N/A'}\n"
+                f"Access Token: {'[REDACTED]' if self.access_token else 'N/A'}\n"
+                f"Websocket Connected: {'Yes' if self.websocket else 'No'}")
+
+class UserRoles(str, Enum):
+    ADMIN = "admin"
+    MOD = "mod"
+    USER = "user"
+
+
+class User:
+    """
+    A data model for a User entry.
+    """
+    def __init__(self,
+                 _id: str,
+                 username: str,
+                 password_hash: str,
+                 role: UserRoles,
+                 api_key: Optional[str] = None):
+        self._id: str = str(_id)
+        self.username: str = username
+        self.password_hash: str = password_hash
+        self.role: UserRoles = role
+        self.api_key: Optional[str] = api_key
+
+    def __repr__(self) -> str:
+        return (f"User(_id='{self._id}', username='{self.username}', role='{self.role.value}', "
+                f"api_key={'<hidden>' if self.api_key else 'None'})")
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "_id": self._id,
+            "username": self.username,
+            "password_hash": self.password_hash,
+            "role": self.role.value,
+            "api_key": self.api_key,
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "User":
+        return cls(
+            _id=data.get("_id"),
+            username=data.get("username", ""),
+            password_hash=data.get("password_hash", ""),
+            role=UserRoles(data.get("role", UserRoles.USER.value)),
+            api_key=data.get("api_key", None),
+        )

app/routers/auth.py

@@ -0,0 +1,141 @@
+import functools
+from quart import Blueprint, jsonify, request, current_app, abort
+from werkzeug.security import generate_password_hash, check_password_hash
+from quart_jwt_extended import create_access_token, jwt_required, get_jwt_identity
+from internal.auth_wrappers import UserDbController
+from internal.types import UserRoles
+from uuid import uuid4
+from datetime import timedelta
+
+# Import the centralized JWTManager instance
+from config.jwt_config import jwt as jwt_manager_instance # Renamed to avoid confusion with jwt_required
+
+
+auth_bp = Blueprint('auth', __name__)
+
+# Decorator for role-based access control
+def role_required(required_role: UserRoles):
+    def wrapper(fn):
+        @functools.wraps(fn)
+        @jwt_required
+        async def decorated_view(*args, **kwargs):
+            current_user_identity = get_jwt_identity()
+            user_id = current_user_identity['id']
+
+            auth_type = current_user_identity['type']
+
+            if auth_type == "node":
+                node = current_app.active_clients.get(user_id)
+                print("Node", node)
+
+                if not node:
+                    abort(401, "Node not found or invalid token.")
+
+            if auth_type == "user":
+                # Make a DB call to get the user and their role
+                user = await current_app.user_db_h.find_user({"_id": user_id})
+
+                if not user:
+                    abort(401, "User not found or invalid token.") # User corresponding to token not found
+
+                user_role = user.role # Get the role from the fetched user object
+
+                role_order = {UserRoles.USER: 0, UserRoles.MOD: 1, UserRoles.ADMIN: 2}
+
+                if role_order[user_role] < role_order[required_role]:
+                    abort(403, "Permission denied: Insufficient role.")
+
+            return await fn(*args, **kwargs) # Directly await the original async function
+
+        return decorated_view
+    return wrapper
+
+@auth_bp.route('/register', methods=['POST'])
+async def register_user():
+    data = await request.get_json()
+    username = data.get('username')
+    password = data.get('password')
+    role = data.get('role', UserRoles.USER.value) # Default to 'user' role
+
+    if not username or not password:
+        abort(400, "Username and password are required")
+
+    existing_user = await current_app.user_db_h.find_user({"username": username})
+    if existing_user:
+        abort(409, "Username already exists")
+
+    hashed_password = generate_password_hash(password)
+
+    try:
+        user_role = UserRoles(role)
+    except ValueError:
+        abort(400, f"Invalid role: {role}. Must be one of {list(UserRoles)}")
+
+    user_data = {
+        "username": username,
+        "password_hash": hashed_password,
+        "role": user_role.value
+    }
+
+    new_user = await current_app.user_db_h.create_user(user_data)
+
+    if new_user:
+        return jsonify({"message": "User registered successfully", "user_id": new_user._id}), 201
+    else:
+        abort(500, "Failed to register user")
+
+@auth_bp.route('/login', methods=['POST'])
+async def login_user():
+    data = await request.get_json()
+    username = data.get('username')
+    password = data.get('password')
+
+    if not username or not password:
+        abort(400, "Username and password are required")
+
+    user = await current_app.user_db_h.find_user({"username": username})
+
+    if not user or not check_password_hash(user.password_hash, password):
+        abort(401, "Invalid credentials")
+
+    access_token = create_access_token(identity={"id": user._id, "username": user.username, "type": "user"}, expires_delta=timedelta(hours=1))
+    return jsonify({"access_token": access_token, "role": user.role, "username": user.username, "user_id": user._id }), 200
+
+# DEPRECATED
+@auth_bp.route('/generate_api_key', methods=['POST'])
+@jwt_required
+async def generate_api_key():
+    current_user_identity = get_jwt_identity()
+    user_id = current_user_identity['id']
+    user_role = current_user_identity['role']
+
+    if user_role not in [UserRoles.ADMIN.value, UserRoles.MOD.value]:
+        target_user_id = (await request.get_json()).get('user_id', user_id)
+        if target_user_id != user_id:
+            abort(403, "Permission denied: You can only generate an API key for your own account.")
+    else:
+        target_user_id = (await request.get_json()).get('user_id', user_id)
+
+
+    new_api_key = str(uuid4())
+    updated_count = await current_app.user_db_h.update_user(
+        {"_id": target_user_id},
+        {"$set": {"api_key": new_api_key}}
+    )
+
+    if updated_count:
+        return jsonify({"message": f"API key generated for user {target_user_id}", "api_key": new_api_key}), 200
+    else:
+        abort(404, f"User {target_user_id} not found or unable to update API key.")
+
+@auth_bp.route('/admin_only', methods=['GET'])
+@jwt_required
+@role_required(UserRoles.ADMIN)
+async def admin_only_route():
+    return jsonify({"message": "Welcome, Admin!"}), 200
+
+@auth_bp.route('/mod_or_admin_only', methods=['GET'])
+@jwt_required
+@role_required(UserRoles.MOD)
+async def mod_or_admin_only_route():
+    return jsonify({"message": "Welcome, Mod or Admin!"}), 200

app/routers/bot.py

@@ -1,19 +1,19 @@
 import json
 from quart import Blueprint, jsonify, request, abort, current_app
 from werkzeug.exceptions import HTTPException
-from internal.types import ActiveClient
+from internal.types import ActiveClient, DiscordId, UserRoles
 from internal.db_wrappers import DiscordIdDbController
+from quart_jwt_extended import jwt_required
+from routers.auth import role_required
+from bson.objectid import ObjectId
 
 bot_bp = Blueprint('bot', __name__)
 
 
-@bot_bp.route("/", methods=['GET'])
+# ------- Discord Token Functions
-async def get_online_bots_route():
-    """API endpoint to list bots (by name) currently online."""
-    return jsonify(list(current_app.active_clients.keys()))
-
-
 @bot_bp.route('/request_token', methods=['POST'])
+@jwt_required
+@role_required(UserRoles.MOD)
 async def request_token_route():
     """
     API endpoint to request a token for a client.
@@ -43,12 +43,18 @@ async def request_token_route():
         if not avail_ids:
             abort(404, "No available active Discord IDs found.")
 
-        # --- Logic for selecting a preferred ID based on client_id (TODO) ---
+        selected_id = None
+        # Check for any fixed IDs
+        for avail_id in avail_ids:
+            if avail_id.fixed_node and avail_id.fixed_node == client_id:
+                selected_id = avail_id
+        
+        if not selected_id:
+            selected_id = avail_ids[0]
 
-        selected_id = avail_ids[0]
+        print("Selected Discord ID: ", selected_id)
         current_app.active_clients[client_id].active_token = selected_id
-
+        print(current_app.active_clients[client_id])
-        # --- End of logic for selecting a preferred ID ---
 
         return jsonify(selected_id.to_dict())
 
@@ -57,29 +63,148 @@ async def request_token_route():
         abort(500, f"An internal error occurred: {e}")
 
 
+@bot_bp.route('/tokens/', methods=['GET'])
+@jwt_required
+@role_required(UserRoles.USER)
+async def get_all_discord_tokens():
+    """
+    API endpoint to return all discord IDs
+    """
+    try:
+        # get the available IDs
+        d_ids = await current_app.d_id_db_h.find_discord_ids(active_only=False)
+        return jsonify([d_id.to_dict() for d_id in d_ids])
+
+    except Exception as e:
+        print(f"Error in request_token_route: {e}")
+        abort(500, f"An internal error occurred: {e}")
+
+
+@bot_bp.route('/token/<string:discord_id_param>', methods=['GET'])
+@jwt_required
+@role_required(UserRoles.MOD)
+async def get_discord_token_by_id(discord_id_param: str):
+    """
+    API endpoint to get a single Discord ID by its _id.
+    """
+    try:
+        query = {"_id": ObjectId(discord_id_param)}
+        d_id = await current_app.d_id_db_h.find_discord_id(query)
+        if d_id:
+            return jsonify(d_id.to_dict())
+        else:
+            abort(404, "Discord ID not found.")
+    except Exception as e:
+        print(f"Error in get_discord_token_by_id: {e}")
+        abort(500, f"An internal error occurred: {e}")
+
+
+@bot_bp.route('/token', methods=["POST"])
+@jwt_required
+@role_required(UserRoles.MOD)
+async def create_discord_token():
+    """
+    API Endpoint to create a discord token (adding a new bot)
+    """
+    data = await request.get_json()
+
+    if '_id' in data:
+        id_search_result = await current_app.sys_db_h.find_system({"_id": ObjectId(data["_id"])})
+        if id_search_result:
+            # If _id is provided and exists, return conflict
+            abort(409, f"System with ID '{data['_id']}' already exists")
+
+    temp_name = data.get("name")
+    temp_discord_id = data.get("discord_id")
+    temp_token = data.get("token")
+    temp_active = bool(data.get("active"))
+    temp_guilds = data.get("guild_ids")
+
+    # Create the discord ID object
+    temp_d_id = {
+        "discord_id": temp_discord_id, 
+        "name": temp_name, 
+        "token": temp_token, 
+        "active": temp_active, 
+        "guild_ids": temp_guilds
+        }
+
+    new_d_id = await current_app.d_id_db_h.create_discord_id(temp_d_id)
+
+    if new_d_id:
+        return jsonify(new_d_id.to_dict()), 201
+    else:
+        abort(500, "Failed to create Discord ID.")
+
+
+@bot_bp.route('/token/<string:discord_id_param>', methods=['PUT'])
+@jwt_required
+@role_required(UserRoles.MOD)
+async def update_discord_token(discord_id_param: str):
+    """
+    API endpoint to update a Discord ID by its _id.
+    """
+    try:
+        data = await request.get_json()
+        if not data:
+            abort(400, "No update data provided.")
+
+        query = {"_id": ObjectId(discord_id_param)}
+        update_result = await current_app.d_id_db_h.update_discord_id(query, {"$set": data})
+
+        if update_result is not None:
+            if update_result > 0:
+                # Optionally, fetch the updated document to return it
+                updated_d_id = await current_app.d_id_db_h.find_discord_id(query)
+                return jsonify(updated_d_id.to_dict()), 200
+            else:
+                abort(404, "Discord ID not found or no changes applied.")
+        else:
+            abort(500, "Failed to update Discord ID.")
+    except Exception as e:
+        print(f"Error in update_discord_token: {e}")
+        abort(500, f"An internal error occurred: {e}")
+
+
+@bot_bp.route('/token/<string:discord_id_param>', methods=['DELETE'])
+@jwt_required
+@role_required(UserRoles.MOD)
+async def delete_discord_token(discord_id_param: str):
+    """
+    API endpoint to delete a Discord ID by its _id.
+    """
+    try:
+        query = {"_id": ObjectId(discord_id_param)}
+        delete_count = await current_app.d_id_db_h.delete_discord_id(query)
+
+        if delete_count is not None:
+            if delete_count > 0:
+                return jsonify({"message": f"Successfully deleted {delete_count} Discord ID(s)."}), 200
+            else:
+                abort(404, "Discord ID not found.")
+        else:
+            abort(500, "Failed to delete Discord ID.")
+    except Exception as e:
+        print(f"Error in delete_discord_token: {e}")
+        abort(500, f"An internal error occurred: {e}")
+
+
+# ------- Util Functions
+
 def find_token_in_active_clients(target_token: str) -> bool:
     """
-    Checks if a target_token exists in the active_token of any ActiveClient object in a list.
+    Checks if a target_token exists in the active_token of any online ActiveClient object.
 
     Args:
-        clients: A list of ActiveClient objects.
         target_token: The token string to search for.
 
     Returns:
         True if the token is found in any ActiveClient, False otherwise.
     """
     for client_id in current_app.active_clients:
-        if current_app.active_clients[client_id].active_token == target_token:
+        try:
-            return True
+            if current_app.active_clients[client_id].active_token.token == target_token:
-    return False
+                return True
-
+        except Exception as e:
-
+            pass
-
+    return False
-
-
-
-
-
-    #  query_params = dict(request.args)
-
-    #     systems = await current_app.sys_db_h.find_systems(query_params)

app/routers/nodes.py

@@ -1,38 +1,135 @@
 import json
+import asyncio # Import asyncio
+import websockets
+import uuid
 from quart import Blueprint, jsonify, request, abort, current_app
 from werkzeug.exceptions import HTTPException
 from enum import Enum
-from internal.types import ActiveClient, NodeCommands
+from internal.types import ActiveClient, NodeCommands, UserRoles
+from quart_jwt_extended import create_access_token
+from quart_jwt_extended import jwt_required
+from routers.auth import role_required
 
 nodes_bp = Blueprint('nodes', __name__)
 
+# Dictionary to store pending requests: {request_id: asyncio.Future}
+pending_requests = {}
 
-async def register_client(websocket, client_id):
+
+async def register_client(websocket, client_id, client_nickname, active_token):
     """Registers a new client connection."""
-    current_app.active_clients[client_id] = ActiveClient(websocket)
+    current_app.active_clients[client_id] = ActiveClient()
+    current_app.active_clients[client_id].websocket = websocket
+    current_app.active_clients[client_id].nickname = client_nickname
+    current_app.active_clients[client_id].active_token = active_token
+    current_app.active_clients[client_id].client_id = client_id
     print(f"Client {client_id} connected.")
 
+    # Create a JWT for the client
+    current_app.active_clients[client_id].access_token = create_access_token(identity={"id": client_id, "username": client_nickname, "type": "node"}, expires_delta=False)
+
+    print(current_app.active_clients[client_id])
+
+    # Start a task to listen for messages from this client
+    asyncio.create_task(listen_to_client(websocket, client_id))
+
 
 async def unregister_client(client_id):
     """Unregisters a disconnected client."""
     if client_id in current_app.active_clients:
+        # Also clean up any pending requests for this client
+        for req_id, req_obj in list(pending_requests.items()):
+            if req_obj['client_id'] == client_id:
+                if not req_obj['future'].done():
+                    req_obj['future'].cancel()
         del current_app.active_clients[client_id]
         print(f"Client {client_id} disconnected.")
 
 
-async def send_command_to_client(client_id, command_name, *args):
+async def listen_to_client(websocket, client_id):
-    """Sends a command to a specific client."""
+    """Listens for messages from a specific client."""
-    if client_id in current_app.active_clients:
+    try:
-        websocket = current_app.active_clients[client_id].websocket
+        while True:
-        message = json.dumps({"type": "command", "name": command_name, "args": args})
+            message = await websocket.recv()
-        try:
+            data = json.loads(message)
-            await websocket.send(message)
+            message_type = data.get("type")
-            print(f"Sent command '{command_name}' to client {client_id}")
+            request_id = data.get("request_id")
-        except websockets.exceptions.ConnectionClosedError:
+
-            print(f"Failed to send to client {client_id}: connection closed.")
+            if message_type == "response" and request_id in pending_requests:
-            await unregister_client(client_id)
+                # Retrieve the dictionary containing the future and client_id
-    else:
+                request_info = pending_requests.pop(request_id)
+                
+                # Extract the actual asyncio.Future object
+                future = request_info["future"] 
+
+                if not future.done(): # This will now correctly call .done() on the Future
+                    future.set_result(data.get("payload"))
+            # Add other message types handling here if needed (e.g., unsolicited messages)
+
+    except websockets.exceptions.ConnectionClosedError:
+        print(f"Client {client_id} connection closed while listening.")
+    except json.JSONDecodeError:
+        print(f"Received invalid JSON from client {client_id}.")
+    except Exception as e:
+        print(f"Error listening to client {client_id}: {e}")
+    finally:
+        await unregister_client(client_id)
+
+
+async def send_command_to_client(client_id, command_name, *args, wait_for_response=False, timeout=10):
+    """Sends a command to a specific client and optionally waits for a response."""
+    if client_id not in current_app.active_clients:
         print(f"Client {client_id} not found.")
+        raise ValueError(f"Client {client_id} not found.")
+
+    websocket = current_app.active_clients[client_id].websocket
+    request_id = str(uuid.uuid4()) if wait_for_response else None
+
+    message_payload = {"type": "command", "name": command_name, "args": args}
+    if request_id:
+        message_payload["request_id"] = request_id
+
+    message = json.dumps(message_payload)
+
+    future = None
+    if wait_for_response:
+        future = asyncio.Future()
+        pending_requests[request_id] = {
+            "future": future,
+            "client_id": client_id
+        }
+
+    try:
+        await websocket.send(message)
+        print(f"Sent command '{command_name}' to client {client_id} (Request ID: {request_id})")
+
+        if wait_for_response:
+            try:
+                response = await asyncio.wait_for(future, timeout)
+                print(f"Received response for Request ID {request_id} from client {client_id}")
+                return response
+            except asyncio.TimeoutError:
+                print(f"Timeout waiting for response from client {client_id} for Request ID {request_id}")
+                raise TimeoutError("Client response timed out.")
+            except asyncio.CancelledError:
+                print(f"Waiting for response for Request ID {request_id} was cancelled.")
+                raise ConnectionClosedError("Client disconnected before response was received.")
+            finally:
+                # Clean up the future if it's still there
+                pending_requests.pop(request_id, None)
+
+    except websockets.exceptions.ConnectionClosedError:
+        print(f"Failed to send to client {client_id}: connection closed.")
+        # If the connection closed, and we were waiting for a response, mark the future as cancelled
+        if future and not future.done():
+            future.cancel()
+        await unregister_client(client_id)
+        raise
+    except Exception as e:
+        print(f"An error occurred sending command to client {client_id}: {e}")
+        if future and not future.done():
+            future.cancel()
+        raise
 
 
 async def send_command_to_all_clients(command_name, *args):
@@ -40,9 +137,9 @@ async def send_command_to_all_clients(command_name, *args):
     message = json.dumps({"type": "command", "name": command_name, "args": args})
     # Use a list of items to avoid issues if clients disconnect during iteration
     clients_to_send = list(current_app.active_clients.items())
-    for client_id, websocket in clients_to_send:
+    for client_id, active_client in clients_to_send:
         try:
-            await websocket.send(message)
+            await active_client.websocket.send(message)
             print(f"Sent command '{command_name}' to client {client_id}")
         except websockets.exceptions.ConnectionClosedError:
             print(f"Failed to send to client {client_id}: connection closed.")
@@ -50,20 +147,61 @@ async def send_command_to_all_clients(command_name, *args):
 
 
 @nodes_bp.route("/", methods=['GET'])
+@jwt_required
+@role_required(UserRoles.USER)
 async def get_nodes():
     """API endpoint to list currently connected client IDs."""
-    return jsonify(list(current_app.active_clients.keys()))
+    return jsonify([current_app.active_clients[client_id].to_dict() for client_id in current_app.active_clients])
 
 
-@nodes_bp.route("/join", methods=['POST'])
+@nodes_bp.route("/online", methods=['GET'])
-async def join():
+@jwt_required
+@role_required(UserRoles.USER)
+async def get_online_bots():
+    active_bots = []
+    for client_id, active_client in current_app.active_clients.items():
+        if active_client.active_token:
+            active_bots.append({client_id: active_client.active_token.to_dict()})
+    return jsonify(active_bots)
+
+
+@nodes_bp.route("/<client_id>/status", methods=["GET"])
+@jwt_required
+@role_required(UserRoles.USER)
+async def status(client_id):
+    """
+    Get the status from a given client
+    """
+    # Check to make sure the client is online
+    if client_id not in current_app.active_clients:
+        return jsonify({"error": f"Client {client_id} not found, it might be offline"}), 404
+
+    try:
+        # Send the command and wait for a response
+        status_data = await send_command_to_client(client_id, NodeCommands.STATUS, wait_for_response=True, timeout=5)
+        return jsonify({
+            "active_client": current_app.active_clients[client_id].to_dict(), 
+            "status": status_data['status']
+        }), 200
+
+    except TimeoutError:
+        return jsonify({"error": f"Client {client_id} did not respond within the timeout period."}), 504
+    except ConnectionClosedError:
+        return jsonify({"error": f"Client {client_id} disconnected before status could be retrieved."}), 503
+    except Exception as e:
+        return jsonify({"error": f"Failed to get status from client: {e}"}), 500
+
+
+@nodes_bp.route("/<client_id>/join", methods=['POST'])
+@jwt_required
+@role_required(UserRoles.MOD)
+async def join(client_id):
     """
     Send a join command to the specific system specified
     """
     data = await request.get_json()
 
-    client_id = data.get("client_id")
+    system_id = data.get("system_id", None)
-    system_id = data.get("system_id")
     guild_id = data.get("guild_id")
     channel_id = data.get("channel_id")
 
@@ -86,14 +224,15 @@ async def join():
         return jsonify({"error": f"Failed to send command: {e}"}), 500
 
 
-@nodes_bp.route("/leave", methods=['POST'])
+@nodes_bp.route("/<client_id>/leave", methods=['POST'])
-async def leave():
+@jwt_required
+@role_required(UserRoles.MOD)
+async def leave(client_id):
     """
-    Send a join command to the specific system specified
+    Send a leave command to the specific node
     """
     data = await request.get_json()
 
-    client_id = data.get("client_id")
     guild_id = data.get("guild_id")
 
     # Check to make sure the client is online
@@ -111,5 +250,80 @@ async def leave():
 
         return jsonify({"status": "command sent", "client_id": client_id, "command": NodeCommands.LEAVE}), 200
 
+    except Exception as e:
+        return jsonify({"error": f"Failed to send command: {e}"}), 500
+
+
+@nodes_bp.route("/<client_id>/op25_start", methods=['POST'])
+@jwt_required
+@role_required(UserRoles.MOD)
+async def op25_start(client_id):
+    """
+    Send an OP25 start command to the specific node
+    """
+    # Check to make sure the client is online
+    if client_id not in current_app.active_clients:
+        return jsonify({"error": f"Client {client_id} not found, it might be offline"}), 404
+
+    try:
+        # Send the command asynchronously
+        await send_command_to_client(client_id, NodeCommands.OP25_START)
+
+        return jsonify({"status": "command sent", "client_id": client_id, "command": NodeCommands.OP25_START}), 200
+
+    except Exception as e:
+        return jsonify({"error": f"Failed to send command: {e}"}), 500
+
+
+@nodes_bp.route("/<client_id>/op25_stop", methods=['POST'])
+@jwt_required
+@role_required(UserRoles.MOD)
+async def op25_stop(client_id):
+    """
+    Send an OP25 stop command to the specific node
+    """
+    # Check to make sure the client is online
+    if client_id not in current_app.active_clients:
+        return jsonify({"error": f"Client {client_id} not found, it might be offline"}), 404
+
+    try:
+        # Send the command asynchronously
+        await send_command_to_client(client_id, NodeCommands.OP25_STOP)
+
+        return jsonify({"status": "command sent", "client_id": client_id, "command": NodeCommands.OP25_STOP}), 200
+
+    except Exception as e:
+        return jsonify({"error": f"Failed to send command: {e}"}), 500
+
+
+@nodes_bp.route("/<client_id>/op25_set", methods=['POST'])
+@jwt_required
+@role_required(UserRoles.MOD)
+async def op25_set(client_id):
+    """
+    Send an OP25 set config command to the specific node
+    """
+    data = await request.get_json()
+
+    system_id = data.get("system_id")
+
+    # Check to make sure the client is online
+    if client_id not in current_app.active_clients:
+        return jsonify({"error": f"Client {client_id} not found, it might be offline"}), 404
+
+    if not system_id:
+        return jsonify({"error":"No System ID supplied"}), 400
+
+    try:
+        args = [system_id]
+
+        if not isinstance(args, list):
+             return jsonify({"error": "'args' must be a list"}), 400
+
+        # Send the command asynchronously
+        await send_command_to_client(client_id, NodeCommands.OP25_SET, *args)
+
+        return jsonify({"status": "command sent", "client_id": client_id, "command": NodeCommands.OP25_SET}), 200
+
     except Exception as e:
         return jsonify({"error": f"Failed to send command: {e}"}), 500

app/routers/systems.py

@@ -1,11 +1,15 @@
 from quart import Blueprint, jsonify, request, abort, current_app
 from werkzeug.exceptions import HTTPException
-from internal.types import System
+from internal.types import System, UserRoles
+from quart_jwt_extended import jwt_required
+from routers.auth import role_required
 
 systems_bp = Blueprint('systems', __name__)
 
 
 @systems_bp.route("/", methods=['POST'])
+@jwt_required
+@role_required(UserRoles.MOD)
 async def create_system_route():
     """API endpoint to create a new system."""
     print("\n--- Handling POST /systems ---")
@@ -38,7 +42,7 @@ async def create_system_route():
 
         if created_system:
             print("Created new system:", created_system)
-            return jsonify(created_system), 201
+            return jsonify(created_system.to_dict()), 201
         else:
             abort(500, "Failed to create system in the database.")
     except HTTPException:
@@ -50,6 +54,8 @@ async def create_system_route():
 
 
 @systems_bp.route('/', methods=['GET'])
+@jwt_required
+@role_required(UserRoles.USER)
 async def list_systems_route():
     """API endpoint to get a list of all systems."""
     print("\n--- Handling GET /systems ---")
@@ -65,6 +71,8 @@ async def list_systems_route():
 
 
 @systems_bp.route('/<string:system_id>', methods=['GET'])
+@jwt_required
+@role_required(UserRoles.USER)
 async def get_system_route(system_id: str):
     """API endpoint to get details for a specific system by ID."""
     print(f"\n--- Handling GET /systems/{system_id} ---")
@@ -86,6 +94,8 @@ async def get_system_route(system_id: str):
 
 
 @systems_bp.route('/client/<string:client_id>', methods=['GET'])
+@jwt_required
+@role_required(UserRoles.USER)
 async def get_system_by_client_route(client_id: str):
     """API endpoint to get details for a specific system by ID."""
     print(f"\n--- Handling GET /systems/client/{client_id} ---")
@@ -107,9 +117,17 @@ async def get_system_by_client_route(client_id: str):
 
 
 @systems_bp.route('/<string:system_id>', methods=['PUT'])
-async def update_system_route(system_id: str, updated_system_data):
+@jwt_required
+@role_required(UserRoles.MOD)
+async def update_system_route(system_id: str):
     try:
-        update_system = await current_app.sys_db_h.update_system({"_id", system_id}, updated_system_data)
+        updated_system_data = await request.get_json()
+
+        if not updated_system_data:
+            abort(400, "No update data provided.")
+
+        query = {"_id": system_id}
+        update_system = await current_app.sys_db_h.update_system(query, {"$set": updated_system_data})
 
         if update_system:
             print("Updated system:", update_system)
@@ -126,25 +144,28 @@ async def update_system_route(system_id: str, updated_system_data):
 
 
 @systems_bp.route('/<string:system_id>', methods=['DELETE'])
+@jwt_required
+@role_required(UserRoles.MOD)
 async def delete_system_route(system_id: str):
     try:
-        deleted_system = await current_app.sys_db_h.delete_system({"_id", system_id})
+        query = {"_id": system_id}
+        delete_count = await current_app.sys_db_h.delete_system(query)
 
-        if deleted_system:
+        if delete_count is not None:
-            print("Deleted system:", deleted_system)
+            if delete_count > 0:
-            return jsonify(deleted_system), 201
+                return jsonify({"message": f"Successfully deleted {delete_count} systems(s)."}), 200
+            else:
+                abort(404, "System not found.")
         else:
-            abort(500, "Failed to delete system in the database.")
+            abort(500, "Failed to delete System.")
-
-    except HTTPException:
-        raise
     except Exception as e:
-        print(f"Error deleting system: {e}")
+        print(f"Error in delete_system_route: {e}")
-        # Catch any other unexpected errors
+        abort(500, f"An internal error occurred: {e}")
-        abort(500, f"Internal server error: {e}")
 
 
 @systems_bp.route('/<string:system_id>/assign', methods=['POST'])
+@jwt_required
+@role_required(UserRoles.MOD)
 async def assign_client_to_system_route(system_id: str):
     """
     API endpoint to assign a client ID to a system's available_on_nodes list.
@@ -201,6 +222,8 @@ async def assign_client_to_system_route(system_id: str):
 
 
 @systems_bp.route('/<string:system_id>/dismiss', methods=['POST'])
+@jwt_required
+@role_required(UserRoles.MOD)
 async def dismiss_client_from_system_route(system_id: str):
     """
     API endpoint to dismiss (remove) a client ID from a system's available_on_nodes list.
@@ -259,6 +282,8 @@ async def dismiss_client_from_system_route(system_id: str):
 
 
 @systems_bp.route('/search', methods=['GET'])
+@jwt_required
+@role_required(UserRoles.MOD)
 async def search_systems_route():
     """
     API endpoint to search for systems based on query parameters.

app/server.py

@@ -1,40 +1,61 @@
+# server.py
 import asyncio
 import websockets
 import json
 import uuid
-from quart import Quart, jsonify, request
+from quart import Quart, jsonify, request, abort
+from quart_cors import cors
 from routers.systems import systems_bp
 from routers.nodes import nodes_bp, register_client, unregister_client
 from routers.bot import bot_bp
-from internal.db_wrappers import SystemDbController, DiscordIdDbController
+from routers.auth import auth_bp
+from internal.db_wrappers import SystemDbController, DiscordIdDbController #
+from internal.auth_wrappers import UserDbController #
+from config.jwt_config import jwt, configure_jwt
+
 
 # --- WebSocket Server Components ---
-# Dictionary to store active clients: {client_id: websocket}
 active_clients = {}
 
 
+# --- Quart API Components ---
+app = Quart(__name__)
+app = cors(app, allow_origin="*")
+
+websocket_server_instance = None
+app.active_clients = active_clients
+
+# Create and attach the DB wrappers
+app.sys_db_h = SystemDbController() #
+app.d_id_db_h = DiscordIdDbController() #
+app.user_db_h = UserDbController() #
+
+configure_jwt(app)
+jwt.init_app(app)
+
+
 async def websocket_server_handler(websocket):
-    """Handles incoming WebSocket connections and messages from clients."""
     client_id = None
     try:
-        # Handshake: Receive the first message which should contain the client ID
         handshake_message = await websocket.recv()
         handshake_data = json.loads(handshake_message)
-
         if handshake_data.get("type") == "handshake" and "id" in handshake_data:
             client_id = handshake_data["id"]
-            await register_client(websocket, client_id)
+            client_nickname = handshake_data.get("nickname")
-            await websocket.send(json.dumps({"type": "handshake_ack", "status": "success"})) # Acknowledge handshake
+            client_active_token = handshake_data.get("active_token")
-
+            # Get the DiscordId for the passed token
-            # Keep the connection alive and listen for potential messages from the client
+            if client_active_token:
-            # (Though in this server-commanded model, clients might not send much)
+                active_discord_id = await app.d_id_db_h.find_discord_id({"token":client_active_token})
-            # We primarily wait for the client to close the connection
+                if active_discord_id:
+                    client_active_token = active_discord_id
+            await register_client(websocket, client_id, client_nickname, client_active_token)
+            if not app.active_clients[client_id].access_token:
+                abort(500, "Error retrieving access token")
+            await websocket.send(json.dumps({"type": "handshake_ack", "status": "success", "access_token": app.active_clients[client_id].access_token}))
             await websocket.wait_closed()
-
         else:
             print(f"Received invalid handshake from {websocket.remote_address}. Closing connection.")
             await websocket.close()
-
     except websockets.exceptions.ConnectionClosedError:
         print(f"Client connection closed unexpectedly for {client_id}.")
     except json.JSONDecodeError:
@@ -45,62 +66,59 @@ async def websocket_server_handler(websocket):
         if client_id:
             await unregister_client(client_id)
 
-# --- Quart API Components ---
-app = Quart(__name__)
-
-# Store the websocket server instance
-websocket_server_instance = None
-
-# Make active_clients accessible via the app instance.
-app.active_clients = active_clients
-
-# Create and attach the DB wrappers
-app.sys_db_h = SystemDbController()
-app.d_id_db_h = DiscordIdDbController()
 
 @app.before_serving
-async def startup_websocket_server():
+async def startup_tasks(): # Combined startup logic
-    """Starts the WebSocket server when the Quart app starts."""
+    """Starts the WebSocket server and prepares other resources."""
     global websocket_server_instance
     websocket_server_address = "0.0.0.0"
     websocket_server_port = 8765
-
-    # Start the WebSocket server task
     websocket_server_instance = await websockets.serve(
         websocket_server_handler,
         websocket_server_address,
         websocket_server_port
     )
     print(f"WebSocket server started on ws://{websocket_server_address}:{websocket_server_port}")
+    # Database connections are now established on first use by MongoHandler's __aenter__/connect
+    # No explicit connect calls needed here unless desired for early failure detection.
+    print("Application startup complete. DB connections will be initialized on first use.")
 
 @app.after_serving
-async def shutdown_websocket_server():
+async def shutdown_tasks(): # Combined shutdown logic
-    """Shuts down the WebSocket server when the Quart app stops."""
+    """Shuts down services and closes connections."""
     global websocket_server_instance
     if websocket_server_instance:
         websocket_server_instance.close()
         await websocket_server_instance.wait_closed()
         print("WebSocket server shut down.")
 
+    # Close database connections
+    if hasattr(app, 'user_db_h') and app.user_db_h:
+        print("Closing User DB connection...")
+        await app.user_db_h.close_db_connection() #
+    if hasattr(app, 'sys_db_h') and app.sys_db_h:
+        print("Closing System DB connection...")
+        await app.sys_db_h.close_db_connection() #
+    if hasattr(app, 'd_id_db_h') and app.d_id_db_h:
+        print("Closing Discord ID DB connection...")
+        await app.d_id_db_h.close_db_connection() #
+    print("All database connections have been signaled to close.")
+
 
 app.register_blueprint(systems_bp, url_prefix="/systems")
 app.register_blueprint(nodes_bp, url_prefix="/nodes")
 app.register_blueprint(bot_bp, url_prefix="/bots")
+app.register_blueprint(auth_bp, url_prefix="/auth")
 
 @app.route('/')
 async def index():
     return "Welcome to the Radio App Server API!"
 
-# --- Main Execution ---
 if __name__ == "__main__":
-    # Quart's app.run() will start the asyncio event loop and manage it.
-    # The @app.before_serving decorator ensures the websocket server starts within that loop.
-    # We removed asyncio.run(main()) and the main() function itself.
     print("Starting Quart API server...")
     app.run(
         host="0.0.0.0",
         port=5000,
-        debug=False # Set to True for development
+        debug=False
     )
-    print("Quart API server stopped.")
+    print("Quart API server stopped.")
-

requirements.txt

@@ -1,4 +1,7 @@
 websockets
 quart
+quart-cors
 motor
-fastapi
+fastapi
+quart-jwt-extended
+passlib[bcrypt]