macltok

use std::{sync::Arc, time::Duration};

use futures_util::StreamExt;

use matrix_sdk::{

    Client, Room, SessionChange,

    ruma::events::{AnyMessageLikeEventContent, room::message::MessageType},

    send_queue::{LocalEchoContent, RoomSendQueueUpdate},

};

use matrix_sdk_ui::sync_service::{State as SyncState, SyncService};

use tokio::{

    sync::{Notify, broadcast::error::RecvError, watch},

    task::JoinSet,

};

use crate::{

    ConnectionState, Event, encryption::encryption_state, error::classify_sdk_error,

    transport::EventSender,

};

const SYNC_RESTART_MIN: Duration = Duration::from_secs(1);

const SYNC_RESTART_MAX: Duration = Duration::from_secs(60);

const SEND_QUEUE_RETRY: Duration = Duration::from_secs(5);

pub(super) async fn session(

    mut changes: tokio::sync::broadcast::Receiver<SessionChange>,

    events: EventSender,

    stop: Arc<Notify>,

) {

    loop {

        let event = match changes.recv().await {

            Ok(SessionChange::TokensRefreshed) => Event::SessionTokensChanged,

            Ok(SessionChange::UnknownToken(data)) => {

                let _ = events

                    .send(Event::SessionAccessRejected {

                        soft_logout: data.soft_logout,

                    })

                    .await;

                stop.notify_one();

                return;

            }

            // a missed refresh costs one needless save, a missed rejection comes round again

            Err(RecvError::Lagged(_)) => Event::SessionTokensChanged,

            Err(RecvError::Closed) => return,

        };

        if events.send(event).await.is_err() {

            return;

        }

    }

}

/// `states` must be taken before sync starts, a late subscriber never hears the transition it missed

pub(super) async fn connection(

    mut states: impl futures_util::Stream<Item = SyncState> + Unpin,

    sync: Arc<SyncService>,

    client: Client,

    events: EventSender,

    connection: watch::Sender<ConnectionState>,

) {

    let mut backoff = SYNC_RESTART_MIN;

    while let Some(state) = states.next().await {

        let (state, restart) = match state {

            SyncState::Idle => (ConnectionState::Idle, false),

            SyncState::Running => {

                backoff = SYNC_RESTART_MIN;

                // the sdk parks a room queue on any send error and never unparks it

                client.send_queue().set_enabled(true).await;

                (ConnectionState::Online, false)

            }

            SyncState::Offline => (ConnectionState::Offline, false),

            SyncState::Terminated => (ConnectionState::Offline, true),

            SyncState::Error(error) => {

                tracing::warn!(%error, "sync service reported an error");

                (ConnectionState::Offline, true)

            }

        };

        if *connection.borrow() != state {

            connection.send_replace(state);

            if events.send(Event::ConnectionChanged(state)).await.is_err() {

                return;

            }

        }

        if restart {

            tokio::time::sleep(backoff).await;

            backoff = (backoff * 2).min(SYNC_RESTART_MAX);

            sync.start().await;

        }

    }

}

/// a recoverable failure while we're online is a hiccup, so give the room another go

pub(super) async fn send_queue_errors(

    client: Client,

    mut errors: tokio::sync::broadcast::Receiver<matrix_sdk::send_queue::SendQueueRoomError>,

    online: watch::Receiver<ConnectionState>,

) {

    let mut retries = JoinSet::new();

    loop {

        let error = tokio::select! {

            error = errors.recv() => error,

            Some(_) = retries.join_next(), if !retries.is_empty() => continue,

        };

        match error {

            Ok(error) if error.is_recoverable => {

                let client = client.clone();

                let online = online.clone();

                retries.spawn(async move {

                    tokio::time::sleep(SEND_QUEUE_RETRY).await;

                    if *online.borrow() == ConnectionState::Online

                        && let Some(room) = client.get_room(&error.room_id)

                    {

                        room.send_queue().set_enabled(true);

                    }

                });

            }

            Ok(_) | Err(RecvError::Lagged(_)) => {}

            Err(RecvError::Closed) => return,

        }

    }

}

/// upload progress only goes out on the room's own channel, never the client-wide one

