import json
import os
import subprocess
import tempfile
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path


SCRIPT = Path(__file__).with_name("movetdarr.sh")
state_lock = threading.Lock()
next_job = 0
open_leases = 0
max_open_leases = 0
job_ids = {}


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        global next_job, open_leases, max_open_leases
        data = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))))
        action = data.get("action")
        if action == "start":
            with state_lock:
                next_job += 1
                job_id = f"job-{next_job}"
                job_ids[data["directory"]] = job_id
                open_leases += 1
                max_open_leases = max(max_open_leases, open_leases)
            time.sleep(0.3)
            result = {"ok": True, "jobId": job_id, "state": "registered"}
        elif action == "moved":
            with state_lock:
                open_leases -= 1
            result = {"ok": True, "jobId": data["jobId"], "state": "processing"}
        elif action == "status":
            result = {"ok": True, "jobId": data["jobId"], "state": "success"}
        else:
            result = {"ok": False, "state": "failed"}
        body = json.dumps(result).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *_args):
        pass


def start_script(root, name, server_port):
    source_root = root / "source"
    source = source_root / "serien" / name
    source.mkdir(parents=True)
    (source / f"{name}.mkv").write_bytes(b"video")
    env = os.environ | {
        "SOURCE_ROOT": str(source_root),
        "DEST_ROOT": str(root / "dest"),
        "LOGFILE": str(root / "postprocess.log"),
        "LEASE_LOCK": str(root / "lease.lock"),
        "HANDOFF_URL": f"http://127.0.0.1:{server_port}/media/handoff",
        "POLL_SECONDS": "0",
        "MAX_POLLS": "2",
    }
    return subprocess.Popen(
        ["bash", str(SCRIPT), str(source), name + ".nzb", name, "", "serien", "", "0"],
        env=env,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )


def test_parallel_lease_blocks_are_serialized():
    global next_job, open_leases, max_open_leases, job_ids
    next_job = 0
    open_leases = 0
    max_open_leases = 0
    job_ids = {}
    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        with tempfile.TemporaryDirectory() as td:
            root = Path(td)
            first = start_script(root, "Show.S01E01", server.server_port)
            second = start_script(root, "Show.S01E02", server.server_port)
            first_result = first.communicate(timeout=20)
            second_result = second.communicate(timeout=20)
            assert first.returncode == 0, first_result
            assert second.returncode == 0, second_result
            assert len(set(job_ids.values())) == 2
            assert max_open_leases == 1, "start→move→moved blocks overlapped"
    finally:
        server.shutdown()
        server.server_close()


def test_lock_is_held_on_a_file_descriptor_until_register_returns():
    source = SCRIPT.read_text()
    assert 'exec 9>"$LEASE_LOCK"' in source
    assert 'flock -n 9' in source
    assert 'flock -u 9' in source


if __name__ == "__main__":
    test_lock_is_held_on_a_file_descriptor_until_register_returns()
    test_parallel_lease_blocks_are_serialized()
    print("PASS: parallel start→move→moved blocks are serialized")