from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Dict, List, Optional from uuid import uuid4 import asyncio import base64 import binascii import contextlib import json import logging import mimetypes import os import httpx import uvicorn from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from fastapi.staticfiles import StaticFiles def load_env_file(path: str = ".env") -> None: env_path = Path(path) if not env_path.exists(): return for raw_line in env_path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'")) load_env_file() logging.basicConfig(level=logging.INFO) logger = logging.getLogger("trichat") MAX_HISTORY_MESSAGES = 100 MAX_FILE_SIZE = 5 * 1024 * 1024 ROOM_HISTORY_HOURS = int(os.getenv("ROOM_HISTORY_HOURS", "5")) HISTORY_CACHE_SECONDS = int(os.getenv("HISTORY_CACHE_SECONDS", "30")) CLEANUP_INTERVAL_SECONDS = int(os.getenv("CLEANUP_INTERVAL_SECONDS", "600")) SUPABASE_URL = os.getenv("SUPABASE_URL", "").rstrip("/") SUPABASE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "") SUPABASE_BUCKET = os.getenv("SUPABASE_BUCKET", "chat-files") app = FastAPI(title="Tri-Chat API", description="Temporary anonymous chat rooms") app.mount("/static", StaticFiles(directory="static"), name="static") def now_utc() -> datetime: return datetime.now(timezone.utc) def utc_now() -> str: return now_utc().isoformat() def expiry_time() -> str: return (now_utc() + timedelta(hours=ROOM_HISTORY_HOURS)).isoformat() def clean_text(value: object, limit: int, default: str = "") -> str: if not isinstance(value, str): return default return value.strip()[:limit] def safe_file_name(file_name: str) -> str: cleaned = Path(file_name).name.strip()[:100] return cleaned or "upload.bin" def storage_path_for(room: str, file_name: str) -> str: room_prefix = "".join(char if char.isalnum() or char in ("-", "_") else "_" for char in room) return f"{room_prefix}/{uuid4().hex}-{safe_file_name(file_name)}" def parse_iso_datetime(value: str) -> Optional[datetime]: if not value: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None def is_expired(message: dict) -> bool: expires_at = parse_iso_datetime(message.get("expiresAt", "")) return bool(expires_at and expires_at <= now_utc()) class SupabaseStore: def __init__(self, url: str, key: str, bucket: str): self.url = url self.key = key self.bucket = bucket @property def enabled(self) -> bool: return bool(self.url and self.key) @property def headers(self) -> Dict[str, str]: return { "apikey": self.key, "Authorization": f"Bearer {self.key}", } async def fetch_history(self, room: str) -> List[dict]: if not self.enabled: return [] params = { "select": "*", "room": f"eq.{room}", "expires_at": f"gt.{utc_now()}", "order": "created_at.desc", "limit": str(MAX_HISTORY_MESSAGES), } async with httpx.AsyncClient(timeout=10) as client: response = await client.get( f"{self.url}/rest/v1/messages", headers=self.headers, params=params, ) response.raise_for_status() rows = list(reversed(response.json())) return [self.row_to_message(row) for row in rows] async def save_message(self, message: dict) -> None: if not self.enabled or message.get("type") == "system": return payload = { "room": message["room"], "username": message["username"], "message_type": message["type"], "text": message.get("text"), "file_url": message.get("fileUrl"), "file_path": message.get("filePath"), "file_name": message.get("fileName"), "file_type": message.get("fileType"), "file_size": message.get("fileSize"), "created_at": message["timestamp"], "expires_at": message["expiresAt"], } async with httpx.AsyncClient(timeout=10) as client: response = await client.post( f"{self.url}/rest/v1/messages", headers={**self.headers, "Content-Type": "application/json"}, json=payload, ) response.raise_for_status() async def upload_file(self, room: str, file_name: str, file_type: str, file_bytes: bytes) -> tuple[str, str]: if not self.enabled: raise RuntimeError("Supabase is not configured") path = storage_path_for(room, file_name) content_type = file_type or mimetypes.guess_type(file_name)[0] or "application/octet-stream" async with httpx.AsyncClient(timeout=30) as client: response = await client.post( f"{self.url}/storage/v1/object/{self.bucket}/{path}", headers={ **self.headers, "Content-Type": content_type, "x-upsert": "false", }, content=file_bytes, ) response.raise_for_status() public_url = f"{self.url}/storage/v1/object/public/{self.bucket}/{path}" return public_url, path async def delete_storage_objects(self, paths: List[str]) -> None: if not self.enabled or not paths: return async with httpx.AsyncClient(timeout=30) as client: response = await client.request( "DELETE", f"{self.url}/storage/v1/object/{self.bucket}", headers={**self.headers, "Content-Type": "application/json"}, json={"prefixes": paths}, ) response.raise_for_status() async def cleanup_expired(self) -> int: if not self.enabled: return 0 current_time = utc_now() async with httpx.AsyncClient(timeout=30) as client: select_response = await client.get( f"{self.url}/rest/v1/messages", headers=self.headers, params={ "select": "id,file_path", "expires_at": f"lte.{current_time}", "limit": "500", }, ) select_response.raise_for_status() expired_rows = select_response.json() if not expired_rows: return 0 file_paths = [row["file_path"] for row in expired_rows if row.get("file_path")] if file_paths: await self.delete_storage_objects(file_paths) delete_response = await client.delete( f"{self.url}/rest/v1/messages", headers=self.headers, params={"expires_at": f"lte.{current_time}"}, ) delete_response.raise_for_status() return len(expired_rows) def row_to_message(self, row: dict) -> dict: message = { "type": row["message_type"], "username": row["username"], "timestamp": row["created_at"], "expiresAt": row["expires_at"], "room": row["room"], } if row["message_type"] == "text": message["text"] = row.get("text") or "" elif row["message_type"] == "file": message.update( { "fileUrl": row.get("file_url") or "", "fileName": row.get("file_name") or "download", "fileType": row.get("file_type") or "application/octet-stream", "fileSize": row.get("file_size") or 0, } ) return message store = SupabaseStore(SUPABASE_URL, SUPABASE_KEY, SUPABASE_BUCKET) class ConnectionManager: def __init__(self): self.active_connections: Dict[str, List[Dict]] = {} self.fallback_history: Dict[str, List[Dict]] = {} self.history_cache: Dict[str, Dict] = {} async def connect(self, websocket: WebSocket, room: str, username: str): await websocket.accept() if room not in self.active_connections: self.active_connections[room] = [] self.fallback_history[room] = [] self.active_connections[room].append( { "websocket": websocket, "username": username, "joined_at": utc_now(), } ) for message in await self.get_history(room): await websocket.send_text(json.dumps(message)) await self.broadcast_to_room( room, { "type": "system", "message": f"{username} joined the room", "timestamp": utc_now(), "room": room, }, persist=False, ) await self.broadcast_user_list(room) async def get_history(self, room: str) -> List[dict]: self.prune_fallback_history(room) cached = self.history_cache.get(room) if cached and cached["expires_at"] > now_utc(): return cached["messages"] if store.enabled: try: messages = await store.fetch_history(room) self.cache_history(room, messages) return messages except httpx.HTTPError as exc: logger.warning("Could not fetch Supabase history: %s", exc) messages = self.fallback_history.get(room, []) self.cache_history(room, messages) return messages def cache_history(self, room: str, messages: List[dict]) -> None: self.history_cache[room] = { "messages": [message for message in messages if not is_expired(message)], "expires_at": now_utc() + timedelta(seconds=HISTORY_CACHE_SECONDS), } def append_to_cache(self, room: str, message: dict) -> None: cached = self.history_cache.get(room) if not cached: return cached["messages"].append(message) cached["messages"] = cached["messages"][-MAX_HISTORY_MESSAGES:] def prune_fallback_history(self, room: str) -> None: if room in self.fallback_history: self.fallback_history[room] = [ message for message in self.fallback_history[room] if not is_expired(message) ][-MAX_HISTORY_MESSAGES:] def disconnect(self, websocket: WebSocket, room: str) -> Optional[str]: if room not in self.active_connections: return None for conn in list(self.active_connections[room]): if conn["websocket"] == websocket: self.active_connections[room].remove(conn) username = conn["username"] if not self.active_connections[room]: self.active_connections.pop(room, None) return username return None async def broadcast_to_room(self, room: str, message: dict, persist: bool = True): if persist: if store.enabled: try: await store.save_message(message) except httpx.HTTPError as exc: logger.warning("Could not save message to Supabase: %s", exc) self.add_fallback_message(room, message) else: self.add_fallback_message(room, message) self.append_to_cache(room, message) if room not in self.active_connections: return disconnected = [] for connection_info in list(self.active_connections[room]): try: await connection_info["websocket"].send_text(json.dumps(message)) except RuntimeError: disconnected.append(connection_info) for conn in disconnected: if conn in self.active_connections.get(room, []): self.active_connections[room].remove(conn) def add_fallback_message(self, room: str, message: dict) -> None: self.fallback_history.setdefault(room, []).append(message) self.prune_fallback_history(room) async def broadcast_user_list(self, room: str): if room not in self.active_connections: return users_message = { "type": "user_list", "users": self.get_room_users(room), "room": room, } disconnected = [] for connection_info in list(self.active_connections[room]): try: await connection_info["websocket"].send_text(json.dumps(users_message)) except RuntimeError: disconnected.append(connection_info) for conn in disconnected: if conn in self.active_connections.get(room, []): self.active_connections[room].remove(conn) def get_room_users(self, room: str) -> List[str]: if room not in self.active_connections: return [] return [conn["username"] for conn in self.active_connections[room]] def clear_cache(self) -> None: self.history_cache.clear() manager = ConnectionManager() cleanup_task: Optional[asyncio.Task] = None async def cleanup_loop() -> None: while True: try: deleted_count = await store.cleanup_expired() if deleted_count: manager.clear_cache() logger.info("Deleted %s expired messages/files", deleted_count) except httpx.HTTPError as exc: logger.warning("Expired cleanup failed: %s", exc) await asyncio.sleep(CLEANUP_INTERVAL_SECONDS) @app.on_event("startup") async def startup_event(): global cleanup_task if store.enabled: cleanup_task = asyncio.create_task(cleanup_loop()) logger.info("Temporary cleanup is running every %s seconds", CLEANUP_INTERVAL_SECONDS) else: logger.warning("Supabase is not configured. Using in-memory fallback only.") @app.on_event("shutdown") async def shutdown_event(): if cleanup_task: cleanup_task.cancel() with contextlib.suppress(asyncio.CancelledError): await cleanup_task @app.get("/", response_class=HTMLResponse) async def get_chat_page(): try: with open("templates/index.html", "r", encoding="utf-8") as f: return HTMLResponse(content=f.read()) except FileNotFoundError: return HTMLResponse( content="

