//! [`Syncer`](super::Syncer) actor ingress.

use super::Artifact;
use crate::stateful::{
    Application,
    actor::{BlockDigest, SyncTargets},
    db::{Anchor, TipUpdate},
};
use commonware_actor::mailbox::{Overflow, Policy, Sender};
use commonware_utils::channel::oneshot;

pub(crate) enum Message<E, A>
where
    A: Application<E>,
{
    Retarget {
        update: TipUpdate<BlockDigest<A, E>, SyncTargets<A, E>>,
        response: oneshot::Sender<Option<Artifact<E, A>>>,
    },
}

impl<E, A> Overflow<Message<E, A>> for Option<Message<E, A>>
where
    A: Application<E>,
{
    fn is_empty(&self) -> bool {
        self.is_none()
    }

    fn drain<F>(&mut self, mut push: F)
    where
        F: FnMut(Message<E, A>) -> Self,
    {
        if let Some(message) = self.take()
            && let Some(message) = push(message)
        {
            *self = Some(message);
        }
    }
}

// Replacing a queued update drops its response sender. `Mailbox::retarget` has one caller, which
// awaits each update before sending the next, so a live response is never dropped.
impl<E, A> Policy for Message<E, A>
where
    A: Application<E>,
{
    type Overflow = Option<Self>;

    fn handle(overflow: &mut Self::Overflow, message: Self) {
        *overflow = Some(message);
    }
}

/// Ingress mailbox for the [`Syncer`](super::Syncer) actor.
pub struct Mailbox<E, A>
where
    A: Application<E>,
{
    sender: Sender<Message<E, A>>,
}

impl<E, A> Mailbox<E, A>
where
    A: Application<E>,
{
    pub const fn new(sender: Sender<Message<E, A>>) -> Self {
        Self { sender }
    }

    /// Sends a target update and waits until the sync coordinator records it.
    ///
    /// Returns `None` once the update is recorded, or the converged [`Artifact`] if state sync
    /// finished first.
    ///
    /// Panics if the syncer stops without responding.
    pub async fn retarget(
        &self,
        anchor: Anchor<BlockDigest<A, E>>,
        targets: SyncTargets<A, E>,
    ) -> Option<Artifact<E, A>> {
        loop {
            let (update, observed) = TipUpdate::with_observation(anchor, targets.clone());
            let (response, receiver) = oneshot::channel();
            let _ = self.sender.enqueue(Message::Retarget { update, response });

            match receiver.await.expect("Syncer should respond to retarget") {
                Some(artifact) => return Some(artifact),
                None => {
                    // Enqueueing can race with convergence, so wait for the coordinator to record it.
                    if observed.await.is_ok() {
                        return None;
                    }

                    // The update was dropped unrecorded. Retry until it is recorded or the
                    // converged artifact is returned.
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{Mailbox, Message};
    use crate::stateful::{
        actor::syncer::Artifact,
        tests::mocks::{TestApp, anchor, test_databases},
    };
    use commonware_actor::mailbox as actor_mailbox;
    use commonware_runtime::{Runner as _, Supervisor as _, deterministic};
    use commonware_utils::NZUsize;
    use futures::FutureExt;

    #[test]
    fn retarget_retries_when_observation_is_dropped() {
        deterministic::Runner::default().start(|context| async move {
            let (sender, mut receiver) = actor_mailbox::new(context.child("mailbox"), NZUsize!(1));
            let mailbox = Mailbox::<deterministic::Context, TestApp>::new(sender);
            let mut retarget = Box::pin(mailbox.retarget(anchor(7, 9), 7));

            assert!(retarget.as_mut().now_or_never().is_none());

            let Some(Message::Retarget { update, response }) = receiver.recv().await else {
                panic!("first update should be sent");
            };
            assert!(
                response.send(None).is_ok(),
                "response receiver should be alive"
            );
            drop(update);

            assert!(retarget.as_mut().now_or_never().is_none());

            let expected = Artifact::<deterministic::Context, TestApp> {
                databases: test_databases(),
                anchor: anchor(8, 10),
            };
            let Some(Message::Retarget { response, .. }) = receiver.recv().await else {
                panic!("dropped observation should trigger a retry");
            };
            assert!(
                response.send(Some(expected.clone())).is_ok(),
                "response receiver should be alive"
            );

            let result = retarget.await;
            assert_eq!(
                result.expect("retry should return artifact").anchor,
                expected.anchor
            );
        });
    }

    #[test]
    fn retarget_returns_none_only_after_observation_is_recorded() {
        deterministic::Runner::default().start(|context| async move {
            let (sender, mut receiver) = actor_mailbox::new(context.child("mailbox"), NZUsize!(1));
            let mailbox = Mailbox::<deterministic::Context, TestApp>::new(sender);
            let mut retarget = Box::pin(mailbox.retarget(anchor(7, 9), 7));

            assert!(retarget.as_mut().now_or_never().is_none());

            let Some(Message::Retarget { update, response }) = receiver.recv().await else {
                panic!("update should be sent");
            };
            assert!(
                response.send(None).is_ok(),
                "response receiver should be alive"
            );

            assert!(retarget.as_mut().now_or_never().is_none());

            update.record(|_, _| {});

            assert!(retarget.await.is_none());
        });
    }
}