"""Local account-bound conversation registry listing."""

from __future__ import annotations

import json
import multiprocessing
import os
import stat

import pytest

import ask_gpt
import chat
import registry
import solwebd
from fake_browser import seat_for


def _id(n: int) -> str:
    return f"{n:08x}-0000-4000-8000-000000000000"


def _append_batch(start: int, count: int, barrier) -> None:
    barrier.wait()
    for number in range(start, start + count):
        registry.append("a@example.com", _id(number), f"title {number}")


def _compete_for_owner(account: str, conversation_id: str, barrier, results) -> None:
    barrier.wait()
    try:
        registry.append(account, conversation_id, account)
        results.put(account)
    except registry.RegistryError:
        results.put("rejected")


@pytest.fixture
def local_registry(monkeypatch, tmp_path):
    monkeypatch.chdir(tmp_path)
    monkeypatch.setattr(registry, "STATE", tmp_path / "state")
    monkeypatch.setattr(registry, "PATH", tmp_path / "state" / "conversations.jsonl")
    monkeypatch.setattr(registry, "LOCK", tmp_path / "state" / "conversations.lock")
    return tmp_path


def test_list_is_local_only(local_registry, monkeypatch, capsys):
    registry.append("a@example.com", _id(1), "first")
    registry.append("a@example.com", _id(2), "second")
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: (_ for _ in ()).throw(AssertionError()))
    monkeypatch.setattr(chat, "bootstrap_registry_standalone", lambda **_: (_ for _ in ()).throw(AssertionError()))
    assert ask_gpt.main(["--list"]) == 0
    assert capsys.readouterr().out.splitlines() == [f"{_id(2)}\tsecond", f"{_id(1)}\tfirst"]


def test_list_rejects_nonpositive_limit(local_registry, capsys):
    with pytest.raises(SystemExit, match="2"):
        ask_gpt.main(["--list", "0"])
    assert "--list N must be greater than zero" in capsys.readouterr().err


@pytest.mark.parametrize("limit", ["-1", "-20"])
def test_list_rejects_negative_limit(local_registry, capsys, limit):
    with pytest.raises(SystemExit, match="2"):
        ask_gpt.main(["--list", limit])
    assert "--list N must be greater than zero" in capsys.readouterr().err


def test_search_uses_regex_then_glob_locally(local_registry, capsys):
    registry.append("a@example.com", _id(1), "fix deploy")
    registry.append("a@example.com", _id(2), "other")
    assert ask_gpt.main(["--search", "*deploy*"]) == 0
    assert capsys.readouterr().out.strip() == f"{_id(1)}\tfix deploy"


def test_list_all_writes_jsonl_copy_only(local_registry, monkeypatch, capsys):
    registry.append("a@example.com", _id(1), "first")
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: (_ for _ in ()).throw(AssertionError()))
    assert ask_gpt.main(["--list-all"]) == 0
    assert capsys.readouterr().out.strip() == "Log: ./ask-gpt/conversations.jsonl"
    assert json.loads((local_registry / "ask-gpt" / "conversations.jsonl").read_text()) == {
        "account": "a@example.com", "conversation_id": _id(1), "title": "first"}


def test_bootstrap_once_then_refuses_without_browser(local_registry, monkeypatch, capsys):
    calls = []
    monkeypatch.setattr(chat, "bootstrap_registry_standalone", lambda **_: (calls.append(1) or (
        "a@example.com", [{"id": _id(1), "title": "imported"}])))
    assert ask_gpt.main(["--bootstrap-registry"]) == 0
    assert calls == [1]
    monkeypatch.setattr(chat, "bootstrap_registry_standalone", lambda **_: (_ for _ in ()).throw(AssertionError()))
    assert ask_gpt.main(["--bootstrap-registry"]) == 1
    assert "already contains records" in capsys.readouterr().err


def test_daemon_bootstrap_reestablishes_root_before_auth_and_sidebar(monkeypatch):
    seat, marionette = seat_for([])
    engine = solwebd.Engine(seat)
    seat.open()
    marionette.navigations.clear()
    order = []

    monkeypatch.setattr(solwebd.chat, "wait_ready", lambda browser: order.append("ready"))
    monkeypatch.setattr(solwebd.chat, "authenticated_account", lambda browser: order.append("auth") or "a@example.com")
    monkeypatch.setattr(solwebd.chat, "list_all_threads", lambda browser: order.append("sidebar") or [])

    assert engine._bootstrap_registry() == {"account": "a@example.com", "threads": []}
    assert marionette.navigations == ["https://chatgpt.com/"]
    assert order == ["ready", "auth", "sidebar"]


