#!/usr/bin/env python3
"""Verify a generated MySQL service against a disposable local Docker database."""

import argparse
import json
import os
from pathlib import Path
import secrets
import signal
import socket
import subprocess
import tempfile
import time
import urllib.error
import urllib.request
import uuid

from verifyScaffold import MODULE, run


def waitFor(check, description, timeout=20):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if check():
            return
        time.sleep(0.1)
    raise AssertionError(f"timed out: {description}")


class MySQL:
    def __init__(self, directory, env, image):
        self.directory, self.env, self.image = directory, dict(env), image
        self.name = "go-omni-check-" + uuid.uuid4().hex[:12]
        self.container = None
        self.password = secrets.token_hex(18)
        self.env.update(MYSQL_ROOT_PASSWORD=secrets.token_hex(18), MYSQL_PASSWORD=self.password)
        self.env["MYSQL_PWD"] = self.env["MYSQL_ROOT_PASSWORD"]

    def sql(self, statement, database=None):
        args = ["docker", "exec", "-i", "--env", "MYSQL_PWD", self.container,
                "mysql", "--protocol=tcp", "-h127.0.0.1", "-uroot", "--batch", "--skip-column-names"]
        if database:
            args.append(database)
        result = subprocess.run(args, cwd=self.directory, env=self.env, input=statement,
                                text=True, capture_output=True, timeout=10)
        if result.returncode:
            raise RuntimeError(f"test database SQL failed: {result.stderr}")
        return result.stdout.strip()

    def wait(self):
        def ready():
            try:
                return self.sql("SELECT 1") == "1"
            except (RuntimeError, subprocess.TimeoutExpired):
                return False
        waitFor(ready, "MySQL ready", timeout=90)

    def start(self):
        endpoint = self.env.get("DOCKER_HOST") or run(
            ["docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"],
            self.directory, self.env).strip()
        if not endpoint.startswith(("unix://", "npipe://")):
            raise RuntimeError("database verification requires a local Docker context")
        # Docker may assign a different random published port after restart.
        # Choose the loopback port once so outage/recovery keeps the same DSN.
        with socket.socket() as listener:
            listener.bind(("127.0.0.1", 0))
            hostPort = listener.getsockname()[1]
        print("Starting isolated MySQL container", flush=True)
        self.container = run(["docker", "run", "--detach", "--name", self.name,
                              "--label", "go-omni.test=database", "--publish", f"127.0.0.1:{hostPort}:3306",
                              "--env", "MYSQL_ROOT_PASSWORD", "--env", "MYSQL_PASSWORD",
                              "--env", "MYSQL_DATABASE=omni_test", "--env", "MYSQL_USER=omni_test",
                              self.image], self.directory, self.env).strip()
        self.wait()
        port = run(["docker", "port", self.container, "3306/tcp"], self.directory, self.env).strip().rsplit(":", 1)[1]
        self.dsn = (f"omni_test:{self.password}@tcp(127.0.0.1:{port})/omni_test"
                    "?parseTime=true&timeout=2s&readTimeout=2s&writeTimeout=2s")
        self.sql("CREATE DATABASE omni_empty; GRANT ALL ON omni_empty.* TO 'omni_test'@'%';")

    def close(self):
        if self.container:
            # Remove only the container and anonymous volume created by this run.
            run(["docker", "rm", "--force", "--volumes", self.container], self.directory, self.env)


