server.py

import math
import sys
import io
import hashlib
import json
import os
import socket
import sqlite3
import time

from tokengate.operations_coordinator import OperationsCoordinator
from tokengate.token_system import task_token_guard
from constants import GenerationConstants
from physics import PhysicsEngine
from world_generation import (
    GLOBAL_HEIGHTMAP, GLOBAL_BIOME_MAP,
    compute_chunk_hash, fetch_chunk_threaded,
    finished_chunks_queue, set_active_database
)

# Fix Windows console encoding for Unicode arrows/emoji in comments/tracebacks
if sys.stdout and hasattr(sys.stdout, 'buffer'):
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
if sys.stderr and hasattr(sys.stderr, 'buffer'):
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')

set_active_database("server_chunk_cache.db")

SAVES_DIR        = "saves"
os.makedirs(SAVES_DIR, exist_ok=True)

LEDGER_PATH      = os.path.join(SAVES_DIR, "world_ledger.db")
STATS_PATH       = os.path.join(SAVES_DIR, "server_player_stats.jsonl")
SUSPECT_LOG      = os.path.join(SAVES_DIR, "suspect_log.jsonl")

HOST = '0.0.0.0'
PORT = 5555

consts = GenerationConstants

player_profiles  = {}
world_changes    = {}
water_changes    = {}
active_chunk_hosts = {}

water_chunk_last_saved: dict = {}   # Format: {(cx,cy,cz): timestamp}
WATER_CHUNK_SAVE_COOLDOWN = 10.0    # Don't re-save a chunk more than once per 10 seconds

pending_block_writes = []
pending_water_writes = []

# ----------------------------------------------------------------
# CHUNK HASH REGISTRY
# Populated as clients report chunk_loaded.
# Format: {(cx, cy, cz): "expected_hash_hex"}
#
# Chunks with accepted server deltas are in dirty_server_chunks —
# hash validation is skipped for these since their volume differs
# from the pristine procedural baseline.
# ----------------------------------------------------------------
chunk_hash_registry:  dict[tuple, str]  = {}
dirty_server_chunks:  set[tuple]        = set()


def _log_suspect(secure_id: str, cid: str, chunk: tuple, reason: str,
                 is_flying: bool, current_time: float):
    """Append a tamper event to the suspect log for admin review."""
    entry = {
        "timestamp":  current_time,
        "secure_id":  secure_id,
        "cid":        cid,
        "chunk":      list(chunk),
        "reason":     reason,
        "is_flying":  is_flying,
    }
    try:
        with open(SUSPECT_LOG, "a") as f:
            f.write(json.dumps(entry) + "\n")
        print(f"[SUSPECT] {cid} flagged: {reason} at chunk {chunk} "
              f"| flying={is_flying}")
    except Exception as e:
        print(f"[SUSPECT LOG ERROR] {e}")


def _precompute_spawn_region_hashes():
    """
    Pre-generates chunks around the world spawn point and caches their
    structural hashes so the server can validate the first players
    immediately without waiting for a client to report them.

    This also ensures cave geometry exists in server RAM at startup.
    Runs in the background via the existing fetch_chunk_threaded pipeline.
    """
    spawn_x = int(consts.WORLD_AREA / 2)
    spawn_z = int(consts.WORLD_AREA / 2)
    cx_center = spawn_x // consts.CHUNK_SIZE
    cz_center = spawn_z // consts.CHUNK_SIZE

    # Pre-warm a 3x3x3 region around spawn — enough for immediate validation
    WARM_RADIUS = 2
    queued = 0
    for nx in range(cx_center - WARM_RADIUS, cx_center + WARM_RADIUS + 1):
        for ny in range(-2, 4):   # covers underground caves to sky
            for nz in range(cz_center - WARM_RADIUS, cz_center + WARM_RADIUS + 1):
                fetch_chunk_threaded(nx, ny, nz, {}, {})
                queued += 1

    print(f"[SERVER] Queued {queued} spawn-region chunks for hash pre-computation.")
    return queued


