from __future__ import annotations

import json
import os
import socket
import ssl
import subprocess
import sys
import tarfile
import threading
import time
import urllib.request
from contextlib import contextmanager
from pathlib import Path

import pytest

if os.getenv("STACKCHAIN_RUN_RELEASE_E2E") != "1":
    pytest.skip("release artifact browser journey runs only in its gated CI job", allow_module_level=True)
pytest.importorskip("playwright.sync_api")
from playwright.sync_api import Page, expect, sync_playwright

from fake_gitea import FakeGiteaServer


ROOT = Path(__file__).resolve().parents[2]
TITLE = "Offline artifact journey 863."
BODY = "Captured on a phone, retained offline, delivered exactly once."
ACCESS_TOKEN = "artifact-browser-access-token-863"


def free_port() -> int:
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return int(sock.getsockname()[1])


def wait_for_health(url: str, process: subprocess.Popen[str]) -> None:
    deadline = time.monotonic() + 20
    tls = ssl._create_unverified_context()
    while time.monotonic() < deadline:
        if process.poll() is not None:
            stdout, stderr = process.communicate()
            raise AssertionError(f"release server exited early\nstdout:\n{stdout}\nstderr:\n{stderr}")
        try:
            with urllib.request.urlopen(url, timeout=0.5, context=tls) as response:
                if response.status == 200:
                    return
        except OSError:
            time.sleep(0.05)
    raise AssertionError("release server did not become healthy")


