"""Synthetic lost-POST-response cases over a real loopback HTTP forward proxy.

No external destinations, automatic retries, durable storage or real jobs.
Python 3.11+; install the accompanying pinned requirements.
"""

import argparse
import asyncio
import csv
from datetime import datetime, timezone
from importlib.metadata import version
import io
import json
from pathlib import Path
import platform
import sys
from time import monotonic
from urllib.parse import urlsplit

PAYLOAD = {"item": "synthetic-widget", "quantity": 1}


def require(condition, message):
    if not condition:
        raise AssertionError(message)


def encode(value):
    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")


async def read_message(reader):
    head = await reader.readuntil(b"\r\n\r\n")
    require(len(head) <= 8192, "fixture header too large")
    lines = head[:-4].decode("ascii").split("\r\n")
    headers = {}
    for line in lines[1:]:
        name, value = line.split(":", 1)
        name = name.lower()
        require(name not in headers, "duplicate fixture header")
        headers[name] = value.strip()
    require("transfer-encoding" not in headers, "fixture requires Content-Length framing")
    size = int(headers.get("content-length", "0"))
    require(0 <= size <= 4096, "fixture body too large")
    return lines[0], headers, await reader.readexactly(size), head


class RetryFixture:
    """One temporary origin, proxy and caller scope; deliberately memory-only."""

    def __init__(self, contract_enabled, *, drop_first_post=True, gate_posts=0):
        self.contract_enabled = contract_enabled
        self.drop_next_post = drop_first_post
        self.gate_posts = gate_posts
        self.apply_gate = asyncio.Event()
        if not gate_posts:
            self.apply_gate.set()
        self.lock = asyncio.Lock()
        self.saved = {}
        self.jobs = []
        self.dropped_response = None
        self.events = []
        self.receipts = {"proxy": {"POST": 0, "GET": 0}, "origin": {"POST": 0, "GET": 0}}
        self.writers = set()
        self.tasks = set()
        self.servers = []
        self.errors = []
        self.forced_cancellations = 0
        self.started = monotonic()

    def log(self, side, event, **fields):
        self.events.append({"sequence": len(self.events) + 1,
                            "seconds": round(monotonic() - self.started, 6),
                            "side": side, "event": event, **fields})

    async def __aenter__(self):
        try:
            origin = await asyncio.start_server(self.accept("origin"), "127.0.0.1", 0)
            self.servers.append(origin)
            self.origin_port = origin.sockets[0].getsockname()[1]
            proxy = await asyncio.start_server(self.accept("proxy"), "127.0.0.1", 0)
            self.servers.append(proxy)
            self.proxy_port = proxy.sockets[0].getsockname()[1]
            self.origin_url = f"http://127.0.0.1:{self.origin_port}"
            self.proxy_url = f"http://127.0.0.1:{self.proxy_port}"
            return self
        except BaseException:
            await self.__aexit__(None, None, None)
            raise

    def accept(self, side):
        def completed(task):
            self.tasks.discard(task)
            if not task.cancelled() and task.exception() is not None:
                self.errors.append({"side": side, "error": type(task.exception()).__name__})

        def connected(reader, writer):
            self.writers.add(writer)
            task = asyncio.create_task(self.serve(side, reader, writer))
            self.tasks.add(task)
            task.add_done_callback(completed)
        return connected

    async def close_writer(self, writer):
        writer.close()
        try:
            await asyncio.wait_for(writer.wait_closed(), 1.0)
        except ConnectionError:
            pass
        finally:
            self.writers.discard(writer)

    async def serve(self, side, reader, writer):
        try:
            async with asyncio.timeout(8.0):
                await getattr(self, side)(reader, writer)
        except (ConnectionError, asyncio.IncompleteReadError):
            self.log(side, "peer_closed")
        except Exception as error:
            self.errors.append({"side": side, "error": type(error).__name__})
        finally:
            await self.close_writer(writer)

    async def apply_job(self, path, key, operation_ref, payload):
        # Synthetic contract: one caller, POST path, key, and parsed input/ref.
        scope = ("POST", path, key)
        fingerprint = encode({"operation_ref": operation_ref, "payload": payload})
        async with self.lock:
            if self.contract_enabled and key and scope in self.saved:
                previous_input, response = self.saved[scope]
                if fingerprint != previous_input:
                    self.log("origin", "key_conflict", key=key)
                    return 409, encode({"error": "key_payload_mismatch"})
                self.log("origin", "saved_result_replayed", key=key,
                         job_id=json.loads(response)["job"]["id"])
                return 201, response
            # Tests hold the first task here until another POST reaches origin.
            # Keeping this await inside the lock makes the concurrency test useful.
            await self.apply_gate.wait()
            job = {"id": f"job-{len(self.jobs) + 1}", "operation_ref": operation_ref, "payload": payload}
            response = encode({"job": job})
            self.jobs.append(job)
            if self.contract_enabled and key:
                self.saved[scope] = (fingerprint, response)
            self.log("origin", "job_applied", job_id=job["id"], operation_ref=operation_ref)
            return 201, response

    async def origin(self, reader, writer):
        line, headers, body, _ = await read_message(reader)
        method, path, protocol = line.split()
        require(method in ("GET", "POST") and protocol == "HTTP/1.1", "unsupported fixture request")
        self.receipts["origin"][method] += 1
        self.log("origin", "request_received", method=method, path=path,
                 key=headers.get("idempotency-key"), operation_ref=headers.get("x-operation-ref"))
        if method == "POST":
            require(path == "/jobs", "unknown job endpoint")
            if self.receipts["origin"]["POST"] >= self.gate_posts:
                self.apply_gate.set()
            operation_ref = headers.get("x-operation-ref", "")
            require(operation_ref and len(operation_ref) <= 100, "missing bounded operation reference")
            payload = json.loads(body)
            require(isinstance(payload, dict), "fixture requires a JSON object")
            status, response = await self.apply_job(path, headers.get("idempotency-key"), operation_ref, payload)
        else:
            require(path.startswith("/operations/"), "unknown lookup endpoint")
            operation_ref = path[len("/operations/"):]
            jobs = [job for job in self.jobs if job["operation_ref"] == operation_ref]
            status = 200 if jobs else 404
            response = encode({"operation_ref": operation_ref, "state": "applied" if jobs else "not_found", "jobs": jobs})
            self.log("origin", "lookup_completed", operation_ref=operation_ref, found_jobs=len(jobs))
        reason = {200: "OK", 201: "Created", 409: "Conflict", 404: "Not Found"}[status]
        head = f"HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {len(response)}\r\nConnection: close\r\n\r\n".encode("ascii")
        writer.write(head + response)
        await writer.drain()
        self.log("origin", "response_sent", method=method, status=status, body=json.loads(response))

    async def proxy(self, reader, writer):
        line, headers, body, _ = await read_message(reader)
        method, target, protocol = line.split()
        url = urlsplit(target)
        require(method in ("GET", "POST") and protocol == "HTTP/1.1", "unsupported fixture proxy request")
        require(url.scheme == "http" and url.hostname == "127.0.0.1" and url.port == self.origin_port
                and not url.username and not url.password and not url.query and not url.fragment,
                "fixture refuses destinations outside its own loopback origin")
        path = url.path or "/"
        self.receipts["proxy"][method] += 1
        self.log("proxy", "request_received", method=method, path=path,
                 key=headers.get("idempotency-key"), operation_ref=headers.get("x-operation-ref"))
        upstream_reader, upstream_writer = await asyncio.open_connection("127.0.0.1", self.origin_port)
        self.writers.add(upstream_writer)
        try:
            forwarded = [f"{method} {path} HTTP/1.1", f"Host: 127.0.0.1:{self.origin_port}",
                         f"Content-Length: {len(body)}", "Content-Type: application/json", "Connection: close"]
            for name in ("idempotency-key", "x-operation-ref"):
                if name in headers:
                    forwarded.append(f"{name}: {headers[name]}")
            upstream_writer.write(("\r\n".join(forwarded) + "\r\n\r\n").encode("ascii") + body)
            await upstream_writer.drain()
            response_line, _, response_body, response_head = await read_message(upstream_reader)
            status = int(response_line.split()[1])
            self.log("proxy", "complete_origin_response_received", method=method, status=status, body=json.loads(response_body))
            if method == "POST" and self.drop_next_post:
                self.drop_next_post = False
                self.dropped_response = {"status": status, "body": json.loads(response_body)}
                self.log("proxy", "response_dropped_before_client_headers", method=method, status=status)
                return  # serve() closes the downstream socket without any response.
            writer.write(response_head + response_body)
            await writer.drain()
            self.log("proxy", "response_forwarded", method=method, status=status)
        finally:
            await self.close_writer(upstream_writer)

    async def __aexit__(self, *_):
        for server in self.servers:
            server.close()
        for server in self.servers:
            await server.wait_closed()
        for writer in tuple(self.writers):
            writer.close()
        if self.tasks:
            _, pending = await asyncio.wait(tuple(self.tasks), timeout=1.0)
            self.forced_cancellations += len(pending)
            for task in pending:
                task.cancel()
            await asyncio.gather(*pending, return_exceptions=True)
        self.log("fixture", "closed", handlers=len(self.tasks), writers=len(self.writers),
                 forced_cancellations=self.forced_cancellations)

    def assert_clean(self):
        require(not self.tasks and not self.writers, "fixture leaked tasks or writers")
        require(not self.forced_cancellations and not self.errors, "fixture cleanup failed")
        require(not any(server.is_serving() for server in self.servers), "fixture listener still open")


