app.py

import gradio as gr
import asyncio
import websockets
import json
import uuid
import argparse
import urllib.parse
from datetime import datetime
import logging
import sys
import os
import time
from pathlib import Path

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("chat-node")

# Dictionary to store active connections
active_connections = {}
# Dictionary to store message history for each chat room (in-memory cache)
chat_history = {}
# Dictionary to track file modification times
file_modification_times = {}
# Dictionary to track users in each room/sector
sector_users = {}
# Global variables for event loop and queue
main_event_loop = None
message_queue = []

# Grid dimensions for 2D sector map
GRID_WIDTH = 10
GRID_HEIGHT = 10

# Directory to store persistent chat history
HISTORY_DIR = "chat_history"

# Create history directory if it doesn't exist
os.makedirs(HISTORY_DIR, exist_ok=True)

# README.md file that won't be listed or deleted
README_PATH = os.path.join(HISTORY_DIR, "README.md")
if not os.path.exists(README_PATH):
    with open(README_PATH, "w") as f:
        f.write("# Chat History\n\nThis directory contains persistent chat history files.\n")

# Get node name from URL or command line
def get_node_name():
    parser = argparse.ArgumentParser(description='Start a chat node with a specific name')
    parser.add_argument('--node-name', type=str, default=None, help='Name for this chat node')
    parser.add_argument('--port', type=int, default=7860, help='Port to run the Gradio interface on')
    
    args = parser.parse_args()
    node_name = args.node_name
    port = args.port
    
    # If no node name specified, generate a random one
    if not node_name:
        node_name = f"node-{uuid.uuid4().hex[:8]}"
        
    return node_name, port

def get_room_history_file(room_id):
    """Get the filename for a room's history."""
    # Create timestamp-based log files
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    return os.path.join(HISTORY_DIR, f"{room_id}_{timestamp}.jsonl")

def get_all_room_history_files(room_id):
    """Get all history files for a specific room."""
    files = []
    for file in os.listdir(HISTORY_DIR):
        if file.startswith(f"{room_id}_") and file.endswith(".jsonl"):
            files.append(os.path.join(HISTORY_DIR, file))
    # Sort by modification time (newest first)
    files.sort(key=lambda x: os.path.getmtime(x), reverse=True)
    return files

def get_all_history_files():
    """Get a list of all unique room IDs from history files, sorted by modification time (newest first)."""
    try:
        rooms = {}  # room_id -> (newest_file_path, mod_time)
        
        for file in os.listdir(HISTORY_DIR):
            if file.endswith(".jsonl"):
                file_path = os.path.join(HISTORY_DIR, file)
                mod_time = os.path.getmtime(file_path)
                
                # Extract room_id from filename (format: roomid_timestamp.jsonl)
                parts = file.split('_', 1)
                if len(parts) > 0:
                    room_id = parts[0]
                    
                    # Keep track of newest file for each room
                    if room_id not in rooms or mod_time > rooms[room_id][1]:
                        rooms[room_id] = (file_path, mod_time)
        
        # Convert to list and sort by modification time
        history_files = [(room_id, file_path, mod_time) for room_id, (file_path, mod_time) in rooms.items()]
        history_files.sort(key=lambda x: x[2], reverse=True)
        
        return history_files
    except Exception as e:
        logger.error(f"Error in get_all_history_files: {e}")
        return []  # Return empty list on error

def load_room_history(room_id):
    """Load chat history for a room from all persistent storage files."""
    if room_id not in chat_history:
        chat_history[room_id] = []
        
        # Get all history files for this room
        history_files = get_all_room_history_files(room_id)
        
        # Track file modification times
        for file in history_files:
            if file not in file_modification_times:
                file_modification_times[file] = os.path.getmtime(file)
        
        # Load messages from all files
        messages = []
        for history_file in history_files:
            try:
                with open(history_file, 'r') as f:
                    for line in f:
                        line = line.strip()
                        if line:  # Skip empty lines
                            try:
                                data = json.loads(line)
                                messages.append(data)
                            except json.JSONDecodeError:
                                logger.error(f"Error parsing JSON line in {history_file}")
            except Exception as e:
                logger.error(f"Error loading history from {history_file}: {e}")
        
        # Sort by timestamp
        messages.sort(key=lambda x: x.get("timestamp", ""), reverse=False)
        chat_history[room_id] = messages
        
        logger.info(f"Loaded {len(messages)} messages from {len(history_files)} files for room {room_id}")
    
    # Track users in this sector
    if room_id not in sector_users:
        sector_users[room_id] = set()
    
    return chat_history[room_id]

