class_name PoseServer
extends Node

## T-WIN-003 / T-AND-011：WebSocket 关键点接收服务
##
## 契约 contracts/websocket-keypoints.md v0.2.0 §1：
##   端点  ws://<windows-host>:8765/pose
##   子协议 opendance.pose.v1
##   二进制帧 = 239 字节关键点（见 KeypointFrame）
##   文本帧  = JSON 控制消息（hello / welcome / config / stats / ping / pong / bye）
##
## 本端为**服务端**，Android 辅助 App 主动连入。
##
## 说明：Godot 的 WebSocketPeer 不暴露握手请求路径，因此无法校验客户端
## 连接的 path 是否为 /pose；本实现接受任意 path，仅校验子协议。
##
## T-AND-011 对齐要点（契约 v0.2.0 §1.1 / §4.1 / §8）：
##   1. 在 welcome / config 中**声明**静默超时与期望心跳间隔（silentTimeoutMs /
##      heartbeatIntervalMs），且 heartbeatIntervalMs < silentTimeoutMs / 2；
##   2. 处理应用层 ping（协议层 Sec-WebSocket-Ping 在 Godot 里对应用层不可见，
##      不能用来续命），可选回 pong 供对端算 RTT；
##   3. 按 hello.deviceId **去重**：同一设备的重复连接替换旧连接（1000 关闭），
##      设备数按去重后计数，避免"反复重连"被显示成多台设备；
##   4. 日志与状态面板使用契约 §8 的同名连接状态描述每个客户端。

signal listening(port: int)
signal listen_failed(port: int, reason: String)
signal client_connected(client_id: int, device: String)
signal client_disconnected(client_id: int, reason: String)
signal frame_received(frame: KeypointFrame.PoseFrame)
signal control_received(client_id: int, message: Dictionary)
signal stats_updated(stats: Dictionary)
## 同一 deviceId 的旧连接被新连接替换（契约 §4.1）
signal device_replaced(device_id: String)

const KeypointFrame = preload("res://src/net/keypoint_frame.gd")

const DEFAULT_PORT := 8765
const DEFAULT_SUBPROTOCOL := "opendance.pose.v1"

## 静默超时（毫秒）：超过该时长没有收到**任何数据帧或文本帧**即断开。
## 契约 §1.1 要求本端在 welcome/config 中声明该值，客户端据此定心跳节奏。
## 取 15 秒：Android 每秒一次 stats、每帧关键点都算活动，15 秒足以容忍
## 相机暂停 / 画面中无人 / 推理短暂卡顿，又不会让死连接长期占用。
const CLIENT_TIMEOUT_MS := 15000
## 期望心跳间隔（毫秒）——必须 < CLIENT_TIMEOUT_MS / 2（契约 §1.1）
const HEARTBEAT_INTERVAL_MS := 5000
## 每秒向客户端推送一次 config 的间隔（0 表示不主动推送）
const CONFIG_PUSH_INTERVAL_MS := 2000
## degraded 状态的保持时长：这段时间内没有新的丢帧就回到 streaming。
## 做成可覆盖的实例字段，便于测试在不依赖真实等待的前提下核对状态回收。
const DEGRADED_HOLD_MS := 3000
## 契约 §8 的连接状态名（本端为接收端，逐客户端维护）
enum ClientState { IDLE, RESOLVING, CONNECTING, HANDSHAKING, STREAMING, DEGRADED, RECONNECTING, STOPPED }

const STATE_NAMES := {
	ClientState.IDLE: "idle",
	ClientState.RESOLVING: "resolving",
	ClientState.CONNECTING: "connecting",
	ClientState.HANDSHAKING: "handshaking",
	ClientState.STREAMING: "streaming",
	ClientState.DEGRADED: "degraded",
	ClientState.RECONNECTING: "reconnecting",
	ClientState.STOPPED: "stopped",
}

var subprotocol: String = DEFAULT_SUBPROTOCOL
var expected_fps: int = 30
var mirror: bool = false
var upper_body_only: bool = false
var resolution: String = "640x480"

