use std::collections::{HashMap, HashSet};

use models::dir::dir_overwrite_group;
use models::dir::dir_overwrite_users;
use models::dir::dirs::{self, DirKind};
use models::space::space_groups;
use models::space::space_member_groups;
use models::space::space_members::{self, MemberStatus};
use sea_orm::{
    ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, Set,
    TransactionTrait,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;

use crate::errors::SpaceError;
use crate::services::service::SpaceService;
use crate::services::spaces::UserDto;
use crate::share::authz::{
    MANAGE_CHANNELS, MANAGE_OVERWRITES, effective_permissions, require_perm,
};
use crate::share::lookup::lookup_space_by_name;

pub const MAX_DIR_NODE_NAME: usize = 255;
pub const PERM_DIR_VISIBLE: i64 = 1 << 0;

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct DirNodeDto {
    pub id: Uuid,
    pub name: String,
    pub kind: DirKind,
    pub parent_dir_id: Option<Uuid>,
    pub color: Option<String>,
    pub position: i64,
    pub icon: Option<String>,
    pub topic: Option<String>,
    pub purpose: Option<String>,
    pub is_archived: bool,
    pub archived_at: Option<OffsetDateTime>,
    pub created_at: OffsetDateTime,
    pub updated_at: OffsetDateTime,
    pub created_by_id: i64,
    #[schema(no_recursion)]
    pub children: Vec<DirNodeDto>,
}

impl From<dirs::Model> for DirNodeDto {
    fn from(m: dirs::Model) -> Self {
        Self {
            id: m.id,
            name: m.name,
            kind: m.kind,
            parent_dir_id: m.parent_dir_id,
            color: m.color,
            position: m.position,
            icon: m.icon,
            topic: m.topic,
            purpose: m.purpose,
            is_archived: m.is_archived,
            archived_at: m.archived_at,
            created_at: m.created_at,
            updated_at: m.updated_at,
            created_by_id: m.created_by_uid,
            children: Vec::new(),
        }
    }
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateDirCmd {
    pub name: String,
    #[serde(default)]
    pub parent_dir_id: Option<Uuid>,
    #[serde(default)]
    pub position: Option<i64>,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateDirCmd {
    pub name: String,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct MoveDirCmd {
    #[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
    pub parent_dir_id: Option<Option<Uuid>>,
    #[serde(default)]
    pub position: Option<i64>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct RepairDirResp {
    pub scanned: i64,
    pub fixed: i64,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct GroupOverwriteDto {
    pub dir_id: Uuid,
    pub group_id: Uuid,
    pub allow_mask: i64,
    pub deny_mask: i64,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct UserOverwriteDto {
    pub dir_id: Uuid,
    pub user_id: i64,
    pub user: Option<UserDto>,
    pub allow_mask: i64,
    pub deny_mask: i64,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct SetOverwriteCmd {
    pub allow_mask: i64,
    pub deny_mask: i64,
}

#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct DirPath {
    pub space_name: String,
    pub dir_id: Uuid,
}

fn resolve_node_visible(
    caller_uid: i64,
    my_group_ids: &HashSet<Uuid>,
    group_ows: &[dir_overwrite_group::Model],
    user_ows: &[dir_overwrite_users::Model],
) -> bool {
    if group_ows.is_empty() && user_ows.is_empty() {
        return true;
    }

    if let Some(uo) = user_ows.iter().find(|u| u.user_id == caller_uid) {
        if (uo.deny_mask & PERM_DIR_VISIBLE) != 0 {
            return false;
        }
        if (uo.allow_mask & PERM_DIR_VISIBLE) != 0 {
            return true;
        }
    }

    let matching: Vec<&dir_overwrite_group::Model> = group_ows
        .iter()
        .filter(|g| my_group_ids.contains(&g.group_id))
        .collect();
    if matching
        .iter()
        .any(|g| (g.deny_mask & PERM_DIR_VISIBLE) != 0)
    {
        return false;
    }
    if matching
        .iter()
        .any(|g| (g.allow_mask & PERM_DIR_VISIBLE) != 0)
    {
        return true;
    }
    true
}

impl SpaceService {
    async fn dir_in_space(&self, space_id: Uuid, dir_id: Uuid) -> Result<dirs::Model, SpaceError> {
        dirs::Entity::find_by_id(dir_id)
            .filter(dirs::Column::SpaceId.eq(space_id))
            .one(self.db().read())
            .await?
            .ok_or(SpaceError::NotFound)
    }

    #[tracing::instrument(name = "space.create_dir_node", skip_all, fields(caller = %caller, space = %space_name))]
    pub async fn create_dir_node(
        &self,
        caller: i64,
        space_name: &str,
        cmd: CreateDirCmd,
    ) -> Result<DirNodeDto, SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_perm(self, &space, caller, MANAGE_CHANNELS).await?;

        let name = cmd.name.trim();
        if name.is_empty() || name.chars().count() > MAX_DIR_NODE_NAME {
            return Err(SpaceError::BadRequest(
                "dir name is empty or too long".to_string(),
            ));
        }

        if let Some(p) = cmd.parent_dir_id {
            let parent_node = self.dir_in_space(space.id, p).await?;
            if parent_node.kind != DirKind::Dir {
                return Err(SpaceError::BadRequest(
                    "parent node must be a dir".to_string(),
                ));
            }
        }

        let position = match cmd.position {
            Some(pos) => {
                if pos < 0 {
                    return Err(SpaceError::BadRequest(
                        "position must be non-negative".to_string(),
                    ));
                }
                pos
            }
            None => {
                let max_node = dirs::Entity::find()
                    .filter(dirs::Column::SpaceId.eq(space.id))
                    .filter(dirs::Column::ParentDirId.eq(cmd.parent_dir_id))
                    .order_by_desc(dirs::Column::Position)
                    .one(self.db().read())
                    .await?;
                max_node.map(|n| n.position + 1).unwrap_or(0)
            }
        };

        let now = OffsetDateTime::now_utc();
        let row = dirs::ActiveModel {
            id: Set(Uuid::now_v7()),
            space_id: Set(Some(space.id)),
            name: Set(name.to_string()),
            kind: Set(DirKind::Dir),
            parent_dir_id: Set(cmd.parent_dir_id),
            color: Set(None),
            position: Set(position),
            icon: Set(None),
            topic: Set(None),
            purpose: Set(None),
            is_archived: Set(false),
            archived_at: Set(None),
            archived_by: Set(None),
            created_at: Set(now),
            updated_at: Set(now),
            created_by_uid: Set(caller),
        };

        let inserted = row.insert(self.db().write()).await?;
        Ok(DirNodeDto::from(inserted))
    }

    pub async fn get_space_dir_tree(
        &self,
        caller: i64,
        space_name: &str,
        parent_id: Option<Uuid>,
    ) -> Result<Vec<DirNodeDto>, SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        let perms = effective_permissions(&self.state, &space, caller).await?;

        let can_manage = perms.has(MANAGE_CHANNELS);

        if parent_id.is_none() {
            let total_nodes = dirs::Entity::find()
                .filter(dirs::Column::SpaceId.eq(space.id))
                .count(self.db().read())
                .await? as u64;
            let max_nodes = self.config().space.max_dir_nodes;
            if total_nodes > max_nodes {
                return Err(SpaceError::BadRequest(format!(
                    "directory tree has {total_nodes} nodes, exceeds max_dir_nodes {max_nodes}; query with parent_id to load one level"
                )));
            }
        }

        let mut sel = dirs::Entity::find()
            .filter(dirs::Column::SpaceId.eq(space.id))
            .order_by_asc(dirs::Column::Position);
        if let Some(pid) = parent_id {
            sel = sel.filter(dirs::Column::ParentDirId.eq(Some(pid)));
        }
        let nodes = sel.all(self.db().read()).await?;

        if nodes.is_empty() {
            return Ok(Vec::new());
        }

        let node_ids: Vec<Uuid> = nodes.iter().map(|n| n.id).collect();

        let group_ows = dir_overwrite_group::Entity::find()
            .filter(dir_overwrite_group::Column::DirId.is_in(node_ids.clone()))
            .all(self.db().read())
            .await?;

        let user_ows = dir_overwrite_users::Entity::find()
            .filter(dir_overwrite_users::Column::DirId.is_in(node_ids.clone()))
            .all(self.db().read())
            .await?;

        let my_member_groups = space_member_groups::Entity::find()
            .filter(space_member_groups::Column::SpaceId.eq(space.id))
            .filter(space_member_groups::Column::UserId.eq(caller))
            .all(self.db().read())
            .await?;
        let my_group_ids: HashSet<Uuid> =
            my_member_groups.into_iter().map(|mg| mg.group_id).collect();

        let mut group_ow_map: HashMap<Uuid, Vec<dir_overwrite_group::Model>> = HashMap::new();
        for go in group_ows {
            group_ow_map.entry(go.dir_id).or_default().push(go);
        }

        let mut user_ow_map: HashMap<Uuid, Vec<dir_overwrite_users::Model>> = HashMap::new();
        for uo in user_ows {
            user_ow_map.entry(uo.dir_id).or_default().push(uo);
        }

        let mut direct_visible: HashMap<Uuid, bool> = HashMap::new();
        let mut parent_map: HashMap<Uuid, Option<Uuid>> = HashMap::new();
        let mut node_map: HashMap<Uuid, dirs::Model> = HashMap::new();

        for n in &nodes {
            parent_map.insert(n.id, n.parent_dir_id);
            node_map.insert(n.id, n.clone());

            if can_manage {
                direct_visible.insert(n.id, true);
                continue;
            }

            let group_ows = group_ow_map.get(&n.id).map(Vec::as_slice).unwrap_or(&[]);
            let user_ows = user_ow_map.get(&n.id).map(Vec::as_slice).unwrap_or(&[]);
            direct_visible.insert(
                n.id,
                resolve_node_visible(caller, &my_group_ids, group_ows, user_ows),
            );
        }

        fn is_truly_visible(
            id: Uuid,
            parent_map: &HashMap<Uuid, Option<Uuid>>,
            direct_visible: &HashMap<Uuid, bool>,
            memo: &mut HashMap<Uuid, bool>,
            visiting: &mut HashSet<Uuid>,
        ) -> bool {
            if let Some(&vis) = memo.get(&id) {
                return vis;
            }
            if visiting.contains(&id) {
                return false;
            }
            if !direct_visible.get(&id).copied().unwrap_or(false) {
                memo.insert(id, false);
                return false;
            }
            visiting.insert(id);
            let parent_opt = parent_map.get(&id).copied().flatten();
            let parent_vis = match parent_opt {
                Some(pid) if parent_map.contains_key(&pid) => {
                    is_truly_visible(pid, parent_map, direct_visible, memo, visiting)
                }
                _ => true,
            };
            visiting.remove(&id);
            memo.insert(id, parent_vis);
            parent_vis
        }

        let mut memo = HashMap::new();
        let mut truly_visible_set = HashSet::new();
        for &nid in &node_ids {
            let mut visiting = HashSet::new();
            if is_truly_visible(nid, &parent_map, &direct_visible, &mut memo, &mut visiting) {
                truly_visible_set.insert(nid);
            }
        }

        let mut children_map: HashMap<Option<Uuid>, Vec<DirNodeDto>> = HashMap::new();
        for n in nodes {
            if !truly_visible_set.contains(&n.id) {
                continue;
            }
            let parent_id = match n.parent_dir_id {
                Some(p) if truly_visible_set.contains(&p) => Some(p),
                _ => None,
            };
            children_map
                .entry(parent_id)
                .or_default()
                .push(DirNodeDto::from(n));
        }

        fn assemble_tree(
            parent_id: Option<Uuid>,
            children_map: &mut HashMap<Option<Uuid>, Vec<DirNodeDto>>,
        ) -> Vec<DirNodeDto> {
            let mut list = children_map.remove(&parent_id).unwrap_or_default();
            list.sort_by(|a, b| {
                a.position
                    .cmp(&b.position)
                    .then_with(|| a.created_at.cmp(&b.created_at))
            });
            for item in &mut list {
                item.children = assemble_tree(Some(item.id), children_map);
            }
            list
        }

        let forest = assemble_tree(None, &mut children_map);
        Ok(forest)
    }

    #[tracing::instrument(name = "space.update_dir_node", skip_all, fields(caller = %caller, space = %space_name, dir_id = %dir_id))]
    pub async fn update_dir_node(
        &self,
        caller: i64,
        space_name: &str,
        dir_id: Uuid,
        cmd: UpdateDirCmd,
    ) -> Result<DirNodeDto, SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_perm(self, &space, caller, MANAGE_CHANNELS).await?;

        let node: dirs::Model = self.dir_in_space(space.id, dir_id).await?;

        let name = cmd.name.trim();
        if name.is_empty() || name.chars().count() > MAX_DIR_NODE_NAME {
            return Err(SpaceError::BadRequest(
                "dir name is empty or too long".to_string(),
            ));
        }

        let now = OffsetDateTime::now_utc();
        let mut am: dirs::ActiveModel = node.into();
        am.name = Set(name.to_string());
        am.updated_at = Set(now);
        let updated = am.update(self.db().write()).await?;

        Ok(DirNodeDto::from(updated))
    }

    pub async fn move_dir_node(
        &self,
        caller: i64,
        space_name: &str,
        dir_id: Uuid,
        cmd: MoveDirCmd,
    ) -> Result<DirNodeDto, SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_perm(self, &space, caller, MANAGE_CHANNELS).await?;

        let node = self.dir_in_space(space.id, dir_id).await?;

        let new_parent_id = match cmd.parent_dir_id {
            None => node.parent_dir_id,
            Some(None) => None,
            Some(Some(p)) => {
                if p == dir_id {
                    return Err(SpaceError::BadRequest(
                        "cannot set parent to self".to_string(),
                    ));
                }
                let parent_node = self.dir_in_space(space.id, p).await?;
                if parent_node.kind != DirKind::Dir {
                    return Err(SpaceError::BadRequest(
                        "parent node must be a dir".to_string(),
                    ));
                }

                let mut curr = parent_node.parent_dir_id;
                while let Some(curr_p) = curr {
                    if curr_p == dir_id {
                        return Err(SpaceError::BadRequest(
                            "cannot move node into its own descendant".to_string(),
                        ));
                    }
                    let next = dirs::Entity::find_by_id(curr_p)
                        .filter(dirs::Column::SpaceId.eq(space.id))
                        .one(self.db().read())
                        .await?;
                    curr = next.and_then(|n| n.parent_dir_id);
                }

                Some(p)
            }
        };

        let new_position = match cmd.position {
            Some(pos) => {
                if pos < 0 {
                    return Err(SpaceError::BadRequest(
                        "position must be non-negative".to_string(),
                    ));
                }
                pos
            }
            None if new_parent_id != node.parent_dir_id => {
                let max_node = dirs::Entity::find()
                    .filter(dirs::Column::SpaceId.eq(space.id))
                    .filter(dirs::Column::ParentDirId.eq(new_parent_id))
                    .order_by_desc(dirs::Column::Position)
                    .one(self.db().read())
                    .await?;
                max_node.map(|n| n.position + 1).unwrap_or(0)
            }
            None => node.position,
        };

        let now = OffsetDateTime::now_utc();
        let mut am: dirs::ActiveModel = node.into();
        am.parent_dir_id = Set(new_parent_id);
        am.position = Set(new_position);
        am.updated_at = Set(now);
        let updated = am.update(self.db().write()).await?;

        Ok(DirNodeDto::from(updated))
    }

    pub async fn repair_dir_parents(
        &self,
        caller: i64,
        space_name: &str,
    ) -> Result<RepairDirResp, SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        if space.owner_id != caller {
            return Err(SpaceError::Forbidden);
        }

        let nodes = dirs::Entity::find()
            .filter(dirs::Column::SpaceId.eq(space.id))
            .all(self.db().read())
            .await?;

        let scanned = nodes.len() as i64;
        let mut parent_map: HashMap<Uuid, Option<Uuid>> = HashMap::new();
        for n in &nodes {
            parent_map.insert(n.id, n.parent_dir_id);
        }

        let mut fixed_nodes: HashSet<Uuid> = HashSet::new();

        for n in &nodes {
            if let Some(p) = n.parent_dir_id
                && (p == n.id || !parent_map.contains_key(&p))
            {
                fixed_nodes.insert(n.id);
                parent_map.insert(n.id, None);
            }
        }

        let mut state: HashMap<Uuid, u8> = HashMap::new();
        let keys: Vec<Uuid> = parent_map.keys().copied().collect();

        for nid in keys {
            if state.get(&nid).copied().unwrap_or(0) != 0 {
                continue;
            }

            let mut curr = Some(nid);
            let mut path = Vec::new();

            while let Some(c) = curr {
                let st = state.get(&c).copied().unwrap_or(0);
                if st == 1 {
                    fixed_nodes.insert(c);
                    parent_map.insert(c, None);
                    break;
                } else if st == 2 {
                    break;
                }

                state.insert(c, 1);
                path.push(c);
                curr = parent_map.get(&c).copied().flatten();
            }

            for p in path {
                state.insert(p, 2);
            }
        }

        let now = OffsetDateTime::now_utc();
        let fixed = fixed_nodes.len() as i64;

        if !fixed_nodes.is_empty() {
            self.db()
                .primary()
                .transaction::<_, (), SpaceError>(|txn| {
                    Box::pin(async move {
                        for fid in fixed_nodes {
                            let node = dirs::Entity::find_by_id(fid).one(txn).await?;
                            if let Some(n) = node {
                                let mut am: dirs::ActiveModel = n.into();
                                am.parent_dir_id = Set(None);
                                am.updated_at = Set(now);
                                am.update(txn).await?;
                            }
                        }
                        Ok(())
                    })
                })
                .await?;
        }

        Ok(RepairDirResp { scanned, fixed })
    }

    pub async fn list_dir_group_overwrites(
        &self,
        caller: i64,
        space_name: &str,
        dir_id: Uuid,
    ) -> Result<Vec<GroupOverwriteDto>, SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_perm(self, &space, caller, MANAGE_OVERWRITES).await?;
        self.dir_in_space(space.id, dir_id).await?;

        let rows = dir_overwrite_group::Entity::find()
            .filter(dir_overwrite_group::Column::DirId.eq(dir_id))
            .all(self.db().read())
            .await?;

        Ok(rows
            .into_iter()
            .map(|r| GroupOverwriteDto {
                dir_id: r.dir_id,
                group_id: r.group_id,
                allow_mask: r.allow_mask,
                deny_mask: r.deny_mask,
            })
            .collect())
    }

    #[tracing::instrument(name = "space.set_dir_group_overwrite", skip_all, fields(caller = %caller, space = %space_name, dir_id = %dir_id, group_id = %group_id))]
    pub async fn set_dir_group_overwrite(
        &self,
        caller: i64,
        space_name: &str,
        dir_id: Uuid,
        group_id: Uuid,
        cmd: SetOverwriteCmd,
    ) -> Result<GroupOverwriteDto, SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_perm(self, &space, caller, MANAGE_OVERWRITES).await?;
        self.dir_in_space(space.id, dir_id).await?;

        space_groups::Entity::find_by_id(group_id)
            .filter(space_groups::Column::SpaceId.eq(space.id))
            .one(self.db().read())
            .await?
            .ok_or(SpaceError::NotFound)?;

        if cmd.allow_mask < 0 || cmd.deny_mask < 0 {
            return Err(SpaceError::BadRequest(
                "masks must be non-negative".to_string(),
            ));
        }

        let existing = dir_overwrite_group::Entity::find_by_id((dir_id, group_id))
            .one(self.db().read())
            .await?;

        if let Some(row) = existing {
            let mut am: dir_overwrite_group::ActiveModel = row.into();
            am.allow_mask = Set(cmd.allow_mask);
            am.deny_mask = Set(cmd.deny_mask);
            let updated = am.update(self.db().write()).await?;
            Ok(GroupOverwriteDto {
                dir_id: updated.dir_id,
                group_id: updated.group_id,
                allow_mask: updated.allow_mask,
                deny_mask: updated.deny_mask,
            })
        } else {
            let am = dir_overwrite_group::ActiveModel {
                dir_id: Set(dir_id),
                group_id: Set(group_id),
                allow_mask: Set(cmd.allow_mask),
                deny_mask: Set(cmd.deny_mask),
            };
            let inserted = am.insert(self.db().write()).await?;
            Ok(GroupOverwriteDto {
                dir_id: inserted.dir_id,
                group_id: inserted.group_id,
                allow_mask: inserted.allow_mask,
                deny_mask: inserted.deny_mask,
            })
        }
    }

    #[tracing::instrument(name = "space.remove_dir_group_overwrite", skip_all, fields(caller = %caller, space = %space_name, dir_id = %dir_id, group_id = %group_id))]
    pub async fn remove_dir_group_overwrite(
        &self,
        caller: i64,
        space_name: &str,
        dir_id: Uuid,
        group_id: Uuid,
    ) -> Result<(), SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_perm(self, &space, caller, MANAGE_OVERWRITES).await?;
        self.dir_in_space(space.id, dir_id).await?;

        let res = dir_overwrite_group::Entity::delete_many()
            .filter(dir_overwrite_group::Column::DirId.eq(dir_id))
            .filter(dir_overwrite_group::Column::GroupId.eq(group_id))
            .exec(self.db().write())
            .await?;

        if res.rows_affected == 0 {
            return Err(SpaceError::NotFound);
        }
        Ok(())
    }

    pub async fn list_dir_user_overwrites(
        &self,
        caller: i64,
        space_name: &str,
        dir_id: Uuid,
    ) -> Result<Vec<UserOverwriteDto>, SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_perm(self, &space, caller, MANAGE_OVERWRITES).await?;
        self.dir_in_space(space.id, dir_id).await?;

        let rows = dir_overwrite_users::Entity::find()
            .filter(dir_overwrite_users::Column::DirId.eq(dir_id))
            .all(self.db().read())
            .await?;

        let user_ids: Vec<i64> = rows.iter().map(|r| r.user_id).collect();
        let user_map: HashMap<i64, UserDto> = self.load_user_briefs(&user_ids).await?;

        Ok(rows
            .into_iter()
            .map(|r| {
                let user_brief = user_map.get(&r.user_id).cloned();
                UserOverwriteDto {
                    dir_id: r.dir_id,
                    user_id: r.user_id,
                    user: user_brief,
                    allow_mask: r.allow_mask,
                    deny_mask: r.deny_mask,
                }
            })
            .collect())
    }

    #[tracing::instrument(name = "space.set_dir_user_overwrite", skip_all, fields(caller = %caller, space = %space_name, dir_id = %dir_id, user_id = %user_id))]
    pub async fn set_dir_user_overwrite(
        &self,
        caller: i64,
        space_name: &str,
        dir_id: Uuid,
        user_id: i64,
        cmd: SetOverwriteCmd,
    ) -> Result<UserOverwriteDto, SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_perm(self, &space, caller, MANAGE_OVERWRITES).await?;
        self.dir_in_space(space.id, dir_id).await?;

        if space.owner_id != user_id {
            space_members::Entity::find_by_id((space.id, user_id))
                .one(self.db().read())
                .await?
                .filter(|m| m.status == MemberStatus::Active)
                .ok_or(SpaceError::BadRequest(
                    "target user is not an active member of this space".to_string(),
                ))?;
        }

        if cmd.allow_mask < 0 || cmd.deny_mask < 0 {
            return Err(SpaceError::BadRequest(
                "masks must be non-negative".to_string(),
            ));
        }

        let existing = dir_overwrite_users::Entity::find_by_id((dir_id, user_id))
            .one(self.db().read())
            .await?;

        let (final_allow, final_deny) = if let Some(row) = existing {
            let mut am: dir_overwrite_users::ActiveModel = row.into();
            am.allow_mask = Set(cmd.allow_mask);
            am.deny_mask = Set(cmd.deny_mask);
            let updated = am.update(self.db().write()).await?;
            (updated.allow_mask, updated.deny_mask)
        } else {
            let am = dir_overwrite_users::ActiveModel {
                dir_id: Set(dir_id),
                user_id: Set(user_id),
                allow_mask: Set(cmd.allow_mask),
                deny_mask: Set(cmd.deny_mask),
            };
            let inserted = am.insert(self.db().write()).await?;
            (inserted.allow_mask, inserted.deny_mask)
        };

        let user_map: HashMap<i64, UserDto> = self.load_user_briefs(&[user_id]).await?;
        let user_brief = user_map.get(&user_id).cloned();

        Ok(UserOverwriteDto {
            dir_id,
            user_id,
            user: user_brief,
            allow_mask: final_allow,
            deny_mask: final_deny,
        })
    }

    #[tracing::instrument(name = "space.remove_dir_user_overwrite", skip_all, fields(caller = %caller, space = %space_name, dir_id = %dir_id, user_id = %user_id))]
    pub async fn remove_dir_user_overwrite(
        &self,
        caller: i64,
        space_name: &str,
        dir_id: Uuid,
        user_id: i64,
    ) -> Result<(), SpaceError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_perm(self, &space, caller, MANAGE_OVERWRITES).await?;
        self.dir_in_space(space.id, dir_id).await?;

        let res = dir_overwrite_users::Entity::delete_many()
            .filter(dir_overwrite_users::Column::DirId.eq(dir_id))
            .filter(dir_overwrite_users::Column::UserId.eq(user_id))
            .exec(self.db().write())
            .await?;

        if res.rows_affected == 0 {
            return Err(SpaceError::NotFound);
        }
        Ok(())
    }
}