pixel/server/rotation_worker.py

#!/usr/bin/env python3
"""Authoritative server-side exhibition rotation policy for Pixel Island.

This worker treats assets as permanent library works and placements as temporary
island exhibition objects. The browser may preview the same policy locally, but
production should accept this worker/database layer as the source of truth.

Policy:
- Account required to publish.
- First 24h: 5 public placements/hour. Day 2+: 10 placements/hour.
- Island exhibition cap: 250 active objects.
- 150 slots are newest/recent-publication slots.
- 100 slots are random revival slots.
- Newest and revival buckets are selected independently; duplicates are removed.
- Upvote rank effect is capped at +50 votes.
- Downvotes advance rotation-out. Extreme downvote hides use the same user-facing
  wording as capacity rotation, unless an admin marks a violation.
- Reports are local-user hides in the client. Server-side moderation creates
  permanent_hidden / violation_hidden only after admin/policy action.
- permanent_hidden can be restored by an admin.
"""
from __future__ import annotations

import argparse
import json
import random
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Iterable, List, MutableMapping, Optional, Tuple

DISPLAY_LIMIT = 250
NEWEST_SLOTS = 150
REVIVAL_SLOTS = 100
REVIVAL_SAMPLE_SIZE = 50
REVIVAL_PICK_COUNT = 20
UPVOTE_DELAY_SLOTS = 20
DOWNVOTE_ADVANCE_SLOTS = 25
UPVOTE_RANK_CAP = 50
FIRST_DAY_LIMIT = 5
TRUSTED_LIMIT = 10
ONE_HOUR_MS = 60 * 60 * 1000
ONE_DAY_MS = 24 * ONE_HOUR_MS
EXTREME_DOWNVOTES = 10
EXTREME_MARGIN = 8

ACTIVE = "active"
HIDDEN_ROTATION = "hidden_rotation"
PERMANENT_HIDDEN = "permanent_hidden"
VIOLATION_HIDDEN = "violation_hidden"

PUBLIC_REASON_ROTATION = "island_exhibition_full"
PUBLIC_REASON_VIOLATION = "moderation_violation"

@dataclass
class RotationConfig:
    display_limit: int = DISPLAY_LIMIT
    newest_slots: int = NEWEST_SLOTS
    revival_slots: int = REVIVAL_SLOTS
    revival_sample_size: int = REVIVAL_SAMPLE_SIZE
    revival_pick_count: int = REVIVAL_PICK_COUNT
    upvote_delay_slots: int = UPVOTE_DELAY_SLOTS
    downvote_advance_slots: int = DOWNVOTE_ADVANCE_SLOTS
    upvote_rank_cap: int = UPVOTE_RANK_CAP
    extreme_downvotes: int = EXTREME_DOWNVOTES
    extreme_margin: int = EXTREME_MARGIN

    def __post_init__(self) -> None:
        self.display_limit = max(0, int(self.display_limit))
        self.newest_slots = max(0, int(self.newest_slots))
        self.revival_slots = max(0, int(self.revival_slots))
        if self.newest_slots + self.revival_slots > self.display_limit:
            self.revival_slots = max(0, self.display_limit - self.newest_slots)
            self.newest_slots = min(self.newest_slots, self.display_limit)

    @classmethod
    def from_mapping(cls, values: MutableMapping[str, Any] | None) -> "RotationConfig":
        data = values or {}
        allowed = set(cls.__dataclass_fields__)
        clean = {key: int(value) for key, value in data.items() if key in allowed and value is not None}
        return cls(**clean)


def load_config(path: Optional[Path], overrides: MutableMapping[str, Any] | None = None) -> RotationConfig:
    data: Dict[str, Any] = {}
    if path:
        with path.open("r", encoding="utf-8") as f:
            loaded = json.load(f)
        if not isinstance(loaded, MutableMapping):
            raise ValueError("Rotation config JSON root must be an object")
        data.update(loaded)
    data.update({k: v for k, v in (overrides or {}).items() if v is not None})
    return RotationConfig.from_mapping(data)


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


