pixel/server/world_policy.py

#!/usr/bin/env python3
"""Authoritative world mutation policy for Pixel Island.

The browser may preview local changes, but shared worlds should apply only the
server events produced here. This module focuses on ownership and destructive
mutation safety; rotation and moderation policy remain separate.
"""
from __future__ import annotations

import time
from copy import deepcopy
from typing import Any, Dict, Iterable, List, MutableMapping, Optional, Tuple

ACTIVE = "active"
VIOLATION_HIDDEN = "violation_hidden"

OBJECT_KEYS = {"static": "placed", "dynamic": "dynamicSummons"}
DAY_MS = 10 * 60 * 1000


def now_ms() -> int:
    return int(time.time() * 1000)


def account_id(account: MutableMapping[str, Any] | None) -> str:
    return str((account or {}).get("id") or "")


def is_admin(account: MutableMapping[str, Any] | None) -> bool:
    return bool((account or {}).get("admin") or (account or {}).get("role") == "admin")


def normalize_tombstones(state: MutableMapping[str, Any]) -> MutableMapping[str, Any]:
    tombstones = state.setdefault("tombstones", {})
    tombstones.setdefault("assets", {})
    tombstones.setdefault("objects", {})
    return tombstones


def iter_objects(state: MutableMapping[str, Any]) -> Iterable[Tuple[str, MutableMapping[str, Any]]]:
    for kind, key in OBJECT_KEYS.items():
        for obj in state.get(key) or []:
            if isinstance(obj, MutableMapping) and obj.get("id"):
                yield kind, obj


def find_object(state: MutableMapping[str, Any], object_id: str, kind: Optional[str] = None) -> Tuple[Optional[str], Optional[MutableMapping[str, Any]]]:
    kinds = [kind] if kind in OBJECT_KEYS else list(OBJECT_KEYS)
    for item_kind in kinds:
        for obj in state.get(OBJECT_KEYS[item_kind]) or []:
            if str(obj.get("id") or "") == str(object_id):
                return item_kind, obj
    return None, None


def find_asset(state: MutableMapping[str, Any], asset_id: str) -> Optional[MutableMapping[str, Any]]:
    for asset in state.get("assets") or []:
        if isinstance(asset, MutableMapping) and str(asset.get("id") or "") == str(asset_id):
            return asset
    return None


def owner_of_object(state: MutableMapping[str, Any], obj: MutableMapping[str, Any]) -> str:
    if obj.get("ownerAccountId"):
        return str(obj.get("ownerAccountId"))
    asset = find_asset(state, str(obj.get("assetId") or ""))
    return str((asset or {}).get("ownerAccountId") or "")


def can_modify_object(state: MutableMapping[str, Any], account: MutableMapping[str, Any] | None, object_id: str) -> Tuple[bool, Dict[str, Any]]:
    actor = account_id(account)
    if not actor:
        return False, {"reason": "account_required"}
    kind, obj = find_object(state, object_id)
    if not obj:
        return False, {"reason": "object_not_found"}
    owner = owner_of_object(state, obj)
    if actor == owner or is_admin(account):
        return True, {"reason": None, "kind": kind, "ownerAccountId": owner}
    return False, {"reason": "not_object_owner", "ownerAccountId": owner}


def can_delete_asset(state: MutableMapping[str, Any], account: MutableMapping[str, Any] | None, asset_id: str) -> Tuple[bool, Dict[str, Any]]:
    actor = account_id(account)
    if not actor:
        return False, {"reason": "account_required"}
    asset = find_asset(state, asset_id)
    if not asset:
        return False, {"reason": "asset_not_found"}
    owner = str(asset.get("ownerAccountId") or "")
    return True, {"reason": None, "ownerAccountId": owner}


def is_object_tombstoned(state: MutableMapping[str, Any], object_id: str, version: int = 0) -> bool:
    tomb = normalize_tombstones(state)["objects"].get(str(object_id))
    return bool(tomb and int(tomb.get("version") or 0) >= int(version or 0))


