package cluster

import (
	"encoding/base64"
	"fmt"
	"net"
	"net/http"
	"os"
	"strings"
)

// NewJoinScriptHandler serves a self-contained shell script that bootstraps
// a new node to join the cluster. The script follows a strict Day-1 protocol:
//
//	Phase 0: Preflight       — token, identity, targeted service stops
//	Phase 1: Binaries        — install binaries and unit files, no state write
//	Phase 2: PKI/Identity    — CA cert, service cert, TLS verification
//	Phase 3: Connectivity    — DNS, /etc/hosts, libnss, journald, config.json
//	Phase 4: Download        — binaries, workflows, apt sources
//	Phase 5: etcd join       — ghost cleanup, member add, yaml, start, verify
//	Phase 6: node-agent      — start after etcd verified, wait for :11000
//	Phase 7: Cluster join    — join RPC, DNS A record
//	Phase 8: MinIO hold      — enforce objectstore topology contract
//
// Repair mode (etcd WAL damage):
//
//	curl -sfL https://<gateway>:8443/join -k | sudo bash -s -- --token <TOKEN> --repair-etcd
//
// Usage (fresh Day-1 join):
//
//	curl -sfL https://<gateway>:8443/join -k | sudo bash -s -- --token <join-token>
func NewJoinScriptHandler(controllerAddr string, gatewayPort int) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodGet {
			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
			return
		}

		// Read CA certificate to embed in the script.
		caCert, err := os.ReadFile("/var/lib/globular/pki/ca.crt")
		if err != nil {
			http.Error(w, "CA certificate not found", http.StatusInternalServerError)
			return
		}
		caB64 := base64.StdEncoding.EncodeToString(caCert)

		// Read ScyllaDB GPG key if available on this node.
		scyllaGpgB64 := ""
		if gpgKey, err := os.ReadFile("/etc/apt/keyrings/scylladb.gpg"); err == nil {
			scyllaGpgB64 = base64.StdEncoding.EncodeToString(gpgKey)
		}

		// Read ScyllaDB apt source line if configured.
		scyllaAptSource := ""
		if src, err := os.ReadFile("/etc/apt/sources.list.d/scylla.list"); err == nil {
			scyllaAptSource = strings.TrimSpace(string(src))
		}

		// Determine the gateway address the client connected to.
		gatewayHost := r.Host
		if gatewayHost == "" {
			gatewayHost = r.URL.Host
		}

		reqHost, reqPort, err := net.SplitHostPort(gatewayHost)
		if err != nil {
			reqHost = gatewayHost
			reqPort = fmt.Sprintf("%d", gatewayPort)
		}
		if reqHost == "" || reqHost == "127.0.0.1" || reqHost == "::1" || reqHost == "localhost" {
			// Never embed a loopback address — the joining node is remote.
			if h, _, splitErr := net.SplitHostPort(r.RemoteAddr); splitErr == nil {
				reqHost = h
			}
		}

		gatewayAddr := net.JoinHostPort(reqHost, reqPort)

		ctrlAddr := controllerAddr
		ctrlHost, _, _ := net.SplitHostPort(ctrlAddr)
		if ctrlAddr == "" || ctrlHost == "127.0.0.1" || ctrlHost == "localhost" || ctrlHost == "::1" {
			ctrlAddr = fmt.Sprintf("%s:12000", reqHost)
		}

		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
		fmt.Fprintf(w, joinScript, caB64, gatewayAddr, ctrlAddr, scyllaGpgB64, scyllaAptSource)
	})
}