def make_client(fixture):
    import httpx
    return httpx.AsyncClient(proxy=fixture.proxy_url, trust_env=False, http2=False,
                             follow_redirects=False,
                             timeout=httpx.Timeout(connect=3.0, read=3.0, write=3.0, pool=1.0),
                             limits=httpx.Limits(max_connections=4, max_keepalive_connections=0))


def request_headers(key=None, operation_ref="operation-1"):
    headers = {"X-Operation-Ref": operation_ref}
    if key is not None:
        headers["Idempotency-Key"] = key
    return headers


CASES = (
    ("blind_retry", False, None, None, "repeat_post"),
    ("ignored_stable_key", False, "key-1", "key-1", "repeat_post"),
    ("supported_stable_key", True, "key-1", "key-1", "repeat_post"),
    ("new_key_on_retry", True, "key-1", "key-2", "repeat_post"),
    ("changed_payload", True, "key-1", "key-1", "change_payload"),
    ("reconcile_confirmed_effect", False, None, None, "lookup_only"),
)


async def run_case(case, contract_enabled, first_key, second_key, action):
    import httpx
    baseline_tasks = set(asyncio.all_tasks())
    fixture = RetryFixture(contract_enabled)
    async with fixture:
        async with make_client(fixture) as client:
            fixture.log("client", "explicit_post", attempt=1, key=first_key, payload=PAYLOAD)
            try:
                await client.post(fixture.origin_url + "/jobs", headers=request_headers(first_key), json=PAYLOAD)
            except httpx.HTTPError as error:
                first_error = type(error).__name__
                fixture.log("client", "request_failed", attempt=1, error=first_error)
            else:
                raise AssertionError("first response was not lost")
            require(first_error == "RemoteProtocolError", "unexpected observed error phase")
            require(len(fixture.jobs) == 1, "first POST did not create exactly one job")
            chain = ("job_applied", "response_sent", "complete_origin_response_received",
                     "response_dropped_before_client_headers", "request_failed")
            sequence = [next(event["sequence"] for event in fixture.events if event["event"] == name)
                        for name in chain]
            require(sequence == sorted(sequence), "response-loss ordering failed")
            require(fixture.dropped_response["status"] == 201, "proxy did not drop a complete created-job result")
            if action == "lookup_only":
                fixture.log("client", "explicit_lookup", operation_ref="operation-1")
                response = await client.get(fixture.origin_url + "/operations/operation-1")
                require(response.json()["state"] == "applied", "positive reconciliation failed")
            else:
                payload = {**PAYLOAD, "quantity": 2} if action == "change_payload" else PAYLOAD
                fixture.log("client", "explicit_post", attempt=2, key=second_key, payload=payload)
                response = await client.post(fixture.origin_url + "/jobs", headers=request_headers(second_key), json=payload)
            result = {"status": response.status_code, "body": response.json()}
            fixture.log("client", "response_received", **result)
            if case == "supported_stable_key":
                require(result == fixture.dropped_response, "stable key did not replay the same result")
    fixture.assert_clean()
    require(fixture.receipts["proxy"] == fixture.receipts["origin"], "proxy and origin receipt counts differ")
    require(client.is_closed and not set(asyncio.all_tasks()) - baseline_tasks, "client or async work leaked")
    return {"case": case, "contract_enabled": contract_enabled, "first_client_error": first_error,
            "next_action": action, "second_action_status": response.status_code,
            "proxy_post_receipts": fixture.receipts["proxy"]["POST"],
            "proxy_get_receipts": fixture.receipts["proxy"]["GET"],
            "origin_post_receipts": fixture.receipts["origin"]["POST"],
            "origin_get_receipts": fixture.receipts["origin"]["GET"],
            "applied_jobs": len(fixture.jobs), "jobs": fixture.jobs,
            "dropped_origin_response": fixture.dropped_response, "client_received_result": result,
            "cleanup": {"handlers": len(fixture.tasks), "writers": len(fixture.writers),
                        "forced_cancellations": fixture.forced_cancellations,
                        "client_closed": client.is_closed, "errors": fixture.errors},
            "events": fixture.events}