Error: templates/index.html not found

", status_code=404, ) @app.websocket("/ws/{room}") async def websocket_endpoint(websocket: WebSocket, room: str, username: str): username = clean_text(username, 20) room = clean_text(room, 30, "global") or "global" if not username: await websocket.close(code=1008, reason="Username is required") return await manager.connect(websocket, room, username) try: while True: data = await websocket.receive_text() try: message_data = json.loads(data) except json.JSONDecodeError: await websocket.send_text(json.dumps({"type": "error", "message": "Invalid message"})) continue message_type = message_data.get("type") if message_type == "text": await handle_text_message(room, username, message_data) elif message_type == "file": await handle_file_message(websocket, room, username, message_data) except WebSocketDisconnect: disconnected_username = manager.disconnect(websocket, room) if disconnected_username: await manager.broadcast_to_room( room, { "type": "system", "message": f"{disconnected_username} left the room", "timestamp": utc_now(), "room": room, }, persist=False, ) await manager.broadcast_user_list(room) async def handle_text_message(room: str, username: str, message_data: dict): text_content = clean_text(message_data.get("text"), 500) if not text_content: return await manager.broadcast_to_room( room, { "type": "text", "username": username, "text": text_content, "timestamp": utc_now(), "expiresAt": expiry_time(), "room": room, }, ) async def handle_file_message(websocket: WebSocket, room: str, username: str, message_data: dict): file_name = safe_file_name(clean_text(message_data.get("fileName"), 100, "upload.bin")) file_type = clean_text(message_data.get("fileType"), 120, "application/octet-stream") file_data = message_data.get("fileData", "") if not store.enabled: await websocket.send_text(json.dumps({"type": "error", "message": "File uploads need Supabase configured"})) return try: file_bytes = base64.b64decode(file_data, validate=True) except (binascii.Error, TypeError): await websocket.send_text(json.dumps({"type": "error", "message": "Invalid file data"})) return if len(file_bytes) > MAX_FILE_SIZE: await websocket.send_text(json.dumps({"type": "error", "message": "File size exceeds 5MB limit"})) return try: file_url, file_path = await store.upload_file(room, file_name, file_type, file_bytes) except httpx.HTTPError as exc: logger.warning("File upload failed: %s", exc) await websocket.send_text(json.dumps({"type": "error", "message": "File upload failed"})) return await manager.broadcast_to_room( room, { "type": "file", "username": username, "fileName": file_name, "fileType": file_type, "fileSize": len(file_bytes), "fileUrl": file_url, "filePath": file_path, "timestamp": utc_now(), "expiresAt": expiry_time(), "room": room, }, ) @app.get("/api/rooms") async def get_active_rooms(): rooms = [] for room_name, connections in manager.active_connections.items(): if connections: rooms.append( { "name": room_name, "user_count": len(connections), "users": [conn["username"] for conn in connections], } ) return {"rooms": rooms} @app.get("/api/rooms/{room}/users") async def get_room_users(room: str): users = manager.get_room_users(room) return { "room": room, "users": users, "user_count": len(users), } if __name__ == "__main__": port = int(os.getenv("PORT", 7860)) uvicorn.run(app, host="0.0.0.0", port=port)