#!/usr/bin/env bash
set -euo pipefail

repo_root=$(cd "$(dirname "$0")/.." && pwd -P)
go_image='golang:1.26.5-bookworm@sha256:1ecb7edf62a0408027bd5729dfd6b1b8766e578e8df93995b225dfd0944eb651'
temp_root=
current_case=setup
failure_reported=false
registered_pids=()

report_pass() {
  printf 'case=%s status=PASS\n' "$1"
}

fail_case() {
  current_case=$1
  failure_reported=true
  printf 'case=%s status=FAIL\n' "$current_case" >&2
  exit 1
}

register_pid() {
  registered_pids+=("$1")
}

cleanup() {
  local pid
  set +e
  for pid in "${registered_pids[@]}"; do
    stop_registered_pid "$pid" || true
  done
  registered_pids=()
  if [[ "$temp_root" == /tmp/thothii-dwh-auth-nginx-integration.* && -d "$temp_root" ]]; then
    rm -rf -- "$temp_root"
  fi
}

on_exit() {
  local exit_code=$?
  cleanup
  if ((exit_code != 0)) && [[ "$failure_reported" != true ]]; then
    printf 'case=%s status=FAIL\n' "$current_case" >&2
  fi
  exit "$exit_code"
}
trap on_exit EXIT
trap 'exit 130' INT TERM

wait_for_socket() {
  local socket=$1
  local attempt
  for attempt in $(seq 1 100); do
    [[ -S "$socket" ]] && return 0
    sleep 0.05
  done
  return 1
}

wait_for_file() {
  local file=$1
  local attempt
  for attempt in $(seq 1 100); do
    [[ -s "$file" ]] && return 0
    sleep 0.05
  done
  return 1
}

build_dwh_auth() {
  local output=$1
  if command -v go >/dev/null 2>&1; then
    (
      cd "$repo_root/tools/dwh-auth"
      go build -o "$output" ./cmd/dwh-auth
    )
    return
  fi
  command -v docker >/dev/null 2>&1 || return 1
  docker run --rm --network none --user "$(id -u):$(id -g)" \
    --volume "$repo_root:/work:ro" \
    --volume "$temp_root:/out" \
    --workdir /work/tools/dwh-auth \
    "$go_image" \
    /bin/sh -ec 'GOCACHE=/out/go-cache CGO_ENABLED=0 go build -o /out/dwh-auth ./cmd/dwh-auth'
}

