import os import json import asyncio import random import math import time from typing import Set, Dict, Optional from fastapi import FastAPI, WebSocket, WebSocketDisconnect from contextlib import asynccontextmanager import aioredis REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0") CRASH_SPEED = 0.08 CRASH_MIN_WAIT = 5.0 CRASH_MAX_WAIT = 10.0 CRASH_RUNNING_TIME = 60.0 CRASH_COOLDOWN = 3.0 app = FastAPI(title="CS2 Simulator WebSocket Server") class RedisPubSub: def __init__(self): self.pub: Optional[aioredis.Redis] = None self.sub: Optional[aioredis.Redis] = None async def connect(self): self.pub = await aioredis.from_url(REDIS_URL, decode_responses=True) self.sub = await aioredis.from_url(REDIS_URL, decode_responses=True) async def publish(self, channel: str, message: dict): if self.pub: await self.pub.publish(channel, json.dumps(message, default=str)) async def subscribe(self, channel: str): if self.sub: psub = self.sub.pubsub() await psub.subscribe(channel) return psub return None async def close(self): if self.pub: await self.pub.close() if self.sub: await self.sub.close() redis_pubsub = RedisPubSub() class ConnectionManager: def __init__(self): self.active_connections: Dict[int, Set[WebSocket]] = {} self._lock = asyncio.Lock() async def connect(self, websocket: WebSocket, user_id: int): await websocket.accept() async with self._lock: if user_id not in self.active_connections: self.active_connections[user_id] = set() self.active_connections[user_id].add(websocket) async def disconnect(self, websocket: WebSocket, user_id: int): async with self._lock: if user_id in self.active_connections: self.active_connections[user_id].discard(websocket) if not self.active_connections[user_id]: del self.active_connections[user_id] async def send_personal(self, user_id: int, message: dict): async with self._lock: if user_id in self.active_connections: dead = set() for ws in self.active_connections[user_id]: try: await ws.send_json(message) except Exception: dead.add(ws) self.active_connections[user_id] -= dead if not self.active_connections[user_id]: del self.active_connections[user_id] async def broadcast(self, message: dict): async with self._lock: dead: Dict[int, Set[WebSocket]] = {} for uid, sockets in self.active_connections.items(): for ws in sockets: try: await ws.send_json(message) except Exception: dead.setdefault(uid, set()).add(ws) for uid, dset in dead.items(): self.active_connections[uid] -= dset if not self.active_connections[uid]: del self.active_connections[uid] @property def online_count(self) -> int: return len(self.active_connections) manager = ConnectionManager() class CrashGame: def __init__(self): self.running = False self.current_multiplier = 1.0 self.crash_point = 1.0 self.bets: Dict[int, float] = {} self.cashed_out: Dict[int, float] = {} self.phase = "waiting" self.timer = 0.0 self.round = 0 self.history: list = [] self._start_time = 0.0 def generate_crash_point(self) -> float: r = random.random() cp = 1.0 / max(0.01, 1.0 - r * 0.99) return round(min(cp, 1000.0), 2) def get_multiplier(self, elapsed: float) -> float: return round(math.exp(elapsed * CRASH_SPEED), 2) def to_dict(self): return { "phase": self.phase, "round": self.round, "current_multiplier": self.current_multiplier, "crash_point": self.crash_point, "timer": int(self.timer), "bet_count": len(self.bets), "total_bets": round(sum(self.bets.values()), 2), "cashed_out": {str(k): v for k, v in self.cashed_out.items()}, "history": self.history[-20:], } crash = CrashGame() async def crash_game_loop(): while True: if crash.phase == "waiting" or crash.phase == "cooldown": wait_time = random.uniform(CRASH_MIN_WAIT, CRASH_MAX_WAIT) for i in range(int(wait_time)): crash.timer = wait_time - i await manager.broadcast({"type": "crash_state", **crash.to_dict()}) await asyncio.sleep(1) crash.timer = 0 crash.phase = "betting" crash.cashed_out.clear() await manager.broadcast({"type": "crash_state", **crash.to_dict()}) await asyncio.sleep(5) crash.phase = "running" crash.round += 1 crash.crash_point = crash.generate_crash_point() crash.current_multiplier = 1.0 crash._start_time = time.time() await manager.broadcast({"type": "crash_state", **crash.to_dict()}) if crash.phase == "running": while True: elapsed = time.time() - crash._start_time mult = crash.get_multiplier(elapsed) crash.current_multiplier = mult await manager.broadcast({"type": "crash_state", **crash.to_dict()}) if mult >= crash.crash_point: crash.phase = "crashed" crash.history.append({ "round": crash.round, "crash_point": crash.crash_point, "timestamp": time.time(), }) if len(crash.history) > 100: crash.history = crash.history[-100:] await manager.broadcast({ "type": "crash_crashed", "crash_point": crash.crash_point, "round": crash.round, }) crash.bets.clear() crash.phase = "cooldown" break if mult > 2.0: await asyncio.sleep(0.15) elif mult > 5.0: await asyncio.sleep(0.08) else: await asyncio.sleep(0.2) await asyncio.sleep(CRASH_COOLDOWN) async def listen_activity(): psub = await redis_pubsub.subscribe("activity") if psub: async for msg in psub.listen(): if msg["type"] == "message": data = json.loads(msg["data"]) await manager.broadcast(data) @asynccontextmanager async def lifespan(app_instance): await redis_pubsub.connect() asyncio.create_task(crash_game_loop()) asyncio.create_task(listen_activity()) yield await redis_pubsub.close() app.router.lifespan_context = lifespan @app.websocket("/ws/{user_id}") async def websocket_endpoint(websocket: WebSocket, user_id: int): await manager.connect(websocket, user_id) await websocket.send_json({"type": "connected", "user_id": user_id}) await websocket.send_json({"type": "crash_state", **crash.to_dict()}) try: while True: data = await websocket.receive_json() msg_type = data.get("type") if msg_type == "crash_bet": amount = float(data.get("amount", 0)) if amount > 0 and crash.phase == "betting": crash.bets[user_id] = crash.bets.get(user_id, 0) + amount await manager.broadcast({"type": "crash_state", **crash.to_dict()}) elif msg_type == "crash_cashout": if crash.phase == "running" and user_id in crash.bets: bet = crash.bets.pop(user_id) crash.cashed_out[user_id] = crash.current_multiplier payout = round(bet * crash.current_multiplier, 2) await redis_pubsub.publish("crash_payout", { "user_id": user_id, "bet": bet, "multiplier": crash.current_multiplier, "payout": payout, }) await manager.broadcast({"type": "crash_state", **crash.to_dict()}) elif msg_type == "ping": await websocket.send_json({"type": "pong"}) except WebSocketDisconnect: await manager.disconnect(websocket, user_id) crash.bets.pop(user_id, None) await manager.broadcast({"type": "crash_state", **crash.to_dict()}) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=13670)