macltok

use std::{future::Future, panic::AssertUnwindSafe};

use futures_util::FutureExt;

use matrix_sdk::encryption::verification::VerificationRequest;

use matrix_sdk::{

    Client,

    notification_settings::{IsEncrypted, IsOneToOne, RoomNotificationMode as SdkMode},

    ruma::{

        OwnedDeviceId, OwnedEventId, OwnedMxcUri, RoomId, TransactionId,

        api::client::receipt::create_receipt::v3::ReceiptType, events::receipt::ReceiptThread,

    },

    send_queue::{LocalEchoContent, SendHandle},

};

use tokio::sync::mpsc;

use crate::{

    EncryptionCommand, Error, ErrorKind, InteractiveAuth, InteractiveAuthChallenge,

    NotificationCommand, RoomNotificationMode, TimelineId,

    encryption::VerificationController,

    transport::{EventSender, Responder},

};

pub(super) struct Job<T> {

    pub(super) command: T,

    pub(super) reply: Responder,

}

/// a panicking command answers with an error instead of taking its worker down with it

pub(super) async fn guarded<T>(

    label: &'static str,

    work: impl Future<Output = Result<T, Error>>,

) -> Result<T, Error> {

    AssertUnwindSafe(work)

        .catch_unwind()

        .await

        .unwrap_or_else(|_| Err(Error::new(ErrorKind::Other, format!("{label} panicked"))))

}

pub(super) async fn run_notifications(

    client: Client,

    mut jobs: mpsc::Receiver<Job<NotificationCommand>>,

) {

    while let Some(Job { command, reply }) = jobs.recv().await {

        reply.done(guarded("notification command", notification(&client, command)).await);

    }

}

pub(super) async fn run_encryption(

    client: Client,

    events: EventSender,

    mut jobs: mpsc::Receiver<Job<EncryptionCommand>>,

    mut incoming: mpsc::UnboundedReceiver<VerificationRequest>,

) {

    let mut verification = VerificationController::default();

    loop {

        tokio::select! {

            job = jobs.recv() => {

                let Some(Job { command, reply }) = job else { return };

                let work = encryption(&client, &events, &mut verification, command);

                reply.done(guarded("encryption command", work).await);

            }

            Some(request) = incoming.recv() => {

                verification.adopt_incoming(events.clone(), request).await;

            }

        }

    }

}

async fn encryption(

    client: &Client,

    events: &EventSender,

    verification: &mut VerificationController,

    command: EncryptionCommand,

) -> Result<(), Error> {

    let encryption = client.encryption();

    match command {

        EncryptionCommand::Recover(key) => encryption

            .recovery()

            .recover(&key)

            .await

            .map_err(|source| Error::from_recovery("account recovery failed", source)),

        EncryptionCommand::RecoverKeyBackup(key) => encryption

            .recovery()

            .recover_and_fix_backup(&key)

            .await

            .map_err(|source| Error::from_recovery("recovering key backup failed", source)),

        EncryptionCommand::CreateKeyBackup => encryption

            .backups()

            .create()

            .await

            .map_err(|source| Error::from_sdk("creating key backup failed", source)),

        EncryptionCommand::EnableKeyBackup => encryption

            .recovery()

            .enable_backup()

            .await

            .map_err(|source| Error::from_recovery("enabling key backup failed", source)),

        EncryptionCommand::DisableKeyBackup { delete_from_server } => {

            let backups = encryption.backups();

            if delete_from_server {

                backups.disable_and_delete().await

            } else {

                backups.disable().await

            }

            .map_err(|source| Error::from_sdk("disabling key backup failed", source))

        }

        EncryptionCommand::StartVerification(target) => {

            verification.start(client, events.clone(), target).await

        }

        EncryptionCommand::Verify(action) => verification.apply(events.clone(), action).await,

    }

}

async fn notification(client: &Client, command: NotificationCommand) -> Result<(), Error> {

    let settings = client.notification_settings().await;

    match command {

        NotificationCommand::SetRoomMode { room_id, mode } => {

            settings

                .set_room_notification_mode(&room_id, sdk_mode(mode))

                .await

        }

        NotificationCommand::ResetRoomMode(room_id) => {

            settings.delete_user_defined_room_rules(&room_id).await

        }

        NotificationCommand::SetDefaultMode {

            encrypted,

            one_to_one,

            mode,

        } => {

            settings

                .set_default_room_notification_mode(

                    IsEncrypted::from(encrypted),

                    IsOneToOne::from(one_to_one),

                    sdk_mode(mode),

                )

                .await

        }

        NotificationCommand::AddKeyword(keyword) => settings.add_keyword(keyword).await,

        NotificationCommand::RemoveKeyword(keyword) => settings.remove_keyword(&keyword).await,

    }

    .map_err(|source| {

        Error::with_source(

            ErrorKind::Server,

            "updating notification settings failed",

            source,

        )

    })

}

const fn sdk_mode(mode: RoomNotificationMode) -> SdkMode {

    match mode {

        RoomNotificationMode::AllMessages => SdkMode::AllMessages,

        RoomNotificationMode::MentionsAndKeywordsOnly => SdkMode::MentionsAndKeywordsOnly,

        RoomNotificationMode::Mute => SdkMode::Mute,

    }

}

fn room(client: &Client, room_id: &RoomId) -> Result<matrix_sdk::Room, Error> {

    client.get_room(room_id).ok_or_else(Error::room_unknown)

}

/// the closed-room path. without a timeline nothing stops the marker going backwards

