"""Disposable PostgreSQL for explicitly selected integration tests only.

Never accepts a DSN, host, existing data directory or production configuration.
The cluster has no TCP listener; its trust-authenticated Unix socket is contained
in a newly generated owner-only temporary directory. All artifacts are synthetic.
"""

from __future__ import annotations

import os
from pathlib import Path
import shlex
import shutil
import subprocess
import tempfile


PROJECT = Path(__file__).resolve().parents[1]


class TemporaryPostgres:
    def __init__(self):
        self.root = self.data = self.socket = None
        self.started = False
        self._temporary = None
        self._saved_pg = {}

    def __enter__(self):
        import psycopg
        self._driver = psycopg
        self.initdb, self.pg_ctl = shutil.which("initdb"), shutil.which("pg_ctl")
        if not self.initdb or not self.pg_ctl:
            raise RuntimeError("local PostgreSQL initdb/pg_ctl binaries are required")
        self._temporary = tempfile.TemporaryDirectory(prefix="arr-pg-test-", dir="/tmp")
        self.root = Path(self._temporary.name)
        self.data, self.socket = self.root / "cluster", self.root / "socket"
        self.socket.mkdir(mode=0o700)
        # Isolate this opt-in test process from libpq defaults; restore on exit.
        self._saved_pg = {k: v for k, v in os.environ.items() if k.startswith("PG")}
        for key in self._saved_pg:
            os.environ.pop(key)
        try:
            self._command([self.initdb, "-D", str(self.data), "-U", "arr_fixture",
                           "--auth-local=trust", "--auth-host=reject", "--encoding=UTF8", "--locale=C"])
            options = shlex.join(["-k", str(self.socket), "-p", "55432", "-c", "listen_addresses=",
                                  "-c", "unix_socket_permissions=0700", "-c", "fsync=on",
                                  "-c", "synchronous_commit=on", "-c", "full_page_writes=on"])
            self._command([self.pg_ctl, "-D", str(self.data), "-l", str(self.root / "postgres.log"),
                           "-o", options, "-w", "-t", "20", "start"])
            self.started = True
            with self.connect(database="postgres", autocommit=True) as connection:
                connection.execute("CREATE DATABASE booking_test TEMPLATE template0 ENCODING 'UTF8'")
            return self
        except BaseException:
            self.__exit__(None, None, None)
            raise

    @staticmethod
    def _command(arguments):
        result = subprocess.run(arguments, capture_output=True, text=True, timeout=35)
        if result.returncode:
            raise RuntimeError(f"temporary PostgreSQL command failed: {result.stdout}\n{result.stderr}")

    def connect(self, _dsn=None, *, database="booking_test", autocommit=False):
        connection = self._driver.connect(host=str(self.socket), port=55432, dbname=database,
                                          user="arr_fixture", connect_timeout=5, autocommit=True)
        try:
            directory = connection.execute("SHOW data_directory").fetchone()[0]
            if Path(directory).resolve() != self.data.resolve():
                raise RuntimeError("refusing non-fixture PostgreSQL instance")
            if connection.execute("SHOW listen_addresses").fetchone()[0] != "":
                raise RuntimeError("fixture PostgreSQL must not listen on TCP")
            connection.autocommit = autocommit
            return connection
        except BaseException:
            connection.close()
            raise

    def reset_database(self):
        # Only this owned cluster has passed data_directory/socket checks above.
        with self.connect(database="postgres", autocommit=True) as connection:
            connection.execute("DROP DATABASE booking_test WITH (FORCE)")
            connection.execute("CREATE DATABASE booking_test TEMPLATE template0 ENCODING 'UTF8'")
        with self.connect(autocommit=True) as connection:
            existing = connection.execute("SELECT nspname FROM pg_namespace WHERE nspname IN ('ingestion','finance','booking','reporting')").fetchall()
            if existing:
                raise RuntimeError("schema bootstrap requires an empty fixture database")
            paths = sorted(p for p in (PROJECT / "database").glob("[0-9][0-9][0-9]_*.sql")
                           if ".down." not in p.name and 8 <= int(p.name[:3]) <= 19)
            if len(paths) != 12:
                raise RuntimeError("unexpected fixture migration set")
            for path in paths:
                connection.execute(path.read_text(), prepare=False)

    def __exit__(self, *_):
        try:
            # A failed start can still leave a postmaster; stop it before cleanup.
            if self.data and (self.data / "postmaster.pid").exists():
                self._command([self.pg_ctl, "-D", str(self.data), "-m", "fast", "-w", "-t", "20", "stop"])
            self.started = False
            if self._temporary is not None:
                self._temporary.cleanup()
        finally:
            os.environ.update(self._saved_pg)