File

from pathlib import Path

from collections import deque

import json

import queue

import sys

import threading

import time

ROOT = Path(__file__).resolve().parents[1]

sys.path.insert(0, str(ROOT))

import sheet_relay as sr

LIVE_REVISION = "20260830-107-LL2"

sr.REVISION = LIVE_REVISION

POLL_WORKERS = 3

POLL_STAGGER_SECONDS = 0.70

POLL_ERROR_BACKOFF_SECONDS = 0.75

HEARTBEAT_INTERVAL_SECONDS = 60.0

MONITOR_INTERVAL_SECONDS = 60.0

EXECUTOR_IDLE_SECONDS = 0.25

MAX_SEEN_COMMANDS = 4096

command_queue = queue.Queue()

stop_event = threading.Event()

seen_lock = threading.Lock()

seen_ids = set()

seen_order = deque()

def enqueue_once(command):

    command_id = str(command.get("id") or "")

    if not command_id:

        sr.log("LL2 dropped command without id")

        return False

    with seen_lock:

        if command_id in seen_ids:

            sr.log(f"LL2 duplicate claim suppressed id={command_id}")

            return False

        seen_ids.add(command_id)

        seen_order.append(command_id)

        while len(seen_order) > MAX_SEEN_COMMANDS:

            seen_ids.discard(seen_order.popleft())

    command_queue.put(command)

    return True

def process_command(cfg, command):

    sr.log(f"claimed command id={command.get('id')} action={command.get('action')}")

    task_id = ""

    try:

        task_id = str(json.loads(command.get("payload_json") or "{}").get("task_id", ""))

    except (TypeError, ValueError):

        pass

    sr.record_task_event(

        task_id,

        "command_started",

        command_id=command.get("id"),

        action=command.get("action"),

    )

    try:

        command_result = sr.execute_command(cfg, command)

        sr.record_task_event(

            task_id,

            "command_completed" if command_result.get("status") == "done" else "command_failed",

            command_id=command.get("id"),

            action=command.get("action"),

            relay_status=command_result.get("status"),

            http_status=command_result.get("http_status"),

        )

        sr.send_result(cfg, command_result)

    except Exception as exc:

        sr.record_task_event(

            task_id,

            "command_failed",

            command_id=command.get("id"),

            action=command.get("action"),

            error=type(exc).__name__,

        )

        sr.send_result(

            cfg,

            sr.result(command.get("id", ""), "error", "", "", str(exc), time.time()),

        )

def poll_worker(cfg, worker_index):

    if worker_index:

        time.sleep(POLL_STAGGER_SECONDS * worker_index)

    sr.log(f"LL2 poll worker {worker_index} online")

    while not stop_event.is_set():

        try:

            status, data = sr.http_json(sr.web_app_url(cfg, "poll"), timeout=45)

            if status != 200:

                sr.log(f"LL2 poll worker {worker_index} http={status}")

                time.sleep(POLL_ERROR_BACKOFF_SECONDS)

                continue

            command = data.get("command")

            if command:

                enqueue_once(command)

        except Exception as exc:

            sr.log(f"LL2 poll worker {worker_index} error: {type(exc).__name__}: {exc}")

            time.sleep(POLL_ERROR_BACKOFF_SECONDS)

def maintenance_worker(cfg):

    next_heartbeat = 0.0

    next_monitor = 0.0

    sr.log("LL2 maintenance worker online")

    while not stop_event.is_set():

        now = time.monotonic()

        if now >= next_heartbeat:

            try:

                sr.heartbeat(cfg, sr.bridge_health(cfg))

            except Exception as exc:

                sr.log(f"LL2 heartbeat error: {type(exc).__name__}: {exc}")

            next_heartbeat = time.monotonic() + HEARTBEAT_INTERVAL_SECONDS

        if time.monotonic() >= next_monitor:

            try:

                sr.monitor_verifies(cfg)

            except Exception as exc:

                sr.log(f"LL2 monitor error: {type(exc).__name__}: {exc}")

            next_monitor = time.monotonic() + MONITOR_INTERVAL_SECONDS

        stop_event.wait(0.25)

def run_loop():

    cfg = sr.require_config()

    sr.log(

        f"low-latency relay starting revision={LIVE_REVISION} "

        f"poll_workers={POLL_WORKERS} stagger={POLL_STAGGER_SECONDS:.2f}s "

        "maintenance=background execution=serial dedup=command-id"

    )

    threading.Thread(

        target=maintenance_worker,

        args=(cfg,),

        name="sheet-maintenance",

        daemon=True,

    ).start()

    for worker_index in range(POLL_WORKERS):

        threading.Thread(

            target=poll_worker,

            args=(cfg, worker_index),

            name=f"sheet-poll-{worker_index}",

            daemon=True,

        ).start()

    try:

        while True:

            try:

                command = command_queue.get(timeout=EXECUTOR_IDLE_SECONDS)

            except queue.Empty:

                continue

            try:

                process_command(cfg, command)

            finally:

                command_queue.task_done()

    except KeyboardInterrupt:

        sr.log("low-latency relay LL2 stopped by keyboard interrupt")

    finally:

        stop_event.set()

if __name__ == "__main__":

    run_loop()