async def run_all():
    observed_at = datetime.now(timezone.utc).isoformat()
    async with asyncio.timeout(30.0):
        cases = [await run_case(*case) for case in CASES]
    expected = [(2, 0, 2, 201), (2, 0, 2, 201), (2, 0, 1, 201),
                (2, 0, 2, 201), (2, 0, 1, 409), (1, 1, 1, 200)]
    actual = [(c["origin_post_receipts"], c["origin_get_receipts"], c["applied_jobs"], c["second_action_status"]) for c in cases]
    require(actual == expected, "scenario matrix failed")
    return {"outcome": "pass", "observation": "synthetic-loopback-http1-post-response-loss",
            "observed_at_utc": observed_at,
            "runtime": {"python": platform.python_version(), "implementation": platform.python_implementation(),
                        "system": platform.system(), "machine": platform.machine(),
                        "packages": {name: version(name) for name in ("httpx", "httpcore", "anyio", "h11", "certifi", "idna", "typing_extensions")}},
            "cases": cases}


def matrix_csv(report):
    fields = ("case", "first_client_error", "next_action", "proxy_post_receipts", "proxy_get_receipts", "origin_post_receipts",
              "origin_get_receipts", "applied_jobs", "second_action_status")
    output = io.StringIO(newline="")
    writer = csv.DictWriter(output, fieldnames=fields, lineterminator="\n", extrasaction="ignore")
    writer.writeheader()
    writer.writerows(report["cases"])
    return output.getvalue()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--csv", type=Path, help="derive CSV from an existing demo JSON; no network requests")
    args = parser.parse_args()
    try:
        if args.csv:
            print(matrix_csv(json.loads(args.csv.read_text())), end="")
        else:
            print(json.dumps(asyncio.run(run_all()), indent=2))
        return 0
    except Exception as error:
        print(json.dumps({"outcome": "fail", "error": type(error).__name__}))
        return 1


if __name__ == "__main__":
    sys.exit(main())