pub(super) async fn mark_read(

    client: &Client,

    timeline_id: &TimelineId,

    event_id: OwnedEventId,

) -> Result<(), Error> {

    let room = room(client, timeline_id.room_id())?;

    let receipt_failed = |source| Error::from_sdk("sending read receipt failed", source);

    let thread = match timeline_id.thread_root() {

        None => ReceiptThread::Main,

        Some(root) => ReceiptThread::Thread(root.clone()),

    };

    room.send_single_receipt(ReceiptType::Read, thread, event_id.clone())

        .await

        .map_err(receipt_failed)?;

    if timeline_id.thread_root().is_none() {

        room.send_single_receipt(ReceiptType::FullyRead, ReceiptThread::Unthreaded, event_id)

            .await

            .map_err(receipt_failed)?;

    }

    Ok(())

}

pub(super) async fn set_typing(

    client: &Client,

    room_id: &RoomId,

    typing: bool,

) -> Result<(), Error> {

    room(client, room_id)?

        .typing_notice(typing)

        .await

        .map_err(|source| Error::from_sdk("typing update failed", source))

}

async fn send_handle(

    client: &Client,

    room_id: &RoomId,

    transaction_id: &TransactionId,

) -> Result<SendHandle, Error> {

    let (echoes, _) = room(client, room_id)?

        .send_queue()

        .subscribe()

        .await

        .map_err(|source| {

            Error::with_source(ErrorKind::Store, "reading send queue failed", source)

        })?;

    echoes

        .into_iter()

        .find_map(|echo| match echo.content {

            LocalEchoContent::Event { send_handle, .. }

                if echo.transaction_id.as_str() == transaction_id.as_str() =>

            {

                Some(send_handle)

            }

            _ => None,

        })

        .ok_or_else(|| {

            Error::new(

                ErrorKind::Invalid,

                "send is no longer pending or is not a message",

            )

        })

}

pub(super) async fn cancel_send(

    client: &Client,

    room_id: &RoomId,

    transaction_id: &TransactionId,

) -> Result<(), Error> {

    match send_handle(client, room_id, transaction_id)

        .await?

        .abort()

        .await

    {

        Ok(true) => Ok(()),

        Ok(false) => Err(Error::new(

            ErrorKind::Invalid,

            "send could not be cancelled because it has already been sent",

        )),

        Err(source) => Err(Error::with_source(

            ErrorKind::Store,

            "cancelling send failed",

            source,

        )),

    }

}

pub(super) async fn retry_send(

    client: &Client,

    room_id: &RoomId,

    transaction_id: &TransactionId,

) -> Result<(), Error> {

    send_handle(client, room_id, transaction_id)

        .await?

        .unwedge()

        .await

        .map_err(|source| Error::with_source(ErrorKind::Store, "retrying send failed", source))?;

    room(client, room_id)?.send_queue().set_enabled(true);

    Ok(())

}

pub(super) async fn upload_avatar(

    client: &Client,

    mimetype: &str,

    bytes: Vec<u8>,

    max_bytes: usize,

) -> Result<(), Error> {

    if bytes.len() > max_bytes {

        return Err(Error::new(

            ErrorKind::MediaTooLarge,

            "avatar exceeds the configured byte limit",

        ));

    }

    let mimetype: mime::Mime = mimetype

        .parse()

        .map_err(|_| Error::new(ErrorKind::Invalid, "avatar MIME type is invalid"))?;

    if mimetype.type_() != mime::IMAGE {

        return Err(Error::new(

            ErrorKind::Invalid,

            "an avatar has to be an image",

        ));

    }

    client

        .account()

        .upload_avatar(&mimetype, bytes)

        .await

        .map(drop)

        .map_err(|source| Error::from_sdk("uploading avatar failed", source))

}

pub(super) async fn upload_media(

    client: &Client,

    mimetype: &str,

    bytes: Vec<u8>,

    max_bytes: usize,

) -> Result<OwnedMxcUri, Error> {

    if bytes.len() > max_bytes {

        return Err(Error::new(

            ErrorKind::MediaTooLarge,

            "media exceeds the configured byte limit",

        ));

    }

    let mimetype: mime::Mime = mimetype

        .parse()

        .map_err(|_| Error::new(ErrorKind::Invalid, "media MIME type is invalid"))?;

    client

        .media()

        .upload(&mimetype, bytes, None)

        .await

        .map(|response| response.content_uri)

        .map_err(|source| Error::from_sdk("uploading media failed", source))

}

pub(super) async fn rename_device(

    client: &Client,

    device_id: &OwnedDeviceId,

    display_name: &str,

) -> Result<(), Error> {

    client

        .rename_device(device_id, display_name)

        .await

        .map(drop)

        .map_err(|source| Error::from_http("renaming device failed", source))

}

pub(super) async fn delete_devices(

    client: &Client,

    device_ids: &[OwnedDeviceId],

    auth: Option<InteractiveAuth>,

) -> Result<(), Error> {

    let user_id = client

        .user_id()

        .ok_or_else(|| Error::new(ErrorKind::Unauthenticated, "the client is not logged in"))?;

    let auth = auth

        .map(|auth| auth.into_sdk(user_id))

        .transpose()

        .map_err(|message| {

            Error::new(

                ErrorKind::Invalid,

                format!("interactive authentication data is invalid, {message}"),

            )

        })?;

    match client.delete_devices(device_ids, auth).await {

        Ok(_) => Ok(()),

        Err(source) => match source.as_uiaa_response() {

            Some(info) => Err(Error::interactive_auth(InteractiveAuthChallenge::from(

                info,

            ))),

            None => Err(Error::from_http("deleting devices failed", source)),

        },

    }

}