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

script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
# shellcheck source=scripts/cluster/common.sh
source "$script_dir/common.sh"
cluster_root=${cluster_root:?}

usage() {
  cat <<'EOF'
Usage:
  scripts/cluster/monitor-production-release.sh \
    --run-id <production-run-id> --release-manifest dist/releases/<SHA>.json

Runs a finite 24-hour post-promotion monitor: 10-second samples for the first
2 hours, then 60-second samples. Test-only duration overrides are available via
AI_GATEWAY_RELEASE_MONITOR_DENSE_SECONDS and
AI_GATEWAY_RELEASE_MONITOR_TOTAL_SECONDS.
EOF
}

[[ ${1:-} == --run-id && ${3:-} == --release-manifest && $# -eq 4 ]] || {
  usage >&2
  exit 64
}
run_id=$2
release_manifest=$4
[[ $run_id =~ ^[0-9a-f-]{36}$ && -f $release_manifest && ! -L $release_manifest ]] || {
  echo 'invalid monitor Run ID or release manifest' >&2
  exit 64
}

load_cluster_env
require_commands curl jq node openssl
: "${AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN:?}"
: "${AI_GATEWAY_RELEASE_MONITOR_DENSE_SECONDS:=7200}"
: "${AI_GATEWAY_RELEASE_MONITOR_TOTAL_SECONDS:=86400}"
: "${AI_GATEWAY_RELEASE_MONITOR_DENSE_INTERVAL_SECONDS:=10}"
: "${AI_GATEWAY_RELEASE_MONITOR_SPARSE_INTERVAL_SECONDS:=60}"
[[ $AI_GATEWAY_RELEASE_MONITOR_DENSE_SECONDS =~ ^[0-9]+$ &&
   $AI_GATEWAY_RELEASE_MONITOR_TOTAL_SECONDS =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_RELEASE_MONITOR_DENSE_INTERVAL_SECONDS =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_RELEASE_MONITOR_SPARSE_INTERVAL_SECONDS =~ ^[1-9][0-9]*$ &&
   $AI_GATEWAY_RELEASE_MONITOR_DENSE_SECONDS -le $AI_GATEWAY_RELEASE_MONITOR_TOTAL_SECONDS ]] || {
  echo 'invalid monitor duration configuration' >&2
  exit 64
}

node "$cluster_root/scripts/release-manifest.mjs" validate "$release_manifest" >/dev/null
release_sha=$(node "$cluster_root/scripts/release-manifest.mjs" get "$release_manifest" sourceSha)
api_image=$(node "$cluster_root/scripts/release-manifest.mjs" get "$release_manifest" images.api)
api_digest=${api_image##*@}
namespace=${AI_GATEWAY_K3S_NAMESPACE:-easyai}
report_root=${AI_GATEWAY_ACCEPTANCE_REPORT_DIR:-"$cluster_root/dist/acceptance/$run_id"}
overall_report=$report_root/acceptance-report.json
monitor_report=$report_root/post-promotion-monitor.json
observations=$report_root/post-promotion-observations.csv
capacity_snapshot=$report_root/pre-acceptance-capacity.json
[[ -f $overall_report && ! -L $overall_report &&
   -f $capacity_snapshot && ! -L $capacity_snapshot ]] || {
  echo 'monitor requires the overall report and pre-acceptance capacity snapshot' >&2
  exit 1
}

temporary_root=$(mktemp -d)
chmod 0700 "$temporary_root"
trap '[[ $temporary_root == /tmp/* || $temporary_root == /var/folders/* ]] && rm -rf -- "$temporary_root"' EXIT

remote_kubectl() {
  local command_text
  printf -v command_text '%q ' "$@"
  cluster_ssh "$CLUSTER_NINGBO_HOST" "k3s kubectl $command_text"
}

database_query() {
  local sql=$1 primary
  primary=$(remote_kubectl get cluster easyai-postgres -n "$namespace" \
    -o 'jsonpath={.status.currentPrimary}')
  remote_kubectl exec -n "$namespace" "$primary" -c postgres -- \
    psql -X -v ON_ERROR_STOP=1 -U postgres -d easyai_ai_gateway -At -c "$sql"
}

admin_request() {
  local method=$1 path=$2 body=${3:-} output=$4
  local service_ip service_port token_encoded body_encoded response status payload script command
  service_ip=$(remote_kubectl get service api-ningbo-edge -n "$namespace" -o 'jsonpath={.spec.clusterIP}')
  service_port=$(remote_kubectl get service api-ningbo-edge -n "$namespace" -o 'jsonpath={.spec.ports[0].port}')
  token_encoded=$(printf '%s' "$AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN" | openssl base64 -A)
  body_encoded=$(printf '%s' "$body" | openssl base64 -A)
  # shellcheck disable=SC2016
  script='
set -euo pipefail
token=$(printf "%s" "$4" | base64 -d)
body=$(printf "%s" "$5" | base64 -d)
args=(-sS --max-time 30 -w "\n%{http_code}" -X "$1" -H "Authorization: Bearer $token")
if [[ -n $body ]]; then args+=(-H "Content-Type: application/json" --data-binary "$body"); fi
curl "${args[@]}" "http://$2:$3$6"
'
  printf -v command 'bash -c %q -- %q %q %q %q %q %q' \
    "$script" "$method" "$service_ip" "$service_port" "$token_encoded" "$body_encoded" "$path"
  response=$(cluster_ssh "$CLUSTER_NINGBO_HOST" "$command")
  status=${response##*$'\n'}
  payload=${response%$'\n'*}
  printf '%s' "$payload" >"$output"
  [[ $status =~ ^2[0-9][0-9]$ ]]
}

metric_sum() {
  local pattern=$1 sum=0 pod value metrics
  while read -r pod; do
    [[ -n $pod ]] || continue
    metrics=$(remote_kubectl exec -n "$namespace" "$pod" -- \
      wget -qO- http://127.0.0.1:8088/metrics)
    value=$(awk -v pattern="$pattern" '$0 ~ pattern {sum += $NF} END {print sum+0}' <<<"$metrics")
    sum=$(awk -v left="$sum" -v right="$value" 'BEGIN {print left+right}')
  done < <(remote_kubectl get pods -n "$namespace" \
    -l app.kubernetes.io/name=easyai-worker \
    -o 'jsonpath={range .items[*]}{.metadata.name}{"\n"}{end}')
  printf '%s\n' "$sum"
}

pool_saturated_pods() {
  local pod metrics max acquired idle saturated=0
  while read -r pod; do
    [[ -n $pod ]] || continue
    metrics=$(remote_kubectl exec -n "$namespace" "$pod" -- \
      wget -qO- http://127.0.0.1:8088/metrics)
    max=$(awk '$1=="easyai_gateway_postgres_pool_max_connections" {print int($2)}' <<<"$metrics")
    acquired=$(awk '$1=="easyai_gateway_postgres_pool_acquired_connections" {print int($2)}' <<<"$metrics")
    idle=$(awk '$1=="easyai_gateway_postgres_pool_idle_connections" {print int($2)}' <<<"$metrics")
    [[ -z $max || -z $acquired || -z $idle ]] || {
      (( acquired == max && idle == 0 )) && saturated=$((saturated + 1))
    }
  done < <(remote_kubectl get pods -n "$namespace" \
    -l 'app.kubernetes.io/part-of=easyai-ai-gateway' \
    -o 'jsonpath={range .items[*]}{.metadata.name}{"\n"}{end}')
  printf '%s\n' "$saturated"
}

check_wireguard() {
  local -a hosts=("$CLUSTER_NINGBO_HOST" "$CLUSTER_HONGKONG_HOST" "$CLUSTER_LOS_ANGELES_HOST")
  local -a ips=(10.77.0.1 10.77.0.2 10.77.0.3)
  [[ $CLUSTER_WORKER_NODE_4_ENABLED != true ]] || {
    hosts+=("$CLUSTER_WORKER_NODE_4_HOST")
    ips+=(10.77.0.4)
  }
  local source target output loss average handshakes peer_count
  peer_count=$((${#hosts[@]} - 1))
  for source in "${!hosts[@]}"; do
    handshakes=$(cluster_ssh "${hosts[$source]}" \
      "wg show wg0 latest-handshakes | awk '\$2 > 0 && systime()-\$2 < 180 {count++} END {print count+0}'")
    (( handshakes == peer_count )) || return 1
    for target in "${!hosts[@]}"; do
      (( source == target )) && continue
      output=$(cluster_ssh "${hosts[$source]}" "ping -q -c 10 -W 2 ${ips[$target]}") || return 1
      loss=$(awk -F',' '/packet loss/ {gsub(/[^0-9.]/,"",$3); print $3}' <<<"$output")
      average=$(awk -F'=' '/min\\/avg\\/max/ {split($2,v,"/"); gsub(/ /,"",v[2]); print v[2]}' <<<"$output")
      if (( source == 2 || target == 2 )); then
        awk -v loss="$loss" -v average="$average" 'BEGIN {exit !(loss < 1 && average < 300)}' || return 1
      else
        awk -v loss="$loss" -v average="$average" 'BEGIN {exit !(loss < 1 && average < 80)}' || return 1
      fi
    done
  done
}

restore_previous_capacity() {
  local remote_file="/root/easyai-pre-acceptance-$run_id.json"
  cluster_scp "$capacity_snapshot" "$CLUSTER_NINGBO_HOST:$remote_file" >/dev/null
  cluster_ssh "$CLUSTER_NINGBO_HOST" \
    "k3s kubectl apply -f '$remote_file' >/dev/null && unlink '$remote_file'"
  for deployment in easyai-api-ningbo easyai-api-hongkong easyai-worker-ningbo \
    easyai-worker-hongkong easyai-capacity-controller; do
    remote_kubectl rollout status deployment/"$deployment" -n "$namespace" --timeout=600s
  done
}

pause_and_restore() {
  local gate_id=$1 mode_file=$temporary_root/mode.json pause_file=$temporary_root/pause.json body
  admin_request GET /api/admin/system/acceptance/traffic-mode '' "$mode_file"
  body=$(jq -cn \
    --argjson revision "$(jq -r '.revision' "$mode_file")" \
    --arg releaseSha "$release_sha" \
    --arg apiDigest "$api_digest" \
    --arg reason "$gate_id" \
    '{
      revision:$revision,
      releaseSha:$releaseSha,
      apiImageDigest:$apiDigest,
      workerImageDigest:$apiDigest,
      reason:$reason
    }')
  admin_request POST /api/admin/system/acceptance/traffic-mode/pause "$body" "$pause_file"
  [[ $(jq -r '.mode' "$pause_file") == validation ]]
  restore_previous_capacity
}

started_epoch=$(date +%s)
started_at=$(date -u '+%Y-%m-%dT%H:%M:%SZ')
monitor_since=$(date -u '+%Y-%m-%d %H:%M:%S+00')
baseline_lease_failure=$(metric_sum 'outcome="failure"')
baseline_lease_lost=$(metric_sum 'outcome="lost"')
samples=0
queue_growth_streak=0
pool_saturation_streak=0
previous_queue=0
failure_gate_id=
last_wireguard_epoch=0
printf 'timestamp,queue_depth,oldest_wait_seconds,connections,max_connections,saturated_pods,node_max_memory_percent,lease_failure,lease_lost\n' >"$observations"

while (( $(date +%s) - started_epoch < AI_GATEWAY_RELEASE_MONITOR_TOTAL_SECONDS )); do
  now_epoch=$(date +%s)
  elapsed=$((now_epoch - started_epoch))
  interval=$AI_GATEWAY_RELEASE_MONITOR_SPARSE_INTERVAL_SECONDS
  (( elapsed < AI_GATEWAY_RELEASE_MONITOR_DENSE_SECONDS )) &&
    interval=$AI_GATEWAY_RELEASE_MONITOR_DENSE_INTERVAL_SECONDS
  samples=$((samples + 1))

  if ! curl -fsS --max-time 10 https://ai.51easyai.com/api/v1/healthz >/dev/null ||
     ! curl -fsS --max-time 10 https://ai.51easyai.com/api/v1/readyz >/dev/null; then
    failure_gate_id=public_health
  fi
  ready_nodes=$(remote_kubectl get nodes -o json |
    jq '[.items[] | select(any(.status.conditions[]; .type=="Ready" and .status=="True"))] | length')
  pressure_nodes=$(remote_kubectl get nodes -o json |
    jq '[.items[] | select(any(.status.conditions[]; .type=="MemoryPressure" and .status=="True"))] | length')
  [[ $ready_nodes == 3 && $pressure_nodes == 0 ]] || failure_gate_id=${failure_gate_id:-node_health}
  node_max_memory=$(remote_kubectl top nodes --no-headers |
    awk '{gsub(/%/,"",$5); if ($5>max) max=$5} END {print max+0}')
  (( node_max_memory < 85 )) || failure_gate_id=${failure_gate_id:-node_memory_hard}
  unhealthy=$(remote_kubectl get pods -n "$namespace" -o json |
    jq '[.items[]
      | select(.metadata.labels["app.kubernetes.io/name"] == "easyai-api" or
               .metadata.labels["app.kubernetes.io/name"] == "easyai-worker")
      | .status.containerStatuses[]?
      | select(.restartCount > 0 or .lastState.terminated.reason == "OOMKilled" or
               .state.terminated.reason == "OOMKilled")] | length')
  [[ $unhealthy == 0 ]] || failure_gate_id=${failure_gate_id:-pod_restart_or_oom}
  pod_max_memory=$(remote_kubectl top pods -n "$namespace" \
    -l 'app.kubernetes.io/part-of=easyai-ai-gateway' --no-headers |
    awk '{
      value=$3
      if (value ~ /Gi$/) {sub(/Gi$/,"",value); value*=1024}
      else if (value ~ /Mi$/) {sub(/Mi$/,"",value)}
      else if (value ~ /Ki$/) {sub(/Ki$/,"",value); value/=1024}
      if (value>max) max=value
    } END {printf "%.0f",max}')
  (( pod_max_memory < 1536 )) || failure_gate_id=${failure_gate_id:-pod_memory_hard}
  [[ $(remote_kubectl get cluster easyai-postgres -n "$namespace" \
    -o 'jsonpath={.status.readyInstances}') == 2 ]] || failure_gate_id=${failure_gate_id:-postgres_ready}
  sync_state=$(database_query "SELECT COALESCE(string_agg(sync_state,','),'') FROM pg_stat_replication;")
  [[ ",$sync_state," == *,sync,* || ",$sync_state," == *,quorum,* ]] ||
    failure_gate_id=${failure_gate_id:-postgres_replication}
  connections=$(database_query "SELECT count(*) FROM pg_stat_activity WHERE backend_type='client backend';")
  max_connections=$(database_query "SELECT setting::int FROM pg_settings WHERE name='max_connections';")
  (( connections < 150 && connections * 4 < max_connections * 3 )) ||
    failure_gate_id=${failure_gate_id:-postgres_connections}
  queue_state=$(database_query "
SELECT count(*) FILTER (WHERE status='queued')||':'||
       COALESCE(EXTRACT(EPOCH FROM now()-MIN(created_at)) FILTER (WHERE status='queued'),0)::bigint
FROM gateway_tasks
WHERE run_mode='production';")
  IFS=: read -r queue_depth oldest_wait <<<"$queue_state"
  if (( queue_depth > previous_queue && queue_depth > 0 )); then
    queue_growth_streak=$((queue_growth_streak + 1))
  else
    queue_growth_streak=0
  fi
  previous_queue=$queue_depth
  (( oldest_wait <= 900 && queue_growth_streak < 4 )) ||
    failure_gate_id=${failure_gate_id:-queue_growth_or_oldest_wait}
  saturated=$(pool_saturated_pods)
  if (( saturated > 0 )); then
    pool_saturation_streak=$((pool_saturation_streak + 1))
  else
    pool_saturation_streak=0
  fi
  (( pool_saturation_streak * interval < 30 )) ||
    failure_gate_id=${failure_gate_id:-postgres_pool_saturation}
  lease_failure=$(metric_sum 'outcome="failure"')
  lease_lost=$(metric_sum 'outcome="lost"')
  awk -v current="$lease_failure" -v baseline="$baseline_lease_failure" 'BEGIN {exit !(current<=baseline)}' ||
    failure_gate_id=${failure_gate_id:-lease_renewal_failure}
  awk -v current="$lease_lost" -v baseline="$baseline_lease_lost" 'BEGIN {exit !(current<=baseline)}' ||
    failure_gate_id=${failure_gate_id:-lease_lost}
  duplicates=$(database_query "
SELECT
  (SELECT count(*) FROM (
    SELECT remote_task_id
    FROM gateway_tasks
    WHERE run_mode='production' AND created_at >= '$monitor_since'::timestamptz
      AND remote_task_id IS NOT NULL
    GROUP BY remote_task_id HAVING count(*)>1
  ) d)
  +
  (SELECT count(*) FROM (
    SELECT reference_id,transaction_type
    FROM gateway_wallet_transactions
    WHERE created_at >= '$monitor_since'::timestamptz
      AND reference_type='gateway_task'
    GROUP BY reference_id,transaction_type HAVING count(*)>1
  ) d);")
  [[ $duplicates == 0 ]] || failure_gate_id=${failure_gate_id:-duplicate_execution_or_billing}
  if (( now_epoch - last_wireguard_epoch >= 300 )); then
    check_wireguard || failure_gate_id=${failure_gate_id:-wireguard}
    last_wireguard_epoch=$now_epoch
  fi
  printf '%s,%s,%s,%s,%s,%s,%s,%s,%s\n' \
    "$(date -u '+%Y-%m-%dT%H:%M:%SZ')" "$queue_depth" "$oldest_wait" \
    "$connections" "$max_connections" "$saturated" "$node_max_memory" \
    "$lease_failure" "$lease_lost" >>"$observations"
  [[ -z $failure_gate_id ]] || break
  sleep "$interval"
done

finished_at=$(date -u '+%Y-%m-%dT%H:%M:%SZ')
passed=true
traffic_mode=live
restored=false
if [[ -n $failure_gate_id ]]; then
  passed=false
  pause_and_restore "$failure_gate_id"
  traffic_mode=validation
  restored=true
fi
jq -n \
  --arg runId "$run_id" \
  --arg releaseSha "$release_sha" \
  --arg startedAt "$started_at" \
  --arg finishedAt "$finished_at" \
  --arg failureGateId "$failure_gate_id" \
  --arg trafficMode "$traffic_mode" \
  --argjson samples "$samples" \
  --argjson passed "$passed" \
  --argjson restored "$restored" \
  '{
    schemaVersion:"acceptance-monitor-report/v1",
    runId:$runId,
    releaseSha:$releaseSha,
    startedAt:$startedAt,
    finishedAt:$finishedAt,
    samples:$samples,
    passed:$passed,
    failureGateId:($failureGateId | if length>0 then . else null end),
    trafficMode:$trafficMode,
    previousCapacityRestored:$restored,
    secretSafe:true
  }' >"$monitor_report"
chmod 0600 "$monitor_report" "$observations"
node "$cluster_root/scripts/acceptance/report.mjs" attach-monitor \
  --input "$overall_report" --monitor "$monitor_report" --output "$overall_report" >/dev/null
if [[ $passed == true ]]; then
  echo "production_release_monitor=PASS run_id=$run_id samples=$samples duration_seconds=$AI_GATEWAY_RELEASE_MONITOR_TOTAL_SECONDS traffic_mode=live"
else
  echo "production_release_monitor=FAIL run_id=$run_id gate_id=$failure_gate_id traffic_mode=validation previous_capacity_restored=true" >&2
  exit 1
fi