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

ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
TEST_DIR="$(mktemp -d)"
STUB_PID=""
PASS_COUNT=0

cleanup() {
  if [[ -n "$STUB_PID" ]]; then
    kill "$STUB_PID" >/dev/null 2>&1 || true
    wait "$STUB_PID" >/dev/null 2>&1 || true
  fi
  rm -rf -- "$TEST_DIR"
}
trap cleanup EXIT

pass() {
  PASS_COUNT=$((PASS_COUNT + 1))
  echo "[PASS] $1"
}

expect_failure() {
  local label="$1"
  shift
  if "$@"; then
    echo "[FAIL] expected failure: $label" >&2
    exit 1
  fi
  pass "$label"
}

capture_lines() {
  local capture="$1"
  if [[ -f "$capture" ]]; then
    wc -l <"$capture" | tr -d ' '
  else
    echo 0
  fi
}

new_state_dir() {
  local target="$1"
  mkdir -p "$target"
  chmod 0700 "$target"
}

set_mode() {
  printf '%s\n' "$1" >"$TEST_DIR/mode"
}

probe() {
  local state_dir="$1"
  local capture="$2"
  shift 2
  env HYB_WECOM_ALERT_TEST_MODE=1 HYB_WECOM_ALERT_TEST_CAPTURE="$capture" \
    python3 "$ROOT_DIR/scripts/check-public-app-health.py" \
      --ready-url "https://localhost:${STUB_PORT}/ready" \
      --state-dir "$state_dir" \
      --webhook-file "$TEST_DIR/wecom-webhook" \
      --alert-helper "$ROOT_DIR/scripts/send-wecom-group-alert.py" \
      --failure-threshold 2 \
      --timeout-seconds 1 \
      --ca-file "$TEST_DIR/cert.pem" \
      "$@"
}

openssl req -x509 -newkey rsa:2048 -nodes -sha256 -days 1 \
  -subj '/CN=localhost' -addext 'subjectAltName=DNS:localhost' \
  -keyout "$TEST_DIR/key.pem" -out "$TEST_DIR/cert.pem" >/dev/null 2>&1
chmod 0600 "$TEST_DIR/key.pem" "$TEST_DIR/cert.pem"
WECOM_TEST_HOST='qyapi.weixin.qq.com'
WECOM_TEST_PATH='/cgi-bin/webhook/send'
printf 'https://%s%s?key=%s\n' "$WECOM_TEST_HOST" "$WECOM_TEST_PATH" \
  '01234567-89ab-cdef-0123-456789abcdef' >"$TEST_DIR/wecom-webhook"
chmod 0600 "$TEST_DIR/wecom-webhook"
printf '%s\n' healthy >"$TEST_DIR/mode"

python3 - "$TEST_DIR/cert.pem" "$TEST_DIR/key.pem" "$TEST_DIR/mode" "$TEST_DIR/port" <<'PY' >/dev/null 2>&1 &
import http.server
import json
import ssl
import sys
import time
from pathlib import Path

cert_file, key_file, mode_file, port_file = map(Path, sys.argv[1:])


class Handler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        mode = mode_file.read_text(encoding="utf-8").strip()
        if self.path != "/ready":
            self.send_response(404)
            self.end_headers()
            return
        if mode == "timeout":
            time.sleep(2)
        if mode == "unavailable":
            status, body = 503, b'{"error":{"details":{"db":"do-not-leak"}}}'
        elif mode == "invalid-json":
            status, body = 200, b'not-json-private-payload'
        elif mode == "bad-contract":
            status, body = 200, json.dumps({"data": {"ok": False}}).encode()
        elif mode == "oversized":
            status, body = 200, b"x" * (64 * 1024 + 1)
        else:
            status, body = 200, json.dumps({"data": {"ok": True, "buildSha": "abcdef123456"}}).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *_args):
        pass


server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(certfile=cert_file, keyfile=key_file)
server.socket = context.wrap_socket(server.socket, server_side=True)
port_file.write_text(str(server.server_port), encoding="utf-8")
server.serve_forever()
PY
STUB_PID=$!

for _ in $(seq 1 50); do
  [[ -s "$TEST_DIR/port" ]] && break
  sleep 0.1