def save_message_to_history(room_id, message):
    """Save a single message to the newest history file for a room."""
    # Get the newest history file or create a new one
    history_files = get_all_room_history_files(room_id)
    
    if not history_files:
        # Create a new file
        history_file = get_room_history_file(room_id)
    else:
        # Use the newest file if it's less than 1 MB, otherwise create a new one
        newest_file = history_files[0]
        if os.path.getsize(newest_file) > 1024 * 1024:  # 1 MB
            history_file = get_room_history_file(room_id)
        else:
            history_file = newest_file
    
    try:
        # Append the message as a single line of JSON
        with open(history_file, 'a') as f:
            f.write(json.dumps(message) + '\n')
        
        # Update modification time
        file_modification_times[history_file] = os.path.getmtime(history_file)
        
        logger.debug(f"Saved message to {history_file}")
    except Exception as e:
        logger.error(f"Error saving message to {history_file}: {e}")

def check_for_new_messages():
    """Check for new messages in all history files."""
    updated_rooms = set()
    
    # Check all files in the history directory
    for file in os.listdir(HISTORY_DIR):
        if file.endswith(".jsonl"):
            file_path = os.path.join(HISTORY_DIR, file)
            current_mtime = os.path.getmtime(file_path)
            
            # Check if this file is new or has been modified
            if file_path not in file_modification_times or current_mtime > file_modification_times[file_path]:
                # Extract room_id from filename
                parts = file.split('_', 1)
                if len(parts) > 0:
                    room_id = parts[0]
                    updated_rooms.add(room_id)
                    
                # Update tracked modification time
                file_modification_times[file_path] = current_mtime
    
    # Reload history for updated rooms
    for room_id in updated_rooms:
        if room_id in chat_history:
            # Remember we had this room loaded
            old_history_len = len(chat_history[room_id])
            # Clear and reload
            chat_history[room_id] = []
            load_room_history(room_id)
            new_history_len = len(chat_history[room_id])
            
            if new_history_len > old_history_len:
                logger.info(f"Found {new_history_len - old_history_len} new messages for room {room_id}")
    
    return updated_rooms

def get_sector_coordinates(room_id):
    """Convert a room ID to grid coordinates, or assign new ones."""
    try:
        # Try to parse room ID as "x,y"
        if ',' in room_id:
            x, y = map(int, room_id.split(','))
            return max(0, min(x, GRID_WIDTH-1)), max(0, min(y, GRID_HEIGHT-1))
    except:
        pass
    
    # Hash the room_id string to get stable coordinates
    hash_val = hash(room_id)
    x = abs(hash_val) % GRID_WIDTH
    y = abs(hash_val >> 8) % GRID_HEIGHT
    
    return x, y

def generate_sector_map():
    """Generate an ASCII representation of the sector map."""
    # Initialize empty grid
    grid = [[' ' for _ in range(GRID_WIDTH)] for _ in range(GRID_HEIGHT)]
    
    # Place active rooms with user counts
    for room_id, users in sector_users.items():
        if users:  # Only show rooms with users
            x, y = get_sector_coordinates(room_id)
            user_count = len(users)
            grid[y][x] = str(min(user_count, 9)) if user_count < 10 else '+'
    
    # Create ASCII representation
    header = '  ' + ''.join([str(i % 10) for i in range(GRID_WIDTH)])
    map_str = header + '\n'
    
    for y in range(GRID_HEIGHT):
        row = f"{y % 10}|"
        for x in range(GRID_WIDTH):
            row += grid[y][x]
        row += '|'
        map_str += row + '\n'
    
    footer = '  ' + ''.join([str(i % 10) for i in range(GRID_WIDTH)])
    map_str += footer
    
    return f"```\n{map_str}\n```\n\nLegend: Number indicates users in sector. '+' means 10+ users."

