"""W4P worker executes Pages Functions before the platform catalog."""

import json
import hashlib
import shutil
import subprocess
from pathlib import Path

import pytest

from src.domain.hosting.runtime.worker_source import (
    ApplicationRouteError,
    application_prelude,
    build_worker_source,
)
from src.domain.hosting.w4p_workflow import W4P_UPLOAD_SCRIPT

RUNTIME = Path("src/domain/codegen/runtime")
_FETCH = """
import { readFileSync } from 'node:fs';
const mod = await import('data:text/javascript,' + encodeURIComponent(readFileSync(process.argv[1], 'utf8')));
const seen = [];
const waits = [];
globalThis.fetch = async (url) => {
  seen.push(String(url));
  if (String(url).includes("unavailable")) throw new Error("upstream down");
  return new Response(JSON.stringify({ ok: true }), {
    status: 200,
    headers: { 'content-type': 'application/json' },
  });
};
const env = {
  ASSETS: { fetch: async () => new Response('asset', { status: 200 }) },
  PROJECT_ID: 'project-1',
  CLOUD_GATEWAY_URL: 'https://cloud.example',
};
const method = process.argv[3] || 'GET';
const init = { method };
if (method !== 'GET' && method !== 'HEAD') init.body = process.argv[4] || '{}';
const ctx = { waitUntil(promise) { waits.push(Promise.resolve(promise)); } };
const response = await mod.default.fetch(new Request(process.argv[2], init), env, ctx);
await Promise.all(waits);
process.stdout.write(JSON.stringify({
  status: response.status,
  body: await response.text(),
  seen,
  waits: waits.length,
}));
"""


def _write(root: Path, rel: str, text: str) -> None:
    path = root / rel
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")


def _lock_application_route(root: Path, rel: str, route: str) -> None:
    target = root / rel
    lock = root / "p2hs" / "platform-ports.lock.json"
    lock.parent.mkdir(parents=True, exist_ok=True)
    lock.write_text(json.dumps({
        "files": {rel: hashlib.sha256(target.read_bytes()).hexdigest()},
        "routes": [route],
    }), encoding="utf-8")


def _fetch(source: str, url: str, method: str = "GET", body: str = "") -> dict:
    # The worker source is larger than Windows' command-line limit once the
    # session bridge is prepended. Node reads it from a file instead of argv.
    import tempfile

    with tempfile.NamedTemporaryFile("w", suffix=".mjs", delete=False, encoding="utf-8") as handle:
        handle.write(source)
        source_path = handle.name
    try:
        result = subprocess.run(
            ["node", "--input-type=module", "-e", _FETCH, source_path, url, method, body],
            capture_output=True,
            text=True,
            encoding="utf-8",
        )
    finally:
        Path(source_path).unlink(missing_ok=True)
    assert result.returncode == 0, result.stderr or result.stdout
    return json.loads(result.stdout)


def _worker(root: Path, routes: list | None = None) -> str:
    return build_worker_source(routes or [], repo_root=root)


def test_upload_script_compiles_application_routes_from_the_repo():
    assert "bundleApplicationRoutes" in W4P_UPLOAD_SCRIPT
    assert "dispatchApplication" in W4P_UPLOAD_SCRIPT
    assert "${{" not in W4P_UPLOAD_SCRIPT
    assert "server/api" not in W4P_UPLOAD_SCRIPT


def test_application_route_runs_before_the_catalog(tmp_path: Path):
    _write(tmp_path, "functions/api/forms/[formKey]/submissions.js", """
export async function onRequest({ params }) {
  return Response.json({ formKey: params.formKey, owner: "application" });
}
""")
    _lock_application_route(tmp_path, "functions/api/forms/[formKey]/submissions.js", "/api/forms/:form_key/submissions")
    source = _worker(tmp_path, [{
        "id": "forms.submit",
        "method": "POST",
        "public_path": "/api/forms/:form_key/submissions",
        "auth": "public",
        "upstream_path": "/v1/projects/:project_id/forms/:form_key/submissions",
    }])
    hit = _fetch(source, "https://proj.example/api/forms/quote/submissions", "POST", "{}")
    assert hit["status"] == 200
    assert json.loads(hit["body"]) == {"formKey": "quote", "owner": "application"}
    assert hit["seen"] == []
    catalog = _fetch(source, "https://proj.example/api/records/orders/query", "POST", "{}")
    assert catalog["status"] == 404
    page = _fetch(source, "https://proj.example/")
    assert page["body"] == "asset"


