"""Real private PostgreSQL/RabbitMQ control plane, no user program execution."""

import asyncio
import json
import os
import secrets
import subprocess
import sys
from pathlib import Path
from urllib.parse import urlsplit
from uuid import uuid4

import aio_pika
import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from test_problems import import_fixture, platform_role

from kf_oj.api.app import create_app
from kf_oj.contracts.submissions import CreateSubmission
from kf_oj.core.settings import Settings
from kf_oj.infra.messaging.publisher import QUEUE, ConfirmedBroker, OutboxRepository


def test_actual_confirm_duplicate_lock_recovery_and_broker_failure(
    postgres_url, tmp_path, monkeypatch
):
    asyncio.run(outbox_acceptance(postgres_url, tmp_path, monkeypatch))


async def outbox_acceptance(postgres_url, tmp_path, monkeypatch):
    config = os.environ.get("KF_OJ_TEST_AMQP_CONFIG")
    if config is None:
        pytest.skip("private RabbitMQ control acceptance requires KF_OJ_TEST_AMQP_CONFIG")
    url = json.loads(Path(config).read_text())["amqp_url"]
    parsed = urlsplit(url)
    assert parsed.hostname == "127.0.0.1" and parsed.port == 26572
    _, policy = import_fixture(tmp_path / "import")
    engine = create_async_engine(postgres_url)
    platform_role(engine)
    monkeypatch.setattr("kf_oj.api.app.build_engine", lambda _: engine)
    app = create_app(
        Settings(
            _env_file=None,
            environment="test",
            database_url=postgres_url,
            artifact_root=tmp_path / "artifacts",
            judge_admission=policy,
        )
    )
    broker = ConfirmedBroker(url)
    connection = await aio_pika.connect(url, timeout=5)
    try:
        async with app.router.lifespan_context(app):
            owner = await app.state.identity.create_account(
                "outbox_test", secrets.token_urlsafe(24)
            )
            version = await app.state.problems.import_directory(tmp_path / "import")
            created = await app.state.submissions.create(
                owner,
                CreateSubmission(
                    problem_version_id=version,
                    language="c17",
                    idempotency_key=uuid4(),
                    source="synthetic control event; not compiled",
                ),
            )
            repo = OutboxRepository(app.state.sessions)
            claimed = await asyncio.gather(repo.claim(), repo.claim())
            assert sum(v is not None for v in claimed) == 1
            item = next(v for v in claimed if v is not None)
            assert item[1] == created.evaluation_id
            # Kill a separate real publisher process after broker confirm, before DB mark.
            code = """import asyncio,json,sys,signal
from uuid import UUID
from pathlib import Path
from kf_oj.infra.messaging.publisher import ConfirmedBroker
async def main():
 item=tuple(UUID(x) for x in json.loads(sys.stdin.readline()))
 broker=ConfirmedBroker(json.loads(Path(sys.argv[1]).read_text())["amqp_url"])
 await broker.publish(item)
 print("CONFIRMED",flush=True)
 await asyncio.Event().wait()
asyncio.run(main())
"""
            child = subprocess.Popen(
                [sys.executable, "-c", code, config],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
            )
            try:
                child.stdin.write(json.dumps([str(x) for x in item]) + "\n")
                child.stdin.flush()
                assert (
                    await asyncio.wait_for(asyncio.to_thread(child.stdout.readline), 10)
                    == "CONFIRMED\n"
                )
            finally:
                child.kill()
                await asyncio.to_thread(child.wait, 5)
                child.stdin.close()
                child.stdout.close()
                child.stderr.close()
            assert child.returncode == -9
            channel = await connection.channel()
            queue = await channel.get_queue(QUEUE)
            first = await queue.get(timeout=5)
            assert json.loads(first.body) == {
                "schema_version": 1,
                "kind": "judge.evaluation",
                "evaluation_id": str(created.evaluation_id),
            }
            assert len(first.body) <= 512 and first.delivery_mode == 2
            # A publisher lost after confirm cannot mark DB; actual DB clock lock expires.
            assert await repo.claim() is None
            await asyncio.sleep(15.1)
            retry = await repo.claim()
            assert retry[:2] == item[:2] and retry[2] != item[2]
            assert not await repo.confirmed(item)
            await broker.publish(retry)
            assert await repo.confirmed(retry)
            second = await queue.get(timeout=5)
            assert second.body == first.body and second.message_id == first.message_id
            await first.ack()
            await second.ack()
            assert await repo.claim() is None
            async with app.state.sessions() as s:
                assert (
                    await s.scalar(
                        text("SELECT count(*) FROM outbox WHERE delivered_at IS NOT NULL")
                    )
                    == 1
                )
            # A second transaction succeeds even when a separate broker is unavailable.
            await app.state.submissions.create(
                owner,
                CreateSubmission(
                    problem_version_id=version,
                    language="c17",
                    idempotency_key=uuid4(),
                    source="broker failure synthetic event",
                ),
            )
            failed = await repo.claim()
            unavailable = ConfirmedBroker(url.replace(":26572/", ":26579/"))
            try:
                with pytest.raises((ConnectionError, OSError)):
                    await unavailable.publish(failed)
            finally:
                await unavailable.close()
            await repo.failed(failed)
            assert await repo.claim() is None
            async with app.state.sessions() as s:
                row = (
                    await s.execute(
                        text("""SELECT failures,last_error_code,
                    next_attempt_at>clock_timestamp() deferred,delivered_at IS NULL pending,
                    lock_owner IS NULL unlocked FROM outbox WHERE id=:id"""),
                        {"id": failed[0]},
                    )
                ).one()
                assert tuple(row) == (1, "broker_unavailable", True, True, True)
            # Preserve the broker-failure row for inspection while testing another crash window.
            async with app.state.sessions.begin() as s:
                await s.execute(
                    text(
                        "UPDATE outbox SET next_attempt_at=clock_timestamp()+interval '1 hour' "
                        "WHERE id=:id"
                    ),
                    {"id": failed[0]},
                )
            before = await app.state.submissions.create(
                owner,
                CreateSubmission(
                    problem_version_id=version,
                    language="c17",
                    idempotency_key=uuid4(),
                    source="before-confirm synthetic event",
                ),
            )
            pending = await repo.claim()
            assert pending[1] == before.evaluation_id
            preconfirm = code.replace("await broker.publish(item)", "await broker.open()")
            child = subprocess.Popen(
                [sys.executable, "-c", preconfirm, config],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
            )
            try:
                child.stdin.write(json.dumps([str(x) for x in pending]) + "\n")
                child.stdin.flush()
                assert (
                    await asyncio.wait_for(asyncio.to_thread(child.stdout.readline), 10)
                    == "CONFIRMED\n"
                )
            finally:
                child.kill()
                await asyncio.to_thread(child.wait, 5)
                child.stdin.close()
                child.stdout.close()
                child.stderr.close()
            assert child.returncode == -9
            assert await queue.get(fail=False, timeout=5) is None
            await asyncio.sleep(15.1)
            retry = await repo.claim()
            assert retry[:2] == pending[:2]
            await broker.publish(retry)
            assert await repo.confirmed(retry)
            delivered = await queue.get(timeout=5)
            assert json.loads(delivered.body)["evaluation_id"] == str(before.evaluation_id)
            await delivered.ack()

    finally:
        await broker.close()
        await connection.close()


def test_actual_transport_failure_does_not_log_connection_credentials(caplog):
    """Local protocol refusal exercises descendant client loggers without a broker/DB."""

    async def refused_connection():
        touched = asyncio.Event()

        async def refuse(reader, writer):
            touched.set()
            writer.close()
            await writer.wait_closed()

        listener = await asyncio.start_server(refuse, "127.0.0.1", 0)
        port = listener.sockets[0].getsockname()[1]
        broker = ConfirmedBroker(f"amqp://fixture:private-sentinel@127.0.0.1:{port}/")
        try:
            with pytest.raises(aio_pika.exceptions.AMQPConnectionError):
                await broker.publish((uuid4(), uuid4(), uuid4()))
            assert touched.is_set(), "must reach actual local transport refusal"
        finally:
            await broker.close()
            listener.close()
            await listener.wait_closed()

    asyncio.run(refused_connection())
    assert "private-sentinel" not in caplog.text
    assert "amqp://" not in caplog.text