var _tcp := TCPServer.new()
var _clients: Array = []
var _next_client_id: int = 1
var _port: int = 0
var _running: bool = false
var _last_config_push_ms: int = 0
## 最近一次广播出去的配置指纹（用于「配置是否变化」判断）
var _last_config_signature: String = ""
## 见 DEGRADED_HOLD_MS；运行期可调（测试用）
var degraded_hold_ms: int = DEGRADED_HOLD_MS

## 累计统计
var frames_total: int = 0
var frames_dropped: int = 0
var malformed_frames: int = 0
var last_frame: KeypointFrame.PoseFrame = null
var last_frame_at_ms: int = 0


func is_running() -> bool:
	return _running


func port() -> int:
	return _port


func client_count() -> int:
	return _clients.size()


## 已握手完成的连接数
func connected_count() -> int:
	var n := 0
	for entry in _clients:
		if entry["handshaken"]:
			n += 1
	return n


func listen(host: String = "0.0.0.0", listen_port: int = DEFAULT_PORT) -> Error:
	stop()

	var bind_host := host
	if bind_host == "" or bind_host == "0.0.0.0":
		bind_host = "*"

	var err := _tcp.listen(listen_port, bind_host)
	if err != OK:
		listen_failed.emit(listen_port, error_string(err))
		push_error("[PoseServer] 监听 %s:%d 失败：%s" % [host, listen_port, error_string(err)])
		return err

	_port = listen_port
	_running = true
	set_process(true)
	listening.emit(listen_port)
	return OK


func stop() -> void:
	_running = false
	for entry in _clients:
		var peer: WebSocketPeer = entry["peer"]
		entry["state"] = ClientState.STOPPED
		if peer.get_ready_state() == WebSocketPeer.STATE_OPEN:
			peer.close(1001, "server shutdown")
	_tcp.stop()
	_clients.clear()
	set_process(false)


func _ready() -> void:
	set_process(false)


func _process(_delta: float) -> void:
	if not _running:
		return

	_accept_new_connections()
	_poll_clients()
	_maybe_push_config()


func _accept_new_connections() -> void:
	while _tcp.is_connection_available():
		var conn := _tcp.take_connection()
		if conn == null:
			continue
		var peer := WebSocketPeer.new()
		peer.supported_protocols = PackedStringArray([subprotocol])
		var err := peer.accept_stream(conn)
		if err != OK:
			push_warning("[PoseServer] accept_stream 失败：%s" % error_string(err))
			continue
		_clients.append({
			"id": _next_client_id,
			"peer": peer,
			"handshaken": false,
			"device": "",
			"device_id": "",
			"state": ClientState.CONNECTING,
			"selected_protocol": "",
			"connected_at_ms": Time.get_ticks_msec(),
			"handshake_at_ms": 0,
			"first_frame_at_ms": 0,
			"last_seen_ms": Time.get_ticks_msec(),
			"last_frame_id": -1,
			"frames": 0,
			"dropped": 0,
			"degraded_until_ms": 0,
			"close_code": 0,
			"close_reason": "",
			"fps_window_start_ms": Time.get_ticks_msec(),
			"fps_window_frames": 0,
			"fps": 0.0,
		})
		_next_client_id += 1