def test_catalog_route_still_proxies_when_the_app_does_not_own_it(tmp_path: Path):
    _write(tmp_path, "functions/api/bookings/hold.js", """
export async function onRequest() {
  return Response.json({ hold: true });
}
""")
    source = _worker(tmp_path, [{
        "id": "records.query",
        "method": "POST",
        "public_path": "/api/records/:entity/query",
        "auth": "session",
        "upstream_path": "/v1/projects/:project_id/records/query",
    }])
    owned = _fetch(source, "https://proj.example/api/bookings/hold", "POST", "{}")
    assert json.loads(owned["body"]) == {"hold": True}
    assert owned["seen"] == []
    proxied = _fetch(source, "https://proj.example/api/records/orders/query", "POST", '{"limit":1}')
    assert proxied["status"] == 200
    assert any("/records/query" in url for url in proxied["seen"])


def test_catalog_upstream_failure_is_normalized(tmp_path: Path):
    source = _worker(tmp_path, [{
        "id": "probe.failure",
        "method": "GET",
        "public_path": "/api/probe/failure",
        "auth": "public",
        "upstream_path": "/v1/unavailable",
    }])
    hit = _fetch(source, "https://proj.example/api/probe/failure")
    assert hit["status"] == 502
    body = json.loads(hit["body"])
    assert body["error"] == "platform_backend_unavailable"
    assert body["correlation_id"]


def test_unlocked_application_handler_cannot_shadow_a_platform_route(tmp_path: Path):
    _write(tmp_path, "functions/api/forms/[formKey]/submissions.js", """
export async function onRequest() {
  return Response.json({ owner: "unauthorized" });
}
""")
    with pytest.raises(ApplicationRouteError, match="protected_platform_route_collision"):
        _worker(tmp_path, [{
            "id": "forms.submit",
            "method": "POST",
            "public_path": "/api/forms/:form_key/submissions",
            "auth": "public",
            "upstream_path": "/v1/projects/:project_id/forms/:form_key/submissions",
        }])


def test_exact_route_wins_over_an_optional_catch_all(tmp_path: Path):
    _write(tmp_path, "functions/api/custom-files.js", """
export async function onRequest() {
  return Response.json({ exact: true });
}
""")
    _write(tmp_path, "functions/api/custom-files/[[path]].js", """
export async function onRequest({ params }) {
  return Response.json({ path: params.path });
}
""")
    source = _worker(tmp_path)
    exact = _fetch(source, "https://proj.example/api/custom-files")
    nested = _fetch(source, "https://proj.example/api/custom-files/a/b")
    assert json.loads(exact["body"]) == {"exact": True}
    assert json.loads(nested["body"]) == {"path": ["a", "b"]}


def test_dynamic_method_route_receives_params_and_wait_until(tmp_path: Path):
    _write(tmp_path, "functions/api/bookings/[id].post.js", """
export async function onRequestPost({ params, waitUntil }) {
  waitUntil(Promise.resolve("done"));
  return Response.json({ id: params.id });
}
""")
    hit = _fetch(_worker(tmp_path), "https://proj.example/api/bookings/r-42", "POST")
    assert hit["status"] == 200
    assert json.loads(hit["body"]) == {"id": "r-42"}
    assert hit["waits"] == 1


def test_application_handler_failure_is_normalized(tmp_path: Path):
    _write(tmp_path, "functions/api/booking/fail.js", """
export async function onRequest() { throw new Error("secret detail"); }
""")
    hit = _fetch(_worker(tmp_path), "https://proj.example/api/booking/fail")
    assert hit["status"] == 500
    assert json.loads(hit["body"]) == {"error": "application_handler_failed", "route": "functions/api/booking/fail.js"}


def test_two_application_files_on_one_path_fail_the_build(tmp_path: Path):
    handler = "export async function onRequest() { return new Response('x'); }\n"
    _write(tmp_path, "functions/api/forms/[formKey]/submissions.js", handler)
    _write(tmp_path, "functions/api/forms/[form_key]/submissions.js", handler)
    with pytest.raises(ApplicationRouteError, match="application_route_collision"):
        application_prelude(tmp_path)