def list_available_rooms():
    """List all available chat rooms with their last activity time and user count."""
    try:
        history_files = get_all_history_files()
        
        if not history_files:
            return "No chat rooms available yet. Create one by joining a room!"
        
        room_list = "### Available Chat Rooms (Sectors)\n\n"
        room_list += "| Room ID | Sector | Users | Last Activity |\n"
        room_list += "|---------|--------|-------|---------------|\n"
        
        for room_id, file_path, mod_time in history_files:
            x, y = get_sector_coordinates(room_id)
            user_count = len(sector_users.get(room_id, set()))
            last_activity = datetime.fromtimestamp(mod_time).strftime("%Y-%m-%d %H:%M:%S")
            room_list += f"| {room_id} | ({x},{y}) | {user_count} | {last_activity} |\n"
        
        room_list += "\n\n### Sector Map\n\n" + generate_sector_map()
        
        return room_list
    except Exception as e:
        logger.error(f"Error in list_available_rooms: {e}")
        return f"Error listing rooms: {str(e)}"

async def broadcast_message(message, room_id):
    """Broadcast a message to all clients in a room."""
    if room_id in active_connections:
        disconnected_clients = []
        
        for client_id, websocket in active_connections[room_id].items():
            try:
                await websocket.send(json.dumps(message))
            except websockets.exceptions.ConnectionClosed:
                disconnected_clients.append(client_id)
        
        # Clean up disconnected clients
        for client_id in disconnected_clients:
            del active_connections[room_id][client_id]

async def start_websocket_server(host='0.0.0.0', port=8765):
    """Start the WebSocket server."""
    server = await websockets.serve(websocket_handler, host, port)
    logger.info(f"WebSocket server started on ws://{host}:{port}")
    return server

def send_message(message, username, room_id):
    """Function to send a message from the Gradio interface."""
    if not message.strip():
        return None
    
    global message_queue
    
    msg_data = {
        "type": "chat",
        "content": message,
        "username": username,
        "room_id": room_id
    }
    
    # Add to queue for processing by the main loop
    message_queue.append(msg_data)
    
    # Format the message for display in the UI
    formatted_msg = f"{username}: {message}"
    return formatted_msg

def send_clear_command():
    """Send a command to clear all chat history."""
    global message_queue
    
    msg_data = {
        "type": "command",
        "command": "clear_history",
        "username": "System"
    }
    
    # Add to queue for processing by the main loop
    message_queue.append(msg_data)
    
    return "🧹 Clearing all chat history..."

async def clear_all_history():
    """Clear all chat history for all rooms."""
    global chat_history, sector_users
    
    # Clear in-memory history
    chat_history = {}
    sector_users = {}
    
    # Delete all history files except README.md
    for file in os.listdir(HISTORY_DIR):
        if file.endswith(".jsonl"):
            try:
                os.remove(os.path.join(HISTORY_DIR, file))
            except Exception as e:
                logger.error(f"Error removing file {file}: {e}")
    
    # Broadcast clear message to all rooms
    clear_msg = {
        "type": "system",
        "content": "🧹 All chat history has been cleared by a user",
        "timestamp": datetime.now().isoformat(),
        "sender": "system"
    }
    
    for room_id in list(active_connections.keys()):
        clear_msg["room_id"] = room_id
        await broadcast_message(clear_msg, room_id)
    
    logger.info("All chat history cleared")
    return "All chat history cleared"

def join_room(room_id, chat_history_output):
    """Join a specific chat room."""
    if not room_id.strip():
        return "Please enter a valid room ID", chat_history_output
    
    # Sanitize the room ID
    room_id = urllib.parse.quote(room_id.strip())
    
    # Load room history from persistent storage
    history = load_room_history(room_id)
    
    # Get sector coordinates
    x, y = get_sector_coordinates(room_id)
    
    # Format existing messages
    formatted_history = [f"You are now in Sector ({x},{y}) - Room ID: {room_id}"]
    formatted_history.append(f"Sector Map:\n{generate_sector_map()}")
    
    for msg in history:
        if msg.get("type") == "chat":
            sender_node = f" [{msg.get('sender_node', 'unknown')}]" if "sender_node" in msg else ""
            time_str = ""
            if "timestamp" in msg:
                try:
                    dt = datetime.fromisoformat(msg["timestamp"])
                    time_str = f"[{dt.strftime('%H:%M:%S')}] "
                except:
                    pass
            formatted_history.append(f"{time_str}{msg.get('username', 'Anonymous')}{sender_node}: {msg.get('content', '')}")
        elif msg.get("type") == "system":
            formatted_history.append(f"System: {msg.get('content', '')}")
    
    return f"Joined room: {room_id}", formatted_history