def _drain_precomputed_hashes(timeout: float = 30.0):
    """
    Drains the finished_chunks_queue after _precompute_spawn_region_hashes()
    and populates chunk_hash_registry.

    Called once during startup before the main loop begins.
    Blocks for up to `timeout` seconds — increase on slow hardware.
    """
    deadline = time.perf_counter() + timeout
    drained  = 0

    while time.perf_counter() < deadline:
        if not finished_chunks_queue.empty():
            try:
                cx, cy, cz, _instances, volume, _water = finished_chunks_queue.get()
                chunk = (cx, cy, cz)
                chunk_hash_registry[chunk] = compute_chunk_hash(cx, cy, cz, volume)
                drained += 1
            except Exception as e:
                print(f"[SERVER] Hash pre-compute error: {e}")
        else:
            time.sleep(0.005)

    print(f"[SERVER] Pre-computed {drained} structural hashes for spawn region.")


def _init_db():
    """Sets up the SQLite database with Write-Ahead Logging for extreme concurrency."""
    with sqlite3.connect(LEDGER_PATH) as conn:
        conn.execute("PRAGMA journal_mode=WAL;")
        conn.execute("""
            CREATE TABLE IF NOT EXISTS world_blocks (
                x INTEGER,
                y INTEGER,
                z INTEGER,
                block_id INTEGER,
                water_vol REAL,
                last_updated REAL,
                PRIMARY KEY (x, y, z)
            )
        """)
        # Spatial indexing makes Octree lookups practically instant
        conn.execute("CREATE INDEX IF NOT EXISTS idx_spatial ON world_blocks(x, y, z);")



@task_token_guard(operation_type='load_player_data', tags={'weight': 'medium'})
def _load_player_profiles():
    """Reads the JSONL to restore player stats and LAST LOCATIONS!"""
    if os.path.exists(STATS_PATH):
        print("Restoring Player Profiles...")
        with open(STATS_PATH, "r") as f:
            for line in f:
                try:
                    data = json.loads(line)
                    # Because it's append-only, this naturally overwrites old data with the newest save
                    for k, v in data.items():
                        player_profiles[k] = v
                except Exception:
                    pass


@task_token_guard(operation_type='load_ledger_data', tags={'weight': 'medium'})
def _load_world_ledger():
    """Pulls the definitive state of the world straight into RAM."""
    _init_db()
    if os.path.exists(LEDGER_PATH):
        print("Reconstructing world from SQLite ledger...")
        with sqlite3.connect(LEDGER_PATH) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT x, y, z, block_id, water_vol FROM world_blocks")

            for x, y, z, block_id, water_vol in cursor.fetchall():
                if block_id is not None:
                    world_changes[(x, y, z)] = block_id
                if water_vol is not None and water_vol > 0:
                    water_changes[(x, y, z)] = water_vol


@task_token_guard(operation_type='save_player', tags={'weight': 'medium', 'storage_speed': 'SLOW'})
def _save_player_profile(secure_id, cdata=None):
    if secure_id in player_profiles:
        # --- NEW: Save their exact coordinates on disconnect ---
        if cdata:
            player_profiles[secure_id]['x'] = cdata['x']
            player_profiles[secure_id]['y'] = cdata['y']
            player_profiles[secure_id]['z'] = cdata['z']
            player_profiles[secure_id]['r'] = cdata['r']

        with  open(STATS_PATH, "a") as f:
            f.write(json.dumps({secure_id: player_profiles[secure_id]}) + "\n")