done
[[ -s "$TEST_DIR/port" ]]
STUB_PORT="$(cat "$TEST_DIR/port")"

MAIN_STATE="$TEST_DIR/main-state"
MAIN_CAPTURE="$TEST_DIR/main-capture.jsonl"
new_state_dir "$MAIN_STATE"
probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
[[ "$(capture_lines "$MAIN_CAPTURE")" == "0" ]]
[[ "$(stat -c %a "$MAIN_STATE/app-health-probe-state.json")" == "600" ]]
pass "healthy baseline writes state without sending an alert"

set_mode unavailable
expect_failure "first failure stays pending" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
[[ "$(capture_lines "$MAIN_CAPTURE")" == "0" ]]
expect_failure "second consecutive failure triggers alert evaluation" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
[[ "$(capture_lines "$MAIN_CAPTURE")" == "1" ]]
expect_failure "identical continued failure remains non-zero" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
[[ "$(capture_lines "$MAIN_CAPTURE")" == "1" ]]
pass "identical application failure is deduplicated"

set_mode invalid-json
expect_failure "changed failure category remains non-zero" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
[[ "$(capture_lines "$MAIN_CAPTURE")" == "2" ]]
pass "changed failure category sends one updated alert"

set_mode healthy
probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
[[ "$(capture_lines "$MAIN_CAPTURE")" == "2" ]]
pass "first healthy sample keeps recovery pending"
probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
[[ "$(capture_lines "$MAIN_CAPTURE")" == "3" ]]
probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
[[ "$(capture_lines "$MAIN_CAPTURE")" == "3" ]]
pass "second healthy sample sends exactly one recovery notification"

set_mode invalid-json
expect_failure "first same-category recurrence stays pending" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
[[ "$(capture_lines "$MAIN_CAPTURE")" == "3" ]]
expect_failure "second same-category recurrence enters flapping" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
expect_failure "continued flapping failure remains deduplicated" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
set_mode healthy
probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
python3 - "$MAIN_STATE/app-health-probe-state.json" <<'PY'
import json, os, sys, time
path = sys.argv[1]
payload = json.load(open(path, encoding="utf-8"))
assert payload["phase"] == "flapping", payload
payload["healthySince"] = int(time.time()) - 1800
with open(path, "w", encoding="utf-8") as handle:
    json.dump(payload, handle)
    handle.write("\n")
os.chmod(path, 0o600)
PY
probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
python3 - "$MAIN_STATE/app-health-probe-state.json" <<'PY'
import json, sys
payload = json.load(open(sys.argv[1], encoding="utf-8"))
assert payload["phase"] == "healthy", payload
assert payload["flapTransitions"] == 0, payload
PY
pass "flapping sends once and exits silently after thirty healthy minutes"

DIFFERENT_STATE="$TEST_DIR/different-state"
DIFFERENT_CAPTURE="$TEST_DIR/different-capture.jsonl"
new_state_dir "$DIFFERENT_STATE"
set_mode healthy
probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null
set_mode unavailable
expect_failure "different-case first failure stays pending" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
expect_failure "different-case second failure alerts" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
set_mode healthy
probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null
probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null
[[ "$(capture_lines "$DIFFERENT_CAPTURE")" == "2" ]]
set_mode invalid-json
expect_failure "new-category recurrence first sample stays pending" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
[[ "$(capture_lines "$DIFFERENT_CAPTURE")" == "2" ]]
expect_failure "new-category recurrence becomes a normal alert" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
[[ "$(capture_lines "$DIFFERENT_CAPTURE")" == "3" ]]
python3 - "$DIFFERENT_STATE/app-health-probe-state.json" <<'PY'
import json, sys
payload = json.load(open(sys.argv[1], encoding="utf-8"))
assert payload["phase"] == "alerted", payload
assert payload["activeCategory"] == "response_contract_invalid", payload
PY
pass "different failure category is not swallowed by same-category flapping debounce"

