macltok

use std::sync::Arc;

use matrix_sdk::{BoxFuture, Client, SessionChange};

use ruma::OwnedDeviceId;

use tokio::sync::{

    broadcast::{self, error::RecvError},

    mpsc, watch,

};

use super::Logouts;

use crate::{Error, Event, Termination, transport::EventSender};

#[cfg(not(target_family = "wasm"))]

pub(crate) type SaveRecord = dyn Fn(Client) -> BoxFuture<'static, Result<(), Error>> + Send + Sync;

#[cfg(target_family = "wasm")]

pub(crate) type SaveRecord = dyn Fn(Client) -> BoxFuture<'static, Result<(), Error>>;

/// how the engine keeps the saved record current, handed over by `Session::build`

pub(crate) struct Persistence {

    pub(crate) save: Box<SaveRecord>,

    /// what the sdk's own save callback made of each refresh, when the store could write

    /// there and then

    pub(crate) inline: mpsc::UnboundedReceiver<Result<(), Error>>,

}

impl Persistence {

    async fn persist(&mut self, client: &Client) -> Result<(), Error> {

        let mut inline = None;

        while let Ok(result) = self.inline.try_recv() {

            inline = Some(result);

        }

        if let Some(Ok(())) = inline {

            return Ok(());

        }

        // an inline save for a newer refresh can land while this one writes, so go again

        // until what's on disk is what the client holds

        loop {

            let tokens = client.session_tokens();

            (self.save)(client.clone()).await?;

            if client.session_tokens() == tokens {

                return Ok(());

            }

        }

    }

}

pub(super) enum Notice {

    Event(Event),

    Rejected(Option<OwnedDeviceId>),

}

/// saves and notices are split, so a full event stream never holds a save back

pub(super) async fn session(

    client: Client,

    mut changes: broadcast::Receiver<SessionChange>,

    mut persistence: Option<Persistence>,

    notices: mpsc::UnboundedSender<Notice>,

    device_id: OwnedDeviceId,

    logouts: Arc<watch::Sender<Logouts>>,

) {

    loop {

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

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

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

                match persistence.as_mut() {

                    None => Event::SessionTokensChanged,

                    Some(persistence) => match persistence.persist(&client).await {

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

                        Err(error) => Event::SessionSaveFailed {

                            message: error.message,

                        },

                    },

                }

            }

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

                let soft_logout = data.soft_logout.then_some(device_id);

                let mut held = false;

                // our own logout revokes the token, so a rejection meanwhile is its doing

                logouts.send_if_modified(|logouts| {

                    held = logouts.in_flight > 0;

                    if held {

                        logouts.rejected = Some(soft_logout.clone());

                    }

                    held

                });

                if !held {

                    let _ = notices.send(Notice::Rejected(soft_logout));

                }

                return;

            }

            Err(RecvError::Closed) => return,

        };

        if notices.send(Notice::Event(event)).is_err() {

            return;

        }

    }

}

pub(super) async fn session_notices(

    mut notices: mpsc::UnboundedReceiver<Notice>,

    events: EventSender,

    stop: mpsc::UnboundedSender<Termination>,

) {

    while let Some(notice) = notices.recv().await {

        match notice {

            Notice::Event(event) => {

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

                    return;

                }

            }

            Notice::Rejected(soft_logout) => {

                return reject_session(&events, &stop, soft_logout).await;

            }

        }

    }

}

pub(super) async fn reject_session(

    events: &EventSender,

    stop: &mpsc::UnboundedSender<Termination>,

    soft_logout: Option<OwnedDeviceId>,

) {

    let _ = events

        .send(Event::SessionAccessRejected {

            soft_logout: soft_logout.clone(),

        })

        .await;

    let _ = stop.send(match soft_logout {

        Some(device_id) => Termination::SoftLoggedOut { device_id },

        None => Termination::AccessRevoked,

    });

}