@task_token_guard(operation_type='ledger_write', tags={'weight': 'medium', 'storage_speed': 'MODERATE'})
def _write_ledger_entry(blocks_batch=None, water_batch=None, chunk_water_dict=None, chunk_coord=None):
    """Executes massive bulk inserts into SQLite in a single transaction."""
    try:
        # 3.0s timeout ensures WAL mode doesn't lock out parallel readers
        with sqlite3.connect(LEDGER_PATH, timeout=3.0) as conn:

            # 1. Flush Standard Block Updates
            if blocks_batch:
                conn.executemany("""
                    INSERT INTO world_blocks (x, y, z, block_id, last_updated)
                    VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT(x, y, z) DO UPDATE SET 
                        block_id=excluded.block_id,
                        last_updated=excluded.last_updated;
                """, blocks_batch)

            # 2. Flush Standard Water Bucket Pours/Scoops
            if water_batch:
                conn.executemany("""
                    INSERT INTO world_blocks (x, y, z, water_vol, last_updated)
                    VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT(x, y, z) DO UPDATE SET 
                        water_vol=excluded.water_vol,
                        last_updated=excluded.last_updated;
                """, water_batch)

            # 3. Flush the MMO Settled Water Arrays (The Ghost Trail Fix)
            if chunk_water_dict is not None and chunk_coord is not None:
                cx, cy, cz = chunk_coord
                min_x, max_x = cx * 16, (cx + 1) * 16
                min_y, max_y = cy * 16, (cy + 1) * 16
                min_z, max_z = cz * 16, (cz + 1) * 16

                # Erase old chunk water data
                conn.execute("""
                    UPDATE world_blocks SET water_vol = 0.0
                    WHERE x >= ? AND x < ? AND y >= ? AND y < ? AND z >= ? AND z < ?
                """, (min_x, max_x, min_y, max_y, min_z, max_z))

                # Insert the fresh fluid state
                ts = time.perf_counter()
                records = [
                    (int(coord.split(',')[0]), int(coord.split(',')[1]), int(coord.split(',')[2]), vol, ts)
                    for coord, vol in chunk_water_dict.items()
                ]
                conn.executemany("""
                    INSERT INTO world_blocks (x, y, z, water_vol, last_updated)
                    VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT(x, y, z) DO UPDATE SET 
                        water_vol=excluded.water_vol,
                        last_updated=excluded.last_updated;
                """, records)

        # Committing once per batch drastically reduces I/O overhead
        conn.commit()

    except Exception as e:
        print(f"Failed to batch-write to SQLite ledger: {e}")


def _generate_player_id(ip_address):
    salt = "TokenTorch_Secure_Salt_2026"
    return hashlib.sha256(f"{ip_address}_{salt}".encode()).hexdigest()[:16]


def _respawn_at_hash(secure_id, death_count):
    seed_string = f"{secure_id}_{death_count}"
    spawn_hash = int(hashlib.md5(seed_string.encode()).hexdigest(), 16)

    for offset in range(100):
        probe_x = (spawn_hash + offset * 73856) % consts.WORLD_AREA
        probe_z = (spawn_hash + offset * 19349) % consts.WORLD_AREA

        surface_y = float(GLOBAL_HEIGHTMAP[int(probe_z), int(probe_x)])
        biome_val = GLOBAL_BIOME_MAP[int(probe_z), int(probe_x)]

        if consts.SEA_LEVEL < surface_y <= 45 and biome_val < 0.8:
            return probe_x, surface_y + 2.0, probe_z

    center = consts.WORLD_AREA / 2
    return center, float(GLOBAL_HEIGHTMAP[int(center), int(center)]) + 2.0, center