@contextmanager
def release_server(archive: Path, tmp_path: Path, gitea_url: str):
    release_root = tmp_path / "release"
    release_root.mkdir()
    with tarfile.open(archive, "r:gz") as bundle:
        bundle.extractall(release_root, filter="data")
    assert (release_root / "release-manifest.json").is_file()
    assert not (release_root / "tests").exists(), "journey must run the packaged artifact, not checkout code"

    port = free_port()
    state_dir = tmp_path / "state"
    state_dir.mkdir()
    key_file = tmp_path / "localhost.key"
    certificate_file = tmp_path / "localhost.crt"
    subprocess.run(
        [
            "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
            "-keyout", str(key_file), "-out", str(certificate_file), "-days", "1",
            "-subj", "/CN=127.0.0.1", "-addext", "subjectAltName=IP:127.0.0.1",
        ],
        check=True,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    origin = f"https://127.0.0.1:{port}"
    env = {
        **os.environ,
        "PYTHONPATH": str(release_root),
        "GITEA_URL": gitea_url,
        "GITEA_TOKEN": "deterministic-e2e-token",
        "STACKCHAIN_DASHBOARD_AUTH_MODE": "operator",
        "STACKCHAIN_DASHBOARD_ACCESS_TOKEN": ACCESS_TOKEN,
        "STACKCHAIN_DASHBOARD_SESSION_SECRET": "artifact-browser-independent-session-secret-863",
        "STACKCHAIN_UNFILED_DRAFT_ENCRYPTION_KEY": (
            "ZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGQ="
        ),
        "STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN": origin,
        "STACKCHAIN_STATE_DIR": str(tmp_path / "state"),
    }
    process = subprocess.Popen(
        [
            sys.executable, "-m", "uvicorn", "src.main:app",
            "--host", "127.0.0.1", "--port", str(port),
            "--ssl-keyfile", str(key_file), "--ssl-certfile", str(certificate_file),
        ],
        cwd=release_root,
        env=env,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    try:
        wait_for_health(origin + "/healthz", process)
        yield origin
    finally:
        process.terminate()
        try:
            process.communicate(timeout=10)
        except subprocess.TimeoutExpired:
            process.kill()
            process.communicate()


def hydrate_workspace(page) -> None:
    page.evaluate(
        "async () => { const lifecycle = await window.stackchainWorkspaceLifecycle; "
        "await lifecycle.hydrateWorkspace(); await lifecycle.workspaceReady; }"
    )


def indexed_issue_records(page: Page) -> list[dict]:
    return page.evaluate(
        """async () => {
          const db = await new Promise((resolve, reject) => {
            const request = indexedDB.open('stackchain-background-outbox-v1', 1);
            request.onsuccess = () => resolve(request.result);
            request.onerror = () => reject(request.error);
          });
          return await new Promise((resolve, reject) => {
            const request = db.transaction('issues', 'readonly').objectStore('issues').getAll();
            request.onsuccess = () => resolve(request.result);
            request.onerror = () => reject(request.error);
          });
        }"""
    )


def test_release_artifact_files_one_mobile_issue_exactly_once_after_offline_reload(tmp_path: Path):
    archives = sorted((ROOT / "dist").glob("stackchain-dashboard-*.tar.gz"))
    assert len(archives) == 1, "browser job must download exactly one assembled release archive"

    fake = FakeGiteaServer(("127.0.0.1", 0))
    fake_thread = threading.Thread(target=fake.serve_forever, daemon=True)
    fake_thread.start()
    fake_url = f"http://127.0.0.1:{fake.server_port}"
    browser_errors: list[str] = []
    failed_responses: list[str] = []


    try:
        with release_server(archives[0], tmp_path, fake_url) as origin, sync_playwright() as playwright:
            browser = playwright.chromium.launch(args=["--ignore-certificate-errors"])
            context = browser.new_context(
                viewport={"width": 390, "height": 844},
                service_workers="allow",
                ignore_https_errors=True,
            )
            context.add_init_script("""
              class DeterministicSpeechRecognition {
                start() { window.__voiceRecognition = this; }
                stop() { if (this.onend) this.onend(); }
                abort() { if (this.onend) this.onend(); }
              }
              window.SpeechRecognition = DeterministicSpeechRecognition;
            """)
            page = context.new_page()
            page.on("pageerror", lambda error: browser_errors.append(error.stack or str(error)))
            page.on("console", lambda message: browser_errors.append(message.text) if message.type == "error" else None)
            page.on("response", lambda response: failed_responses.append(
                f"{response.status} {response.url}"
            ) if response.status >= 400 else None)
            page.goto(origin + "/", wait_until="networkidle")
            expect(page.locator("#sign-in")).to_be_visible()
            page.locator('input[name="device_label"]').fill("Release journey phone")
            page.locator('input[name="access_token"]').fill(ACCESS_TOKEN)
            page.locator("#submit-sign-in").click()
            page.wait_for_url(origin + "/", wait_until="networkidle")
            new_action = page.locator('[data-mobile-task="new"]')
            expect(new_action).to_be_visible()
            service_worker_ready = page.evaluate(
                """() => Promise.race([
                  navigator.serviceWorker.ready.then(() => true),
                  new Promise(resolve => setTimeout(() => resolve(false), 10000)),
                ])"""
            )
            assert service_worker_ready, (
                "release artifact did not register its service worker; browser errors=" + repr(browser_errors)
                + "; failed responses=" + repr(failed_responses)
            )
            if not page.evaluate("() => navigator.serviceWorker.controller !== null"):
                page.reload(wait_until="networkidle")
                new_action = page.locator('[data-mobile-task="new"]')
            assert page.evaluate("() => navigator.serviceWorker.controller !== null")

            new_action.click()
            expect(page.locator("#create-issue-sheet")).to_have_class("create-issue-sheet open")
            assert page.evaluate("document.activeElement?.id") == "cancel-new-issue"
            assert page.evaluate("document.querySelector('main').inert") is True
            page.keyboard.press("Shift+Tab")
            assert page.evaluate("document.activeElement.closest('#create-issue-sheet')?.id") == "create-issue-sheet"
            page.keyboard.press("Escape")
            expect(page.locator("#create-issue-sheet")).not_to_have_class("create-issue-sheet open")
            assert page.evaluate("document.activeElement?.dataset.mobileTask") == "new"
            assert page.evaluate("document.querySelector('main').inert") is False
            new_action.click()
            expect(page.locator("#create-issue-sheet")).to_have_class("create-issue-sheet open")
            assert page.evaluate("document.activeElement?.id") == "cancel-new-issue"
            create_navigation = page.locator(".mobile-create-issue-nav")
            expect(create_navigation).to_be_visible()
            create_navigation_buttons = create_navigation.locator("button")
            assert create_navigation_buttons.count() == 3
            for control in create_navigation_buttons.all():
                bounds = control.bounding_box()
                assert bounds and bounds["height"] >= 44
            expect(create_navigation.locator('[data-create-issue-section="describe"]')).to_have_attribute(
                "aria-current", "location"
            )
            expect(create_navigation.locator('[data-create-issue-section="file"]')).to_have_attribute(
                "aria-disabled", "true"
            )
            create_navigation.locator('[data-create-issue-section="evidence"]').click()
            expect(create_navigation.locator('[data-create-issue-section="evidence"]')).to_have_attribute(
                "aria-current", "location"
            )
            for control in page.locator(".photo-evidence-actions .issue-attachment-trigger").all():
                bounds = control.bounding_box()
                assert bounds and bounds["height"] >= 44
            assert page.evaluate("document.documentElement.scrollWidth <= window.innerWidth")
            expect(page.locator("#voice-issue-capture")).to_be_visible()
            page.locator("#start-voice-issue-capture").click()
            page.evaluate("""([title, body]) => {
              const result = [{ transcript: title + ' ' + body }];
              result.isFinal = true;
              window.__voiceRecognition.onresult({ results: [result] });
            }""", [TITLE, BODY])
            expect(page.locator("#voice-issue-review")).to_be_visible()
            expect(page.locator("#append-voice-issue-transcript")).to_have_text("Use transcript")
            page.locator("#append-voice-issue-transcript").click()
            expect(page.locator("#create-issue-title")).to_have_value(TITLE)
            expect(page.locator("#create-issue-body")).to_have_value(BODY)
            page.locator("#file-new-issue").click()
            expect(page.locator("#create-issue-filing")).to_be_visible()
            expect(create_navigation.locator('[data-create-issue-section="file"]')).to_have_attribute(
                "aria-current", "location"
            )
            expect(create_navigation.locator('[data-create-issue-section="file"]')).not_to_have_attribute(
                "aria-disabled", "true"
            )
            page.locator("#create-issue-repository").select_option("acme/mobile")
            expect(page.locator("#submit-new-issue")).to_be_enabled()

            # Page offline emulation can leave service-worker requests connected.
            # Abort the mutation at the browser boundary too, matching a real
            # transport outage without poisoning server-side idempotency state.
            context.route("**/api/v1/repos/acme/mobile/issues", lambda route: route.abort())
            context.set_offline(True)
            page.locator("#submit-new-issue").click()
            expect(page.locator("#issue-filing-review")).to_be_visible()
            page.locator("#confirm-issue-filing").click()
            page.wait_for_timeout(1_000)
            filing_statuses = page.locator(
                "#my-work-action-status, #create-issue-status, #issue-filing-review-status"
            ).all_inner_texts()
            assert any(
                marker in " ".join(filing_statuses).lower()
                for marker in ("queued", "saved for", "background delivery")
            ), (filing_statuses, browser_errors, failed_responses)

            local_record = json.loads(page.evaluate("localStorage.getItem('stackchain.issue-outbox.v1')"))
            assert [(item["title"], item["body"], item["repository"]) for item in local_record["items"]] == [
                (TITLE, BODY, "acme/mobile")
            ]
            durable = indexed_issue_records(page)
            assert [(item["title"], item["body"], item["repository"]) for item in durable] == [
                (TITLE, BODY, "acme/mobile")
            ]
            operation_id = durable[0]["operationId"]
            assert operation_id and operation_id == local_record["items"][0]["operationId"]
            assert fake.created_issues == []

            page.reload(wait_until="domcontentloaded")
            hydrate_workspace(page)
            expect(page.locator('[data-mobile-task="new"]')).to_be_visible()
            durable_after_reload = indexed_issue_records(page)
            assert [(item["title"], item["body"]) for item in durable_after_reload] == [(TITLE, BODY)]
            assert durable_after_reload[0]["operationId"] == operation_id
            assert fake.created_issues == []

            browser_errors.clear()  # Chromium reports expected network errors while the context is offline.
            failed_responses.clear()
            fake.issue_creation_enabled = True
            fake.issue_creation_ready.set()
            context.unroute("**/api/v1/repos/acme/mobile/issues")
            context.set_offline(False)
            page.evaluate("window.dispatchEvent(new Event('online'))")
            # Containerized Actions runners can take longer than a local browser to wake the
            # service worker and complete its identity handshake after reconnect.
            for _ in range(160):
                if fake.created_issues:
                    break
                page.wait_for_timeout(250)
            local_debug = page.evaluate("localStorage.getItem('stackchain.issue-outbox.v1')")
            durable_debug = indexed_issue_records(page)
            assert fake.created_issues == [
                {"title": TITLE, "body": BODY, "assignee": "timmy"}
            ], (
                f"url={page.url} ready={page.evaluate('document.readyState')} "
                f"status={page.locator('#my-work-action-status').inner_text()!r} "
                f"errors={browser_errors[-5:]!r} responses={failed_responses[-10:]!r} "
                f"local={local_debug!r} durable={durable_debug!r} "
                f"requests={fake.requests[-20:]!r}"
            )
            # Deferred workspace startup can report its already-issued offline request
            # after the pre-reconnect clear. Chromium uses either its explicit offline
            # code or a generic fetch rejection for that same request, depending on
            # whether the service worker or page observes the disconnect first.
            browser_errors[:] = [
                error for error in browser_errors
                if "ERR_INTERNET_DISCONNECTED" not in error
                and not error.startswith("context failed TypeError: Failed to fetch")
            ]

            for _ in range(40):
                durable_completion = indexed_issue_records(page)
                if durable_completion and all(item.get("status") == "sent" for item in durable_completion):
                    break
                page.wait_for_timeout(100)
            assert durable_completion and all(item.get("status") == "sent" for item in durable_completion)
            page.reload(wait_until="networkidle")
            hydrate_workspace(page)
            page.evaluate("window.dispatchEvent(new Event('online'))")
            for _ in range(40):
                completed_local = json.loads(page.evaluate("localStorage.getItem('stackchain.issue-outbox.v1')"))
                if completed_local["items"] == []:
                    break
                page.wait_for_timeout(100)
            assert completed_local["items"] == []
            page.wait_for_timeout(500)
            assert len(fake.created_issues) == 1
            assert browser_errors == []
            assert failed_responses == []
            assert page.evaluate("document.documentElement.scrollWidth <= window.innerWidth")
            browser.close()
    finally:
        fake.issue_creation_ready.set()
        fake.shutdown()
        fake.server_close()
        fake_thread.join(timeout=5)