TRANSIENT_STATE="$TEST_DIR/transient-state"
TRANSIENT_CAPTURE="$TEST_DIR/transient-capture.jsonl"
new_state_dir "$TRANSIENT_STATE"
set_mode healthy
probe "$TRANSIENT_STATE" "$TRANSIENT_CAPTURE" >/dev/null
set_mode unavailable
expect_failure "single transient outage is recorded" probe "$TRANSIENT_STATE" "$TRANSIENT_CAPTURE" >/dev/null 2>&1
set_mode healthy
probe "$TRANSIENT_STATE" "$TRANSIENT_CAPTURE" >/dev/null
[[ "$(capture_lines "$TRANSIENT_CAPTURE")" == "0" ]]
pass "single transient outage does not send failure or recovery noise"

for mode in bad-contract oversized timeout; do
  CASE_STATE="$TEST_DIR/${mode}-state"
  new_state_dir "$CASE_STATE"
  set_mode "$mode"
  expect_failure "$mode is rejected" probe "$CASE_STATE" "$TEST_DIR/${mode}-capture.jsonl" >/dev/null 2>&1
  python3 - "$CASE_STATE/app-health-probe-state.json" "$mode" <<'PY'
import json, sys
expected = {
    "bad-contract": "response_contract_invalid",
    "oversized": "response_too_large",
    "timeout": "request_timeout",
}[sys.argv[2]]
payload = json.load(open(sys.argv[1], encoding="utf-8"))
assert payload["activeCategory"] == expected, payload
PY
done

python3 - "$ROOT_DIR/scripts/check-public-app-health.py" <<'PY'
import importlib.util
import socket
import sys
import urllib.error

spec = importlib.util.spec_from_file_location("app_health", sys.argv[1])
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
assert module.classify_url_error(urllib.error.URLError(socket.gaierror(-2, "private-dns-text"))) == "dns_resolution_failed"
assert module.classify_url_error(urllib.error.URLError(ConnectionRefusedError())) == "tcp_connection_failed"

watch = module.new_probe_state()
watch.update({
    "phase": "recovery_watch",
    "lastRecoveredCategory": "http_unavailable",
    "recoveredAt": 1000,
})
pending, candidate = module.transition_probe_state(watch, "http_unavailable", 4599, 2)
assert pending["phase"] == "failure_pending" and candidate is None, (pending, candidate)
flapping, candidate = module.transition_probe_state(pending, "http_unavailable", 4599, 2)
assert flapping["phase"] == "flapping", flapping
assert candidate == ("FAIL_APP_HEALTH_FLAPPING", "http_unavailable"), candidate

expired, candidate = module.transition_probe_state(watch, "http_unavailable", 4600, 2)
assert expired["phase"] == "failure_pending" and expired["recoveredAt"] is None, expired
alerted, candidate = module.transition_probe_state(expired, "http_unavailable", 4600, 2)
assert alerted["phase"] == "alerted", alerted
assert candidate == ("FAIL_APP_HEALTH", "http_unavailable"), candidate

flapping["healthySince"] = 5000
flapping["consecutiveFailures"] = 0
flapping["consecutiveSuccesses"] = 1
stable, candidate = module.transition_probe_state(flapping, None, 6800, 2)
assert stable["phase"] == "healthy" and candidate is None, (stable, candidate)
PY
pass "DNS/TCP classification and exact thirty/sixty-minute boundaries are deterministic"

UNUSED_PORT="$(python3 - <<'PY'
import socket
sock = socket.socket()
sock.bind(("127.0.0.1", 0))
print(sock.getsockname()[1])
sock.close()
PY
)"
CONNECT_STATE="$TEST_DIR/connect-state"
new_state_dir "$CONNECT_STATE"
expect_failure "connection failure is rejected" env \
  HYB_WECOM_ALERT_TEST_MODE=1 HYB_WECOM_ALERT_TEST_CAPTURE="$TEST_DIR/connect-capture.jsonl" \
  python3 "$ROOT_DIR/scripts/check-public-app-health.py" \
    --ready-url "https://localhost:${UNUSED_PORT}/ready" --state-dir "$CONNECT_STATE" \
    --webhook-file "$TEST_DIR/wecom-webhook" --alert-helper "$ROOT_DIR/scripts/send-wecom-group-alert.py" \
    --failure-threshold 2 --timeout-seconds 1 --ca-file "$TEST_DIR/cert.pem" >/dev/null 2>&1