async def websocket_handler(websocket, path):
    """Handle WebSocket connections."""
    client_id = str(uuid.uuid4())
    room_id = "default"  # Default initialization to avoid reference errors
    
    try:
        # Extract room_id from path if present
        path_parts = path.strip('/').split('/')
        room_id = path_parts[0] if path_parts else "default"
        
        # Register the new client
        if room_id not in active_connections:
            active_connections[room_id] = {}
        
        active_connections[room_id][client_id] = websocket
        
        # Add user to sector map
        if room_id not in sector_users:
            sector_users[room_id] = set()
        sector_users[room_id].add(client_id)
        
        # Get sector coordinates
        x, y = get_sector_coordinates(room_id)
        
        # Load or initialize chat history
        room_history = load_room_history(room_id)
        
        # Send welcome message
        welcome_msg = {
            "type": "system",
            "content": f"Welcome to room '{room_id}' (Sector {x},{y})! Connected from node '{NODE_NAME}'",
            "timestamp": datetime.now().isoformat(),
            "sender": "system",
            "room_id": room_id
        }
        await websocket.send(json.dumps(welcome_msg))
        
        # Send sector map
        map_msg = {
            "type": "system",
            "content": f"Sector Map:\n{generate_sector_map()}",
            "timestamp": datetime.now().isoformat(),
            "sender": "system",
            "room_id": room_id
        }
        await websocket.send(json.dumps(map_msg))
        
        # Send chat history
        for msg in room_history:
            await websocket.send(json.dumps(msg))
        
        # Broadcast join notification
        join_msg = {
            "type": "system",
            "content": f"User joined the room (Sector {x},{y}) - {len(sector_users[room_id])} users now present",
            "timestamp": datetime.now().isoformat(),
            "sender": "system",
            "room_id": room_id
        }
        await broadcast_message(join_msg, room_id)
        save_message_to_history(room_id, join_msg)
        
        logger.info(f"New client {client_id} connected to room {room_id} (Sector {x},{y})")
        
        # Handle messages from this client
        async for message in websocket:
            try:
                data = json.loads(message)
                
                # Check for clear command
                if data.get("type") == "command" and data.get("command") == "clear_history":
                    result = await clear_all_history()
                    continue
                
                # Check for map request
                if data.get("type") == "command" and data.get("command") == "show_map":
                    map_msg = {
                        "type": "system",
                        "content": f"Sector Map:\n{generate_sector_map()}",
                        "timestamp": datetime.now().isoformat(),
                        "sender": "system",
                        "room_id": room_id
                    }
                    await websocket.send(json.dumps(map_msg))
                    continue
                
                # Add metadata to the message
                data["timestamp"] = datetime.now().isoformat()
                data["sender_node"] = NODE_NAME
                data["room_id"] = room_id
                
                # Store in history
                chat_history[room_id].append(data)
                if len(chat_history[room_id]) > 500:  # Increased limit to 500 messages
                    chat_history[room_id] = chat_history[room_id][-500:]
                
                # Save to persistent storage
                save_message_to_history(room_id, data)
                
                # Broadcast to all clients in the room
                await broadcast_message(data, room_id)
                
            except json.JSONDecodeError:
                error_msg = {
                    "type": "error",
                    "content": "Invalid JSON format",
                    "timestamp": datetime.now().isoformat(),
                    "sender": "system",
                    "room_id": room_id
                }
                await websocket.send(json.dumps(error_msg))
    
    except websockets.exceptions.ConnectionClosed:
        logger.info(f"Client {client_id} disconnected from room {room_id}")
    finally:
        # Remove the client when disconnected
        if room_id in active_connections and client_id in active_connections[room_id]:
            del active_connections[room_id][client_id]
            
            # Remove user from sector map
            if room_id in sector_users and client_id in sector_users[room_id]:
                sector_users[room_id].remove(client_id)
            
            # Get sector coordinates
            x, y = get_sector_coordinates(room_id)
            
            # Broadcast leave notification
            leave_msg = {
                "type": "system",
                "content": f"User left the room (Sector {x},{y}) - {len(sector_users.get(room_id, set()))} users remaining",
                "timestamp": datetime.now().isoformat(),
                "sender": "system",
                "room_id": room_id
            }
            await broadcast_message(leave_msg, room_id)
            save_message_to_history(room_id, leave_msg)
            
            # Clean up empty rooms (but keep history)
            if not active_connections[room_id]:
                del active_connections[room_id]