def is_asset_tombstoned(state: MutableMapping[str, Any], asset_id: str, version: int = 0) -> bool:
    tomb = normalize_tombstones(state)["assets"].get(str(asset_id))
    return bool(tomb and int(tomb.get("version") or 0) >= int(version or 0))


def _server_event(event_type: str, actor: str, payload: Dict[str, Any], at: Optional[int] = None) -> Dict[str, Any]:
    stamp = at or now_ms()
    return {
        "serverEventId": f"sev_{stamp}_{event_type.replace('.', '_')}",
        "serverAt": stamp,
        "actorAccountId": actor,
        "type": event_type,
        **payload,
    }


def _accepted(event: Dict[str, Any]) -> Dict[str, Any]:
    return {"accepted": True, "event": event}


def phase_key_for_time(world_time_ms: int, day_ms: int = DAY_MS) -> str:
    progress = (int(world_time_ms) % int(day_ms or DAY_MS)) / float(day_ms or DAY_MS)
    if progress < 0.10:
        return "pre_dawn"
    if progress < 0.20:
        return "sunrise"
    if progress < 0.60:
        return "day"
    if progress < 0.72:
        return "sunset"
    return "night"


def make_world_phase_event(at: Optional[int] = None, day_ms: int = DAY_MS) -> Dict[str, Any]:
    """Create the authoritative day/night clock event emitted by the server."""
    stamp = at or now_ms()
    cycle = int(day_ms or DAY_MS)
    world_time_ms = int(stamp)
    progress = (world_time_ms % cycle) / float(cycle)
    return _server_event(
        "world.phase",
        "server",
        {
            "worldTimeMs": world_time_ms,
            "dayMs": cycle,
            "phase": {"key": phase_key_for_time(world_time_ms, cycle), "progress": progress},
        },
        stamp,
    )


def make_dynamic_move_event(
    state: MutableMapping[str, Any],
    object_id: str,
    x: float,
    y: float,
    target_x: Optional[float] = None,
    target_y: Optional[float] = None,
    at: Optional[int] = None,
    actor: str = "server",
) -> Optional[Dict[str, Any]]:
    """Create an authoritative dynamic-object movement event.

    Clients may ask for a target, but shared-world dynamic positions are only
    changed after this server event is applied.
    """
    stamp = at or now_ms()
    kind, obj = find_object(state, object_id, "dynamic")
    if kind != "dynamic" or not obj:
        return None
    next_object = deepcopy(dict(obj))
    target = {
        "objectId": str(object_id),
        "x": float(x),
        "y": float(y),
        "targetX": float(target_x if target_x is not None else x),
        "targetY": float(target_y if target_y is not None else y),
        "homeX": float(next_object.get("homeX", x)),
        "homeY": float(next_object.get("homeY", y)),
        "serverAt": stamp,
    }
    next_object["serverState"] = target
    next_object["version"] = int(next_object.get("version") or 0) + 1
    return _server_event(
        "dynamic.move",
        actor or "server",
        {
            "objectId": str(object_id),
            "object": next_object,
            "x": target["x"],
            "y": target["y"],
            "targetX": target["targetX"],
            "targetY": target["targetY"],
            "homeX": target["homeX"],
            "homeY": target["homeY"],
            "objectVersion": next_object["version"],
        },
        stamp,
    )


def _rejected(reason: str, **extra: Any) -> Dict[str, Any]:
    return {"accepted": False, "reason": reason, **extra}


def _copy_owned_asset(asset: MutableMapping[str, Any], owner: str) -> Dict[str, Any]:
    copied = deepcopy(dict(asset))
    copied["ownerAccountId"] = owner
    copied.setdefault("author", owner)
    copied.setdefault("version", 1)
    return copied


def _copy_owned_object(obj: MutableMapping[str, Any], owner: str) -> Dict[str, Any]:
    copied = deepcopy(dict(obj))
    copied["ownerAccountId"] = owner
    copied["version"] = int(copied.get("version") or 0) + 1
    copied.setdefault("status", ACTIVE)
    return copied