v1_key_id() {
  local value=$1
  local remainder=${value#thtdwh_v1.}
  printf '%s' "${remainder%%.*}"
}

request_status() {
  local body=$1
  shift
  curl --silent --show-error --noproxy '*' \
    --unix-socket "$nginx_socket" \
    --output "$body" \
    --write-out '%{http_code}' \
    "$@" 2>"$temp_root/curl.stderr"
}

expect_status() {
  local name=$1
  local expected=$2
  local body=$3
  shift 3
  local status
  current_case=$name
  if ! status=$(request_status "$body" "$@"); then
    fail_case "$name"
  fi
  [[ "$status" == "$expected" ]] || fail_case "$name"
}

unregister_pid() {
  local target=$1
  local candidate
  local retained=()
  for candidate in "${registered_pids[@]}"; do
    [[ "$candidate" == "$target" ]] || retained+=("$candidate")
  done
  registered_pids=("${retained[@]}")
}

pid_exited_or_zombie() {
  local pid=$1
  local state
  [[ "$pid" =~ ^[0-9]+$ ]] || return 1
  [[ ! -d "/proc/$pid" ]] && return 0
  state=$(awk "{print \$3}" "/proc/$pid/stat" 2>/dev/null) || return 0
  [[ "$state" == Z* ]]
}

pid_is_direct_child() {
  local pid=$1
  local parent
  [[ -r "/proc/$pid/stat" ]] || return 1
  parent=$(awk "{print \$4}" "/proc/$pid/stat" 2>/dev/null) || return 1
  [[ "$parent" == "$$" ]]
}

reap_if_direct_child() {
  local pid=$1
  if pid_is_direct_child "$pid"; then
    wait "$pid" 2>/dev/null || true
  fi
}

wait_for_exit_or_zombie() {
  local pid=$1
  local attempts=${2:-10}
  local attempt
  for ((attempt = 0; attempt < attempts; attempt++)); do
    pid_exited_or_zombie "$pid" && return 0
    sleep 0.05
  done
  pid_exited_or_zombie "$pid"
}

tcp_listener_for_pid() {
  local pid=$1
  [[ "$pid" =~ ^[0-9]+$ ]] || return 1
  ss -ltnpH 2>/dev/null | grep -Eq "(^|[^0-9])pid=$pid([,)]|$)"
}

wait_for_tcp_listener_pid() {
  local pid=$1
  local attempt
  for ((attempt = 0; attempt < 10; attempt++)); do
    tcp_listener_for_pid "$pid" && return 0
    sleep 0.05
  done
  tcp_listener_for_pid "$pid"
}

stop_registered_pid() {
  local pid=$1
  [[ "$pid" =~ ^[0-9]+$ ]] || return 1
  if pid_exited_or_zombie "$pid"; then
    reap_if_direct_child "$pid"
    unregister_pid "$pid"
    return 0
  fi
  if ! kill -TERM "$pid" 2>/dev/null; then
    wait_for_exit_or_zombie "$pid" 1 || return 1
  fi
  if ! wait_for_exit_or_zombie "$pid"; then
    kill -KILL "$pid" 2>/dev/null || return 1
    wait_for_exit_or_zombie "$pid" || return 1
  fi
  reap_if_direct_child "$pid"
  unregister_pid "$pid"
}

run_term_ignored_regression() {
  local child_pid started_seconds registered_pid
  current_case=cleanup_term_ignored_bounded
  python3 -c "import signal; signal.signal(signal.SIGTERM, signal.SIG_IGN); signal.pause()" >"$temp_root/term-ignored.log" 2>&1 &
  child_pid=$!
  register_pid "$child_pid"
  started_seconds=$SECONDS
  if ! stop_registered_pid "$child_pid"; then
    fail_case cleanup_term_ignored_bounded
  fi
  if kill -0 "$child_pid" 2>/dev/null; then
    fail_case cleanup_term_ignored_bounded
  fi
  for registered_pid in "${registered_pids[@]}"; do
    [[ "$registered_pid" != "$child_pid" ]] || fail_case cleanup_term_ignored_bounded
  done
  ((SECONDS - started_seconds < 3)) || fail_case cleanup_term_ignored_bounded
  report_pass cleanup_term_ignored_bounded
}

run_tcp_listener_detector_positive() {
  local probe_pid
  current_case=tcp_listener_detector_positive
  python3 - >"$temp_root/tcp-listener.log" 2>&1 <<PY &
import signal
import socket

listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.bind(("127.0.0.1", 0))
listener.listen()
signal.pause()
PY
  probe_pid=$!
  register_pid "$probe_pid"
  wait_for_tcp_listener_pid "$probe_pid" || fail_case tcp_listener_detector_positive
  report_pass tcp_listener_detector_positive
  stop_registered_pid "$probe_pid" || fail_case tcp_listener_detector_positive
  ! tcp_listener_for_pid "$probe_pid" || fail_case tcp_listener_detector_positive
}

for command in nginx curl python3 ss date; do
  command -v "$command" >/dev/null 2>&1 || fail_case "missing_${command}"
done

nginx_version=$(nginx -v 2>&1)
[[ "$nginx_version" == *nginx/1.24.* ]] || fail_case nginx_version
report_pass nginx_1_24
temp_root=$(mktemp -d /tmp/thothii-dwh-auth-nginx-integration.XXXXXXXX) || fail_case fixture_root
chmod 0700 "$temp_root" || fail_case fixture_root
umask 077

run_term_ignored_regression

dwh_auth="$temp_root/dwh-auth"
current_case=build_dwh_auth
build_dwh_auth "$dwh_auth" >"$temp_root/build.log" 2>&1 || fail_case build_dwh_auth
[[ -x "$dwh_auth" ]] || fail_case build_dwh_auth
report_pass build_dwh_auth

registry_root="$temp_root/registry"
socket_parent="$temp_root/socket"
service_socket="$socket_parent/verify.sock"
auth_proxy_socket="$socket_parent/nginx-auth.sock"
nginx_prefix="$temp_root/nginx"
nginx_socket="$nginx_prefix/listener.sock"
nginx_config="$nginx_prefix/nginx.conf"
runtime_http="$nginx_prefix/http.conf"
runtime_location="$nginx_prefix/location.conf"
marker_port_file="$temp_root/marker-port"
marker_observations="$temp_root/marker-observations.jsonl"
auth_observations="$temp_root/auth-observations.jsonl"

mkdir "$registry_root" "$socket_parent" "$nginx_prefix" || fail_case fixture_directories
chmod 0750 "$registry_root"
chmod 0700 "$socket_parent" "$nginx_prefix"

v1_file="$temp_root/v1.key"
legacy_file="$temp_root/legacy.key"
revoked_file="$temp_root/revoked.key"
expired_file="$temp_root/expired.key"

current_case=registry_setup
"$dwh_auth" --registry-root "$registry_root" key create --installation-id synthetic-primary --output "$v1_file" \
  >"$temp_root/cli.log" 2>&1 || fail_case registry_setup
printf '%s' 'synthetic-legacy-ordinary' >"$legacy_file"
chmod 0600 "$legacy_file"
"$dwh_auth" --registry-root "$registry_root" key import --legacy-raw --installation-id legacy-shared --from-file "$legacy_file" \
  >"$temp_root/cli.log" 2>&1 || fail_case registry_setup
"$dwh_auth" --registry-root "$registry_root" key create --installation-id synthetic-revoked --output "$revoked_file" \
  >"$temp_root/cli.log" 2>&1 || fail_case registry_setup
revoked_key=$(<"$revoked_file")
revoked_id=$(v1_key_id "$revoked_key")
"$dwh_auth" --registry-root "$registry_root" key revoke --key-id "$revoked_id" --reason synthetic \
  >"$temp_root/cli.log" 2>&1 || fail_case registry_setup
expires_at=$(date --utc --date='2 seconds' '+%Y-%m-%dT%H:%M:%SZ')
"$dwh_auth" --registry-root "$registry_root" key create --installation-id synthetic-expired --expires-at "$expires_at" --output "$expired_file" \
  >"$temp_root/cli.log" 2>&1 || fail_case registry_setup
sleep 3
v1_key=$(<"$v1_file")
legacy_key=$(<"$legacy_file")
expired_key=$(<"$expired_file")
report_pass registry_setup

current_case=verifier_start
"$dwh_auth" serve --registry-root "$registry_root" --socket "$service_socket" \
  >"$temp_root/verifier.log" 2>&1 &
verifier_pid=$!
register_pid "$verifier_pid"
wait_for_socket "$service_socket" || fail_case verifier_start
report_pass verifier_start

run_tcp_listener_detector_positive

current_case=synthetic_upstreams
AUTH_PROXY_SOCKET="$auth_proxy_socket" \
VERIFIER_SOCKET="$service_socket" \
MARKER_PORT_FILE="$marker_port_file" \
MARKER_OBSERVATIONS="$marker_observations" \
AUTH_OBSERVATIONS="$auth_observations" \
python3 - >"$temp_root/upstreams.log" 2>&1 <<'PY' &
import http.client
import http.server
import json
import os
import signal
import socket
import socketserver
import sys
import threading

proxy_socket = os.environ["AUTH_PROXY_SOCKET"]
verifier_socket = os.environ["VERIFIER_SOCKET"]
marker_port_file = os.environ["MARKER_PORT_FILE"]
marker_observations = os.environ["MARKER_OBSERVATIONS"]
auth_observations = os.environ["AUTH_OBSERVATIONS"]


def append_json(path, value):
    with open(path, "a", encoding="utf-8") as handle:
        handle.write(json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n")


class UnixHTTPConnection(http.client.HTTPConnection):
    def __init__(self, path):
        super().__init__("localhost")
        self.path = path

    def connect(self):
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.sock.connect(self.path)


class AuthProxy(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, _format, *_args):
        pass

    def do_GET(self):
        names = sorted(
            name.lower()
            for name in self.headers.keys()
            if name.lower() not in {"host", "connection"}
        )
        append_json(
            auth_observations,
            {
                "client_header_names": names,
                "has_authorization": "authorization" in self.headers,
                "has_cookie": "cookie" in self.headers,
                "has_dwh_key_id": "x-dwh-key-id" in self.headers,
                "method": self.command,
                "path": self.path,
            },
        )
        try:
            connection = UnixHTTPConnection(verifier_socket)
            connection.request(self.command, self.path, headers=dict(self.headers.items()))
            response = connection.getresponse()
            payload = response.read()
            self.send_response(response.status)
            for name, value in response.getheaders():
                if name.lower() not in {"connection", "transfer-encoding", "content-length"}:
                    self.send_header(name, value)
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)
            connection.close()
        except OSError:
            self.send_response(500)
            self.send_header("Content-Length", "0")
            self.end_headers()


class UnixHTTPServer(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
    daemon_threads = True


class Marker(http.server.BaseHTTPRequestHandler):
    def log_message(self, _format, *_args):
        pass

    def do_GET(self):
        append_json(
            marker_observations,
            {
                "has_api_key": "x-api-key" in self.headers,
                "has_dwh_key_id": "x-dwh-key-id" in self.headers,
                "path": self.path,
            },
        )
        payload = b"postgrest-marker\n"
        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)


for path in (proxy_socket,):
    try:
        os.unlink(path)
    except FileNotFoundError:
        pass

marker = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Marker)
auth_proxy = UnixHTTPServer(proxy_socket, AuthProxy)
os.chmod(proxy_socket, 0o600)
with open(marker_port_file, "w", encoding="ascii") as handle:
    handle.write(str(marker.server_address[1]))