python3 - "$CONNECT_STATE/app-health-probe-state.json" <<'PY'
import json, sys
payload = json.load(open(sys.argv[1], encoding="utf-8"))
assert payload["activeCategory"] == "tcp_connection_failed", payload
PY
pass "TCP connection failure is stored separately from DNS resolution failure"

TLS_STATE="$TEST_DIR/tls-state"
new_state_dir "$TLS_STATE"
expect_failure "untrusted TLS certificate is rejected" env \
  HYB_WECOM_ALERT_TEST_MODE=1 HYB_WECOM_ALERT_TEST_CAPTURE="$TEST_DIR/tls-capture.jsonl" \
  python3 "$ROOT_DIR/scripts/check-public-app-health.py" \
    --ready-url "https://localhost:${STUB_PORT}/ready" --state-dir "$TLS_STATE" \
    --webhook-file "$TEST_DIR/wecom-webhook" --alert-helper "$ROOT_DIR/scripts/send-wecom-group-alert.py" \
    --failure-threshold 2 --timeout-seconds 1 >/dev/null 2>&1
python3 - "$TLS_STATE/app-health-probe-state.json" <<'PY'
import json, sys
payload = json.load(open(sys.argv[1], encoding="utf-8"))
assert payload["activeCategory"] == "tls_validation_failed", payload
PY
pass "TLS failure is stored as a stable category"

MIGRATION_STATE="$TEST_DIR/migration-state"
MIGRATION_CAPTURE="$TEST_DIR/migration-capture.jsonl"
new_state_dir "$MIGRATION_STATE"
printf '%s\n' '{"schemaVersion":1,"consecutiveFailures":0,"lastCategory":null}' \
  >"$MIGRATION_STATE/app-health-probe-state.json"
chmod 0600 "$MIGRATION_STATE/app-health-probe-state.json"
set_mode healthy
probe "$MIGRATION_STATE" "$MIGRATION_CAPTURE" >/dev/null
python3 - "$MIGRATION_STATE/app-health-probe-state.json" <<'PY'
import json, sys
payload = json.load(open(sys.argv[1], encoding="utf-8"))
assert payload["schemaVersion"] == 2, payload
assert payload["phase"] == "healthy", payload
PY
[[ "$(capture_lines "$MIGRATION_CAPTURE")" == "0" ]]
pass "valid legacy state migrates without backfilling a notification"

INVALID_SCHEMA_STATE="$TEST_DIR/invalid-schema-state"
new_state_dir "$INVALID_SCHEMA_STATE"
printf '%s\n' '{"schemaVersion":2,"phase":"healthy","activeCategory":null,"lastRecoveredCategory":null,"consecutiveFailures":1000001,"consecutiveSuccesses":0,"recoveredAt":null,"healthySince":null,"flapTransitions":0}' \
  >"$INVALID_SCHEMA_STATE/app-health-probe-state.json"
chmod 0600 "$INVALID_SCHEMA_STATE/app-health-probe-state.json"
set_mode healthy
expect_failure "out-of-range v2 state fails closed" probe "$INVALID_SCHEMA_STATE" \
  "$TEST_DIR/invalid-schema-capture.jsonl" >"$TEST_DIR/invalid-schema-stdout" 2>"$TEST_DIR/invalid-schema-stderr"
grep -Fxq '[app-health][ERROR] local_state_invalid' "$TEST_DIR/invalid-schema-stderr"
! grep -Fq '1000001' "$TEST_DIR/invalid-schema-stderr"
pass "v2 state schema rejects inconsistent or out-of-range fields without disclosure"

CORRUPT_STATE="$TEST_DIR/corrupt-state"
new_state_dir "$CORRUPT_STATE"
printf '%s\n' 'private-payload-must-not-appear' >"$CORRUPT_STATE/app-health-probe-state.json"
chmod 0600 "$CORRUPT_STATE/app-health-probe-state.json"
set_mode healthy
expect_failure "corrupt local state fails closed" probe "$CORRUPT_STATE" "$TEST_DIR/corrupt-capture.jsonl" \
  >"$TEST_DIR/corrupt-stdout" 2>"$TEST_DIR/corrupt-stderr"