class Server:
    def __init__(self, binary, directory, env):
        self.binary, self.directory, self.env = binary, directory, dict(env)
        with socket.socket() as listener:
            listener.bind(("127.0.0.1", 0))
            self.address = f"127.0.0.1:{listener.getsockname()[1]}"
        self.env["GO_OMNI_HTTP_ADDR"] = self.address
        self.process = None
        self.logPath = directory / f"server-{uuid.uuid4().hex[:8]}.log"
        self.client = urllib.request.build_opener(urllib.request.ProxyHandler({}))

    def request(self, method, path, body=None, expected=200):
        req = urllib.request.Request(f"http://{self.address}{path}", method=method,
                                     data=None if body is None else json.dumps(body).encode(),
                                     headers={"Content-Type": "application/json", "X-Request-ID": "databaseCheck"})
        try:
            response = self.client.open(req, timeout=3)
        except urllib.error.HTTPError as error:
            response = error
        with response:
            raw = response.read()
            if response.status != expected:
                raise AssertionError(f"{method} {path}: expected {expected}, got {response.status}: {raw!r}")
            if response.headers.get("X-Request-ID") != "databaseCheck":
                raise AssertionError("missing requestId header")
            if not raw:
                return None
            result = json.loads(raw)
            if result.get("requestId") != "databaseCheck":
                raise AssertionError("missing camelCase requestId")
            return result

    def start(self):
        with self.logPath.open("w") as logs:
            self.process = subprocess.Popen([str(self.binary)], cwd=self.directory, env=self.env,
                                            stdout=logs, stderr=subprocess.STDOUT)
        def ready():
            if self.process.poll() is not None:
                raise RuntimeError(f"service exited: {self.logPath.read_text()}")
            try:
                return self.request("GET", "/health/live")["data"]["status"] == "ok"
            except (OSError, urllib.error.URLError):
                return False
        waitFor(ready, "service startup")

    def stop(self):
        if self.process and self.process.poll() is None:
            self.process.send_signal(signal.SIGTERM)
            code = self.process.wait(timeout=10)
            if code != 0:
                raise AssertionError(f"service exit {code}: {self.logPath.read_text()}")

    def close(self):
        if self.process and self.process.poll() is None:
            self.process.kill()
            self.process.wait(timeout=5)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--go", default="go")
    parser.add_argument("--version", default="v0.6.0-preview")
    parser.add_argument("--cli", help="Use a locally built CLI")
    parser.add_argument("--framework-dir", help="Local framework source for pre-release checks")
    parser.add_argument("--image", default="mysql:8.4")
    args = parser.parse_args()
    env = {key: value for key, value in os.environ.items() if not key.startswith("GO_OMNI_")}
    env["GOWORK"] = "off"
    env["GOPRIVATE"] = ",".join(filter(None, [env.get("GOPRIVATE"), "git.yuntuo.hk"]))
    with tempfile.TemporaryDirectory(prefix="go omni database ") as temporary:
        directory = Path(temporary).resolve()
        if args.cli:
            cli = Path(args.cli).resolve()
        else:
            env["GOBIN"] = str(directory / "bin")
            run([args.go, "install", f"{MODULE}/cmd/go-omni@{args.version}"], directory, env)
            cli = directory / "bin" / "go-omni"
        command = [str(cli), "new", "--module", "example.com/databaseCheck", "--database", "mysql"]
        if args.framework_dir:
            command += ["--framework-dir", str(Path(args.framework_dir).resolve())]
        project = directory / "billing"
        run(command + [str(project)], directory, env)
        module = json.loads(run([args.go, "mod", "edit", "-json"], project, env))
        if not args.framework_dir and module.get("Replace"):
            raise AssertionError("remote MySQL project contains a local replace")
        if {entry["Path"]: entry["Version"] for entry in module["Require"]}.get(MODULE) != args.version:
            raise AssertionError("MySQL project did not pin the CLI version")
        run([args.go, "mod", "tidy"], project, env)
        binary = directory / "server"
        run([args.go, "build", "-o", str(binary), "./cmd/server"], project, env)
        database = MySQL(directory, env, args.image)
        servers = []
        try:
            database.start()
            database.sql((project / "schema.sql").read_text(), "omni_test")
            testEnv = dict(env, GO_OMNI_TEST_MYSQL_DSN=database.dsn)
            output = run([args.go, "test", "-race", "-v", "./..."], project, testEnv)
            for name in ("TestDatabaseTransactions", "TestDatabasePoolHealthAndClose"):
                if f"--- PASS: {name}" not in output:
                    raise AssertionError(f"integration test did not pass: {name}\n{output}")
            print("PASS real MySQL: commit, error/panic/cancel rollback, pool limits, health timeout, close", flush=True)

            serverEnv = dict(env, GO_OMNI_DB_DSN=database.dsn,
                             GO_OMNI_DB_PING_TIMEOUT="500ms", GO_OMNI_ENABLE_LOGGER="true")
            server = Server(binary, project, serverEnv)
            servers.append(server)
            server.start()
            ready = server.request("GET", "/health/ready")
            if ready["code"] != "READY" or ready["data"]["checks"][0]["name"] != "database":
                raise AssertionError(f"database probe not registered: {ready}")
            server.request("POST", "/users", {"name": "", "email": "invalid"}, expected=400)
            created = server.request("POST", "/users", {"name": " Alice ", "email": "ALICE@example.com"}, expected=201)
            user = created["data"]
            if user["email"] != "alice@example.com" or not user.get("createdAt"):
                raise AssertionError(f"normalization/time parsing failed: {user}")
            path = f"/users/{user['id']}"
            duplicate = server.request("POST", "/users", {"name": "Duplicate", "email": "alice@example.com"}, expected=409)
            if duplicate["code"] != "EMAIL_TAKEN":
                raise AssertionError("duplicate email error was not mapped")
            if server.request("GET", path)["data"]["name"] != "Alice":
                raise AssertionError("GET did not load created user")
            page = server.request("GET", "/users?page=1&pageSize=1")["data"]
            if page["total"] != 1 or page["pageSize"] != 1:
                raise AssertionError(f"pagination mismatch: {page}")
            server.request("PUT", path, {"name": "Updated", "email": "updated@example.com"})
            server.stop()
            waitFor(lambda: database.sql("SELECT COUNT(*) FROM information_schema.processlist WHERE USER='omni_test'") == "0",
                    "connection pool closed on service exit")
            server = Server(binary, project, serverEnv)
            servers.append(server)
            server.start()
            if server.request("GET", path)["data"]["name"] != "Updated":
                raise AssertionError("data did not survive service restart")
            print("PASS HTTP validation, unique email, create/read/update/list, restart persistence and pool cleanup", flush=True)

            failureEnv = dict(serverEnv, GO_OMNI_DB_DSN=database.dsn.replace("/omni_test?", "/omni_empty?"))
            failed = subprocess.run([str(binary)], cwd=project, env=failureEnv, capture_output=True, text=True, timeout=10)
            if failed.returncode == 0 or "apply schema.sql" not in failed.stderr:
                raise AssertionError("missing schema did not fail startup before listening")
            waitFor(lambda: database.sql("SELECT COUNT(*) FROM information_schema.processlist WHERE DB='omni_empty'") == "0",
                    "connection pool closed on schema failure")
            print("PASS missing schema fails startup and releases connections", flush=True)

            run(["docker", "stop", "--time", "10", database.container], directory, database.env)
            readiness = server.request("GET", "/health/ready", expected=503)
            if readiness["code"] != "NOT_READY" or readiness["data"]["checks"][0]["status"] != "not_ready":
                raise AssertionError("database outage did not fail readiness")
            server.request("GET", "/health/live")
            unavailable = server.request("GET", path, expected=500)
            if unavailable["code"] != "INTERNAL_ERROR" or database.password in json.dumps(unavailable):
                raise AssertionError("database outage leaked an internal error")
            failed = subprocess.run([str(binary)], cwd=project, env=serverEnv, capture_output=True, text=True, timeout=10)
            if failed.returncode == 0 or "ping database" not in failed.stderr:
                raise AssertionError("offline database did not fail startup")
            run(["docker", "start", database.container], directory, database.env)
            database.wait()
            def recovered():
                try:
                    return server.request("GET", "/health/ready")["code"] == "READY"
                except (AssertionError, OSError, urllib.error.URLError):
                    return False
            waitFor(recovered, "application reconnects after MySQL restart")
            if server.request("GET", path)["data"]["name"] != "Updated":
                raise AssertionError("database restart lost persistent data")
            server.request("DELETE", path, expected=204)
            server.request("GET", path, expected=404)
            server.stop()
            waitFor(lambda: database.sql("SELECT COUNT(*) FROM information_schema.processlist WHERE USER='omni_test'") == "0",
                    "final connection cleanup")
            print("PASS database outage: readiness 503, liveness 200, startup failure; recovery 200; delete 204/404", flush=True)
        finally:
            for server in servers:
                server.close()
            database.close()
            print("Removed test container, anonymous volume and temporary project", flush=True)


if __name__ == "__main__":
    main()