macltok

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

use futures_util::StreamExt;

use matrix_sdk::{

    Client, Room,

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

    send_queue::{LocalEchoContent, RoomSendQueueUpdate, SendQueueUpdate},

};

use matrix_sdk_base::RoomInfoNotableUpdateReasons;

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

use ruma::{

    OwnedRoomId, OwnedTransactionId,

    events::{AnyGlobalAccountDataEvent, AnySyncStateEvent},

    serde::Raw,

    time::Instant,

};

use tokio::sync::{Notify, broadcast::error::RecvError, mpsc, watch};

use super::executor::TaskSet;

use crate::{

    ConnectionState, Event, UploadOutcome,

    encryption::encryption_state,

    error::classify_sdk_error,

    normalise::{ENABLED_ROOMS, ROOM_PACKS, USER_PACK},

    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);

const SYNC_STABLE: Duration = Duration::from_secs(30);

pub(super) async fn room_subscriptions(

    sync: Arc<SyncService>,

    mut rooms: watch::Receiver<Vec<ruma::OwnedRoomId>>,

) {

    while rooms.changed().await.is_ok() {

        let room_ids = rooms.borrow_and_update().clone();

        let room_ids = room_ids.iter().map(AsRef::as_ref).collect::<Vec<_>>();

        sync.room_list_service().subscribe_to_rooms(&room_ids).await;

    }

}

/// `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;

    let mut online_since: Option<Instant> = None;

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

        // only a run that lasted counts as recovered, or a server that answers /versions

        // but refuses sync would have us bounce between offline and online forever

        let stable = online_since

            .take()

            .map(|since| since.elapsed() >= SYNC_STABLE);

        if stable == Some(true) {

            backoff = SYNC_RESTART_MIN;

        }

        let (state, restart) = match state {

            // only ever us stopping it for a restart, nothing a client needs to hear

            SyncState::Idle => continue,

            SyncState::Running => {

                online_since = Some(Instant::now());

                // 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, stable == Some(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 {

            sync.stop().await;

            matrix_sdk::sleep::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 = TaskSet::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 {

                    matrix_sdk::sleep::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.

/// the room's sends and cancels come along behind it, so an upload's end can't

/// overtake its own progress

pub(super) async fn room_uploads(

    room: Room,

    tracker: mpsc::Sender<RoomUpload>,

    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 report = match updates.recv().await {

            Ok(RoomSendQueueUpdate::MediaUpload {

                related_to,

                file,

                progress,

                ..

            }) => RoomUpload::Progress {

                upload: (room_id.clone(), related_to),

                uploaded_bytes: progress.current as u64,

                total_bytes: progress.total as u64,

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

            },

            Ok(RoomSendQueueUpdate::SentEvent {

                transaction_id,

                event_id,

            }) => RoomUpload::Ended(

                (room_id.clone(), transaction_id),

                UploadOutcome::Sent { event_id },

            ),

            Ok(RoomSendQueueUpdate::CancelledLocalEvent { transaction_id }) => {

                RoomUpload::Ended((room_id.clone(), transaction_id), UploadOutcome::Cancelled)

            }

            Ok(_) => continue,

            Err(RecvError::Lagged(_)) => RoomUpload::Lagged,

            Err(RecvError::Closed) => return,

        };

        if tracker.send(report).await.is_err() {

            return;

        }

    }

}

pub(super) enum RoomUpload {

    Progress {

        upload: Upload,

        uploaded_bytes: u64,

        total_bytes: u64,

        complete: bool,

    },

    Ended(Upload, UploadOutcome),

    /// an end may have been lost with the ticks

    Lagged,

}

/// the one place upload events leave from, so an upload is never reported after it

/// ended. progress comes in from the room forwarders, the rest from the client-wide

/// channel, which exists before any upload does

pub(super) async fn uploads(

    client: Client,

    mut updates: tokio::sync::broadcast::Receiver<SendQueueUpdate>,

    mut rooms_reports: mpsc::Receiver<RoomUpload>,

    events: EventSender,

) {

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

    let mut tracker = UploadTracker {

        live: attachment_echoes(&client).await,

        ended: HashSet::new(),

    };

    loop {

        let reports = tokio::select! {

            update = updates.recv() => match update {

                Ok(update) => tracker.update(update).into_iter().collect(),

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

                    tracing::warn!(skipped, "upload tracker lagged");

                    tracker.resnapshot(attachment_echoes(&client).await)

                }

                Err(RecvError::Closed) => return,

            },

            Some(report) = rooms_reports.recv() => match report {

                RoomUpload::Lagged => tracker.resnapshot(attachment_echoes(&client).await),

                report => tracker.room(report).into_iter().collect(),

            },

            update = rooms.next() => {

                if update.is_none() {

                    return;

                }

                tracker.finish_where(

                    |(room_id, _)| client.get_room(room_id).is_none(),

                    &UploadOutcome::RoomForgotten,

                )

            }

        };

        for event in reports {

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

                return;

            }

        }

    }

}

type Upload = (OwnedRoomId, OwnedTransactionId);

/// everything reported gets exactly one `MediaUploadFinished`. an upload ended from