def test_node_import_and_process_fail_the_build(tmp_path: Path):
    _write(tmp_path, "functions/api/bookings/hold.js", """
import { readFile } from "node:fs";
export async function onRequest() { return new Response(String(readFile)); }
""")
    with pytest.raises(ApplicationRouteError, match="unsupported_import:node:fs"):
        application_prelude(tmp_path)
    _write(tmp_path, "functions/api/bookings/hold.js", """
export async function onRequest() {
  return Response.json({ value: process.env.SECRET });
}
""")
    with pytest.raises(ApplicationRouteError, match="unsupported_node_api"):
        application_prelude(tmp_path)


def test_typescript_pages_function_fails_the_build(tmp_path: Path):
    _write(
        tmp_path,
        "functions/api/bookings/hold.ts",
        "export async function onRequest() { return new Response('x'); }\n",
    )
    with pytest.raises(ApplicationRouteError, match="unsupported_typescript"):
        application_prelude(tmp_path)


def test_commented_process_env_and_real_form_handler_still_bundle(tmp_path: Path):
    lib = tmp_path / "functions" / "_lib"
    lib.mkdir(parents=True)
    destination = tmp_path / "functions" / "api" / "forms" / "[formKey]" / "submissions.js"
    destination.parent.mkdir(parents=True)
    shutil.copy(RUNTIME / "p2hs-records.mjs", lib / "p2hs-records.mjs")
    shutil.copy(RUNTIME / "p2hs-forms-submit.mjs", lib / "p2hs-forms-submit.mjs")
    shutil.copy(RUNTIME / "forms-submit-pages-fn.js", destination)
    _lock_application_route(tmp_path, "functions/api/forms/[formKey]/submissions.js", "/api/forms/:form_key/submissions")
    source = _worker(tmp_path)
    rejected = _fetch(
        source,
        "https://proj.example/api/forms/quote/submissions",
        "POST",
        json.dumps({"project_id": "spoofed"}),
    )
    assert rejected["status"] == 400
    assert json.loads(rejected["body"])["error"] == "invalid_request"
    assert rejected["seen"] == []


def test_magic_link_consume_stays_on_the_runtime_when_the_app_proxies_it(tmp_path: Path):
    _write(tmp_path, "functions/api/auth/magic-link/consume.js", """
export function onRequest() {
  return Response.json({ owner: "application-proxy" });
}
""")
    _lock_application_route(
        tmp_path,
        "functions/api/auth/magic-link/consume.js",
        "/api/auth/magic-link/consume",
    )
    source = _worker(tmp_path, [{
        "id": "auth.magic_link_consume",
        "method": "GET",
        "public_path": "/api/auth/magic-link/consume",
        "auth": "public",
        "upstream_path": "/v1/projects/:project_id/auth/magic-link/consume",
    }])
    sign_in = _fetch(source, "https://proj.example/api/auth/magic-link/consume?token=expired")
    assert sign_in["status"] == 401
    assert "expired" in sign_in["body"].lower()
    assert "application-proxy" not in sign_in["body"]


def test_customer_route_cannot_claim_the_edge_function_facade(tmp_path: Path):
    _write(tmp_path, "functions/api/fn/[slug].js", """
export async function onRequest() {
  return Response.json({ owner: "application" });
}
""")
    with pytest.raises(ApplicationRouteError, match="edge_function_facade_reserved"):
        _worker(tmp_path)


def test_customer_route_cannot_open_a_second_cloud_client(tmp_path: Path):
    _write(tmp_path, "functions/api/quotes/price.js", """
export async function onRequest() {
  return fetch("https://cloud.example/v1/projects/project-1/records/query");
}
""")
    with pytest.raises(ApplicationRouteError, match="customer_platform_boundary:direct_cloud_route"):
        _worker(tmp_path)


def test_customer_route_uses_the_platform_sdk_for_billable_calls(tmp_path: Path):
    _write(tmp_path, "functions/api/quotes/price.js", """
export async function onRequest({ platform }) {
  return platform.call(new Request("https://proj.example/api/records/orders/query", {
    method: "POST",
    headers: { authorization: "Bearer user", "content-type": "application/json", "idempotency-key": "quote-1" },
    body: JSON.stringify({ limit: 1 }),
  }));
}
""")
    source = _worker(tmp_path, [{
        "id": "records.query",
        "method": "POST",
        "public_path": "/api/records/:entity/query",
        "auth": "session",
        "upstream_path": "/v1/projects/:project_id/records/query",
    }])
    hit = _fetch(source, "https://proj.example/api/quotes/price", "POST", "{}")
    assert hit["status"] == 200
    assert any("/v1/projects/project-1/records/query" in url for url in hit["seen"])
    assert "application-executed-function" not in hit["body"]