func _poll_clients() -> void:
	var now := Time.get_ticks_msec()
	var survivors: Array = []

	for entry in _clients:
		var peer: WebSocketPeer = entry["peer"]
		peer.poll()

		var state := peer.get_ready_state()
		if state == WebSocketPeer.STATE_CLOSED:
			entry["close_code"] = peer.get_close_code()
			entry["close_reason"] = peer.get_close_reason()
			entry["state"] = ClientState.STOPPED
			client_disconnected.emit(entry["id"], _close_reason_text(entry))
			continue

		if state == WebSocketPeer.STATE_OPEN and not entry["handshaken"]:
			entry["handshaken"] = true
			entry["handshake_at_ms"] = now
			entry["state"] = ClientState.HANDSHAKING
			var selected := peer.get_selected_protocol()
			entry["selected_protocol"] = selected
			if selected != "" and selected != subprotocol:
				push_warning("[PoseServer] 客户端选择子协议 %s，与期望 %s 不符" % [selected, subprotocol])
			client_connected.emit(entry["id"], entry["device"])

		while peer.get_available_packet_count() > 0:
			_handle_packet(entry, peer)
			entry["last_seen_ms"] = now

		# 降级状态到期后回到 streaming（契约 §8）
		if entry["state"] == ClientState.DEGRADED and now >= int(entry["degraded_until_ms"]):
			entry["state"] = ClientState.STREAMING if int(entry["first_frame_at_ms"]) > 0 else ClientState.HANDSHAKING

		# 超时剔除：契约 §1.1，客户端应按 heartbeatIntervalMs 保活
		if entry["handshaken"] and now - int(entry["last_seen_ms"]) > CLIENT_TIMEOUT_MS:
			var silent_ms := now - int(entry["last_seen_ms"])
			entry["state"] = ClientState.STOPPED
			client_disconnected.emit(
				entry["id"],
				"静默超时 %d ms（阈值 %d ms，期望心跳 %d ms）" % [silent_ms, CLIENT_TIMEOUT_MS, HEARTBEAT_INTERVAL_MS],
			)
			peer.close(1001, "timeout")
			continue

		survivors.append(entry)

	_clients = survivors
	_dedup_devices()


## 契约 §4.1：同一 deviceId 只保留最新连接，旧连接以 1000 关闭（对端不重连）
func _dedup_devices() -> void:
	if _clients.size() < 2:
		return
	var seen := {}
	var keep: Array = []
	# 倒序扫描：后加入的条目更新，先被保留
	for i in range(_clients.size() - 1, -1, -1):
		var entry: Dictionary = _clients[i]
		var device := str(entry["device_id"])
		if device == "" or not entry["handshaken"]:
			keep.append(entry)
			continue
		if seen.has(device):
			entry["state"] = ClientState.STOPPED
			var peer: WebSocketPeer = entry["peer"]
			if peer.get_ready_state() == WebSocketPeer.STATE_OPEN:
				peer.close(1000, "replaced by new connection")
			client_disconnected.emit(entry["id"], "被同一设备的新连接替换（deviceId=%s）" % device)
			device_replaced.emit(device)
			continue
		seen[device] = entry
		keep.append(entry)
	keep.reverse()
	_clients = keep


static func _close_reason_text(entry: Dictionary) -> String:
	var code := int(entry["close_code"])
	var reason := str(entry["close_reason"])
	if code == 0:
		return "连接已关闭（未收到关闭帧）"
	if reason == "":
		return "连接已关闭（code=%d）" % code
	return "连接已关闭（code=%d，%s）" % [code, reason]


func _handle_packet(entry: Dictionary, peer: WebSocketPeer) -> void:
	var packet := peer.get_packet()
	var is_text := peer.was_string_packet()

	if is_text:
		_handle_control(entry, packet.get_string_from_utf8())
		return

	# 二进制帧：契约要求恰好 239 字节
	var problem := KeypointFrame.validate(packet)
	if problem != "":
		malformed_frames += 1
		push_warning("[PoseServer] 客户端 %d 发来非法帧：%s" % [entry["id"], problem])
		return

	var frame := KeypointFrame.decode(packet)
	if frame == null:
		malformed_frames += 1
		return

	_update_frame_stats(entry, frame)
	frames_total += 1
	last_frame = frame
	last_frame_at_ms = Time.get_ticks_msec()
	frame_received.emit(frame)