def test_registry_rejects_malformed_and_conflicting_ownership(local_registry):
    first = _id(1)
    registry.PATH.parent.mkdir(parents=True)
    registry.PATH.write_text(f'{{"account":"a@example.com","conversation_id":"{first}","title":"one"}}\nnot-json\n')
    with pytest.raises(registry.RegistryError, match="invalid JSON"):
        registry.conversations()
    registry.PATH.write_text(
        f'{{"account":"a@example.com","conversation_id":"{first}","title":"one"}}\n'
        f'{{"account":"b@example.com","conversation_id":"{first}","title":"two"}}\n')
    with pytest.raises(registry.RegistryError, match="both"):
        registry.conversations()


def test_registry_normalizes_account_and_conversation_id(local_registry):
    item = registry.append(" Owner@Example.COM ", f" {_id(1).upper()} ", " title ")
    assert item.account == "owner@example.com"
    assert item.conversation_id == _id(1)
    assert registry.lookup(f" {_id(1).upper()} ") == item


@pytest.mark.parametrize("conversation_id", ["x", "00000000-0000-0000-0000-00000000000g", "prefix-" + _id(1)])
def test_registry_rejects_invalid_conversation_id(local_registry, conversation_id):
    with pytest.raises(registry.RegistryError, match="invalid conversation_id"):
        registry.append("a@example.com", conversation_id, "title")


@pytest.mark.parametrize(("field", "value"), [
    ("account", "a@example.com\nforged"),
    ("title", "safe\x1b[2Jforged"),
    ("url", "https://chatgpt.com/\tforged"),
])
def test_registry_rejects_control_characters(local_registry, field, value):
    values = {"account": "a@example.com", "conversation_id": _id(1),
              "title": "title", "url": "https://chatgpt.com/"}
    values[field] = value
    with pytest.raises(registry.RegistryError, match="control character"):
        registry.append(**values)


def test_registry_files_are_private(local_registry):
    previous = os.umask(0)
    try:
        registry.append("a@example.com", _id(1), "title")
    finally:
        os.umask(previous)
    assert stat.S_IMODE(registry.PATH.stat().st_mode) == 0o600
    assert stat.S_IMODE(registry.LOCK.stat().st_mode) == 0o600


def test_registry_repairs_existing_file_permissions(local_registry):
    registry.PATH.parent.mkdir(parents=True)
    registry.PATH.touch()
    registry.LOCK.touch()
    registry.PATH.chmod(0o644)
    registry.LOCK.chmod(0o644)

    registry.append("a@example.com", _id(1), "title")

    assert stat.S_IMODE(registry.PATH.stat().st_mode) == 0o600
    assert stat.S_IMODE(registry.LOCK.stat().st_mode) == 0o600


def test_registry_read_repairs_existing_jsonl_permissions(local_registry):
    registry.append("a@example.com", _id(1), "title")
    registry.PATH.chmod(0o644)

    assert len(registry.conversations()) == 1

    assert stat.S_IMODE(registry.PATH.stat().st_mode) == 0o600


def test_concurrent_registry_writers_keep_complete_unique_records(local_registry):
    context = multiprocessing.get_context("fork")
    barrier = context.Barrier(4)
    writers = [context.Process(target=_append_batch, args=(index * 25 + 1, 25, barrier))
               for index in range(4)]
    for writer in writers:
        writer.start()
    for writer in writers:
        writer.join(10)

    assert [writer.exitcode for writer in writers] == [0, 0, 0, 0]
    raw_records = [json.loads(line) for line in registry.PATH.read_text().splitlines()]
    assert len(raw_records) == 100
    assert len({record["conversation_id"] for record in raw_records}) == 100
    assert len(registry.conversations()) == 100


def test_concurrent_registry_writers_cannot_change_ownership(local_registry):
    context = multiprocessing.get_context("fork")
    barrier = context.Barrier(2)
    results = context.Queue()
    writers = [context.Process(target=_compete_for_owner,
                               args=(account, _id(1), barrier, results))
               for account in ("a@example.com", "b@example.com")]
    for writer in writers:
        writer.start()
    for writer in writers:
        writer.join(10)

    assert [writer.exitcode for writer in writers] == [0, 0]
    outcomes = {results.get(timeout=1), results.get(timeout=1)}
    assert outcomes == {"rejected", registry.lookup(_id(1)).account}
    assert len(registry.PATH.read_text().splitlines()) == 1


def test_registry_latest_title_and_order(local_registry):
    registry.append("a@example.com", _id(1), "old")
    registry.append("a@example.com", _id(2), "other")
    registry.append("a@example.com", _id(1), "new")
    assert [(x.conversation_id, x.title) for x in registry.conversations()] == [(_id(1), "new"), (_id(2), "other")]