macltok

use std::{

    collections::{HashMap, HashSet, VecDeque},

    sync::Arc,

};

use matrix_sdk::{

    Client, Room,

    ruma::{

        Int,

        api::client::{

            profile::{AvatarUrl, DisplayName},

            room::{

                Visibility as SdkVisibility,

                create_room::v3::{Request as CreateRoomRequest, RoomPreset as SdkPreset},

            },

            state::get_state_event_for_key,

        },

        events::{

            AnyGlobalAccountDataEventContent, GlobalAccountDataEventType, MessageLikeEventType,

            StateEventType,

            ignored_user_list::{IgnoredUser, IgnoredUserListEventContent},

            room::power_levels::RoomPowerLevelsEventContent,

        },

        serde::{JsonObject, Raw},

    },

};

use ruma::{OwnedRoomId, OwnedUserId};

use serde_json::Value as JsonValue;

use tokio::sync::mpsc;

use super::commands::{Job, guarded};

use super::executor::TaskSet;

use crate::{

    CommandOutput, CreateRoomOptions, Error, ErrorKind, PowerLevelChanges, RoomCommand, RoomPreset,

    RoomVisibility, normalise::stored_space_order, space_order,

};

/// state changes that must not race each other. same room, same alias, same dm

/// partner or same account data type run in order, everything else runs at once

pub(super) enum Mutation {

    Room(RoomCommand),

    GlobalAccountData {

        event_type: GlobalAccountDataEventType,

        content: Raw<AnyGlobalAccountDataEventContent>,

    },

    Ignore {

        user_id: OwnedUserId,

        ignored: bool,

    },

    ReorderSpaces(Vec<OwnedRoomId>),

}

pub(super) type MutationJob = Job<Mutation>;

#[derive(Clone, Debug, Eq, Hash, PartialEq)]

enum Key {

    Room(OwnedRoomId),

    Alias(String),

    Direct(OwnedUserId),

    AccountData(String),

    Independent(u64),

}

pub(super) async fn run_scheduler(

    client: Client,

    mut jobs: mpsc::Receiver<MutationJob>,

    queue_limit: usize,

) {

    let fresh = Fresh::default();

    let mut waiting = HashMap::<Key, VecDeque<MutationJob>>::new();

    let mut running = HashSet::<Key>::new();

    let mut tasks = TaskSet::<Key>::new();

    let mut queued = 0_usize;

    let mut independent = 0_u64;

    let mut receiving = true;

    while receiving || !tasks.is_empty() {

        tokio::select! {

            biased;

            Some(finished) = tasks.join_next(), if !tasks.is_empty() => {

                // a job's own panic is caught inside it, so this is only ever a key

                let Ok(key) = finished else { continue };

                let next = waiting.get_mut(&key).and_then(VecDeque::pop_front);

                if waiting.get(&key).is_some_and(VecDeque::is_empty) {

                    waiting.remove(&key);

                }

                match next {

                    Some(job) => {

                        queued -= 1;

                        spawn(&mut tasks, &client, &fresh, key, job);

                    }

                    None => {

                        running.remove(&key);

                    }

                }

            }

            job = jobs.recv(), if receiving => {

                let Some(job) = job else {

                    receiving = false;

                    continue;

                };

                let key = key(&job.command, client.user_id().map(ruma::UserId::server_name)).unwrap_or_else(|| {

                    independent += 1;

                    Key::Independent(independent)

                });

                if running.insert(key.clone()) {

                    spawn(&mut tasks, &client, &fresh, key, job);

                } else if queued < queue_limit {

                    waiting.entry(key).or_default().push_back(job);

                    queued += 1;

                } else {

                    job.reply.done(Err(Error::new(

                        ErrorKind::QueueFull,

                        "room command queue is full",

                    )));

                }

            }

        }

    }

}

/// `server` is ours, since a new alias lives there