func _update_frame_stats(entry: Dictionary, frame: KeypointFrame.PoseFrame) -> void:
	if int(entry["first_frame_at_ms"]) == 0:
		entry["first_frame_at_ms"] = Time.get_ticks_msec()
	var previous := int(entry["last_frame_id"])
	if previous >= 0 and frame.frame_id > previous + 1:
		# u32 回绕保护：差距过大视为计数器重置而非丢包
		var gap := frame.frame_id - previous - 1
		if gap < 100000:
			entry["dropped"] = int(entry["dropped"]) + gap
			frames_dropped += gap
			# 契约 §8：仍在连接但出现丢帧 -> degraded（3 秒无新丢帧后自动回到 streaming）
			entry["state"] = ClientState.DEGRADED
			entry["degraded_until_ms"] = Time.get_ticks_msec() + degraded_hold_ms
	entry["last_frame_id"] = frame.frame_id
	entry["frames"] = int(entry["frames"]) + 1
	if entry["state"] != ClientState.DEGRADED:
		entry["state"] = ClientState.STREAMING

	entry["fps_window_frames"] = int(entry["fps_window_frames"]) + 1
	var now := Time.get_ticks_msec()
	var elapsed := now - int(entry["fps_window_start_ms"])
	if elapsed >= 1000:
		entry["fps"] = float(entry["fps_window_frames"]) * 1000.0 / float(elapsed)
		entry["fps_window_frames"] = 0
		entry["fps_window_start_ms"] = now

	var stats := {
		"client_id": entry["id"],
		"fps": entry["fps"],
		"frames": entry["frames"],
		"dropped": entry["dropped"],
		"device": entry["device"],
		"device_id": entry["device_id"],
		"state": state_name(entry["state"]),
		"total_frames": frames_total,
		"total_dropped": frames_dropped,
		"malformed": malformed_frames,
	}
	stats_updated.emit(stats)


func _handle_control(entry: Dictionary, text: String) -> void:
	var parsed: Variant = JSON.parse_string(text)
	if typeof(parsed) != TYPE_DICTIONARY:
		push_warning("[PoseServer] 客户端 %d 发来非法控制帧" % entry["id"])
		return

	var message: Dictionary = parsed
	var type := str(message.get("type", ""))
	control_received.emit(entry["id"], message)

	match type:
		"hello":
			entry["device"] = str(message.get("device", ""))
			entry["device_id"] = str(message.get("deviceId", ""))
			var protocol := str(message.get("protocol", subprotocol))
			if protocol != subprotocol:
				push_warning("[PoseServer] 协议版本不匹配：客户端 %s，本端 %s" % [protocol, subprotocol])
				_send_json(entry, {
					"type": "bye",
					"reason": "protocol_mismatch: 本端要求 %s" % subprotocol,
				})
				var peer: WebSocketPeer = entry["peer"]
				peer.close(1002, "protocol mismatch")
				return
			if message.has("targetFps"):
				expected_fps = int(message["targetFps"])
			entry["state"] = ClientState.STREAMING if int(entry["first_frame_at_ms"]) > 0 else ClientState.HANDSHAKING
			# 顺序固定：先 welcome（对 hello 的应答），再 config（当前镜像/帧率/保活参数）。
			# 也可以等周期 tick 广播，但那样"新连接的第一条消息"可能变成 config，
			# 让只认"握手应答"的实现困惑 —— 契约没规定顺序，但顺序自然更好排障。
			_send_json(entry, welcome_payload())
			_send_json(entry, config_payload())
		"ping":
			# 契约 §3.6：应用层心跳（协议层 ping 对应用层不可见，无法续命）
			_send_json(entry, {
				"type": "pong",
				"sentAtMs": int(message.get("sentAtMs", 0)),
				"serverAtMs": Time.get_ticks_msec(),
			})
		"bye":
			entry["state"] = ClientState.STOPPED
			client_disconnected.emit(entry["id"], str(message.get("reason", "客户端主动断开")))
			var peer2: WebSocketPeer = entry["peer"]
			peer2.close(1000, "bye")
		_:
			pass


## 契约 §3.2：welcome 必须声明保活参数，客户端据此定心跳节奏
func welcome_payload() -> Dictionary:
	return {
		"type": "welcome",
		"protocol": subprotocol,
		"sessionId": _session_id(),
		"expectedFps": expected_fps,
		"silentTimeoutMs": CLIENT_TIMEOUT_MS,
		"heartbeatIntervalMs": HEARTBEAT_INTERVAL_MS,
	}


