İlk commit: Stepstead backend (FastAPI)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
commit
ea049b8ccd
45 changed files with 6751 additions and 0 deletions
280
routers/ws.py
Normal file
280
routers/ws.py
Normal file
|
|
@ -0,0 +1,280 @@
|
|||
"""
|
||||
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})
|
||||
Loading…
Add table
Add a link
Reference in a new issue