use std::collections::HashMap;

use livekit_api::access_token::{AccessToken, VideoGrants};
use livekit_api::services::LiveKitApi;
use livekit_api::services::room::{CreateRoomOptions, UpdateParticipantOptions};
use livekit_protocol as proto;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;

use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::lookup::{lookup_space_by_name, require_room_readable};
use crate::share::realtime::RoomEvent;

/// 举手状态。审批通过后 approved，被收回后 revoked（可再次举手）。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum HandState {
    Pending,
    Approved,
    Denied,
    Revoked,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct HandEntry {
    pub state: HandState,
    pub requested_at: i64,
    pub decided_by: Option<i64>,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct HandDto {
    pub user_id: i64,
    pub display_name: String,
    pub avatar_url: Option<String>,
    pub state: HandState,
    pub requested_at: i64,
}

/// 一个参与者窗口在画布里的摆位，坐标按画布宽高归一化（0~1），
/// 这样不同分辨率、不同窗口大小的成员看到的是同一套布局。
#[derive(Debug, Clone, Copy, Serialize, Deserialize, ToSchema)]
pub struct VoiceTileDto {
    pub x: f64,
    pub y: f64,
    /// 归一化宽度。
    pub w: f64,
    /// 归一化高度，和 `w` 一样按画布的可用宽高分别归一，因此窗口可以自由
    /// 拉伸而不是只能等比缩放。旧记录没有这个字段，读出来是 `None`，客户端
    /// 按宽度的 3/4 补一个经典比例。
    #[serde(default)]
    pub h: Option<f64>,
    /// 服务端写入时刻（秒）。请求里被忽略，响应里供客户端判断“谁更新”，
    /// 这样并发拖拽时一个迟到的 GET 不会把刚摆好的窗口拽回旧位置。
    #[serde(default)]
    pub updated_at: i64,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize, ToSchema)]
pub struct VoiceLayoutDto {
    /// key 为参与者用户 id。
    #[serde(default)]
    pub tiles: HashMap<String, VoiceTileDto>,
}

/// 布局写入是**按 key 合并**的补丁而不是整表替换：每个客户端只提交自己刚拖的
/// 那一格，收回的格子里列在 `removed`，两个人同时摆不同窗口互不吞没。
#[derive(Debug, Clone, Default, Deserialize, ToSchema)]
pub struct VoiceLayoutPatchDto {
    /// 需要新增或覆盖的窗口（key = 参与者用户 id）。
    #[serde(default)]
    pub tiles: HashMap<String, VoiceTileDto>,
    /// 需要收回自动网格的窗口 key。
    #[serde(default)]
    pub removed: Vec<String>,
}

/// 本次写入的服务端时刻，客户端用它给本地副本盖戳。
#[derive(Debug, Clone, Copy, Serialize, ToSchema)]
pub struct VoiceLayoutSavedDto {
    pub updated_at: i64,
}