async def process_message_queue():
    """Process messages in the queue and broadcast them."""
    global message_queue
    
    while True:
        # Check if there are messages to process
        if message_queue:
            # Get the oldest message
            msg_data = message_queue.pop(0)
            
            # Check for clear command
            if msg_data.get("type") == "command" and msg_data.get("command") == "clear_history":
                await clear_all_history()
            elif "room_id" in msg_data:
                # Regular message - add to history and broadcast
                room_id = msg_data["room_id"]
                
                # Add timestamp if not present
                if "timestamp" not in msg_data:
                    msg_data["timestamp"] = datetime.now().isoformat()
                
                # Add sender node if not present
                if "sender_node" not in msg_data:
                    msg_data["sender_node"] = NODE_NAME
                
                # Store in memory history
                if room_id not in chat_history:
                    chat_history[room_id] = []
                chat_history[room_id].append(msg_data)
                
                # Save to file
                save_message_to_history(room_id, msg_data)
                
                # Broadcast to all clients in the room
                await broadcast_message(msg_data, room_id)
        
        # Check for file changes every second
        updated_rooms = check_for_new_messages()
        
        # If any rooms were updated, notify clients
        for room_id in updated_rooms:
            if room_id in active_connections:
                # Find the newest messages that clients might not have seen
                # This is a simplification - in a real app, you'd track which messages each client has seen
                if room_id in chat_history and chat_history[room_id]:
                    # Get the latest 5 messages as an example
                    latest_messages = chat_history[room_id][-5:]
                    for msg in latest_messages:
                        # Only broadcast messages from other nodes (to avoid duplicates)
                        if msg.get("sender_node") != NODE_NAME:
                            await broadcast_message(msg, room_id)
        
        # Sleep to avoid busy-waiting
        await asyncio.sleep(1.0)  # Check every second

