280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""
|
||
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})
|