def start_server():
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind((HOST, PORT))
    sock.setblocking(False)

    coordinator = OperationsCoordinator()
    coordinator.start()

    _load_player_profiles()
    _load_world_ledger()

    # Pre-generate spawn region caves and cache structural hashes.
    # This ensures cave geometry exists server-side before any client connects.
    print("[SERVER] Pre-warming spawn region...")
    _precompute_spawn_region_hashes()
    _drain_precomputed_hashes(timeout=30.0)

    print(f"UDP Game Server running on {HOST}:{PORT}")

    clients = {}
    engine = PhysicsEngine()

    # --- Network Tick Control ---
    NETWORK_TICK_RATE = 1.0 / 30.0  # Send outbound updates 30 times a second
    last_network_tick = 0.0

    def process_physics(dt):
        pass

    def process_inbound():
        current_time = time.perf_counter()
        while True:
            try:
                data, addr = sock.recvfrom(2048)  # Increased buffer slightly for batching
                packet = json.loads(data.decode('utf-8'))
                cid = packet.get('id')
                ip_addr = addr[0]

                if cid and cid not in clients:
                    secure_id = _generate_player_id(ip_addr)
                    if secure_id not in player_profiles:
                        player_profiles[secure_id] = {'deaths': 0, 'kills': 0}

                    print(f"New player {cid} connected [SECURE_ID: {secure_id}]")
                    clients[cid] = {
                        'addr': addr, 'vy': 0.0, 'hp': 100.0,
                        'secure_id': secure_id, 'last_seen': current_time,
                        'x': 0, 'y': 0, 'z': 0, 'r': 0,
                        'last_x': 0, 'last_y': 0, 'last_z': 0, 'last_r': 0,
                        'invulnerable_until': 0.0
                    }

                    # --- Force Teleport if they have a saved location! ---
                    if 'x' in player_profiles[secure_id]:
                        sock.sendto(json.dumps({
                            "type": "respawn",  # Reusing the respawn logic for login teleport!
                            "x": player_profiles[secure_id]['x'],
                            "y": player_profiles[secure_id]['y'],
                            "z": player_profiles[secure_id]['z'],
                            "hp": 100.0
                        }).encode('utf-8'), addr)

                    # FRAGMENTED UDP WORLD SYNC
                    changes_list = list(world_changes.items())
                    water_list = list(water_changes.items())

                    # 1. Send Solid Blocks in safe batches of 200
                    for i in range(0, len(changes_list), 200):
                        batch = changes_list[i:i + 200]
                        sync_packet = {
                            "type": "world_sync",
                            "changes": {f"{k[0]},{k[1]},{k[2]}": v for k, v in batch}
                        }
                        try:
                            sock.sendto(json.dumps(sync_packet).encode('utf-8'), addr)
                        except Exception as e:
                            print(f"[Warning] Failed to send block sync chunk: {e}")

                    # 2. Send Water Blocks in safe batches of 200
                    for i in range(0, len(water_list), 200):
                        batch = water_list[i:i + 200]
                        sync_packet = {
                            "type": "world_sync",
                            "water_changes": {f"{k[0]},{k[1]},{k[2]}": v for k, v in batch}
                        }
                        try:
                            sock.sendto(json.dumps(sync_packet).encode('utf-8'), addr)
                        except Exception as e:
                            print(f"[Warning] Failed to send water sync chunk: {e}")

                if cid in clients:
                    clients[cid]['last_seen'] = current_time

                # Catching player movement
                if packet.get("type") == "player" or "x" in packet:
                    clients[cid].update({
                        'x': packet['x'], 'y': packet['y'], 'z': packet['z'], 'r': packet.get('r', 0),
                        'item': packet.get('item', 0),  # NEW: What are they holding?
                        'firing': packet.get('firing', False)  # NEW: Are they shooting?
                    })

                # 1. A client tells us they just generated a chunk
                elif packet.get("type") == "chunk_loaded":
                    cx, cy, cz = packet["chunk"]
                    chunk = (cx, cy, cz)

                    # ---- STRUCTURAL HASH VALIDATION ----
                    client_hash = packet.get("chunk_hash")
                    is_flying = clients[cid].get('is_flying', False)

                    if client_hash and chunk not in dirty_server_chunks:
                        expected = chunk_hash_registry.get(chunk)

                        if expected is None:
                            # We haven't seen this chunk yet — store the client's
                            # report as the reference. First honest client wins.
                            chunk_hash_registry[chunk] = client_hash

                        elif expected != client_hash:
                            # Hash mismatch — client is reporting different terrain
                            # than the seed produces. Flag and continue (no kick yet —
                            # admin reviews suspect_log.jsonl).
                            secure_id = clients[cid].get('secure_id', 'unknown')
                            _log_suspect(
                                secure_id, cid, chunk,
                                reason=f"chunk_hash_mismatch expected={expected} got={client_hash}",
                                is_flying=is_flying,
                                current_time=current_time
                            )
                            # Still serve them the correct world state —
                            # don't reveal that we flagged them.

                    # ---- WELCOME SYNC (unchanged) ----
                    chunk_blocks = {}
                    chunk_water = {}

                    # Calculate absolute world boundaries for this chunk
                    min_x, max_x = cx * 16, (cx + 1) * 16
                    min_y, max_y = cy * 16, (cy + 1) * 16
                    min_z, max_z = cz * 16, (cz + 1) * 16

                    # Scrape RAM for blocks in this specific chunk
                    for (x, y, z), b_id in list(world_changes.items()):
                        if min_x <= x < max_x and min_y <= y < max_y and min_z <= z < max_z:
                            chunk_blocks[f"{x},{y},{z}"] = b_id

                    for (x, y, z), vol in list(water_changes.items()):
                        if min_x <= x < max_x and min_y <= y < max_y and min_z <= z < max_z:
                            chunk_water[f"{x},{y},{z}"] = vol

                    # If the server knows anything about this chunk, send it immediately!
                    if chunk_blocks or chunk_water:
                        sync_packet = {
                            "type": "world_sync",  # Re-using the login sync perfectly patches the client!
                            "changes": chunk_blocks,
                            "water_changes": chunk_water
                        }
                        try:
                            sock.sendto(json.dumps(sync_packet).encode('utf-8'), clients[cid]['addr'])
                        except Exception:
                            pass

                    if chunk not in active_chunk_hosts:
                        # They are the first one here! They become the Authority.
                        active_chunk_hosts[chunk] = cid
                    else:
                        host_id = active_chunk_hosts[chunk]
                        if host_id in clients and host_id != cid:
                            # Someone else is already here. Ask the Host to sync the new guy!
                            req = {
                                "type": "request_fluid_state",
                                "chunk": [cx, cy, cz],
                                "target": cid
                            }
                            sock.sendto(json.dumps(req).encode('utf-8'), clients[host_id]['addr'])
                        else:
                            # The old host disconnected. The new guy takes over!
                            active_chunk_hosts[chunk] = cid

                # --- Local Chat Routing ---
                elif packet.get("type") == "chat":
                    sender_id = packet.get("id")
                    sender_name = packet.get("name", "Unknown")
                    text = packet.get("text", "")

                    # Find sender's location
                    sx, sy, sz = 0, 0, 0
                    if sender_id in clients:
                        sx, sy, sz = clients[sender_id]['x'], clients[sender_id]['y'], clients[sender_id]['z']

                    chat_packet = json.dumps({"type": "chat", "name": sender_name, "text": text}).encode('utf-8')

                    # Broadcast ONLY to players/NPCs within 50 blocks!
                    for other_cid, other_cdata in clients.items():
                        dist = math.hypot(other_cdata['x'] - sx, other_cdata['z'] - sz)
                        if dist < 50.0:
                            try:
                                sock.sendto(chat_packet, other_cdata['addr'])
                            except Exception:
                                pass

                # 2. The Host sends the mid-air physics, we relay it to the NEW GUY ONLY!
                elif packet.get("type") == "relay_fluid_state":
                    target_id = packet.get("target_id")

                    if target_id in clients:
                        relay = {
                            "type": "water_consensus_sync",
                            "water_data": packet.get("water_data", {})
                        }
                        # SENDTO SPECIFIC TARGET ONLY - NO FOR LOOPS!
                        sock.sendto(json.dumps(relay).encode('utf-8'), clients[target_id]['addr'])

                elif packet.get("type") == "block_update":
                    # --- Force cast to INT so floats don't corrupt the sync strings! ---
                    bx, by, bz = int(packet["bx"]), int(packet["by"]), int(packet["bz"])
                    block_id = packet["block_id"]

                    # 1. Update server RAM
                    world_changes[(bx, by, bz)] = block_id

                    # 2. DROP IT IN THE BUFFER (DO NOT SAVE TO DB YET!)
                    pending_block_writes.append((bx, by, bz, block_id, current_time))

                    # --- Bounce block update to other clients! ---
                    encoded_update = json.dumps(packet).encode('utf-8')
                    for other_cid, other_cdata in clients.items():
                        if other_cid != cid:
                            try:
                                sock.sendto(encoded_update, other_cdata['addr'])
                            except Exception:
                                pass

                elif packet.get("type") == "water_update":
                    # --- Force cast to INT here too just to be perfectly safe ---
                    bx, by, bz = int(packet["bx"]), int(packet["by"]), int(packet["bz"])
                    vol = packet["volume"]

                    # 1. Update server RAM
                    water_changes[(bx, by, bz)] = vol

                    # 2. DROP IT IN THE BUFFER
                    pending_water_writes.append((bx, by, bz, vol, current_time))

                    # 3. Bounce packet to other clients
                    # (The duplicate copy was removed from here!)
                    encoded_update = json.dumps(packet).encode('utf-8')
                    for other_cid, other_cdata in clients.items():
                        if other_cid != cid:
                            try:
                                sock.sendto(encoded_update, other_cdata['addr'])
                            except Exception:
                                pass

                elif packet.get("type") == "batched_water_sync":
                    water_data = packet.get("water_data", {})
                    chunk_coord_list = packet.get("chunk")

                    if chunk_coord_list:
                        cx, cy, cz = chunk_coord_list
                        chunk_key = (cx, cy, cz)

                        min_x, max_x = cx * 16, (cx + 1) * 16
                        min_y, max_y = cy * 16, (cy + 1) * 16
                        min_z, max_z = cz * 16, (cz + 1) * 16

                        # Erase old RAM state for this chunk
                        keys_to_delete = [
                            k for k in water_changes.keys()
                            if min_x <= k[0] < max_x and min_y <= k[1] < max_y and min_z <= k[2] < max_z
                        ]
                        for k in keys_to_delete:
                            del water_changes[k]

                        # ── DB WRITE GATE ──────────────────────────────────────────────
                        # Only write to SQLite if this chunk hasn't been saved recently.
                        # RAM is always current — this just throttles disk I/O.
                        last_saved = water_chunk_last_saved.get(chunk_key, 0.0)
                        if current_time - last_saved >= WATER_CHUNK_SAVE_COOLDOWN:
                            water_chunk_last_saved[chunk_key] = current_time
                            _write_ledger_entry(
                                chunk_water_dict=water_data,
                                chunk_coord=chunk_coord_list
                            )

                    # Apply fresh water state to RAM regardless of DB cooldown
                    for coord_str, vol in water_data.items():
                        x, y, z = map(int, coord_str.split(','))
                        water_changes[(x, y, z)] = vol

                elif packet.get("type") == "damage":
                    target_id = packet.get("target_id")
                    damage_amount = packet.get("amount", 0)

                    if target_id in clients and cid in clients:
                        # --- I-FRAME CHECK: Is the target invincible right now? ---
                        if current_time < clients[target_id].get('invulnerable_until', 0.0):
                            continue  # Ignore the damage entirely!

                        clients[target_id]['hp'] -= damage_amount
                        print(f"[PvP] {cid} blasted {target_id}! Target HP: {clients[target_id]['hp']}")

                        sock.sendto(json.dumps({
                            "type": "health_update",
                            "hp": clients[target_id]['hp']
                        }).encode('utf-8'), clients[target_id]['addr'])

                        if clients[target_id]['hp'] <= 0:
                            clients[target_id]['hp'] = 100.0

                            # Grant 2.5 seconds of Invulnerability so late packets don't double kill!
                            clients[target_id]['invulnerable_until'] = current_time + 2.5

                            victim_secure_id = clients[target_id]['secure_id']
                            attacker_secure_id = clients[cid]['secure_id']

                            player_profiles[victim_secure_id]['deaths'] += 1
                            player_profiles[attacker_secure_id]['kills'] += 1

                            rx, ry, rz = _respawn_at_hash(victim_secure_id, player_profiles[victim_secure_id]['deaths'])

                            sock.sendto(json.dumps({
                                "type": "respawn",
                                "x": rx, "y": ry, "z": rz,
                                "hp": 100.0
                            }).encode('utf-8'), clients[target_id]['addr'])

                            print(f"[PvP] {target_id} Obliterated! Sent to Grassy Respawn.")

            except BlockingIOError:
                break
            except Exception:
                pass

    def process_outbound():
        nonlocal last_network_tick
        # --- 1Hz Heartbeat Timer ---
        if not hasattr(process_outbound, 'last_heartbeat'):
            process_outbound.last_heartbeat = 0.0

        current_time = time.perf_counter()

        if current_time - last_network_tick < NETWORK_TICK_RATE:
            return

        last_network_tick = current_time

        # Override the AFK sleep every 1 second
        force_heartbeat = False
        if current_time - process_outbound.last_heartbeat > 1.0:
            force_heartbeat = True
            process_outbound.last_heartbeat = current_time

        # --- REACTIVE LEDGER FLUSH ---
        # Initialize the timer on the function object if it doesn't exist
        if not hasattr(process_outbound, 'last_db_flush'):
            process_outbound.last_db_flush = current_time

        buffer_size = len(pending_block_writes) + len(pending_water_writes)
        time_since_last_flush = current_time - process_outbound.last_db_flush

        # Trigger DB write ONLY IF we have 100+ items, OR 5 seconds have passed with pending items
        if buffer_size >= 100 or (buffer_size > 0 and time_since_last_flush > 5.0):
            process_outbound.last_db_flush = current_time

            # Safely copy the lists
            b_batch = pending_block_writes[:]
            w_batch = pending_water_writes[:]

            # Clear the RAM buckets immediately so the simulation isn't blocked
            pending_block_writes.clear()
            pending_water_writes.clear()

            # Ship it using explicit keyword arguments to avoid parameter mismatched
            _write_ledger_entry(blocks_batch=b_batch, water_batch=w_batch)

        timeout_disconnects = []
        active_players = []

        for cid, cdata in list(clients.items()):
            if current_time - cdata['last_seen'] > 15.0:
                print(f"[Disconnect] {cid} timed out. Dumping stats & pos...")
                # Pass cdata so the server can extract their coordinates!
                _save_player_profile(cdata['secure_id'], cdata)
                timeout_disconnects.append(cid)

                # --- Broadcast Leave Event to cleanup ghosts ---
                leave_packet = json.dumps({"type": "player_leave", "id": cid}).encode('utf-8')
                for other_cid, other_cdata in clients.items():
                    if other_cid != cid:
                        try:
                            sock.sendto(leave_packet, other_cdata['addr'])
                        except Exception:
                            pass
                continue

            dx = abs(cdata['x'] - cdata['last_x'])
            dy = abs(cdata['y'] - cdata['last_y'])
            dz = abs(cdata['z'] - cdata['last_z'])
            dr = abs(cdata['r'] - cdata['last_r'])

            # --- Heartbeat Override ---
            if dx > 0.01 or dy > 0.01 or dz > 0.01 or dr > 0.01 or cdata.get('firing', False) or force_heartbeat:
                cdata['last_x'], cdata['last_y'] = cdata['x'], cdata['y']
                cdata['last_z'], cdata['last_r'] = cdata['z'], cdata['r']

                active_players.append({
                    "id": cid, "x": cdata['x'], "y": cdata['y'],
                    "z": cdata['z'], "r": cdata['r'], "hp": cdata['hp'],
                    "item": cdata.get('item', 0),  # NEW
                    "firing": cdata.get('firing', False)  # NEW
                })

        # Clean up dead connections
        for cid in timeout_disconnects:
            del clients[cid]

        # 2. BATCH & SHIP PHASE
        # We only send outbound if AT LEAST ONE player moved.
        if active_players:
            # Create a SINGLE packet containing a list of ALL moving players
            batch_packet = json.dumps({
                "type": "batch_players",
                "players": active_players
            }).encode('utf-8')

            # Send that one packet to everyone
            for cdata in clients.values():
                try:
                    sock.sendto(batch_packet, cdata['addr'])
                except Exception:
                    pass

        # Always send the master clock
        encoded_world = json.dumps({"type": "world", "time": time.perf_counter()}).encode('utf-8')
        for cdata in clients.values():
            try:
                sock.sendto(encoded_world, cdata['addr'])
            except Exception:
                pass

    try:
        while True:
            engine.process_loop(process_physics, process_inbound, process_outbound)
            time.sleep(0.001)

    except KeyboardInterrupt:
        print("\nShutting down server gracefully. Dumping ALL profiles...")
        for cdata in clients.values():
            _save_player_profile(cdata['secure_id'])
        sock.close()

        # --- Safely stop the event bus on server shutdown ---
        print("Stopping Operations Coordinator...")
        coordinator.stop()

if __name__ == "__main__":
    start_server()