for server in (marker, auth_proxy):
    threading.Thread(target=server.serve_forever, daemon=True).start()

signal.pause()
PY
upstreams_pid=$!
register_pid "$upstreams_pid"
wait_for_socket "$auth_proxy_socket" || fail_case synthetic_upstreams
wait_for_file "$marker_port_file" || fail_case synthetic_upstreams
marker_port=$(<"$marker_port_file")
[[ "$marker_port" =~ ^[0-9]+$ ]] || fail_case synthetic_upstreams
report_pass synthetic_upstreams

current_case=render_nginx
cp -- "$repo_root/deploy/dwh-auth/nginx-http.conf.example" "$runtime_http"
sed \
  -e "s|http://unix:/run/dwh-auth/verify.sock:/verify|http://unix:$auth_proxy_socket:/verify|" \
  -e "s|http://127.0.0.1:3001|http://127.0.0.1:$marker_port|" \
  "$repo_root/deploy/dwh-auth/nginx-dwh-location.conf.example" >"$runtime_location"
cat >"$nginx_config" <<EOF
worker_processes 1;
pid $nginx_prefix/nginx.pid;
error_log $nginx_prefix/error.log crit;

events {
    worker_connections 16;
}

http {
    access_log $nginx_prefix/access.log;
    include $runtime_http;

    server {
        listen unix:$nginx_socket;
        server_name synthetic;
        include $runtime_location;
    }
}
EOF
nginx -t -p "$nginx_prefix" -c "$nginx_config" >"$temp_root/nginx-test.log" 2>&1 || fail_case render_nginx
report_pass composite_nginx_config