fn key(mutation: &Mutation, server: Option<&ruma::ServerName>) -> Option<Key> {

    let room_id = match mutation {

        Mutation::GlobalAccountData { event_type, .. } => {

            return Some(Key::AccountData(event_type.to_string()));

        }

        Mutation::Ignore { .. } => {

            return Some(Key::AccountData(

                GlobalAccountDataEventType::IgnoredUserList.to_string(),

            ));

        }

        Mutation::ReorderSpaces(_) => {

            return Some(Key::AccountData(space_order::EVENT_TYPE.to_owned()));

        }

        Mutation::Room(RoomCommand::Join { room, .. }) => {

            return Some(match OwnedRoomId::try_from(room.clone()) {

                Ok(room_id) => Key::Room(room_id),

                Err(alias) => Key::Alias(alias.to_string()),

            });

        }

        Mutation::Room(RoomCommand::CreateDirect(user_id)) => {

            return Some(Key::Direct(user_id.clone()));

        }

        Mutation::Room(RoomCommand::Create(options)) => {

            return match (&options.alias_localpart, options.invite.as_slice()) {

                (Some(localpart), _) => Some(Key::Alias(format!("#{localpart}:{}", server?))),

                (None, [user_id]) if options.direct => Some(Key::Direct(user_id.clone())),

                _ => None,

            };

        }

        Mutation::Room(

            RoomCommand::Leave(room_id)

            | RoomCommand::Forget(room_id)

            | RoomCommand::AcceptInvite(room_id)

            | RoomCommand::RejectInvite(room_id)

            | RoomCommand::InviteUser { room_id, .. }

            | RoomCommand::KickUser { room_id, .. }

            | RoomCommand::BanUser { room_id, .. }

            | RoomCommand::UnbanUser { room_id, .. }

            | RoomCommand::SetPowerLevels { room_id, .. }

            | RoomCommand::SetUserPowerLevel { room_id, .. }

            | RoomCommand::MuteUser { room_id, .. }

            | RoomCommand::SetFavourite { room_id, .. }

            | RoomCommand::SetOwnDisplayName { room_id, .. }

            | RoomCommand::SetOwnAvatar { room_id, .. }

            | RoomCommand::SetAccountData { room_id, .. },

        ) => room_id,

        Mutation::Room(_) => return None,

    };

    Some(Key::Room(room_id.clone()))

}

/// dms made here that the next sync hasn't told the sdk about yet, since `get_dm_room`

/// only knows `m.direct` once it comes back round

type FreshDms = Arc<std::sync::Mutex<HashMap<OwnedUserId, OwnedRoomId>>>;

/// space orders written here that sync hasn't echoed into the store yet

type FreshOrders = Arc<std::sync::Mutex<HashMap<OwnedRoomId, String>>>;

#[derive(Clone, Default)]

struct Fresh {

    dms: FreshDms,

    orders: FreshOrders,

}

fn spawn(tasks: &mut TaskSet<Key>, client: &Client, fresh: &Fresh, key: Key, job: MutationJob) {

    let client = client.clone();

    let fresh = fresh.clone();

    tasks.spawn(async move {

        let Job { command, reply } = job;

        reply.send(guarded("room command", execute(&client, &fresh, command)).await);

        key

    });

}

