""" WebSocket multiplayer — canlı oyuncu pozisyonları. Mimari: uvicorn 2 worker ile çalıştığı için bağlantılar süreçlere dağılır; eventler Redis pub/sub kanalı üzerinden tüm worker'lara yayılır, her worker kendi yerel bağlantılarına iletir. Online varlık (presence) Redis'te TTL'li key olarak tutulur (heartbeat ile yenilenir). Event türleri (JSON, hepsinde "type" alanı): - online_list → sadece yeni bağlanana: {"players": [player, ...]} - chat_history → sadece yeni bağlanana: {"messages": [chat, ...]} (son 30, eskiden yeniye) - player_joined / player_moved → player payload (char_id, name, level, chunk_x, chunk_y, x, y) - player_left → {"char_id": int} - chat → {"char_id", "name", "text", "ts", "channel", "location_code?"} (gönderene de gider) - pong → ping yanıtı İstemciden gelenler: - {"type":"ping"} - {"type":"chat","text":"...","channel":"global"|"local"} (local = sadece aynı lokasyondakilere) - {"type":"get_local_history"} → bulunduğun lokasyonun son mesajları (chat_history, channel=local) """ import asyncio import json import logging import time from datetime import datetime, timezone from fastapi import APIRouter, WebSocket, WebSocketDisconnect from sqlalchemy import select, text from cache import redis_client from models import AsyncSessionLocal, Character from security import decode_token log = logging.getLogger("stepstead.ws") router = APIRouter(tags=["ws"]) CHANNEL = "ss:world_events" ONLINE_KEY_PREFIX = "ss:online:" ONLINE_TTL = 120 # saniye; istemci ping'i (25sn) ile yenilenir CHAT_CHANNEL = "global" CHAT_MAX_LEN = 300 CHAT_HISTORY_LIMIT = 30 CHAT_MIN_INTERVAL = 1.0 # saniye — basit flood koruması # Bu event türleri üreticisine geri gönderilmez (chat gönderene de gider) NO_ECHO_TYPES = {"player_joined", "player_moved", "player_left"} def player_payload(char: Character, location_code: str | None) -> dict: return { "char_id": char.id, "name": char.name, "level": char.level, "location_code": location_code, } async def publish_event(event: dict) -> None: try: await redis_client.publish(CHANNEL, json.dumps(event)) except Exception: log.exception("Redis publish başarısız") async def set_online(char_id: int, data: dict) -> None: try: await redis_client.set(ONLINE_KEY_PREFIX + str(char_id), json.dumps(data), ex=ONLINE_TTL) except Exception: log.exception("Online key yazılamadı") class Hub: """Bu worker'a bağlı WebSocket'ler + Redis pub/sub dinleyicisi.""" def __init__(self) -> None: self.local: dict[int, WebSocket] = {} # char_id → lokasyon kodu (yerel chat filtrelemesi için; player_moved ile güncellenir) self.locs: dict[int, str | None] = {} self._listener: asyncio.Task | None = None def ensure_listener(self) -> None: if self._listener is None or self._listener.done(): self._listener = asyncio.create_task(self._listen()) async def _listen(self) -> None: pubsub = redis_client.pubsub() await pubsub.subscribe(CHANNEL) try: async for msg in pubsub.listen(): if msg["type"] != "message": continue await self._broadcast_local(msg["data"]) except Exception: log.exception("pub/sub dinleyici düştü") finally: await pubsub.aclose() async def _broadcast_local(self, raw: str) -> None: # Pozisyon eventlerini üreten oyuncuya geri gönderme (echo önleme) origin: int | None = None local_chat_loc: str | None = None try: evt = json.loads(raw) etype = evt.get("type") if etype in NO_ECHO_TYPES: origin = evt.get("char_id") # Seyahat eden oyuncunun lokasyonunu takip et (yerel chat için) if etype == "player_moved" and evt.get("char_id") in self.locs: self.locs[evt["char_id"]] = evt.get("location_code") if etype == "chat" and evt.get("channel") == "local": local_chat_loc = evt.get("location_code") except (json.JSONDecodeError, AttributeError): pass dead: list[int] = [] for cid, ws in list(self.local.items()): if cid == origin: continue # Yerel chat sadece aynı lokasyondakilere gider if local_chat_loc is not None and self.locs.get(cid) != local_chat_loc: continue try: await ws.send_text(raw) except Exception: dead.append(cid) for cid in dead: self.local.pop(cid, None) hub = Hub() async def _chat_history(channel_key: str = CHAT_CHANNEL) -> list[dict]: async with AsyncSessionLocal() as db: rows = (await db.execute(text(""" SELECT m.sender_id, c.name, m.text, m.ts FROM chat_messages m JOIN characters c ON c.id = m.sender_id WHERE m.channel = :ch ORDER BY m.ts DESC LIMIT :lim """), {"ch": channel_key, "lim": CHAT_HISTORY_LIMIT})).all() return [ {"char_id": r[0], "name": r[1], "text": r[2], "ts": r[3].isoformat()} for r in reversed(rows) ] async def _save_chat(char_id: int, chat_text: str, channel_key: str = CHAT_CHANNEL) -> None: async with AsyncSessionLocal() as db: await db.execute( text("INSERT INTO chat_messages (channel, sender_id, text) VALUES (:ch, :sid, :t)"), {"ch": channel_key, "sid": char_id, "t": chat_text}, ) await db.commit() @router.websocket("/ws") async def world_ws(ws: WebSocket): # Auth: token query param (WebSocket'te Authorization header her istemcide desteklenmez) token = ws.query_params.get("token", "") try: payload = decode_token(token) user_id = int(payload["sub"]) except Exception: await ws.close(code=4401) return async with AsyncSessionLocal() as db: char = (await db.execute( select(Character).where(Character.user_id == user_id) )).scalar_one_or_none() loc_code = None if char is not None and char.location_id is not None: row = (await db.execute( text("SELECT code FROM locations WHERE id = :i"), {"i": char.location_id} )).first() loc_code = row[0] if row else None if char is None: await ws.close(code=4404) return await ws.accept() hub.ensure_listener() hub.local[char.id] = ws hub.locs[char.id] = loc_code me = player_payload(char, loc_code) # Mevcut online oyuncular sadece bu istemciye online: list[dict] = [] try: async for key in redis_client.scan_iter(match=ONLINE_KEY_PREFIX + "*"): raw = await redis_client.get(key) if not raw: continue try: p = json.loads(raw) except json.JSONDecodeError: continue if p.get("char_id") != char.id: online.append(p) except Exception: log.exception("Online liste okunamadı") await ws.send_text(json.dumps({"type": "online_list", "players": online})) try: history = await _chat_history() except Exception: log.exception("Chat history okunamadı") history = [] await ws.send_text(json.dumps({"type": "chat_history", "channel": "global", "messages": history})) await set_online(char.id, me) await publish_event({"type": "player_joined", **me}) last_chat = 0.0 try: while True: raw = await ws.receive_text() try: msg = json.loads(raw) except json.JSONDecodeError: continue msg_type = msg.get("type") if msg_type == "ping": # TTL yenile; key düştüyse (worker restart vs.) tekrar yaz refreshed = await redis_client.expire(ONLINE_KEY_PREFIX + str(char.id), ONLINE_TTL) if not refreshed: await set_online(char.id, me) await ws.send_text('{"type": "pong"}') elif msg_type == "chat": now_mono = time.monotonic() if now_mono - last_chat < CHAT_MIN_INTERVAL: continue chat_text = str(msg.get("text", "")).strip()[:CHAT_MAX_LEN] if not chat_text: continue last_chat = now_mono channel = "local" if msg.get("channel") == "local" else "global" cur_loc = hub.locs.get(char.id) if channel == "local" and not cur_loc: continue channel_key = f"loc:{cur_loc}" if channel == "local" else CHAT_CHANNEL try: await _save_chat(char.id, chat_text, channel_key) except Exception: log.exception("Chat kaydedilemedi") continue event = { "type": "chat", "channel": channel, "char_id": char.id, "name": char.name, "text": chat_text, "ts": datetime.now(timezone.utc).isoformat(), } if channel == "local": event["location_code"] = cur_loc await publish_event(event) elif msg_type == "get_local_history": cur_loc = hub.locs.get(char.id) messages: list[dict] = [] if cur_loc: try: messages = await _chat_history(f"loc:{cur_loc}") except Exception: log.exception("Yerel chat history okunamadı") await ws.send_text(json.dumps( {"type": "chat_history", "channel": "local", "messages": messages})) except WebSocketDisconnect: pass except Exception: log.exception("WS döngü hatası (char_id=%s)", char.id) finally: hub.local.pop(char.id, None) hub.locs.pop(char.id, None) try: await redis_client.delete(ONLINE_KEY_PREFIX + str(char.id)) except Exception: pass await publish_event({"type": "player_left", "char_id": char.id})