current_case=nginx_start
nginx -p "$nginx_prefix" -c "$nginx_config" >"$temp_root/nginx-start.log" 2>&1 || fail_case nginx_start
wait_for_file "$nginx_prefix/nginx.pid" || fail_case nginx_start
nginx_pid=$(<"$nginx_prefix/nginx.pid")
[[ "$nginx_pid" =~ ^[0-9]+$ ]] || fail_case nginx_start
register_pid "$nginx_pid"
wait_for_socket "$nginx_socket" || fail_case nginx_start
report_pass nginx_start

current_case=auth_socket_unix_only
[[ -S "$service_socket" ]] || fail_case auth_socket_unix_only
ss -xl >"$temp_root/ss-unix.log" 2>&1 || fail_case auth_socket_unix_only
grep -Fq -- "$service_socket" "$temp_root/ss-unix.log" || fail_case auth_socket_unix_only
! tcp_listener_for_pid "$verifier_pid" || fail_case auth_socket_unix_only
report_pass auth_socket_unix_only

probe_body="$temp_root/probe.body"
expect_status verifier_not_public 404 "$probe_body" 'http://synthetic/_check_dwh_key'
report_pass verifier_not_public

route_url='http://synthetic/dwh/?keep=exact&second=two'
expect_status valid_v1 200 "$probe_body" \
  -H "X-API-Key: $v1_key" \
  -H 'Cookie: synthetic-session=one' \
  -H 'Authorization: Bearer synthetic' \
  -H 'X-DWH-Key-ID: client-spoof' \
  "$route_url"