async fn execute(

    client: &Client,

    fresh: &Fresh,

    mutation: Mutation,

) -> Result<CommandOutput, Error> {

    let command = match mutation {

        Mutation::GlobalAccountData {

            event_type,

            content,

        } => {

            client

                .account()

                .set_account_data_raw(event_type, content)

                .await

                .map_err(|source| Error::from_sdk("writing account data failed", source))?;

            return Ok(CommandOutput::Done);

        }

        Mutation::Ignore { user_id, ignored } => {

            return set_ignored(client, user_id, ignored).await.map(done);

        }

        Mutation::ReorderSpaces(order) => {

            return reorder_spaces(client, &fresh.orders, order).await.map(done);

        }

        Mutation::Room(command) => command,

    };

    let fresh_dms = &fresh.dms;

    let room = |room_id: &OwnedRoomId| client.get_room(room_id).ok_or_else(Error::room_unknown);

    let joined = |room: Room| CommandOutput::Room(room.room_id().to_owned());

    let sdk = |message: &'static str| move |source| Error::from_sdk(message, source);

    match command {

        RoomCommand::Join { room, via } => client

            .join_room_by_id_or_alias(&room, &via)

            .await

            .map(joined)

            .map_err(sdk("joining room failed")),

        RoomCommand::Leave(room_id) => room(&room_id)?

            .leave()

            .await

            .map(done)

            .map_err(sdk("leaving room failed")),

        RoomCommand::Forget(room_id) => room(&room_id)?

            .forget()

            .await

            .map(done)

            .map_err(sdk("forgetting room failed")),

        RoomCommand::AcceptInvite(room_id) => room(&room_id)?

            .join()

            .await

            .map(done)

            .map_err(sdk("accepting invite failed")),

        RoomCommand::RejectInvite(room_id) => room(&room_id)?

            .leave()

            .await

            .map(done)

            .map_err(sdk("rejecting invite failed")),

        RoomCommand::InviteUser { room_id, user_id } => room(&room_id)?

            .invite_user_by_id(&user_id)

            .await

            .map(done)

            .map_err(sdk("inviting user failed")),

        RoomCommand::KickUser {

            room_id,

            user_id,

            reason,

        } => room(&room_id)?

            .kick_user(&user_id, reason.as_deref())

            .await

            .map(done)

            .map_err(sdk("kicking user failed")),

        RoomCommand::BanUser {

            room_id,

            user_id,

            reason,

        } => room(&room_id)?

            .ban_user(&user_id, reason.as_deref())

            .await

            .map(done)

            .map_err(sdk("banning user failed")),

        RoomCommand::UnbanUser {

            room_id,

            user_id,

            reason,

        } => room(&room_id)?

            .unban_user(&user_id, reason.as_deref())

            .await

            .map(done)

            .map_err(sdk("unbanning user failed")),

        RoomCommand::SetPowerLevels { room_id, changes } => {

            apply_power_level_changes(&room(&room_id)?, changes)

                .await

                .map(done)

        }

        RoomCommand::SetUserPowerLevel {

            room_id,

            user_id,

            level,

        } => room(&room_id)?

            .update_power_levels(vec![(&user_id, power_level(level)?)])

            .await

            .map(done)

            .map_err(sdk("updating user power level failed")),

        RoomCommand::MuteUser {

            room_id,

            user_id,

            muted,

        } => mute_user(client, &room(&room_id)?, &user_id, muted)

            .await

            .map(done),

        RoomCommand::SetFavourite { room_id, favourite } => room(&room_id)?

            .set_is_favourite(favourite, None)

            .await

            .map(done)

            .map_err(sdk("setting favourite failed")),

        RoomCommand::SetOwnDisplayName { room_id, name } => {

            set_own_member_field(client, &room(&room_id)?, ProfileField::DisplayName, name)

                .await

                .map(done)

        }

        RoomCommand::SetOwnAvatar { room_id, avatar } => set_own_member_field(

            client,

            &room(&room_id)?,

            ProfileField::Avatar,

            avatar.map(String::from),

        )

        .await

        .map(done),

        RoomCommand::SetAccountData {

            room_id,

            event_type,

            content,

        } => room(&room_id)?

            .set_account_data_raw(event_type, content)

            .await

            .map(done)

            .map_err(sdk("writing room account data failed")),

        RoomCommand::Create(options) => client

            .create_room(create_room_request(options))

            .await

            .map(joined)

            .map_err(sdk("creating room failed")),

        RoomCommand::CreateDirect(user_id) => {

            if let Some(existing) = client.get_dm_room(&user_id) {

                return Ok(joined(existing));

            }

            let fresh = fresh_dms

                .lock()

                .expect("nothing panics holding this")

                .get(&user_id)

                .cloned();

            if let Some(room_id) = fresh.filter(|room_id| client.get_room(room_id).is_some()) {

                return Ok(CommandOutput::Room(room_id));

            }

            let room = client

                .create_dm(&user_id)

                .await

                .map_err(sdk("creating direct room failed"))?;

            fresh_dms

                .lock()

                .expect("nothing panics holding this")

                .insert(user_id, room.room_id().to_owned());

            Ok(joined(room))

        }

        _ => Err(Error::new(ErrorKind::Other, "room command was misrouted")),

    }

}