// joinScript is the shell script template served to joining nodes.
// Format args (all positional to avoid ambiguity with shell ${} syntax):
//
//	%[1]s = base64 CA cert
//	%[2]s = gateway address (host:port)
//	%[3]s = controller address (host:port)
//	%[4]s = base64 ScyllaDB GPG key (may be empty)
//	%[5]s = ScyllaDB apt source line (may be empty)
//
// IMPORTANT: any literal %% in this string becomes % in the output script.
// Use %% wherever the shell script needs a literal % (e.g. date format strings).
const joinScript = `#!/usr/bin/env bash
set -euo pipefail

# ============================================================================
# Globular Day-1 Node Join Script
# Generated by the bootstrap gateway — do not edit manually.
#
# Phases:
#   0 — Preflight:   token required, identity detection, targeted service stops
#   1 — Directories: create layout, globular user (no state wipe)
#   2 — PKI/Identity: CA cert, service cert, TLS smoke-test
#   3 — Connectivity: DNS, /etc/hosts MinIO fallback, libnss, journald
#   4 — Download:    binaries, workflows, ScyllaDB apt source
#   5 — etcd join:   ghost cleanup → member add → etcd.yaml → start → verify
#   6 — node-agent:  start AFTER etcd healthy, wait for :11000
#   7 — Cluster join: join RPC, DNS A record
#   8 — MinIO hold:  enforce objectstore topology contract
#
# After Phase 6 the controller drives all further convergence:
#   admitted → infra_preparing → etcd_ready → xds_ready →
#   envoy_ready → storage_joining → workload_ready
#
# Repair mode (etcd WAL corrupt or permanently removed from cluster):
#   ... | sudo bash -s -- --token <TOKEN> --repair-etcd
#   This mode REQUIRES explicit operator intent. It backs up and wipes
#   /var/lib/globular/etcd before rejoining. Never wiped without this flag.
# ============================================================================

# ── Embedded cluster parameters ─────────────────────────────────────────────
CA_B64="%[1]s"
GATEWAY="%[2]s"
CONTROLLER="%[3]s"
SCYLLA_GPG_B64="%[4]s"
SCYLLA_APT_SOURCE="%[5]s"

# ── Runtime defaults ─────────────────────────────────────────────────────────
JOIN_TOKEN=""
REPAIR_ETCD=0
INSTALL_DIR="/usr/lib/globular/bin"
STATE_DIR="/var/lib/globular"
SYSTEMD_DIR="/etc/systemd/system"

# ── Argument parsing ─────────────────────────────────────────────────────────
while [[ $# -gt 0 ]]; do
  case "$1" in
    --token)       JOIN_TOKEN="$2"; shift 2 ;;
    --controller)  CONTROLLER="$2"; shift 2 ;;
    --repair-etcd) REPAIR_ETCD=1; shift ;;
    *) echo "Unknown argument: $1" >&2; exit 1 ;;
  esac
done

[[ -n "${JOIN_TOKEN}" ]] || {
  echo "Error: --token is required" >&2
  echo "Usage: curl -sfL https://<gateway>:8443/join -k | sudo bash -s -- --token <TOKEN>" >&2
  exit 1
}

# ── Helpers ──────────────────────────────────────────────────────────────────
die()       { echo "  FAIL: $*" >&2; exit 1; }
log_ok()    { echo "  OK:   $*"; }
log_info()  { echo "  ...   $*"; }
log_warn()  { echo "  WARN: $*" >&2; }
log_phase() { echo ""; echo "══════════════════════════════════════════════════════"; echo "PHASE $*"; echo "══════════════════════════════════════════════════════"; echo ""; }

[[ "$(id -u)" -eq 0 ]] || die "This script must be run as root (sudo)"

BOOTSTRAP_HOST=$(echo "${GATEWAY}" | cut -d: -f1)
NODE_HOSTNAME=$(hostname)
NODE_IP=$(hostname -I | awk '{print $1}')
PKI_DIR="${STATE_DIR}/pki/issued/services"
ETCD_CACERT="${STATE_DIR}/pki/ca.crt"
ETCD_CERT="${PKI_DIR}/service.crt"
ETCD_KEY="${PKI_DIR}/service.key"

echo ""
echo "╔══════════════════════════════════════════════════════════════════╗"
echo "║          GLOBULAR DAY-1 NODE JOIN                               ║"
echo "╚══════════════════════════════════════════════════════════════════╝"
echo ""
echo "  Controller: ${CONTROLLER}"
echo "  Gateway:    ${GATEWAY}"
echo "  Bootstrap:  ${BOOTSTRAP_HOST}"
echo "  This node:  ${NODE_HOSTNAME} (${NODE_IP})"
if [[ ${REPAIR_ETCD} -eq 1 ]]; then
  echo "  Mode:       REPAIR-ETCD (WAL backup + wipe authorized)"
else
  echo "  Mode:       fresh Day-1 join"
fi
echo ""

# ============================================================================
log_phase "0 — Preflight"
# ============================================================================
# Stop ONLY the services that must be reset for this join sequence.
# Do not wildcard-stop all globular-* services.

log_info "[0.1] Stopping globular-etcd and globular-node-agent if running..."
for unit in globular-etcd.service globular-node-agent.service; do
  if systemctl is-active --quiet "${unit}" 2>/dev/null; then
    log_info "Stopping ${unit}..."
    systemctl stop "${unit}" 2>/dev/null || true
  fi
done

# Repair mode: back up etcd WAL then wipe — ONLY with explicit --repair-etcd.
# Without this flag the script FAILS if etcd data exists, forcing an
# explicit operator decision rather than silently destroying cluster state.
if [[ ${REPAIR_ETCD} -eq 1 ]]; then
  log_info "[0.2] Repair mode: backing up etcd data before wipe..."
  if [[ -d "${STATE_DIR}/etcd" ]]; then
    ETCD_BACKUP="${STATE_DIR}/etcd.bak.$(date +%%Y%%m%%dT%%H%%M%%S)"
    cp -a "${STATE_DIR}/etcd" "${ETCD_BACKUP}" \
      || die "etcd backup to ${ETCD_BACKUP} failed — refusing to wipe without a successful backup"
    log_ok "etcd backup written: ${ETCD_BACKUP}"
    rm -rf "${STATE_DIR}/etcd"
    log_ok "etcd data dir wiped (backup at ${ETCD_BACKUP})"
  else
    log_info "no etcd data dir found — nothing to wipe"
  fi
elif [[ -d "${STATE_DIR}/etcd" ]] && [[ -n "$(ls -A "${STATE_DIR}/etcd" 2>/dev/null)" ]]; then
  die "etcd data directory exists at ${STATE_DIR}/etcd and --repair-etcd was not given.
  If this node had a failed or broken etcd join, use --repair-etcd to
  authorize a WAL backup and wipe. Never wipe without explicit operator intent."
fi

log_ok "Preflight complete"

# ============================================================================
log_phase "1 — Directories and globular user"
# ============================================================================
# Create directory layout and the globular system user.
# No etcd.yaml, no PKI material, no node identity is written here.
# State wipe already handled in Phase 0 (repair only) or by clean-node.sh.

log_info "[1.1] Creating directories..."
mkdir -p "${INSTALL_DIR}" \
         "${STATE_DIR}/pki/issued/services" \
         "${STATE_DIR}/services" \
         "${STATE_DIR}/node_agent" \
         "${STATE_DIR}/config" \
         "${STATE_DIR}/workflows" \
         "${STATE_DIR}/keys"

if ! id -u globular &>/dev/null; then
  useradd --system --no-create-home --home-dir "${STATE_DIR}" --shell /usr/sbin/nologin globular
fi
log_ok "Directories and user ready"

# ============================================================================
log_phase "2 — PKI / Identity"
# ============================================================================

log_info "[2.1] Installing cluster CA certificate..."
echo "${CA_B64}" | base64 -d > "${STATE_DIR}/pki/ca.crt"
cp "${STATE_DIR}/pki/ca.crt" "${STATE_DIR}/pki/ca.pem"
chmod 644 "${STATE_DIR}/pki/ca.crt" "${STATE_DIR}/pki/ca.pem"

# Install into system trust store so curl and Go TLS work with the cluster CA.
cp "${STATE_DIR}/pki/ca.crt" /usr/local/share/ca-certificates/globular-ca.crt
update-ca-certificates >/dev/null 2>&1 || true
log_ok "CA certificate installed and trusted"

log_info "[2.2] Generating service certificate..."
openssl genrsa -out "${PKI_DIR}/service.key" 2048 2>/dev/null

cat > /tmp/globular-san.cnf <<SANEOF
[req]
req_extensions     = v3_req
distinguished_name = req_dn
prompt             = no
[req_dn]
CN = ${NODE_HOSTNAME}
O  = globular.internal
[v3_req]
subjectAltName = DNS:${NODE_HOSTNAME},DNS:${NODE_HOSTNAME}.globular.internal,DNS:localhost,IP:${NODE_IP},IP:127.0.0.1
SANEOF

openssl req -new -key "${PKI_DIR}/service.key" \
  -config /tmp/globular-san.cnf \
  -out /tmp/globular-join.csr 2>/dev/null

CSR_B64=$(base64 -w0 /tmp/globular-join.csr)
curl -sfL --cacert "${STATE_DIR}/pki/ca.crt" \
  "https://${GATEWAY}/sign_ca_certificate?csr=${CSR_B64}" \
  -o "${PKI_DIR}/service.crt"
rm -f /tmp/globular-join.csr /tmp/globular-san.cnf

[[ -s "${PKI_DIR}/service.crt" ]] \
  || die "Certificate signing failed — controller returned empty certificate"

chmod 600 "${PKI_DIR}/service.key"
chmod 644 "${PKI_DIR}/service.crt"
log_ok "Service certificate issued for ${NODE_HOSTNAME} (${NODE_IP})"

log_info "[2.3] Verifying TLS access to bootstrap gateway..."
if curl -sf --cacert "${STATE_DIR}/pki/ca.crt" -o /dev/null \
  "https://${GATEWAY}/get_ca_certificate" 2>/dev/null; then
  log_ok "Bootstrap gateway reachable via TLS"
else
  log_warn "Gateway TLS check failed — DNS may not be set yet; continuing"
fi

# ============================================================================
log_phase "3 — Connectivity"
# ============================================================================

log_info "[3.1] Configuring DNS resolver (*.globular.internal → bootstrap)..."
mkdir -p /etc/systemd/resolved.conf.d
cat > /etc/systemd/resolved.conf.d/globular.conf <<DNSCONF
[Resolve]
DNS=${BOOTSTRAP_HOST}
Domains=~globular.internal
DNSCONF
systemctl restart systemd-resolved 2>/dev/null || true
log_ok "DNS resolver → ${BOOTSTRAP_HOST} for globular.internal"

# -- MinIO /etc/hosts fallback (objectstore topology contract) ---------------
#
# This node is NOT automatically a MinIO/objectstore member on Day-1.
# Objectstore membership is governed solely by ObjectStoreDesiredState in
# etcd, written by the controller after explicit 'globular objectstore
# apply-topology'. No topology mutation happens here.
#
# The bootstrap node is written as the ONLY /etc/hosts MinIO entry so that
# artifact fetches resolve before DNS propagates. This node's own IP is
# never written as a MinIO endpoint by the join script.
log_info "[3.2] Adding MinIO /etc/hosts fallback (bootstrap objectstore only)..."
sed -i '/minio\.globular\.internal/d' /etc/hosts
echo "${BOOTSTRAP_HOST} minio.globular.internal  # MinIO bootstrap fallback — ObjectStoreDesiredState governs topology" >> /etc/hosts
log_ok "minio.globular.internal → ${BOOTSTRAP_HOST} (bootstrap; topology contract governs membership)"

log_info "[3.3] Ensuring libnss-resolve (systemd-resolved bridge for Go binaries)..."
if ! dpkg -l libnss-resolve 2>/dev/null | grep -q '^ii'; then
  apt-get install -y libnss-resolve >/dev/null 2>&1 || true
fi
if grep -q 'mdns4_minimal.*NOTFOUND.*return' /etc/nsswitch.conf; then
  sed -i 's/^hosts:.*/hosts:          files resolve dns myhostname/' /etc/nsswitch.conf
fi
log_ok "libnss-resolve configured"

log_info "[3.4] Configuring log rate limits (prevent syslog/journal disk fill)..."
mkdir -p /etc/rsyslog.d
cat > /etc/rsyslog.d/50-globular-rate-limit.conf <<'RSYSLOG'
$SystemLogRateLimitInterval 10
$SystemLogRateLimitBurst 500
RSYSLOG
systemctl restart rsyslog 2>/dev/null || true

mkdir -p /etc/systemd/journald.conf.d
cat > /etc/systemd/journald.conf.d/50-globular.conf <<'JOURNALD'
[Journal]
SystemMaxUse=2G
SystemKeepFree=10G
MaxRetentionSec=7day
JOURNALD
systemctl restart systemd-journald 2>/dev/null || true
log_ok "rsyslog rate-limited, journald capped at 2GB/7days"

log_info "[3.5] Writing node-agent seed config.json..."
cat > "${STATE_DIR}/config.json" <<CFGJSON
{
  "Address": "${NODE_IP}",
  "AlternateDomains": [],
  "Domain": "globular.internal",
  "EnablePeerUpserts": false,
  "EtcdClientPort": "2379",
  "EtcdConfigPath": "${STATE_DIR}/config/etcd.yaml",
  "EtcdDataDir": "${STATE_DIR}/etcd",
  "EtcdEnabled": true,
  "EtcdMode": "existing",
  "EtcdName": "${NODE_HOSTNAME}",
  "EtcdPeerPort": "2380",
  "MutateHostsFile": false,
  "MutateResolvConf": false,
  "Name": "${NODE_HOSTNAME}",
  "Peers": [],
  "PortHTTP": 8080,
  "PortHTTPS": 8443,
  "Protocol": "https",
  "Services": {}
}
CFGJSON

chown -R globular:globular "${STATE_DIR}"
chown globular:globular "${STATE_DIR}/keys"
log_ok "Connectivity configured"

# ============================================================================
log_phase "4 — Download Binaries and Workflows"
# ============================================================================

log_info "[4.1] Downloading core binaries..."
for BIN in node_agent_server globularcli etcd etcdctl; do
  curl -sfL --cacert "${STATE_DIR}/pki/ca.crt" \
    "https://${GATEWAY}/join/bin/${BIN}" \
    -o "${INSTALL_DIR}/${BIN}"
  chmod +x "${INSTALL_DIR}/${BIN}"
  log_info "${BIN} installed"
done
ln -sf "${INSTALL_DIR}/globularcli" /usr/local/bin/globular
ln -sf "${INSTALL_DIR}/etcdctl" /usr/local/bin/etcdctl
log_ok "Binaries installed"

log_info "[4.2] Downloading workflow definitions..."
for WF in node.join.yaml node.bootstrap.yaml node.repair.yaml \
           release.apply.package.yaml release.apply.infrastructure.yaml \
           release.remove.package.yaml cluster.reconcile.yaml day0.bootstrap.yaml; do
  if curl -sfL --cacert "${STATE_DIR}/pki/ca.crt" \
    "https://${GATEWAY}/join/workflows/${WF}" \
    -o "${STATE_DIR}/workflows/${WF}" 2>/dev/null; then
    log_info "${WF} installed"
  fi
done
cp "${STATE_DIR}/workflows/node.join.yaml" /tmp/node.join.yaml 2>/dev/null || true
chown -R globular:globular "${STATE_DIR}/workflows" 2>/dev/null || true
log_ok "Workflows installed"

log_info "[4.3] Preparing ScyllaDB apt source (workflow owns the actual install)..."
if [[ -n "${SCYLLA_GPG_B64}" && -n "${SCYLLA_APT_SOURCE}" ]]; then
  if [[ ! -f /etc/apt/keyrings/scylladb.gpg ]]; then
    mkdir -p /etc/apt/keyrings
    echo "${SCYLLA_GPG_B64}" | base64 -d > /etc/apt/keyrings/scylladb.gpg
    chmod 644 /etc/apt/keyrings/scylladb.gpg
  fi
  echo "${SCYLLA_APT_SOURCE}" > /etc/apt/sources.list.d/scylladb.list
  log_ok "ScyllaDB apt source configured (install driven by controller workflow)"
else
  log_info "ScyllaDB GPG not provided — workflow will configure apt source on demand"
fi

# ============================================================================
log_phase "5 — etcd Join"
# ============================================================================
# Contract (always enforced):
#   1. Remove ghost/stale member for this node's IP BEFORE member add.
#   2. member add exactly once.
#   3. etcd.yaml written with initial-cluster-state: existing — NEVER "new".
#   4. etcd must reach healthy state or this script FAILS — no "continuing".
#   5. Verify this node appears as a named member after join.

export ETCDCTL_API=3
BOOTSTRAP_ETCD="https://${BOOTSTRAP_HOST}:2379"
ETCD_NAME=$(echo "${NODE_HOSTNAME}" | sed 's/[^a-zA-Z0-9_-]/-/g; s/^-//; s/-$//')
[[ -n "${ETCD_NAME}" ]] || ETCD_NAME="node"

mkdir -p "${STATE_DIR}/etcd"
chown -R globular:globular "${STATE_DIR}/config" "${STATE_DIR}/etcd"

# [5.1] Remove ghost member (unnamed or matching this IP) BEFORE member add.
# A previous failed join may have left a peer URL registered without a name
# (ghost member). This would block a new member add with the same peer URL.
log_info "[5.1] Checking for ghost/stale etcd member at ${NODE_IP}:2380..."
GHOST_ID=$("${INSTALL_DIR}/etcdctl" \
  --endpoints="${BOOTSTRAP_ETCD}" \
  --cacert="${ETCD_CACERT}" \
  --cert="${ETCD_CERT}" \
  --key="${ETCD_KEY}" \
  member list 2>/dev/null \
  | grep "${NODE_IP}" | awk -F',' '{print $1}' | xargs || true)

if [[ -n "${GHOST_ID}" ]]; then
  log_info "Removing stale member ${GHOST_ID} for ${NODE_IP}..."
  "${INSTALL_DIR}/etcdctl" \
    --endpoints="${BOOTSTRAP_ETCD}" \
    --cacert="${ETCD_CACERT}" \
    --cert="${ETCD_CERT}" \
    --key="${ETCD_KEY}" \
    member remove "${GHOST_ID}" \
    || log_warn "member remove returned non-zero (may already be gone)"
  log_ok "Stale member ${GHOST_ID} removed"
else
  log_info "No stale member found for ${NODE_IP}"
fi

# [5.2] Read current member list (named members only, excluding ghosts).
# Build initial-cluster string BEFORE adding this node.
log_info "[5.2] Reading current named etcd members..."
MEMBER_LIST=$("${INSTALL_DIR}/etcdctl" \
  --endpoints="${BOOTSTRAP_ETCD}" \
  --cacert="${ETCD_CACERT}" \
  --cert="${ETCD_CERT}" \
  --key="${ETCD_KEY}" \
  member list 2>/dev/null || true)

INITIAL_CLUSTER=""
while IFS=',' read -r _id _status name peerURLs _rest; do
  name=$(echo "${name}" | xargs)
  peerURLs=$(echo "${peerURLs}" | xargs)
  # Skip unnamed (ghost) members — they corrupt the initial-cluster string.
  [[ -z "${name}" ]] && continue
  # Normalise any loopback peer URLs to the bootstrap host's real IP.
  peerURLs=$(echo "${peerURLs}" | sed -E "s#https?://(127\\.0\\.0\\.1|localhost|\\[::1\\])#https://${BOOTSTRAP_HOST}#g")
  [[ -n "${INITIAL_CLUSTER}" ]] && INITIAL_CLUSTER="${INITIAL_CLUSTER},"
  INITIAL_CLUSTER="${INITIAL_CLUSTER}${name}=${peerURLs}"
done <<< "${MEMBER_LIST}"

# [5.2.5] Acquire join-in-progress lock to protect the new member from being
# evicted by the controller's removeStaleMembers loop. There is a structural
# race between this script's "etcdctl member add" (which immediately changes
# raft membership) and the joining node's node-agent heartbeat (which puts
# the node into /globular/nodes/). Between those two events the controller's
# desired set does NOT contain this hostname, so the next stale-member sweep
# classifies the freshly-added member as stale and removes it.
#
# Live incident: 2026-05-14, globule-nuc evicted within ~100ms of member-add.
#
# The lock is a leased key at /globular/etcd_joins/<sanitized_hostname>. The
# controller's removeStaleMembers reads this prefix and skips eviction for
# any member whose name appears in it. 300s lease covers the longest healthy
# Day-1 join we've observed; if the script dies mid-way, the TTL self-cleans
# the lock so a retry isn't blocked.
#
# Failure to acquire the lock is logged but not fatal — the original race is
# narrow enough that most joins still succeed without it, and refusing to
# proceed would convert a flaky-controller incident into a hard outage.
log_info "[5.2.5] Acquiring etcd-join lock for ${ETCD_NAME}..."
LOCK_LEASE_ID=$("${INSTALL_DIR}/etcdctl" \
  --endpoints="${BOOTSTRAP_ETCD}" \
  --cacert="${ETCD_CACERT}" \
  --cert="${ETCD_CERT}" \
  --key="${ETCD_KEY}" \
  lease grant 300 2>/dev/null | awk '/granted/ {print $2}')

if [[ -n "${LOCK_LEASE_ID}" ]]; then
  if "${INSTALL_DIR}/etcdctl" \
    --endpoints="${BOOTSTRAP_ETCD}" \
    --cacert="${ETCD_CACERT}" \
    --cert="${ETCD_CERT}" \
    --key="${ETCD_KEY}" \
    put "/globular/etcd_joins/${ETCD_NAME}" \
    "{\"peer_url\":\"https://${NODE_IP}:2380\",\"started_unix\":$(date -u +%%s)}" \
    --lease="${LOCK_LEASE_ID}" >/dev/null 2>&1; then
    log_ok "join-lock acquired (lease=${LOCK_LEASE_ID}, ttl=300s)"
  else
    log_warn "failed to put join-lock — controller may evict during join"
  fi
else
  log_warn "lease grant failed — proceeding without join-lock; if eviction occurs, retry with --repair-etcd"
fi

# [5.3] Add this node as a new etcd member.
log_info "[5.3] Adding ${ETCD_NAME} (https://${NODE_IP}:2380) to etcd cluster..."
ADD_OUT=$("${INSTALL_DIR}/etcdctl" \
  --endpoints="${BOOTSTRAP_ETCD}" \
  --cacert="${ETCD_CACERT}" \
  --cert="${ETCD_CERT}" \
  --key="${ETCD_KEY}" \
  member add "${ETCD_NAME}" --peer-urls="https://${NODE_IP}:2380" 2>&1) || {
  if echo "${ADD_OUT}" | grep -qi "already exists\|Peer URLs already exists"; then
    log_info "peer URL already registered — continuing (previous partial join)"
  else
    die "etcd member add failed: ${ADD_OUT}"
  fi
}
log_ok "etcd member add: ${ETCD_NAME}"

# Append this node to the initial-cluster string.
[[ -n "${INITIAL_CLUSTER}" ]] && INITIAL_CLUSTER="${INITIAL_CLUSTER},"
INITIAL_CLUSTER="${INITIAL_CLUSTER}${ETCD_NAME}=https://${NODE_IP}:2380"

# [5.4] Write etcd.yaml.
# INVARIANT: initial-cluster-state MUST be "existing" for Day-1 joins.
# "new" is only for the founding single-node bootstrap (Day-0). Writing "new"
# on a joining node would fork the etcd cluster and destroy quorum.
log_info "[5.4] Writing etcd.yaml (initial-cluster-state: existing)..."
cat > "${STATE_DIR}/config/etcd.yaml" <<ETCDCFG
name: "${ETCD_NAME}"
data-dir: "${STATE_DIR}/etcd"

listen-client-urls: "https://${NODE_IP}:2379"
advertise-client-urls: "https://${NODE_IP}:2379"
listen-peer-urls: "https://${NODE_IP}:2380"
initial-advertise-peer-urls: "https://${NODE_IP}:2380"

initial-cluster: "${INITIAL_CLUSTER}"
initial-cluster-state: "existing"
initial-cluster-token: "globular-etcd-cluster"

quota-backend-bytes: 8589934592
auto-compaction-mode: "periodic"
auto-compaction-retention: "1h"
snapshot-count: 10000

# Client TLS — no client-cert-auth so etcdctl works without requiring a client cert.
client-transport-security:
  cert-file: "${ETCD_CERT}"
  key-file: "${ETCD_KEY}"
  client-cert-auth: false
  auto-tls: false

# Peer TLS — mutual auth between etcd members requires trusted-ca-file.
peer-transport-security:
  cert-file: "${ETCD_CERT}"
  key-file: "${ETCD_KEY}"
  client-cert-auth: false
  trusted-ca-file: "${ETCD_CACERT}"
  auto-tls: false

logger: "zap"
ETCDCFG
chown globular:globular "${STATE_DIR}/config/etcd.yaml"

# Write etcd endpoint list for other services that dial it directly.
printf '%%s\n' "https://${BOOTSTRAP_HOST}:2379" "https://${NODE_IP}:2379" \
  > "${STATE_DIR}/config/etcd_endpoints"
chown globular:globular "${STATE_DIR}/config/etcd_endpoints"

# [5.5] Install and start globular-etcd.service.
log_info "[5.5] Installing globular-etcd.service..."
cat > "${SYSTEMD_DIR}/globular-etcd.service" <<ETCDUNIT
[Unit]
Description=Globular etcd
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=globular
Group=globular
ExecStartPre=+/usr/bin/mkdir -p ${STATE_DIR}/etcd
ExecStartPre=+/usr/bin/chown globular:globular ${STATE_DIR}/etcd
ExecStartPre=+/usr/bin/chmod 0750 ${STATE_DIR}/etcd
ExecStartPre=+/bin/sh -c 'chown globular:globular ${STATE_DIR}/pki/issued/services/service.key ${STATE_DIR}/pki/issued/services/service.crt 2>/dev/null || true'
ExecStart=${INSTALL_DIR}/etcd --config-file ${STATE_DIR}/config/etcd.yaml
Restart=on-failure
RestartSec=2
LimitNOFILE=524288

[Install]
WantedBy=multi-user.target
ETCDUNIT

systemctl daemon-reload
systemctl enable globular-etcd.service
systemctl start globular-etcd.service

# [5.6] Wait for etcd to become healthy.
# This is a HARD requirement — node-agent must not start until etcd is up.
# If etcd does not become healthy, print the journal and fail loudly.
log_info "[5.6] Waiting for etcd to become healthy (max 60s)..."
ETCD_OK=0
for i in $(seq 1 30); do
  if "${INSTALL_DIR}/etcdctl" \
    --endpoints="https://${NODE_IP}:2379" \
    --cacert="${ETCD_CACERT}" \
    --cert="${ETCD_CERT}" \
    --key="${ETCD_KEY}" \
    endpoint health 2>/dev/null | grep -q "is healthy"; then
    ETCD_OK=1
    break
  fi
  sleep 2
done

if [[ ${ETCD_OK} -ne 1 ]]; then
  echo "" >&2
  echo "  FAIL: etcd did not become healthy after 60s" >&2
  echo "  Last journal entries:" >&2
  journalctl -u globular-etcd.service --no-pager -n 40 >&2 || true
  echo "" >&2
  die "etcd join failed. If the WAL is damaged, retry with --repair-etcd."
fi
log_ok "etcd healthy"

# [5.7] Verify this node appears as a named member — queried from the BOOTSTRAP
# cluster, not local etcd. A locally-healthy etcd that forked its own cluster
# would pass a local check; only the bootstrap view proves actual cluster join.
log_info "[5.7] Verifying named member presence..."
MEMBER_CHECK=$("${INSTALL_DIR}/etcdctl" \
  --endpoints="${BOOTSTRAP_ETCD}" \
  --cacert="${ETCD_CACERT}" \
  --cert="${ETCD_CERT}" \
  --key="${ETCD_KEY}" \
  member list 2>/dev/null || true)

if echo "${MEMBER_CHECK}" | grep -q "${ETCD_NAME}"; then
  log_ok "etcd member '${ETCD_NAME}' confirmed in cluster"
else
  die "etcd join failed: '${ETCD_NAME}' not visible from bootstrap cluster (${BOOTSTRAP_ETCD}). Local etcd may have forked its own cluster — use --repair-etcd to retry."
fi

# ============================================================================
log_phase "6 — node-agent"
# ============================================================================
# Start the node-agent ONLY after etcd is verified healthy.
# The node-agent drives all subsequent package installs and service starts.

log_info "[6.1] Installing globular-node-agent.service..."
cat > "${SYSTEMD_DIR}/globular-node-agent.service" <<UNIT
[Unit]
Description=Globular node_agent
After=network-online.target globular-etcd.service
Wants=network-online.target
Requires=globular-etcd.service

[Service]
Type=simple
User=root
Group=root
WorkingDirectory=${STATE_DIR}/node_agent
ExecStartPre=/bin/sh -c 'mkdir -p ${STATE_DIR}/node_agent'
ExecStartPre=/bin/sh -c 'for i in \$(seq 1 60); do [ -f ${PKI_DIR}/service.crt ] && exit 0; sleep 1; done; echo "TLS cert not ready" >&2; exit 1'
ExecStart=${INSTALL_DIR}/node_agent_server
Restart=on-failure
RestartSec=2
LimitNOFILE=524288

[Install]
WantedBy=multi-user.target
UNIT

systemctl daemon-reload
systemctl enable globular-node-agent.service
systemctl start globular-node-agent.service

log_info "[6.2] Waiting for node-agent to listen on :11000..."
AGENT_UP=0
for i in $(seq 1 30); do
  if ss -tlnp 2>/dev/null | grep -q ":11000 "; then
    AGENT_UP=1
    break
  fi
  sleep 1
done
if [[ ${AGENT_UP} -ne 1 ]]; then
  log_warn "node-agent not yet listening on :11000 — controller will retry join RPC"
fi
log_ok "node-agent running"

# ============================================================================
log_phase "7 — Cluster Join RPC and DNS"
# ============================================================================

log_info "[7.1] Sending join request to controller..."
globular cluster join \
  --join-token "${JOIN_TOKEN}" \
  --controller "${CONTROLLER}" \
  --node "${NODE_IP}:11000" \
  --insecure
log_ok "Join request sent"

log_info "[7.2] Registering DNS A record..."
globular dns a set "${NODE_HOSTNAME}.globular.internal." "${NODE_IP}" \
  --dns "${BOOTSTRAP_HOST}:10007" 2>/dev/null \
  && log_ok "DNS ${NODE_HOSTNAME}.globular.internal → ${NODE_IP}" \
  || log_warn "DNS registration failed — controller will reconcile"

# ============================================================================
log_phase "8 — MinIO / Objectstore Hold Enforcement"
# ============================================================================
# The join script MUST NOT start globular-minio.service.
# Objectstore topology is exclusively governed by ObjectStoreDesiredState in
# etcd via 'globular objectstore apply-topology'.
# The controller and node-agent start MinIO ONLY when this node is admitted
# as an objectstore member by an approved TopologyTransition.

log_info "[8.1] Enforcing MinIO hold (objectstore topology contract)..."
if systemctl is-active --quiet globular-minio.service 2>/dev/null; then
  log_warn "globular-minio.service is active — stopping (topology contract violation)"
  systemctl stop globular-minio.service 2>/dev/null || true
fi
if systemctl is-enabled --quiet globular-minio.service 2>/dev/null; then
  systemctl disable globular-minio.service 2>/dev/null || true
  log_info "globular-minio.service disabled (controller drives objectstore topology)"
fi
log_ok "MinIO held — controller owns objectstore topology"

# ============================================================================
echo ""
echo "╔══════════════════════════════════════════════════════════════════╗"
echo "║          NODE JOINED SUCCESSFULLY                               ║"
echo "╚══════════════════════════════════════════════════════════════════╝"
echo ""
echo "  Identity:     ${NODE_HOSTNAME} (${NODE_IP})"
echo "  etcd:         healthy, member '${ETCD_NAME}'"
echo "  node-agent:   running"
echo "  MinIO:        held (not a topology member)"
echo ""
echo "  The controller now drives bootstrap convergence:"
echo "    admitted → infra_preparing → etcd_ready →"
echo "    xds_ready → envoy_ready → storage_joining → workload_ready"
echo ""
echo "  Monitor: globular cluster nodes list"
echo ""
`
