#!/usr/bin/env python3
"""API + MySQL + storage business acceptance, using only disposable local resources.

--version installs the real remote CLI and tests consumers with no local replace.
No deployment is performed. Secrets and signed URLs are excluded from reports.
"""
import argparse
import hashlib
import json
import os
from pathlib import Path
import secrets
import shutil
import signal
import socket
import subprocess
import tempfile
import time
import urllib.parse
import urllib.request
import urllib.error

from verifyDatabase import MySQL, waitFor

ROOT = Path(__file__).resolve().parents[1]
MODULE = "git.yuntuo.hk/k2safe/go-omni"


def run(args, cwd, env, timeout=180, expect_ok=True):
    proc = subprocess.run([str(x) for x in args], cwd=cwd, env=env, text=True,
                          stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout)
    if (proc.returncode == 0) != expect_ok:
        # Commands contain only paths and module versions, never credentials.
        output = proc.stdout
        for key, value in env.items():
            if value and any(word in key for word in ("TOKEN", "PASSWORD", "SECRET", "DSN", "SIGNING_KEY")):
                output = output.replace(value, "[REDACTED]")
        raise RuntimeError(f"{args[0]} failed ({proc.returncode}):\n{output}")
    return proc.stdout


class LimitedMySQL(MySQL):
    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("a local Docker context is required")
        with socket.socket() as sock:
            sock.bind(("127.0.0.1", 0)); port = sock.getsockname()[1]
        self.container = run(["docker", "run", "--detach", "--name", self.name,
            "--label", "go-omni.test=apiWorkflow", "--cpus=1", "--memory=768m", "--pids-limit=256",
            "--publish", f"127.0.0.1:{port}: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()
        self.dsn = f"omni_test:{self.password}@tcp(127.0.0.1:{port})/omni_test?parseTime=true&timeout=2s&readTimeout=2s&writeTimeout=2s"


def exercise(binary, project, env, db, transport):
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0)); port = sock.getsockname()[1]
    token, signing = secrets.token_hex(24), secrets.token_hex(24)
    environment = dict(env, GO_OMNI_HTTP_ADDR=f"127.0.0.1:{port}", GO_OMNI_TRANSPORT=transport,
        USER_SERVICE_API_TOKEN=token, GO_OMNI_STORAGE_SIGNING_KEY=signing,
        GO_OMNI_STORAGE_ROOT=str(project / (transport + "Files")),
        GO_OMNI_STORAGE_PUBLIC_URL=f"http://127.0.0.1:{port}/storage/object")
    logs = project / "var/log/server.log"
    logs.parent.mkdir(parents=True, exist_ok=True)
    # Keep each transport's business log checks separate.
    logs.write_text("")
    checks = []
    client = urllib.request.build_opener(urllib.request.ProxyHandler({}))
    def request(method, path, body=None, expected=200, auth=True):
        headers = {"Content-Type": "application/json", "X-Request-ID": "apiWorkflow"}
        if auth: headers["Authorization"] = "Bearer " + token
        req = urllib.request.Request(f"http://127.0.0.1:{port}" + path,
            data=None if body is None else json.dumps(body).encode(), headers=headers, method=method)
        try: response = client.open(req, timeout=5)
        except urllib.error.HTTPError as exc: response = exc
        with response:
            data = response.read()
            if response.status != expected: raise AssertionError(f"{method} {path}: expected {expected}, got {response.status}")
            if response.headers.get("X-Request-ID") != "apiWorkflow": raise AssertionError("requestId header absent")
            if not data: return None
            value = json.loads(data)
            if path != "/openapi.json" and value.get("requestId") != "apiWorkflow": raise AssertionError("requestId body absent")
            return value
    def start():
        process = subprocess.Popen([str(binary)], cwd=project, env=environment, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        def ready():
            if process.poll() is not None: raise AssertionError("example failed startup")
            try: return request("GET", "/health/ready")["code"] == "READY"
            except (OSError, urllib.error.URLError): return False
        try: waitFor(ready, "business service ready", 15)
        except BaseException:
            process.kill();process.wait(timeout=5);raise
        return process
    process = start()
    try:
        request("GET", "/api/v1/users", expected=401, auth=False); checks.append("authBeforeService")
        empty = request("GET", "/api/v1/users")["data"]
        assert empty["items"] == [] and empty["total"] == 0
        checks.append("emptyPageAndQueryDefaults")
        request("POST", "/api/v1/users", {"name": "X", "email": "invalid"}, 400)
        request("POST", "/api/v1/users", {"name": "X", "email": "x@example.com", "unknown": 1}, 400)
        request("GET", "/api/v1/users?pageSize=101", expected=400)
        request("GET", "/api/v1/users/0", expected=400); checks.append("contractValidation")
        user = request("POST", "/api/v1/users", {"name": "  Alice  ", "email": "Alice@Example.com"}, 201)["data"]
        assert user["name"] == "Alice" and user["email"] == "alice@example.com"
        uid = user["id"]; path = f"/api/v1/users/{uid}"
        assert request("GET", path)["data"] == user; checks.append("createReadNormalize")
        assert request("POST", "/api/v1/users", {"name": "Other", "email": "alice@example.com"}, 409)["code"] == "EMAIL_TAKEN"
        checks.append("uniqueEmailConflict")
        user = request("PUT", path, {"name": "Updated", "email": "updated@example.com"})["data"]
        assert user["name"] == "Updated"
        page = request("GET", "/api/v1/users?page=1&pageSize=1")["data"]
        assert page["total"] == 1 and page["items"][0]["id"] == uid
        checks.append("updateAndPagination")
        # Force the second SQL write to fail: create/update/delete must roll back.
        db.sql("CREATE TRIGGER reject_audit BEFORE INSERT ON user_audit FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='acceptance rollback';", "omni_test")
        try:
            request("POST", "/api/v1/users", {"name": "Rollback", "email": "rollback@example.com"}, 500)
            request("PUT", path, {"name": "Rollback", "email": "rollback@example.com"}, 500)
            request("DELETE", path, expected=500)
            assert request("GET", path)["data"] == user
            assert db.sql("SELECT COUNT(*) FROM users WHERE email='rollback@example.com'", "omni_test") == "0"
            assert db.sql("SELECT COUNT(*) FROM user_audit", "omni_test") == "2"
        finally: db.sql("DROP TRIGGER reject_audit", "omni_test")
        checks.append("createUpdateDeleteRollbackOnAuditFailure")
        link = request("POST", path + "/export")["data"]["downloadUrl"]
        parsed = urllib.parse.urlparse(link)
        assert parsed.hostname == "127.0.0.1" and parsed.port == port
        with client.open(link, timeout=3) as response: exported = json.load(response)
        assert exported == user; checks.append("storageUploadAndSignedDownload")
        tampered = link.replace("signature=", "signature=bad")
        if tampered == link:
            query = urllib.parse.parse_qs(parsed.query)
            key = next(k for k in query if k.lower() in ("sig", "signature"))
            query[key] = ["bad"]
            tampered = urllib.parse.urlunparse(parsed._replace(query=urllib.parse.urlencode(query, doseq=True)))
        try: client.open(tampered, timeout=3); raise AssertionError("tampered signature accepted")
        except urllib.error.HTTPError as exc: assert exc.code == 403
        checks.append("signedDownloadTamperRejected")
        doc = request("GET", "/openapi.json")
        assert doc == json.loads((project / "docs/openapi.json").read_text()); checks.append("openAPIMatchesContract")
        run(["docker", "stop", "--time=3", db.container], project, db.env)
        request("GET", "/health/ready", expected=503)
        run(["docker", "start", db.container], project, db.env); db.wait()
        waitFor(lambda: request("GET", "/health/ready")["code"] == "READY", "database recovery")
        checks.append("readinessOutageAndRecovery")
        process.send_signal(signal.SIGTERM); assert process.wait(timeout=12) == 0
        process = start()
        assert request("GET", path)["data"] == user; checks.append("persistenceAfterServiceRestart")
        request("DELETE", path, expected=204)
        assert request("GET", path, expected=404)["code"] == "USER_NOT_FOUND"
        assert request("GET", "/api/v1/users")["data"]["items"] == []
        assert db.sql("SELECT COUNT(*) FROM user_audit WHERE request_id='apiWorkflow'", "omni_test") == "3"
        checks.append("deleteAndTransactionalAudit")
        process.send_signal(signal.SIGTERM); assert process.wait(timeout=12) == 0
        with socket.socket() as sock: assert sock.connect_ex(("127.0.0.1", port)) != 0
        checks.append("gracefulStopAndListenerRelease")
        text = logs.read_text()
        entries = [json.loads(line) for line in text.splitlines() if line.startswith("{")]
        assert any(e.get("msg") == "user.created" and e.get("requestId") == "apiWorkflow" for e in entries)
        assert all(secret not in text for secret in (token, signing, env["GO_OMNI_DB_DSN"], link, "updated@example.com"))
        checks.append("businessFileLogsAndSecretExclusion")
        return {"transport": transport, "checks": checks}
    finally:
        if process.poll() is None: process.kill(); process.wait(timeout=5)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--version")
    parser.add_argument("--report", type=Path, default=ROOT / "performance/apiWorkflowLocal.json")
    args = parser.parse_args()
    env = {k: v for k, v in os.environ.items() if not k.startswith(("GO_OMNI_", "USER_SERVICE_"))}
    env.update(GOWORK="off", GOFLAGS="", GOPRIVATE="git.yuntuo.hk", GONOSUMDB="git.yuntuo.hk")
    started = time.monotonic()
    resolved = {}
    with tempfile.TemporaryDirectory(prefix="goOmniAPIWorkflow-") as temp:
        temp = Path(temp); cli = temp / "bin/go-omni"; cli.parent.mkdir()
        if args.version:
            env.update(GOBIN=str(cli.parent), GOMODCACHE=str(temp / "modcache"))
            run(["go", "install", MODULE + "/cmd/go-omni@" + args.version], temp, env, timeout=300)
        else: run(["go", "build", "-o", cli, "./cmd/go-omni"], ROOT, env)
        project = temp / "userService"
        source = ROOT / "examples/userService"
        shutil.copytree(source, project, ignore=lambda directory, names: {"var", "server"} if Path(directory) == source else set())
        if args.version:
            run(["go", "mod", "edit", "-dropreplace=" + MODULE, "-require=" + MODULE + "@" + args.version], project, env)
        else: run(["go", "mod", "edit", "-replace=" + MODULE + "=" + str(ROOT)], project, env)
        run(["go", "mod", "tidy"], project, env)
        if args.version:
            assert "replace " not in (project / "go.mod").read_text()
            modules = run(["go", "list", "-m", "-json", MODULE], project, env)
            resolved = json.loads(modules)
            assert resolved["Version"] == args.version and "Replace" not in resolved
            # Installed CLI must pin its own version without a source checkout.
            fresh = temp / "fresh"
            run([cli, "api", "new", "--module", "example.com/fresh", "--database", "mysql", "--storage", "local", fresh], temp, env)
            assert args.version in (fresh / "go.mod").read_text() and "replace " not in (fresh / "go.mod").read_text()
            run(["go", "test", "-mod=mod", "./..."], fresh, env)
        # Regeneration must preserve all developer-owned code and config.
        manifest = json.loads((project / ".goOmni/apiManifest.json").read_text())
        owned = set(manifest["generated"]) | {".goOmni/apiManifest.json"}
        def hand_hashes():
            return {str(p.relative_to(project)): hashlib.sha256(p.read_bytes()).hexdigest()
                    for p in project.rglob("*") if p.is_file() and str(p.relative_to(project)) not in owned}
        before = hand_hashes()
        run([cli, "api", "generate", "--api", "api/user.api", "--dir", ".", "--check"], project, env)
        run([cli, "api", "generate", "--api", "api/user.api", "--dir", "."], project, env)
        assert before == hand_hashes()
        db = LimitedMySQL(temp, env, "mysql:8.4")
        try:
            print("Starting bounded, isolated MySQL", flush=True); db.start()
            env.update(GO_OMNI_DB_DSN=db.dsn, USER_SERVICE_TEST_DATABASE="1")
            run([cli, "migrate", "--dir", "migrations", "--apply"], project, env)
            run(["go", "test", "-race", "-count=1", "./..."], project, env)
            run(["go", "vet", "./..."], project, env)
            binary = project / "server";run(["go", "build", "-o", binary, "./cmd/server"], project, env)
            groups = []
            for transport in ("nethttp", "native"):
                db.sql("DELETE FROM user_audit; DELETE FROM users;", "omni_test")
                groups.append(exercise(binary, project, env, db, transport))
                print("PASS business workflow: " + transport, flush=True)
        finally: db.close()
        version = args.version or "local"
        report = {"passed": True, "frameworkVersion": version, "noLocalReplace": bool(args.version),
                  "durationSeconds": round(time.monotonic() - started, 2),
                  "regenerationPreservedHandwrittenFiles": len(before), "groups": groups,
                  "transactionTests": ["commit", "errorRollback", "panicRollback", "cancellationRollback"],
                  "startupTests": ["storageFailureClosesDatabase", "idempotentClose", "closedContextCannotReopen"],
                  "migrationCLI": True,
                  "frameworkSum": resolved.get("Sum"),
                  "sourceHashes": {str(p.relative_to(ROOT)): hashlib.sha256(p.read_bytes()).hexdigest()
                      for base in (ROOT / "cmd/go-omni", ROOT / "database", ROOT / "examples/userService")
                      for p in sorted(base.rglob("*")) if p.is_file() and p.suffix in (".go", ".tmpl", ".api", ".sql", ".yaml")}}
        args.report.write_text(json.dumps(report, indent=2) + "\n")
        print(f"PASS API workflow ({version}): {sum(len(g['checks']) for g in groups)} HTTP/lifecycle checks", flush=True)


if __name__ == "__main__":
    def interrupt(signum, frame):
        raise KeyboardInterrupt(f"signal {signum}")
    for sig in (signal.SIGTERM, signal.SIGHUP): signal.signal(sig, interrupt)
    main()