/// one state event carrying every requested change, since the sdk's own patcher is private

async fn apply_power_level_changes(room: &Room, changes: PowerLevelChanges) -> Result<(), Error> {

    let mut levels = room

        .power_levels()

        .await

        .map_err(|source| Error::from_sdk("reading power levels failed", source.into()))?;

    for (target, value) in [

        (&mut levels.ban, changes.ban),

        (&mut levels.invite, changes.invite),

        (&mut levels.kick, changes.kick),

        (&mut levels.redact, changes.redact),

        (&mut levels.events_default, changes.events_default),

        (&mut levels.state_default, changes.state_default),

        (&mut levels.users_default, changes.users_default),

        (&mut levels.notifications.room, changes.room_notification),

    ] {

        if let Some(value) = value {

            *target = power_level(value)?;

        }

    }

    for (event_type, value) in [

        (StateEventType::RoomName.into(), changes.room_name),

        (StateEventType::RoomAvatar.into(), changes.room_avatar),

        (StateEventType::RoomTopic.into(), changes.room_topic),

        (StateEventType::SpaceChild.into(), changes.space_child),

        (MessageLikeEventType::Beacon.into(), changes.beacon),

        (StateEventType::BeaconInfo.into(), changes.beacon_info),

    ] {

        if let Some(value) = value {

            levels.events.insert(event_type, power_level(value)?);

        }

    }

    let content = RoomPowerLevelsEventContent::try_from(levels).map_err(|source| {

        Error::with_source(

            ErrorKind::Invalid,

            "normalising power levels failed",

            source,

        )

    })?;

    room.send_state_event(content)

        .await

        .map(drop)

        .map_err(|source| Error::from_sdk("updating power levels failed", source))

}

/// reads the list from the server, since the store only catches up on the next sync

async fn set_ignored(client: &Client, user_id: OwnedUserId, ignored: bool) -> Result<(), Error> {

    if ignored && client.user_id() == Some(&*user_id) {

        return Err(Error::new(ErrorKind::Invalid, "you can't ignore yourself"));

    }

    let account = client.account();

    let mut list = account

        .fetch_account_data_static::<IgnoredUserListEventContent>()

        .await

        .map_err(|source| Error::from_sdk("reading the ignore list failed", source))?

        .map(|raw| raw.deserialize())

        .transpose()

        .map_err(|source| {

            Error::with_source(ErrorKind::Invalid, "the ignore list is malformed", source)

        })?

        .unwrap_or_default();

    let changed = if ignored {

        list.ignored_users

            .insert(user_id, IgnoredUser::new())

            .is_none()

    } else {

        list.ignored_users.remove(&user_id).is_some()

    };

    if !changed {

        return Ok(());

    }

    account

        .set_account_data(list)

        .await

        .map(drop)

        .map_err(|source| Error::from_sdk("writing the ignore list failed", source))

}

async fn reorder_spaces(

    client: &Client,

    fresh: &FreshOrders,

    order: Vec<OwnedRoomId>,

) -> Result<(), Error> {

    if order.iter().collect::<HashSet<_>>().len() != order.len() {

        return Err(Error::new(ErrorKind::Invalid, "a space is listed twice"));

    }

    let mut rooms = HashMap::new();

    let mut spaces = Vec::with_capacity(order.len());

    for room_id in order {

        let room = client.get_room(&room_id).ok_or_else(Error::room_unknown)?;

        let stored = stored_space_order(&room).await;

        let current = {

            let mut fresh = fresh.lock().expect("nothing panics holding this");

            match fresh.get(&room_id) {

                Some(written) if stored.as_ref() != Some(written) => Some(written.clone()),

                Some(_) => fresh.remove(&room_id),

                None => stored,

            }

        };

        rooms.insert(room_id.clone(), room);

        spaces.push((room_id, current));

    }

    for (room_id, order) in space_order::plan(&spaces) {

        let content = Raw::new(&serde_json::json!({ "order": order }))

            .expect("a string map always serialises")

            .cast_unchecked();

        rooms[&room_id]