def create_gradio_interface():
    """Create and return the Gradio interface."""
    with gr.Blocks(title=f"Chat Node: {NODE_NAME}") as interface:
        gr.Markdown(f"# Chat Node: {NODE_NAME}")
        gr.Markdown("Join a room by entering a room ID below or create a new one.")
        
        # Room list and management
        with gr.Row():
            with gr.Column(scale=3):
                room_list = gr.Markdown(value="Loading available rooms...")
                refresh_button = gr.Button("🔄 Refresh Room List")
            with gr.Column(scale=1):
                clear_button = gr.Button("🧹 Clear All Chat History", variant="stop")
        
        # Join room controls with 2D grid input
        with gr.Row():
            with gr.Column(scale=2):
                room_id_input = gr.Textbox(label="Room ID", placeholder="Enter room ID or use x,y coordinates")
                join_button = gr.Button("Join Room")
            with gr.Column(scale=1):
                with gr.Row():
                    x_coord = gr.Number(label="X", value=0, minimum=0, maximum=GRID_WIDTH-1, step=1)
                    y_coord = gr.Number(label="Y", value=0, minimum=0, maximum=GRID_HEIGHT-1, step=1)
                grid_join_button = gr.Button("Join by Coordinates")
        
        # Chat area with multiline support
        chat_history_output = gr.Textbox(label="Chat History", lines=20, max_lines=20)
        
        # Message controls with multiline support
        with gr.Row():
            username_input = gr.Textbox(label="Username", placeholder="Enter your username", value="User")
            with gr.Column(scale=3):
                message_input = gr.Textbox(
                    label="Message", 
                    placeholder="Type your message here. Press Shift+Enter for new line, Enter to send.", 
                    lines=3
                )
            with gr.Column(scale=1):
                send_button = gr.Button("Send")
                map_button = gr.Button("🗺️ Show Map")
        
        # Current room display
        current_room_display = gr.Textbox(label="Current Room", value="Not joined any room yet")
        
        # Event handlers
        refresh_button.click(
            list_available_rooms,
            inputs=[],
            outputs=[room_list]
        )
        
        clear_button.click(
            send_clear_command,
            inputs=[],
            outputs=[room_list]
        )
        
        def join_by_coordinates(x, y):
            """Join a room using grid coordinates."""
            room_id = f"{int(x)},{int(y)}"
            return room_id
        
        # Link grid coordinates to room ID
        grid_join_button.click(
            join_by_coordinates,
            inputs=[x_coord, y_coord],
            outputs=[room_id_input]
        ).then(
            join_room,
            inputs=[room_id_input, chat_history_output],
            outputs=[current_room_display, chat_history_output]
        )
        
        join_button.click(
            join_room,
            inputs=[room_id_input, chat_history_output],
            outputs=[current_room_display, chat_history_output]
        )
        
        def send_and_clear(message, username, room_id):
            if not room_id.startswith("Joined room:"):
                return "Please join a room first", message
            
            actual_room_id = room_id.replace("Joined room: ", "").strip()
            
            # Support for multi-line messages
            message_lines = message.strip().split("\n")
            formatted_msg = ""
            
            for line in message_lines:
                if line.strip():  # Skip empty lines
                    sent_msg = send_message(line.strip(), username, actual_room_id)
                    if sent_msg:
                        formatted_msg += sent_msg + "\n"
            
            if formatted_msg:
                return "", formatted_msg
            return message, None
        
        send_button.click(
            send_and_clear,
            inputs=[message_input, username_input, current_room_display],
            outputs=[message_input, chat_history_output]
        )
        
        def show_sector_map(room_id):
            if not room_id.startswith("Joined room:"):
                return "Please join a room first to view the map"
            
            return generate_sector_map()
        
        map_button.click(
            show_sector_map,
            inputs=[current_room_display],
            outputs=[chat_history_output]
        )
        
        # Handle Enter key for sending, Shift+Enter for new line
        def on_message_submit(message, username, room_id):
            # Simply call send_and_clear
            return send_and_clear(message, username, room_id)
        
        message_input.submit(
            on_message_submit,
            inputs=[message_input, username_input, current_room_display],
            outputs=[message_input, chat_history_output]
        )
        
        # On load, populate room list
        interface.load(
            list_available_rooms,
            inputs=[],
            outputs=[room_list]
        )
    
    return interface

async def main():
    """Main function to start the application."""
    global NODE_NAME, main_event_loop
    NODE_NAME, port = get_node_name()
    
    # Store the main event loop for later use
    main_event_loop = asyncio.get_running_loop()
    
    # Start WebSocket server
    server = await start_websocket_server()
    
    # Start message queue processor
    asyncio.create_task(process_message_queue())
    
    # Create and launch Gradio interface
    interface = create_gradio_interface()
    
    # Custom middleware to extract node name from URL query parameters
    from starlette.middleware.base import BaseHTTPMiddleware
    
    class NodeNameMiddleware(BaseHTTPMiddleware):
        async def dispatch(self, request, call_next):
            global NODE_NAME
            query_params = dict(request.query_params)
            if "node_name" in query_params:
                NODE_NAME = query_params["node_name"]
                logger.info(f"Node name set to {NODE_NAME} from URL parameter")
            
            response = await call_next(request)
            return response
    
    # Apply middleware
    app = gr.routes.App.create_app(interface)
    app.add_middleware(NodeNameMiddleware)
    
    # Launch with the modified app
    gr.routes.mount_gradio_app(app, interface, path="/")
    
    # Run the FastAPI app with uvicorn
    import uvicorn
    config = uvicorn.Config(app, host="0.0.0.0", port=port)
    server = uvicorn.Server(config)
    
    logger.info(f"Starting Gradio interface on http://0.0.0.0:{port} with node name '{NODE_NAME}'")
    logger.info("Starting message queue processor")
    
    # Start the server
    await server.serve()

if __name__ == "__main__":
    asyncio.run(main())