UPDATE: Added functionality that updates database on stream start and on stream end. Adds new stream to database

nginx/nginx.conf

@@ -16,6 +16,7 @@ rtmp {
             push rtmp://127.0.0.1:1935/hls-live;
 
             on_publish http://web_server:5000/publish_stream;
+            on_publish_done http://web_server:5000/end_stream;
 
         }
 

web_server/blueprints/streams.py

@@ -3,6 +3,7 @@ from utils.stream_utils import streamer_live_status, streamer_most_recent_stream
 from utils.user_utils import get_user_id
 from blueprints.utils import login_required
 from database.database import Database
+from datetime import datetime
 stream_bp = Blueprint("stream", __name__)
 
 
@@ -171,6 +172,8 @@ def stream_thumbnail_snapshot(streamer_id):
     """
     return
 
+
+## RTMP Server Routes
 @stream_bp.route("/publish_stream", methods=["POST"])
 def publish_stream():
     """
@@ -181,13 +184,22 @@ def publish_stream():
     # Check if stream key is valid
     db = Database()
     db.create_connection()
-    user_info = db.fetchone("""SELECT user_id, username 
+    user_info = db.fetchone("""SELECT user_id, username, current_stream_title, current_selected_category_id 
                                FROM users 
                                WHERE stream_key = ?""", (stream_key,))
 
     if not user_info:
         return "Unauthorized", 403
     
+    # Insert stream into database
+    db.execute("""INSERT INTO streams (user_id, title, category_id, start_time, isLive)
+                  VALUES (?, ?, ?, ?, ?)""", (user_info["user_id"], 
+                                           user_info["current_stream_title"],
+                                           1,
+                                           datetime.now(),
+                                           1))
+
+    
     return redirect(f"/{user_info['username']}")
 
 @stream_bp.route("/end_stream", methods=["POST"])
@@ -195,4 +207,14 @@ def end_stream():
     """
     Ends a stream
     """
-    return
+    db = Database()
+    db.create_connection()
+    user_info = db.fetchone("""SELECT user_id FROM users WHERE stream_key = ?""", (request.form.get("name"),))
+
+    if not user_info:
+        return "Unauthorized", 403
+    
+    # Set stream to not live
+    db.execute("""UPDATE streams SET isLive = 0 WHERE user_id = ? AND isLive = 1""", (user_info["user_id"],))
+
+    return "Stream ended", 200

web_server/database/database.py

@@ -30,7 +30,21 @@ class Database:
 
         result = self.cursor.fetchone()
         return self.convert_to_list_dict(result)
-    
+
+    def execute(self, query: str, parameters=None) -> None:
+        """
+        Executes a command (e.g., INSERT, UPDATE, DELETE) and commits the changes.
+        """
+        try:
+            if parameters:
+                self.cursor.execute(query, parameters)
+            else:
+                self.cursor.execute(query)
+            self.commit_data()
+        except Exception as e:
+            print(f"Error executing command: {e}")
+            raise
+
     def convert_to_list_dict(self, result):
         """
         Converts a query result to a list of dictionaries
@@ -47,7 +61,7 @@ class Database:
         else:
             # for fetchall or fetchmany
             return [dict(zip(columns, row)) for row in result]
-    
+
     def commit_data(self):
         try:
             self._conn.commit()