def object_id(obj: MutableMapping[str, Any]) -> str:
    return str(obj.get("id") or "")


def public_at(obj: MutableMapping[str, Any]) -> int:
    return int(obj.get("publishedAt") or obj.get("placedAt") or obj.get("createdAt") or 0)


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


def account_publish_limit(account: MutableMapping[str, Any] | None, now: Optional[int] = None) -> int:
    if not account or not account.get("createdAt"):
        return 0
    now = now or now_ms()
    created = int(account.get("createdAt") or now)
    return FIRST_DAY_LIMIT if now - created < ONE_DAY_MS else TRUSTED_LIMIT


def can_publish(state: MutableMapping[str, Any], account: MutableMapping[str, Any] | None, now: Optional[int] = None) -> Tuple[bool, Dict[str, Any]]:
    """API helper. Production publish endpoints should call this before accepting a placement."""
    now = now or now_ms()
    account_id = author_id_from_account(account)
    limit = account_publish_limit(account, now)
    if not account_id or limit <= 0:
        return False, {"reason": "account_required", "used": 0, "limit": 0}
    log = [entry for entry in state.get("publishLog") or [] if now - int(entry.get("at") or 0) < ONE_DAY_MS]
    used = sum(1 for entry in log if entry.get("author") == account_id and now - int(entry.get("at") or 0) < ONE_HOUR_MS)
    state["publishLog"] = log
    return used < limit, {"reason": None if used < limit else "quota_exceeded", "used": used, "limit": limit}


def record_publish(state: MutableMapping[str, Any], account: MutableMapping[str, Any], obj: MutableMapping[str, Any], kind: str, action: str, now: Optional[int] = None) -> None:
    now = now or now_ms()
    obj["publishedAt"] = now
    obj["status"] = ACTIVE
    obj.pop("hiddenReason", None)
    obj.pop("hiddenAt", None)
    log = state.setdefault("publishLog", [])
    log.append({"at": now, "author": author_id_from_account(account), "kind": kind, "objectId": object_id(obj), "assetId": obj.get("assetId"), "action": action})
    state["publishLog"] = log[-10000:]


def iter_objects(state: MutableMapping[str, Any]) -> Iterable[Tuple[str, MutableMapping[str, Any]]]:
    for obj in state.get("placed") or []:
        if isinstance(obj, MutableMapping) and obj.get("id"):
            yield "static", obj
    for obj in state.get("dynamicSummons") or []:
        if isinstance(obj, MutableMapping) and obj.get("id"):
            yield "dynamic", obj


def vote_counts(state: MutableMapping[str, Any], obj_id: str) -> Tuple[int, int]:
    votes = (state.get("objectVotes") or {}).get(obj_id) or {}
    return int(votes.get("up") or 0), int(votes.get("down") or 0)


def is_moderation_hidden(obj: MutableMapping[str, Any]) -> bool:
    return obj.get("status") in {PERMANENT_HIDDEN, VIOLATION_HIDDEN} or obj.get("permanentHidden") is True


def eligible_for_exhibition(obj: MutableMapping[str, Any]) -> bool:
    return not is_moderation_hidden(obj)


def rotation_entry(state: MutableMapping[str, Any], kind: str, obj: MutableMapping[str, Any], base_index: int, config: RotationConfig) -> Dict[str, Any]:
    up_raw, down = vote_counts(state, object_id(obj))
    up_rank = min(up_raw, config.upvote_rank_cap)
    return {
        "kind": kind,
        "object": obj,
        "id": object_id(obj),
        "publicAt": public_at(obj),
        "baseIndex": base_index,
        "up": up_raw,
        "upRank": up_rank,
        "down": down,
        "effectiveSlot": base_index + up_rank * config.upvote_delay_slots - down * config.downvote_advance_slots,
    }


