# -*- coding: utf-8 -*-
"""
OK ITHAKI V249383 - V411 Live Wall Reactions
Additive module. It does not alter the existing SQLite database, media,
messages, calls, posts, or scheduled publishing.

Persistent V411 reaction events are stored in:
    data/v411_wall_reactions.jsonl
The file is append-only. Device identifiers are never stored directly.
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import os
import secrets
import threading
import time
from collections import defaultdict
from pathlib import Path
from typing import Any

from fastapi import HTTPException, Query, WebSocket, WebSocketDisconnect

V411_VERSION = "V249383-v411-live-wall-reactions"
ALLOWED_REACTIONS = ("❤️", "🤝", "🌟", "🫶", "😮", "🙏")
REACTION_LABELS = {
    "❤️": "Το νιώθω",
    "🤝": "Σε καταλαβαίνω",
    "🌟": "Δύναμη",
    "🫶": "Μαζί σου",
    "😮": "Έκπληξη",
    "🙏": "Ελπίδα",
}


def _clean(value: Any, maximum: int) -> str:
    return str(value or "").strip()[:maximum]


class V411ReactionStore:
    """Append-only, database-independent reaction store."""

    def __init__(self, root: Path):
        self.data_dir = root / "data"
        self.data_dir.mkdir(parents=True, exist_ok=True)
        self.events_path = self.data_dir / "v411_wall_reactions.jsonl"
        self.salt_path = self.data_dir / ".v411_reaction_salt"
        self._lock = threading.RLock()
        self._state: dict[tuple[str, str, str, str], bool] = {}
        self._loaded_signature: tuple[int, int] = (-1, -1)
        self._rate: dict[str, float] = {}
        self._salt = self._load_or_create_salt()
        self._reload(force=True)

    def _load_or_create_salt(self) -> bytes:
        try:
            data = self.salt_path.read_bytes().strip()
            if len(data) >= 24:
                return data
        except FileNotFoundError:
            pass
        data = secrets.token_bytes(32)
        try:
            fd = os.open(str(self.salt_path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            with os.fdopen(fd, "wb") as handle:
                handle.write(data)
                handle.flush()
                os.fsync(handle.fileno())
            return data
        except FileExistsError:
            existing = self.salt_path.read_bytes().strip()
            return existing if existing else data

    def actor_hash(self, device_id: str) -> str:
        raw = _clean(device_id, 256).encode("utf-8", "ignore")
        return hashlib.sha256(self._salt + b":" + raw).hexdigest()

    def _signature(self) -> tuple[int, int]:
        try:
            stat = self.events_path.stat()
            return int(stat.st_size), int(stat.st_mtime_ns)
        except FileNotFoundError:
            return (0, 0)

    def _reload(self, force: bool = False) -> None:
        signature = self._signature()
        if not force and signature == self._loaded_signature:
            return
        state: dict[tuple[str, str, str, str], bool] = {}
        if signature[0] > 0:
            try:
                with self.events_path.open("r", encoding="utf-8", errors="replace") as handle:
                    for line in handle:
                        try:
                            event = json.loads(line)
                            room_id = _clean(event.get("room_id"), 128)
                            post_id = _clean(event.get("post_id"), 180)
                            actor = _clean(event.get("actor"), 80)
                            emoji = _clean(event.get("emoji"), 8)
                            if room_id and post_id and actor and emoji in ALLOWED_REACTIONS:
                                state[(room_id, post_id, actor, emoji)] = bool(event.get("active"))
                        except Exception:
                            continue
            except FileNotFoundError:
                pass
        self._state = state
        self._loaded_signature = signature

    def _append(self, event: dict[str, Any]) -> None:
        line = json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n"
        self.events_path.parent.mkdir(parents=True, exist_ok=True)
        with self.events_path.open("a", encoding="utf-8") as handle:
            try:
                import fcntl
                fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
            except Exception:
                fcntl = None
            handle.write(line)
            handle.flush()
            os.fsync(handle.fileno())
            if fcntl is not None:
                try:
                    fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
                except Exception:
                    pass
        self._loaded_signature = self._signature()

    def _counts_for(self, room_id: str, post_id: str, actor: str = "") -> dict[str, Any]:
        counts = {emoji: 0 for emoji in ALLOWED_REACTIONS}
        mine: list[str] = []
        for (room, post, who, emoji), active in self._state.items():
            if active and room == room_id and post == post_id:
                counts[emoji] += 1
                if actor and who == actor:
                    mine.append(emoji)
        return {"post_id": post_id, "counts": counts, "mine": mine}

    def list_counts(self, room_id: str, post_ids: list[str], device_id: str = "") -> list[dict[str, Any]]:
        with self._lock:
            self._reload()
            actor = self.actor_hash(device_id) if device_id else ""
            return [self._counts_for(room_id, post_id, actor) for post_id in post_ids]

    def toggle(self, room_id: str, post_id: str, device_id: str, emoji: str) -> dict[str, Any]:
        actor = self.actor_hash(device_id)
        rate_key = f"{actor}:{room_id}:{post_id}"
        with self._lock:
            now_mono = time.monotonic()
            previous = self._rate.get(rate_key, 0.0)
            if now_mono - previous < 0.35:
                raise HTTPException(status_code=429, detail="Πολύ γρήγορες αντιδράσεις.")
            self._rate[rate_key] = now_mono
            self._reload(force=True)
            key = (room_id, post_id, actor, emoji)
            active = not bool(self._state.get(key, False))
            event = {
                "event_id": "v411_" + secrets.token_hex(8),
                "version": V411_VERSION,
                "room_id": room_id,
                "post_id": post_id,
                "actor": actor,
                "emoji": emoji,
                "active": active,
                "created_at": time.time(),
            }
            self._append(event)
            self._state[key] = active
            result = self._counts_for(room_id, post_id, actor)
            result.update({
                "ok": True,
                "active": active,
                "emoji": emoji,
                "event_id": event["event_id"],
                "created_at": event["created_at"],
            })
            return result

    def stats(self) -> dict[str, int]:
        with self._lock:
            self._reload()
            active = sum(1 for value in self._state.values() if value)
            posts = len({(room, post) for (room, post, _, _), value in self._state.items() if value})
            return {"active_reactions": active, "posts_with_reactions": posts}


class V411LiveHub:
    def __init__(self):
        self._rooms: dict[str, set[WebSocket]] = defaultdict(set)
        self._lock = asyncio.Lock()

    async def connect(self, room_id: str, websocket: WebSocket) -> int:
        await websocket.accept()
        async with self._lock:
            self._rooms[room_id].add(websocket)
            return len(self._rooms[room_id])

    async def disconnect(self, room_id: str, websocket: WebSocket) -> int:
        async with self._lock:
            self._rooms[room_id].discard(websocket)
            if not self._rooms[room_id]:
                self._rooms.pop(room_id, None)
                return 0
            return len(self._rooms[room_id])

    async def count(self, room_id: str) -> int:
        async with self._lock:
            return len(self._rooms.get(room_id, set()))

    async def broadcast(self, room_id: str, payload: dict[str, Any]) -> None:
        async with self._lock:
            clients = list(self._rooms.get(room_id, set()))
        dead: list[WebSocket] = []
        for websocket in clients:
            try:
                await websocket.send_json(payload)
            except Exception:
                dead.append(websocket)
        if dead:
            async with self._lock:
                for websocket in dead:
                    self._rooms.get(room_id, set()).discard(websocket)

    async def broadcast_presence(self, room_id: str) -> None:
        await self.broadcast(room_id, {
            "type": "v411_presence",
            "room_id": room_id,
            "live_now": await self.count(room_id),
            "created_at": time.time(),
        })


def install_v411(app: Any, root: Path | str) -> None:
    if getattr(app.state, "ithaki_v411_installed", False):
        return

    root_path = Path(root)
    store = V411ReactionStore(root_path)
    hub = V411LiveHub()
    app.state.ithaki_v411_installed = True
    app.state.ithaki_v411_store = store
    app.state.ithaki_v411_hub = hub

    @app.get("/api/v411/wall/status")
    async def v411_wall_status(room_id: str = Query("ithaki-phone-test")):
        room_id = _clean(room_id, 128) or "ithaki-phone-test"
        return {
            "ok": True,
            "version": V411_VERSION,
            "storage": "append-only-jsonl",
            "existing_database_modified": False,
            "allowed_reactions": [
                {"emoji": emoji, "label": REACTION_LABELS[emoji]}
                for emoji in ALLOWED_REACTIONS
            ],
            "live_now": await hub.count(room_id),
            **store.stats(),
        }

    @app.get("/api/v411/wall/reactions")
    async def v411_wall_reactions(
        room_id: str = Query(...),
        post_ids: str = Query(""),
        device_id: str = Query(""),
    ):
        room = _clean(room_id, 128)
        if not room:
            raise HTTPException(status_code=422, detail="room_id required")
        unique: list[str] = []
        seen: set[str] = set()
        for raw in str(post_ids or "").split(","):
            post_id = _clean(raw, 180)
            if post_id and post_id not in seen:
                seen.add(post_id)
                unique.append(post_id)
            if len(unique) >= 120:
                break
        return {
            "ok": True,
            "version": V411_VERSION,
            "room_id": room,
            "live_now": await hub.count(room),
            "items": store.list_counts(room, unique, _clean(device_id, 256)),
        }

    @app.post("/api/v411/wall/reaction")
    async def v411_wall_reaction(payload: dict[str, Any]):
        room_id = _clean(payload.get("room_id"), 128)
        post_id = _clean(payload.get("post_id"), 180)
        device_id = _clean(payload.get("device_id") or payload.get("owner_token"), 256)
        emoji = _clean(payload.get("emoji"), 8)
        if not room_id or not post_id or not device_id:
            raise HTTPException(status_code=422, detail="room_id, post_id and device_id are required")
        if emoji not in ALLOWED_REACTIONS:
            raise HTTPException(status_code=422, detail="Unsupported reaction")
        result = store.toggle(room_id, post_id, device_id, emoji)
        event = {
            "type": "v411_reaction",
            "version": V411_VERSION,
            "room_id": room_id,
            "post_id": post_id,
            "emoji": emoji,
            "active": result["active"],
            "counts": result["counts"],
            "event_id": result["event_id"],
            "created_at": result["created_at"],
        }
        await hub.broadcast(room_id, event)
        result["live_now"] = await hub.count(room_id)
        result["version"] = V411_VERSION
        return result

    @app.websocket("/ws/v411/wall/{room_id}")
    async def v411_wall_socket(websocket: WebSocket, room_id: str):
        room = _clean(room_id, 128) or "ithaki-phone-test"
        live_now = await hub.connect(room, websocket)
        await websocket.send_json({
            "type": "v411_hello",
            "version": V411_VERSION,
            "room_id": room,
            "live_now": live_now,
            "created_at": time.time(),
        })
        await hub.broadcast_presence(room)
        try:
            while True:
                message = await websocket.receive_text()
                if message.strip().lower() in {"ping", "keepalive"}:
                    await websocket.send_json({
                        "type": "v411_pong",
                        "created_at": time.time(),
                    })
        except WebSocketDisconnect:
            pass
        except Exception:
            pass
        finally:
            await hub.disconnect(room, websocket)
            await hub.broadcast_presence(room)