/// 房间成员共享的窗口数量上限，避免布局被当成存储滥用。
const MAX_LAYOUT_TILES: usize = 64;

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceConfigDto {
    pub enabled: bool,
    pub url: String,
    pub recording_enabled: bool,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceTokenDto {
    pub token: String,
    pub url: String,
    pub room_name: String,
    pub expires_in_secs: u64,
    pub can_publish: bool,
    pub is_stage: bool,
    pub hand_state: Option<HandState>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct VoiceTrackDto {
    pub sid: String,
    pub kind: String,
    pub source: String,
    pub muted: bool,
}

/// 房间参与者快照。
///
/// 权威来源是 LiveKit 的 list_participants，Redis 只补主持人静音等本地状态；
/// 这是「谁在这个频道里」的服务端视图，用于侧栏人数与未加入时的预览。
/// 会中界面的发言高亮、静音图标与画面轨道一律直接读浏览器里的 Room，
/// 因为本接口不是事件流，返回值天然是快照。
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceParticipantDto {
    pub user_id: i64,
    pub display_name: String,
    pub avatar_url: Option<String>,
    pub joined_at: i64,
    pub can_publish: bool,
    pub moderator_muted: bool,
    /// 快照值：list_participants 不返回发言状态，因此这里恒为 false；
    /// 实时发言指示由前端从 Room 的 activeSpeakers 读取。
    pub speaking: bool,
    pub tracks: Vec<VoiceTrackDto>,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceRoomStateDto {
    pub room_id: Uuid,
    pub count: u32,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceSpaceStatesDto {
    pub rooms: Vec<VoiceRoomStateDto>,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceHandsDto {
    pub hands: Vec<HandDto>,
    pub can_moderate: bool,
    pub my_state: Option<HandState>,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceModerationDto {
    pub ok: bool,
    pub action: String,
    pub user_id: i64,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceRecordingDto {
    pub egress_id: String,
    pub active: bool,
}

/// Redis 中一位参与者的在场记录。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VoiceMember {
    pub user_id: i64,
    pub display_name: String,
    pub avatar_url: Option<String>,
    pub joined_at: i64,
    pub can_publish: bool,
    #[serde(default)]
    pub moderator_muted: bool,
    #[serde(default)]
    pub tracks: Vec<VoiceTrackDto>,
    pub livekit_sid: Option<String>,
}

pub fn voice_key(room_id: Uuid) -> String {
    format!("chat:voice:{room_id}")
}

pub fn hands_key(room_id: Uuid) -> String {
    format!("chat:voice:{room_id}:hands")
}

/// 布局合并脚本：读写必须在 Redis 里原子完成，否则两个人同时摆不同窗口时，
/// 后写的那次会带着自己读到的旧快照覆盖掉前一次的结果。
const MERGE_LAYOUT_SCRIPT: &str = r"
local key = KEYS[1]
local stamp = tonumber(ARGV[1])
local max_tiles = tonumber(ARGV[3])
local ttl = tonumber(ARGV[4])
local patch = cjson.decode(ARGV[2])
local current = redis.call('GET', key)
local layout = { tiles = {} }
if current then
  local ok, decoded = pcall(cjson.decode, current)
  if ok and type(decoded) == 'table' and type(decoded.tiles) == 'table' then
    layout = decoded
  end
end
local removed = patch.removed
if type(removed) == 'table' then
  for _, id in ipairs(removed) do layout.tiles[id] = nil end
end
local added = patch.tiles
if type(added) == 'table' then
  for id, tile in pairs(added) do
    layout.tiles[id] = {
      x = tonumber(tile.x),
      y = tonumber(tile.y),
      w = tonumber(tile.w),
      -- 旧客户端（或旧记录回写）不带 h：tonumber(nil) 是 nil，这个键就不会写进
      -- JSON，读出来仍是“未指定”，而不是塌成 0。
      h = tonumber(tile.h),
      updated_at = stamp,
    }
  end
end
local count = 0
for _ in pairs(layout.tiles) do count = count + 1 end
if count > max_tiles then return { err = 'too many tiles' } end
redis.call('SET', key, cjson.encode(layout), 'EX', ttl)
return stamp
";

pub fn layout_key(room_id: Uuid) -> String {
    format!("chat:voice:{room_id}:layout")
}

fn now_secs() -> i64 {
    OffsetDateTime::now_utc().unix_timestamp()
}

fn livekit(service: &ChatService) -> Result<LiveKitApi, ChatError> {
    let cfg = &service.settings.livekit;
    if !cfg.enabled {
        return Err(ChatError::BadRequest("livekit is not enabled".to_string()));
    }
    Ok(LiveKitApi::with_api_key(
        cfg.api_url.trim(),
        cfg.api_key.trim(),
        cfg.api_secret.trim(),
    ))
}

/// 语音能力的全局限定：房间必须存在、必须是 Voice/Stage、调用者可读。
pub async fn require_voice_room(
    service: &ChatService,
    room_id: Uuid,
    uid: i64,
) -> Result<models::room::rooms::Model, ChatError> {
    let ctx = require_room_readable(service, room_id, uid).await?;
    if !matches!(
        ctx.room.kind,
        models::room::rooms::RoomKind::Voice | models::room::rooms::RoomKind::Stage
    ) {
        return Err(ChatError::BadRequest(
            "room is not a voice channel".to_string(),
        ));
    }
    Ok(ctx.room)
}

/// 调用者是否能主持（房间角色 >= Moderator，或空间 manage_channels）。
pub async fn can_moderate_voice(
    service: &ChatService,
    room_id: Uuid,
    uid: i64,
) -> Result<bool, ChatError> {
    let ctx = crate::share::lookup::lookup_room_context(service, room_id).await?;
    match &ctx.space {
        Some(space) => crate::share::authz::can_manage_space(service, space, uid).await,
        None => Ok(false),
    }
}

/// Stage 成员是否拥有发言权：房间角色 >= Moderator，或已有 approved 举手。
pub async fn has_publish_grant(
    service: &ChatService,
    room_id: Uuid,
    uid: i64,
) -> Result<bool, ChatError> {
    if can_moderate_voice(service, room_id, uid).await? {
        return Ok(true);
    }
    Ok(matches!(
        read_hand(service, room_id, uid).await?.map(|h| h.state),
        Some(HandState::Approved)
    ))
}

/// 参与者 id 的解析：identity 是 uid 字符串。
pub fn identity_to_uid(identity: &str) -> Option<i64> {
    identity.parse::<i64>().ok()
}

pub async fn display_names(
    service: &ChatService,
    uids: &[i64],
) -> HashMap<i64, (String, Option<String>)> {
    let mut map = HashMap::new();
    let users = match service.audience.load_user_briefs(uids).await {
        Ok(users) => users,
        Err(_) => return map,
    };
    for (uid, user) in users {
        let avatar_url = api::resolve_user_avatar_url(uid, &user.avatar_url);
        let avatar = if avatar_url.trim().is_empty() {
            None
        } else {
            Some(avatar_url)
        };
        let name = if user.display_name.trim().is_empty() {
            user.username
        } else {
            user.display_name
        };
        map.insert(uid, (name, avatar));
    }
    map
}

/// 把 Redis 里的一位成员写回（webhook 与对账共用）。
pub async fn write_member(
    service: &ChatService,
    room_id: Uuid,
    member: &VoiceMember,
) -> Result<(), ChatError> {
    let ttl = service.settings.livekit.presence_ttl_secs.max(60);
    let value = serde_json::to_string(member)?;
    let mut conn = service.participants.redis.connection_clone();
    use cache::RedisConnection;
    let key = voice_key(room_id);
    let field = member.user_id.to_string();
    let result: Result<i64, redis::RedisError> = match &mut conn {
        RedisConnection::Standalone(c) => {
            let written: Result<i64, redis::RedisError> = redis::cmd("HSET")
                .arg(&key)
                .arg(&field)
                .arg(&value)
                .query_async(c)
                .await;
            if written.is_ok() {
                let _: Result<i64, redis::RedisError> =
                    redis::cmd("EXPIRE").arg(&key).arg(ttl).query_async(c).await;
            }
            written
        }
        RedisConnection::Cluster(c) => {
            let written: Result<i64, redis::RedisError> = redis::cmd("HSET")
                .arg(&key)
                .arg(&field)
                .arg(&value)
                .query_async(c)
                .await;
            if written.is_ok() {
                let _: Result<i64, redis::RedisError> =
                    redis::cmd("EXPIRE").arg(&key).arg(ttl).query_async(c).await;
            }
            written
        }
    };
    result.map(|_| ()).map_err(ChatError::from)
}

pub async fn remove_member(
    service: &ChatService,
    room_id: Uuid,
    uid: i64,
) -> Result<(), ChatError> {
    let mut conn = service.participants.redis.connection_clone();
    use cache::RedisConnection;
    let key = voice_key(room_id);
    let field = uid.to_string();
    let _: Result<i64, redis::RedisError> = match &mut conn {
        RedisConnection::Standalone(c) => {
            redis::cmd("HDEL")
                .arg(&key)
                .arg(&field)
                .query_async(c)
                .await
        }
        RedisConnection::Cluster(c) => {
            redis::cmd("HDEL")
                .arg(&key)
                .arg(&field)
                .query_async(c)
                .await
        }
    };
    let hkey = hands_key(room_id);
    let _: Result<i64, redis::RedisError> = match &mut conn {
        RedisConnection::Standalone(c) => {
            redis::cmd("HDEL")
                .arg(&hkey)
                .arg(&field)
                .query_async(c)
                .await
        }
        RedisConnection::Cluster(c) => {
            redis::cmd("HDEL")
                .arg(&hkey)
                .arg(&field)
                .query_async(c)
                .await
        }
    };
    Ok(())
}

pub async fn clear_room(service: &ChatService, room_id: Uuid) -> Result<(), ChatError> {
    let mut conn = service.participants.redis.connection_clone();
    use cache::RedisConnection;
    for key in [voice_key(room_id), hands_key(room_id)] {
        let _: Result<i64, redis::RedisError> = match &mut conn {
            RedisConnection::Standalone(c) => redis::cmd("DEL").arg(&key).query_async(c).await,
            RedisConnection::Cluster(c) => redis::cmd("DEL").arg(&key).query_async(c).await,
        };
    }
    Ok(())
}

/// 读取房间当前在 Redis 中的成员快照。
pub async fn read_members(
    service: &ChatService,
    room_id: Uuid,
) -> Result<Vec<VoiceMember>, ChatError> {
    let mut conn = service.participants.redis.connection_clone();
    use cache::RedisConnection;
    let key = voice_key(room_id);
    let raw: Result<HashMap<String, String>, redis::RedisError> = match &mut conn {
        RedisConnection::Standalone(c) => redis::cmd("HGETALL").arg(&key).query_async(c).await,
        RedisConnection::Cluster(c) => redis::cmd("HGETALL").arg(&key).query_async(c).await,
    };
    let raw = raw.unwrap_or_default();
    let mut members: Vec<VoiceMember> = raw
        .into_values()
        .filter_map(|value| serde_json::from_str::<VoiceMember>(&value).ok())
        .collect();
    members.sort_by_key(|m| m.joined_at);
    Ok(members)
}

pub async fn read_hand(
    service: &ChatService,
    room_id: Uuid,
    uid: i64,
) -> Result<Option<HandEntry>, ChatError> {
    let mut conn = service.participants.redis.connection_clone();
    use cache::RedisConnection;
    let key = hands_key(room_id);
    let field = uid.to_string();
    let raw: Result<Option<String>, redis::RedisError> = match &mut conn {
        RedisConnection::Standalone(c) => {
            redis::cmd("HGET")
                .arg(&key)
                .arg(&field)
                .query_async(c)
                .await
        }
        RedisConnection::Cluster(c) => {
            redis::cmd("HGET")
                .arg(&key)
                .arg(&field)
                .query_async(c)
                .await
        }
    };
    Ok(raw
        .ok()
        .flatten()
        .and_then(|v| serde_json::from_str(&v).ok()))
}

pub async fn read_hands(
    service: &ChatService,
    room_id: Uuid,
) -> Result<HashMap<i64, HandEntry>, ChatError> {
    let mut conn = service.participants.redis.connection_clone();
    use cache::RedisConnection;
    let key = hands_key(room_id);
    let raw: Result<HashMap<String, String>, redis::RedisError> = match &mut conn {
        RedisConnection::Standalone(c) => redis::cmd("HGETALL").arg(&key).query_async(c).await,
        RedisConnection::Cluster(c) => redis::cmd("HGETALL").arg(&key).query_async(c).await,
    };
    let mut map = HashMap::new();
    for (field, value) in raw.unwrap_or_default() {
        if let (Ok(uid), Ok(entry)) = (
            field.parse::<i64>(),
            serde_json::from_str::<HandEntry>(&value),
        ) {
            map.insert(uid, entry);
        }
    }
    Ok(map)
}

pub async fn write_hand(
    service: &ChatService,
    room_id: Uuid,
    uid: i64,
    entry: &HandEntry,
) -> Result<(), ChatError> {
    let ttl = service.settings.livekit.hand_ttl_secs.max(60);
    let value = serde_json::to_string(entry)?;
    let mut conn = service.participants.redis.connection_clone();
    use cache::RedisConnection;
    let key = hands_key(room_id);
    let field = uid.to_string();
    let result: Result<i64, redis::RedisError> = match &mut conn {
        RedisConnection::Standalone(c) => {
            let written: Result<i64, redis::RedisError> = redis::cmd("HSET")
                .arg(&key)
                .arg(&field)
                .arg(&value)
                .query_async(c)
                .await;
            if written.is_ok() {
                let _: Result<i64, redis::RedisError> =
                    redis::cmd("EXPIRE").arg(&key).arg(ttl).query_async(c).await;
            }
            written
        }
        RedisConnection::Cluster(c) => {
            let written: Result<i64, redis::RedisError> = redis::cmd("HSET")
                .arg(&key)
                .arg(&field)
                .arg(&value)
                .query_async(c)
                .await;
            if written.is_ok() {
                let _: Result<i64, redis::RedisError> =
                    redis::cmd("EXPIRE").arg(&key).arg(ttl).query_async(c).await;
            }
            written
        }
    };
    result.map(|_| ()).map_err(ChatError::from)
}

impl ChatService {
    /// 前端搭接参数。
    pub fn voice_config(&self) -> VoiceConfigDto {
        let cfg = &self.settings.livekit;
        VoiceConfigDto {
            enabled: cfg.enabled,
            url: cfg.browser_url().to_string(),
            recording_enabled: cfg.recording_enabled,
        }
    }

    /// 为调用者签发 LiveKit 访问令牌。
    pub async fn voice_token(
        &self,
        caller: i64,
        room_id: Uuid,
    ) -> Result<VoiceTokenDto, ChatError> {
        let cfg = self.settings.livekit.clone();
        if !cfg.enabled {
            return Err(ChatError::BadRequest("livekit is not enabled".to_string()));
        }
        let room = require_voice_room(self, room_id, caller).await?;
        let is_stage = room.kind == models::room::rooms::RoomKind::Stage;
        let can_publish = if is_stage {
            has_publish_grant(self, room_id, caller).await?
        } else {
            true
        };
        let api = livekit(self)?;
        let room_name = room_id.to_string();

        let limit = room
            .user_limit
            .map(|v| v.max(1) as u32)
            .unwrap_or(cfg.default_max_participants);
        if let Ok(participants) = api.room().list_participants(&room_name).await {
            let already_inside = participants
                .iter()
                .any(|p| identity_to_uid(&p.identity) == Some(caller));
            if !already_inside && participants.len() as u32 >= limit {
                return Err(ChatError::Conflict("voice room is full".to_string()));
            }
        }
        let _ = api
            .room()
            .create_room(
                &room_name,
                CreateRoomOptions {
                    empty_timeout: cfg.empty_timeout_secs,
                    departure_timeout: cfg.departure_timeout_secs,
                    max_participants: limit,
                    ..Default::default()
                },
            )
            .await;

        let names = display_names(self, &[caller]).await;
        let (display_name, _) = names
            .get(&caller)
            .cloned()
            .unwrap_or_else(|| (caller.to_string(), None));

        let grants = VideoGrants {
            room_join: true,
            room: room_name.clone(),
            can_subscribe: true,
            can_publish,
            can_publish_data: true,
            can_publish_sources: if can_publish {
                {
                    vec![
                        "camera".to_string(),
                        "microphone".to_string(),
                        "screen_share".to_string(),
                        "screen_share_audio".to_string(),
                    ]
                }
            } else {
                Vec::new()
            },
            ..Default::default()
        };
        let token = AccessToken::with_api_key(cfg.api_key.trim(), cfg.api_secret.trim())
            .with_identity(&caller.to_string())
            .with_name(&display_name)
            .with_ttl(std::time::Duration::from_secs(cfg.token_ttl_secs))
            .with_grants(grants)
            .to_jwt()
            .map_err(|error| ChatError::Internal(format!("livekit token: {error}")))?;

        let hand_state = if is_stage {
            read_hand(self, room_id, caller).await?.map(|h| h.state)
        } else {
            None
        };

        Ok(VoiceTokenDto {
            token,
            url: cfg.browser_url().to_string(),
            room_name,
            expires_in_secs: cfg.token_ttl_secs,
            can_publish,
            is_stage,
            hand_state,
        })
    }

    /// 以 LiveKit 为权威源对账，并回写 Redis。
    pub async fn voice_participants(
        &self,
        caller: i64,
        room_id: Uuid,
    ) -> Result<Vec<VoiceParticipantDto>, ChatError> {
        require_voice_room(self, room_id, caller).await?;
        let api = livekit(self)?;
        let room_name = room_id.to_string();
        let participants = api
            .room()
            .list_participants(&room_name)
            .await
            .map_err(|error| ChatError::Internal(format!("livekit list participants: {error}")))?;

        let uids: Vec<i64> = participants
            .iter()
            .filter_map(|p| identity_to_uid(&p.identity))
            .collect();
        let names = display_names(self, &uids).await;
        let stored: HashMap<i64, VoiceMember> = read_members(self, room_id)
            .await?
            .into_iter()
            .map(|m| (m.user_id, m))
            .collect();

        let mut out = Vec::with_capacity(participants.len());
        for participant in participants {
            let Some(uid) = identity_to_uid(&participant.identity) else {
                continue;
            };
            let can_publish = participant
                .permission
                .as_ref()
                .map(|p| p.can_publish)
                .unwrap_or(true);
            let moderator_muted = stored.get(&uid).map(|m| m.moderator_muted).unwrap_or(false);
            let tracks: Vec<VoiceTrackDto> = participant
                .tracks
                .iter()
                .map(|t| VoiceTrackDto {
                    sid: t.sid.clone(),
                    kind: proto::TrackType::try_from(t.r#type)
                        .map(|k| k.as_str_name().to_string())
                        .unwrap_or_else(|_| "UNKNOWN".to_string()),
                    source: proto::TrackSource::try_from(t.source)
                        .map(|s| s.as_str_name().to_string())
                        .unwrap_or_else(|_| "UNKNOWN".to_string()),
                    muted: t.muted,
                })
                .collect();
            let (display_name, avatar_url) = names
                .get(&uid)
                .cloned()
                .unwrap_or_else(|| (participant.name.clone(), None));
            let member = VoiceMember {
                user_id: uid,
                display_name: display_name.clone(),
                avatar_url: avatar_url.clone(),
                joined_at: participant.joined_at,
                can_publish,
                moderator_muted,
                tracks: tracks.clone(),
                livekit_sid: Some(participant.sid.clone()),
            };
            write_member(self, room_id, &member).await?;
            out.push(VoiceParticipantDto {
                user_id: uid,
                display_name,
                avatar_url,
                joined_at: participant.joined_at,
                can_publish,
                moderator_muted,
                speaking: false,
                tracks,
            });
        }
        out.sort_by_key(|p| p.joined_at);
        Ok(out)
    }

    /// 侧栏用的整空间语音人数（纯 Redis 读，缺房间不报错）。
    pub async fn voice_space_states(
        &self,
        caller: i64,
        space_name: &str,
    ) -> Result<VoiceSpaceStatesDto, ChatError> {
        let space = lookup_space_by_name(self, space_name).await?;
        crate::share::lookup::require_space_member(self, &space, caller).await?;
        let rooms = models::room::rooms::Entity::find()
            .filter(models::room::rooms::Column::SpaceId.eq(space.id()))
            .filter(models::room::rooms::Column::IsActive.eq(true))
            .filter(models::room::rooms::Column::Kind.is_in([
                models::room::rooms::RoomKind::Voice,
                models::room::rooms::RoomKind::Stage,
            ]))
            .all(self.database.read())
            .await?;
        let mut out = Vec::with_capacity(rooms.len());
        for room in rooms {
            let count = read_members(self, room.id).await?.len() as u32;
            out.push(VoiceRoomStateDto {
                room_id: room.id,
                count,
            });
        }
        Ok(VoiceSpaceStatesDto { rooms: out })
    }

    /// 读取房间共享的窗口布局；没有存过就是空布局（前端回落到自动网格）。
    pub async fn voice_layout(
        &self,
        caller: i64,
        room_id: Uuid,
    ) -> Result<VoiceLayoutDto, ChatError> {
        require_voice_room(self, room_id, caller).await?;
        self.voice_layout_raw(room_id).await
    }

    /// 纯读取，不做权限校验：调用方要么刚校验过，要么只用于合并写入。
    async fn voice_layout_raw(&self, room_id: Uuid) -> Result<VoiceLayoutDto, ChatError> {
        let mut conn = self.participants.redis.connection_clone();
        use cache::RedisConnection;
        let key = layout_key(room_id);
        let raw: Result<Option<String>, redis::RedisError> = match &mut conn {
            RedisConnection::Standalone(c) => redis::cmd("GET").arg(&key).query_async(c).await,
            RedisConnection::Cluster(c) => redis::cmd("GET").arg(&key).query_async(c).await,
        };
        let raw = raw.map_err(ChatError::from)?;
        let Some(raw) = raw else {
            return Ok(VoiceLayoutDto::default());
        };
        Ok(serde_json::from_str(&raw).unwrap_or_default())
    }

    /// 按 key 合并布局补丁并广播给房间里的所有人。坐标一律夹到 0~1，越界写入
    /// 不可能发生；返回写入时刻，让客户端能分辨“我发的”和“比我新的”。
    pub async fn save_voice_layout(
        &self,
        caller: i64,
        room_id: Uuid,
        patch: VoiceLayoutPatchDto,
    ) -> Result<VoiceLayoutSavedDto, ChatError> {
        require_voice_room(self, room_id, caller).await?;
        if patch.tiles.is_empty() && patch.removed.is_empty() {
            return Err(ChatError::BadRequest("empty layout patch".to_string()));
        }
        let stamp = now_secs();
        let payload = serde_json::json!({
            "tiles": patch.tiles,
            "removed": patch.removed,
        })
        .to_string();
        let ttl = self.settings.livekit.presence_ttl_secs.max(60);
        let mut conn = self.participants.redis.connection_clone();
        use cache::RedisConnection;
        let key = layout_key(room_id);
        let mut command = redis::cmd("EVAL");
        command
            .arg(MERGE_LAYOUT_SCRIPT)
            .arg(1_u8)
            .arg(&key)
            .arg(stamp)
            .arg(&payload)
            .arg(MAX_LAYOUT_TILES as i64)
            .arg(ttl);
        let written: Result<i64, redis::RedisError> = match &mut conn {
            RedisConnection::Standalone(c) => command.clone().query_async(c).await,
            RedisConnection::Cluster(c) => command.clone().query_async(c).await,
        };
        let written = match written {
            Ok(updated_at) => Ok(updated_at),
            // The script signals a full map with an error reply; turn it back into
            // the same 400 the previous whole-map write produced.
            Err(error) if error.to_string().contains("too many tiles") => Err(
                ChatError::BadRequest(format!("at most {MAX_LAYOUT_TILES} windows can be placed")),
            ),
            Err(error) => Err(ChatError::from(error)),
        }?;
        self.audience
            .realtime
            .publish_room(&RoomEvent::VoiceLayout { room_id })
            .await;
        Ok(VoiceLayoutSavedDto {
            updated_at: written,
        })
    }

    /// 举手。Stage 频道之外、已有发言权都不允许。
    pub async fn raise_hand(&self, caller: i64, room_id: Uuid) -> Result<HandDto, ChatError> {
        let room = require_voice_room(self, room_id, caller).await?;
        if room.kind != models::room::rooms::RoomKind::Stage {
            return Err(ChatError::BadRequest(
                "hands are only available in stage channels".to_string(),
            ));
        }
        if has_publish_grant(self, room_id, caller).await? {
            return Err(ChatError::Conflict("already able to speak".to_string()));
        }
        let entry = HandEntry {
            state: HandState::Pending,
            requested_at: now_secs(),
            decided_by: None,
        };
        write_hand(self, room_id, caller, &entry).await?;
        self.audience
            .realtime
            .publish_room(&RoomEvent::VoiceHands { room_id })
            .await;
        let names = display_names(self, &[caller]).await;
        let (display_name, avatar_url) = names
            .get(&caller)
            .cloned()
            .unwrap_or_else(|| (caller.to_string(), None));
        Ok(HandDto {
            user_id: caller,
            display_name,
            avatar_url,
            state: entry.state,
            requested_at: entry.requested_at,
        })
    }

    pub async fn cancel_hand(&self, caller: i64, room_id: Uuid) -> Result<(), ChatError> {
        require_voice_room(self, room_id, caller).await?;
        let existing = read_hand(self, room_id, caller).await?;
        if matches!(existing.map(|h| h.state), Some(HandState::Approved)) {
            return Err(ChatError::Conflict(
                "an approved speaker must be revoked by a moderator".to_string(),
            ));
        }
        let mut conn = self.participants.redis.connection_clone();
        use cache::RedisConnection;
        let key = hands_key(room_id);
        let field = caller.to_string();
        let _: Result<i64, redis::RedisError> = match &mut conn {
            RedisConnection::Standalone(c) => {
                redis::cmd("HDEL")
                    .arg(&key)
                    .arg(&field)
                    .query_async(c)
                    .await
            }
            RedisConnection::Cluster(c) => {
                redis::cmd("HDEL")
                    .arg(&key)
                    .arg(&field)
                    .query_async(c)
                    .await
            }
        };
        self.audience
            .realtime
            .publish_room(&RoomEvent::VoiceHands { room_id })
            .await;
        Ok(())
    }

    pub async fn list_hands(&self, caller: i64, room_id: Uuid) -> Result<VoiceHandsDto, ChatError> {
        require_voice_room(self, room_id, caller).await?;
        let can_moderate = can_moderate_voice(self, room_id, caller).await?;
        let stored = read_hands(self, room_id).await?;
        let my_state = stored.get(&caller).map(|h| h.state);
        let uids: Vec<i64> = stored.keys().copied().collect();
        let names = display_names(self, &uids).await;
        let mut hands: Vec<HandDto> = stored
            .into_iter()
            .filter(|(_, entry)| can_moderate || matches!(entry.state, HandState::Approved))
            .map(|(uid, entry)| {
                let (display_name, avatar_url) = names
                    .get(&uid)
                    .cloned()
                    .unwrap_or_else(|| (uid.to_string(), None));
                HandDto {
                    user_id: uid,
                    display_name,
                    avatar_url,
                    state: entry.state,
                    requested_at: entry.requested_at,
                }
            })
            .collect();
        hands.sort_by_key(|h| h.requested_at);
        Ok(VoiceHandsDto {
            hands,
            can_moderate,
            my_state,
        })
    }

    async fn decide_hand(
        &self,
        caller: i64,
        room_id: Uuid,
        target: i64,
        approve: bool,
    ) -> Result<VoiceModerationDto, ChatError> {
        require_voice_room(self, room_id, caller).await?;
        if !can_moderate_voice(self, room_id, caller).await? {
            return Err(ChatError::Forbidden);
        }
        let Some(existing) = read_hand(self, room_id, target).await? else {
            return Err(ChatError::NotFound);
        };
        if !matches!(existing.state, HandState::Pending | HandState::Revoked) {
            return Err(ChatError::Conflict("hand is not pending".to_string()));
        }
        let entry = HandEntry {
            state: if approve {
                HandState::Approved
            } else {
                HandState::Denied
            },
            requested_at: existing.requested_at,
            decided_by: Some(caller),
        };
        write_hand(self, room_id, target, &entry).await?;
        if approve {
            let _ = self.apply_publish_permission(room_id, target, true).await;
        }
        self.audience
            .realtime
            .publish_room(&RoomEvent::VoiceHands { room_id })
            .await;
        let action = if approve {
            "speak_granted"
        } else {
            "speak_denied"
        };
        self.audience
            .realtime
            .publish_room(&RoomEvent::VoiceModeration {
                room_id,
                user_id: target,
                action: action.to_string(),
            })
            .await;
        Ok(VoiceModerationDto {
            ok: true,
            action: action.to_string(),
            user_id: target,
        })
    }

    pub async fn approve_hand(
        &self,
        caller: i64,
        room_id: Uuid,
        target: i64,
    ) -> Result<VoiceModerationDto, ChatError> {
        self.decide_hand(caller, room_id, target, true).await
    }

    pub async fn deny_hand(
        &self,
        caller: i64,
        room_id: Uuid,
        target: i64,
    ) -> Result<VoiceModerationDto, ChatError> {
        self.decide_hand(caller, room_id, target, false).await
    }

    /// 收回发言权：翻转 can_publish 并静音已发布的麦克风。
    pub async fn revoke_hand(
        &self,
        caller: i64,
        room_id: Uuid,
        target: i64,
    ) -> Result<VoiceModerationDto, ChatError> {
        require_voice_room(self, room_id, caller).await?;
        if !can_moderate_voice(self, room_id, caller).await? {
            return Err(ChatError::Forbidden);
        }
        let existing = read_hand(self, room_id, target).await?;
        let requested_at = existing
            .as_ref()
            .map(|h| h.requested_at)
            .unwrap_or_else(now_secs);
        write_hand(
            self,
            room_id,
            target,
            &HandEntry {
                state: HandState::Revoked,
                requested_at,
                decided_by: Some(caller),
            },
        )
        .await?;
        let _ = self.apply_publish_permission(room_id, target, false).await;
        self.audience
            .realtime
            .publish_room(&RoomEvent::VoiceHands { room_id })
            .await;
        self.audience
            .realtime
            .publish_room(&RoomEvent::VoiceModeration {
                room_id,
                user_id: target,
                action: "speak_revoked".to_string(),
            })
            .await;
        Ok(VoiceModerationDto {
            ok: true,
            action: "speak_revoked".to_string(),
            user_id: target,
        })
    }

    /// 服务端翻转某人的发布权限（对 LiveKit 与 Redis 同时生效）。
    pub async fn apply_publish_permission(
        &self,
        room_id: Uuid,
        target: i64,
        can_publish: bool,
    ) -> Result<(), ChatError> {
        let api = livekit(self)?;
        let room_name = room_id.to_string();
        let identity = target.to_string();
        let sources: Vec<i32> = if can_publish {
            vec![
                proto::TrackSource::Camera as i32,
                proto::TrackSource::Microphone as i32,
                proto::TrackSource::ScreenShare as i32,
                proto::TrackSource::ScreenShareAudio as i32,
            ]
        } else {
            Vec::new()
        };
        let _ = api
            .room()
            .update_participant(
                &room_name,
                &identity,
                UpdateParticipantOptions {
                    permission: Some(proto::ParticipantPermission {
                        can_subscribe: true,
                        can_publish,
                        can_publish_data: true,
                        can_publish_sources: sources,
                        ..Default::default()
                    }),
                    ..Default::default()
                },
            )
            .await;
        if let Some(mut member) = read_members(self, room_id)
            .await?
            .into_iter()
            .find(|m| m.user_id == target)
        {
            member.can_publish = can_publish;
            write_member(self, room_id, &member).await?;
        }
        Ok(())
    }

    /// 主持人强制静音/解除静音某人的麦克风。
    pub async fn moderate_mute(
        &self,
        caller: i64,
        room_id: Uuid,
        target: i64,
        muted: bool,
    ) -> Result<VoiceModerationDto, ChatError> {
        require_voice_room(self, room_id, caller).await?;
        if !can_moderate_voice(self, room_id, caller).await? {
            return Err(ChatError::Forbidden);
        }
        let api = livekit(self)?;
        let room_name = room_id.to_string();
        let identity = target.to_string();
        if muted {
            let _ = api
                .room()
                .update_participant(
                    &room_name,
                    &identity,
                    UpdateParticipantOptions {
                        permission: Some(proto::ParticipantPermission {
                            can_subscribe: true,
                            can_publish: false,
                            can_publish_data: true,
                            can_publish_sources: Vec::new(),
                            ..Default::default()
                        }),
                        ..Default::default()
                    },
                )
                .await;
            if let Ok(participants) = api.room().list_participants(&room_name).await
                && let Some(participant) = participants.iter().find(|p| p.identity == identity)
            {
                for track in &participant.tracks {
                    if track.source == proto::TrackSource::Microphone as i32 && !track.muted {
                        let _ = api
                            .room()
                            .mute_published_track(&room_name, &identity, &track.sid, true)
                            .await;
                    }
                }
            }
        } else {
            let _ = self.apply_publish_permission(room_id, target, true).await;
        }
        if let Some(mut member) = read_members(self, room_id)
            .await?
            .into_iter()
            .find(|m| m.user_id == target)
        {
            member.moderator_muted = muted;
            write_member(self, room_id, &member).await?;
        }
        let action = if muted { "muted" } else { "unmuted" }.to_string();
        self.audience
            .realtime
            .publish_room(&RoomEvent::VoiceModeration {
                room_id,
                user_id: target,
                action: action.clone(),
            })
            .await;
        Ok(VoiceModerationDto {
            ok: true,
            action,
            user_id: target,
        })
    }

    /// 强制移出通话（可再次加入）。
    pub async fn kick_participant(
        &self,
        caller: i64,
        room_id: Uuid,
        target: i64,
    ) -> Result<VoiceModerationDto, ChatError> {
        require_voice_room(self, room_id, caller).await?;
        if !can_moderate_voice(self, room_id, caller).await? {
            return Err(ChatError::Forbidden);
        }
        if caller == target {
            return Err(ChatError::BadRequest("cannot remove yourself".to_string()));
        }
        let api = livekit(self)?;
        let room_name = room_id.to_string();
        let identity = target.to_string();
        let _ = api.room().remove_participant(&room_name, &identity).await;
        remove_member(self, room_id, target).await?;
        self.audience
            .realtime
            .publish_room(&RoomEvent::VoiceModeration {
                room_id,
                user_id: target,
                action: "removed".to_string(),
            })
            .await;
        Ok(VoiceModerationDto {
            ok: true,
            action: "removed".to_string(),
            user_id: target,
        })
    }

    /// 录制预留：需要 egress 组件，本次只做接口占位。
    pub async fn start_recording(
        &self,
        caller: i64,
        room_id: Uuid,
    ) -> Result<VoiceRecordingDto, ChatError> {
        self.recording_unavailable(caller, room_id).await
    }

    pub async fn stop_recording(
        &self,
        caller: i64,
        room_id: Uuid,
    ) -> Result<VoiceRecordingDto, ChatError> {
        self.recording_unavailable(caller, room_id).await
    }

    async fn recording_unavailable(
        &self,
        caller: i64,
        room_id: Uuid,
    ) -> Result<VoiceRecordingDto, ChatError> {
        require_voice_room(self, room_id, caller).await?;
        if !can_moderate_voice(self, room_id, caller).await? {
            return Err(ChatError::Forbidden);
        }
        if !self.settings.livekit.recording_enabled {
            return Err(ChatError::BadRequest(
                "recording is not enabled".to_string(),
            ));
        }
        Err(ChatError::BadRequest(
            "recording requires the livekit egress component".to_string(),
        ))
    }
}