[[ $(<"$probe_body") == 'postgrest-marker' ]] || fail_case valid_v1
report_pass valid_v1

expect_status valid_legacy 200 "$probe_body" -H "X-API-Key: $legacy_key" 'http://synthetic/dwh/?legacy=one'
[[ $(<"$probe_body") == 'postgrest-marker' ]] || fail_case valid_legacy
report_pass valid_legacy

expect_status invalid_key 401 "$probe_body" \
  -H 'X-API-Key: thtdwh_v1.AAAAAAAAAAAAAAAA.AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA' \
  'http://synthetic/dwh/?invalid=one'
report_pass invalid_key

expect_status revoked_key 401 "$probe_body" -H "X-API-Key: $revoked_key" 'http://synthetic/dwh/?revoked=one'
report_pass revoked_key

expect_status expired_key 401 "$probe_body" -H "X-API-Key: $expired_key" 'http://synthetic/dwh/?expired=one'
report_pass expired_key

expect_status duplicate_v1 401 "$probe_body" \
  -H "X-API-Key: $v1_key" \
  -H "X-API-Key: $v1_key" \
  'http://synthetic/dwh/?duplicate=v1'
report_pass duplicate_v1

expect_status duplicate_legacy 401 "$probe_body" \
  -H "X-API-Key: $legacy_key" \
  -H "X-API-Key: $legacy_key" \
  'http://synthetic/dwh/?duplicate=legacy'
report_pass duplicate_legacy

current_case=stopped_verifier
stop_registered_pid "$verifier_pid" || fail_case stopped_verifier
expect_status stopped_verifier 503 "$probe_body" -H "X-API-Key: $v1_key" 'http://synthetic/dwh/?unavailable=one'
report_pass stopped_verifier

current_case=header_and_path_isolation
AUTH_OBSERVATIONS="$auth_observations" MARKER_OBSERVATIONS="$marker_observations" python3 - >"$temp_root/assertions.log" 2>&1 <<'PY' || fail_case header_and_path_isolation
import json
import os


def load(path):
    with open(path, encoding="utf-8") as handle:
        return [json.loads(line) for line in handle if line.strip()]


auth = load(os.environ["AUTH_OBSERVATIONS"])
marker = load(os.environ["MARKER_OBSERVATIONS"])
assert len(auth) == 8
for request in auth:
    assert request["method"] == "GET"
    assert request["path"] == "/verify"
    assert request["client_header_names"] == ["x-api-key"]
    assert not request["has_authorization"]
    assert not request["has_cookie"]
    assert not request["has_dwh_key_id"]

assert len(marker) == 2
assert marker[0] == {
    "has_api_key": False,
    "has_dwh_key_id": False,
    "path": "/dwh/?keep=exact&second=two",
}
assert marker[1] == {
    "has_api_key": False,
    "has_dwh_key_id": False,
    "path": "/dwh/?legacy=one",
}
PY
report_pass header_and_path_isolation

report_pass summary