def rotation_entries(state: MutableMapping[str, Any], config: RotationConfig) -> List[Dict[str, Any]]:
    pairs = [(kind, obj) for kind, obj in iter_objects(state) if eligible_for_exhibition(obj)]
    pairs.sort(key=lambda pair: (public_at(pair[1]), object_id(pair[1])))
    return [rotation_entry(state, kind, obj, i, config) for i, (kind, obj) in enumerate(pairs)]


def deterministic_random_score(obj_id: str, seed: int) -> float:
    rng = random.Random(f"{seed}:{obj_id}")
    return rng.random()


def select_exhibition_buckets(state: MutableMapping[str, Any], config: RotationConfig, seed: Optional[int] = None) -> Dict[str, List[Dict[str, Any]]]:
    seed = seed if seed is not None else int(state.get("lastRotationAt") or now_ms()) // ONE_DAY_MS
    entries = rotation_entries(state, config)
    newest = sorted(entries, key=lambda e: (e["effectiveSlot"], e["publicAt"], e["id"]), reverse=True)[:config.newest_slots]
    newest_ids = {e["id"] for e in newest}
    revival_pool = [e for e in entries if e["id"] not in newest_ids]

    # Periodic server revival: sample up to 50 previous/excess objects, then take 20 with
    # the best capped score, while keeping randomness in the sample.
    historical = [e for e in revival_pool if e["object"].get("status") == HIDDEN_ROTATION]
    rng = random.Random(seed)
    sample = rng.sample(historical, min(config.revival_sample_size, len(historical))) if historical else []
    picked = sorted(sample, key=lambda e: (e["upRank"], e["upRank"] - e["down"], e["publicAt"]), reverse=True)[:config.revival_pick_count]
    picked_ids = {e["id"] for e in picked}

    rest = [e for e in revival_pool if e["id"] not in picked_ids]
    random_rest = sorted(rest, key=lambda e: (deterministic_random_score(e["id"], seed), e["upRank"] - e["down"]), reverse=True)
    revival = (picked + random_rest)[:config.revival_slots]
    return {"newest": newest, "revival": revival, "entries": entries}


def apply_extreme_downvote_policy(state: MutableMapping[str, Any], config: RotationConfig, now: Optional[int] = None) -> List[Dict[str, Any]]:
    """Hide extreme downvote cases from the island with the same public text as capacity rotation.

    This is not a violation decision. It stays reversible and is separate from admin
    violation hiding.
    """
    now = now or now_ms()
    changed: List[Dict[str, Any]] = []
    for kind, obj in iter_objects(state):
        if is_moderation_hidden(obj):
            continue
        up, down = vote_counts(state, object_id(obj))
        if down >= config.extreme_downvotes and down - up >= config.extreme_margin:
            obj["status"] = HIDDEN_ROTATION
            obj["hiddenReason"] = PUBLIC_REASON_ROTATION
            obj["hiddenAt"] = now
            changed.append({"objectId": object_id(obj), "assetId": obj.get("assetId"), "kind": kind, "reason": "extreme_downvotes_as_rotation", "up": up, "down": down})
    return changed


def apply_exhibition_cap(state: MutableMapping[str, Any], config: RotationConfig, seed: Optional[int] = None, now: Optional[int] = None) -> Dict[str, Any]:
    now = now or now_ms()
    buckets = select_exhibition_buckets(state, config, seed=seed)
    visible_ids = {e["id"] for e in buckets["newest"] + buckets["revival"]}
    newly_active = newly_hidden = 0
    for entry in buckets["entries"]:
        obj = entry["object"]
        if entry["id"] in visible_ids:
            if obj.get("status") != ACTIVE:
                newly_active += 1
            obj["status"] = ACTIVE
            obj.pop("hiddenReason", None)
            obj.pop("hiddenAt", None)
        else:
            if obj.get("status") != HIDDEN_ROTATION:
                newly_hidden += 1
            obj["status"] = HIDDEN_ROTATION
            obj["hiddenReason"] = PUBLIC_REASON_ROTATION
            obj["hiddenAt"] = now
    return {"active": len(visible_ids), "newest": len(buckets["newest"]), "revival": len(buckets["revival"]), "rotationHidden": max(0, len(buckets["entries"]) - len(visible_ids)), "newlyActive": newly_active, "newlyHidden": newly_hidden}