pub(super) async fn room_uploads(room: Room, events: EventSender, ready: watch::Sender<bool>) {

    let room_id = room.room_id().to_owned();

    let mut updates = match room.send_queue().subscribe().await {

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

        Err(error) => {

            tracing::warn!(%room_id, %error, "reading the media send queue failed");

            return;

        }

    };

    ready.send_replace(true);

    loop {

        let update = match updates.recv().await {

            Ok(update) => update,

            Err(RecvError::Lagged(skipped)) => {

                tracing::warn!(%room_id, skipped, "upload watcher lagged");

                continue;

            }

            Err(RecvError::Closed) => return,

        };

        let event = match update {

            RoomSendQueueUpdate::MediaUpload {

                related_to,

                file,

                progress,

                ..

            } => Event::MediaUploadProgress {

                room_id: room_id.clone(),

                transaction_id: related_to,

                uploaded_bytes: progress.current as u64,

                total_bytes: progress.total as u64,

                complete: file.is_some() && progress.current == progress.total,

            },

            RoomSendQueueUpdate::SendError {

                transaction_id,

                error,

                is_recoverable,

            } => {

                let is_attachment = room

                    .send_queue()

                    .subscribe()

                    .await

                    .is_ok_and(|(echoes, _)| {

                        echoes.iter().any(|echo| {

                            echo.transaction_id == transaction_id

                                && is_attachment_echo(&echo.content)

                        })

                    });

                if !is_attachment {

                    continue;

                }

                Event::MediaUploadFailed {

                    room_id: room_id.clone(),

                    transaction_id,

                    kind: classify_sdk_error(&error),

                    recoverable: is_recoverable,

                }

            }

            _ => continue,

        };

        if events.send(event).await.is_err() {

            return;

        }

    }

}

fn is_attachment_echo(content: &LocalEchoContent) -> bool {

    let LocalEchoContent::Event {

        serialized_event, ..

    } = content

    else {

        return false;

    };

    matches!(

        serialized_event.deserialize(),

        Ok(AnyMessageLikeEventContent::RoomMessage(content))

            if matches!(

                content.msgtype,

                MessageType::Audio(_) | MessageType::File(_) | MessageType::Image(_) | MessageType::Video(_)

            )

    )

}

pub(super) async fn encryption(client: Client, events: EventSender) {

    let Some(device_id) = client.device_id().map(ToOwned::to_owned) else {

        tracing::warn!("encryption watcher stopped because the client has no device id");

        return;

    };

    let recovery = client.encryption().recovery();

    let backups = client.encryption().backups();

    let mut recovery_stream = recovery.state_stream();

    let mut backup_stream = backups.state_stream();

    let mut verification_stream = client.encryption().verification_state();

    let mut recovery_state = recovery.state();

    let mut backup_state = backups.state();

    let mut verification_state = verification_stream.get();

    loop {

        let state = encryption_state(

            device_id.clone(),

            recovery_state,

            verification_state,

            backup_state,

        );

        if events.send(Event::EncryptionChanged(state)).await.is_err() {

            return;

        }

        tokio::select! {

            Some(update) = recovery_stream.next() => recovery_state = update,

            Some(update) = verification_stream.next() => verification_state = update,

            Some(update) = backup_stream.next() => {

                backup_state = update.unwrap_or_else(|_| backups.state());

            }

            else => return,

        }

    }

}

pub(super) async fn rooms(client: Client, events: EventSender) {

    let (_, mut rooms) = client.rooms_stream();

    let mut info = client.room_info_notable_update_receiver();

    loop {

        if events.send(Event::RoomsChanged).await.is_err() {

            return;

        }

        tokio::select! {

            update = rooms.next() => if update.is_none() { return },

            update = info.recv() => if matches!(update, Err(RecvError::Closed)) { return },

        }

        // a sync touching fifty rooms is one invalidation, not fifty

        while let Ok(_) | Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) =

            info.try_recv()

        {}

    }

}

pub(super) async fn notification_settings(client: Client, events: EventSender) {

    let settings = client.notification_settings().await;

    let mut changes = settings.subscribe_to_changes();

    while let Ok(()) | Err(RecvError::Lagged(_)) = changes.recv().await {

        if events

            .send(Event::NotificationSettingsChanged)

            .await

            .is_err()

        {

            return;

        }

    }

}