def apply_command(state: MutableMapping[str, Any], account: MutableMapping[str, Any] | None, command: MutableMapping[str, Any], at: Optional[int] = None) -> Dict[str, Any]:
    """Validate a client command and return a server event without mutating state."""
    actor = account_id(account)
    if not actor:
        return _rejected("account_required")
    command_type = str(command.get("type") or "")

    if command_type == "asset.create":
        asset = command.get("asset")
        if not isinstance(asset, MutableMapping) or not asset.get("id"):
            return _rejected("invalid_asset")
        if is_asset_tombstoned(state, str(asset["id"]), int(asset.get("version") or 1)):
            return _rejected("asset_tombstoned")
        return _accepted(_server_event("asset.upsert", actor, {"asset": _copy_owned_asset(asset, actor)}, at))

    if command_type in {"object.publish", "object.move"}:
        kind = str(command.get("kind") or "static")
        obj = command.get("object")
        if kind not in OBJECT_KEYS or not isinstance(obj, MutableMapping) or not obj.get("id"):
            return _rejected("invalid_object")
        asset = find_asset(state, str(obj.get("assetId") or ""))
        if not asset:
            return _rejected("asset_not_found")
        _, existing = find_object(state, str(obj["id"]), kind)
        if existing:
            allowed, meta = can_modify_object(state, account, str(obj["id"]))
            if not allowed:
                return _rejected(meta["reason"], **{k: v for k, v in meta.items() if k != "reason"})
            owner = owner_of_object(state, existing)
        else:
            if str(asset.get("ownerAccountId") or "") != actor and not is_admin(account):
                return _rejected("not_asset_owner", ownerAccountId=asset.get("ownerAccountId"))
            owner = actor
        next_object = _copy_owned_object(obj, owner)
        if is_object_tombstoned(state, str(next_object["id"]), int(next_object.get("version") or 1)):
            return _rejected("object_tombstoned")
        return _accepted(_server_event("object.upsert", actor, {"kind": kind, "object": next_object, "objectVersion": next_object["version"]}, at))

    if command_type == "object.delete":
        object_id = str(command.get("objectId") or "")
        kind = str(command.get("kind") or "")
        allowed, meta = can_modify_object(state, account, object_id)
        if not allowed:
            return _rejected(meta["reason"], **{k: v for k, v in meta.items() if k != "reason"})
        found_kind, obj = find_object(state, object_id, kind if kind in OBJECT_KEYS else None)
        version = int((obj or {}).get("version") or 0) + 1
        tombstone = {"id": object_id, "deletedAt": at or now_ms(), "deletedBy": actor, "version": version}
        return _accepted(_server_event("object.delete", actor, {"kind": found_kind, "objectId": object_id, "objectVersion": version, "tombstone": tombstone}, at))

    if command_type == "asset.delete":
        asset_id = str(command.get("assetId") or "")
        allowed, meta = can_delete_asset(state, account, asset_id)
        if not allowed:
            return _rejected(meta["reason"], **{k: v for k, v in meta.items() if k != "reason"})
        asset = find_asset(state, asset_id) or {}
        version = int(asset.get("version") or 0) + 1
        object_tombstones: List[Dict[str, Any]] = []
        for _, obj in iter_objects(state):
            if str(obj.get("assetId") or "") == asset_id:
                object_tombstones.append({"id": str(obj["id"]), "deletedAt": at or now_ms(), "deletedBy": actor, "version": int(obj.get("version") or 0) + 1})
        tombstone = {"id": asset_id, "deletedAt": at or now_ms(), "deletedBy": actor, "version": version}
        return _accepted(_server_event("asset.delete", actor, {"assetId": asset_id, "assetVersion": version, "tombstone": tombstone, "objectTombstones": object_tombstones}, at))

    if command_type == "admin.hide_violation":
        if not is_admin(account):
            return _rejected("admin_required")
        object_id = str(command.get("objectId") or "")
        kind, obj = find_object(state, object_id)
        if not obj:
            return _rejected("object_not_found")
        next_object = deepcopy(dict(obj))
        next_object["status"] = VIOLATION_HIDDEN
        next_object["permanentHidden"] = True
        next_object["hiddenReason"] = "moderation_violation"
        next_object["version"] = int(next_object.get("version") or 0) + 1
        return _accepted(_server_event("object.upsert", actor, {"kind": kind, "object": next_object, "objectVersion": next_object["version"]}, at))

    return _rejected("unknown_command")