def hide_violation(state: MutableMapping[str, Any], object_ids: List[str], now: Optional[int] = None) -> List[str]:
    now = now or now_ms()
    hidden: List[str] = []
    for _, obj in iter_objects(state):
        if object_id(obj) in object_ids:
            obj["status"] = VIOLATION_HIDDEN
            obj["permanentHidden"] = True
            obj["hiddenReason"] = PUBLIC_REASON_VIOLATION
            obj["hiddenAt"] = now
            hidden.append(object_id(obj))
    return hidden


def restore_permanent(state: MutableMapping[str, Any], object_ids: List[str], now: Optional[int] = None) -> List[str]:
    """Admin restore for permanent/violation hidden placements."""
    now = now or now_ms()
    restored: List[str] = []
    for _, obj in iter_objects(state):
        if object_id(obj) in object_ids and is_moderation_hidden(obj):
            obj["status"] = HIDDEN_ROTATION
            obj["permanentHidden"] = False
            obj["hiddenReason"] = PUBLIC_REASON_ROTATION
            obj["hiddenAt"] = now
            restored.append(object_id(obj))
    return restored


def run_rotation(state: MutableMapping[str, Any], config: RotationConfig, seed: Optional[int] = None) -> Dict[str, Any]:
    now = now_ms()
    extreme = apply_extreme_downvote_policy(state, config, now=now)
    cap = apply_exhibition_cap(state, config, seed=seed, now=now)
    state["lastRotationAt"] = now
    state["rotationPolicy"] = {
        "displayLimit": config.display_limit,
        "newestSlots": config.newest_slots,
        "revivalSlots": config.revival_slots,
        "upvoteRankCap": config.upvote_rank_cap,
        "serverAuthoritative": True,
    }
    return {"extremeDownvoteHiddenAsRotation": extreme, "cap": cap, "lastRotationAt": now}


def load_json(path: Path) -> MutableMapping[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, MutableMapping):
        raise ValueError("World JSON root must be an object")
    return data


def save_json(path: Path, data: MutableMapping[str, Any]) -> None:
    with path.open("w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, separators=(",", ":"))


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run Pixel Island exhibition rotation policy on a world JSON file.")
    parser.add_argument("world_json", type=Path)
    parser.add_argument("--write", action="store_true")
    parser.add_argument("--out", type=Path)
    parser.add_argument("--seed", type=int)
    parser.add_argument("--config", type=Path, help="Optional rotation policy JSON file.")
    parser.add_argument("--display-limit", type=int)
    parser.add_argument("--newest-slots", type=int)
    parser.add_argument("--revival-slots", type=int)
    parser.add_argument("--restore", nargs="*", default=None, help="Admin restore object IDs from permanent/violation hidden to rotation hidden.")
    parser.add_argument("--hide-violation", nargs="*", default=None, help="Admin hide object IDs as violation_hidden.")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    state = load_json(args.world_json)
    config = load_config(args.config, {"display_limit": args.display_limit, "newest_slots": args.newest_slots, "revival_slots": args.revival_slots})
    summary: Dict[str, Any] = {}
    if args.restore:
        summary["restored"] = restore_permanent(state, args.restore)
    if args.hide_violation:
        summary["violationHidden"] = hide_violation(state, args.hide_violation)
    if not args.restore and not args.hide_violation:
        summary = run_rotation(state, config, seed=args.seed)
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    if args.write or args.out:
        save_json(args.out or args.world_json, state)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())