def test_platform_sdk_rejects_cross_origin_customer_requests(tmp_path: Path):
    _write(tmp_path, "functions/api/quotes/price.js", """
export async function onRequest({ platform }) {
  return platform.call(new Request(["https://", "cloud.example", "/v1/", "projects/project-1/records/query"].join(""), {
    method: "POST",
    body: "{}",
  }));
}
""")
    source = _worker(tmp_path)
    hit = _fetch(source, "https://proj.example/api/quotes/price", "POST", "{}")
    assert hit["status"] == 400
    assert json.loads(hit["body"])["error"] == "platform_route_must_be_same_origin_api"
    assert hit["seen"] == []


def test_health_stays_on_the_runtime_when_no_application_route_claims_it(tmp_path: Path):
    source = _worker(tmp_path)
    health = _fetch(source, "https://proj.example/api/__p2hs/health")
    assert health["status"] == 200
    assert "p2hs-application-runtime" in health["body"]


def test_one_worker_serves_static_application_cloud_auth_and_validation(tmp_path: Path):
    lib = tmp_path / "functions" / "_lib"
    lib.mkdir(parents=True)
    destination = tmp_path / "functions" / "api" / "forms" / "[formKey]" / "submissions.js"
    destination.parent.mkdir(parents=True)
    shutil.copy(RUNTIME / "p2hs-records.mjs", lib / "p2hs-records.mjs")
    shutil.copy(RUNTIME / "p2hs-forms-submit.mjs", lib / "p2hs-forms-submit.mjs")
    shutil.copy(RUNTIME / "forms-submit-pages-fn.js", destination)
    _lock_application_route(
        tmp_path,
        "functions/api/forms/[formKey]/submissions.js",
        "/api/forms/:form_key/submissions",
    )
    source = _worker(tmp_path, [{
        "id": "records.query",
        "method": "POST",
        "public_path": "/api/records/:entity/query",
        "auth": "session",
        "upstream_path": "/v1/projects/:project_id/records/query",
    }])
    page = _fetch(source, "https://proj.example/")
    assert page["status"] == 200
    assert page["body"] == "asset"
    invalid = _fetch(
        source,
        "https://proj.example/api/forms/quote/submissions",
        "POST",
        json.dumps({"project_id": "spoofed"}),
    )
    assert invalid["status"] == 400
    assert json.loads(invalid["body"])["error"] == "invalid_request"
    assert invalid["seen"] == []
    cloud = _fetch(source, "https://proj.example/api/records/orders/query", "POST", "{}")
    assert cloud["status"] == 200
    assert any("/records/query" in url for url in cloud["seen"])
    sign_in = _fetch(source, "https://proj.example/api/auth/magic-link/consume?token=expired")
    assert sign_in["status"] == 401
    assert "expired" in sign_in["body"].lower()


def _assert_pages_functions_bundle(root: Path) -> None:
    prelude = application_prelude(root, [])
    assert "async function dispatchApplication" in prelude
    for path in (root / "functions").rglob("*"):
        if not path.is_file() or path.suffix not in {".js", ".mjs"} or ".p2hs" in path.name:
            continue
        text = path.read_text(encoding="utf-8")
        assert "export default" not in text, path.relative_to(root)
        assert "import contract from" not in text, path.relative_to(root)


def test_new_pages_projects_and_boilerplates_bundle_for_w4p(tmp_path):
    import shutil

    from src.domain.codegen.port_adapter import install_catalog_adapters
    from src.domain.starter_library.runtime_substrate import write_generated_project_tree

    for project_type in ("html", "react_app", "nuxt_app"):
        created = tmp_path / "created" / project_type
        write_generated_project_tree(str(created), project_type)
        install_catalog_adapters(str(created), project_type, {"legacy_contact_submit": False})
        _assert_pages_functions_bundle(created)

        live = Path("boilerplates") / project_type
        if not (live / "functions").is_dir():
            continue
        _assert_pages_functions_bundle(live)
        copied = tmp_path / "boilerplate" / project_type
        shutil.copytree(
            live,
            copied,
            ignore=shutil.ignore_patterns("node_modules", ".git", "dist", ".output"),
        )
        install_catalog_adapters(str(copied), project_type, {"legacy_contact_submit": False})
        _assert_pages_functions_bundle(copied)