## 契约 §3.3 的 config 载荷（同时用作「配置是否变化」的指纹来源）
func config_payload() -> Dictionary:
	return {
		"type": "config",
		"mirror": mirror,
		"targetFps": expected_fps,
		"upperBodyOnly": upper_body_only,
		"resolution": resolution,
		# 契约 §3.3：运行中可一并调整保活参数
		"silentTimeoutMs": CLIENT_TIMEOUT_MS,
		"heartbeatIntervalMs": HEARTBEAT_INTERVAL_MS,
	}


## 契约 §3.3 的语义是「**运行中调整**时下发」，不是心跳。
##
## 原实现每 `CONFIG_PUSH_INTERVAL_MS`（2 秒）**无条件**广播一次 config ——
## 对一次 30 分钟的推流是 900 条没有任何信息量的消息，也让「收到 config」
## 无法被当成「配置变了」的信号（发送端想据此做点什么就无从判断）。
## 现在改为：
##   1. 有新客户端完成握手 → 立刻补发一次（它需要知道镜像/目标帧率）；
##   2. 之后仅当**载荷真的变化**才发，并用 `CONFIG_PUSH_INTERVAL_MS` 做去抖
##      （连续拖动滑块时不会每个中间值都发一条）。
func _maybe_push_config() -> void:
	if CONFIG_PUSH_INTERVAL_MS <= 0:
		return
	var now := Time.get_ticks_msec()
	var signature := JSON.stringify(config_payload())

	if signature == _last_config_signature:
		return
	if now - _last_config_push_ms < CONFIG_PUSH_INTERVAL_MS:
		return
	_last_config_signature = signature
	_last_config_push_ms = now
	broadcast_config()


func broadcast_config() -> void:
	var payload := config_payload()
	# 显式广播（UI 改了镜像/帧率等）也要更新指纹，否则下一个 tick 会再发一遍同样的内容
	_last_config_signature = JSON.stringify(payload)
	_last_config_push_ms = Time.get_ticks_msec()
	for entry in _clients:
		if entry["handshaken"]:
			_send_json(entry, payload)


func _send_json(entry: Dictionary, payload: Dictionary) -> void:
	var peer: WebSocketPeer = entry["peer"]
	if peer.get_ready_state() != WebSocketPeer.STATE_OPEN:
		return
	peer.send_text(JSON.stringify(payload))


func _session_id() -> String:
	return "%d-%d" % [Time.get_unix_time_from_system(), _port]


## 状态名（契约 §8，两端同名，便于对照日志）
static func state_name(state: int) -> String:
	return str(STATE_NAMES.get(state, "unknown"))


## 逐客户端信息（界面 / 日志用）
func clients_info() -> Array:
	var out: Array = []
	for entry in _clients:
		out.append({
			"id": entry["id"],
			"device": entry["device"],
			"device_id": entry["device_id"],
			"state": state_name(entry["state"]),
			"protocol": entry["selected_protocol"],
			"frames": entry["frames"],
			"dropped": entry["dropped"],
			"fps": entry["fps"],
			"first_frame_at_ms": entry["first_frame_at_ms"],
		})
	return out


## 去重后的设备数（契约 §4.1：同一 deviceId 只算一台）
func device_count() -> int:
	var ids := {}
	var anonymous := 0
	for entry in _clients:
		if not entry["handshaken"]:
			continue
		var device := str(entry["device_id"])
		if device == "":
			anonymous += 1
		else:
			ids[device] = true
	return ids.size() + anonymous


func snapshot() -> Dictionary:
	return {
		"running": _running,
		"port": _port,
		"clients": _clients.size(),
		"connected": connected_count(),
		"devices": device_count(),
		"silent_timeout_ms": CLIENT_TIMEOUT_MS,
		"heartbeat_interval_ms": HEARTBEAT_INTERVAL_MS,
		"frames_total": frames_total,
		"frames_dropped": frames_dropped,
		"malformed_frames": malformed_frames,
	}