transient.chat/blueprints/rooms.py

from flask import Blueprint, request, jsonify, Response
from datetime import datetime, timezone
from typing import Dict, Any, Tuple
import time

from blueprints.auth import verify_user_key
from models.state import (
    room_users,
    room_passwords,
    authorized_room_users,
    room_owners,
    kicked_users,
    nuked_rooms,
    banned_ips,
    banned_usernames,
    user_ips,
    video_state,
    hash_room_password,
    verify_room_password,
)

rooms_bp = Blueprint("rooms", __name__)


@rooms_bp.route("/rooms", methods=["GET", "POST"])
def get_or_create_rooms() -> Tuple[Response, int] | Response:
    if request.method == "POST":
        data: Dict[str, Any] = request.get_json() or {}
        room_name: str = (data.get("name") or data.get("room", "")).strip()
        password: str = str(data.get("password", "")).strip()
        username: str = str(data.get("username", "")).strip()
        key: str = str(data.get("key", "")).strip()

        return handle_room_join_or_create(room_name, password, username, key)

    # GET request: Return ONLY active rooms
    active_rooms = []
    all_users = set()

    for room, users in list(room_users.items()):
        user_count = len(users)
        if user_count > 0:
            all_users.update(users.keys())
            has_wp = bool(
                room in video_state
                and (video_state[room].get("is_active") or video_state[room].get("url"))
            )
            active_rooms.append(
                {
                    "name": room,
                    "users": user_count,
                    "has_watchparty": has_wp,
                    "type": "watchparty" if has_wp else "chat",
                }
            )

    return jsonify(
        {
            "active_rooms": active_rooms,
            "unique_users": len(all_users),
            # Maintain backward compatibility aliases
            "public": active_rooms,
            "regional": [],
            "topical": [],
        }
    )


@rooms_bp.route("/rooms/join", methods=["POST"])
def join_room() -> Tuple[Response, int] | Response:
    data: Dict[str, Any] = request.get_json() or {}
    room_name: str = (data.get("room") or data.get("name", "")).strip()
    password: str = str(data.get("password", "")).strip()
    username: str = str(data.get("username", "")).strip()
    key: str = str(data.get("key", "")).strip()

    return handle_room_join_or_create(room_name, password, username, key)


def handle_room_join_or_create(
    room_name: str, password: str, username: str, key: str
) -> Tuple[Response, int] | Response:
    if not room_name:
        return jsonify({"error": "Room name is required"}), 400

    if not password:
        return jsonify({"error": "Room password is required for private encryption"}), 400

    if not username or not key:
        return jsonify({"error": "User credentials are required"}), 401

    if not verify_user_key(username, key):
        return jsonify({"error": "Invalid user credentials"}), 403

    user_ip = request.remote_addr
    if (
        user_ip in banned_ips
        or username in banned_usernames
        or (username in user_ips and user_ips[username] in banned_ips)
    ):
        return jsonify({"error": "User is banned"}), 403

    if room_name in nuked_rooms:
        if time.time() < nuked_rooms[room_name]:
            return jsonify({"error": "Room is currently unavailable"}), 403
        else:
            nuked_rooms.pop(room_name, None)

    if username in kicked_users.get(room_name, set()):
        return jsonify({"error": "You were kicked from this room"}), 403

    # On-demand room creation vs existing room verification
    if room_name not in room_passwords:
        # Create new on-demand encrypted room
        room_passwords[room_name] = hash_room_password(password)
        room_owners[room_name] = username
    else:
        # Verify password for existing room
        if not verify_room_password(password, room_passwords[room_name]):
            return jsonify({"error": "Incorrect room password"}), 401

    # Authorize user session for this room
    authorized_room_users.setdefault(room_name, set()).add(username)

    # Register user presence
    room_users.setdefault(room_name, {})[username] = datetime.now(timezone.utc).isoformat()

    return jsonify(
        {
            "status": "ok",
            "name": room_name,
            "room": room_name,
            "owner": room_owners.get(room_name),
            "is_owner": room_owners.get(room_name) == username,
        }
    )