def apply_server_event(state: MutableMapping[str, Any], event: MutableMapping[str, Any]) -> bool:
    """Apply only server-issued events. Returns True when state changed."""
    if not event or not event.get("serverEventId") or not event.get("serverAt"):
        return False
    normalize_tombstones(state)
    event_type = str(event.get("type") or "")

    if event_type == "world.phase":
        clock = state.setdefault("serverSync", {}).setdefault("clock", {})
        day_ms = int(event.get("dayMs") or DAY_MS)
        world_time_ms = int(event.get("worldTimeMs") or event.get("serverNow") or event.get("serverAt") or now_ms())
        clock.clear()
        clock.update({
            "worldTimeMs": world_time_ms,
            "syncedAt": int(event.get("serverAt") or now_ms()),
            "dayMs": day_ms,
            "phase": event.get("phase") or {"key": phase_key_for_time(world_time_ms, day_ms), "progress": (world_time_ms % day_ms) / float(day_ms)},
            "serverEventId": event["serverEventId"],
        })
        state["serverSync"]["lastServerEventId"] = event["serverEventId"]
        return True

    if event_type == "dynamic.move":
        object_id = str(event.get("objectId") or event.get("dynamicId") or "")
        obj = event.get("object")
        if not object_id and isinstance(obj, MutableMapping):
            object_id = str(obj.get("id") or "")
        if not object_id:
            return False
        object_version = int((obj or {}).get("version") or event.get("objectVersion") or 1) if isinstance(obj, MutableMapping) else int(event.get("objectVersion") or 1)
        if is_object_tombstoned(state, object_id, object_version):
            return False
        target = {
            "objectId": object_id,
            "x": float(event.get("x", (obj or {}).get("serverState", {}).get("x", (obj or {}).get("homeX", 0))) if isinstance(obj, MutableMapping) else event.get("x", 0)),
            "y": float(event.get("y", (obj or {}).get("serverState", {}).get("y", (obj or {}).get("homeY", 0))) if isinstance(obj, MutableMapping) else event.get("y", 0)),
            "targetX": float(event.get("targetX", event.get("x", 0))),
            "targetY": float(event.get("targetY", event.get("y", 0))),
            "homeX": float(event.get("homeX", (obj or {}).get("homeX", 0)) if isinstance(obj, MutableMapping) else event.get("homeX", 0)),
            "homeY": float(event.get("homeY", (obj or {}).get("homeY", 0)) if isinstance(obj, MutableMapping) else event.get("homeY", 0)),
            "serverAt": int(event.get("serverAt") or now_ms()),
        }
        server_sync = state.setdefault("serverSync", {})
        server_sync.setdefault("dynamicTargets", {})[object_id] = target
        if isinstance(obj, MutableMapping) and obj.get("id"):
            rows = state.setdefault("dynamicSummons", [])
            next_object = deepcopy(dict(obj))
            next_object["serverState"] = target
            index = next((i for i, item in enumerate(rows) if str(item.get("id") or "") == object_id), -1)
            if index >= 0:
                rows[index] = next_object
            else:
                rows.append(next_object)
        server_sync["lastServerEventId"] = event["serverEventId"]
        return True

    if event_type == "asset.upsert":
        asset = event.get("asset")
        if not isinstance(asset, MutableMapping) or not asset.get("id"):
            return False
        if is_asset_tombstoned(state, str(asset["id"]), int(asset.get("version") or 1)):
            return False
        assets = state.setdefault("assets", [])
        index = next((i for i, item in enumerate(assets) if item.get("id") == asset.get("id")), -1)
        if index >= 0:
            assets[index] = deepcopy(dict(asset))
        else:
            assets.insert(0, deepcopy(dict(asset)))
        state.setdefault("serverSync", {})["lastServerEventId"] = event["serverEventId"]
        return True

    if event_type == "object.upsert":
        obj = event.get("object")
        kind = str(event.get("kind") or "static")
        if kind not in OBJECT_KEYS or not isinstance(obj, MutableMapping) or not obj.get("id"):
            return False
        if is_object_tombstoned(state, str(obj["id"]), int(obj.get("version") or event.get("objectVersion") or 1)):
            return False
        rows = state.setdefault(OBJECT_KEYS[kind], [])
        index = next((i for i, item in enumerate(rows) if item.get("id") == obj.get("id")), -1)
        if index >= 0:
            rows[index] = deepcopy(dict(obj))
        else:
            rows.append(deepcopy(dict(obj)))
        state.setdefault("serverSync", {})["lastServerEventId"] = event["serverEventId"]
        return True

    if event_type == "object.delete":
        object_id = str(event.get("objectId") or "")
        kind = str(event.get("kind") or "")
        if not object_id or kind not in OBJECT_KEYS:
            return False
        tombstone = event.get("tombstone") or {"id": object_id, "deletedAt": event["serverAt"], "deletedBy": event.get("actorAccountId"), "version": event.get("objectVersion") or 1}
        state["tombstones"]["objects"][object_id] = tombstone
        state[OBJECT_KEYS[kind]] = [obj for obj in state.get(OBJECT_KEYS[kind]) or [] if str(obj.get("id") or "") != object_id]
        state.setdefault("objectVotes", {}).pop(object_id, None)
        state.setdefault("hiddenObjects", {}).pop(object_id, None)
        state.setdefault("serverSync", {}).setdefault("dynamicTargets", {}).pop(object_id, None)
        state["moderationReports"] = [report for report in state.get("moderationReports") or [] if report.get("objectId") != object_id]
        state.setdefault("serverSync", {})["lastServerEventId"] = event["serverEventId"]
        return True

    if event_type == "asset.delete":
        asset_id = str(event.get("assetId") or "")
        if not asset_id:
            return False
        state["tombstones"]["assets"][asset_id] = event.get("tombstone") or {"id": asset_id, "deletedAt": event["serverAt"], "deletedBy": event.get("actorAccountId"), "version": event.get("assetVersion") or 1}
        for tombstone in event.get("objectTombstones") or []:
            if tombstone.get("id"):
                state["tombstones"]["objects"][str(tombstone["id"])] = tombstone
        removed_object_ids = {str(t.get("id")) for t in event.get("objectTombstones") or [] if t.get("id")}
        state["assets"] = [asset for asset in state.get("assets") or [] if str(asset.get("id") or "") != asset_id]
        for key in OBJECT_KEYS.values():
            state[key] = [obj for obj in state.get(key) or [] if str(obj.get("id") or "") not in removed_object_ids]
        state.setdefault("assetVotes", {}).pop(asset_id, None)
        state.setdefault("hiddenAssets", {}).pop(asset_id, None)
        dynamic_targets = state.setdefault("serverSync", {}).setdefault("dynamicTargets", {})
        for object_id in removed_object_ids:
            state.setdefault("objectVotes", {}).pop(object_id, None)
            state.setdefault("hiddenObjects", {}).pop(object_id, None)
            dynamic_targets.pop(object_id, None)
        state["moderationReports"] = [report for report in state.get("moderationReports") or [] if report.get("objectId") not in removed_object_ids]
        state.setdefault("serverSync", {})["lastServerEventId"] = event["serverEventId"]
        return True

    return False


def can_import_full_state(state: MutableMapping[str, Any], account: MutableMapping[str, Any] | None) -> Tuple[bool, Dict[str, Any]]:
    if state.get("worldMode") == "shared":
        return False, {"reason": "shared_full_import_disabled"}
    return True, {"reason": None}