/// anywhere but its room's forwarder goes in `ended` until the forwarder's own end for

/// it arrives, since until then the forwarder can still be holding progress for it

struct UploadTracker {

    live: HashSet<Upload>,

    ended: HashSet<Upload>,

}

impl UploadTracker {

    fn update(&mut self, update: SendQueueUpdate) -> Option<Event> {

        let room_id = update.room_id;

        let (transaction_id, outcome) = match update.update {

            RoomSendQueueUpdate::NewLocalEvent(echo) => {

                let upload = (room_id, echo.transaction_id);

                if is_attachment_echo(&echo.content) && !self.ended.contains(&upload) {

                    self.live.insert(upload);

                }

                return None;

            }

            RoomSendQueueUpdate::SendError {

                transaction_id,

                error,

                is_recoverable,

            } => {

                let upload = (room_id, transaction_id);

                if !self.live.contains(&upload) {

                    return None;

                }

                let (room_id, transaction_id) = upload;

                return Some(Event::MediaUploadFailed {

                    room_id,

                    transaction_id,

                    kind: classify_sdk_error(&error),

                    will_retry: is_recoverable,

                });

            }

            RoomSendQueueUpdate::SentEvent {

                transaction_id,

                event_id,

            } => (transaction_id, UploadOutcome::Sent { event_id }),

            RoomSendQueueUpdate::CancelledLocalEvent { transaction_id } => {

                (transaction_id, UploadOutcome::Cancelled)

            }

            _ => return None,

        };

        self.finish((room_id, transaction_id), outcome, true)

    }

    fn room(&mut self, report: RoomUpload) -> Option<Event> {

        match report {

            RoomUpload::Progress {

                upload,

                uploaded_bytes,

                total_bytes,

                complete,

            } => {

                if self.ended.contains(&upload) {

                    return None;

                }

                self.live.insert(upload.clone());

                let (room_id, transaction_id) = upload;

                Some(Event::MediaUploadProgress {

                    room_id,

                    transaction_id,

                    uploaded_bytes,

                    total_bytes,

                    complete,

                })

            }

            RoomUpload::Ended(upload, outcome) => {

                self.ended.remove(&upload);

                self.finish(upload, outcome, false)

            }

            RoomUpload::Lagged => None,

        }

    }

    fn resnapshot(&mut self, current: HashSet<Upload>) -> Vec<Event> {

        let reports = self.finish_where(|upload| !current.contains(upload), &UploadOutcome::Missed);

        let ended = &self.ended;

        self.live

            .extend(current.into_iter().filter(|upload| !ended.contains(upload)));

        reports

    }

    fn finish_where(

        &mut self,

        ends: impl Fn(&Upload) -> bool,

        outcome: &UploadOutcome,

    ) -> Vec<Event> {

        let ending = self

            .live

            .iter()

            .filter(|upload| ends(upload))

            .cloned()

            .collect::<Vec<_>>();

        ending

            .into_iter()

            .filter_map(|upload| self.finish(upload, outcome.clone(), true))

            .collect()

    }

    fn finish(&mut self, upload: Upload, outcome: UploadOutcome, remember: bool) -> Option<Event> {

        if !self.live.remove(&upload) {

            return None;

        }

        if remember {

            self.ended.insert(upload.clone());

        }

        let (room_id, transaction_id) = upload;

        Some(Event::MediaUploadFinished {

            room_id,

            transaction_id,

            outcome,

        })

    }

}

async fn attachment_echoes(client: &Client) -> HashSet<Upload> {

    let Ok(echoes) = client.send_queue().local_echoes().await else {

        return HashSet::new();

    };

    echoes

        .into_iter()

        .flat_map(|(room_id, echoes)| {

            echoes

                .into_iter()

                .filter(|echo| is_attachment_echo(&echo.content))

                .map(move |echo| (room_id.clone(), echo.transaction_id))

        })

        .collect()

}

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()

        {}

    }

}

/// packs live in room state and account data, and leaving a room takes its packs with it

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

    let changed = Arc::new(Notify::new());

    let handlers = [

        client.add_event_handler({

            let changed = changed.clone();

            move |event: Raw<AnySyncStateEvent>| {

                let changed = changed.clone();

                async move {

                    if event_type(&event).as_deref() == Some(ROOM_PACKS) {

                        changed.notify_one();

                    }

                }

            }

        }),

        client.add_event_handler({

            let changed = changed.clone();

            move |event: Raw<AnyGlobalAccountDataEvent>| {

                let changed = changed.clone();

                async move {

                    if matches!(

                        event_type(&event).as_deref(),

                        Some(USER_PACK | ENABLED_ROOMS)

                    ) {

                        changed.notify_one();

                    }

                }

            }

        }),

    ];

    let mut info = client.room_info_notable_update_receiver();

    loop {

        tokio::select! {

            () = changed.notified() => {}

            update = info.recv() => match update {

                Ok(update) if !update.reasons.contains(RoomInfoNotableUpdateReasons::MEMBERSHIP) => {

                    continue;

                }

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

                Err(RecvError::Closed) => break,

            },

        }

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

            break;

        }

    }

    for handler in handlers {

        client.remove_event_handler(handler);

    }

}