grep -Fxq '[app-health][ERROR] local_state_invalid' "$TEST_DIR/corrupt-stderr"
! grep -Fq 'private-payload' "$TEST_DIR/corrupt-stderr"
! grep -Fq "$TEST_DIR" "$TEST_DIR/corrupt-stderr"
pass "local state error output is stable and redacted"

python3 - "$MAIN_CAPTURE" <<'PY'
import json, sys
messages = [json.loads(line)["text"]["content"] for line in open(sys.argv[1], encoding="utf-8")]
assert len(messages) == 4, messages
assert "FAIL_APP_HEALTH" in messages[0]
assert "http_unavailable" in messages[0]
assert "response_contract_invalid" in messages[1]
assert "PASS_APP_HEALTH_READY" in messages[2]
assert "FAIL_APP_HEALTH_FLAPPING" in messages[3]
assert messages[0].startswith("OA 公网健康告警\n"), messages[0]
assert messages[2].startswith("OA 公网健康恢复\n"), messages[2]
assert messages[3].startswith("OA 公网健康链路反复抖动\n"), messages[3]
for forbidden in ("do-not-leak", "localhost", "https://", "/tmp/", "not-json-private-payload"):
    assert all(forbidden not in message for message in messages), (forbidden, messages)
PY
pass "captured application alerts use explicit source titles and stable redacted categories"

expect_failure "plain HTTP readiness URL is rejected" env \
  HYB_WECOM_ALERT_TEST_MODE=1 HYB_WECOM_ALERT_TEST_CAPTURE="$TEST_DIR/http-capture.jsonl" \
  python3 "$ROOT_DIR/scripts/check-public-app-health.py" \
    --ready-url "http://localhost:${STUB_PORT}/ready" --state-dir "$TEST_DIR/http-state" \
    --webhook-file "$TEST_DIR/wecom-webhook" --alert-helper "$ROOT_DIR/scripts/send-wecom-group-alert.py" \
    >/dev/null 2>&1

PYTHONPYCACHEPREFIX="$TEST_DIR/pycache" python3 -m py_compile \
  "$ROOT_DIR/scripts/check-public-app-health.py" "$ROOT_DIR/scripts/send-wecom-group-alert.py"
bash -n "$ROOT_DIR/deploy/trialrun/install-offhost-app-health-monitor.sh"
grep -Fq 'OnCalendar=*:0/5' "$ROOT_DIR/deploy/trialrun/install-offhost-app-health-monitor.sh"
systemd-analyze calendar '*:0/5' >/dev/null
grep -Fq 'disabled application health monitor installed' "$ROOT_DIR/deploy/trialrun/install-offhost-app-health-monitor.sh"
grep -Fq 'install -o root -g root -m 0755 "$ALERT_HELPER_SOURCE" "$ALERT_HELPER_TARGET"' "$ROOT_DIR/deploy/trialrun/install-offhost-app-health-monitor.sh"
grep -Fq 'state and shared alert files retained' "$ROOT_DIR/deploy/trialrun/install-offhost-app-health-monitor.sh"
grep -Fq 'PROBE_TARGET must remain /usr/local/sbin/hyb-oa-public-app-health' "$ROOT_DIR/deploy/trialrun/install-offhost-app-health-monitor.sh"
grep -Fq 'ALERT_HELPER_TARGET must remain /usr/local/sbin/hyb-oa-wecom-group-alert' "$ROOT_DIR/deploy/trialrun/install-offhost-app-health-monitor.sh"
grep -Fq 'STATE_DIR must remain /var/lib/hyb-oa-app-health' "$ROOT_DIR/deploy/trialrun/install-offhost-app-health-monitor.sh"
grep -Fq 'SERVICE_NAME must remain hyb-oa-public-app-health' "$ROOT_DIR/deploy/trialrun/install-offhost-app-health-monitor.sh"
grep -Fq 'SYSTEMD_DIR must remain /etc/systemd/system' "$ROOT_DIR/deploy/trialrun/install-offhost-app-health-monitor.sh"
pass "installer is five-minute, disabled-by-default, exact-target restricted, and preserves shared state"

printf '[PASS] public application health contract completed (%d checks)\n' "$PASS_COUNT"